Add Prometheus metrics

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)

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) {