func (p *pluginControl) SetMonitorOptions(options ...monitorOption) {
	p.pluginRunner.Monitor().Option(options...)
}
SetMonitorOptions exposes monitors options
package control
import (
	"crypto/sha256"
	"errors"
	"fmt"
	"io/ioutil"
	"net"
	"os"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"time"
	log "github.com/sirupsen/logrus"
	"google.golang.org/grpc"
	"github.com/intelsdi-x/gomit"
	"github.com/intelsdi-x/snap/control/plugin"
	"github.com/intelsdi-x/snap/control/plugin/client"
	"github.com/intelsdi-x/snap/control/strategy"
	"github.com/intelsdi-x/snap/core"
	"github.com/intelsdi-x/snap/core/cdata"
	"github.com/intelsdi-x/snap/core/control_event"
	"github.com/intelsdi-x/snap/core/ctypes"
	"github.com/intelsdi-x/snap/core/serror"
	"github.com/intelsdi-x/snap/grpc/controlproxy/rpc"
	"github.com/intelsdi-x/snap/pkg/aci"
	"github.com/intelsdi-x/snap/pkg/psigning"
)
const (
	PluginTrustDisabled int = iota
	PluginTrustEnabled
	PluginTrustWarn
)
var (
	controlLogger = log.WithFields(log.Fields{
		"_module": "control",
	})
	ErrLoadedPluginNotFound = errors.New("Loaded plugin not found")
	ErrControllerNotStarted = errors.New("Must start Controller before use")
)
type pluginControl struct {
	Started bool
	Config  *Config
	autodiscoverPaths []string
	eventManager      *gomit.EventController
	pluginManager  managesPlugins
	metricCatalog  catalogsMetrics
	pluginRunner   runsPlugins
	signingManager managesSigning
	pluginTrust  int
	keyringFiles []string
	grpcServer  *grpc.Server
	closingChan chan bool
	wg          sync.WaitGroup
	subscriptionGroups ManagesSubscriptionGroups
	grpcSecurity       client.GRPCSecurity
}
type subscribedPlugin struct {
	typeName string
	name     string
	version  int
	config   *cdata.ConfigDataNode
}
func (s subscribedPlugin) TypeName() string              { return s.typeName }
func (s subscribedPlugin) Name() string                  { return s.name }
func (s subscribedPlugin) Version() int                  { return s.version }
func (s subscribedPlugin) Config() *cdata.ConfigDataNode { return s.config }
type runsPlugins interface {
	Start() error
	Stop() []error
	AvailablePlugins() *availablePlugins
	AddDelegates(...gomit.Delegator)
	SetEmitter(gomit.Emitter)
	SetMetricCatalog(catalogsMetrics)
	SetPluginManager(managesPlugins)
	Monitor() *monitor
	runPlugin(string, *pluginDetails) error
	SetPluginLoadTimeout(int)
}
type managesPlugins interface {
	teardown()
	get(string) (*loadedPlugin, error)
	all() map[string]*loadedPlugin
	LoadPlugin(*pluginDetails, gomit.Emitter) (*loadedPlugin, serror.SnapError)
	UnloadPlugin(core.Plugin) (*loadedPlugin, serror.SnapError)
	SetMetricCatalog(catalogsMetrics)
	GenerateArgs(logLevel int) plugin.Arg
	SetPluginConfig(*pluginConfig)
	GetPluginConfig() *pluginConfig
	SetPluginTags(map[string]map[string]string)
	AddStandardAndWorkflowTags(core.Metric, map[string]map[string]string) core.Metric
	SetPluginLoadTimeout(int)
}
type catalogsMetrics interface {
	GetMetric(core.Namespace, int) (*metricType, error)
	GetMetrics(core.Namespace, int) ([]*metricType, error)
	Add(*metricType)
	AddLoadedMetricType(*loadedPlugin, core.Metric) error
	RmUnloadedPluginMetrics(lp *loadedPlugin)
	GetVersions(core.Namespace) ([]*metricType, error)
	Fetch(core.Namespace) ([]*metricType, error)
	Keys() []string
	Subscribe([]string, int) error
	Unsubscribe([]string, int) error
	GetPlugin(core.Namespace, int) (core.CatalogedPlugin, error)
	GetPlugins(core.Namespace) ([]core.CatalogedPlugin, error)
}
type managesSigning interface {
	ValidateSignature([]string, string, []byte) error
}
type PluginControlOpt func(*pluginControl)
func MaxRunningPlugins(m int) PluginControlOpt {
	return func(c *pluginControl) {
		strategy.MaximumRunningPlugins = m
	}
}
func CacheExpiration(t time.Duration) PluginControlOpt {
	return func(c *pluginControl) {
		strategy.GlobalCacheExpiration = t
	}
}
func OptSetConfig(cfg *Config) PluginControlOpt {
	return func(c *pluginControl) {
		c.Config = cfg
		c.pluginManager.SetPluginConfig(cfg.Plugins)
		c.pluginManager.SetPluginLoadTimeout(c.Config.PluginLoadTimeout)
		c.pluginRunner.SetPluginLoadTimeout(c.Config.PluginLoadTimeout)
	}
}
func OptSetTags(tags map[string]map[string]string) PluginControlOpt {
	return func(c *pluginControl) {
		c.pluginManager.SetPluginTags(tags)
	}
}
func MaxPluginRestarts(cfg *Config) PluginControlOpt {
	return func(*pluginControl) {
		MaxPluginRestartCount = cfg.MaxPluginRestarts
	}
}
func New(cfg *Config) *pluginControl {
	opts := []PluginControlOpt{
		MaxRunningPlugins(cfg.MaxRunningPlugins),
		CacheExpiration(cfg.CacheExpiration.Duration),
		OptSetConfig(cfg),
		OptSetTags(cfg.Tags),
		MaxPluginRestarts(cfg),
	}
	c := &pluginControl{}
	c.Config = cfg
	c.eventManager = gomit.NewEventController()
	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("pevent controller created")
	c.metricCatalog = newMetricCatalog()
	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("metric catalog created")
	managerOpts := []pluginManagerOpt{
		OptSetPprof(cfg.Pprof),
		OptSetTempDirPath(cfg.TempDirPath),
	}
	runnerOpts := []pluginRunnerOpt{}
	if cfg.IsTLSEnabled() {
		if cfg.CACertPaths != "" {
			certPaths := filepath.SplitList(cfg.CACertPaths)
			c.grpcSecurity = client.SecurityTLSExtended(cfg.TLSCertPath, cfg.TLSKeyPath, client.SecureClient, certPaths)
		} else {
			c.grpcSecurity = client.SecurityTLSEnabled(cfg.TLSCertPath, cfg.TLSKeyPath, client.SecureClient)
		}
		managerOpts = append(managerOpts, OptEnableManagerTLS(c.grpcSecurity))
		runnerOpts = append(runnerOpts, OptEnableRunnerTLS(c.grpcSecurity))
	}
	c.pluginManager = newPluginManager(managerOpts...)
	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("plugin manager created")
	c.pluginManager.SetMetricCatalog(c.metricCatalog)
	c.signingManager = &psigning.SigningManager{}
	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("signing manager created")
	c.pluginRunner = newRunner(runnerOpts...)
	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("runner created")
	c.pluginRunner.AddDelegates(c.eventManager)
	c.pluginRunner.SetEmitter(c.eventManager)
	c.pluginRunner.SetMetricCatalog(c.metricCatalog)
	c.pluginRunner.SetPluginManager(c.pluginManager)
	c.eventManager.RegisterHandler(c.Name(), c)
	c.subscriptionGroups = newSubscriptionGroups(c)
	err := c.pluginRunner.Start()
	if err != nil {
		panic(err)
	}
	for _, opt := range opts {
		opt(c)
	}
	return c
}
func (p *pluginControl) HandleGomitEvent(e gomit.Event) {
	switch v := e.Body.(type) {
	case *control_event.LoadPluginEvent:
		serrs := p.subscriptionGroups.Process()
		if serrs != nil {
			for _, err := range serrs {
				controlLogger.WithFields(log.Fields{
					"_block": "LoadPluginEvent",
				}).Error(err)
			}
		}
	case *control_event.UnloadPluginEvent:
		serrs := p.subscriptionGroups.Process()
		if serrs != nil {
			for _, err := range serrs {
				controlLogger.WithFields(log.Fields{
					"_block": "UnloadPluginEvent",
				}).Error(err)
			}
		}
	default:
		runnerLog.WithFields(log.Fields{
			"_block": "handle-events",
			"event":  v.Namespace(),
		}).Info("Nothing to do for this event")
	}
}
func (p *pluginControl) Name() string {
	return "control"
}
func (p *pluginControl) RegisterEventHandler(name string, h gomit.Handler) error {
	return p.eventManager.RegisterHandler(name, h)
}
func (p *pluginControl) Start() error {
	p.Started = true
	controlLogger.WithFields(log.Fields{
		"_block": "start",
	}).Info("control started")
	if p.Config.AutoDiscoverPath != "" {
		controlLogger.WithFields(log.Fields{
			"_block": "start",
		}).Info("auto discover path is enabled")
		paths := filepath.SplitList(p.Config.AutoDiscoverPath)
		p.SetAutodiscoverPaths(paths)
		for _, pa := range paths {
			fullPath, err := filepath.Abs(pa)
			if err != nil {
				controlLogger.WithFields(log.Fields{
					"_block":           "start",
					"autodiscoverpath": pa,
				}).Fatal(err)
			}
			controlLogger.WithFields(log.Fields{
				"_block": "start",
			}).Info("autoloading plugins from: ", fullPath)
			files, err := ioutil.ReadDir(fullPath)
			if err != nil {
				controlLogger.WithFields(log.Fields{
					"_block":           "start",
					"autodiscoverpath": pa,
				}).Fatal(err)
			}
			for _, file := range files {
				fileName := file.Name()
				statCheck := file
				if file.Mode()&os.ModeSymlink != 0 {
					realPath, err := filepath.EvalSymlinks(filepath.Join(fullPath, fileName))
					if err != nil {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": pa,
							"error":            err,
							"plugin":           fileName,
						}).Error("Cannot follow symlink")
						continue
					}
					statCheck, err = os.Stat(realPath)
					if err != nil {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": pa,
							"error":            err,
							"plugin":           fileName,
							"target-path":      realPath,
						}).Error("Target of symlink inacessible")
						continue
					}
				}
				if statCheck.IsDir() {
					controlLogger.WithFields(log.Fields{
						"_block":           "start",
						"autodiscoverpath": pa,
					}).Warning("Ignoring subdirectory: ", fileName)
					continue
				}
				fname := strings.ToLower(fileName)
				if strings.HasSuffix(fname, ".json") || strings.HasSuffix(fname, ".yaml") || strings.HasSuffix(fname, ".yml") {
					controlLogger.WithFields(log.Fields{
						"_block":           "start",
						"autodiscoverpath": pa,
					}).Warning("Ignoring JSON/Yaml file: ", fileName)
					continue
				}
				if strings.HasSuffix(fileName, ".aci") || !(strings.HasSuffix(fileName, ".asc")) {
					if (statCheck.Mode() & 0111) == 0 {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": pa,
							"plugin":           fileName,
						}).Warn("Auto-loading of plugin '", fileName, "' skipped (plugin not executable)")
						continue
					}
					rp, err := core.NewRequestedPlugin(path.Join(fullPath, fileName), p.GetTempDir(), nil)
					if err != nil {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": pa,
							"plugin":           fileName,
						}).Error(err)
					}
					signatureFile := fileName + ".asc"
					if _, err := os.Stat(path.Join(fullPath, signatureFile)); err == nil {
						err = rp.ReadSignatureFile(path.Join(fullPath, signatureFile))
						if err != nil {
							controlLogger.WithFields(log.Fields{
								"_block":           "start",
								"autodiscoverpath": pa,
								"plugin":           fileName + ".asc",
							}).Error(err)
						}
					}
					pl, err := p.Load(rp)
					if err != nil {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": fullPath,
							"plugin":           fileName,
						}).Error(err)
					} else {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": fullPath,
							"plugin-file-name": fileName,
							"plugin-name":      pl.Name(),
							"plugin-version":   pl.Version(),
							"plugin-type":      pl.TypeName(),
						}).Info("Loading plugin")
					}
				}
			}
		}
	} else {
		controlLogger.WithFields(log.Fields{
			"_block": "start",
		}).Info("auto discover path is disabled")
	}
	lis, err := net.Listen("tcp", fmt.Sprintf("%v:%v", p.Config.ListenAddr, p.Config.ListenPort))
	if err != nil {
		controlLogger.WithField("error", err.Error()).Error("Failed to start control grpc listener")
		return err
	}
	opts := []grpc.ServerOption{}
	p.closingChan = make(chan bool, 1)
	p.grpcServer = grpc.NewServer(opts...)
	rpc.RegisterMetricManagerServer(p.grpcServer, &ControlGRPCServer{p})
	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		err := p.grpcServer.Serve(lis)
		if err != nil {
			select {
			case <-p.closingChan:
			default:
				controlLogger.Fatal(err)
			}
		}
	}()
	return nil
}
func (p *pluginControl) Stop() {
	p.Started = false
	p.closingChan <- true
	p.grpcServer.Stop()
	p.wg.Wait()
	err := p.pluginRunner.Stop()
	if err != nil {
		controlLogger.Error(err)
	}
	for _, rp := range p.pluginRunner.AvailablePlugins().all() {
		controlLogger.Debug("Stopping running plugin")
		rp.Stop("daemon exiting")
	}
	p.pluginManager.teardown()
	controlLogger.WithFields(log.Fields{
		"_block": "stop",
	}).Info("control stopped")
}
func (p *pluginControl) Load(rp *core.RequestedPlugin) (core.CatalogedPlugin, serror.SnapError) {
	f := map[string]interface{}{
		"_block": "load",
	}
	details, serr := p.returnPluginDetails(rp)
	if serr != nil {
		return nil, serr
	}
	if details.IsPackage {
		defer os.RemoveAll(filepath.Dir(details.ExecPath))
	}
	controlLogger.WithFields(f).Info("plugin load called")
	if !p.Started {
		se := serror.New(ErrControllerNotStarted)
		se.SetFields(f)
		controlLogger.WithFields(f).Error(se)
		return nil, se
	}
	pl, se := p.pluginManager.LoadPlugin(details, p.eventManager)
	if se != nil {
		return nil, se
	}
	if pl.Details.IsPackage {
		pl.Details.ExecPath = ""
	}
	event := &control_event.LoadPluginEvent{
		Name:    pl.Meta.Name,
		Version: pl.Meta.Version,
		Type:    int(pl.Meta.Type),
		Signed:  pl.Details.Signed,
	}
	defer p.eventManager.Emit(event)
	return pl, nil
}
func (p *pluginControl) verifySignature(rp *core.RequestedPlugin) (bool, serror.SnapError) {
	f := map[string]interface{}{
		"_block": "verifySignature",
	}
	switch p.pluginTrust {
	case PluginTrustDisabled:
		return false, nil
	case PluginTrustEnabled:
		err := p.signingManager.ValidateSignature(p.keyringFiles, rp.Path(), rp.Signature())
		if err != nil {
			return false, serror.New(err)
		}
	case PluginTrustWarn:
		if rp.Signature() == nil {
			controlLogger.WithFields(f).Warn("Loading unsigned plugin ", rp.Path())
			return false, nil
		}
		err := p.signingManager.ValidateSignature(p.keyringFiles, rp.Path(), rp.Signature())
		if err != nil {
			return false, serror.New(err)
		}
	}
	return true, nil
}
func (p *pluginControl) returnPluginDetails(rp *core.RequestedPlugin) (*pluginDetails, serror.SnapError) {
	details := &pluginDetails{}
	var serr serror.SnapError
	details.Signed, serr = p.verifySignature(rp)
	if serr != nil {
		return nil, serr
	}
	details.Path = rp.Path()
	details.CheckSum = rp.CheckSum()
	details.Signature = rp.Signature()
	details.CertPath = rp.CertPath()
	details.KeyPath = rp.KeyPath()
	details.CACertPaths = rp.CACertPaths()
	details.TLSEnabled = rp.TLSEnabled()
	details.Uri = rp.Uri()
	if rp.Uri() != nil {
	} else if filepath.Ext(rp.Path()) == ".aci" {
		f, err := os.Open(rp.Path())
		if err != nil {
			return nil, serror.New(err)
		}
		defer f.Close()
		if err := aci.Validate(f); err != nil {
			return nil, serror.New(err)
		}
		tempPath, err := aci.Extract(f)
		if err != nil {
			return nil, serror.New(err)
		}
		details.ExecPath = path.Join(tempPath, "rootfs")
		if details.Manifest, err = aci.Manifest(f); err != nil {
			return nil, serror.New(err)
		}
		details.Exec = details.Manifest.App.Exec
		details.IsPackage = true
	} else {
		details.IsPackage = false
		details.Exec = []string{filepath.Base(rp.Path())}
		details.ExecPath = filepath.Dir(rp.Path())
	}
	return details, nil
}
func (p *pluginControl) Unload(pl core.Plugin) (core.CatalogedPlugin, serror.SnapError) {
	up, err := p.pluginManager.get(fmt.Sprintf("%s"+core.Separator+"%s"+core.Separator+"%d", pl.TypeName(), pl.Name(), pl.Version()))
	if err != nil {
		se := serror.New(ErrPluginNotFound, map[string]interface{}{
			"plugin-name":    pl.Name(),
			"plugin-version": pl.Version(),
			"plugin-type":    pl.TypeName(),
		})
		return nil, se
	}
	if errs := p.subscriptionGroups.validatePluginUnloading(up); errs != nil {
		impactOnTasks := []string{}
		for _, err := range errs {
			taskId := err.Fields()["task-id"].(string)
			impactOnTasks = append(impactOnTasks, taskId)
		}
		se := serror.New(errorPluginCannotBeUnloaded(impactOnTasks), map[string]interface{}{
			"plugin-name":    pl.Name(),
			"plugin-version": pl.Version(),
			"plugin-type":    pl.TypeName(),
			"impacted-tasks": impactOnTasks,
		})
		return nil, se
	}
	if _, err := p.pluginManager.UnloadPlugin(pl); err != nil {
		return nil, err
	}
	event := &control_event.UnloadPluginEvent{
		Name:    up.Meta.Name,
		Version: up.Meta.Version,
		Type:    int(up.Meta.Type),
	}
	defer p.eventManager.Emit(event)
	return up, nil
}
func (p *pluginControl) SwapPlugins(in *core.RequestedPlugin, out core.CatalogedPlugin) serror.SnapError {
	details, serr := p.returnPluginDetails(in)
	if serr != nil {
		return serr
	}
	if details.IsPackage {
		defer os.RemoveAll(filepath.Dir(details.ExecPath))
	}
	lp, err := p.pluginManager.LoadPlugin(details, p.eventManager)
	if err != nil {
		return err
	}
	if lp.TypeName() != out.TypeName() || lp.Name() != out.Name() {
		serr := serror.New(errors.New("Plugin types and names must match."))
		serr.SetFields(map[string]interface{}{
			"in-type":  lp.TypeName(),
			"out-type": out.TypeName(),
			"in-name":  lp.Name(),
			"out-name": out.Name(),
		})
		_, err := p.pluginManager.UnloadPlugin(lp)
		if err != nil {
			se := serror.New(errors.New("Failed to rollback after error"))
			se.SetFields(map[string]interface{}{
				"original-unload-error": serr.Error(),
				"rollback-unload-error": err.Error(),
			})
			return se
		}
		return serr
	}
	up, err := p.pluginManager.UnloadPlugin(out)
	if err != nil {
		_, err2 := p.pluginManager.UnloadPlugin(lp)
		if err2 != nil {
			se := serror.New(errors.New("Failed to rollback after error"))
			se.SetFields(map[string]interface{}{
				"original-unload-error": err.Error(),
				"rollback-unload-error": err2.Error(),
			})
			return se
		}
		return err
	}
	event := &control_event.SwapPluginsEvent{
		LoadedPluginName:      lp.Meta.Name,
		LoadedPluginVersion:   lp.Meta.Version,
		UnloadedPluginName:    up.Meta.Name,
		UnloadedPluginVersion: up.Meta.Version,
		PluginType:            int(lp.Meta.Type),
	}
	defer p.eventManager.Emit(event)
	return nil
}
func (p *pluginControl) ValidateDeps(requested []core.RequestedMetric, plugins []core.SubscribedPlugin, configTree *cdata.ConfigDataTree, asserts ...core.SubscribedPluginAssert) []serror.SnapError {
	return p.subscriptionGroups.ValidateDeps(requested, plugins, configTree, asserts...)
}
func (p *pluginControl) SubscribeDeps(id string, requested []core.RequestedMetric, plugins []core.SubscribedPlugin, configTree *cdata.ConfigDataTree) (serrs []serror.SnapError) {
	return p.subscriptionGroups.Add(id, requested, configTree, plugins)
}
func (p *pluginControl) UnsubscribeDeps(id string) []serror.SnapError {
	return p.subscriptionGroups.Remove(id)
}
func (p *pluginControl) verifyPlugin(lp *loadedPlugin) error {
	if lp.Details.Uri != nil {
		if core.IsUri(lp.Details.Uri.String()) {
			return fmt.Errorf(fmt.Sprintf("Remote plugin failed to load: bad uri: (%x)", lp.Details.Uri))
		}
		return nil
	}
	b, err := ioutil.ReadFile(lp.Details.Path)
	if err != nil {
		return err
	}
	cs := sha256.Sum256(b)
	if lp.Details.CheckSum != cs {
		return fmt.Errorf(fmt.Sprintf("Current plugin checksum (%x) does not match checksum when plugin was first loaded (%x).", cs, lp.Details.CheckSum))
	}
	if lp.Details.Signed {
		return p.signingManager.ValidateSignature(p.keyringFiles, lp.Details.Path, lp.Details.Signature)
	}
	return nil
}
func (p *pluginControl) getMetricsAndCollectors(requested []core.RequestedMetric, configTree *cdata.ConfigDataTree) (map[string]metricTypes, []core.SubscribedPlugin, []serror.SnapError) {
	newMetricsGroupedByPlugin := make(map[string]metricTypes)
	newPlugins := []core.SubscribedPlugin{}
	var serrs []serror.SnapError
	for _, r := range requested {
		newMetrics, err := p.metricCatalog.GetMetrics(r.Namespace(), r.Version())
		if err != nil {
			log.WithFields(log.Fields{
				"_block": "control",
				"action": "expanding-requested-metrics",
				"query":  r.Namespace(),
				"err":    err,
			}).Error("error matching requested namespace with metric catalog")
			serrs = append(serrs, serror.New(err))
			continue
		}
		if controlLogger.Level >= log.DebugLevel {
			for _, m := range newMetrics {
				controlLogger.WithFields(log.Fields{
					"_block":  "control",
					"ns":      m.Namespace().String(),
					"version": m.Version(),
				}).Debug("Expanded namespaces found in metric catalog")
			}
		}
		for _, mt := range newMetrics {
			cfg := configTree.Get(mt.Namespace().Strings())
			if cfg == nil {
				cfg = cdata.NewNode()
			}
			mt.config = cfg
			plType, _ := core.ToPluginType(mt.Plugin.TypeName())
			cfgNode := p.pluginManager.GetPluginConfig().getPluginConfigDataNode(plType, mt.Plugin.Name(), mt.Plugin.Version())
			cfg.ApplyDefaults(cfgNode.Table())
			if pluginCfg := mt.Plugin.Policy().Get(mt.Namespace().Strings()); pluginCfg != nil {
				mt.config.ApplyDefaults(pluginCfg.Defaults())
			}
			cp := mt.Plugin
			key := fmt.Sprintf("%s"+core.Separator+"%s"+core.Separator+"%d", cp.TypeName(), cp.Name(), cp.Version())
			pmt, _ := newMetricsGroupedByPlugin[key]
			pmt.plugin = cp
			pmt.metricTypes = append(pmt.metricTypes, mt)
			newMetricsGroupedByPlugin[key] = pmt
			plugin := subscribedPlugin{
				name:     cp.Name(),
				typeName: cp.TypeName(),
				version:  cp.Version(),
				config:   cdata.NewNode(),
			}
			if !containsPlugin(newPlugins, plugin) {
				newPlugins = append(newPlugins, plugin)
			}
		}
	}
	if controlLogger.Level >= log.DebugLevel {
		for _, pmt := range newMetricsGroupedByPlugin {
			for _, m := range pmt.Metrics() {
				log.WithFields(log.Fields{
					"_block": "control",
					"action": "gather",
					"metric": fmt.Sprintf("%s:%d", m.Namespace().String(), m.Version()),
				}).Debug("gathered metrics from workflow request")
			}
		}
		for _, p := range newPlugins {
			log.WithFields(log.Fields{
				"_block": "control",
				"action": "gather",
				"metric": fmt.Sprintf("%s:%s:%d", p.TypeName(), p.Name(), p.Version()),
			}).Debug("gathered plugins from workflow request")
		}
	}
	return newMetricsGroupedByPlugin, newPlugins, serrs
}