From e3b2cab4a4ef11ff804467e0a4d4a7695f744e3a Mon Sep 17 00:00:00 2001 From: PrashantBtkl Date: Mon, 17 Aug 2026 11:02:38 +0530 Subject: [PATCH] collector result metadata --- docs/cli.md | 16 +++- pkg/proxy/discovery/README.md | 7 ++ pkg/proxy/discovery/cloud_identity.go | 2 +- pkg/proxy/discovery/executor.go | 84 +++++++++++------ pkg/proxy/discovery/executor_test.go | 124 ++++++++++++++++++++++++-- 5 files changed, 199 insertions(+), 34 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index fd24219..0f65980 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -161,7 +161,10 @@ The example pack lives at `docs/content-packs/linux-inventory-example.yaml`. "pkgs-dpkg": "acpid\t1:2.0.33-1ubuntu1\tamd64\tinstalled\n...", "machine-id": "ec2403e319a2f3f0ae53a05e3daf084b\n", "os-release": "NAME=\"Ubuntu\"\nID=ubuntu\n..." - } + }, + "collector_exit_codes": {"pkgs-dpkg": 0, "machine-id": 0}, + "collector_truncated": {"pkgs-dpkg": false, "machine-id": false}, + "collector_output_limit_bytes": 1048576 }, {"host": "192.168.1.51", "status": "ssh-auth-failed", "error": "..."} ] @@ -187,6 +190,17 @@ Collector output is returned raw, exactly as the command printed it. Parsing happens server-side, which is why fixing a parser never requires touching a single machine. +Each successfully started collector also reports its command exit status in +`collector_exit_codes`. A non-zero status is preserved as data because some +package-management commands use it to communicate findings (for example, +`100` can mean updates are available); it is not treated as a collector +failure. Commands that cannot complete, including timeouts, are reported in +`collector_errors` and have no exit-code entry. + +`collector_truncated` identifies output that exceeded the per-stream cap shown +in `collector_output_limit_bytes`. The cap applies independently to stdout +and stderr, and the result is marked truncated if either stream exceeded it. + --- ## pack — managing content packs diff --git a/pkg/proxy/discovery/README.md b/pkg/proxy/discovery/README.md index 4b1f444..199d34e 100644 --- a/pkg/proxy/discovery/README.md +++ b/pkg/proxy/discovery/README.md @@ -119,6 +119,13 @@ Response: ]} ``` +Successful collectors also include their command status in +`collector_exit_codes`. Non-zero statuses remain collected data; commands +that fail to complete, including timeouts, are listed in +`collector_errors`. `collector_truncated` reports whether stdout or stderr +exceeded `collector_output_limit_bytes`, the per-stream output cap applied to +the target. + Per-target statuses (`ok`, `ssh-refused`, `ssh-auth-failed`, `timeout`, `error`) map onto the server's coverage states and gap reasons. **One host failing never fails the batch** — "which hosts could we not reach, and why" diff --git a/pkg/proxy/discovery/cloud_identity.go b/pkg/proxy/discovery/cloud_identity.go index 2939ba3..afa4933 100644 --- a/pkg/proxy/discovery/cloud_identity.go +++ b/pkg/proxy/discovery/cloud_identity.go @@ -109,7 +109,7 @@ func enrichCloudIdentity(ctx context.Context, hosts []SweepHost, cfg execConfig) } defer func() { _ = client.Close() }() - out, _, err := runCommand(hostCtx, client, cloudIdentityProbeCmd, cfg) + out, _, _, _, err := runCommand(hostCtx, client, cloudIdentityProbeCmd, cfg) if err != nil || strings.TrimSpace(out) == "" { return } diff --git a/pkg/proxy/discovery/executor.go b/pkg/proxy/discovery/executor.go index b6b9758..90ca9a5 100644 --- a/pkg/proxy/discovery/executor.go +++ b/pkg/proxy/discovery/executor.go @@ -9,6 +9,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "golang.org/x/crypto/ssh" @@ -28,14 +29,17 @@ const ( // is returned raw: parsing lives server-side so that fixing a parser never // requires touching the agent. type TargetResult struct { - Host string `json:"host"` - Status string `json:"status"` - Error string `json:"error,omitempty"` - Facts map[string]string `json:"facts,omitempty"` - Collected map[string]string `json:"collectors,omitempty"` - Failed map[string]string `json:"collector_errors,omitempty"` - Skipped []SkippedCollector `json:"skipped_collectors,omitempty"` - DurationS float64 `json:"duration_seconds"` + Host string `json:"host"` + Status string `json:"status"` + Error string `json:"error,omitempty"` + Facts map[string]string `json:"facts,omitempty"` + Collected map[string]string `json:"collectors,omitempty"` + CollectorExitCodes map[string]int `json:"collector_exit_codes,omitempty"` + CollectorTruncated map[string]bool `json:"collector_truncated,omitempty"` + CollectorOutputLimitBytes int `json:"collector_output_limit_bytes,omitempty"` + Failed map[string]string `json:"collector_errors,omitempty"` + Skipped []SkippedCollector `json:"skipped_collectors,omitempty"` + DurationS float64 `json:"duration_seconds"` } // execConfig is the tunable part of an inventory run. @@ -94,7 +98,7 @@ func inventoryHost(ctx context.Context, host string, pack *Pack, cfg execConfig) defer func() { _ = client.Close() }() // Probe facts first: the pack's guards are evaluated against them. - probeOut, _, err := runCommand(ctx, client, factsProbe, cfg) + probeOut, _, _, _, err := runCommand(ctx, client, factsProbe, cfg) if err != nil { res.Status = StatusError res.Error = fmt.Sprintf("facts probe: %v", err) @@ -105,6 +109,9 @@ func inventoryHost(ctx context.Context, host string, pack *Pack, cfg execConfig) collectors, skipped := pack.Select(res.Facts) res.Skipped = skipped res.Collected = make(map[string]string, len(collectors)) + res.CollectorExitCodes = make(map[string]int, len(collectors)) + res.CollectorTruncated = make(map[string]bool, len(collectors)) + res.CollectorOutputLimitBytes = cfg.maxOutputBytes for _, c := range collectors { if ctx.Err() != nil { @@ -112,7 +119,7 @@ func inventoryHost(ctx context.Context, host string, pack *Pack, cfg execConfig) res.Error = "host timeout during collection" return res } - out, stderr, err := runCommand(ctx, client, c.Cmd, cfg) + out, stderr, exitCode, truncated, err := runCommand(ctx, client, c.Cmd, cfg) if err != nil { if res.Failed == nil { res.Failed = make(map[string]string, 1) @@ -120,6 +127,8 @@ func inventoryHost(ctx context.Context, host string, pack *Pack, cfg execConfig) res.Failed[c.ID] = err.Error() continue } + res.CollectorExitCodes[c.ID] = exitCode + res.CollectorTruncated[c.ID] = truncated // A non-zero exit is not an error here: `dnf needs-restarting -r` // signals its answer through the exit code, and stderr is useful // context for the server-side parser either way. @@ -182,25 +191,30 @@ func classifyDialError(err error) string { } // runCommand executes one command and returns stdout and stderr, each capped -// at cfg.maxOutputBytes. A package list on a large host is big but bounded; -// an unbounded read here would let one host exhaust the forager's memory. -func runCommand(ctx context.Context, client *ssh.Client, cmd string, cfg execConfig) (string, string, error) { +// at cfg.maxOutputBytes. It also returns the command's exit code and whether +// either stream exceeded the cap. A package list on a large host is big but +// bounded; an unbounded read here would let one host exhaust the forager's +// memory. +func runCommand(ctx context.Context, client *ssh.Client, cmd string, cfg execConfig) (string, string, int, bool, error) { + if cfg.maxOutputBytes < 0 { + cfg.maxOutputBytes = 0 + } cmdCtx, cancel := context.WithTimeout(ctx, cfg.commandTimeout) defer cancel() session, err := client.NewSession() if err != nil { - return "", "", fmt.Errorf("new session: %w", err) + return "", "", 0, false, fmt.Errorf("new session: %w", err) } defer func() { _ = session.Close() }() stdoutPipe, err := session.StdoutPipe() if err != nil { - return "", "", fmt.Errorf("stdout pipe: %w", err) + return "", "", 0, false, fmt.Errorf("stdout pipe: %w", err) } stderrPipe, err := session.StderrPipe() if err != nil { - return "", "", fmt.Errorf("stderr pipe: %w", err) + return "", "", 0, false, fmt.Errorf("stderr pipe: %w", err) } // Commands come from a content pack read off local disk and verified @@ -208,14 +222,18 @@ func runCommand(ctx context.Context, client *ssh.Client, cmd string, cfg execCon // request selects a pack by version and cannot supply one, so no part of // an action's payload is executed here. if err := session.Start(cmd); err != nil { - return "", "", fmt.Errorf("start: %w", err) + return "", "", 0, false, fmt.Errorf("start: %w", err) } // stdout and stderr must be drained concurrently. Reading them in // sequence deadlocks whenever the remote fills one pipe's buffer while we // are still blocked on the other — `dnf repolist` writing warnings to // stderr ahead of its stdout is enough to trigger it. - type readResult struct{ stdout, stderr []byte } + type readResult struct { + stdout, stderr []byte + truncated bool + } + var readResultTruncated atomic.Bool readCh := make(chan readResult, 1) go func() { var out, errOut []byte @@ -223,18 +241,26 @@ func runCommand(ctx context.Context, client *ssh.Client, cmd string, cfg execCon readWG.Add(2) go func() { defer readWG.Done() - out, _ = io.ReadAll(io.LimitReader(stdoutPipe, int64(cfg.maxOutputBytes))) + out, _ = io.ReadAll(io.LimitReader(stdoutPipe, int64(cfg.maxOutputBytes)+1)) + if len(out) > cfg.maxOutputBytes { + out = out[:cfg.maxOutputBytes] + readResultTruncated.Store(true) + } // Discard the remainder so a host exceeding the cap cannot block // on a full pipe and hold the session open until timeout. _, _ = io.Copy(io.Discard, stdoutPipe) }() go func() { defer readWG.Done() - errOut, _ = io.ReadAll(io.LimitReader(stderrPipe, int64(cfg.maxOutputBytes))) + errOut, _ = io.ReadAll(io.LimitReader(stderrPipe, int64(cfg.maxOutputBytes)+1)) + if len(errOut) > cfg.maxOutputBytes { + errOut = errOut[:cfg.maxOutputBytes] + readResultTruncated.Store(true) + } _, _ = io.Copy(io.Discard, stderrPipe) }() readWG.Wait() - readCh <- readResult{out, errOut} + readCh <- readResult{out, errOut, readResultTruncated.Load()} }() waitCh := make(chan error, 1) @@ -245,21 +271,25 @@ func runCommand(ctx context.Context, client *ssh.Client, cmd string, cfg execCon case read = <-readCh: case <-cmdCtx.Done(): _ = session.Signal(ssh.SIGKILL) - return "", "", fmt.Errorf("command timed out") + return "", "", 0, false, fmt.Errorf("command timed out") } + exitCode := 0 select { case waitErr := <-waitCh: // Non-zero exit is reported through output, not as a Go error: // `dnf needs-restarting -r` answers through its exit code. - var exitErr *ssh.ExitError - if waitErr != nil && !errors.As(waitErr, &exitErr) { - return "", "", fmt.Errorf("wait: %w", waitErr) + if waitErr != nil { + var exitErr *ssh.ExitError + if !errors.As(waitErr, &exitErr) { + return "", "", 0, read.truncated, fmt.Errorf("wait: %w", waitErr) + } + exitCode = exitErr.ExitStatus() } case <-cmdCtx.Done(): _ = session.Signal(ssh.SIGKILL) - return "", "", fmt.Errorf("command timed out") + return "", "", 0, read.truncated, fmt.Errorf("command timed out") } - return string(read.stdout), string(read.stderr), nil + return string(read.stdout), string(read.stderr), exitCode, read.truncated, nil } diff --git a/pkg/proxy/discovery/executor_test.go b/pkg/proxy/discovery/executor_test.go index d472f40..3991841 100644 --- a/pkg/proxy/discovery/executor_test.go +++ b/pkg/proxy/discovery/executor_test.go @@ -3,6 +3,7 @@ package discovery import ( "context" "crypto/ed25519" + "encoding/json" "fmt" "io" "net" @@ -26,6 +27,10 @@ type fakeSSHServer struct { // responses maps a command to its canned stdout. responses map[string]string + // exitCodes maps a command to its remote exit status. Unspecified + // commands exit successfully. + exitCodes map[string]uint32 + // stderrFor maps a command to stderr written before its stdout, used to // exercise concurrent pipe draining. stderrFor map[string]string @@ -33,6 +38,9 @@ type fakeSSHServer struct { // delay is applied before answering, to exercise concurrency. delay time.Duration + // delayFor overrides delay for individual commands. + delayFor map[string]time.Duration + // rejectAuth makes the server refuse authentication. rejectAuth bool @@ -53,7 +61,7 @@ func newFakeSSHServer(t *testing.T, responses map[string]string) *fakeSSHServer t.Fatalf("creating signer: %v", err) } - s := &fakeSSHServer{t: t, responses: responses} + s := &fakeSSHServer{t: t, responses: responses, exitCodes: make(map[string]uint32)} s.config = &ssh.ServerConfig{ PasswordCallback: func(ssh.ConnMetadata, []byte) (*ssh.Permissions, error) { if s.rejectAuth { @@ -148,14 +156,18 @@ func (s *fakeSSHServer) handleSession(ch ssh.Channel, reqs <-chan *ssh.Request) continue } _ = req.Reply(true, nil) + cmd := string(req.Payload[4:]) done := s.trackExec() - if s.delay > 0 { - time.Sleep(s.delay) + commandDelay := s.delay + if d, ok := s.delayFor[cmd]; ok { + commandDelay = d + } + if commandDelay > 0 { + time.Sleep(commandDelay) } defer done() - cmd := string(req.Payload[4:]) out, known := s.responses[cmd] if !known { _, _ = io.WriteString(ch.Stderr(), "command not found\n") @@ -168,7 +180,7 @@ func (s *fakeSSHServer) handleSession(ch ssh.Channel, reqs <-chan *ssh.Request) _, _ = io.WriteString(ch.Stderr(), errOut) } _, _ = io.WriteString(ch, out) - _, _ = ch.SendRequest("exit-status", false, exitStatusPayload(0)) + _, _ = ch.SendRequest("exit-status", false, exitStatusPayload(s.exitCodes[cmd])) return } } @@ -247,6 +259,108 @@ func TestRunInventory_CollectsFromDebianHost(t *testing.T) { } } +func TestRunInventory_ReportsCollectorExitCode(t *testing.T) { + srv := newFakeSSHServer(t, map[string]string{ + factsProbe: ubuntuOSRelease + "\n---\nx86_64", + "dpkg-query -W": "updates available\n", + "cat /etc/os-release": ubuntuOSRelease, + }) + srv.exitCodes["dpkg-query -W"] = 100 + + r := runInventory(context.Background(), []string{"127.0.0.1"}, debianPack(t), testExecConfig(srv.port(), 2))[0] + if r.Status != StatusOK { + t.Fatalf("status = %s (%s), want ok", r.Status, r.Error) + } + if got := r.CollectorExitCodes["pkgs-dpkg"]; got != 100 { + t.Errorf("collector exit code = %d, want 100", got) + } + if got := r.Collected["pkgs-dpkg"]; got != "updates available\n" { + t.Errorf("collector output = %q, want output preserved", got) + } + encoded, err := json.Marshal(r) + if err != nil { + t.Fatalf("marshal target result: %v", err) + } + var wire map[string]json.RawMessage + if err := json.Unmarshal(encoded, &wire); err != nil { + t.Fatalf("unmarshal target result: %v", err) + } + for _, field := range []string{"collector_exit_codes", "collector_truncated", "collector_output_limit_bytes"} { + if _, ok := wire[field]; !ok { + t.Errorf("wire result missing %q", field) + } + } +} + +func TestRunInventory_ReportsOutputTruncationAtBoundary(t *testing.T) { + for _, tc := range []struct { + name string + outputLen int + truncated bool + }{ + {name: "under cap", outputLen: 999, truncated: false}, + {name: "at cap", outputLen: 1000, truncated: false}, + {name: "over cap", outputLen: 1001, truncated: true}, + } { + t.Run(tc.name, func(t *testing.T) { + srv := newFakeSSHServer(t, map[string]string{ + factsProbe: ubuntuOSRelease + "\n---\nx86_64", + "dpkg-query -W": strings.Repeat("x", tc.outputLen), + "cat /etc/os-release": ubuntuOSRelease, + }) + cfg := testExecConfig(srv.port(), 2) + cfg.maxOutputBytes = 1000 + + r := runInventory(context.Background(), []string{"127.0.0.1"}, debianPack(t), cfg)[0] + if r.Status != StatusOK { + t.Fatalf("status = %s (%s), want ok", r.Status, r.Error) + } + if got := r.CollectorTruncated["pkgs-dpkg"]; got != tc.truncated { + t.Errorf("collector truncated = %t, want %t", got, tc.truncated) + } + if got := len(r.Collected["pkgs-dpkg"]); got != min(tc.outputLen, cfg.maxOutputBytes) { + t.Errorf("collector output length = %d, want %d", got, min(tc.outputLen, cfg.maxOutputBytes)) + } + if got := r.CollectorOutputLimitBytes; got != cfg.maxOutputBytes { + t.Errorf("output limit = %d, want %d", got, cfg.maxOutputBytes) + } + }) + } +} + +func TestRunInventory_NegativeOutputCapDoesNotPanic(t *testing.T) { + srv := newFakeSSHServer(t, map[string]string{ + factsProbe: ubuntuOSRelease + "\n---\nx86_64", + "cat /etc/os-release": ubuntuOSRelease, + }) + cfg := testExecConfig(srv.port(), 2) + cfg.maxOutputBytes = -1 + + r := runInventory(context.Background(), []string{"127.0.0.1"}, debianPack(t), cfg)[0] + if r.Status != StatusOK { + t.Fatalf("status = %s (%s), want ok", r.Status, r.Error) + } +} + +func TestRunInventory_TimeoutIsNotAnExitCode(t *testing.T) { + srv := newFakeSSHServer(t, map[string]string{ + factsProbe: ubuntuOSRelease + "\n---\nx86_64", + "dpkg-query -W": "package output", + "cat /etc/os-release": ubuntuOSRelease, + }) + srv.delayFor = map[string]time.Duration{"dpkg-query -W": 200 * time.Millisecond} + cfg := testExecConfig(srv.port(), 2) + cfg.commandTimeout = 20 * time.Millisecond + + r := runInventory(context.Background(), []string{"127.0.0.1"}, debianPack(t), cfg)[0] + if _, ok := r.CollectorExitCodes["pkgs-dpkg"]; ok { + t.Fatal("timed-out collector reported an exit code") + } + if !strings.Contains(r.Failed["pkgs-dpkg"], "command timed out") { + t.Errorf("collector error = %q, want command timeout", r.Failed["pkgs-dpkg"]) + } +} + // Version strings carry epoch and release suffixes that backport-aware CVE // matching depends on; the agent must not normalize them. func TestRunInventory_PreservesVerbatimVersions(t *testing.T) {