func()

in pkg/sidecar_mounter/sidecar_mounter.go [63:155]

• 70 lines of code
• 15 McCabe index (conditional complexity)

func (m *Mounter) Mount(ctx context.Context, mc *MountConfig) error {
	// Start the token server for HostNetwork enabled pods.
	if mc.TokenServerIdentityProvider != "" {
		tp := filepath.Join(mc.TempDir, TokenFileName)
		klog.Infof("Pod has hostNetwork enabled and token server feature is turned on. Starting Token Server on %s.", tp)
		go StartTokenServer(ctx, tp, mc.TokenServerIdentityProvider)
	}

	klog.Infof("start to mount bucket %q for volume %q", mc.BucketName, mc.VolumeName)

	if err := os.MkdirAll(mc.BufferDir+TempDir, os.ModePerm); err != nil {
		return fmt.Errorf("failed to create temp dir %q: %w", mc.BufferDir+TempDir, err)
	}

	args := []string{}
	for k, v := range mc.FlagMap {
		args = append(args, "--"+k)
		if v != "" {
			args = append(args, v)
		}
	}

	args = append(args, mc.BucketName)
	// gcsfuse supports the `/dev/fd/N` syntax
	// the /dev/fuse is passed as ExtraFiles below, and will always be FD 3
	args = append(args, "/dev/fd/3")

	klog.Infof("gcsfuse mounting with args %v...", args)
	//nolint: gosec
	cmd := exec.CommandContext(ctx, m.mounterPath, args...)
	cmd.ExtraFiles = []*os.File{os.NewFile(uintptr(mc.FileDescriptor), "/dev/fuse")}
	cmd.Stdout = os.Stdout
	cmd.Stderr = io.MultiWriter(os.Stderr, mc.ErrWriter)
	cmd.Cancel = func() error {
		klog.V(4).Infof("sending SIGTERM to gcsfuse process: %v", cmd)

		return cmd.Process.Signal(syscall.SIGTERM)
	}

	// when the ctx.Done() is closed,
	// the main workload containers have exited,
	// so it is safe to force kill the gcsfuse process.
	go func(cmd *exec.Cmd) {
		<-ctx.Done()
		time.Sleep(time.Second * 5)
		if cmd.ProcessState == nil || !cmd.ProcessState.Exited() {
			klog.Warningf("after 5 seconds, process with id %v has not exited, force kill the process", cmd.Process.Pid)
			if err := cmd.Process.Kill(); err != nil {
				klog.Warningf("failed to force kill process with id %v", cmd.Process.Pid)
			}
		}
	}(cmd)

	m.WaitGroup.Add(1)
	go func() {
		defer m.WaitGroup.Done()
		if err := cmd.Start(); err != nil {
			mc.ErrWriter.WriteMsg(fmt.Sprintf("failed to start gcsfuse with error: %v\n", err))

			return
		}

		klog.Infof("gcsfuse for bucket %q, volume %q started with process id %v", mc.BucketName, mc.VolumeName, cmd.Process.Pid)

		loggingSeverity := mc.ConfigFileFlagMap["logging:severity"]
		if loggingSeverity == "debug" || loggingSeverity == "trace" {
			go logMemoryUsage(ctx, cmd.Process.Pid)
			go logVolumeUsage(ctx, mc.BufferDir, mc.CacheDir)
		}

		promPort, ok := mc.FlagMap["prometheus-port"]
		if ok && promPort != "0" {
			klog.Infof("start to collect metrics from port %v for volume %q", promPort, mc.VolumeName)
			go collectMetrics(ctx, promPort, mc.TempDir)
		}

		// Since the gcsfuse has taken over the file descriptor,
		// closing the file descriptor to avoid other process forking it.
		syscall.Close(mc.FileDescriptor)
		if err := cmd.Wait(); err != nil {
			errMsg := fmt.Sprintf("gcsfuse exited with error: %v\n", err)
			if strings.Contains(errMsg, "signal: terminated") {
				klog.Infof("[%v] gcsfuse was terminated.", mc.VolumeName)
			} else {
				mc.ErrWriter.WriteMsg(errMsg)
			}
		} else {
			klog.Infof("[%v] gcsfuse exited normally.", mc.VolumeName)
		}
	}()

	return nil
}