From 78faf8f504eef39166fdede98b897d3fe3fb0b5e Mon Sep 17 00:00:00 2001 From: Kenneth Tran Date: Tue, 16 Jun 2026 14:47:13 -0700 Subject: [PATCH] Add Prometheus metrics --- lib/services/expiry/expiry.go | 91 ++++++++++++++++++++++-- lib/services/expiry/expiry_test.go | 109 +++++++++++++++++++++++++++++ 2 files changed, 193 insertions(+), 7 deletions(-) diff --git a/lib/services/expiry/expiry.go b/lib/services/expiry/expiry.go index b20e66c7072..0fe9110184a 100644 --- a/lib/services/expiry/expiry.go +++ b/lib/services/expiry/expiry.go @@ -25,13 +25,16 @@ import ( "time" "github.com/gravitational/trace" + "github.com/prometheus/client_golang/prometheus" "golang.org/x/sync/errgroup" + "github.com/gravitational/teleport" "github.com/gravitational/teleport/api/types" apievents "github.com/gravitational/teleport/api/types/events" "github.com/gravitational/teleport/api/utils/clientutils" "github.com/gravitational/teleport/api/utils/retryutils" "github.com/gravitational/teleport/lib/events" + "github.com/gravitational/teleport/lib/observability/metrics" "github.com/gravitational/teleport/lib/services" "github.com/gravitational/teleport/lib/tlsca" "github.com/gravitational/teleport/lib/utils/interval" @@ -52,8 +55,45 @@ const ( // maxExpiresPerCycle is an arbitrary limit on the number of resources to expire per cycle // to prevent any one auth server holding the lease for more than a couple of minutes. maxExpiresPerCycle = 120 + + // metricsSubsystem groups Prometheus metrics emitted by this service. + metricsSubsystem = "expiry_service" + + // metricLabelResourceKind partitions the gauges by resource kind + // (e.g. "access_request", "app_session"). + metricLabelResourceKind = "resource_kind" ) +// expiryMetrics holds the Prometheus metrics emitted by the expiry service. +// A new instance is constructed in New() per Service so tests are isolated. +type expiryMetrics struct { + // expiredBeforeScan is the number of expired resources observed at the + // start of the most recent scan, partitioned by resource kind. + expiredBeforeScan *prometheus.GaugeVec + + // expiredAfterScan is the number of expired resources still awaiting + // deletion at the end of the most recent scan. + expiredAfterScan *prometheus.GaugeVec +} + +// newExpiryMetrics constructs a fresh set of unregistered metrics. +func newExpiryMetrics() *expiryMetrics { + return &expiryMetrics{ + expiredBeforeScan: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: teleport.MetricNamespace, + Subsystem: metricsSubsystem, + Name: "expired_resources_before_scan", + Help: "Number of expired resources at the start of scan", + }, []string{metricLabelResourceKind}), + expiredAfterScan: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: teleport.MetricNamespace, + Subsystem: metricsSubsystem, + Name: "expired_resources_after_scan", + Help: "Number of expired resources remaining at the end of scan", + }, []string{metricLabelResourceKind}), + } +} + // AccessPoint is the API used by the expiry service. type AccessPoint interface { // Semaphores provides semaphore operations @@ -113,6 +153,7 @@ type expiryTask struct { type Service struct { *Config expiryTasks []expiryTask + metrics *expiryMetrics } // New initializes an expiry service. @@ -121,8 +162,17 @@ func New(cfg *Config) (*Service, error) { return nil, trace.Wrap(err) } + m := newExpiryMetrics() + if err := metrics.RegisterPrometheusCollectors( + m.expiredBeforeScan, + m.expiredAfterScan, + ); err != nil { + return nil, trace.Wrap(err) + } + s := &Service{ - Config: cfg, + Config: cfg, + metrics: m, } intervalCfg := interval.Config{ @@ -236,7 +286,15 @@ func (s *Service) loop(ctx context.Context, task expiryTask) error { func (s *Service) processRequests(ctx context.Context) { s.Log.DebugContext(ctx, "Cleaning up expired access requests.") + // expiredBefore counts requests waiting to be expired + // requestsExpired counts only successful expirations + expiredBefore := 0 requestsExpired := 0 + defer func() { + s.metrics.expiredBeforeScan.WithLabelValues(types.KindAccessRequest).Set(float64(expiredBefore)) + s.metrics.expiredAfterScan.WithLabelValues(types.KindAccessRequest).Set(float64(expiredBefore - requestsExpired)) + }() + readTime := time.Now() for expiredAccessRequest, err := range clientutils.Resources(ctx, s.AccessPoint.ListExpiredAccessRequests) { if err != nil { @@ -251,16 +309,22 @@ func (s *Service) processRequests(ctx context.Context) { continue } } + expiredBefore++ + + // Keep iterating to count remaining requests but stop expiring them. + if expiredBefore > maxExpiresPerCycle { + continue + } - requestsExpired++ s.Log.DebugContext(ctx, "Expiring access request.", "request", expiredAccessRequest.GetName()) if err := s.expireRequest(ctx, expiredAccessRequest); err != nil { s.Log.ErrorContext(ctx, "Error expiring access request.", "error", err) continue } - if requestsExpired >= maxExpiresPerCycle { + requestsExpired++ + + if requestsExpired == maxExpiresPerCycle { s.Log.DebugContext(ctx, "Cleaned up maximum amount of expired access requests. Will continue in the next run.", "max", maxExpiresPerCycle) - return } } @@ -270,14 +334,27 @@ func (s *Service) processRequests(ctx context.Context) { func (s *Service) processAppSessions(ctx context.Context) { s.Log.DebugContext(ctx, "Cleaning up expired application sessions.") + // expiredBefore counts app sessions waiting to be expired + // sessionsExpired counts only successful expirations + expiredBefore := 0 sessionsExpired := 0 + defer func() { + s.metrics.expiredBeforeScan.WithLabelValues(types.KindAppSession).Set(float64(expiredBefore)) + s.metrics.expiredAfterScan.WithLabelValues(types.KindAppSession).Set(float64(expiredBefore - sessionsExpired)) + }() + for expiredSession, err := range clientutils.Resources(ctx, s.AccessPoint.ListExpiredAppSessions) { if err != nil { s.Log.ErrorContext(ctx, "Error listing expired application sessions.", "error", err) return } + expiredBefore++ + + // Keep iterating to count remaining sessions but stop expiring them. + if expiredBefore > maxExpiresPerCycle { + continue + } - sessionsExpired++ s.Log.DebugContext(ctx, "Expiring application session.", "user", expiredSession.GetUser(), "session_id", expiredSession.GetName()) @@ -286,10 +363,10 @@ func (s *Service) processAppSessions(ctx context.Context) { s.Log.ErrorContext(ctx, "Error expiring application session.", "error", err) continue } + sessionsExpired++ - if sessionsExpired >= maxExpiresPerCycle { + if sessionsExpired == maxExpiresPerCycle { s.Log.DebugContext(ctx, "Cleaned up maximum amount of expired application sessions. Will continue in the next run.", "max", maxExpiresPerCycle) - return } } s.Log.DebugContext(ctx, "Successfully cleaned up expired application sessions.", "count", sessionsExpired) diff --git a/lib/services/expiry/expiry_test.go b/lib/services/expiry/expiry_test.go index a3f375f2f21..84782ebb59b 100644 --- a/lib/services/expiry/expiry_test.go +++ b/lib/services/expiry/expiry_test.go @@ -20,11 +20,13 @@ package expiry import ( "context" + "fmt" "testing" "testing/synctest" "time" "github.com/google/uuid" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" "github.com/gravitational/teleport/api/client/proto" @@ -340,6 +342,113 @@ func TestAppSessionExpiryInterval(t *testing.T) { }) } +// metricsTestResource bundles the per-kind helpers needed by TestExpiryMetrics +// so the table can stay declarative. +type metricsTestResource struct { + // label is the value used for the resource_kind gauge label. + label string + // taskIndex is the position of this resource's task in expiryTasks. + taskIndex int + // create produces one resource that is expired (or expires within ns of + // the scan firing). The index argument disambiguates resources that need + // unique names (e.g. app sessions tied to distinct users). + create func(t *testing.T, authServer *auth.Server, i int) + // remaining returns the number of resources of this kind still in the + // backend. + remaining func(t *testing.T, authServer *auth.Server) int +} + +var ( + appSessionMetricsResource = metricsTestResource{ + label: types.KindAppSession, + taskIndex: 1, + create: func(t *testing.T, authServer *auth.Server, i int) { + createAppSession(t, authServer, fmt.Sprintf("user-%d", i), time.Now().Add(1)) + }, + remaining: func(t *testing.T, authServer *auth.Server) int { + return len(mustListAppSessions(t, authServer)) + }, + } + accessRequestMetricsResource = metricsTestResource{ + label: types.KindAccessRequest, + taskIndex: 0, + create: func(t *testing.T, authServer *auth.Server, _ int) { + _ = createAccessRequest(t, authServer, types.RequestState_DENIED, time.Now().Add(1)) + }, + remaining: func(t *testing.T, authServer *auth.Server) int { + return len(mustListAccessRequests(t, authServer)) + }, + } +) + +// TestExpiryMetrics verifies the before-scan and after-scan gauges for both +// resource kinds and the behavior when expired resources exceed maxExpiresPerCycle. +func TestExpiryMetrics(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + resource metricsTestResource + numCreate int + expectProcessed int // capped by maxExpiresPerCycle + }{ + { + name: "app session: all processed", + resource: appSessionMetricsResource, + numCreate: 3, + expectProcessed: 3, + }, + { + name: "access request: all processed", + resource: accessRequestMetricsResource, + numCreate: 3, + expectProcessed: 3, + }, + { + name: "access request: cap hit, remainder expected", + resource: accessRequestMetricsResource, + numCreate: maxExpiresPerCycle + 5, + expectProcessed: maxExpiresPerCycle, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + const testInterval = time.Hour + + expiry, authServer, emitter := setupExpiryService(t, true) + expiry.expiryTasks[tc.resource.taskIndex].intervalCfg = interval.Config{ + Duration: testInterval, + FirstDuration: testInterval, + } + runExpiryBackground(t, func(ctx context.Context) error { + return expiry.run(ctx, expiry.expiryTasks[tc.resource.taskIndex]) + }) + + for i := range tc.numCreate { + tc.resource.create(t, authServer, i) + } + require.Equal(t, tc.numCreate, tc.resource.remaining(t, authServer)) + + // Trigger exactly one sweep. + time.Sleep(testInterval) + synctest.Wait() + + expectAfter := tc.numCreate - tc.expectProcessed + require.Equal(t, expectAfter, tc.resource.remaining(t, authServer)) + require.Len(t, emitter.Events(), tc.expectProcessed) + + beforeMetric := expiry.metrics.expiredBeforeScan.WithLabelValues(tc.resource.label) + afterMetric := expiry.metrics.expiredAfterScan.WithLabelValues(tc.resource.label) + require.Equal(t, float64(tc.numCreate), testutil.ToFloat64(beforeMetric)) + require.Equal(t, float64(expectAfter), testutil.ToFloat64(afterMetric)) + }) + }) + } +} + // TestAppSessionExpiryTaskRegistration verifies that the app session expiry // task is only registered when the opt-in flag is enabled. func TestAppSessionExpiryTaskRegistration(t *testing.T) {