feat: metrics (#1024)

* refactor: switch from opencensus to opentelemetry

* tempo works as designed nooooot

* fix: log traceids

* with grafana agent

* fix: http tracing

* fix: cleanup files

* chore: remove todo

* fix: bad test

* fix: ignore methods in grpc interceptors

* fix: remove test log

* clean up

* typo

* fix(config): configure tracing endpoint

* fix(span): add error id to span

* feat: metrics package

* feat: metrics package

* fix: counter

* fix: metric

* try metrics

* fix: coutner metrics

* fix: active sessin counter

* fix: active sessin counter

* fix: change current Sequence table

* fix: change current Sequence table

* fix: current sequences

* fix: spooler div metrics

* fix: console view

* fix: merge master

* fix: Last spool run on search result instead of eventtimestamp

* fix: go mod

* Update console/src/assets/i18n/de.json

Co-authored-by: Livio Amstutz <livio.a@gmail.com>

* fix: pr review

* fix: map

* update oidc pkg

* fix: handlers

* fix: value observer

* fix: remove fmt

* fix: handlers

* fix: tests

* fix: handler minimum cycle duration 1s

* fix(spooler): handler channel buffer

* fix interceptors

Co-authored-by: adlerhurst <silvan.reusser@gmail.com>
Co-authored-by: Livio Amstutz <livio.a@gmail.com>

internal/api/api.go

@@ -2,6 +2,12 @@ package api
 
 import (
 	"context"
+	admin_es "github.com/caos/zitadel/internal/admin/repository/eventsourcing"
+	auth_es "github.com/caos/zitadel/internal/auth/repository/eventsourcing"
+	"github.com/caos/zitadel/internal/telemetry/metrics"
+	"github.com/caos/zitadel/internal/telemetry/metrics/otel"
+	view_model "github.com/caos/zitadel/internal/view/model"
+	"go.opentelemetry.io/otel/api/metric"
 	"net/http"
 
 	"github.com/caos/logging"
@@ -16,7 +22,7 @@ import (
 	"github.com/caos/zitadel/internal/config/systemdefaults"
 	"github.com/caos/zitadel/internal/errors"
 	iam_model "github.com/caos/zitadel/internal/iam/model"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 type Config struct {
@@ -30,19 +36,33 @@ type API struct {
 	verifier       *authz.TokenVerifier
 	serverPort     string
 	health         health
+	auth           auth
+	admin          admin
 }
+
 type health interface {
 	Health(ctx context.Context) error
 	IamByID(ctx context.Context) (*iam_model.IAM, error)
 	VerifierClientID(ctx context.Context, appName string) (string, error)
 }
 
-func Create(config Config, authZ authz.Config, authZRepo *authz_es.EsRepository, sd systemdefaults.SystemDefaults) *API {
+type auth interface {
+	ActiveUserSessionCount() int64
+}
+
+type admin interface {
+	GetViews() ([]*view_model.View, error)
+	GetSpoolerDiv(database, viewName string) int64
+}
+
+func Create(config Config, authZ authz.Config, authZRepo *authz_es.EsRepository, authRepo *auth_es.EsRepository, adminRepo *admin_es.EsRepository, sd systemdefaults.SystemDefaults) *API {
 	api := &API{
 		serverPort: config.GRPC.ServerPort,
 	}
 	api.verifier = authz.Start(authZRepo)
 	api.health = authZRepo
+	api.auth = authRepo
+	api.admin = adminRepo
 	api.grpcServer = server.CreateServer(api.verifier, authZ, sd.DefaultLanguage)
 	api.gatewayHandler = server.CreateGatewayHandler(config.GRPC)
 	api.RegisterHandler("", api.healthHandler())
@@ -92,6 +112,7 @@ func (a *API) healthHandler() http.Handler {
 	handler.HandleFunc("/ready", handleReadiness(checks))
 	handler.HandleFunc("/validate", handleValidate(checks))
 	handler.HandleFunc("/clientID", a.handleClientID)
+	handler.Handle("/metrics", a.handleMetrics())
 
 	return handler
 }
@@ -132,6 +153,48 @@ func (a *API) handleClientID(w http.ResponseWriter, r *http.Request) {
 	http_util.MarshalJSON(w, id, nil, http.StatusOK)
 }
 
+func (a *API) handleMetrics() http.Handler {
+	a.registerActiveSessionCounters()
+	a.registerSpoolerDivCounters()
+	return metrics.GetExporter()
+}
+
+func (a *API) registerActiveSessionCounters() {
+	metrics.RegisterValueObserver(
+		metrics.ActiveSessionCounter,
+		metrics.ActiveSessionCounterDescription,
+		func(ctx context.Context, result metric.Int64ObserverResult) {
+			result.Observe(
+				a.auth.ActiveUserSessionCount(),
+			)
+		},
+	)
+}
+
+func (a *API) registerSpoolerDivCounters() {
+	views, err := a.admin.GetViews()
+	if err != nil {
+		logging.Log("API-3M8sd").WithError(err).Error("could not read views for metrics")
+		return
+	}
+	metrics.RegisterValueObserver(
+		metrics.SpoolerDivCounter,
+		metrics.SpoolerDivCounterDescription,
+		func(ctx context.Context, result metric.Int64ObserverResult) {
+			for _, view := range views {
+				labels := map[string]interface{}{
+					metrics.Database: view.Database,
+					metrics.ViewName: view.ViewName,
+				}
+				result.Observe(
+					a.admin.GetSpoolerDiv(view.Database, view.ViewName),
+					otel.MapToKeyValue(labels)...,
+				)
+			}
+		},
+	)
+}
+
 type ValidationFunction func(ctx context.Context) error
 
 func validate(ctx context.Context, validations []ValidationFunction) []error {

internal/api/authz/authorization.go

@@ -7,7 +7,7 @@ import (
 	"strings"
 
 	"github.com/caos/zitadel/internal/errors"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 const (

internal/api/authz/context.go

@@ -6,7 +6,7 @@ import (
 	"github.com/caos/zitadel/internal/api/grpc"
 	http_util "github.com/caos/zitadel/internal/api/http"
 	"github.com/caos/zitadel/internal/errors"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 type key int

internal/api/authz/permissions.go

@@ -4,7 +4,7 @@ import (
 	"context"
 
 	"github.com/caos/zitadel/internal/errors"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 func getUserMethodPermissions(ctx context.Context, t *TokenVerifier, requiredPerm string, authConfig Config) (_ context.Context, _ []string, err error) {

internal/api/authz/token.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 
 	caos_errs "github.com/caos/zitadel/internal/errors"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 const (

internal/api/grpc/admin/administrator.go

@@ -10,7 +10,7 @@ import (
 )
 
 func (s *Server) GetViews(ctx context.Context, _ *empty.Empty) (_ *admin.Views, err error) {
-	views, err := s.administrator.GetViews(ctx)
+	views, err := s.administrator.GetViews()
 	if err != nil {
 		return nil, err
 	}

internal/api/grpc/admin/administrator_converter.go

@@ -26,14 +26,17 @@ func failedEventsFromModel(failedEvents []*view_model.FailedEvent) []*admin.Fail
 }
 
 func viewFromModel(view *view_model.View) *admin.View {
-	timestamp, err := ptypes.TimestampProto(view.CurrentTimestamp)
+	eventTimestamp, err := ptypes.TimestampProto(view.EventTimestamp)
+	logging.Log("GRPC-KSo03").OnError(err).Debug("unable to parse timestamp")
+	lastSpool, err := ptypes.TimestampProto(view.EventTimestamp)
 	logging.Log("GRPC-KSo03").OnError(err).Debug("unable to parse timestamp")
 
 	return &admin.View{
-		Database:          view.Database,
-		ViewName:          view.ViewName,
-		ProcessedSequence: view.CurrentSequence,
-		ViewTimestamp:     timestamp,
+		Database:                 view.Database,
+		ViewName:                 view.ViewName,
+		ProcessedSequence:        view.CurrentSequence,
+		EventTimestamp:           eventTimestamp,
+		LastSuccessfulSpoolerRun: lastSpool,
 	}
 }
 

internal/api/grpc/admin/probes.go

@@ -2,7 +2,6 @@ package admin
 
 import (
 	"context"
-
 	"github.com/golang/protobuf/ptypes/empty"
 )
 

internal/api/grpc/auth/user_converter.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/caos/zitadel/internal/api/authz"
 	"github.com/caos/zitadel/internal/eventstore/models"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 	usr_model "github.com/caos/zitadel/internal/user/model"
 	"github.com/caos/zitadel/pkg/grpc/auth"
 	"github.com/caos/zitadel/pkg/grpc/message"

internal/api/grpc/server/gateway.go

@@ -14,7 +14,7 @@ import (
 	client_middleware "github.com/caos/zitadel/internal/api/grpc/client/middleware"
 	http_util "github.com/caos/zitadel/internal/api/http"
 	http_mw "github.com/caos/zitadel/internal/api/http/middleware"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 const (
@@ -129,7 +129,8 @@ func createDialOptions(g Gateway) []grpc.DialOption {
 }
 
 func addInterceptors(handler http.Handler, g Gateway) http.Handler {
-	handler = http_mw.DefaultTraceHandler(handler)
+	handler = http_mw.DefaultMetricsHandler(handler)
+	handler = http_mw.DefaultTelemetryHandler(handler)
 	handler = http_mw.NoCacheInterceptor(handler)
 	if interceptor, ok := g.(grpcGatewayCustomInterceptor); ok {
 		handler = interceptor.GatewayHTTPInterceptor(handler)

internal/api/grpc/server/middleware/auth_interceptor.go

@@ -10,7 +10,7 @@ import (
 	"github.com/caos/zitadel/internal/api/authz"
 	grpc_util "github.com/caos/zitadel/internal/api/grpc"
 	"github.com/caos/zitadel/internal/api/http"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 func AuthorizationInterceptor(verifier *authz.TokenVerifier, authConfig authz.Config) grpc.UnaryServerInterceptor {

internal/api/grpc/server/middleware/metrics_interceptor.go

@@ -0,0 +1,86 @@
+package middleware
+
+import (
+	"context"
+	"strings"
+
+	"github.com/grpc-ecosystem/grpc-gateway/runtime"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
+
+	_ "github.com/caos/zitadel/internal/statik"
+	"github.com/caos/zitadel/internal/telemetry/metrics"
+)
+
+const (
+	GrpcMethod                         = "grpc_method"
+	ReturnCode                         = "return_code"
+	GrpcRequestCounter                 = "grpc.server.request_counter"
+	GrpcRequestCounterDescription      = "Grpc request counter"
+	TotalGrpcRequestCounter            = "grpc.server.total_request_counter"
+	TotalGrpcRequestCounterDescription = "Total grpc request counter"
+	GrpcStatusCodeCounter              = "grpc.server.grpc_status_code"
+	GrpcStatusCodeCounterDescription   = "Grpc status code counter"
+)
+
+func MetricsHandler(metricTypes []metrics.MetricType, ignoredMethodSuffixes ...string) grpc.UnaryServerInterceptor {
+	return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		return RegisterMetrics(ctx, req, info, handler, metricTypes, ignoredMethodSuffixes...)
+	}
+}
+
+func RegisterMetrics(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler, metricTypes []metrics.MetricType, ignoredMethodSuffixes ...string) (_ interface{}, err error) {
+	if len(metricTypes) == 0 {
+		return handler(ctx, req)
+	}
+
+	for _, ignore := range ignoredMethodSuffixes {
+		if strings.HasSuffix(info.FullMethod, ignore) {
+			return handler(ctx, req)
+		}
+	}
+
+	resp, err := handler(ctx, req)
+	if containsMetricsMethod(metrics.MetricTypeRequestCount, metricTypes) {
+		RegisterGrpcRequestCounter(ctx, info)
+	}
+	if containsMetricsMethod(metrics.MetricTypeTotalCount, metricTypes) {
+		RegisterGrpcTotalRequestCounter(ctx)
+	}
+	if containsMetricsMethod(metrics.MetricTypeStatusCode, metricTypes) {
+		RegisterGrpcRequestCodeCounter(ctx, info, err)
+	}
+	return resp, err
+}
+
+func RegisterGrpcRequestCounter(ctx context.Context, info *grpc.UnaryServerInfo) {
+	var labels = map[string]interface{}{
+		GrpcMethod: info.FullMethod,
+	}
+	metrics.RegisterCounter(GrpcRequestCounter, GrpcRequestCounterDescription)
+	metrics.AddCount(ctx, GrpcRequestCounter, 1, labels)
+}
+
+func RegisterGrpcTotalRequestCounter(ctx context.Context) {
+	metrics.RegisterCounter(TotalGrpcRequestCounter, TotalGrpcRequestCounterDescription)
+	metrics.AddCount(ctx, TotalGrpcRequestCounter, 1, nil)
+}
+
+func RegisterGrpcRequestCodeCounter(ctx context.Context, info *grpc.UnaryServerInfo, err error) {
+	statusCode := status.Code(err)
+	var labels = map[string]interface{}{
+		GrpcMethod: info.FullMethod,
+		ReturnCode: runtime.HTTPStatusFromCode(statusCode),
+	}
+	metrics.RegisterCounter(GrpcStatusCodeCounter, GrpcStatusCodeCounterDescription)
+	metrics.AddCount(ctx, GrpcStatusCodeCounter, 1, labels)
+}
+
+func containsMetricsMethod(metricType metrics.MetricType, metricTypes []metrics.MetricType) bool {
+	for _, m := range metricTypes {
+		if m == metricType {
+			return true
+		}
+	}
+	return false
+}

internal/api/grpc/server/probes.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/caos/zitadel/internal/errors"
 	"github.com/caos/zitadel/internal/proto"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 type ValidationFunction func(ctx context.Context) error

internal/api/grpc/server/server.go

@@ -2,6 +2,8 @@ package server
 
 import (
 	"context"
+	grpc_api "github.com/caos/zitadel/internal/api/grpc"
+	"github.com/caos/zitadel/internal/telemetry/metrics"
 
 	"github.com/caos/logging"
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
@@ -11,7 +13,7 @@ import (
 	"github.com/caos/zitadel/internal/api/authz"
 	"github.com/caos/zitadel/internal/api/grpc/server/middleware"
 	"github.com/caos/zitadel/internal/api/http"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 const (
@@ -27,10 +29,12 @@ type Server interface {
 }
 
 func CreateServer(verifier *authz.TokenVerifier, authConfig authz.Config, lang language.Tag) *grpc.Server {
+	metricTypes := []metrics.MetricType{metrics.MetricTypeTotalCount, metrics.MetricTypeRequestCount, metrics.MetricTypeStatusCode}
 	return grpc.NewServer(
 		grpc.UnaryInterceptor(
 			grpc_middleware.ChainUnaryServer(
 				middleware.DefaultTracingServer(),
+				middleware.MetricsHandler(metricTypes, grpc_api.Probes...),
 				middleware.ErrorHandler(),
 				middleware.AuthorizationInterceptor(verifier, authConfig),
 				middleware.TranslationHandler(lang),

internal/api/http/middleware/metrics_interceptor.go

@@ -0,0 +1,19 @@
+package middleware
+
+import (
+	"github.com/caos/zitadel/internal/telemetry/metrics"
+	"net/http"
+
+	http_utils "github.com/caos/zitadel/internal/api/http"
+)
+
+func DefaultMetricsHandler(handler http.Handler) http.Handler {
+	metricTypes := []metrics.MetricType{metrics.MetricTypeTotalCount}
+	return MetricsHandler(metricTypes, http_utils.Probes...)(handler)
+}
+
+func MetricsHandler(metricTypes []metrics.MetricType, ignoredMethods ...string) func(http.Handler) http.Handler {
+	return func(handler http.Handler) http.Handler {
+		return metrics.NewMetricsHandler(handler, metricTypes, ignoredMethods...)
+	}
+}

internal/api/http/middleware/telemetry_interceptor.go

@@ -0,0 +1,18 @@
+package middleware
+
+import (
+	"github.com/caos/zitadel/internal/telemetry"
+	"net/http"
+
+	http_utils "github.com/caos/zitadel/internal/api/http"
+)
+
+func DefaultTelemetryHandler(handler http.Handler) http.Handler {
+	return TelemetryHandler(http_utils.Probes...)(handler)
+}
+
+func TelemetryHandler(ignoredMethods ...string) func(http.Handler) http.Handler {
+	return func(handler http.Handler) http.Handler {
+		return telemetry.TelemetryHandler(handler, ignoredMethods...)
+	}
+}

internal/api/http/middleware/trace_interceptor.go

@@ -1,18 +0,0 @@
-package middleware
-
-import (
-	"net/http"
-
-	http_utils "github.com/caos/zitadel/internal/api/http"
-	"github.com/caos/zitadel/internal/tracing"
-)
-
-func DefaultTraceHandler(handler http.Handler) http.Handler {
-	return TraceHandler(http_utils.Probes...)(handler)
-}
-
-func TraceHandler(ignoredMethods ...string) func(http.Handler) http.Handler {
-	return func(handler http.Handler) http.Handler {
-		return tracing.TraceHandler(handler, ignoredMethods...)
-	}
-}

internal/api/http/server.go

@@ -5,7 +5,7 @@ import (
 	"net/http"
 
 	"github.com/caos/logging"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 func Serve(ctx context.Context, handler http.Handler, port, servername string) {

internal/api/oidc/auth_request.go

@@ -13,7 +13,7 @@ import (
 	"github.com/caos/zitadel/internal/api/http/middleware"
 	"github.com/caos/zitadel/internal/errors"
 	proj_model "github.com/caos/zitadel/internal/project/model"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 	grant_model "github.com/caos/zitadel/internal/usergrant/model"
 )
 

internal/api/oidc/client.go

@@ -16,7 +16,7 @@ import (
 	"github.com/caos/zitadel/internal/crypto"
 	"github.com/caos/zitadel/internal/errors"
 	proj_model "github.com/caos/zitadel/internal/project/model"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 	user_model "github.com/caos/zitadel/internal/user/model"
 	grant_model "github.com/caos/zitadel/internal/usergrant/model"
 )

internal/api/oidc/op.go

@@ -2,6 +2,7 @@ package oidc
 
 import (
 	"context"
+	"github.com/caos/zitadel/internal/telemetry/metrics"
 	"time"
 
 	"github.com/caos/logging"
@@ -12,7 +13,7 @@ import (
 	"github.com/caos/zitadel/internal/auth/repository"
 	"github.com/caos/zitadel/internal/config/types"
 	"github.com/caos/zitadel/internal/id"
-	"github.com/caos/zitadel/internal/tracing"
+	"github.com/caos/zitadel/internal/telemetry/tracing"
 )
 
 type OPHandlerConfig struct {
@@ -55,12 +56,14 @@ func NewProvider(ctx context.Context, config OPHandlerConfig, repo repository.Re
 	cookieHandler, err := middleware.NewUserAgentHandler(config.UserAgentCookieConfig, id.SonyFlakeGenerator, localDevMode)
 	logging.Log("OIDC-sd4fd").OnError(err).WithField("traceID", tracing.TraceIDFromCtx(ctx)).Panic("cannot user agent handler")
 	config.OPConfig.CodeMethodS256 = true
+	metricTypes := []metrics.MetricType{metrics.MetricTypeRequestCount, metrics.MetricTypeStatusCode, metrics.MetricTypeTotalCount}
 	provider, err := op.NewOpenIDProvider(
 		ctx,
 		config.OPConfig,
 		newStorage(config.StorageConfig, repo),
 		op.WithHttpInterceptors(
-			middleware.TraceHandler(),
+			middleware.MetricsHandler(metricTypes),
+			middleware.TelemetryHandler(),
 			middleware.NoCacheInterceptor,
 			cookieHandler,
 			http_utils.CopyHeadersToContext,