Query piped nodes when counting with -R/-F (#22)

With piped input, `-C` skipped the query and counted the piped lines' second column, so `... | node-lookup -R -C` reported the upstream fact instead of each node's role.

- query each piped node for `-R`/`-F` with `-C` and count the results
- count piped lines directly only for bare `-C` with piped input; `-C` with `-j`/`-1`/`-2` still requires `-R`/`-F`
- validate flags before reading stdin so invalid combos never block on an open pipe
- add tests for piped count paths, rejected combos and validation ordering

Reviewed-on: #22
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>

.woodpecker/build.yaml

@@ -3,7 +3,19 @@ when:
 
 steps:
   - name: build
-    image: golang:1.25
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9
+    environment:
+      GOCACHE_S3_BUCKET: gocache
+      GOCACHE_S3_REGION: us-east-1
+      GOCACHE_S3_ENDPOINT_URL: https://s3.ceph.unkin.net
+      GOCACHE_S3_PATH_STYLE: "true"
+      GOCACHE_KEY_PREFIX: ci-node-lookup
+      GOCACHE_METRICS: "true"
+      GOCACHEPROG: go-cache-plugin --cache-dir=/tmp/gocache
+      AWS_ACCESS_KEY_ID:
+        from_secret: GOCACHE_AWS_ACCESS_KEY_ID
+      AWS_SECRET_ACCESS_KEY:
+        from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
     commands:
       - make build
     backend_options:

.woodpecker/pre-commit.yaml

@@ -3,7 +3,19 @@ when:
 
 steps:
   - name: pre-commit
-    image: git.unkin.net/unkin/almalinux9-gobuilder:20260606
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9
+    environment:
+      GOCACHE_S3_BUCKET: gocache
+      GOCACHE_S3_REGION: us-east-1
+      GOCACHE_S3_ENDPOINT_URL: https://s3.ceph.unkin.net
+      GOCACHE_S3_PATH_STYLE: "true"
+      GOCACHE_KEY_PREFIX: ci-node-lookup
+      GOCACHE_METRICS: "true"
+      GOCACHEPROG: go-cache-plugin --cache-dir=/tmp/gocache
+      AWS_ACCESS_KEY_ID:
+        from_secret: GOCACHE_AWS_ACCESS_KEY_ID
+      AWS_SECRET_ACCESS_KEY:
+        from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
     commands:
       - uvx pre-commit run --all-files
     backend_options:

.woodpecker/release.yaml

@@ -3,7 +3,19 @@ when:
 
 steps:
   - name: test
-    image: golang:1.25
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9
+    environment:
+      GOCACHE_S3_BUCKET: gocache
+      GOCACHE_S3_REGION: us-east-1
+      GOCACHE_S3_ENDPOINT_URL: https://s3.ceph.unkin.net
+      GOCACHE_S3_PATH_STYLE: "true"
+      GOCACHE_KEY_PREFIX: ci-node-lookup
+      GOCACHE_METRICS: "true"
+      GOCACHEPROG: go-cache-plugin --cache-dir=/tmp/gocache
+      AWS_ACCESS_KEY_ID:
+        from_secret: GOCACHE_AWS_ACCESS_KEY_ID
+      AWS_SECRET_ACCESS_KEY:
+        from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
     commands:
       - go test -race ./...
     backend_options:
@@ -21,7 +33,19 @@ steps:
   # cross-platform binaries attached to the Gitea release. Each tool is a
   # separate main package, so they are built individually per os/arch.
   - name: build
-    image: git.unkin.net/unkin/almalinux9-gobuilder:20260606
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9
+    environment:
+      GOCACHE_S3_BUCKET: gocache
+      GOCACHE_S3_REGION: us-east-1
+      GOCACHE_S3_ENDPOINT_URL: https://s3.ceph.unkin.net
+      GOCACHE_S3_PATH_STYLE: "true"
+      GOCACHE_KEY_PREFIX: ci-node-lookup
+      GOCACHE_METRICS: "true"
+      GOCACHEPROG: go-cache-plugin --cache-dir=/tmp/gocache
+      AWS_ACCESS_KEY_ID:
+        from_secret: GOCACHE_AWS_ACCESS_KEY_ID
+      AWS_SECRET_ACCESS_KEY:
+        from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
     commands:
       - make build VERSION=${CI_COMMIT_TAG}
       # Shell variables/expansions are escaped as $$ so Woodpecker leaves them
@@ -51,7 +75,7 @@ steps:
 
   # Package the built binary + generated shell completions into an RPM.
   - name: package
-    image: git.unkin.net/unkin/almalinux9-rpmbuilder:latest
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/rpmbuilder:0.1.0-alma9
     commands:
       - ./scripts/build-rpm.sh ${CI_COMMIT_TAG}
     depends_on: [build]

.woodpecker/test.yaml

@@ -18,7 +18,19 @@ steps:
             cpu: 2
 
   - name: test
-    image: golang:1.25
+    image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9
+    environment:
+      GOCACHE_S3_BUCKET: gocache
+      GOCACHE_S3_REGION: us-east-1
+      GOCACHE_S3_ENDPOINT_URL: https://s3.ceph.unkin.net
+      GOCACHE_S3_PATH_STYLE: "true"
+      GOCACHE_KEY_PREFIX: ci-node-lookup
+      GOCACHE_METRICS: "true"
+      GOCACHEPROG: go-cache-plugin --cache-dir=/tmp/gocache
+      AWS_ACCESS_KEY_ID:
+        from_secret: GOCACHE_AWS_ACCESS_KEY_ID
+      AWS_SECRET_ACCESS_KEY:
+        from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
     commands:
       - go test -v -race ./...
     backend_options:

main.go

@@ -357,15 +357,38 @@ func run(cfg config, nodeName, factName, match string, showRole, partial, invers
 		return nil
 	}
 
-	if (nodeOnly || valueOnly || count || ansible) && !showRole && factName == "" {
+	hasQuery := showRole || factName != ""
-		return fmt.Errorf("-R or -F must be used with -1, -2, -C, or -A")
+	// Bare -C counts already-looked-up piped lines; only it needs stdin to validate.
+	bareCount := count && !hasQuery && !nodeOnly && !valueOnly && !ansible && !jsonMode && nodeName == ""
+	countErr := fmt.Errorf("-R or -F must be used with -1, -2, -C, or -A")
+
+	if (nodeOnly || valueOnly || (count && !bareCount) || ansible) && !hasQuery {
+		return countErr
 	}
-	if (match != "" || partial || inverse) && !showRole && factName == "" {
+	if (match != "" || partial || inverse) && !hasQuery {
 		return fmt.Errorf("-R or -F must be used with -m, -p, or -i")
 	}
 
+	var reader *bufio.Reader
+	piped := false
+	if nodeName == "" {
+		reader, piped = stdinReader(os.Stdin)
+	}
+
+	if bareCount {
+		if !piped {
+			return countErr
+		}
+		var lines []string
+		scanner := bufio.NewScanner(reader)
+		for scanner.Scan() {
+			lines = append(lines, scanner.Text())
+		}
+		fmt.Println(strings.Join(countResults(lines), "\n"))
+		return nil
+	}
+
 	var collected []fact
-	var stdinLines []string
 
 	doQuery := func(node string) error {
 		query := buildQuery(node, factName, match, cfg.RoleFact, showRole, partial, inverse)
@@ -377,21 +400,15 @@ func run(cfg config, nodeName, factName, match string, showRole, partial, invers
 		return nil
 	}
 
-	if reader, ok := stdinReader(os.Stdin); ok && nodeName == "" {
+	if piped {
 		scanner := bufio.NewScanner(reader)
-		if count {
+		for scanner.Scan() {
-			for scanner.Scan() {
+			fields := strings.Fields(scanner.Text())
-				stdinLines = append(stdinLines, scanner.Text())
+			if len(fields) == 0 {
+				continue
 			}
-		} else {
+			if err := doQuery(qualifyNode(fields[0], cfg.Domain)); err != nil {
-			for scanner.Scan() {
+				fmt.Fprintln(os.Stderr, "error:", err)
-				fields := strings.Fields(scanner.Text())
-				if len(fields) == 0 {
-					continue
-				}
-				if err := doQuery(qualifyNode(fields[0], cfg.Domain)); err != nil {
-					fmt.Fprintln(os.Stderr, "error:", err)
-				}
 			}
 		}
 	} else {
@@ -410,11 +427,7 @@ func run(cfg config, nodeName, factName, match string, showRole, partial, invers
 		_ = enc.Encode(factsByHost(collected, factName, cfg.RoleFact, showRole))
 
 	case count:
-		values := stdinLines
+		fmt.Println(strings.Join(countResults(returnData), "\n"))
-		if len(values) == 0 {
-			values = returnData
-		}
-		fmt.Println(strings.Join(countResults(values), "\n"))
 
 	case ansible:
 		// Attach each host's queried fact(s) as inventory host vars, e.g.

main_test.go

@@ -8,7 +8,9 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"gopkg.in/yaml.v3"
 )
@@ -872,3 +874,151 @@ func TestRun_AllFacts_PrintsSortedByName(t *testing.T) {
 		t.Fatalf("facts not sorted by name: %v", lines)
 	}
 }
+
+// ---- run: count with piped stdin --------------------------------------------
+
+func withStdin(t *testing.T, content string) {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "stdin")
+	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	f, err := os.Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	old := os.Stdin
+	os.Stdin = f
+	t.Cleanup(func() { os.Stdin = old; _ = f.Close() })
+}
+
+const pipedLookup = "hosta.main.unkin.net puppetdbapi.service.consul\nhostb.main.unkin.net puppetdbapi.service.consul\n"
+
+func TestRun_CountPipedWithRoleQueriesEachNode(t *testing.T) {
+	withStdin(t, pipedLookup)
+	var queried atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		queried.Add(1)
+		_ = json.NewEncoder(w).Encode([]fact{{Certname: "h", Name: "enc_role", Value: rawJSON("roles::vault")}})
+	}))
+	t.Cleanup(srv.Close)
+	cfg := config{PuppetDBURL: srv.URL, RoleFact: "enc_role"}
+
+	out := strings.TrimSpace(captureStdout(t, func() {
+		if err := run(cfg, "", "", "", true, false, false, false, false, true, false, false, false); err != nil {
+			t.Fatalf("run returned error: %v", err)
+		}
+	}))
+	if out != "2: roles::vault" {
+		t.Fatalf("expected role counts of piped nodes, got: %q", out)
+	}
+	if n := queried.Load(); n != 2 {
+		t.Fatalf("expected one query per piped node, got %d", n)
+	}
+}
+
+func TestRun_CountPipedWithoutQueryCountsInput(t *testing.T) {
+	withStdin(t, pipedLookup+"hostc.main.unkin.net other\n")
+	cfg := config{PuppetDBURL: "http://unused", RoleFact: "enc_role"}
+
+	out := strings.TrimSpace(captureStdout(t, func() {
+		if err := run(cfg, "", "", "", false, false, false, false, false, true, false, false, false); err != nil {
+			t.Fatalf("run returned error: %v", err)
+		}
+	}))
+	if out != "1: other\n2: puppetdbapi.service.consul" {
+		t.Fatalf("expected counts of piped values, got: %q", out)
+	}
+}
+
+func TestRun_CountWithoutStdinRequiresRoleOrFact(t *testing.T) {
+	devnull, err := os.Open(os.DevNull)
+	if err != nil {
+		t.Fatal(err)
+	}
+	old := os.Stdin
+	os.Stdin = devnull
+	t.Cleanup(func() { os.Stdin = old; _ = devnull.Close() })
+
+	err = run(config{PuppetDBURL: "http://unused"}, "", "", "", false, false, false, false, false, true, false, false, false)
+	if err == nil || !strings.Contains(err.Error(), "-R or -F") {
+		t.Fatalf("expected -R/-F requirement error, got: %v", err)
+	}
+}
+
+func TestRun_CountPipedWithFactQueriesEachNode(t *testing.T) {
+	withStdin(t, pipedLookup)
+	var queried atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.Contains(r.URL.Query().Get("query"), "osfamily") {
+			t.Errorf("expected osfamily query, got %q", r.URL.Query().Get("query"))
+		}
+		queried.Add(1)
+		_ = json.NewEncoder(w).Encode([]fact{{Certname: "h", Name: "osfamily", Value: rawJSON("RedHat")}})
+	}))
+	t.Cleanup(srv.Close)
+	cfg := config{PuppetDBURL: srv.URL, RoleFact: "enc_role"}
+
+	out := strings.TrimSpace(captureStdout(t, func() {
+		if err := run(cfg, "", "osfamily", "", false, false, false, false, false, true, false, false, false); err != nil {
+			t.Fatalf("run returned error: %v", err)
+		}
+	}))
+	if out != "2: RedHat" {
+		t.Fatalf("expected fact counts of piped nodes, got: %q", out)
+	}
+	if n := queried.Load(); n != 2 {
+		t.Fatalf("expected one query per piped node, got %d", n)
+	}
+}
+
+func TestRun_CountPipedWithOtherOutputFlagsRequiresRoleOrFact(t *testing.T) {
+	for name, mutate := range map[string]func(*runArgs){
+		"json":      func(a *runArgs) { a.jsonMode = true },
+		"nodeOnly":  func(a *runArgs) { a.nodeOnly = true },
+		"valueOnly": func(a *runArgs) { a.valueOnly = true },
+	} {
+		t.Run(name, func(t *testing.T) {
+			withStdin(t, pipedLookup)
+			a := runArgs{cfg: config{PuppetDBURL: "http://unused"}, count: true}
+			mutate(&a)
+			var err error
+			out := captureStdout(t, func() {
+				err = run(a.cfg, "", "", "", false, false, false, a.nodeOnly, a.valueOnly, true, false, a.jsonMode, false)
+			})
+			if err == nil || !strings.Contains(err.Error(), "-R or -F") {
+				t.Fatalf("expected -R/-F requirement error, got: %v (output %q)", err, out)
+			}
+		})
+	}
+}
+
+func TestRun_InvalidFlagsErrorWithoutReadingOpenPipe(t *testing.T) {
+	for name, a := range map[string]runArgs{
+		"match":     {match: "foo"},
+		"countJSON": {count: true, jsonMode: true},
+	} {
+		t.Run(name, func(t *testing.T) {
+			r, w, err := os.Pipe()
+			if err != nil {
+				t.Fatal(err)
+			}
+			old := os.Stdin
+			os.Stdin = r
+			t.Cleanup(func() { os.Stdin = old; _ = w.Close(); _ = r.Close() })
+
+			done := make(chan error, 1)
+			go func() {
+				done <- run(config{PuppetDBURL: "http://unused"}, "", "", a.match, false, false, false, false, false, a.count, false, a.jsonMode, false)
+			}()
+			select {
+			case err := <-done:
+				if err == nil || !strings.Contains(err.Error(), "-R or -F") {
+					t.Fatalf("expected -R/-F requirement error, got: %v", err)
+				}
+			case <-time.After(2 * time.Second):
+				t.Fatal("run blocked on an open, empty stdin pipe before validating flags")
+			}
+		})
+	}
+}