Add Decouple and Batch Processors to Collector (aws-observability#959)

* Add Decouple and Batch Processors

Add a new processor that decouples the receiver and exporter sides of the pipeline and is aware of lambda lifecycle events.

Also add the Batch processor to the list of available processor to reduce the cost of lambda function invocation at the expense of data being delayed.

* Fix missing go.sum entries

* Add link to lambda lifecycle

* Add additional comments to clarify lifecycle

* Update README.md

* Add Makefile to lambdalifecycle

* Fix race detector error

* Update dependencies

Author: adcharre

collector/go.mod

@@ -4,8 +4,12 @@ go 1.20
 
 replace github.com/open-telemetry/opentelemetry-lambda/collector/lambdacomponents => ./lambdacomponents
 
+replace github.com/open-telemetry/opentelemetry-lambda/collector/lambdalifecycle => ./lambdalifecycle
+
 replace github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor => ./processor/coldstartprocessor
 
+replace github.com/open-telemetry/opentelemetry-lambda/collector/processor/decoupleprocessor => ./processor/decoupleprocessor
+
 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:
@@ -18,6 +22,7 @@ require (
 	github.com/golang-collections/go-datastructures v0.0.0-20150211160725-59788d5eb259
 	github.com/open-telemetry/opentelemetry-collector-contrib/confmap/provider/s3provider v0.88.0
 	github.com/open-telemetry/opentelemetry-lambda/collector/lambdacomponents v0.88.0
+	github.com/open-telemetry/opentelemetry-lambda/collector/lambdalifecycle v0.0.0-00010101000000-000000000000
 	github.com/stretchr/testify v1.8.4
 	go.opentelemetry.io/collector/component v0.88.0
 	go.opentelemetry.io/collector/confmap v0.88.0
@@ -94,6 +99,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourceprocessor v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor v0.88.0 // indirect
+	github.com/open-telemetry/opentelemetry-lambda/collector/processor/decoupleprocessor v0.0.0-00010101000000-000000000000 // indirect
 	github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver v0.88.0 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
@@ -138,6 +144,7 @@ require (
 	go.opentelemetry.io/collector/featuregate v1.0.0-rcv0017 // indirect
 	go.opentelemetry.io/collector/pdata v1.0.0-rcv0017 // indirect
 	go.opentelemetry.io/collector/processor v0.88.0 // indirect
+	go.opentelemetry.io/collector/processor/batchprocessor v0.88.0 // indirect
 	go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.88.0 // indirect
 	go.opentelemetry.io/collector/receiver v0.88.0 // indirect
 	go.opentelemetry.io/collector/receiver/otlpreceiver v0.88.0 // indirect

collector/go.sum

@@ -643,6 +643,8 @@ go.opentelemetry.io/collector/pdata v1.0.0-rcv0017 h1:AgALhc2VenoA5l1DvTdg7mkzaB
 go.opentelemetry.io/collector/pdata v1.0.0-rcv0017/go.mod h1:Rv9fOclA5AtM/JGm0d4jBOIAo1+jBA13UT5Bx0ovXi4=
 go.opentelemetry.io/collector/processor v0.88.0 h1:5BUZaH+RhTpgTVqBZCrBnN/vl0M1CtwQsZ8ek4iH1lc=
 go.opentelemetry.io/collector/processor v0.88.0/go.mod h1:2T5KxgBQxXuuyMu9dh+PIBxQ/geCFYcdnjmlWZx8o3E=
+go.opentelemetry.io/collector/processor/batchprocessor v0.88.0 h1:KEifeRMC9JysHpVhQPEyD29C+gqhP0cHuFpJMJUbE/Y=
+go.opentelemetry.io/collector/processor/batchprocessor v0.88.0/go.mod h1:SQhHxRcZ92/DLufTYzb4xnxnR/uuW5makoqezBlJgJ4=
 go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.88.0 h1:m37nR0I1F7ao3qAJtzHB4GwBr1qrtDaPKNfeBOYzDF0=
 go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.88.0/go.mod h1:p5+E4COkqfUPip3q2HFGuieBZbryI0h8KPP7jNYZvsY=
 go.opentelemetry.io/collector/receiver v0.88.0 h1:MPvVAFOfjl0+Ylka7so8QoK8T2Za2471rv5t3sqbbSY=

collector/internal/lifecycle/manager.go

@@ -17,6 +17,7 @@ package lifecycle
 import (
 	"context"
 	"fmt"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/lambdalifecycle"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -42,11 +43,12 @@ type collectorWrapper interface {
 }
 
 type manager struct {
-	logger          *zap.Logger
-	collector       collectorWrapper
-	extensionClient *extensionapi.Client
-	listener        *telemetryapi.Listener
-	wg              sync.WaitGroup
+	logger             *zap.Logger
+	collector          collectorWrapper
+	extensionClient    *extensionapi.Client
+	listener           *telemetryapi.Listener
+	wg                 sync.WaitGroup
+	lifecycleListeners []lambdalifecycle.Listener
 }
 
 func NewManager(ctx context.Context, logger *zap.Logger, version string) (context.Context, *manager) {
@@ -132,6 +134,7 @@ func (lm *manager) processEvents(ctx context.Context) error {
 			// Exit if we receive a SHUTDOWN event
 			if res.EventType == extensionapi.Shutdown {
 				lm.logger.Info("Received SHUTDOWN event")
+				lm.notifyEnvironmentShutdown()
 				lm.listener.Shutdown()
 				err = lm.collector.Stop()
 				if err != nil {
@@ -142,10 +145,37 @@ func (lm *manager) processEvents(ctx context.Context) error {
 				return err
 			}
 
+			lm.notifyFunctionInvoked()
+
 			err = lm.listener.Wait(ctx, res.RequestID)
 			if err != nil {
 				lm.logger.Error("problem waiting for platform.runtimeDone event", zap.Error(err), zap.String("requestID", res.RequestID))
 			}
+
+			// Check other components are ready before allowing the freezing of the environment.
+			lm.notifyFunctionFinished()
 		}
 	}
 }
+
+func (lm *manager) notifyFunctionInvoked() {
+	for _, listener := range lm.lifecycleListeners {
+		listener.FunctionInvoked()
+	}
+}
+
+func (lm *manager) notifyFunctionFinished() {
+	for _, listener := range lm.lifecycleListeners {
+		listener.FunctionFinished()
+	}
+}
+
+func (lm *manager) notifyEnvironmentShutdown() {
+	for _, listener := range lm.lifecycleListeners {
+		listener.EnvironmentShutdown()
+	}
+}
+
+func (lm *manager) AddListener(listener lambdalifecycle.Listener) {
+	lm.lifecycleListeners = append(lm.lifecycleListeners, listener)
+}

collector/lambdacomponents/default.go

@@ -22,13 +22,15 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/probabilisticsamplerprocessor"
 	"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/decoupleprocessor"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/loggingexporter"
 	"go.opentelemetry.io/collector/exporter/otlpexporter"
 	"go.opentelemetry.io/collector/exporter/otlphttpexporter"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/otelcol"
 	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/batchprocessor"
 	"go.opentelemetry.io/collector/processor/memorylimiterprocessor"
 	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/receiver/otlpreceiver"
@@ -67,6 +69,8 @@ func Components(extensionID string) (otelcol.Factories, error) {
 		resourceprocessor.NewFactory(),
 		spanprocessor.NewFactory(),
 		coldstartprocessor.NewFactory(),
+		decoupleprocessor.NewFactory(),
+		batchprocessor.NewFactory(),
 	)
 	if err != nil {
 		errs = append(errs, err)

collector/lambdacomponents/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourceprocessor v0.88.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor v0.88.0
 	github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor v0.88.0
+	github.com/open-telemetry/opentelemetry-lambda/collector/processor/decoupleprocessor v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver v0.88.0
 	go.opentelemetry.io/collector/exporter v0.88.0
 	go.opentelemetry.io/collector/exporter/loggingexporter v0.88.0
@@ -19,6 +20,7 @@ require (
 	go.opentelemetry.io/collector/extension v0.88.0
 	go.opentelemetry.io/collector/otelcol v0.88.0
 	go.opentelemetry.io/collector/processor v0.88.0
+	go.opentelemetry.io/collector/processor/batchprocessor v0.88.0
 	go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.88.0
 	go.opentelemetry.io/collector/receiver v0.88.0
 	go.opentelemetry.io/collector/receiver/otlpreceiver v0.88.0
@@ -82,6 +84,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheusremotewrite v0.88.0 // indirect
 	github.com/open-telemetry/opentelemetry-lambda/collector v0.88.0 // indirect
+	github.com/open-telemetry/opentelemetry-lambda/collector/lambdalifecycle v0.0.0-00010101000000-000000000000 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus/client_golang v1.17.0 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
@@ -163,6 +166,10 @@ 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/lambdalifecycle => ../lambdalifecycle
+
 replace github.com/open-telemetry/opentelemetry-lambda/collector/processor/coldstartprocessor => ../processor/coldstartprocessor
 
+replace github.com/open-telemetry/opentelemetry-lambda/collector/processor/decoupleprocessor => ../processor/decoupleprocessor
+
 replace github.com/open-telemetry/opentelemetry-lambda/collector/receiver/telemetryapireceiver => ../receiver/telemetryapireceiver

collector/lambdacomponents/go.sum

@@ -626,6 +626,8 @@ go.opentelemetry.io/collector/pdata v1.0.0-rcv0017 h1:AgALhc2VenoA5l1DvTdg7mkzaB
 go.opentelemetry.io/collector/pdata v1.0.0-rcv0017/go.mod h1:Rv9fOclA5AtM/JGm0d4jBOIAo1+jBA13UT5Bx0ovXi4=
 go.opentelemetry.io/collector/processor v0.88.0 h1:5BUZaH+RhTpgTVqBZCrBnN/vl0M1CtwQsZ8ek4iH1lc=
 go.opentelemetry.io/collector/processor v0.88.0/go.mod h1:2T5KxgBQxXuuyMu9dh+PIBxQ/geCFYcdnjmlWZx8o3E=
+go.opentelemetry.io/collector/processor/batchprocessor v0.88.0 h1:KEifeRMC9JysHpVhQPEyD29C+gqhP0cHuFpJMJUbE/Y=
+go.opentelemetry.io/collector/processor/batchprocessor v0.88.0/go.mod h1:SQhHxRcZ92/DLufTYzb4xnxnR/uuW5makoqezBlJgJ4=
 go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.88.0 h1:m37nR0I1F7ao3qAJtzHB4GwBr1qrtDaPKNfeBOYzDF0=
 go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.88.0/go.mod h1:p5+E4COkqfUPip3q2HFGuieBZbryI0h8KPP7jNYZvsY=
 go.opentelemetry.io/collector/receiver v0.88.0 h1:MPvVAFOfjl0+Ylka7so8QoK8T2Za2471rv5t3sqbbSY=

collector/lambdalifecycle/Makefile

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

collector/lambdalifecycle/go.mod

@@ -0,0 +1,3 @@
+module github.com/open-telemetry/opentelemetry-lambda/collector/lambdalifecycle
+
+go 1.20

collector/lambdalifecycle/notifier.go

@@ -0,0 +1,43 @@
+// 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 lambdalifecycle
+
+// Listener interface used to notify objects of Lambda lifecycle events.
+type Listener interface {
+	// FunctionInvoked is called after the extension receives a "Next" notification.
+	FunctionInvoked()
+	// FunctionFinished is called after the extension is notified that the function has completed, but before the environment is frozen.
+	// The environment is only frozen once all listeners have returned.
+	FunctionFinished()
+	// EnvironmentShutdown is called when the extension is notified that the environment is about to shut down.
+	// Shutting down of the collector components only happens after all listeners have returned.
+	EnvironmentShutdown()
+}
+
+type Notifier interface {
+	AddListener(listener Listener)
+}
+
+var (
+	notifier Notifier
+)
+
+func SetNotifier(n Notifier) {
+	notifier = n
+}
+
+func GetNotifier() Notifier {
+	return notifier
+}

collector/main.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/lambdalifecycle"
 	"os"
 
 	"go.uber.org/zap"
@@ -47,6 +48,9 @@ func main() {
 
 	ctx, lm := lifecycle.NewManager(context.Background(), logger, Version)
 
+	// Set the new lifecycle manager as the lifecycle notifier for all other components.
+	lambdalifecycle.SetNotifier(lm)
+
 	// Will block until shutdown event is received or cancelled via the context.
 	logger.Info("done", zap.Error(lm.Run(ctx)))
 }