Add spanmetricsprocessor

Signed-off-by: albertteoh <albert.teoh@logz.io>

cmd/otelcontribcol/components.go

@@ -46,6 +46,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/metricstransformprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourcedetectionprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/routingprocessor"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanmetricsprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsecscontainermetricsreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsxrayreceiver"
@@ -156,6 +157,7 @@ func components() (component.Factories, error) {
 		resourcedetectionprocessor.NewFactory(),
 		routingprocessor.NewFactory(),
 		tailsamplingprocessor.NewFactory(),
+		spanmetricsprocessor.NewFactory(),
 	}
 	for _, pr := range factories.Processors {
 		processors = append(processors, pr)

go.mod

@@ -35,6 +35,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourcedetectionprocessor v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/routingprocessor v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.0.0-00010101000000-000000000000
+	github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanmetricsprocessor v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsecscontainermetricsreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsxrayreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/carbonreceiver v0.0.0-00010101000000-000000000000
@@ -179,3 +180,5 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/metr
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/routingprocessor => ./processor/routingprocessor/
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor => ./processor/tailsamplingprocessor
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanmetricsprocessor => ./processor/spanmetricsprocessor/

processor/spanmetricsprocessor/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

processor/spanmetricsprocessor/README.md

@@ -0,0 +1,106 @@
+# Span Metrics Processor
+
+Aggregates Request, Error and Duration (R.E.D) metrics from span data.
+
+**Request** counts are computed as the number of spans seen per unique set of dimensions, including Errors.
+For example, the following metric shows 142 calls:
+```
+promexample_calls{http_method="GET",http_status_code="200",operation="/Address",service_name="shippingservice",span_kind="SPAN_KIND_SERVER",status_code="STATUS_CODE_UNSET"} 142
+```
+Multiple metrics can be aggregated if, for instance, a user wishes to view call counts just on `service_name` and `operation`.
+
+**Error** counts are computed from the Request counts which have an "Error" Status Code metric dimension.
+For example, the following metric indicates 220 errors:
+```
+promexample_calls{http_method="GET",http_status_code="503",operation="/checkout",service_name="frontend",span_kind="SPAN_KIND_CLIENT",status_code="STATUS_CODE_ERROR"} 220
+```
+
+**Duration** is computed from the difference between the span start and end times and inserted into the
+relevant latency histogram time bucket for each unique set dimensions.
+For example, the following latency buckets indicate the vast majority of spans (9K) have a 100ms latency:
+```
+promexample_latency_bucket{http_method="GET",http_status_code="200",label1="value1",operation="/Address",service_name="shippingservice",span_kind="SPAN_KIND_SERVER",status_code="STATUS_CODE_UNSET",le="2"} 327
+promexample_latency_bucket{http_method="GET",http_status_code="200",label1="value1",operation="/Address",service_name="shippingservice",span_kind="SPAN_KIND_SERVER",status_code="STATUS_CODE_UNSET",le="6"} 751
+promexample_latency_bucket{http_method="GET",http_status_code="200",label1="value1",operation="/Address",service_name="shippingservice",span_kind="SPAN_KIND_SERVER",status_code="STATUS_CODE_UNSET",le="10"} 1195
+promexample_latency_bucket{http_method="GET",http_status_code="200",label1="value1",operation="/Address",service_name="shippingservice",span_kind="SPAN_KIND_SERVER",status_code="STATUS_CODE_UNSET",le="100"} 10180
+promexample_latency_bucket{http_method="GET",http_status_code="200",label1="value1",operation="/Address",service_name="shippingservice",span_kind="SPAN_KIND_SERVER",status_code="STATUS_CODE_UNSET",le="250"} 10180
+...
+```
+
+Each metric will have _at least_ the following dimensions because they are common across all spans:
+- Service name
+- Operation
+- Span kind
+- Status code
+
+This processor lets traces to continue through the pipeline unmodified.
+
+The following settings are required:
+
+- `metrics_exporter`: the name of the exporter that this processor will write metrics to. This exporter **must** be present in a pipeline.
+
+The following settings can be optionally configured:
+
+- `latency_histogram_buckets`: the list of durations defining the latency histogram buckets.
+  - Default: `[2ms, 4ms, 6ms, 8ms, 10ms, 50ms, 100ms, 200ms, 400ms, 800ms, 1s, 1400ms, 2s, 5s, 10s, 15s]`
+- `dimensions`: the list of dimensions to add together with the default dimensions defined above. Each additional dimension is defined with a `name` which is looked up in the span's collection of attributes. If the `name`d attribute is missing in the span, the optional provided `default` is used. If no `default` is provided, this dimension will be **omitted** from the metric.
+
+Example:
+
+```yaml
+receivers:
+  jaeger:
+    protocols:
+      thrift_http:
+        endpoint: "0.0.0.0:14278"
+
+  # Dummy receiver that's never used, because a pipeline is required to have one.
+  otlp/spanmetrics:
+    protocols:
+      grpc:
+        endpoint: "localhost:12345"
+
+  otlp:
+    protocols:
+      grpc:
+        endpoint: "localhost:55677"
+
+processors:
+  spanmetrics:
+    metrics_exporter: otlp/spanmetrics
+    latency_histogram_buckets: [2ms, 6ms, 10ms, 100ms, 250ms]
+    dimensions:
+      - name: http.method
+        default: GET
+      - name: http.status_code
+
+exporters:
+  jaeger:
+    endpoint: localhost:14250
+
+  otlp/spanmetrics:
+    endpoint: "localhost:55677"
+    insecure: true
+
+  prometheus:
+    endpoint: "0.0.0.0:8889"
+    namespace: promexample
+
+pipelines:
+  traces:
+    receivers: [jaeger]
+    processors: [spanmetrics, batch, queued_retry]
+    exporters: [jaeger]
+
+  # The exporter name must match the metrics_exporter name.
+  # The receiver is just a dummy and never used; added to pass validation requiring at least one receiver in a pipeline.
+  metrics/spanmetrics:
+    receivers: [otlp/spanmetrics]
+    exporters: [otlp/spanmetrics]
+
+  metrics:
+    receivers: [otlp]
+    exporters: [prometheus]
+```
+
+The full list of settings exposed for this processor are documented [here](./config.go) with detailed sample configuration [here](./testdata).

processor/spanmetricsprocessor/config.go

@@ -0,0 +1,46 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package spanmetricsprocessor
+
+import (
+	"time"
+
+	"go.opentelemetry.io/collector/config/configmodels"
+)
+
+type Dimension struct {
+	Name    string  `mapstructure:"name"`
+	Default *string `mapstructure:"default"`
+}
+
+type Config struct {
+	configmodels.ProcessorSettings `mapstructure:",squash"`
+
+	// MetricsExporter is the name of the metrics exporter to use to ship metrics.
+	MetricsExporter string `mapstructure:"metrics_exporter"`
+
+	// LatencyHistogramBuckets is the list of durations representing latency histogram buckets.
+	// Default (in ms): [2, 4, 6, 8, 10, 50, 100, 200, 400, 800, 1000, 1400, 2000, 5000, 10_000, 15_000, rest]
+	LatencyHistogramBuckets []time.Duration `mapstructure:"latency_histogram_buckets"`
+
+	// Dimensions defines the list of additional dimensions on top of the provided:
+	// - service.name
+	// - operation
+	// - span.kind
+	// - status.code
+	// The dimensions will be fetched from the span's attributes. Examples of some conventionally used attributes:
+	// https://github.com/open-telemetry/opentelemetry-collector/blob/master/translator/conventions/opentelemetry.go.
+	Dimensions []Dimension `mapstructure:"dimensions"`
+}

processor/spanmetricsprocessor/config_test.go

@@ -0,0 +1,99 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package spanmetricsprocessor
+
+import (
+	"path"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/config/configtest"
+	"go.opentelemetry.io/collector/exporter/jaegerexporter"
+	"go.opentelemetry.io/collector/exporter/otlpexporter"
+	"go.opentelemetry.io/collector/exporter/prometheusexporter"
+	"go.opentelemetry.io/collector/processor/batchprocessor"
+	"go.opentelemetry.io/collector/processor/queuedprocessor"
+	"go.opentelemetry.io/collector/receiver/jaegerreceiver"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver"
+)
+
+func TestLoadConfig(t *testing.T) {
+	defaultMethod := "GET"
+	testcases := []struct {
+		configFile                  string
+		wantMetricsExporter         string
+		wantLatencyHistogramBuckets []time.Duration
+		wantDimensions              []Dimension
+	}{
+		{configFile: "config-2-pipelines.yaml", wantMetricsExporter: "prometheus"},
+		{configFile: "config-3-pipelines.yaml", wantMetricsExporter: "otlp/spanmetrics"},
+		{
+			configFile:          "config-full.yaml",
+			wantMetricsExporter: "otlp/spanmetrics",
+			wantLatencyHistogramBuckets: []time.Duration{
+				2 * time.Millisecond,
+				6 * time.Millisecond,
+				10 * time.Millisecond,
+				100 * time.Millisecond,
+				250 * time.Millisecond,
+			},
+			wantDimensions: []Dimension{
+				{"http.method", &defaultMethod},
+				{"http.status_code", nil},
+			},
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.configFile, func(t *testing.T) {
+			// Prepare
+			factories, err := componenttest.ExampleComponents()
+			require.NoError(t, err)
+
+			factories.Receivers["otlp"] = otlpreceiver.NewFactory()
+			factories.Receivers["jaeger"] = jaegerreceiver.NewFactory()
+
+			factories.Processors[typeStr] = NewFactory()
+			factories.Processors["batch"] = batchprocessor.NewFactory()
+			factories.Processors["queued_retry"] = queuedprocessor.NewFactory()
+
+			factories.Exporters["otlp"] = otlpexporter.NewFactory()
+			factories.Exporters["prometheus"] = prometheusexporter.NewFactory()
+			factories.Exporters["jaeger"] = jaegerexporter.NewFactory()
+
+			// Test
+			cfg, err := configtest.LoadConfigFile(t, path.Join(".", "testdata", tc.configFile), factories)
+
+			// Verify
+			require.NoError(t, err)
+			require.NotNil(t, cfg)
+			assert.Equal(t,
+				&Config{
+					ProcessorSettings: configmodels.ProcessorSettings{
+						NameVal: "spanmetrics",
+						TypeVal: "spanmetrics",
+					},
+					MetricsExporter:         tc.wantMetricsExporter,
+					LatencyHistogramBuckets: tc.wantLatencyHistogramBuckets,
+					Dimensions:              tc.wantDimensions,
+				},
+				cfg.Processors["spanmetrics"],
+			)
+		})
+	}
+}

processor/spanmetricsprocessor/factory.go

@@ -0,0 +1,51 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package spanmetricsprocessor
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/processor/processorhelper"
+)
+
+const (
+	// The value of "type" key in configuration.
+	typeStr = "spanmetrics"
+)
+
+// NewFactory creates a factory for the spanmetrics processor.
+func NewFactory() component.ProcessorFactory {
+	return processorhelper.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		processorhelper.WithTraces(createTraceProcessor),
+	)
+}
+
+func createDefaultConfig() configmodels.Processor {
+	return &Config{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: typeStr,
+			NameVal: typeStr,
+		},
+	}
+}
+
+func createTraceProcessor(_ context.Context, params component.ProcessorCreateParams, cfg configmodels.Processor, nextConsumer consumer.TracesConsumer) (component.TracesProcessor, error) {
+	return newProcessor(params.Logger, cfg, nextConsumer)
+}