orbit/cmd/orbit/metrics_integration_test.go

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 !enabled {
				select {
				case <-done:
					t.Fatal("server exited after metrics was disabled")
				case <-time.After(2200 * time.Millisecond):
				}
			}
			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")
		})
	}
}