feat(metrics_ingestion): implement SendMetrics function to send metrics to BigQuery

Author: leo-marble

models/metrics_collection.go

@@ -3,6 +3,7 @@ package models
 import (
 	"time"
 
+	"cloud.google.com/go/bigquery"
 	"github.com/google/uuid"
 )
 
@@ -34,6 +35,73 @@ type MetricsCollection struct {
 	LicenseKey   *string
 }
 
+// Bigquery schema for metrics
+type MetricEventRow struct {
+	StartTime    time.Time            `bigquery:"start_time"`
+	EndTime      time.Time            `bigquery:"end_time"`
+	DeploymentID uuid.UUID            `bigquery:"deployment_id"`
+	LicenseKey   bigquery.NullString  `bigquery:"license_key"`
+	OrgID        bigquery.NullString  `bigquery:"org_id"`
+	EventType    string               `bigquery:"event_type"`
+	Counter      bigquery.NullFloat64 `bigquery:"counter"`
+	Gauge        bigquery.NullFloat64 `bigquery:"gauge"`
+	Text         bigquery.NullString  `bigquery:"text"`
+}
+
+func AdaptMetricsCollection(metricsCollection MetricsCollection) []*MetricEventRow {
+	metricEventRows := make([]*MetricEventRow, 0, len(metricsCollection.Metrics))
+
+	licenseKey := bigquery.NullString{}
+	if metricsCollection.LicenseKey != nil {
+		licenseKey.StringVal = *metricsCollection.LicenseKey
+		licenseKey.Valid = true
+	}
+
+	for _, metric := range metricsCollection.Metrics {
+		orgID := bigquery.NullString{}
+		if metric.OrganizationID != nil {
+			orgID.StringVal = *metric.OrganizationID
+			orgID.Valid = true
+		}
+
+		counter := bigquery.NullFloat64{}
+		if metric.Numeric != nil {
+			counter.Float64 = *metric.Numeric
+			counter.Valid = true
+		}
+
+		text := bigquery.NullString{}
+		if metric.Text != nil {
+			text.StringVal = *metric.Text
+			text.Valid = true
+		}
+
+		startTime := metricsCollection.Timestamp
+		if metric.From != nil {
+			startTime = *metric.From
+		}
+
+		endTime := metricsCollection.Timestamp
+		if metric.To != nil {
+			endTime = *metric.To
+		}
+
+		metricEventRows = append(metricEventRows, &MetricEventRow{
+			StartTime:    startTime,
+			EndTime:      endTime,
+			DeploymentID: metricsCollection.DeploymentID,
+			LicenseKey:   licenseKey,
+			OrgID:        orgID,
+			EventType:    metric.Name,
+			Counter:      counter,
+			Gauge:        bigquery.NullFloat64{},
+			Text:         text,
+		})
+	}
+
+	return metricEventRows
+}
+
 func NewGlobalMetric(name string, numeric *float64, text *string, from, to *time.Time, frequency MetricCollectionFrequency) MetricData {
 	return MetricData{
 		Name:      name,

repositories/metrics_ingestion_repository.go

@@ -47,3 +47,25 @@ func (repo MetricsIngestionRepository) TestConnection(ctx context.Context) error
 	logger.InfoContext(ctx, "BigQuery connection test successful", "result", row[0])
 	return nil
 }
+
+func (repo MetricsIngestionRepository) SendMetrics(ctx context.Context, metrics models.MetricsCollection) error {
+	logger := utils.LoggerFromContext(ctx)
+
+	logger.DebugContext(ctx, "Sending metrics to BigQuery",
+		"collection_id", metrics.CollectionID,
+		"metrics_count", len(metrics.Metrics),
+	)
+
+	table := repo.bqClient.Client.Dataset("metrics").Table("events")
+	inserter := table.Inserter()
+
+	metricEventRows := models.AdaptMetricsCollection(metrics)
+
+	err := inserter.Put(ctx, metricEventRows)
+	if err != nil {
+		logger.ErrorContext(ctx, "Failed to send metrics to BigQuery", "error", err)
+		return fmt.Errorf("failed to send metrics to BigQuery: %w", err)
+	}
+
+	return nil
+}

usecases/health_usecase.go

@@ -20,9 +20,8 @@ type HealthUsecase struct {
 	executorFactory  executor_factory.ExecutorFactory
 	healthRepository healthRepository
 
-	openSanctionsRepository    OpenSanctionsHealthRepository
-	hasOpensanctionsSetup      bool
-	metricsIngestionRepository MetricsIngestionRepository
+	openSanctionsRepository OpenSanctionsHealthRepository
+	hasOpensanctionsSetup   bool
 }
 
 func (u *HealthUsecase) GetHealthStatus(ctx context.Context) models.HealthStatus {
@@ -43,15 +42,4 @@ func (u *HealthUsecase) GetHealthStatus(ctx context.Context) models.HealthStatus
 			Status: ok && err == nil,
 		})
 	}
-
-	// Check BigQuery health
-	err = u.metricsIngestionRepository.TestConnection(ctx)
-	statuses = append(statuses, models.HealthItemStatus{
-		Name:   models.BigQueryHealthItemName,
-		Status: err == nil,
-	})
-
-	return models.HealthStatus{
-		Statuses: statuses,
-	}
 }

usecases/metrics_ingestion.go

@@ -2,13 +2,14 @@ package usecases
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/checkmarble/marble-backend/models"
 	"github.com/checkmarble/marble-backend/utils"
 )
 
 type MetricsIngestionRepository interface {
-	TestConnection(ctx context.Context) error
+	SendMetrics(ctx context.Context, collection models.MetricsCollection) error
 }
 
 type MetricsIngestionUsecase struct {
@@ -21,24 +22,14 @@ func NewMetricsIngestionUsecase(repository MetricsIngestionRepository) MetricsIn
 	}
 }
 
-func (u *MetricsIngestionUsecase) IngestMetrics(ctx context.Context, metrics models.MetricsCollection) error {
-	// Ingest the collection
-	err := u.ingestCollection(ctx, metrics)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (u *MetricsIngestionUsecase) ingestCollection(ctx context.Context, collection models.MetricsCollection) error {
+func (u *MetricsIngestionUsecase) IngestMetrics(ctx context.Context, collection models.MetricsCollection) error {
 	logger := utils.LoggerFromContext(ctx)
-	logger.DebugContext(ctx, "Ingesting collection", "collection", collection)
+	logger.DebugContext(ctx, "Sending metrics to BigQuery", "collection", collection)
 
-	// TODO: Implement the ingestion logic
-	err := u.repository.TestConnection(ctx)
+	err := u.repository.SendMetrics(ctx, collection)
 	if err != nil {
-		return err
+		logger.ErrorContext(ctx, "Failed to send metrics to BigQuery", "error", err)
+		return fmt.Errorf("failed to send metrics to BigQuery: %w", err)
 	}
 
 	return nil

usecases/scheduled_execution/metrics_collection_job.go

@@ -30,7 +30,8 @@ func NewMetricsCollectionPeriodicJob(config infra.MetricCollectionConfig) *river
 				},
 			}
 		},
-		&river.PeriodicJobOpts{RunOnStart: false},
+		// TODO: RunOnstart should be false for production
+		&river.PeriodicJobOpts{RunOnStart: true},
 	)
 }
 

usecases/usecases.go

@@ -207,11 +207,10 @@ func (usecases *Usecases) NewLivenessUsecase() LivenessUsecase {
 
 func (usecases *Usecases) NewHealthUsecase() HealthUsecase {
 	return HealthUsecase{
-		executorFactory:            usecases.NewExecutorFactory(),
-		healthRepository:           &usecases.Repositories.MarbleDbRepository,
-		hasOpensanctionsSetup:      usecases.hasOpensanctionsSetup,
-		openSanctionsRepository:    &usecases.Repositories.OpenSanctionsRepository,
-		metricsIngestionRepository: usecases.Repositories.MetricsIngestionRepository,
+		executorFactory:         usecases.NewExecutorFactory(),
+		healthRepository:        &usecases.Repositories.MarbleDbRepository,
+		hasOpensanctionsSetup:   usecases.hasOpensanctionsSetup,
+		openSanctionsRepository: &usecases.Repositories.OpenSanctionsRepository,
 	}
 }