From 8865f7cb43610cb6b1591d52422839dd21219506 Mon Sep 17 00:00:00 2001 From: roroghost17 Date: Thu, 11 Jun 2026 18:40:13 +0530 Subject: [PATCH 1/2] fix: fixes customer FK column issue --- framework/configstore/migrations.go | 110 ++++++++++++++++++++++++++++ 1 file changed, 110 insertions(+) diff --git a/framework/configstore/migrations.go b/framework/configstore/migrations.go index 720930a0e0c..55a5bf06edc 100644 --- a/framework/configstore/migrations.go +++ b/framework/configstore/migrations.go @@ -880,6 +880,9 @@ func triggerMigrations(ctx context.Context, db *gorm.DB) error { if err := migrationAddCustomerNameUniqueConstraint(ctx, db); err != nil { return err } + if err := migrationNullLegacyCustomerBudgetID(ctx, db); err != nil { + return err + } return nil } @@ -9838,6 +9841,113 @@ func migrationAddCustomerBudgetsToBudgetsTable(ctx context.Context, db *gorm.DB) return nil } +// migrationNullLegacyCustomerBudgetID clears the legacy governance_customers.budget_id +// values left behind by migrationAddCustomerBudgetsToBudgetsTable. The column and its +// FK (fk_governance_customers_budget) are intentionally kept — dropping either is +// deferred to a major release — but rows still holding a value make DeleteCustomer's +// `DELETE FROM governance_budgets WHERE customer_id = ?` fail that FK check. Ownership +// already lives on governance_budgets.customer_id, so after a defensive backfill the +// legacy values can be nulled; a null reference satisfies the FK unconditionally. +func migrationNullLegacyCustomerBudgetID(ctx context.Context, db *gorm.DB) error { + m := migrator.New(db, migrator.DefaultOptions, []*migrator.Migration{{ + ID: "null_legacy_customer_budget_id_refs", + Migrate: func(tx *gorm.DB) error { + tx = tx.WithContext(ctx) + legacyExists, err := hasColumn(tx, "governance_customers", "budget_id") + if err != nil { + return fmt.Errorf("failed to introspect governance_customers for budget_id: %w", err) + } + if !legacyExists { + return nil + } + // Customers the defensive backfill below will attach a budget to. + // GenerateCustomerHash includes sorted budget IDs, so their stored + // config_hash goes stale once the budget is linked and must be refreshed. + var affectedCustomerIDs []string + if err := tx.Raw(` + SELECT DISTINCT c.id + FROM governance_customers c + JOIN governance_budgets b ON b.id = c.budget_id + WHERE b.customer_id IS NULL + AND b.virtual_key_id IS NULL AND b.team_id IS NULL + AND b.provider_config_id IS NULL AND b.model_config_id IS NULL + `).Scan(&affectedCustomerIDs).Error; err != nil { + return fmt.Errorf("failed to identify customers affected by budget backfill: %w", err) + } + // Defensive backfill (same shape as migrationAddCustomerBudgetsToBudgetsTable) + // in case a budget_id was written after that migration ran, e.g. by an older + // instance in a mixed-version cluster. Only claims budgets with no owner yet. + if err := tx.Exec(` + UPDATE governance_budgets SET customer_id = ( + SELECT id FROM governance_customers + WHERE governance_customers.budget_id = governance_budgets.id + ) WHERE customer_id IS NULL + AND virtual_key_id IS NULL AND team_id IS NULL + AND provider_config_id IS NULL AND model_config_id IS NULL + AND EXISTS ( + SELECT 1 FROM governance_customers + WHERE governance_customers.budget_id = governance_budgets.id + ) + `).Error; err != nil { + return fmt.Errorf("failed to backfill customer budget customer_id: %w", err) + } + // Refresh config_hash for customers whose budgets just got linked, keeping + // migration and runtime hash generation in parity (same as + // migrationAddCustomerBudgetsToBudgetsTable). + for _, customerID := range affectedCustomerIDs { + var customer tables.TableCustomer + if err := tx.Preload("Budgets").First(&customer, "id = ?", customerID).Error; err != nil { + return fmt.Errorf("failed to reload customer %s for hash refresh: %w", customerID, err) + } + hash, err := GenerateCustomerHash(customer) + if err != nil { + return fmt.Errorf("failed to generate hash for customer %s: %w", customerID, err) + } + if err := tx.Model(&tables.TableCustomer{}).Where("id = ?", customerID).Update("config_hash", hash).Error; err != nil { + return fmt.Errorf("failed to update hash for customer %s: %w", customerID, err) + } + } + if err := tx.Exec(`UPDATE governance_customers SET budget_id = NULL WHERE budget_id IS NOT NULL`).Error; err != nil { + return fmt.Errorf("failed to clear legacy governance_customers.budget_id values: %w", err) + } + return nil + }, + // Best-effort inverse: repopulate budget_id from governance_budgets.customer_id. + // The legacy column held a single value while the new model allows several + // budgets per customer, so for multi-budget customers the oldest budget is + // picked — for any customer that predates the pivot that is the original + // legacy budget, since later additions sort newer. + Rollback: func(tx *gorm.DB) error { + tx = tx.WithContext(ctx) + legacyExists, err := hasColumn(tx, "governance_customers", "budget_id") + if err != nil { + return fmt.Errorf("failed to introspect governance_customers for budget_id: %w", err) + } + if !legacyExists { + return nil + } + if err := tx.Exec(` + UPDATE governance_customers SET budget_id = ( + SELECT id FROM governance_budgets + WHERE governance_budgets.customer_id = governance_customers.id + ORDER BY created_at ASC, id ASC + LIMIT 1 + ) WHERE budget_id IS NULL AND EXISTS ( + SELECT 1 FROM governance_budgets + WHERE governance_budgets.customer_id = governance_customers.id + ) + `).Error; err != nil { + return fmt.Errorf("failed to restore legacy governance_customers.budget_id values: %w", err) + } + return nil + }, + }}) + if err := m.Migrate(); err != nil { + return fmt.Errorf("error running null_legacy_customer_budget_id_refs migration: %s", err.Error()) + } + return nil +} + // migrationAddMCPLibraryTable creates the mcp_library table, the synced-only // catalog of discoverable MCP servers. Rows are populated from the external MCP // library datasheet on a configurable interval (mirroring the model-pricing From 069a2a9c5df2c93caec79060db8c261fc40ff44d Mon Sep 17 00:00:00 2001 From: roroghost17 Date: Thu, 11 Jun 2026 19:35:34 +0530 Subject: [PATCH 2/2] feat: adds http metrics to OTEL --- docs/features/observability/otel.mdx | 4 ++++ plugins/otel/main.go | 24 ++++++++++++++++++++++ plugins/otel/metrics.go | 26 ++++++++++++++++-------- transports/bifrost-http/server/server.go | 25 +++++++++++++++++++++++ 4 files changed, 71 insertions(+), 8 deletions(-) diff --git a/docs/features/observability/otel.mdx b/docs/features/observability/otel.mdx index a5992962d88..2cb011dfa23 100644 --- a/docs/features/observability/otel.mdx +++ b/docs/features/observability/otel.mdx @@ -941,6 +941,10 @@ These are the same **Prometheus-style metrics** from the telemetry plugin, pushe | `bifrost_stream_inter_token_latency_seconds` | Histogram | Inter-token latency | | `http_requests_total` | Counter | Total HTTP requests | | `http_request_duration_seconds` | Histogram | HTTP request duration | +| `http_request_size_bytes` | Histogram | HTTP request body size | +| `http_response_size_bytes` | Histogram | HTTP response body size | + +> **Note:** Size metrics are only recorded when the `Content-Length` header is present. Requests or responses without it (e.g., chunked transfer encoding, streaming responses) do not produce data points in these histograms. ### OTEL Collector Configuration diff --git a/plugins/otel/main.go b/plugins/otel/main.go index d82a0576803..b56c42d3572 100644 --- a/plugins/otel/main.go +++ b/plugins/otel/main.go @@ -599,6 +599,30 @@ func (p *OtelPlugin) anyMetricsEnabled() bool { return false } +// RecordHTTPMetrics records HTTP-layer metrics (request count, duration, request/response +// sizes) against every profile's metrics exporter. The HTTP transport's middleware calls +// this once per completed request; it is a no-op when no profile has metrics enabled. +// Non-positive sizes are skipped (fasthttp reports -1 when Content-Length is unknown). +func (p *OtelPlugin) RecordHTTPMetrics(ctx context.Context, path, method, status string, durationSeconds, requestSizeBytes, responseSizeBytes float64) { + if !p.anyMetricsEnabled() { + return + } + attrs := BuildHTTPAttributes(path, method, status) + for _, t := range p.targets { + if t.metricsExporter == nil { + continue + } + t.metricsExporter.RecordHTTPRequest(ctx, attrs...) + t.metricsExporter.RecordHTTPRequestDuration(ctx, durationSeconds, attrs...) + if requestSizeBytes > 0 { + t.metricsExporter.RecordHTTPRequestSize(ctx, requestSizeBytes, attrs...) + } + if responseSizeBytes > 0 { + t.metricsExporter.RecordHTTPResponseSize(ctx, responseSizeBytes, attrs...) + } + } +} + // Inject receives a completed trace and sends it to the OTEL collector. // Implements schemas.ObservabilityPlugin interface. // This method is called asynchronously by TracingMiddleware after the response diff --git a/plugins/otel/metrics.go b/plugins/otel/metrics.go index ccde3635c02..2a6123c7d81 100644 --- a/plugins/otel/metrics.go +++ b/plugins/otel/metrics.go @@ -143,6 +143,14 @@ var ( interTokenLatencyBuckets = []float64{ .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10, } + + // httpBodySizeBuckets: HTTP request/response body sizes, 100B to 1GB + // (matches prometheus.ExponentialBuckets(100, 10, 8) on the Prometheus side). + // The SDK default boundaries top out at 10,000, which would collapse any + // payload over 10KB into +Inf. + httpBodySizeBuckets = []float64{ + 100, 1_000, 10_000, 100_000, 1_000_000, 10_000_000, 100_000_000, 1_000_000_000, + } ) // syncFloat64Histogram wraps metric.Float64Histogram with thread-safe lazy initialization @@ -416,17 +424,19 @@ func (m *MetricsExporter) initMetrics() { } m.httpRequestSizeBytes = &syncFloat64Histogram{ - name: "http_request_size_bytes", - desc: "Size of HTTP requests", - unit: "By", - meter: m.meter, + name: "http_request_size_bytes", + desc: "Size of HTTP requests", + unit: "By", + meter: m.meter, + boundaries: httpBodySizeBuckets, } m.httpResponseSizeBytes = &syncFloat64Histogram{ - name: "http_response_size_bytes", - desc: "Size of HTTP responses", - unit: "By", - meter: m.meter, + name: "http_response_size_bytes", + desc: "Size of HTTP responses", + unit: "By", + meter: m.meter, + boundaries: httpBodySizeBuckets, } } diff --git a/transports/bifrost-http/server/server.go b/transports/bifrost-http/server/server.go index 71938fcc00a..b097ac17694 100644 --- a/transports/bifrost-http/server/server.go +++ b/transports/bifrost-http/server/server.go @@ -9,6 +9,7 @@ import ( "net" "os" "os/signal" + "strconv" "strings" "sync" "syscall" @@ -28,6 +29,7 @@ import ( "github.com/maximhq/bifrost/plugins/governance" "github.com/maximhq/bifrost/plugins/governance/complexity" "github.com/maximhq/bifrost/plugins/logging" + "github.com/maximhq/bifrost/plugins/otel" "github.com/maximhq/bifrost/plugins/prompts" "github.com/maximhq/bifrost/plugins/semanticcache" "github.com/maximhq/bifrost/plugins/telemetry" @@ -1508,6 +1510,29 @@ func (s *BifrostHTTPServer) PrepareCommonMiddlewares() []schemas.BifrostHTTPMidd } else { logger.Warn("prometheus plugin not found, skipping telemetry middleware") } + // OTel HTTP metrics (http_requests_total etc., pushed via OTLP). The otel plugin is + // resolved per request rather than captured here: a config reload swaps in a freshly + // constructed plugin instance, and a pointer captured at startup would keep recording + // against exporters whose meter provider has been shut down. + commonMiddlewares = append(commonMiddlewares, func(next fasthttp.RequestHandler) fasthttp.RequestHandler { + return func(ctx *fasthttp.RequestCtx) { + start := time.Now() + reqSize := float64(ctx.Request.Header.ContentLength()) + next(ctx) + otelPlugin, err := lib.FindPluginAs[*otel.OtelPlugin](s.Config, otel.PluginName) + if err != nil { + return + } + otelPlugin.RecordHTTPMetrics(ctx, + string(ctx.Path()), + string(ctx.Method()), + strconv.Itoa(ctx.Response.StatusCode()), + time.Since(start).Seconds(), + reqSize, + float64(ctx.Response.Header.ContentLength()), + ) + } + }) return commonMiddlewares }