func addStandardAndWorkflowTags(m core.Metric, allTags map[string]map[string]string) core.Metric {
	hostname := hostnameReader.Hostname()

	tags := m.Tags()
	if tags == nil {
		tags = map[string]string{}
	}
	// apply tags from workflow
	for ns, nsTags := range allTags {
		if strings.HasPrefix(m.Namespace().String(), ns) {
			for k, v := range nsTags {
				tags[k] = v
			}
		}
	}
	// apply standard tag
	tags[core.STD_TAG_PLUGIN_RUNNING_ON] = hostname

	metric := plugin.MetricType{
		Namespace_:          m.Namespace(),
		Version_:            m.Version(),
		LastAdvertisedTime_: m.LastAdvertisedTime(),
		Config_:             m.Config(),
		Data_:               m.Data(),
		Tags_:               tags,
		Description_:        m.Description(),
		Unit_:               m.Unit(),
		Timestamp_:          m.Timestamp(),
	}
	return metric
}

func (mc *metricCatalog) AddLoadedMetricType(lp *loadedPlugin, mt core.Metric) error {
	if err := validateMetricNamespace(mt.Namespace()); err != nil {
		log.WithFields(log.Fields{
			"_module": "control",
			"_file":   "metrics.go,",
			"_block":  "add-loaded-metric-type",
			"error":   fmt.Errorf("Metric namespace %s is invalid", mt.Namespace()),
		}).Error("error adding loaded metric type")
		return err
	}
	if lp.ConfigPolicy == nil {
		err := errors.New("Config policy is nil")
		log.WithFields(log.Fields{
			"_module": "control",
			"_file":   "metrics.go,",
			"_block":  "add-loaded-metric-type",
			"error":   err,
		}).Error("error adding loaded metric type")
		return err
	}
	newMt := metricType{
		Plugin:             lp,
		namespace:          mt.Namespace(),
		version:            mt.Version(),
		lastAdvertisedTime: mt.LastAdvertisedTime(),
		tags:               mt.Tags(),
		policy:             lp.ConfigPolicy.Get(mt.Namespace().Strings()),
		description:        mt.Description(),
		unit:               mt.Unit(),
	}
	mc.Add(&newMt)
	return nil
}

func addStandardAndWorkflowTags(m core.Metric, allTags map[string]map[string]string) core.Metric {
	hostname, err := hostnameReader.Hostname()
	if err != nil {
		log.WithFields(log.Fields{
			"_module": "control",
			"_file":   "metrics.go,",
			"_block":  "addStandardAndWorkflowTags",
			"error":   err.Error(),
		}).Error("Unable to determine hostname")
	}
	tags := m.Tags()
	if tags == nil {
		tags = map[string]string{}
	}
	// apply tags from workflow
	for ns, nsTags := range allTags {
		if strings.HasPrefix(m.Namespace().String(), ns) {
			for k, v := range nsTags {
				tags[k] = v
			}
		}
	}
	// apply standard tag
	tags[core.STD_TAG_PLUGIN_RUNNING_ON] = hostname

	metric := plugin.MetricType{
		Namespace_:          m.Namespace(),
		Version_:            m.Version(),
		LastAdvertisedTime_: m.LastAdvertisedTime(),
		Config_:             m.Config(),
		Data_:               m.Data(),
		Tags_:               tags,
		Description_:        m.Description(),
		Unit_:               m.Unit(),
		Timestamp_:          m.Timestamp(),
	}
	return metric
}

func (mc *metricCatalog) AddLoadedMetricType(lp *loadedPlugin, mt core.Metric) {
	if lp.ConfigPolicy == nil {
		panic("NO")
	}

	newMt := metricType{
		Plugin:             lp,
		namespace:          mt.Namespace(),
		version:            mt.Version(),
		lastAdvertisedTime: mt.LastAdvertisedTime(),
		tags:               mt.Tags(),
		labels:             mt.Labels(),
		policy:             lp.ConfigPolicy.Get(mt.Namespace()),
	}
	mc.Add(&newMt)
}

func (s *subscriptionGroups) validateMetric(
	metric core.Metric) (serrs []serror.SnapError) {
	m, err := s.metricCatalog.GetMetric(metric.Namespace(), metric.Version())
	if err != nil {
		serrs = append(serrs, serror.New(err, map[string]interface{}{
			"name":    metric.Namespace().String(),
			"version": metric.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)",
					metric.Namespace(), metric.Version())))
		return serrs
	}

	m.config = metric.Config()

	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(
			s.Config.Plugins.getPluginConfigDataNode(typ,
				m.Plugin.Name(), m.Plugin.Version()))
	} else {
		m.config = s.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 {
			fields := log.Fields{
				"metric":  m.Namespace(),
				"version": m.Version(),
				"plugin":  m.Plugin.Name(),
			}
			serrs = append(serrs, serror.New(ErrConfigRequiredForMetric, fields))
			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
}

// Convert a core.Metric to common.Metric protobuf message
func ToMetric(co core.Metric) *Metric {
	cm := &Metric{
		Namespace: ToNamespace(co.Namespace()),
		Version:   int64(co.Version()),
		Tags:      co.Tags(),
		Timestamp: &Time{
			Sec:  co.Timestamp().Unix(),
			Nsec: int64(co.Timestamp().Nanosecond()),
		},
		LastAdvertisedTime: &Time{
			Sec:  time.Now().Unix(),
			Nsec: int64(time.Now().Nanosecond()),
		},
	}
	if co.Config() != nil {
		cm.Config = ConfigToConfigMap(co.Config())
	}
	switch t := co.Data().(type) {
	case string:
		cm.Data = &Metric_StringData{t}
	case float64:
		cm.Data = &Metric_Float64Data{t}
	case float32:
		cm.Data = &Metric_Float32Data{t}
	case int32:
		cm.Data = &Metric_Int32Data{t}
	case int:
		cm.Data = &Metric_Int64Data{int64(t)}
	case int64:
		cm.Data = &Metric_Int64Data{t}
	case []byte:
		cm.Data = &Metric_BytesData{t}
	case bool:
		cm.Data = &Metric_BoolData{t}
	case nil:
		cm.Data = nil
	default:
		panic(fmt.Sprintf("unsupported type: %s", t))
	}
	return cm
}