add migrating dockerstats metadata

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Released TBA
 
+- feat(sumologicschemaprocessor): add translating docker stats resource attributes [#1081]
+
+[#1081]: https://github.com/SumoLogic/sumologic-otel-collector/pull/1081
 [Unreleased]: https://github.com/SumoLogic/sumologic-otel-collector/compare/v0.73.0-sumo-1...main
 
 ## [v0.73.0-sumo-1]

docs/migration.md

@@ -61,6 +61,7 @@ You should manually migrate your Sources to an OpenTelemetry Configuration.
     - [Fields](#fields-3)
     - [Scan interval](#scan-interval)
     - [Metrics](#metrics)
+    - [Metadata](#metadata)
   - [Script Source](#script-source)
   - [Streaming Metrics Source](#streaming-metrics-source)
     - [Overall example](#overall-example-3)
@@ -69,14 +70,14 @@ You should manually migrate your Sources to an OpenTelemetry Configuration.
     - [Protocol and Port](#protocol-and-port-1)
     - [Content Type](#content-type)
     - [Source Category](#source-category-3)
-    - [Metadata](#metadata)
+    - [Metadata](#metadata-1)
   - [Host Metrics Source](#host-metrics-source)
     - [Overall Example](#overall-example-4)
     - [Name](#name-6)
     - [Description](#description-6)
     - [Source Host](#source-host-4)
     - [Source Category](#source-category-4)
-    - [Metadata](#metadata-1)
+    - [Metadata](#metadata-2)
     - [Scan Interval](#scan-interval-1)
     - [Metrics](#metrics-1)
       - [CPU](#cpu)
@@ -1522,6 +1523,28 @@ processors:
     translate_docker_metrics: true
 ```
 
+#### Metadata
+
+The metadata sent by Installed Collector correspond to the metadata sent by OpenTelemetry Collector in the following way:
+
+- `container.FullID` corresponds to `container.id`
+- `container.ID`, which is a shorter version of `container.FullID` is not being emitted - if needed, `transform` processor can be used to trim it
+- `container.ImageName` corresponds to `container.image.name`
+- `container.Name` corresponds to `container.name`
+- `container.ImageID` and `container.ImageFullID` are not being emitted
+
+These metadata is represented as resource attributes and can be translated by using [Sumo Logic Schema Processor][sumologicschemaprocessor]
+in the same way as for translating metric names, by using the following config:
+
+```yaml
+processors:
+  sumologicschema/dockerstats:
+    translate_docker_metrics: true
+```
+
+In addition, there is some additional metadata sent by the OpenTelemetry Collector. Full list of it can be seen [here]
+[dockerstatsmetrics].
+
 ### Script Source
 
 Script Source is not supported by the OpenTelemetry Collector.

pkg/processor/sumologicschemaprocessor/README.md

@@ -36,6 +36,7 @@ processors:
     # Specifies whether docker stats metric names should be translated to match
     # Sumo Logic conventions expected in Sumo Logic host related apps (for example
     # `container.cpu.usage.kernelmode` => `cpu_usage.usage_in_kernelmode` or `container.memory.usage.max` => `max_usage`).
+    # It also translates some resource attribute names (for example `container.id` => `container.FullID`).
     # See `translate_docker_metrics_processor.go` for full list of translations.
     # default = false
     translate_docker_metrics: {true, false}

pkg/processor/sumologicschemaprocessor/processor_test.go

@@ -667,21 +667,43 @@ func TestTranslateTelegrafMetrics(t *testing.T) {
 
 func TestTranslateDockerMetrics(t *testing.T) {
 	testCases := []struct {
-		testName        string
-		originalNames   []string
-		translatedNames []string
-		shouldTranslate bool
+		testName                     string
+		originalNames                []string
+		translatedNames              []string
+		originalResourceAttributes   map[string]string
+		translatedResourceAttributes map[string]interface{}
+		shouldTranslate              bool
 	}{
 		{
 			testName:        "translates two names",
 			originalNames:   []string{"container.cpu.usage.percpu", "container.blockio.io_serviced_recursive"},
 			translatedNames: []string{"cpu_usage.percpu_usage", "io_serviced_recursive"},
+			originalResourceAttributes: map[string]string{
+				"container.id":         "a",
+				"container.image.name": "a",
+				"container.name":       "a",
+			},
+			translatedResourceAttributes: map[string]interface{}{
+				"container.FullID":    "a",
+				"container.ImageName": "a",
+				"container.Name":      "a",
+			},
 			shouldTranslate: true,
 		},
 		{
 			testName:        "does not translate",
 			originalNames:   []string{"container.cpu.usage.percpu", "container.blockio.io_serviced_recursive"},
 			translatedNames: []string{"container.cpu.usage.percpu", "container.blockio.io_serviced_recursive"},
+			originalResourceAttributes: map[string]string{
+				"container.id":         "a",
+				"container.image.name": "a",
+				"container.name":       "a",
+			},
+			translatedResourceAttributes: map[string]interface{}{
+				"container.id":         "a",
+				"container.image.name": "a",
+				"container.name":       "a",
+			},
 			shouldTranslate: false,
 		},
 	}
@@ -698,6 +720,12 @@ func TestTranslateDockerMetrics(t *testing.T) {
 				metrics.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty().SetName(name)
 			}
 
+			// Prepare resource attributes
+			ra := metrics.ResourceMetrics().At(0).Resource().Attributes()
+			for k, v := range testCase.originalResourceAttributes {
+				ra.PutStr(k, v)
+			}
+
 			// Act
 			resultMetrics, err := processor.processMetrics(context.Background(), metrics)
 			require.NoError(t, err)
@@ -706,6 +734,7 @@ func TestTranslateDockerMetrics(t *testing.T) {
 			for index, name := range testCase.translatedNames {
 				assert.Equal(t, name, resultMetrics.ResourceMetrics().At(index).ScopeMetrics().At(0).Metrics().At(0).Name())
 			}
+			assert.Equal(t, testCase.translatedResourceAttributes, metrics.ResourceMetrics().At(0).Resource().Attributes().AsRaw())
 		})
 	}
 }

pkg/processor/sumologicschemaprocessor/translate_docker_metrics_processor.go

@@ -15,6 +15,7 @@
 package sumologicschemaprocessor
 
 import (
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
@@ -78,6 +79,12 @@ var dockerMetricsTranslations = map[string]string{
 	"container.blockio.sectors_recursive":             "sectors_recursive",
 }
 
+var dockerReasourceAttributeTranslations = map[string]string{
+	"container.id":         "container.FullID",
+	"container.image.name": "container.ImageName",
+	"container.name":       "container.Name",
+}
+
 func newTranslateDockerMetricsProcessor(shouldTranslate bool) (*translateDockerMetricsProcessor, error) {
 	return &translateDockerMetricsProcessor{
 		shouldTranslate: shouldTranslate,
@@ -93,6 +100,7 @@ func (proc *translateDockerMetricsProcessor) processMetrics(metrics pmetric.Metr
 	if proc.shouldTranslate {
 		for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
 			rm := metrics.ResourceMetrics().At(i)
+			translateDockerResourceAttributes(rm.Resource().Attributes())
 
 			for j := 0; j < rm.ScopeMetrics().Len(); j++ {
 				metricsSlice := rm.ScopeMetrics().At(j).Metrics()
@@ -127,3 +135,32 @@ func translateDockerMetric(m pmetric.Metric) {
 		m.SetName(name)
 	}
 }
+
+func translateDockerResourceAttributes(attributes pcommon.Map) {
+	result := pcommon.NewMap()
+	result.EnsureCapacity(attributes.Len())
+
+	attributes.Range(func(otKey string, value pcommon.Value) bool {
+		if sumoKey, ok := dockerReasourceAttributeTranslations[otKey]; ok {
+			// Only insert if it doesn't exist yet to prevent overwriting.
+			// We have to do it this way since the final return value is not
+			// ready yet to rely on .Insert() not overwriting.
+			if _, exists := attributes.Get(sumoKey); !exists {
+				if _, ok := result.Get(sumoKey); !ok {
+					value.CopyTo(result.PutEmpty(sumoKey))
+				}
+			} else {
+				if _, ok := result.Get(otKey); !ok {
+					value.CopyTo(result.PutEmpty(otKey))
+				}
+			}
+		} else {
+			if _, ok := result.Get(otKey); !ok {
+				value.CopyTo(result.PutEmpty(otKey))
+			}
+		}
+		return true
+	})
+
+	result.CopyTo(attributes)
+}

pkg/processor/sumologicschemaprocessor/translate_docker_metrics_processor_test.go

@@ -18,6 +18,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 )
 
@@ -86,3 +87,26 @@ func TestTranslateDockerMetric_NamesAreTranslatedCorrectly(t *testing.T) {
 		})
 	}
 }
+
+func TestTranslateDockerMetric_ResourceAttrbutesAreTranslatedCorrectly(t *testing.T) {
+	testcases := []struct {
+		nameIn  string
+		nameOut string
+	}{
+		{nameIn: "container.id", nameOut: "container.FullID"},
+		{nameIn: "container.image.name", nameOut: "container.ImageName"},
+		{nameIn: "container.name", nameOut: "container.Name"},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.nameIn+"-"+tc.nameOut, func(t *testing.T) {
+			actual := pcommon.NewMap()
+			actual.PutStr(tc.nameIn, "a")
+			translateDockerResourceAttributes(actual)
+
+			res, ok := actual.Get(tc.nameOut)
+			assert.True(t, ok)
+			assert.Equal(t, res.AsString(), "a")
+		})
+	}
+}