Update OTel dependencies

Author: dhurley

go.mod

@@ -69,6 +69,7 @@ require (
 	go.opentelemetry.io/collector/receiver v1.30.0
 	go.opentelemetry.io/collector/receiver/otlpreceiver v0.124.0
 	go.opentelemetry.io/collector/receiver/receivertest v0.124.0
+	go.opentelemetry.io/collector/scraper v0.124.0
 	go.opentelemetry.io/collector/scraper/scraperhelper v0.124.0
 	go.opentelemetry.io/otel v1.35.0
 	go.uber.org/goleak v1.3.0
@@ -262,7 +263,6 @@ require (
 	go.opentelemetry.io/collector/processor/xprocessor v0.124.0 // indirect
 	go.opentelemetry.io/collector/receiver/receiverhelper v0.124.0 // indirect
 	go.opentelemetry.io/collector/receiver/xreceiver v0.124.0 // indirect
-	go.opentelemetry.io/collector/scraper v0.124.0 // indirect
 	go.opentelemetry.io/collector/semconv v0.124.0 // indirect
 	go.opentelemetry.io/collector/service v0.124.0 // indirect
 	go.opentelemetry.io/collector/service/hostcapabilities v0.124.0 // indirect

internal/collector/nginxossreceiver/factory.go

@@ -9,30 +9,29 @@ import (
 	"context"
 	"errors"
 
+	"go.opentelemetry.io/collector/scraper"
+
 	"github.com/nginx/agent/v3/internal/collector/nginxossreceiver/internal/config"
 	"github.com/nginx/agent/v3/internal/collector/nginxossreceiver/internal/metadata"
 	"github.com/nginx/agent/v3/internal/collector/nginxossreceiver/internal/scraper/accesslog"
 	"github.com/nginx/agent/v3/internal/collector/nginxossreceiver/internal/scraper/stubstatus"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/receiver"
+	"go.opentelemetry.io/collector/scraper/scraperhelper"
 )
 
 // nolint: ireturn
 func NewFactory() receiver.Factory {
-	stubStatusOption := receiver.WithMetrics(createStubStatusReceiver, metadata.MetricsStability)
-	accessLogOptions := receiver.WithMetrics(createAccessLogReceiver, metadata.MetricsStability)
-
 	return receiver.NewFactory(
 		metadata.Type,
 		config.CreateDefaultConfig,
-		stubStatusOption,
-		accessLogOptions,
+		receiver.WithMetrics(createMetrics, metadata.MetricsStability),
 	)
 }
 
 // nolint: ireturn
-func createStubStatusReceiver(
+func createMetrics(
 	_ context.Context,
 	params receiver.Settings,
 	rConf component.Config,
@@ -43,20 +42,39 @@ func createStubStatusReceiver(
 		return nil, errors.New("cast to metrics receiver config")
 	}
 
-	return stubstatus.NewScraper(params, cfg), nil
-}
+	var controllers []scraperhelper.ControllerOption
 
-// nolint: ireturn
-func createAccessLogReceiver(
-	_ context.Context,
-	params receiver.Settings,
-	rConf component.Config,
-	cons consumer.Metrics,
-) (receiver.Metrics, error) {
-	cfg, ok := rConf.(*config.Config)
-	if !ok {
-		return nil, errors.New("cast to metrics receiver config")
+	stubStatusScraper := stubstatus.NewScraper(params, cfg)
+	stubStatusMetrics, stubStatusMetricsError := scraper.NewMetrics(
+		stubStatusScraper.Scrape,
+		scraper.WithStart(stubStatusScraper.Start),
+		scraper.WithShutdown(stubStatusScraper.Shutdown),
+	)
+	if stubStatusMetricsError != nil {
+		return nil, stubStatusMetricsError
 	}
 
-	return accesslog.NewScraper(params, cfg)
+	controllers = append(controllers, scraperhelper.AddScraper(metadata.Type, stubStatusMetrics))
+
+	if len(cfg.AccessLogs) > 0 {
+		accessLogScraper := accesslog.NewScraper(params, cfg)
+
+		accessLogMetrics, accessLogMetricsError := scraper.NewMetrics(
+			accessLogScraper.Scrape,
+			scraper.WithStart(accessLogScraper.Start),
+			scraper.WithShutdown(accessLogScraper.Shutdown),
+		)
+		if accessLogMetricsError != nil {
+			return nil, accessLogMetricsError
+		}
+
+		controllers = append(controllers, scraperhelper.AddScraper(metadata.Type, accessLogMetrics))
+	}
+
+	return scraperhelper.NewMetricsController(
+		&cfg.ControllerConfig,
+		params,
+		cons,
+		controllers...,
+	)
 }

internal/collector/nginxossreceiver/factory_test.go

@@ -10,8 +10,6 @@ import (
 	"testing"
 	"time"
 
-	"go.opentelemetry.io/collector/component"
-
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
@@ -55,25 +53,3 @@ func TestCreateMetricsReceiver(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, metricsReceiver)
 }
-
-func TestCreateStubStatusReceiver(t *testing.T) {
-	metrics, err := createStubStatusReceiver(
-		context.Background(),
-		receivertest.NewNopSettings(component.Type{}),
-		NewFactory().CreateDefaultConfig(),
-		consumertest.NewNop(),
-	)
-	require.NoError(t, err)
-	require.NotNil(t, metrics)
-}
-
-func TestCreateAccessLogReceiver(t *testing.T) {
-	metrics, err := createAccessLogReceiver(
-		context.Background(),
-		receivertest.NewNopSettings(component.Type{}),
-		NewFactory().CreateDefaultConfig(),
-		consumertest.NewNop(),
-	)
-	require.NoError(t, err)
-	require.NotNil(t, metrics)
-}

internal/collector/nginxossreceiver/internal/scraper/accesslog/nginx_log_scraper.go

@@ -36,17 +36,18 @@ const Percentage = 100
 
 type (
 	NginxLogScraper struct {
-		outChan  <-chan []*entry.Entry
-		cfg      *config.Config
-		settings receiver.Settings
-		logger   *zap.Logger
-		mb       *metadata.MetricsBuilder
-		rb       *metadata.ResourceBuilder
-		pipe     *pipeline.DirectedPipeline
-		wg       *sync.WaitGroup
-		cancel   context.CancelFunc
-		entries  []*entry.Entry
-		mut      sync.Mutex
+		outChan   <-chan []*entry.Entry
+		cfg       *config.Config
+		settings  receiver.Settings
+		logger    *zap.Logger
+		mb        *metadata.MetricsBuilder
+		rb        *metadata.ResourceBuilder
+		pipe      *pipeline.DirectedPipeline
+		wg        *sync.WaitGroup
+		cancel    context.CancelFunc
+		entries   []*entry.Entry
+		operators []operator.Config
+		mut       sync.Mutex
 	}
 
 	NginxMetrics struct {
@@ -65,7 +66,7 @@ type (
 func NewScraper(
 	settings receiver.Settings,
 	cfg *config.Config,
-) (*NginxLogScraper, error) {
+) *NginxLogScraper {
 	logger := settings.Logger
 	logger.Info("Creating NGINX access log scraper")
 
@@ -85,24 +86,17 @@ func NewScraper(
 	}
 
 	nls := &NginxLogScraper{
-		cfg:      cfg,
-		logger:   logger,
-		settings: settings,
-		mb:       mb,
-		rb:       rb,
-		mut:      sync.Mutex{},
-		wg:       &sync.WaitGroup{},
+		cfg:       cfg,
+		logger:    logger,
+		settings:  settings,
+		mb:        mb,
+		rb:        rb,
+		mut:       sync.Mutex{},
+		wg:        &sync.WaitGroup{},
+		operators: operators,
 	}
 
-	if len(operators) > 0 {
-		var err error
-		nls.pipe, err = nls.initStanzaPipeline(operators, settings.Logger)
-		if err != nil {
-			return nil, fmt.Errorf("init stanza pipeline: %w", err)
-		}
-	}
-
-	return nls, nil
+	return nls
 }
 
 func (nls *NginxLogScraper) ID() component.ID {
@@ -114,16 +108,20 @@ func (nls *NginxLogScraper) Start(parentCtx context.Context, _ component.Host) e
 	ctx, cancel := context.WithCancel(parentCtx)
 	nls.cancel = cancel
 
-	if nls.pipe != nil {
-		err := nls.pipe.Start(storage.NewNopClient())
-		if err != nil {
-			return fmt.Errorf("stanza pipeline start: %w", err)
-		}
+	var err error
+	nls.pipe, err = nls.initStanzaPipeline(nls.operators, nls.logger)
+	if err != nil {
+		return fmt.Errorf("init stanza pipeline: %w", err)
+	}
 
-		nls.wg.Add(1)
-		go nls.runConsumer(ctx)
+	startError := nls.pipe.Start(storage.NewNopClient())
+	if startError != nil {
+		return fmt.Errorf("stanza pipeline start: %w", startError)
 	}
 
+	nls.wg.Add(1)
+	go nls.runConsumer(ctx)
+
 	return nil
 }
 
@@ -208,11 +206,7 @@ func (nls *NginxLogScraper) Shutdown(_ context.Context) error {
 	}
 	nls.wg.Wait()
 
-	if nls.pipe != nil {
-		return nls.pipe.Stop()
-	}
-
-	return nil
+	return nls.pipe.Stop()
 }
 
 func (nls *NginxLogScraper) initStanzaPipeline(

internal/collector/nginxossreceiver/internal/scraper/accesslog/nginx_log_scraper_test.go

@@ -55,14 +55,13 @@ func TestAccessLogScraper(t *testing.T) {
 		},
 	}
 
-	accessLogScraper, err := NewScraper(receivertest.NewNopSettings(component.Type{}), cfg)
-	require.NoError(t, err)
+	accessLogScraper := NewScraper(receivertest.NewNopSettings(component.Type{}), cfg)
 	defer func() {
 		shutdownError := accessLogScraper.Shutdown(ctx)
 		require.NoError(t, shutdownError)
 	}()
 
-	err = accessLogScraper.Start(context.Background(), componenttest.NewNopHost())
+	err := accessLogScraper.Start(context.Background(), componenttest.NewNopHost())
 	require.NoError(t, err)
 
 	go simulateLogging(t, testDataFilePath, testAccessLogPath, 250*time.Millisecond)
@@ -83,6 +82,15 @@ func TestAccessLogScraper(t *testing.T) {
 		pmetrictest.IgnoreResourceAttributeValue("instance.id")))
 }
 
+func TestAccessLogScraperError(t *testing.T) {
+	t.Run("include config missing", func(tt *testing.T) {
+		logScraper := NewScraper(receivertest.NewNopSettings(component.Type{}), &config.Config{})
+		err := logScraper.Start(context.Background(), componenttest.NewNopHost())
+		require.Error(tt, err)
+		assert.Contains(tt, err.Error(), "init stanza pipeline")
+	})
+}
+
 // Copies the contents of one file to another with the given delay. Used to simulate writing log entries to a log file.
 // Reason for nolint: we must use testify's assert instead of require,
 // for more info see https://github.com/stretchr/testify/issues/772#issuecomment-945166599