Function testScrapeLoopRun

scrape/scrape_test.go:1145–1219

(t *testing.T, appV2 bool)

}

func testScrapeLoopRun(t *testing.T, appV2 bool) {
	t.Parallel()
	var (
		signal = make(chan struct{}, 1)
		errc   = make(chan error)
	)

	ctx, cancel := context.WithCancel(t.Context())
	sl, scraper := newTestScrapeLoop(t, withCtx(ctx), withAppendable(teststorage.NewAppendable(), appV2))
	// The loop must terminate during the initial offset if the context
	// is canceled.
	scraper.offsetDur = time.Hour

	go func() {
		sl.run(errc)
		signal <- struct{}{}
	}()

	// Wait to make sure we are actually waiting on the offset.
	time.Sleep(1 * time.Second)

	cancel()
	select {
	case <-signal:
	case <-time.After(5 * time.Second):
		require.FailNow(t, "Cancellation during initial offset failed.")
	case err := <-errc:
		require.FailNow(t, "Unexpected error", "err: %s", err)
	}

	ctx, cancel = context.WithCancel(t.Context())
	sl, scraper = newTestScrapeLoop(t, withAppendable(teststorage.NewAppendable(), appV2), func(sl *scrapeLoop) {
		sl.ctx = ctx
		sl.timeout = 100 * time.Millisecond
	})
	// The provided timeout must cause cancellation of the context passed down to the
	// scraper. The scraper has to respect the context.
	scraper.offsetDur = 0

	blockCtx, blockCancel := context.WithCancel(t.Context())
	scraper.scrapeFunc = func(ctx context.Context, _ io.Writer) error {
		select {
		case <-blockCtx.Done():
			cancel()
		case <-ctx.Done():
			return ctx.Err()
		}
		return nil
	}

	go func() {
		sl.run(errc)
		signal <- struct{}{}
	}()

	select {
	case err := <-errc:
		require.ErrorIs(t, err, context.DeadlineExceeded)