Skip to content

Commit f9d2c94

Browse files
axwflorianlcodeboten
authored
[chore] [scraperhelper] Thread context and error through scrapeFunc (#15152)
#### Description Change the scraper controller's scrapeFunc signature to accept a context and return an error. This is preparation for scraper controller extensions (#12449). NewController wraps the user-supplied scrapeFunc to apply Timeout to the incoming context when non-zero, so scrapeMetrics/scrapeLogs/ scrapeProfiles no longer build their own context, and WithScrapeContext is removed. Scraper and consumer errors are now returned. The timer path discards the returned error, preserving existing behavior. Assisted-by: Claude Opus 4.7 #### Link to tracking issue Relates to #12449 #### Testing Unit tests updated #### Documentation N/A --------- Co-authored-by: Florian Lehner <florianl@users.noreply.github.com> Co-authored-by: Alex Boten <223565+codeboten@users.noreply.github.com>
1 parent 98f555d commit f9d2c94

4 files changed

Lines changed: 157 additions & 60 deletions

File tree

‎scraper/scraperhelper/controller.go‎

Lines changed: 23 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ package scraperhelper // import "go.opentelemetry.io/collector/scraper/scraperhe
55

66
import (
77
"context"
8+
"errors"
89
"time"
910

1011
"go.opentelemetry.io/collector/component"
@@ -112,7 +113,9 @@ func NewLogsController(cfg *ControllerConfig,
112113
scrapers = append(scrapers, s)
113114
}
114115
return controller.NewController[scraper.Logs](
115-
cfg, rSet, scrapers, func(c *controller.Controller[scraper.Logs]) { scrapeLogs(c, nextConsumer) }, co.tickerCh)
116+
cfg, rSet, scrapers, func(ctx context.Context, c *controller.Controller[scraper.Logs]) error {
117+
return scrapeLogs(ctx, c, nextConsumer)
118+
}, co.tickerCh)
116119
}
117120

118121
// NewMetricsController creates a receiver.Metrics with the configured options, that can control multiple scraper.Metrics.
@@ -136,18 +139,21 @@ func NewMetricsController(cfg *ControllerConfig,
136139
scrapers = append(scrapers, s)
137140
}
138141
return controller.NewController[scraper.Metrics](
139-
cfg, rSet, scrapers, func(c *controller.Controller[scraper.Metrics]) { scrapeMetrics(c, nextConsumer) }, co.tickerCh)
142+
cfg, rSet, scrapers, func(ctx context.Context, c *controller.Controller[scraper.Metrics]) error {
143+
return scrapeMetrics(ctx, c, nextConsumer)
144+
}, co.tickerCh)
140145
}
141146

142-
func scrapeLogs(c *controller.Controller[scraper.Logs], nextConsumer consumer.Logs) {
143-
ctx, done := controller.WithScrapeContext(c.Timeout)
144-
defer done()
145-
147+
func scrapeLogs(ctx context.Context, c *controller.Controller[scraper.Logs], nextConsumer consumer.Logs) error {
148+
var errs []error
146149
logs := plog.NewLogs()
147150
for i := range c.Scrapers {
148151
md, err := c.Scrapers[i].ScrapeLogs(ctx)
149-
if err != nil && !scrapererror.IsPartialScrapeError(err) {
150-
continue
152+
if err != nil {
153+
errs = append(errs, err)
154+
if !scrapererror.IsPartialScrapeError(err) {
155+
continue
156+
}
151157
}
152158
md.ResourceLogs().MoveAndAppendTo(logs.ResourceLogs())
153159
}
@@ -156,17 +162,19 @@ func scrapeLogs(c *controller.Controller[scraper.Logs], nextConsumer consumer.Lo
156162
ctx = c.Obsrecv.StartLogsOp(ctx)
157163
err := nextConsumer.ConsumeLogs(ctx, logs)
158164
c.Obsrecv.EndLogsOp(ctx, "", logRecordCount, err)
165+
return errors.Join(append(errs, err)...)
159166
}
160167

161-
func scrapeMetrics(c *controller.Controller[scraper.Metrics], nextConsumer consumer.Metrics) {
162-
ctx, done := controller.WithScrapeContext(c.Timeout)
163-
defer done()
164-
168+
func scrapeMetrics(ctx context.Context, c *controller.Controller[scraper.Metrics], nextConsumer consumer.Metrics) error {
169+
var errs []error
165170
metrics := pmetric.NewMetrics()
166171
for i := range c.Scrapers {
167172
md, err := c.Scrapers[i].ScrapeMetrics(ctx)
168-
if err != nil && !scrapererror.IsPartialScrapeError(err) {
169-
continue
173+
if err != nil {
174+
errs = append(errs, err)
175+
if !scrapererror.IsPartialScrapeError(err) {
176+
continue
177+
}
170178
}
171179
md.ResourceMetrics().MoveAndAppendTo(metrics.ResourceMetrics())
172180
}
@@ -175,6 +183,7 @@ func scrapeMetrics(c *controller.Controller[scraper.Metrics], nextConsumer consu
175183
ctx = c.Obsrecv.StartMetricsOp(ctx)
176184
err := nextConsumer.ConsumeMetrics(ctx, metrics)
177185
c.Obsrecv.EndMetricsOp(ctx, "", dataPointCount, err)
186+
return errors.Join(append(errs, err)...)
178187
}
179188

180189
func getOptions(options []ControllerOption) controllerOptions {

‎scraper/scraperhelper/internal/controller/controller.go‎

Lines changed: 12 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ type Controller[T component.Component] struct {
2525
Timeout time.Duration
2626

2727
Scrapers []T
28-
scrapeFunc func(*Controller[T])
28+
scrapeFunc func(context.Context, *Controller[T]) error
2929
tickerCh <-chan time.Time
3030

3131
done chan struct{}
@@ -38,7 +38,7 @@ func NewController[T component.Component](
3838
cfg *ControllerConfig,
3939
rSet receiver.Settings,
4040
scrapers []T,
41-
scrapeFunc func(*Controller[T]),
41+
scrapeFunc func(context.Context, *Controller[T]) error,
4242
tickerCh <-chan time.Time,
4343
) (*Controller[T], error) {
4444
obsrecv, err := receiverhelper.NewObsReport(receiverhelper.ObsReportSettings{
@@ -61,6 +61,14 @@ func NewController[T component.Component](
6161
Obsrecv: obsrecv,
6262
}
6363

64+
if cfg.Timeout > 0 {
65+
cs.scrapeFunc = func(ctx context.Context, c *Controller[T]) error {
66+
ctx, cancel := context.WithTimeout(ctx, cfg.Timeout)
67+
defer cancel()
68+
return scrapeFunc(ctx, c)
69+
}
70+
}
71+
6472
return cs, nil
6573
}
6674

@@ -110,11 +118,11 @@ func (sc *Controller[T]) startScraping() {
110118
// Call scrape method during initialization to ensure
111119
// that scrapers start from when the component starts
112120
// instead of waiting for the full duration to start.
113-
sc.scrapeFunc(sc)
121+
_ = sc.scrapeFunc(context.Background(), sc)
114122
for {
115123
select {
116124
case <-sc.tickerCh:
117-
sc.scrapeFunc(sc)
125+
_ = sc.scrapeFunc(context.Background(), sc)
118126
case <-sc.done:
119127
return
120128
}
@@ -132,13 +140,3 @@ func GetSettings(sType component.Type, rSet receiver.Settings) scraper.Settings
132140
BuildInfo: rSet.BuildInfo,
133141
}
134142
}
135-
136-
// WithScrapeContext will return a context that has no deadline if timeout is 0
137-
// which implies no explicit timeout had occurred, otherwise, a context
138-
// with a deadline of the provided timeout is returned.
139-
func WithScrapeContext(timeout time.Duration) (context.Context, context.CancelFunc) {
140-
if timeout == 0 {
141-
return context.WithCancel(context.Background())
142-
}
143-
return context.WithTimeout(context.Background(), timeout)
144-
}

‎scraper/scraperhelper/internal/controller/controller_test.go‎

Lines changed: 110 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ func newTestController(t *testing.T, cfg *ControllerConfig, scrapers []component
3030
cfg,
3131
receivertest.NewNopSettings(receivertest.NopType),
3232
scrapers,
33-
func(*Controller[component.Component]) {},
33+
func(context.Context, *Controller[component.Component]) error { return nil },
3434
tickerCh,
3535
)
3636
require.NoError(t, err)
@@ -40,7 +40,7 @@ func newTestController(t *testing.T, cfg *ControllerConfig, scrapers []component
4040
func TestNewController(t *testing.T) {
4141
t.Parallel()
4242

43-
scrapeFunc := func(*Controller[component.Component]) {}
43+
scrapeFunc := func(context.Context, *Controller[component.Component]) error { return nil }
4444

4545
for _, tc := range []struct {
4646
name string
@@ -145,8 +145,9 @@ func TestStartScrapingWithNilTickerCh(t *testing.T) {
145145
t.Parallel()
146146

147147
var scrapeCount atomic.Int32
148-
scrapeFunc := func(*Controller[component.Component]) {
148+
scrapeFunc := func(context.Context, *Controller[component.Component]) error {
149149
scrapeCount.Add(1)
150+
return nil
150151
}
151152

152153
cfg := &ControllerConfig{
@@ -226,8 +227,9 @@ func TestStartScraping(t *testing.T) {
226227

227228
tickerCh := make(chan time.Time)
228229
var scrapeCount atomic.Int32
229-
scrapeFunc := func(*Controller[component.Component]) {
230+
scrapeFunc := func(context.Context, *Controller[component.Component]) error {
230231
scrapeCount.Add(1)
232+
return nil
231233
}
232234

233235
cfg := &ControllerConfig{
@@ -263,8 +265,9 @@ func TestStartScrapingWithInitialDelay(t *testing.T) {
263265

264266
tickerCh := make(chan time.Time)
265267
var scrapeCount atomic.Int32
266-
scrapeFunc := func(*Controller[component.Component]) {
268+
scrapeFunc := func(context.Context, *Controller[component.Component]) error {
267269
scrapeCount.Add(1)
270+
return nil
268271
}
269272

270273
cfg := &ControllerConfig{
@@ -294,8 +297,9 @@ func TestStartScrapingShutdownDuringInitialDelay(t *testing.T) {
294297
t.Parallel()
295298

296299
var scraped atomic.Bool
297-
scrapeFunc := func(*Controller[component.Component]) {
300+
scrapeFunc := func(context.Context, *Controller[component.Component]) error {
298301
scraped.Store(true)
302+
return nil
299303
}
300304

301305
cfg := &ControllerConfig{
@@ -330,25 +334,107 @@ func TestGetSettings(t *testing.T) {
330334
assert.Equal(t, rSet.BuildInfo, sSet.BuildInfo)
331335
}
332336

333-
func TestWithScrapeContext(t *testing.T) {
337+
func TestScrapeFuncAppliesTimeout(t *testing.T) {
334338
t.Parallel()
335339

336-
t.Run("zero timeout returns context without deadline", func(t *testing.T) {
337-
t.Parallel()
338-
ctx, cancel := WithScrapeContext(0)
339-
defer cancel()
340-
_, hasDeadline := ctx.Deadline()
341-
assert.False(t, hasDeadline)
342-
})
340+
timeout := 5 * time.Second
341+
var deadline time.Time
342+
var hasDeadline bool
343+
scrapeFunc := func(ctx context.Context, _ *Controller[component.Component]) error {
344+
deadline, hasDeadline = ctx.Deadline()
345+
return nil
346+
}
343347

344-
t.Run("positive timeout returns context with deadline", func(t *testing.T) {
345-
t.Parallel()
346-
timeout := 5 * time.Second
347-
ctx, cancel := WithScrapeContext(timeout)
348-
defer cancel()
349-
deadline, hasDeadline := ctx.Deadline()
350-
assert.True(t, hasDeadline)
351-
// The deadline should be approximately now + timeout.
352-
assert.WithinDuration(t, time.Now().Add(timeout), deadline, time.Second)
353-
})
348+
cfg := &ControllerConfig{
349+
CollectionInterval: time.Minute,
350+
Timeout: timeout,
351+
}
352+
ctrl, err := NewController(
353+
cfg,
354+
receivertest.NewNopSettings(receivertest.NopType),
355+
[]component.Component{},
356+
scrapeFunc,
357+
nil,
358+
)
359+
require.NoError(t, err)
360+
361+
require.NoError(t, ctrl.scrapeFunc(context.Background(), ctrl))
362+
assert.True(t, hasDeadline)
363+
assert.WithinDuration(t, time.Now().Add(timeout), deadline, time.Second)
364+
}
365+
366+
func TestScrapeFuncNoTimeout(t *testing.T) {
367+
t.Parallel()
368+
369+
var hasDeadline bool
370+
scrapeFunc := func(ctx context.Context, _ *Controller[component.Component]) error {
371+
_, hasDeadline = ctx.Deadline()
372+
return nil
373+
}
374+
375+
cfg := &ControllerConfig{
376+
CollectionInterval: time.Minute,
377+
}
378+
ctrl, err := NewController(
379+
cfg,
380+
receivertest.NewNopSettings(receivertest.NopType),
381+
[]component.Component{},
382+
scrapeFunc,
383+
nil,
384+
)
385+
require.NoError(t, err)
386+
387+
require.NoError(t, ctrl.scrapeFunc(context.Background(), ctrl))
388+
assert.False(t, hasDeadline)
389+
}
390+
391+
func TestScrapeFuncPropagatesParentCancellation(t *testing.T) {
392+
t.Parallel()
393+
394+
var gotErr error
395+
scrapeFunc := func(ctx context.Context, _ *Controller[component.Component]) error {
396+
gotErr = ctx.Err()
397+
return nil
398+
}
399+
400+
cfg := &ControllerConfig{
401+
CollectionInterval: time.Minute,
402+
Timeout: time.Hour,
403+
}
404+
ctrl, err := NewController(
405+
cfg,
406+
receivertest.NewNopSettings(receivertest.NopType),
407+
[]component.Component{},
408+
scrapeFunc,
409+
nil,
410+
)
411+
require.NoError(t, err)
412+
413+
ctx, cancel := context.WithCancel(context.Background())
414+
cancel()
415+
require.NoError(t, ctrl.scrapeFunc(ctx, ctrl))
416+
assert.ErrorIs(t, gotErr, context.Canceled)
417+
}
418+
419+
func TestScrapeFuncReturnsError(t *testing.T) {
420+
t.Parallel()
421+
422+
scrapeErr := errors.New("scrape failed")
423+
scrapeFunc := func(context.Context, *Controller[component.Component]) error {
424+
return scrapeErr
425+
}
426+
427+
cfg := &ControllerConfig{
428+
CollectionInterval: time.Minute,
429+
}
430+
ctrl, err := NewController(
431+
cfg,
432+
receivertest.NewNopSettings(receivertest.NopType),
433+
[]component.Component{},
434+
scrapeFunc,
435+
nil,
436+
)
437+
require.NoError(t, err)
438+
439+
assert.ErrorIs(t, ctrl.scrapeFunc(context.Background(), ctrl), scrapeErr)
354440
}

‎scraper/scraperhelper/xscraperhelper/controller.go‎

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ package xscraperhelper // import "go.opentelemetry.io/collector/scraper/scraperh
66

77
import (
88
"context"
9+
"errors"
910
"time"
1011

1112
"go.opentelemetry.io/collector/component"
@@ -106,7 +107,9 @@ func NewProfilesController(cfg *scraperhelper.ControllerConfig,
106107
scrapers = append(scrapers, s)
107108
}
108109
return controller.NewController[xscraper.Profiles](
109-
cfg, rSet, scrapers, func(c *controller.Controller[xscraper.Profiles]) { scrapeProfiles(c, nextConsumer) }, co.tickerCh)
110+
cfg, rSet, scrapers, func(ctx context.Context, c *controller.Controller[xscraper.Profiles]) error {
111+
return scrapeProfiles(ctx, c, nextConsumer)
112+
}, co.tickerCh)
110113
}
111114

112115
func getOptions(options []ControllerOption) controllerOptions {
@@ -117,15 +120,16 @@ func getOptions(options []ControllerOption) controllerOptions {
117120
return co
118121
}
119122

120-
func scrapeProfiles(c *controller.Controller[xscraper.Profiles], nextConsumer xconsumer.Profiles) {
121-
ctx, done := controller.WithScrapeContext(c.Timeout)
122-
defer done()
123-
123+
func scrapeProfiles(ctx context.Context, c *controller.Controller[xscraper.Profiles], nextConsumer xconsumer.Profiles) error {
124+
var errs []error
124125
profiles := pprofile.NewProfiles()
125126
for i := range c.Scrapers {
126127
md, err := c.Scrapers[i].ScrapeProfiles(ctx)
127-
if err != nil && !scrapererror.IsPartialScrapeError(err) {
128-
continue
128+
if err != nil {
129+
errs = append(errs, err)
130+
if !scrapererror.IsPartialScrapeError(err) {
131+
continue
132+
}
129133
}
130134
if mergeErr := md.MergeTo(profiles); mergeErr != nil {
131135
continue
@@ -134,5 +138,5 @@ func scrapeProfiles(c *controller.Controller[xscraper.Profiles], nextConsumer xc
134138

135139
// TODO: Add proper receiver observability for profiles when receiverhelper supports it
136140
// For now, we skip the obs report and just consume the profiles directly
137-
_ = nextConsumer.ConsumeProfiles(ctx, profiles)
141+
return errors.Join(append(errs, nextConsumer.ConsumeProfiles(ctx, profiles))...)
138142
}

0 commit comments

Comments
 (0)