func (s *SignalFx) convertToProto(incomingMetric *metric.Metric) *DataPoint {
	outname := s.Prefix() + incomingMetric.Name

	datapoint := new(DataPoint)
	datapoint.Metric = &outname
	datapoint.Value = &Datum{
		DoubleValue: &incomingMetric.Value,
	}
	datapoint.Source = new(string)
	*datapoint.Source = "fullerite"

	switch incomingMetric.MetricType {
	case metric.Gauge:
		datapoint.MetricType = MetricType_GAUGE.Enum()
	case metric.Counter:
		datapoint.MetricType = MetricType_COUNTER.Enum()
	case metric.CumulativeCounter:
		datapoint.MetricType = MetricType_CUMULATIVE_COUNTER.Enum()
	}

	dimensions := incomingMetric.GetDimensions(s.DefaultDimensions())
	for key, value := range dimensions {
		dim := Dimension{
			Key:   &key,
			Value: &value,
		}
		datapoint.Dimensions = append(datapoint.Dimensions, &dim)
	}

	return datapoint
}

func addDimensionsFromName(m *metric.Metric, dimensions []string) {
	var dimension []string
	for i := 1; i < len(dimensions); i++ {
		dimension = strings.Split(dimensions[i], "=")
		m.AddDimension(dimension[0], dimension[1])
	}

}

func (g Graphite) getSanitizedDimensions(incomingMetric metric.Metric) map[string]string {
	dimSanitized := make(map[string]string)
	dimensions := incomingMetric.GetDimensions(g.DefaultDimensions())
	for key, value := range dimensions {
		dimSanitized[graphiteSanitize(key)] = graphiteSanitize(value)
	}
	return dimSanitized
}

func (s SignalFx) getSanitizedDimensions(incomingMetric metric.Metric) map[string]string {
	dimSanitized := make(map[string]string)
	dimensions := incomingMetric.GetDimensions(s.DefaultDimensions())
	for key, value := range dimensions {
		dimSanitized[signalFxKeySanitize(key)] = signalFxValueSanitize(value)
	}
	return dimSanitized
}

func (k Kairos) convertToKairos(incomingMetric metric.Metric) (datapoint KairosMetric) {
	km := new(KairosMetric)
	km.Name = k.Prefix() + incomingMetric.Name
	km.Value = incomingMetric.Value
	km.MetricType = "double"
	km.Timestamp = time.Now().Unix() * 1000 // Kairos require timestamps to be milliseconds
	km.Tags = incomingMetric.GetDimensions(k.DefaultDimensions())
	return *km
}

func (s *Scribe) createScribeMetric(m metric.Metric) scribeMetric {
	return scribeMetric{
		Name:       m.Name,
		Value:      m.Value,
		MetricType: m.MetricType,
		Timestamp:  time.Now().Unix(),
		Dimensions: m.GetDimensions(s.DefaultDimensions()),
	}

}

// Collect reads metrics collected from Diamond collectors, converts
// them to fullerite's Metric type and publishes them to handlers.
func (d Diamond) Collect() {
	for line := range d.incoming {
		var metric metric.Metric
		if err := json.Unmarshal(line, &metric); err != nil {
			log.Error("Cannot unmarshal metric line from diamond:", line)
			continue
		}
		metric.AddDimension("diamond", "yes")
		d.Channel() <- metric
	}
}

func (d *Diamond) parseMetric(line []byte) (metric.Metric, bool) {
	var metric metric.Metric
	if err := json.Unmarshal(line, &metric); err != nil {
		d.log.Error("Cannot unmarshal metric line from diamond:", line)
		return metric, false
	}
	// All diamond metric_types are reported in uppercase, lets make them
	// fullerite compatible
	metric.MetricType = strings.ToLower(metric.MetricType)
	metric.AddDimension("diamond", "yes")
	return metric, true
}

func (k Kairos) convertToKairos(incomingMetric metric.Metric) (datapoint KairosMetric) {
	km := new(KairosMetric)
	km.Name = k.Prefix() + kairosSanitize(incomingMetric.Name)
	km.Value = incomingMetric.Value
	km.MetricType = "double"
	km.Timestamp = time.Now().Unix() * 1000 // Kairos require timestamps to be milliseconds
	km.Tags = make(map[string]string)
	for key, value := range incomingMetric.GetDimensions(k.DefaultDimensions()) {
		km.Tags[kairosSanitize(key)] = kairosSanitize(value)
	}
	return *km
}

func (d *Datadog) convertToDatadog(incomingMetric metric.Metric) (datapoint datadogMetric) {
	dog := new(datadogMetric)
	dog.Metric = incomingMetric.Name
	dog.Points = makeDatadogPoints(incomingMetric)
	dog.MetricType = incomingMetric.MetricType
	if host, ok := incomingMetric.GetDimensionValue("host", d.DefaultDimensions()); ok {
		dog.Host = host
	} else {
		dog.Host = "unknown"
	}
	dog.Tags = d.serializedDimensions(incomingMetric)
	return *dog
}

// Collect reads metrics collected from Diamond collectors, converts
// them to fullerite's Metric type and publishes them to handlers.
func (d *Diamond) Collect() {
	if !d.serverStarted {
		d.serverStarted = true
		go d.collectDiamond()
	}

	for line := range d.incoming {
		var metric metric.Metric
		if err := json.Unmarshal(line, &metric); err != nil {
			d.log.Error("Cannot unmarshal metric line from diamond:", line)
			continue
		}
		metric.AddDimension("diamond", "yes")
		d.Channel() <- metric
	}
}

func (g *Graphite) convertToGraphite(incomingMetric metric.Metric) (datapoint string) {
	//orders dimensions so datapoint keeps consistent name
	var keys []string
	dimensions := incomingMetric.GetDimensions(g.DefaultDimensions())
	for k := range dimensions {
		keys = append(keys, k)
	}
	sort.Strings(keys)

	datapoint = g.Prefix() + incomingMetric.Name
	for _, key := range keys {
		datapoint = fmt.Sprintf("%s.%s.%s", datapoint, key, dimensions[key])
	}
	datapoint = fmt.Sprintf("%s %f %d\n", datapoint, incomingMetric.Value, time.Now().Unix())
	return datapoint
}

func (s SignalFx) convertToProto(incomingMetric metric.Metric) *DataPoint {
	// Create a new values for the Datapoint that requires pointers.
	outname := s.Prefix() + incomingMetric.Name
	value := incomingMetric.Value

	now := time.Now().UnixNano() / int64(time.Millisecond)
	datapoint := new(DataPoint)
	datapoint.Timestamp = &now
	datapoint.Metric = &outname
	datapoint.Value = &Datum{
		DoubleValue: &value,
	}
	datapoint.Source = new(string)
	*datapoint.Source = "fullerite"

	switch incomingMetric.MetricType {
	case metric.Gauge:
		datapoint.MetricType = MetricType_GAUGE.Enum()
	case metric.Counter:
		datapoint.MetricType = MetricType_COUNTER.Enum()
	case metric.CumulativeCounter:
		datapoint.MetricType = MetricType_CUMULATIVE_COUNTER.Enum()
	}

	dimensions := incomingMetric.GetDimensions(s.DefaultDimensions())
	for key, value := range dimensions {
		// Dimension (protobuf) require a pointer to string
		// values. We need to create new string objects in the
		// scope of this for loop not to repeatedly add the
		// same key:value pairs to the the datapoint.
		dimensionKey := key
		dimensionValue := value
		dim := Dimension{
			Key:   &dimensionKey,
			Value: &dimensionValue,
		}
		datapoint.Dimensions = append(datapoint.Dimensions, &dim)
	}

	return datapoint
}

func writeToHandlers(handlers []handler.Handler, metric metric.Metric) {
	for _, handler := range handlers {
		value, ok := metric.GetDimensionValue("collector")
		isBlackListed, _ := handler.IsCollectorBlackListed(value)
		if ok && isBlackListed {
			// This collector is black listed by
			// this handler. Therefore we are dropping this
			log.Debug("Not forwarding metrics from", value, "collector to", handler.Name(), "handler, since it has blacklisted this collector")
		} else {
			handler.Channel() <- metric
		}
	}
}

// Collect calls smem periodically
func (s *SmemStats) Collect() {
	if s.whitelistedProcs == "" || s.user == "" || s.smemPath == "" {
		return
	}

	for _, stat := range getSmemStats(s) {
		dims := s.getCustomDimensions(stat.pid)
		for _, element := range s.whitelistedMetrics {
			var m metric.Metric
			switch element {
			case "pss":
				m = metric.WithValue(stat.proc+".smem.pss", stat.pss)
			case "uss":
				m = metric.WithValue(stat.proc+".smem.uss", stat.uss)
			case "vss":
				m = metric.WithValue(stat.proc+".smem.vss", stat.vss)
			case "rss":
				m = metric.WithValue(stat.proc+".smem.rss", stat.rss)
			}
			m.AddDimensions(dims)
			s.Channel() <- m
		}
	}
}

func (g *Graphite) convertToGraphite(metric *metric.Metric) string {
	outname := g.Prefix() + (*metric).Name
	dimensions := metric.GetDimensions(g.DefaultDimensions())
	// for key in dimensions, generate a new metric data point, add to a list, return
	//what timestamp to use?

	//orders keys so datapoint keeps consistent name
	var keys []string
	for k := range dimensions {
		keys = append(keys, k)
	}
	sort.Strings(keys)

	for _, key := range keys {
		//create a list of datapoints for this metric, then append that list the a global list
		outname = fmt.Sprintf("%s.%s.%s", outname, key, dimensions[key])
	}

	outname = fmt.Sprintf("%s %f %d", outname, metric.Value, time.Now().Unix())

	return outname
}

func canSendMetric(handler handler.Handler, metric metric.Metric) bool {
	// If the handler's whitelist is set, then only metrics from collectors in it will be emitted. If the same
	// collector is also in the blacklist, it will be skipped.
	// If the handler's whitelist is not set and its blacklist is not empty, only metrics from collectors not in
	// the blacklist will be emitted.
	value, _ := metric.GetDimensionValue("collector")
	isWhiteListed, _ := handler.IsCollectorWhiteListed(value)
	isBlackListed, _ := handler.IsCollectorBlackListed(value)

	// If the handler's whitelist is not nil and not empty, only the whitelisted collectors should be considered
	if handler.CollectorWhiteList() != nil && len(handler.CollectorWhiteList()) > 0 {
		if isWhiteListed && !isBlackListed {
			return true
		}
		return false
	}

	// If the handler's whitelist is nil, all collector except the ones in the blacklist are enabled
	if !isBlackListed {
		return true
	}
	return false
}

func (d *Datadog) serializedDimensions(m metric.Metric) (dimensions []string) {
	for name, value := range m.GetDimensions(d.DefaultDimensions()) {
		dimensions = append(dimensions, name+":"+value)
	}
	return dimensions
}