🌱 add grpc subscribers metric. (#148)

* add grpc subscribers metric.

Signed-off-by: morvencao <[email protected]>

rh-pre-commit.version: 2.3.2
rh-pre-commit.check-secrets: ENABLED

* remove origin source label for source side metrics.

Signed-off-by: morvencao <[email protected]>

rh-pre-commit.version: 2.3.2
rh-pre-commit.check-secrets: ENABLED

Author: morvencao

pkg/cloudevents/generic/metrics_collector.go

@@ -47,12 +47,11 @@ var cloudeventsReceivedByClientMetricsLabels = []string{
 
 // cloudeventsSentFromSourceMetricsLabels - Array of labels added to cloudevents sent from source metrics:
 var cloudeventsSentFromSourceMetricsLabels = []string{
-	metricsSourceLabel,         // source
-	metricsOriginalSourceLabel, // original source, if no, set to "none"
-	metricsConsumerLabel,       // consumer
-	metricsDataTypeLabel,       // data type, e.g. manifests, manifestbundles
-	metricsSubResourceLabel,    // subresource, eg, spec or status
-	metricsActionLabel,         // action, eg, create, update, delete, resync_request, resync_response
+	metricsSourceLabel,      // source
+	metricsConsumerLabel,    // consumer
+	metricsDataTypeLabel,    // data type, e.g. manifests, manifestbundles
+	metricsSubResourceLabel, // subresource, eg, spec or status
+	metricsActionLabel,      // action, eg, create, update, delete, resync_request, resync_response
 }
 
 // cloudeventsSentFromClientMetricsLabels - Array of labels added to cloudevents sent from client metrics:
@@ -125,7 +124,7 @@ var cloudeventsReceivedByClientCounterMetric = prometheus.NewCounterVec(
 // The cloudevents sent from source counter metric is a counter with a base metric name of 'sent_from_source_total'
 // and a help string of 'The total number of CloudEvents sent from source.'
 // For example, 1 cloudevent sent from source1 to consumer1 with data type manifestbundles for resource spec create would result in the following metrics:
-// cloudevents_sent_total{source="source1",original_source="none",consumer="consumer1",type="io.open-cluster-management.works.v1alpha1.manifestbundles",subresource="spec",action="create"} 1
+// cloudevents_sent_total{source="source1",consumer="consumer1",type="io.open-cluster-management.works.v1alpha1.manifestbundles",subresource="spec",action="create"} 1
 var cloudeventsSentFromSourceCounterMetric = prometheus.NewCounterVec(
 	prometheus.CounterOpts{
 		Subsystem: cloudeventsMetricsSubsystem,
@@ -294,17 +293,13 @@ func increaseCloudEventsReceivedByAgentCounter(source, dataType, subresource, ac
 }
 
 // increaseCloudEventsSentFromSourceCounter increases the cloudevents sent from source counter metric:
-func increaseCloudEventsSentFromSourceCounter(source, originalSource, consumer, dataType, subresource, action string) {
-	if originalSource == "" {
-		originalSource = noneOriginalSource
-	}
+func increaseCloudEventsSentFromSourceCounter(source, consumer, dataType, subresource, action string) {
 	labels := prometheus.Labels{
-		metricsSourceLabel:         source,
-		metricsOriginalSourceLabel: originalSource,
-		metricsConsumerLabel:       consumer,
-		metricsDataTypeLabel:       dataType,
-		metricsSubResourceLabel:    subresource,
-		metricsActionLabel:         action,
+		metricsSourceLabel:      source,
+		metricsConsumerLabel:    consumer,
+		metricsDataTypeLabel:    dataType,
+		metricsSubResourceLabel: subresource,
+		metricsActionLabel:      action,
 	}
 	cloudeventsSentFromSourceCounterMetric.With(labels).Inc()
 }

pkg/cloudevents/generic/metrics_collector_test.go

@@ -100,7 +100,7 @@ func TestCloudEventsMetrics(t *testing.T) {
 			time.Sleep(time.Second)
 
 			// ensure metrics are updated
-			sentTotal := cloudeventsSentFromSourceCounterMetric.WithLabelValues(c.sourceID, noneOriginalSource, c.clusterName, c.dataType.String(), string(c.subresource), string(c.action))
+			sentTotal := cloudeventsSentFromSourceCounterMetric.WithLabelValues(c.sourceID, c.clusterName, c.dataType.String(), string(c.subresource), string(c.action))
 			require.Equal(t, len(c.resources), int(toFloat64Counter(sentTotal)))
 			receivedTotal := cloudeventsReceivedByClientCounterMetric.WithLabelValues(c.sourceID, c.dataType.String(), string(c.subresource), string(c.action))
 			require.Equal(t, len(c.resources), int(toFloat64Counter(receivedTotal)))
@@ -239,7 +239,7 @@ func TestResyncSpecMetrics(t *testing.T) {
 				require.Greater(t, sum, 0.0)
 				require.Less(t, sum, 1.0)
 
-				sentTotal := cloudeventsSentFromSourceCounterMetric.WithLabelValues(c.sourceID, noneOriginalSource, c.clusterName, c.dataType.String(), string(types.SubResourceSpec), string(types.ResyncResponseAction))
+				sentTotal := cloudeventsSentFromSourceCounterMetric.WithLabelValues(c.sourceID, c.clusterName, c.dataType.String(), string(types.SubResourceSpec), string(types.ResyncResponseAction))
 				require.Equal(t, len(c.resources), int(toFloat64Counter(sentTotal)))
 			}
 

pkg/cloudevents/generic/sourceclient.go

@@ -105,7 +105,7 @@ func (c *CloudEventSourceClient[T]) Resync(ctx context.Context, clusterName stri
 		return err
 	}
 
-	increaseCloudEventsSentFromSourceCounter(evt.Source(), "", clusterName, c.codec.EventDataType().String(), string(eventType.SubResource), string(eventType.Action))
+	increaseCloudEventsSentFromSourceCounter(evt.Source(), clusterName, c.codec.EventDataType().String(), string(eventType.SubResource), string(eventType.Action))
 
 	return nil
 }
@@ -130,7 +130,7 @@ func (c *CloudEventSourceClient[T]) Publish(ctx context.Context, eventType types
 	}
 
 	clusterName := evt.Context.GetExtensions()[types.ExtensionClusterName].(string)
-	increaseCloudEventsSentFromSourceCounter(evt.Source(), "", clusterName, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))
+	increaseCloudEventsSentFromSourceCounter(evt.Source(), clusterName, eventType.CloudEventsDataType.String(), string(eventType.SubResource), string(eventType.Action))
 
 	return nil
 }
@@ -290,7 +290,7 @@ func (c *CloudEventSourceClient[T]) respondResyncSpecRequest(
 		if err := c.publish(ctx, evt); err != nil {
 			return err
 		}
-		increaseCloudEventsSentFromSourceCounter(evt.Source(), "", fmt.Sprintf("%s", clusterName), evtDataType.String(), string(eventType.SubResource), string(eventType.Action))
+		increaseCloudEventsSentFromSourceCounter(evt.Source(), fmt.Sprintf("%s", clusterName), evtDataType.String(), string(eventType.SubResource), string(eventType.Action))
 	}
 
 	return nil

pkg/cloudevents/server/grpc/broker.go

@@ -4,10 +4,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/heartbeat"
 	"sync"
 	"time"
 
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/heartbeat"
+	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/metrics"
+
 	"k8s.io/apimachinery/pkg/api/errors"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -134,6 +136,7 @@ func (bkr *GRPCBroker) register(
 	}
 
 	klog.V(4).Infof("register a subscriber %s (cluster name = %s)", id, clusterName)
+	metrics.IncGRPCCESubscribersMetric(clusterName, dataType.String())
 
 	return id, errChan
 }
@@ -147,6 +150,7 @@ func (bkr *GRPCBroker) unregister(id string) {
 	if sub, exists := bkr.subscribers[id]; exists {
 		close(sub.errChan)
 		delete(bkr.subscribers, id)
+		metrics.DecGRPCCESubscribersMetric(sub.clusterName, sub.dataType.String())
 	}
 }
 

pkg/cloudevents/server/grpc/metrics/metrics.go

@@ -9,6 +9,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/cloudevents/sdk-go/v2/binding"
+	cetypes "github.com/cloudevents/sdk-go/v2/types"
 	pbv1 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protobuf/v1"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protocol"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
@@ -21,37 +22,51 @@ const grpcCEMetricsSubsystem = "grpc_server_ce"
 
 // Names of the labels added to metrics:
 const (
+	grpcCEMetricsClusterLabel  = "consumer"
 	grpcCEMetricsDataTypeLabel = "data_type"
 	grpcCEMetricsMethodLabel   = "method"
 	grpcMetricsCodeLabel       = "grpc_code"
 )
 
-// grpcCEMetricsLabels - Array of labels added to grpc server metrics for cloudevents:
-var grpcCEMetricsLabels = []string{
+// grpcCEMetricsCommonLabels - Array of common labels added to grpc server metrics for cloudevents:
+var grpcCEMetricsCommonLabels = []string{
+	grpcCEMetricsClusterLabel,
 	grpcCEMetricsDataTypeLabel,
-	grpcCEMetricsMethodLabel,
 }
 
+// grpcCEMetricsHandlerLabels - Array of handler labels added to grpc server metrics for cloudevents:
+var grpcCEMetricsHandlerLabels = append(grpcCEMetricsCommonLabels, grpcCEMetricsMethodLabel)
+
 // grpcCEMetricsAllLabels - Array of all labels added to grpc server metrics for cloudevents:
-var grpcCEMetricsAllLabels = append(grpcCEMetricsLabels, grpcMetricsCodeLabel)
+var grpcCEMetricsAllLabels = append(grpcCEMetricsHandlerLabels, grpcMetricsCodeLabel)
 
 // Names of the grpc server metrics for cloudevents:
 const (
+	subscribersMetric          = "subscribers"
 	calledCountMetric          = "called_total"
 	processedCountMetric       = "processed_total"
 	processingDurationMetric   = "processing_duration_seconds"
 	messageReceivedCountMetric = "msg_received_total"
 	messageSentCountMetric     = "msg_sent_total"
 )
 
+// grpcCESubscribersMetric is a gauge metric that tracks the number of registered
+// subscribers for cloudevents on the gRPC server.
+var grpcCESubscribersMetric = k8smetrics.NewGaugeVec(&k8smetrics.GaugeOpts{
+	Subsystem:      grpcCEMetricsSubsystem,
+	Name:           subscribersMetric,
+	StabilityLevel: k8smetrics.ALPHA,
+	Help:           "Number of registered subscribers for cloudevents on the grpc server.",
+}, grpcCEMetricsCommonLabels)
+
 // grpcCECalledCountMetric is a counter metric that tracks the total number of
 // RPC requests for cloudevents called on the gRPC server.
 var grpcCECalledCountMetric = k8smetrics.NewCounterVec(&k8smetrics.CounterOpts{
 	Subsystem:      grpcCEMetricsSubsystem,
 	Name:           calledCountMetric,
 	StabilityLevel: k8smetrics.ALPHA,
 	Help:           "Total number of RPC requests for cloudevents called on the grpc server.",
-}, grpcCEMetricsLabels)
+}, grpcCEMetricsHandlerLabels)
 
 // grpcCEMessageReceivedCountMetric is a counter metric that tracks the total number of
 // messages for cloudevents received on the gRPC server.
@@ -60,7 +75,7 @@ var grpcCEMessageReceivedCountMetric = k8smetrics.NewCounterVec(&k8smetrics.Coun
 	Name:           messageReceivedCountMetric,
 	StabilityLevel: k8smetrics.ALPHA,
 	Help:           "Total number of messages for cloudevents received on the gRPC server.",
-}, grpcCEMetricsLabels)
+}, grpcCEMetricsHandlerLabels)
 
 // grpcCEMessageSentCountMetric is a counter metric that tracks the total number of
 // messages for cloudevents sent by the gRPC server.
@@ -69,7 +84,7 @@ var grpcCEMessageSentCountMetric = k8smetrics.NewCounterVec(&k8smetrics.CounterO
 	Name:           messageSentCountMetric,
 	StabilityLevel: k8smetrics.ALPHA,
 	Help:           "Total number of messages for cloudevents sent by the gRPC server.",
-}, grpcCEMetricsLabels)
+}, grpcCEMetricsHandlerLabels)
 
 // grpcCEProcessedCountMetric is a counter metric that tracks the total number of
 // RPC requests for cloudevents processed on the server, regardless of success or failure.
@@ -106,61 +121,72 @@ func NewCloudEventsMetricsUnaryInterceptor() grpc.UnaryServerInterceptor {
 		}
 
 		// initialize defaults for error cases
+		cluster := "unknown"
 		dataType := "unknown"
 
 		pubReq, ok := req.(*pbv1.PublishRequest)
 		if !ok {
 			err := fmt.Errorf("invalid request type for Publish method")
-			recordCloudEventsMetrics(dataType, method, err, startTime)
+			recordCloudEventsMetrics(cluster, dataType, method, err, startTime)
 			return nil, err
 		}
 		// convert the request to cloudevent and extract the source
 		evt, err := binding.ToEvent(ctx, protocol.NewMessage(pubReq.Event))
 		if err != nil {
 			err = fmt.Errorf("failed to convert to cloudevent: %v", err)
-			recordCloudEventsMetrics(dataType, method, err, startTime)
+			recordCloudEventsMetrics(cluster, dataType, method, err, startTime)
+			return nil, err
+		}
+
+		// extract the cluster name from event extensions
+		clusterVal, err := cetypes.ToString(evt.Context.GetExtensions()[types.ExtensionClusterName])
+		if err != nil {
+			err = fmt.Errorf("failed to get clustername extension: %v", err)
+			recordCloudEventsMetrics(cluster, dataType, method, err, startTime)
 			return nil, err
 		}
+		cluster = clusterVal
 
 		// extract the data type from event type
 		eventType, err := types.ParseCloudEventsType(evt.Type())
 		if err != nil {
 			err = fmt.Errorf("failed to parse cloud event type %s, %v", evt.Type(), err)
-			recordCloudEventsMetrics(dataType, method, err, startTime)
+			recordCloudEventsMetrics(cluster, dataType, method, err, startTime)
 			return nil, err
 		}
 		dataType = eventType.CloudEventsDataType.String()
 
-		grpcCECalledCountMetric.WithLabelValues(dataType, method).Inc()
-		grpcCEMessageReceivedCountMetric.WithLabelValues(dataType, method).Inc()
+		grpcCECalledCountMetric.WithLabelValues(cluster, dataType, method).Inc()
+		grpcCEMessageReceivedCountMetric.WithLabelValues(cluster, dataType, method).Inc()
 		// call rpc handler to handle RPC request
 		resp, err := handler(ctx, req)
 		duration := time.Since(startTime).Seconds()
-		grpcCEMessageSentCountMetric.WithLabelValues(dataType, method).Inc()
+		grpcCEMessageSentCountMetric.WithLabelValues(cluster, dataType, method).Inc()
 
 		// get status code from error
 		status := statusFromError(err)
 		code := status.Code()
-		grpcCEProcessedCountMetric.WithLabelValues(dataType, method, code.String()).Inc()
-		grpcCEProcessingDurationMetric.WithLabelValues(dataType, method, code.String()).Observe(duration)
+		grpcCEProcessedCountMetric.WithLabelValues(cluster, dataType, method, code.String()).Inc()
+		grpcCEProcessingDurationMetric.WithLabelValues(cluster, dataType, method, code.String()).Observe(duration)
 
 		return resp, err
 	}
 }
 
-func recordCloudEventsMetrics(dataType, method string, err error, startTime time.Time) {
+func recordCloudEventsMetrics(cluster, dataType, method string, err error, startTime time.Time) {
 	duration := time.Since(startTime).Seconds()
 	status := statusFromError(err)
 	code := status.Code()
-	grpcCEProcessedCountMetric.WithLabelValues(dataType, method, code.String()).Inc()
-	grpcCEProcessingDurationMetric.WithLabelValues(dataType, method, code.String()).Observe(duration)
+	grpcCEProcessedCountMetric.WithLabelValues(cluster, dataType, method, code.String()).Inc()
+	grpcCEProcessingDurationMetric.WithLabelValues(cluster, dataType, method, code.String()).Observe(duration)
 }
 
 // wrappedCloudEventsMetricsStream wraps a grpc.ServerStream, capturing the request source
 // emitting metrics for the stream interceptor.
 type wrappedCloudEventsMetricsStream struct {
-	dataType *string
-	method   string
+	clusterName *string
+	dataType    *string
+	method      string
 	grpc.ServerStream
 	ctx context.Context
 }
@@ -178,10 +204,11 @@ func (w *wrappedCloudEventsMetricsStream) RecvMsg(m interface{}) error {
 		return fmt.Errorf("invalid request type for Subscribe method")
 	}
 
-	if w.dataType != nil {
+	if w.clusterName != nil && w.dataType != nil {
+		*w.clusterName = subReq.ClusterName
 		*w.dataType = subReq.DataType
-		grpcCECalledCountMetric.WithLabelValues(*w.dataType, w.method).Inc()
-		grpcCEMessageReceivedCountMetric.WithLabelValues(*w.dataType, w.method).Inc()
+		grpcCECalledCountMetric.WithLabelValues(*w.clusterName, *w.dataType, w.method).Inc()
+		grpcCEMessageReceivedCountMetric.WithLabelValues(*w.clusterName, *w.dataType, w.method).Inc()
 	}
 
 	return nil
@@ -194,16 +221,16 @@ func (w *wrappedCloudEventsMetricsStream) SendMsg(m interface{}) error {
 		return err
 	}
 
-	if w.dataType != nil && *w.dataType != "" {
-		grpcCEMessageSentCountMetric.WithLabelValues(*w.dataType, w.method).Inc()
+	if w.clusterName != nil && w.dataType != nil && *w.clusterName != "" && *w.dataType != "" {
+		grpcCEMessageSentCountMetric.WithLabelValues(*w.clusterName, *w.dataType, w.method).Inc()
 	}
 
 	return nil
 }
 
-// newWrappedCloudEventsMetricsStream creates a wrappedCloudEventsMetricsStream with the specified type reference.
-func newWrappedCloudEventsMetricsStream(dataType *string, method string, ctx context.Context, ss grpc.ServerStream) grpc.ServerStream {
-	return &wrappedCloudEventsMetricsStream{dataType, method, ss, ctx}
+// newWrappedCloudEventsMetricsStream creates a wrappedCloudEventsMetricsStream with the specified type and cluster reference.
+func newWrappedCloudEventsMetricsStream(clusterName, dataType *string, method string, ctx context.Context, ss grpc.ServerStream) grpc.ServerStream {
+	return &wrappedCloudEventsMetricsStream{clusterName, dataType, method, ss, ctx}
 }
 
 // NewCloudEventsMetricsStreamInterceptor creates a stream server interceptor for server metrics.
@@ -217,15 +244,16 @@ func NewCloudEventsMetricsStreamInterceptor() grpc.StreamServerInterceptor {
 		}
 
 		dataType := ""
+		cluster := ""
 		// create a wrapped stream to capture the source and emit metrics
-		wrappedCEMetricsStream := newWrappedCloudEventsMetricsStream(&dataType, method, stream.Context(), stream)
+		wrappedCEMetricsStream := newWrappedCloudEventsMetricsStream(&cluster, &dataType, method, stream.Context(), stream)
 		// call rpc handler to handle RPC request
 		err := handler(srv, wrappedCEMetricsStream)
 
 		// get status code from error
 		status := statusFromError(err)
 		code := status.Code()
-		grpcCEProcessedCountMetric.WithLabelValues(dataType, method, code.String()).Inc()
+		grpcCEProcessedCountMetric.WithLabelValues(cluster, dataType, method, code.String()).Inc()
 
 		return err
 	}
@@ -264,10 +292,21 @@ func SplitMethod(fullMethod string) (service, method string) {
 // Register all the grpc server metrics for cloudevents.
 func CloudEventsGRPCMetrics() []k8smetrics.Registerable {
 	return []k8smetrics.Registerable{
+		grpcCESubscribersMetric,
 		grpcCECalledCountMetric,
 		grpcCEProcessedCountMetric,
 		grpcCEProcessingDurationMetric,
 		grpcCEMessageReceivedCountMetric,
 		grpcCEMessageSentCountMetric,
 	}
 }
+
+// IncGRPCCESubscribersMetric increments the grpcCESubscribersMetric by 1 for the given cluster and dataType.
+func IncGRPCCESubscribersMetric(cluster, dataType string) {
+	grpcCESubscribersMetric.WithLabelValues(cluster, dataType).Inc()
+}
+
+// DecGRPCCESubscribersMetric decrements the grpcCESubscribersMetric by 1 for the given cluster and dataType.
+func DecGRPCCESubscribersMetric(cluster, dataType string) {
+	grpcCESubscribersMetric.WithLabelValues(cluster, dataType).Dec()
+}