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
}

// Read an input configuration file, parsing it (as YAML or JSON)
// into the input 'interface{}', v
func Read(path string, v interface{}, schema string) []serror.SnapError {
	// read bytes from file
	b, err := cfgReader.ReadFile(path)
	if err != nil {
		return []serror.SnapError{serror.New(err)}
	}
	// convert from YAML to JSON (remember, JSON is actually valid YAML)
	jb, err := yaml.YAMLToJSON(b)
	if err != nil {
		return []serror.SnapError{serror.New(fmt.Errorf("error converting YAML to JSON: %v", err))}
	}
	// validate the resulting JSON against the input the schema
	if errors := cfgValidator.validateSchema(schema, string(jb)); errors != nil {
		// if invalid, construct (and return?) a SnapError from the errors identified
		// during schema validation
		return errors
	}
	// if valid, parse the JSON byte-stream (above)
	if parseErr := json.Unmarshal(jb, v); parseErr != nil {
		// remove any YAML-specific prefix that might have been added by then
		// yaml.Unmarshal() method or JSON-specific prefix that might have been
		// added if the resulting JSON string could not be marshalled into our
		// input interface correctly (note, if there is no match to either of
		// these prefixes then the error message will be passed through unchanged)
		tmpErr := strings.TrimPrefix(parseErr.Error(), "error converting YAML to JSON: yaml: ")
		errRet := strings.TrimPrefix(tmpErr, "error unmarshaling JSON: json: ")
		return []serror.SnapError{serror.New(fmt.Errorf("Error while parsing configuration file: %v", errRet))}
	}
	return nil
}

// and define an implementation for that type that performs the schema validation
func (r *schemaValidatorType) validateSchema(schema, cfg string) []serror.SnapError {
	schemaLoader := gojsonschema.NewStringLoader(schema)
	testDoc := gojsonschema.NewStringLoader(cfg)
	result, err := gojsonschema.Validate(schemaLoader, testDoc)
	var serrors []serror.SnapError
	// Check for invalid json
	if err != nil {
		serrors = append(serrors, serror.New(err))
		return serrors
	}
	// check if result passes validation
	if result.Valid() {
		return nil
	}
	for _, err := range result.Errors() {
		serr := serror.New(errors.New("Validate schema error"))
		serr.SetFields(map[string]interface{}{
			"value":       err.Value(),
			"context":     err.Context().String("::"),
			"description": err.Description(),
		})
		serrors = append(serrors, serr)
	}
	return serrors
}

// groupMetricTypesByPlugin groups metricTypes by a plugin.Key() and returns appropriate structure
func groupMetricTypesByPlugin(cat catalogsMetrics, metricTypes []core.Metric) (map[string]pluginMetricTypes, serror.SnapError) {
	pmts := make(map[string]pluginMetricTypes)
	// For each plugin type select a matching available plugin to call
	for _, mt := range metricTypes {
		version := mt.Version()
		if version == 0 {
			// If the version is not provided we will choose the latest
			version = -1
		}

		lp, err := cat.GetPlugin(mt.Namespace(), version)
		if err != nil {
			return nil, serror.New(err)
		}
		// if loaded plugin is nil, we have failed.  return error
		if lp == nil {
			return nil, serror.New(errorMetricNotFound(mt.Namespace()))
		}

		key := lp.Key()
		pmt, _ := pmts[key]
		pmt.plugin = lp
		pmt.metricTypes = append(pmt.metricTypes, mt)
		pmts[key] = pmt
	}
	return pmts, nil
}

func (p *pluginControl) validatePluginSubscription(pl core.SubscribedPlugin) []serror.SnapError {
	var serrs = []serror.SnapError{}
	controlLogger.WithFields(log.Fields{
		"_block": "validate-plugin-subscription",
		"plugin": fmt.Sprintf("%s:%d", pl.Name(), pl.Version()),
	}).Info(fmt.Sprintf("validating dependencies for plugin %s:%d", pl.Name(), pl.Version()))
	lp, err := p.pluginManager.get(fmt.Sprintf("%s:%s:%d", pl.TypeName(), pl.Name(), pl.Version()))
	if err != nil {
		se := serror.New(fmt.Errorf("Plugin not found: type(%s) name(%s) version(%d)", pl.TypeName(), pl.Name(), pl.Version()))
		se.SetFields(map[string]interface{}{
			"name":    pl.Name(),
			"version": pl.Version(),
			"type":    pl.TypeName(),
		})
		serrs = append(serrs, se)
		return serrs
	}

	if lp.ConfigPolicy != nil {
		ncd := lp.ConfigPolicy.Get([]string{""})
		_, errs := ncd.Process(pl.Config().Table())
		if errs != nil && errs.HasErrors() {
			for _, e := range errs.Errors() {
				se := serror.New(e)
				se.SetFields(map[string]interface{}{"name": pl.Name(), "version": pl.Version()})
				serrs = append(serrs, se)
			}
		}
	}
	return serrs
}

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
		} else {
			err := p.signingManager.ValidateSignature(p.keyringFiles, rp.Path(), rp.Signature())
			if err != nil {
				return false, serror.New(err)
			}
		}
	}
	return true, nil

}

// 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 (ap *availablePlugins) getPool(key string) (strategy.Pool, serror.SnapError) {
	ap.RLock()
	defer ap.RUnlock()
	pool, ok := ap.table[key]
	if !ok {
		tnv := strings.Split(key, ":")
		if len(tnv) != 3 {
			return nil, serror.New(ErrBadKey, map[string]interface{}{
				"key": key,
			})
		}

		v, err := strconv.Atoi(tnv[2])
		if err != nil {
			return nil, serror.New(ErrBadKey, map[string]interface{}{
				"key": key,
			})
		}

		if v < 1 {
			return ap.findLatestPool(tnv[0], tnv[1])
		}
		// No key found
		return nil, serror.New(ErrBadKey, map[string]interface{}{"key": key})
	}

	return pool, nil
}

// UnloadPlugin unloads a plugin from the LoadedPlugins table
func (p *pluginManager) UnloadPlugin(pl core.Plugin) (*loadedPlugin, serror.SnapError) {

	plugin, err := p.loadedPlugins.get(fmt.Sprintf("%s:%s:%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 plugin.State != LoadedState {
		se := serror.New(ErrPluginNotInLoadedState, map[string]interface{}{
			"plugin-name":    plugin.Name(),
			"plugin-version": plugin.Version(),
			"plugin-type":    pl.TypeName(),
		})
		return nil, se
	}

	// If the plugin has been uploaded via REST API
	// aka, was not auto loaded from auto_discover_path
	// nor loaded from tests
	// then do clean up
	if !plugin.Details.IsAutoLoaded {
		pmLogger.WithFields(log.Fields{
			"plugin-type":    plugin.TypeName(),
			"plugin-name":    plugin.Name(),
			"plugin-version": plugin.Version(),
			"plugin-path":    plugin.Details.Path,
		}).Debugf("Removing plugin")
		if err := os.RemoveAll(filepath.Dir(plugin.Details.Path)); err != nil {
			pmLogger.WithFields(log.Fields{
				"plugin-type":    plugin.TypeName(),
				"plugin-name":    plugin.Name(),
				"plugin-version": plugin.Version(),
				"plugin-path":    plugin.Details.Path,
			}).Error(err)
			se := serror.New(err)
			se.SetFields(map[string]interface{}{
				"plugin-type":    plugin.TypeName(),
				"plugin-name":    plugin.Name(),
				"plugin-version": plugin.Version(),
				"plugin-path":    plugin.Details.Path,
			})
			return nil, se
		}
	}

	p.loadedPlugins.remove(plugin.Key())

	// Remove any metrics from the catalog if this was a collector
	if plugin.TypeName() == "collector" {
		p.metricCatalog.RmUnloadedPluginMetrics(plugin)
	}

	return plugin, nil
}

func (s *scheduler) stopTask(id, source string) []serror.SnapError {
	logger := schedulerLogger.WithFields(log.Fields{
		"_block": "stop-task",
		"source": source,
	})
	t, err := s.getTask(id)
	if err != nil {
		logger.WithFields(log.Fields{
			"_error":  err.Error(),
			"task-id": id,
		}).Error("error stopping task")
		return []serror.SnapError{
			serror.New(err),
		}
	}

	if t.state == core.TaskStopped {
		logger.WithFields(log.Fields{
			"task-id":    t.ID(),
			"task-state": t.State(),
		}).Info("task is already stopped")
		return []serror.SnapError{
			serror.New(ErrTaskAlreadyStopped),
		}
	}

	// Group depndencies by the host they live on and
	// unsubscirbe them since task is stopping.
	depGroupMap := s.gatherMetricsAndPlugins(t.workflow)

	var errs []serror.SnapError
	for k := range depGroupMap {
		mgr, err := t.RemoteManagers.Get(k)
		if err != nil {
			errs = append(errs, serror.New(err))
		} else {
			uerrs := mgr.UnsubscribeDeps(t.ID(), depGroupMap[k].Metrics, returnCorePlugin(depGroupMap[k].Plugins))
			if len(uerrs) > 0 {
				errs = append(errs, uerrs...)
			}
		}
	}
	if len(errs) > 0 {
		return errs
	}

	event := &scheduler_event.TaskStoppedEvent{
		TaskID: t.ID(),
		Source: source,
	}
	defer s.eventManager.Emit(event)
	t.Stop()
	logger.WithFields(log.Fields{
		"task-id":    t.ID(),
		"task-state": t.State(),
	}).Info("task stopped")
	return nil
}

func (p *pluginControl) validateMetricTypeSubscription(mt core.RequestedMetric, cd *cdata.ConfigDataNode) []serror.SnapError {
	var serrs []serror.SnapError
	controlLogger.WithFields(log.Fields{
		"_block":    "validate-metric-subscription",
		"namespace": mt.Namespace(),
		"version":   mt.Version(),
	}).Info("subscription called on metric")

	m, err := p.metricCatalog.Get(mt.Namespace(), mt.Version())

	if err != nil {
		serrs = append(serrs, serror.New(err, map[string]interface{}{
			"name":    mt.Namespace().String(),
			"version": mt.Version(),
		}))
		return serrs
	}

	// No metric found return error.
	if m == nil {
		serrs = append(serrs, serror.New(fmt.Errorf("no metric found cannot subscribe: (%s) version(%d)", mt.Namespace(), mt.Version())))
		return serrs
	}

	m.config = cd

	typ, serr := core.ToPluginType(m.Plugin.TypeName())
	if serr != nil {
		return []serror.SnapError{serror.New(err)}
	}

	// merge global plugin config
	if m.config != nil {
		m.config.ReverseMerge(p.Config.Plugins.getPluginConfigDataNode(typ, m.Plugin.Name(), m.Plugin.Version()))
	} else {
		m.config = p.Config.Plugins.getPluginConfigDataNode(typ, m.Plugin.Name(), m.Plugin.Version())
	}

	// When a metric is added to the MetricCatalog, the policy of rules defined by the plugin is added to the metric's policy.
	// If no rules are defined for a metric, we set the metric's policy to an empty ConfigPolicyNode.
	// Checking m.policy for nil will not work, we need to check if rules are nil.
	if m.policy.HasRules() {
		if m.Config() == nil {
			serrs = append(serrs, serror.New(fmt.Errorf("Policy defined for metric, (%s) version (%d), but no config defined in manifest", mt.Namespace(), mt.Version())))
			return serrs
		}
		ncdTable, errs := m.policy.Process(m.Config().Table())
		if errs != nil && errs.HasErrors() {
			for _, e := range errs.Errors() {
				serrs = append(serrs, serror.New(e))
			}
			return serrs
		}
		m.config = cdata.FromTable(*ncdTable)
	}

	return serrs
}

func (ap *availablePlugins) collectMetrics(pluginKey string, metricTypes []core.Metric) ([]core.Metric, error) {
	var results []core.Metric
	pool, serr := ap.getPool(pluginKey)
	if serr != nil {
		return nil, serr
	}
	if pool == nil {
		return nil, serror.New(ErrPoolNotFound, map[string]interface{}{"pool-key": pluginKey})
	}

	metricsToCollect, metricsFromCache := pool.CheckCache(metricTypes)

	if len(metricsToCollect) == 0 {
		return metricsFromCache, nil
	}

	pool.RLock()
	defer pool.RUnlock()
	p, serr := pool.selectAP()
	if serr != nil {
		return nil, serr
	}

	// cast client to PluginCollectorClient
	cli, ok := p.client.(client.PluginCollectorClient)
	if !ok {
		return nil, serror.New(errors.New("unable to cast client to PluginCollectorClient"))
	}

	// collect metrics
	metrics, err := cli.CollectMetrics(metricsToCollect)
	if err != nil {
		return nil, serror.New(err)
	}

	pool.UpdateCache(metrics)

	results = make([]core.Metric, len(metricsFromCache)+len(metrics))
	idx := 0
	for _, m := range metrics {
		results[idx] = m
		idx++
	}
	for _, m := range metricsFromCache {
		results[idx] = m
		idx++
	}

	// update plugin stats
	p.hitCount++
	p.lastHitTime = time.Now()

	return metrics, nil
}

func (s *subscriptionGroup) subscribePlugins(id string,
	plugins []core.SubscribedPlugin) (serrs []serror.SnapError) {
	plgs := make([]*loadedPlugin, len(plugins))
	// First range through plugins to verify if all required plugins
	// are available
	for i, sub := range plugins {
		plg, err := s.pluginManager.get(key(sub))
		if err != nil {
			serrs = append(serrs, serror.New(err))
			return serrs
		}
		plgs[i] = plg
	}

	// If all plugins are available, subscribe to pools and start
	// plugins as needed
	for _, plg := range plgs {
		controlLogger.WithFields(log.Fields{
			"name":    plg.Name(),
			"type":    plg.TypeName(),
			"version": plg.Version(),
			"_block":  "subscriptionGroup.subscribePlugins",
		}).Debug("plugin subscription")
		pool, err := s.pluginRunner.AvailablePlugins().getOrCreatePool(plg.Key())
		if err != nil {
			serrs = append(serrs, serror.New(err))
			return serrs
		}
		pool.Subscribe(id)
		if pool.Eligible() {
			err = s.verifyPlugin(plg)
			if err != nil {
				serrs = append(serrs, serror.New(err))
				return serrs
			}
			err = s.pluginRunner.runPlugin(plg.Details)
			if err != nil {
				serrs = append(serrs, serror.New(err))
				return serrs
			}
		}

		serr := s.sendPluginSubscriptionEvent(id, plg)
		if serr != nil {
			serrs = append(serrs, serr)
			return serrs
		}
	}
	return serrs
}

func (m *subscriptionManager) SubscribeDeps(taskID string, reqs []core.RequestedMetric, cps []core.SubscribedPlugin, cdt *cdata.ConfigDataTree) []serror.SnapError {
	if m.Fail {
		return []serror.SnapError{serror.New(errors.New("error"))}
	}
	m.SubscribeCallCount += 1
	return nil
}

func (s *Server) getMember(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
	tribeLogger = tribeLogger.WithField("_block", "getMember")
	name := p.ByName("name")
	member := s.tr.GetMember(name)
	if member == nil {
		fields := map[string]interface{}{
			"name": name,
		}
		tribeLogger.WithFields(fields).Error(ErrMemberNotFound)
		respond(404, rbody.FromSnapError(serror.New(ErrMemberNotFound, fields)), w)
		return
	}
	resp := &rbody.TribeMemberShow{
		Name: member.Name,
		Tags: member.Tags,
	}
	if member.PluginAgreement != nil {
		resp.PluginAgreement = member.PluginAgreement.Name
	}
	for k, t := range member.TaskAgreements {
		if len(t.Tasks) > 0 {
			resp.TaskAgreements = append(resp.TaskAgreements, k)
		}
	}
	respond(200, resp, w)
}

func loadPlg(c *pluginControl, paths ...string) (core.CatalogedPlugin, serror.SnapError) {
	// This is a Travis optimized loading of plugins. From time to time, tests will error in Travis
	// due to a timeout when waiting for a response from a plugin. We are going to attempt loading a plugin
	// 3 times before letting the error through. Hopefully this cuts down on the number of Travis failures
	var e serror.SnapError
	var p core.CatalogedPlugin
	rp, err := core.NewRequestedPlugin(paths[0])
	if err != nil {
		return nil, serror.New(err)
	}
	if len(paths) > 1 {
		rp.SetSignature([]byte{00, 00, 00})
	}
	for i := 0; i < 3; i++ {
		p, e = c.Load(rp)
		if e == nil {
			break
		}
		if e != nil && i == 2 {
			return nil, e

		}
	}
	return p, nil
}

func (p *subscriptionGroups) validatePluginSubscription(pl core.SubscribedPlugin, mergedConfig *cdata.ConfigDataNode) []serror.SnapError {
	var serrs = []serror.SnapError{}
	controlLogger.WithFields(log.Fields{
		"_block": "validate-plugin-subscription",
		"plugin": fmt.Sprintf("%s:%d", pl.Name(), pl.Version()),
	}).Info(fmt.Sprintf("validating dependencies for plugin %s:%d", pl.Name(), pl.Version()))
	lp, err := p.pluginManager.get(key(pl))
	if err != nil {
		serrs = append(serrs, pluginNotFoundError(pl))
		return serrs
	}

	if lp.ConfigPolicy != nil {
		ncd := lp.ConfigPolicy.Get([]string{""})
		_, errs := ncd.Process(mergedConfig.Table())
		if errs != nil && errs.HasErrors() {
			for _, e := range errs.Errors() {
				se := serror.New(e)
				se.SetFields(map[string]interface{}{"name": pl.Name(), "version": pl.Version()})
				serrs = append(serrs, se)
			}
		}
	}
	return serrs
}

// Converts SnapError protobuf messages to serror.Snaperrors
func ConvertSnapErrors(s []*SnapError) []serror.SnapError {
	rerrs := make([]serror.SnapError, len(s))
	for i, err := range s {
		rerrs[i] = serror.New(errors.New(err.ErrorString), GetFields(err))
	}
	return rerrs
}

func (ap *availablePlugins) processMetrics(contentType string, content []byte, pluginName string, pluginVersion int, config map[string]ctypes.ConfigValue, taskID string) (string, []byte, []error) {
	var errs []error
	key := strings.Join([]string{plugin.ProcessorPluginType.String(), pluginName, strconv.Itoa(pluginVersion)}, ":")
	pool, serr := ap.getPool(key)
	if serr != nil {
		errs = append(errs, serr)
		return "", nil, errs
	}
	if pool == nil {
		return "", nil, []error{serror.New(ErrPoolNotFound, map[string]interface{}{"pool-key": key})}
	}

	pool.RLock()
	defer pool.RUnlock()
	p, err := pool.SelectAP(taskID, config)
	if err != nil {
		errs = append(errs, err)
		return "", nil, errs
	}

	cli, ok := p.(*availablePlugin).client.(client.PluginProcessorClient)
	if !ok {
		return "", nil, []error{errors.New("unable to cast client to PluginProcessorClient")}
	}

	ct, c, errp := cli.Process(contentType, content, config)
	if errp != nil {
		return "", nil, []error{errp}
	}
	p.(*availablePlugin).hitCount++
	p.(*availablePlugin).lastHitTime = time.Now()
	return ct, c, nil
}

func (ap *availablePlugins) publishMetrics(metrics []core.Metric, pluginName string, pluginVersion int, config map[string]ctypes.ConfigValue, taskID string) []error {
	var errs []error
	key := strings.Join([]string{plugin.PublisherPluginType.String(), pluginName, strconv.Itoa(pluginVersion)}, core.Separator)
	pool, serr := ap.getPool(key)
	if serr != nil {
		errs = append(errs, serr)
		return errs
	}
	if pool == nil {
		return []error{serror.New(ErrPoolNotFound, map[string]interface{}{"pool-key": key})}
	}

	pool.RLock()
	defer pool.RUnlock()

	p, err := pool.SelectAP(taskID, config)
	if err != nil {
		errs = append(errs, err)
		return errs
	}

	cli, ok := p.(*availablePlugin).client.(client.PluginPublisherClient)
	if !ok {
		return []error{errors.New("unable to cast client to PluginPublisherClient")}
	}

	errp := cli.Publish(metrics, config)
	if errp != nil {
		return []error{errp}
	}
	p.(*availablePlugin).hitCount++
	p.(*availablePlugin).lastHitTime = time.Now()
	return nil
}