Add metrics for IP-to-services lookups (#997)

Adds a `service_lookups_total` metric which tracks IP-to-services lookups (via
entity cache). To differentiate between successful and failed lookups,
status=OK/Error label can be used.

Resolves #882

pkg/metrics/schema.go

@@ -68,6 +68,9 @@ const (
 	// TokenBucketAvailableMetricName - a gauge that tracks the number of tokens available in token bucket.
 	TokenBucketAvailableMetricName = "token_bucket_available_tokens_total"
 
+	// ServiceLookupsMetricName - counter for IP to services lookups.
+	ServiceLookupsMetricName = "service_lookups_total"
+
 	// FlowControlRequestsMetricName - counter for Check requests for flowcontrol.
 	FlowControlRequestsMetricName = "flowcontrol_requests_total"
 	// FlowControlDecisionsMetricName - counter for Check requests per decision type.
@@ -115,6 +118,12 @@ const (
 	MethodLabel = "http_method"
 	// HandlerName - name of the http handler. Defaults to 'default'.
 	HandlerName = "handler_name"
+	// ServiceLookupsStatusLabel - status for ServiceLookupsMetricName.
+	ServiceLookupsStatusLabel = "status"
+	// ServiceLookupsStatusOK - service lookup status OK.
+	ServiceLookupsStatusOK = FlowStatusOK
+	// ServiceLookupsStatusError - service lookup status Error.
+	ServiceLookupsStatusError = FlowStatusError
 	// FlowStatusLabel - flow status.
 	FlowStatusLabel = "flow_status"
 	// FlowStatusOK - flow status OK.

pkg/policies/flowcontrol/provide.go

@@ -17,8 +17,8 @@ func Module() fx.Option {
 		fluxmeter.Module(),
 		classifier.Module(),
 		service.Module(),
+		servicegetter.Module,
 		fx.Provide(
-			servicegetter.ProvideFromEntityCache,
 			NewEngine,
 		),
 	)

pkg/policies/flowcontrol/servicegetter/metrics.go

@@ -0,0 +1,43 @@
+package servicegetter
+
+import (
+	"github.com/fluxninja/aperture/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Metrics is used for collecting metrics about servicegetter
+//
+// nil value of Metrics should be always usable.
+type Metrics struct {
+	okTotal     prometheus.Counter
+	errorsTotal prometheus.Counter
+}
+
+// NewMetrics creates new Metrics, registering counters in given registry.
+func NewMetrics(registry *prometheus.Registry) (*Metrics, error) {
+	total := prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: metrics.ServiceLookupsMetricName,
+			Help: "Number of IP to services lookups",
+		},
+		[]string{metrics.ServiceLookupsStatusLabel},
+	)
+	if err := registry.Register(total); err != nil {
+		return nil, err
+	}
+	return &Metrics{
+		okTotal:     total.WithLabelValues(metrics.ServiceLookupsStatusOK),
+		errorsTotal: total.WithLabelValues(metrics.ServiceLookupsStatusError),
+	}, nil
+}
+
+func (m *Metrics) inc(ok bool) {
+	if m == nil {
+		return
+	}
+	if ok {
+		m.okTotal.Inc()
+	} else {
+		m.errorsTotal.Inc()
+	}
+}

pkg/policies/flowcontrol/servicegetter/provide.go

@@ -0,0 +1,9 @@
+package servicegetter
+
+import "go.uber.org/fx"
+
+// Module is a set of default providers for servicegetter components.
+var Module = fx.Options(
+	fx.Provide(ProvideFromEntityCache),
+	fx.Provide(NewMetrics),
+)

pkg/policies/flowcontrol/servicegetter/servicegetter.go

@@ -27,27 +27,34 @@ func NewEmpty() ServiceGetter { return emptyServiceGetter{} }
 type ecServiceGetter struct {
 	entityCache    *entitycache.EntityCache
 	ecHasDiscovery bool
+	metrics        *Metrics
 }
 
 // ServicesFromContext returns list of services associated with IP extracted from context
 //
 // The returned list of services depends only on state of entityCache.
 // However, emitted warnings will depend on whether service discovery is enabled or not.
 func (sg *ecServiceGetter) ServicesFromContext(ctx context.Context) []string {
+	svcs, ok := sg.servicesFromContext(ctx)
+	sg.metrics.inc(ok)
+	return svcs
+}
+
+func (sg *ecServiceGetter) servicesFromContext(ctx context.Context) (svcs []string, ok bool) {
 	rpcPeer, peerExists := peer.FromContext(ctx)
 	if !peerExists {
 		if sg.ecHasDiscovery {
 			log.Bug().Msg("cannot get client info from context")
 		}
-		return nil
+		return nil, false
 	}
 
 	tcpAddr, isTCPAddr := rpcPeer.Addr.(*net.TCPAddr)
 	if !isTCPAddr {
 		if sg.ecHasDiscovery {
 			log.Bug().Msg("client addr is not TCP")
 		}
-		return nil
+		return nil, false
 
 	}
 
@@ -58,27 +65,35 @@ func (sg *ecServiceGetter) ServicesFromContext(ctx context.Context) []string {
 			log.Sample(noEntitySampler).Warn().Err(err).Str("clientIP", clientIP).
 				Msg("cannot get services")
 		}
-		return nil
+		return nil, false
 	}
 
-	return entity.Services
+	return entity.Services, true
 }
 
 var noEntitySampler = log.NewRatelimitingSampler()
 
+// FxIn are FX arguments to ProvideFromEntityCache.
+type FxIn struct {
+	fx.In
+	Lifecycle      fx.Lifecycle
+	EntityCache    *entitycache.EntityCache
+	EntityTrackers *entitycache.EntityTrackers
+	Metrics        *Metrics `optional:"true"`
+}
+
 // ProvideFromEntityCache provides an EntityCache-powered ServiceGetter.
-func ProvideFromEntityCache(
-	entityCache *entitycache.EntityCache,
-	entityTrackers *entitycache.EntityTrackers,
-	lc fx.Lifecycle,
-) ServiceGetter {
-	sg := &ecServiceGetter{entityCache: entityCache}
-
-	lc.Append(fx.Hook{
+func ProvideFromEntityCache(in FxIn) ServiceGetter {
+	sg := &ecServiceGetter{
+		entityCache: in.EntityCache,
+		metrics:     in.Metrics,
+	}
+
+	in.Lifecycle.Append(fx.Hook{
 		OnStart: func(context.Context) error {
 			// Checking this flag on OnStart so that all registrations done in
 			// provide/invoke stage would be visible.
-			sg.ecHasDiscovery = entityTrackers.HasDiscovery()
+			sg.ecHasDiscovery = in.EntityTrackers.HasDiscovery()
 			return nil
 		},
 		OnStop: func(context.Context) error { return nil },