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/session/spec.go | 183 ++++++++++++++++++++++++++++++++++++++++++ internal/session/spec_test.go | 117 +++++++++++++++++++++++++++ 2 files changed, 300 insertions(+) create mode 100644 internal/session/spec.go create mode 100644 internal/session/spec_test.go (limited to 'internal/session') diff --git a/internal/session/spec.go b/internal/session/spec.go new file mode 100644 index 0000000..44203df --- /dev/null +++ b/internal/session/spec.go @@ -0,0 +1,183 @@ +package session + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "strings" + + "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/omode" + "github.com/mimecast/dtail/internal/regex" +) + +// Spec captures the mutable, per-connection workload a DTail client wants to run. +type Spec struct { + Mode omode.Mode `json:"mode"` + Files []string `json:"files"` + Options string `json:"options,omitempty"` + Query string `json:"query,omitempty"` + Regex string `json:"regex,omitempty"` + RegexInvert bool `json:"regex_invert,omitempty"` + Timeout int `json:"timeout,omitempty"` +} + +// NewSpec returns a session specification from client args. +func NewSpec(args config.Args) Spec { + files := splitFiles(args.What) + if args.Serverless && len(files) == 0 && supportsServerlessPipe(args.Mode) { + files = []string{"-"} + } + + return Spec{ + Mode: args.Mode, + Files: files, + Options: args.SerializeOptions(), + Query: strings.TrimSpace(args.QueryStr), + Regex: args.RegexStr, + RegexInvert: args.RegexInvert, + Timeout: args.Timeout, + } +} + +// Commands returns the legacy command stream for this session specification. +func (s Spec) Commands() ([]string, error) { + switch { + case s.Mode == omode.HealthClient: + return []string{"health"}, nil + case s.Query != "": + return s.queryCommands() + default: + return s.readCommands(s.Mode.String()) + } +} + +// HasJournalFiles reports whether this session reads any systemd journal target. +func (s Spec) HasJournalFiles() bool { + for _, file := range s.Files { + if strings.HasPrefix(strings.TrimSpace(file), "journal:") { + return true + } + } + return false +} + +// StartCommand returns the SESSION START command for this specification. +func (s Spec) StartCommand() (string, error) { + payload, err := s.encodedPayload() + if err != nil { + return "", err + } + + return fmt.Sprintf("SESSION START %s", payload), nil +} + +// UpdateCommand returns the SESSION UPDATE command for this specification. +func (s Spec) UpdateCommand(generation uint64) (string, error) { + payload, err := s.encodedPayload() + if err != nil { + return "", err + } + + if generation == 0 { + return fmt.Sprintf("SESSION UPDATE %s", payload), nil + } + + return fmt.Sprintf("SESSION UPDATE %d %s", generation, payload), nil +} + +func (s Spec) queryCommands() ([]string, error) { + if s.Mode != omode.MapClient && s.Mode != omode.TailClient { + return nil, fmt.Errorf("session spec query mode requires map or tail mode, got %s", s.Mode) + } + + regexValue, err := s.serializedRegex() + if err != nil { + return nil, err + } + + commands := []string{fmt.Sprintf("map:%s %s", s.Options, s.Query)} + readMode := "cat" + if s.Mode == omode.TailClient { + readMode = "tail" + } + + for _, file := range s.Files { + if s.Timeout > 0 { + commands = append(commands, fmt.Sprintf("timeout %d %s %s %s", s.Timeout, readMode, file, regexValue)) + continue + } + commands = append(commands, fmt.Sprintf("%s:%s %s %s", readMode, s.Options, file, regexValue)) + } + + return commands, nil +} + +func (s Spec) readCommands(mode string) ([]string, error) { + switch s.Mode { + case omode.TailClient, omode.CatClient, omode.GrepClient: + default: + return nil, fmt.Errorf("unsupported session mode %s", s.Mode) + } + + regexValue, err := s.serializedRegex() + if err != nil { + return nil, err + } + + var commands []string + for _, file := range s.Files { + commands = append(commands, fmt.Sprintf("%s:%s %s %s", mode, s.Options, file, regexValue)) + } + + return commands, nil +} + +func (s Spec) serializedRegex() (string, error) { + flag := regex.Default + if s.RegexInvert { + flag = regex.Invert + } + + re, err := regex.New(s.Regex, flag) + if err != nil { + return "", err + } + + return re.Serialize() +} + +func splitFiles(what string) []string { + if strings.TrimSpace(what) == "" { + return nil + } + + rawFiles := strings.Split(what, ",") + files := make([]string, 0, len(rawFiles)) + for _, file := range rawFiles { + file = strings.TrimSpace(file) + if file == "" { + continue + } + files = append(files, file) + } + return files +} + +func supportsServerlessPipe(mode omode.Mode) bool { + switch mode { + case omode.TailClient, omode.CatClient, omode.GrepClient, omode.MapClient: + return true + default: + return false + } +} + +func (s Spec) encodedPayload() (string, error) { + payload, err := json.Marshal(s) + if err != nil { + return "", fmt.Errorf("marshal session spec: %w", err) + } + + return base64.StdEncoding.EncodeToString(payload), nil +} diff --git a/internal/session/spec_test.go b/internal/session/spec_test.go new file mode 100644 index 0000000..9d4a9c9 --- /dev/null +++ b/internal/session/spec_test.go @@ -0,0 +1,117 @@ +package session + +import ( + "encoding/base64" + "encoding/json" + "reflect" + "strings" + "testing" + + "github.com/mimecast/dtail/internal/omode" +) + +func TestSpecStartCommandEncodesPayload(t *testing.T) { + t.Parallel() + + spec := Spec{ + Mode: omode.TailClient, + Files: []string{"/var/log/app.log"}, + Options: "plain=true", + Regex: "ERROR", + Timeout: 15, + } + + command, err := spec.StartCommand() + if err != nil { + t.Fatalf("StartCommand() error = %v", err) + } + if !strings.HasPrefix(command, "SESSION START ") { + t.Fatalf("unexpected start command prefix: %q", command) + } + + var decoded Spec + if err := decodeSpecPayload(strings.TrimPrefix(command, "SESSION START "), &decoded); err != nil { + t.Fatalf("decode start payload: %v", err) + } + if !reflect.DeepEqual(decoded, spec) { + t.Fatalf("unexpected decoded spec: got %#v want %#v", decoded, spec) + } +} + +func TestSpecUpdateCommandIncludesGeneration(t *testing.T) { + t.Parallel() + + spec := Spec{ + Mode: omode.MapClient, + Files: []string{"/var/log/app.log"}, + Query: "from STATS select count(*)", + } + + command, err := spec.UpdateCommand(7) + if err != nil { + t.Fatalf("UpdateCommand() error = %v", err) + } + if !strings.HasPrefix(command, "SESSION UPDATE 7 ") { + t.Fatalf("unexpected update command prefix: %q", command) + } + + var decoded Spec + if err := decodeSpecPayload(strings.TrimPrefix(command, "SESSION UPDATE 7 "), &decoded); err != nil { + t.Fatalf("decode update payload: %v", err) + } + if !reflect.DeepEqual(decoded, spec) { + t.Fatalf("unexpected decoded spec: got %#v want %#v", decoded, spec) + } +} + +func TestSpecHasJournalFiles(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + files []string + want bool + }{ + { + name: "journal file", + files: []string{"journal:ssh.service"}, + want: true, + }, + { + name: "journal file with surrounding spaces", + files: []string{" /var/log/app.log ", " journal:nginx.service "}, + want: true, + }, + { + name: "regular file", + files: []string{"/var/log/app.log"}, + want: false, + }, + { + name: "journal substring is not prefix", + files: []string{"/var/log/journal:ssh.service.log"}, + want: false, + }, + { + name: "empty files", + want: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + spec := Spec{Files: tc.files} + if got := spec.HasJournalFiles(); got != tc.want { + t.Fatalf("HasJournalFiles() = %v, want %v", got, tc.want) + } + }) + } +} + +func decodeSpecPayload(payload string, out *Spec) error { + raw, err := base64.StdEncoding.DecodeString(payload) + if err != nil { + return err + } + return json.Unmarshal(raw, out) +} -- cgit v1.2.3