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, 136 insertions, 0 deletions

diff --git a/internal/clients/client_benchmark_test.go b/internal/clients/client_benchmark_test.go
new file mode 100644
index 0000000..9423274
--- /dev/null
+++ b/internal/clients/client_benchmark_test.go
@@ -0,0 +1,136 @@
+package clients
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/mimecast/dtail/internal/config"
+	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/source"
+	"sync"
+)
+
+func setupBenchmarkData(b *testing.B, lines int) string {
+	b.Helper()
+
+	tmpDir := b.TempDir()
+	testFile := filepath.Join(tmpDir, "benchmark_data.log")
+
+	f, err := os.Create(testFile)
+	if err != nil {
+		b.Fatalf("Failed to create test file: %v", err)
+	}
+	defer f.Close()
+
+	// Create test data
+	for i := 0; i < lines; i++ {
+		line := fmt.Sprintf("INFO|1002-071143|1|test.go:%d|8|%d|7|0.21|471h0m21s|MAPREDUCE:STATS|currentConnections=%d|lifetimeConnections=%d|pattern=test-%d|data=%s\n",
+			i%100, i%50, i%10, i, i%5, "some-test-data-that-makes-the-line-longer")
+		f.WriteString(line)
+	}
+
+	return testFile
+}
+
+func BenchmarkDGrep(b *testing.B) {
+	benchmarkDGrepWithSize(b, 100000) // 100k lines
+}
+
+// Benchmark with different file sizes
+func BenchmarkDGrepSmallFile(b *testing.B) {
+	benchmarkDGrepWithSize(b, 1000) // 1k lines
+}
+
+func BenchmarkDGrepMediumFile(b *testing.B) {
+	benchmarkDGrepWithSize(b, 50000) // 50k lines
+}
+
+func BenchmarkDGrepLargeFile(b *testing.B) {
+	benchmarkDGrepWithSize(b, 500000) // 500k lines
+}
+
+func benchmarkDGrepWithSize(b *testing.B, lines int) {
+	// Setup config. The direct-output read path is the only runtime path.
+	config.Server = &config.ServerConfig{
+		MaxConcurrentCats:  10,
+		MaxConcurrentTails: 50,
+		MaxLineLength:      1024 * 1024,
+	}
+
+	config.Common = &config.CommonConfig{
+		Logger:   "none",
+		LogLevel: "error",
+	}
+
+	config.Client = &config.ClientConfig{
+		TermColorsEnable: false,
+	}
+
+	// Initialize logging
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	dlog.Start(ctx, wg, source.Client)
+
+	// Create test data
+	testFile := setupBenchmarkData(b, lines)
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		// Create grep client
+		args := config.Args{
+			ServersStr: "serverless",
+			QueryStr:   "",
+			What:       testFile,
+			RegexStr:   "pattern=test-1",
+			Serverless: true,
+			Plain:      true,
+		}
+
+		client, err := NewGrepClient(args)
+		if err != nil {
+			b.Fatalf("Failed to create grep client: %v", err)
+		}
+
+		// Capture output
+		oldStdout := os.Stdout
+		r, w, _ := os.Pipe()
+		os.Stdout = w
+
+		// Run grep
+		statusCh := make(chan int, 1)
+		go func() {
+			status := client.Start(ctx, nil) // nil for statsCh
+			statusCh <- status
+		}()
+
+		// Wait for completion or timeout
+		select {
+		case status := <-statusCh:
+			if status != 0 {
+				b.Errorf("Grep failed with status: %d", status)
+			}
+		case <-time.After(30 * time.Second):
+			b.Error("Grep timed out")
+		}
+
+		// Restore stdout
+		w.Close()
+		os.Stdout = oldStdout
+
+		// Read captured output
+		var buf bytes.Buffer
+		buf.ReadFrom(r)
+	}
+
+	// Report custom metrics
+	b.ReportMetric(float64(lines), "lines/op")
+	b.ReportMetric(float64(lines)/b.Elapsed().Seconds(), "lines/sec")
+}