internal/plugin/manager/pluginmanager.go

//  Copyright 2024 Google LLC
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.

package manager

import (
	"context"
	"encoding/gob"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"

	"github.com/GoogleCloudPlatform/galog"
	acpb "github.com/GoogleCloudPlatform/google-guest-agent/internal/acp/proto/google_guest_agent/acp"
	"github.com/GoogleCloudPlatform/google-guest-agent/internal/acs/client"
	"github.com/GoogleCloudPlatform/google-guest-agent/internal/boundedlist"
	"github.com/GoogleCloudPlatform/google-guest-agent/internal/cfg"
	"github.com/GoogleCloudPlatform/google-guest-agent/internal/ps"
	"github.com/GoogleCloudPlatform/google-guest-agent/internal/scheduler"
	"github.com/GoogleCloudPlatform/google-guest-agent/internal/utils/file"
	"google.golang.org/grpc/codes"
	"google.golang.org/protobuf/proto"
	tpb "google.golang.org/protobuf/types/known/timestamppb"
)

const (
	// agentStateDir is where all agent state is stored. This includes plugin
	// information and other information agent might want to store.
	agentStateDir = "agent_state"
	// pluginInfoDir is where all plugin information is stored within agent state.
	pluginInfoDir = "plugin_info"
	// manifestFile is the name of the file containing the plugin manifest.
	manifestFile = "manifest.binpb"
	// healthCheckFrequency is the frequency at which plugin health check is
	// executed.
	healthCheckFrequency = 10 * time.Second
	// metricsCheckFrequency is the default frequency at which plugin metrics
	// check is executed.
	metricsCheckFrequency = 10 * time.Second
	// maxMetricDatapoints is the maximum number of datapoints to be stored in the
	// metric list in memory.
	maxMetricDatapoints = 60
	// CorePluginName is the name of the guest agent core plugin.
	CorePluginName = "GuestAgentCorePlugin"
)

var (
	// pluginManager is the instance of plugin manager.
	pluginManager *PluginManager

	// isUDSSupported checks if UDS is supported on the host. This is stubbed out
	// for testing.
	isUDSSupported = defaultIsUDSSupported
)

// PluginManager struct represents the plugins that plugin manager manages.
type PluginManager struct {
	// mu protects the plugins map.
	mu sync.RWMutex
	// plugins is the map of plugin name and plugin managed by plugin manager.
	plugins map[string]*Plugin
	// pluginMonitorMu protects the pluginMonitors map.
	pluginMonitorMu sync.Mutex
	// pluginMonitors is the map of plugin and plugin monitor ID monitoring
	// plugin.
	pluginMonitors map[string]string
	// pluginMetricsMu protects the pluginMetrics map.
	pluginMetricsMu sync.Mutex
	// pluginMetricsMonitors is the map of plugin and plugin metrics ID monitoring
	// plugin.
	pluginMetricsMonitors map[string]string
	// scheduler is the scheduler used by plugin manager to schedule plugins
	// monitoring.
	scheduler *scheduler.Scheduler
	// protocol is the protocol used by plugin manager to communicate with all
	// plugins.
	protocol string

	// pendingPluginRevisionsMu protects the pendingPluginRevisions map.
	pendingPluginRevisionsMu sync.RWMutex
	// inProgressPluginRequests keep tracks of the pending plugin requests. This
	// allows us to ignore new requests if previous request is still in progress
	// for the same plugin.
	inProgressPluginRequests map[string]bool

	// requestCountMu protects the requestCount map.
	requestCountMu sync.Mutex
	// requestCount keeps track of the number of requests received for each
	// action and also the number of successful and failures requests.
	requestCount map[acpb.ConfigurePluginStates_Action]map[bool]int

	// instanceIDMu protects the instanceID.
	instanceIDMu sync.Mutex
	// instanceID is the instance ID of the VM plugin manager is running on.
	instanceID string

	// IsInitialized indicates if plugin manager is initialized.
	IsInitialized atomic.Bool
}

// LocalPluginInstallation contains the configuration for a local plugin
// installation.
type LocalPluginInstallation struct {
	// Enable indicates if the local plugin should be enabled.
	Enable bool
	// OnReady is the callback when the local plugin is running.
	OnReady func(context.Context)
}

// agentPluginState returns the path to the directory where agent maintains
// plugin state.
func agentPluginState() string {
	return filepath.Join(baseState(), agentStateDir, pluginInfoDir)
}

// Instance returns the previously initialized instance of plugin manager.
func Instance() *PluginManager {
	return pluginManager
}

func init() {
	pluginManager = &PluginManager{
		protocol:                 tcpProtocol,
		pluginMonitors:           make(map[string]string),
		pluginMetricsMonitors:    make(map[string]string),
		scheduler:                scheduler.Instance(),
		inProgressPluginRequests: make(map[string]bool),
		requestCount:             make(map[acpb.ConfigurePluginStates_Action]map[bool]int),
	}
}

func (m *PluginManager) setInstanceID(id string) {
	m.instanceIDMu.Lock()
	defer m.instanceIDMu.Unlock()
	m.instanceID = id
}

func (m *PluginManager) currentInstanceID() string {
	m.instanceIDMu.Lock()
	defer m.instanceIDMu.Unlock()
	return m.instanceID
}

// InitAdHocPluginManager initializes and returns a PluginManager instance for
// ad-hoc requests that does not go through ACS. This is generally used by other
// local processes to temporarily leverage plugin manager functionality. Use
// this instead of InitPluginManager if there's no active plugin management
// required. InitAdHocPluginManager skips some initialization steps like
// scheduling plugin monitors that are not required for ad-hoc requests.
func InitAdHocPluginManager(ctx context.Context, instanceID string) (*PluginManager, error) {
	galog.Infof("Initializing ad-hoc plugin manager for instance %q", instanceID)

	pluginManager.setInstanceID(instanceID)
	plugins, err := load(agentPluginState())
	if err != nil {
		return nil, fmt.Errorf("unable to load existing plugin state: %w", err)
	}
	pluginManager.plugins = plugins

	return pluginManager, nil
}

// StopPlugin stops the plugin. This is used for ad-hoc requests that does not
// go through ACS. In case the plugin is not found or is not running, it returns
// nil error and is a no-op.
func (m *PluginManager) StopPlugin(ctx context.Context, name string) error {
	galog.Infof("Stopping plugin %q", name)

	plugin, err := m.Fetch(name)
	if err != nil {
		galog.Infof("Plugin %q state not found [err: %v], skipping stop request", name, err)
		return nil
	}

	if err := plugin.Connect(ctx); err != nil {
		galog.Infof("Failed to connect to plugin %q [err: %v], skipping graceful stop", plugin.FullName(), err)
	}

	return m.stopAndRemovePlugin(ctx, plugin)
}

// InitPluginManager initializes and returns a PluginManager instance.
// Plugin Manager can be initialized and used to support core plugins even if
// ACS is disabled. Plugin Manager will be initialized during early Guest Agent
// startup to configure the core plugins.
func InitPluginManager(ctx context.Context, instanceID string) (*PluginManager, error) {
	version := cfg.Retrieve().Core.Version
	galog.Infof("Initializing plugin manager for instance %q, agent version: %q", instanceID, version)
	pluginManager.setInstanceID(instanceID)

	// Cleanup old plugin state in a separate goroutine. This operation is not
	// critical for plugin manager initialization and should not block it.
	go func() {
		if err := pluginManager.cleanupOldState(ctx, filepath.Dir(baseState())); err != nil {
			galog.Errorf("Failed to cleanup old plugin state: %v", err)
		}
	}()

	// Load the existing plugin state.
	plugins, err := load(agentPluginState())
	if err != nil {
		return nil, fmt.Errorf("unable to load existing plugin state: %w", err)
	}

	pluginManager.plugins = plugins

	// Subscribe to cleanup event. This needs to happen after plugin manager is
	// initialized, or we may accidentally cleanup plugins that shouldn't be removed.
	scheduler.ScheduleJobs(ctx, []scheduler.Job{newCleanupJob(pluginManager)}, false)

	if err := RegisterCmdHandler(ctx); err != nil {
		return nil, fmt.Errorf("failed to register plugin command handler: %w", err)
	}

	wg := sync.WaitGroup{}
	pluginManager.pendingPluginRevisionsMu.Lock()

	for _, p := range plugins {
		pluginManager.inProgressPluginRequests[p.Name] = true
		wg.Add(1)

		go func(p *Plugin) {
			// Regardless of the outcome, we should remove the plugin from the pending
			// list as request is no longer in process for this plugin.
			defer func() {
				pluginManager.pendingPluginRevisionsMu.Lock()
				defer pluginManager.pendingPluginRevisionsMu.Unlock()
				delete(pluginManager.inProgressPluginRequests, p.Name)
				wg.Done()
			}()

			if p.Name == CorePluginName && p.Revision != version {
				// If core plugin is already running, just set the state to running and
				// return. If the revision is different, it means the core plugin is
				// being updated and there's no need to relaunch if not running.
				if p.IsRunning(ctx) {
					p.setState(acpb.CurrentPluginStates_RUNNING)
				}
				galog.Infof("Plugin %q state is different from expected revision %q, skipping relaunch", p.FullName(), version)
				return
			}

			if err := connectOrReLaunch(ctx, p); err != nil {
				galog.Errorf("Failed to connect or relaunch plugin %q: %v", p.FullName(), err)
			} else {
				pluginManager.startPluginSchedulers(ctx, p)
			}
		}(p)
	}
	pluginManager.IsInitialized.Store(true)
	pluginManager.pendingPluginRevisionsMu.Unlock()
	wg.Wait()

	// TCP is always supported, so if the forced connection type is TCP, we can
	// return early.
	connType := cfg.Retrieve().Plugin.ConnectionType
	if connType == tcpProtocol {
		pluginManager.protocol = tcpProtocol
		return pluginManager, nil
	}

	// If the connection type is not TCP, check if UDS is supported and set the
	// protocol to UDS if it is. Otherwise, fallback to TCP.
	if isUDSSupported() {
		pluginManager.protocol = udsProtocol
	} else {
		galog.Debugf("UDS is not supported, fallback to TCP")
		pluginManager.protocol = tcpProtocol
	}
	return pluginManager, nil
}

// RemoveAllDynamicPlugins filters out core plugins and triggers plugin manager
// to remove all dynamic plugins on the host. It also removes the base state
// directory to ensure that the entire state is cleaned up.
func (m *PluginManager) RemoveAllDynamicPlugins(ctx context.Context) error {
	var reqs []*acpb.ConfigurePluginStates_ConfigurePlugin
	galog.Infof("Removing all dynamic plugins")

	var toRemove []*Plugin
	for _, p := range m.list() {
		if p.IsLocal() {
			galog.Debugf("Skipping core plugin %q, it will be removed by package manager", p.Name)
			continue
		}
		toRemove = append(toRemove, p)
	}

	errChan := make(chan error, len(reqs))
	wg := sync.WaitGroup{}

	for _, plugin := range toRemove {
		wg.Add(1)
		go func(p *Plugin) {
			var err error
			defer wg.Done()

			if err = m.StopPlugin(ctx, p.Name); err != nil {
				galog.Warnf("Remove plugin %q completed with error: [%v]", p.FullName(), err)
				errChan <- err
			}

		}(plugin)
	}

	go func() {
		wg.Wait()
		baseStateDir := baseState()
		if file.Exists(baseStateDir, file.TypeDir) {
			if err := os.RemoveAll(baseStateDir); err != nil {
				errChan <- fmt.Errorf("failed to remove file %q: %w", baseStateDir, err)
			}
		}
		close(errChan)
	}()

	var errs error
	for err := range errChan {
		errs = errors.Join(errs, err)
	}

	return errs
}

// ListPluginStates returns the plugin states and cached health check
// information.
func (m *PluginManager) ListPluginStates(ctx context.Context, req *acpb.ListPluginStates) *acpb.CurrentPluginStates {
	galog.Debugf("Handling list plugin state request: %+v", req)
	var states []*acpb.CurrentPluginStates_DaemonPluginState
	plugins := m.list()

	for _, p := range plugins {
		status := &acpb.CurrentPluginStates_Status{Status: p.State()}
		h := p.healthInfo()
		if h != nil {
			status.ResponseCode = h.responseCode
			status.Results = h.messages
			status.UpdateTime = tpb.New(h.timestamp)
		}

		p.RuntimeInfo.metricsMu.Lock()
		var pluginMetrics []*acpb.CurrentPluginStates_Metric
		for _, metric := range p.RuntimeInfo.metrics.All() {
			monitorMetric := &acpb.CurrentPluginStates_Metric{
				Timestamp:   metric.timestamp,
				CpuUsage:    metric.cpuUsage,
				MemoryUsage: metric.memoryUsage,
			}
			pluginMetrics = append(pluginMetrics, monitorMetric)
		}

		state := &acpb.CurrentPluginStates_DaemonPluginState{
			Name:                 p.Name,
			CurrentRevisionId:    p.Revision,
			CurrentPluginStatus:  status,
			CurrentPluginMetrics: pluginMetrics,
			ConfigHash:           p.configHash(),
		}

		pendingStatus := p.pendingStatus()
		if pendingStatus != nil {
			state.PendingRevisionId = pendingStatus.revision
		}

		// Flush the metrics array.
		p.RuntimeInfo.metrics.Reset()

		// Release the metrics lock.
		p.RuntimeInfo.metricsMu.Unlock()

		// Append the state to the list.
		states = append(states, state)
	}

	return &acpb.CurrentPluginStates{DaemonPluginStates: states}
}

// filterPendingPluginRevisions filters out plugin requests that already have a
// request in progress. This is to prevent multiple concurrent requests for the
// same plugin.
//
// setInProgress determines if the in-progress plugin state should be set for
// the filtered requests. This should be set to true only when the caller intends
// to process the filtered requests immediately.
func (m *PluginManager) filterPendingPluginRevisions(ctx context.Context, req *acpb.ConfigurePluginStates, setInProgress bool) []*acpb.ConfigurePluginStates_ConfigurePlugin {
	var filtered []*acpb.ConfigurePluginStates_ConfigurePlugin
	// Check if there's already a request in progress for this plugin.
	m.pendingPluginRevisionsMu.Lock()
	defer m.pendingPluginRevisionsMu.Unlock()
	for _, req := range req.GetConfigurePlugins() {
		if _, inProgress := m.inProgressPluginRequests[req.GetPlugin().GetName()]; inProgress {
			galog.Infof("Ignoring request to %v plugin %q, another request already in progress", req.GetAction(), req.GetPlugin().GetName())
			continue
		} else {
			filtered = append(filtered, req)
			if setInProgress {
				// Set state to in progress only if the caller intends to process the
				// filtered requests immediately.
				m.inProgressPluginRequests[req.GetPlugin().GetName()] = true
			}
		}
	}
	return filtered
}

// ConfigurePluginStates configures the plugin states as stated in the request.
func (m *PluginManager) ConfigurePluginStates(ctx context.Context, req *acpb.ConfigurePluginStates) {
	galog.Debugf("Handling configure plugin state request: %+v", req)
	wg := sync.WaitGroup{}

	toProcess := m.filterPendingPluginRevisions(ctx, req, true)

	for _, req := range toProcess {
		wg.Add(1)
		go func(req *acpb.ConfigurePluginStates_ConfigurePlugin) {
			defer wg.Done()
			m.configurePlugin(ctx, req)
		}(req)
	}

	wg.Wait()
	galog.Debugf("Configure plugin state request completed")
}

// list returns the list of currently managed plugins.
func (m *PluginManager) list() []*Plugin {
	m.mu.RLock()
	defer m.mu.RUnlock()
	var plugins []*Plugin
	for _, p := range m.plugins {
		plugins = append(plugins, p)
	}
	return plugins
}

// Fetch returns the plugin instance with the given name.
func (m *PluginManager) Fetch(name string) (*Plugin, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	p, ok := m.plugins[name]
	if !ok {
		return nil, fmt.Errorf("plugin %q not found", name)
	}
	return p, nil
}

// add stores the plugin instance with the given name.
func (m *PluginManager) add(p *Plugin) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.plugins[p.Name] = p
}

// delete deletes the plugin instance with the given name.
func (m *PluginManager) delete(name string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	delete(m.plugins, name)
}

func (m *PluginManager) configurePlugin(ctx context.Context, req *acpb.ConfigurePluginStates_ConfigurePlugin) {
	// Regardless of the outcome, we should remove the plugin from the pending
	// list as request is no longer in process for this plugin.
	defer func() {
		m.pendingPluginRevisionsMu.Lock()
		defer m.pendingPluginRevisionsMu.Unlock()
		delete(m.inProgressPluginRequests, req.GetPlugin().GetName())
	}()

	var success bool
	switch req.GetAction() {
	case acpb.ConfigurePluginStates_INSTALL:
		if err := m.installPlugin(ctx, req); err != nil {
			galog.Errorf("Failed to install plugin %q, revision %q: %v", req.GetPlugin().GetName(), req.GetPlugin().GetRevisionId(), err)
		} else {
			success = true
		}
	case acpb.ConfigurePluginStates_REMOVE:
		// If we want to avoid launching the core plugin by reading the local config
		// dynamically, just remove it without reading the config.
		if cfg.Retrieve().Core.EnableLocalPlugins && req.GetPlugin().GetName() == CorePluginName {
			if err := m.removePlugin(ctx, req); err != nil {
				galog.Errorf("Failed to remove plugin %q, revision %q: %v", req.GetPlugin().GetName(), req.GetPlugin().GetRevisionId(), err)
			} else {
				success = true
			}
		} else {
			// Find if there's a local installation of the plugin. If found, upgrade to
			// it, otherwise remove the plugin.
			localInstallation, err := m.GetLocalPlugin(ctx, req.GetPlugin().GetName())
			if err != nil {
				galog.Errorf("Failed to get local plugin %q: %v", req.GetPlugin().GetName(), err)
			} else {
				if localInstallation != nil {
					galog.Debugf("Local installation of plugin %q with revision %q found, attempting to install.", localInstallation.GetPlugin().GetName(), localInstallation.GetPlugin().GetRevisionId())
					if err = m.installPlugin(ctx, localInstallation); err != nil {
						galog.Errorf("Failed to install local plugin %q, revision %q: %v", localInstallation.GetPlugin().GetName(), localInstallation.GetPlugin().GetRevisionId(), err)
					}
				} else {
					galog.Debugf("Local installation of plugin %q not found, removing.", req.GetPlugin().GetName())
					if err := m.removePlugin(ctx, req); err != nil {
						galog.Errorf("Failed to remove plugin %q, revision %q: %v", req.GetPlugin().GetName(), req.GetPlugin().GetRevisionId(), err)
					} else {
						success = true
					}
				}
			}
		}
	case acpb.ConfigurePluginStates_APPLY:
		if err := m.applyConfig(ctx, req); err != nil {
			galog.Errorf("Failed to apply config for plugin %q, revision %q: %v", req.GetPlugin().GetName(), req.GetPlugin().GetRevisionId(), err)
		} else {
			success = true
		}
	default:
		galog.Warnf("Unknown action (%s) for configure plugin state request, ignoring", req.GetAction().String())
	}

	m.requestCountMu.Lock()
	defer m.requestCountMu.Unlock()
	if m.requestCount[req.GetAction()] == nil {
		m.requestCount[req.GetAction()] = make(map[bool]int)
	}
	m.requestCount[req.GetAction()][success]++
}

// setMetricConfig sets the default metric configurations for the plugin or
// overrides it from request proto if provided.
func (p *Plugin) setMetricConfig(req *acpb.ConfigurePluginStates_ConfigurePlugin) {
	p.Manifest.MetricsInterval = metricsCheckFrequency
	p.Manifest.MaxMetricDatapoints = maxMetricDatapoints

	if points := req.GetManifest().GetMaxMetricDatapoints(); points != 0 {
		p.Manifest.MaxMetricDatapoints = uint(points)
	}

	if interval := req.GetManifest().GetMetricsInterval().GetSeconds(); interval != 0 {
		p.Manifest.MetricsInterval = time.Duration(interval) * time.Second
	}

	p.RuntimeInfo.metrics = boundedlist.New[Metric](p.Manifest.MaxMetricDatapoints)
}

func setConfig(manifest *Manifest, req *acpb.ConfigurePluginStates_ConfigurePlugin) error {
	manifest.StartConfig = &ServiceConfig{}

	// If no config is provided, return early.
	if req.GetManifest().Config == nil {
		galog.Infof("No config provided for plugin %q, setting empty config", req.GetPlugin().GetName())
		return nil
	}

	switch req.GetManifest().Config.(type) {
	case *acpb.ConfigurePluginStates_Manifest_StringConfig:
		galog.Infof("Setting string config for plugin %q", req.GetPlugin().GetName())
		manifest.StartConfig.Simple = req.GetManifest().GetStringConfig()
	case *acpb.ConfigurePluginStates_Manifest_StructConfig:
		// Marshal the service config to a byte array to persist it on disk. This
		// will be un-marshaled to use at plugin launch time.
		bytes, err := proto.Marshal(req.GetManifest().GetStructConfig())
		if err != nil {
			return fmt.Errorf("unable to marshal service config: %w", err)
		}
		galog.Infof("Setting structured config for plugin %q", req.GetPlugin().GetName())
		manifest.StartConfig.Structured = bytes
	}
	return nil
}

// newPluginManifest generates agent representation of the manifest from the
// install request.
func newPluginManifest(req *acpb.ConfigurePluginStates_ConfigurePlugin) (*Manifest, error) {
	manifest := &Manifest{
		StartAttempts:          int(req.GetManifest().GetStartAttemptCount()),
		MaxMemoryUsage:         req.GetManifest().GetMaxMemoryUsageBytes(),
		MaxCPUUsage:            req.GetManifest().GetMaxCpuUsagePercentage(),
		StopTimeout:            time.Duration(req.GetManifest().GetStopTimeout().GetSeconds()) * time.Second,
		StartTimeout:           time.Duration(req.GetManifest().GetStartTimeout().GetSeconds()) * time.Second,
		StartConfig:            &ServiceConfig{},
		PluginType:             req.GetManifest().GetPluginType(),
		PluginInstallationType: req.GetManifest().GetPluginInstallationType(),
	}

	if err := setConfig(manifest, req); err != nil {
		return nil, err
	}

	return manifest, nil
}

// newPlugin creates a new plugin instance from the request.
// Rest of the plugin instance values are set at run time when install
// steps are executed on it.
func newPlugin(req *acpb.ConfigurePluginStates_ConfigurePlugin) (*Plugin, error) {
	p := &Plugin{
		Name:        req.GetPlugin().GetName(),
		Revision:    req.GetPlugin().GetRevisionId(),
		RuntimeInfo: &RuntimeInfo{},
	}

	manifest, err := newPluginManifest(req)
	if err != nil {
		return nil, fmt.Errorf("unable to generate plugin service config: %w", err)
	}

	p.Manifest = manifest

	if p.IsLocal() {
		// Dynamic plugins are installed in a specific directory, that install
		// workflow sets its install path. In case of local plugins since they're
		// already present on disk and directory is known set its install path here
		// itself.
		p.InstallPath = filepath.Dir(req.GetPlugin().GetEntryPoint())
	}

	p.setMetricConfig(req)

	return p, nil
}

// installPlugin installs checks if the plugin already exists and does a
// fresh install or removes existing plugin revision and installs a new one.
func (m *PluginManager) installPlugin(ctx context.Context, req *acpb.ConfigurePluginStates_ConfigurePlugin) error {
	galog.Infof("Installing plugin %q, revision %q", req.GetPlugin().GetName(), req.GetPlugin().GetRevisionId())

	plugin, err := newPlugin(req)
	if err != nil {
		return fmt.Errorf("failed to create new plugin instance: %w", err)
	}

	sendEvent(ctx, plugin, acpb.PluginEventMessage_PLUGIN_CONFIG_INSTALL, "Received request to install a plugin.")
	currPlugin, err := m.Fetch(req.GetPlugin().GetName())
	if err == nil && currPlugin.Revision == req.GetPlugin().GetRevisionId() {
		sendEvent(ctx, currPlugin, acpb.PluginEventMessage_PLUGIN_INSTALL_FAILED, "Plugin is already installed or being processed.")
		return fmt.Errorf("plugin %q is already installed or being processed", currPlugin.FullName())
	}

	if currPlugin != nil {
		return m.upgradePlugin(ctx, req)
	}

	steps := m.generateInstallWorkflow(ctx, req)
	return m.runlaunchPluginSteps(ctx, plugin, steps)
}

// startPluginSchedulers starts all scheduler jobs for the plugin.
func (m *PluginManager) startPluginSchedulers(ctx context.Context, plugin *Plugin) {
	// At this point plugin is already running, run them in a separate go routine
	// to avoid blocking the main thread. These jobs are configured to start
	// immediately scheduling them synchronously would block the caller until they
	// finish first run.
	go func() {
		m.startMonitoring(ctx, plugin)
		m.startMetricsMonitoring(ctx, plugin)
	}()
}

// runlaunchPluginSteps runs the steps to launch the plugin. Steps differ for
// install and upgrade but the reaction to success/failure is the same.
func (m *PluginManager) runlaunchPluginSteps(ctx context.Context, plugin *Plugin, steps []Step) error {
	// Store the plugin in the manager as soon as we start processing so
	// [ListPluginStates] can send intermediate states as well.
	m.add(plugin)

	if err := plugin.runSteps(ctx, steps); err != nil {
		// Delete the plugin from the list in case it fails, agent will start fresh
		// if ACS requests to install the plugin again.
		m.delete(plugin.Name)
		sendEvent(ctx, plugin, acpb.PluginEventMessage_PLUGIN_INSTALL_FAILED, fmt.Sprintf("Failed to install plugin: %v", err))
		// If the installation fails, try cleaning up the process to prevent
		// potential conflicts or errors that unmanaged processes could cause.
		if err := ps.KillProcess(plugin.pid(), ps.KillModeNoWait); err != nil {
			// Just log the error, process might have crashed, already exited or not
			// successfully launched at all.
			galog.Warnf("Stop plugin %q finished with error: %v", plugin.FullName(), err)
		}
		// Return original install error.
		return fmt.Errorf("install plugin %q: %w", plugin.FullName(), err)
	}

	m.startPluginSchedulers(ctx, plugin)
	sendEvent(ctx, plugin, acpb.PluginEventMessage_PLUGIN_INSTALLED, "Successfully installed the plugin.")

	galog.Infof("Successfully installed plugin %q", plugin.FullName())

	return nil
}

// upgradePlugin handles the plugin revision upgrades. It downloads and unpacks
// the new plugin revision, stops the old plugin revision and then launches the
// new one.
func (m *PluginManager) upgradePlugin(ctx context.Context, req *acpb.ConfigurePluginStates_ConfigurePlugin) error {
	galog.Debugf("Upgrading plugin %q, revision %q", req.GetPlugin().GetName(), req.GetPlugin().GetRevisionId())

	plugin, err := newPlugin(req)
	if err != nil {
		return fmt.Errorf("failed to create new plugin instance: %w", err)
	}

	currPlugin, err := m.Fetch(req.GetPlugin().GetName())
	if err != nil {
		return fmt.Errorf("fetching current instance for %q: %w", req.GetPlugin().GetName(), err)
	}

	// If new revision fails pre-launch steps, current plugin would keep running
	// and new revision process would be aborted. If new revision succeeds,
	// current plugin would be stopped and removed. Reset pending plugin status to
	// avoid showing plugin install in progress.
	defer currPlugin.resetPendingStatus()

	// Current plugin will be removed as soon as new plugin launch is started
	// below. Set the pending status to show new plugin revision install in
	// progress. This will be captured by [ListPluginStates] and sent to ACS.
	currPlugin.setPendingStatus(plugin.Revision, acpb.CurrentPluginStates_INSTALLING)

	// Two plugin revisions can co-exist on the same host, but only one of them
	// can be running. Run pre-launch steps on new plugin revision to reduce
	// plugin downtime and make sure it can be launched.
	steps := m.preLaunchWorkflow(ctx, req)
	galog.Infof("Running pre-upgrade steps for plugin %q", plugin.FullName())
	if err := plugin.runSteps(ctx, steps); err != nil {
		sendEvent(ctx, plugin, acpb.PluginEventMessage_PLUGIN_INSTALL_FAILED, fmt.Sprintf("Failed to run pre-upgrade steps: %v", err))
		return fmt.Errorf("failed to run pre-upgrade steps: %w", err)
	}

	// Previously installed plugin revision already exists, remove before
	// installing a new one.
	galog.Infof("Stopping and removing old plugin %q", currPlugin.FullName())
	if err := m.stopAndRemovePlugin(ctx, currPlugin); err != nil {
		sendEvent(ctx, currPlugin, acpb.PluginEventMessage_PLUGIN_INSTALL_FAILED, fmt.Sprintf("Failed to remove plugin: %v", err))
		return fmt.Errorf("failed to remove plugin: %w", err)
	}

	return m.runlaunchPluginSteps(ctx, plugin, []Step{m.newLaunchStep(req)})
}

// stopAndRemovePlugin stops the given plugin, all of its schedulers and removes
// it from the manager.
func (m *PluginManager) stopAndRemovePlugin(ctx context.Context, p *Plugin) error {
	sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_CONFIG_REMOVE, "Received request to remove a plugin.")

	// Stop all schedulers running on the plugin so it doesn't interfere with the
	// stop process.
	m.stopMonitoring(p)
	m.stopMetricsMonitoring(p)

	if err := p.runSteps(ctx, []Step{&stopStep{cleanup: true}}); err != nil {
		sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_REMOVE_FAILED, fmt.Sprintf("Failed to remove plugin: %v", err))
		return fmt.Errorf("unable to remove plugin %q: %w", p.FullName(), err)
	}

	sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_REMOVED, "Successfully removed the plugin.")
	m.delete(p.Name)

	galog.Infof("Successfully removed plugin %q", p.FullName())
	return nil
}

// applyConfig applies the config to the plugin and sends events on success or
// failure.
func (m *PluginManager) applyConfig(ctx context.Context, req *acpb.ConfigurePluginStates_ConfigurePlugin) error {
	galog.Infof("Applying config to plugin %q, revision %s", req.GetPlugin().GetName(), req.GetPlugin().GetRevisionId())

	p, err := m.Fetch(req.GetPlugin().GetName())
	if err != nil {
		sendEvent(ctx, &Plugin{Name: req.GetPlugin().GetName(), Revision: req.GetPlugin().GetRevisionId()}, acpb.PluginEventMessage_PLUGIN_CONFIG_APPLY_FAILED, "Plugin not found.")
		return fmt.Errorf("plugin %q not found", req.GetPlugin().GetName())
	}

	sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_CONFIG_APPLY, "Received request to apply a config to a plugin.")

	if err := setConfig(p.Manifest, req); err != nil {
		sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_CONFIG_APPLY_FAILED, fmt.Sprintf("Failed to set config: %v", err))
		return fmt.Errorf("failed to set config: %w", err)
	}

	// Update the plugin state file with the new config.
	galog.Infof("Updating on-disk config for plugin %q", p.FullName())
	if err := p.Store(); err != nil {
		sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_CONFIG_APPLY_FAILED, fmt.Sprintf("Failed to store config: %v", err))
		return fmt.Errorf("failed to store config: %w", err)
	}

	// Reset start config hash.
	p.Manifest.startConfigMu.Lock()
	p.Manifest.startConfigHash = ""
	p.Manifest.startConfigMu.Unlock()
	p.configHash()

	_, status := p.Apply(ctx, p.Manifest.StartConfig)
	if status.Err() == nil {
		sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_CONFIG_APPLIED, "Successfully applied the config to the plugin.")
		return nil
	}

	if status.Code() == codes.Unimplemented {
		// Plugin doesn't support applying config via Apply RPC. This could be
		// because plugin doesn't support config update at runtime or it
		// deliberately needs a restart to apply the config. In either case
		// retrying won't help and we should attempt restart workflow.
		galog.Infof("Plugin %q returned unimplemented error for apply config request, attempting the restart workflow", p.FullName())
		if err := p.runSteps(ctx, relaunchWorkflow(ctx, p)); err != nil {
			p.setState(acpb.CurrentPluginStates_CRASHED)
			return fmt.Errorf("failed to relaunch plugin %q: %w", p.FullName(), err)
		}
	} else {
		sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_CONFIG_APPLY_FAILED, fmt.Sprintf("Failed to apply config: %v", status.Err()))
		return fmt.Errorf("failed to apply config: %w", status.Err())
	}

	sendEvent(ctx, p, acpb.PluginEventMessage_PLUGIN_CONFIG_APPLIED, "Successfully applied the config to the plugin.")
	return nil
}

// removePlugin removes the plugin revision or ignores the request if plugin
// does not exist.
func (m *PluginManager) removePlugin(ctx context.Context, req *acpb.ConfigurePluginStates_ConfigurePlugin) error {
	galog.Infof("Removing plugin %q, revision %s", req.GetPlugin().GetName(), req.GetPlugin().GetRevisionId())

	p, err := m.Fetch(req.GetPlugin().GetName())
	if err != nil {
		sendEvent(ctx, &Plugin{Name: req.GetPlugin().GetName(), Revision: req.GetPlugin().GetRevisionId()}, acpb.PluginEventMessage_PLUGIN_REMOVE_FAILED, "Plugin not found.")
		return fmt.Errorf("plugin %q not found", req.GetPlugin().GetName())
	}

	if err := m.stopAndRemovePlugin(ctx, p); err != nil {
		return fmt.Errorf("failed to remove plugin %q: %w", p.FullName(), err)
	}

	// State directory should persist across plugin revisions and should be
	// removed only when plugin is explicitly removed.
	stateDir := p.stateDir()
	if !file.Exists(stateDir, file.TypeDir) {
		galog.Debugf("Plugin state directory %q does not exist, nothing to remove", stateDir)
		return nil
	}

	return os.RemoveAll(p.stateDir())
}

// startMonitoring schedules a plugin monitoring job that ensures the
// plugin is running. If plugin is found unhealthy it is restarted.
func (m *PluginManager) startMonitoring(ctx context.Context, p *Plugin) {
	galog.Infof("Starting plugin monitor job for plugin %q", p.FullName())
	pm := NewPluginMonitor(p, healthCheckFrequency)

	m.pluginMonitorMu.Lock()
	m.pluginMonitors[p.FullName()] = pm.ID()
	m.pluginMonitorMu.Unlock()

	// ScheduleJob() throws error only if we try to schedule a job that should
	// not be enabled (ShouldEnable() returns false). Ignoring this error
	// here as in case of monitoring pm.ShouldEnable() always returns true.
	m.scheduler.ScheduleJob(ctx, pm)
}

// stopMonitoring stops/removes the plugin monitoring job.
func (m *PluginManager) stopMonitoring(p *Plugin) {
	galog.Infof("Removing plugin monitor job for plugin %q", p.FullName())
	pm, ok := m.pluginMonitors[p.FullName()]
	if !ok {
		galog.Warnf("Plugin monitor not found for %q, ignoring stop monitor request", p.FullName())
		return
	}
	m.scheduler.UnscheduleJob(pm)
	m.pluginMonitorMu.Lock()
	defer m.pluginMonitorMu.Unlock()
	delete(m.pluginMonitors, p.FullName())
}

// startMetricsMonitoring schedules a plugin resource metrics collection job
// that collects and stores them in memory.
func (m *PluginManager) startMetricsMonitoring(ctx context.Context, p *Plugin) {
	galog.Infof("Starting plugin metrics monitor job for plugin %q", p.FullName())
	pm := NewPluginMetrics(p, p.Manifest.MetricsInterval)
	// ScheduleJob() throws error only if we try to schedule a job that should
	// not be enabled (ShouldEnable() returns false). Ignoring this error
	// here as in case of monitoring pm.ShouldEnable() always returns true.
	m.scheduler.ScheduleJob(ctx, pm)
	m.pluginMetricsMu.Lock()
	defer m.pluginMetricsMu.Unlock()
	m.pluginMetricsMonitors[p.FullName()] = pm.ID()
}

// stopMetricsMonitoring stops/removes the plugin metrics monitoring job.
func (m *PluginManager) stopMetricsMonitoring(p *Plugin) {
	galog.Infof("Removing plugin metrics monitor job for plugin %q", p.FullName())

	m.pluginMetricsMu.Lock()
	pm, ok := m.pluginMetricsMonitors[p.FullName()]
	m.pluginMetricsMu.Unlock()

	if !ok {
		galog.Warnf("Plugin metrics monitor not found for %q, ignoring stop monitor request", p.FullName())
		return
	}
	m.scheduler.UnscheduleJob(pm)
	m.pluginMetricsMu.Lock()
	defer m.pluginMetricsMu.Unlock()
	delete(m.pluginMetricsMonitors, p.FullName())
}

// connectOrReLaunch connects to the plugin and launches the plugin if needed.
func connectOrReLaunch(ctx context.Context, p *Plugin) error {
	if p.IsRunning(ctx) {
		p.setState(acpb.CurrentPluginStates_RUNNING)
		return nil
	}
	galog.Debugf("Plugin %q is not running, relaunching", p.FullName())
	if err := p.runSteps(ctx, relaunchWorkflow(ctx, p)); err != nil {
		p.setState(acpb.CurrentPluginStates_CRASHED)
		return fmt.Errorf("failed to relaunch plugin %q: %w", p.FullName(), err)
	}

	return nil
}

// load loads the plugin information from directory and returns a map of
// plugins.
func load(stateDir string) (map[string]*Plugin, error) {
	galog.Debugf("Loading plugin state from %s", stateDir)

	plugins := make(map[string]*Plugin)

	files, err := os.ReadDir(stateDir)
	if err != nil {
		if os.IsNotExist(err) {
			// Plugin state might not exist yet to load from disk just log.
			galog.Debugf("Plugin state directory %q does not exist, nothing to load", stateDir)
			return plugins, nil
		}
		return nil, fmt.Errorf("unable to load plugin state from directory %s: %v", stateDir, err)
	}

	for _, f := range files {
		if f.IsDir() {
			galog.Debugf("Found unknown directory %q in %q, ignoring", f.Name(), stateDir)
			continue
		}
		file := filepath.Join(stateDir, f.Name())
		fh, err := os.Open(file)
		if err != nil {
			return nil, fmt.Errorf("unabled to read plugin state from %s: %w", file, err)
		}
		defer fh.Close()

		plugin := &Plugin{}
		if err := gob.NewDecoder(fh).Decode(plugin); err != nil {
			return nil, fmt.Errorf("unable to decode plugin state file %s: %w", f, err)
		}
		plugin.RuntimeInfo.metrics = boundedlist.New[Metric](plugin.Manifest.MaxMetricDatapoints)
		plugins[plugin.Name] = plugin

		// This is a hack to ensure that the core plugin is always treated as a
		// daemon and local plugin. This is because previous versions of the agent
		// did not store this information in the plugin state. Hence, when the agent
		// is upgraded, the core plugin will be treated as a dynamic plugin, which
		// would cause the agent to clean up the core plugin binary and folder.
		if plugin.Name == CorePluginName {
			plugin.Manifest.PluginType = acpb.PluginType_DAEMON
			plugin.Manifest.PluginInstallationType = acpb.PluginInstallationType_LOCAL_INSTALLATION
		}
	}
	return plugins, nil
}

// sendEvent sends a plugin event on ACS channel.
func sendEvent(ctx context.Context, p *Plugin, evType acpb.PluginEventMessage_PluginEventType, details string) {
	event := &acpb.PluginEventMessage{
		PluginName:     p.Name,
		RevisionId:     p.Revision,
		EventType:      evType,
		EventTimestamp: tpb.New(time.Now()),
		EventDetails:   []byte(details),
	}
	// This might do a retry on the client side if it fails no point in blocking
	// the caller.
	go func() {
		if err := client.Notify(ctx, event); err != nil {
			// Just log the error, Notify() internally handles retrying the request
			// if this fails there's nothing really we can do.
			galog.Errorf("Failed to sent event notification [%+v]: %v", event, err)
		}
	}()
}

// GetLocalPlugin returns the local plugin configuration for the given plugin
// name. If the plugin does not have a local installation, this will return nil.
func (m *PluginManager) GetLocalPlugin(ctx context.Context, name string) (*acpb.ConfigurePluginStates_ConfigurePlugin, error) {
	galog.Debugf("Getting local plugin %q", name)
	localPluginDir := cfg.Retrieve().Plugin.LocalPluginDir
	textprotoFile := filepath.Join(localPluginDir, name, manifestFile)
	contentBytes, err := os.ReadFile(textprotoFile) // NOLINT
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			galog.Debugf("Local plugin %q does not have local manifest file %q, returning nil", name, textprotoFile)
			return nil, nil
		}
		return nil, fmt.Errorf("failed to read local plugin manifest file %q: %w", textprotoFile, err)
	}
	req := new(acpb.ConfigurePluginStates_ConfigurePlugin)
	if err := proto.Unmarshal(contentBytes, req); err != nil {
		return nil, fmt.Errorf("failed to unmarshal manifest file %q into plugin request: %w", textprotoFile, err)
	}

	// Correct the fields that are specific to local plugin manifests.
	req.GetManifest().PluginInstallationType = acpb.PluginInstallationType_LOCAL_INSTALLATION
	req.Action = acpb.ConfigurePluginStates_INSTALL

	// Only for core plugin, set the revision ID to the guest agent version.
	if req.GetPlugin().GetRevisionId() == "" && req.GetPlugin().GetName() == CorePluginName {
		version := cfg.Retrieve().Core.Version
		galog.Debugf("Setting revision ID for core plugin to guest agent version %q", version)
		req.GetPlugin().RevisionId = version
	}
	galog.Debugf("Local plugin %q has local manifest file %q, returning request: %+v", name, textprotoFile, req)
	return req, nil
}

// StartLocalPlugins starts all local plugins that are not running and for which
// there are no dynamically installed versions of the plugin.
//
// The enable map is used to determine which plugins to start. By default, a
// local plugin will be run if no dynamically installed version exists. If the
// enable map is provided, and a local plugin has an entry disabling it, then
// the local plugin will not be started.
func (m *PluginManager) StartLocalPlugins(ctx context.Context, config map[string]LocalPluginInstallation) error {
	galog.Infof("Starting local plugins")
	localPluginDir := cfg.Retrieve().Plugin.LocalPluginDir
	files, err := os.ReadDir(localPluginDir)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			galog.Debugf("Local plugins directory %q does not exist, nothing to start", localPluginDir)
			return nil
		}
		return fmt.Errorf("unable to read local plugins directory %s: %w", localPluginDir, err)
	}

	var installPlugins []string
	configurePlugins := &acpb.ConfigurePluginStates{}
	for _, f := range files {
		if !f.IsDir() {
			continue
		}
		galog.V(1).Debugf("Found local plugin directory %q", f.Name())
		req, err := m.GetLocalPlugin(ctx, f.Name())
		if err != nil {
			return fmt.Errorf("failed to get local plugin %q: %w", f.Name(), err)
		}
		if req == nil {
			galog.Warnf("Local plugin %q does not have local manifest file, skipping local installation", f.Name())
			continue
		}
		pluginName := req.GetPlugin().GetName()

		// Skip plugins that are not enabled.
		installation, ok := config[pluginName]
		if ok && !installation.Enable {
			galog.Infof("Local plugin %q is not enabled, skipping local installation", pluginName)
			continue
		}

		// Check if the plugin is already installed.
		p, err := m.Fetch(pluginName)
		if err == nil {
			// If a dynamically installed version exists, skip the local installation,
			// since the dynamic version will always take precedence.
			if p.IsDynamic() {
				galog.Infof("Plugin %q already dynamically installed with revision %q, skipping local installation", pluginName, p.Revision)
				if installation, ok := config[req.GetPlugin().GetName()]; ok && installation.OnReady != nil {
					galog.Debugf("Running OnReady function for plugin %q", pluginName)
					installation.OnReady(ctx)
				}
				continue
			}
			// If the local version is already installed, check if it's the same
			// revision as the one in the manifest file. If not, we install the new
			// revision over the old one.
			if p.IsLocal() && p.Revision == req.GetPlugin().GetRevisionId() {
				galog.Infof("Plugin %q already locally installed with revision %q, skipping local installation", pluginName, p.Revision)
				// Run the OnReady function for the plugin.
				if installation, ok := config[req.GetPlugin().GetName()]; ok && installation.OnReady != nil {
					galog.Debugf("Running OnReady function for plugin %q", pluginName)
					installation.OnReady(ctx)
				}
				continue
			}
		}
		// Make sure the core plugin is always the first in the list.
		if pluginName == CorePluginName {
			configurePlugins.ConfigurePlugins = append([]*acpb.ConfigurePluginStates_ConfigurePlugin{req}, configurePlugins.ConfigurePlugins...)
		} else {
			configurePlugins.ConfigurePlugins = append(configurePlugins.ConfigurePlugins, req)
		}

		// Add the plugin name to the list of plugins to install.
		installPlugins = append(installPlugins, pluginName)
	}

	if len(configurePlugins.GetConfigurePlugins()) > 0 {
		galog.Infof("Installing local plugins: %v", installPlugins)
		toProcess := m.filterPendingPluginRevisions(ctx, configurePlugins, true)
		var wg sync.WaitGroup
		var nonBootCriticalPlugins []*acpb.ConfigurePluginStates_ConfigurePlugin
		var errs []error
		var errMu sync.Mutex
		for _, req := range toProcess {
			// Each plugin installation is done in parallel. This is to avoid cases
			// where one slow plugin installation blocks the readiness of other
			// plugins.
			//
			// Only block on boot critical plugins. Other plugins can be installed
			// in the background.
			if req.Manifest.ExecutionModel != nil && req.Manifest.ExecutionModel.BootCritical {
				wg.Add(1)
				go func(ctx context.Context, req *acpb.ConfigurePluginStates_ConfigurePlugin, config map[string]LocalPluginInstallation) {
					defer wg.Done()
					if err := m.configureAndVerifyPlugin(ctx, req, config); err != nil {
						errMu.Lock()
						errs = append(errs, err)
						errMu.Unlock()
					}
				}(ctx, req, config)
			} else {
				nonBootCriticalPlugins = append(nonBootCriticalPlugins, req)
			}
		}
		wg.Wait()
		if err := errors.Join(errs...); err != nil {
			return err
		}

		// Run the non-boot critical plugins in the background.
		for _, req := range nonBootCriticalPlugins {
			go func(ctx context.Context, req *acpb.ConfigurePluginStates_ConfigurePlugin, config map[string]LocalPluginInstallation) {
				if err := m.configureAndVerifyPlugin(ctx, req, config); err != nil {
					galog.Warnf("Failed to configure and verify local plugin %q: %v", req.GetPlugin().GetName(), err)
				}
			}(ctx, req, config)
		}
	}
	galog.Infof("No local plugins to install")
	return nil
}

// configureAndVerify configures the plugin and verifies that it is running.
func (m *PluginManager) configureAndVerifyPlugin(ctx context.Context, req *acpb.ConfigurePluginStates_ConfigurePlugin, config map[string]LocalPluginInstallation) error {
	m.configurePlugin(ctx, req)

	// Verify the plugin is running after installation.
	if err := m.VerifyPluginRunning(ctx, req); err != nil {
		return fmt.Errorf("failed to verify local plugin %q is running: %v", req.GetPlugin().GetName(), err)
	}

	// Run the OnReady function if it exists for the plugin.
	galog.Debugf("Running OnReady function for plugin %q", req.GetPlugin().GetName())
	if installation, ok := config[req.GetPlugin().GetName()]; ok && installation.OnReady != nil {
		installation.OnReady(ctx)
	}
	return nil
}

// VerifyPluginRunning verifies that the configured plugins are running.
func (m *PluginManager) VerifyPluginRunning(ctx context.Context, plugin *acpb.ConfigurePluginStates_ConfigurePlugin) error {
	currPlugins := m.list()

	var found bool
	for _, currPlugin := range currPlugins {
		if currPlugin.Name == plugin.GetPlugin().GetName() {
			if currPlugin.State() != acpb.CurrentPluginStates_RUNNING || currPlugin.Revision != plugin.GetPlugin().GetRevisionId() {
				return fmt.Errorf("plugin %q with revision %q is not running, current status: %+v", plugin.GetPlugin().GetName(), plugin.GetPlugin().GetRevisionId(), currPlugin.State())
			}
			found = true
			break
		}
	}
	if !found {
		return fmt.Errorf("plugin %q with revision %q not found", plugin.GetPlugin().GetName(), plugin.GetPlugin().GetRevisionId())
	}
	return nil
}