[scraperhelper] Add extension-based scraper controller support

Implement the extensionscrapercontroller package and integrate it with
the scraper controller, allowing extensions to trigger scrapes based on
external events instead of (or in addition to) a fixed timer interval.

- Add extensionscrapercontroller.ControllerExtension interface and
  DeregisterFunc helper in extension/xextension/extensionscrapercontroller
- Add Controllers field to ControllerConfig for specifying extension IDs
- Allow CollectionInterval=0 when Controllers are configured
- Register scrapers with controller extensions on Start, deregister on
  Shutdown
- Update RFC to remove scraperID parameter from RegisterScraper

Assisted-by: Claude Opus 4.5

Author: axw

.chloggen/extensionscrapercontroller.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/otlp)
+component: pkg/scraperhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add ControllerExtension interface for extensible scraper controllers
+
+# One or more tracking issues or pull requests related to the change
+issues: [12449]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

cmd/mdatagen/go.mod

@@ -88,8 +88,10 @@ require (
 	go.opentelemetry.io/collector/config/configtls v1.56.0 // indirect
 	go.opentelemetry.io/collector/confmap/xconfmap v0.150.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.150.0 // indirect
+	go.opentelemetry.io/collector/extension v1.56.0 // indirect
 	go.opentelemetry.io/collector/extension/extensionauth v1.56.0 // indirect
 	go.opentelemetry.io/collector/extension/extensionmiddleware v0.150.0 // indirect
+	go.opentelemetry.io/collector/extension/xextension v0.150.0 // indirect
 	go.opentelemetry.io/collector/internal/componentalias v0.150.0 // indirect
 	go.opentelemetry.io/collector/internal/fanoutconsumer v0.150.0 // indirect
 	go.opentelemetry.io/collector/pdata/testdata v0.150.0 // indirect

docs/rfcs/scraper-controller-extension.md

@@ -40,12 +40,19 @@ type ControllerExtension interface {
     // RegisterScraper registers a scraper with this controller extension.
     // The extension will invoke the provided scrape function according to its implementation.
     // Returns a registration handle that can be used to deregister the scraper.
-    RegisterScraper(ctx context.Context, scraperID component.ID, scrapeFunc func(context.Context) error) (RegistrationHandle, error)
+    //
+    // Implementations may call scrapeFunc concurrently. After Deregister on the
+    // returned handle returns, implementations must not start new invocations of
+    // scrapeFunc; already in-flight invocations may continue to run.
+    RegisterScraper(ctx context.Context, scrapeFunc func(context.Context) error) (RegistrationHandle, error)
 }
 
 // RegistrationHandle provides a way to deregister a scraper from a controller extension
 type RegistrationHandle interface {
-    // Deregister removes the scraper from the controller extension
+    // Deregister removes the scraper from the controller extension.
+    // After Deregister returns, the extension must not start any new
+    // invocations of the associated scrapeFunc. Deregister need not wait
+    // for in-flight invocations to complete.
     Deregister(ctx context.Context) error
 }
 ```
@@ -76,8 +83,8 @@ Modify `ControllerConfig` to support:
 ```golang
 type ControllerConfig struct {
     // CollectionInterval sets how frequently the scraper should be called.
-    // If zero or negative, the timer-based scraping is disabled.
-    // At least one controller extension must be configured if timer is disabled.
+    // Must be positive, or zero to disable timer-based scraping. If zero,
+    // at least one controller extension must be configured.
     CollectionInterval time.Duration `mapstructure:"collection_interval"`
 
     // InitialDelay sets the initial start delay for the scraper timer.
@@ -96,7 +103,7 @@ type ControllerConfig struct {
 ### 3. Controller Implementation Changes
 
 Modify `controller.Controller` to:
-- Support disabling the timer when `CollectionInterval <= 0`
+- Support disabling the timer when `CollectionInterval == 0` (negative values remain invalid)
 - Register scrapers with configured controller extensions during `Start()`
 - Deregister scrapers during `Shutdown()`
 - Validate that at least one controller extension is configured if timer is disabled

extension/xextension/extensionscrapercontroller/doc.go

@@ -0,0 +1,6 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+// Package extensionscrapercontroller defines the interface for extensions
+// that control when scraper-based receivers invoke their scrapers.
+package extensionscrapercontroller // import "go.opentelemetry.io/collector/extension/xextension/extensionscrapercontroller"

extension/xextension/extensionscrapercontroller/extensionscrapercontroller.go

@@ -0,0 +1,51 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package extensionscrapercontroller // import "go.opentelemetry.io/collector/extension/xextension/extensionscrapercontroller"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/extension"
+)
+
+// ControllerExtension is an extension that controls when scraper-based
+// receivers invoke their scrapers.
+type ControllerExtension interface {
+	extension.Extension
+
+	// RegisterScraper registers a scraper with the extension. The extension
+	// will call scrapeFunc when it determines a scrape should occur.
+	// The returned RegistrationHandle must be used to deregister the scraper
+	// during shutdown.
+	//
+	// Implementations may call scrapeFunc concurrently. After Deregister on
+	// the returned handle returns, implementations must not start new
+	// invocations of scrapeFunc; already in-flight invocations may continue
+	// to run.
+	RegisterScraper(ctx context.Context, scrapeFunc func(context.Context) error) (RegistrationHandle, error)
+}
+
+// RegistrationHandle is returned by ControllerExtension.RegisterScraper and
+// is used to deregister the scraper during shutdown.
+type RegistrationHandle interface {
+	// Deregister removes the scraper registration from the extension.
+	// After Deregister returns, the extension must not start any new
+	// invocations of the associated scrapeFunc. Deregister need not wait
+	// for in-flight invocations to complete.
+	Deregister(ctx context.Context) error
+}
+
+// DeregisterFunc is a function that implements RegistrationHandle.
+// A nil DeregisterFunc is valid and returns nil on Deregister.
+type DeregisterFunc func(ctx context.Context) error
+
+var _ RegistrationHandle = DeregisterFunc(nil)
+
+// Deregister calls the underlying function. If the receiver is nil, it returns nil.
+func (f DeregisterFunc) Deregister(ctx context.Context) error {
+	if f == nil {
+		return nil
+	}
+	return f(ctx)
+}

extension/xextension/extensionscrapercontroller/extensionscrapercontroller_test.go

@@ -0,0 +1,30 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package extensionscrapercontroller
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDeregisterFuncNil(t *testing.T) {
+	var f DeregisterFunc
+	assert.NoError(t, f.Deregister(context.Background()))
+}
+
+func TestDeregisterFuncDelegates(t *testing.T) {
+	called := false
+	expectedErr := errors.New("deregister error")
+	f := DeregisterFunc(func(context.Context) error {
+		called = true
+		return expectedErr
+	})
+	err := f.Deregister(context.Background())
+	require.True(t, called)
+	assert.Equal(t, expectedErr, err)
+}

scraper/scraperhelper/controller_test.go

@@ -22,6 +22,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/extension/xextension/extensionscrapercontroller"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/receiver"
@@ -832,3 +833,214 @@ func TestNewMetricsController_ScraperIDInErrorLogs(t *testing.T) {
 	assert.Equal(t, "test log from receiver", receiverLog.Message)
 	assert.NotContains(t, receiverLog.ContextMap(), "scraper")
 }
+
+// mockHost implements component.Host with configurable extensions.
+type mockHost struct {
+	ext map[component.ID]component.Component
+}
+
+func (h *mockHost) GetExtensions() map[component.ID]component.Component {
+	return h.ext
+}
+
+// mockControllerExtension implements extensionscrapercontroller.ControllerExtension.
+type mockControllerExtension struct {
+	component.StartFunc
+	component.ShutdownFunc
+	scrapeFunc   func(context.Context) error
+	deregistered bool
+}
+
+func (m *mockControllerExtension) RegisterScraper(_ context.Context, scrapeFunc func(context.Context) error) (extensionscrapercontroller.RegistrationHandle, error) {
+	m.scrapeFunc = scrapeFunc
+	return extensionscrapercontroller.DeregisterFunc(func(context.Context) error {
+		m.deregistered = true
+		return nil
+	}), nil
+}
+
+func TestExtensionTriggersMetricsScrape(t *testing.T) {
+	t.Parallel()
+
+	scrapeCh := make(chan int, 10)
+	ts := &testScrape{ch: scrapeCh}
+
+	scp, err := scraper.NewMetrics(ts.scrapeMetrics)
+	require.NoError(t, err)
+
+	extID := component.MustNewID("myext")
+	mockExt := &mockControllerExtension{}
+
+	cfg := &ControllerConfig{
+		CollectionInterval: 0,
+		InitialDelay:       0,
+		Controllers:        []component.ID{extID},
+	}
+
+	recv, err := NewMetricsController(
+		cfg,
+		receivertest.NewNopSettings(receivertest.NopType),
+		new(consumertest.MetricsSink),
+		AddMetricsScraper(component.MustNewType("scraper"), scp),
+	)
+	require.NoError(t, err)
+
+	host := &mockHost{ext: map[component.ID]component.Component{extID: mockExt}}
+	require.NoError(t, recv.Start(context.Background(), host))
+
+	// Extension triggers a scrape
+	require.NotNil(t, mockExt.scrapeFunc, "scrapeFunc should have been registered")
+	require.NoError(t, mockExt.scrapeFunc(context.Background()))
+
+	// Verify scrape was called
+	select {
+	case <-scrapeCh:
+	case <-time.After(time.Second):
+		t.Fatal("expected scrape to be triggered by extension")
+	}
+
+	require.NoError(t, recv.Shutdown(context.Background()))
+	assert.True(t, mockExt.deregistered, "expected deregister to be called on shutdown")
+}
+
+func TestExtensionAndTimerBothTriggerScrapes(t *testing.T) {
+	t.Parallel()
+
+	scrapeCh := make(chan int, 10)
+	ts := &testScrape{ch: scrapeCh}
+
+	scp, err := scraper.NewMetrics(ts.scrapeMetrics)
+	require.NoError(t, err)
+
+	extID := component.MustNewID("myext")
+	mockExt := &mockControllerExtension{}
+
+	tickerCh := make(chan time.Time)
+
+	cfg := &ControllerConfig{
+		CollectionInterval: time.Second,
+		InitialDelay:       0,
+		Controllers:        []component.ID{extID},
+	}
+
+	recv, err := NewMetricsController(
+		cfg,
+		receivertest.NewNopSettings(receivertest.NopType),
+		new(consumertest.MetricsSink),
+		AddMetricsScraper(component.MustNewType("scraper"), scp),
+		WithTickerChannel(tickerCh),
+	)
+	require.NoError(t, err)
+
+	host := &mockHost{ext: map[component.ID]component.Component{extID: mockExt}}
+	require.NoError(t, recv.Start(context.Background(), host))
+
+	// Initial scrape on start (from ticker goroutine)
+	<-scrapeCh
+
+	// Extension triggers a scrape
+	require.NoError(t, mockExt.scrapeFunc(context.Background()))
+	<-scrapeCh
+
+	// Ticker triggers a scrape
+	tickerCh <- time.Now()
+	<-scrapeCh
+
+	require.NoError(t, recv.Shutdown(context.Background()))
+	assert.True(t, mockExt.deregistered)
+}
+
+func TestExtensionNotFound(t *testing.T) {
+	t.Parallel()
+
+	scp, err := scraper.NewMetrics(func(context.Context) (pmetric.Metrics, error) {
+		return pmetric.NewMetrics(), nil
+	})
+	require.NoError(t, err)
+
+	cfg := &ControllerConfig{
+		CollectionInterval: 0,
+		InitialDelay:       0,
+		Controllers:        []component.ID{component.MustNewID("missing")},
+	}
+
+	recv, err := NewMetricsController(
+		cfg,
+		receivertest.NewNopSettings(receivertest.NopType),
+		new(consumertest.MetricsSink),
+		AddMetricsScraper(component.MustNewType("scraper"), scp),
+	)
+	require.NoError(t, err)
+
+	host := &mockHost{ext: map[component.ID]component.Component{}}
+	err = recv.Start(context.Background(), host)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), `extension "missing" not found`)
+}
+
+func TestExtensionWrongType(t *testing.T) {
+	t.Parallel()
+
+	scp, err := scraper.NewMetrics(func(context.Context) (pmetric.Metrics, error) {
+		return pmetric.NewMetrics(), nil
+	})
+	require.NoError(t, err)
+
+	extID := component.MustNewID("wrongtype")
+
+	cfg := &ControllerConfig{
+		CollectionInterval: 0,
+		InitialDelay:       0,
+		Controllers:        []component.ID{extID},
+	}
+
+	recv, err := NewMetricsController(
+		cfg,
+		receivertest.NewNopSettings(receivertest.NopType),
+		new(consumertest.MetricsSink),
+		AddMetricsScraper(component.MustNewType("scraper"), scp),
+	)
+	require.NoError(t, err)
+
+	// Use a plain component that does not implement ControllerExtension
+	host := &mockHost{ext: map[component.ID]component.Component{extID: &struct {
+		component.StartFunc
+		component.ShutdownFunc
+	}{}}}
+	err = recv.Start(context.Background(), host)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), `extension "wrongtype" is not a scraper controller extension`)
+}
+
+func TestDeregisterOnShutdown(t *testing.T) {
+	t.Parallel()
+
+	scp, err := scraper.NewMetrics(func(context.Context) (pmetric.Metrics, error) {
+		return pmetric.NewMetrics(), nil
+	})
+	require.NoError(t, err)
+
+	extID := component.MustNewID("myext")
+	mockExt := &mockControllerExtension{}
+
+	cfg := &ControllerConfig{
+		CollectionInterval: 0,
+		InitialDelay:       0,
+		Controllers:        []component.ID{extID},
+	}
+
+	recv, err := NewMetricsController(
+		cfg,
+		receivertest.NewNopSettings(receivertest.NopType),
+		new(consumertest.MetricsSink),
+		AddMetricsScraper(component.MustNewType("scraper"), scp),
+	)
+	require.NoError(t, err)
+
+	host := &mockHost{ext: map[component.ID]component.Component{extID: mockExt}}
+	require.NoError(t, recv.Start(context.Background(), host))
+	assert.False(t, mockExt.deregistered)
+
+	require.NoError(t, recv.Shutdown(context.Background()))
+	assert.True(t, mockExt.deregistered)
+}