[release-1.19] Make autoscaler instrument async so we can remove metric attributes when revisions go away (#16301)

* switch autoscaler metrics to be async - allowing removal

* allow removing kpa metrics when a revision goes away

---------

Co-authored-by: Dave Protasowski <dprotaso@gmail.com>

Author: knative-prow-robot

pkg/autoscaler/metrics/stats_scraper.go

@@ -203,7 +203,8 @@ func newServiceScraperWithClient(
 			semconv.K8SNamespaceName(m.ObjectMeta.Namespace),
 			metrics.ServiceNameKey.With(svcName),
 			metrics.ConfigurationNameKey.With(cfgName),
-			metrics.RevisionNameKey.With(revisionName),
+			// TODO: Re-enable when OTel has the ability to remove attributes
+			// metrics.RevisionNameKey.With(revisionName),
 		),
 	}
 }

pkg/autoscaler/scaling/autoscaler.go

@@ -91,7 +91,7 @@ func newAutoscaler(
 	deciderSpec *DeciderSpec,
 	delayWindow *max.TimeWindow,
 ) *autoscaler {
-	metrics := newMetrics(mp, attrs)
+	metrics := newMetrics(mp, deciderSpec.ScalingMetric, attrs)
 	// We always start in the panic mode, if the deployment is scaled up over 1 pod.
 	// If the scale is 0 or 1, normal Autoscaler behavior is fine.
 	// When Autoscaler restarts we lose metric history, which causes us to
@@ -137,6 +137,10 @@ func (a *autoscaler) Update(deciderSpec *DeciderSpec) {
 	a.deciderSpec = deciderSpec
 }
 
+func (a *autoscaler) OnDelete() {
+	a.metrics.OnDelete()
+}
+
 // Scale calculates the desired scale based on current statistics given the current time.
 // desiredPodCount is the calculated pod count the autoscaler would like to set.
 // validScale signifies whether the desiredPodCount should be applied or not.
@@ -293,24 +297,13 @@ func (a *autoscaler) Scale(logger *zap.SugaredLogger, now time.Time) ScaleResult
 			observedPanicValue, spec.TargetBurstCapacity, excessBCF))
 	}
 
-	switch spec.ScalingMetric {
-	case autoscaling.RPS:
-		a.metrics.RecordRPS(
-			excessBCF,
-			int64(desiredPodCount),
-			observedStableValue,
-			observedPanicValue,
-			spec.TargetValue,
-		)
-	default:
-		a.metrics.RecordConcurrency(
-			excessBCF,
-			int64(desiredPodCount),
-			observedStableValue,
-			observedPanicValue,
-			spec.TargetValue,
-		)
-	}
+	a.metrics.Record(
+		excessBCF,
+		int64(desiredPodCount),
+		observedStableValue,
+		observedPanicValue,
+		spec.TargetValue,
+	)
 
 	return ScaleResult{
 		DesiredPodCount:     desiredPodCount,

pkg/autoscaler/scaling/autoscaler_test.go

@@ -217,6 +217,7 @@ func expectedEBC(totCap, targetBC, recordedConcurrency, numPods float64) int32 {
 }
 
 func TestAutoscalerStartMetrics(t *testing.T) {
+	expectedAttrs := attribute.NewSet(attribute.String("foo", "bar"))
 	metrics := &metricClient{StableConcurrency: 50.0, PanicConcurrency: 50.0}
 
 	_, _, reader := newTestAutoscalerWithScalingMetric(
@@ -227,39 +228,16 @@ func TestAutoscalerStartMetrics(t *testing.T) {
 
 	metricstest.AssertMetrics(
 		t, reader,
-		metricstest.MetricsEqual(scopeName, panicMetric(true)),
-	)
-}
-
-func TestAutoscalerMetrics(t *testing.T) {
-	metrics := &metricClient{StableConcurrency: 50.0, PanicConcurrency: 50.0}
-	a, _, reader := newTestAutoscaler(10, 100, metrics)
-	expectedAttrs := attribute.NewSet(attribute.String("foo", "bar"))
-
-	// Non-panic created autoscaler.
-	metricstest.AssertMetrics(
-		t, reader,
-		metricstest.MetricsEqual(scopeName, panicMetric(false)),
-	)
-
-	ebc := expectedEBC(10, 100, 50, 1)
-	expectScale(t, a, time.Now(), ScaleResult{5, ebc, true})
-	spec := a.currentSpec()
-
-	metricstest.AssertMetrics(
-		t, reader,
-		metricstest.MetricsEqual(
-			scopeName,
+		metricstest.MetricsEqual(scopeName,
 			panicMetric(true),
-
 			metricdata.Metrics{
 				Name:        "kn.revision.pods.desired",
 				Description: "Number of pods the autoscaler wants to allocate",
 				Unit:        "{item}",
 				Data: metricdata.Gauge[int64]{
 					DataPoints: []metricdata.DataPoint[int64]{{
 						Attributes: expectedAttrs,
-						Value:      5,
+						Value:      0,
 					}},
 				},
 			},
@@ -271,7 +249,7 @@ func TestAutoscalerMetrics(t *testing.T) {
 				Data: metricdata.Gauge[float64]{
 					DataPoints: []metricdata.DataPoint[float64]{{
 						Attributes: expectedAttrs,
-						Value:      float64(ebc),
+						Value:      float64(0),
 					}},
 				},
 			},
@@ -283,7 +261,7 @@ func TestAutoscalerMetrics(t *testing.T) {
 				Data: metricdata.Gauge[float64]{
 					DataPoints: []metricdata.DataPoint[float64]{{
 						Attributes: expectedAttrs,
-						Value:      50,
+						Value:      0,
 					}},
 				},
 			},
@@ -295,7 +273,7 @@ func TestAutoscalerMetrics(t *testing.T) {
 				Data: metricdata.Gauge[float64]{
 					DataPoints: []metricdata.DataPoint[float64]{{
 						Attributes: expectedAttrs,
-						Value:      50,
+						Value:      0,
 					}},
 				},
 			},
@@ -307,14 +285,39 @@ func TestAutoscalerMetrics(t *testing.T) {
 				Data: metricdata.Gauge[float64]{
 					DataPoints: []metricdata.DataPoint[float64]{{
 						Attributes: expectedAttrs,
-						Value:      float64(spec.TargetValue),
+						Value:      float64(0),
 					}},
 				},
 			},
 		),
 	)
 }
 
+func TestAutoscalerMetrics(t *testing.T) {
+	metrics := &metricClient{StableConcurrency: 50.0, PanicConcurrency: 50.0}
+	a, _, reader := newTestAutoscaler(10, 100, metrics)
+	expectedAttrs := attribute.NewSet(attribute.String("foo", "bar"))
+
+	expectedMetrics := concurrencyDataPoints(expectedAttrs, false, 0, 0, 0, 0, 0)
+
+	// Non-panic created autoscaler.
+	metricstest.AssertMetrics(
+		t, reader,
+		metricstest.MetricsEqual(scopeName, expectedMetrics...),
+	)
+
+	ebc := expectedEBC(10, 100, 50, 1)
+	expectScale(t, a, time.Now(), ScaleResult{5, ebc, true})
+	spec := a.currentSpec()
+
+	expectedMetrics = concurrencyDataPoints(expectedAttrs, true, 5, float64(ebc), 50, 50, spec.TargetValue)
+
+	metricstest.AssertMetrics(
+		t, reader,
+		metricstest.MetricsEqual(scopeName, expectedMetrics...),
+	)
+}
+
 func TestAutoscalerMetricsWithRPS(t *testing.T) {
 	expectedAttrs := attribute.NewSet(attribute.String("foo", "bar"))
 	metrics := &metricClient{PanicRPS: 99.0, StableRPS: 100}
@@ -936,3 +939,68 @@ func panicMetric(panic bool) metricdata.Metrics {
 		},
 	}
 }
+
+func concurrencyDataPoints(attrs attribute.Set, isPanic bool, pods int64, ebc, stable, panic, target float64) []metricdata.Metrics {
+	return []metricdata.Metrics{
+		panicMetric(isPanic),
+		{
+			Name:        "kn.revision.pods.desired",
+			Description: "Number of pods the autoscaler wants to allocate",
+			Unit:        "{item}",
+			Data: metricdata.Gauge[int64]{
+				DataPoints: []metricdata.DataPoint[int64]{{
+					Attributes: attrs,
+					Value:      pods,
+				}},
+			},
+		},
+
+		{
+			Name:        "kn.revision.capacity.excess",
+			Description: "Excess burst capacity observed over the stable window",
+			Unit:        "{concurrency}",
+			Data: metricdata.Gauge[float64]{
+				DataPoints: []metricdata.DataPoint[float64]{{
+					Attributes: attrs,
+					Value:      ebc,
+				}},
+			},
+		},
+
+		{
+			Name:        "kn.revision.concurrency.stable",
+			Description: "Average of request count per observed pod over the stable window",
+			Unit:        "{concurrency}",
+			Data: metricdata.Gauge[float64]{
+				DataPoints: []metricdata.DataPoint[float64]{{
+					Attributes: attrs,
+					Value:      stable,
+				}},
+			},
+		},
+
+		{
+			Name:        "kn.revision.concurrency.panic",
+			Description: "Average of request count per observed pod over the panic window",
+			Unit:        "{concurrency}",
+			Data: metricdata.Gauge[float64]{
+				DataPoints: []metricdata.DataPoint[float64]{{
+					Attributes: attrs,
+					Value:      panic,
+				}},
+			},
+		},
+
+		{
+			Name:        "kn.revision.concurrency.target",
+			Description: "The desired concurrent requests for each pod",
+			Unit:        "{concurrency}",
+			Data: metricdata.Gauge[float64]{
+				DataPoints: []metricdata.DataPoint[float64]{{
+					Attributes: attrs,
+					Value:      target,
+				}},
+			},
+		},
+	}
+}