feat: 完善服务监控与下游重投

2026-08-14 17:21:29 +08:00
parent d30d9ea4d0
commit 1ef4380422
50 changed files with 1279 additions and 82 deletions
@@ -201,6 +201,13 @@ func onlineAccounts() []string {
 	return accounts
 }

+// ActiveConnectionCount exposes only an aggregate gauge; account and remote-IP labels are intentionally excluded.
+func ActiveConnectionCount() int {
+	downstreamRegistry.RLock()
+	defer downstreamRegistry.RUnlock()
+	return len(downstreamRegistry.byConn)
+}
+
 // DisconnectAccount closes every live downstream CMPP session for an
 // application account. The normal connection-close callback removes registry
 // and presence state and reports the disconnect to the API.
@@ -0,0 +1,80 @@
+package metrics
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"runtime"
+	"sync/atomic"
+	"time"
+)
+
+var startedAt = time.Now()
+var submitAccepted atomic.Uint64
+var submitFailed atomic.Uint64
+var submitDurationNanoseconds atomic.Uint64
+
+type Snapshot struct {
+	UpstreamDesired       int
+	UpstreamConnected     int
+	DownstreamConnected   int
+	SubmitWorkerUp        bool
+	QueueAvailable        bool
+	QueuePending          int64
+	QueueLag              int64
+	QueueOldestAgeSeconds float64
+}
+
+type SnapshotFunc func(context.Context) Snapshot
+
+func ObserveSubmit(accepted bool, duration time.Duration) {
+	if accepted {
+		submitAccepted.Add(1)
+	} else {
+		submitFailed.Add(1)
+	}
+	submitDurationNanoseconds.Add(uint64(max(duration, 0)))
+}
+
+func Handler(load SnapshotFunc) http.Handler {
+	return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
+		if request.Method != http.MethodGet || request.URL.Path != "/metrics" {
+			response.WriteHeader(http.StatusNotFound)
+			return
+		}
+		ctx, cancel := context.WithTimeout(request.Context(), time.Second)
+		defer cancel()
+		snapshot := Snapshot{}
+		if load != nil {
+			snapshot = load(ctx)
+		}
+		var memory runtime.MemStats
+		runtime.ReadMemStats(&memory)
+		accepted := submitAccepted.Load()
+		failed := submitFailed.Load()
+		count := accepted + failed
+		response.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
+		response.Header().Set("Cache-Control", "no-store")
+		fmt.Fprintf(response, "# HELP cmpp_gateway_process_uptime_seconds Gateway process uptime.\n# TYPE cmpp_gateway_process_uptime_seconds gauge\ncmpp_gateway_process_uptime_seconds %f\n", time.Since(startedAt).Seconds())
+		fmt.Fprintf(response, "# HELP cmpp_gateway_go_goroutines Current goroutine count.\n# TYPE cmpp_gateway_go_goroutines gauge\ncmpp_gateway_go_goroutines %d\n", runtime.NumGoroutine())
+		fmt.Fprintf(response, "# HELP cmpp_gateway_go_heap_alloc_bytes Current Go heap allocation.\n# TYPE cmpp_gateway_go_heap_alloc_bytes gauge\ncmpp_gateway_go_heap_alloc_bytes %d\n", memory.HeapAlloc)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_submit_total Upstream submit attempts by bounded result.\n# TYPE cmpp_gateway_submit_total counter\ncmpp_gateway_submit_total{result=\"accepted\"} %d\ncmpp_gateway_submit_total{result=\"failed\"} %d\n", accepted, failed)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_submit_duration_seconds_sum Total upstream submit duration.\n# TYPE cmpp_gateway_submit_duration_seconds_sum counter\ncmpp_gateway_submit_duration_seconds_sum %f\n", float64(submitDurationNanoseconds.Load())/float64(time.Second))
+		fmt.Fprintf(response, "# HELP cmpp_gateway_submit_duration_seconds_count Total measured upstream submits.\n# TYPE cmpp_gateway_submit_duration_seconds_count counter\ncmpp_gateway_submit_duration_seconds_count %d\n", count)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_upstream_connections Desired and live supplier connections.\n# TYPE cmpp_gateway_upstream_connections gauge\ncmpp_gateway_upstream_connections{state=\"desired\"} %d\ncmpp_gateway_upstream_connections{state=\"connected\"} %d\n", snapshot.UpstreamDesired, snapshot.UpstreamConnected)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_downstream_connections Authenticated client connections.\n# TYPE cmpp_gateway_downstream_connections gauge\ncmpp_gateway_downstream_connections %d\n", snapshot.DownstreamConnected)
+		fmt.Fprintf(response, "# HELP cmpp_gateway_submit_worker_up Whether the submit worker was initialized.\n# TYPE cmpp_gateway_submit_worker_up gauge\ncmpp_gateway_submit_worker_up %d\n", boolNumber(snapshot.SubmitWorkerUp))
+		if snapshot.QueueAvailable {
+			fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_pending Pending entries owned by the consumer group.\n# TYPE cmpp_gateway_submit_queue_pending gauge\ncmpp_gateway_submit_queue_pending %d\n", snapshot.QueuePending)
+			fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_lag Undelivered entries for the consumer group.\n# TYPE cmpp_gateway_submit_queue_lag gauge\ncmpp_gateway_submit_queue_lag %d\n", snapshot.QueueLag)
+			fmt.Fprintf(response, "# HELP cmpp_gateway_submit_queue_oldest_pending_age_seconds Age of the oldest pending entry.\n# TYPE cmpp_gateway_submit_queue_oldest_pending_age_seconds gauge\ncmpp_gateway_submit_queue_oldest_pending_age_seconds %f\n", snapshot.QueueOldestAgeSeconds)
+		}
+	})
+}
+
+func boolNumber(value bool) int {
+	if value {
+		return 1
+	}
+	return 0
+}
@@ -0,0 +1,29 @@
+package metrics
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestMetricsHandlerExportsOnlyAggregateGatewayState(t *testing.T) {
+	ObserveSubmit(true, 20*time.Millisecond)
+	request := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+	response := httptest.NewRecorder()
+	Handler(func(context.Context) Snapshot {
+		return Snapshot{UpstreamDesired: 2, UpstreamConnected: 1, DownstreamConnected: 3, SubmitWorkerUp: true, QueueAvailable: true, QueuePending: 4, QueueLag: 5, QueueOldestAgeSeconds: 12}
+	}).ServeHTTP(response, request)
+
+	body := response.Body.String()
+	if response.Code != http.StatusOK || !strings.Contains(body, "cmpp_gateway_submit_queue_pending 4") || !strings.Contains(body, "cmpp_gateway_upstream_connections{state=\"connected\"} 1") {
+		t.Fatalf("unexpected metrics response: code=%d body=%s", response.Code, body)
+	}
+	for _, forbidden := range []string{"phone_number", "message_id", "channel_id"} {
+		if strings.Contains(body, forbidden) {
+			t.Fatalf("metrics expose forbidden label %q", forbidden)
+		}
+	}
+}
@@ -12,6 +12,7 @@ import (
 	"sync"
 	"time"

+	"cmpp-platform/gateway/internal/metrics"
 	"cmpp-platform/gateway/internal/queue"
 	"cmpp-platform/gateway/internal/ratelimit"
 	"cmpp-platform/gateway/internal/upstream"
@@ -206,6 +207,7 @@ func (w *Worker) processMessage(ctx context.Context, message redis.XMessage) err
 }

 func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand) error {
+	startedAt := time.Now()
 	if w.Limiter != nil {
 		if _, err := w.Limiter.Wait(ctx, command.ChannelID, command.Route.RateLimitPerSecond); err != nil {
 			return err
@@ -219,6 +221,8 @@ func (w *Worker) handleCommand(ctx context.Context, command queue.SubmitCommand)
 		submit = w.Upstream.Submit
 	}
 	result, err := submit(ctx, command)
+	accepted := err == nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted")
+	metrics.ObserveSubmit(accepted, time.Since(startedAt))
 	if err != nil && (result.SubmitStatus == "" || result.SubmitStatus == "accepted") {
 		return err
 	}
@@ -146,6 +146,21 @@ func (m *Manager) ensureDefaultsLocked() {
 	}
 }

+// ConnectionCounts returns bounded platform totals without channel identifiers to prevent time-series cardinality growth.
+func (m *Manager) ConnectionCounts() (desired int, connected int) {
+	m.mu.Lock()
+	pools := make([]*connectionPool, 0, len(m.conns))
+	for _, pool := range m.conns {
+		pools = append(pools, pool)
+	}
+	m.mu.Unlock()
+	for _, pool := range pools {
+		desired += max(pool.config.DesiredConnections, 1)
+		connected += pool.countActiveConnections()
+	}
+	return desired, connected
+}
+
 func (m *Manager) newConnectionPool(channelID string, connectionID string, config queue.UpstreamConfig) *connectionPool {
 	return &connectionPool{
 		channelID:    channelID,