-
Notifications
You must be signed in to change notification settings - Fork 21
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: morvencao <[email protected]>
- Loading branch information
Showing
8 changed files
with
539 additions
and
50 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,246 @@ | ||
| package server | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/cloudevents/sdk-go/v2/binding" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/status" | ||
| pbv1 "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protobuf/v1" | ||
| grpcprotocol "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/grpc/protocol" | ||
| ) | ||
|
|
||
| func init() { | ||
| // Register the metrics: | ||
| RegisterGRPCMetrics() | ||
| } | ||
|
|
||
| // NewMetricsUnaryInterceptor creates a unary server interceptor for server metrics. | ||
| // Currently supports the Publish method with PublishRequest. | ||
| func newMetricsUnaryInterceptor() grpc.UnaryServerInterceptor { | ||
| return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { | ||
| // extract the type from the method name | ||
| methodInfo := strings.Split(info.FullMethod, "/") | ||
| if len(methodInfo) != 3 || methodInfo[2] != "Publish" { | ||
| return nil, fmt.Errorf("invalid method name: %s", info.FullMethod) | ||
| } | ||
| t := methodInfo[2] | ||
| pubReq, ok := req.(*pbv1.PublishRequest) | ||
| if !ok { | ||
| return nil, fmt.Errorf("invalid request type for Publish method") | ||
| } | ||
| // convert the request to cloudevent and extract the source | ||
| evt, err := binding.ToEvent(ctx, grpcprotocol.NewMessage(pubReq.Event)) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to convert to cloudevent: %v", err) | ||
| } | ||
| source := evt.Source() | ||
| grpcStartCountMetric.WithLabelValues(t, source).Inc() | ||
|
|
||
| grpcMsgReceivedCountMetric.WithLabelValues(t, source).Inc() | ||
| startTime := time.Now() | ||
| resp, err := handler(ctx, req) | ||
| duration := time.Since(startTime).Seconds() | ||
| grpcMsgSentCountMetric.WithLabelValues(t, source).Inc() | ||
|
|
||
| // get status code from error | ||
| status := statusFromError(err) | ||
| code := status.Code() | ||
| grpcHandledCountMetric.WithLabelValues(t, source, code.String()).Inc() | ||
| grpcHandledDurationMetric.WithLabelValues(t, source).Observe(duration) | ||
|
|
||
| return resp, err | ||
| } | ||
| } | ||
|
|
||
| // wrappedStream wraps a grpc.ServerStream, capturing the request source | ||
| // emitting metrics for the stream interceptor. | ||
| type wrappedStream struct { | ||
| t string | ||
| source *string | ||
| grpc.ServerStream | ||
| ctx context.Context | ||
| } | ||
|
|
||
| // RecvMsg wraps the RecvMsg method of the embedded grpc.ServerStream. | ||
| // It captures the source from the SubscriptionRequest and emits metrics. | ||
| func (w *wrappedStream) RecvMsg(m interface{}) error { | ||
| err := w.ServerStream.RecvMsg(m) | ||
| subReq, ok := m.(*pbv1.SubscriptionRequest) | ||
| if !ok { | ||
| return fmt.Errorf("invalid request type for Subscribe method") | ||
| } | ||
| *w.source = subReq.Source | ||
| grpcStartCountMetric.WithLabelValues(w.t, subReq.Source).Inc() | ||
| grpcMsgReceivedCountMetric.WithLabelValues(w.t, subReq.Source).Inc() | ||
|
|
||
| return err | ||
| } | ||
|
|
||
| // SendMsg wraps the SendMsg method of the embedded grpc.ServerStream. | ||
| func (w *wrappedStream) SendMsg(m interface{}) error { | ||
| err := w.ServerStream.SendMsg(m) | ||
| grpcMsgSentCountMetric.WithLabelValues(w.t, *w.source).Inc() | ||
| return err | ||
| } | ||
|
|
||
| // newWrappedStream creates a wrappedStream with the specified type and source reference. | ||
| func newWrappedStream(t string, source *string, ctx context.Context, ss grpc.ServerStream) grpc.ServerStream { | ||
| return &wrappedStream{t, source, ss, ctx} | ||
| } | ||
|
|
||
| // newMetricsStreamInterceptor creates a stream server interceptor for server metrics. | ||
| // Currently supports the Subscribe method with SubscriptionRequest. | ||
| func newMetricsStreamInterceptor() grpc.StreamServerInterceptor { | ||
| return func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { | ||
| // extract the type from the method name | ||
| if !info.IsServerStream || info.IsClientStream { | ||
| return fmt.Errorf("invalid stream type for stream method: %s", info.FullMethod) | ||
| } | ||
| methodInfo := strings.Split(info.FullMethod, "/") | ||
| if len(methodInfo) != 3 || methodInfo[2] != "Subscribe" { | ||
| return fmt.Errorf("invalid method name for stream method: %s", info.FullMethod) | ||
| } | ||
| t := methodInfo[2] | ||
| source := "" | ||
|
|
||
| // create a wrapped stream to capture the source and emit metrics | ||
| wrappedStream := newWrappedStream(t, &source, stream.Context(), stream) | ||
| startTime := time.Now() | ||
| err := handler(srv, wrappedStream) | ||
| duration := time.Since(startTime).Seconds() | ||
|
|
||
| // get status code from error | ||
| status := statusFromError(err) | ||
| code := status.Code() | ||
| grpcHandledCountMetric.WithLabelValues(t, source, code.String()).Inc() | ||
| grpcHandledDurationMetric.WithLabelValues(t, source).Observe(duration) | ||
|
|
||
| return err | ||
| } | ||
| } | ||
|
|
||
| // statusFromError returns a grpc status. If the error code is neither a valid grpc status | ||
| // nor a context error, codes.Unknown will be set. | ||
| func statusFromError(err error) *status.Status { | ||
| s, ok := status.FromError(err) | ||
| // Mirror what the grpc server itself does, i.e. also convert context errors to status | ||
| if !ok { | ||
| s = status.FromContextError(err) | ||
| } | ||
| return s | ||
| } | ||
|
|
||
| // Subsystem used to define the metrics: | ||
| const grpcMetricsSubsystem = "maestro_grpc_server" | ||
|
|
||
| // Names of the labels added to metrics: | ||
| const ( | ||
| grpcMetricsTypeLabel = "type" | ||
| grpcMetricsSourceLabel = "source" | ||
| grpcMetricsCodeLabel = "code" | ||
| ) | ||
|
|
||
| // grpcMetricsLabels - Array of labels added to metrics: | ||
| var grpcMetricsLabels = []string{ | ||
| grpcMetricsTypeLabel, | ||
| grpcMetricsSourceLabel, | ||
| } | ||
|
|
||
| // grpcMetricsAllLabels - Array of all labels added to metrics: | ||
| var grpcMetricsAllLabels = []string{ | ||
| grpcMetricsTypeLabel, | ||
| grpcMetricsSourceLabel, | ||
| grpcMetricsCodeLabel, | ||
| } | ||
|
|
||
| // Names of the metrics: | ||
| const ( | ||
| startCountMetric = "started_total" | ||
| handledCountMetric = "handled_total" | ||
| handledDurationMetric = "handled_duration_seconds" | ||
| msgReceivedCountMetric = "msg_received_total" | ||
| msgSentCountMetric = "msg_sent_total" | ||
| ) | ||
|
|
||
| // Register the metrics: | ||
| func RegisterGRPCMetrics() { | ||
| prometheus.MustRegister(grpcStartCountMetric) | ||
| prometheus.MustRegister(grpcHandledCountMetric) | ||
| prometheus.MustRegister(grpcHandledDurationMetric) | ||
| prometheus.MustRegister(grpcMsgReceivedCountMetric) | ||
| prometheus.MustRegister(grpcMsgSentCountMetric) | ||
| } | ||
|
|
||
| // Unregister the metrics: | ||
| func UnregisterGRPCMetrics() { | ||
| prometheus.Unregister(grpcStartCountMetric) | ||
| prometheus.Unregister(grpcHandledCountMetric) | ||
| prometheus.Unregister(grpcHandledDurationMetric) | ||
| prometheus.Unregister(grpcMsgReceivedCountMetric) | ||
| prometheus.Unregister(grpcMsgSentCountMetric) | ||
| } | ||
|
|
||
| // Reset the metrics: | ||
| func ResetGRPCMetrics() { | ||
| grpcStartCountMetric.Reset() | ||
| grpcHandledCountMetric.Reset() | ||
| grpcHandledDurationMetric.Reset() | ||
| grpcMsgReceivedCountMetric.Reset() | ||
| grpcMsgSentCountMetric.Reset() | ||
| } | ||
|
|
||
| // Description of the gRPC start count metric: | ||
| var grpcStartCountMetric = prometheus.NewCounterVec( | ||
| prometheus.CounterOpts{ | ||
| Subsystem: grpcMetricsSubsystem, | ||
| Name: startCountMetric, | ||
| Help: "Total number of RPCs started on the server.", | ||
| }, | ||
| grpcMetricsLabels, | ||
| ) | ||
|
|
||
| // Description of the gRPC handled count metric: | ||
| var grpcHandledCountMetric = prometheus.NewCounterVec( | ||
| prometheus.CounterOpts{ | ||
| Subsystem: grpcMetricsSubsystem, | ||
| Name: handledCountMetric, | ||
| Help: "Total number of RPCs completed on the server, regardless of success or failure.", | ||
| }, | ||
| grpcMetricsAllLabels, | ||
| ) | ||
|
|
||
| // Description of the gRPC handled duration metric: | ||
| var grpcHandledDurationMetric = prometheus.NewHistogramVec( | ||
| prometheus.HistogramOpts{ | ||
| Subsystem: grpcMetricsSubsystem, | ||
| Name: handledDurationMetric, | ||
| Help: "Histogram of response latency (seconds) by the server.", | ||
| Buckets: prometheus.DefBuckets, | ||
| }, | ||
| grpcMetricsLabels, | ||
| ) | ||
|
|
||
| // Description of the gRPC message received count metric: | ||
| var grpcMsgReceivedCountMetric = prometheus.NewCounterVec( | ||
| prometheus.CounterOpts{ | ||
| Subsystem: grpcMetricsSubsystem, | ||
| Name: msgReceivedCountMetric, | ||
| Help: "Total number of cloudevents received on the server.", | ||
| }, | ||
| grpcMetricsLabels, | ||
| ) | ||
|
|
||
| // Description of the gRPC message sent count metric: | ||
| var grpcMsgSentCountMetric = prometheus.NewCounterVec( | ||
| prometheus.CounterOpts{ | ||
| Subsystem: grpcMetricsSubsystem, | ||
| Name: msgSentCountMetric, | ||
| Help: "Total number of cloudevents sent by the server.", | ||
| }, | ||
| grpcMetricsLabels, | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.