feat(metrics): Store all UMA metrics regardless of feature mapping (#1742)

* feat(metrics): Store all UMA metrics regardless of feature mapping

Previously, the UMA metric consumer would only store histogram metrics that could be successfully mapped to an existing web feature ID. This meant that metrics for new or unmapped features were discarded.

This change modifies the data ingestion logic to ensure all metrics are saved:

1.  All histogram metrics for a given day are now unconditionally saved to the `DailyChromiumHistogramMetrics` table.
2.  A separate step then attempts to associate these metrics with known web features in the `FeatureChromiumHistogramMetrics` table.
3.  The `LatestFeatureChromiumHistogramMetrics` table is now updated using a synchronization operation, ensuring it accurately reflects the latest data and handles removals correctly.

This separation makes the data ingestion more robust, preventing data loss and allowing feature mappings to be corrected in the future without losing historical metric data.

* manual changes

Author: jcscottiii

lib/gcpspanner/client.go

@@ -59,12 +59,6 @@ var (
 	// ErrUsageMetricUpsertNoHistogramFound indicates that the chromium histogram metric
 	// was not found when attempting to upsert a usage metric.
 	ErrUsageMetricUpsertNoHistogramFound = errors.New("histogram not found when upserting usage metric")
-
-	// ErrUsageMetricUpsertNoHistogramEnumFound indicates that the chromium histogram enum
-	// was not found when attempting to upsert a usage metric. This typically occurs when
-	// the histogram name associated with the metric is not found, possibly due to
-	// a draft or obsolete feature for which the corresponding enum ID has not been created.
-	ErrUsageMetricUpsertNoHistogramEnumFound = errors.New("histogram enum not found when upserting usage metric")
 )
 
 // entitySynchronizer specific errors.

lib/gcpspanner/daily_chromium_histogram_metrics.go

@@ -19,12 +19,10 @@ import (
 	"errors"
 	"log/slog"
 	"math/big"
-	"time"
 
 	"cloud.google.com/go/civil"
 	"cloud.google.com/go/spanner"
 	"github.com/GoogleChrome/webstatus.dev/lib/metricdatatypes"
-	"google.golang.org/api/iterator"
 )
 
 const dailyChromiumHistogramMetricsTable = "DailyChromiumHistogramMetrics"
@@ -46,179 +44,162 @@ type spannerDailyChromiumHistogramMetric struct {
 	ChromiumHistogramEnumValueID string `spanner:"ChromiumHistogramEnumValueID"`
 }
 
-// getLatestDailyChromiumMetricDate retrieves the Date of the latest usage stats for the given feature.
-func getLatestDailyChromiumMetricDate(
+// StoreDailyChromiumHistogramMetrics stores a slice of daily chromium histogram metrics in a bulk insert.
+func (c *Client) StoreDailyChromiumHistogramMetrics(
 	ctx context.Context,
-	txn *spanner.ReadWriteTransaction,
-	chromiumHistogramEnumValueID string) (*civil.Date, error) {
-	stmt := spanner.NewStatement(`
-        SELECT
-			dchm.Day as Date
-        FROM LatestDailyChromiumHistogramMetrics l
-        JOIN DailyChromiumHistogramMetrics dchm
-		ON l.ChromiumHistogramEnumValueID = dchm.ChromiumHistogramEnumValueID
-        WHERE l.ChromiumHistogramEnumValueID = @chromiumHistogramEnumValueID`)
-
-	stmt.Params = map[string]interface{}{
-		"chromiumHistogramEnumValueID": chromiumHistogramEnumValueID,
-	}
+	histogramName metricdatatypes.HistogramName,
+	metrics map[int64]DailyChromiumHistogramMetric) error {
+	chromiumHistogramEnumID, err := c.GetIDFromChromiumHistogramKey(ctx, string(histogramName))
+	if err != nil {
+		slog.ErrorContext(ctx, "unable to find histogram key id from histogram name", "name", string(histogramName))
 
-	iter := txn.Query(ctx, stmt)
-	defer iter.Stop()
+		return errors.Join(err, ErrUsageMetricUpsertNoHistogramFound)
+	}
 
-	row, err := iter.Next()
-	if err != nil {
-		if errors.Is(err, iterator.Done) {
-			// No row found, return zero time
-			zeroTime := time.Time{}
-			zeroDate := civil.DateOf(zeroTime)
+	producerFn := func(metricChan chan<- spannerDailyChromiumHistogramMetric) {
+		for bucketID, metric := range metrics {
+			chromiumHistogramEnumValueID, err := c.GetIDFromChromiumHistogramEnumValueKey(
+				ctx, *chromiumHistogramEnumID, bucketID)
+			if err != nil {
+				slog.WarnContext(ctx, "unable to find histogram value id. likely a draft or obsolete feature. will skip",
+					"id", *chromiumHistogramEnumID,
+					"bucketID", bucketID)
 
-			return &zeroDate, errors.Join(ErrQueryReturnedNoResults, err)
+				continue
+			}
+			metricChan <- spannerDailyChromiumHistogramMetric{
+				DailyChromiumHistogramMetric: metric,
+				ChromiumHistogramEnumValueID: *chromiumHistogramEnumValueID,
+			}
 		}
-		slog.ErrorContext(ctx, "error querying for latest run time", "error", err)
+	}
 
-		return nil, err
+	toMutationFn := func(m spannerDailyChromiumHistogramMetric) (*spanner.Mutation, error) {
+		return spanner.InsertOrUpdateStruct(dailyChromiumHistogramMetricsTable, m)
 	}
 
-	var date civil.Date
-	if err := row.Columns(&date); err != nil {
-		slog.ErrorContext(ctx, "error extracting date", "error", err)
+	return runConcurrentBatch(ctx, c, producerFn, dailyChromiumHistogramMetricsTable, toMutationFn)
+}
 
-		return nil, err
-	}
+// Implements the syncableEntityMapper interface for WebFeature and SpannerLatestDailyChromiumHistogramMetric.
+type latestDailyChromiumHistogramMetricMapper struct{}
 
-	return &date, nil
+// Key for the latestDailyChromiumHistogramMetricMapper.
+type latestDailyChromiumHistogramMetricKey struct {
+	WebFeatureID                 string
+	ChromiumHistogramEnumValueID string
 }
 
-func getWebFeatureIDByChromiumHistogramEnumValueID(
-	ctx context.Context,
-	txn *spanner.ReadWriteTransaction,
-	chromiumHistogramEnumValueID string) (*string, error) {
-	stmt := spanner.NewStatement(`
-		SELECT
-			WebFeatureID
-		FROM WebFeatureChromiumHistogramEnumValues
-		WHERE chromiumHistogramEnumValueID = @chromiumHistogramEnumValueID`)
+// Table returns the name of the Spanner table.
+func (m latestDailyChromiumHistogramMetricMapper) Table() string {
+	return LatestDailyChromiumHistogramMetricsTable
+}
 
-	stmt.Params = map[string]interface{}{
-		"chromiumHistogramEnumValueID": chromiumHistogramEnumValueID,
-	}
+// SelectAll returns a statement to select all LatestDailyChromiumHistogramMetrics.
+func (m latestDailyChromiumHistogramMetricMapper) SelectAll() spanner.Statement {
+	return spanner.NewStatement(`
+		SELECT
+			WebFeatureID, ChromiumHistogramEnumValueID, Day
+		FROM LatestDailyChromiumHistogramMetrics`)
+}
 
-	iter := txn.Query(ctx, stmt)
-	defer iter.Stop()
+// GetKeyFromExternal returns the business key from an external struct.
+func (m latestDailyChromiumHistogramMetricMapper) GetKeyFromExternal(
+	in SpannerLatestDailyChromiumHistogramMetric) latestDailyChromiumHistogramMetricKey {
+	return latestDailyChromiumHistogramMetricKey{in.WebFeatureID, in.ChromiumHistogramEnumValueID}
+}
 
-	row, err := iter.Next()
-	if err != nil {
-		if errors.Is(err, iterator.Done) {
-			return nil, errors.Join(ErrQueryReturnedNoResults, err)
-		}
-		slog.ErrorContext(ctx, "error querying for web feature ID", "error", err)
+// GetKeyFromInternal returns the business key from an internal Spanner struct.
+func (m latestDailyChromiumHistogramMetricMapper) GetKeyFromInternal(
+	in SpannerLatestDailyChromiumHistogramMetric) latestDailyChromiumHistogramMetricKey {
+	return latestDailyChromiumHistogramMetricKey{in.WebFeatureID, in.ChromiumHistogramEnumValueID}
+}
 
-		return nil, err
+// MergeAndCheckChanged will merge the entity and return if the entity has changed.
+func (m latestDailyChromiumHistogramMetricMapper) MergeAndCheckChanged(
+	in SpannerLatestDailyChromiumHistogramMetric,
+	existing SpannerLatestDailyChromiumHistogramMetric,
+) (SpannerLatestDailyChromiumHistogramMetric, bool) {
+	merged := SpannerLatestDailyChromiumHistogramMetric{
+		WebFeatureID:                 existing.WebFeatureID,
+		ChromiumHistogramEnumValueID: existing.ChromiumHistogramEnumValueID,
+		Day:                          in.Day,
 	}
-	var featureID string
-	if err := row.Columns(&featureID); err != nil {
-		slog.ErrorContext(ctx, "error extracting date", "error", err)
 
-		return nil, err
-	}
+	hasChanged := merged.Day != existing.Day
 
-	return &featureID, nil
+	return merged, hasChanged
 }
 
-// shouldUpsertLatestDailyChromiumUsageMetric determines whether the latest metric should be upserted based on
-// date comparison.
-func shouldUpsertLatestDailyChromiumUsageMetric(existingDate *civil.Date, newDate civil.Date) bool {
-	return existingDate == nil || existingDate.IsZero() || newDate.After(*existingDate)
+// GetChildDeleteKeyMutations is a no-op for this table.
+func (m latestDailyChromiumHistogramMetricMapper) GetChildDeleteKeyMutations(
+	_ context.Context,
+	_ *Client,
+	_ []SpannerLatestDailyChromiumHistogramMetric,
+) ([]ChildDeleteKeyMutations, error) {
+	return nil, nil
 }
 
-// UpsertDailyChromiumHistogramMetric upserts a daily chromium histogram metric.
-//
-// Errors:
-//   - ErrQueryReturnedNoResults: If the histogram key or value ID is not found.
-//   - ErrInternalQueryFailure: If any internal query fails during the process.
-//   - ErrUsageMetricUpsertNoFeatureIDFound: If no feature ID is found while
-//     attempting to upsert the latest daily chromium usage metric.
-//   - ErrUsageMetricUpsertNoHistogramFound: If the histogram is not found
-//   - ErrUsageMetricUpsertNoHistogramEnumFound: If a particular enum in the histogram is not found.
-func (c *Client) UpsertDailyChromiumHistogramMetric(
-	ctx context.Context,
-	histogramName metricdatatypes.HistogramName,
-	bucketID int64,
-	metric DailyChromiumHistogramMetric) error {
-	// TODO: When we have a generic way to do batch upserts, change this to accept an array of metrics.
-	chromiumHistogramEnumID, err := c.GetIDFromChromiumHistogramKey(ctx, string(histogramName))
-	if err != nil {
-		slog.ErrorContext(ctx, "unable to find histogram key id from histogram name", "name", string(histogramName))
-
-		return errors.Join(err, ErrUsageMetricUpsertNoHistogramFound)
-	}
-	chromiumHistogramEnumValueID, err := c.GetIDFromChromiumHistogramEnumValueKey(
-		ctx, *chromiumHistogramEnumID, bucketID)
-	if err != nil {
-		if errors.Is(err, ErrQueryReturnedNoResults) {
-			slog.WarnContext(ctx, "unable to find histogram value id. likely a draft or obsolete feature. will skip",
-				"id", *chromiumHistogramEnumID,
-				"bucketID", bucketID)
-
-			return errors.Join(err, ErrUsageMetricUpsertNoHistogramEnumFound)
-		}
+// DeleteMutation creates a Spanner delete mutation.
+func (m latestDailyChromiumHistogramMetricMapper) DeleteMutation(
+	in SpannerLatestDailyChromiumHistogramMetric) *spanner.Mutation {
+	return spanner.Delete(LatestDailyChromiumHistogramMetricsTable,
+		spanner.Key{in.WebFeatureID, in.ChromiumHistogramEnumValueID})
+}
 
-		slog.ErrorContext(ctx, "unable to find histogram value id",
-			"id", *chromiumHistogramEnumID,
-			"bucketID", bucketID)
+// SyncLatestDailyChromiumHistogramMetrics reconciles the LatestDailyChromiumHistogramMetrics table.
+func (c *Client) SyncLatestDailyChromiumHistogramMetrics(ctx context.Context) error {
+	slog.InfoContext(ctx, "Starting latest daily chromium histogram metrics synchronization")
+	synchronizer := newEntitySynchronizer[latestDailyChromiumHistogramMetricMapper,
+		SpannerLatestDailyChromiumHistogramMetric,
+		SpannerLatestDailyChromiumHistogramMetric,
+		latestDailyChromiumHistogramMetricKey](c)
 
+	desiredState, err := c.getDesiredLatestDailyChromiumHistogramMetrics(ctx)
+	if err != nil {
 		return err
 	}
 
-	_, err = c.ReadWriteTransaction(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
-		var mutations []*spanner.Mutation
-		m0, err := spanner.InsertOrUpdateStruct(
-			dailyChromiumHistogramMetricsTable,
-			spannerDailyChromiumHistogramMetric{
-				DailyChromiumHistogramMetric: metric,
-				ChromiumHistogramEnumValueID: *chromiumHistogramEnumValueID,
-			})
-		if err != nil {
-			return err
-		}
-		mutations = append(mutations, m0)
+	return synchronizer.Sync(ctx, desiredState)
+}
 
-		existingDate, err := getLatestDailyChromiumMetricDate(ctx, txn, *chromiumHistogramEnumValueID)
-		if err != nil {
-			if !errors.Is(err, ErrQueryReturnedNoResults) { // Handle errors other than "not found"
-				return errors.Join(ErrInternalQueryFailure, err)
-			}
-		}
+func (c *Client) getDesiredLatestDailyChromiumHistogramMetrics(
+	ctx context.Context) ([]SpannerLatestDailyChromiumHistogramMetric, error) {
+	txn := c.ReadOnlyTransaction()
+	defer txn.Close()
 
-		if shouldUpsertLatestDailyChromiumUsageMetric(existingDate, metric.Day) {
-			featureID, err := getWebFeatureIDByChromiumHistogramEnumValueID(ctx, txn, *chromiumHistogramEnumValueID)
-			if err != nil {
-				if errors.Is(err, ErrQueryReturnedNoResults) {
-					return errors.Join(err, ErrUsageMetricUpsertNoFeatureIDFound)
-				}
+	stmt := spanner.NewStatement(`
+		WITH LatestMetrics AS (
+			SELECT
+				ChromiumHistogramEnumValueID,
+				MAX(Day) AS MaxDay
+			FROM DailyChromiumHistogramMetrics
+			GROUP BY ChromiumHistogramEnumValueID
+		)
+		SELECT
+			w.WebFeatureID,
+			w.ChromiumHistogramEnumValueID,
+			l.MaxDay AS Day
+		FROM WebFeatureChromiumHistogramEnumValues w
+		JOIN LatestMetrics l ON w.ChromiumHistogramEnumValueID = l.ChromiumHistogramEnumValueID
+	`)
 
-				return errors.Join(ErrInternalQueryFailure, err)
-			}
-			m1, err := spanner.InsertOrUpdateStruct(
-				LatestDailyChromiumHistogramMetricsTable,
-				SpannerLatestDailyChromiumHistogramMetric{
-					WebFeatureID:                 *featureID,
-					ChromiumHistogramEnumValueID: *chromiumHistogramEnumValueID,
-					Day:                          metric.Day,
-				})
-			if err != nil {
-				return errors.Join(ErrInternalQueryFailure, err)
-			}
-			mutations = append(mutations, m1)
-		}
-		err = txn.BufferWrite(mutations)
-		if err != nil {
-			return errors.Join(ErrInternalQueryFailure, err)
+	iter := txn.Query(ctx, stmt)
+	defer iter.Stop()
+
+	var desiredState []SpannerLatestDailyChromiumHistogramMetric
+	err := iter.Do(func(r *spanner.Row) error {
+		var metric SpannerLatestDailyChromiumHistogramMetric
+		if err := r.ToStruct(&metric); err != nil {
+			return err
 		}
+		desiredState = append(desiredState, metric)
 
 		return nil
 	})
+	if err != nil {
+		return nil, err
+	}
 
-	return err
+	return desiredState, nil
 }