From e5e7a8073518cc3593b3ee4e7f8f4ecd9407305f Mon Sep 17 00:00:00 2001 From: Ryan Cook Date: Sun, 11 Oct 2026 16:26:15 -0400 Subject: [PATCH] Add opt-in fleet metrics with Git sync and host identity Signed-off-by: Ryan Cook --- .github/workflows/docker-image.yml | 3 + .github/workflows/unit-tests.yml | 2 +- docs/index.rst | 1 + docs/metrics.rst | 206 ++++++++++++++ docs/status.rst | 19 ++ pkg/engine/common.go | 6 + pkg/engine/fetchit.go | 13 +- pkg/engine/metrics.go | 216 +++++++++++++++ pkg/engine/metrics_test.go | 284 ++++++++++++++++++++ pkg/engine/quadlet.go | 10 +- pkg/engine/status.go | 23 +- tests/integration/live-update/git-server.py | 3 + tests/integration/metrics/README.md | 25 ++ tests/integration/metrics/test-metrics.sh | 122 +++++++++ 14 files changed, 926 insertions(+), 7 deletions(-) create mode 100644 docs/metrics.rst create mode 100644 pkg/engine/metrics.go create mode 100644 pkg/engine/metrics_test.go create mode 100644 tests/integration/metrics/README.md create mode 100644 tests/integration/metrics/test-metrics.sh diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml index 87a049ea..472b4217 100644 --- a/.github/workflows/docker-image.yml +++ b/.github/workflows/docker-image.yml @@ -1489,6 +1489,9 @@ jobs: - name: Reconcile a commit from an isolated Git repository run: sudo bash tests/integration/live-update/test-live-update.sh + - name: Validate metrics through Git failure and recovery + run: sudo bash tests/integration/metrics/test-metrics.sh + loader-validate: runs-on: ubuntu-26.04 needs: [ build, pull-and-archive ] diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml index 6c413e5b..0b7a9b7a 100644 --- a/.github/workflows/unit-tests.yml +++ b/.github/workflows/unit-tests.yml @@ -36,7 +36,7 @@ jobs: AGE_TEST_BINARY: /usr/bin/age-keygen run: go test -mod=readonly -tags "$BUILDTAGS" ./... - name: Engine concurrency race checks - run: go test -race -mod=readonly -tags "$BUILDTAGS" ./pkg/engine -run 'TestStatus|TestMirror|TestCommonGit|TestHelperContainer|TestRetirement|TestConfigReloadCan|TestRollback|TestDisconnected|TestFileTransfer|TestShutdown' + run: go test -race -mod=readonly -tags "$BUILDTAGS" ./pkg/engine -run 'TestStatus|TestMetrics|TestMirror|TestCommonGit|TestHelperContainer|TestRetirement|TestConfigReloadCan|TestRollback|TestDisconnected|TestFileTransfer|TestShutdown' fedora: runs-on: ubuntu-26.04 diff --git a/docs/index.rst b/docs/index.rst index 95db6e96..d2e63b1d 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -25,6 +25,7 @@ images containing those changes; an older published image may lack new options. mirrors quadlet status + metrics release_notes running documentation diff --git a/docs/metrics.rst b/docs/metrics.rst new file mode 100644 index 00000000..6d31bed0 --- /dev/null +++ b/docs/metrics.rst @@ -0,0 +1,206 @@ +Fleet metrics +============= + +FetchIt provides an optional, read-only Prometheus endpoint for +`issue #418 `_. An existing +remote Prometheus server scrapes FetchIt; FetchIt does not push metrics or need +a Prometheus server URL. Deploying Prometheus is outside FetchIt's scope. + +Enable the endpoint +------------------- + +Set ``FETCHIT_METRICS_ADDR`` to an explicit listen address. Unset means no +metrics listener. Port ``8080`` is the convention used in these examples; +you can choose another port. For a native process, begin with loopback: + +.. code-block:: bash + + FETCHIT_METRICS_ADDR=127.0.0.1:8080 FETCHIT_HOSTNAME=edge-01.example.invalid fetchit start + curl -i http://127.0.0.1:8080/metrics + +``FETCHIT_HOSTNAME`` defaults to the operating system hostname. For containers, +set it to the underlying host's unique hostname, since the container hostname +usually identifies the container. Allowed values contain 1–253 ASCII letters, +digits, dots, underscores or hyphens, starting with a letter or digit. An invalid +value fails startup. Host identity comes from configuration, never request +headers. ``/metrics`` includes the ``hostname`` label and an +``X-Fetchit-Hostname`` response header. The separate :doc:`status` listener +also includes that header, and ``/status`` adds a top-level ``hostname`` field. + +For an existing container deployment, add these options to ``podman run``: + +.. code-block:: bash + + -e FETCHIT_METRICS_ADDR=:8080 \ + -e FETCHIT_HOSTNAME=edge-01.example.invalid \ + -p 127.0.0.1:8080:8080 + +Keep the deployment's existing config, state volume, and Podman socket mounts. +These options listen inside the container and publish only the host's loopback +port. With host networking, omit ``-p`` and bind ``127.0.0.1:8080`` directly. + +Remote collection and security +------------------------------ + +To scrape remotely, publish on the host's monitoring-network address instead +of loopback, or bind that address directly with host networking. For example, +if ``10.0.1.20`` is an address assigned to the host: + +.. code-block:: bash + + -e FETCHIT_METRICS_ADDR=:8080 \ + -e FETCHIT_HOSTNAME=edge-01.example.invalid \ + -p 10.0.1.20:8080:8080 + +Permit inbound TCP 8080 only from the collector using the host firewall and +network security rules. In AWS, restrict the instance security group's rule to +the collector's security group or a specific private source address. Verify +both intended access and denial from other hosts. A private address alone is +not access control. Loopback does not authenticate local processes; published +loopback ports do not isolate a service from container-network peers. + +The listener serves plain HTTP with no authentication or native TLS. Use only +on a deliberately restricted trusted monitoring network. For traffic crossing +untrusted networks, use an authenticated TLS proxy or tunnel, with certificate +verification at the collector. This endpoint supplies neither encryption nor +credentials. Do not publish it on the public internet. + +The metrics listener serves only ``GET /metrics`` (also ``HEAD``). It does not +expose ``/status``, ``/healthz``, debug, or administration routes. Leave +``FETCHIT_STATUS_ADDR`` unset unless its separate endpoint is needed: +``/status`` contains raw method names and repository URLs. The two listeners +must use different addresses/ports if both are enabled. + +Scrapes read an in-memory snapshot. They do not fetch Git, invoke Podman, or +change workloads. The output contains bounded kinds/states and the configured +hostname; no repository URLs, method names, paths, SHAs, credentials, decrypted +contents, or free-form errors become labels. A Git SHA identifies a revision +shared by many hosts, not a host identity. Keep hostnames unique within the +fleet; Prometheus also attaches its own ``job`` and ``instance`` target labels. + +Configure an existing Prometheus server: + +.. code-block:: yaml + + scrape_configs: + - job_name: fetchit + metrics_path: /metrics + scrape_interval: 30s + static_configs: + - targets: ['edge-01.example.invalid:8080'] + +Replace the target with the reachable, access-controlled endpoint. Prometheus's +own port ``9090`` does not determine FetchIt's listening port. + +Metrics and synchronization +--------------------------- + +All metrics are gauges. A configuration means a **method declaration**, not a +repository, file, or deployed resource. One repository may contain several +independently scheduled methods. Counts describe the current configuration; +removed methods disappear and reloads reset synchronization to unknown until +startup replay and reconciliation complete. + +.. list-table:: Metric families + :header-rows: 1 + :widths: 45 55 + + * - Metric + - Meaning and labels + * - ``fetchit_host_info`` + - Value 1; ``hostname``. + * - ``fetchit_configured_methods`` + - All configured method declarations; ``hostname``, ``kind``. + * - ``fetchit_git_methods`` + - Connected Git methods; ``hostname``, ``kind``, ``state``. + * - ``fetchit_git_oldest_observation_timestamp_seconds`` + - Oldest successful Git observation among methods of a kind, in Unix + seconds; ``hostname``, ``kind``. Zero if any has never observed Git. + +``kind`` is one of ``raw``, ``kube``, ``ansible``, ``filetransfer``, ``systemd``, +``quadlet``, ``image``, ``prune``, ``config`` or ``other``. Only kinds +present in configuration are emitted. For connected Git methods, all three +state series are emitted, including zero values: + +* ``synced``: the latest successfully fetched and accepted branch revision was + applied successfully, and the saved Git baseline matches it. Startup replay + must also succeed. An unchanged revision is confirmed by a successful check. +* ``out_of_sync``: a desired revision was observed but its application or + baseline update has not completed successfully. Partial application, failed + application, skipped bad commits, and rollback do not establish sync. +* ``unknown``: no completed observation/application establishes the current + state, including startup, checks in progress, clone/replay failure, or source + access failure. A cancelled run cannot report success. + +Image, prune, config reload, auto-update, and disconnected archive/device +methods count toward configured declarations, but not connected Git sync +counts. Disconnected operation cannot establish whether a remote Git branch +is current. Mirror-enabled targets observe the accepted configured source +using FetchIt's existing mirror selection and commit verification rules. + +Git is checked on each method's schedule. ``synced`` means up to date with the +**last observed** repository/branch revision; commits published afterward are +unknown until the next check. Freshness advances only after a successful Git +fetch/reference checkout and any required commit verification. An apply failure +can have fresh Git knowledge while remaining out of sync. Source failure keeps +the previous timestamp, or zero if none exists. This is configuration +synchronization, not workload readiness or detection of manual workload drift. + +Example output for one successful raw method: + +.. code-block:: text + + fetchit_host_info{hostname="edge-01.example.invalid"} 1 + fetchit_configured_methods{hostname="edge-01.example.invalid",kind="raw"} 1 + fetchit_git_methods{hostname="edge-01.example.invalid",kind="raw",state="synced"} 1 + fetchit_git_methods{hostname="edge-01.example.invalid",kind="raw",state="out_of_sync"} 0 + fetchit_git_methods{hostname="edge-01.example.invalid",kind="raw",state="unknown"} 0 + fetchit_git_oldest_observation_timestamp_seconds{hostname="edge-01.example.invalid",kind="raw"} 1791700000.000 + +Queries and alert examples +-------------------------- + +Total configured declarations per host: + +.. code-block:: promql + + sum by (hostname) (fetchit_configured_methods) + +Pending or unknown connected Git methods: + +.. code-block:: promql + + sum by (hostname) (fetchit_git_methods{state=~"out_of_sync|unknown"}) > 0 + +For methods checking every minute, a five-minute observation deadline can be a +starting point. Adjust for the longest configured interval, skew, and expected +operation duration. Require the condition to persist to avoid alerts during +normal checks or startup: + +.. code-block:: yaml + + groups: + - name: fetchit + rules: + - alert: FetchitGitNotSynced + expr: sum by (hostname) (fetchit_git_methods{state=~"out_of_sync|unknown"}) > 0 + for: 5m + - alert: FetchitGitObservationStale + expr: time() - fetchit_git_oldest_observation_timestamp_seconds > 300 + for: 5m + - alert: FetchitScrapeUnavailable + expr: up{job="fetchit"} == 0 + for: 2m + +A successful scrape (``up == 1``) proves endpoint availability, not successful +Git observation or application. Zero freshness is deliberately stale. Missing +series after removal are different from a zero-valued state series; the +``up`` alert detects a failed scrape, while removal of a target from Prometheus +requires inventory monitoring outside FetchIt. + +References +---------- + +See `Prometheus scraping `_, +the `security model `_, and +`jobs and instances `_. diff --git a/docs/status.rst b/docs/status.rst index b93a85df..e9c0d09d 100644 --- a/docs/status.rst +++ b/docs/status.rst @@ -1,6 +1,11 @@ Health and status endpoint ========================== +For Prometheus configuration synchronization metrics, see :doc:`metrics`. +The separate metrics listener exposes no repository metadata. ``/healthz`` and +``/status`` describe liveness and invocation activity, not successful +configuration application, Git synchronization, or workload health. + FetchIt can expose an optional HTTP endpoint for process liveness and scheduled method activity, including Quadlet methods. Set ``FETCHIT_STATUS_ADDR`` to enable it; leaving the variable unset opens no listener. For a host installation, use @@ -18,12 +23,25 @@ and exposes method names and repository URLs; use trusted access or an authenticated reverse proxy for remote monitoring. Repository URLs should not contain credentials. +The loopback publication above restricts the published host port; it does not +authenticate local clients or prevent access from peers that can reach the +container's network. Keep the engine off networks shared with untrusted +containers. With host networking, bind the listener itself to +``127.0.0.1:8080``. Host-local processes that can connect to the listener can +read its output. Do not expose ``/status`` remotely merely to collect metrics: +the current response includes configured names and URLs without redaction. + .. code-block:: shell curl --fail http://127.0.0.1:8080/healthz curl --fail http://127.0.0.1:8080/status ``GET /healthz`` returns ``200 OK`` with ``ok`` while the HTTP server can respond. +Responses include an ``X-Fetchit-Hostname`` header; ``/status`` also reports +``hostname`` in JSON. Set ``FETCHIT_HOSTNAME`` explicitly to the underlying +hostname when running in a container; otherwise it defaults to the operating +system hostname. See :doc:`metrics` for validation rules. + It is a liveness check: it does not assert Podman connectivity, successful reconciliation, or workload readiness. @@ -33,6 +51,7 @@ reconciliation, or workload readiness. { "status": "running", + "hostname": "edge-01.example.invalid", "startedAt": "2026-10-08T12:00:00Z", "uptimeSeconds": 312, "methods": [ diff --git a/pkg/engine/common.go b/pkg/engine/common.go index bb080ab0..dbc80a6b 100644 --- a/pkg/engine/common.go +++ b/pkg/engine/common.go @@ -103,11 +103,13 @@ func currentToLatest(ctx, conn context.Context, m Method, target *Target, tag *[ return fmt.Errorf("Failed to get latest commit: %v", err) } + observeGit(ctx, "unknown", true) current, err := getCurrent(target, m.GetKind(), m.GetName()) if err != nil { return fmt.Errorf("Failed to get current commit: %v", err) } + observeGit(ctx, "out_of_sync", false) if latest != current { if err := applyWithRecovery(ctx, conn, m, current, latest, tag); err != nil { return fmt.Errorf("Failed to apply changes: %w", err) @@ -120,6 +122,10 @@ func currentToLatest(ctx, conn context.Context, m Method, target *Target, tag *[ logger.Infof("No changes applied to git target %s this run, %s currently at %s", directory, m.GetKind(), current.String()[:hashReportLen]) } + if err := ctx.Err(); err != nil { + return err + } + observeGit(ctx, "synced", false) return nil } diff --git a/pkg/engine/fetchit.go b/pkg/engine/fetchit.go index 8605070a..93adbd04 100644 --- a/pkg/engine/fetchit.go +++ b/pkg/engine/fetchit.go @@ -137,12 +137,12 @@ func (f *Fetchit) runMethod(m Method, ctx, conn context.Context, skew int) { return } status.recordRun(m) - m.Process(ctx, conn, skew) + m.Process(metrics.begin(ctx, m), conn, skew) return } defer f.runMu.RUnlock() status.recordRun(m) - m.Process(ctx, conn, skew) + m.Process(metrics.begin(ctx, m), conn, skew) } func readConfig(v *viper.Viper) (*FetchitConfig, bool, error) { @@ -426,6 +426,10 @@ func (f *Fetchit) startTargets() error { if err := parent.Err(); err != nil { return err } + hostname, err := configuredHostname() + if err != nil { + return err + } store, err := loadRemovalStore(defaultRemovalsPath) if err != nil { return err @@ -439,7 +443,12 @@ func (f *Fetchit) startTargets() error { } ctx, cancel := context.WithCancel(parent) f.runCancel = cancel + metrics.replace(f.methodTargetScheds, hostname) status.replace(f.methodTargetScheds) + status.setHostname(hostname) + if err := startMetricsServer(); err != nil { + return err + } startStatusServer() for method := range f.methodTargetScheds { if err := ctx.Err(); err != nil { diff --git a/pkg/engine/metrics.go b/pkg/engine/metrics.go new file mode 100644 index 00000000..ae425d49 --- /dev/null +++ b/pkg/engine/metrics.go @@ -0,0 +1,216 @@ +package engine + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "regexp" + "sort" + "strings" + "sync" + "time" +) + +const metricsAddrEnv = "FETCHIT_METRICS_ADDR" +const hostnameEnv = "FETCHIT_HOSTNAME" + +// No repository, method name, revision, path, or error text becomes a label. +// The set of kinds and states is bounded independently of configuration. +func metricKind(kind string) string { + switch kind { + case rawMethod, kubeMethod, ansibleMethod, filetransferMethod, systemdMethod, quadletMethod, imageMethod, pruneMethod, configFileMethod: + return kind + default: + return "other" + } +} + +type gitMetric struct { + kind string + state string + observed time.Time + run uint64 +} +type metricsRegistry struct { + mu sync.Mutex + hostname string + methods map[Method]*gitMetric +} + +var metrics = &metricsRegistry{} +var metricsServerOnce sync.Once +var hostnamePattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]*$`) + +func configuredHostname() (string, error) { + host := os.Getenv(hostnameEnv) + if host == "" { + var err error + host, err = os.Hostname() + if err != nil { + return "", err + } + } + if len(host) > 253 || !hostnamePattern.MatchString(host) { + return "", errors.New("FETCHIT_HOSTNAME must be a hostname of 1–253 ASCII letters, digits, dots, underscores or hyphens") + } + return host, nil +} + +func gitBacked(m Method) bool { + t := m.GetTarget() + return t != nil && t.url != "" && !t.disconnected +} + +func (r *metricsRegistry) replace(methods map[Method]SchedInfo, hostname string) { + r.mu.Lock() + defer r.mu.Unlock() + r.hostname = hostname + r.methods = make(map[Method]*gitMetric, len(methods)) + for m := range methods { + state := "not_applicable" + if gitBacked(m) { + state = "unknown" + } + r.methods[m] = &gitMetric{kind: metricKind(m.GetKind()), state: state} + } +} + +type observationKey struct{} +type gitObservation struct { + registry *metricsRegistry + method Method + entry *gitMetric + run uint64 +} + +// Each run gets a generation-specific token. A retired run cannot publish into +// a replacement configuration, even if its name and URL are unchanged. An +// overlapping older run also cannot overwrite a newer run's state. +func (r *metricsRegistry) begin(ctx context.Context, m Method) context.Context { + r.mu.Lock() + entry := r.methods[m] + var run uint64 + if entry != nil && entry.state != "not_applicable" { + entry.state = "unknown" + entry.run++ + run = entry.run + } + r.mu.Unlock() + return context.WithValue(ctx, observationKey{}, gitObservation{r, m, entry, run}) +} + +// Process returns no error, and its callers cannot infer success from return. +// Only the Git reconciliation paths explicitly publish verified observations +// and successful application/baseline completion through this context token. +func observeGit(ctx context.Context, state string, fetched bool) { + observation, ok := ctx.Value(observationKey{}).(gitObservation) + if !ok || observation.entry == nil || ctx.Err() != nil { + return + } + r := observation.registry + r.mu.Lock() + defer r.mu.Unlock() + entry := r.methods[observation.method] + if entry != observation.entry || entry.run != observation.run || entry.state == "not_applicable" { + return + } + entry.state = state + if fetched { + entry.observed = time.Now() + } +} + +func (r *metricsRegistry) writeMetrics(w http.ResponseWriter) { + type totals struct { + configured int + states map[string]int + oldest time.Time + git int + } + r.mu.Lock() + hostname := r.hostname + counts := map[string]*totals{} + for _, entry := range r.methods { + group := counts[entry.kind] + if group == nil { + group = &totals{states: map[string]int{}} + counts[entry.kind] = group + } + group.configured++ + if entry.state == "not_applicable" { + continue + } + group.states[entry.state]++ + if group.git == 0 || entry.observed.Before(group.oldest) { + group.oldest = entry.observed + } + group.git++ + } + r.mu.Unlock() + kinds := make([]string, 0, len(counts)) + for kind := range counts { + kinds = append(kinds, kind) + } + sort.Strings(kinds) + var output strings.Builder + fmt.Fprintf(&output, "# HELP fetchit_host_info Configured host identity.\n# TYPE fetchit_host_info gauge\nfetchit_host_info{hostname=%q} 1\n", hostname) + output.WriteString("# HELP fetchit_configured_methods Number of configured method declarations.\n# TYPE fetchit_configured_methods gauge\n") + for _, kind := range kinds { + fmt.Fprintf(&output, "fetchit_configured_methods{hostname=%q,kind=%q} %d\n", hostname, kind, counts[kind].configured) + } + output.WriteString("# HELP fetchit_git_methods Number of connected Git method declarations by synchronization state.\n# TYPE fetchit_git_methods gauge\n") + for _, kind := range kinds { + if counts[kind].git == 0 { + continue + } + for _, state := range []string{"synced", "out_of_sync", "unknown"} { + fmt.Fprintf(&output, "fetchit_git_methods{hostname=%q,kind=%q,state=%q} %d\n", hostname, kind, state, counts[kind].states[state]) + } + } + output.WriteString("# HELP fetchit_git_oldest_observation_timestamp_seconds Oldest successful Git observation per kind, or zero if any method has never observed Git.\n# TYPE fetchit_git_oldest_observation_timestamp_seconds gauge\n") + for _, kind := range kinds { + if counts[kind].git == 0 { + continue + } + timestamp := float64(0) + if !counts[kind].oldest.IsZero() { + timestamp = float64(counts[kind].oldest.UnixNano()) / 1e9 + } + fmt.Fprintf(&output, "fetchit_git_oldest_observation_timestamp_seconds{hostname=%q,kind=%q} %.3f\n", hostname, kind, timestamp) + } + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + w.Header().Set("X-Fetchit-Hostname", hostname) + _, _ = fmt.Fprint(w, output.String()) +} + +func newMetricsServer(addr string, r *metricsRegistry) *http.Server { + mux := http.NewServeMux() + mux.HandleFunc("GET /metrics", func(w http.ResponseWriter, _ *http.Request) { r.writeMetrics(w) }) + return &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 5 * time.Second, WriteTimeout: 10 * time.Second, IdleTimeout: 60 * time.Second, MaxHeaderBytes: 16 * 1024} +} + +func startMetricsServer() error { + addr := os.Getenv(metricsAddrEnv) + if addr == "" { + return nil + } + var result error + metricsServerOnce.Do(func() { + listener, err := net.Listen("tcp", addr) + if err != nil { + result = fmt.Errorf("metrics listener: %w", err) + return + } + server := newMetricsServer(addr, metrics) + logger.Infof("Metrics server listening on %s (/metrics)", listener.Addr()) + go func() { + if err := server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) { + logger.Errorf("metrics server: %v", err) + } + }() + }) + return result +} diff --git a/pkg/engine/metrics_test.go b/pkg/engine/metrics_test.go new file mode 100644 index 00000000..30a867d1 --- /dev/null +++ b/pkg/engine/metrics_test.go @@ -0,0 +1,284 @@ +package engine + +import ( + "context" + "errors" + "net/http/httptest" + "os" + "strings" + "sync" + "testing" + + "github.com/go-git/go-git/v5/plumbing" +) + +func metricOutput(r *metricsRegistry) string { + response := httptest.NewRecorder() + r.writeMetrics(response) + return response.Body.String() +} +func requireMetric(t *testing.T, r *metricsRegistry, expected string) { + t.Helper() + if output := metricOutput(r); !strings.Contains(output, expected) { + t.Fatalf("missing %q in:\n%s", expected, output) + } +} + +func TestMetricsReconciliationAndFreshness(t *testing.T) { + target, source, sourcePath := mirrorTestTarget(t) + mirrorCommit(t, source, "initial") + m := &lifecycleFake{CommonMethod: CommonMethod{Name: "secret-name", target: target, initialRun: true}, kind: rawMethod} + fail := true + m.apply = func(_, _ plumbing.Hash) error { + if fail { + return errors.New("secret apply failure") + } + return nil + } + r := &metricsRegistry{} + r.replace(map[Method]SchedInfo{m: {}}, "edge-01.example.invalid") + label := `hostname="edge-01.example.invalid",kind="raw",state=` + requireMetric(t, r, `fetchit_git_methods{`+label+`"unknown"} 1`) + run := func() { ctx := r.begin(context.Background(), m); m.processGit(ctx, ctx, m, 0, nil) } + run() + requireMetric(t, r, `fetchit_git_methods{`+label+`"out_of_sync"} 1`) + firstObservation := r.methods[m].observed + if firstObservation.IsZero() { + t.Fatal("successful fetch not observed") + } + fail = false + run() + requireMetric(t, r, `fetchit_git_methods{`+label+`"synced"} 1`) + if !r.methods[m].observed.After(firstObservation) { + t.Fatal("freshness did not advance") + } + // A previously synced cache cannot be mistaken for a successful new fetch. + observed := r.methods[m].observed + if err := os.Rename(sourcePath, sourcePath+"-offline"); err != nil { + t.Fatal(err) + } + run() + requireMetric(t, r, `fetchit_git_methods{`+label+`"unknown"} 1`) + if !r.methods[m].observed.Equal(observed) { + t.Fatal("failed fetch refreshed freshness") + } + if err := os.Rename(sourcePath+"-offline", sourcePath); err != nil { + t.Fatal(err) + } + run() + requireMetric(t, r, `fetchit_git_methods{`+label+`"synced"} 1`) + // Startup replay is required before equality of saved and desired revisions + // can establish synchronization after reload. + m.initialRun = true + fail = true + run() + requireMetric(t, r, `fetchit_git_methods{`+label+`"unknown"} 1`) +} + +func TestMetricsRetirementCancellationAndMetadata(t *testing.T) { + target := &Target{url: "https://user:secret-password@example.invalid/private.git", branch: "main"} + m := &lifecycleFake{CommonMethod: CommonMethod{Name: "secret-name", target: target}, kind: rawMethod} + nonGit := &lifecycleFake{CommonMethod: CommonMethod{target: &Target{}}, kind: imageMethod} + disconnected := &lifecycleFake{CommonMethod: CommonMethod{target: &Target{url: "secret-archive", disconnected: true}}, kind: kubeMethod} + r := &metricsRegistry{} + methods := map[Method]SchedInfo{m: {}, nonGit: {}, disconnected: {}} + r.replace(methods, "edge-01") + old := r.begin(context.Background(), m) + observeGit(old, "synced", true) + r.replace(methods, "edge-01") + observeGit(old, "synced", true) + requireMetric(t, r, `fetchit_git_methods{hostname="edge-01",kind="raw",state="unknown"} 1`) + older := r.begin(context.Background(), m) + newer := r.begin(context.Background(), m) + observeGit(older, "synced", true) + if !r.methods[m].observed.IsZero() { + t.Fatal("superseded run published success") + } + observeGit(newer, "out_of_sync", false) + cancelled, cancel := context.WithCancel(r.begin(context.Background(), m)) + cancel() + observeGit(cancelled, "synced", true) + if !r.methods[m].observed.IsZero() { + t.Fatal("retired or cancelled run changed replacement") + } + output := metricOutput(r) + for _, secret := range []string{"secret", "private.git", "https:", "branch", "state=\"not_applicable\""} { + if strings.Contains(output, secret) { + t.Fatalf("metadata leaked: %s", secret) + } + } + requireMetric(t, r, `fetchit_configured_methods{hostname="edge-01",kind="image"} 1`) + if strings.Contains(output, `fetchit_git_methods{hostname="edge-01",kind="image"`) || strings.Contains(output, `fetchit_git_methods{hostname="edge-01",kind="kube"`) { + t.Fatal("non-connected Git counted as synchronized") + } + r.replace(nil, "edge-01") + if strings.Contains(metricOutput(r), `kind="raw"`) { + t.Fatal("removed method remains") + } +} + +func TestMetricsRoutesAndHostname(t *testing.T) { + r := &metricsRegistry{} + r.replace(nil, "edge-01") + server := newMetricsServer("127.0.0.1:8080", r) + for _, test := range []struct { + method, path string + code int + }{{"GET", "/metrics", 200}, {"HEAD", "/metrics", 200}, {"POST", "/metrics", 405}, {"GET", "/status", 404}, {"GET", "/healthz", 404}, {"GET", "/debug/pprof/", 404}} { + response := httptest.NewRecorder() + server.Handler.ServeHTTP(response, httptest.NewRequest(test.method, test.path, nil)) + if response.Code != test.code { + t.Fatalf("%s %s: %d", test.method, test.path, response.Code) + } + if test.code == 200 && response.Header().Get("X-Fetchit-Hostname") != "edge-01" { + t.Fatal("host identity missing") + } + } + if server.ReadTimeout == 0 || server.WriteTimeout == 0 || server.ReadHeaderTimeout == 0 || server.MaxHeaderBytes > 16384 { + t.Fatal("unbounded listener") + } + t.Setenv(hostnameEnv, "edge-01.example.invalid") + if host, err := configuredHostname(); err != nil || host != "edge-01.example.invalid" { + t.Fatal(host, err) + } + for _, host := range []string{"bad\nmetric 1", "bad\"label", strings.Repeat("a", 254), "-invalid", "é"} { + t.Setenv(hostnameEnv, host) + if _, err := configuredHostname(); err == nil { + t.Fatalf("invalid hostname accepted: %q", host) + } + } +} + +func TestMetricsConcurrentScrapesAndReloads(t *testing.T) { + m := &lifecycleFake{CommonMethod: CommonMethod{target: &Target{url: "secret"}}, kind: rawMethod} + r := &metricsRegistry{} + r.replace(map[Method]SchedInfo{m: {}}, "edge-01") + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 100; j++ { + ctx := r.begin(context.Background(), m) + observeGit(ctx, "out_of_sync", true) + observeGit(ctx, "synced", false) + _ = metricOutput(r) + r.replace(map[Method]SchedInfo{m: {}}, "edge-01") + } + }() + } + wg.Wait() + // Aggregate freshness is governed by the oldest method, including never fetched. + second := &lifecycleFake{CommonMethod: CommonMethod{target: &Target{url: "secret2"}}, kind: rawMethod} + r.replace(map[Method]SchedInfo{m: {}, second: {}}, "edge-01") + observeGit(r.begin(context.Background(), m), "synced", true) + requireMetric(t, r, `fetchit_git_oldest_observation_timestamp_seconds{hostname="edge-01",kind="raw"} 0.000`) + observeGit(r.begin(context.Background(), second), "synced", true) + if strings.Contains(metricOutput(r), `kind="raw"} 0.000`) { + t.Fatal("freshness not recorded") + } +} + +func TestMetricsRollbackAndSkippedBadCommitStayOutOfSync(t *testing.T) { + target, source, _ := mirrorTestTarget(t) + initial := mirrorCommit(t, source, "initial") + target.rollback = true + target.trackBadCommits = true + m := &lifecycleFake{CommonMethod: CommonMethod{Name: "rollback", target: target, initialRun: true}, kind: rawMethod} + m.apply = func(_, _ plumbing.Hash) error { return nil } + r := &metricsRegistry{} + r.replace(map[Method]SchedInfo{m: {}}, "edge-01") + run := func() { ctx := r.begin(context.Background(), m); m.processGit(ctx, ctx, m, 0, nil) } + run() + desired := mirrorCommit(t, source, "bad") + calls := 0 + m.apply = func(from, to plumbing.Hash) error { + calls++ + if to == desired { + return errors.New("bad apply") + } + if from != desired || to != initial { + t.Fatalf("unexpected rollback %s -> %s", from, to) + } + return nil + } + run() + requireMetric(t, r, `fetchit_git_methods{hostname="edge-01",kind="raw",state="out_of_sync"} 1`) + current, err := getCurrent(target, rawMethod, m.Name) + if err != nil || current != initial || calls != 2 { + t.Fatalf("rollback baseline: %s %v calls=%d", current, err, calls) + } + run() + requireMetric(t, r, `fetchit_git_methods{hostname="edge-01",kind="raw",state="out_of_sync"} 1`) + if calls != 2 { + t.Fatal("known bad commit was retried") + } +} + +func TestMetricsQuadletForcedReplayMustSucceed(t *testing.T) { + target, source, _ := mirrorTestTarget(t) + worktree, err := source.Worktree() + if err != nil { + t.Fatal(err) + } + directory := worktree.Filesystem.Root() + if err := os.MkdirAll(directory+"/bundle", 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(directory+"/bundle/app.container", []byte("[Container]\nImage=quay.io/podman/hello\n"), 0644); err != nil { + t.Fatal(err) + } + if _, err := worktree.Add("bundle/app.container"); err != nil { + t.Fatal(err) + } + mirrorCommit(t, source, "bundle") + q := &Quadlet{CommonMethod: CommonMethod{Name: "quadlet", TargetPath: "bundle", target: target, initialRun: true}, Root: true} + fail := false + q.runHost = func(context.Context, context.Context, quadletPlan) error { + if fail { + return errors.New("replay failed") + } + return nil + } + r := &metricsRegistry{} + r.replace(map[Method]SchedInfo{q: {}}, "edge-01") + run := func() { ctx := r.begin(context.Background(), q); q.Process(ctx, ctx, 0) } + run() + requireMetric(t, r, `fetchit_git_methods{hostname="edge-01",kind="quadlet",state="synced"} 1`) + q.initialRun = true + fail = true + run() + requireMetric(t, r, `fetchit_git_methods{hostname="edge-01",kind="quadlet",state="out_of_sync"} 1`) +} + +func TestStatusHostnameIdentity(t *testing.T) { + r := &statusRegistry{methods: map[string]*methodStatus{}} + r.setHostname("edge-01") + server := newStatusServer("", r) + for _, path := range []string{"/status", "/healthz"} { + response := httptest.NewRecorder() + server.Handler.ServeHTTP(response, httptest.NewRequest("GET", path, nil)) + if response.Header().Get("X-Fetchit-Hostname") != "edge-01" { + t.Fatal("hostname header missing") + } + if path == "/status" && !strings.Contains(response.Body.String(), `"hostname": "edge-01"`) { + t.Fatal("JSON hostname missing") + } + } +} + +func TestMetricsDefaultsAndDisabledListener(t *testing.T) { + t.Setenv(hostnameEnv, "") + expected, err := os.Hostname() + if err != nil { + t.Fatal(err) + } + actual, err := configuredHostname() + if err != nil || actual != expected { + t.Fatalf("native hostname: %q %v", actual, err) + } + t.Setenv(metricsAddrEnv, "") + if err := startMetricsServer(); err != nil { + t.Fatal(err) + } +} diff --git a/pkg/engine/quadlet.go b/pkg/engine/quadlet.go index c01baf49..5895245f 100644 --- a/pkg/engine/quadlet.go +++ b/pkg/engine/quadlet.go @@ -100,10 +100,12 @@ func (q *Quadlet) reconcile(ctx, conn context.Context) error { if err != nil { return err } + observeGit(ctx, "unknown", true) current, err := getCurrent(target, q.GetKind(), q.GetName()) if err != nil { return err } + observeGit(ctx, "out_of_sync", false) q.forceReconcile = q.initialRun || q.dirty err = applyWithRecovery(ctx, conn, q, current, latest, nil) q.forceReconcile = false @@ -111,8 +113,14 @@ func (q *Quadlet) reconcile(ctx, conn context.Context) error { return err } if latest != current { - return updateCurrent(ctx, target, latest, q.GetKind(), q.GetName()) + if err := updateCurrent(ctx, target, latest, q.GetKind(), q.GetName()); err != nil { + return err + } + } + if err := ctx.Err(); err != nil { + return err } + observeGit(ctx, "synced", false) return nil } diff --git a/pkg/engine/status.go b/pkg/engine/status.go index cf67ca4c..71169350 100644 --- a/pkg/engine/status.go +++ b/pkg/engine/status.go @@ -26,9 +26,10 @@ type methodStatus struct { } type statusRegistry struct { - mu sync.Mutex - started time.Time - methods map[string]*methodStatus + mu sync.Mutex + started time.Time + hostname string + methods map[string]*methodStatus } var status = &statusRegistry{methods: make(map[string]*methodStatus)} @@ -72,6 +73,17 @@ func (r *statusRegistry) recordRun(m Method) { s.LastRun = &now } +func (r *statusRegistry) setHostname(hostname string) { + r.mu.Lock() + defer r.mu.Unlock() + r.hostname = hostname +} +func (r *statusRegistry) hostName() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.hostname +} + func (r *statusRegistry) writeJSON(w http.ResponseWriter) { r.mu.Lock() // Copy by value under the lock so concurrent recordRun/register calls @@ -81,6 +93,7 @@ func (r *statusRegistry) writeJSON(w http.ResponseWriter) { ms = append(ms, *s) } started := r.started + hostname := r.hostname r.mu.Unlock() sort.Slice(ms, func(i, j int) bool { @@ -109,17 +122,20 @@ func (r *statusRegistry) writeJSON(w http.ResponseWriter) { } resp := struct { Status string `json:"status"` + Hostname string `json:"hostname"` StartedAt *time.Time `json:"startedAt,omitempty"` UptimeSeconds int64 `json:"uptimeSeconds"` Methods []methodStatus `json:"methods"` }{ Status: state, + Hostname: hostname, StartedAt: startedAt, UptimeSeconds: uptime, Methods: ms, } w.Header().Set("Content-Type", "application/json") + w.Header().Set("X-Fetchit-Hostname", hostname) enc := json.NewEncoder(w) enc.SetIndent("", " ") if err := enc.Encode(resp); err != nil { @@ -154,6 +170,7 @@ func newStatusServer(addr string, registry *statusRegistry) *http.Server { mux := http.NewServeMux() mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.Header().Set("X-Fetchit-Hostname", registry.hostName()) if _, err := w.Write([]byte("ok\n")); err != nil { logger.Debugf("status: failed to write health response: %v", err) } diff --git a/tests/integration/live-update/git-server.py b/tests/integration/live-update/git-server.py index d83f9b45..8ef644e9 100644 --- a/tests/integration/live-update/git-server.py +++ b/tests/integration/live-update/git-server.py @@ -17,6 +17,9 @@ def do_POST(self): self.serve_git() def serve_git(self): + if Path(self.server.repository_root, "offline").exists(): + self.send_error(503, "Git fixture offline") + return request = urlsplit(self.path) if request.path not in ("/live.git/info/refs", "/live.git/git-upload-pack"): self.send_error(404) diff --git a/tests/integration/metrics/README.md b/tests/integration/metrics/README.md new file mode 100644 index 00000000..862759a2 --- /dev/null +++ b/tests/integration/metrics/README.md @@ -0,0 +1,25 @@ +Run the production image against a temporary, loopback-only smart HTTP Git +repository and real rootful Podman: + +```sh +sudo systemctl enable --now podman.socket +sudo podman pull quay.io/fetchit/fetchit-sample-app:latest +sudo env FETCHIT_TEST_ENGINE_IMAGE=localhost/fetchit:metrics \ + bash tests/integration/metrics/test-metrics.sh +``` + +Build the engine using the repository Dockerfile and vendored dependencies +first. The test needs Bash, Python 3, Git, curl, awk, and a free TCP port 8080. +It removes its own engine, workload, state volume, Git server, and temporary +files on exit. Allow roughly five minutes for scheduled checks. + +The test verifies scheduled Git updates, an invalid workload remaining out of +sync, source failure retaining the previous observation timestamp, recovery, +hostname reporting, absence of repository metadata, rejected administration +routes, and read-only scrapes. + +For a collector on a second private host, override +`FETCHIT_TEST_METRICS_ADDR=PRIVATE_HOST_IP:8080` and +`FETCHIT_TEST_METRICS_URL=http://PRIVATE_HOST_IP:8080/metrics`. Restrict inbound +8080 to that collector before running. Configure the existing collector to +scrape that address. The test neither installs nor starts Prometheus. diff --git a/tests/integration/metrics/test-metrics.sh b/tests/integration/metrics/test-metrics.sh new file mode 100644 index 00000000..ab9793d9 --- /dev/null +++ b/tests/integration/metrics/test-metrics.sh @@ -0,0 +1,122 @@ +#!/usr/bin/env bash +# Exercise the production engine with a private Git fixture and real Podman. +set -euo pipefail +engine_image=${FETCHIT_TEST_ENGINE_IMAGE:-quay.io/fetchit/fetchit-amd:latest} +sample_image=quay.io/fetchit/fetchit-sample-app:latest +listen=${FETCHIT_TEST_METRICS_ADDR:-127.0.0.1:8080} +endpoint=${FETCHIT_TEST_METRICS_URL:-http://127.0.0.1:8080/metrics} +scratch=$(mktemp -d) +fixture_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")/../live-update" && pwd) +engine="fetchit-metrics-$$" +workload="$engine-app" +state="$engine-state" +server_pid= +cleanup() { + status=$? + podman logs "$engine" || true + cat "$scratch/server.log" || true + podman rm -f "$engine" "$workload" >/dev/null 2>&1 || true + podman volume rm "$state" >/dev/null 2>&1 || true + if [[ -n "$server_pid" ]]; then kill "$server_pid" 2>/dev/null || true; wait "$server_pid" 2>/dev/null || true; fi + rm -rf "$scratch" + exit "$status" +} +trap cleanup EXIT +podman image exists "$engine_image" +podman image exists "$sample_image" +mkdir -p "$scratch/server" "$scratch/seed/raw" +git init --bare --initial-branch=main "$scratch/server/live.git" +git -C "$scratch/server/live.git" config http.receivepack false +git init --initial-branch=main "$scratch/seed" +git -C "$scratch/seed" config user.name 'FetchIt metrics fixture' +git -C "$scratch/seed" config user.email 'ci@example.invalid' +git -C "$scratch/seed" remote add origin "$scratch/server/live.git" +write_workload() { + cat > "$scratch/seed/raw/app.json" < "$scratch/server.log" 2>&1 & +server_pid=$! +for ((attempt=0;attempt<50;attempt++)); do + if [[ -s "$scratch/port" ]]; then break; fi + kill -0 "$server_pid" + sleep 0.1 +done +port=$(cat "$scratch/port") +cat > "$scratch/config.yaml" </dev/null) != true ]]; then return 1; fi + if [[ $(metric "fetchit_git_methods{hostname=\"edge-metrics-test\",kind=\"raw\",state=\"$wanted\"}" || true) == 1 ]]; then return 0; fi + sleep 1 + done + echo "Timed out waiting for $wanted" >&2 + return 1 +} +color() { podman exec "$workload" printenv APP_COLOR; } +wait_state synced +test "$(color)" = blue +output=$(curl --fail --silent "$endpoint") +! [[ "$output" =~ private-method-name|live.git|127.0.0.1|app.json|APP_COLOR ]] +base=${endpoint%/metrics} +test "$(curl -s -o /dev/null -w '%{http_code}' "$base/status")" = 404 +test "$(curl -s -o /dev/null -w '%{http_code}' "$base/debug/pprof/")" = 404 +test "$(curl -s -X POST -o /dev/null -w '%{http_code}' "$endpoint")" = 405 +curl --fail --silent -I "$endpoint" | awk 'tolower($0) ~ /^x-fetchit-hostname: edge-metrics-test/ {found=1} END {exit !found}' +# Repeated scrapes must not invoke reconciliation or recreate the workload. +old_id=$(podman inspect --format '{{.Id}}' "$workload") +for ((i=0;i<30;i++)); do curl --fail --silent "$endpoint" >/dev/null; done +test "$(podman inspect --format '{{.Id}}' "$workload")" = "$old_id" +# Observe an actual scheduled update, without restarting the engine. +write_workload pink +publish update +# State may remain synced until the next scheduled check; wait for the workload. +for ((deadline=SECONDS+150;SECONDS/dev/null) == pink ]]; then break; fi + sleep 1 +done +test "$(color)" = pink +wait_state synced +# Malformed workload produces fresh Git knowledge and a failed application. +printf '{invalid-json\n' > "$scratch/seed/raw/app.json" +publish invalid +wait_state out_of_sync +# Source failure must not advance the observed timestamp. +first=$(metric 'fetchit_git_oldest_observation_timestamp_seconds{hostname="edge-metrics-test",kind="raw"}') +touch "$scratch/server/offline" +wait_state unknown +test "$(metric 'fetchit_git_oldest_observation_timestamp_seconds{hostname="edge-metrics-test",kind="raw"}')" = "$first" +write_workload green +publish recover +rm "$scratch/server/offline" +wait_state synced +test "$(color)" = green +second=$(metric 'fetchit_git_oldest_observation_timestamp_seconds{hostname="edge-metrics-test",kind="raw"}') +awk -v first="$first" -v second="$second" 'BEGIN {exit !(second > first)}' +test "$(podman inspect --format '{{.RestartCount}}' "$engine")" = 0 +echo 'Metrics integration passed: Git update, apply failure, source failure, recovery, hostname and read-only routes.'