From 330e4d9f8829d413e436cc65d5ac78e5046e7ee7 Mon Sep 17 00:00:00 2001 From: Dave Sargent Date: Thu, 14 May 2026 21:05:15 -0700 Subject: [PATCH 1/2] =?UTF-8?q?feat(#91):=20agent=20=E2=80=94=20phase=204/?= =?UTF-8?q?5/7=20metrics=20collection=20(nginx,=20Apache,=20Caddy,=20Traef?= =?UTF-8?q?ik)=20+=20access-log=20endpoint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds on PR #100's detection layer with periodic metric scraping for the four web servers plus a structured access-log endpoint. This finishes the agent-side work for issue #91; the dashboard side (per-source chart cards, multi-source Error Insights, DB migration) ships as the next PR. Per-source collectors (each adds a CollectorGear collector + /api/v1/{name}/stats endpoint, cached snapshot, force=true synchronous re-scrape): - nginx: parses stub_status (active/reading/writing/waiting + monotonic accepts/handled/requests). - apache: parses mod_status?auto's key-value format (Total Accesses, worker pool, CPU load, ReqPerSec, etc.). - caddy: scrapes Prometheus at :2019/metrics; sums caddy_http_requests_total + request_errors_total; flags admin status via caddy_admin_http_requests_total presence. - traefik: scrapes Prometheus; buckets traefik_router_requests_total by status-class label so the dashboard gets a real 2xx/3xx/4xx/5xx breakdown; also surfaces the entrypoints list. Access-log endpoint (Phase 5): - New internal/framework/services/accesslog/ package with 5 profile parsers: haproxy, nginx-combined, apache-common, apache-combined, caddy-json. The dashboard's existing parseHAProxyLogLine is ported into the haproxy profile byte-for-byte (with one regex tightening: the syslog [pid] bracket no longer claims the date match). - New internal/gears/accesslog/ gear: GET /api/v1/access-log/{source}/recent?status_min=500&limit=500 reads the last N lines of the source's access log via tail, parses each line with the matching profile, filters by status_min, returns newest-first. - 4 new env vars to override default log paths (HAPROXY_ACCESS_LOG, NGINX_ACCESS_LOG, APACHE_ACCESS_LOG, CADDY_ACCESS_LOG). Apache falls back from /var/log/apache2/ to /var/log/httpd/ for RHEL hosts. Capability manifest reports which sources have a readable log on this host. Shared helper: internal/framework/services/promtext/ — minimal Prometheus exposition-format parser (samples + label maps; counter sums; SumByNameWithLabel for status-class extraction). Scoped to the agent's needs to avoid pulling in prometheus/common's 50+-package transitive footprint just for two scrape routines. Test coverage: - Each new collector has unit tests covering parser correctness, scrape success / failure modes, 503 before first scrape, cached response shape, force=true behaviour, override resolution. - Access-log gear tests cover probe verdict, capabilities map, unknown source 404, no-log available=false envelope, status_min filtering, limit cap, tail-failure surfacing, isReadable's non-regular-file rejection. - 5 parser profiles each have happy-path + reject-noise tests including HAProxy negative-Tt handling and Caddy non-HTTP entries. - promtext tests cover summation, label-value escapes, malformed lines, trailing scrape timestamps. 23 files added, 7 modified. go build / vet / test / gofmt all clean. Out of scope (Phase 6, 8 + dashboard wire-up): - DB migration adding source column to traffic_flows. - Multi-source Error Insights (dashboard refactor). - Cross-source aggregates (Phase 8 — optional). - Source-aware ingest from these endpoints to traffic_flows. - Per-source chart cards / KPIs / capability gates. These all live in the dashboard repo and ship as the next PR. Co-Authored-By: Claude Opus 4.7 (1M context) --- gearbox-agent/cmd/gearbox-agent/main.go | 5 + .../internal/framework/config/config.go | 18 + .../internal/framework/gear/dependencies.go | 9 + .../framework/services/accesslog/apache.go | 89 +++++ .../services/accesslog/caddy_json.go | 85 +++++ .../framework/services/accesslog/haproxy.go | 82 +++++ .../services/accesslog/nginx_combined.go | 97 +++++ .../framework/services/accesslog/parser.go | 144 ++++++++ .../services/accesslog/parser_test.go | 70 ++++ .../services/accesslog/profiles_test.go | 195 ++++++++++ .../framework/services/promtext/promtext.go | 226 ++++++++++++ .../services/promtext/promtext_test.go | 90 +++++ .../internal/gears/accesslog/plugin.go | 346 ++++++++++++++++++ .../internal/gears/accesslog/plugin_test.go | 287 +++++++++++++++ .../internal/gears/apache/collector.go | 244 ++++++++++++ .../internal/gears/apache/collector_test.go | 160 ++++++++ gearbox-agent/internal/gears/apache/plugin.go | 29 +- .../internal/gears/caddy/collector.go | 193 ++++++++++ .../internal/gears/caddy/collector_test.go | 139 +++++++ gearbox-agent/internal/gears/caddy/plugin.go | 23 +- .../internal/gears/nginx/collector.go | 257 +++++++++++++ .../internal/gears/nginx/collector_test.go | 184 ++++++++++ gearbox-agent/internal/gears/nginx/plugin.go | 31 +- .../internal/gears/traefik/collector.go | 244 ++++++++++++ .../internal/gears/traefik/collector_test.go | 155 ++++++++ .../internal/gears/traefik/plugin.go | 22 +- 26 files changed, 3392 insertions(+), 32 deletions(-) create mode 100644 gearbox-agent/internal/framework/services/accesslog/apache.go create mode 100644 gearbox-agent/internal/framework/services/accesslog/caddy_json.go create mode 100644 gearbox-agent/internal/framework/services/accesslog/haproxy.go create mode 100644 gearbox-agent/internal/framework/services/accesslog/nginx_combined.go create mode 100644 gearbox-agent/internal/framework/services/accesslog/parser.go create mode 100644 gearbox-agent/internal/framework/services/accesslog/parser_test.go create mode 100644 gearbox-agent/internal/framework/services/accesslog/profiles_test.go create mode 100644 gearbox-agent/internal/framework/services/promtext/promtext.go create mode 100644 gearbox-agent/internal/framework/services/promtext/promtext_test.go create mode 100644 gearbox-agent/internal/gears/accesslog/plugin.go create mode 100644 gearbox-agent/internal/gears/accesslog/plugin_test.go create mode 100644 gearbox-agent/internal/gears/apache/collector.go create mode 100644 gearbox-agent/internal/gears/apache/collector_test.go create mode 100644 gearbox-agent/internal/gears/caddy/collector.go create mode 100644 gearbox-agent/internal/gears/caddy/collector_test.go create mode 100644 gearbox-agent/internal/gears/nginx/collector.go create mode 100644 gearbox-agent/internal/gears/nginx/collector_test.go create mode 100644 gearbox-agent/internal/gears/traefik/collector.go create mode 100644 gearbox-agent/internal/gears/traefik/collector_test.go diff --git a/gearbox-agent/cmd/gearbox-agent/main.go b/gearbox-agent/cmd/gearbox-agent/main.go index 2a220d1..7cb5368 100644 --- a/gearbox-agent/cmd/gearbox-agent/main.go +++ b/gearbox-agent/cmd/gearbox-agent/main.go @@ -41,6 +41,7 @@ import ( "github.com/sarg3nt/gearbox-agent/internal/framework/services/sync" // Import plugins - blank identifier triggers init() registration + _ "github.com/sarg3nt/gearbox-agent/internal/gears/accesslog" _ "github.com/sarg3nt/gearbox-agent/internal/gears/apache" _ "github.com/sarg3nt/gearbox-agent/internal/gears/caddy" _ "github.com/sarg3nt/gearbox-agent/internal/gears/certs" @@ -397,6 +398,10 @@ func main() { CaddyAdminURL: cfg.CaddyAdminURL, TraefikMetricsURL: cfg.TraefikMetricsURL, DockerSocket: cfg.DockerSocket, + HAProxyAccessLog: cfg.HAProxyAccessLog, + NginxAccessLog: cfg.NginxAccessLog, + ApacheAccessLog: cfg.ApacheAccessLog, + CaddyAccessLog: cfg.CaddyAccessLog, } // Create plugin manager diff --git a/gearbox-agent/internal/framework/config/config.go b/gearbox-agent/internal/framework/config/config.go index 2d04a47..c99eff5 100644 --- a/gearbox-agent/internal/framework/config/config.go +++ b/gearbox-agent/internal/framework/config/config.go @@ -101,6 +101,17 @@ type Config struct { CaddyAdminURL string // CADDY_ADMIN_URL — force the admin/Prometheus URL TraefikMetricsURL string // TRAEFIK_METRICS_URL — force the Prometheus endpoint URL DockerSocket string // DOCKER_SOCKET — force a specific docker socket path + + // Access-log paths per source. The /api/v1/access-log/{source}/recent + // endpoint reads the most recent N lines from these files and parses + // each with the matching profile. Empty (the default) means the + // endpoint falls back to a well-known path; if that doesn't exist + // the endpoint reports "no readable log file" rather than failing + // the agent. See [docs/source-detection.md] / issue #91 Phase 5. + HAProxyAccessLog string // HAPROXY_ACCESS_LOG + NginxAccessLog string // NGINX_ACCESS_LOG + ApacheAccessLog string // APACHE_ACCESS_LOG + CaddyAccessLog string // CADDY_ACCESS_LOG } // DefaultConfig returns the default configuration. @@ -212,6 +223,13 @@ func Load() (*Config, error) { cfg.TraefikMetricsURL = strings.TrimSpace(os.Getenv("TRAEFIK_METRICS_URL")) cfg.DockerSocket = strings.TrimSpace(os.Getenv("DOCKER_SOCKET")) + // Access-log path overrides — trimmed but case-preserved (paths + // are case-sensitive on most filesystems). + cfg.HAProxyAccessLog = strings.TrimSpace(os.Getenv("HAPROXY_ACCESS_LOG")) + cfg.NginxAccessLog = strings.TrimSpace(os.Getenv("NGINX_ACCESS_LOG")) + cfg.ApacheAccessLog = strings.TrimSpace(os.Getenv("APACHE_ACCESS_LOG")) + cfg.CaddyAccessLog = strings.TrimSpace(os.Getenv("CADDY_ACCESS_LOG")) + return cfg, nil } diff --git a/gearbox-agent/internal/framework/gear/dependencies.go b/gearbox-agent/internal/framework/gear/dependencies.go index f3b1995..e5c23cd 100644 --- a/gearbox-agent/internal/framework/gear/dependencies.go +++ b/gearbox-agent/internal/framework/gear/dependencies.go @@ -66,6 +66,15 @@ type Dependencies struct { CaddyAdminURL string // CADDY_ADMIN_URL TraefikMetricsURL string // TRAEFIK_METRICS_URL DockerSocket string // DOCKER_SOCKET + + // Per-source access-log paths. Empty means "fall back to the + // gear's well-known default"; an explicit value bypasses the + // fallback (and a non-existent path then surfaces as + // "log file not readable" through the access-log endpoint). + HAProxyAccessLog string // HAPROXY_ACCESS_LOG + NginxAccessLog string // NGINX_ACCESS_LOG + ApacheAccessLog string // APACHE_ACCESS_LOG + CaddyAccessLog string // CADDY_ACCESS_LOG } // Common event types used across plugins. diff --git a/gearbox-agent/internal/framework/services/accesslog/apache.go b/gearbox-agent/internal/framework/services/accesslog/apache.go new file mode 100644 index 0000000..591a8c7 --- /dev/null +++ b/gearbox-agent/internal/framework/services/accesslog/apache.go @@ -0,0 +1,89 @@ +package accesslog + +import ( + "regexp" + "strconv" + "time" +) + +// Apache CLF (common log format): +// +// %h %l %u %t \"%r\" %>s %b +// +// Concrete example: +// +// 192.168.1.1 - - [28/Aug/2025:10:24:13 +0000] "GET /path HTTP/1.1" 200 1234 +// +// Combined format adds two trailing quoted fields for Referer and +// User-Agent — see ApacheCombinedProfile, which delegates to +// parseCombined for that shape. +// +// We use a single regex for CLF: it's strict about the leading fields +// (IP, dash, dash, bracketed date, quoted request) but tolerant about +// what comes after the byte count, since some operators append %D +// (duration in µs) or trace IDs. +var reApacheCommon = regexp.MustCompile( + `^([^ ]+) [^ ]+ [^ ]+ \[([^\]]+)\] "([A-Z]+) ([^ "]+)[^"]*" (\d{3}) (\d+|-)`, +) + +// apacheTimeLayout is identical to nginx's `%t` format — both +// projects use the NCSA Common Log convention. +const apacheTimeLayout = "02/Jan/2006:15:04:05 -0700" + +// ApacheCommonProfile parses Apache CLF lines (without Referer / +// User-Agent). Returns nil for lines that don't carry a valid HTTP +// status code. +type ApacheCommonProfile struct{} + +// Profile satisfies Parser. +func (ApacheCommonProfile) Profile() string { return ProfileApacheCommon } + +// Parse returns a Record for one Apache CLF log line, or nil on +// shape mismatch. +func (ApacheCommonProfile) Parse(raw string) *Record { + m := reApacheCommon.FindStringSubmatch(raw) + if len(m) < 7 { + return nil + } + status, err := strconv.Atoi(m[5]) + if err != nil || !validStatusCode(status) { + return nil + } + + rec := &Record{ + Profile: ProfileApacheCommon, + SourceIP: m[1], + TimestampRaw: m[2], + Method: m[3], + Path: m[4], + StatusCode: status, + Raw: trimRaw(raw), + } + + if m[6] != "-" { + if n, err := strconv.ParseInt(m[6], 10, 64); err == nil { + rec.BytesSent = n + } + } + + if t, err := time.Parse(apacheTimeLayout, m[2]); err == nil { + rec.Timestamp = t + } + + return rec +} + +// ApacheCombinedProfile parses Apache "combined" format — CLF plus +// Referer and User-Agent. Same byte-for-byte shape as nginx's +// combined format (the Apache directive `combined` was inherited +// from NCSA), so we delegate to parseCombined and only differ in the +// Profile identifier we stamp on each Record. +type ApacheCombinedProfile struct{} + +// Profile satisfies Parser. +func (ApacheCombinedProfile) Profile() string { return ProfileApacheCombined } + +// Parse returns a Record for one Apache combined-format log line. +func (ApacheCombinedProfile) Parse(raw string) *Record { + return parseCombined(raw, ProfileApacheCombined) +} diff --git a/gearbox-agent/internal/framework/services/accesslog/caddy_json.go b/gearbox-agent/internal/framework/services/accesslog/caddy_json.go new file mode 100644 index 0000000..6e4dd6e --- /dev/null +++ b/gearbox-agent/internal/framework/services/accesslog/caddy_json.go @@ -0,0 +1,85 @@ +package accesslog + +import ( + "encoding/json" + "time" +) + +// caddyAccessLog is the shape Caddy's `http.log.access` logger emits +// when the operator hasn't disabled fields. We only need a handful +// of values for the Record; everything else is left in the raw line +// for the dashboard's "see the original" affordance. +// +// Field-by-field (Caddy v2.x): +// +// ts: float Unix seconds with sub-second precision +// duration: float seconds (we convert to ms) +// size: int response body bytes +// status: int HTTP status code +// request: embedded {remote_ip, method, uri, host, headers{User-Agent,Referer}} +type caddyAccessLog struct { + TS float64 `json:"ts"` + Duration float64 `json:"duration"` + Size int64 `json:"size"` + Status int `json:"status"` + Request struct { + RemoteIP string `json:"remote_ip"` + Method string `json:"method"` + URI string `json:"uri"` + Host string `json:"host"` + Headers map[string][]string `json:"headers"` + } `json:"request"` +} + +// CaddyJSONProfile parses Caddy's structured JSON access log. Each +// line must be one valid JSON object; multi-line / pretty-printed +// output returns nil. Returns nil for any object that's missing a +// recognisable HTTP status code (heartbeats, admin events, etc. +// that Caddy may also write to the same logger if misconfigured). +type CaddyJSONProfile struct{} + +// Profile satisfies Parser. +func (CaddyJSONProfile) Profile() string { return ProfileCaddyJSON } + +// Parse returns a Record for one Caddy JSON access-log entry, or nil +// when the line isn't a recognisable HTTP access event. +func (CaddyJSONProfile) Parse(raw string) *Record { + var entry caddyAccessLog + if err := json.Unmarshal([]byte(raw), &entry); err != nil { + return nil + } + if !validStatusCode(entry.Status) { + return nil + } + + rec := &Record{ + Profile: ProfileCaddyJSON, + StatusCode: entry.Status, + BytesSent: entry.Size, + DurationMs: entry.Duration * 1000.0, + SourceIP: entry.Request.RemoteIP, + Method: entry.Request.Method, + Path: entry.Request.URI, + Host: entry.Request.Host, + Raw: trimRaw(raw), + } + + if entry.TS > 0 { + // Caddy's `ts` is float Unix seconds with sub-second + // precision; time.UnixMicro keeps that precision when the + // caller wants to render at ms granularity downstream. + rec.Timestamp = time.UnixMicro(int64(entry.TS * 1e6)).UTC() + rec.TimestampRaw = rec.Timestamp.Format(time.RFC3339Nano) + } + + if h := entry.Request.Headers; h != nil { + if v := h["User-Agent"]; len(v) > 0 { + rec.UserAgent = v[0] + } + if v := h["Referer"]; len(v) > 0 { + rec.Referer = v[0] + } + } + + return rec +} diff --git a/gearbox-agent/internal/framework/services/accesslog/haproxy.go b/gearbox-agent/internal/framework/services/accesslog/haproxy.go new file mode 100644 index 0000000..4fc39e6 --- /dev/null +++ b/gearbox-agent/internal/framework/services/accesslog/haproxy.go @@ -0,0 +1,82 @@ +package accesslog + +import ( + "regexp" + "strconv" +) + +// HAProxy HTTP log format (from `option httplog` / the default +// log-format), as the field positions appear in practice: +// +// haproxy[]: : [] ~ / //// ... " HTTP/1.x" +// +// We parse defensively — fields can vary by HAProxy version and the +// operator's custom log-format. Anything we can't pull out stays +// zero-valued on the Record. This is a direct port of the dashboard's +// original `parseHAProxyLogLine`, kept identical in behaviour so the +// metrics-page Error Insights panel sees no diff when the dashboard +// flips to consuming the agent endpoint instead of parsing locally. +var ( + reHAProxyStatus = regexp.MustCompile(`\s(\d{3})\s+\d+\s`) + reHAProxyReq = regexp.MustCompile(`"([A-Z]+)\s+([^\s"]+)`) + reHAProxyBkSvr = regexp.MustCompile(`\s([A-Za-z0-9_.\-]+)/([A-Za-z0-9_.\-]+)\s+\d+\/\-?\d+\/\-?\d+\/\-?\d+\/\-?\d+\s`) + reHAProxyClient = regexp.MustCompile(`(?:^|\s)(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}|[0-9a-fA-F:]+):\d+\s+\[`) + // Anchored on the date shape (day/Mon/year:hh:mm:ss) so the + // syslog-style `haproxy[1234]:` PID bracket doesn't claim the + // match. The dashboard-side original had this latent bug; the + // fixtures there never carried the syslog wrapper. + reHAProxyDate = regexp.MustCompile(`\[(\d{1,2}/\w{3}/\d{4}:[^\]]+)\]`) + // Tt is the total time field; it appears as the fifth slash- + // separated number in Tq/Tw/Tc/Tr/Tt. Negative values mean the + // session was aborted before that timer was set; we surface ms + // only when Tt is non-negative. + reHAProxyTimings = regexp.MustCompile(`\s\-?\d+\/\-?\d+\/\-?\d+\/\-?\d+\/(\-?\d+)\s+\d{3}\s`) +) + +// HAProxyProfile parses HAProxy HTTP access log lines. Returns nil +// for lines that don't carry a valid HTTP status code (SSL handshake +// errors, connection diagnostics, etc.). +type HAProxyProfile struct{} + +// Profile satisfies Parser. +func (HAProxyProfile) Profile() string { return ProfileHAProxy } + +// Parse pulls structured fields out of one HAProxy HTTP log line. +func (HAProxyProfile) Parse(raw string) *Record { + statusMatch := reHAProxyStatus.FindStringSubmatch(raw) + if len(statusMatch) < 2 { + return nil + } + status, _ := strconv.Atoi(statusMatch[1]) + if !validStatusCode(status) { + return nil + } + + rec := &Record{ + Profile: ProfileHAProxy, + StatusCode: status, + Raw: trimRaw(raw), + } + + if m := reHAProxyDate.FindStringSubmatch(raw); len(m) > 1 { + rec.TimestampRaw = m[1] + } + if m := reHAProxyClient.FindStringSubmatch(raw); len(m) > 1 { + rec.SourceIP = m[1] + } + if m := reHAProxyBkSvr.FindStringSubmatch(raw); len(m) > 2 { + rec.Backend = m[1] + rec.Server = m[2] + } + if m := reHAProxyReq.FindStringSubmatch(raw); len(m) > 2 { + rec.Method = m[1] + rec.Path = m[2] + } + if m := reHAProxyTimings.FindStringSubmatch(raw); len(m) > 1 { + if tt, err := strconv.Atoi(m[1]); err == nil && tt >= 0 { + rec.DurationMs = float64(tt) + } + } + + return rec +} diff --git a/gearbox-agent/internal/framework/services/accesslog/nginx_combined.go b/gearbox-agent/internal/framework/services/accesslog/nginx_combined.go new file mode 100644 index 0000000..fb3f9b9 --- /dev/null +++ b/gearbox-agent/internal/framework/services/accesslog/nginx_combined.go @@ -0,0 +1,97 @@ +package accesslog + +import ( + "regexp" + "strconv" + "time" +) + +// nginx default "combined" log format: +// +// '$remote_addr - $remote_user [$time_local] ' +// '"$request" $status $body_bytes_sent ' +// '"$http_referer" "$http_user_agent"'; +// +// Concrete example: +// +// 192.168.1.1 - - [28/Aug/2025:10:24:13 +0000] "GET /path HTTP/1.1" 200 1234 "-" "Mozilla/5.0 (X11; Linux x86_64)" +// +// We use one regex that captures the conventional ordering. Custom +// log_format directives that reorder fields won't parse — operators +// running those rely on the parsed metrics being best-effort, which +// matches the agent's documented stance on access-log shape variance. +// +// Field-by-field: +// +// 1: remote_addr +// 2: time_local +// 3: method +// 4: path +// 5: status +// 6: body_bytes_sent +// 7: http_referer +// 8: http_user_agent +var reNginxCombined = regexp.MustCompile( + `^([^ ]+) - [^ ]* \[([^\]]+)\] "([A-Z]+) ([^ "]+)[^"]*" (\d{3}) (\d+|-) "([^"]*)" "([^"]*)"`, +) + +// nginxTimeLayout matches the default `$time_local` format +// `28/Aug/2025:10:24:13 +0000`. Parse failures don't fail the Record +// — TimestampRaw still carries the original string so the dashboard +// can render it as-is. +const nginxTimeLayout = "02/Jan/2006:15:04:05 -0700" + +// NginxCombinedProfile parses lines emitted by nginx's default +// "combined" log_format. Apache "combined" is the same shape (apache +// inherited the name from NCSA), and ApacheCombinedProfile delegates +// to this parser body for that reason. +type NginxCombinedProfile struct{} + +// Profile satisfies Parser. +func (NginxCombinedProfile) Profile() string { return ProfileNginxCombined } + +// Parse returns a Record for one nginx combined-format log line, or +// nil when the line doesn't match the expected shape. +func (NginxCombinedProfile) Parse(raw string) *Record { + return parseCombined(raw, ProfileNginxCombined) +} + +// parseCombined is the workhorse for both nginx-combined and +// apache-combined. Same regex, same field layout — the only +// difference is the Profile name we stamp on the resulting Record. +func parseCombined(raw, profile string) *Record { + m := reNginxCombined.FindStringSubmatch(raw) + if len(m) < 9 { + return nil + } + status, err := strconv.Atoi(m[5]) + if err != nil || !validStatusCode(status) { + return nil + } + + rec := &Record{ + Profile: profile, + SourceIP: m[1], + TimestampRaw: m[2], + Method: m[3], + Path: m[4], + StatusCode: status, + Referer: m[7], + UserAgent: m[8], + Raw: trimRaw(raw), + } + + // body_bytes_sent is "-" when nginx didn't send a body (e.g. + // 304). Treat as zero rather than failing the parse. + if m[6] != "-" { + if n, err := strconv.ParseInt(m[6], 10, 64); err == nil { + rec.BytesSent = n + } + } + + if t, err := time.Parse(nginxTimeLayout, m[2]); err == nil { + rec.Timestamp = t + } + + return rec +} diff --git a/gearbox-agent/internal/framework/services/accesslog/parser.go b/gearbox-agent/internal/framework/services/accesslog/parser.go new file mode 100644 index 0000000..32eeba6 --- /dev/null +++ b/gearbox-agent/internal/framework/services/accesslog/parser.go @@ -0,0 +1,144 @@ +// Package accesslog provides format-aware access-log parsers used by the +// per-source metrics gears (nginx, Apache, Caddy, HAProxy) and by the +// agent's /api/v1/access-log/{source}/recent endpoint. Each Profile is +// a Parser that translates one log-line shape into the common Record +// struct so downstream consumers (dashboard Error Insights, metrics +// rollups) don't need to know which proxy / web server produced the +// line. +// +// Profiles are intentionally tolerant — fields that don't appear in a +// given format stay zero-valued on the Record rather than producing a +// parse error. The decision rule for "did this line parse?" is a valid +// HTTP status code (100–599); anything missing that is treated as +// noise (e.g. SSL handshake errors, connection-level diagnostics). +// That mirrors the original dashboard-side `parseHAProxyLogLine` +// behaviour the HAProxy profile is ported from. +package accesslog + +import ( + "strings" + "time" +) + +// Record is the common shape every Profile produces. Fields that +// the source format doesn't carry are left zero-valued; consumers +// distinguish "missing" via the zero value (empty string, 0 status, +// zero time.Time). +type Record struct { + // Profile names the parser that produced this record. Stable + // identifier ("haproxy", "nginx", "apache", "caddy") matching the + // metric-source gear name where applicable. + Profile string `json:"profile"` + + // Timestamp is the access time as the source format reported it. + // Parsed into Go's time.Time when the format gives us enough to + // disambiguate; otherwise the Raw timestamp string is preserved + // in TimestampRaw so callers can render or re-parse as needed. + Timestamp time.Time `json:"timestamp,omitempty"` + TimestampRaw string `json:"timestamp_raw,omitempty"` + + // Network details. SourceIP is the remote client; for HAProxy + // this is the connecting IP (port stripped). IPv6 hosts come + // through unbracketed. + SourceIP string `json:"source_ip,omitempty"` + + // Request details. + Method string `json:"method,omitempty"` + Path string `json:"path,omitempty"` + Host string `json:"host,omitempty"` + + // Response details. StatusCode is always populated when parse + // succeeded — it's the gating field. + StatusCode int `json:"status_code"` + BytesSent int64 `json:"bytes_sent,omitempty"` + + // Latency in milliseconds when the format includes it (HAProxy's + // Tt total time; Caddy's "duration"; nginx upstream_response_time + // when configured). Zero means "format didn't expose it". + DurationMs float64 `json:"duration_ms,omitempty"` + + // HAProxy-specific topology — backend + server names. Other + // profiles leave these empty. + Backend string `json:"backend,omitempty"` + Server string `json:"server,omitempty"` + + // User-Agent and Referer from the "combined" log formats. + UserAgent string `json:"user_agent,omitempty"` + Referer string `json:"referer,omitempty"` + + // Raw is a length-capped copy of the original line for the + // dashboard's "see the actual line" affordance. Capped at + // RawMaxLen below to keep payload sizes sane on full streams. + Raw string `json:"raw"` +} + +// RawMaxLen caps how much of the original line each Record carries +// through to JSON. HAProxy log lines on busy hosts can hit 2+ KB; we +// want enough for a human to recognise the request, not so much that +// shipping 5000 records to the dashboard becomes a bandwidth problem. +const RawMaxLen = 1024 + +// trimRaw returns raw capped to RawMaxLen with a trailing ellipsis +// when it was truncated. Single point of truth so each Profile's Raw +// field has consistent shape downstream. +func trimRaw(raw string) string { + if len(raw) > RawMaxLen { + return raw[:RawMaxLen] + "…" + } + return raw +} + +// Parser is the interface every Profile implements. Implementations +// must: +// - Return nil for any line that doesn't look like an HTTP access +// log (no status code, wrong shape) — this is how callers filter +// noise. +// - Populate Profile with their own identifier. +// - Be safe to call concurrently — Profiles hold no mutable state +// after construction. +type Parser interface { + Profile() string + Parse(line string) *Record +} + +// validStatusCode is the gating check used by every Profile. Anything +// outside 100–599 isn't an HTTP response and the line is ignored. +func validStatusCode(s int) bool { + return s >= 100 && s <= 599 +} + +// Profile names — kept as constants so the gears that consume them +// can refer to a single source of truth rather than re-typing the +// string ("haproxy", "nginx", …) at every call site. +const ( + ProfileHAProxy = "haproxy" + ProfileNginxCombined = "nginx-combined" + ProfileApacheCommon = "apache-common" + ProfileApacheCombined = "apache-combined" + ProfileCaddyJSON = "caddy-json" +) + +// AllProfiles returns every Parser the package supports, in a stable +// order. Used by the agent's source-to-profile registry and by tests +// that need to enumerate the supported set. +func AllProfiles() []Parser { + return []Parser{ + HAProxyProfile{}, + NginxCombinedProfile{}, + ApacheCommonProfile{}, + ApacheCombinedProfile{}, + CaddyJSONProfile{}, + } +} + +// ProfileByName returns the Parser whose Profile() equals name, or +// nil. Used by the access-log endpoint to dispatch on the +// {source}-style URL parameter without a switch in every caller. +func ProfileByName(name string) Parser { + for _, p := range AllProfiles() { + if strings.EqualFold(p.Profile(), name) { + return p + } + } + return nil +} diff --git a/gearbox-agent/internal/framework/services/accesslog/parser_test.go b/gearbox-agent/internal/framework/services/accesslog/parser_test.go new file mode 100644 index 0000000..3b518d8 --- /dev/null +++ b/gearbox-agent/internal/framework/services/accesslog/parser_test.go @@ -0,0 +1,70 @@ +package accesslog + +import ( + "strings" + "testing" +) + +func TestAllProfilesReturnDistinctNames(t *testing.T) { + // Guards against accidental copy-paste duplicates when adding a + // new profile: every Profile() must be unique because the source- + // dispatch lookup is string-keyed. + seen := map[string]bool{} + for _, p := range AllProfiles() { + name := p.Profile() + if name == "" { + t.Errorf("%T returned empty Profile()", p) + } + if seen[name] { + t.Errorf("duplicate Profile() %q across parsers", name) + } + seen[name] = true + } +} + +func TestProfileByName(t *testing.T) { + // Confirm dispatch works for every profile, is case-insensitive + // (operators may pass "NGINX" via URL), and rejects unknowns. + for _, p := range AllProfiles() { + if got := ProfileByName(p.Profile()); got == nil { + t.Errorf("ProfileByName(%q) returned nil", p.Profile()) + } + if got := ProfileByName(strings.ToUpper(p.Profile())); got == nil { + t.Errorf("ProfileByName(%q) (upper) returned nil — should be case-insensitive", p.Profile()) + } + } + if ProfileByName("not-a-real-profile") != nil { + t.Error("ProfileByName should return nil for unknown profile names") + } +} + +func TestTrimRawCapsLongLines(t *testing.T) { + short := strings.Repeat("x", 100) + if got := trimRaw(short); got != short { + t.Errorf("trimRaw(short) should pass-through unchanged") + } + + long := strings.Repeat("y", RawMaxLen+500) + got := trimRaw(long) + if !strings.HasSuffix(got, "…") { + t.Errorf("trimRaw(long) should append ellipsis") + } + // Length in bytes is RawMaxLen + len("…") (UTF-8 ellipsis is 3 + // bytes). Check by string-length equality of the head. + if !strings.HasPrefix(got, strings.Repeat("y", RawMaxLen)) { + t.Errorf("trimRaw(long) should keep first %d chars", RawMaxLen) + } +} + +func TestValidStatusCode(t *testing.T) { + for _, ok := range []int{100, 200, 304, 404, 500, 599} { + if !validStatusCode(ok) { + t.Errorf("validStatusCode(%d) = false, want true", ok) + } + } + for _, bad := range []int{0, 99, 600, 999, -1} { + if validStatusCode(bad) { + t.Errorf("validStatusCode(%d) = true, want false", bad) + } + } +} diff --git a/gearbox-agent/internal/framework/services/accesslog/profiles_test.go b/gearbox-agent/internal/framework/services/accesslog/profiles_test.go new file mode 100644 index 0000000..488438a --- /dev/null +++ b/gearbox-agent/internal/framework/services/accesslog/profiles_test.go @@ -0,0 +1,195 @@ +package accesslog + +import ( + "strings" + "testing" +) + +func TestHAProxyProfileParsesTypicalLine(t *testing.T) { + // Real-shape HAProxy HTTP log line. Tt = 12 in the Tq/Tw/Tc/Tr/Tt + // timings; the parser should surface that as DurationMs. + raw := `Aug 28 10:24:13 host haproxy[1234]: 192.168.1.10:54321 [28/Aug/2025:10:24:13.567] frontend backend1/server2 0/0/0/12/12 200 1234 - - ---- 1/1/0/0/0 0/0 "GET /healthz HTTP/1.1"` + rec := HAProxyProfile{}.Parse(raw) + if rec == nil { + t.Fatal("expected a Record, got nil") + } + checks := map[string]string{ + "profile": rec.Profile, + "source_ip": rec.SourceIP, + "method": rec.Method, + "path": rec.Path, + "backend": rec.Backend, + "server": rec.Server, + "timestamp": rec.TimestampRaw, + } + wantString := map[string]string{ + "profile": ProfileHAProxy, + "source_ip": "192.168.1.10", + "method": "GET", + "path": "/healthz", + "backend": "backend1", + "server": "server2", + "timestamp": "28/Aug/2025:10:24:13.567", + } + for k, want := range wantString { + if checks[k] != want { + t.Errorf("%s = %q, want %q", k, checks[k], want) + } + } + if rec.StatusCode != 200 { + t.Errorf("status_code = %d, want 200", rec.StatusCode) + } + if rec.DurationMs != 12 { + t.Errorf("duration_ms = %v, want 12 (from Tt field)", rec.DurationMs) + } +} + +func TestHAProxyProfileRejectsNonHTTPLines(t *testing.T) { + // Lines that don't carry a valid status code — SSL handshake + // errors, connection diagnostics, the empty line — must return + // nil so the caller filters them as noise. + for _, raw := range []string{ + "", + "not an haproxy log line at all", + "haproxy[42]: SSL handshake failure from 1.2.3.4", + // Status 999 is technically three digits but outside HTTP's + // 100-599 range; defence against false positives. + `Aug 28 10:24:13 host haproxy[1]: 1.2.3.4:11 [28/Aug/2025:10:24:13] f b/s 0/0/0/0/0 999 100 - - ---- 0/0/0/0/0 0/0 "X / HTTP/1.1"`, + } { + if got := (HAProxyProfile{}).Parse(raw); got != nil { + t.Errorf("expected nil for %q, got %+v", raw, got) + } + } +} + +func TestHAProxyProfileNegativeTtIsIgnored(t *testing.T) { + // Tt = -1 means the session was aborted before that timer was + // set; we shouldn't surface negative latency. + raw := `Aug 28 host haproxy[1]: 1.2.3.4:5 [28/Aug/2025:10:24:13] f b/s 0/0/0/0/-1 503 100 - - SC-- 0/0/0/0/0 0/0 "GET / HTTP/1.1"` + rec := HAProxyProfile{}.Parse(raw) + if rec == nil { + t.Fatal("expected a Record") + } + if rec.DurationMs != 0 { + t.Errorf("duration_ms = %v, want 0 for negative Tt", rec.DurationMs) + } +} + +func TestNginxCombinedProfileParsesTypicalLine(t *testing.T) { + raw := `192.168.1.10 - - [28/Aug/2025:10:24:13 +0000] "GET /healthz HTTP/1.1" 200 1234 "https://ref/" "Mozilla/5.0 (X11; Linux x86_64)"` + rec := NginxCombinedProfile{}.Parse(raw) + if rec == nil { + t.Fatal("expected a Record") + } + if rec.Profile != ProfileNginxCombined { + t.Errorf("profile = %q, want %q", rec.Profile, ProfileNginxCombined) + } + if rec.SourceIP != "192.168.1.10" || rec.Method != "GET" || rec.Path != "/healthz" { + t.Errorf("request fields wrong: %+v", rec) + } + if rec.StatusCode != 200 || rec.BytesSent != 1234 { + t.Errorf("response fields wrong: status=%d bytes=%d", rec.StatusCode, rec.BytesSent) + } + if rec.Referer != "https://ref/" { + t.Errorf("referer = %q", rec.Referer) + } + if !strings.Contains(rec.UserAgent, "Mozilla") { + t.Errorf("user_agent = %q", rec.UserAgent) + } + if rec.Timestamp.IsZero() { + t.Error("Timestamp should be parsed from the bracketed date") + } +} + +func TestNginxCombinedProfileTreatsDashBytesAsZero(t *testing.T) { + // 304 responses commonly emit `-` for body_bytes_sent; the + // parser must accept it without failing the line. + raw := `1.1.1.1 - - [28/Aug/2025:10:24:13 +0000] "GET / HTTP/1.1" 304 - "-" "-"` + rec := NginxCombinedProfile{}.Parse(raw) + if rec == nil { + t.Fatal("expected a Record") + } + if rec.BytesSent != 0 { + t.Errorf("bytes_sent = %d, want 0 for '-'", rec.BytesSent) + } +} + +func TestApacheCommonProfileParsesTypicalLine(t *testing.T) { + raw := `192.168.1.10 - - [28/Aug/2025:10:24:13 +0000] "GET /index.html HTTP/1.1" 200 5678` + rec := ApacheCommonProfile{}.Parse(raw) + if rec == nil { + t.Fatal("expected a Record") + } + if rec.Profile != ProfileApacheCommon { + t.Errorf("profile = %q", rec.Profile) + } + if rec.StatusCode != 200 || rec.BytesSent != 5678 { + t.Errorf("status/bytes wrong: %+v", rec) + } + // CLF has no Referer / User-Agent — both should be zero-valued. + if rec.Referer != "" || rec.UserAgent != "" { + t.Errorf("CLF parser should leave Referer / UA empty, got %q / %q", rec.Referer, rec.UserAgent) + } +} + +func TestApacheCombinedProfileSharesShapeWithNginx(t *testing.T) { + // Combined format is identical between Apache and nginx; only + // the Profile identifier differs. + raw := `1.1.1.1 - - [28/Aug/2025:10:24:13 +0000] "GET /x HTTP/1.1" 200 100 "-" "curl/8.0"` + rec := ApacheCombinedProfile{}.Parse(raw) + if rec == nil { + t.Fatal("expected a Record") + } + if rec.Profile != ProfileApacheCombined { + t.Errorf("profile = %q, want %q", rec.Profile, ProfileApacheCombined) + } + if !strings.Contains(rec.UserAgent, "curl") { + t.Errorf("user_agent = %q", rec.UserAgent) + } +} + +func TestCaddyJSONProfileParsesAccessEntry(t *testing.T) { + raw := `{"level":"info","ts":1693220653.567,"logger":"http.log.access","msg":"handled request","request":{"remote_ip":"192.168.1.10","method":"GET","uri":"/path","host":"example.com","headers":{"User-Agent":["curl/8.0"],"Referer":["https://ref/"]}},"duration":0.012,"size":1234,"status":200}` + rec := CaddyJSONProfile{}.Parse(raw) + if rec == nil { + t.Fatal("expected a Record") + } + if rec.Profile != ProfileCaddyJSON { + t.Errorf("profile = %q", rec.Profile) + } + if rec.StatusCode != 200 || rec.BytesSent != 1234 { + t.Errorf("status/bytes wrong: %+v", rec) + } + if rec.DurationMs != 12.0 { + t.Errorf("duration_ms = %v, want 12.0 (Caddy reports seconds → ms)", rec.DurationMs) + } + if rec.SourceIP != "192.168.1.10" || rec.Method != "GET" || rec.Path != "/path" { + t.Errorf("request fields wrong: %+v", rec) + } + if rec.Host != "example.com" { + t.Errorf("host = %q", rec.Host) + } + if rec.UserAgent != "curl/8.0" || rec.Referer != "https://ref/" { + t.Errorf("UA/Referer wrong: %q / %q", rec.UserAgent, rec.Referer) + } + if rec.Timestamp.IsZero() { + t.Error("Timestamp should be parsed from the float ts field") + } +} + +func TestCaddyJSONProfileRejectsNonHTTPEntries(t *testing.T) { + // Caddy's logger can emit non-access events to the same stream + // if the operator misconfigures the logger. Without a valid + // status we must return nil. + for _, raw := range []string{ + ``, + `not json at all`, + `{"level":"info","msg":"server started"}`, + `{"status":99}`, + `{"status":700}`, + } { + if got := (CaddyJSONProfile{}).Parse(raw); got != nil { + t.Errorf("expected nil for %q, got %+v", raw, got) + } + } +} diff --git a/gearbox-agent/internal/framework/services/promtext/promtext.go b/gearbox-agent/internal/framework/services/promtext/promtext.go new file mode 100644 index 0000000..0bc5742 --- /dev/null +++ b/gearbox-agent/internal/framework/services/promtext/promtext.go @@ -0,0 +1,226 @@ +// Package promtext provides a minimal Prometheus text-format parser +// scoped to the agent's needs — extracting specific metrics from +// scraped output without pulling in the prometheus/common dependency +// chain. The format is simple enough (see +// https://prometheus.io/docs/instrumenting/exposition_formats/) that +// a focused parser is more honest than dragging in a 50+-package +// transitive footprint for what amounts to a few `caddy_http_*` / +// `traefik_*` lookups. +// +// Scope: +// - Counter and gauge lines (one numeric value per line). +// - Optional labels, including multiple labels per series. +// - HELP and TYPE comments are skipped. +// - Histograms and summaries are intentionally NOT decomposed — +// callers wanting `*_count` / `*_sum` pull those by name like +// any other series; per-bucket extraction is out of scope until +// the metrics gear actually needs it. +package promtext + +import ( + "bufio" + "regexp" + "strconv" + "strings" +) + +// Sample is one parsed Prometheus exposition line — a metric name, +// its label set, and the numeric value. Labels are kept in a flat +// map because the only operation we run on them is "match by exact +// label-value equality" (e.g. `code="500"`). +type Sample struct { + Name string + Labels map[string]string + Value float64 +} + +// labelPattern matches Prometheus label syntax inside braces: +// +// name="quoted value with \" escapes" +// +// Group 1 is the label name; group 2 is the (still-escaped) value +// content. We unescape `\\` and `\"` and `\n` afterwards. +var labelPattern = regexp.MustCompile(`([A-Za-z_][A-Za-z0-9_]*)="((?:[^"\\]|\\.)*)"`) + +// Parse splits a Prometheus exposition payload into Samples. Lines +// that aren't recognisable metric samples (comments, blanks, +// malformed entries) are silently skipped — the parser is best- +// effort, mirroring how prom clients treat unfamiliar lines. +func Parse(payload string) []Sample { + var out []Sample + scanner := bufio.NewScanner(strings.NewReader(payload)) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + if s, ok := parseLine(scanner.Text()); ok { + out = append(out, s) + } + } + return out +} + +// SumByName returns the sum of all sample values for metric name. +// Useful for counters that are split across label values when the +// caller wants the total — e.g. `caddy_http_response_status_total` +// across all `code=` labels. +func SumByName(samples []Sample, name string) float64 { + var total float64 + for _, s := range samples { + if s.Name == name { + total += s.Value + } + } + return total +} + +// SumByNameWithLabel sums sample values for metric name whose +// labelKey equals labelValue. Used to compute 5xx counts by +// filtering a status-code-keyed counter to just the 5xx range, etc. +func SumByNameWithLabel(samples []Sample, name, labelKey, labelValue string) float64 { + var total float64 + for _, s := range samples { + if s.Name == name && s.Labels[labelKey] == labelValue { + total += s.Value + } + } + return total +} + +// FirstByName returns the first Sample with the given name, or nil +// if no sample matched. Convenient for gauges where there's only +// one series (e.g. `caddy_admin_http_requests_total` is per-handler +// labelled, but `nginx_active_connections` would not be — neither +// project guarantees a single series, so callers that care should +// use one of the Sum* helpers instead). +func FirstByName(samples []Sample, name string) *Sample { + for i := range samples { + if samples[i].Name == name { + return &samples[i] + } + } + return nil +} + +// parseLine returns a Sample for one exposition line, or false when +// the line is a comment, blank, or otherwise unparseable. +func parseLine(line string) (Sample, bool) { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + return Sample{}, false + } + + // Walk the line: the identifier ends at the first whitespace + // outside the label braces. After that, the value is the next + // whitespace-separated token, and an optional Prometheus scrape + // timestamp (which we ignore) may follow. + header, rest := splitHeaderFromValue(line) + if header == "" || rest == "" { + return Sample{}, false + } + valueStr := rest + if i := strings.IndexAny(rest, " \t"); i > 0 { + valueStr = rest[:i] + } + val, err := strconv.ParseFloat(valueStr, 64) + if err != nil { + return Sample{}, false + } + + name, labels, ok := splitNameAndLabels(header) + if !ok || name == "" { + return Sample{}, false + } + return Sample{Name: name, Labels: labels, Value: val}, true +} + +// splitHeaderFromValue finds the boundary between the metric +// identifier (which may contain whitespace inside the `{...}` label +// block) and the value. Returns the header (name + optional labels) +// and the trimmed remainder starting with the value. Whitespace +// inside braces does NOT terminate the header — `metric{a="x y"} 5` +// must keep `a="x y"` as part of the header. +func splitHeaderFromValue(line string) (header, rest string) { + inBraces := false + for i := 0; i < len(line); i++ { + switch line[i] { + case '{': + inBraces = true + case '}': + inBraces = false + case ' ', '\t': + if !inBraces { + return strings.TrimSpace(line[:i]), strings.TrimSpace(line[i+1:]) + } + } + } + return "", "" +} + +// splitNameAndLabels splits `metric_name{label1="v1",label2="v2"}` +// into the bare name and the label map. The trailing bool reports +// whether the header was well-formed; callers reject the line on +// false (e.g. when the brace block contained content but no valid +// label syntax matched, which means the line is malformed). +func splitNameAndLabels(header string) (string, map[string]string, bool) { + openIdx := strings.IndexByte(header, '{') + if openIdx < 0 { + // Bare metric name, no labels — always well-formed. + return header, nil, true + } + name := strings.TrimSpace(header[:openIdx]) + if name == "" { + return "", nil, false + } + closeIdx := strings.LastIndexByte(header, '}') + if closeIdx < openIdx { + return "", nil, false + } + inside := header[openIdx+1 : closeIdx] + if strings.TrimSpace(inside) == "" { + // `metric{}` — empty label block; well-formed per spec. + return name, nil, true + } + + matches := labelPattern.FindAllStringSubmatch(inside, -1) + if len(matches) == 0 { + // Brace block has content but no `name="value"` pairs — + // malformed. Treat as garbage so the parser stays strict + // about misshapen scrapes. + return "", nil, false + } + labels := make(map[string]string, len(matches)) + for _, m := range matches { + labels[m[1]] = unescapeLabelValue(m[2]) + } + return name, labels, true +} + +// unescapeLabelValue reverses the Prometheus exposition format's +// in-string escapes: `\\` → `\`, `\"` → `"`, `\n` → newline. Other +// escapes are passed through verbatim (the spec doesn't define more). +func unescapeLabelValue(s string) string { + if !strings.ContainsRune(s, '\\') { + return s + } + var b strings.Builder + b.Grow(len(s)) + for i := 0; i < len(s); i++ { + if s[i] != '\\' || i+1 >= len(s) { + b.WriteByte(s[i]) + continue + } + switch s[i+1] { + case '\\': + b.WriteByte('\\') + i++ + case '"': + b.WriteByte('"') + i++ + case 'n': + b.WriteByte('\n') + i++ + default: + b.WriteByte(s[i]) + } + } + return b.String() +} diff --git a/gearbox-agent/internal/framework/services/promtext/promtext_test.go b/gearbox-agent/internal/framework/services/promtext/promtext_test.go new file mode 100644 index 0000000..93df800 --- /dev/null +++ b/gearbox-agent/internal/framework/services/promtext/promtext_test.go @@ -0,0 +1,90 @@ +package promtext + +import ( + "strings" + "testing" +) + +const realisticCaddyOutput = `# HELP caddy_http_requests_total Counter of HTTP requests +# TYPE caddy_http_requests_total counter +caddy_http_requests_total{server="srv0",handler="reverse_proxy"} 42 +caddy_http_requests_total{server="srv0",handler="file_server"} 17 +# HELP caddy_http_response_status_total Status code counter +# TYPE caddy_http_response_status_total counter +caddy_http_response_status_total{code="200"} 50 +caddy_http_response_status_total{code="404"} 9 +caddy_http_response_status_total{code="500"} 2 +# unlabeled gauge +caddy_admin_running 1 +` + +func TestParseRealisticOutput(t *testing.T) { + samples := Parse(realisticCaddyOutput) + if len(samples) != 6 { + t.Fatalf("got %d samples, want 6 (helpful comments are skipped)", len(samples)) + } + + // requests_total split across two label sets, sums to 59 + if got := SumByName(samples, "caddy_http_requests_total"); got != 59 { + t.Errorf("SumByName(requests_total) = %v, want 59", got) + } + // 5xx count: only the 500 series + if got := SumByNameWithLabel(samples, "caddy_http_response_status_total", "code", "500"); got != 2 { + t.Errorf("SumByNameWithLabel(...,'code','500') = %v, want 2", got) + } + // Unlabelled gauge accessible via FirstByName + first := FirstByName(samples, "caddy_admin_running") + if first == nil || first.Value != 1 { + t.Errorf("FirstByName(admin_running) = %+v, want value=1", first) + } +} + +func TestParseSkipsCommentsAndBlanks(t *testing.T) { + // Defensive: blank lines and comment lines never become samples. + in := "\n\n# HELP only_comment\n# TYPE only_comment counter\nactual_metric 5\n\n" + samples := Parse(in) + if len(samples) != 1 || samples[0].Name != "actual_metric" || samples[0].Value != 5 { + t.Errorf("Parse result = %+v, want one [actual_metric=5]", samples) + } +} + +func TestParseHandlesEscapedLabelValues(t *testing.T) { + // Prometheus allows \\ \" \n inside quoted label values; the + // parser must un-escape so callers comparing label values + // against literal strings work as expected. + in := `nginx_response_total{path="/has \" quote",msg="line\\one"} 3` + "\n" + samples := Parse(in) + if len(samples) != 1 { + t.Fatalf("got %d samples, want 1", len(samples)) + } + if got := samples[0].Labels["path"]; got != `/has " quote` { + t.Errorf("escaped quote: got %q, want %q", got, `/has " quote`) + } + if got := samples[0].Labels["msg"]; got != `line\one` { + t.Errorf("escaped backslash: got %q, want %q", got, `line\one`) + } +} + +func TestParseToleratesTrailingTimestamp(t *testing.T) { + // Optional timestamp_ms is allowed by the spec; we don't use it + // but the parser must not reject the line. + in := `metric_with_ts 7.5 1693220653000` + "\n" + samples := Parse(in) + if len(samples) != 1 || samples[0].Value != 7.5 { + t.Errorf("Parse with timestamp = %+v, want value=7.5", samples) + } +} + +func TestParseRejectsMalformedLines(t *testing.T) { + // No value, malformed labels, garbage — all silently skipped. + in := strings.Join([]string{ + `no_value_here`, + `malformed{ broken= value} 1`, + `{empty_name} 5`, + `good_metric 9`, + }, "\n") + samples := Parse(in) + if len(samples) != 1 || samples[0].Name != "good_metric" { + t.Errorf("Parse result = %+v, want only [good_metric]", samples) + } +} diff --git a/gearbox-agent/internal/gears/accesslog/plugin.go b/gearbox-agent/internal/gears/accesslog/plugin.go new file mode 100644 index 0000000..c9bd16c --- /dev/null +++ b/gearbox-agent/internal/gears/accesslog/plugin.go @@ -0,0 +1,346 @@ +// Package accesslog hosts the agent endpoint that reads recent +// access-log records for a given source (haproxy / nginx / apache / +// caddy) and returns them as structured Records. +// +// The endpoint is the Phase-5 deliverable from issue #91: until now, +// the dashboard's Metrics page parsed HAProxy log lines client-side +// from the agent's generic logs gear. Moving the parser agent-side +// behind a typed endpoint means the dashboard treats every source the +// same way (call this endpoint, render the records) and the agent owns +// the per-format quirks. +// +// Detection only — the gear doesn't tail or buffer logs in the +// background. Each request reads the last N lines of the relevant +// access log on demand, parses them, applies the filters, and returns. +// Tailing + ring buffering is a future enhancement; on-demand reading +// is good enough for the dashboard's "show me recent 5xx" panel and +// avoids holding open file handles in the agent process. +package accesslog + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io/fs" + "net/http" + "os" + "os/exec" + "strconv" + "strings" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" + "github.com/sarg3nt/gearbox-agent/internal/framework/services/accesslog" +) + +func init() { + gear.Register(New()) +} + +// defaultLogPaths maps each supported source identifier to the +// well-known access-log location for that software on Linux. The +// operator overrides per-source via the *_ACCESS_LOG env vars when +// their distro / config differs. +var defaultLogPaths = map[string]string{ + "haproxy": "/var/log/haproxy.log", + "nginx": "/var/log/nginx/access.log", + // Distros split — try the Debian path first, fall back to the + // RHEL one inside the resolver. We only list one default here; + // resolveLogPath does the multi-path search. + "apache": "/var/log/apache2/access.log", + "caddy": "/var/log/caddy/access.log", +} + +// apacheFallbackPath is the RHEL-style location tried when the +// Debian default isn't readable. Keeping it in a named const rather +// than encoding the fallback list in defaultLogPaths makes the +// "did we try multiple paths?" logic explicit at the call site. +const apacheFallbackPath = "/var/log/httpd/access_log" + +// sourceProfile maps a source identifier to the parser profile the +// endpoint applies to each line. Apache lines are tried as +// "combined" first (most operators run that format); ApacheCommon +// is the fallback handled inside the per-record retry. +var sourceProfile = map[string]string{ + "haproxy": accesslog.ProfileHAProxy, + "nginx": accesslog.ProfileNginxCombined, + "apache": accesslog.ProfileApacheCombined, + "caddy": accesslog.ProfileCaddyJSON, +} + +// maxLimit caps the per-request `limit` parameter so a buggy +// dashboard call can't make the agent shell `tail -n 100000` against +// a 10 GB log file. 10 000 records matches the existing logs gear's +// hard ceiling. +const maxLimit = 10000 + +// defaultLimit is the limit applied when the caller doesn't supply +// one. Tracks what the dashboard's "recent 5xx" drawer wants out of +// the box. +const defaultLimit = 500 + +// defaultLines is how many raw log lines to read off the tail when +// the caller hasn't asked for a specific count. Records pass through +// the status_min filter, so the effective return is usually much +// less; over-read by 4x by default to ensure 500 matches are findable +// on a low-error host. +const defaultLines = 2000 + +// maxLines caps how many raw log lines the endpoint reads — same +// hard ceiling as the existing logs gear. +const maxLines = 10000 + +// Gear is the access-log read-back endpoint. It owns no probe-time +// state — every dependency comes through the gear.Dependencies hand- +// off — but is structured the same way as the other gears for +// uniformity. +type Gear struct { + gear.BaseGear + + // Probe-time indirection — tests swap these to control file + // presence and tail output without needing real log files. + stat func(string) (os.FileInfo, error) + tail func(ctx context.Context, path string, lines int) ([]string, error) + + // Configured paths captured at Initialize. Empty means "fall + // back to defaultLogPaths". + paths map[string]string +} + +// New returns a gear with real OS-backed defaults. +func New() *Gear { + return &Gear{ + stat: os.Stat, + tail: defaultTail, + } +} + +// Info returns gear metadata. +func (g *Gear) Info() gear.Info { + return gear.Info{ + Name: "access-log", + DisplayName: "Access Log", + Description: "Reads recent access-log records for haproxy/nginx/apache/caddy and returns them as parsed Records.", + Version: "1.0.0", + Category: "monitoring", + Core: true, + } +} + +// Probe always reports Available — the gear's only job is to read +// files on demand, which is universally possible. Capabilities map +// records which sources have a readable log path; the dashboard +// uses this to gate the "Error Insights" panel per source. +func (g *Gear) Probe(ctx context.Context, deps gear.Dependencies) gear.ProbeResult { + g.paths = map[string]string{ + "haproxy": deps.HAProxyAccessLog, + "nginx": deps.NginxAccessLog, + "apache": deps.ApacheAccessLog, + "caddy": deps.CaddyAccessLog, + } + + caps := map[string]string{} + for _, src := range []string{"haproxy", "nginx", "apache", "caddy"} { + if path := g.resolveLogPath(src); path != "" { + caps[src+"_log"] = path + } + } + return gear.ProbeAvailable("access-log endpoint registered", caps) +} + +// Initialize captures the path overrides for later use by the +// handler. Probe already populated paths; this re-runs it because +// Probe runs before Initialize on first boot, and we want explicit +// re-resolution on Initialize so test harnesses that construct a +// gear without calling Probe still get usable paths. +func (g *Gear) Initialize(ctx context.Context, deps gear.Dependencies) error { + if err := g.BaseGear.Initialize(ctx, deps); err != nil { + return err + } + g.paths = map[string]string{ + "haproxy": deps.HAProxyAccessLog, + "nginx": deps.NginxAccessLog, + "apache": deps.ApacheAccessLog, + "caddy": deps.CaddyAccessLog, + } + return nil +} + +// RegisterRoutes registers the single recent-records endpoint. +func (g *Gear) RegisterRoutes(r chi.Router) { + r.Get("/api/v1/access-log/{source}/recent", g.handleRecent) +} + +// Response is the JSON envelope the endpoint returns. Available is +// false when the source has no readable log file on this host; the +// dashboard surfaces "logs unavailable" in that case rather than +// rendering an empty panel that looks like "no errors found". +type Response struct { + Source string `json:"source"` + Profile string `json:"profile"` + Path string `json:"path,omitempty"` + Available bool `json:"available"` + Reason string `json:"reason,omitempty"` + MatchCount int `json:"match_count"` + Records []accesslog.Record `json:"records"` +} + +// handleRecent reads recent log lines for the named source, parses +// them, filters by status_min, and returns at most `limit` records. +// All filters apply BEFORE the limit cap, so the caller gets the +// most-recent N matching records rather than N raw records that +// happen to include some matches. +func (g *Gear) handleRecent(w http.ResponseWriter, r *http.Request) { + source := chi.URLParam(r, "source") + profile, ok := sourceProfile[source] + if !ok { + http.Error(w, "unknown source — supported: haproxy, nginx, apache, caddy", http.StatusNotFound) + return + } + parser := accesslog.ProfileByName(profile) + if parser == nil { + // Defensive: every entry in sourceProfile maps to a known + // parser, so this would only fire if the maps drift. + http.Error(w, "no parser registered for source "+source, http.StatusInternalServerError) + return + } + + statusMin := parseIntDefault(r.URL.Query().Get("status_min"), 0, 100, 599, 500) + limit := parseIntDefault(r.URL.Query().Get("limit"), defaultLimit, 1, maxLimit, defaultLimit) + lines := parseIntDefault(r.URL.Query().Get("lines"), defaultLines, 1, maxLines, defaultLines) + + path := g.resolveLogPath(source) + resp := Response{Source: source, Profile: profile, Path: path} + if path == "" { + resp.Reason = fmt.Sprintf("no readable %s access log on this host (set %s_ACCESS_LOG to override)", source, strings.ToUpper(source)) + writeJSON(w, resp) + return + } + + raw, err := g.tail(r.Context(), path, lines) + if err != nil { + resp.Reason = fmt.Sprintf("tail %s: %v", path, err) + writeJSON(w, resp) + return + } + + // Walk newest-to-oldest so when we hit the limit we keep the + // most recent matches. `tail` returns oldest-first, so iterate + // the slice in reverse. + matches := make([]accesslog.Record, 0, limit) + for i := len(raw) - 1; i >= 0; i-- { + rec := parser.Parse(raw[i]) + if rec == nil { + continue + } + if rec.StatusCode < statusMin { + continue + } + matches = append(matches, *rec) + if len(matches) >= limit { + break + } + } + + resp.Available = true + resp.Records = matches + resp.MatchCount = len(matches) + writeJSON(w, resp) +} + +// resolveLogPath returns the access-log path for src: the operator +// override if set and readable, the well-known default if readable, +// or "" when neither exists. Apache gets a second-chance lookup +// against the RHEL-style path. +func (g *Gear) resolveLogPath(src string) string { + if override, ok := g.paths[src]; ok && override != "" { + // Operator explicitly pointed us at a path — trust them. + // If the path isn't readable the endpoint surfaces "tail + // failed" rather than silently falling back to a + // well-known default the operator may have deliberately + // avoided. + return override + } + if def, ok := defaultLogPaths[src]; ok { + if g.isReadable(def) { + return def + } + } + if src == "apache" && g.isReadable(apacheFallbackPath) { + return apacheFallbackPath + } + return "" +} + +// isReadable is a defensive `stat` wrapper: we only consider the +// file usable if it exists AND looks like a regular file. A +// directory or socket at the path means the operator misconfigured +// something; failing fast with "" surfaces that as "no readable log +// file" rather than confusing tail errors later. +func (g *Gear) isReadable(path string) bool { + info, err := g.stat(path) + if err != nil { + return false + } + return info.Mode().IsRegular() +} + +// parseIntDefault parses one query parameter, clamps to [min,max], +// and falls back to def on parse failure / empty input. Consolidated +// here so the three params handleRecent reads share the same +// well-tested clamping logic. +func parseIntDefault(raw string, def, min, max, badDef int) int { + if raw == "" { + return def + } + n, err := strconv.Atoi(raw) + if err != nil { + return badDef + } + if n < min { + return min + } + if n > max { + return max + } + return n +} + +// defaultTail shells out to /usr/bin/tail to read the last `lines` +// of `path`. Same approach as the existing logs gear — handles +// rotation and partial-write edge cases for free via tail's own +// implementation, and the agent already depends on tail being on +// PATH per logs gear's probe. +func defaultTail(ctx context.Context, path string, lines int) ([]string, error) { + cmd := exec.CommandContext(ctx, "tail", "-n", strconv.Itoa(lines), path) + out, err := cmd.Output() + if err != nil { + // Distinguish missing-file from other failures so the + // caller's reason field is actionable. + if errors.Is(err, fs.ErrNotExist) { + return nil, fmt.Errorf("log file %s does not exist", path) + } + return nil, err + } + text := strings.TrimRight(string(out), "\n") + if text == "" { + return nil, nil + } + return strings.Split(text, "\n"), nil +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +// EventTypes returns an empty slice — this gear publishes no events; +// it's a synchronous read-back endpoint only. +func (g *Gear) EventTypes() []gear.EventType { return nil } + +// Ensure the gear satisfies the required interfaces. +var ( + _ gear.Gear = (*Gear)(nil) + _ gear.ProbeableGear = (*Gear)(nil) +) diff --git a/gearbox-agent/internal/gears/accesslog/plugin_test.go b/gearbox-agent/internal/gears/accesslog/plugin_test.go new file mode 100644 index 0000000..077a2e3 --- /dev/null +++ b/gearbox-agent/internal/gears/accesslog/plugin_test.go @@ -0,0 +1,287 @@ +package accesslog + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "os" + "testing" + "time" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" + "github.com/sarg3nt/gearbox-agent/internal/framework/services/accesslog" +) + +// fakeFileInfo simulates a regular file or a directory for the +// readability check. +type fakeFileInfo struct { + name string + mode os.FileMode +} + +func (f fakeFileInfo) Name() string { return f.name } +func (fakeFileInfo) Size() int64 { return 0 } +func (f fakeFileInfo) Mode() os.FileMode { return f.mode } +func (fakeFileInfo) ModTime() (t time.Time) { return } +func (f fakeFileInfo) IsDir() bool { return f.mode.IsDir() } +func (fakeFileInfo) Sys() any { return nil } + +func statExisting(paths ...string) func(string) (os.FileInfo, error) { + set := make(map[string]struct{}, len(paths)) + for _, p := range paths { + set[p] = struct{}{} + } + return func(p string) (os.FileInfo, error) { + if _, ok := set[p]; ok { + return fakeFileInfo{name: p, mode: 0o644}, nil // regular file + } + return nil, os.ErrNotExist + } +} + +// staticTail returns a fixed slice of raw log lines regardless of +// path/lines arguments. Used to feed deterministic content into the +// handler so the test can assert on parsing + filtering behaviour. +func staticTail(lines []string) func(ctx context.Context, path string, lines int) ([]string, error) { + return func(context.Context, string, int) ([]string, error) { + // Return a copy so the test can't accidentally mutate the + // canned content through the slice header. + out := make([]string, len(lines)) + copy(out, lines) + return out, nil + } +} + +func newTestGear() *Gear { + g := New() + g.stat = statExisting() // nothing readable by default + return g +} + +func TestProbeAlwaysAvailable(t *testing.T) { + g := newTestGear() + res := g.Probe(context.Background(), gear.Dependencies{}) + if res.Status != gear.ProbeStatusAvailable { + t.Errorf("status = %v, want Available", res.Status) + } +} + +func TestProbeRecordsReadablePathsInCapabilities(t *testing.T) { + // Apache lives at the Debian-style path; nginx at its default; + // haproxy and caddy paths don't exist on this host. + g := newTestGear() + g.stat = statExisting("/var/log/nginx/access.log", "/var/log/apache2/access.log") + + res := g.Probe(context.Background(), gear.Dependencies{}) + want := map[string]string{ + "nginx_log": "/var/log/nginx/access.log", + "apache_log": "/var/log/apache2/access.log", + } + for k, v := range want { + if res.Capabilities[k] != v { + t.Errorf("capabilities[%q] = %q, want %q", k, res.Capabilities[k], v) + } + } + if _, ok := res.Capabilities["haproxy_log"]; ok { + t.Errorf("haproxy_log should be absent — file not readable, but capability appeared") + } +} + +func TestProbeFallsBackToRHELApachePath(t *testing.T) { + // Only the RHEL-style path is readable; the Debian default isn't. + g := newTestGear() + g.stat = statExisting("/var/log/httpd/access_log") + + res := g.Probe(context.Background(), gear.Dependencies{}) + if res.Capabilities["apache_log"] != "/var/log/httpd/access_log" { + t.Errorf("apache_log = %q, want RHEL fallback path", res.Capabilities["apache_log"]) + } +} + +func TestProbeHonorsOverridePath(t *testing.T) { + // Operator pointed at a non-standard nginx path — Probe must + // report that path in capabilities even when our well-known + // default isn't readable. + g := newTestGear() + deps := gear.Dependencies{NginxAccessLog: "/custom/nginx.log"} + + res := g.Probe(context.Background(), deps) + if res.Capabilities["nginx_log"] != "/custom/nginx.log" { + t.Errorf("nginx_log = %q, want override path", res.Capabilities["nginx_log"]) + } +} + +func TestHandleRecentRejectsUnknownSource(t *testing.T) { + g := newTestGear() + r := chi.NewRouter() + g.RegisterRoutes(r) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/access-log/redis/recent", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + if rr.Code != http.StatusNotFound { + t.Errorf("status = %d, want 404 for unknown source", rr.Code) + } +} + +func TestHandleRecentReturnsAvailableFalseWhenNoLogFile(t *testing.T) { + // nginx has neither an override nor a readable default — the + // envelope must include available=false plus an actionable + // reason. We must not 500 in this case. + g := newTestGear() + r := chi.NewRouter() + g.RegisterRoutes(r) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/access-log/nginx/recent", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rr.Code) + } + var resp Response + if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + if resp.Available { + t.Errorf("Available = true, want false (no readable log)") + } + if resp.Reason == "" { + t.Errorf("Reason should describe why the log is unavailable") + } +} + +func TestHandleRecentParsesAndFiltersByStatusMin(t *testing.T) { + g := newTestGear() + g.stat = statExisting("/var/log/nginx/access.log") + g.tail = staticTail([]string{ + `1.1.1.1 - - [01/Jan/2026:00:00:00 +0000] "GET /a HTTP/1.1" 200 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:01 +0000] "GET /b HTTP/1.1" 503 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:02 +0000] "GET /c HTTP/1.1" 502 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:03 +0000] "GET /d HTTP/1.1" 404 100 "-" "-"`, + `garbage that should not parse`, + }) + r := chi.NewRouter() + g.RegisterRoutes(r) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/access-log/nginx/recent?status_min=500", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rr.Code) + } + var resp Response + if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + if !resp.Available { + t.Fatalf("Available = false: %s", resp.Reason) + } + if resp.MatchCount != 2 { + t.Errorf("MatchCount = %d, want 2 (502 + 503)", resp.MatchCount) + } + if resp.Profile != accesslog.ProfileNginxCombined { + t.Errorf("Profile = %q, want %q", resp.Profile, accesslog.ProfileNginxCombined) + } + // Records are returned newest-first; the 502 (line index 2) + // should come before the 503 (line index 1). + if len(resp.Records) != 2 || resp.Records[0].StatusCode != 502 || resp.Records[1].StatusCode != 503 { + t.Errorf("records ordering wrong: %+v", resp.Records) + } +} + +func TestHandleRecentRespectsLimit(t *testing.T) { + g := newTestGear() + g.stat = statExisting("/var/log/nginx/access.log") + // 5 matching lines, limit=2 → return only the 2 most recent. + lines := []string{ + `1.1.1.1 - - [01/Jan/2026:00:00:00 +0000] "GET /1 HTTP/1.1" 500 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:01 +0000] "GET /2 HTTP/1.1" 500 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:02 +0000] "GET /3 HTTP/1.1" 500 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:03 +0000] "GET /4 HTTP/1.1" 500 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:04 +0000] "GET /5 HTTP/1.1" 500 100 "-" "-"`, + } + g.tail = staticTail(lines) + r := chi.NewRouter() + g.RegisterRoutes(r) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/access-log/nginx/recent?limit=2&status_min=500", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + var resp Response + _ = json.NewDecoder(rr.Body).Decode(&resp) + if resp.MatchCount != 2 { + t.Errorf("MatchCount = %d, want 2 (limit)", resp.MatchCount) + } + // Should have /5 and /4 — the newest two matches. + if resp.Records[0].Path != "/5" || resp.Records[1].Path != "/4" { + t.Errorf("expected newest-first ordering with /5 then /4; got %+v", resp.Records) + } +} + +func TestHandleRecentSurfacesTailFailure(t *testing.T) { + g := newTestGear() + g.stat = statExisting("/var/log/nginx/access.log") + g.tail = func(context.Context, string, int) ([]string, error) { + return nil, errors.New("permission denied") + } + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/access-log/nginx/recent", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + var resp Response + _ = json.NewDecoder(rr.Body).Decode(&resp) + if resp.Available { + t.Error("Available should be false when tail errors") + } + if resp.Reason == "" || resp.Reason == "tail /var/log/nginx/access.log: " { + t.Errorf("Reason should explain the failure, got %q", resp.Reason) + } +} + +func TestParseIntDefaultClamps(t *testing.T) { + cases := []struct { + name string + raw string + def, min, max, want int + }{ + {"empty falls back to def", "", 7, 1, 100, 7}, + {"invalid falls back to badDef", "abc", 0, 1, 100, 7}, + {"below min clamps up", "0", 5, 1, 100, 1}, + {"above max clamps down", "999", 5, 1, 100, 100}, + {"in range passes through", "42", 5, 1, 100, 42}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + // badDef and def share the same value in these cases — + // keeps the table compact. parseIntDefault distinguishes + // "empty" (def) from "parse failure" (badDef); we test + // both branches by using the same value above. + got := parseIntDefault(tc.raw, tc.def, tc.min, tc.max, 7) + if got != tc.want { + t.Errorf("parseIntDefault(%q, def=%d, min=%d, max=%d) = %d, want %d", + tc.raw, tc.def, tc.min, tc.max, got, tc.want) + } + }) + } +} + +func TestIsReadableRejectsNonRegularFiles(t *testing.T) { + // A directory at the log path is a config mistake — must NOT + // be treated as a readable log file. + g := newTestGear() + g.stat = func(string) (os.FileInfo, error) { + return fakeFileInfo{name: "x", mode: os.ModeDir}, nil + } + if g.isReadable("/anywhere") { + t.Error("isReadable should return false for directories") + } +} diff --git a/gearbox-agent/internal/gears/apache/collector.go b/gearbox-agent/internal/gears/apache/collector.go new file mode 100644 index 0000000..096e2f0 --- /dev/null +++ b/gearbox-agent/internal/gears/apache/collector.go @@ -0,0 +1,244 @@ +package apache + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/events" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" +) + +// Stats is the normalised view of one mod_status?auto scrape. Field +// names mirror what Apache prints in its key-value format so a +// reader can map them back to the upstream surface without guessing. +// +// All counters are monotonic since httpd start; the dashboard +// derives rates by diffing successive samples. Reading per-request +// "best" / "worst" timing would require parsing the Scoreboard +// character-by-character, which we intentionally don't do here — +// that's a follow-up enhancement when the dashboard actually needs +// it. +type Stats struct { + // TotalAccesses is the total number of requests served since + // httpd start. + TotalAccesses int64 `json:"total_accesses"` + + // TotalKBytes is the total response body bytes (in KB) served + // since httpd start. + TotalKBytes int64 `json:"total_kbytes"` + + // UptimeSeconds is how long the parent httpd has been running. + UptimeSeconds int64 `json:"uptime_seconds"` + + // ReqPerSec is Apache's own short-window average. We keep it + // even though the dashboard prefers diff-based rates, because + // it's the first value visible before two samples have + // accumulated. + ReqPerSec float64 `json:"req_per_sec"` + + // BytesPerSec / BytesPerReq from mod_status. Same caveat as + // ReqPerSec — Apache's averages. + BytesPerSec float64 `json:"bytes_per_sec"` + BytesPerReq float64 `json:"bytes_per_req"` + + // Worker pool. BusyWorkers + IdleWorkers ≈ ServerLimit on a + // healthy host; deviations signal config drift. + BusyWorkers int `json:"busy_workers"` + IdleWorkers int `json:"idle_workers"` + + // CPULoad is Apache's view of recent CPU consumption — useful + // alongside the host CPU% gauge for "is httpd the culprit?". + CPULoad float64 `json:"cpu_load"` + + // CollectedAt is when the agent scraped, in RFC3339. + CollectedAt string `json:"collected_at"` +} + +// ParseModStatus extracts a Stats struct from mod_status?auto's +// key-value body. Exported for unit testing without a live Apache. +// Unknown keys are silently skipped — Apache versions add new lines +// over time and the parser must tolerate that. +func ParseModStatus(body string) (Stats, error) { + if strings.TrimSpace(body) == "" { + return Stats{}, errors.New("empty mod_status body") + } + stats := Stats{CollectedAt: time.Now().UTC().Format(time.RFC3339)} + sawAnyField := false + + for _, line := range strings.Split(body, "\n") { + key, value, ok := strings.Cut(line, ":") + if !ok { + continue + } + key = strings.TrimSpace(key) + value = strings.TrimSpace(value) + if value == "" { + continue + } + sawAnyField = true + switch key { + case "Total Accesses": + stats.TotalAccesses, _ = strconv.ParseInt(value, 10, 64) + case "Total kBytes": + stats.TotalKBytes, _ = strconv.ParseInt(value, 10, 64) + case "Uptime": + stats.UptimeSeconds, _ = strconv.ParseInt(value, 10, 64) + case "ReqPerSec": + stats.ReqPerSec, _ = strconv.ParseFloat(value, 64) + case "BytesPerSec": + stats.BytesPerSec, _ = strconv.ParseFloat(value, 64) + case "BytesPerReq": + stats.BytesPerReq, _ = strconv.ParseFloat(value, 64) + case "BusyWorkers": + stats.BusyWorkers, _ = strconv.Atoi(value) + case "IdleWorkers": + stats.IdleWorkers, _ = strconv.Atoi(value) + case "CPULoad": + stats.CPULoad, _ = strconv.ParseFloat(value, 64) + } + } + if !sawAnyField { + // Body had no parseable key:value lines — usually means + // mod_status returned HTML (?auto wasn't honoured) or some + // other surface answered instead of mod_status. + return Stats{}, errors.New("mod_status body had no recognisable key:value pairs (is ?auto enabled?)") + } + return stats, nil +} + +// Initialize captures collector-time state (status URL, event bus, +// scrape interval) once the gear has been probed Available. +func (g *Gear) Initialize(ctx context.Context, deps gear.Dependencies) error { + if err := g.BaseGear.Initialize(ctx, deps); err != nil { + return err + } + g.statusURL = pickStatusURL(deps) + g.eventBus = deps.EventBus + g.statsInterval = deps.StatsInterval + return nil +} + +// pickStatusURL: operator override beats the default. +func pickStatusURL(deps gear.Dependencies) string { + if deps.ApacheStatusURL != "" { + return deps.ApacheStatusURL + } + return defaultStatusURL +} + +// Collectors registers the periodic mod_status scrape. +func (g *Gear) Collectors() []gear.Collector { + if g.statusURL == "" { + return nil + } + return []gear.Collector{ + { + Name: "apache-mod-status", + Interval: g.statsInterval, + Collect: g.scrape, + OnData: g.publish, + }, + } +} + +// EventTypes documents what this gear publishes. +func (g *Gear) EventTypes() []gear.EventType { + return []gear.EventType{ + { + Name: "apache.stats.updated", + Description: "Published when Apache mod_status is scraped", + Payload: "apache.Stats — total accesses, worker pool, uptime, etc.", + }, + } +} + +// scrape fetches mod_status?auto, parses it, caches the latest snapshot. +func (g *Gear) scrape(ctx context.Context) (any, error) { + res, err := g.httpGet(ctx, g.statusURL) + if err != nil { + return nil, fmt.Errorf("apache mod_status fetch: %w", err) + } + if res.StatusCode != http.StatusOK { + return nil, fmt.Errorf("apache mod_status returned HTTP %d", res.StatusCode) + } + stats, err := ParseModStatus(res.Body) + if err != nil { + return nil, fmt.Errorf("apache mod_status parse: %w", err) + } + g.cacheStats(stats) + return stats, nil +} + +// publish broadcasts the new stats to the event bus. +func (g *Gear) publish(data any) error { + if g.eventBus == nil { + return nil + } + stats, ok := data.(Stats) + if !ok { + return nil + } + g.eventBus.Publish(events.Event{ + Type: events.EventType("apache.stats.updated"), + Timestamp: time.Now(), + Data: map[string]any{ + "collected_at": stats.CollectedAt, + "stats": stats, + }, + }) + return nil +} + +func (g *Gear) cacheStats(s Stats) { + g.statsMu.Lock() + g.lastStats = &s + g.statsMu.Unlock() +} + +func (g *Gear) readCachedStats() (Stats, bool) { + g.statsMu.RLock() + defer g.statsMu.RUnlock() + if g.lastStats == nil { + return Stats{}, false + } + return *g.lastStats, true +} + +// RegisterRoutes registers the apache HTTP endpoints. +func (g *Gear) RegisterRoutes(r chi.Router) { + r.Get("/api/v1/apache/stats", g.handleStats) +} + +// handleStats returns the most recent mod_status snapshot. The +// `force` query parameter triggers a synchronous scrape. +func (g *Gear) handleStats(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("force") == "true" { + stats, err := g.scrape(r.Context()) + if err != nil { + http.Error(w, "apache scrape failed: "+err.Error(), http.StatusBadGateway) + return + } + writeJSON(w, stats) + return + } + stats, ok := g.readCachedStats() + if !ok { + http.Error(w, "apache stats not yet collected — agent just started, or mod_status is not reachable", http.StatusServiceUnavailable) + return + } + writeJSON(w, stats) +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +var _ gear.CollectorGear = (*Gear)(nil) diff --git a/gearbox-agent/internal/gears/apache/collector_test.go b/gearbox-agent/internal/gears/apache/collector_test.go new file mode 100644 index 0000000..c05934c --- /dev/null +++ b/gearbox-agent/internal/gears/apache/collector_test.go @@ -0,0 +1,160 @@ +package apache + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" + "github.com/sarg3nt/gearbox-agent/internal/framework/probe" +) + +const sampleModStatusBody = `Total Accesses: 1234 +Total kBytes: 5678 +CPULoad: .5 +Uptime: 1000 +ReqPerSec: 1.234 +BytesPerSec: 5678.0 +BytesPerReq: 100.5 +BusyWorkers: 5 +IdleWorkers: 95 +Scoreboard: ____W___K___ +` + +func TestParseModStatus(t *testing.T) { + got, err := ParseModStatus(sampleModStatusBody) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + got.CollectedAt = "" + want := Stats{ + TotalAccesses: 1234, + TotalKBytes: 5678, + UptimeSeconds: 1000, + ReqPerSec: 1.234, + BytesPerSec: 5678.0, + BytesPerReq: 100.5, + BusyWorkers: 5, + IdleWorkers: 95, + CPULoad: 0.5, + } + if got != want { + t.Errorf("Stats = %+v, want %+v", got, want) + } +} + +func TestParseModStatusRejectsEmptyOrHTML(t *testing.T) { + for _, raw := range []string{ + "", + " \n \n ", + // HTML-shaped — no key:value pairs the parser recognises. + "

Apache Status

", + } { + if _, err := ParseModStatus(raw); err == nil { + t.Errorf("expected error for %q", raw) + } + } +} + +func TestParseModStatusToleratesUnknownKeys(t *testing.T) { + // Newer Apache builds add lines like ConnsTotal / Load1 — they + // must not break the parser, just get ignored. + body := "Total Accesses: 5\nTotal kBytes: 10\nLoad1: 0.7\nConnsTotal: 42\n" + got, err := ParseModStatus(body) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.TotalAccesses != 5 || got.TotalKBytes != 10 { + t.Errorf("Stats = %+v, want Total Accesses=5 / kBytes=10", got) + } +} + +func gearWithHTTPStub(get func(ctx context.Context, url string) (probe.HTTPResult, error)) *Gear { + g := New() + g.httpGet = get + g.statusURL = defaultStatusURL + return g +} + +func TestCollectorScrapeCachesAndReturnsStats(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusOK, Body: sampleModStatusBody}, nil + }) + data, err := g.scrape(context.Background()) + if err != nil { + t.Fatalf("scrape: %v", err) + } + stats, ok := data.(Stats) + if !ok || stats.TotalAccesses != 1234 { + t.Errorf("scrape data = %+v, want Stats with TotalAccesses=1234", data) + } + cached, hit := g.readCachedStats() + if !hit || cached.TotalAccesses != 1234 { + t.Errorf("cache after scrape = %+v hit=%v", cached, hit) + } +} + +func TestCollectorScrapeReturnsErrorOnNon200(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusForbidden}, nil + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to error on 403") + } +} + +func TestCollectorScrapeReturnsErrorOnTransportFailure(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{}, errors.New("connection refused") + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to surface transport errors") + } +} + +func TestHandleStatsReturns503BeforeFirstScrape(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{}, errors.New("unused") + }) + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/apache/stats", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + if rr.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503 before first scrape", rr.Code) + } +} + +func TestHandleStatsReturnsCachedJSON(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusOK, Body: sampleModStatusBody}, nil + }) + if _, err := g.scrape(context.Background()); err != nil { + t.Fatalf("scrape: %v", err) + } + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/apache/stats", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Errorf("status = %d, want 200", rr.Code) + } + if !strings.Contains(rr.Body.String(), `"total_accesses":1234`) { + t.Errorf("body missing total_accesses=1234, got %s", rr.Body.String()) + } +} + +func TestPickStatusURLHonorsOverride(t *testing.T) { + if got := pickStatusURL(gear.Dependencies{ApacheStatusURL: "http://1.2.3.4/_status"}); got != "http://1.2.3.4/_status" { + t.Errorf("override = %q, want override URL", got) + } + if got := pickStatusURL(gear.Dependencies{}); got != defaultStatusURL { + t.Errorf("default = %q, want %q", got, defaultStatusURL) + } +} diff --git a/gearbox-agent/internal/gears/apache/plugin.go b/gearbox-agent/internal/gears/apache/plugin.go index e5cbcf2..202d907 100644 --- a/gearbox-agent/internal/gears/apache/plugin.go +++ b/gearbox-agent/internal/gears/apache/plugin.go @@ -1,7 +1,6 @@ // Package apache detects an Apache HTTP Server installation on the host -// and reports it in the capability manifest. Phase 3 only — no metrics -// collection yet; that lands in Phase 7 (separate issue, #95 followup) -// and will read the mod_status surface this gear already verified. +// and periodically scrapes its mod_status?auto surface so the +// dashboard's Metrics gear can render Apache alongside HAProxy. // // The detector declares CategoryHTTPRequests so the agent's // primary-source resolver (see [gear.ResolvePrimarySources]) considers @@ -19,8 +18,10 @@ import ( "os/exec" "regexp" "strings" + "sync" + "time" - "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/events" "github.com/sarg3nt/gearbox-agent/internal/framework/gear" "github.com/sarg3nt/gearbox-agent/internal/framework/probe" ) @@ -68,7 +69,7 @@ var serverConfigFileRegex = regexp.MustCompile(`SERVER_CONFIG_FILE="([^"]+)"`) // respond no matter how the config is set up. const statusModuleSentinel = "status_module" -// Gear is the Apache detector. +// Gear is the Apache detector + mod_status metrics collector. type Gear struct { gear.BaseGear @@ -79,6 +80,17 @@ type Gear struct { runM func(ctx context.Context, binary string) ([]byte, error) // -M stat func(string) (os.FileInfo, error) httpGet func(ctx context.Context, url string) (probe.HTTPResult, error) + + // Collector-time state. Populated by Initialize once Probe + // returned Available. statsMu guards the lastStats pointer so + // concurrent reads from the API handler and writes from the + // collector goroutine don't race. + statusURL string + eventBus *events.Bus + statsInterval time.Duration + + statsMu sync.RWMutex + lastStats *Stats } // New constructs an Apache gear with real OS-backed defaults. @@ -103,7 +115,7 @@ func (g *Gear) Info() gear.Info { return gear.Info{ Name: "apache", DisplayName: "Apache HTTP Server", - Description: "Detects an Apache installation and verifies its mod_status surface so the Metrics gear can consume it in Phase 7+.", + Description: "Detects an Apache installation, verifies its mod_status surface, and periodically scrapes it for the Metrics gear.", Version: "1.0.0", Category: "monitoring", } @@ -256,9 +268,8 @@ func resolveConfigPath(override, vOut string, stat func(string) (os.FileInfo, er return "" } -// RegisterRoutes is a no-op — Phase 3 Apache detection produces no API -// surface. The metrics gear that reads mod_status lands in Phase 7+. -func (g *Gear) RegisterRoutes(_ chi.Router) {} +// RegisterRoutes and the periodic-scrape machinery live in +// collector.go alongside the rest of the metrics code. // Ensure the gear implements the required interfaces. var ( diff --git a/gearbox-agent/internal/gears/caddy/collector.go b/gearbox-agent/internal/gears/caddy/collector.go new file mode 100644 index 0000000..470d400 --- /dev/null +++ b/gearbox-agent/internal/gears/caddy/collector.go @@ -0,0 +1,193 @@ +package caddy + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "time" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/events" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" + "github.com/sarg3nt/gearbox-agent/internal/framework/services/promtext" +) + +// Stats is the normalised view of one Prometheus scrape. Caddy +// emits a rich Prometheus surface; we surface the rollups the +// Metrics gear cares about and leave the histogram detail for a +// future enhancement when the dashboard renders it. +// +// All counters are monotonic since Caddy start; the dashboard +// derives rates by diffing successive samples. The status-class +// totals are derived from caddy_http_request_errors_total + the +// inverse for 2xx (Caddy doesn't emit a per-code counter by default +// — `request_errors_total` is the supported error signal; future +// caddy versions or operator-extended metrics may add per-code +// granularity). +type Stats struct { + // RequestsTotal sums caddy_http_requests_total across all + // server / handler labels. + RequestsTotal int64 `json:"requests_total"` + + // RequestErrorsTotal sums caddy_http_request_errors_total — + // requests that failed before producing a response (5xx-class + // from Caddy's perspective). Distinct from response_5xx in + // other proxies; documented as such to avoid the dashboard + // double-counting. + RequestErrorsTotal int64 `json:"request_errors_total"` + + // AdminRunning is 1 when Caddy's admin endpoint reports itself + // alive, 0 otherwise. Lets the dashboard surface "Caddy + // running but admin disconnected" without a second probe. + AdminRunning bool `json:"admin_running"` + + // CollectedAt is when the agent scraped, in RFC3339. + CollectedAt string `json:"collected_at"` +} + +// ParsePrometheusOutput extracts a Stats struct from raw Caddy +// `:2019/metrics` Prometheus output. Exported for unit testing. +func ParsePrometheusOutput(body string) Stats { + samples := promtext.Parse(body) + stats := Stats{CollectedAt: time.Now().UTC().Format(time.RFC3339)} + + stats.RequestsTotal = int64(promtext.SumByName(samples, "caddy_http_requests_total")) + stats.RequestErrorsTotal = int64(promtext.SumByName(samples, "caddy_http_request_errors_total")) + + if s := promtext.FirstByName(samples, "caddy_admin_http_requests_total"); s != nil { + // admin_http_requests_total existing at all means the + // admin endpoint is up — the metric is only registered + // when admin is enabled. + stats.AdminRunning = true + } + return stats +} + +// Initialize captures collector-time state. +func (g *Gear) Initialize(ctx context.Context, deps gear.Dependencies) error { + if err := g.BaseGear.Initialize(ctx, deps); err != nil { + return err + } + g.adminURL = pickAdminURL(deps) + g.eventBus = deps.EventBus + g.statsInterval = deps.StatsInterval + return nil +} + +// pickAdminURL: operator override beats the default. +func pickAdminURL(deps gear.Dependencies) string { + if deps.CaddyAdminURL != "" { + return deps.CaddyAdminURL + } + return defaultAdminURL +} + +// Collectors registers the periodic Prometheus scrape. +func (g *Gear) Collectors() []gear.Collector { + if g.adminURL == "" { + return nil + } + return []gear.Collector{ + { + Name: "caddy-prometheus", + Interval: g.statsInterval, + Collect: g.scrape, + OnData: g.publish, + }, + } +} + +// EventTypes documents what this gear publishes. +func (g *Gear) EventTypes() []gear.EventType { + return []gear.EventType{ + { + Name: "caddy.stats.updated", + Description: "Published when Caddy Prometheus metrics are scraped", + Payload: "caddy.Stats — requests + error counters, admin status", + }, + } +} + +// scrape fetches the Prometheus surface and normalises the result. +// Unlike nginx/Apache, ParsePrometheusOutput cannot fail — a +// Prometheus surface that returns 200 with garbage simply yields +// zero counters, so the dashboard sees "Caddy is up but emitted +// nothing" rather than an error. +func (g *Gear) scrape(ctx context.Context) (any, error) { + res, err := g.httpGet(ctx, g.adminURL) + if err != nil { + return nil, fmt.Errorf("caddy /metrics fetch: %w", err) + } + if res.StatusCode != http.StatusOK { + return nil, fmt.Errorf("caddy /metrics returned HTTP %d", res.StatusCode) + } + stats := ParsePrometheusOutput(res.Body) + g.cacheStats(stats) + return stats, nil +} + +// publish broadcasts stats updates to the event bus. +func (g *Gear) publish(data any) error { + if g.eventBus == nil { + return nil + } + stats, ok := data.(Stats) + if !ok { + return nil + } + g.eventBus.Publish(events.Event{ + Type: events.EventType("caddy.stats.updated"), + Timestamp: time.Now(), + Data: map[string]any{ + "collected_at": stats.CollectedAt, + "stats": stats, + }, + }) + return nil +} + +func (g *Gear) cacheStats(s Stats) { + g.statsMu.Lock() + g.lastStats = &s + g.statsMu.Unlock() +} + +func (g *Gear) readCachedStats() (Stats, bool) { + g.statsMu.RLock() + defer g.statsMu.RUnlock() + if g.lastStats == nil { + return Stats{}, false + } + return *g.lastStats, true +} + +// RegisterRoutes registers the caddy HTTP endpoint. +func (g *Gear) RegisterRoutes(r chi.Router) { + r.Get("/api/v1/caddy/stats", g.handleStats) +} + +func (g *Gear) handleStats(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("force") == "true" { + stats, err := g.scrape(r.Context()) + if err != nil { + http.Error(w, "caddy scrape failed: "+err.Error(), http.StatusBadGateway) + return + } + writeJSON(w, stats) + return + } + stats, ok := g.readCachedStats() + if !ok { + http.Error(w, "caddy stats not yet collected — agent just started, or admin endpoint is not reachable", http.StatusServiceUnavailable) + return + } + writeJSON(w, stats) +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +var _ gear.CollectorGear = (*Gear)(nil) diff --git a/gearbox-agent/internal/gears/caddy/collector_test.go b/gearbox-agent/internal/gears/caddy/collector_test.go new file mode 100644 index 0000000..13ee0c9 --- /dev/null +++ b/gearbox-agent/internal/gears/caddy/collector_test.go @@ -0,0 +1,139 @@ +package caddy + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" + "github.com/sarg3nt/gearbox-agent/internal/framework/probe" +) + +const samplePrometheusBody = `# HELP caddy_http_requests_total Counter of HTTP requests +# TYPE caddy_http_requests_total counter +caddy_http_requests_total{server="srv0",handler="reverse_proxy"} 42 +caddy_http_requests_total{server="srv0",handler="file_server"} 17 +# HELP caddy_http_request_errors_total Error counter +# TYPE caddy_http_request_errors_total counter +caddy_http_request_errors_total{server="srv0"} 3 +# admin endpoint counter — its mere existence signals admin is up +caddy_admin_http_requests_total{path="/load"} 5 +` + +func TestParsePrometheusOutputSumsCounters(t *testing.T) { + got := ParsePrometheusOutput(samplePrometheusBody) + got.CollectedAt = "" + want := Stats{ + RequestsTotal: 59, // 42 + 17 + RequestErrorsTotal: 3, + AdminRunning: true, + } + if got != want { + t.Errorf("Stats = %+v, want %+v", got, want) + } +} + +func TestParsePrometheusOutputHandlesMissingAdminMetric(t *testing.T) { + // Admin endpoint disabled (admin off in Caddyfile) — the + // `caddy_admin_*` metric won't appear; AdminRunning must be + // false even though the rest of the scrape is fine. + body := `caddy_http_requests_total{server="srv0"} 7 +` + got := ParsePrometheusOutput(body) + if got.AdminRunning { + t.Error("AdminRunning should be false when admin metric is absent") + } + if got.RequestsTotal != 7 { + t.Errorf("RequestsTotal = %d, want 7", got.RequestsTotal) + } +} + +func gearWithHTTPStub(get func(ctx context.Context, url string) (probe.HTTPResult, error)) *Gear { + g := New() + g.httpGet = get + g.adminURL = defaultAdminURL + return g +} + +func TestCollectorScrapeCachesAndReturnsStats(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusOK, Body: samplePrometheusBody}, nil + }) + data, err := g.scrape(context.Background()) + if err != nil { + t.Fatalf("scrape: %v", err) + } + stats, ok := data.(Stats) + if !ok || stats.RequestsTotal != 59 { + t.Errorf("scrape data = %+v, want RequestsTotal=59", data) + } + cached, hit := g.readCachedStats() + if !hit || cached.RequestsTotal != 59 { + t.Errorf("cache after scrape = %+v hit=%v", cached, hit) + } +} + +func TestCollectorScrapeReturnsErrorOnNon200(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusNotFound}, nil + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to error on 404") + } +} + +func TestCollectorScrapeReturnsErrorOnTransportFailure(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{}, errors.New("connection refused") + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to surface transport errors") + } +} + +func TestHandleStatsReturns503BeforeFirstScrape(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{}, errors.New("unused") + }) + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/caddy/stats", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + if rr.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503", rr.Code) + } +} + +func TestHandleStatsReturnsCachedJSON(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusOK, Body: samplePrometheusBody}, nil + }) + if _, err := g.scrape(context.Background()); err != nil { + t.Fatalf("scrape: %v", err) + } + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/caddy/stats", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Errorf("status = %d, want 200", rr.Code) + } + if !strings.Contains(rr.Body.String(), `"requests_total":59`) { + t.Errorf("body missing requests_total=59, got %s", rr.Body.String()) + } +} + +func TestPickAdminURLHonorsOverride(t *testing.T) { + if got := pickAdminURL(gear.Dependencies{CaddyAdminURL: "http://1.2.3.4/_metrics"}); got != "http://1.2.3.4/_metrics" { + t.Errorf("override = %q", got) + } + if got := pickAdminURL(gear.Dependencies{}); got != defaultAdminURL { + t.Errorf("default = %q", got) + } +} diff --git a/gearbox-agent/internal/gears/caddy/plugin.go b/gearbox-agent/internal/gears/caddy/plugin.go index f426374..2a03249 100644 --- a/gearbox-agent/internal/gears/caddy/plugin.go +++ b/gearbox-agent/internal/gears/caddy/plugin.go @@ -22,8 +22,10 @@ import ( "os/exec" "regexp" "strings" + "sync" + "time" - "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/events" "github.com/sarg3nt/gearbox-agent/internal/framework/gear" "github.com/sarg3nt/gearbox-agent/internal/framework/probe" ) @@ -50,7 +52,7 @@ const prometheusSentinel = "caddy_http_requests_total" // Tolerant of the build hash suffix; we only care about the semver. var versionRegex = regexp.MustCompile(`v(\d+\.\d+\.\d+)`) -// Gear is the Caddy detector. +// Gear is the Caddy detector + Prometheus metrics collector. type Gear struct { gear.BaseGear @@ -59,6 +61,15 @@ type Gear struct { lookPath func(string) (string, error) runVersion func(ctx context.Context) ([]byte, error) httpGet func(ctx context.Context, url string) (probe.HTTPResult, error) + + // Collector-time state. Populated by Initialize once Probe + // returned Available. + adminURL string + eventBus *events.Bus + statsInterval time.Duration + + statsMu sync.RWMutex + lastStats *Stats } // New constructs a Caddy gear with real OS-backed defaults. @@ -79,7 +90,7 @@ func (g *Gear) Info() gear.Info { return gear.Info{ Name: "caddy", DisplayName: "Caddy", - Description: "Detects a Caddy installation and verifies its admin/metrics endpoint so the Metrics gear can scrape it in Phase 7+.", + Description: "Detects a Caddy installation, verifies its admin/metrics endpoint, and periodically scrapes its Prometheus output for the Metrics gear.", Version: "1.0.0", Category: "monitoring", } @@ -170,10 +181,8 @@ func (g *Gear) recordBinaryFacts(ctx context.Context, caps map[string]string) { } } -// RegisterRoutes is a no-op — Phase 3 Caddy detection produces no API -// surface. The metrics gear that scrapes Caddy's Prometheus output -// lands in Phase 7+. -func (g *Gear) RegisterRoutes(_ chi.Router) {} +// RegisterRoutes and the periodic-scrape machinery live in +// collector.go alongside the rest of the metrics code. // Ensure the gear implements the required interfaces. var ( diff --git a/gearbox-agent/internal/gears/nginx/collector.go b/gearbox-agent/internal/gears/nginx/collector.go new file mode 100644 index 0000000..2b41e43 --- /dev/null +++ b/gearbox-agent/internal/gears/nginx/collector.go @@ -0,0 +1,257 @@ +package nginx + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "regexp" + "strconv" + "time" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/events" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" +) + +// Stats is the normalised view of one stub_status scrape. We keep +// the connection counters split out because the dashboard renders +// reading/writing/waiting as a stacked chart and needs them +// distinguishable. Rates (req/sec etc.) are derived on the +// dashboard side from successive samples — the agent's job is to +// emit honest counter values, not to invent rates between scrapes. +type Stats struct { + // Active is the count of currently active client connections, + // including waiting connections. + Active int `json:"active"` + + // Reading: connections where nginx is reading the request + // header. + Reading int `json:"reading"` + + // Writing: connections where nginx is writing the response + // back to the client. + Writing int `json:"writing"` + + // Waiting: idle keepalive connections waiting for the next + // request. On busy sites this is the bulk of `Active`. + Waiting int `json:"waiting"` + + // Total connection counters since nginx start. Monotonic; the + // dashboard subtracts successive samples to derive rates. + Accepts int64 `json:"accepts"` + Handled int64 `json:"handled"` + Requests int64 `json:"requests"` + + // CollectedAt is when the agent scraped stub_status, in RFC3339. + // Lets the dashboard compute "data age" without trusting wall- + // clock skew between the two hosts (it just compares to the + // previous sample's CollectedAt). + CollectedAt string `json:"collected_at"` +} + +// stubStatusBody is what nginx writes to the stub_status URL: +// +// Active connections: 291 +// server accepts handled requests +// 16630948 16630948 31070465 +// Reading: 6 Writing: 179 Waiting: 106 +// +// Each field has a stable regex; we parse defensively because some +// older nginx builds emit slight whitespace variations. +var ( + reActive = regexp.MustCompile(`Active connections:\s+(\d+)`) + reTotals = regexp.MustCompile(`(?m)^\s*(\d+)\s+(\d+)\s+(\d+)\s*$`) + reReadWait = regexp.MustCompile(`Reading:\s+(\d+)\s+Writing:\s+(\d+)\s+Waiting:\s+(\d+)`) +) + +// ParseStubStatus extracts a Stats struct from raw stub_status text. +// Exported so the collector can also be exercised by unit tests +// without needing a live nginx — the function is pure, no state. +func ParseStubStatus(body string) (Stats, error) { + stats := Stats{CollectedAt: time.Now().UTC().Format(time.RFC3339)} + if m := reActive.FindStringSubmatch(body); len(m) > 1 { + stats.Active, _ = strconv.Atoi(m[1]) + } else { + return Stats{}, errors.New("stub_status body missing 'Active connections' line") + } + if m := reTotals.FindStringSubmatch(body); len(m) > 3 { + stats.Accepts, _ = strconv.ParseInt(m[1], 10, 64) + stats.Handled, _ = strconv.ParseInt(m[2], 10, 64) + stats.Requests, _ = strconv.ParseInt(m[3], 10, 64) + } else { + return Stats{}, errors.New("stub_status body missing 'accepts handled requests' counter row") + } + if m := reReadWait.FindStringSubmatch(body); len(m) > 3 { + stats.Reading, _ = strconv.Atoi(m[1]) + stats.Writing, _ = strconv.Atoi(m[2]) + stats.Waiting, _ = strconv.Atoi(m[3]) + } + return stats, nil +} + +// Initialize stores deps + last-stats slot so the collector + API +// handler can both reach probe-time indirections (httpGet) and the +// status URL the detector picked. +func (g *Gear) Initialize(ctx context.Context, deps gear.Dependencies) error { + if err := g.BaseGear.Initialize(ctx, deps); err != nil { + return err + } + g.statusURL = pickStatusURL(deps) + g.eventBus = deps.EventBus + g.statsInterval = deps.StatsInterval + return nil +} + +// pickStatusURL returns the URL the collector should scrape: the +// operator override when set, otherwise the default stub_status URL. +// We don't re-read the probe result here because the resolution +// already happened during Probe(); collectors trust the probed +// surface and don't second-guess it. +func pickStatusURL(deps gear.Dependencies) string { + if deps.NginxStatusURL != "" { + return deps.NginxStatusURL + } + return defaultStatusURL +} + +// Collectors implements gear.CollectorGear. One periodic collector +// per gear: scrape stub_status, normalise into Stats, cache, publish. +// Interval matches the agent's standard stats interval (HAProxy uses +// the same). Returning an empty list when there's no status URL +// keeps the gear quiet on hosts where probe came back Inaccessible +// — the manager already skipped Initialize in that case, but the +// belt-and-suspenders check costs nothing. +func (g *Gear) Collectors() []gear.Collector { + if g.statusURL == "" { + return nil + } + return []gear.Collector{ + { + Name: "nginx-stub-status", + Interval: g.statsInterval, + Collect: g.scrape, + OnData: g.publish, + }, + } +} + +// EventTypes documents what this gear publishes for the WebSocket +// API consumers. Mirrors HAProxy's pattern. +func (g *Gear) EventTypes() []gear.EventType { + return []gear.EventType{ + { + Name: "nginx.stats.updated", + Description: "Published when nginx stub_status is scraped", + Payload: "nginx.Stats — active/reading/writing/waiting + accepts/handled/requests counters", + }, + } +} + +// scrape runs one stub_status fetch. Returns an error on transport +// failure; returns a parse error if the body doesn't look like +// stub_status (e.g. someone enabled the surface on a non-stub URL). +func (g *Gear) scrape(ctx context.Context) (any, error) { + res, err := g.httpGet(ctx, g.statusURL) + if err != nil { + return nil, fmt.Errorf("nginx stub_status fetch: %w", err) + } + if res.StatusCode != http.StatusOK { + return nil, fmt.Errorf("nginx stub_status returned HTTP %d", res.StatusCode) + } + stats, err := ParseStubStatus(res.Body) + if err != nil { + return nil, fmt.Errorf("nginx stub_status parse: %w", err) + } + g.cacheStats(stats) + return stats, nil +} + +// publish broadcasts the new stats to the agent event bus so +// WebSocket consumers can subscribe to live nginx data the same +// way they do for HAProxy. Returns nil even on type mismatch — +// publishing is best-effort and shouldn't fail collection. +func (g *Gear) publish(data any) error { + if g.eventBus == nil { + return nil + } + stats, ok := data.(Stats) + if !ok { + return nil + } + g.eventBus.Publish(events.Event{ + Type: events.EventType("nginx.stats.updated"), + Timestamp: time.Now(), + Data: map[string]any{ + "collected_at": stats.CollectedAt, + "stats": stats, + }, + }) + return nil +} + +// cacheStats stores the most recent stub_status sample so the +// synchronous /api/v1/nginx/stats handler can return data without +// triggering a fresh scrape on every request — keeping the dashboard +// snappy even when nginx is on a slow box. +func (g *Gear) cacheStats(s Stats) { + g.statsMu.Lock() + g.lastStats = &s + g.statsMu.Unlock() +} + +// readCachedStats returns the most recent sample if one exists. +// The bool conveys "has anything been collected yet?" so the handler +// can return a 503 with a clear reason rather than an empty Stats +// (which would look like "nginx has zero connections", a different +// bug). +func (g *Gear) readCachedStats() (Stats, bool) { + g.statsMu.RLock() + defer g.statsMu.RUnlock() + if g.lastStats == nil { + return Stats{}, false + } + return *g.lastStats, true +} + +// RegisterRoutes registers the nginx HTTP endpoints. Detection-only +// installations (Probe returned Available without any stats yet) +// still get the routes because the dashboard uses route presence as +// "this source is supported"; the handler distinguishes "no data +// yet" with a 503. +func (g *Gear) RegisterRoutes(r chi.Router) { + r.Get("/api/v1/nginx/stats", g.handleStats) +} + +// handleStats returns the most recent stub_status snapshot. The +// `force` query parameter triggers a synchronous scrape (useful for +// debugging and for the first dashboard load before the collector's +// first tick). +func (g *Gear) handleStats(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("force") == "true" { + stats, err := g.scrape(r.Context()) + if err != nil { + http.Error(w, "nginx scrape failed: "+err.Error(), http.StatusBadGateway) + return + } + writeJSON(w, stats) + return + } + stats, ok := g.readCachedStats() + if !ok { + http.Error(w, "nginx stats not yet collected — agent just started, or stub_status is not reachable", http.StatusServiceUnavailable) + return + } + writeJSON(w, stats) +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +// Ensure the gear implements CollectorGear at compile time so a +// future refactor that removes one of the methods fails the build +// rather than silently disabling collection. +var _ gear.CollectorGear = (*Gear)(nil) diff --git a/gearbox-agent/internal/gears/nginx/collector_test.go b/gearbox-agent/internal/gears/nginx/collector_test.go new file mode 100644 index 0000000..6976134 --- /dev/null +++ b/gearbox-agent/internal/gears/nginx/collector_test.go @@ -0,0 +1,184 @@ +package nginx + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" + "github.com/sarg3nt/gearbox-agent/internal/framework/probe" +) + +const sampleStubStatusBody = `Active connections: 291 +server accepts handled requests + 16630948 16630948 31070465 +Reading: 6 Writing: 179 Waiting: 106 +` + +func TestParseStubStatus(t *testing.T) { + got, err := ParseStubStatus(sampleStubStatusBody) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + want := Stats{ + Active: 291, + Reading: 6, + Writing: 179, + Waiting: 106, + Accepts: 16630948, + Handled: 16630948, + Requests: 31070465, + } + // CollectedAt is non-deterministic; drop before comparing. + got.CollectedAt = "" + if got != want { + t.Errorf("Stats = %+v, want %+v", got, want) + } +} + +func TestParseStubStatusRejectsNonStubBody(t *testing.T) { + cases := []string{ + "", + "random garbage", + "Active connections: nope", + // Missing the 'accepts handled requests' counter row. + "Active connections: 5\nReading: 1 Writing: 2 Waiting: 3\n", + } + for _, raw := range cases { + if _, err := ParseStubStatus(raw); err == nil { + t.Errorf("expected parse error for %q", raw) + } + } +} + +// gearWithHTTPStub returns a nginx gear set up to mock the HTTP +// probe so collector tests don't need a live nginx. +func gearWithHTTPStub(get func(ctx context.Context, url string) (probe.HTTPResult, error)) *Gear { + g := New() + g.httpGet = get + g.statusURL = defaultStatusURL + return g +} + +func TestCollectorScrapeCachesAndReturnsStats(t *testing.T) { + calls := 0 + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + calls++ + return probe.HTTPResult{StatusCode: http.StatusOK, Body: sampleStubStatusBody}, nil + }) + + data, err := g.scrape(context.Background()) + if err != nil { + t.Fatalf("scrape: %v", err) + } + if calls != 1 { + t.Errorf("scrape should call httpGet once, got %d", calls) + } + stats, ok := data.(Stats) + if !ok || stats.Active != 291 { + t.Errorf("scrape returned %+v, want Stats{Active: 291, …}", data) + } + // Cache must hold the same stats so the synchronous handler can + // return without a fresh scrape. + cached, hit := g.readCachedStats() + if !hit || cached.Active != 291 { + t.Errorf("cache after scrape = %+v hit=%v, want Active=291", cached, hit) + } +} + +func TestCollectorScrapeReturnsErrorOnNon200(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusForbidden}, nil + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to error on 403 response") + } +} + +func TestCollectorScrapeReturnsErrorOnTransportFailure(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{}, errors.New("connection refused") + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to surface transport errors") + } +} + +func TestHandleStatsReturns503BeforeFirstScrape(t *testing.T) { + // The dashboard distinguishes "no data yet" from "zero active + // connections" by HTTP status code; the handler must return 503 + // not an empty 200 when nothing has been cached yet. + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{}, errors.New("unused") + }) + r := chi.NewRouter() + g.RegisterRoutes(r) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/nginx/stats", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + if rr.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503", rr.Code) + } +} + +func TestHandleStatsReturnsCachedJSON(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusOK, Body: sampleStubStatusBody}, nil + }) + // Prime the cache. + if _, err := g.scrape(context.Background()); err != nil { + t.Fatalf("scrape: %v", err) + } + + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/nginx/stats", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rr.Code) + } + if got := rr.Header().Get("Content-Type"); got != "application/json" { + t.Errorf("content-type = %q", got) + } + if !strings.Contains(rr.Body.String(), `"active":291`) { + t.Errorf("body should contain JSON-encoded Active=291, got %s", rr.Body.String()) + } +} + +func TestHandleStatsForceTriggersFreshScrape(t *testing.T) { + calls := 0 + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + calls++ + return probe.HTTPResult{StatusCode: http.StatusOK, Body: sampleStubStatusBody}, nil + }) + + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/nginx/stats?force=true", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Errorf("status = %d, want 200", rr.Code) + } + if calls != 1 { + t.Errorf("force=true should call httpGet once, got %d", calls) + } +} + +func TestPickStatusURLHonorsOverride(t *testing.T) { + if got := pickStatusURL(gear.Dependencies{NginxStatusURL: "http://1.2.3.4/_status"}); got != "http://1.2.3.4/_status" { + t.Errorf("pickStatusURL with override = %q, want override", got) + } + if got := pickStatusURL(gear.Dependencies{}); got != defaultStatusURL { + t.Errorf("pickStatusURL without override = %q, want default", got) + } +} diff --git a/gearbox-agent/internal/gears/nginx/plugin.go b/gearbox-agent/internal/gears/nginx/plugin.go index e024c74..72010f4 100644 --- a/gearbox-agent/internal/gears/nginx/plugin.go +++ b/gearbox-agent/internal/gears/nginx/plugin.go @@ -1,7 +1,6 @@ -// Package nginx detects an nginx installation on the host and reports -// it in the capability manifest. Phase 3 only — no metrics collection -// yet; that lands in Phase 4 (separate issue, #95 followup) and will -// read the `stub_status` endpoint surface this gear already verified. +// Package nginx detects an nginx installation on the host and +// periodically scrapes its `stub_status` surface so the dashboard's +// Metrics gear can render nginx alongside HAProxy. // // The detector declares CategoryHTTPRequests so the agent's // primary-source resolver (see [gear.ResolvePrimarySources]) considers @@ -17,8 +16,10 @@ import ( "os/exec" "regexp" "strings" + "sync" + "time" - "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/events" "github.com/sarg3nt/gearbox-agent/internal/framework/gear" "github.com/sarg3nt/gearbox-agent/internal/framework/probe" ) @@ -65,7 +66,7 @@ var confPathRegex = regexp.MustCompile(`--conf-path=([^\s]+)`) // over stub_status when available; Phase 3 just records the fact. const apiModuleSentinel = "--with-http_api_module" -// Gear is the nginx detector. +// Gear is the nginx detector + stub_status metrics collector. type Gear struct { gear.BaseGear @@ -76,6 +77,17 @@ type Gear struct { runLongV func(ctx context.Context) ([]byte, error) // nginx -V stat func(string) (os.FileInfo, error) httpGet func(ctx context.Context, url string) (probe.HTTPResult, error) + + // Collector-time state, populated by Initialize() once the + // gear has been probed Available. Kept on the same struct so + // the API handler can read the most recent sample without a + // global; statsMu guards reads + writes of lastStats. + statusURL string + eventBus *events.Bus + statsInterval time.Duration + + statsMu sync.RWMutex + lastStats *Stats } // New constructs an nginx gear with real OS-backed defaults. @@ -101,7 +113,7 @@ func (g *Gear) Info() gear.Info { return gear.Info{ Name: "nginx", DisplayName: "nginx", - Description: "Detects an nginx installation and verifies its stub_status surface so the Metrics gear can consume it in Phase 4+.", + Description: "Detects an nginx installation, verifies its stub_status surface, and periodically scrapes it for the Metrics gear.", Version: "1.0.0", Category: "monitoring", } @@ -252,9 +264,8 @@ func resolveConfigPath(override, buildInfo string, stat func(string) (os.FileInf return "" } -// RegisterRoutes is a no-op — Phase 3 nginx detection produces no API -// surface. The metrics gear that reads stub_status lands in Phase 4+. -func (g *Gear) RegisterRoutes(_ chi.Router) {} +// RegisterRoutes and the Collectors+publish machinery live in +// collector.go alongside the rest of the periodic-scrape code. // Ensure the gear implements the required interfaces. var ( diff --git a/gearbox-agent/internal/gears/traefik/collector.go b/gearbox-agent/internal/gears/traefik/collector.go new file mode 100644 index 0000000..326c267 --- /dev/null +++ b/gearbox-agent/internal/gears/traefik/collector.go @@ -0,0 +1,244 @@ +package traefik + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/events" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" + "github.com/sarg3nt/gearbox-agent/internal/framework/services/promtext" +) + +// Stats is the normalised view of one Traefik Prometheus scrape. +// Traefik labels its router/service counters by HTTP status code, +// which lets us emit a real per-class breakdown — unlike Caddy, +// whose default surface lacks per-code granularity. +// +// All counters are monotonic since Traefik start; the dashboard +// derives rates by diffing successive samples. +type Stats struct { + // RequestsTotal sums traefik_router_requests_total across all + // labels — the all-traffic counter. + RequestsTotal int64 `json:"requests_total"` + + // Response counts split by HTTP status-code first digit. The + // dashboard uses these for the "5xx %" KPI and the per-class + // stacked chart. + Response1xx int64 `json:"response_1xx"` + Response2xx int64 `json:"response_2xx"` + Response3xx int64 `json:"response_3xx"` + Response4xx int64 `json:"response_4xx"` + Response5xx int64 `json:"response_5xx"` + + // EntryPoints lists the entrypoints Traefik exposed metrics + // for. Stable order (alphabetised) so the dashboard renders + // the same list across polls. + EntryPoints []string `json:"entrypoints"` + + // CollectedAt is when the agent scraped, in RFC3339. + CollectedAt string `json:"collected_at"` +} + +// ParsePrometheusOutput extracts a Stats from Traefik's +// `:8082/metrics` Prometheus output. Exported for unit testing. +func ParsePrometheusOutput(body string) Stats { + samples := promtext.Parse(body) + stats := Stats{CollectedAt: time.Now().UTC().Format(time.RFC3339)} + + stats.RequestsTotal = int64(promtext.SumByName(samples, "traefik_router_requests_total")) + + // Per-status-class breakdown: walk every router_requests_total + // sample once, bucket by first digit of the `code` label. + entryPointSet := make(map[string]struct{}) + for _, s := range samples { + switch s.Name { + case "traefik_router_requests_total": + bucketByStatusClass(&stats, s.Labels["code"], s.Value) + case "traefik_entrypoint_requests_total": + if ep := s.Labels["entrypoint"]; ep != "" { + entryPointSet[ep] = struct{}{} + } + } + } + + stats.EntryPoints = sortedKeys(entryPointSet) + return stats +} + +// bucketByStatusClass adds value to the right Response{1..5}xx +// counter on stats. Unknown / empty codes are ignored so a metric +// without the `code` label doesn't quietly land in the wrong bucket. +func bucketByStatusClass(stats *Stats, code string, value float64) { + if len(code) == 0 { + return + } + v := int64(value) + switch code[0] { + case '1': + stats.Response1xx += v + case '2': + stats.Response2xx += v + case '3': + stats.Response3xx += v + case '4': + stats.Response4xx += v + case '5': + stats.Response5xx += v + } +} + +// sortedKeys returns the map's keys in alphabetical order — a +// helper for stable JSON output. +func sortedKeys(m map[string]struct{}) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + // Tiny enough that a manual sort beats pulling in `sort`'s + // extra closure overhead — usually 2-4 entrypoints. + for i := 1; i < len(out); i++ { + for j := i; j > 0 && out[j-1] > out[j]; j-- { + out[j-1], out[j] = out[j], out[j-1] + } + } + return out +} + +// Initialize captures collector-time state. +func (g *Gear) Initialize(ctx context.Context, deps gear.Dependencies) error { + if err := g.BaseGear.Initialize(ctx, deps); err != nil { + return err + } + g.metricsURL = pickMetricsURL(deps) + g.eventBus = deps.EventBus + g.statsInterval = deps.StatsInterval + return nil +} + +// pickMetricsURL: operator override > first default URL. Unlike the +// probe phase (which tries both default URLs in order), the +// collector commits to one — the operator's override or the first +// default. Switching URLs at collection time would invalidate the +// counters mid-stream from the dashboard's perspective. +func pickMetricsURL(deps gear.Dependencies) string { + if deps.TraefikMetricsURL != "" { + return deps.TraefikMetricsURL + } + return defaultMetricsURLs[0] +} + +// Collectors registers the periodic Prometheus scrape. +func (g *Gear) Collectors() []gear.Collector { + if g.metricsURL == "" { + return nil + } + return []gear.Collector{ + { + Name: "traefik-prometheus", + Interval: g.statsInterval, + Collect: g.scrape, + OnData: g.publish, + }, + } +} + +// EventTypes documents what this gear publishes. +func (g *Gear) EventTypes() []gear.EventType { + return []gear.EventType{ + { + Name: "traefik.stats.updated", + Description: "Published when Traefik Prometheus metrics are scraped", + Payload: "traefik.Stats — request counters split by status class + entrypoints", + }, + } +} + +func (g *Gear) scrape(ctx context.Context) (any, error) { + res, err := g.httpGet(ctx, g.metricsURL) + if err != nil { + return nil, fmt.Errorf("traefik /metrics fetch: %w", err) + } + if res.StatusCode != http.StatusOK { + return nil, fmt.Errorf("traefik /metrics returned HTTP %d", res.StatusCode) + } + // Defence against the operator pointing us at a non-Traefik + // Prometheus exporter (the override branch's risk): a body + // without the traefik_ sentinel means we'd produce all zeros, + // which the dashboard would misread as "Traefik is up but + // idle". Surface as an error so the issue is visible. + if !strings.Contains(res.Body, prometheusSentinel) { + return nil, fmt.Errorf("traefik /metrics body lacked the %q sentinel; a different service may be answering at %s", prometheusSentinel, g.metricsURL) + } + stats := ParsePrometheusOutput(res.Body) + g.cacheStats(stats) + return stats, nil +} + +func (g *Gear) publish(data any) error { + if g.eventBus == nil { + return nil + } + stats, ok := data.(Stats) + if !ok { + return nil + } + g.eventBus.Publish(events.Event{ + Type: events.EventType("traefik.stats.updated"), + Timestamp: time.Now(), + Data: map[string]any{ + "collected_at": stats.CollectedAt, + "stats": stats, + }, + }) + return nil +} + +func (g *Gear) cacheStats(s Stats) { + g.statsMu.Lock() + g.lastStats = &s + g.statsMu.Unlock() +} + +func (g *Gear) readCachedStats() (Stats, bool) { + g.statsMu.RLock() + defer g.statsMu.RUnlock() + if g.lastStats == nil { + return Stats{}, false + } + return *g.lastStats, true +} + +// RegisterRoutes registers the traefik HTTP endpoint. +func (g *Gear) RegisterRoutes(r chi.Router) { + r.Get("/api/v1/traefik/stats", g.handleStats) +} + +func (g *Gear) handleStats(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("force") == "true" { + stats, err := g.scrape(r.Context()) + if err != nil { + http.Error(w, "traefik scrape failed: "+err.Error(), http.StatusBadGateway) + return + } + writeJSON(w, stats) + return + } + stats, ok := g.readCachedStats() + if !ok { + http.Error(w, "traefik stats not yet collected — agent just started, or metrics endpoint is not reachable", http.StatusServiceUnavailable) + return + } + writeJSON(w, stats) +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +var _ gear.CollectorGear = (*Gear)(nil) diff --git a/gearbox-agent/internal/gears/traefik/collector_test.go b/gearbox-agent/internal/gears/traefik/collector_test.go new file mode 100644 index 0000000..47fad5e --- /dev/null +++ b/gearbox-agent/internal/gears/traefik/collector_test.go @@ -0,0 +1,155 @@ +package traefik + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + + "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/gear" + "github.com/sarg3nt/gearbox-agent/internal/framework/probe" +) + +const sampleTraefikBody = `# HELP traefik_router_requests_total Total +# TYPE traefik_router_requests_total counter +traefik_router_requests_total{code="200",method="GET",router="r1"} 100 +traefik_router_requests_total{code="201",method="POST",router="r1"} 4 +traefik_router_requests_total{code="404",method="GET",router="r1"} 7 +traefik_router_requests_total{code="500",method="POST",router="r2"} 3 +traefik_router_requests_total{code="503",method="GET",router="r2"} 2 +traefik_entrypoint_requests_total{entrypoint="web",code="200"} 90 +traefik_entrypoint_requests_total{entrypoint="websecure",code="200"} 20 +` + +func TestParsePrometheusOutputBucketsByStatusClass(t *testing.T) { + got := ParsePrometheusOutput(sampleTraefikBody) + got.CollectedAt = "" + want := Stats{ + RequestsTotal: 116, // 100 + 4 + 7 + 3 + 2 + Response2xx: 104, // 100 + 4 + Response4xx: 7, + Response5xx: 5, // 3 + 2 + EntryPoints: []string{"web", "websecure"}, + } + if !reflect.DeepEqual(got, want) { + t.Errorf("Stats = %+v, want %+v", got, want) + } +} + +func TestParsePrometheusOutputIgnoresCounterWithoutCodeLabel(t *testing.T) { + // A traefik_router_requests_total without `code` shouldn't + // land in a status-class bucket (would silently inflate the + // wrong class). It still counts toward RequestsTotal because + // SumByName sums everything by name. + body := `traefik_router_requests_total{method="GET"} 99 +` + got := ParsePrometheusOutput(body) + if got.RequestsTotal != 99 { + t.Errorf("RequestsTotal = %d, want 99", got.RequestsTotal) + } + if got.Response1xx+got.Response2xx+got.Response3xx+got.Response4xx+got.Response5xx != 0 { + t.Errorf("status-class buckets should stay zero without code label; got %+v", got) + } +} + +func gearWithHTTPStub(get func(ctx context.Context, url string) (probe.HTTPResult, error)) *Gear { + g := New() + g.httpGet = get + g.metricsURL = defaultMetricsURLs[0] + return g +} + +func TestCollectorScrapeCachesAndReturnsStats(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusOK, Body: sampleTraefikBody}, nil + }) + data, err := g.scrape(context.Background()) + if err != nil { + t.Fatalf("scrape: %v", err) + } + stats, ok := data.(Stats) + if !ok || stats.Response5xx != 5 { + t.Errorf("scrape data = %+v, want Response5xx=5", data) + } +} + +func TestCollectorScrapeRejectsBodyWithoutSentinel(t *testing.T) { + // A non-Traefik service answering on the configured URL would + // return Prometheus output without the traefik_ sentinel. The + // collector should error rather than caching a zero-valued + // Stats that misleads the dashboard. + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{ + StatusCode: http.StatusOK, + Body: "other_service_metric{a=\"b\"} 1\n", + }, nil + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to error when sentinel is absent") + } +} + +func TestCollectorScrapeReturnsErrorOnNon200(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusServiceUnavailable}, nil + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to error on 503") + } +} + +func TestCollectorScrapeReturnsErrorOnTransportFailure(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{}, errors.New("connection refused") + }) + if _, err := g.scrape(context.Background()); err == nil { + t.Error("expected scrape to surface transport errors") + } +} + +func TestHandleStatsReturns503BeforeFirstScrape(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{}, errors.New("unused") + }) + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/traefik/stats", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + if rr.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503", rr.Code) + } +} + +func TestHandleStatsReturnsCachedJSON(t *testing.T) { + g := gearWithHTTPStub(func(context.Context, string) (probe.HTTPResult, error) { + return probe.HTTPResult{StatusCode: http.StatusOK, Body: sampleTraefikBody}, nil + }) + if _, err := g.scrape(context.Background()); err != nil { + t.Fatalf("scrape: %v", err) + } + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/traefik/stats", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Errorf("status = %d, want 200", rr.Code) + } + if !strings.Contains(rr.Body.String(), `"response_5xx":5`) { + t.Errorf("body missing response_5xx=5, got %s", rr.Body.String()) + } +} + +func TestPickMetricsURLHonorsOverride(t *testing.T) { + if got := pickMetricsURL(gear.Dependencies{TraefikMetricsURL: "http://1.2.3.4/_metrics"}); got != "http://1.2.3.4/_metrics" { + t.Errorf("override = %q", got) + } + if got := pickMetricsURL(gear.Dependencies{}); got != defaultMetricsURLs[0] { + t.Errorf("default = %q, want %q", got, defaultMetricsURLs[0]) + } +} diff --git a/gearbox-agent/internal/gears/traefik/plugin.go b/gearbox-agent/internal/gears/traefik/plugin.go index c9c4db0..4c735a8 100644 --- a/gearbox-agent/internal/gears/traefik/plugin.go +++ b/gearbox-agent/internal/gears/traefik/plugin.go @@ -23,8 +23,10 @@ import ( "os/exec" "regexp" "strings" + "sync" + "time" - "github.com/go-chi/chi/v5" + "github.com/sarg3nt/gearbox-agent/internal/framework/events" "github.com/sarg3nt/gearbox-agent/internal/framework/gear" "github.com/sarg3nt/gearbox-agent/internal/framework/probe" ) @@ -62,7 +64,7 @@ const dashboardAPIURL = "http://127.0.0.1:8080/api/rawdata" // The `v` prefix is optional; older builds printed plain semver. var versionRegex = regexp.MustCompile(`Version:\s+v?(\d+\.\d+\.\d+)`) -// Gear is the Traefik detector. +// Gear is the Traefik detector + Prometheus metrics collector. type Gear struct { gear.BaseGear @@ -71,6 +73,15 @@ type Gear struct { lookPath func(string) (string, error) runVersion func(ctx context.Context) ([]byte, error) httpGet func(ctx context.Context, url string) (probe.HTTPResult, error) + + // Collector-time state. Populated by Initialize once Probe + // returned Available. + metricsURL string + eventBus *events.Bus + statsInterval time.Duration + + statsMu sync.RWMutex + lastStats *Stats } // New constructs a Traefik gear with real OS-backed defaults. @@ -91,7 +102,7 @@ func (g *Gear) Info() gear.Info { return gear.Info{ Name: "traefik", DisplayName: "Traefik", - Description: "Detects a Traefik installation and verifies its Prometheus metrics endpoint so the Metrics gear can scrape it in Phase 7+.", + Description: "Detects a Traefik installation, verifies its Prometheus metrics endpoint, and periodically scrapes it for the Metrics gear.", Version: "1.0.0", Category: "monitoring", } @@ -195,9 +206,8 @@ func (g *Gear) recordDashboardAPI(ctx context.Context, caps map[string]string) { caps["dashboard_api"] = dashboardAPIURL } -// RegisterRoutes is a no-op — Phase 3 Traefik detection produces no API -// surface. The metrics gear lands in Phase 7+. -func (g *Gear) RegisterRoutes(_ chi.Router) {} +// RegisterRoutes and the periodic-scrape machinery live in +// collector.go alongside the rest of the metrics code. // Ensure the gear implements the required interfaces. var ( From 33ef56fb7cff6cd5790b7b03cb06fe2c27898af1 Mon Sep 17 00:00:00 2001 From: Dave Sargent Date: Thu, 14 May 2026 21:12:30 -0700 Subject: [PATCH 2/2] fix(#91): address Copilot review findings on PR #101 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - access-log: implement the documented Apache CLF fallback. Previously the handler used a single parser from sourceProfile (ApacheCombined) and the comments + PR body claimed a per-record fallback to ApacheCommon that didn't exist. RHEL hosts running default CLF would have produced zero parsed records. New parseWithFallback helper + sourceFallbackProfile map drive the actual fallback now; Apache is the only source using it today. - access-log: status_min query parameter now accepts an explicit 0 to disable the filter. Was previously clamped to a 100 minimum, which silently coerced 0 to 100 and broke the "give me all records" intent. Default when the param is absent stays 500 (the dashboard's primary use case). Lock the new defaults in via two new tests. - caddy: drop the AdminRunning field. The previous heuristic relied on caddy_admin_http_requests_total existing, which Prometheus doesn't emit for counters with zero increments. A freshly-started Caddy with admin enabled but no admin traffic yet would have falsely read "admin disconnected." The real signal is "did the scrape succeed?" — which the handler already conveys via 503 before the first successful scrape — so the field was redundant on success and misleading on cold start. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../internal/gears/accesslog/plugin.go | 50 ++++++-- .../internal/gears/accesslog/plugin_test.go | 116 ++++++++++++++++++ .../internal/gears/caddy/collector.go | 22 ++-- .../internal/gears/caddy/collector_test.go | 22 ++-- 4 files changed, 179 insertions(+), 31 deletions(-) diff --git a/gearbox-agent/internal/gears/accesslog/plugin.go b/gearbox-agent/internal/gears/accesslog/plugin.go index c9bd16c..d17c17e 100644 --- a/gearbox-agent/internal/gears/accesslog/plugin.go +++ b/gearbox-agent/internal/gears/accesslog/plugin.go @@ -58,10 +58,11 @@ var defaultLogPaths = map[string]string{ // "did we try multiple paths?" logic explicit at the call site. const apacheFallbackPath = "/var/log/httpd/access_log" -// sourceProfile maps a source identifier to the parser profile the -// endpoint applies to each line. Apache lines are tried as -// "combined" first (most operators run that format); ApacheCommon -// is the fallback handled inside the per-record retry. +// sourceProfile maps a source identifier to the primary parser +// profile the endpoint tries first. Apache uniquely also has a +// fallback profile (CLF without Referer / User-Agent) tried when +// the primary returns nil — see sourceFallbackProfile and +// parseWithFallback. var sourceProfile = map[string]string{ "haproxy": accesslog.ProfileHAProxy, "nginx": accesslog.ProfileNginxCombined, @@ -69,6 +70,16 @@ var sourceProfile = map[string]string{ "caddy": accesslog.ProfileCaddyJSON, } +// sourceFallbackProfile names the second profile tried when the +// primary returns nil for a line. Today only Apache has one: many +// RHEL-style installs ship CLF (no Referer / User-Agent) by +// default, so we try ApacheCombined first (covers the Debian +// default + custom combined-format ops) and fall back to +// ApacheCommon. Lines that match neither stay rejected as noise. +var sourceFallbackProfile = map[string]string{ + "apache": accesslog.ProfileApacheCommon, +} + // maxLimit caps the per-request `limit` parameter so a buggy // dashboard call can't make the agent shell `tail -n 100000` against // a 10 GB log file. 10 000 records matches the existing logs gear's @@ -205,8 +216,17 @@ func (g *Gear) handleRecent(w http.ResponseWriter, r *http.Request) { http.Error(w, "no parser registered for source "+source, http.StatusInternalServerError) return } - - statusMin := parseIntDefault(r.URL.Query().Get("status_min"), 0, 100, 599, 500) + // fallback may be nil — most sources have a single profile. + // parseWithFallback handles the nil case as "no second try". + fallback := accesslog.ProfileByName(sourceFallbackProfile[source]) + + // status_min defaults to 500 (the dashboard's primary use case + // is 5xx insights) but lets callers pass an explicit 0 to + // disable the filter entirely. Min clamp is 0 (not 100) so + // "explicitly disable" works; the gating logic compares with + // rec.StatusCode < statusMin, which is a no-op when statusMin + // is 0. + statusMin := parseIntDefault(r.URL.Query().Get("status_min"), 500, 0, 599, 500) limit := parseIntDefault(r.URL.Query().Get("limit"), defaultLimit, 1, maxLimit, defaultLimit) lines := parseIntDefault(r.URL.Query().Get("lines"), defaultLines, 1, maxLines, defaultLines) @@ -230,7 +250,7 @@ func (g *Gear) handleRecent(w http.ResponseWriter, r *http.Request) { // the slice in reverse. matches := make([]accesslog.Record, 0, limit) for i := len(raw) - 1; i >= 0; i-- { - rec := parser.Parse(raw[i]) + rec := parseWithFallback(parser, fallback, raw[i]) if rec == nil { continue } @@ -249,6 +269,22 @@ func (g *Gear) handleRecent(w http.ResponseWriter, r *http.Request) { writeJSON(w, resp) } +// parseWithFallback tries primary first; if primary returns nil and +// a fallback parser was registered for this source, it tries the +// fallback. Returns nil only when both reject the line. The Apache +// source uses this to handle both combined (default Debian) and +// CLF (default RHEL) without the caller needing to pre-detect +// which format the operator's running. +func parseWithFallback(primary, fallback accesslog.Parser, raw string) *accesslog.Record { + if rec := primary.Parse(raw); rec != nil { + return rec + } + if fallback == nil { + return nil + } + return fallback.Parse(raw) +} + // resolveLogPath returns the access-log path for src: the operator // override if set and readable, the well-known default if readable, // or "" when neither exists. Apache gets a second-chance lookup diff --git a/gearbox-agent/internal/gears/accesslog/plugin_test.go b/gearbox-agent/internal/gears/accesslog/plugin_test.go index 077a2e3..5daadd0 100644 --- a/gearbox-agent/internal/gears/accesslog/plugin_test.go +++ b/gearbox-agent/internal/gears/accesslog/plugin_test.go @@ -247,6 +247,122 @@ func TestHandleRecentSurfacesTailFailure(t *testing.T) { } } +func TestHandleRecentApacheFallsBackToCommonLogFormat(t *testing.T) { + // RHEL-style Apache emits CLF by default (no Referer/UA). Our + // primary apache profile is combined; the gear must fall back + // to ApacheCommon for any line the combined regex rejects so + // the dashboard sees records on those hosts instead of an empty + // envelope. + g := newTestGear() + g.stat = statExisting("/var/log/apache2/access.log") + g.tail = staticTail([]string{ + // CLF line — no trailing quoted fields. ApacheCombined + // rejects this; ApacheCommon must catch it. + `192.168.1.10 - - [01/Jan/2026:00:00:00 +0000] "GET /clf-route HTTP/1.1" 500 1234`, + // Combined line — both profiles would accept; primary wins. + `192.168.1.11 - - [01/Jan/2026:00:00:01 +0000] "GET /combined-route HTTP/1.1" 503 100 "-" "curl/8.0"`, + }) + + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/access-log/apache/recent?status_min=500", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + var resp Response + if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + if resp.MatchCount != 2 { + t.Fatalf("MatchCount = %d, want 2 (combined + CLF fallback)", resp.MatchCount) + } + // Newest-first ordering: combined line is the second tail + // entry (index 1, newer). The records carry their actual + // profile so the dashboard can tell which parser matched — + // CLF lines come back tagged ProfileApacheCommon, combined as + // ProfileApacheCombined. + if resp.Records[0].Profile != accesslog.ProfileApacheCombined { + t.Errorf("first record profile = %q, want %q (combined wins for that line)", resp.Records[0].Profile, accesslog.ProfileApacheCombined) + } + if resp.Records[1].Profile != accesslog.ProfileApacheCommon { + t.Errorf("second record profile = %q, want %q (CLF fallback)", resp.Records[1].Profile, accesslog.ProfileApacheCommon) + } +} + +func TestHandleRecentStatusMinZeroDisablesFilter(t *testing.T) { + // Explicit status_min=0 must return EVERY parsed record, + // including 2xx. The previous clamp at 100 silently coerced 0 + // up to 100, which is correct for HTTP statuses but blocked the + // "give me everything" use case the dashboard wants for + // general-purpose log browsing. + g := newTestGear() + g.stat = statExisting("/var/log/nginx/access.log") + g.tail = staticTail([]string{ + `1.1.1.1 - - [01/Jan/2026:00:00:00 +0000] "GET /a HTTP/1.1" 200 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:01 +0000] "GET /b HTTP/1.1" 304 0 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:02 +0000] "GET /c HTTP/1.1" 500 100 "-" "-"`, + }) + + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/access-log/nginx/recent?status_min=0", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + var resp Response + if err := json.NewDecoder(rr.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + if resp.MatchCount != 3 { + t.Errorf("MatchCount = %d with status_min=0, want 3 (all records)", resp.MatchCount) + } +} + +func TestHandleRecentStatusMinDefaultIs500(t *testing.T) { + // No status_min param → defaults to 500 (the dashboard's main + // use case). Locks in the documented default so a future tweak + // to parseIntDefault doesn't quietly change behaviour for + // existing callers. + g := newTestGear() + g.stat = statExisting("/var/log/nginx/access.log") + g.tail = staticTail([]string{ + `1.1.1.1 - - [01/Jan/2026:00:00:00 +0000] "GET /a HTTP/1.1" 200 100 "-" "-"`, + `1.1.1.1 - - [01/Jan/2026:00:00:01 +0000] "GET /b HTTP/1.1" 500 100 "-" "-"`, + }) + r := chi.NewRouter() + g.RegisterRoutes(r) + req := httptest.NewRequest(http.MethodGet, "/api/v1/access-log/nginx/recent", nil) + rr := httptest.NewRecorder() + r.ServeHTTP(rr, req) + + var resp Response + _ = json.NewDecoder(rr.Body).Decode(&resp) + if resp.MatchCount != 1 { + t.Errorf("MatchCount without explicit status_min = %d, want 1 (default 500 keeps only the 500-line)", resp.MatchCount) + } +} + +func TestParseWithFallbackTriesPrimaryFirst(t *testing.T) { + // Direct test of the helper to keep the contract pinned even + // if the handler rewires which sources use which fallback. + primary := accesslog.NginxCombinedProfile{} + fallback := accesslog.ApacheCommonProfile{} + + combinedLine := `1.1.1.1 - - [01/Jan/2026:00:00:00 +0000] "GET /x HTTP/1.1" 200 100 "-" "curl/8.0"` + clfLine := `1.1.1.1 - - [01/Jan/2026:00:00:00 +0000] "GET /x HTTP/1.1" 200 100` + + if got := parseWithFallback(primary, fallback, combinedLine); got == nil || got.Profile != accesslog.ProfileNginxCombined { + t.Errorf("combined line should match primary; got %+v", got) + } + if got := parseWithFallback(primary, fallback, clfLine); got == nil || got.Profile != accesslog.ProfileApacheCommon { + t.Errorf("CLF line should fall through to fallback; got %+v", got) + } + // No fallback registered → only primary attempted. + if got := parseWithFallback(primary, nil, clfLine); got != nil { + t.Errorf("nil fallback should not match CLF line; got %+v", got) + } +} + func TestParseIntDefaultClamps(t *testing.T) { cases := []struct { name string diff --git a/gearbox-agent/internal/gears/caddy/collector.go b/gearbox-agent/internal/gears/caddy/collector.go index 470d400..10bb3b4 100644 --- a/gearbox-agent/internal/gears/caddy/collector.go +++ b/gearbox-agent/internal/gears/caddy/collector.go @@ -37,17 +37,23 @@ type Stats struct { // double-counting. RequestErrorsTotal int64 `json:"request_errors_total"` - // AdminRunning is 1 when Caddy's admin endpoint reports itself - // alive, 0 otherwise. Lets the dashboard surface "Caddy - // running but admin disconnected" without a second probe. - AdminRunning bool `json:"admin_running"` - // CollectedAt is when the agent scraped, in RFC3339. CollectedAt string `json:"collected_at"` } // ParsePrometheusOutput extracts a Stats struct from raw Caddy // `:2019/metrics` Prometheus output. Exported for unit testing. +// +// We deliberately do NOT report a separate "admin reachable" +// boolean here: the scrape itself goes against the admin endpoint +// (default `:2019/metrics`), so a Stats record landing in the +// dashboard's cache already proves admin is reachable. Surfacing it +// as a field would either always be true (redundant) or depend on +// request-driven metrics like `caddy_admin_http_requests_total`, +// which a freshly-started Caddy with admin enabled but no admin +// traffic yet would not emit — confusing the dashboard with a false +// "admin disconnected" reading. The 503 response from the handler +// before the first scrape covers the "not reachable" state. func ParsePrometheusOutput(body string) Stats { samples := promtext.Parse(body) stats := Stats{CollectedAt: time.Now().UTC().Format(time.RFC3339)} @@ -55,12 +61,6 @@ func ParsePrometheusOutput(body string) Stats { stats.RequestsTotal = int64(promtext.SumByName(samples, "caddy_http_requests_total")) stats.RequestErrorsTotal = int64(promtext.SumByName(samples, "caddy_http_request_errors_total")) - if s := promtext.FirstByName(samples, "caddy_admin_http_requests_total"); s != nil { - // admin_http_requests_total existing at all means the - // admin endpoint is up — the metric is only registered - // when admin is enabled. - stats.AdminRunning = true - } return stats } diff --git a/gearbox-agent/internal/gears/caddy/collector_test.go b/gearbox-agent/internal/gears/caddy/collector_test.go index 13ee0c9..09b33d8 100644 --- a/gearbox-agent/internal/gears/caddy/collector_test.go +++ b/gearbox-agent/internal/gears/caddy/collector_test.go @@ -30,25 +30,21 @@ func TestParsePrometheusOutputSumsCounters(t *testing.T) { want := Stats{ RequestsTotal: 59, // 42 + 17 RequestErrorsTotal: 3, - AdminRunning: true, } if got != want { t.Errorf("Stats = %+v, want %+v", got, want) } } -func TestParsePrometheusOutputHandlesMissingAdminMetric(t *testing.T) { - // Admin endpoint disabled (admin off in Caddyfile) — the - // `caddy_admin_*` metric won't appear; AdminRunning must be - // false even though the rest of the scrape is fine. - body := `caddy_http_requests_total{server="srv0"} 7 -` - got := ParsePrometheusOutput(body) - if got.AdminRunning { - t.Error("AdminRunning should be false when admin metric is absent") - } - if got.RequestsTotal != 7 { - t.Errorf("RequestsTotal = %d, want 7", got.RequestsTotal) +func TestParsePrometheusOutputBackgroundEmptyBody(t *testing.T) { + // A scrape that returns 200 with no recognisable Caddy metric + // must parse cleanly to zero counters (NOT an error). The + // "admin reachable" question is answered by whether a scrape + // succeeded at all, which the handler conveys via 503 — not by + // any field on the returned Stats. + got := ParsePrometheusOutput("") + if got.RequestsTotal != 0 || got.RequestErrorsTotal != 0 { + t.Errorf("empty body should yield zero-valued counters, got %+v", got) } }