[collector] move lifecyclemanager and collector in internal (aws-observability#409)

* [collector] move lifecyclemanager and collector in internal

Moving lifecyclemanager and the collector components in their own internal packages. Will follow this up with some tests for each package.

Signed-off-by: Alex Boten <[email protected]>

* Apply suggestions from code review

Co-authored-by: Anthony Mirabella <[email protected]>

* [chore] update makefile to include test/tidy (aws-observability#402)

* [chore] update makefile to include test/tidy

This updates the build command to also run tests and tidy.

Signed-off-by: Alex Boten <[email protected]>

* separate test/tidy into their own gh jobs

* Apply suggestions from code review

Co-authored-by: Anthony Mirabella <[email protected]>

Signed-off-by: Alex Boten <[email protected]>
Co-authored-by: Anthony Mirabella <[email protected]>

* cleanup additional references

* rename file

Signed-off-by: Alex Boten <[email protected]>
Co-authored-by: Anthony Mirabella <[email protected]>

Author: Alex Boten

collector/collector.go renamed to collector/internal/collector/collector.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package main
+package collector
 
 import (
 	"context"
@@ -33,14 +33,6 @@ import (
 	"go.uber.org/zap/zapcore"
 )
 
-var (
-	// Version variable will be replaced at link time after `make` has been run.
-	Version = "latest"
-
-	// GitHash variable will be replaced at link time after `make` has been run.
-	GitHash = "<NOT PROPERLY GENERATED>"
-)
-
 // Collector implements the OtelcolRunner interfaces running a single otelcol as a go routine within the
 // same process as the test executor.
 type Collector struct {
@@ -50,6 +42,7 @@ type Collector struct {
 	appDone        chan struct{}
 	stopped        bool
 	logger         *zap.Logger
+	version        string
 }
 
 func getConfig(logger *zap.Logger) string {
@@ -61,7 +54,7 @@ func getConfig(logger *zap.Logger) string {
 	return val
 }
 
-func NewCollector(logger *zap.Logger, factories component.Factories) *Collector {
+func NewCollector(logger *zap.Logger, factories component.Factories, version string) *Collector {
 	l := logger.Named("NewCollector")
 	providers := []confmap.Provider{fileprovider.New(), envprovider.New(), yamlprovider.New(), httpprovider.New(), s3provider.New()}
 	mapProvider := make(map[string]confmap.Provider, len(providers))
@@ -87,6 +80,7 @@ func NewCollector(logger *zap.Logger, factories component.Factories) *Collector
 		factories:      factories,
 		configProvider: cfgProvider,
 		logger:         logger,
+		version:        version,
 	}
 	return col
 }
@@ -96,7 +90,7 @@ func (c *Collector) Start(ctx context.Context) error {
 		BuildInfo: component.BuildInfo{
 			Command:     "otelcol-lambda",
 			Description: "Lambda Collector",
-			Version:     Version,
+			Version:     c.version,
 		},
 		ConfigProvider: c.configProvider,
 		Factories:      c.factories,

collector/internal/lifecycle/manager.go

@@ -0,0 +1,134 @@
+// 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 lifecycle
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"sync"
+	"syscall"
+
+	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/collector"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/extensionapi"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/telemetryapi"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/lambdacomponents"
+	"go.uber.org/zap"
+)
+
+var (
+	extensionName = filepath.Base(os.Args[0]) // extension name has to match the filename
+)
+
+type manager struct {
+	logger          *zap.Logger
+	collector       *collector.Collector
+	extensionClient *extensionapi.Client
+	listener        *telemetryapi.Listener
+	wg              sync.WaitGroup
+}
+
+func NewManager(ctx context.Context, logger *zap.Logger, version string) (context.Context, *manager) {
+	ctx, cancel := context.WithCancel(ctx)
+
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT)
+	go func() {
+		s := <-sigs
+		cancel()
+		logger.Info("received signal", zap.String("signal", s.String()))
+	}()
+
+	extensionClient := extensionapi.NewClient(logger, os.Getenv("AWS_LAMBDA_RUNTIME_API"))
+	res, err := extensionClient.Register(ctx, extensionName)
+	if err != nil {
+		logger.Fatal("Cannot register extension", zap.Error(err))
+	}
+
+	listener := telemetryapi.NewListener(logger)
+	addr, err := listener.Start()
+	if err != nil {
+		logger.Fatal("Cannot start Telemetry API Listener", zap.Error(err))
+	}
+
+	telemetryClient := telemetryapi.NewClient(logger)
+	_, err = telemetryClient.Subscribe(ctx, res.ExtensionID, addr)
+	if err != nil {
+		logger.Fatal("Cannot register Telemetry API client", zap.Error(err))
+	}
+
+	lm := &manager{
+		logger:          logger.Named("lifecycle.manager"),
+		extensionClient: extensionClient,
+		listener:        listener,
+	}
+
+	go lm.processEvents(ctx)
+
+	factories, _ := lambdacomponents.Components()
+	lm.collector = collector.NewCollector(logger, factories, version)
+
+	return ctx, lm
+}
+
+func (lm *manager) Run(ctx context.Context) error {
+	if err := lm.collector.Start(ctx); err != nil {
+		lm.logger.Warn("Failed to start the extension", zap.Error(err))
+		lm.extensionClient.InitError(ctx, fmt.Sprintf("failed to start the collector: %v", err))
+		return err
+	}
+
+	lm.wg.Wait()
+	return nil
+}
+
+func (lm *manager) processEvents(ctx context.Context) {
+	lm.wg.Add(1)
+	defer lm.wg.Done()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			lm.logger.Debug("Waiting for event...")
+			res, err := lm.extensionClient.NextEvent(ctx)
+			if err != nil {
+				lm.logger.Warn("error waiting for extension event", zap.Error(err))
+				lm.extensionClient.ExitError(ctx, fmt.Sprintf("error waiting for extension event: %v", err))
+				return
+			}
+
+			lm.logger.Debug("Received ", zap.Any("event :", res))
+			// Exit if we receive a SHUTDOWN event
+			if res.EventType == extensionapi.Shutdown {
+				lm.logger.Info("Received SHUTDOWN event")
+				lm.listener.Shutdown()
+				err = lm.collector.Stop()
+				if err != nil {
+					lm.extensionClient.ExitError(ctx, fmt.Sprintf("error stopping collector: %v", err))
+				}
+				return
+			}
+
+			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))
+			}
+		}
+	}
+}

collector/main.go

@@ -16,131 +16,29 @@ package main
 
 import (
 	"context"
-	"fmt"
 	"os"
-	"os/signal"
-	"path/filepath"
-	"sync"
-	"syscall"
 
-	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/extensionapi"
-	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/telemetryapi"
-	"github.com/open-telemetry/opentelemetry-lambda/collector/lambdacomponents"
+	"github.com/open-telemetry/opentelemetry-lambda/collector/internal/lifecycle"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
 
 var (
-	extensionName = filepath.Base(os.Args[0]) // extension name has to match the filename
+	// Version variable will be replaced at link time after `make` has been run.
+	Version = "latest"
+
+	// GitHash variable will be replaced at link time after `make` has been run.
+	GitHash = "<NOT PROPERLY GENERATED>"
 )
 
 func main() {
 	logger := initLogger()
 	logger.Info("Launching OpenTelemetry Lambda extension", zap.String("version", Version))
 
-	ctx, lm := newLifecycleManager(context.Background(), logger)
+	ctx, lm := lifecycle.NewManager(context.Background(), logger, Version)
 
 	// Will block until shutdown event is received or cancelled via the context.
-	logger.Info("done", zap.Error(lm.run(ctx)))
-}
-
-type lifecycleManager struct {
-	logger          *zap.Logger
-	collector       *Collector
-	extensionClient *extensionapi.Client
-	listener        *telemetryapi.Listener
-	wg              sync.WaitGroup
-}
-
-func newLifecycleManager(ctx context.Context, logger *zap.Logger) (context.Context, *lifecycleManager) {
-	ctx, cancel := context.WithCancel(ctx)
-
-	sigs := make(chan os.Signal, 1)
-	signal.Notify(sigs, syscall.SIGTERM, syscall.SIGINT)
-	go func() {
-		s := <-sigs
-		cancel()
-		logger.Info("received signal", zap.String("signal", s.String()))
-	}()
-
-	extensionClient := extensionapi.NewClient(logger, os.Getenv("AWS_LAMBDA_RUNTIME_API"))
-	res, err := extensionClient.Register(ctx, extensionName)
-	if err != nil {
-		logger.Fatal("Cannot register extension", zap.Error(err))
-	}
-
-	listener := telemetryapi.NewListener(logger)
-	addr, err := listener.Start()
-	if err != nil {
-		logger.Fatal("Cannot start Telemetry API Listener", zap.Error(err))
-	}
-
-	telemetryClient := telemetryapi.NewClient(logger)
-	_, err = telemetryClient.Subscribe(ctx, res.ExtensionID, addr)
-	if err != nil {
-		logger.Fatal("Cannot register Telemetry API client", zap.Error(err))
-	}
-
-	lm := &lifecycleManager{
-		logger:          logger.Named("lifecycleManager"),
-		extensionClient: extensionClient,
-		listener:        listener,
-	}
-
-	go lm.processEvents(ctx)
-
-	factories, _ := lambdacomponents.Components()
-	lm.collector = NewCollector(logger, factories)
-
-	return ctx, lm
-}
-
-func (lm *lifecycleManager) run(ctx context.Context) error {
-	if err := lm.collector.Start(ctx); err != nil {
-		lm.logger.Warn("Failed to start the extension", zap.Error(err))
-		lm.extensionClient.InitError(ctx, fmt.Sprintf("failed to start the collector: %v", err))
-		return err
-	}
-
-	lm.wg.Wait()
-	return nil
-}
-
-func (lm *lifecycleManager) processEvents(ctx context.Context) {
-	lm.wg.Add(1)
-	defer lm.wg.Done()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-			lm.logger.Debug("Waiting for event...")
-			res, err := lm.extensionClient.NextEvent(ctx)
-			if err != nil {
-				lm.logger.Warn("error waiting for extension event", zap.Error(err))
-				lm.extensionClient.ExitError(ctx, fmt.Sprintf("error waiting for extension event: %v", err))
-				return
-			}
-
-			lm.logger.Debug("Received ", zap.Any("event :", res))
-			// Exit if we receive a SHUTDOWN event
-			if res.EventType == extensionapi.Shutdown {
-				lm.logger.Info("Received SHUTDOWN event")
-				lm.listener.Shutdown()
-				err = lm.collector.Stop()
-				if err != nil {
-					lm.extensionClient.ExitError(ctx, fmt.Sprintf("error stopping collector: %v", err))
-				}
-				return
-			}
-
-			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))
-			}
-		}
-	}
+	logger.Info("done", zap.Error(lm.Run(ctx)))
 }
 
 func initLogger() *zap.Logger {