control.go

/*
http://www.apache.org/licenses/LICENSE-2.0.txt


Copyright 2015-2016 Intel Corporation

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

    http://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 control

import (
	"crypto/sha256"
	"errors"
	"fmt"
	"io/ioutil"
	"net"
	"os"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"google.golang.org/grpc"

	log "github.com/Sirupsen/logrus"

	"github.com/intelsdi-x/gomit"

	"github.com/intelsdi-x/snap/control/plugin"
	"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 - enum representing plugin trust disabled
	PluginTrustDisabled int = iota
	// PluginTrustEnabled - enum representing plugin trust enabled
	PluginTrustEnabled
	// PluginTrustWarn - enum representing plugin trust warning
	PluginTrustWarn
)

var (
	controlLogger = log.WithFields(log.Fields{
		"_module": "control",
	})

	// ErrLoadedPluginNotFound - error message when a loaded plugin is not found
	ErrLoadedPluginNotFound = errors.New("Loaded plugin not found")

	// ErrControllerNotStarted - error message when the Controller was not started
	ErrControllerNotStarted = errors.New("Must start Controller before use")
)

type pluginControl struct {
	// TODO, going to need coordination on changing of these
	Started bool
	Config  *Config

	autodiscoverPaths []string
	eventManager      *gomit.EventController

	pluginManager  managesPlugins
	metricCatalog  catalogsMetrics
	pluginRunner   runsPlugins
	signingManager managesSigning

	pluginTrust  int
	keyringFiles []string
	// used to cleanly shutdown the GRPC server
	grpcServer  *grpc.Server
	closingChan chan bool
	wg          sync.WaitGroup

	subscriptionGroups ManagesSubscriptionGroups
}

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(*pluginDetails) error
}

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)
	SetPluginLoadTimeout(int)
}

type catalogsMetrics interface {
	Get(core.Namespace, int) (*metricType, error)
	GetQueriedNamespaces(core.Namespace) ([]core.Namespace, error)
	UpdateQueriedNamespaces(core.Namespace)
	Add(*metricType)
	AddLoadedMetricType(*loadedPlugin, core.Metric) error
	RmUnloadedPluginMetrics(lp *loadedPlugin)
	GetVersions(core.Namespace) ([]*metricType, error)
	Fetch(core.Namespace) ([]*metricType, error)
	Item() (string, []*metricType)
	Next() bool
	Keys() []string
	Subscribe([]string, int) error
	Unsubscribe([]string, int) error
	GetPlugin(core.Namespace, int) (*loadedPlugin, error)
}

type managesSigning interface {
	ValidateSignature([]string, string, []byte) error
}

// PluginControlOpt is used to set optional parameters on the pluginControl struct
type PluginControlOpt func(*pluginControl)

// MaxRunningPlugins sets the maximum number of plugins to run per pool
func MaxRunningPlugins(m int) PluginControlOpt {
	return func(c *pluginControl) {
		strategy.MaximumRunningPlugins = m
	}
}

// CacheExpiration is the PluginControlOpt which sets the global metric cache TTL
func CacheExpiration(t time.Duration) PluginControlOpt {
	return func(c *pluginControl) {
		strategy.GlobalCacheExpiration = t
	}
}

// OptSetConfig sets the plugin control configuration.
func OptSetConfig(cfg *Config) PluginControlOpt {
	return func(c *pluginControl) {
		c.Config = cfg
		c.pluginManager.SetPluginConfig(cfg.Plugins)
		c.pluginManager.SetPluginLoadTimeout(c.Config.PluginLoadTimeout)
	}
}

// New returns a new pluginControl instance
func New(cfg *Config) *pluginControl {
	// construct a slice of options from the input configuration
	opts := []PluginControlOpt{
		MaxRunningPlugins(cfg.MaxRunningPlugins),
		CacheExpiration(cfg.CacheExpiration.Duration),
		OptSetConfig(cfg),
	}
	c := &pluginControl{}
	c.Config = cfg
	// Initialize components
	// Event Manager
	c.eventManager = gomit.NewEventController()

	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("pevent controller created")

	// Metric Catalog
	c.metricCatalog = newMetricCatalog()
	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("metric catalog created")

	// Plugin Manager
	c.pluginManager = newPluginManager()
	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("plugin manager created")
	// Plugin Manager needs a reference to the metric catalog
	c.pluginManager.SetMetricCatalog(c.metricCatalog)

	// Signing Manager
	c.signingManager = &psigning.SigningManager{}
	controlLogger.WithFields(log.Fields{
		"_block": "new",
	}).Debug("signing manager created")

	// Plugin Runner
	c.pluginRunner = newRunner()
	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)

	// Pass runner events to control main module
	c.eventManager.RegisterHandler(c.Name(), c)

	// Create subscription group - used for managing a group of subscriptions
	c.subscriptionGroups = newSubscriptionGroups(c)

	// Start stuff
	err := c.pluginRunner.Start()
	if err != nil {
		panic(err)
	}

	// apply options

	// it is important that this happens last, as an option may
	// require that an internal member of c be constructed.
	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)
}

// Begin handling load, unload, and inventory
func (p *pluginControl) Start() error {
	// Start pluginManager when pluginControl starts
	p.Started = true
	controlLogger.WithFields(log.Fields{
		"_block": "start",
	}).Info("control started")

	//Autodiscover
	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 {
				if file.IsDir() {
					controlLogger.WithFields(log.Fields{
						"_block":           "start",
						"autodiscoverpath": pa,
					}).Warning("Ignoring subdirectory: ", file.Name())
					continue
				}
				// Ignore tasks files (JSON and YAML)
				fname := strings.ToLower(file.Name())
				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: ", file.Name())
					continue
				}
				// if the file is a plugin package (which would have a suffix of '.aci') or if the file
				// is not a plugin signing file (which would have a suffix of '.asc'), then attempt to
				// automatically load the file as a plugin
				if strings.HasSuffix(file.Name(), ".aci") || !(strings.HasSuffix(file.Name(), ".asc")) {
					// check to makd sure the file is executable by someone (even if it isn't you); if no one
					// can execute this file then skip it (and include a warning in the log output)
					if (file.Mode() & 0111) == 0 {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": pa,
							"plugin":           file,
						}).Warn("Auto-loading of plugin '", file.Name(), "' skipped (plugin not executable)")
						continue
					}
					rp, err := core.NewRequestedPlugin(path.Join(fullPath, file.Name()))
					if err != nil {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": pa,
							"plugin":           file,
						}).Error(err)
					}
					signatureFile := file.Name() + ".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":           file.Name() + ".asc",
							}).Error(err)
						}
					}
					pl, err := p.Load(rp)
					if err != nil {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": fullPath,
							"plugin":           file,
						}).Error(err)
					} else {
						controlLogger.WithFields(log.Fields{
							"_block":           "start",
							"autodiscoverpath": fullPath,
							"plugin-file-name": file.Name(),
							"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:
			// If we called Stop() then there will be a value in p.closingChan, so
			// we'll get here and we can exit without showing the error.
			default:
				controlLogger.Fatal(err)
			}
		}
	}()

	return nil
}

func (p *pluginControl) Stop() {
	// set the Started flag to false (since we're stopping the server)
	p.Started = false

	// and add a boolean to the p.closingChan (used for error handling in the
	// goroutine that is listening for connections)
	p.closingChan <- true

	// stop GRPC server
	p.grpcServer.Stop()
	p.wg.Wait()

	// stop runner
	err := p.pluginRunner.Stop()
	if err != nil {
		controlLogger.Error(err)
	}

	// stop running plugins
	for _, rp := range p.pluginRunner.AvailablePlugins().all() {
		controlLogger.Debug("Stopping running plugin")
		rp.Stop("daemon exiting")
	}

	// unload plugins
	p.pluginManager.teardown()

	// log that we've stopped the control module
	controlLogger.WithFields(log.Fields{
		"_block": "stop",
	}).Info("control stopped")

}

// Load is the public method to load a plugin into
// the LoadedPlugins array and issue an event when
// successful.
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 plugin was loaded from a package, remove ExecPath for
	// the temporary plugin that was used for load
	if pl.Details.IsPackage {
		pl.Details.ExecPath = ""
	}

	// defer sending event
	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
	//Check plugin signing
	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.IsAutoLoaded = rp.AutoLoaded()

	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[0]
		details.IsPackage = true
	} else {
		details.IsPackage = false
		details.Exec = 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.UnloadPlugin(pl)
	if 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
	}

	// Make sure plugin types and names are the same
	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
}

// MatchQueryToNamespaces performs the process of matching the 'ns' with namespaces of all cataloged metrics
func (p *pluginControl) matchQueryToNamespaces(ns core.Namespace) ([]core.Namespace, serror.SnapError) {
	// carry out the matching process
	p.metricCatalog.UpdateQueriedNamespaces(ns)
	nss, err := p.metricCatalog.GetQueriedNamespaces(ns)
	if err != nil {
		return nil, serror.New(err)
	}
	return nss, nil
}

func (p *pluginControl) ValidateDeps(requested []core.RequestedMetric, plugins []core.SubscribedPlugin, configTree *cdata.ConfigDataTree) []serror.SnapError {
	return p.subscriptionGroups.ValidateDeps(requested, plugins, configTree)
}

// SubscribeDeps will subscribe to collectors, processors and publishers.  The collectors are subscribed by mapping the provided
// array of core.RequestedMetrics to the corresponding plugins while processors and publishers provided in the array of core.Plugin
// will be subscribed directly.  The ID provides a logical grouping of subscriptions.
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)
}

// UnsubscribeDeps unsubscribes a group of dependencies provided the subscription group ID
func (p *pluginControl) UnsubscribeDeps(id string) []serror.SnapError {
	// update view and unsubscribe to plugins
	return p.subscriptionGroups.Remove(id)
}

func (p *pluginControl) verifyPlugin(lp *loadedPlugin) error {
	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) ([]core.Metric, []core.SubscribedPlugin, []serror.SnapError) {
	newMetrics := []core.Metric{}
	newPlugins := []core.SubscribedPlugin{}
	var serrs []serror.SnapError
	for _, r := range requested {
		// get expanded namespaces from requested metrics
		newNss, err := p.matchQueryToNamespaces(r.Namespace())
		if err != nil {
			log.WithFields(log.Fields{
				"_block": "control",
				"action": "expanding-requested-metrics",
				"query":  r.Namespace(),
			}).Error("error matching requested namespace with metric catalog")
			serrs = append(serrs, err)
			continue
		}

		if controlLogger.Level >= log.DebugLevel {
			for _, n := range newNss {
				controlLogger.WithFields(log.Fields{
					"_block": "control",
					"ns":     n.String(),
				}).Debug("Expanded namespaces found")
			}
		}

		if len(newNss) > 0 {
			for _, ns := range newNss {
				// Get metric types from metric catalog
				m, err := p.metricCatalog.Get(ns, r.Version())
				if err != nil {
					log.WithFields(log.Fields{
						"_block":    "control",
						"action":    "expanding requested metrics",
						"namespace": ns.String(),
						"version":   r.Version(),
					}).Error("error retreiving metric given a namespace and version.")
					serrs = append(serrs, serror.New(fmt.Errorf("error retreiving metric %s:%d", ns.String(), r.Version())))
					continue
				}
				// in case config tree doesn't have any configuration for current namespace
				// it's needed to initialize config, otherwise it will stay nil and panic later on
				config := configTree.Get(ns.Strings())
				if config == nil {
					config = cdata.NewNode()
				}
				newMetrics = append(newMetrics, &metric{
					namespace: m.Namespace(),
					version:   m.Version(),
					config:    config,
				})

				config = configTree.Get([]string{""})
				if config == nil {
					config = cdata.NewNode()
				}
				plugin := subscribedPlugin{
					name:     m.Plugin.Name(),
					typeName: m.Plugin.TypeName(),
					version:  m.Plugin.Version(),
					config:   config,
				}

				if !containsPlugin(newPlugins, plugin) {
					newPlugins = append(newPlugins, plugin)
				}
			}
		}
	}

	if controlLogger.Level >= log.DebugLevel {
		for _, m := range newMetrics {
			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 newMetrics, newPlugins, serrs
}

// SetMonitorOptions exposes monitors options
func (p *pluginControl) SetMonitorOptions(options ...monitorOption) {
	p.pluginRunner.Monitor().Option(options...)
}

// returns the loaded plugin collection
// NOTE: The returned data from this function should be considered constant and read only
func (p *pluginControl) PluginCatalog() core.PluginCatalog {
	table := p.pluginManager.all()
	plugins := make([]core.CatalogedPlugin, len(table))
	i := 0
	for _, plugin := range table {
		plugins[i] = plugin
		i++
	}
	return plugins
}

// AvailablePlugins returns pointers to all the running plugins in the pools
// NOTE: The returned data from this function should be considered constant and read only
func (p *pluginControl) AvailablePlugins() []core.AvailablePlugin {
	var caps []core.AvailablePlugin
	for _, ap := range p.pluginRunner.AvailablePlugins().all() {
		caps = append(caps, ap)
	}
	return caps
}

// MetricCatalog returns the entire metric catalog
// NOTE: The returned data from this function should be considered constant and read only
func (p *pluginControl) MetricCatalog() ([]core.CatalogedMetric, error) {
	return p.FetchMetrics(core.Namespace{}, 0)
}

// FetchMetrics returns the metrics which fall under the given namespace
// NOTE: The returned data from this function should be considered constant and read only
func (p *pluginControl) FetchMetrics(ns core.Namespace, version int) ([]core.CatalogedMetric, error) {
	mts, err := p.metricCatalog.Fetch(ns)
	if err != nil {
		return nil, err
	}
	cmt := make([]core.CatalogedMetric, 0, len(mts))
	nsMap := map[string]struct{}{}
	for _, mt := range mts {
		if version > 0 {
			// a version is specified
			if mt.version == version {
				cmt = append(cmt, mt)
			}
		} else if version < 0 {
			// -1 (or less) is specified meaning return the latest
			if _, ok := nsMap[mt.Namespace().String()]; !ok {
				mt, err = p.metricCatalog.Get(mt.Namespace(), version)
				if err != nil {
					return nil, err
				}
				cmt = append(cmt, mt)
				nsMap[mt.Namespace().String()] = struct{}{}
			}
		} else {
			// no version is specified return all metric versions
			cmt = append(cmt, mt)

		}
	}
	return cmt, nil
}

func (p *pluginControl) GetMetric(ns core.Namespace, ver int) (core.CatalogedMetric, error) {
	return p.metricCatalog.Get(ns, ver)
}

func (p *pluginControl) GetMetricVersions(ns core.Namespace) ([]core.CatalogedMetric, error) {
	mts, err := p.metricCatalog.GetVersions(ns)
	if err != nil {
		return nil, err
	}

	rmts := make([]core.CatalogedMetric, len(mts))
	for i, m := range mts {
		rmts[i] = m
	}
	return rmts, nil
}

func (p *pluginControl) MetricExists(mns core.Namespace, ver int) bool {
	_, err := p.metricCatalog.Get(mns, ver)
	if err == nil {
		return true
	}
	return false
}

// CollectMetrics is a blocking call to collector plugins returning a collection
// of metrics and errors.  If an error is encountered no metrics will be
// returned.
func (p *pluginControl) CollectMetrics(id string, allTags map[string]map[string]string) (metrics []core.Metric, errs []error) {
	// If control is not started we don't want tasks to be able to
	// go through a workflow.
	if !p.Started {
		return nil, []error{ErrControllerNotStarted}
	}

	// Subscription groups are processed anytime a plugin is loaded/unloaded.
	results, serrs, err := p.subscriptionGroups.Get(id)
	if err != nil {
		controlLogger.WithFields(log.Fields{
			"_block":                "CollectorMetrics",
			"subscription-group-id": id,
		}).Error(err)
		errs = append(errs, err)
		return
	}
	// If We received errors when the requested metrics were last processed
	// against the metric catalog we need to return them to the caller.
	if serrs != nil {
		for _, e := range serrs {
			errs = append(errs, e)
		}
	}

	for ns, nsTags := range allTags {
		for k, v := range nsTags {
			log.WithFields(log.Fields{
				"_module": "control",
				"block":   "CollectMetrics",
				"type":    "pluginCollector",
				"ns":      ns,
				"tag-key": k,
				"tag-val": v,
			}).Debug("Tags in CollectMetrics")
		}
	}

	pluginToMetricMap, err := groupMetricTypesByPlugin(p.metricCatalog, results)
	if err != nil {
		errs = append(errs, err)
		return
	}

	cMetrics := make(chan []core.Metric)
	cError := make(chan error)
	var wg sync.WaitGroup

	// For each available plugin call available plugin using RPC client and wait for response (goroutines)
	for pluginKey, pmt := range pluginToMetricMap {
		// merge global plugin config into the config for the metric
		for _, mt := range pmt.metricTypes {
			if mt.Config() != nil {
				mt.Config().ReverseMerge(p.Config.Plugins.getPluginConfigDataNode(core.CollectorPluginType, pmt.plugin.Name(), pmt.plugin.Version()))
			}
		}

		wg.Add(1)

		go func(pluginKey string, mt []core.Metric) {
			mts, err := p.pluginRunner.AvailablePlugins().collectMetrics(pluginKey, mt, id)
			if err != nil {
				cError <- err
			} else {
				cMetrics <- mts
			}
		}(pluginKey, pmt.metricTypes)
	}

	go func() {
		for m := range cMetrics {
			// Reapply standard tags after collection as a precaution.  It is common for
			// plugin authors to inadvertently overwrite or not pass along the data
			// passed to CollectMetrics so we will help them out here.
			for i := range m {
				m[i] = addStandardAndWorkflowTags(m[i], allTags)
			}
			metrics = append(metrics, m...)
			wg.Done()
		}
	}()

	go func() {
		for e := range cError {
			errs = append(errs, e)
			wg.Done()
		}
	}()

	wg.Wait()
	close(cMetrics)
	close(cError)

	if len(errs) > 0 {
		return nil, errs
	}
	return
}

// PublishMetrics
func (p *pluginControl) PublishMetrics(metrics []core.Metric, config map[string]ctypes.ConfigValue, taskID, pluginName string, pluginVersion int) []error {
	// If control is not started we don't want tasks to be able to
	// go through a workflow.
	if !p.Started {
		return []error{ErrControllerNotStarted}
	}
	// merge global plugin config into the config for this request
	// without over-writing the task specific config
	cfg := p.Config.Plugins.getPluginConfigDataNode(core.PublisherPluginType, pluginName, pluginVersion).Table()
	merged := make(map[string]ctypes.ConfigValue)
	for k, v := range cfg {
		merged[k] = v
	}
	for k, v := range config {
		merged[k] = v
	}

	return p.pluginRunner.AvailablePlugins().publishMetrics(metrics, pluginName, pluginVersion, merged, taskID)
}

// ProcessMetrics
func (p *pluginControl) ProcessMetrics(metrics []core.Metric, config map[string]ctypes.ConfigValue, taskID, pluginName string, pluginVersion int) ([]core.Metric, []error) {
	// If control is not started we don't want tasks to be able to
	// go through a workflow.
	if !p.Started {
		return nil, []error{ErrControllerNotStarted}
	}
	// merge global plugin config into the config for this request
	// without over-writing the task specific config
	cfg := p.Config.Plugins.getPluginConfigDataNode(core.ProcessorPluginType, pluginName, pluginVersion).Table()
	merged := make(map[string]ctypes.ConfigValue)
	for k, v := range cfg {
		merged[k] = v
	}
	for k, v := range config {
		merged[k] = v
	}

	return p.pluginRunner.AvailablePlugins().processMetrics(metrics, pluginName, pluginVersion, merged, taskID)
}

func (p *pluginControl) SetAutodiscoverPaths(paths []string) {
	p.autodiscoverPaths = paths
}

func (p *pluginControl) GetAutodiscoverPaths() []string {
	return p.autodiscoverPaths
}

func (p *pluginControl) SetPluginTrustLevel(trust int) {
	p.pluginTrust = trust
}

func (p *pluginControl) SetKeyringFile(keyring string) {
	p.keyringFiles = append(p.keyringFiles, keyring)
}

type requestedPlugin struct {
	name    string
	version int
	config  *cdata.ConfigDataNode
}

func (r *requestedPlugin) Name() string {
	return r.name
}

func (r *requestedPlugin) Version() int {
	return r.version
}

func (r *requestedPlugin) Config() *cdata.ConfigDataNode {
	return r.config
}

// ------------------- helper struct and function for grouping metrics types ------

// just a tuple of loadedPlugin and metricType slice
type metricTypes struct {
	plugin      *loadedPlugin
	metricTypes []core.Metric
}

func (p *metricTypes) Count() int {
	return len(p.metricTypes)
}

// groupMetricTypesByPlugin groups metricTypes by a plugin.Key() and returns appropriate structure
func groupMetricTypesByPlugin(cat catalogsMetrics, mts []core.Metric) (map[string]metricTypes, serror.SnapError) {
	pmts := make(map[string]metricTypes)
	// For each plugin type select a matching available plugin to call
	for _, incomingmt := range mts {
		version := incomingmt.Version()
		if version == 0 {
			// If the version is not provided we will choose the latest
			version = -1
		}
		catalogedmt, err := cat.Get(incomingmt.Namespace(), version)
		if err != nil {
			return nil, serror.New(err)
		}
		returnedmt := plugin.MetricType{
			Namespace_:          catalogedmt.Namespace(),
			LastAdvertisedTime_: catalogedmt.LastAdvertisedTime(),
			Version_:            incomingmt.Version(),
			Tags_:               catalogedmt.Tags(),
			Config_:             incomingmt.Config(),
			Unit_:               catalogedmt.Unit(),
		}
		lp := catalogedmt.Plugin
		if lp == nil {
			return nil, serror.New(errorMetricNotFound(incomingmt.Namespace().String()))
		}
		key := lp.Key()
		pmt, _ := pmts[key]
		pmt.plugin = lp
		pmt.metricTypes = append(pmt.metricTypes, returnedmt)
		pmts[key] = pmt
	}
	return pmts, nil
}

func containsPlugin(slice []core.SubscribedPlugin, lookup subscribedPlugin) bool {
	for _, plugin := range slice {
		if plugin.Name() == lookup.Name() &&
			plugin.Version() == lookup.Version() &&
			plugin.TypeName() == lookup.TypeName() {
			return true
		}
	}
	return false
}