From 49bf42303b4e94b7608eb39e14b1e6fea7a771f7 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Thu, 9 Jul 2026 13:08:53 +1000 Subject: [PATCH 1/3] Replace service request logs with metrics --- agent/internal/agent/agent.go | 2 + agent/internal/agent/drift.go | 10 +- agent/internal/agent/reporting.go | 8 + agent/internal/agent/run.go | 65 +++ agent/internal/container/stats.go | 247 +++++++++++ agent/internal/container/stats_test.go | 119 ++++++ agent/internal/metrics/victoria.go | 93 ++++- agent/internal/metrics/victoria_test.go | 41 ++ agent/internal/traefik/static.go | 113 +++++ agent/internal/traefik/static_test.go | 50 +++ .../[id]/{request-stats => metrics}/route.ts | 28 +- .../details/service-details-overview.tsx | 392 +++++++++++++----- web/lib/victoria-logs.ts | 287 ------------- web/lib/victoria-metrics.ts | 362 +++++++++++++++- web/public/setup.sh | 25 ++ web/tests/victoria-logs-request-stats.test.ts | 161 ------- .../victoria-metrics-service-metrics.test.ts | 125 ++++++ 17 files changed, 1554 insertions(+), 574 deletions(-) create mode 100644 agent/internal/container/stats.go create mode 100644 agent/internal/container/stats_test.go create mode 100644 agent/internal/metrics/victoria_test.go create mode 100644 agent/internal/traefik/static_test.go rename web/app/api/services/[id]/{request-stats => metrics}/route.ts (61%) delete mode 100644 web/tests/victoria-logs-request-stats.test.ts create mode 100644 web/tests/victoria-metrics-service-metrics.test.ts diff --git a/agent/internal/agent/agent.go b/agent/internal/agent/agent.go index 6a3b4f0b..a1a4f7fb 100644 --- a/agent/internal/agent/agent.go +++ b/agent/internal/agent/agent.go @@ -127,6 +127,8 @@ func NewAgent( type MetricsSender interface { SendSystemStats(stats *health.SystemStats, collectedAt time.Time) error + SendContainerStats(stats []container.ResourceStats, collectedAt time.Time) error + SendPrometheusMetrics(data []byte, extraLabels map[string]string) error } func (a *Agent) GetState() AgentState { diff --git a/agent/internal/agent/drift.go b/agent/internal/agent/drift.go index 6d083884..18216d87 100644 --- a/agent/internal/agent/drift.go +++ b/agent/internal/agent/drift.go @@ -575,13 +575,19 @@ func (a *Agent) updateTraefik() error { } needsRestart := false + metricsRestart, err := traefik.EnsureMetricsConfig() + if err != nil { + return fmt.Errorf("failed to ensure Traefik metrics config: %w", err) + } + needsRestart = metricsRestart + if len(tcpPorts) > 0 || len(udpPorts) > 0 { log.Printf("[reconcile] ensuring L4 entry points: %d TCP, %d UDP", len(tcpPorts), len(udpPorts)) - var err error - needsRestart, err = traefik.EnsureEntryPoints(tcpPorts, udpPorts) + entryPointsRestart, err := traefik.EnsureEntryPoints(tcpPorts, udpPorts) if err != nil { return fmt.Errorf("failed to ensure entry points: %w", err) } + needsRestart = needsRestart || entryPointsRestart } log.Printf("[reconcile] updating Traefik routes (HTTP: %d, TCP: %d, UDP: %d)", len(expectedHttpRoutes), len(tcpRoutes), len(udpRoutes)) diff --git a/agent/internal/agent/reporting.go b/agent/internal/agent/reporting.go index d83db657..ab4c6574 100644 --- a/agent/internal/agent/reporting.go +++ b/agent/internal/agent/reporting.go @@ -53,6 +53,14 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport if err := a.MetricsSender.SendSystemStats(systemStats, collectedAt); err != nil { log.Printf("[metrics] failed to send system stats: %v", err) } + containerStats, err := container.CollectResourceStats() + if err != nil { + log.Printf("[metrics] failed to collect container stats: %v", err) + return + } + if err := a.MetricsSender.SendContainerStats(containerStats, collectedAt); err != nil { + log.Printf("[metrics] failed to send container stats: %v", err) + } }() } report.NetworkHealth = health.CollectNetworkHealth("wg0") diff --git a/agent/internal/agent/run.go b/agent/internal/agent/run.go index 220a2982..fbf35ca2 100644 --- a/agent/internal/agent/run.go +++ b/agent/internal/agent/run.go @@ -2,13 +2,22 @@ package agent import ( "context" + "fmt" + "io" "log" + "net/http" "time" "techulus/cloud-agent/internal/container" "techulus/cloud-agent/internal/serverless" ) +const ( + traefikMetricsURL = "http://127.0.0.1:9100/metrics" + traefikMetricsInterval = 15 * time.Second + traefikMetricsMaxBytes = 8 * 1024 * 1024 +) + func (a *Agent) Run(ctx context.Context) { if a.Config.RegistryURL != "" && a.Config.RegistryUsername != "" && a.Config.RegistryPassword != "" { if err := container.Login(a.Config.RegistryURL, a.Config.RegistryUsername, a.Config.RegistryPassword, a.Config.RegistryInsecure); err != nil { @@ -33,6 +42,10 @@ func (a *Agent) Run(ctx context.Context) { a.TraefikLogCollector.Start() } + if a.IsProxy && a.MetricsSender != nil { + go a.TraefikMetricsLoop(ctx) + } + if a.IsProxy { gateway := serverless.NewGateway(a) if err := gateway.Start(ctx); err != nil { @@ -81,6 +94,58 @@ func (a *Agent) Run(ctx context.Context) { } } +func (a *Agent) TraefikMetricsLoop(ctx context.Context) { + ticker := time.NewTicker(traefikMetricsInterval) + defer ticker.Stop() + + if err := a.ForwardTraefikMetrics(ctx); err != nil { + log.Printf("[traefik-metrics] initial scrape failed: %v", err) + } + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := a.ForwardTraefikMetrics(ctx); err != nil { + log.Printf("[traefik-metrics] scrape failed: %v", err) + } + } + } +} + +func (a *Agent) ForwardTraefikMetrics(ctx context.Context) error { + if a.MetricsSender == nil { + return nil + } + + request, err := http.NewRequestWithContext(ctx, http.MethodGet, traefikMetricsURL, nil) + if err != nil { + return err + } + + client := &http.Client{Timeout: 10 * time.Second} + response, err := client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + + if response.StatusCode != http.StatusOK { + return fmt.Errorf("unexpected status %d", response.StatusCode) + } + + body, err := io.ReadAll(io.LimitReader(response.Body, traefikMetricsMaxBytes)) + if err != nil { + return err + } + + return a.MetricsSender.SendPrometheusMetrics(body, map[string]string{ + "job": "traefik", + "server_id": a.Config.ServerID, + }) +} + func (a *Agent) StatusReportLoop(ctx context.Context) { a.reportStatus("startup") diff --git a/agent/internal/container/stats.go b/agent/internal/container/stats.go new file mode 100644 index 00000000..960a2051 --- /dev/null +++ b/agent/internal/container/stats.go @@ -0,0 +1,247 @@ +package container + +import ( + "bytes" + "encoding/json" + "fmt" + "math" + "os/exec" + "strconv" + "strings" + "unicode" +) + +type ResourceStats struct { + ContainerID string + ServiceID string + DeploymentID string + CPUUsagePercent float64 + MemoryUsagePercent float64 + MemoryUsedBytes float64 + NetworkReceiveBytes float64 + NetworkTransmitBytes float64 +} + +func CollectResourceStats() ([]ResourceStats, error) { + containers, err := List() + if err != nil { + return nil, err + } + + running := make([]Container, 0, len(containers)) + args := []string{"stats", "--no-stream", "--format", "json"} + for _, c := range containers { + if c.State != "running" || c.ServiceID == "" || c.DeploymentID == "" { + continue + } + running = append(running, c) + args = append(args, c.ID) + } + if len(running) == 0 { + return nil, nil + } + + cmd := exec.Command("podman", args...) + var stderr bytes.Buffer + cmd.Stderr = &stderr + output, err := cmd.Output() + if err != nil { + return nil, fmt.Errorf("failed to collect container stats: %s: %w", stderr.String(), err) + } + + return parsePodmanStatsOutput(output, running) +} + +func parsePodmanStatsOutput(output []byte, containers []Container) ([]ResourceStats, error) { + rows, err := parseStatsRows(output) + if err != nil { + return nil, err + } + + stats := make([]ResourceStats, 0, len(rows)) + for _, row := range rows { + containerID := firstRowString(row, "ID", "Id", "id", "ContainerID", "Container") + container := findStatsContainerByID(containerID, containers) + if container == nil { + name := firstRowString(row, "Name", "Names", "name") + container = findStatsContainerByName(name, containers) + } + if container == nil { + continue + } + + rx, tx := parseNetIO(firstRowString(row, "NetIO", "NetIOBytes", "net_io")) + stats = append(stats, ResourceStats{ + ContainerID: container.ID, + ServiceID: container.ServiceID, + DeploymentID: container.DeploymentID, + CPUUsagePercent: parsePercent(firstRowString(row, "CPUPerc", "CPU", "cpu_percent")), + MemoryUsagePercent: parsePercent(firstRowString(row, "MemPerc", "MEMPerc", "mem_percent")), + MemoryUsedBytes: parseMemUsed(firstRowString(row, "MemUsage", "MemUse", "mem_usage")), + NetworkReceiveBytes: rx, + NetworkTransmitBytes: tx, + }) + } + + return stats, nil +} + +func parseStatsRows(output []byte) ([]map[string]interface{}, error) { + trimmed := strings.TrimSpace(string(output)) + if trimmed == "" { + return nil, nil + } + + if strings.HasPrefix(trimmed, "[") { + var rows []map[string]interface{} + if err := json.Unmarshal([]byte(trimmed), &rows); err != nil { + return nil, fmt.Errorf("failed to parse podman stats JSON array: %w", err) + } + return rows, nil + } + + var rows []map[string]interface{} + for _, line := range strings.Split(trimmed, "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var row map[string]interface{} + if err := json.Unmarshal([]byte(line), &row); err != nil { + return nil, fmt.Errorf("failed to parse podman stats JSON row: %w", err) + } + rows = append(rows, row) + } + return rows, nil +} + +func findStatsContainerByID(value string, containers []Container) *Container { + value = strings.TrimSpace(value) + if value == "" { + return nil + } + + for i := range containers { + containerID := strings.TrimSpace(containers[i].ID) + if value == containerID { + return &containers[i] + } + if containerID != "" && (strings.HasPrefix(containerID, value) || strings.HasPrefix(value, containerID)) { + return &containers[i] + } + } + return nil +} + +func findStatsContainerByName(value string, containers []Container) *Container { + value = strings.TrimPrefix(strings.TrimSpace(value), "/") + if value == "" { + return nil + } + + for i := range containers { + containerName := strings.TrimPrefix(strings.TrimSpace(containers[i].Name), "/") + if value == containerName { + return &containers[i] + } + } + return nil +} + +func firstRowString(row map[string]interface{}, keys ...string) string { + for _, key := range keys { + value, ok := row[key] + if !ok || value == nil { + continue + } + switch v := value.(type) { + case string: + return v + case []interface{}: + if len(v) > 0 { + return fmt.Sprint(v[0]) + } + default: + return fmt.Sprint(v) + } + } + return "" +} + +func parsePercent(value string) float64 { + value = strings.TrimSpace(strings.TrimSuffix(value, "%")) + if value == "" || value == "--" { + return 0 + } + parsed, err := strconv.ParseFloat(value, 64) + if err != nil || !isFinite(parsed) { + return 0 + } + return parsed +} + +func parseMemUsed(value string) float64 { + parts := strings.Split(value, "/") + if len(parts) == 0 { + return 0 + } + return parseByteQuantity(parts[0]) +} + +func parseNetIO(value string) (float64, float64) { + parts := strings.Split(value, "/") + if len(parts) != 2 { + return 0, 0 + } + return parseByteQuantity(parts[0]), parseByteQuantity(parts[1]) +} + +func parseByteQuantity(value string) float64 { + value = strings.TrimSpace(value) + if value == "" || value == "--" { + return 0 + } + + compact := strings.ReplaceAll(value, " ", "") + splitAt := len(compact) + for i, r := range compact { + if !(unicode.IsDigit(r) || r == '.' || r == '-') { + splitAt = i + break + } + } + + numberText := compact[:splitAt] + unit := strings.ToLower(compact[splitAt:]) + parsed, err := strconv.ParseFloat(numberText, 64) + if err != nil || !isFinite(parsed) { + return 0 + } + + switch unit { + case "", "b": + return parsed + case "kb", "k", "kib", "ki": + return parsed * unitMultiplier(unit, 1) + case "mb", "m", "mib", "mi": + return parsed * unitMultiplier(unit, 2) + case "gb", "g", "gib", "gi": + return parsed * unitMultiplier(unit, 3) + case "tb", "t", "tib", "ti": + return parsed * unitMultiplier(unit, 4) + default: + return parsed + } +} + +func unitMultiplier(unit string, power float64) float64 { + base := 1000.0 + if strings.Contains(unit, "i") { + base = 1024.0 + } + return math.Pow(base, power) +} + +func isFinite(value float64) bool { + return !math.IsNaN(value) && !math.IsInf(value, 0) +} diff --git a/agent/internal/container/stats_test.go b/agent/internal/container/stats_test.go new file mode 100644 index 00000000..1edfd61b --- /dev/null +++ b/agent/internal/container/stats_test.go @@ -0,0 +1,119 @@ +package container + +import "testing" + +func TestParsePodmanStatsOutputArray(t *testing.T) { + containers := []Container{ + { + ID: "abcdef1234567890", + Name: "api", + State: "running", + ServiceID: "svc_1", + DeploymentID: "dep_1", + }, + } + + stats, err := parsePodmanStatsOutput([]byte(`[ + { + "ID": "abcdef123456", + "Name": "api", + "CPUPerc": "12.34%", + "MemUsage": "64MiB / 512MiB", + "MemPerc": "12.50%", + "NetIO": "1.5MB / 2.5MB" + } + ]`), containers) + if err != nil { + t.Fatalf("parse stats: %v", err) + } + if len(stats) != 1 { + t.Fatalf("expected 1 stat, got %d", len(stats)) + } + + stat := stats[0] + if stat.ContainerID != "abcdef1234567890" { + t.Fatalf("container id = %q", stat.ContainerID) + } + if stat.ServiceID != "svc_1" || stat.DeploymentID != "dep_1" { + t.Fatalf("unexpected service/deployment labels: %#v", stat) + } + if stat.CPUUsagePercent != 12.34 { + t.Fatalf("cpu = %f", stat.CPUUsagePercent) + } + if stat.MemoryUsagePercent != 12.5 { + t.Fatalf("memory percent = %f", stat.MemoryUsagePercent) + } + if stat.MemoryUsedBytes != 64*1024*1024 { + t.Fatalf("memory bytes = %f", stat.MemoryUsedBytes) + } + if stat.NetworkReceiveBytes != 1.5*1000*1000 { + t.Fatalf("rx bytes = %f", stat.NetworkReceiveBytes) + } + if stat.NetworkTransmitBytes != 2.5*1000*1000 { + t.Fatalf("tx bytes = %f", stat.NetworkTransmitBytes) + } +} + +func TestParsePodmanStatsOutputJSONLines(t *testing.T) { + containers := []Container{ + {ID: "1234567890abcdef", Name: "worker", State: "running", ServiceID: "svc_2", DeploymentID: "dep_2"}, + } + + stats, err := parsePodmanStatsOutput([]byte(`{"ContainerID":"1234567890","CPUPerc":"0%","MemUsage":"128MB / 1GB","MemPerc":"10%","NetIO":"0B / 32kB"}`), containers) + if err != nil { + t.Fatalf("parse stats: %v", err) + } + if len(stats) != 1 { + t.Fatalf("expected 1 stat, got %d", len(stats)) + } + if stats[0].MemoryUsedBytes != 128*1000*1000 { + t.Fatalf("memory bytes = %f", stats[0].MemoryUsedBytes) + } + if stats[0].NetworkTransmitBytes != 32*1000 { + t.Fatalf("tx bytes = %f", stats[0].NetworkTransmitBytes) + } +} + +func TestParsePodmanStatsOutputNameFallbackDoesNotUseIDPrefix(t *testing.T) { + containers := []Container{ + {ID: "api1234567890", Name: "backend", State: "running", ServiceID: "wrong", DeploymentID: "wrong_dep"}, + {ID: "fedcba987654", Name: "api", State: "running", ServiceID: "svc_3", DeploymentID: "dep_3"}, + } + + stats, err := parsePodmanStatsOutput([]byte(`[ + { + "Name": "api", + "CPUPerc": "7%", + "MemUsage": "1MiB / 128MiB", + "MemPerc": "1%", + "NetIO": "0B / 0B" + } + ]`), containers) + if err != nil { + t.Fatalf("parse stats: %v", err) + } + if len(stats) != 1 { + t.Fatalf("expected 1 stat, got %d", len(stats)) + } + if stats[0].ServiceID != "svc_3" || stats[0].DeploymentID != "dep_3" { + t.Fatalf("unexpected attribution: %#v", stats[0]) + } +} + +func TestParseByteQuantity(t *testing.T) { + tests := map[string]float64{ + "42B": 42, + "1 kB": 1000, + "1KiB": 1024, + "1.5GB": 1.5 * 1000 * 1000 * 1000, + "2 MiB": 2 * 1024 * 1024, + "--": 0, + "broken": 0, + } + + for input, expected := range tests { + if actual := parseByteQuantity(input); actual != expected { + t.Fatalf("%q = %f, want %f", input, actual, expected) + } + } +} diff --git a/agent/internal/metrics/victoria.go b/agent/internal/metrics/victoria.go index 8d90e677..d74fad37 100644 --- a/agent/internal/metrics/victoria.go +++ b/agent/internal/metrics/victoria.go @@ -5,9 +5,11 @@ import ( "fmt" "net/http" "net/url" + "sort" "strings" "time" + "techulus/cloud-agent/internal/container" "techulus/cloud-agent/internal/health" ) @@ -78,8 +80,97 @@ func (v *VictoriaMetricsSender) SendSystemStats(stats *health.SystemStats, colle return nil } +func (v *VictoriaMetricsSender) SendContainerStats(stats []container.ResourceStats, collectedAt time.Time) error { + if len(stats) == 0 { + return nil + } + + timestampMs := collectedAt.UnixMilli() + serverID := escapeLabelValue(v.serverID) + + var buf bytes.Buffer + for _, stat := range stats { + labels := map[string]string{ + "server_id": serverID, + "service_id": escapeLabelValue(stat.ServiceID), + "deployment_id": escapeLabelValue(stat.DeploymentID), + "container_id": escapeLabelValue(stat.ContainerID), + } + writeGaugeWithLabels(&buf, "techulus_service_cpu_usage_percent", labels, stat.CPUUsagePercent, timestampMs) + writeGaugeWithLabels(&buf, "techulus_service_memory_usage_percent", labels, stat.MemoryUsagePercent, timestampMs) + writeGaugeWithLabels(&buf, "techulus_service_memory_used_bytes", labels, stat.MemoryUsedBytes, timestampMs) + writeGaugeWithLabels(&buf, "techulus_service_network_receive_bytes_total", labels, stat.NetworkReceiveBytes, timestampMs) + writeGaugeWithLabels(&buf, "techulus_service_network_transmit_bytes_total", labels, stat.NetworkTransmitBytes, timestampMs) + } + + return v.postPrometheusImport(buf.Bytes(), nil) +} + +func (v *VictoriaMetricsSender) SendPrometheusMetrics(data []byte, extraLabels map[string]string) error { + if len(bytes.TrimSpace(data)) == 0 { + return nil + } + return v.postPrometheusImport(data, extraLabels) +} + +func (v *VictoriaMetricsSender) postPrometheusImport(data []byte, extraLabels map[string]string) error { + requestURL, err := url.Parse(v.endpoint + "/api/v1/import/prometheus") + if err != nil { + return fmt.Errorf("failed to parse metrics endpoint: %w", err) + } + query := requestURL.Query() + for _, key := range sortedKeys(extraLabels) { + if key == "" || extraLabels[key] == "" { + continue + } + query.Add("extra_label", fmt.Sprintf("%s=%s", key, extraLabels[key])) + } + requestURL.RawQuery = query.Encode() + + req, err := http.NewRequest("POST", requestURL.String(), bytes.NewReader(data)) + if err != nil { + return fmt.Errorf("failed to create metrics request: %w", err) + } + req.Header.Set("Content-Type", "text/plain; version=0.0.4") + if v.username != "" { + req.SetBasicAuth(v.username, v.password) + } + + resp, err := v.client.Do(req) + if err != nil { + return fmt.Errorf("failed to send metrics: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + return fmt.Errorf("unexpected metrics status code: %d", resp.StatusCode) + } + + return nil +} + func writeGauge(buf *bytes.Buffer, name, serverID string, value float64, timestampMs int64) { - fmt.Fprintf(buf, "%s{server_id=\"%s\"} %f %d\n", name, serverID, value, timestampMs) + writeGaugeWithLabels(buf, name, map[string]string{"server_id": serverID}, value, timestampMs) +} + +func writeGaugeWithLabels(buf *bytes.Buffer, name string, labels map[string]string, value float64, timestampMs int64) { + fmt.Fprintf(buf, "%s{", name) + for i, key := range sortedKeys(labels) { + if i > 0 { + buf.WriteByte(',') + } + fmt.Fprintf(buf, "%s=\"%s\"", key, labels[key]) + } + fmt.Fprintf(buf, "} %f %d\n", value, timestampMs) +} + +func sortedKeys(values map[string]string) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys } func escapeLabelValue(value string) string { diff --git a/agent/internal/metrics/victoria_test.go b/agent/internal/metrics/victoria_test.go new file mode 100644 index 00000000..5854f197 --- /dev/null +++ b/agent/internal/metrics/victoria_test.go @@ -0,0 +1,41 @@ +package metrics + +import ( + "io" + "net/http" + "net/http/httptest" + "testing" +) + +func TestSendPrometheusMetricsAddsExtraLabels(t *testing.T) { + var gotPath string + var gotQuery string + var gotBody string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + gotQuery = r.URL.RawQuery + body, _ := io.ReadAll(r.Body) + gotBody = string(body) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + sender := NewVictoriaMetricsSender(server.URL, "server-1") + err := sender.SendPrometheusMetrics([]byte("sample_metric 1\n"), map[string]string{ + "server_id": "server-1", + "job": "traefik", + }) + if err != nil { + t.Fatalf("send prometheus metrics: %v", err) + } + + if gotPath != "/api/v1/import/prometheus" { + t.Fatalf("path = %q", gotPath) + } + if gotQuery != "extra_label=job%3Dtraefik&extra_label=server_id%3Dserver-1" { + t.Fatalf("query = %q", gotQuery) + } + if gotBody != "sample_metric 1\n" { + t.Fatalf("body = %q", gotBody) + } +} diff --git a/agent/internal/traefik/static.go b/agent/internal/traefik/static.go index 3c7702ff..9fa007dd 100644 --- a/agent/internal/traefik/static.go +++ b/agent/internal/traefik/static.go @@ -5,11 +5,35 @@ import ( "log" "os" "os/exec" + "reflect" "time" "gopkg.in/yaml.v3" ) +const ( + metricsEntryPointName = "metrics" + metricsEntryPointAddr = "127.0.0.1:9100" +) + +var prometheusLatencyBuckets = []interface{}{ + 0.005, + 0.01, + 0.025, + 0.05, + 0.075, + 0.1, + 0.25, + 0.5, + 0.75, + 1.0, + 2.5, + 5.0, + 10.0, + 30.0, + 60.0, +} + func validateStaticConfig(data []byte) error { var config map[string]interface{} if err := yaml.Unmarshal(data, &config); err != nil { @@ -112,6 +136,95 @@ func EnsureEntryPoints(tcpPorts []int, udpPorts []int) (needsRestart bool, err e return true, nil } +func EnsureMetricsConfig() (needsRestart bool, err error) { + originalData, err := os.ReadFile(traefikStaticConfigPath) + if err != nil { + return false, fmt.Errorf("failed to read static config: %w", err) + } + + var config map[string]interface{} + if err := yaml.Unmarshal(originalData, &config); err != nil { + return false, fmt.Errorf("failed to parse static config: %w", err) + } + + if !ensurePrometheusMetricsConfig(config) { + return false, nil + } + + newData, err := yaml.Marshal(config) + if err != nil { + return false, fmt.Errorf("failed to marshal static config: %w", err) + } + + if err := validateStaticConfig(newData); err != nil { + return false, fmt.Errorf("config validation failed: %w", err) + } + + if err := atomicWrite(traefikStaticConfigPath, newData, 0644); err != nil { + return false, fmt.Errorf("failed to write static config: %w", err) + } + + log.Printf("[traefik] metrics config updated, restart required") + return true, nil +} + +func ensurePrometheusMetricsConfig(config map[string]interface{}) bool { + modified := false + + entryPoints, ok := config["entryPoints"].(map[string]interface{}) + if !ok { + entryPoints = make(map[string]interface{}) + config["entryPoints"] = entryPoints + modified = true + } + + metricsEntryPoint, ok := entryPoints[metricsEntryPointName].(map[string]interface{}) + if !ok { + metricsEntryPoint = make(map[string]interface{}) + entryPoints[metricsEntryPointName] = metricsEntryPoint + modified = true + } + if setMapValue(metricsEntryPoint, "address", metricsEntryPointAddr) { + modified = true + } + + metricsConfig, ok := config["metrics"].(map[string]interface{}) + if !ok { + metricsConfig = make(map[string]interface{}) + config["metrics"] = metricsConfig + modified = true + } + + prometheusConfig, ok := metricsConfig["prometheus"].(map[string]interface{}) + if !ok { + prometheusConfig = make(map[string]interface{}) + metricsConfig["prometheus"] = prometheusConfig + modified = true + } + + for key, value := range map[string]interface{}{ + "entryPoint": metricsEntryPointName, + "addEntryPointsLabels": false, + "addRoutersLabels": false, + "addServicesLabels": true, + "buckets": prometheusLatencyBuckets, + } { + if setMapValue(prometheusConfig, key, value) { + modified = true + } + } + + return modified +} + +func setMapValue(values map[string]interface{}, key string, value interface{}) bool { + if reflect.DeepEqual(values[key], value) { + return false + } + values[key] = value + return true +} + func ReloadTraefik() error { cmd := exec.Command("systemctl", "restart", "traefik") if err := cmd.Run(); err != nil { diff --git a/agent/internal/traefik/static_test.go b/agent/internal/traefik/static_test.go new file mode 100644 index 00000000..7f2b47af --- /dev/null +++ b/agent/internal/traefik/static_test.go @@ -0,0 +1,50 @@ +package traefik + +import "testing" + +func TestEnsurePrometheusMetricsConfigAddsPrivateMetricsEndpoint(t *testing.T) { + config := map[string]interface{}{ + "entryPoints": map[string]interface{}{ + "web": map[string]interface{}{"address": ":80"}, + }, + } + + if !ensurePrometheusMetricsConfig(config) { + t.Fatal("expected config to be modified") + } + + entryPoints := config["entryPoints"].(map[string]interface{}) + metricsEntryPoint := entryPoints["metrics"].(map[string]interface{}) + if metricsEntryPoint["address"] != "127.0.0.1:9100" { + t.Fatalf("metrics address = %#v", metricsEntryPoint["address"]) + } + + metricsConfig := config["metrics"].(map[string]interface{}) + prometheusConfig := metricsConfig["prometheus"].(map[string]interface{}) + if prometheusConfig["entryPoint"] != "metrics" { + t.Fatalf("entryPoint = %#v", prometheusConfig["entryPoint"]) + } + if prometheusConfig["addServicesLabels"] != true { + t.Fatalf("addServicesLabels = %#v", prometheusConfig["addServicesLabels"]) + } + if prometheusConfig["addRoutersLabels"] != false { + t.Fatalf("addRoutersLabels = %#v", prometheusConfig["addRoutersLabels"]) + } + if prometheusConfig["addEntryPointsLabels"] != false { + t.Fatalf("addEntryPointsLabels = %#v", prometheusConfig["addEntryPointsLabels"]) + } + if len(prometheusConfig["buckets"].([]interface{})) == 0 { + t.Fatal("expected latency buckets") + } +} + +func TestEnsurePrometheusMetricsConfigIsStable(t *testing.T) { + config := map[string]interface{}{} + + if !ensurePrometheusMetricsConfig(config) { + t.Fatal("expected first call to modify config") + } + if ensurePrometheusMetricsConfig(config) { + t.Fatal("expected second call to be stable") + } +} diff --git a/web/app/api/services/[id]/request-stats/route.ts b/web/app/api/services/[id]/metrics/route.ts similarity index 61% rename from web/app/api/services/[id]/request-stats/route.ts rename to web/app/api/services/[id]/metrics/route.ts index 3b25c5be..1d8cc791 100644 --- a/web/app/api/services/[id]/request-stats/route.ts +++ b/web/app/api/services/[id]/metrics/route.ts @@ -2,11 +2,12 @@ import { headers } from "next/headers"; import { getService } from "@/db/queries"; import { auth } from "@/lib/auth"; import { - createEmptyHttpRequestStats, - isLoggingEnabled, - parseRequestStatsRange, - queryHttpRequestStats, -} from "@/lib/victoria-logs"; + createEmptyServiceMetrics, + isMetricsEnabled, + parseMetricRange, + queryServiceMetrics, + warnMissingMetricsConfig, +} from "@/lib/victoria-metrics"; const SERVICE_ID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; @@ -25,7 +26,7 @@ export async function GET( const { id: serviceId } = await params; const url = new URL(request.url); - const range = parseRequestStatsRange(url.searchParams.get("range")); + const range = parseMetricRange(url.searchParams.get("range")); if (!SERVICE_ID_PATTERN.test(serviceId)) { return Response.json({ message: "Invalid service id" }, { status: 400 }); @@ -36,20 +37,17 @@ export async function GET( return Response.json({ message: "Service not found" }, { status: 404 }); } - if (!isLoggingEnabled()) { - return Response.json({ - loggingEnabled: false, - ...createEmptyHttpRequestStats(range), - }); + if (!isMetricsEnabled()) { + warnMissingMetricsConfig("service"); + return Response.json(createEmptyServiceMetrics(range)); } try { - const stats = await queryHttpRequestStats({ serviceId, range }); - return Response.json({ loggingEnabled: true, ...stats }); + return Response.json(await queryServiceMetrics({ serviceId, range })); } catch (error) { - console.error("[logs:request-stats] failed to query HTTP stats:", error); + console.error("[metrics:service] failed to query service metrics:", error); return Response.json( - { message: "Request stats unavailable" }, + { message: "Service metrics unavailable" }, { status: 502 }, ); } diff --git a/web/components/service/details/service-details-overview.tsx b/web/components/service/details/service-details-overview.tsx index e385f341..1d1f30ce 100644 --- a/web/components/service/details/service-details-overview.tsx +++ b/web/components/service/details/service-details-overview.tsx @@ -28,8 +28,8 @@ import { isObservedStarting } from "@/lib/deployment-status"; import { fetcher } from "@/lib/fetcher"; import { cn } from "@/lib/utils"; -type RequestStatsResponse = { - loggingEnabled: boolean; +type ServiceMetricsResponse = { + metricsEnabled: boolean; range: string; windowStart: string; windowEnd: string; @@ -40,6 +40,15 @@ type RequestStatsResponse = { timestamp: string; totalRequests: number; statuses: Record; + p50ResponseTimeMs: number | null; + p90ResponseTimeMs: number | null; + p95ResponseTimeMs: number | null; + p99ResponseTimeMs: number | null; + ingressBytesPerSecond: number | null; + egressBytesPerSecond: number | null; + cpuUsagePercent: number | null; + memoryUsagePercent: number | null; + memoryUsedBytes: number | null; }>; }; @@ -85,20 +94,18 @@ type ServiceStatus = { type ChartRow = { timestamp: string; totalRequests: number; -} & Record; +} & Record; -type RequestChartMode = "rate" | "total"; +type ServiceChartMode = "requests" | "latency" | "traffic" | "resources"; type StatusSeries = { status: string; - rateDataKey: string; totalDataKey: string; color: string; - averageRequestsPerSecond: number; totalRequests: number; }; -type RequestTooltipPayload = { +type ServiceMetricsTooltipPayload = { name?: string; value?: unknown; color?: string; @@ -106,11 +113,18 @@ type RequestTooltipPayload = { payload?: ChartRow; }; -type RequestTooltipProps = { +type ServiceMetricsTooltipProps = { active?: boolean; label?: string | number; - mode: RequestChartMode; - payload?: readonly RequestTooltipPayload[]; + mode: ServiceChartMode; + payload?: readonly ServiceMetricsTooltipPayload[]; +}; + +type MetricSeries = { + key: string; + label: string; + color: string; + valueFormatter: (value: number) => string; }; const STATUS_TONE_CLASSES: Record< @@ -146,26 +160,27 @@ export function ServiceDetailsOverview({ service }: { service: Service }) { () => buildOverviewData(service, proxyDomain), [service, proxyDomain], ); - const requestStatsUrl = - overview.publicHttpCount > 0 - ? `/api/services/${service.id}/request-stats?range=week` + const serviceMetricsUrl = + overview.runningDeployments > 0 + ? `/api/services/${service.id}/metrics?range=24h` : null; const { - data: requestStats, - error: requestStatsError, - isLoading: isRequestStatsLoading, - } = useSWR(requestStatsUrl, fetcher, { + data: serviceMetrics, + error: serviceMetricsError, + isLoading: isServiceMetricsLoading, + } = useSWR(serviceMetricsUrl, fetcher, { refreshInterval: 60000, }); return (
- 0} - stats={requestStats} - error={requestStatsError} - isLoading={isRequestStatsLoading} + hasRunningDeployments={overview.runningDeployments > 0} + stats={serviceMetrics} + error={serviceMetricsError} + isLoading={isServiceMetricsLoading} /> @@ -174,23 +189,33 @@ export function ServiceDetailsOverview({ service }: { service: Service }) { ); } -function RequestStatsPanel({ +function ServiceMetricsPanel({ hasPublicHttp, + hasRunningDeployments, stats, error, isLoading, }: { hasPublicHttp: boolean; - stats?: RequestStatsResponse; + hasRunningDeployments: boolean; + stats?: ServiceMetricsResponse; error?: unknown; isLoading: boolean; }) { - const [chartMode, setChartMode] = useState("total"); + const [chartMode, setChartMode] = useState("requests"); const chartRows = useMemo(() => buildChartRows(stats), [stats]); const statusSeries = useMemo(() => buildStatusSeries(stats), [stats]); - const hasChartData = chartRows.some((row) => row.totalRequests > 0); - const isUnavailable = Boolean(error) || stats?.loggingEnabled === false; - const hasMetricData = hasPublicHttp && stats && !isUnavailable; + const activeSeries = useMemo( + () => buildMetricSeries(chartMode, statusSeries), + [chartMode, statusSeries], + ); + const hasChartData = hasMetricDataForMode(chartRows, chartMode, activeSeries); + const isUnavailable = Boolean(error) || stats?.metricsEnabled === false; + const hasMetricData = stats && !isUnavailable; + const p95 = getLatestValue(chartRows, "p95ResponseTimeMs"); + const egress = getLatestValue(chartRows, "egressBytesPerSecond"); + const cpu = getLatestValue(chartRows, "cpuUsagePercent"); + const memory = getLatestValue(chartRows, "memoryUsedBytes"); return (
@@ -201,7 +226,7 @@ function RequestStatsPanel({
- ) : hasPublicHttp ? ( + ) : hasRunningDeployments ? (

@@ -209,9 +234,7 @@ function RequestStatsPanel({ ? formatCompactNumber(stats.totalRequests) : "-"}

-

- requests this week -

+

requests in 24h

@@ -219,9 +242,33 @@ function RequestStatsPanel({ ? formatRate(getAverageRequestsPerSecond(stats)) : "-"}

-

- avg RPS this week +

avg RPS

+
+
+

+ {hasMetricData && p95 != null ? formatDurationMs(p95) : "-"} +

+

p95 latency

+
+
+

+ {hasMetricData && cpu != null ? `${formatRate(cpu)}%` : "-"}

+

CPU

+
+
+

+ {hasMetricData && memory != null ? formatBytes(memory) : "-"} +

+

memory

+
+
+

+ {hasMetricData && egress != null + ? `${formatBytes(egress)}/s` + : "-"} +

+

egress

) : ( @@ -229,33 +276,39 @@ function RequestStatsPanel({

-

-

- no public HTTP ingress -

+

not deployed

)}

{formatToday()}

-
- {!hasPublicHttp ? ( - + {!hasRunningDeployments ? ( + ) : isLoading ? ( ) : isUnavailable ? ( - + ) : !hasChartData ? ( - + ) : ( @@ -289,30 +342,26 @@ function RequestStatsPanel({ + formatAxisTick(Number(value), chartMode) } className="text-xs" /> ( - )} /> - {statusSeries.map((series) => ( + {activeSeries.map((series) => ( - {statusSeries.length > 0 && ( + {activeSeries.length > 0 && (
- {statusSeries.map((series) => ( + {activeSeries.map((series) => ( ))}
@@ -349,18 +390,20 @@ function RequestStatsPanel({ ); } -function RequestChartModeToggle({ +function ServiceChartModeToggle({ value, onChange, disabled, }: { - value: RequestChartMode; - onChange: (value: RequestChartMode) => void; + value: ServiceChartMode; + onChange: (value: ServiceChartMode) => void; disabled: boolean; }) { - const options: Array<{ value: RequestChartMode; label: string }> = [ - { value: "total", label: "Total" }, - { value: "rate", label: "RPS" }, + const options: Array<{ value: ServiceChartMode; label: string }> = [ + { value: "requests", label: "Requests" }, + { value: "latency", label: "Latency" }, + { value: "traffic", label: "Traffic" }, + { value: "resources", label: "Resources" }, ]; return ( @@ -616,7 +659,7 @@ function LegendMetric({ ); } -function RequestStatsState({ message }: { message: string }) { +function ServiceMetricsState({ message }: { message: string }) { return (
{message} @@ -624,16 +667,18 @@ function RequestStatsState({ message }: { message: string }) { ); } -function RequestStatsTooltip({ +function ServiceMetricsTooltip({ active, payload, label, mode, -}: RequestTooltipProps) { +}: ServiceMetricsTooltipProps) { if (!active || !payload?.length) return null; const row = payload[0]?.payload; - const visiblePayload = payload.filter((item) => Number(item.value) > 0); + const visiblePayload = payload.filter( + (item) => item.value != null && Number.isFinite(Number(item.value)), + ); const items = visiblePayload.length > 0 ? visiblePayload : payload; return ( @@ -824,28 +869,34 @@ function getSourceInfo(service: Service): SourceInfo { }; } -function buildChartRows(stats?: RequestStatsResponse): ChartRow[] { +function buildChartRows(stats?: ServiceMetricsResponse): ChartRow[] { if (!stats) return []; return stats.buckets.map((bucket) => { const row: ChartRow = { timestamp: bucket.timestamp, totalRequests: bucket.totalRequests, + p50ResponseTimeMs: bucket.p50ResponseTimeMs, + p90ResponseTimeMs: bucket.p90ResponseTimeMs, + p95ResponseTimeMs: bucket.p95ResponseTimeMs, + p99ResponseTimeMs: bucket.p99ResponseTimeMs, + ingressBytesPerSecond: bucket.ingressBytesPerSecond, + egressBytesPerSecond: bucket.egressBytesPerSecond, + cpuUsagePercent: bucket.cpuUsagePercent, + memoryUsagePercent: bucket.memoryUsagePercent, + memoryUsedBytes: bucket.memoryUsedBytes, }; for (const status of stats.statusCodes) { const requests = bucket.statuses[status] ?? 0; - row[getStatusRateDataKey(status)] = requests / stats.stepSeconds; - row[getStatusTotalDataKey(status)] = requests; + row[getStatusDataKey(status)] = requests; } return row; }); } -function buildStatusSeries(stats?: RequestStatsResponse): StatusSeries[] { +function buildStatusSeries(stats?: ServiceMetricsResponse): StatusSeries[] { if (!stats) return []; - const totalSeconds = getStatsDurationSeconds(stats); - return stats.statusCodes.map((status, index) => { const totalRequests = stats.buckets.reduce( (total, bucket) => total + (bucket.statuses[status] ?? 0), @@ -854,21 +905,14 @@ function buildStatusSeries(stats?: RequestStatsResponse): StatusSeries[] { return { status, - rateDataKey: getStatusRateDataKey(status), - totalDataKey: getStatusTotalDataKey(status), + totalDataKey: getStatusDataKey(status), color: getStatusColor(status, index), - averageRequestsPerSecond: - totalSeconds > 0 ? totalRequests / totalSeconds : 0, totalRequests, }; }); } -function getStatusRateDataKey(status: string): string { - return `${getStatusDataKeyBase(status)}_rate`; -} - -function getStatusTotalDataKey(status: string): string { +function getStatusDataKey(status: string): string { return `${getStatusDataKeyBase(status)}_total`; } @@ -977,7 +1021,116 @@ function formatRate(value: number): string { return value.toFixed(2).replace(/\.?0+$/, ""); } -function getAverageRequestsPerSecond(stats: RequestStatsResponse): number { +function buildMetricSeries( + mode: ServiceChartMode, + statusSeries: StatusSeries[], +): MetricSeries[] { + if (mode === "requests") { + return statusSeries.map((series) => ({ + key: series.totalDataKey, + label: series.status, + color: series.color, + valueFormatter: formatRequestCount, + })); + } + + if (mode === "latency") { + return [ + { + key: "p99ResponseTimeMs", + label: "p99", + color: "#ef4444", + valueFormatter: formatDurationMs, + }, + { + key: "p95ResponseTimeMs", + label: "p95", + color: "#ec4899", + valueFormatter: formatDurationMs, + }, + { + key: "p90ResponseTimeMs", + label: "p90", + color: "#f59e0b", + valueFormatter: formatDurationMs, + }, + { + key: "p50ResponseTimeMs", + label: "p50", + color: "#3b82f6", + valueFormatter: formatDurationMs, + }, + ]; + } + + if (mode === "traffic") { + return [ + { + key: "ingressBytesPerSecond", + label: "Ingress", + color: "#0ea5e9", + valueFormatter: (value) => `${formatBytes(value)}/s`, + }, + { + key: "egressBytesPerSecond", + label: "Egress", + color: "#10b981", + valueFormatter: (value) => `${formatBytes(value)}/s`, + }, + ]; + } + + return [ + { + key: "cpuUsagePercent", + label: "CPU", + color: "#8b5cf6", + valueFormatter: (value) => `${formatRate(value)}%`, + }, + { + key: "memoryUsagePercent", + label: "Memory", + color: "#14b8a6", + valueFormatter: (value) => `${formatRate(value)}%`, + }, + ]; +} + +function hasMetricDataForMode( + rows: ChartRow[], + mode: ServiceChartMode, + series: MetricSeries[], +): boolean { + if (mode === "requests") { + return rows.some((row) => row.totalRequests > 0); + } + return rows.some((row) => + series.some((item) => { + const value = row[item.key]; + return typeof value === "number" && Number.isFinite(value); + }), + ); +} + +function getLatestValue(rows: ChartRow[], key: string): number | null { + for (let index = rows.length - 1; index >= 0; index--) { + const value = rows[index][key]; + if (typeof value === "number" && Number.isFinite(value)) { + return value; + } + } + return null; +} + +function formatLatestSeriesValue( + rows: ChartRow[], + series: MetricSeries, +): string { + const value = getLatestValue(rows, series.key); + return value == null ? "-" : series.valueFormatter(value); +} + +function getAverageRequestsPerSecond(stats: ServiceMetricsResponse): number { const totalSeconds = getStatsDurationSeconds(stats); if (totalSeconds <= 0) return 0; @@ -985,7 +1138,7 @@ function getAverageRequestsPerSecond(stats: RequestStatsResponse): number { return stats.totalRequests / totalSeconds; } -function getStatsDurationSeconds(stats: RequestStatsResponse): number { +function getStatsDurationSeconds(stats: ServiceMetricsResponse): number { const windowStart = new Date(stats.windowStart).getTime(); const windowEnd = new Date(stats.windowEnd).getTime(); @@ -1000,18 +1153,34 @@ function getStatsDurationSeconds(stats: RequestStatsResponse): number { return stats.buckets.length * stats.stepSeconds; } -function formatChartValue(value: number, mode: RequestChartMode): string { - if (mode === "rate") return `${formatRate(value)}/s`; - +function formatChartValue(value: number, mode: ServiceChartMode): string { + if (!Number.isFinite(value)) return "-"; + if (mode === "latency") return formatDurationMs(value); + if (mode === "traffic") return `${formatBytes(value)}/s`; + if (mode === "resources") return `${formatRate(value)}%`; return formatRequestCount(value); } +function formatAxisTick(value: number, mode: ServiceChartMode): string { + if (mode === "latency") return formatDurationMs(value); + if (mode === "traffic") return formatBytes(value); + if (mode === "resources") return `${formatRateTick(value)}%`; + return formatRequestTick(value); +} + function formatRateTick(value: number): string { if (value >= 100) return value.toFixed(0); if (value >= 10) return value.toFixed(0); return value.toFixed(1); } +function getYAxisMargin(mode: ServiceChartMode): number { + if (mode === "traffic") return -4; + if (mode === "latency") return -12; + if (mode === "resources") return -24; + return -20; +} + function formatRequestTick(value: number): string { return formatCompactNumber(value); } @@ -1024,6 +1193,29 @@ function formatRequestCount(value: number): string { }).format(value); } +function formatDurationMs(value: number): string { + if (!Number.isFinite(value)) return "-"; + if (value >= 1000) { + return `${(value / 1000).toFixed(value >= 10000 ? 0 : 1)}s`; + } + if (value >= 100) return `${value.toFixed(0)}ms`; + if (value >= 10) return `${value.toFixed(1)}ms`; + return `${value.toFixed(2).replace(/\.?0+$/, "")}ms`; +} + +function formatBytes(value: number): string { + if (!Number.isFinite(value)) return "-"; + const units = ["B", "KB", "MB", "GB", "TB"]; + let unitIndex = 0; + let scaled = Math.max(0, value); + while (scaled >= 1000 && unitIndex < units.length - 1) { + scaled /= 1000; + unitIndex++; + } + const maximumFractionDigits = scaled >= 100 || unitIndex === 0 ? 0 : 1; + return `${scaled.toFixed(maximumFractionDigits)} ${units[unitIndex]}`; +} + function formatShortDate(value: string): string { return new Intl.DateTimeFormat(undefined, { month: "short", diff --git a/web/lib/victoria-logs.ts b/web/lib/victoria-logs.ts index 32484641..9540568d 100644 --- a/web/lib/victoria-logs.ts +++ b/web/lib/victoria-logs.ts @@ -1,35 +1,6 @@ const VICTORIA_LOGS_URL = process.env.VICTORIA_LOGS_URL; const VICTORIA_LOGS_PRIVATE_URL = process.env.VICTORIA_LOGS_PRIVATE_URL; -export const REQUEST_STATS_RANGE_OPTIONS = { - "1h": { durationMs: 60 * 60 * 1000, stepSeconds: 60, stepLogSql: "1m" }, - "6h": { - durationMs: 6 * 60 * 60 * 1000, - stepSeconds: 5 * 60, - stepLogSql: "5m", - }, - "24h": { - durationMs: 24 * 60 * 60 * 1000, - stepSeconds: 5 * 60, - stepLogSql: "5m", - }, - "7d": { - durationMs: 7 * 24 * 60 * 60 * 1000, - stepSeconds: 30 * 60, - stepLogSql: "30m", - }, - week: { - durationMs: 0, - stepSeconds: 30 * 60, - stepLogSql: "30m", - weekToDate: true, - }, -} as const; - -export type RequestStatsRange = keyof typeof REQUEST_STATS_RANGE_OPTIONS; - -export const DEFAULT_REQUEST_STATS_RANGE: RequestStatsRange = "7d"; - type EndpointConfig = { url: string; username?: string; @@ -84,260 +55,6 @@ export function isLoggingEnabled(): boolean { return !!(VICTORIA_LOGS_PRIVATE_URL || VICTORIA_LOGS_URL); } -export type HttpRequestStatsBucket = { - timestamp: string; - totalRequests: number; - statuses: Record; -}; - -export type HttpRequestStats = { - range: RequestStatsRange; - windowStart: string; - windowEnd: string; - stepSeconds: number; - totalRequests: number; - statusCodes: string[]; - buckets: HttpRequestStatsBucket[]; -}; - -type RequestStatsRow = Record; - -export function parseRequestStatsRange( - value: string | null | undefined, -): RequestStatsRange { - if (value && value in REQUEST_STATS_RANGE_OPTIONS) { - return value as RequestStatsRange; - } - return DEFAULT_REQUEST_STATS_RANGE; -} - -export function createEmptyHttpRequestStats( - range: RequestStatsRange = DEFAULT_REQUEST_STATS_RANGE, - now = new Date(), -): HttpRequestStats { - const window = getRequestStatsWindow(range, now); - return { - range, - windowStart: window.start.toISOString(), - windowEnd: window.end.toISOString(), - stepSeconds: window.stepSeconds, - totalRequests: 0, - statusCodes: [], - buckets: [], - }; -} - -export function getRequestStatsWindow( - range: RequestStatsRange, - now = new Date(), -): { - start: Date; - end: Date; - durationMs: number; - stepSeconds: number; - stepLogSql: string; -} { - const config = REQUEST_STATS_RANGE_OPTIONS[range]; - const start = - "weekToDate" in config && config.weekToDate - ? getStartOfUtcWeek(now) - : new Date(now.getTime() - config.durationMs); - - return { - start, - end: now, - durationMs: Math.max(0, now.getTime() - start.getTime()), - stepSeconds: config.stepSeconds, - stepLogSql: config.stepLogSql, - }; -} - -export function parseRequestStatsRows(text: string): RequestStatsRow[] { - const lines = text.trim().split("\n").filter(Boolean); - const rows: RequestStatsRow[] = []; - - for (const line of lines) { - try { - const parsed = JSON.parse(line) as unknown; - if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { - rows.push(parsed as RequestStatsRow); - } - } catch { - // VictoriaLogs should return JSON lines, but a single malformed row should - // not break the service overview card. - } - } - - return rows; -} - -export function buildRequestStatsBuckets( - rows: RequestStatsRow[], - options: { start: Date; end: Date; stepSeconds: number }, -): HttpRequestStatsBucket[] { - const stepMs = options.stepSeconds * 1000; - const startMs = floorToStep(options.start.getTime(), stepMs); - const endMs = floorToStep(options.end.getTime(), stepMs); - - if (!Number.isFinite(startMs) || !Number.isFinite(endMs) || startMs > endMs) { - return []; - } - - const byBucket = new Map< - number, - { totalRequests: number; statuses: Record } - >(); - for (const row of rows) { - const timestamp = parseTimestamp(row._time ?? row.time ?? row.timestamp); - if (!timestamp) continue; - - const bucketMs = floorToStep(timestamp.getTime(), stepMs); - const bucket = byBucket.get(bucketMs) ?? { - totalRequests: 0, - statuses: {}, - }; - const requests = parseStatNumber(row.requests); - const status = normalizeStatus(row.status); - bucket.totalRequests += requests; - bucket.statuses[status] = (bucket.statuses[status] ?? 0) + requests; - byBucket.set(bucketMs, bucket); - } - - const buckets: HttpRequestStatsBucket[] = []; - for (let timestampMs = startMs; timestampMs <= endMs; timestampMs += stepMs) { - const bucket = byBucket.get(timestampMs) ?? { - totalRequests: 0, - statuses: {}, - }; - buckets.push({ - timestamp: new Date(timestampMs).toISOString(), - totalRequests: bucket.totalRequests, - statuses: bucket.statuses, - }); - } - - return buckets; -} - -export async function queryHttpRequestStats(options: { - serviceId: string; - range: RequestStatsRange; - now?: Date; -}): Promise { - const now = options.now ?? new Date(); - const window = getRequestStatsWindow(options.range, now); - const statusCodeLimit = 64; - const bucketLimit = - (Math.ceil(window.durationMs / (window.stepSeconds * 1000)) + 5) * - statusCodeLimit; - const baseFilter = [ - formatLogSqlExactFilter("service_id", options.serviceId), - "log_type:http", - formatLogSqlTimeRange(window.start, window.end), - ].join(" "); - const bucketQuery = `${baseFilter} | stats by (_time:${window.stepLogSql}, status) count() as requests | sort by (_time)`; - - const bucketText = await queryLogSql(bucketQuery, bucketLimit); - - const buckets = buildRequestStatsBuckets(parseRequestStatsRows(bucketText), { - start: window.start, - end: window.end, - stepSeconds: window.stepSeconds, - }); - const statusCodes = sortStatusCodes([ - ...buckets.flatMap((bucket) => Object.keys(bucket.statuses)), - ]); - - return { - range: options.range, - windowStart: window.start.toISOString(), - windowEnd: window.end.toISOString(), - stepSeconds: window.stepSeconds, - totalRequests: buckets.reduce( - (sum, bucket) => sum + bucket.totalRequests, - 0, - ), - statusCodes, - buckets, - }; -} - -async function queryLogSql(query: string, limit: number): Promise { - const endpoint = getQueryEndpoint(); - if (!endpoint) { - throw new Error("VICTORIA_LOGS_URL is not configured"); - } - - const url = new URL(`${endpoint.url}/select/logsql/query`); - url.searchParams.set("query", query); - url.searchParams.set("limit", String(limit)); - - const response = await fetch(url.toString(), buildFetchOptions(endpoint)); - - if (!response.ok) { - throw new Error( - `Failed to query logs: ${response.status} ${response.statusText}`, - ); - } - - return response.text(); -} - -function parseStatNumber(value: unknown): number { - if (typeof value === "number" && Number.isFinite(value)) { - return value; - } - if (typeof value === "string" && value.trim() !== "") { - const parsed = Number(value); - return Number.isFinite(parsed) ? parsed : 0; - } - return 0; -} - -function normalizeStatus(value: unknown): string { - if (typeof value === "number" && Number.isFinite(value)) { - return String(value); - } - if (typeof value === "string" && value.trim() !== "") { - return value.trim(); - } - return "unknown"; -} - -function sortStatusCodes(statuses: string[]): string[] { - return Array.from(new Set(statuses)).sort((a, b) => { - const statusA = Number(a); - const statusB = Number(b); - if (Number.isFinite(statusA) && Number.isFinite(statusB)) { - return statusA - statusB; - } - return a.localeCompare(b); - }); -} - -function parseTimestamp(value: unknown): Date | null { - if (typeof value !== "string" && typeof value !== "number") { - return null; - } - - const timestamp = new Date(value); - return Number.isFinite(timestamp.getTime()) ? timestamp : null; -} - -function floorToStep(timestampMs: number, stepMs: number): number { - return Math.floor(timestampMs / stepMs) * stepMs; -} - -function getStartOfUtcWeek(value: Date): Date { - const start = new Date( - Date.UTC(value.getUTCFullYear(), value.getUTCMonth(), value.getUTCDate()), - ); - const day = start.getUTCDay(); - const daysSinceMonday = day === 0 ? 6 : day - 1; - start.setUTCDate(start.getUTCDate() - daysSinceMonday); - return start; -} - function formatLogSqlExactFilter(field: string, value: string): string { const trimmed = value.trim(); if (!/^[a-zA-Z0-9_-]+$/.test(trimmed)) { @@ -347,10 +64,6 @@ function formatLogSqlExactFilter(field: string, value: string): string { return `${field}:${trimmed}`; } -function formatLogSqlTimeRange(start: Date, end: Date): string { - return `_time:[${start.toISOString()}, ${end.toISOString()})`; -} - type QueryLogsByServiceOptions = { serviceId: string; limit: number; diff --git a/web/lib/victoria-metrics.ts b/web/lib/victoria-metrics.ts index 23984e7f..0fb62f7d 100644 --- a/web/lib/victoria-metrics.ts +++ b/web/lib/victoria-metrics.ts @@ -6,8 +6,6 @@ import { export { METRIC_RANGE_OPTIONS, type MetricRange, parseMetricRange }; -const VICTORIA_METRICS_URL = process.env.VICTORIA_METRICS_URL; -const VICTORIA_METRICS_PRIVATE_URL = process.env.VICTORIA_METRICS_PRIVATE_URL; let hasWarnedMissingMetricsConfig = false; type EndpointConfig = { @@ -27,13 +25,15 @@ type VictoriaInstantResponse = { error?: string; }; +type VictoriaMatrixResult = { + metric: Record; + values: Array<[number, string]>; +}; + type VictoriaMatrixResponse = { status: string; data?: { - result?: Array<{ - metric: Record; - values: Array<[number, string]>; - }>; + result?: VictoriaMatrixResult[]; }; error?: string; }; @@ -51,6 +51,32 @@ export type NodeMetricPoint = { value: number; }; +export type ServiceMetricsBucket = { + timestamp: string; + totalRequests: number; + statuses: Record; + p50ResponseTimeMs: number | null; + p90ResponseTimeMs: number | null; + p95ResponseTimeMs: number | null; + p99ResponseTimeMs: number | null; + ingressBytesPerSecond: number | null; + egressBytesPerSecond: number | null; + cpuUsagePercent: number | null; + memoryUsagePercent: number | null; + memoryUsedBytes: number | null; +}; + +export type ServiceMetrics = { + metricsEnabled: boolean; + range: MetricRange; + windowStart: string; + windowEnd: string; + stepSeconds: number; + totalRequests: number; + statusCodes: string[]; + buckets: ServiceMetricsBucket[]; +}; + export type NodeMetricsHistory = { cpuUsagePercent: NodeMetricPoint[]; memoryUsagePercent: NodeMetricPoint[]; @@ -85,7 +111,9 @@ function parseEndpoint(endpoint: string): EndpointConfig { } function getQueryEndpoint(): EndpointConfig | undefined { - const endpoint = VICTORIA_METRICS_PRIVATE_URL || VICTORIA_METRICS_URL; + const endpoint = + process.env.VICTORIA_METRICS_PRIVATE_URL || + process.env.VICTORIA_METRICS_URL; if (!endpoint) return undefined; return parseEndpoint(endpoint); } @@ -101,7 +129,9 @@ function buildFetchOptions(config: EndpointConfig): RequestInit { } export function isMetricsEnabled(): boolean { - return !!(VICTORIA_METRICS_PRIVATE_URL || VICTORIA_METRICS_URL); + return !!( + process.env.VICTORIA_METRICS_PRIVATE_URL || process.env.VICTORIA_METRICS_URL + ); } export function warnMissingMetricsConfig(context: string) { @@ -262,6 +292,180 @@ export async function queryServersMetricsHistory(options: { })); } +export function createEmptyServiceMetrics( + range: MetricRange, + now = new Date(), +): ServiceMetrics { + const window = getMetricWindow(range, now); + return { + metricsEnabled: false, + range, + windowStart: window.start.toISOString(), + windowEnd: window.end.toISOString(), + stepSeconds: window.stepSeconds, + totalRequests: 0, + statusCodes: [], + buckets: [], + }; +} + +export async function queryServiceMetrics(options: { + serviceId: string; + range: MetricRange; + now?: Date; +}): Promise { + const endpoint = getQueryEndpoint(); + const now = options.now ?? new Date(); + const window = getMetricWindow(options.range, now); + if (!endpoint) return createEmptyServiceMetrics(options.range, now); + + const serviceMatcher = buildTraefikServiceMatcher(options.serviceId); + const serviceId = escapePromQL(options.serviceId); + const rangeWindow = formatPromDuration(window.stepSeconds); + const traefikFilter = `service=~"${serviceMatcher}"`; + const queryStart = new Date( + window.start.getTime() + window.stepSeconds * 1000, + ); + + const [ + requestResults, + p50Results, + p90Results, + p95Results, + p99Results, + ingressResults, + egressResults, + cpuResults, + memoryPercentResults, + memoryBytesResults, + ] = await Promise.all([ + queryRangePromQL(endpoint, { + query: `sum by (code) (increase(traefik_service_requests_total{${traefikFilter}}[${rangeWindow}]))`, + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: responseTimeQuery(0.5, traefikFilter, rangeWindow), + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: responseTimeQuery(0.9, traefikFilter, rangeWindow), + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: responseTimeQuery(0.95, traefikFilter, rangeWindow), + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: responseTimeQuery(0.99, traefikFilter, rangeWindow), + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: `sum(rate(traefik_service_requests_bytes_total{${traefikFilter}}[${rangeWindow}]))`, + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: `sum(rate(traefik_service_responses_bytes_total{${traefikFilter}}[${rangeWindow}]))`, + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: `sum(avg_over_time(techulus_service_cpu_usage_percent{service_id="${serviceId}"}[${rangeWindow}]))`, + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: `avg(avg_over_time(techulus_service_memory_usage_percent{service_id="${serviceId}"}[${rangeWindow}]))`, + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + queryRangePromQL(endpoint, { + query: `sum(avg_over_time(techulus_service_memory_used_bytes{service_id="${serviceId}"}[${rangeWindow}]))`, + start: queryStart, + end: window.end, + stepSeconds: window.stepSeconds, + }).catch(() => []), + ]); + + const buckets = createServiceMetricBuckets(window); + const bucketsByTimestamp = new Map( + buckets.map((bucket) => [bucket.timestamp, bucket]), + ); + const statusCodes = new Set(); + + for (const result of requestResults) { + const status = result.metric.code || "unknown"; + statusCodes.add(status); + for (const point of matrixResultToPoints(result)) { + const bucket = bucketsByTimestamp.get(point.timestamp); + if (!bucket) continue; + const requests = Math.max(0, Math.round(point.value)); + bucket.statuses[status] = (bucket.statuses[status] ?? 0) + requests; + bucket.totalRequests += requests; + } + } + + applySingleSeries(bucketsByTimestamp, p50Results, (bucket, value) => { + bucket.p50ResponseTimeMs = value * 1000; + }); + applySingleSeries(bucketsByTimestamp, p90Results, (bucket, value) => { + bucket.p90ResponseTimeMs = value * 1000; + }); + applySingleSeries(bucketsByTimestamp, p95Results, (bucket, value) => { + bucket.p95ResponseTimeMs = value * 1000; + }); + applySingleSeries(bucketsByTimestamp, p99Results, (bucket, value) => { + bucket.p99ResponseTimeMs = value * 1000; + }); + applySingleSeries(bucketsByTimestamp, ingressResults, (bucket, value) => { + bucket.ingressBytesPerSecond = value; + }); + applySingleSeries(bucketsByTimestamp, egressResults, (bucket, value) => { + bucket.egressBytesPerSecond = value; + }); + applySingleSeries(bucketsByTimestamp, cpuResults, (bucket, value) => { + bucket.cpuUsagePercent = value; + }); + applySingleSeries( + bucketsByTimestamp, + memoryPercentResults, + (bucket, value) => { + bucket.memoryUsagePercent = value; + }, + ); + applySingleSeries(bucketsByTimestamp, memoryBytesResults, (bucket, value) => { + bucket.memoryUsedBytes = value; + }); + + return { + metricsEnabled: true, + range: options.range, + windowStart: window.start.toISOString(), + windowEnd: window.end.toISOString(), + stepSeconds: window.stepSeconds, + totalRequests: buckets.reduce( + (total, bucket) => total + bucket.totalRequests, + 0, + ), + statusCodes: sortStatusCodes([...statusCodes]), + buckets, + }; +} + async function queryInstantMetric( endpoint: EndpointConfig, metricName: string, @@ -414,6 +618,39 @@ async function queryRangeMetricGroup( return byServer; } +async function queryRangePromQL( + endpoint: EndpointConfig, + options: { + query: string; + start: Date; + end: Date; + stepSeconds: number; + }, +): Promise { + const url = new URL(`${endpoint.url}/api/v1/query_range`); + url.searchParams.set("query", options.query); + url.searchParams.set( + "start", + String(Math.floor(options.start.getTime() / 1000)), + ); + url.searchParams.set("end", String(Math.floor(options.end.getTime() / 1000))); + url.searchParams.set("step", String(options.stepSeconds)); + + const response = await fetch(url.toString(), buildFetchOptions(endpoint)); + if (!response.ok) { + throw new Error( + `Failed to query metrics range: ${response.status} ${response.statusText}`, + ); + } + + const data = (await response.json()) as VictoriaMatrixResponse; + if (data.status !== "success") { + throw new Error(data.error || "Failed to query metrics range"); + } + + return data.data?.result ?? []; +} + export function emptyHistory(): NodeMetricsHistory { return { cpuUsagePercent: [], @@ -424,9 +661,118 @@ export function emptyHistory(): NodeMetricsHistory { }; } +export function getMetricWindow( + range: MetricRange, + now = new Date(), +): { + start: Date; + end: Date; + durationMs: number; + stepSeconds: number; +} { + const option = METRIC_RANGE_OPTIONS[range]; + const end = new Date(Math.floor(now.getTime() / 1000) * 1000); + const start = new Date(end.getTime() - option.durationMs); + return { + start, + end, + durationMs: option.durationMs, + stepSeconds: option.stepSeconds, + }; +} + +export function buildTraefikServiceMatcher(serviceId: string): string { + return `^${escapePromRegex(serviceId)}(@file)?$`; +} + +export function formatPromDuration(seconds: number): string { + if (seconds % 86400 === 0) return `${seconds / 86400}d`; + if (seconds % 3600 === 0) return `${seconds / 3600}h`; + if (seconds % 60 === 0) return `${seconds / 60}m`; + return `${seconds}s`; +} + +function responseTimeQuery( + quantile: number, + traefikFilter: string, + rangeWindow: string, +) { + return `histogram_quantile(${quantile}, sum by (le) (rate(traefik_service_request_duration_seconds_bucket{${traefikFilter}}[${rangeWindow}])))`; +} + +function createServiceMetricBuckets(window: { + start: Date; + end: Date; + stepSeconds: number; +}): ServiceMetricsBucket[] { + const buckets: ServiceMetricsBucket[] = []; + const stepMs = window.stepSeconds * 1000; + for ( + let timestampMs = window.start.getTime() + stepMs; + timestampMs <= window.end.getTime(); + timestampMs += stepMs + ) { + buckets.push({ + timestamp: new Date(timestampMs).toISOString(), + totalRequests: 0, + statuses: {}, + p50ResponseTimeMs: null, + p90ResponseTimeMs: null, + p95ResponseTimeMs: null, + p99ResponseTimeMs: null, + ingressBytesPerSecond: null, + egressBytesPerSecond: null, + cpuUsagePercent: null, + memoryUsagePercent: null, + memoryUsedBytes: null, + }); + } + return buckets; +} + +function applySingleSeries( + bucketsByTimestamp: Map, + results: VictoriaMatrixResult[], + apply: (bucket: ServiceMetricsBucket, value: number) => void, +) { + const result = results[0]; + if (!result) return; + for (const point of matrixResultToPoints(result)) { + const bucket = bucketsByTimestamp.get(point.timestamp); + if (!bucket) continue; + apply(bucket, point.value); + } +} + +function matrixResultToPoints(result: { + values: Array<[number, string]>; +}): NodeMetricPoint[] { + return result.values + .map(([timestamp, rawValue]) => ({ + timestamp: new Date(timestamp * 1000).toISOString(), + value: Number.parseFloat(rawValue), + })) + .filter((point) => Number.isFinite(point.value)); +} + +function sortStatusCodes(statuses: string[]): string[] { + return Array.from(new Set(statuses)).sort((a, b) => { + const statusA = Number(a); + const statusB = Number(b); + if (Number.isFinite(statusA) && Number.isFinite(statusB)) { + return statusA - statusB; + } + return a.localeCompare(b); + }); +} + function escapePromQL(value: string) { return value .replace(/\\/g, "\\\\") .replace(/"/g, '\\"') .replace(/\n/g, "\\n"); } + +function escapePromRegex(value: string) { + return value.replace(/[\\^$.*+?()[\]{}|]/g, "\\$&"); +} diff --git a/web/public/setup.sh b/web/public/setup.sh index 30725e66..1509fe3a 100644 --- a/web/public/setup.sh +++ b/web/public/setup.sh @@ -407,6 +407,29 @@ accessLog: names: User-Agent: keep +metrics: + prometheus: + entryPoint: metrics + addEntryPointsLabels: false + addRoutersLabels: false + addServicesLabels: true + buckets: + - 0.005 + - 0.01 + - 0.025 + - 0.05 + - 0.075 + - 0.1 + - 0.25 + - 0.5 + - 0.75 + - 1 + - 2.5 + - 5 + - 10 + - 30 + - 60 + api: dashboard: false insecure: false @@ -418,6 +441,8 @@ experimental: version: v1.5.0 entryPoints: + metrics: + address: "127.0.0.1:9100" web: address: ":80" websecure: diff --git a/web/tests/victoria-logs-request-stats.test.ts b/web/tests/victoria-logs-request-stats.test.ts deleted file mode 100644 index ca732e25..00000000 --- a/web/tests/victoria-logs-request-stats.test.ts +++ /dev/null @@ -1,161 +0,0 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; -import { - buildRequestStatsBuckets, - createEmptyHttpRequestStats, - getRequestStatsWindow, - parseRequestStatsRange, - parseRequestStatsRows, -} from "@/lib/victoria-logs"; - -const SERVICE_ID = "123e4567-e89b-42d3-a456-426614174000"; - -describe("VictoriaLogs request stats", () => { - afterEach(() => { - vi.restoreAllMocks(); - vi.unstubAllEnvs(); - vi.unstubAllGlobals(); - vi.resetModules(); - }); - - it("parses JSON lines and ignores malformed rows", () => { - const rows = parseRequestStatsRows( - [ - '{"_time":"2026-07-02T00:00:00Z","status":200,"requests":"10"}', - "not-json", - '{"_time":"2026-07-02T00:01:00Z","status":"500","requests":5}', - ].join("\n"), - ); - - expect(rows).toEqual([ - { - _time: "2026-07-02T00:00:00Z", - status: 200, - requests: "10", - }, - { - _time: "2026-07-02T00:01:00Z", - status: "500", - requests: 5, - }, - ]); - }); - - it("builds complete buckets and fills gaps with zeroes", () => { - const buckets = buildRequestStatsBuckets( - [ - { _time: "2026-07-02T00:00:10Z", status: "200", requests: "3" }, - { _time: "2026-07-02T00:02:05Z", status: "404", requests: 7 }, - { _time: "2026-07-02T00:02:15Z", status: "500", requests: "2" }, - ], - { - start: new Date("2026-07-02T00:00:00Z"), - end: new Date("2026-07-02T00:02:30Z"), - stepSeconds: 60, - }, - ); - - expect(buckets).toEqual([ - { - timestamp: "2026-07-02T00:00:00.000Z", - totalRequests: 3, - statuses: { "200": 3 }, - }, - { - timestamp: "2026-07-02T00:01:00.000Z", - totalRequests: 0, - statuses: {}, - }, - { - timestamp: "2026-07-02T00:02:00.000Z", - totalRequests: 9, - statuses: { "404": 7, "500": 2 }, - }, - ]); - }); - - it("falls back to the default range for unsupported values", () => { - expect(parseRequestStatsRange("7d")).toBe("7d"); - expect(parseRequestStatsRange("week")).toBe("week"); - expect(parseRequestStatsRange("90d")).toBe("7d"); - expect(parseRequestStatsRange(null)).toBe("7d"); - }); - - it("uses a Monday UTC start for the week-to-date range", () => { - const window = getRequestStatsWindow( - "week", - new Date("2026-07-02T12:34:00Z"), - ); - - expect(window.start.toISOString()).toBe("2026-06-29T00:00:00.000Z"); - expect(window.end.toISOString()).toBe("2026-07-02T12:34:00.000Z"); - expect(window.stepSeconds).toBe(1800); - }); - - it("handles Sunday and Monday UTC week boundaries", () => { - expect( - getRequestStatsWindow( - "week", - new Date("2026-07-05T23:59:59Z"), - ).start.toISOString(), - ).toBe("2026-06-29T00:00:00.000Z"); - expect( - getRequestStatsWindow( - "week", - new Date("2026-07-06T00:00:00Z"), - ).start.toISOString(), - ).toBe("2026-07-06T00:00:00.000Z"); - }); - - it("queries a week-to-date stats window with one LogSQL range filter", async () => { - vi.stubEnv("VICTORIA_LOGS_URL", "http://victoria.test"); - vi.stubEnv("VICTORIA_LOGS_PRIVATE_URL", ""); - vi.resetModules(); - - const fetchMock = vi.fn(async (input: string | URL | Request) => { - const url = new URL(String(input)); - const query = url.searchParams.get("query"); - - expect(url.pathname).toBe("/select/logsql/query"); - expect(url.searchParams.get("limit")).toBe("11200"); - expect(query).toBe( - `service_id:${SERVICE_ID} log_type:http _time:[2026-06-29T00:00:00.000Z, 2026-07-02T12:34:00.000Z) | stats by (_time:30m, status) count() as requests | sort by (_time)`, - ); - - return new Response( - '{"_time":"2026-07-02T12:30:00Z","status":"200","requests":"3"}\n', - ); - }); - vi.stubGlobal("fetch", fetchMock); - - const { queryHttpRequestStats } = await import("@/lib/victoria-logs"); - const stats = await queryHttpRequestStats({ - serviceId: SERVICE_ID, - range: "week", - now: new Date("2026-07-02T12:34:00Z"), - }); - - expect(fetchMock).toHaveBeenCalledTimes(1); - expect(stats).toMatchObject({ - range: "week", - windowStart: "2026-06-29T00:00:00.000Z", - windowEnd: "2026-07-02T12:34:00.000Z", - stepSeconds: 1800, - totalRequests: 3, - statusCodes: ["200"], - }); - }); - - it("creates an empty non-breaking stats payload", () => { - expect( - createEmptyHttpRequestStats("1h", new Date("2026-07-02T01:00:00Z")), - ).toMatchObject({ - range: "1h", - windowStart: "2026-07-02T00:00:00.000Z", - windowEnd: "2026-07-02T01:00:00.000Z", - stepSeconds: 60, - totalRequests: 0, - statusCodes: [], - buckets: [], - }); - }); -}); diff --git a/web/tests/victoria-metrics-service-metrics.test.ts b/web/tests/victoria-metrics-service-metrics.test.ts new file mode 100644 index 00000000..9e0f1fd1 --- /dev/null +++ b/web/tests/victoria-metrics-service-metrics.test.ts @@ -0,0 +1,125 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + buildTraefikServiceMatcher, + createEmptyServiceMetrics, + formatPromDuration, + queryServiceMetrics, +} from "@/lib/victoria-metrics"; + +const SERVICE_ID = "123e4567-e89b-42d3-a456-426614174000"; +const END_TS = Date.parse("2026-07-02T12:34:00Z") / 1000; + +describe("VictoriaMetrics service metrics", () => { + afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + vi.unstubAllGlobals(); + vi.resetModules(); + }); + + it("formats Prometheus durations", () => { + expect(formatPromDuration(60)).toBe("1m"); + expect(formatPromDuration(300)).toBe("5m"); + expect(formatPromDuration(7200)).toBe("2h"); + expect(formatPromDuration(45)).toBe("45s"); + }); + + it("matches Traefik service labels with optional provider suffix", () => { + expect(buildTraefikServiceMatcher(SERVICE_ID)).toBe( + `^${SERVICE_ID}(@file)?$`, + ); + expect(buildTraefikServiceMatcher("svc.1")).toBe("^svc\\.1(@file)?$"); + }); + + it("creates an empty metrics payload", () => { + expect( + createEmptyServiceMetrics("1h", new Date("2026-07-02T01:00:00Z")), + ).toMatchObject({ + metricsEnabled: false, + range: "1h", + windowStart: "2026-07-02T00:00:00.000Z", + windowEnd: "2026-07-02T01:00:00.000Z", + stepSeconds: 60, + totalRequests: 0, + statusCodes: [], + buckets: [], + }); + }); + + it("queries service metrics from VictoriaMetrics only", async () => { + vi.stubEnv("VICTORIA_METRICS_URL", "http://victoria.test"); + vi.stubEnv("VICTORIA_METRICS_PRIVATE_URL", ""); + + const queries: string[] = []; + const starts: string[] = []; + const fetchMock = vi.fn(async (input: string | URL | Request) => { + const url = new URL(String(input)); + const query = url.searchParams.get("query") || ""; + queries.push(query); + starts.push(url.searchParams.get("start") || ""); + + expect(url.pathname).toBe("/api/v1/query_range"); + + if (query.includes("traefik_service_requests_total")) { + return jsonResponse([ + { + metric: { code: "200" }, + values: [[END_TS, "3"]], + }, + ]); + } + if (query.includes("histogram_quantile(0.95")) { + return jsonResponse([{ metric: {}, values: [[END_TS, "0.123"]] }]); + } + if (query.includes("traefik_service_responses_bytes_total")) { + return jsonResponse([{ metric: {}, values: [[END_TS, "2048"]] }]); + } + if (query.includes("techulus_service_cpu_usage_percent")) { + return jsonResponse([{ metric: {}, values: [[END_TS, "42"]] }]); + } + + return jsonResponse([]); + }); + vi.stubGlobal("fetch", fetchMock); + + const stats = await queryServiceMetrics({ + serviceId: SERVICE_ID, + range: "24h", + now: new Date("2026-07-02T12:34:00Z"), + }); + + expect(fetchMock).toHaveBeenCalledTimes(10); + expect(queries.some((query) => query.includes("LogSQL"))).toBe(false); + expect(queries).toContain( + `sum by (code) (increase(traefik_service_requests_total{service=~"^${SERVICE_ID}(@file)?$"}[5m]))`, + ); + expect(queries).toContain( + `sum(avg_over_time(techulus_service_cpu_usage_percent{service_id="${SERVICE_ID}"}[5m]))`, + ); + expect( + starts.every((start) => start === String(END_TS - 24 * 60 * 60 + 5 * 60)), + ).toBe(true); + expect(stats.totalRequests).toBe(3); + expect(stats.statusCodes).toEqual(["200"]); + expect(stats.buckets).toHaveLength(288); + expect(stats.buckets[0]?.timestamp).toBe("2026-07-01T12:39:00.000Z"); + + const lastBucket = stats.buckets.at(-1); + expect(lastBucket).toMatchObject({ + totalRequests: 3, + statuses: { "200": 3 }, + p95ResponseTimeMs: 123, + egressBytesPerSecond: 2048, + cpuUsagePercent: 42, + }); + }); +}); + +function jsonResponse(result: unknown[]) { + return new Response( + JSON.stringify({ + status: "success", + data: { result }, + }), + ); +} From f88b2c4a6e153b07f736cefe93c2ca97fd2fd4ce Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Thu, 9 Jul 2026 14:53:22 +1000 Subject: [PATCH 2/3] Reduce service metric label cardinality --- agent/internal/metrics/victoria.go | 53 ++++++++++++++++-- agent/internal/metrics/victoria_test.go | 54 +++++++++++++++++++ web/lib/victoria-metrics.ts | 2 +- .../victoria-metrics-service-metrics.test.ts | 3 ++ 4 files changed, 106 insertions(+), 6 deletions(-) diff --git a/agent/internal/metrics/victoria.go b/agent/internal/metrics/victoria.go index d74fad37..c7c5c721 100644 --- a/agent/internal/metrics/victoria.go +++ b/agent/internal/metrics/victoria.go @@ -21,6 +21,15 @@ type VictoriaMetricsSender struct { client *http.Client } +type serviceResourceStats struct { + ServiceID string + CPUUsagePercent float64 + MemoryUsagePercent float64 + MemoryUsedBytes float64 + NetworkReceiveBytes float64 + NetworkTransmitBytes float64 +} + func NewVictoriaMetricsSender(endpoint, serverID string) *VictoriaMetricsSender { var username, password string cleanEndpoint := strings.TrimRight(endpoint, "/") @@ -85,16 +94,19 @@ func (v *VictoriaMetricsSender) SendContainerStats(stats []container.ResourceSta return nil } + aggregates := aggregateContainerStats(stats) + if len(aggregates) == 0 { + return nil + } + timestampMs := collectedAt.UnixMilli() serverID := escapeLabelValue(v.serverID) var buf bytes.Buffer - for _, stat := range stats { + for _, stat := range aggregates { labels := map[string]string{ - "server_id": serverID, - "service_id": escapeLabelValue(stat.ServiceID), - "deployment_id": escapeLabelValue(stat.DeploymentID), - "container_id": escapeLabelValue(stat.ContainerID), + "server_id": serverID, + "service_id": escapeLabelValue(stat.ServiceID), } writeGaugeWithLabels(&buf, "techulus_service_cpu_usage_percent", labels, stat.CPUUsagePercent, timestampMs) writeGaugeWithLabels(&buf, "techulus_service_memory_usage_percent", labels, stat.MemoryUsagePercent, timestampMs) @@ -106,6 +118,37 @@ func (v *VictoriaMetricsSender) SendContainerStats(stats []container.ResourceSta return v.postPrometheusImport(buf.Bytes(), nil) } +func aggregateContainerStats(stats []container.ResourceStats) []serviceResourceStats { + byService := make(map[string]*serviceResourceStats) + for _, stat := range stats { + if stat.ServiceID == "" { + continue + } + aggregate := byService[stat.ServiceID] + if aggregate == nil { + aggregate = &serviceResourceStats{ServiceID: stat.ServiceID} + byService[stat.ServiceID] = aggregate + } + aggregate.CPUUsagePercent += stat.CPUUsagePercent + aggregate.MemoryUsagePercent += stat.MemoryUsagePercent + aggregate.MemoryUsedBytes += stat.MemoryUsedBytes + aggregate.NetworkReceiveBytes += stat.NetworkReceiveBytes + aggregate.NetworkTransmitBytes += stat.NetworkTransmitBytes + } + + serviceIDs := make([]string, 0, len(byService)) + for serviceID := range byService { + serviceIDs = append(serviceIDs, serviceID) + } + sort.Strings(serviceIDs) + + aggregates := make([]serviceResourceStats, 0, len(serviceIDs)) + for _, serviceID := range serviceIDs { + aggregates = append(aggregates, *byService[serviceID]) + } + return aggregates +} + func (v *VictoriaMetricsSender) SendPrometheusMetrics(data []byte, extraLabels map[string]string) error { if len(bytes.TrimSpace(data)) == 0 { return nil diff --git a/agent/internal/metrics/victoria_test.go b/agent/internal/metrics/victoria_test.go index 5854f197..9e8792f7 100644 --- a/agent/internal/metrics/victoria_test.go +++ b/agent/internal/metrics/victoria_test.go @@ -4,7 +4,11 @@ import ( "io" "net/http" "net/http/httptest" + "strings" "testing" + "time" + + "techulus/cloud-agent/internal/container" ) func TestSendPrometheusMetricsAddsExtraLabels(t *testing.T) { @@ -39,3 +43,53 @@ func TestSendPrometheusMetricsAddsExtraLabels(t *testing.T) { t.Fatalf("body = %q", gotBody) } } + +func TestSendContainerStatsAggregatesStableServiceLabels(t *testing.T) { + var gotBody string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + gotBody = string(body) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + sender := NewVictoriaMetricsSender(server.URL, "server-1") + err := sender.SendContainerStats([]container.ResourceStats{ + { + ContainerID: "container-a", + ServiceID: "svc-a", + DeploymentID: "dep-a", + CPUUsagePercent: 10, + MemoryUsagePercent: 1.5, + MemoryUsedBytes: 1024, + NetworkReceiveBytes: 100, + NetworkTransmitBytes: 200, + }, + { + ContainerID: "container-b", + ServiceID: "svc-a", + DeploymentID: "dep-b", + CPUUsagePercent: 20, + MemoryUsagePercent: 2.5, + MemoryUsedBytes: 2048, + NetworkReceiveBytes: 300, + NetworkTransmitBytes: 400, + }, + }, time.UnixMilli(1_700_000_000_000)) + if err != nil { + t.Fatalf("send container stats: %v", err) + } + + if strings.Contains(gotBody, "container_id") || strings.Contains(gotBody, "deployment_id") { + t.Fatalf("unexpected churn labels in body:\n%s", gotBody) + } + if !strings.Contains(gotBody, `techulus_service_cpu_usage_percent{server_id="server-1",service_id="svc-a"} 30.000000 1700000000000`) { + t.Fatalf("missing aggregated CPU metric:\n%s", gotBody) + } + if !strings.Contains(gotBody, `techulus_service_memory_usage_percent{server_id="server-1",service_id="svc-a"} 4.000000 1700000000000`) { + t.Fatalf("missing aggregated memory percent metric:\n%s", gotBody) + } + if !strings.Contains(gotBody, `techulus_service_memory_used_bytes{server_id="server-1",service_id="svc-a"} 3072.000000 1700000000000`) { + t.Fatalf("missing aggregated memory bytes metric:\n%s", gotBody) + } +} diff --git a/web/lib/victoria-metrics.ts b/web/lib/victoria-metrics.ts index 0fb62f7d..2e68ad3d 100644 --- a/web/lib/victoria-metrics.ts +++ b/web/lib/victoria-metrics.ts @@ -388,7 +388,7 @@ export async function queryServiceMetrics(options: { stepSeconds: window.stepSeconds, }).catch(() => []), queryRangePromQL(endpoint, { - query: `avg(avg_over_time(techulus_service_memory_usage_percent{service_id="${serviceId}"}[${rangeWindow}]))`, + query: `sum(avg_over_time(techulus_service_memory_usage_percent{service_id="${serviceId}"}[${rangeWindow}]))`, start: queryStart, end: window.end, stepSeconds: window.stepSeconds, diff --git a/web/tests/victoria-metrics-service-metrics.test.ts b/web/tests/victoria-metrics-service-metrics.test.ts index 9e0f1fd1..5f971156 100644 --- a/web/tests/victoria-metrics-service-metrics.test.ts +++ b/web/tests/victoria-metrics-service-metrics.test.ts @@ -96,6 +96,9 @@ describe("VictoriaMetrics service metrics", () => { expect(queries).toContain( `sum(avg_over_time(techulus_service_cpu_usage_percent{service_id="${SERVICE_ID}"}[5m]))`, ); + expect(queries).toContain( + `sum(avg_over_time(techulus_service_memory_usage_percent{service_id="${SERVICE_ID}"}[5m]))`, + ); expect( starts.every((start) => start === String(END_TS - 24 * 60 * 60 + 5 * 60)), ).toBe(true); From 1ee14b0c24dd9fdc3237c131c5977fe6cfe82553 Mon Sep 17 00:00:00 2001 From: Techulus Agent <291950465+techulus-agent@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:21:45 +1000 Subject: [PATCH 3/3] Add agent process resource metrics --- agent/internal/agent/agent.go | 1 + agent/internal/agent/reporting.go | 6 ++ agent/internal/health/health.go | 100 ++++++++++++++++++++++++ agent/internal/health/health_test.go | 67 ++++++++++++++++ agent/internal/metrics/victoria.go | 16 ++++ agent/internal/metrics/victoria_test.go | 31 ++++++++ 6 files changed, 221 insertions(+) create mode 100644 agent/internal/health/health_test.go diff --git a/agent/internal/agent/agent.go b/agent/internal/agent/agent.go index a1a4f7fb..fa85f72d 100644 --- a/agent/internal/agent/agent.go +++ b/agent/internal/agent/agent.go @@ -127,6 +127,7 @@ func NewAgent( type MetricsSender interface { SendSystemStats(stats *health.SystemStats, collectedAt time.Time) error + SendAgentStats(stats *health.AgentProcessStats, collectedAt time.Time) error SendContainerStats(stats []container.ResourceStats, collectedAt time.Time) error SendPrometheusMetrics(data []byte, extraLabels map[string]string) error } diff --git a/agent/internal/agent/reporting.go b/agent/internal/agent/reporting.go index ab4c6574..70f75cc3 100644 --- a/agent/internal/agent/reporting.go +++ b/agent/internal/agent/reporting.go @@ -53,6 +53,12 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport if err := a.MetricsSender.SendSystemStats(systemStats, collectedAt); err != nil { log.Printf("[metrics] failed to send system stats: %v", err) } + agentStats, err := health.CollectAgentProcessStats() + if err != nil { + log.Printf("[metrics] failed to collect agent stats: %v", err) + } else if err := a.MetricsSender.SendAgentStats(agentStats, collectedAt); err != nil { + log.Printf("[metrics] failed to send agent stats: %v", err) + } containerStats, err := container.CollectResourceStats() if err != nil { log.Printf("[metrics] failed to collect container stats: %v", err) diff --git a/agent/internal/health/health.go b/agent/internal/health/health.go index 3d140cdc..f23ce9d7 100644 --- a/agent/internal/health/health.go +++ b/agent/internal/health/health.go @@ -1,14 +1,19 @@ package health import ( + "math" + "os" "os/exec" + "runtime" "strconv" "strings" + "sync" "time" "github.com/shirou/gopsutil/v3/cpu" "github.com/shirou/gopsutil/v3/disk" "github.com/shirou/gopsutil/v3/mem" + "github.com/shirou/gopsutil/v3/process" ) type SystemStats struct { @@ -19,6 +24,12 @@ type SystemStats struct { DiskUsedGb int `json:"diskUsedGb"` } +type AgentProcessStats struct { + CPUUsagePercent float64 + MemoryUsagePercent float64 + MemoryUsedBytes uint64 +} + type NetworkPeerHealth struct { ID string `json:"id"` LastSeenSecs int `json:"lastSeenSecs"` @@ -45,6 +56,12 @@ type AgentHealthInfo struct { LastSyncAt string `json:"lastSyncAt"` } +var ( + agentProcessCPUMu sync.Mutex + agentProcessLastCPUTimes *cpu.TimesStat + agentProcessLastCPUTime time.Time +) + func CollectSystemStats() *SystemStats { stats := &SystemStats{} @@ -68,6 +85,89 @@ func CollectSystemStats() *SystemStats { return stats } +func CollectAgentProcessStats() (*AgentProcessStats, error) { + proc, err := process.NewProcess(int32(os.Getpid())) + if err != nil { + return nil, err + } + + stats := &AgentProcessStats{} + + cpuPercent, err := collectAgentCPUPercent(proc) + if err != nil { + return nil, err + } + stats.CPUUsagePercent = cpuPercent + + memInfo, err := proc.MemoryInfo() + if err != nil { + return nil, err + } + stats.MemoryUsedBytes = memInfo.RSS + + memPercent, err := proc.MemoryPercent() + if err != nil { + return nil, err + } + stats.MemoryUsagePercent = float64(memPercent) + + return stats, nil +} + +func collectAgentCPUPercent(proc *process.Process) (float64, error) { + cpuTimes, err := proc.Times() + if err != nil { + return 0, err + } + now := time.Now() + + agentProcessCPUMu.Lock() + defer agentProcessCPUMu.Unlock() + + if agentProcessLastCPUTimes == nil || agentProcessLastCPUTime.IsZero() { + // Delta-based CPU metrics need one sample to establish the baseline. + agentProcessLastCPUTimes = cpuTimes + agentProcessLastCPUTime = now + return 0, nil + } + + elapsedSeconds := now.Sub(agentProcessLastCPUTime).Seconds() + percent := calculateAgentCPUUsagePercent( + agentProcessLastCPUTimes, + cpuTimes, + elapsedSeconds, + runtime.NumCPU(), + ) + agentProcessLastCPUTimes = cpuTimes + agentProcessLastCPUTime = now + + return percent, nil +} + +func calculateAgentCPUUsagePercent(previous, current *cpu.TimesStat, elapsedSeconds float64, cpuCount int) float64 { + if previous == nil || current == nil { + return 0 + } + + cpuDeltaSeconds := processCPUTotal(current) - processCPUTotal(previous) + if elapsedSeconds <= 0 || cpuDeltaSeconds < 0 { + return 0 + } + if cpuCount <= 0 { + cpuCount = 1 + } + + percent := (cpuDeltaSeconds / elapsedSeconds) * 100 / float64(cpuCount) + if math.IsNaN(percent) || math.IsInf(percent, 0) { + return 0 + } + return percent +} + +func processCPUTotal(times *cpu.TimesStat) float64 { + return times.User + times.System +} + func CollectNetworkHealth(interfaceName string) *NetworkHealth { health := &NetworkHealth{ TunnelUp: false, diff --git a/agent/internal/health/health_test.go b/agent/internal/health/health_test.go new file mode 100644 index 00000000..1e169e9a --- /dev/null +++ b/agent/internal/health/health_test.go @@ -0,0 +1,67 @@ +package health + +import ( + "testing" + + "github.com/shirou/gopsutil/v3/cpu" +) + +func TestCalculateAgentCPUUsagePercentNormalizesByCPUCount(t *testing.T) { + previous := &cpu.TimesStat{User: 10, System: 5} + current := &cpu.TimesStat{User: 12, System: 7} + + got := calculateAgentCPUUsagePercent(previous, current, 1, 4) + if got != 100 { + t.Fatalf("cpu percent = %f, want 100", got) + } +} + +func TestCalculateAgentCPUUsagePercentWarmupOrInvalidInputs(t *testing.T) { + current := &cpu.TimesStat{User: 12, System: 7} + + tests := map[string]struct { + previous *cpu.TimesStat + current *cpu.TimesStat + elapsedSeconds float64 + cpuCount int + }{ + "missing previous sample": { + previous: nil, + current: current, + elapsedSeconds: 1, + cpuCount: 4, + }, + "missing current sample": { + previous: &cpu.TimesStat{User: 10, System: 5}, + current: nil, + elapsedSeconds: 1, + cpuCount: 4, + }, + "zero elapsed": { + previous: &cpu.TimesStat{User: 10, System: 5}, + current: current, + elapsedSeconds: 0, + cpuCount: 4, + }, + "negative counter delta": { + previous: current, + current: &cpu.TimesStat{User: 10, System: 5}, + elapsedSeconds: 1, + cpuCount: 4, + }, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + got := calculateAgentCPUUsagePercent( + test.previous, + test.current, + test.elapsedSeconds, + test.cpuCount, + ) + if got != 0 { + t.Fatalf("cpu percent = %f, want 0", got) + } + }) + } +} diff --git a/agent/internal/metrics/victoria.go b/agent/internal/metrics/victoria.go index c7c5c721..1bedd144 100644 --- a/agent/internal/metrics/victoria.go +++ b/agent/internal/metrics/victoria.go @@ -89,6 +89,22 @@ func (v *VictoriaMetricsSender) SendSystemStats(stats *health.SystemStats, colle return nil } +func (v *VictoriaMetricsSender) SendAgentStats(stats *health.AgentProcessStats, collectedAt time.Time) error { + if stats == nil { + return nil + } + + timestampMs := collectedAt.UnixMilli() + serverID := escapeLabelValue(v.serverID) + + var buf bytes.Buffer + writeGauge(&buf, "techulus_agent_cpu_usage_percent", serverID, stats.CPUUsagePercent, timestampMs) + writeGauge(&buf, "techulus_agent_memory_usage_percent", serverID, stats.MemoryUsagePercent, timestampMs) + writeGauge(&buf, "techulus_agent_memory_used_bytes", serverID, float64(stats.MemoryUsedBytes), timestampMs) + + return v.postPrometheusImport(buf.Bytes(), nil) +} + func (v *VictoriaMetricsSender) SendContainerStats(stats []container.ResourceStats, collectedAt time.Time) error { if len(stats) == 0 { return nil diff --git a/agent/internal/metrics/victoria_test.go b/agent/internal/metrics/victoria_test.go index 9e8792f7..ff691eb2 100644 --- a/agent/internal/metrics/victoria_test.go +++ b/agent/internal/metrics/victoria_test.go @@ -9,6 +9,7 @@ import ( "time" "techulus/cloud-agent/internal/container" + "techulus/cloud-agent/internal/health" ) func TestSendPrometheusMetricsAddsExtraLabels(t *testing.T) { @@ -44,6 +45,36 @@ func TestSendPrometheusMetricsAddsExtraLabels(t *testing.T) { } } +func TestSendAgentStatsWritesStableServerLabels(t *testing.T) { + var gotBody string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + gotBody = string(body) + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + sender := NewVictoriaMetricsSender(server.URL, "server-1") + err := sender.SendAgentStats(&health.AgentProcessStats{ + CPUUsagePercent: 1.25, + MemoryUsagePercent: 0.5, + MemoryUsedBytes: 64 * 1024 * 1024, + }, time.UnixMilli(1_700_000_000_000)) + if err != nil { + t.Fatalf("send agent stats: %v", err) + } + + if !strings.Contains(gotBody, `techulus_agent_cpu_usage_percent{server_id="server-1"} 1.250000 1700000000000`) { + t.Fatalf("missing agent CPU metric:\n%s", gotBody) + } + if !strings.Contains(gotBody, `techulus_agent_memory_usage_percent{server_id="server-1"} 0.500000 1700000000000`) { + t.Fatalf("missing agent memory percent metric:\n%s", gotBody) + } + if !strings.Contains(gotBody, `techulus_agent_memory_used_bytes{server_id="server-1"} 67108864.000000 1700000000000`) { + t.Fatalf("missing agent memory bytes metric:\n%s", gotBody) + } +} + func TestSendContainerStatsAggregatesStableServiceLabels(t *testing.T) { var gotBody string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {