Merge pull request #208 from coroot/collector_nonstring_attributes

collector: add support for non-string attribute values

collector/logs.go

@@ -128,21 +128,17 @@ func (b *LogsBatch) Add(req *v1.ExportLogsServiceRequest) {
 
 	for _, l := range req.GetResourceLogs() {
 		var serviceName string
-		resourceAttributes := map[string]string{}
-		for _, attr := range l.GetResource().GetAttributes() {
-			if attr.GetKey() == semconv.AttributeServiceName {
-				serviceName = attr.GetValue().GetStringValue()
+		resourceAttributes := attributesToMap(l.GetResource().GetAttributes())
+		for k, v := range resourceAttributes {
+			if k == semconv.AttributeServiceName {
+				serviceName = v
 			}
-			resourceAttributes[attr.GetKey()] = attr.GetValue().GetStringValue()
 		}
 		for _, sl := range l.GetScopeLogs() {
 			scopeName := sl.GetScope().GetName()
 			scopeVersion := sl.GetScope().GetVersion()
 			for _, lr := range sl.GetLogRecords() {
-				logAttributes := map[string]string{}
-				for _, attr := range lr.GetAttributes() {
-					logAttributes[attr.GetKey()] = attr.GetValue().GetStringValue()
-				}
+				logAttributes := attributesToMap(lr.GetAttributes())
 				if scopeName != "" {
 					logAttributes[semconv.AttributeOtelScopeName] = scopeName
 				}

collector/traces.go

@@ -147,21 +147,17 @@ func (b *TracesBatch) Add(req *v1.ExportTraceServiceRequest) {
 
 	for _, rs := range req.GetResourceSpans() {
 		var serviceName string
-		resourceAttributes := map[string]string{}
-		for _, attr := range rs.GetResource().GetAttributes() {
-			if attr.GetKey() == semconv.AttributeServiceName {
-				serviceName = attr.GetValue().GetStringValue()
+		resourceAttributes := attributesToMap(rs.GetResource().GetAttributes())
+		for k, v := range resourceAttributes {
+			if k == semconv.AttributeServiceName {
+				serviceName = v
 			}
-			resourceAttributes[attr.GetKey()] = attr.GetValue().GetStringValue()
 		}
 		for _, ss := range rs.GetScopeSpans() {
 			scopeName := ss.GetScope().GetName()
 			scopeVersion := ss.GetScope().GetVersion()
 			for _, s := range ss.GetSpans() {
-				spanAttributes := map[string]string{}
-				for _, attr := range s.GetAttributes() {
-					spanAttributes[attr.GetKey()] = attr.GetValue().GetStringValue()
-				}
+				spanAttributes := attributesToMap(s.GetAttributes())
 				if scopeName != "" {
 					spanAttributes[semconv.AttributeOtelScopeName] = scopeName
 				}
@@ -174,11 +170,7 @@ func (b *TracesBatch) Add(req *v1.ExportTraceServiceRequest) {
 				for _, e := range s.GetEvents() {
 					eventTimestamps = append(eventTimestamps, time.Unix(0, int64(e.GetTimeUnixNano())))
 					eventNames = append(eventNames, e.GetName())
-					attrs := map[string]string{}
-					for _, a := range e.GetAttributes() {
-						attrs[a.GetKey()] = a.GetValue().GetStringValue()
-					}
-					eventAttributes = append(eventAttributes, attrs)
+					eventAttributes = append(eventAttributes, attributesToMap(e.GetAttributes()))
 				}
 				var linkTraceIds []string
 				var linkSpanIds []string
@@ -188,11 +180,7 @@ func (b *TracesBatch) Add(req *v1.ExportTraceServiceRequest) {
 					linkTraceIds = append(linkTraceIds, hex.EncodeToString(l.GetTraceId()))
 					linkSpanIds = append(linkSpanIds, hex.EncodeToString(l.GetSpanId()))
 					linkTraceStates = append(linkTraceStates, l.GetTraceState())
-					attrs := map[string]string{}
-					for _, a := range l.GetAttributes() {
-						attrs[a.GetKey()] = a.GetValue().GetStringValue()
-					}
-					linkAttributes = append(linkAttributes, attrs)
+					linkAttributes = append(linkAttributes, attributesToMap(l.GetAttributes()))
 				}
 
 				b.Timestamp.Append(time.Unix(0, int64(s.GetStartTimeUnixNano())))

collector/utils.go

@@ -0,0 +1,60 @@
+package collector
+
+import (
+	"encoding/json"
+	"fmt"
+	"math"
+	"strconv"
+
+	v1 "go.opentelemetry.io/proto/otlp/common/v1"
+)
+
+func attributesToMap(kv []*v1.KeyValue) map[string]string {
+	res := make(map[string]string, len(kv))
+	for _, attr := range kv {
+		res[attr.GetKey()] = valueToString(attr.GetValue())
+	}
+	return res
+}
+
+func valueToString(value *v1.AnyValue) string {
+	switch value.Value.(type) {
+	case *v1.AnyValue_StringValue:
+		return value.GetStringValue()
+	case *v1.AnyValue_BoolValue:
+		return strconv.FormatBool(value.GetBoolValue())
+	case *v1.AnyValue_IntValue:
+		return strconv.FormatInt(value.GetIntValue(), 10)
+	case *v1.AnyValue_DoubleValue:
+		v := value.GetDoubleValue()
+		if math.IsNaN(v) {
+			return "NaN"
+		}
+		if math.IsInf(v, 1) {
+			return "+Inf"
+		}
+		if math.IsInf(v, -1) {
+			return "-Inf"
+		}
+		return strconv.FormatFloat(v, 'f', -1, 64)
+	case *v1.AnyValue_BytesValue:
+		return string(value.GetBytesValue())
+	case *v1.AnyValue_ArrayValue:
+		vs := value.GetArrayValue().GetValues()
+		s := make([]string, 0, len(vs))
+		for _, v := range vs {
+			s = append(s, valueToString(v))
+		}
+		j, _ := json.Marshal(s)
+		return string(j)
+	case *v1.AnyValue_KvlistValue:
+		vs := value.GetKvlistValue().Values
+		s := make(map[string]string, len(vs))
+		for _, kv := range vs {
+			s[kv.GetKey()] = valueToString(kv.GetValue())
+		}
+		j, _ := json.Marshal(s)
+		return string(j)
+	}
+	return fmt.Sprintf("unknown attribute value type: %T", value.Value)
+}