From 849951be1d1a7ee9f9302006ccb187bf5b4e36f3 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 22 Jul 2026 23:51:18 +0300 Subject: =?UTF-8?q?feat:=20DTail=20fork=20=E2=80=94=20server/client=20feat?= =?UTF-8?q?ure=20development?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- internal/discovery/comma.go | 15 ++- internal/discovery/discovery.go | 34 ++++--- internal/discovery/discovery_test.go | 176 +++++++++++++++++++++++++++++++++++ 3 files changed, 210 insertions(+), 15 deletions(-) create mode 100644 internal/discovery/discovery_test.go (limited to 'internal/discovery') diff --git a/internal/discovery/comma.go b/internal/discovery/comma.go index 9bea89c..126cbec 100644 --- a/internal/discovery/comma.go +++ b/internal/discovery/comma.go @@ -9,5 +9,18 @@ import ( // ServerListFromCOMMA retrieves a list of servers from comma separated input list. func (d *Discovery) ServerListFromCOMMA() []string { dlog.Common.Debug("Retrieving server list from comma separated list", d.server) - return strings.Split(d.server, ",") + + rawServers := strings.Split(d.server, ",") + servers := make([]string, 0, len(rawServers)) + for _, server := range rawServers { + if server == "" { + continue + } + servers = append(servers, server) + } + if len(servers) == 0 && d.server == "" { + return rawServers + } + + return servers } diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go index f709501..bebd798 100644 --- a/internal/discovery/discovery.go +++ b/internal/discovery/discovery.go @@ -35,14 +35,14 @@ type Discovery struct { } // New returns a new discovery method. -func New(method, server string, order ServerOrder) *Discovery { +func New(method, server string, order ServerOrder) (*Discovery, error) { module := method options := "" if strings.Contains(module, ":") { - s := strings.Split(module, ":") - if len(s) != 2 { - dlog.Common.FatalPanic("Unable to parse discovery module", module) + s := strings.SplitN(module, ":", 2) + if len(s) != 2 || s[0] == "" || s[1] == "" { + return nil, fmt.Errorf("unable to parse discovery module %q", module) } module = s[0] options = s[1] @@ -59,7 +59,7 @@ func New(method, server string, order ServerOrder) *Discovery { d.initRegex() } - return &d + return &d, nil } func (d *Discovery) initRegex() { @@ -102,6 +102,9 @@ func (d *Discovery) serverListFromModule() []string { if d.module != "" { return d.serverListFromReflectedModule() } + if d.server == "" && d.regex != nil { + return []string{} + } if _, err := os.Stat(d.server); err == nil { // Appears to be a file name, now try to read from that file. return d.ServerListFromFILE() @@ -153,19 +156,22 @@ func (d *Discovery) dedupList(servers []string) (deduped []string) { return } -// Randomly shuffle the server list. +// Randomly shuffle the server list. A copy of the input slice is made so the +// caller's backing array is never modified. The seed uses UnixNano so that +// two callers started within the same wall-clock second still get different +// shuffle orders, which is important for client-side load spreading. func (d *Discovery) shuffleList(servers []string) []string { dlog.Common.Debug("Shuffling server list") - r := rand.New(rand.NewSource(time.Now().Unix())) - shuffled := make([]string, len(servers)) - n := len(servers) + // Copy to avoid mutating the caller's backing array. + shuffled := append([]string(nil), servers...) - for i := 0; i < n; i++ { - randIndex := r.Intn(len(servers)) - shuffled[i] = servers[randIndex] - servers = append(servers[:randIndex], servers[randIndex+1:]...) - } + // Seed with nanosecond resolution to avoid identical orderings when + // multiple clients start within the same second. + r := rand.New(rand.NewSource(time.Now().UnixNano())) + r.Shuffle(len(shuffled), func(i, j int) { + shuffled[i], shuffled[j] = shuffled[j], shuffled[i] + }) return shuffled } diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go new file mode 100644 index 0000000..afc5d31 --- /dev/null +++ b/internal/discovery/discovery_test.go @@ -0,0 +1,176 @@ +package discovery + +import ( + "reflect" + "testing" + + "github.com/mimecast/dtail/internal/io/dlog" +) + +func TestMain(m *testing.M) { + dlog.Common = &dlog.DLog{} + m.Run() +} + +// TestShuffleListDoesNotMutateInput verifies that shuffleList never writes +// back into the caller's backing array, and that repeated calls within the +// same sub-second window produce different orderings (the chance of a false +// failure with 20 elements is 1/20! ≈ 4×10⁻¹⁹). +func TestShuffleListDoesNotMutateInput(t *testing.T) { + t.Parallel() + + d := &Discovery{} + + // Build a stable reference slice. + original := make([]string, 20) + for i := range original { + original[i] = string(rune('a' + i)) + } + + snapshot := append([]string(nil), original...) + + // First shuffle — must not touch original. + first := d.shuffleList(original) + if !reflect.DeepEqual(original, snapshot) { + t.Fatalf("shuffleList mutated caller's slice: got %v, want %v", original, snapshot) + } + if len(first) != len(original) { + t.Fatalf("shuffleList returned wrong length: got %d, want %d", len(first), len(original)) + } + + // Second shuffle — different seed path; overwhelmingly likely to differ. + second := d.shuffleList(original) + if reflect.DeepEqual(first, second) { + t.Fatal("shuffleList returned identical order on two consecutive calls; seed resolution may be too coarse") + } + + // Both results must contain exactly the same elements as the input. + countFirst := make(map[string]int, len(original)) + countSecond := make(map[string]int, len(original)) + for i := range original { + countFirst[first[i]]++ + countSecond[second[i]]++ + } + for _, s := range original { + if countFirst[s] != 1 { + t.Errorf("element %q appears %d times in first shuffle, want 1", s, countFirst[s]) + } + if countSecond[s] != 1 { + t.Errorf("element %q appears %d times in second shuffle, want 1", s, countSecond[s]) + } + } +} + +func TestNewParsesModuleOptionsWithAdditionalColons(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + method string + wantMod string + wantOpt string + wantErr bool + }{ + { + name: "plain module", + method: "file", + wantMod: "FILE", + }, + { + name: "options with additional colons", + method: "method:host:port:extra", + wantMod: "METHOD", + wantOpt: "host:port:extra", + }, + { + name: "empty options rejected", + method: "method:", + wantErr: true, + }, + { + name: "missing module rejected", + method: ":host:port", + wantErr: true, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, err := New(tt.method, "server", Shuffle) + if tt.wantErr { + if err == nil { + t.Fatalf("New(%q) error = nil, want error", tt.method) + } + return + } + if err != nil { + t.Fatalf("New(%q) error = %v, want nil", tt.method, err) + } + if got.module != tt.wantMod { + t.Fatalf("module = %q, want %q", got.module, tt.wantMod) + } + if got.options != tt.wantOpt { + t.Fatalf("options = %q, want %q", got.options, tt.wantOpt) + } + }) + } +} + +func TestServerListIgnoresEmptyServerEntries(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + method string + server string + want []string + wantErr bool + }{ + { + name: "regex without module yields no phantom host", + server: "/.*/", + want: []string{}, + }, + { + name: "comma list filters empty entries", + server: "alpha,,beta,", + want: []string{"alpha", "beta"}, + }, + { + name: "empty server input preserves serverless sentinel", + server: "", + want: []string{""}, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got, err := New(tt.method, tt.server, ServerOrder(99)) + if tt.wantErr { + if err == nil { + t.Fatalf("New(%q, %q) error = nil, want error", tt.method, tt.server) + } + return + } + if err != nil { + t.Fatalf("New(%q, %q) error = %v, want nil", tt.method, tt.server, err) + } + + servers := got.ServerList() + if len(servers) != len(tt.want) { + t.Fatalf("ServerList() len = %d, want %d (%v)", len(servers), len(tt.want), servers) + } + for i := range tt.want { + if servers[i] != tt.want[i] { + t.Fatalf("ServerList()[%d] = %q, want %q (full=%v)", i, servers[i], tt.want[i], servers) + } + } + }) + } +} -- cgit v1.2.3