[processor/coldstart] add coldstart processor (aws-observability#346)

Alex Boten · web-flow · commit 9ebe1089e34e · 2023-01-19T13:56:34.000-08:00
Follow up to aws-observability#345 This processor associates cold start timing information generated by the telemetryapireceiver with the first span processed by the Collector extension containing a faas.execution attribute. Signed-off-by: Alex Boten aboten@lightstep.com
diff --git a/collector/go.mod b/collector/go.mod
@@ -4,6 +4,8 @@ go 1.18
 
 replace github.com/open-telemetry/opentelemetry-lambda/collector/lambdacomponents => ./lambdacomponents
 
+replace github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor => ./processor/coldstartprocessor
+
 replace github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver => ./receiver/telemetryapireceiver
 
 // fixes ambiguous import error: found package cloud.google.com/go/compute/metadata in multiple modules:
@@ -88,6 +90,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/probabilisticsamplerprocessor v0.69.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourceprocessor v0.69.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor v0.69.0 // indirect
+	github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor v0.69.0 // indirect
 	github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver v0.69.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
diff --git a/collector/go.sum b/collector/go.sum
@@ -440,6 +440,7 @@ github.com/cenkalti/backoff/v4 v4.1.1/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInq
 github.com/cenkalti/backoff/v4 v4.2.0 h1:HN5dHm3WBOgndBH6E8V0q2jIYIR3s9yglV8k/+MN3u4=
 github.com/cenkalti/backoff/v4 v4.2.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
+github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.0/go.mod h1:dgIUBU3pDso/gPgZ1osOZ0iQf77oPR28Tjxl5dIMyVM=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
diff --git a/collector/lambdacomponents/default.go b/collector/lambdacomponents/default.go
@@ -35,6 +35,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourceprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor"
 
+	"github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor"
 	"github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver"
 )
 
@@ -66,6 +67,7 @@ func Components(extensionID string) (otelcol.Factories, error) {
 		probabilisticsamplerprocessor.NewFactory(),
 		resourceprocessor.NewFactory(),
 		spanprocessor.NewFactory(),
+		coldstartprocessor.NewFactory(),
 	)
 	if err != nil {
 		errs = append(errs, err)
diff --git a/collector/lambdacomponents/go.mod b/collector/lambdacomponents/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/probabilisticsamplerprocessor v0.69.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourceprocessor v0.69.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor v0.69.0
+	github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor v0.69.0
 	github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver v0.69.0
 	go.opentelemetry.io/collector v0.69.1
 	go.opentelemetry.io/collector/exporter/loggingexporter v0.69.1
@@ -129,4 +130,6 @@ replace cloud.google.com/go => cloud.google.com/go v0.107.0
 
 replace github.com/open-telemetry/opentelemetry-lambda/collector => ../
 
+replace github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor => ../processor/coldstartprocessor
+
 replace github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver => ../receiver/telemetryapireceiver
diff --git a/collector/lambdacomponents/go.sum b/collector/lambdacomponents/go.sum
@@ -426,6 +426,7 @@ github.com/cenkalti/backoff/v4 v4.1.1/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInq
 github.com/cenkalti/backoff/v4 v4.2.0 h1:HN5dHm3WBOgndBH6E8V0q2jIYIR3s9yglV8k/+MN3u4=
 github.com/cenkalti/backoff/v4 v4.2.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
+github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.0/go.mod h1:dgIUBU3pDso/gPgZ1osOZ0iQf77oPR28Tjxl5dIMyVM=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
diff --git a/collector/processor/coldstartprocessor/Makefile b/collector/processor/coldstartprocessor/Makefile
@@ -0,0 +1 @@
+include ../../Makefile.Common
diff --git a/collector/processor/coldstartprocessor/README.md b/collector/processor/coldstartprocessor/README.md
@@ -0,0 +1,32 @@
+# Coldstart Processor
+
+| Status                   |                 |
+| ------------------------ |-----------------|
+| Stability                | [in development]|
+| Supported pipeline types | traces          |
+| Distributions            | [extension]     |
+
+This processor associates cold start information generated by the [telemetryapireceiver](../../receiver/telemetryapireceiver) with incoming span data processed by
+the Collector extension. It reads the following of incoming Lambda execution spans identified by the `faas.execution` attribute:
+
+- trace ID
+- parent span ID
+- span scope
+- resource attributes
+
+This information is stored until the first coldstart span identified by the `faas.coldstart` attribute
+is received. That span's trace ID, parent span ID are updated to match the execution span. This
+allows the coldstart span to be part of the same trace as the operation that triggered the cold start. Additionally,
+the span scope and resource attributes of the span scope containing the coldstart span
+are replaced with the span scope and resource attributes of the execution span as
+they contain more details.
+
+There are currently no configuration parameters available for this processor. It can be enabled via the following configuration:
+
+```yaml
+processors:
+    coldstart:
+```
+
+[in development]: https://github.com/open-telemetry/opentelemetry-collector#development
+[extension]: https://github.com/open-telemetry/opentelemetry-lambda/collector
diff --git a/collector/processor/coldstartprocessor/config.go b/collector/processor/coldstartprocessor/config.go
@@ -0,0 +1,23 @@
+// 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 coldstartprocessor // import "github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor"
+
+// Config defines the configuration for the various elements of the processor.
+type Config struct{}
+
+// Validate validates the configuration by checking for missing or invalid fields
+func (cfg *Config) Validate() error {
+	return nil
+}
diff --git a/collector/processor/coldstartprocessor/config_test.go b/collector/processor/coldstartprocessor/config_test.go
@@ -0,0 +1,47 @@
+// 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 coldstartprocessor // import "github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor"
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestValidate(t *testing.T) {
+	testCases := []struct {
+		desc        string
+		cfg         *Config
+		expectedErr error
+	}{
+		{
+			desc:        "valid config",
+			cfg:         &Config{},
+			expectedErr: nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			actualErr := tc.cfg.Validate()
+			if tc.expectedErr != nil {
+				require.EqualError(t, actualErr, tc.expectedErr.Error())
+			} else {
+				require.NoError(t, actualErr)
+			}
+
+		})
+	}
+}
diff --git a/collector/processor/coldstartprocessor/doc.go b/collector/processor/coldstartprocessor/doc.go
@@ -0,0 +1,31 @@
+// 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 coldstartprocessor correlates cold start information generated by the telemetryapireceiver
+// with incoming span data.
+//
+// It reads the following of incoming Lambda execution spans identified by the faas.execution attribute:
+//
+//   - trace ID
+//   - parent span ID
+//   - span scope
+//   - resource attributes
+//
+// This information is stored until the first coldstart span identified by the faas.coldstart attribute
+// is received. That span's trace ID, parent span ID are updated to match the execution span. This
+// allows the coldstart span to be part of the same trace as the operation that triggered the cold start. Additionally,
+// the span scope and resource attributes of the span scope containing the coldstart span
+// are replaced with the span scope and resource attributes of the execution span as
+// they contain more details.
+package coldstartprocessor // import "github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor"
diff --git a/collector/processor/coldstartprocessor/factory.go b/collector/processor/coldstartprocessor/factory.go
@@ -0,0 +1,68 @@
+// 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 coldstartprocessor // import "github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor"
+
+import (
+	"context"
+	"errors"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorhelper"
+)
+
+const (
+	typeStr   = "coldstart"
+	stability = component.StabilityLevelDevelopment
+)
+
+var (
+	errConfigNotColdstart = errors.New("config was not a Coldstart processor config")
+	processorCapabilities = consumer.Capabilities{MutatesData: true}
+)
+
+func NewFactory() processor.Factory {
+	return processor.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		processor.WithTraces(createTracesProcessor, stability),
+	)
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{}
+}
+
+func createTracesProcessor(ctx context.Context, params processor.CreateSettings, rConf component.Config, next consumer.Traces) (processor.Traces, error) {
+	cfg, ok := rConf.(*Config)
+	if !ok {
+		return nil, errConfigNotColdstart
+	}
+
+	cp, err := newColdstartProcessor(cfg, next, params)
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewTracesProcessor(
+		ctx,
+		params,
+		cfg,
+		next,
+		cp.processTraces,
+		processorhelper.WithCapabilities(processorCapabilities),
+	)
+
+}
diff --git a/collector/processor/coldstartprocessor/factory_test.go b/collector/processor/coldstartprocessor/factory_test.go
@@ -0,0 +1,70 @@
+// 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 coldstartprocessor // import "github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor"
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/processor/processortest"
+)
+
+func TestNewFactory(t *testing.T) {
+	testCases := []struct {
+		desc     string
+		testFunc func(*testing.T)
+	}{
+		{
+			desc: "creates a new factory with correct type",
+			testFunc: func(t *testing.T) {
+				factory := NewFactory()
+				require.EqualValues(t, typeStr, factory.Type())
+			},
+		},
+		{
+			desc: "creates a new factory and CreateTracesProcessor returns no error",
+			testFunc: func(t *testing.T) {
+				factory := NewFactory()
+				cfg := factory.CreateDefaultConfig()
+				_, err := factory.CreateTracesProcessor(
+					context.Background(),
+					processortest.NewNopCreateSettings(),
+					cfg,
+					consumertest.NewNop(),
+				)
+				require.NoError(t, err)
+			},
+		},
+		{
+			desc: "creates a new factory and CreateTracesProcessor returns error with incorrect config",
+			testFunc: func(t *testing.T) {
+				factory := NewFactory()
+				_, err := factory.CreateTracesProcessor(
+					context.Background(),
+					processortest.NewNopCreateSettings(),
+					nil,
+					consumertest.NewNop(),
+				)
+				require.ErrorIs(t, err, errConfigNotColdstart)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, tc.testFunc)
+	}
+}
diff --git a/collector/processor/coldstartprocessor/go.mod b/collector/processor/coldstartprocessor/go.mod
@@ -0,0 +1,43 @@
+module github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor
+
+go 1.19
+
+require (
+	github.com/cespare/xxhash v1.1.0
+	github.com/stretchr/testify v1.8.1
+	go.opentelemetry.io/collector v0.69.1
+	go.opentelemetry.io/collector/component v0.69.1
+	go.opentelemetry.io/collector/consumer v0.69.1
+	go.opentelemetry.io/collector/pdata v1.0.0-rc3
+	go.opentelemetry.io/collector/semconv v0.69.1
+	go.uber.org/multierr v1.9.0
+	go.uber.org/zap v1.24.0
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/knadh/koanf v1.4.4 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/mapstructure v1.5.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/collector/confmap v0.69.1 // indirect
+	go.opentelemetry.io/collector/featuregate v0.69.1 // indirect
+	go.opentelemetry.io/otel v1.11.2 // indirect
+	go.opentelemetry.io/otel/metric v0.34.0 // indirect
+	go.opentelemetry.io/otel/trace v1.11.2 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
+	golang.org/x/net v0.4.0 // indirect
+	golang.org/x/sys v0.3.0 // indirect
+	golang.org/x/text v0.5.0 // indirect
+	google.golang.org/genproto v0.0.0-20211208223120-3a66f561d7aa // indirect
+	google.golang.org/grpc v1.51.0 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
diff --git a/collector/processor/coldstartprocessor/go.sum b/collector/processor/coldstartprocessor/go.sum
diff --git a/collector/processor/coldstartprocessor/processor.go b/collector/processor/coldstartprocessor/processor.go
diff --git a/collector/processor/coldstartprocessor/processor_test.go b/collector/processor/coldstartprocessor/processor_test.go

Original file line number	Diff line number	Diff line change
`@@ -35,6 +35,7 @@ import (`
`35`	`35`	`"github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourceprocessor"`
`36`	`36`	`"github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor"`
`37`	`37`
	`38`	`+ "github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor"`
`38`	`39`	`"github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver"`
`39`	`40`	`)`
`40`	`41`
`@@ -66,6 +67,7 @@ func Components(extensionID string) (otelcol.Factories, error) {`
`66`	`67`	`probabilisticsamplerprocessor.NewFactory(),`
`67`	`68`	`resourceprocessor.NewFactory(),`
`68`	`69`	`spanprocessor.NewFactory(),`
	`70`	`+ coldstartprocessor.NewFactory(),`
`69`	`71`	`)`
`70`	`72`	`if err != nil {`
`71`	`73`	`errs = append(errs, err)`