Merge pull request #88609 from yue9944882/chore/follow-up-metrics

Preserve legacy inflight metrics and fixes registration

staging/src/k8s.io/apiserver/go.mod

@@ -29,7 +29,6 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
 	github.com/pkg/errors v0.8.1
 	github.com/pquerna/cachecontrol v0.0.0-20171018203845-0dec1b30a021 // indirect
-	github.com/prometheus/client_golang v1.0.0
 	github.com/prometheus/client_model v0.2.0
 	github.com/sirupsen/logrus v1.4.2 // indirect
 	github.com/spf13/pflag v1.0.5

staging/src/k8s.io/apiserver/pkg/server/filters/priority-and-fairness.go

@@ -20,11 +20,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
-
-	// TODO: decide whether to also generate the old metrics, which
-	// categorize according to mutating vs readonly.
-
-	// "k8s.io/apiserver/pkg/endpoints/metrics"
+	"sync/atomic"
 
 	fcv1a1 "k8s.io/api/flowcontrol/v1alpha1"
 	apitypes "k8s.io/apimachinery/pkg/types"
@@ -38,8 +34,8 @@ type priorityAndFairnessKeyType int
 const priorityAndFairnessKey priorityAndFairnessKeyType = iota
 
 const (
-	responseHeaderMatchedPriorityLevelConfigurationUID = "X-Kubernetes-PF-PriorityLevelUID"
-	responseHeaderMatchedFlowSchemaUID                 = "X-Kubernetes-PF-FlowSchemaUID"
+	responseHeaderMatchedPriorityLevelConfigurationUID = "X-Kubernetes-PF-PriorityLevel-UID"
+	responseHeaderMatchedFlowSchemaUID                 = "X-Kubernetes-PF-FlowSchema-UID"
 )
 
 // PriorityAndFairnessClassification identifies the results of
@@ -57,6 +53,8 @@ func GetClassification(ctx context.Context) *PriorityAndFairnessClassification {
 	return ctx.Value(priorityAndFairnessKey).(*PriorityAndFairnessClassification)
 }
 
+var atomicMutatingLen, atomicNonMutatingLen int32
+
 // WithPriorityAndFairness limits the number of in-flight
 // requests in a fine-grained way.
 func WithPriorityAndFairness(
@@ -68,7 +66,7 @@ func WithPriorityAndFairness(
 		klog.Warningf("priority and fairness support not found, skipping")
 		return handler
 	}
-
+	startOnce.Do(startRecordingUsage)
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
 		requestInfo, ok := apirequest.RequestInfoFrom(ctx)
@@ -97,8 +95,25 @@ func WithPriorityAndFairness(
 				PriorityLevelName: pl.Name,
 				PriorityLevelUID:  pl.UID}
 		}
+
 		var served bool
+		isMutatingRequest := !nonMutatingRequestVerbs.Has(requestInfo.Verb)
 		execute := func() {
+			var mutatingLen, readOnlyLen int
+			if isMutatingRequest {
+				mutatingLen = int(atomic.AddInt32(&atomicMutatingLen, 1))
+			} else {
+				readOnlyLen = int(atomic.AddInt32(&atomicNonMutatingLen, 1))
+			}
+			defer func() {
+				if isMutatingRequest {
+					atomic.AddInt32(&atomicMutatingLen, -11)
+					watermark.recordMutating(mutatingLen)
+				} else {
+					atomic.AddInt32(&atomicNonMutatingLen, -1)
+					watermark.recordReadOnly(readOnlyLen)
+				}
+			}()
 			served = true
 			innerCtx := context.WithValue(ctx, priorityAndFairnessKey, classification)
 			innerReq := r.Clone(innerCtx)

staging/src/k8s.io/apiserver/pkg/server/routes/BUILD

@@ -26,6 +26,7 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/endpoints/metrics:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/mux:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/etcd3/metrics:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics:go_default_library",
         "//staging/src/k8s.io/component-base/metrics/legacyregistry:go_default_library",
         "//vendor/github.com/emicklei/go-restful:go_default_library",
         "//vendor/github.com/go-openapi/spec:go_default_library",

staging/src/k8s.io/apiserver/pkg/server/routes/metrics.go

@@ -23,6 +23,7 @@ import (
 	apimetrics "k8s.io/apiserver/pkg/endpoints/metrics"
 	"k8s.io/apiserver/pkg/server/mux"
 	etcd3metrics "k8s.io/apiserver/pkg/storage/etcd3/metrics"
+	flowcontrolmetrics "k8s.io/apiserver/pkg/util/flowcontrol/metrics"
 	"k8s.io/component-base/metrics/legacyregistry"
 )
 
@@ -58,4 +59,5 @@ func (m MetricsWithReset) Install(c *mux.PathRecorderMux) {
 func register() {
 	apimetrics.Register()
 	etcd3metrics.Register()
+	flowcontrolmetrics.Register()
 }

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/BUILD

@@ -6,7 +6,10 @@ go_library(
     importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/util/flowcontrol/metrics",
     importpath = "k8s.io/apiserver/pkg/util/flowcontrol/metrics",
     visibility = ["//visibility:public"],
-    deps = ["//vendor/github.com/prometheus/client_golang/prometheus:go_default_library"],
+    deps = [
+        "//staging/src/k8s.io/component-base/metrics:go_default_library",
+        "//staging/src/k8s.io/component-base/metrics/legacyregistry:go_default_library",
+    ],
 )
 
 filegroup(

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/metrics/metrics.go

@@ -17,9 +17,11 @@ limitations under the License.
 package metrics
 
 import (
+	"sync"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
+	compbasemetrics "k8s.io/component-base/metrics"
+	"k8s.io/component-base/metrics/legacyregistry"
 )
 
 const (
@@ -37,28 +39,29 @@ var (
 	requestDurationSecondsBuckets = []float64{0, 0.005, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 30}
 )
 
-func init() {
-	prometheus.MustRegister(apiserverRejectedRequests)
-	prometheus.MustRegister(apiserverCurrentInqueueRequests)
-	prometheus.MustRegister(apiserverRequestQueueLength)
-	prometheus.MustRegister(apiserverRequestConcurrencyLimit)
-	prometheus.MustRegister(apiserverCurrentExecutingRequests)
-	prometheus.MustRegister(apiserverRequestWaitingSeconds)
-	prometheus.MustRegister(apiserverRequestExecutionSeconds)
+var registerMetrics sync.Once
+
+// Register all metrics.
+func Register() {
+	registerMetrics.Do(func() {
+		for _, metric := range metrics {
+			legacyregistry.MustRegister(metric)
+		}
+	})
 }
 
 var (
-	apiserverRejectedRequests = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
+	apiserverRejectedRequestsTotal = compbasemetrics.NewCounterVec(
+		&compbasemetrics.CounterOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
-			Name:      "rejected_requests",
+			Name:      "rejected_requests_total",
 			Help:      "Number of rejected requests by api priority and fairness system",
 		},
 		[]string{priorityLevel, "reason"},
 	)
-	apiserverCurrentInqueueRequests = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
+	apiserverCurrentInqueueRequests = compbasemetrics.NewGaugeVec(
+		&compbasemetrics.GaugeOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "current_inqueue_requests",
@@ -66,8 +69,8 @@ var (
 		},
 		[]string{priorityLevel},
 	)
-	apiserverRequestQueueLength = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
+	apiserverRequestQueueLength = compbasemetrics.NewHistogramVec(
+		&compbasemetrics.HistogramOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "request_queue_length",
@@ -76,8 +79,8 @@ var (
 		},
 		[]string{priorityLevel},
 	)
-	apiserverRequestConcurrencyLimit = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
+	apiserverRequestConcurrencyLimit = compbasemetrics.NewGaugeVec(
+		&compbasemetrics.GaugeOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "request_concurrency_limit",
@@ -85,8 +88,8 @@ var (
 		},
 		[]string{priorityLevel},
 	)
-	apiserverCurrentExecutingRequests = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
+	apiserverCurrentExecutingRequests = compbasemetrics.NewGaugeVec(
+		&compbasemetrics.GaugeOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "current_executing_requests",
@@ -94,8 +97,8 @@ var (
 		},
 		[]string{priorityLevel},
 	)
-	apiserverRequestWaitingSeconds = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
+	apiserverRequestWaitingSeconds = compbasemetrics.NewHistogramVec(
+		&compbasemetrics.HistogramOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "request_wait_duration_seconds",
@@ -104,8 +107,8 @@ var (
 		},
 		[]string{priorityLevel, flowSchema, "execute"},
 	)
-	apiserverRequestExecutionSeconds = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
+	apiserverRequestExecutionSeconds = compbasemetrics.NewHistogramVec(
+		&compbasemetrics.HistogramOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "request_execution_seconds",
@@ -114,6 +117,15 @@ var (
 		},
 		[]string{priorityLevel, flowSchema},
 	)
+	metrics = []compbasemetrics.Registerable{
+		apiserverRejectedRequestsTotal,
+		apiserverCurrentInqueueRequests,
+		apiserverRequestQueueLength,
+		apiserverRequestConcurrencyLimit,
+		apiserverCurrentExecutingRequests,
+		apiserverRequestWaitingSeconds,
+		apiserverRequestExecutionSeconds,
+	}
 )
 
 // UpdateFlowControlRequestsInQueue updates the value for the # of requests in the specified queues in flow control
@@ -133,7 +145,7 @@ func UpdateSharedConcurrencyLimit(priorityLevel string, limit int) {
 
 // AddReject increments the # of rejected requests for flow control
 func AddReject(priorityLevel string, reason string) {
-	apiserverRejectedRequests.WithLabelValues(priorityLevel, reason).Add(1)
+	apiserverRejectedRequestsTotal.WithLabelValues(priorityLevel, reason).Add(1)
 }
 
 // ObserveQueueLength observes the queue length for flow control