[receiver/collectd] Migrate from OpenCensus data model to pdata (#24770)

**Description:**
Migrate from OpenCensus data model to pdata on collectdreceiver. Using
golden to compare metrics for testing.

Since wavefrontreceiver depends on collectdreceiver, adjustments are
necessary in the go.mod and go.sum files.

**Link to tracking Issue:** [20760](#20760)

---------

Co-authored-by: Antoine Toulme <[email protected]>

Author: ZenoCC-Peng

.chloggen/collectdreceiver-migrate-from-opensus-to-pdata.yaml

@@ -0,0 +1,15 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: receiver/collectdreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Migrate from opencensus to pdata, change collectd, test to match pdata format.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [20760]

receiver/collectdreceiver/collectd.go

@@ -9,19 +9,12 @@ import (
 	"strings"
 	"time"
 
-	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
-	"google.golang.org/protobuf/types/known/timestamppb"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/sanitize"
 )
 
-const (
-	collectDMetricDerive   = "derive"
-	collectDMetricGauge    = "gauge"
-	collectDMetricCounter  = "counter"
-	collectDMetricAbsolute = "absolute"
-)
-
 type collectDRecord struct {
 	Dsnames        []*string              `json:"dsnames"`
 	Dstypes        []*string              `json:"dstypes"`
@@ -38,31 +31,31 @@ type collectDRecord struct {
 	Severity       *string                `json:"severity"`
 }
 
-func (r *collectDRecord) isEvent() bool {
-	return r.Time != nil && r.Severity != nil && r.Message != nil
+func (cdr *collectDRecord) isEvent() bool {
+	return cdr.Time != nil && cdr.Severity != nil && cdr.Message != nil
 }
 
-func (r *collectDRecord) protoTime() *timestamppb.Timestamp {
-	if r.Time == nil {
-		return nil
+func (cdr *collectDRecord) protoTime() pcommon.Timestamp {
+	// Return 1970-01-01 00:00:00 +0000 UTC.
+	if cdr.Time == nil {
+		return pcommon.NewTimestampFromTime(time.Unix(0, 0))
 	}
-	ts := time.Unix(0, int64(float64(time.Second)**r.Time))
-	return timestamppb.New(ts)
+	ts := time.Unix(0, int64(float64(time.Second)**cdr.Time))
+	return pcommon.NewTimestampFromTime(ts)
 }
 
-func (r *collectDRecord) startTimestamp(mdType metricspb.MetricDescriptor_Type) *timestamppb.Timestamp {
-	if mdType == metricspb.MetricDescriptor_CUMULATIVE_DISTRIBUTION || mdType == metricspb.MetricDescriptor_CUMULATIVE_DOUBLE || mdType == metricspb.MetricDescriptor_CUMULATIVE_INT64 {
-		return timestamppb.New(time.Unix(0, int64((*r.Time-*r.Interval)*float64(time.Second))))
+func (cdr *collectDRecord) startTimestamp(metricType string) pcommon.Timestamp {
+	if metricType == "cumulative" {
+		return pcommon.NewTimestampFromTime(time.Unix(0, int64((*cdr.Time-*cdr.Interval)*float64(time.Second))))
 	}
-	return nil
+	return pcommon.NewTimestampFromTime(time.Unix(0, 0))
 }
 
-func (r *collectDRecord) appendToMetrics(metrics []*metricspb.Metric, defaultLabels map[string]string) ([]*metricspb.Metric, error) {
+func (cdr *collectDRecord) appendToMetrics(scopeMetrics pmetric.ScopeMetrics, defaultLabels map[string]string) error {
 	// Ignore if record is an event instead of data point
-	if r.isEvent() {
+	if cdr.isEvent() {
 		recordEventsReceived()
-		return metrics, nil
-
+		return nil
 	}
 
 	recordMetricsReceived()
@@ -71,125 +64,120 @@ func (r *collectDRecord) appendToMetrics(metrics []*metricspb.Metric, defaultLab
 		labels[k] = v
 	}
 
-	for i := range r.Dsnames {
-		if i < len(r.Dstypes) && i < len(r.Values) && r.Values[i] != nil {
-			dsType, dsName, val := r.Dstypes[i], r.Dsnames[i], r.Values[i]
-			metricName, usedDsName := r.getReasonableMetricName(i, labels)
+	for i := range cdr.Dsnames {
+		if i < len(cdr.Dstypes) && i < len(cdr.Values) && cdr.Values[i] != nil {
+			dsType, dsName, val := cdr.Dstypes[i], cdr.Dsnames[i], cdr.Values[i]
+			metricName, usedDsName := cdr.getReasonableMetricName(i, labels)
 
-			addIfNotNullOrEmpty(labels, "plugin", r.Plugin)
-			parseAndAddLabels(labels, r.PluginInstance, r.Host)
+			addIfNotNullOrEmpty(labels, "plugin", cdr.Plugin)
+			parseAndAddLabels(labels, cdr.PluginInstance, cdr.Host)
 			if !usedDsName {
 				addIfNotNullOrEmpty(labels, "dsname", dsName)
 			}
 
-			metric, err := r.newMetric(metricName, dsType, val, labels)
+			metric, err := cdr.newMetric(metricName, dsType, val, labels)
 			if err != nil {
-				return metrics, fmt.Errorf("error processing metric %s: %w", sanitize.String(metricName), err)
+				return fmt.Errorf("error processing metric %s: %w", sanitize.String(metricName), err)
 			}
-			metrics = append(metrics, metric)
-
+			newMetric := scopeMetrics.Metrics().AppendEmpty()
+			metric.MoveTo(newMetric)
 		}
 	}
-	return metrics, nil
+	return nil
 }
 
-func (r *collectDRecord) newMetric(name string, dsType *string, val *json.Number, labels map[string]string) (*metricspb.Metric, error) {
-	metric := &metricspb.Metric{}
-	point, isDouble, err := r.newPoint(val)
+// Create new metric, get labels, then setting attribute and metric info
+func (cdr *collectDRecord) newMetric(name string, dsType *string, val *json.Number, labels map[string]string) (pmetric.Metric, error) {
+	attributes := setAttributes(labels)
+	metric, err := cdr.setMetric(name, dsType, val, attributes)
 	if err != nil {
-		return metric, fmt.Errorf("error processing metric %s: %w", name, err)
+		return pmetric.Metric{}, fmt.Errorf("error processing metric %s: %w", name, err)
 	}
+	return metric, nil
+}
 
-	lKeys, lValues := labelKeysAndValues(labels)
-	metricType := r.metricType(dsType, isDouble)
-	metric.MetricDescriptor = &metricspb.MetricDescriptor{
-		Name:      name,
-		Type:      metricType,
-		LabelKeys: lKeys,
-	}
-	metric.Timeseries = []*metricspb.TimeSeries{
-		{
-			StartTimestamp: r.startTimestamp(metricType),
-			LabelValues:    lValues,
-			Points:         []*metricspb.Point{point},
-		},
+func setAttributes(labels map[string]string) pcommon.Map {
+	attributes := pcommon.NewMap()
+	for k, v := range labels {
+		attributes.PutStr(k, v)
 	}
-
-	return metric, nil
+	return attributes
 }
 
-func (r *collectDRecord) metricType(dsType *string, isDouble bool) metricspb.MetricDescriptor_Type {
-	val := ""
+// Set new metric info with name, datapoint, time, attributes
+func (cdr *collectDRecord) setMetric(name string, dsType *string, val *json.Number, atr pcommon.Map) (pmetric.Metric, error) {
+
+	typ := ""
+	metric := pmetric.NewMetric()
+
 	if dsType != nil {
-		val = *dsType
+		typ = *dsType
 	}
 
-	switch val {
-	case collectDMetricCounter, collectDMetricDerive:
-		return metricCumulative(isDouble)
+	metric.SetName(name)
+	dataPoint := setDataPoint(typ, metric)
+	dataPoint.SetTimestamp(cdr.protoTime())
+	atr.CopyTo(dataPoint.Attributes())
 
-	// Prometheus collectd exporter just ignores it. We use gauge for it as it seems the
-	// closes type. https://github.com/prometheus/collectd_exporter/blob/master/main.go#L109-L129
-	case collectDMetricGauge, collectDMetricAbsolute:
-		return metricGauge(isDouble)
+	if pointVal, err := val.Int64(); err == nil {
+		dataPoint.SetIntValue(pointVal)
+	} else if pointVal, err := val.Float64(); err == nil {
+		dataPoint.SetDoubleValue(pointVal)
+	} else {
+		return pmetric.Metric{}, fmt.Errorf("value could not be decoded: %w", err)
 	}
-	return metricGauge(isDouble)
+	return metric, nil
 }
 
-func (r *collectDRecord) newPoint(val *json.Number) (*metricspb.Point, bool, error) {
-	p := &metricspb.Point{
-		Timestamp: r.protoTime(),
-	}
-
-	isDouble := true
-	if v, err := val.Int64(); err == nil {
-		isDouble = false
-		p.Value = &metricspb.Point_Int64Value{Int64Value: v}
-	} else {
-		v, err := val.Float64()
-		if err != nil {
-			return nil, isDouble, fmt.Errorf("value could not be decoded: %w", err)
-		}
-		p.Value = &metricspb.Point_DoubleValue{DoubleValue: v}
+// check type to decide metric type and return data point
+func setDataPoint(typ string, metric pmetric.Metric) pmetric.NumberDataPoint {
+	var dataPoint pmetric.NumberDataPoint
+	switch typ {
+	case "derive", "counter":
+		sum := metric.SetEmptySum()
+		sum.SetIsMonotonic(true)
+		dataPoint = sum.DataPoints().AppendEmpty()
+	default:
+		dataPoint = metric.SetEmptyGauge().DataPoints().AppendEmpty()
 	}
-	return p, isDouble, nil
+	return dataPoint
 }
 
 // getReasonableMetricName creates metrics names by joining them (if non empty) type.typeinstance
 // if there are more than one dsname append .dsname for the particular uint. if there's only one it
 // becomes a dimension.
-func (r *collectDRecord) getReasonableMetricName(index int, attrs map[string]string) (string, bool) {
+func (cdr *collectDRecord) getReasonableMetricName(index int, attrs map[string]string) (string, bool) {
 	usedDsName := false
 	capacity := 0
-	if r.TypeS != nil {
-		capacity += len(*r.TypeS)
+	if cdr.TypeS != nil {
+		capacity += len(*cdr.TypeS)
 	}
-	if r.TypeInstance != nil {
-		capacity += len(*r.TypeInstance)
+	if cdr.TypeInstance != nil {
+		capacity += len(*cdr.TypeInstance)
 	}
 	parts := make([]byte, 0, capacity)
 
-	if !isNilOrEmpty(r.TypeS) {
-		parts = append(parts, *r.TypeS...)
+	if !isNilOrEmpty(cdr.TypeS) {
+		parts = append(parts, *cdr.TypeS...)
 	}
-	parts = r.pointTypeInstance(attrs, parts)
-	if r.Dsnames != nil && !isNilOrEmpty(r.Dsnames[index]) && len(r.Dsnames) > 1 {
+	parts = cdr.pointTypeInstance(attrs, parts)
+	if cdr.Dsnames != nil && !isNilOrEmpty(cdr.Dsnames[index]) && len(cdr.Dsnames) > 1 {
 		if len(parts) > 0 {
 			parts = append(parts, '.')
 		}
-		parts = append(parts, *r.Dsnames[index]...)
+		parts = append(parts, *cdr.Dsnames[index]...)
 		usedDsName = true
 	}
 	return string(parts), usedDsName
 }
 
 // pointTypeInstance extracts information from the TypeInstance field and appends to the metric name when possible.
-func (r *collectDRecord) pointTypeInstance(attrs map[string]string, parts []byte) []byte {
-	if isNilOrEmpty(r.TypeInstance) {
+func (cdr *collectDRecord) pointTypeInstance(attrs map[string]string, parts []byte) []byte {
+	if isNilOrEmpty(cdr.TypeInstance) {
 		return parts
 	}
 
-	instanceName, extractedAttrs := LabelsFromName(r.TypeInstance)
+	instanceName, extractedAttrs := LabelsFromName(cdr.TypeInstance)
 	if instanceName != "" {
 		if len(parts) > 0 {
 			parts = append(parts, '.')
@@ -275,29 +263,3 @@ func parseNameForLabels(labels map[string]string, key string, val *string) {
 	}
 	addIfNotNullOrEmpty(labels, key, &instanceName)
 }
-
-func labelKeysAndValues(labels map[string]string) ([]*metricspb.LabelKey, []*metricspb.LabelValue) {
-	keys := make([]*metricspb.LabelKey, len(labels))
-	values := make([]*metricspb.LabelValue, len(labels))
-	i := 0
-	for k, v := range labels {
-		keys[i] = &metricspb.LabelKey{Key: k}
-		values[i] = &metricspb.LabelValue{Value: v, HasValue: true}
-		i++
-	}
-	return keys, values
-}
-
-func metricCumulative(isDouble bool) metricspb.MetricDescriptor_Type {
-	if isDouble {
-		return metricspb.MetricDescriptor_CUMULATIVE_DOUBLE
-	}
-	return metricspb.MetricDescriptor_CUMULATIVE_INT64
-}
-
-func metricGauge(isDouble bool) metricspb.MetricDescriptor_Type {
-	if isDouble {
-		return metricspb.MetricDescriptor_GAUGE_DOUBLE
-	}
-	return metricspb.MetricDescriptor_GAUGE_INT64
-}