feat: add bounded process metrics and private monitoring endpoint

cmd/orbit/main.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"git.yuntuo.hk/k2safe/orbit/internal/config"
+	"git.yuntuo.hk/k2safe/orbit/internal/observability"
 	"git.yuntuo.hk/k2safe/orbit/internal/storage/mysql"
 	"git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver"
 	core "git.yuntuo.hk/k2safe/orbit/model"
@@ -53,12 +54,24 @@ func run() error {
 		return err
 	}
 	cancel()
-	server := httpserver.New(cfg.HTTPAddr, func(ctx context.Context) error {
+	var metrics *observability.Registry
+	var metricsListener net.Listener
+	if cfg.MetricsAddr != "" {
+		metricsListener, err = net.Listen("tcp", cfg.MetricsAddr)
+		if err != nil {
+			return errors.New("metrics listener could not bind")
+		}
+		defer metricsListener.Close()
+		metrics = observability.NewRegistry()
+		metrics.RecordDatabase(db.Stats())
+	}
+	ready := func(ctx context.Context) error {
 		if err := db.PingContext(ctx); err != nil {
 			return err
 		}
 		return mysql.CheckSchema(ctx, db)
-	}, service.New(mysql.NewTaskStore(db)))
+	}
+	server := httpserver.New(cfg.HTTPAddr, ready, service.New(mysql.NewTaskStore(db)))
 	httpserver.RegisterApplicationResources(server, resources.New(mysql.NewTaskStore(db)))
 	node, err := core.NewID()
 	if err != nil {
@@ -86,6 +99,7 @@ func run() error {
 	httpserver.RegisterSchedules(server, schedule.New(schedules))
 	httpserver.RegisterWorkflowDefinitions(server, definition.New(mysql.NewWorkflowDefinitionStore(db)))
 	httpserver.RegisterWorkflows(server, instance.New(mysql.NewWorkflowInstanceStore(db)), control.New(mysql.NewWorkflowControlStore(db)), weavesignal.New(mysql.NewWorkflowSignalStore(db)), recovery.New(mysql.NewWorkflowRecoveryStore(db)))
+	httpserver.WrapMetrics(server, metrics)
 	workflows, err := mysql.NewWorkflowRuntime(db, node)
 	if err != nil {
 		return err
@@ -101,9 +115,11 @@ func run() error {
 			case <-workflowCtx.Done():
 				return
 			case <-ticker.C:
+				started := time.Now()
 				call, cancel := context.WithTimeout(workflowCtx, 5*time.Second)
 				_, err := workflows.Tick(call, 100)
 				cancel()
+				metrics.RecordBatch("workflow", started, err)
 				if err != nil && workflowCtx.Err() == nil {
 					slog.Error("workflow batch failed", "node_id", node)
 				}
@@ -122,9 +138,11 @@ func run() error {
 			case <-scheduleCtx.Done():
 				return
 			case <-ticker.C:
+				started := time.Now()
 				call, cancel := context.WithTimeout(scheduleCtx, 5*time.Second)
 				_, err := schedules.Tick(call, 100)
 				cancel()
+				metrics.RecordBatch("schedule", started, err)
 				if err != nil && scheduleCtx.Err() == nil {
 					slog.Error("schedule batch failed", "node_id", node)
 				}
@@ -148,9 +166,11 @@ func run() error {
 			case <-recoveryCtx.Done():
 				return
 			case <-ticker.C:
+				started := time.Now()
 				call, cancel := context.WithTimeout(recoveryCtx, 5*time.Second)
 				_, err := executions.Recover(call, 100)
 				cancel()
+				metrics.RecordBatch("execution_recovery", started, err)
 				if err != nil && recoveryCtx.Err() == nil {
 					slog.Error("execution recovery batch failed", "node_id", node)
 				}
@@ -158,11 +178,38 @@ func run() error {
 		}
 	}()
 	defer func() { stopRecovery(); <-recoveryDone }()
+	if metrics != nil {
+		metricsCtx, stopMetrics := context.WithCancel(root)
+		metricsDone := make(chan struct{})
+		go func() {
+			defer close(metricsDone)
+			ticker := time.NewTicker(5 * time.Second)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-metricsCtx.Done():
+					return
+				case <-ticker.C:
+					metrics.RecordDatabase(db.Stats())
+					metrics.RecordProcess()
+				}
+			}
+		}()
+		defer func() { stopMetrics(); <-metricsDone }()
+	}
 	listener, err := net.Listen("tcp", cfg.HTTPAddr)
 	if err != nil {
 		return err
 	}
-	done := make(chan error, 1)
+	defer server.Close()
+	var metricsServer *http.Server
+	done := make(chan error, 2)
+	if metrics != nil {
+		metricsServer = httpserver.NewMetricsServer(cfg.MetricsAddr, metrics)
+		defer metricsServer.Close()
+		go func() { done <- metricsServer.Serve(metricsListener) }()
+		slog.Info("Orbit metrics server starting", "metrics_address", metricsListener.Addr().String())
+	}
 	go func() { done <- server.Serve(listener) }()
 	slog.Info("Orbit HTTP server starting", "address", listener.Addr().String(), "scope", "health, task, queue, schedule, subscription, workflow and Worker APIs", "node_id", node)
 	select {
@@ -178,10 +225,20 @@ func run() error {
 			_ = server.Close()
 			return err
 		}
-		err := <-done
-		if errors.Is(err, http.ErrServerClosed) {
-			return nil
+		if metricsServer != nil {
+			if err := metricsServer.Shutdown(shutdown); err != nil {
+				return err
+			}
 		}
-		return err
+		count := 1
+		if metricsServer != nil {
+			count++
+		}
+		for i := 0; i < count; i++ {
+			if err := <-done; !errors.Is(err, http.ErrServerClosed) {
+				return err
+			}
+		}
+		return nil
 	}
 }

cmd/orbit/metrics_integration_test.go

@@ -0,0 +1,162 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+)
+
+// TestIntegrationMetricsProcess uses the same compiled binary and disposable
+// database as the full regression. It never reads an existing business DSN.
+func TestIntegrationMetricsProcess(t *testing.T) {
+	binary, dsn := os.Getenv("ORBIT_TEST_SERVER_BINARY"), os.Getenv("ORBIT_TEST_MYSQL_DSN")
+	if binary == "" || dsn == "" {
+		t.Skip("requires compiled Orbit and disposable database")
+	}
+	for _, enabled := range []bool{true, false} {
+		t.Run(map[bool]string{true: "enabled", false: "disabled"}[enabled], func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+			defer cancel()
+			logPath := filepath.Join(t.TempDir(), "server.log")
+			log, err := os.Create(logPath)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer log.Close()
+			cmd := exec.CommandContext(ctx, binary)
+			for _, entry := range os.Environ() {
+				if strings.HasPrefix(entry, "ORBIT_MYSQL_DSN=") || strings.HasPrefix(entry, "ORBIT_HTTP_ADDR=") || strings.HasPrefix(entry, "ORBIT_METRICS_ADDR=") {
+					continue
+				}
+				cmd.Env = append(cmd.Env, entry)
+			}
+			addr := ""
+			if enabled {
+				addr = "127.0.0.1:0"
+			}
+			cmd.Env = append(cmd.Env, "ORBIT_MYSQL_DSN="+dsn, "ORBIT_HTTP_ADDR=127.0.0.1:0", "ORBIT_METRICS_ADDR="+addr)
+			cmd.Stdout = log
+			cmd.Stderr = log
+			if err = cmd.Start(); err != nil {
+				t.Fatal(err)
+			}
+			done := make(chan struct{})
+			var exit error
+			go func() { exit = cmd.Wait(); close(done) }()
+			defer func() { cancel(); <-done }()
+			var api, metrics string
+			for api == "" || (enabled && metrics == "") {
+				data, _ := os.ReadFile(logPath)
+				for _, line := range strings.Split(string(data), "\n") {
+					var event struct {
+						Message        string `json:"msg"`
+						Address        string `json:"address"`
+						MetricsAddress string `json:"metrics_address"`
+					}
+					if json.Unmarshal([]byte(line), &event) != nil {
+						continue
+					}
+					if event.Message == "Orbit HTTP server starting" {
+						api = "http://" + event.Address
+					}
+					if event.Message == "Orbit metrics server starting" {
+						metrics = "http://" + event.MetricsAddress
+					}
+				}
+				select {
+				case <-done:
+					t.Fatal("server exited before listeners became ready")
+				case <-ctx.Done():
+					t.Fatal("listener startup deadline")
+				case <-time.After(10 * time.Millisecond):
+				}
+			}
+			client := &http.Client{Timeout: 2 * time.Second}
+			get := func(url string, want int) string {
+				t.Helper()
+				resp, err := client.Get(url)
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer resp.Body.Close()
+				raw, err := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
+				if err != nil || resp.StatusCode != want {
+					t.Fatalf("unexpected response: status=%d error=%v", resp.StatusCode, err)
+				}
+				return string(raw)
+			}
+			get(api+"/health/ready", 200)
+			get(api+"/v1/tasks/private-business-id?business_key=private-business-key", 401)
+			get(api+"/metrics", 404)
+			if enabled {
+				if api == metrics {
+					t.Fatal("metrics shares business listener")
+				}
+				get(metrics+"/health/ready", 404)
+				get(metrics+"/v1/tasks", 404)
+				get(metrics+"/metrics?token=private-key", 400)
+				for {
+					body := get(metrics+"/metrics", 200)
+					for _, secret := range []string{dsn, "private-business", "private-key"} {
+						if strings.Contains(body, secret) {
+							t.Fatal("metric data contains request data or credentials")
+						}
+					}
+					if !strings.Contains(body, "orbit_http_requests_total{") || !strings.Contains(body, "orbit_db_connections_max_open 20\n") {
+						t.Fatal("process metrics not connected")
+					}
+					observed := false
+					for _, line := range strings.Split(body, "\n") {
+						prefix := `orbit_runtime_batch_total{component="execution_recovery",outcome="success"} `
+						if strings.HasPrefix(line, prefix) {
+							n, err := strconv.ParseFloat(strings.TrimPrefix(line, prefix), 64)
+							observed = err == nil && n > 0
+						}
+					}
+					if observed {
+						break
+					}
+					select {
+					case <-ctx.Done():
+						t.Fatal("background batch was not observed")
+					case <-time.After(50 * time.Millisecond):
+					}
+				}
+			} else if metrics != "" {
+				t.Fatal("disabled metrics listener was started")
+			}
+			if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
+				t.Fatal(err)
+			}
+			select {
+			case <-done:
+				if exit != nil {
+					t.Fatal("graceful exit failed", exit)
+				}
+			case <-ctx.Done():
+				t.Fatal("shutdown hung")
+			}
+			for _, url := range []string{api, metrics} {
+				if url == "" {
+					continue
+				}
+				c, err := net.DialTimeout("tcp", strings.TrimPrefix(url, "http://"), time.Second)
+				if err == nil {
+					c.Close()
+					t.Fatal("listener remained after shutdown")
+				}
+			}
+			t.Log("real process: metrics enable/disable, separate listeners, HTTP/database/background samples, request privacy and graceful closure verified")
+		})
+	}
+}