func()

in internal/plugin/manager/pluginlauncher.go [76:178]
79 lines of code
12 McCabe index (conditional complexity)

func (l *launchStep) Run(ctx context.Context, p *Plugin) error {
	policy := retry.Policy{MaxAttempts: l.startAttempts, BackoffFactor: 1, Jitter: time.Second}

	addr, err := address(ctx, l.protocol, p.FullName(), policy)
	if err != nil {
		return fmt.Errorf("failed to get address: %w", err)
	}

	p.Address = addr
	p.Manifest.MaxMemoryUsage = l.maxMemoryUsage
	p.Manifest.MaxCPUUsage = l.maxCPUUsage
	p.Manifest.StartAttempts = l.startAttempts
	p.EntryPath = l.entryPath
	p.Protocol = l.protocol
	p.Manifest.LaunchArguments = l.extraArgs

	stateDir := p.stateDir()
	// Create state directory for the plugin.
	if err := os.MkdirAll(stateDir, 0755); err != nil {
		return fmt.Errorf("create state directory %q for plugin %q: %w", stateDir, p.FullName(), err)
	}

	// Update plugin path symlink to point to latest. Core plugin install path is
	// static and does not change across revisions. For instance on Linux it is
	// always /usr/lib/google/guest_agent/core_plugin. Skip symlink setup for core
	// plugin.
	if p.PluginType != PluginTypeCore {
		if err := file.UpdateSymlink(p.staticInstallPath(), p.InstallPath); err != nil {
			return fmt.Errorf("UpdateSymlink(%q, %q) failed for plugin %q: %w", p.staticInstallPath(), p.InstallPath, p.FullName(), err)
		}
	}

	args := p.Manifest.LaunchArguments
	args = append(args, fmt.Sprintf("--protocol=%s", l.protocol), fmt.Sprintf("--address=%s", p.Address), fmt.Sprintf("--errorlogfile=%s", p.logfile()))

	launchFunc := func() error {
		opts := run.Options{
			Name:     p.EntryPath,
			Args:     args,
			ExecMode: run.ExecModeDetach,
		}
		res, err := run.WithContext(ctx, opts)
		if err != nil {
			return fmt.Errorf("failed to launch plugin from %q: %w", p.EntryPath, err)
		}

		p.setPid(res.Pid)
		return nil
	}

	// Launch the plugin.
	if err := retry.Run(ctx, policy, launchFunc); err != nil {
		sendEvent(ctx, p, acmpb.PluginEventMessage_PLUGIN_START_FAILED, fmt.Sprintf("Failed to launch plugin: [%v]. Plugin logs: %s", err, readPluginLogs(p.logfile())))
		p.setState(acmpb.CurrentPluginStates_DaemonPluginState_CRASHED)
		return err
	}

	pluginPid := p.pid()

	constraintFunc := func() error {
		constraint := resource.Constraint{
			PID:            pluginPid,
			Name:           p.FullName(),
			MaxMemoryUsage: p.Manifest.MaxMemoryUsage,
			MaxCPUUsage:    p.Manifest.MaxCPUUsage,
		}
		if err := resource.Apply(constraint); err != nil {
			return err
		}
		return nil
	}

	// Apply resource constraint.
	if err := retry.Run(ctx, policy, constraintFunc); err != nil {
		sendEvent(ctx, p, acmpb.PluginEventMessage_PLUGIN_START_FAILED, fmt.Sprintf("Failed to apply resource constraint: [%v]", err))
		p.setState(acmpb.CurrentPluginStates_DaemonPluginState_CRASHED)
		return err
	}

	galog.Debugf("Launched a plugin process from %q with pid %d", p.EntryPath, pluginPid)

	if err := p.Connect(ctx); err != nil {
		p.setState(acmpb.CurrentPluginStates_DaemonPluginState_CRASHED)
		return fmt.Errorf("failed to connect plugin %s: %w", p.FullName(), err)
	}

	_, status := p.Start(ctx)
	if status.Err() != nil {
		sendEvent(ctx, p, acmpb.PluginEventMessage_PLUGIN_START_FAILED, fmt.Sprintf("Failed to start plugin: [%+v]. Plugin logs: %s", status, readPluginLogs(p.logfile())))
		p.setState(acmpb.CurrentPluginStates_DaemonPluginState_CRASHED)
		return fmt.Errorf("failed to start plugin %s with error: %w", p.FullName(), status.Err())
	}

	sendEvent(ctx, p, acmpb.PluginEventMessage_PLUGIN_STARTED, "Successfully started the plugin.")
	p.setState(acmpb.CurrentPluginStates_DaemonPluginState_RUNNING)
	galog.Infof("Successfully started plugin %q", p.FullName())

	if err := p.Store(); err != nil {
		return fmt.Errorf("store plugin %s info failed: %w", p.FullName(), err)
	}

	return nil
}