feat: DTail fork — server/client feature development

Squashed development of the snonux/dtail fork's product code (internal/, cmd/)
since diverging from mimecast/dtail. Major areas:

- Read/output path: the former "turbo" channel-less path is now the single,
  default server-side read/output path for cat/grep/tail and MapReduce; the old
  channel-based path and its config/env toggles were removed.
- MapReduce: single aggregate implementation (server + serverless) fed directly
  by a processor pipeline, with input-exhausted finalization via the shutdown
  coordinator; high-concurrency and data-race fixes.
- Journal source reads (journal:unit.service) via journalctl, Linux-gated behind
  a journal-v1 capability.
- Auth-key fast reconnect: in-memory per-user public-key cache with TTL/max-keys,
  registered over an authenticated session (AUTHKEY), checked before
  authorized_keys.
- Interactive query reload (--interactive-query) with SESSION START/UPDATE
  generation boundaries and capability negotiation.
- Client-side deadlines: --timeout / --shutdownAfter as context deadlines;
  follow shutdown handling.
- Client logging: diagnostics-only daily log by default, opt-in payload tee via
  --log-payload.
- Numerous correctness fixes (buffer-pool double-recycle races, EOF-sentinel
  leaks, glob-expansion cap, TOCTOU in CSV parsing) with accompanying unit tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

1 files changed, 176 insertions, 93 deletions

diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go
index 440cb91..7ef3705 100644
--- a/internal/clients/maprclient.go
+++ b/internal/clients/maprclient.go
@@ -4,15 +4,17 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
+	"os"
 	"runtime"
-	"strings"
 	"time"
 
 	"github.com/mimecast/dtail/internal/clients/handlers"
-	"github.com/mimecast/dtail/internal/color"
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/dlog"
 	"github.com/mimecast/dtail/internal/mapr"
+	maprclient "github.com/mimecast/dtail/internal/mapr/client"
+	"github.com/mimecast/dtail/internal/mapr/logformat"
 	"github.com/mimecast/dtail/internal/omode"
 )
 
@@ -31,14 +33,10 @@ const (
 // MaprClient is used for running mapreduce aggregations on remote files.
 type MaprClient struct {
 	baseClient
-	// Global group set for merged mapr aggregation results
-	globalGroup *mapr.GlobalGroupSet
-	// The query object (constructed from queryStr)
-	query *mapr.Query
-	// Additative result or new result every interval run?
-	cumulative bool
-	// The last result string received
-	lastResult string
+	// Shared mapreduce state for all handlers and reporting paths.
+	session *maprclient.SessionState
+	// Selected cumulative reporting mode.
+	mode MaprClientMode
 }
 
 // NewMaprClient returns a new mapreduce client.
@@ -52,44 +50,33 @@ func NewMaprClient(args config.Args, maprClientMode MaprClientMode) (*MaprClient
 		dlog.Client.FatalPanic(args.QueryStr, "Can't parse mapr query", err)
 	}
 
+	// Warn once, at plan time, about $-variables the selected parser cannot
+	// populate. This runs in the user's client process for both server and
+	// serverless mode, so the warning reaches the user's stderr even though the
+	// actual field resolution happens server-side.
+	warnUnknownQueryVariables(os.Stderr, query)
+
 	// Don't retry connection if in tail mode and no outfile specified.
 	retry := args.Mode == omode.TailClient && !query.HasOutfile()
 
-	var cumulative bool
-	switch maprClientMode {
-	case CumulativeMode:
-		cumulative = true
-	case NonCumulativeMode:
-		cumulative = false
-	default:
-		// Result is comulative if we are in MapClient mode or with outfile
-		cumulative = args.Mode == omode.MapClient || query.HasOutfile()
-	}
-
-	dlog.Client.Debug("Cumulative mapreduce mode?", cumulative)
-
 	c := MaprClient{
 		baseClient: baseClient{
+			mu:         newBaseClientMu(),
 			Args:       args,
 			throttleCh: make(chan struct{}, args.ConnectionsPerCPU*runtime.NumCPU()),
 			retry:      retry,
+			runtime:    newClientRuntimeBoundary(config.CurrentRuntime()),
 		},
-		query:      query,
-		cumulative: cumulative,
+		session: maprclient.NewSessionState(query),
+		mode:    maprClientMode,
 	}
+	dlog.Client.Debug("Cumulative mapreduce mode?", c.isCumulative(query))
 
-	switch c.query.Table {
-	case "", ".":
-		c.RegexStr = "."
-	case "*":
-		c.RegexStr = "\\|MAPREDUCE:\\|"
-	default:
-		c.RegexStr = fmt.Sprintf("\\|MAPREDUCE:%s\\|", c.query.Table)
-	}
-
-	c.globalGroup = mapr.NewGlobalGroupSet()
+	c.setRegexForQuery(query)
 	c.baseClient.init()
-	c.baseClient.makeConnections(c)
+	if err := c.baseClient.makeConnections(&c); err != nil {
+		return nil, err
+	}
 
 	return &c, nil
 }
@@ -99,9 +86,14 @@ func (c *MaprClient) Start(ctx context.Context, statsCh <-chan string) (status i
 	go c.periodicReportResults(ctx)
 
 	status = c.baseClient.Start(ctx, statsCh)
-	if c.cumulative {
-		dlog.Client.Debug("Received final mapreduce result")
-		c.reportResults(true)
+
+	// Always write final result for cumulative mode (includes outfile case)
+	if snapshot := c.session.Snapshot(); c.isCumulative(snapshot.Query) {
+		dlog.Client.Debug("Writing final mapreduce result")
+		if err := c.reportResults(true); err != nil {
+			dlog.Client.Error("Unable to write final mapreduce result", err)
+		}
+		dlog.Client.Debug("Final result written")
 	}
 
 	return
@@ -109,96 +101,121 @@ func (c *MaprClient) Start(ctx context.Context, statsCh <-chan string) (status i
 
 // NEXT: Make this a callback function rather trying to use polymorphism to call
 // this. This applies to all clients. It will make the code easier to read.
-func (c MaprClient) makeHandler(server string) handlers.Handler {
-	return handlers.NewMaprHandler(server, c.query, c.globalGroup)
+func (c *MaprClient) makeHandler(server string) handlers.Handler {
+	return handlers.NewMaprHandler(server, c.session)
 }
 
-func (c MaprClient) makeCommands() (commands []string) {
-	commands = append(commands, fmt.Sprintf("map %s", c.query.RawQuery))
-	modeStr := "cat"
-	if c.Mode == omode.TailClient {
-		modeStr = "tail"
+func (c *MaprClient) makeSessionSpec() (SessionSpec, error) {
+	sessionSpec := NewSessionSpec(c.Args)
+	if snapshot := c.session.Snapshot(); snapshot.Query != nil {
+		sessionSpec.Query = snapshot.Query.RawQuery
 	}
+	return sessionSpec, nil
+}
 
-	for _, file := range strings.Split(c.What, ",") {
-		regex, err := c.Regex.Serialize()
-		if err != nil {
-			dlog.Client.FatalPanic(err)
-		}
-		if c.Timeout > 0 {
-			commands = append(commands, fmt.Sprintf("timeout %d %s %s %s", c.Timeout,
-				modeStr, file, regex))
-			continue
-		}
-		commands = append(commands, fmt.Sprintf("%s:%s %s %s",
-			modeStr, c.Args.SerializeOptions(), file, regex))
+func (c *MaprClient) makeCommands() (commands []string) {
+	sessionSpec, err := c.makeSessionSpec()
+	if err != nil {
+		dlog.Client.FatalPanic("unable to build map session spec", err)
 	}
-	return
+	commands, err = sessionSpec.Commands()
+	if err != nil {
+		dlog.Client.FatalPanic("unable to build map commands from session spec", err)
+	}
+	return commands
 }
 
 func (c *MaprClient) periodicReportResults(ctx context.Context) {
-	rampUpSleep := c.query.Interval / 2
-	dlog.Client.Debug("Ramp up sleeping before processing mapreduce results", rampUpSleep)
-	time.Sleep(rampUpSleep)
+	var (
+		lastGeneration uint64
+		seenGeneration bool
+	)
 
 	for {
+		snapshot := c.session.Snapshot()
+		rampUp := !seenGeneration || snapshot.Generation != lastGeneration
+		lastGeneration = snapshot.Generation
+		seenGeneration = true
+
+		delay := c.reportDelay(snapshot.Query, rampUp)
+		dlog.Client.Debug("Sleeping before processing mapreduce results", "generation", snapshot.Generation, "delay", delay)
+
+		timer := time.NewTimer(delay)
 		select {
-		case <-time.After(c.query.Interval):
+		case <-timer.C:
 			dlog.Client.Debug("Gathering interim mapreduce result")
-			c.reportResults(false)
+			if err := c.reportResults(false); err != nil {
+				dlog.Client.Error("Unable to gather mapreduce result", err)
+			}
+		case <-c.session.Changes():
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:
+				}
+			}
+			dlog.Client.Debug("Mapreduce query generation changed, recalculating report interval")
 		case <-ctx.Done():
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:
+				}
+			}
 			return
 		}
 	}
 }
 
-func (c *MaprClient) reportResults(finalResult bool) {
-	if c.query.HasOutfile() {
-		c.writeResultsToOutfile(finalResult)
-		return
+func (c *MaprClient) reportResults(finalResult bool) error {
+	snapshot := c.session.Snapshot()
+	if snapshot.Query == nil || snapshot.GlobalGroup == nil {
+		return nil
 	}
-	c.printResults()
+
+	if snapshot.Query.HasOutfile() {
+		return c.writeResultsToOutfile(snapshot, finalResult)
+	}
+	return c.printResults(snapshot)
 }
 
-func (c *MaprClient) printResults() {
+func (c *MaprClient) printResults(snapshot maprclient.SessionSnapshot) error {
 	var result string
 	var err error
 	var numRows int
 	rowsLimit := -1
 
-	if c.query.Limit == -1 {
+	if snapshot.Query.Limit == -1 {
 		// Limit output to 10 rows when the result is printed to stdout.
 		// This can be overriden with the limit clause though.
 		rowsLimit = 10
 	}
 
-	if c.cumulative {
-		result, numRows, err = c.globalGroup.Result(c.query, rowsLimit)
+	if c.isCumulative(snapshot.Query) {
+		result, numRows, err = snapshot.GlobalGroup.Result(snapshot.Query, rowsLimit, c.runtime.output.MaprResultRenderer())
 	} else {
-		result, numRows, err = c.globalGroup.SwapOut().Result(c.query, rowsLimit)
+		result, numRows, err = snapshot.GlobalGroup.SwapOut().Result(snapshot.Query, rowsLimit, c.runtime.output.MaprResultRenderer())
 	}
 	if err != nil {
-		dlog.Client.FatalPanic(err)
+		return fmt.Errorf("unable to render mapreduce result: %w", err)
 	}
 
-	if result == c.lastResult {
+	changed, ok := c.session.CommitRenderedResult(snapshot.Generation, result)
+	if !ok {
+		dlog.Client.Debug("Discarding stale mapreduce result", "generation", snapshot.Generation)
+		return nil
+	}
+	if !changed {
 		dlog.Client.Debug("Result hasn't changed compared to last time...")
-		return
+		return nil
 	}
-	c.lastResult = result
 
 	if numRows == 0 {
 		dlog.Client.Debug("Empty result set this time...")
-		return
+		return nil
 	}
 
-	rawQuery := c.query.RawQuery
-	if config.Client.TermColorsEnable {
-		rawQuery = color.PaintStrWithAttr(rawQuery,
-			config.Client.TermColors.MaprTable.RawQueryFg,
-			config.Client.TermColors.MaprTable.RawQueryBg,
-			config.Client.TermColors.MaprTable.RawQueryAttr)
-	}
+	rawQuery := c.runtime.output.PaintMaprRawQuery(snapshot.Query.RawQuery)
 	dlog.Client.Raw(fmt.Sprintf("%s\n", rawQuery))
 
 	if rowsLimit > 0 && numRows > rowsLimit {
@@ -206,16 +223,82 @@ func (c *MaprClient) printResults() {
 			"to %d rows! Use 'limit' clause to override!", numRows, rowsLimit))
 	}
 	dlog.Client.Raw(fmt.Sprintf("%s\n", result))
+	return nil
 }
 
-func (c *MaprClient) writeResultsToOutfile(finalResult bool) {
-	if c.cumulative {
-		if err := c.globalGroup.WriteResult(c.query, finalResult); err != nil {
-			dlog.Client.FatalPanic(err)
+func (c *MaprClient) writeResultsToOutfile(snapshot maprclient.SessionSnapshot, finalResult bool) error {
+	cumulative := c.isCumulative(snapshot.Query)
+	dlog.Client.Debug("writeResultsToOutfile called", "finalResult", finalResult, "cumulative", cumulative, "generation", snapshot.Generation)
+	if cumulative {
+		if err := snapshot.GlobalGroup.WriteResult(snapshot.Query, finalResult); err != nil {
+			return fmt.Errorf("unable to write cumulative mapreduce result: %w", err)
 		}
-		return
+		dlog.Client.Debug("WriteResult completed for cumulative mode")
+		return nil
+	}
+	if err := snapshot.GlobalGroup.SwapOut().WriteResult(snapshot.Query, true); err != nil {
+		return fmt.Errorf("unable to write non-cumulative mapreduce result: %w", err)
 	}
-	if err := c.globalGroup.SwapOut().WriteResult(c.query, true); err != nil {
-		dlog.Client.FatalPanic(err)
+	dlog.Client.Debug("WriteResult completed for non-cumulative mode")
+	return nil
+}
+
+func (c *MaprClient) commitSessionSpec(spec SessionSpec, generation uint64) error {
+	if spec.Query == "" {
+		return errors.New("missing mapreduce query")
+	}
+
+	query, err := c.session.CommitQuery(spec.Query, generation)
+	if err != nil {
+		return err
+	}
+
+	c.Args.QueryStr = spec.Query
+	c.setRegexForQuery(query)
+	return nil
+}
+
+func (c *MaprClient) isCumulative(query *mapr.Query) bool {
+	switch c.mode {
+	case CumulativeMode:
+		return true
+	case NonCumulativeMode:
+		return false
+	default:
+		return c.Args.Mode == omode.MapClient || (query != nil && query.HasOutfile())
+	}
+}
+
+func (c *MaprClient) setRegexForQuery(query *mapr.Query) {
+	c.RegexStr = maprRegexForQuery(query)
+}
+
+// warnUnknownQueryVariables writes a plan-time warning to w for each
+// $-variable the query references that the selected parser cannot populate.
+// The client cannot know a remote server's configured MapreduceLogFormat, so it
+// assumes the DTail default ("default") for the from-TABLE case; this matches
+// the stock server configuration. Unknown/custom server formats therefore do
+// not get this diagnostic (see PlanVariableWarnings, which is a no-op for
+// non-enumerable formats).
+func warnUnknownQueryVariables(w io.Writer, query *mapr.Query) {
+	logFormat := query.EffectiveLogFormat("")
+	for _, warning := range logformat.PlanVariableWarnings(query, logFormat) {
+		fmt.Fprintln(w, warning)
+	}
+}
+
+func (c *MaprClient) reportDelay(query *mapr.Query, rampUp bool) time.Duration {
+	interval := time.Second
+	if query != nil && query.Interval > 0 {
+		interval = query.Interval
+	}
+	if !rampUp {
+		return interval
+	}
+
+	delay := interval / 2
+	if delay <= 0 {
+		return interval
 	}
+	return delay
 }