change implementation to use sks mode directly

Author: Alexander-Kita

config/core/300-resources/podautoscaler.yaml

@@ -181,6 +181,9 @@ spec:
                   description: DesiredScale shows the current desired number of replicas for the revision.
                   type: integer
                   format: int32
+                metricsPaused:
+                  description: MetricsPaused to determine whether metric scraping should be paused
+                  type: boolean
                 metricsServiceName:
                   description: |-
                     MetricsServiceName is the K8s Service name that provides revision metrics.

docs/serving-api.md

@@ -497,6 +497,17 @@ int32
 <p>ActualScale shows the actual number of replicas for the revision.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>metricsPaused</code><br/>
+<em>
+bool
+</em>
+</td>
+<td>
+<p>MetricsPaused to determine whether metric scraping should be paused</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="autoscaling.internal.knative.dev/v1alpha1.PodScalable">PodScalable

pkg/apis/autoscaling/v1alpha1/pa_types.go

@@ -127,6 +127,9 @@ type PodAutoscalerStatus struct {
 
 	// ActualScale shows the actual number of replicas for the revision.
 	ActualScale *int32 `json:"actualScale,omitempty"`
+
+	// MetricsPaused to determine whether metric scraping should be paused
+	MetricsPaused bool `json:"metricsPaused,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/autoscaler/metrics/collector.go

@@ -71,6 +71,10 @@ type Collector interface {
 	// Watch registers a singleton function to call when a specific collector's status changes.
 	// The passed name is the namespace/name of the metric owned by the respective collector.
 	Watch(func(types.NamespacedName))
+	// Pause metric collection
+	Pause(metric *autoscalingv1alpha1.Metric)
+	// Resume metric collection
+	Resume(metric *autoscalingv1alpha1.Metric)
 }
 
 // MetricClient surfaces the metrics that can be obtained via the collector.
@@ -82,12 +86,6 @@ 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.
@@ -172,7 +170,9 @@ func (c *MetricCollector) Record(key types.NamespacedName, now time.Time, stat S
 	}
 }
 
-func (c *MetricCollector) Pause(key types.NamespacedName) {
+func (c *MetricCollector) Pause(metric *autoscalingv1alpha1.Metric) {
+	key := types.NamespacedName{Namespace: metric.Namespace, Name: metric.Name}
+
 	c.collectionsMutex.RLock()
 	defer c.collectionsMutex.RUnlock()
 
@@ -181,7 +181,9 @@ func (c *MetricCollector) Pause(key types.NamespacedName) {
 	}
 }
 
-func (c *MetricCollector) Resume(key types.NamespacedName) {
+func (c *MetricCollector) Resume(metric *autoscalingv1alpha1.Metric) {
+	key := types.NamespacedName{Namespace: metric.Namespace, Name: metric.Name}
+
 	c.collectionsMutex.RLock()
 	defer c.collectionsMutex.RUnlock()
 
@@ -272,11 +274,12 @@ type (
 		rpsPanicBuckets         windowAverager
 
 		// Fields relevant for metric scraping specifically.
-		scraper StatsScraper
-		lastErr error
-		grp     sync.WaitGroup
-		paused  bool
-		stopCh  chan struct{}
+		creationTime time.Time
+		scraper      StatsScraper
+		lastErr      error
+		grp          sync.WaitGroup
+		paused       bool
+		stopCh       chan struct{}
 	}
 )
 
@@ -309,6 +312,7 @@ func newCollection(metric *autoscalingv1alpha1.Metric, scraper StatsScraper, clo
 		}
 	}
 
+	creationTime := time.Now()
 	c := &collection{
 		metric: metric,
 		concurrencyBuckets: bucketCtor(
@@ -321,8 +325,9 @@ func newCollection(metric *autoscalingv1alpha1.Metric, scraper StatsScraper, clo
 			metric.Spec.PanicWindow, config.BucketSize),
 		scraper: scraper,
 
-		stopCh: make(chan struct{}),
-		paused: false,
+		stopCh:       make(chan struct{}),
+		creationTime: creationTime,
+		paused:       false,
 	}
 
 	key := types.NamespacedName{Namespace: metric.Namespace, Name: metric.Name}
@@ -339,7 +344,11 @@ func newCollection(metric *autoscalingv1alpha1.Metric, scraper StatsScraper, clo
 			case <-c.stopCh:
 				return
 			case <-scrapeTicker.C():
-				if c.getPaused() {
+				now := time.Now()
+				timeSinceCreation := now.Sub(c.creationTime)
+				stableWindow := 2 * c.metric.Spec.StableWindow
+				// initially wait two stable windows to allow initial scale to zero due to activator behavior
+				if timeSinceCreation > 2*stableWindow && c.getPaused() {
 					continue
 				}
 				scraper := c.getScraper()

pkg/autoscaler/scaling/autoscaler.go

@@ -297,11 +297,6 @@ 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)
-	}
-
 	a.metrics.Record(
 		excessBCF,
 		int64(desiredPodCount),
@@ -310,11 +305,6 @@ func (a *autoscaler) Scale(logger *zap.SugaredLogger, now time.Time) ScaleResult
 		spec.TargetValue,
 	)
 
-	// 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

@@ -871,51 +871,6 @@ 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,
@@ -924,29 +879,15 @@ 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
@@ -972,20 +913,6 @@ 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)

pkg/reconciler/autoscaling/kpa/kpa.go

@@ -104,13 +104,16 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, pa *autoscalingv1alpha1.
 			return fmt.Errorf("error reconciling SKS: %w", err)
 		}
 		pa.Status.MarkSKSNotReady(noPrivateServiceName) // In both cases this is true.
+		pa.Status.MetricsPaused = false                 // unpause metrics
 		c.computeStatus(ctx, pa, podCounts{want: scaleUnknown}, logger)
 		return nil
 	}
 
 	pa.Status.MetricsServiceName = sks.Status.PrivateServiceName
 	decider, err := c.reconcileDecider(ctx, pa)
 	if err != nil {
+		// unpause incase of this error
+		pa.Status.MetricsPaused = false
 		return fmt.Errorf("error reconciling Decider: %w", err)
 	}
 
@@ -171,6 +174,27 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, pa *autoscalingv1alpha1.
 		pa.Status.MarkSKSNotReady(sks.Status.GetCondition(nv1alpha1.ServerlessServiceConditionReady).GetMessage())
 	}
 
+	shouldPause := false
+	isSKSReady := pa.Status.GetCondition(autoscalingv1alpha1.PodAutoscalerConditionSKSReady).IsTrue()
+	// only try probe if sks should be in proxy mode and SKS is ready, otherwise always unpause
+	if sks.Spec.Mode == nv1alpha1.SKSOperationModeProxy && isSKSReady {
+		r, err := c.scaler.activatorProbe(pa, c.scaler.transport)
+		if err != nil {
+			shouldPause = false
+		} else {
+			if r {
+				logger.Debug("activator probed, pausing metrics")
+			}
+			shouldPause = r
+		}
+	} else {
+		shouldPause = false
+	}
+
+	if shouldPause != pa.Status.MetricsPaused {
+		pa.Status.MetricsPaused = shouldPause
+	}
+
 	logger.Infof("PA scale got=%d, want=%d, desiredPods=%d ebc=%d", ready, want,
 		decider.Status.DesiredScale, decider.Status.ExcessBurstCapacity)