feat: Add missing-key metrics for online feature lookups (#359)

* feat: Add missing-key metrics for online feature lookups (Go + Python)

Emit DogStatsD counters (feature_lookup_not_found, feature_lookup_null_or_expired)
when online store reads return NOT_FOUND, NULL_VALUE, or OUTSIDE_MAX_AGE statuses.
Gated behind ENABLE_MISSING_KEY_METRICS env var

Author: vanitabhagwat

go/embedded/online_features.go

@@ -311,7 +311,7 @@ func (s *OnlineFeatureService) StartGrpcServerWithLogging(host string, port int,
 	if err != nil {
 		return err
 	}
-	ser := server.NewGrpcServingServiceServer(s.fs, loggingService)
+	ser := server.NewGrpcServingServiceServer(s.fs, loggingService, nil, nil)
 	log.Printf("Starting a gRPC server on host %s port %d\n", host, port)
 	lis, err := net.Listen("tcp", fmt.Sprintf("%s:%d", host, port))
 	if err != nil {
@@ -366,7 +366,7 @@ func (s *OnlineFeatureService) StartHttpServerWithLogging(host string, port int,
 	if err != nil {
 		return err
 	}
-	ser := server.NewHttpServer(s.fs, loggingService)
+	ser := server.NewHttpServer(s.fs, loggingService, nil, nil)
 	log.Printf("Starting a HTTP server on host %s port %d\n", host, port)
 
 	go func() {

go/internal/feast/metrics/client.go

@@ -0,0 +1,12 @@
+package metrics
+
+// StatsdClient wraps DogStatsD so tests can inject a fake.
+// The real github.com/DataDog/datadog-go/v5/statsd.Client satisfies this interface.
+type StatsdClient interface {
+	Count(name string, value int64, tags []string, rate float64) error
+}
+
+// NoOpStatsdClient does nothing when metrics are disabled.
+type NoOpStatsdClient struct{}
+
+func (n *NoOpStatsdClient) Count(string, int64, []string, float64) error { return nil }

go/internal/feast/metrics/config.go

@@ -0,0 +1,37 @@
+package metrics
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/feast-dev/feast/go/internal/feast/registry"
+)
+
+func IsMissingKeyMetricsEnabled() bool {
+	return strings.ToLower(os.Getenv("ENABLE_MISSING_KEY_METRICS")) == "true"
+}
+
+func GetOnlineStoreType(config *registry.RepoConfig) string {
+	if storeType, ok := config.OnlineStore["type"]; ok {
+		return fmt.Sprintf("%v", storeType)
+	}
+	return "unknown"
+}
+
+// GetStatsDAddress returns the DogStatsD address from environment variables.
+// Returns empty string if DD_AGENT_HOST is not set.
+// Port can be configured via DD_DOGSTATSD_PORT (defaults to 8125).
+func GetStatsDAddress() string {
+	host := os.Getenv("DD_AGENT_HOST")
+	if host == "" {
+		return ""
+	}
+
+	port := os.Getenv("DD_DOGSTATSD_PORT")
+	if port == "" {
+		port = "8125"
+	}
+
+	return fmt.Sprintf("%s:%s", host, port)
+}

go/internal/feast/metrics/config_test.go

@@ -0,0 +1,70 @@
+package metrics
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetStatsDAddress(t *testing.T) {
+	tests := []struct {
+		name     string
+		host     string
+		port     string
+		expected string
+	}{
+		{
+			name:     "no env vars set",
+			host:     "",
+			port:     "",
+			expected: "",
+		},
+		{
+			name:     "only host set, default port",
+			host:     "datadog-agent",
+			port:     "",
+			expected: "datadog-agent:8125",
+		},
+		{
+			name:     "host and custom port",
+			host:     "datadog-agent",
+			port:     "9125",
+			expected: "datadog-agent:9125",
+		},
+		{
+			name:     "localhost with default port",
+			host:     "localhost",
+			port:     "",
+			expected: "localhost:8125",
+		},
+		{
+			name:     "IP address with custom port",
+			host:     "10.0.0.5",
+			port:     "8126",
+			expected: "10.0.0.5:8126",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Set env vars
+			if tt.host != "" {
+				os.Setenv("DD_AGENT_HOST", tt.host)
+				defer os.Unsetenv("DD_AGENT_HOST")
+			} else {
+				os.Unsetenv("DD_AGENT_HOST")
+			}
+
+			if tt.port != "" {
+				os.Setenv("DD_DOGSTATSD_PORT", tt.port)
+				defer os.Unsetenv("DD_DOGSTATSD_PORT")
+			} else {
+				os.Unsetenv("DD_DOGSTATSD_PORT")
+			}
+
+			result := GetStatsDAddress()
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}

go/internal/feast/metrics/lookup_metrics.go

@@ -0,0 +1,140 @@
+package metrics
+
+import (
+	"math/rand"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/feast-dev/feast/go/internal/feast/onlineserving"
+	"github.com/feast-dev/feast/go/protos/feast/serving"
+)
+
+// extractFeatureView extracts the feature view name from a full feature name.
+// Feature names follow the format: feature_view__feature_name
+// Example: "hotel_fv__price" -> "hotel_fv"
+func extractFeatureView(featureName string) string {
+	parts := strings.SplitN(featureName, "__", 2)
+	if len(parts) == 2 {
+		return parts[0]
+	}
+	return "unknown"
+}
+
+type LookupMetricsAggregator struct {
+	notFound      map[string]int64
+	nullOrExpired map[string]int64
+	project       string
+	onlineStore   string
+	client        StatsdClient
+	sampleRate    float64
+}
+
+func NewLookupMetricsAggregator(
+	project, onlineStore string,
+	client StatsdClient,
+) *LookupMetricsAggregator {
+	if client == nil {
+		return nil
+	}
+
+	// Read sampling rate from environment (default: 1.0 = no sampling)
+	sampleRate := 1.0
+	if rateStr := os.Getenv("FEAST_METRICS_SAMPLE_RATE"); rateStr != "" {
+		if rate, err := strconv.ParseFloat(rateStr, 64); err == nil {
+			if rate > 0 && rate <= 1.0 {
+				sampleRate = rate
+			}
+		}
+	}
+
+	return &LookupMetricsAggregator{
+		notFound:      make(map[string]int64),
+		nullOrExpired: make(map[string]int64),
+		project:       project,
+		onlineStore:   onlineStore,
+		client:        client,
+		sampleRate:    sampleRate,
+	}
+}
+
+func (m *LookupMetricsAggregator) Record(featureID string, status serving.FieldStatus) {
+	if m == nil {
+		return
+	}
+	switch status {
+	case serving.FieldStatus_NOT_FOUND:
+		m.notFound[featureID]++
+	case serving.FieldStatus_NULL_VALUE, serving.FieldStatus_OUTSIDE_MAX_AGE:
+		m.nullOrExpired[featureID]++
+	}
+}
+
+func (m *LookupMetricsAggregator) RecordFromFeatureVectors(vectors []*onlineserving.FeatureVector) {
+	if m == nil {
+		return
+	}
+	for _, vector := range vectors {
+		for _, status := range vector.Statuses {
+			m.Record(vector.Name, status)
+		}
+	}
+}
+
+func (m *LookupMetricsAggregator) RecordFromRangeFeatureVectors(vectors []*onlineserving.RangeFeatureVector) {
+	if m == nil {
+		return
+	}
+	for _, vector := range vectors {
+		for _, entityStatuses := range vector.RangeStatuses {
+			for _, status := range entityStatuses {
+				m.Record(vector.Name, status)
+			}
+		}
+	}
+}
+
+func (m *LookupMetricsAggregator) Emit() {
+	if m == nil || m.client == nil {
+		return
+	}
+
+	// Probabilistic sampling: skip this request's metrics based on sample_rate
+	if m.sampleRate < 1.0 && rand.Float64() > m.sampleRate {
+		return
+	}
+
+	// Calculate multiplier to preserve statistical accuracy
+	// If sampleRate=0.1, we only emit 10% of the time, so multiply counts by 10
+	multiplier := 1.0 / m.sampleRate
+
+	baseTags := []string{
+		"project:" + m.project,
+		"online_store_type:" + m.onlineStore,
+	}
+
+	for featureID, count := range m.notFound {
+		if count == 0 {
+			continue
+		}
+		// Adjust count to preserve accuracy when sampling
+		adjustedCount := int64(float64(count) * multiplier)
+		tags := make([]string, len(baseTags)+2)
+		copy(tags, baseTags)
+		tags[len(baseTags)] = "feature:" + featureID
+		tags[len(baseTags)+1] = "feature_view:" + extractFeatureView(featureID)
+		m.client.Count("mlpfs.featureserver.feature_lookup_not_found", adjustedCount, tags, 1.0)
+	}
+
+	for featureID, count := range m.nullOrExpired {
+		if count == 0 {
+			continue
+		}
+		adjustedCount := int64(float64(count) * multiplier)
+		tags := make([]string, len(baseTags)+2)
+		copy(tags, baseTags)
+		tags[len(baseTags)] = "feature:" + featureID
+		tags[len(baseTags)+1] = "feature_view:" + extractFeatureView(featureID)
+		m.client.Count("mlpfs.featureserver.feature_lookup_null_or_expired", adjustedCount, tags, 1.0)
+	}
+}