orca/internal/cli/logs.go

package cli

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"iter"
	"log/slog"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/spf13/cobra"

	"git.cloudinit.dev/coreci/orca/internal/certpaths"
	"git.cloudinit.dev/coreci/orca/internal/model"
	"git.cloudinit.dev/coreci/orca/internal/sshpush"
	"git.cloudinit.dev/coreci/orca/internal/store"
)

// logsExecer is the SSH command-execution seam used by `orca logs`.
// *sshpush.Transport satisfies it via Exec; tests inject a mock
// without a real SSH server (same pattern as drain.go / stepca mockExec).
type logsExecer interface {
	Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
}

// logsExecOverride is the package-level exec seam. When non-nil it
// replaces the production transport; tests set it and restore nil in
// cleanup. nil means "build the real transport on first use".
var logsExecOverride logsExecer

func logsExecFromCtx(_ context.Context) (logsExecer, error) {
	if logsExecOverride != nil {
		return logsExecOverride, nil
	}
	keyPath := certpaths.SSHKeyPath()
	khPath := certpaths.KnownHostsPath()
	return sshpush.NewTransport(keyPath, khPath), nil
}

// LogLine is a single journald log entry parsed from journalctl --output
// json. Host is the peer the line came from (set by the aggregator).
type LogLine struct {
	Host      string    `json:"host"`
	Timestamp time.Time `json:"timestamp"`
	Unit      string    `json:"unit"`
	Message   string    `json:"message"`
	Priority  string    `json:"priority"`
}

// journalRaw is the subset of journalctl --output json fields we
// decode. Extra fields are ignored.
type journalRaw struct {
	Realtime   int64  `json:"__REALTIME_TIMESTAMP"`
	Unit       string `json:"_SYSTEMD_UNIT"`
	Identifier string `json:"SYSLOG_IDENTIFIER"`
	Comm       string `json:"_COMM"`
	Message    string `json:"MESSAGE"`
	Priority   any    `json:"PRIORITY"`
}

func (j journalRaw) unit() string {
	if j.Unit != "" {
		return strings.TrimSuffix(j.Unit, ".service")
	}
	if j.Identifier != "" {
		return j.Identifier
	}
	if j.Comm != "" {
		return j.Comm
	}
	return ""
}

func (j journalRaw) priority() string {
	switch p := j.Priority.(type) {
	case string:
		return p
	case float64:
		return fmt.Sprintf("%v", int(p))
	default:
		return ""
	}
}

func (j journalRaw) timestamp() time.Time {
	if j.Realtime == 0 {
		return time.Time{}
	}
	return time.Unix(0, j.Realtime).UTC()
}

var (
	logsAllNodes bool
	logsNode     string
	logsJob      string
	logsSince    string
	logsJSON     bool
)

var logsCmd = &cobra.Command{
	Use:   "logs",
	Short: "Aggregate journald logs across nodes (REQ-117)",
	Long: `Aggregate journald logs across registered nodes via SSH fanout.

  orca logs --all-nodes --since 5m
  orca logs --node web-1 --since 1h
  orca logs --all-nodes --job web --since 30m --json

Runs 'journalctl -u 'orca-alloc-*' --since <dur> --output json' on each
peer, parses the JSON-per-line output, and streams the entries with a
[<hostname>] prefix (multi-node) or raw (single-node). --json outputs
the raw journalctl JSON lines verbatim.

Use --all-nodes to fan out to every registered node, or --node <host>
for a single node. --job <name> filters units to orca-alloc-<name>-*.

Ctrl-C cancels the fan-out via signal.NotifyContext.`,
	RunE: func(cmd *cobra.Command, args []string) error {
		if !logsAllNodes && logsNode == "" {
			return fmt.Errorf("specify --all-nodes or --node <host>")
		}
		if logsAllNodes && logsNode != "" {
			return fmt.Errorf("--all-nodes and --node are mutually exclusive")
		}
		// F1: validate --job before interpolation into the journalctl
		// unit pattern. Go's %q does not escape backticks and bash
		// executes command substitution inside double quotes, so an
		// unvalidated job name is a remote RCE vector.
		if logsJob != "" && !validSafeName(logsJob) {
			return fmt.Errorf("logs: --job %q contains disallowed characters (allowed: A-Z a-z 0-9 _ -)", logsJob)
		}
		since, err := parseSince(logsSince)
		if err != nil {
			return err
		}

		ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
		defer cancel()

		nodes, err := resolveLogNodes(ctx)
		if err != nil {
			return err
		}
		if len(nodes) == 0 {
			return fmt.Errorf("no nodes to query")
		}

		ex, err := logsExecFromCtx(cmd.Context())
		if err != nil {
			return fmt.Errorf("ssh transport: %w", err)
		}

		out := cmd.OutOrStdout()
		multi := len(nodes) > 1
		for line := range streamLogs(ctx, ex, nodes, since, logsJob) {
			if logsJSON {
				raw, _ := json.Marshal(line)
				fmt.Fprintln(out, string(raw))
				continue
			}
			if multi {
				fmt.Fprintf(out, "[%s] %s %s\n", line.Host, line.Timestamp.Format(time.RFC3339), line.Message)
			} else {
				fmt.Fprintln(out, line.Message)
			}
		}
		return nil
	},
}

// parseSince parses a duration string like "5m", "1h30m", "500ms". The
// returned time is time.Now().UTC().Add(-d). An empty string defaults
// to 5 minutes.
func parseSince(s string) (time.Time, error) {
	if s == "" {
		s = "5m"
	}
	d, err := time.ParseDuration(s)
	if err != nil {
		return time.Time{}, fmt.Errorf("--since %q: %w", s, err)
	}
	if d <= 0 {
		return time.Time{}, fmt.Errorf("--since must be positive, got %s", d)
	}
	return time.Now().UTC().Add(-d), nil
}

// resolveLogNodes returns the set of nodes to query. For --all-nodes it
// lists every registered node; for --node <host> it resolves a single
// node by name or id.
func resolveLogNodes(ctx context.Context) ([]*model.Node, error) {
	db, closer, err := openDB()
	if err != nil {
		return nil, err
	}
	defer closer()
	reg := store.NewNodeRepo(db)
	if logsAllNodes {
		return reg.List(ctx)
	}
	n, err := reg.GetByName(ctx, logsNode)
	if err == nil {
		return []*model.Node{n}, nil
	}
	if !errors.Is(err, store.ErrNotFound) {
		return nil, fmt.Errorf("lookup node %q: %w", logsNode, err)
	}
	n, err = reg.Get(ctx, logsNode)
	if err == nil {
		return []*model.Node{n}, nil
	}
	if errors.Is(err, store.ErrNotFound) {
		return nil, fmt.Errorf("node %q not found", logsNode)
	}
	return nil, fmt.Errorf("lookup node %q: %w", logsNode, err)
}

// streamLogs fans out journalctl across nodes and yields parsed
// LogLine values as they arrive. It runs each node's exec in its own
// goroutine, scans the output line-by-line, and yields each parsed
// JSON entry immediately. The stream ends when every node has
// completed (or the context is cancelled). The caller drives the
// iteration via range-over-func (D-017 iter.Seq pattern).
func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since time.Time, job string) iter.Seq[LogLine] {
	return func(yield func(LogLine) bool) {
		merged := make(chan LogLine)
		var wg sync.WaitGroup
		for _, n := range nodes {
			wg.Add(1)
			go func(n *model.Node) {
				defer wg.Done()
				streamNodeLines(ctx, ex, n, since, job, merged)
			}(n)
		}
		done := make(chan struct{})
		go func() {
			wg.Wait()
			close(done)
		}()

		// Pump merged lines to the yield function until either all
		// nodes finish or the consumer stops pulling (yield==false)
		// or the context is cancelled.
		for {
			select {
			case <-done:
				return
			case <-ctx.Done():
				return
			case line := <-merged:
				if !yield(line) {
					return
				}
			}
		}
	}
}

// streamNodeLines runs journalctl on a single node and pushes each
// parsed line into out. It blocks until the exec completes (or the
// context is cancelled); the caller is responsible for waiting on the
// goroutine. Send is non-blocking via select on ctx.Done so a slow
// consumer does not stall the fanout forever.
func streamNodeLines(ctx context.Context, ex logsExecer, n *model.Node, since time.Time, job string, out chan<- LogLine) {
	peer := peerAddrForNode(n)
	if peer == "" {
		slog.Default().Warn("logs: cannot resolve SSH address for node", "node", n.Name)
		return
	}
	unitPattern := "orca-alloc-*"
	if job != "" {
		unitPattern = "orca-alloc-" + job + "-*"
	}
	sinceStr := since.Format("2006-01-02 15:04:05")
	// F1: shellQuote (single-quote wrap) instead of %q — %q does not
	// escape backticks, enabling command substitution in double quotes.
	cmd := fmt.Sprintf("journalctl -u %s --since %s --output json --no-pager", shellQuote(unitPattern), shellQuote(sinceStr))
	raw, err := ex.Exec(ctx, peer, cmd)
	if err != nil {
		slog.Default().Warn("logs: exec failed", "node", n.Name, "peer", peer, "error", err)
		return
	}
	host := n.Name
	for _, line := range strings.Split(string(raw), "\n") {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		ll, perr := parseJournalLine(line)
		if perr != nil {
			continue
		}
		ll.Host = host
		select {
		case out <- ll:
		case <-ctx.Done():
			return
		}
	}
}

// parseJournalLine decodes a single journalctl --output json line into
// a LogLine. Unknown fields are ignored.
func parseJournalLine(s string) (LogLine, error) {
	var j journalRaw
	if err := json.Unmarshal([]byte(s), &j); err != nil {
		return LogLine{}, fmt.Errorf("parse journal line: %w", err)
	}
	return LogLine{
		Timestamp: j.timestamp(),
		Unit:      j.unit(),
		Message:   j.Message,
		Priority:  j.priority(),
	}, nil
}

func init() {
	logsCmd.Flags().BoolVar(&logsAllNodes, "all-nodes", false, "fan out to all registered nodes")
	logsCmd.Flags().StringVar(&logsNode, "node", "", "restrict to a single node (name or id)")
	logsCmd.Flags().StringVar(&logsJob, "job", "", "filter by job name (matches orca-alloc-<name>-* units)")
	logsCmd.Flags().StringVar(&logsSince, "since", "5m", "duration lookback (e.g. 5m, 1h, 30m); default 5m")
	logsCmd.Flags().BoolVar(&logsJSON, "json", false, "output raw JSON (one LogLine per line)")
	rootCmd.AddCommand(logsCmd)
}