diff --git a/GLOSSARY.md b/GLOSSARY.md
index d221851d6..16132c213 100644
--- a/GLOSSARY.md
+++ b/GLOSSARY.md
@@ -230,6 +230,25 @@ A time-based trigger that fires an action — sending a message or dispatching (
 _Avoid_: cron job (recurring only), scheduled message (too narrow), reminder, timer
 _See also_: Dispatch
 
+## Observability
+
+Scion produces two distinct families of metrics. They serve different audiences, use different prefixes, and flow through different pipelines — but both export to the same Cloud Monitoring backend.
+
+**Infrastructure metrics**:
+Operational health metrics for Scion as a system — the Hub process, its database connections, dispatch pipeline, broker authentication, and GCP token minting. These answer "is Scion itself healthy?" and are consumed by platform operators. Prefixes: `scion.hub.*`, `scion.db.*`, `scion.dispatch.*`. Produced by the Hub process; exported directly to Cloud Monitoring via an OTel MeterProvider with a GCP exporter.
+_Avoid_: system metrics, platform metrics, server metrics
+_See also_: Agent metrics (the other family)
+
+**Agent metrics**:
+Telemetry about what agents and their harnesses are doing — token usage, tool calls, model API latency, session counts, and cost signals. These answer "what are the agents doing and what do they cost?" and are consumed by users and project owners. Prefixes: `gen_ai.*`, `agent.*` (following OpenTelemetry Generative AI semantic conventions). Produced inside agent containers by the harness and sciontool; exported to Cloud Monitoring via the telemetry pipeline (`pkg/sciontool/telemetry`).
+_Avoid_: harness metrics, user metrics, LLM metrics
+_See also_: Infrastructure metrics (the other family), Telemetry pipeline
+
+**Telemetry pipeline**:
+The in-container OTLP receiver and forwarding pipeline (`pkg/sciontool/telemetry`) that collects traces, metrics, and logs from the harness and exports them to a cloud backend (GCP Cloud Monitoring, Cloud Trace, Cloud Logging). Requires the `scion-telemetry-gcp-credentials` secret for cloud export; runs in local-only mode without it.
+_Avoid_: metrics pipeline, collector, OTel collector
+_See also_: Agent metrics
+
 ## Potential Future Additions
 
 Terms that recur in the codebase and may warrant canonical entries, but are **not yet defined** here. Listed so they aren't lost; promote to full entries (verified against the code) as the glossary matures.
diff --git a/cmd/server_foreground.go b/cmd/server_foreground.go
index 0dd5b4f61..7d9bbac6b 100644
--- a/cmd/server_foreground.go
+++ b/cmd/server_foreground.go
@@ -44,6 +44,8 @@ import (
 	"github.com/GoogleCloudPlatform/scion/pkg/harness"
 	"github.com/GoogleCloudPlatform/scion/pkg/hub"
 	"github.com/GoogleCloudPlatform/scion/pkg/observability/dbmetrics"
+	"github.com/GoogleCloudPlatform/scion/pkg/observability/dispatchmetrics"
+	"github.com/GoogleCloudPlatform/scion/pkg/observability/hubmetrics"
 	scionplugin "github.com/GoogleCloudPlatform/scion/pkg/plugin"
 	"github.com/GoogleCloudPlatform/scion/pkg/runtime"
 	"github.com/GoogleCloudPlatform/scion/pkg/runtimebroker"
@@ -211,6 +213,7 @@ func runServerStart(cmd *cobra.Command, args []string) error {
 	// 11. Start Hub
 	var hubSrv *hub.Server
 	var secretBackend secret.SecretBackend
+	var hubDBRec dbmetrics.Recorder
 	if enableHub {
 		// Initialize secret backend early so signing keys can be loaded from it
 		// during hub server creation. This prevents the previous bug where
@@ -232,13 +235,62 @@ func runServerStart(cmd *cobra.Command, args []string) error {
 			log.Fatalf("Hub server failed to start: %v", hubInitErr)
 		}
 
+		// Wire hub OTel metrics export to Cloud Monitoring.
+		if cfg.Hub.GCPProjectID != "" {
+			mp, mpErr := hubmetrics.NewMeterProvider(ctx, cfg.Hub.GCPProjectID,
+				hubmetrics.WithHubID(hubSrv.HubID()),
+			)
+			if mpErr != nil {
+				log.Printf("WARNING: hub metrics export disabled: %v", mpErr)
+			} else {
+				defer func() {
+					shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+					defer cancel()
+					_ = mp.Shutdown(shutdownCtx)
+				}()
+
+				dbRec, dbErr := dbmetrics.New(mp)
+				if dbErr != nil {
+					log.Printf("WARNING: hub db metrics disabled: %v", dbErr)
+				} else {
+					hubDBRec = dbRec
+					hubSrv.SetDBMetrics(dbRec)
+				}
+
+				dispRec, dispErr := dispatchmetrics.New(mp)
+				if dispErr != nil {
+					log.Printf("WARNING: hub dispatch metrics disabled: %v", dispErr)
+				} else {
+					hubSrv.SetDispatchMetrics(dispRec)
+				}
+
+				if hubSrv.GetBrokerAuthService() != nil {
+					otelMetrics, otelAuthErr := hub.NewOTelMetricsRecorder(mp)
+					if otelAuthErr != nil {
+						log.Printf("WARNING: hub auth metrics OTel export disabled: %v", otelAuthErr)
+					} else {
+						hubSrv.SetMetrics(otelMetrics)
+					}
+				}
+
+				otelGCP, otelGCPErr := hub.NewOTelGCPTokenMetrics(mp)
+				if otelGCPErr != nil {
+					log.Printf("WARNING: hub GCP token metrics OTel export disabled: %v", otelGCPErr)
+				} else {
+					hubSrv.SetGCPTokenMetrics(otelGCP)
+				}
+
+				log.Printf("Hub OTel metrics export enabled (project: %s)", cfg.Hub.GCPProjectID)
+			}
+		}
+
 		// Wire command bus for cross-node dispatch (B2-4).
 		cmdBus := newCommandBus(ctx, cfg, hubSrv)
 		hubSrv.SetCommandBus(cmdBus)
 
 		if !enableWeb {
 			// Hub runs its own HTTP server (standalone mode).
-			eventPub := newEventPublisher(ctx, cfg)
+			eventPub := newEventPublisher(ctx, cfg, hubDBRec)
 			hubSrv.SetEventPublisher(eventPub)
 
 			log.Printf("Starting Hub API server on %s:%d", cfg.Hub.Host, cfg.Hub.Port)
@@ -266,7 +318,7 @@ func runServerStart(cmd *cobra.Command, args []string) error {
 	// 12. Start Web
 	var webSrv *hub.WebServer
 	if enableWeb {
-		webSrv = initWebServer(ctx, cfg, hubSrv, devAuthToken, adminEmailList, adminMode, maintenanceMessage, requestLogger)
+		webSrv = initWebServer(ctx, cfg, hubSrv, devAuthToken, adminEmailList, adminMode, maintenanceMessage, requestLogger, hubDBRec)
 
 		// In combined mode, start Hub background services now that the
 		// ChannelEventPublisher has been wired by initWebServer.
@@ -1167,11 +1219,12 @@ func initHubStorage(ctx context.Context, hubSrv *hub.Server, cfg *config.GlobalC
 // ChannelEventPublisher. If the Postgres publisher cannot be started it falls
 // back to the in-process publisher so a single instance still functions, logging
 // a prominent warning since cross-replica SSE delivery will be unavailable.
-func newEventPublisher(ctx context.Context, cfg *config.GlobalConfig) hub.EventPublisher {
+func newEventPublisher(ctx context.Context, cfg *config.GlobalConfig, dbRec dbmetrics.Recorder) hub.EventPublisher {
 	if strings.EqualFold(cfg.Database.Driver, "postgres") {
-		// Metrics export is wired separately (see pkg/observability/dbmetrics);
-		// use a disabled recorder until a MeterProvider is configured.
-		pub, err := hub.NewPostgresEventPublisher(ctx, cfg.Database.URL, dbmetrics.NewDisabled(), logging.Subsystem("hub.events"))
+		if dbRec == nil {
+			dbRec = dbmetrics.NewDisabled()
+		}
+		pub, err := hub.NewPostgresEventPublisher(ctx, cfg.Database.URL, dbRec, logging.Subsystem("hub.events"))
 		if err != nil {
 			log.Printf("WARNING: failed to start Postgres event publisher (%v); falling back to in-process events. Cross-replica SSE will not work.", err)
 			return hub.NewChannelEventPublisher()
@@ -1208,7 +1261,7 @@ func newCommandBus(ctx context.Context, cfg *config.GlobalConfig, hubSrv *hub.Se
 // initWebServer creates and configures the Web server. The provided context is
 // threaded to the event publisher so that the Postgres LISTEN/NOTIFY goroutine
 // is cancelled cleanly on shutdown, preventing connection leaks.
-func initWebServer(ctx context.Context, cfg *config.GlobalConfig, hubSrv *hub.Server, devAuthToken string, adminEmailList []string, adminMode bool, maintenanceMessage string, requestLogger *slog.Logger) *hub.WebServer {
+func initWebServer(ctx context.Context, cfg *config.GlobalConfig, hubSrv *hub.Server, devAuthToken string, adminEmailList []string, adminMode bool, maintenanceMessage string, requestLogger *slog.Logger, dbRec dbmetrics.Recorder) *hub.WebServer {
 	webHost := cfg.Hub.Host
 	if webHost == "" {
 		webHost = "0.0.0.0"
@@ -1264,7 +1317,7 @@ func initWebServer(ctx context.Context, cfg *config.GlobalConfig, hubSrv *hub.Se
 	webSrv.SetRequestLogger(requestLogger)
 
 	// Create shared event publisher for real-time SSE
-	eventPub := newEventPublisher(ctx, cfg)
+	eventPub := newEventPublisher(ctx, cfg, dbRec)
 	webSrv.SetEventPublisher(eventPub)
 
 	// Wire Hub services into WebServer if Hub is enabled
diff --git a/pkg/hub/gcp_metrics.go b/pkg/hub/gcp_metrics.go
index 0c91c4a4c..0cbcd0077 100644
--- a/pkg/hub/gcp_metrics.go
+++ b/pkg/hub/gcp_metrics.go
@@ -20,6 +20,14 @@ import (
 	"time"
 )
 
+// GCPTokenMetricsRecorder is the interface for recording GCP token metrics.
+type GCPTokenMetricsRecorder interface {
+	RecordAccessTokenRequest(success bool, latency time.Duration)
+	RecordIDTokenRequest(success bool, latency time.Duration)
+	RecordRateLimitRejection()
+	GetSnapshot() *GCPTokenMetricsSnapshot
+}
+
 // GCPTokenMetrics tracks metrics for GCP token operations.
 type GCPTokenMetrics struct {
 	// Access token counters
diff --git a/pkg/hub/otel_gcp_metrics.go b/pkg/hub/otel_gcp_metrics.go
new file mode 100644
index 000000000..cbca83d7a
--- /dev/null
+++ b/pkg/hub/otel_gcp_metrics.go
@@ -0,0 +1,125 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package hub
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.opentelemetry.io/otel/metric"
+)
+
+// OTelGCPTokenMetrics implements GCPTokenMetricsRecorder using OTel
+// instruments for Cloud Monitoring export. It embeds a GCPTokenMetrics for
+// the /api/metrics JSON snapshot endpoint (dual-write).
+type OTelGCPTokenMetrics struct {
+	accessRequests   metric.Int64Counter
+	accessSuccesses  metric.Int64Counter
+	accessFailures   metric.Int64Counter
+	idRequests       metric.Int64Counter
+	idSuccesses      metric.Int64Counter
+	idFailures       metric.Int64Counter
+	rateLimitRejects metric.Int64Counter
+	iamDuration      metric.Float64Histogram
+
+	snap *GCPTokenMetrics
+}
+
+var _ GCPTokenMetricsRecorder = (*OTelGCPTokenMetrics)(nil)
+
+// NewOTelGCPTokenMetrics creates an OTel-backed GCP token metrics recorder.
+func NewOTelGCPTokenMetrics(mp metric.MeterProvider) (*OTelGCPTokenMetrics, error) {
+	m := mp.Meter(instrumentationScope)
+	r := &OTelGCPTokenMetrics{snap: NewGCPTokenMetrics()}
+
+	var err error
+
+	if r.accessRequests, err = m.Int64Counter("scion.hub.gcp.token.access.requests",
+		metric.WithUnit("{request}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating gcp.token.access.requests counter: %w", err)
+	}
+	if r.accessSuccesses, err = m.Int64Counter("scion.hub.gcp.token.access.successes",
+		metric.WithUnit("{request}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating gcp.token.access.successes counter: %w", err)
+	}
+	if r.accessFailures, err = m.Int64Counter("scion.hub.gcp.token.access.failures",
+		metric.WithUnit("{request}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating gcp.token.access.failures counter: %w", err)
+	}
+	if r.idRequests, err = m.Int64Counter("scion.hub.gcp.token.identity.requests",
+		metric.WithUnit("{request}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating gcp.token.identity.requests counter: %w", err)
+	}
+	if r.idSuccesses, err = m.Int64Counter("scion.hub.gcp.token.identity.successes",
+		metric.WithUnit("{request}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating gcp.token.identity.successes counter: %w", err)
+	}
+	if r.idFailures, err = m.Int64Counter("scion.hub.gcp.token.identity.failures",
+		metric.WithUnit("{request}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating gcp.token.identity.failures counter: %w", err)
+	}
+	if r.rateLimitRejects, err = m.Int64Counter("scion.hub.gcp.token.ratelimit.rejections",
+		metric.WithUnit("{rejection}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating gcp.token.ratelimit.rejections counter: %w", err)
+	}
+	if r.iamDuration, err = m.Float64Histogram("scion.hub.gcp.iam.duration",
+		metric.WithUnit("ms"),
+	); err != nil {
+		return nil, fmt.Errorf("creating gcp.iam.duration histogram: %w", err)
+	}
+
+	return r, nil
+}
+
+func (r *OTelGCPTokenMetrics) RecordAccessTokenRequest(success bool, latency time.Duration) {
+	ctx := context.Background()
+	r.accessRequests.Add(ctx, 1)
+	if success {
+		r.accessSuccesses.Add(ctx, 1)
+	} else {
+		r.accessFailures.Add(ctx, 1)
+	}
+	r.iamDuration.Record(ctx, float64(latency.Milliseconds()))
+	r.snap.RecordAccessTokenRequest(success, latency)
+}
+
+func (r *OTelGCPTokenMetrics) RecordIDTokenRequest(success bool, latency time.Duration) {
+	ctx := context.Background()
+	r.idRequests.Add(ctx, 1)
+	if success {
+		r.idSuccesses.Add(ctx, 1)
+	} else {
+		r.idFailures.Add(ctx, 1)
+	}
+	r.iamDuration.Record(ctx, float64(latency.Milliseconds()))
+	r.snap.RecordIDTokenRequest(success, latency)
+}
+
+func (r *OTelGCPTokenMetrics) RecordRateLimitRejection() {
+	r.rateLimitRejects.Add(context.Background(), 1)
+	r.snap.RecordRateLimitRejection()
+}
+
+func (r *OTelGCPTokenMetrics) GetSnapshot() *GCPTokenMetricsSnapshot {
+	return r.snap.GetSnapshot()
+}
diff --git a/pkg/hub/otel_gcp_metrics_test.go b/pkg/hub/otel_gcp_metrics_test.go
new file mode 100644
index 000000000..59d14248b
--- /dev/null
+++ b/pkg/hub/otel_gcp_metrics_test.go
@@ -0,0 +1,185 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package hub
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+var _ GCPTokenMetricsRecorder = (*OTelGCPTokenMetrics)(nil)
+
+func newTestGCPRecorder(t *testing.T) (*OTelGCPTokenMetrics, *metric.ManualReader) {
+	t.Helper()
+	reader := metric.NewManualReader()
+	mp := metric.NewMeterProvider(metric.WithReader(reader))
+	t.Cleanup(func() { _ = mp.Shutdown(context.Background()) })
+
+	rec, err := NewOTelGCPTokenMetrics(mp)
+	if err != nil {
+		t.Fatalf("NewOTelGCPTokenMetrics: %v", err)
+	}
+	return rec, reader
+}
+
+func collectGCPMetrics(t *testing.T, reader *metric.ManualReader) map[string]metricdata.Metrics {
+	t.Helper()
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(context.Background(), &rm); err != nil {
+		t.Fatalf("collecting metrics: %v", err)
+	}
+	result := make(map[string]metricdata.Metrics)
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			result[m.Name] = m
+		}
+	}
+	return result
+}
+
+func gcpSumCounter(m metricdata.Metrics) int64 {
+	sum, ok := m.Data.(metricdata.Sum[int64])
+	if !ok {
+		return 0
+	}
+	var total int64
+	for _, dp := range sum.DataPoints {
+		total += dp.Value
+	}
+	return total
+}
+
+func TestOTelGCPRecordAccessTokenRequest(t *testing.T) {
+	rec, reader := newTestGCPRecorder(t)
+
+	rec.RecordAccessTokenRequest(true, 30*time.Millisecond)
+	rec.RecordAccessTokenRequest(false, 50*time.Millisecond)
+
+	metrics := collectGCPMetrics(t, reader)
+
+	if got := gcpSumCounter(metrics["scion.hub.gcp.token.access.requests"]); got != 2 {
+		t.Errorf("access.requests = %d, want 2", got)
+	}
+	if got := gcpSumCounter(metrics["scion.hub.gcp.token.access.successes"]); got != 1 {
+		t.Errorf("access.successes = %d, want 1", got)
+	}
+	if got := gcpSumCounter(metrics["scion.hub.gcp.token.access.failures"]); got != 1 {
+		t.Errorf("access.failures = %d, want 1", got)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.AccessTokenRequests != 2 {
+		t.Errorf("snapshot AccessTokenRequests = %d, want 2", snap.AccessTokenRequests)
+	}
+	if snap.AccessTokenSuccesses != 1 {
+		t.Errorf("snapshot AccessTokenSuccesses = %d, want 1", snap.AccessTokenSuccesses)
+	}
+	if snap.AccessTokenFailures != 1 {
+		t.Errorf("snapshot AccessTokenFailures = %d, want 1", snap.AccessTokenFailures)
+	}
+}
+
+func TestOTelGCPRecordIDTokenRequest(t *testing.T) {
+	rec, reader := newTestGCPRecorder(t)
+
+	rec.RecordIDTokenRequest(true, 20*time.Millisecond)
+	rec.RecordIDTokenRequest(false, 40*time.Millisecond)
+
+	metrics := collectGCPMetrics(t, reader)
+
+	if got := gcpSumCounter(metrics["scion.hub.gcp.token.identity.requests"]); got != 2 {
+		t.Errorf("identity.requests = %d, want 2", got)
+	}
+	if got := gcpSumCounter(metrics["scion.hub.gcp.token.identity.successes"]); got != 1 {
+		t.Errorf("identity.successes = %d, want 1", got)
+	}
+	if got := gcpSumCounter(metrics["scion.hub.gcp.token.identity.failures"]); got != 1 {
+		t.Errorf("identity.failures = %d, want 1", got)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.IDTokenRequests != 2 {
+		t.Errorf("snapshot IDTokenRequests = %d, want 2", snap.IDTokenRequests)
+	}
+	if snap.IDTokenSuccesses != 1 {
+		t.Errorf("snapshot IDTokenSuccesses = %d, want 1", snap.IDTokenSuccesses)
+	}
+	if snap.IDTokenFailures != 1 {
+		t.Errorf("snapshot IDTokenFailures = %d, want 1", snap.IDTokenFailures)
+	}
+}
+
+func TestOTelGCPRecordRateLimitRejection(t *testing.T) {
+	rec, reader := newTestGCPRecorder(t)
+
+	rec.RecordRateLimitRejection()
+	rec.RecordRateLimitRejection()
+
+	metrics := collectGCPMetrics(t, reader)
+
+	if got := gcpSumCounter(metrics["scion.hub.gcp.token.ratelimit.rejections"]); got != 2 {
+		t.Errorf("ratelimit.rejections = %d, want 2", got)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.RateLimitRejections != 2 {
+		t.Errorf("snapshot RateLimitRejections = %d, want 2", snap.RateLimitRejections)
+	}
+}
+
+func TestOTelGCPIAMDurationHistogram(t *testing.T) {
+	rec, reader := newTestGCPRecorder(t)
+
+	rec.RecordAccessTokenRequest(true, 42*time.Millisecond)
+
+	metrics := collectGCPMetrics(t, reader)
+	m, ok := metrics["scion.hub.gcp.iam.duration"]
+	if !ok {
+		t.Fatal("scion.hub.gcp.iam.duration not found")
+	}
+	hist, ok := m.Data.(metricdata.Histogram[float64])
+	if !ok {
+		t.Fatal("iam.duration is not a histogram")
+	}
+	if len(hist.DataPoints) == 0 {
+		t.Fatal("histogram has no data points")
+	}
+	if hist.DataPoints[0].Sum <= 0 {
+		t.Errorf("histogram sum = %f, want > 0", hist.DataPoints[0].Sum)
+	}
+}
+
+func TestOTelGCPGetSnapshot(t *testing.T) {
+	rec, _ := newTestGCPRecorder(t)
+
+	rec.RecordAccessTokenRequest(true, 10*time.Millisecond)
+	rec.RecordIDTokenRequest(false, 20*time.Millisecond)
+	rec.RecordRateLimitRejection()
+
+	snap := rec.GetSnapshot()
+	if snap.AccessTokenRequests != 1 {
+		t.Errorf("AccessTokenRequests = %d, want 1", snap.AccessTokenRequests)
+	}
+	if snap.IDTokenRequests != 1 {
+		t.Errorf("IDTokenRequests = %d, want 1", snap.IDTokenRequests)
+	}
+	if snap.RateLimitRejections != 1 {
+		t.Errorf("RateLimitRejections = %d, want 1", snap.RateLimitRejections)
+	}
+}
diff --git a/pkg/hub/otel_metrics.go b/pkg/hub/otel_metrics.go
new file mode 100644
index 000000000..751697289
--- /dev/null
+++ b/pkg/hub/otel_metrics.go
@@ -0,0 +1,174 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package hub
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+const instrumentationScope = "github.com/GoogleCloudPlatform/scion/pkg/hub"
+
+// OTelMetricsRecorder implements MetricsRecorder using OTel instruments for
+// Cloud Monitoring export. It embeds a BrokerAuthMetrics for the /api/metrics
+// JSON snapshot endpoint (dual-write).
+type OTelMetricsRecorder struct {
+	authAttempts     metric.Int64Counter
+	authSuccesses    metric.Int64Counter
+	authFailures     metric.Int64Counter
+	authDuration     metric.Float64Histogram
+	registrations    metric.Int64Counter
+	joins            metric.Int64Counter
+	joinFailures     metric.Int64Counter
+	rotations        metric.Int64Counter
+	dispatchAttempts metric.Int64Counter
+	dispatchFailures metric.Int64Counter
+	connectedBrokers metric.Int64Gauge
+
+	snap *BrokerAuthMetrics
+}
+
+var _ MetricsRecorder = (*OTelMetricsRecorder)(nil)
+
+// NewOTelMetricsRecorder creates an OTel-backed MetricsRecorder. All
+// instruments are registered under the hub instrumentation scope.
+func NewOTelMetricsRecorder(mp metric.MeterProvider) (*OTelMetricsRecorder, error) {
+	m := mp.Meter(instrumentationScope)
+	r := &OTelMetricsRecorder{snap: NewBrokerAuthMetrics()}
+
+	var err error
+
+	if r.authAttempts, err = m.Int64Counter("scion.hub.auth.attempts",
+		metric.WithUnit("{attempt}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating auth.attempts counter: %w", err)
+	}
+	if r.authSuccesses, err = m.Int64Counter("scion.hub.auth.successes",
+		metric.WithUnit("{attempt}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating auth.successes counter: %w", err)
+	}
+	if r.authFailures, err = m.Int64Counter("scion.hub.auth.failures",
+		metric.WithUnit("{attempt}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating auth.failures counter: %w", err)
+	}
+	if r.authDuration, err = m.Float64Histogram("scion.hub.auth.duration",
+		metric.WithUnit("ms"),
+	); err != nil {
+		return nil, fmt.Errorf("creating auth.duration histogram: %w", err)
+	}
+	if r.registrations, err = m.Int64Counter("scion.hub.registration.count",
+		metric.WithUnit("{registration}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating registration.count counter: %w", err)
+	}
+	if r.joins, err = m.Int64Counter("scion.hub.join.attempts",
+		metric.WithUnit("{attempt}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating join.attempts counter: %w", err)
+	}
+	if r.joinFailures, err = m.Int64Counter("scion.hub.join.failures",
+		metric.WithUnit("{attempt}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating join.failures counter: %w", err)
+	}
+	if r.rotations, err = m.Int64Counter("scion.hub.rotation.count",
+		metric.WithUnit("{rotation}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating rotation.count counter: %w", err)
+	}
+	if r.dispatchAttempts, err = m.Int64Counter("scion.hub.dispatch.attempts",
+		metric.WithUnit("{attempt}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating dispatch.attempts counter: %w", err)
+	}
+	if r.dispatchFailures, err = m.Int64Counter("scion.hub.dispatch.failures",
+		metric.WithUnit("{attempt}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating dispatch.failures counter: %w", err)
+	}
+	if r.connectedBrokers, err = m.Int64Gauge("scion.hub.brokers.connected",
+		metric.WithUnit("{broker}"),
+	); err != nil {
+		return nil, fmt.Errorf("creating brokers.connected gauge: %w", err)
+	}
+
+	return r, nil
+}
+
+func (r *OTelMetricsRecorder) RecordAuthAttempt(brokerID string, success bool, latency time.Duration) {
+	ctx := context.Background()
+	attrs := metric.WithAttributes(attribute.String("broker_id", brokerID))
+	r.authAttempts.Add(ctx, 1, attrs)
+	if success {
+		r.authSuccesses.Add(ctx, 1, attrs)
+	} else {
+		r.authFailures.Add(ctx, 1, attrs)
+	}
+	r.authDuration.Record(ctx, float64(latency.Milliseconds()), attrs)
+	r.snap.RecordAuthAttempt(brokerID, success, latency)
+}
+
+func (r *OTelMetricsRecorder) RecordRegistration(brokerID string) {
+	ctx := context.Background()
+	attrs := metric.WithAttributes(attribute.String("broker_id", brokerID))
+	r.registrations.Add(ctx, 1, attrs)
+	r.snap.RecordRegistration(brokerID)
+}
+
+func (r *OTelMetricsRecorder) RecordJoin(brokerID string, success bool) {
+	ctx := context.Background()
+	attrs := metric.WithAttributes(attribute.String("broker_id", brokerID))
+	r.joins.Add(ctx, 1, attrs)
+	if !success {
+		r.joinFailures.Add(ctx, 1, attrs)
+	}
+	r.snap.RecordJoin(brokerID, success)
+}
+
+func (r *OTelMetricsRecorder) RecordRotation(brokerID string) {
+	ctx := context.Background()
+	attrs := metric.WithAttributes(attribute.String("broker_id", brokerID))
+	r.rotations.Add(ctx, 1, attrs)
+	r.snap.RecordRotation(brokerID)
+}
+
+func (r *OTelMetricsRecorder) RecordDispatch(brokerID string, operation string, success bool, latency time.Duration) {
+	ctx := context.Background()
+	attrs := metric.WithAttributes(
+		attribute.String("broker_id", brokerID),
+		attribute.String("operation", operation),
+	)
+	r.dispatchAttempts.Add(ctx, 1, attrs)
+	if !success {
+		r.dispatchFailures.Add(ctx, 1, attrs)
+	}
+	r.snap.RecordDispatch(brokerID, operation, success, latency)
+}
+
+func (r *OTelMetricsRecorder) SetConnectedBrokers(count int64) {
+	ctx := context.Background()
+	r.connectedBrokers.Record(ctx, count)
+	r.snap.SetConnectedBrokers(count)
+}
+
+func (r *OTelMetricsRecorder) GetSnapshot() *MetricsSnapshot {
+	return r.snap.GetSnapshot()
+}
diff --git a/pkg/hub/otel_metrics_test.go b/pkg/hub/otel_metrics_test.go
new file mode 100644
index 000000000..212cd40f9
--- /dev/null
+++ b/pkg/hub/otel_metrics_test.go
@@ -0,0 +1,224 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package hub
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+var _ MetricsRecorder = (*OTelMetricsRecorder)(nil)
+
+func newTestRecorder(t *testing.T) (*OTelMetricsRecorder, *metric.ManualReader) {
+	t.Helper()
+	reader := metric.NewManualReader()
+	mp := metric.NewMeterProvider(metric.WithReader(reader))
+	t.Cleanup(func() { _ = mp.Shutdown(context.Background()) })
+
+	rec, err := NewOTelMetricsRecorder(mp)
+	if err != nil {
+		t.Fatalf("NewOTelMetricsRecorder: %v", err)
+	}
+	return rec, reader
+}
+
+func collectMetrics(t *testing.T, reader *metric.ManualReader) map[string]metricdata.Metrics {
+	t.Helper()
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(context.Background(), &rm); err != nil {
+		t.Fatalf("collecting metrics: %v", err)
+	}
+	result := make(map[string]metricdata.Metrics)
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			result[m.Name] = m
+		}
+	}
+	return result
+}
+
+func sumCounter(m metricdata.Metrics) int64 {
+	sum, ok := m.Data.(metricdata.Sum[int64])
+	if !ok {
+		return 0
+	}
+	var total int64
+	for _, dp := range sum.DataPoints {
+		total += dp.Value
+	}
+	return total
+}
+
+func TestOTelRecordAuthAttempt(t *testing.T) {
+	rec, reader := newTestRecorder(t)
+
+	rec.RecordAuthAttempt("broker-1", true, 50*time.Millisecond)
+	rec.RecordAuthAttempt("broker-1", false, 100*time.Millisecond)
+
+	metrics := collectMetrics(t, reader)
+
+	if got := sumCounter(metrics["scion.hub.auth.attempts"]); got != 2 {
+		t.Errorf("auth.attempts = %d, want 2", got)
+	}
+	if got := sumCounter(metrics["scion.hub.auth.successes"]); got != 1 {
+		t.Errorf("auth.successes = %d, want 1", got)
+	}
+	if got := sumCounter(metrics["scion.hub.auth.failures"]); got != 1 {
+		t.Errorf("auth.failures = %d, want 1", got)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.AuthAttempts != 2 {
+		t.Errorf("snapshot AuthAttempts = %d, want 2", snap.AuthAttempts)
+	}
+	if snap.AuthSuccesses != 1 {
+		t.Errorf("snapshot AuthSuccesses = %d, want 1", snap.AuthSuccesses)
+	}
+	if snap.AuthFailures != 1 {
+		t.Errorf("snapshot AuthFailures = %d, want 1", snap.AuthFailures)
+	}
+}
+
+func TestOTelAuthDurationHistogram(t *testing.T) {
+	rec, reader := newTestRecorder(t)
+
+	rec.RecordAuthAttempt("broker-1", true, 42*time.Millisecond)
+
+	metrics := collectMetrics(t, reader)
+	m, ok := metrics["scion.hub.auth.duration"]
+	if !ok {
+		t.Fatal("scion.hub.auth.duration not found")
+	}
+	hist, ok := m.Data.(metricdata.Histogram[float64])
+	if !ok {
+		t.Fatal("auth.duration is not a histogram")
+	}
+	if len(hist.DataPoints) == 0 {
+		t.Fatal("histogram has no data points")
+	}
+	if hist.DataPoints[0].Sum <= 0 {
+		t.Errorf("histogram sum = %f, want > 0", hist.DataPoints[0].Sum)
+	}
+}
+
+func TestOTelRecordRegistration(t *testing.T) {
+	rec, reader := newTestRecorder(t)
+
+	rec.RecordRegistration("broker-1")
+	rec.RecordRegistration("broker-2")
+
+	metrics := collectMetrics(t, reader)
+	if got := sumCounter(metrics["scion.hub.registration.count"]); got != 2 {
+		t.Errorf("registration.count = %d, want 2", got)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.Registrations != 2 {
+		t.Errorf("snapshot Registrations = %d, want 2", snap.Registrations)
+	}
+}
+
+func TestOTelRecordJoin(t *testing.T) {
+	rec, reader := newTestRecorder(t)
+
+	rec.RecordJoin("broker-1", true)
+	rec.RecordJoin("broker-2", false)
+
+	metrics := collectMetrics(t, reader)
+	if got := sumCounter(metrics["scion.hub.join.attempts"]); got != 2 {
+		t.Errorf("join.attempts = %d, want 2", got)
+	}
+	if got := sumCounter(metrics["scion.hub.join.failures"]); got != 1 {
+		t.Errorf("join.failures = %d, want 1", got)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.Joins != 2 {
+		t.Errorf("snapshot Joins = %d, want 2", snap.Joins)
+	}
+	if snap.JoinFailures != 1 {
+		t.Errorf("snapshot JoinFailures = %d, want 1", snap.JoinFailures)
+	}
+}
+
+func TestOTelRecordRotation(t *testing.T) {
+	rec, reader := newTestRecorder(t)
+
+	rec.RecordRotation("broker-1")
+
+	metrics := collectMetrics(t, reader)
+	if got := sumCounter(metrics["scion.hub.rotation.count"]); got != 1 {
+		t.Errorf("rotation.count = %d, want 1", got)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.Rotations != 1 {
+		t.Errorf("snapshot Rotations = %d, want 1", snap.Rotations)
+	}
+}
+
+func TestOTelRecordDispatch(t *testing.T) {
+	rec, reader := newTestRecorder(t)
+
+	rec.RecordDispatch("broker-1", "create", true, 10*time.Millisecond)
+	rec.RecordDispatch("broker-1", "create", false, 20*time.Millisecond)
+
+	metrics := collectMetrics(t, reader)
+	if got := sumCounter(metrics["scion.hub.dispatch.attempts"]); got != 2 {
+		t.Errorf("dispatch.attempts = %d, want 2", got)
+	}
+	if got := sumCounter(metrics["scion.hub.dispatch.failures"]); got != 1 {
+		t.Errorf("dispatch.failures = %d, want 1", got)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.DispatchAttempts != 2 {
+		t.Errorf("snapshot DispatchAttempts = %d, want 2", snap.DispatchAttempts)
+	}
+	if snap.DispatchFailures != 1 {
+		t.Errorf("snapshot DispatchFailures = %d, want 1", snap.DispatchFailures)
+	}
+}
+
+func TestOTelSetConnectedBrokers(t *testing.T) {
+	rec, reader := newTestRecorder(t)
+
+	rec.SetConnectedBrokers(5)
+
+	metrics := collectMetrics(t, reader)
+	m, ok := metrics["scion.hub.brokers.connected"]
+	if !ok {
+		t.Fatal("scion.hub.brokers.connected not found")
+	}
+	gauge, ok := m.Data.(metricdata.Gauge[int64])
+	if !ok {
+		t.Fatal("brokers.connected is not a gauge")
+	}
+	if len(gauge.DataPoints) == 0 {
+		t.Fatal("gauge has no data points")
+	}
+	if gauge.DataPoints[0].Value != 5 {
+		t.Errorf("gauge value = %d, want 5", gauge.DataPoints[0].Value)
+	}
+
+	snap := rec.GetSnapshot()
+	if snap.ConnectedBrokers != 5 {
+		t.Errorf("snapshot ConnectedBrokers = %d, want 5", snap.ConnectedBrokers)
+	}
+}
diff --git a/pkg/hub/server.go b/pkg/hub/server.go
index 9294747f9..8c214322a 100644
--- a/pkg/hub/server.go
+++ b/pkg/hub/server.go
@@ -602,7 +602,7 @@ type Server struct {
 	gcpTokenRateLimiter *GCPTokenRateLimiter
 
 	// GCP token metrics tracker (nil = disabled)
-	gcpTokenMetrics *GCPTokenMetrics
+	gcpTokenMetrics GCPTokenMetricsRecorder
 
 	// Database connection-pool / notify metrics recorder (P0-5). Defaults to a
 	// disabled no-op recorder; SetDBMetrics wires a real exporter. Drives the
@@ -1472,6 +1472,13 @@ func (s *Server) SetDispatchMetrics(rec dispatchmetrics.Recorder) {
 	s.dispatchMetrics = rec
 }
 
+// SetGCPTokenMetrics wires the GCP token metrics recorder.
+func (s *Server) SetGCPTokenMetrics(m GCPTokenMetricsRecorder) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.gcpTokenMetrics = m
+}
+
 // GetMaintenanceState returns the runtime maintenance state.
 func (s *Server) GetMaintenanceState() *MaintenanceState {
 	return s.maintenance
diff --git a/pkg/observability/hubmetrics/hubmetrics.go b/pkg/observability/hubmetrics/hubmetrics.go
new file mode 100644
index 000000000..5d3228d1d
--- /dev/null
+++ b/pkg/observability/hubmetrics/hubmetrics.go
@@ -0,0 +1,146 @@
+/*
+Copyright 2026 The Scion Authors.
+*/
+
+// Package hubmetrics creates the OpenTelemetry MeterProvider used by hub-side
+// metric recorders (dbmetrics, dispatchmetrics). It exports directly to GCP
+// Cloud Monitoring via Application Default Credentials.
+package hubmetrics
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	mexporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/resource"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
+)
+
+const defaultExportInterval = 60 * time.Second
+
+// MetricGroup identifies a logical group of hub metrics that can be
+// independently enabled or disabled.
+type MetricGroup struct {
+	EnvVar      string
+	NamePattern string
+}
+
+var metricGroups = []MetricGroup{
+	{EnvVar: "SCION_METRICS_DB_NOTIFY", NamePattern: "scion.db.notify.*"},
+	{EnvVar: "SCION_METRICS_DB_POOL", NamePattern: "scion.db.pool.*"},
+	{EnvVar: "SCION_METRICS_DISPATCH", NamePattern: "scion.dispatch.*"},
+	{EnvVar: "SCION_METRICS_HUB_AUTH", NamePattern: "scion.hub.auth.*"},
+	{EnvVar: "SCION_METRICS_HUB_AUTH", NamePattern: "scion.hub.registration.*"},
+	{EnvVar: "SCION_METRICS_HUB_AUTH", NamePattern: "scion.hub.join.*"},
+	{EnvVar: "SCION_METRICS_HUB_AUTH", NamePattern: "scion.hub.rotation.*"},
+	{EnvVar: "SCION_METRICS_HUB_AUTH", NamePattern: "scion.hub.brokers.*"},
+	{EnvVar: "SCION_METRICS_HUB_AUTH", NamePattern: "scion.hub.dispatch.*"},
+	{EnvVar: "SCION_METRICS_HUB_GCP", NamePattern: "scion.hub.gcp.*"},
+}
+
+// Option configures the MeterProvider.
+type Option func(*options)
+
+type options struct {
+	exportInterval time.Duration
+	hubID          string
+}
+
+// WithExportInterval sets the periodic reader interval. Defaults to 60s.
+func WithExportInterval(d time.Duration) Option {
+	return func(o *options) { o.exportInterval = d }
+}
+
+// WithHubID sets the scion.hub.id resource attribute.
+func WithHubID(id string) Option {
+	return func(o *options) { o.hubID = id }
+}
+
+// NewMeterProvider creates an OTel SDK MeterProvider that exports to GCP Cloud
+// Monitoring. It uses Application Default Credentials (workload identity on
+// Cloud Run, attached SA on GCE).
+//
+// If gcpProjectID is empty, an error is returned — callers should fall back to
+// disabled recorders.
+func NewMeterProvider(ctx context.Context, gcpProjectID string, opts ...Option) (*metric.MeterProvider, error) {
+	if gcpProjectID == "" {
+		return nil, fmt.Errorf("GCP project ID is required for hub metrics export")
+	}
+
+	o := &options{exportInterval: defaultExportInterval}
+	for _, fn := range opts {
+		fn(o)
+	}
+
+	exporter, err := mexporter.New(mexporter.WithProjectID(gcpProjectID))
+	if err != nil {
+		return nil, fmt.Errorf("creating GCP metric exporter: %w", err)
+	}
+
+	resAttrs := []attribute.KeyValue{
+		semconv.ServiceName("scion-hub"),
+	}
+	if o.hubID != "" {
+		resAttrs = append(resAttrs, attribute.String("scion.hub.id", o.hubID))
+	}
+	if envHubID := os.Getenv("SCION_HUB_ID"); envHubID != "" && o.hubID == "" {
+		resAttrs = append(resAttrs, attribute.String("scion.hub.id", envHubID))
+	}
+
+	res, err := resource.New(ctx,
+		resource.WithAttributes(resAttrs...),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("creating OTel resource: %w", err)
+	}
+
+	mpOpts := []metric.Option{
+		metric.WithResource(res),
+		metric.WithReader(metric.NewPeriodicReader(exporter,
+			metric.WithInterval(o.exportInterval),
+		)),
+	}
+
+	mpOpts = append(mpOpts, groupDropViews()...)
+
+	return metric.NewMeterProvider(mpOpts...), nil
+}
+
+// groupDropViews returns OTel View options that drop instruments belonging to
+// disabled metric groups. A group is disabled when its env var is set to
+// "false" or "0". All groups are enabled by default.
+func groupDropViews() []metric.Option {
+	var opts []metric.Option
+	for _, g := range metricGroups {
+		if isGroupDisabled(g.EnvVar) {
+			opts = append(opts, metric.WithView(metric.NewView(
+				metric.Instrument{Name: g.NamePattern},
+				metric.Stream{Aggregation: metric.AggregationDrop{}},
+			)))
+		}
+	}
+	return opts
+}
+
+func isGroupDisabled(envVar string) bool {
+	v := strings.ToLower(strings.TrimSpace(os.Getenv(envVar)))
+	return v == "false" || v == "0"
+}
+
+// GroupScopes returns the instrumentation scopes for each metric group, useful
+// for testing and documentation.
+func GroupScopes() []MetricGroup {
+	return append([]MetricGroup(nil), metricGroups...)
+}
+
+// InstrumentationScope returns a scope matching the dbmetrics or
+// dispatchmetrics package, useful for building Views in tests.
+func InstrumentationScope(name string) instrumentation.Scope {
+	return instrumentation.Scope{Name: name}
+}
diff --git a/pkg/observability/hubmetrics/hubmetrics_test.go b/pkg/observability/hubmetrics/hubmetrics_test.go
new file mode 100644
index 000000000..68567f0a4
--- /dev/null
+++ b/pkg/observability/hubmetrics/hubmetrics_test.go
@@ -0,0 +1,172 @@
+/*
+Copyright 2026 The Scion Authors.
+*/
+
+package hubmetrics
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+
+	"github.com/GoogleCloudPlatform/scion/pkg/observability/dbmetrics"
+	"github.com/GoogleCloudPlatform/scion/pkg/observability/dispatchmetrics"
+)
+
+func TestNewMeterProviderEmptyProjectID(t *testing.T) {
+	_, err := NewMeterProvider(context.Background(), "")
+	if err == nil {
+		t.Fatal("expected error for empty project ID")
+	}
+}
+
+func TestGroupDropViewsAllEnabled(t *testing.T) {
+	for _, g := range metricGroups {
+		if err := os.Unsetenv(g.EnvVar); err != nil {
+			t.Fatalf("Unsetenv(%s): %v", g.EnvVar, err)
+		}
+	}
+	views := groupDropViews()
+	if len(views) != 0 {
+		t.Errorf("expected 0 drop views when all groups enabled, got %d", len(views))
+	}
+}
+
+func TestGroupDropViewsDisabled(t *testing.T) {
+	t.Setenv("SCION_METRICS_DB_NOTIFY", "false")
+
+	views := groupDropViews()
+	if len(views) != 1 {
+		t.Errorf("expected 1 drop view, got %d", len(views))
+	}
+}
+
+func TestGroupDropViewsDisabledZero(t *testing.T) {
+	t.Setenv("SCION_METRICS_DISPATCH", "0")
+
+	views := groupDropViews()
+	if len(views) != 1 {
+		t.Errorf("expected 1 drop view, got %d", len(views))
+	}
+}
+
+func TestIsGroupDisabled(t *testing.T) {
+	tests := []struct {
+		value string
+		want  bool
+	}{
+		{"", false},
+		{"true", false},
+		{"1", false},
+		{"false", true},
+		{"0", true},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.value, func(t *testing.T) {
+			envVar := "SCION_METRICS_TEST_GROUP"
+			if tc.value != "" {
+				t.Setenv(envVar, tc.value)
+			} else {
+				if err := os.Unsetenv(envVar); err != nil {
+					t.Fatalf("Unsetenv(%s): %v", envVar, err)
+				}
+			}
+			if got := isGroupDisabled(envVar); got != tc.want {
+				t.Errorf("isGroupDisabled(%q=%q) = %v, want %v", envVar, tc.value, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestRecordersEnabledWithRealProvider(t *testing.T) {
+	reader := metric.NewManualReader()
+	mp := metric.NewMeterProvider(metric.WithReader(reader))
+	t.Cleanup(func() { _ = mp.Shutdown(context.Background()) })
+
+	dbRec, err := dbmetrics.New(mp)
+	if err != nil {
+		t.Fatalf("dbmetrics.New: %v", err)
+	}
+	if !dbRec.Enabled() {
+		t.Error("dbmetrics.Recorder should be enabled with real MeterProvider")
+	}
+
+	dispRec, err := dispatchmetrics.New(mp)
+	if err != nil {
+		t.Fatalf("dispatchmetrics.New: %v", err)
+	}
+	if !dispRec.Enabled() {
+		t.Error("dispatchmetrics.Recorder should be enabled with real MeterProvider")
+	}
+}
+
+func TestDropViewPreventsExport(t *testing.T) {
+	t.Setenv("SCION_METRICS_DB_NOTIFY", "false")
+
+	reader := metric.NewManualReader()
+	mpOpts := []metric.Option{metric.WithReader(reader)}
+	mpOpts = append(mpOpts, groupDropViews()...)
+	mp := metric.NewMeterProvider(mpOpts...)
+	t.Cleanup(func() { _ = mp.Shutdown(context.Background()) })
+
+	dbRec, err := dbmetrics.New(mp)
+	if err != nil {
+		t.Fatalf("dbmetrics.New: %v", err)
+	}
+
+	ctx := context.Background()
+	dbRec.IncPublished(ctx, 1)
+	dbRec.IncDelivered(ctx, 1)
+
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(ctx, &rm); err != nil {
+		t.Fatalf("collecting metrics: %v", err)
+	}
+
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name == dbmetrics.MetricNotificationsPublished ||
+				m.Name == dbmetrics.MetricNotificationsDelivered {
+				t.Errorf("metric %q should have been dropped by view, but was exported", m.Name)
+			}
+		}
+	}
+}
+
+func TestPoolMetricsNotDroppedWhenNotifyDisabled(t *testing.T) {
+	t.Setenv("SCION_METRICS_DB_NOTIFY", "false")
+
+	reader := metric.NewManualReader()
+	mpOpts := []metric.Option{metric.WithReader(reader)}
+	mpOpts = append(mpOpts, groupDropViews()...)
+	mp := metric.NewMeterProvider(mpOpts...)
+	t.Cleanup(func() { _ = mp.Shutdown(context.Background()) })
+
+	dbRec, err := dbmetrics.New(mp)
+	if err != nil {
+		t.Fatalf("dbmetrics.New: %v", err)
+	}
+
+	ctx := context.Background()
+	dbRec.ObservePoolStats(ctx, dbmetrics.PoolStats{Active: 5, Idle: 3, Waiting: 0, Max: 10})
+
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(ctx, &rm); err != nil {
+		t.Fatalf("collecting metrics: %v", err)
+	}
+
+	names := make(map[string]bool)
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			names[m.Name] = true
+		}
+	}
+
+	if !names[dbmetrics.MetricPoolConnectionsActive] {
+		t.Error("pool metric should still be exported when only db-notify is disabled")
+	}
+}
diff --git a/pkg/sciontool/log/log.go b/pkg/sciontool/log/log.go
index e2dfaf3df..6311207e7 100644
--- a/pkg/sciontool/log/log.go
+++ b/pkg/sciontool/log/log.go
@@ -200,8 +200,24 @@ func (h *slogHandler) Enabled(_ context.Context, level slog.Level) bool {
 func (h *slogHandler) Handle(_ context.Context, r slog.Record) error {
 	level := r.Level.String()
 	msg := r.Message
-	// In a real implementation we might want to include attributes,
-	// but for sciontool we keep it simple for now.
+	if r.NumAttrs() > 0 || len(h.attrs) > 0 {
+		var buf []byte
+		buf = append(buf, msg...)
+		for _, a := range h.attrs {
+			buf = append(buf, ' ')
+			buf = append(buf, a.Key...)
+			buf = append(buf, '=')
+			buf = append(buf, a.Value.String()...)
+		}
+		r.Attrs(func(a slog.Attr) bool {
+			buf = append(buf, ' ')
+			buf = append(buf, a.Key...)
+			buf = append(buf, '=')
+			buf = append(buf, a.Value.String()...)
+			return true
+		})
+		msg = string(buf)
+	}
 	write(level, "slog", "%s", msg)
 	return nil
 }
diff --git a/pkg/sciontool/telemetry/pipeline.go b/pkg/sciontool/telemetry/pipeline.go
index 5b09c2f92..bce065d5b 100644
--- a/pkg/sciontool/telemetry/pipeline.go
+++ b/pkg/sciontool/telemetry/pipeline.go
@@ -6,24 +6,35 @@ package telemetry
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"log/slog"
 	"os"
+	"strings"
 	"sync"
+	"time"
 
 	"github.com/GoogleCloudPlatform/scion/pkg/sciontool/log"
+	"go.opentelemetry.io/otel/attribute"
+	otelmetric "go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/metric/noop"
 	logspb "go.opentelemetry.io/proto/otlp/logs/v1"
 	metricpb "go.opentelemetry.io/proto/otlp/metrics/v1"
 	tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
+	"google.golang.org/api/googleapi"
 )
 
 // Pipeline orchestrates the telemetry collection and forwarding.
 type Pipeline struct {
-	config   *Config
-	receiver *Receiver
-	exporter *CloudExporter
-	filter   *Filter
-	mu       sync.Mutex
-	running  bool
+	config       *Config
+	receiver     *Receiver
+	exporter     *CloudExporter
+	filter       *Filter
+	mu           sync.Mutex
+	running      bool
+	healthCancel context.CancelFunc
+	exportErrors otelmetric.Int64Counter
+	meter        otelmetric.Meter
 }
 
 // New creates a new telemetry pipeline.
@@ -69,10 +80,21 @@ func (p *Pipeline) Start(ctx context.Context) error {
 		if envVal := os.Getenv(EnvGCPCredentials); envVal == "" {
 			source = "well-known-path"
 		}
-		log.Info("GCP telemetry credentials: %s (source: %s, project: %s)",
-			p.config.GCPCredentialsFile, source, p.config.ProjectID)
+		slog.Info("telemetry pipeline credential resolution",
+			"credentials_file", p.config.GCPCredentialsFile,
+			"source", source,
+			"project_id", p.config.ProjectID,
+			"provider", p.config.CloudProvider,
+			"cloud_configured", p.config.IsCloudConfigured(),
+		)
 	} else if p.config.IsCloudConfigured() {
-		log.Info("GCP telemetry credentials: none (using ADC fallback)")
+		slog.Info("telemetry pipeline credential resolution",
+			"credentials_file", "",
+			"source", "adc",
+			"project_id", p.config.ProjectID,
+			"provider", p.config.CloudProvider,
+			"cloud_configured", true,
+		)
 	}
 
 	// Create cloud exporter if configured
@@ -93,7 +115,11 @@ func (p *Pipeline) Start(ctx context.Context) error {
 			log.Info("Cloud exporter initialized (%s, project: %s)", mode, p.config.ProjectID)
 		}
 	} else {
-		log.Debug("Cloud export not configured - telemetry will only be received locally")
+		slog.Warn("telemetry cloud export not configured",
+			"reason", "no credentials or endpoint",
+			"env_checked", EnvGCPCredentials,
+			"well_known_path", WellKnownGCPCredentialsPath,
+		)
 	}
 
 	// Create receiver with span and metric handlers
@@ -108,6 +134,12 @@ func (p *Pipeline) Start(ctx context.Context) error {
 	}
 
 	p.running = true
+
+	// Register pipeline health gauge and export error counter.
+	if p.config.IsCloudConfigured() && p.exporter != nil {
+		p.initSelfMetrics(ctx)
+	}
+
 	log.Info("Telemetry pipeline started (gRPC: %d, HTTP: %d)", p.config.GRPCPort, p.config.HTTPPort)
 
 	return nil
@@ -128,6 +160,12 @@ func (p *Pipeline) Stop(ctx context.Context) error {
 
 	var errs []error
 
+	// Stop health gauge ticker
+	if p.healthCancel != nil {
+		p.healthCancel()
+		p.healthCancel = nil
+	}
+
 	// Stop receiver first
 	if p.receiver != nil {
 		if err := p.receiver.Stop(ctx); err != nil {
@@ -188,6 +226,7 @@ func (p *Pipeline) handleSpans(ctx context.Context, resourceSpans []*tracepb.Res
 	// Forward to cloud exporter if available
 	if p.exporter != nil {
 		if err := p.exporter.ExportProtoSpans(ctx, filtered); err != nil {
+			p.recordExportError(ctx, "spans", err)
 			log.Error("Failed to export spans to cloud: %v", err)
 			return err
 		}
@@ -249,6 +288,7 @@ func (p *Pipeline) handleMetrics(ctx context.Context, resourceMetrics []*metricp
 	// directly via a MeterProvider.
 	if p.exporter != nil {
 		if err := p.exporter.ExportProtoMetrics(ctx, resourceMetrics); err != nil {
+			p.recordExportError(ctx, "metrics", err)
 			log.Error("Failed to export metrics to cloud: %v", err)
 			return err
 		}
@@ -274,6 +314,7 @@ func (p *Pipeline) handleLogs(ctx context.Context, resourceLogs []*logspb.Resour
 	// Forward to cloud exporter if available
 	if p.exporter != nil {
 		if err := p.exporter.ExportProtoLogs(ctx, resourceLogs); err != nil {
+			p.recordExportError(ctx, "logs", err)
 			log.Error("Failed to export logs to cloud: %v", err)
 			return err
 		}
@@ -282,3 +323,126 @@ func (p *Pipeline) handleLogs(ctx context.Context, resourceLogs []*logspb.Resour
 
 	return nil
 }
+
+// initSelfMetrics creates a minimal MeterProvider for self-monitoring metrics
+// (pipeline health gauge and export error counter) and starts the health ticker.
+func (p *Pipeline) initSelfMetrics(ctx context.Context) {
+	providers, err := NewProviders(ctx, p.config, true)
+	if err != nil || providers == nil || providers.MeterProvider == nil {
+		log.Debug("Could not create MeterProvider for pipeline self-metrics: %v", err)
+		p.meter = noop.Meter{}
+	} else {
+		// Shut down TracerProvider and LoggerProvider immediately — we only
+		// need the MeterProvider for self-monitoring metrics.
+		if providers.TracerProvider != nil {
+			_ = providers.TracerProvider.Shutdown(ctx)
+		}
+		if providers.LoggerProvider != nil {
+			_ = providers.LoggerProvider.Shutdown(ctx)
+		}
+		p.meter = providers.MeterProvider.Meter("github.com/GoogleCloudPlatform/scion/pkg/sciontool/telemetry")
+	}
+
+	p.exportErrors, err = p.meter.Int64Counter("scion.telemetry.export.errors",
+		otelmetric.WithDescription("Count of telemetry export failures by signal type"),
+		otelmetric.WithUnit("{error}"),
+	)
+	if err != nil {
+		log.Debug("Failed to create export error counter: %v", err)
+	}
+
+	p.startHealthGauge(ctx, providers)
+}
+
+// startHealthGauge registers the scion.telemetry.pipeline.status gauge and
+// starts a background ticker that reports value 1 every 60 seconds.
+func (p *Pipeline) startHealthGauge(ctx context.Context, providers *Providers) {
+	gauge, err := p.meter.Int64Gauge("scion.telemetry.pipeline.status",
+		otelmetric.WithDescription("Pipeline health status (1=running)"),
+		otelmetric.WithUnit("{status}"),
+	)
+	if err != nil {
+		log.Debug("Failed to create pipeline health gauge: %v", err)
+		if providers != nil && providers.MeterProvider != nil {
+			_ = providers.MeterProvider.Shutdown(ctx)
+		}
+		return
+	}
+
+	attrs := otelmetric.WithAttributes(
+		attribute.String("scion.telemetry.provider", p.config.CloudProvider),
+		attribute.String("scion.telemetry.project_id", p.config.ProjectID),
+	)
+
+	healthCtx, cancel := context.WithCancel(ctx)
+	p.healthCancel = cancel
+
+	gauge.Record(healthCtx, 1, attrs)
+
+	go func() {
+		ticker := time.NewTicker(60 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-healthCtx.Done():
+				if providers != nil && providers.MeterProvider != nil {
+					shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
+					_ = providers.MeterProvider.Shutdown(shutdownCtx)
+					shutdownCancel()
+				}
+				return
+			case <-ticker.C:
+				gauge.Record(healthCtx, 1, attrs)
+			}
+		}
+	}()
+}
+
+// recordExportError increments the export error counter if registered.
+func (p *Pipeline) recordExportError(ctx context.Context, signal string, err error) {
+	if p.exportErrors == nil {
+		return
+	}
+	p.exportErrors.Add(ctx, 1,
+		otelmetric.WithAttributes(
+			attribute.String("signal", signal),
+			attribute.String("error_type", classifyError(err)),
+		),
+	)
+}
+
+// classifyError buckets an export error into a category for metric attributes.
+func classifyError(err error) string {
+	if err == nil {
+		return "none"
+	}
+
+	if errors.Is(err, context.DeadlineExceeded) {
+		return "timeout"
+	}
+	if errors.Is(err, context.Canceled) {
+		return "timeout"
+	}
+
+	var gapiErr *googleapi.Error
+	if errors.As(err, &gapiErr) {
+		switch gapiErr.Code {
+		case 401, 403:
+			return "auth"
+		case 429:
+			return "quota"
+		}
+	}
+
+	msg := strings.ToLower(err.Error())
+	switch {
+	case strings.Contains(msg, "unauthorized") || strings.Contains(msg, "unauthenticated") || strings.Contains(msg, "permission denied"):
+		return "auth"
+	case strings.Contains(msg, "quota") || strings.Contains(msg, "rate limit") || strings.Contains(msg, "resource exhausted"):
+		return "quota"
+	case strings.Contains(msg, "deadline exceeded") || strings.Contains(msg, "timeout"):
+		return "timeout"
+	}
+
+	return "other"
+}
diff --git a/pkg/sciontool/telemetry/pipeline_health_test.go b/pkg/sciontool/telemetry/pipeline_health_test.go
new file mode 100644
index 000000000..b37755650
--- /dev/null
+++ b/pkg/sciontool/telemetry/pipeline_health_test.go
@@ -0,0 +1,189 @@
+/*
+Copyright 2025 The Scion Authors.
+*/
+
+package telemetry
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+	"time"
+
+	"google.golang.org/api/googleapi"
+)
+
+func TestPipeline_HealthGauge_Registers(t *testing.T) {
+	clearTelemetryEnv()
+	os.Setenv(EnvEnabled, "true")
+	os.Setenv(EnvCloudEnabled, "false")
+	os.Setenv(EnvGRPCPort, "54401")
+	os.Setenv(EnvHTTPPort, "54402")
+	defer clearTelemetryEnv()
+
+	cfg := &Config{
+		Enabled:       true,
+		CloudEnabled:  false,
+		GRPCPort:      54401,
+		HTTPPort:      54402,
+		CloudProvider: "",
+	}
+	pipeline := NewWithConfig(cfg)
+	if pipeline == nil {
+		t.Fatal("Expected non-nil pipeline")
+	}
+
+	ctx := context.Background()
+	if err := pipeline.Start(ctx); err != nil {
+		t.Fatalf("Failed to start pipeline: %v", err)
+	}
+	defer func() {
+		stopCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		defer cancel()
+		if err := pipeline.Stop(stopCtx); err != nil {
+			t.Errorf("pipeline.Stop: %v", err)
+		}
+	}()
+
+	// Without cloud configured, health gauge should not be started
+	if pipeline.healthCancel != nil {
+		t.Error("Health gauge should not be started without cloud exporter")
+	}
+}
+
+func TestPipeline_HealthGauge_StopsOnStop(t *testing.T) {
+	clearTelemetryEnv()
+	os.Setenv(EnvEnabled, "true")
+	os.Setenv(EnvCloudEnabled, "false")
+	os.Setenv(EnvGRPCPort, "54403")
+	os.Setenv(EnvHTTPPort, "54404")
+	defer clearTelemetryEnv()
+
+	cfg := &Config{
+		Enabled:  true,
+		GRPCPort: 54403,
+		HTTPPort: 54404,
+	}
+	pipeline := NewWithConfig(cfg)
+	if pipeline == nil {
+		t.Fatal("Expected non-nil pipeline")
+	}
+
+	ctx := context.Background()
+	if err := pipeline.Start(ctx); err != nil {
+		t.Fatalf("Failed to start pipeline: %v", err)
+	}
+
+	// Stop the pipeline and verify healthCancel is cleared
+	stopCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	if err := pipeline.Stop(stopCtx); err != nil {
+		t.Fatalf("Failed to stop pipeline: %v", err)
+	}
+
+	if pipeline.healthCancel != nil {
+		t.Error("healthCancel should be nil after Stop()")
+	}
+	if pipeline.IsRunning() {
+		t.Error("Pipeline should not be running after Stop()")
+	}
+}
+
+func TestPipeline_ExportErrors_NilCounter(t *testing.T) {
+	cfg := &Config{
+		Enabled:  true,
+		GRPCPort: 54405,
+		HTTPPort: 54406,
+	}
+	pipeline := NewWithConfig(cfg)
+	if pipeline == nil {
+		t.Fatal("Expected non-nil pipeline")
+	}
+
+	// recordExportError should be safe to call with nil counter
+	pipeline.recordExportError(context.Background(), "metrics", errors.New("test error"))
+}
+
+func TestClassifyError(t *testing.T) {
+	tests := []struct {
+		name     string
+		err      error
+		expected string
+	}{
+		{
+			name:     "nil error",
+			err:      nil,
+			expected: "none",
+		},
+		{
+			name:     "deadline exceeded",
+			err:      context.DeadlineExceeded,
+			expected: "timeout",
+		},
+		{
+			name:     "context canceled",
+			err:      context.Canceled,
+			expected: "timeout",
+		},
+		{
+			name:     "wrapped deadline exceeded",
+			err:      errors.Join(errors.New("export failed"), context.DeadlineExceeded),
+			expected: "timeout",
+		},
+		{
+			name:     "googleapi 401",
+			err:      &googleapi.Error{Code: 401, Message: "unauthorized"},
+			expected: "auth",
+		},
+		{
+			name:     "googleapi 403",
+			err:      &googleapi.Error{Code: 403, Message: "forbidden"},
+			expected: "auth",
+		},
+		{
+			name:     "googleapi 429",
+			err:      &googleapi.Error{Code: 429, Message: "too many requests"},
+			expected: "quota",
+		},
+		{
+			name:     "permission denied string",
+			err:      errors.New("rpc error: code = PermissionDenied desc = permission denied"),
+			expected: "auth",
+		},
+		{
+			name:     "unauthenticated string",
+			err:      errors.New("rpc error: code = Unauthenticated"),
+			expected: "auth",
+		},
+		{
+			name:     "quota string",
+			err:      errors.New("resource exhausted: quota exceeded"),
+			expected: "quota",
+		},
+		{
+			name:     "rate limit string",
+			err:      errors.New("rate limit exceeded"),
+			expected: "quota",
+		},
+		{
+			name:     "timeout string",
+			err:      errors.New("request timeout"),
+			expected: "timeout",
+		},
+		{
+			name:     "generic error",
+			err:      errors.New("connection refused"),
+			expected: "other",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := classifyError(tt.err)
+			if result != tt.expected {
+				t.Errorf("classifyError(%v) = %q, want %q", tt.err, result, tt.expected)
+			}
+		})
+	}
+}