prometheus: scrapers are not stopped when receiver is shutdown (#3450)

During receiver shutdown discovery is stopped. However, any previously
discovered targets will continue to be scraped. This calls `Stop` on the scrape
manager to stop the scraping on shutdown as well.

**Issue:** https://github.com/signalfx/splunk-otel-collector/issues/471

**Testing:** 
- added check to Prometheus end to end test that scraping is actually stopped
- tested end to end with receivercreator,  saw that scraping actually stopped after pod was deleted

Author: jrcamp

receiver/prometheusreceiver/internal/metadata.go

@@ -16,6 +16,7 @@ package internal
 
 import (
 	"errors"
+	"sync"
 
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
@@ -33,10 +34,27 @@ type ScrapeManager interface {
 }
 
 type metadataService struct {
-	sm ScrapeManager
+	sync.Mutex
+	stopped bool
+	sm      ScrapeManager
+}
+
+func (s *metadataService) Close() {
+	s.Lock()
+	s.stopped = true
+	s.Unlock()
 }
 
 func (s *metadataService) Get(job, instance string) (MetadataCache, error) {
+	s.Lock()
+	defer s.Unlock()
+
+	// If we're already stopped return early so that we don't call scrapeManager.TargetsAll()
+	// which will result in deadlock if scrapeManager is being stopped.
+	if s.stopped {
+		return nil, errAlreadyStopped
+	}
+
 	targetGroup, ok := s.sm.TargetsAll()[job]
 	if !ok {
 		return nil, errors.New("unable to find a target group with job=" + job)

receiver/prometheusreceiver/internal/ocastore.go

@@ -106,9 +106,11 @@ func (o *OcaStore) Appender(context.Context) storage.Appender {
 	return noop
 }
 
-func (o *OcaStore) Close() error {
-	atomic.CompareAndSwapInt32(&o.running, runningStateReady, runningStateStop)
-	return nil
+// Close OcaStore as well as the internal metadataService.
+func (o *OcaStore) Close() {
+	if atomic.CompareAndSwapInt32(&o.running, runningStateReady, runningStateStop) {
+		o.mc.Close()
+	}
 }
 
 // noopAppender, always return error on any operations

receiver/prometheusreceiver/internal/ocastore_test.go

@@ -33,7 +33,7 @@ func TestOcaStore(t *testing.T) {
 	app := o.Appender(context.Background())
 	require.NotNil(t, app, "Expecting app")
 
-	_ = o.Close()
+	o.Close()
 
 	app = o.Appender(context.Background())
 	assert.Equal(t, noop, app)

receiver/prometheusreceiver/metrics_receiver.go

@@ -36,7 +36,9 @@ type pReceiver struct {
 	consumer   consumer.Metrics
 	cancelFunc context.CancelFunc
 
-	logger *zap.Logger
+	logger        *zap.Logger
+	scrapeManager *scrape.Manager
+	ocaStore      *internal.OcaStore
 }
 
 // New creates a new prometheus.Receiver reference.
@@ -79,7 +81,7 @@ func (r *pReceiver) Start(_ context.Context, host component.Host) error {
 	// Per component.Component Start instructions, for async operations we should not use the
 	// incoming context, it may get cancelled.
 	receiverCtx := obsreport.ReceiverContext(context.Background(), r.cfg.ID(), transport)
-	ocaStore := internal.NewOcaStore(
+	r.ocaStore = internal.NewOcaStore(
 		receiverCtx,
 		r.consumer,
 		r.logger,
@@ -89,13 +91,13 @@ func (r *pReceiver) Start(_ context.Context, host component.Host) error {
 		r.cfg.ID(),
 		r.cfg.PrometheusConfig.GlobalConfig.ExternalLabels,
 	)
-	scrapeManager := scrape.NewManager(logger, ocaStore)
-	ocaStore.SetScrapeManager(scrapeManager)
-	if err := scrapeManager.ApplyConfig(r.cfg.PrometheusConfig); err != nil {
+	r.scrapeManager = scrape.NewManager(logger, r.ocaStore)
+	r.ocaStore.SetScrapeManager(r.scrapeManager)
+	if err := r.scrapeManager.ApplyConfig(r.cfg.PrometheusConfig); err != nil {
 		return err
 	}
 	go func() {
-		if err := scrapeManager.Run(discoveryManager.SyncCh()); err != nil {
+		if err := r.scrapeManager.Run(discoveryManager.SyncCh()); err != nil {
 			r.logger.Error("Scrape manager failed", zap.Error(err))
 			host.ReportFatalError(err)
 		}
@@ -106,5 +108,11 @@ func (r *pReceiver) Start(_ context.Context, host component.Host) error {
 // Shutdown stops and cancels the underlying Prometheus scrapers.
 func (r *pReceiver) Shutdown(context.Context) error {
 	r.cancelFunc()
+	// ocaStore (and internally metadataService) needs to stop first to prevent deadlocks.
+	// When stopping scrapeManager it waits for all scrapes to terminate. However during
+	// scraping metadataService calls scrapeManager.AllTargets() which acquires
+	// the same lock that's acquired when scrapeManager is stopped.
+	r.ocaStore.Close()
+	r.scrapeManager.Stop()
 	return nil
 }

receiver/prometheusreceiver/metrics_receiver_test.go

@@ -33,6 +33,7 @@ import (
 	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
 	gokitlog "github.com/go-kit/kit/log"
 	promcfg "github.com/prometheus/prometheus/config"
+	"github.com/prometheus/prometheus/scrape"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
@@ -1403,7 +1404,12 @@ func testEndToEnd(t *testing.T, targets []*testData, useStartTimeMetric bool) {
 		UseStartTimeMetric: useStartTimeMetric}, cms)
 
 	require.NoError(t, rcvr.Start(context.Background(), componenttest.NewNopHost()), "Failed to invoke Start: %v", err)
-	t.Cleanup(func() { require.NoError(t, rcvr.Shutdown(context.Background())) })
+	t.Cleanup(func() {
+		// verify state after shutdown is called
+		assert.Lenf(t, flattenTargets(rcvr.scrapeManager.TargetsAll()), len(targets), "expected %v targets to be running", len(targets))
+		require.NoError(t, rcvr.Shutdown(context.Background()))
+		assert.Len(t, flattenTargets(rcvr.scrapeManager.TargetsAll()), 0, "expected scrape manager to have no targets")
+	})
 
 	// wait for all provided data to be scraped
 	mp.wg.Wait()
@@ -1436,6 +1442,15 @@ func testEndToEnd(t *testing.T, targets []*testData, useStartTimeMetric bool) {
 	}
 }
 
+// flattenTargets takes a map of jobs to target and flattens to a list of targets
+func flattenTargets(targets map[string][]*scrape.Target) []*scrape.Target {
+	var flatTargets []*scrape.Target
+	for _, target := range targets {
+		flatTargets = append(flatTargets, target...)
+	}
+	return flatTargets
+}
+
 var startTimeMetricRegexPage = `
 # HELP go_threads Number of OS threads created
 # TYPE go_threads gauge