[wip] [feedback wanted] Do not scrape pods when activator in path

pkg/autoscaler/metrics/collector.go

@@ -82,6 +82,12 @@ type MetricClient interface {
 	// StableAndPanicRPS returns both the stable and the panic RPS
 	// for the given replica as of the given time.
 	StableAndPanicRPS(key types.NamespacedName, now time.Time) (float64, float64, error)
+
+	// Pause metric collection
+	Pause(key types.NamespacedName)
+
+	// Resume metric collection
+	Resume(key types.NamespacedName)
 }
 
 // MetricCollector manages collection of metrics for many entities.
@@ -166,6 +172,24 @@ func (c *MetricCollector) Record(key types.NamespacedName, now time.Time, stat S
 	}
 }
 
+func (c *MetricCollector) Pause(key types.NamespacedName) {
+	c.collectionsMutex.RLock()
+	defer c.collectionsMutex.RUnlock()
+
+	if collection, exists := c.collections[key]; exists {
+		collection.setPause(true)
+	}
+}
+
+func (c *MetricCollector) Resume(key types.NamespacedName) {
+	c.collectionsMutex.RLock()
+	defer c.collectionsMutex.RUnlock()
+
+	if collection, exists := c.collections[key]; exists {
+		collection.setPause(false)
+	}
+}
+
 // Watch registers a singleton function to call when collector status changes.
 func (c *MetricCollector) Watch(fn func(types.NamespacedName)) {
 	c.watcherMutex.Lock()
@@ -251,6 +275,7 @@ type (
 		scraper StatsScraper
 		lastErr error
 		grp     sync.WaitGroup
+		paused  bool
 		stopCh  chan struct{}
 	}
 )
@@ -297,6 +322,7 @@ func newCollection(metric *autoscalingv1alpha1.Metric, scraper StatsScraper, clo
 		scraper: scraper,
 
 		stopCh: make(chan struct{}),
+		paused: false,
 	}
 
 	key := types.NamespacedName{Namespace: metric.Namespace, Name: metric.Name}
@@ -313,6 +339,9 @@ func newCollection(metric *autoscalingv1alpha1.Metric, scraper StatsScraper, clo
 			case <-c.stopCh:
 				return
 			case <-scrapeTicker.C():
+				if c.getPaused() {
+					continue
+				}
 				scraper := c.getScraper()
 				if scraper == nil {
 					// Don't scrape empty target service.
@@ -345,6 +374,24 @@ func (c *collection) close() {
 	c.grp.Wait()
 }
 
+// pause the scraper, happens when activator in path
+func (c *collection) setPause(pause bool) {
+	c.mux.Lock()
+	defer c.mux.Unlock()
+
+	if c.paused != pause {
+		c.paused = pause
+	}
+}
+
+// pause the scraper, happens when activator in path
+func (c *collection) getPaused() bool {
+	c.mux.Lock()
+	defer c.mux.Unlock()
+
+	return c.paused
+}
+
 // updateMetric safely updates the metric stored in the collection.
 func (c *collection) updateMetric(metric *autoscalingv1alpha1.Metric) {
 	c.mux.Lock()

pkg/autoscaler/scaling/autoscaler.go

@@ -293,6 +293,11 @@ func (a *autoscaler) Scale(logger *zap.SugaredLogger, now time.Time) ScaleResult
 			observedPanicValue, spec.TargetBurstCapacity, excessBCF))
 	}
 
+	// Resume pod scraping if excess burst capacity >= 0
+	if excessBCF >= 0 {
+		a.metricClient.Resume(metricKey)
+	}
+
 	switch spec.ScalingMetric {
 	case autoscaling.RPS:
 		a.metrics.RecordRPS(
@@ -312,6 +317,11 @@ func (a *autoscaler) Scale(logger *zap.SugaredLogger, now time.Time) ScaleResult
 		)
 	}
 
+	// pause after recording concurrency if excess burst capacity is < 0
+	if excessBCF < 0 {
+		a.metricClient.Pause(metricKey)
+	}
+
 	return ScaleResult{
 		DesiredPodCount:     desiredPodCount,
 		ExcessBurstCapacity: int32(excessBCF),

pkg/autoscaler/scaling/autoscaler_test.go

@@ -868,6 +868,51 @@ func TestNewFail(t *testing.T) {
 	}
 }
 
+func TestPausingCollection(t *testing.T) {
+	reader := metric.NewManualReader()
+	key := types.NamespacedName{Namespace: testNamespace, Name: testRevision}
+	mp := metric.NewMeterProvider(metric.WithReader(reader))
+	attrs := attribute.NewSet(attribute.String("foo", "bar"))
+	testColl := &testCollection{
+		paused: false,
+	}
+	metrics := &metricClient{
+		StableConcurrency: 9.0,
+		PanicConcurrency:  9.0,
+		StableRPS:         9.0,
+		PanicRPS:          9.0,
+		collections: map[types.NamespacedName]*testCollection{
+			key: testColl,
+		},
+	}
+	deciderSpec := &DeciderSpec{
+		TargetValue:         3,
+		TotalValue:          4,
+		TargetBurstCapacity: 1,
+		PanicThreshold:      2,
+		MaxScaleUpRate:      10,
+		MaxScaleDownRate:    10,
+		StableWindow:        stableWindow,
+	}
+
+	pc := fakePodCounter{
+		readyCount: 1,
+	}
+	a := newAutoscaler(attrs, mp, testNamespace, testRevision, metrics, pc, deciderSpec, nil)
+	now := time.Now()
+	_ = a.Scale(logtesting.TestLogger(t), now)
+	if metrics.collections[key].getPaused() != true {
+		t.Errorf("metric collection should be paused but is not")
+	}
+	metrics.SetStableAndPanicConcurrency(3, 3)
+	a.Update(deciderSpec)
+	now = time.Now()
+	_ = a.Scale(logtesting.TestLogger(t), now)
+	if metrics.collections[key].getPaused() != false {
+		t.Errorf("metric collection should be resumed but was paused")
+	}
+}
+
 // staticMetricClient returns stable/panic concurrency and RPS with static value, i.e. 10.
 var staticMetricClient = metricClient{
 	StableConcurrency: 10.0,
@@ -876,15 +921,29 @@ var staticMetricClient = metricClient{
 	PanicRPS:          10.0,
 }
 
+// test collection type
+type testCollection struct {
+	paused bool
+}
+
 // metricClient is a fake implementation of autoscaler.metricClient for testing.
 type metricClient struct {
 	StableConcurrency float64
 	PanicConcurrency  float64
 	StableRPS         float64
 	PanicRPS          float64
+	collections       map[types.NamespacedName]*testCollection
 	ErrF              func(key types.NamespacedName, now time.Time) error
 }
 
+func (c *testCollection) setPause(paused bool) {
+	c.paused = paused
+}
+
+func (c *testCollection) getPaused() bool {
+	return c.paused
+}
+
 // SetStableAndPanicConcurrency sets the stable and panic concurrencies.
 func (mc *metricClient) SetStableAndPanicConcurrency(s, p float64) {
 	mc.StableConcurrency, mc.PanicConcurrency = s, p
@@ -910,6 +969,20 @@ func (mc *metricClient) StableAndPanicRPS(key types.NamespacedName, now time.Tim
 	return mc.StableRPS, mc.PanicRPS, err
 }
 
+// Pauses metric collection
+func (mc *metricClient) Pause(key types.NamespacedName) {
+	if collection, exists := mc.collections[key]; exists {
+		collection.setPause(true)
+	}
+}
+
+// Resumes metric collection
+func (mc *metricClient) Resume(key types.NamespacedName) {
+	if collection, exists := mc.collections[key]; exists {
+		collection.setPause(false)
+	}
+}
+
 func BenchmarkAutoscaler(b *testing.B) {
 	metrics := &metricClient{StableConcurrency: 50.0, PanicConcurrency: 10}
 	a := newTestAutoscalerNoPC(10, 101, metrics)