feat: adds http metrics to OTEL

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
 

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

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

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,
 	}
 }
 

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
 }