-
Notifications
You must be signed in to change notification settings - Fork 74
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(server): return event count as prom metrics (#1135)
Signed-off-by: Peter Marton <[email protected]> Co-authored-by: Krisztian Gacsal <[email protected]>
- Loading branch information
1 parent
e394f7e
commit 7d35c53
Showing
16 changed files
with
900 additions
and
350 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
Oops, something went wrong.
Large diffs are not rendered by default.
Oops, something went wrong.
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
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,90 @@ | ||
| package debug | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| dto "github.com/prometheus/client_model/go" | ||
| "github.com/prometheus/common/expfmt" | ||
| "google.golang.org/protobuf/proto" | ||
| "google.golang.org/protobuf/types/known/timestamppb" | ||
|
|
||
| "github.com/openmeterio/openmeter/internal/streaming" | ||
| ) | ||
|
|
||
| // DebugConnector is a connector for debug metrics. | ||
| type DebugConnector interface { | ||
| GetDebugMetrics(ctx context.Context, namespace string) (string, error) | ||
| } | ||
|
|
||
| // debugConnector is the internal implementation of the DebugConnector interface. | ||
| type debugConnector struct { | ||
| streaming streaming.Connector | ||
| } | ||
|
|
||
| // NewDebugConnector creates a new DebugConnector. | ||
| func NewDebugConnector(streaming streaming.Connector) DebugConnector { | ||
| return &debugConnector{ | ||
| streaming: streaming, | ||
| } | ||
| } | ||
|
|
||
| // GetDebugMetrics returns metrics in an OpenMetrics (Prometheus) format for debugging purposes. | ||
| // It is useful to monitor the number of events ingested on the vendor side. | ||
| func (c *debugConnector) GetDebugMetrics(ctx context.Context, namespace string) (string, error) { | ||
| // Start from the beginning of the day | ||
| queryParams := streaming.CountEventsParams{ | ||
| From: time.Now().Truncate(time.Hour * 24).UTC(), | ||
| } | ||
|
|
||
| // Query events counts | ||
| rows, err := c.streaming.CountEvents(ctx, namespace, queryParams) | ||
| if err != nil { | ||
| return "", fmt.Errorf("connector count events: %w", err) | ||
| } | ||
|
|
||
| // Convert to Prometheus metrics | ||
| var metrics []*dto.Metric | ||
| for _, row := range rows { | ||
| metric := &dto.Metric{ | ||
| Label: []*dto.LabelPair{ | ||
| { | ||
| Name: proto.String("subject"), | ||
| Value: proto.String(row.Subject), | ||
| }, | ||
| }, | ||
| Counter: &dto.Counter{ | ||
| // We can lose precision here | ||
| Value: proto.Float64(float64(row.Count)), | ||
| CreatedTimestamp: timestamppb.New(time.Now()), | ||
| }, | ||
| } | ||
|
|
||
| if row.IsError { | ||
| metric.Label = append(metric.Label, &dto.LabelPair{ | ||
| Name: proto.String("error"), | ||
| Value: proto.String("true"), | ||
| }) | ||
| } | ||
|
|
||
| metrics = append(metrics, metric) | ||
| } | ||
|
|
||
| family := &dto.MetricFamily{ | ||
| Name: proto.String("openmeter_events_total"), | ||
| Help: proto.String("Number of ingested events"), | ||
| Type: dto.MetricType_COUNTER.Enum(), | ||
| Unit: proto.String("events"), | ||
| Metric: metrics, | ||
| } | ||
|
|
||
| var out bytes.Buffer | ||
| _, err = expfmt.MetricFamilyToOpenMetrics(&out, family) | ||
| if err != nil { | ||
| return "", fmt.Errorf("convert metric family to OpenMetrics: %w", err) | ||
| } | ||
|
|
||
| return out.String(), nil | ||
| } |
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,89 @@ | ||
| package httpdriver | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "net/http" | ||
|
|
||
| "github.com/openmeterio/openmeter/internal/debug" | ||
| "github.com/openmeterio/openmeter/internal/namespace/namespacedriver" | ||
| "github.com/openmeterio/openmeter/pkg/framework/commonhttp" | ||
| "github.com/openmeterio/openmeter/pkg/framework/transport/httptransport" | ||
| "github.com/openmeterio/openmeter/pkg/models" | ||
| ) | ||
|
|
||
| type DebugHandler interface { | ||
| GetMetrics() GetMetricsHandler | ||
| } | ||
|
|
||
| type debugHandler struct { | ||
| namespaceDecoder namespacedriver.NamespaceDecoder | ||
| debugConnector debug.DebugConnector | ||
| options []httptransport.HandlerOption | ||
| } | ||
|
|
||
| func NewDebugHandler( | ||
| namespaceDecoder namespacedriver.NamespaceDecoder, | ||
| debugConnector debug.DebugConnector, | ||
| options ...httptransport.HandlerOption, | ||
| ) DebugHandler { | ||
| return &debugHandler{ | ||
| namespaceDecoder: namespaceDecoder, | ||
| debugConnector: debugConnector, | ||
| options: options, | ||
| } | ||
| } | ||
|
|
||
| type GetMetricsHandlerRequestParams struct { | ||
| Namespace string | ||
| } | ||
|
|
||
| type GetMetricsHandlerRequest struct { | ||
| params GetMetricsHandlerRequestParams | ||
| } | ||
| type GetMetricsHandlerResponse = string | ||
| type GetMetricsHandlerParams struct{} | ||
| type GetMetricsHandler httptransport.HandlerWithArgs[GetMetricsHandlerRequest, GetMetricsHandlerResponse, GetMetricsHandlerParams] | ||
|
|
||
| func (h *debugHandler) GetMetrics() GetMetricsHandler { | ||
| return httptransport.NewHandlerWithArgs[GetMetricsHandlerRequest, string, GetMetricsHandlerParams]( | ||
| func(ctx context.Context, r *http.Request, params GetMetricsHandlerParams) (GetMetricsHandlerRequest, error) { | ||
| ns, err := h.resolveNamespace(ctx) | ||
| if err != nil { | ||
| return GetMetricsHandlerRequest{}, err | ||
| } | ||
|
|
||
| return GetMetricsHandlerRequest{ | ||
| params: GetMetricsHandlerRequestParams{ | ||
| Namespace: ns, | ||
| }, | ||
| }, nil | ||
| }, | ||
| func(ctx context.Context, request GetMetricsHandlerRequest) (string, error) { | ||
| return h.debugConnector.GetDebugMetrics(ctx, request.params.Namespace) | ||
| }, | ||
| commonhttp.PlainTextResponseEncoder[string], | ||
| httptransport.AppendOptions( | ||
| h.options, | ||
| httptransport.WithErrorEncoder(func(ctx context.Context, err error, w http.ResponseWriter) bool { | ||
| if _, ok := err.(*models.GenericUserError); ok { | ||
| commonhttp.NewHTTPError( | ||
| http.StatusBadRequest, | ||
| err, | ||
| ).EncodeError(ctx, w) | ||
| return true | ||
| } | ||
| return false | ||
| }), | ||
| )..., | ||
| ) | ||
| } | ||
|
|
||
| func (h *debugHandler) resolveNamespace(ctx context.Context) (string, error) { | ||
| ns, ok := h.namespaceDecoder.GetNamespace(ctx) | ||
| if !ok { | ||
| return "", commonhttp.NewHTTPError(http.StatusInternalServerError, errors.New("internal server error")) | ||
| } | ||
|
|
||
| return ns, nil | ||
| } |
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,13 @@ | ||
| package router | ||
|
|
||
| import ( | ||
| "net/http" | ||
|
|
||
| "github.com/openmeterio/openmeter/internal/debug/httpdriver" | ||
| ) | ||
|
|
||
| // Get debug metrics | ||
| // (GET /api/v1/debug/metrics) | ||
| func (a *Router) GetDebugMetrics(w http.ResponseWriter, r *http.Request) { | ||
| a.debugHandler.GetMetrics().With(httpdriver.GetMetricsHandlerParams{}).ServeHTTP(w, r) | ||
| } |
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.