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/mapr/logformat/csv.go | 97 ++++++--- internal/mapr/logformat/csv_test.go | 170 ++++++++++++++- internal/mapr/logformat/custom1.go | 5 +- internal/mapr/logformat/custom2.go | 5 +- internal/mapr/logformat/default.go | 239 ++++++++++++++++++---- internal/mapr/logformat/default_benchmark_test.go | 44 ++++ internal/mapr/logformat/default_test.go | 42 +++- internal/mapr/logformat/delimited.go | 12 ++ internal/mapr/logformat/generic.go | 15 +- internal/mapr/logformat/generickv.go | 36 ++-- internal/mapr/logformat/mimecast.go | 5 +- internal/mapr/logformat/parser.go | 123 ++++++++--- internal/mapr/logformat/parser_test.go | 69 +++++++ internal/mapr/logformat/variables.go | 107 ++++++++++ internal/mapr/logformat/variables_test.go | 145 +++++++++++++ 15 files changed, 981 insertions(+), 133 deletions(-) create mode 100644 internal/mapr/logformat/default_benchmark_test.go create mode 100644 internal/mapr/logformat/delimited.go create mode 100644 internal/mapr/logformat/parser_test.go create mode 100644 internal/mapr/logformat/variables.go create mode 100644 internal/mapr/logformat/variables_test.go (limited to 'internal/mapr/logformat') diff --git a/internal/mapr/logformat/csv.go b/internal/mapr/logformat/csv.go index ea85ca9..d82b238 100644 --- a/internal/mapr/logformat/csv.go +++ b/internal/mapr/logformat/csv.go @@ -2,52 +2,103 @@ package logformat import ( "fmt" - "strings" + "sync" "github.com/mimecast/dtail/internal/protocol" ) +// csvParser parses CSV log lines. The first line encountered for a given +// sourceID is treated as the column header and stored so that subsequent +// lines from the same source can be mapped to named fields. State is kept +// per sourceID because a single parser instance is shared across every +// file/stream processed within a mapreduce session; without this, the +// header row of every file after the first one would silently be mapped +// as a data row, corrupting aggregates. type csvParser struct { defaultParser - header []string - hasHeader bool + mu sync.RWMutex + headers map[string][]string } +var _ Parser = (*csvParser)(nil) + func newCSVParser(hostname, timeZoneName string, timeZoneOffset int) (*csvParser, error) { defaultParser, err := newDefaultParser(hostname, timeZoneName, timeZoneOffset) if err != nil { return &csvParser{}, err } - return &csvParser{defaultParser: *defaultParser}, nil + return &csvParser{ + defaultParser: *defaultParser, + headers: make(map[string][]string), + }, nil } -func (p *csvParser) MakeFields(maprLine string) (map[string]string, error) { - if !p.hasHeader { - p.parseHeader(maprLine) +func (p *csvParser) MakeFields(maprLine, sourceID string) (map[string]string, error) { + header, installed := p.ensureHeader(sourceID, maprLine) + if installed { return nil, ErrIgnoreFields } - fields := make(map[string]string, 7+len(p.header)) - fields["*"] = "*" - fields["$hostname"] = p.hostname - fields["$server"] = p.hostname - fields["$line"] = maprLine - fields["$empty"] = "" - fields["$timezone"] = p.timeZoneName - fields["$timeoffset"] = p.timeZoneOffset - - splitted := strings.Split(maprLine, protocol.CSVDelimiter) - for i, value := range splitted { - if i >= len(p.header) { + fields := make(map[string]string, p.fieldsCapacity) + p.addDefaultFields(fields, maprLine) + start := 0 + column := 0 + delimiter := protocol.CSVDelimiter[0] + + for { + value, next, done := scanDelimitedField(maprLine, start, delimiter) + if column >= len(header) { return fields, fmt.Errorf("CSV file seems corrupted, more fields than header values?") } - fields[p.header[i]] = value + p.addDynamicField(fields, header[column], value) + column++ + if done { + break + } + start = next } return fields, nil } -func (p *csvParser) parseHeader(maprLine string) { - p.header = strings.Split(maprLine, protocol.CSVDelimiter) - p.hasHeader = true +// ensureHeader atomically checks for, and if necessary installs, the header +// for sourceID. It returns the effective header for the source and whether +// this call was the one that installed it. Only the goroutine that actually +// installs the header should tell its caller to ignore the current line +// (i.e. return ErrIgnoreFields); any racing goroutine on the same sourceID +// sees installed=false and proceeds to map its line against the installed +// header. The previous implementation split the check (RLock) from the +// install (Lock), so two goroutines could both observe "missing" and both +// report ErrIgnoreFields, silently dropping the loser's data row. +func (p *csvParser) ensureHeader(sourceID, maprLine string) ([]string, bool) { + p.mu.RLock() + if header, ok := p.headers[sourceID]; ok { + p.mu.RUnlock() + return header, false + } + p.mu.RUnlock() + + p.mu.Lock() + defer p.mu.Unlock() + if header, ok := p.headers[sourceID]; ok { + return header, false + } + header := parseHeaderLine(maprLine) + p.headers[sourceID] = header + return header, true +} + +func parseHeaderLine(maprLine string) []string { + var header []string + start := 0 + delimiter := protocol.CSVDelimiter[0] + for { + field, next, done := scanDelimitedField(maprLine, start, delimiter) + header = append(header, field) + if done { + break + } + start = next + } + return header } diff --git a/internal/mapr/logformat/csv_test.go b/internal/mapr/logformat/csv_test.go index 1baf032..fa85a99 100644 --- a/internal/mapr/logformat/csv_test.go +++ b/internal/mapr/logformat/csv_test.go @@ -2,6 +2,7 @@ package logformat import ( "strings" + "sync" "testing" "github.com/mimecast/dtail/internal/protocol" @@ -23,13 +24,15 @@ func TestCSVLogFormat(t *testing.T) { strings.Join(dataLine2, protocol.CSVDelimiter), } + const sourceID = "file-a" + // First line is the header! - if _, err := parser.MakeFields(inputs[0]); err != ErrIgnoreFields { + if _, err := parser.MakeFields(inputs[0], sourceID); err != ErrIgnoreFields { t.Errorf("Unable to parse the CSV header") } // First data line - fields, err := parser.MakeFields(inputs[1]) + fields, err := parser.MakeFields(inputs[1], sourceID) if err != nil { t.Errorf("Unable to parse first CSV data line: %s", err.Error()) } @@ -41,7 +44,7 @@ func TestCSVLogFormat(t *testing.T) { } // Second data line - fields, err = parser.MakeFields(inputs[2]) + fields, err = parser.MakeFields(inputs[2], sourceID) if err != nil { t.Errorf("Unable to parse first CSV data line: %s", err.Error()) } @@ -52,3 +55,164 @@ func TestCSVLogFormat(t *testing.T) { t.Errorf("Expected 'color' to be 'Black' but got '%s'", val) } } + +// TestCSVLogFormatMultiFileHeaders reproduces the bug where a single +// csvParser instance (as used by the Aggregate for every file +// in a mapreduce session) was treating the header row of every file after +// the first as a data row, silently corrupting aggregates. +func TestCSVLogFormatMultiFileHeaders(t *testing.T) { + parser, err := NewParser("csv", nil) + if err != nil { + t.Fatalf("Unable to create parser: %s", err.Error()) + } + + headersA := []string{"name", "value"} + headersB := []string{"color", "count"} + + fileA := []string{ + strings.Join(headersA, protocol.CSVDelimiter), + strings.Join([]string{"alpha", "1"}, protocol.CSVDelimiter), + strings.Join([]string{"beta", "2"}, protocol.CSVDelimiter), + } + fileB := []string{ + strings.Join(headersB, protocol.CSVDelimiter), + strings.Join([]string{"orange", "3"}, protocol.CSVDelimiter), + strings.Join([]string{"black", "4"}, protocol.CSVDelimiter), + } + + const sourceA = "file-a" + const sourceB = "file-b" + + // First line of file A is its header. + if _, err := parser.MakeFields(fileA[0], sourceA); err != ErrIgnoreFields { + t.Fatalf("Expected header line of file A to be ignored, got err=%v", err) + } + for _, line := range fileA[1:] { + fields, err := parser.MakeFields(line, sourceA) + if err != nil { + t.Fatalf("Unable to parse data line %q of file A: %s", line, err.Error()) + } + if _, ok := fields["name"]; !ok { + t.Errorf("Expected file A field 'name' for line %q, got %v", line, fields) + } + } + + // First line of file B MUST also be treated as a header, not a data row. + if _, err := parser.MakeFields(fileB[0], sourceB); err != ErrIgnoreFields { + t.Fatalf("Expected header line of file B to be ignored (bug: header is being consumed as a data row), got err=%v", err) + } + + // Data lines of file B must be mapped against file B's headers, not + // file A's. + for _, line := range fileB[1:] { + fields, err := parser.MakeFields(line, sourceB) + if err != nil { + t.Fatalf("Unable to parse data line %q of file B: %s", line, err.Error()) + } + if _, ok := fields["color"]; !ok { + t.Errorf("Expected file B field 'color' for line %q, got %v", line, fields) + } + if _, ok := fields["name"]; ok { + t.Errorf("File B line %q should not carry file A field 'name'; got %v", + line, fields) + } + } +} + +// TestCSVLogFormatConcurrentSameSourceInstall reproduces a TOCTOU bug in +// csvParser.MakeFields: the original code first called headerFor under +// RLock, and only if the header was missing did it call parseHeader under +// Lock. Two goroutines racing on the same sourceID could both observe +// "missing" and both return ErrIgnoreFields, silently dropping the loser's +// data row (the installer wrote the header; the non-installer still +// signalled "this line was a header" to the caller). +// +// With the fix, the check-and-install is a single critical section and +// exactly one of the two concurrent calls reports ErrIgnoreFields; the +// other maps its line against the installed header. +func TestCSVLogFormatConcurrentSameSourceInstall(t *testing.T) { + header := strings.Join([]string{"name", "value"}, protocol.CSVDelimiter) + data := strings.Join([]string{"alpha", "1"}, protocol.CSVDelimiter) + + const attempts = 500 + const sourceID = "file-race" + + for attempt := 0; attempt < attempts; attempt++ { + parser, err := NewParser("csv", nil) + if err != nil { + t.Fatalf("attempt %d: unable to create parser: %s", attempt, err.Error()) + } + + lines := [2]string{header, data} + var results [2]struct { + fields map[string]string + err error + } + + start := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(2) + for i := 0; i < 2; i++ { + i := i + go func() { + defer wg.Done() + <-start + results[i].fields, results[i].err = parser.MakeFields(lines[i], sourceID) + }() + } + close(start) + wg.Wait() + + ignored := 0 + for _, r := range results { + if r.err == ErrIgnoreFields { + ignored++ + } + } + if ignored != 1 { + t.Fatalf("attempt %d: expected exactly one ErrIgnoreFields across two racing calls on the same sourceID, got %d; results=%+v", + attempt, ignored, results) + } + } +} + +// TestCSVLogFormatConcurrentSources ensures the per-source header store is +// safe for concurrent access across multiple sourceIDs, matching how the +// aggregator drives the parser from batched lines across files. +func TestCSVLogFormatConcurrentSources(t *testing.T) { + parser, err := NewParser("csv", nil) + if err != nil { + t.Fatalf("Unable to create parser: %s", err.Error()) + } + + header := strings.Join([]string{"name", "value"}, protocol.CSVDelimiter) + data := strings.Join([]string{"alpha", "1"}, protocol.CSVDelimiter) + + const workers = 16 + const iterations = 64 + + var wg sync.WaitGroup + wg.Add(workers) + for w := 0; w < workers; w++ { + go func(id int) { + defer wg.Done() + sourceID := "source-" + string(rune('a'+id)) + if _, err := parser.MakeFields(header, sourceID); err != ErrIgnoreFields { + t.Errorf("worker %d: expected header to be ignored, got err=%v", id, err) + return + } + for i := 0; i < iterations; i++ { + fields, err := parser.MakeFields(data, sourceID) + if err != nil { + t.Errorf("worker %d: parse err=%v", id, err) + return + } + if fields["name"] != "alpha" { + t.Errorf("worker %d: expected name=alpha, got %q", id, fields["name"]) + return + } + } + }(w) + } + wg.Wait() +} diff --git a/internal/mapr/logformat/custom1.go b/internal/mapr/logformat/custom1.go index 7229f3e..a159328 100644 --- a/internal/mapr/logformat/custom1.go +++ b/internal/mapr/logformat/custom1.go @@ -2,15 +2,18 @@ package logformat import "errors" +// ErrCustom1NotImplemented indicates custom1 parser is only a template. var ErrCustom1NotImplemented error = errors.New("custom1 log format is not implemented") // Template for creating a custom log format. type custom1Parser struct{} +var _ Parser = (*custom1Parser)(nil) + func newCustom1Parser(hostname, timeZoneName string, timeZoneOffset int) (*custom1Parser, error) { return &custom1Parser{}, ErrCustom1NotImplemented } -func (p *custom1Parser) MakeFields(maprLine string) (map[string]string, error) { +func (p *custom1Parser) MakeFields(maprLine, _ string) (map[string]string, error) { return nil, ErrCustom1NotImplemented } diff --git a/internal/mapr/logformat/custom2.go b/internal/mapr/logformat/custom2.go index 262c721..a1a732b 100644 --- a/internal/mapr/logformat/custom2.go +++ b/internal/mapr/logformat/custom2.go @@ -2,15 +2,18 @@ package logformat import "errors" +// ErrCustom2NotImplemented indicates custom2 parser is only a template. var ErrCustom2NotImplemented error = errors.New("custom2 log format is not implemented") // Template for creating a custom log format. type custom2Parser struct{} +var _ Parser = (*custom2Parser)(nil) + func newCustom2Parser(hostname, timeZoneName string, timeZoneOffset int) (*custom2Parser, error) { return &custom2Parser{}, ErrCustom2NotImplemented } -func (p *custom2Parser) MakeFields(maprLine string) (map[string]string, error) { +func (p *custom2Parser) MakeFields(maprLine, _ string) (map[string]string, error) { return nil, ErrCustom2NotImplemented } diff --git a/internal/mapr/logformat/default.go b/internal/mapr/logformat/default.go index a499bc5..c1e6e44 100644 --- a/internal/mapr/logformat/default.go +++ b/internal/mapr/logformat/default.go @@ -4,6 +4,7 @@ import ( "fmt" "strings" + "github.com/mimecast/dtail/internal/mapr" "github.com/mimecast/dtail/internal/protocol" ) @@ -11,62 +12,216 @@ type defaultParser struct { hostname string timeZoneName string timeZoneOffset string + fieldsCapacity int + + wantStar bool + wantLine bool + wantEmpty bool + wantHostname bool + wantServer bool + wantTimezone bool + wantTimeOffset bool + wantSeverity bool + wantLogLevel bool + wantTime bool + wantDate bool + wantHour bool + wantMinute bool + wantSecond bool + wantPID bool + wantCaller bool + wantCPUs bool + wantGoroutines bool + wantCGOCalls bool + wantLoadAvg bool + wantUptime bool + + allDynamicFields bool + dynamicFields map[string]struct{} } func newDefaultParser(hostname, timeZoneName string, timeZoneOffset int) (*defaultParser, error) { - return &defaultParser{ + parser := &defaultParser{ hostname: hostname, timeZoneName: timeZoneName, timeZoneOffset: fmt.Sprintf("%d", timeZoneOffset), - }, nil + } + parser.configureFieldPlan(mapr.ParserFieldPlan{AllFields: true}) + return parser, nil +} + +func (p *defaultParser) setQuery(query *mapr.Query) { + p.configureFieldPlan(query.ParserFieldPlan()) } -func (p *defaultParser) MakeFields(maprLine string) (map[string]string, error) { - splitted := strings.Split(maprLine, protocol.FieldDelimiter) +func (p *defaultParser) MakeFields(maprLine, _ string) (map[string]string, error) { + fields := make(map[string]string, p.fieldsCapacity) + tokenIndex := 0 + start := 0 + delimiter := protocol.FieldDelimiter[0] - if len(splitted) < 11 || !strings.HasPrefix(splitted[9], "MAPREDUCE:") || - !strings.HasPrefix(splitted[0], "INFO") { + for { + token, next, done := scanDelimitedField(maprLine, start, delimiter) + switch { + case tokenIndex == 0: + if !strings.HasPrefix(token, "INFO") { + return nil, ErrIgnoreFields + } + p.addDefaultFields(fields, maprLine) + if p.wantSeverity { + fields["$severity"] = token + } + if p.wantLogLevel { + fields["$loglevel"] = token + } + case tokenIndex == 1: + if p.wantTime { + fields["$time"] = token + } + if len(token) == 15 { + // Example: 20211002-071209 + if p.wantDate { + fields["$date"] = token[0:8] + } + if p.wantHour { + fields["$hour"] = token[9:11] + } + if p.wantMinute { + fields["$minute"] = token[11:13] + } + if p.wantSecond { + fields["$second"] = token[13:] + } + } + case tokenIndex == 2: + if p.wantPID { + fields["$pid"] = token + } + case tokenIndex == 3: + if p.wantCaller { + fields["$caller"] = token + } + case tokenIndex == 4: + if p.wantCPUs { + fields["$cpus"] = token + } + case tokenIndex == 5: + if p.wantGoroutines { + fields["$goroutines"] = token + } + case tokenIndex == 6: + if p.wantCGOCalls { + fields["$cgocalls"] = token + } + case tokenIndex == 7: + if p.wantLoadAvg { + fields["$loadavg"] = token + } + case tokenIndex == 8: + if p.wantUptime { + fields["$uptime"] = token + } + case tokenIndex == 9: + if !strings.HasPrefix(token, "MAPREDUCE:") { + return nil, ErrIgnoreFields + } + default: + if err := p.addKeyValueField(fields, token); err != nil { + return fields, err + } + } + + tokenIndex++ + if done { + break + } + start = next + } + + if tokenIndex < 11 { // Not a DTail mapreduce log line. return nil, ErrIgnoreFields } - fields := make(map[string]string, len(splitted)+8) - - fields["*"] = "*" - fields["$line"] = maprLine - fields["$empty"] = "" - fields["$hostname"] = p.hostname - fields["$server"] = p.hostname - fields["$timezone"] = p.timeZoneName - fields["$timeoffset"] = p.timeZoneOffset - - fields["$severity"] = splitted[0] - fields["$loglevel"] = splitted[0] - - time := splitted[1] - fields["$time"] = time - if len(time) == 15 { - // Example: 20211002-071209 - fields["$date"] = time[0:8] - fields["$hour"] = time[9:11] - fields["$minute"] = time[11:13] - fields["$second"] = time[13:] + return fields, nil +} + +func (p *defaultParser) addDefaultFields(fields map[string]string, maprLine string) { + if p.wantStar { + fields["*"] = "*" } - fields["$pid"] = splitted[2] - fields["$caller"] = splitted[3] - fields["$cpus"] = splitted[4] - fields["$goroutines"] = splitted[5] - fields["$cgocalls"] = splitted[6] - fields["$loadavg"] = splitted[7] - fields["$uptime"] = splitted[8] - - for _, kv := range splitted[10:] { - keyAndValue := strings.SplitN(kv, "=", 2) - if len(keyAndValue) != 2 { - return fields, fmt.Errorf("Unable to parse key-value token '%s'", kv) - } - fields[keyAndValue[0]] = keyAndValue[1] + if p.wantLine { + fields["$line"] = maprLine + } + if p.wantEmpty { + fields["$empty"] = "" + } + if p.wantHostname { + fields["$hostname"] = p.hostname + } + if p.wantServer { + fields["$server"] = p.hostname } + if p.wantTimezone { + fields["$timezone"] = p.timeZoneName + } + if p.wantTimeOffset { + fields["$timeoffset"] = p.timeZoneOffset + } +} - return fields, nil +func (p *defaultParser) addDynamicField(fields map[string]string, key string, value string) { + if p.allDynamicFields { + fields[key] = value + return + } + if _, ok := p.dynamicFields[key]; ok { + fields[key] = value + } +} + +func (p *defaultParser) addKeyValueField(fields map[string]string, token string) error { + keyAndValueIndex := strings.IndexByte(token, '=') + if keyAndValueIndex < 0 { + return fmt.Errorf("Unable to parse key-value token '%s'", token) + } + p.addDynamicField(fields, token[:keyAndValueIndex], token[keyAndValueIndex+1:]) + return nil +} + +func (p *defaultParser) configureFieldPlan(plan mapr.ParserFieldPlan) { + p.fieldsCapacity = plan.Capacity() + p.dynamicFields = nil + p.allDynamicFields = plan.AllFields + + p.wantStar = plan.Needs("*") + p.wantLine = plan.Needs("$line") + p.wantEmpty = plan.Needs("$empty") + p.wantHostname = plan.Needs("$hostname") + p.wantServer = plan.Needs("$server") + p.wantTimezone = plan.Needs("$timezone") + p.wantTimeOffset = plan.Needs("$timeoffset") + p.wantSeverity = plan.Needs("$severity") + p.wantLogLevel = plan.Needs("$loglevel") + p.wantTime = plan.Needs("$time") + p.wantDate = plan.Needs("$date") + p.wantHour = plan.Needs("$hour") + p.wantMinute = plan.Needs("$minute") + p.wantSecond = plan.Needs("$second") + p.wantPID = plan.Needs("$pid") + p.wantCaller = plan.Needs("$caller") + p.wantCPUs = plan.Needs("$cpus") + p.wantGoroutines = plan.Needs("$goroutines") + p.wantCGOCalls = plan.Needs("$cgocalls") + p.wantLoadAvg = plan.Needs("$loadavg") + p.wantUptime = plan.Needs("$uptime") + + if plan.AllFields { + return + } + + p.dynamicFields = make(map[string]struct{}, len(plan.Fields)) + for field := range plan.Fields { + p.dynamicFields[field] = struct{}{} + } } diff --git a/internal/mapr/logformat/default_benchmark_test.go b/internal/mapr/logformat/default_benchmark_test.go new file mode 100644 index 0000000..2eb468e --- /dev/null +++ b/internal/mapr/logformat/default_benchmark_test.go @@ -0,0 +1,44 @@ +package logformat + +import ( + "testing" + + "github.com/mimecast/dtail/internal/mapr" +) + +func BenchmarkDefaultParserMakeFields(b *testing.B) { + input := "INFO|20211002-072342|1|default_benchmark_test.go:0|8|14|7|0.21|471h0m21s|" + + "MAPREDUCE:STATS|foo=bar|bar=baz|qux=quux|alpha=beta|gamma=delta" + + b.Run("all_fields", func(b *testing.B) { + parser, err := NewParser("default", nil) + if err != nil { + b.Fatalf("Unable to create parser: %s", err.Error()) + } + + b.ReportAllocs() + for i := 0; i < b.N; i++ { + if _, err := parser.MakeFields(input, ""); err != nil { + b.Fatalf("Unable to parse input: %s", err.Error()) + } + } + }) + + b.Run("query_specific", func(b *testing.B) { + q, err := mapr.NewQuery(`select count(foo) from STATS where bar eq "baz"`) + if err != nil { + b.Fatalf("Unable to create query: %s", err.Error()) + } + parser, err := NewParser("default", q) + if err != nil { + b.Fatalf("Unable to create parser: %s", err.Error()) + } + + b.ReportAllocs() + for i := 0; i < b.N; i++ { + if _, err := parser.MakeFields(input, ""); err != nil { + b.Fatalf("Unable to parse input: %s", err.Error()) + } + } + }) +} diff --git a/internal/mapr/logformat/default_test.go b/internal/mapr/logformat/default_test.go index 4eae81b..992a052 100644 --- a/internal/mapr/logformat/default_test.go +++ b/internal/mapr/logformat/default_test.go @@ -3,6 +3,8 @@ package logformat import ( "fmt" "testing" + + "github.com/mimecast/dtail/internal/mapr" ) func TestDefaultLogFormat(t *testing.T) { @@ -23,7 +25,7 @@ func TestDefaultLogFormat(t *testing.T) { } for _, input := range inputs { - fields, err := parser.MakeFields(input) + fields, err := parser.MakeFields(input, "") if err != nil { t.Errorf("Parser unable to make fields: %s", err.Error()) @@ -86,12 +88,46 @@ func TestDefaultLogFormat(t *testing.T) { } } - fields, err := parser.MakeFields("foozoo=bar|bazbay") + fields, err := parser.MakeFields("foozoo=bar|bazbay", "") if err != nil && err != ErrIgnoreFields { - t.Errorf(err.Error()) + t.Errorf("%s", err.Error()) } if _, ok := fields["foo"]; ok { t.Errorf("Expected fiending field 'foo', but found it\n") } } + +func TestDefaultLogFormatQuerySpecificFields(t *testing.T) { + q, err := mapr.NewQuery(`select count(foo) from STATS where $hostname eq "testhost"`) + if err != nil { + t.Fatalf("Unable to create query: %s", err.Error()) + } + + parser, err := NewParser("default", q) + if err != nil { + t.Fatalf("Unable to create parser: %s", err.Error()) + } + + fields, err := parser.MakeFields( + "INFO|20211002-072342|1|default_test.go:0|8|14|7|0.21|471h0m21s|MAPREDUCE:STATS|foo=bar|bar=baz", + "", + ) + if err != nil { + t.Fatalf("Parser unable to make fields: %s", err.Error()) + } + + requiredFields := []string{"foo", "$hostname"} + for _, field := range requiredFields { + if _, ok := fields[field]; !ok { + t.Errorf("Expected query-specific field '%s' to be present", field) + } + } + + omittedFields := []string{"bar", "$time", "$pid", "$line"} + for _, field := range omittedFields { + if _, ok := fields[field]; ok { + t.Errorf("Expected query-specific field '%s' to be omitted", field) + } + } +} diff --git a/internal/mapr/logformat/delimited.go b/internal/mapr/logformat/delimited.go new file mode 100644 index 0000000..2fa0639 --- /dev/null +++ b/internal/mapr/logformat/delimited.go @@ -0,0 +1,12 @@ +package logformat + +import "strings" + +func scanDelimitedField(input string, start int, delimiter byte) (token string, next int, done bool) { + index := strings.IndexByte(input[start:], delimiter) + if index < 0 { + return input[start:], len(input), true + } + end := start + index + return input[start:end], end + 1, false +} diff --git a/internal/mapr/logformat/generic.go b/internal/mapr/logformat/generic.go index 32d9b4a..4f6d58b 100644 --- a/internal/mapr/logformat/generic.go +++ b/internal/mapr/logformat/generic.go @@ -4,6 +4,8 @@ type genericParser struct { defaultParser } +var _ Parser = (*genericParser)(nil) + func newGenericParser(hostname, timeZoneName string, timeZoneOffset int) (*genericParser, error) { defaultParser, err := newDefaultParser(hostname, timeZoneName, timeZoneOffset) if err != nil { @@ -12,16 +14,9 @@ func newGenericParser(hostname, timeZoneName string, timeZoneOffset int) (*gener return &genericParser{defaultParser: *defaultParser}, nil } -func (p *genericParser) MakeFields(maprLine string) (map[string]string, error) { - fields := make(map[string]string, 3) - - fields["*"] = "*" - fields["$hostname"] = p.hostname - fields["$server"] = p.hostname - fields["$line"] = maprLine - fields["$empty"] = "" - fields["$timezone"] = p.timeZoneName - fields["$timeoffset"] = p.timeZoneOffset +func (p *genericParser) MakeFields(maprLine, _ string) (map[string]string, error) { + fields := make(map[string]string, p.fieldsCapacity) + p.addDefaultFields(fields, maprLine) return fields, nil } diff --git a/internal/mapr/logformat/generickv.go b/internal/mapr/logformat/generickv.go index 9c3de92..5a0595d 100644 --- a/internal/mapr/logformat/generickv.go +++ b/internal/mapr/logformat/generickv.go @@ -1,15 +1,13 @@ package logformat -import ( - "strings" - - "github.com/mimecast/dtail/internal/protocol" -) +import "github.com/mimecast/dtail/internal/protocol" type genericKVParser struct { defaultParser } +var _ Parser = (*genericKVParser)(nil) + func newGenericKVParser(hostname, timeZoneName string, timeZoneOffset int) (*genericKVParser, error) { defaultParser, err := newDefaultParser(hostname, timeZoneName, timeZoneOffset) if err != nil { @@ -18,25 +16,21 @@ func newGenericKVParser(hostname, timeZoneName string, timeZoneOffset int) (*gen return &genericKVParser{defaultParser: *defaultParser}, nil } -func (p *genericKVParser) MakeFields(maprLine string) (map[string]string, error) { - splitted := strings.Split(maprLine, protocol.FieldDelimiter) - fields := make(map[string]string, len(splitted)) +func (p *genericKVParser) MakeFields(maprLine, _ string) (map[string]string, error) { + fields := make(map[string]string, p.fieldsCapacity) + p.addDefaultFields(fields, maprLine) + start := 0 + delimiter := protocol.FieldDelimiter[0] - fields["*"] = "*" - fields["$line"] = maprLine - fields["$empty"] = "" - fields["$hostname"] = p.hostname - fields["$server"] = p.hostname - fields["$timezone"] = p.timeZoneName - fields["$timeoffset"] = p.timeZoneOffset - - for _, kv := range splitted[0:] { - keyAndValue := strings.SplitN(kv, "=", 2) - if len(keyAndValue) != 2 { - //dlog.Common.Debug("Unable to parse key-value token, ignoring it", kv) + for { + token, next, done := scanDelimitedField(maprLine, start, delimiter) + if err := p.addKeyValueField(fields, token); err != nil { continue } - fields[keyAndValue[0]] = keyAndValue[1] + if done { + break + } + start = next } return fields, nil diff --git a/internal/mapr/logformat/mimecast.go b/internal/mapr/logformat/mimecast.go index cf6b333..249e2db 100644 --- a/internal/mapr/logformat/mimecast.go +++ b/internal/mapr/logformat/mimecast.go @@ -1,5 +1,4 @@ //go:build !proprietary -// +build !proprietary package logformat @@ -10,6 +9,8 @@ var ErrMimecastNotAvailable error = errors.New("The mimecast logformat is not av type mimecastParser struct{} +var _ Parser = (*mimecastParser)(nil) + func newMimecastParser(hostname, timeZoneName string, timeZoneOffset int) (*mimecastParser, error) { return &mimecastParser{}, ErrMimecastNotAvailable } @@ -18,6 +19,6 @@ func newMimecastGenericParser(hostname, timeZoneName string, timeZoneOffset int) return &mimecastParser{}, ErrMimecastNotAvailable } -func (p *mimecastParser) MakeFields(maprLine string) (map[string]string, error) { +func (p *mimecastParser) MakeFields(maprLine, _ string) (map[string]string, error) { return nil, ErrMimecastNotAvailable } diff --git a/internal/mapr/logformat/parser.go b/internal/mapr/logformat/parser.go index 37d7a63..0556c8b 100644 --- a/internal/mapr/logformat/parser.go +++ b/internal/mapr/logformat/parser.go @@ -3,6 +3,8 @@ package logformat import ( "errors" "fmt" + "strings" + "sync" "time" "github.com/mimecast/dtail/internal/config" @@ -14,8 +16,72 @@ var ErrIgnoreFields error = errors.New("Ignore this field set") // Parser is used to parse the mapreduce information from the server log files. type Parser interface { - // MakeFields creates a field map from an input log line. - MakeFields(string) (map[string]string, error) + // MakeFields creates a field map from an input log line. The sourceID + // identifies the log file (or stream) the line belongs to so that + // stateful parsers (e.g. CSV with per-file headers) can key their + // state per source instead of smearing it across every file in a + // session. + MakeFields(maprLine, sourceID string) (map[string]string, error) +} + +type queryAwareParser interface { + setQuery(*mapr.Query) +} + +// ParserFactory builds a Parser for a specific log format. +type ParserFactory func(hostname, timeZoneName string, timeZoneOffset int) (Parser, error) + +var parserFactories = make(map[string]ParserFactory) +var parserFactoriesMu sync.RWMutex + +func init() { + registerBuiltInParsers() +} + +// RegisterParser registers or replaces a parser factory for a log format name. +func RegisterParser(logFormatName string, factory ParserFactory) error { + name := strings.TrimSpace(logFormatName) + if name == "" { + return errors.New("log format name cannot be empty") + } + if factory == nil { + return errors.New("parser factory cannot be nil") + } + + parserFactoriesMu.Lock() + defer parserFactoriesMu.Unlock() + parserFactories[name] = factory + return nil +} + +func getParserFactory(logFormatName string) (ParserFactory, bool) { + parserFactoriesMu.RLock() + defer parserFactoriesMu.RUnlock() + factory, found := parserFactories[logFormatName] + return factory, found +} + +func registerBuiltInParsers() { + mustRegisterParser("generic", wrapParserFactory(newGenericParser)) + mustRegisterParser("generickv", wrapParserFactory(newGenericKVParser)) + mustRegisterParser("csv", wrapParserFactory(newCSVParser)) + mustRegisterParser("mimecast", wrapParserFactory(newMimecastParser)) + mustRegisterParser("mimecastgeneric", wrapParserFactory(newMimecastGenericParser)) + mustRegisterParser("default", wrapParserFactory(newDefaultParser)) + mustRegisterParser("custom1", wrapParserFactory(newCustom1Parser)) + mustRegisterParser("custom2", wrapParserFactory(newCustom2Parser)) +} + +func mustRegisterParser(logFormatName string, factory ParserFactory) { + if err := RegisterParser(logFormatName, factory); err != nil { + panic(err) + } +} + +func wrapParserFactory[T Parser](factory func(string, string, int) (T, error)) ParserFactory { + return func(hostname, timeZoneName string, timeZoneOffset int) (Parser, error) { + return factory(hostname, timeZoneName, timeZoneOffset) + } } // NewParser returns a new log parser. @@ -27,30 +93,33 @@ func NewParser(logFormatName string, query *mapr.Query) (Parser, error) { now := time.Now() timeZoneName, timeZoneOffset := now.Zone() - // Extend this for adding more log formats! - switch logFormatName { - case "generic": - return newGenericParser(hostname, timeZoneName, timeZoneOffset) - case "generickv": - return newGenericKVParser(hostname, timeZoneName, timeZoneOffset) - case "csv": - return newCSVParser(hostname, timeZoneName, timeZoneOffset) - case "mimecast": - return newMimecastParser(hostname, timeZoneName, timeZoneOffset) - case "mimecastgeneric": - return newMimecastGenericParser(hostname, timeZoneName, timeZoneOffset) - case "default": - return newDefaultParser(hostname, timeZoneName, timeZoneOffset) - case "custom1": - return newCustom1Parser(hostname, timeZoneName, timeZoneOffset) - case "custom2": - return newCustom2Parser(hostname, timeZoneName, timeZoneOffset) - default: - p, err := newDefaultParser(hostname, timeZoneName, timeZoneOffset) - if err != nil { - return p, fmt.Errorf("No '%s' mapr log format and problem creating default one: %v", - logFormatName, err) - } - return p, fmt.Errorf("No '%s' mapr log format", logFormatName) + if parserFactory, found := getParserFactory(logFormatName); found { + parser, err := parserFactory(hostname, timeZoneName, timeZoneOffset) + configureParserQuery(parser, query) + return parser, err + } + + defaultFactory, found := getParserFactory("default") + if !found { + return nil, fmt.Errorf("No '%s' mapr log format and no default parser registered", logFormatName) + } + + p, err := defaultFactory(hostname, timeZoneName, timeZoneOffset) + if err != nil { + return p, fmt.Errorf("No '%s' mapr log format and problem creating default one: %v", + logFormatName, err) + } + configureParserQuery(p, query) + return p, fmt.Errorf("No '%s' mapr log format", logFormatName) +} + +func configureParserQuery(parser Parser, query *mapr.Query) { + if parser == nil { + return + } + queryAware, ok := parser.(queryAwareParser) + if !ok { + return } + queryAware.setQuery(query) } diff --git a/internal/mapr/logformat/parser_test.go b/internal/mapr/logformat/parser_test.go new file mode 100644 index 0000000..3b319c4 --- /dev/null +++ b/internal/mapr/logformat/parser_test.go @@ -0,0 +1,69 @@ +package logformat + +import ( + "strings" + "testing" +) + +type testParser struct{} + +func (p *testParser) MakeFields(maprLine, _ string) (map[string]string, error) { + return map[string]string{"line": maprLine}, nil +} + +func TestRegisterParserValidation(t *testing.T) { + if err := RegisterParser("", wrapParserFactory(newDefaultParser)); err == nil { + t.Errorf("Expected error when registering parser with empty name") + } + + if err := RegisterParser("test-nil-factory", nil); err == nil { + t.Errorf("Expected error when registering parser with nil factory") + } +} + +func TestNewParserUsesRegistry(t *testing.T) { + const parserName = "unit-test-registry-parser" + + if err := RegisterParser(parserName, func(string, string, int) (Parser, error) { + return &testParser{}, nil + }); err != nil { + t.Fatalf("Unable to register parser: %s", err.Error()) + } + + parser, err := NewParser(parserName, nil) + if err != nil { + t.Fatalf("Unable to create parser from registry: %s", err.Error()) + } + + fields, err := parser.MakeFields("hello", "") + if err != nil { + t.Fatalf("Unable to parse line: %s", err.Error()) + } + if fields["line"] != "hello" { + t.Errorf("Expected custom parser output, got '%s'", fields["line"]) + } +} + +func TestNewParserFallbackToDefault(t *testing.T) { + parser, err := NewParser("missing-parser-format", nil) + if err == nil { + t.Fatalf("Expected NewParser to return error for missing parser format") + } + if !strings.Contains(err.Error(), "No 'missing-parser-format' mapr log format") { + t.Errorf("Unexpected error message: %s", err.Error()) + } + if parser == nil { + t.Fatalf("Expected default parser fallback when format is missing") + } + + fields, parseErr := parser.MakeFields( + "INFO|20211002-072342|1|parser_test.go:0|8|14|7|0.21|471h0m21s|MAPREDUCE:STATS|foo=bar", + "", + ) + if parseErr != nil { + t.Fatalf("Fallback parser failed to parse line: %s", parseErr.Error()) + } + if val, ok := fields["$severity"]; !ok || val != "INFO" { + t.Errorf("Fallback parser did not behave like default parser") + } +} diff --git a/internal/mapr/logformat/variables.go b/internal/mapr/logformat/variables.go new file mode 100644 index 0000000..541386b --- /dev/null +++ b/internal/mapr/logformat/variables.go @@ -0,0 +1,107 @@ +package logformat + +import ( + "fmt" + "strings" + + "github.com/mimecast/dtail/internal/mapr" +) + +// The $-variable sets below enumerate every $-prefixed variable a given parser +// is able to populate. They are the single source of truth for the plan-time +// "unknown $-variable" diagnostic (see PlanVariableWarnings). +// +// Keep them in sync with defaultParser in default.go: +// - addDefaultFields populates commonVariables for every parser that embeds +// defaultParser (generic, generickv, csv and default itself). +// - defaultParser.MakeFields additionally populates defaultOnlyVariables from +// the positional fields of DTail's own MAPREDUCE log line layout; the +// lighter parsers (generic/generickv/csv) override MakeFields and therefore +// do NOT populate these. + +// commonVariables are the $-variables set by defaultParser.addDefaultFields and +// are therefore available in every built-in log format. +var commonVariables = map[string]struct{}{ + "$line": {}, + "$empty": {}, + "$hostname": {}, + "$server": {}, + "$timezone": {}, + "$timeoffset": {}, +} + +// defaultOnlyVariables are the extra $-variables that only the "default" parser +// extracts from DTail's MAPREDUCE log lines (see defaultParser.MakeFields). +var defaultOnlyVariables = map[string]struct{}{ + "$severity": {}, + "$loglevel": {}, + "$time": {}, + "$date": {}, + "$hour": {}, + "$minute": {}, + "$second": {}, + "$pid": {}, + "$caller": {}, + "$cpus": {}, + "$goroutines": {}, + "$cgocalls": {}, + "$loadavg": {}, + "$uptime": {}, +} + +// knownVariables returns the set of $-variables the named parser can populate, +// and whether that set is enumerable at all. For log formats whose variable set +// cannot be determined statically (proprietary, stub or unknown formats) it +// returns (nil, false) so callers skip the diagnostic entirely rather than emit +// false-positive warnings for variables that might in fact be valid. +func knownVariables(logFormatName string) (map[string]struct{}, bool) { + switch logFormatName { + case "default": + known := make(map[string]struct{}, len(commonVariables)+len(defaultOnlyVariables)) + for name := range commonVariables { + known[name] = struct{}{} + } + for name := range defaultOnlyVariables { + known[name] = struct{}{} + } + return known, true + case "generic", "generickv", "csv": + return commonVariables, true + default: + return nil, false + } +} + +// PlanVariableWarnings returns one warning per unknown $-variable referenced by +// the query, given the parser selected by logFormatName. A $-variable is +// "unknown" when the selected parser cannot populate it and the query does not +// define it via a `set` clause; at runtime such a variable silently resolves to +// the empty string, collapsing an aggregation into a single empty group with no +// other diagnostic. This is a WARNING and not an error: sparse dynamic fields +// and built-ins like $empty are legitimate, so resolution behaviour is +// unchanged. +// +// No warnings are produced when the parser's variable set is not enumerable +// (see knownVariables); guessing there would risk false positives, which are +// worse than none because they train users to ignore warnings. +func PlanVariableWarnings(query *mapr.Query, logFormatName string) []string { + if query == nil { + return nil + } + known, enumerable := knownVariables(logFormatName) + if !enumerable { + return nil + } + + var warnings []string + for _, variable := range query.ReferencedVariables() { + if _, ok := known[variable]; ok { + continue + } + warnings = append(warnings, fmt.Sprintf( + "warning: %s is not a known variable for log format %q; "+ + "did you mean bareword %s?", + variable, logFormatName, strings.TrimPrefix(variable, "$"))) + } + return warnings +} diff --git a/internal/mapr/logformat/variables_test.go b/internal/mapr/logformat/variables_test.go new file mode 100644 index 0000000..7843855 --- /dev/null +++ b/internal/mapr/logformat/variables_test.go @@ -0,0 +1,145 @@ +package logformat + +import ( + "strings" + "testing" + + "github.com/mimecast/dtail/internal/mapr" +) + +// TestPlanVariableWarnings locks the plan-time unknown-$-variable diagnostic. +// The warning exists to surface the silent footgun where a mistyped or wrongly +// $-prefixed field (e.g. "$service" instead of the bareword "service", or +// "$time" without a "from"/logformat clause) collapses an aggregation into a +// single empty group with zero other diagnostics. +func TestPlanVariableWarnings(t *testing.T) { + tests := []struct { + name string + // query is parsed and its parser resolved via EffectiveLogFormat so the + // test exercises the same selection rule the client uses. + query string + // wantVars lists the $-variables that must be flagged (each appears in + // exactly one warning line). + wantVars []string + // unwantVars lists names that must NOT appear in any warning. + unwantVars []string + }{ + { + name: "ys0 footgun: $-prefixed dynamic field with from clause", + query: "from STATS select $service,sum($bytes) group by $service", + wantVars: []string{"$service", "$bytes"}, + unwantVars: []string{"service", "bytes"}, + }, + { + name: "generic parser without from warns for default-format $time", + query: "select $time,$line", + wantVars: []string{"$time"}, + unwantVars: []string{"$line"}, + }, + { + name: "valid default-format variables do not warn", + query: "from STATS select $hostname,max($goroutines),$time group by $hostname", + wantVars: nil, + unwantVars: []string{"$hostname", "$goroutines", "$time"}, + }, + { + name: "barewords never warn", + query: "from STATS select service,sum(bytes) group by service", + wantVars: nil, + unwantVars: []string{"service", "bytes", "$service", "$bytes"}, + }, + { + name: "built-in $empty does not warn", + query: "from STATS select $empty,count($line) group by $empty", + wantVars: nil, + unwantVars: []string{"$empty", "$line"}, + }, + { + name: "set-defined variable does not warn", + query: "from STATS select $masked group by $masked set $masked = maskdigits($line)", + wantVars: nil, + unwantVars: []string{"$masked"}, + }, + { + name: "unknown variable referenced only in where clause warns", + query: "from STATS select $line where $bogus eq foo", + wantVars: []string{"$bogus"}, + unwantVars: []string{"$line"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + query, err := mapr.NewQuery(tc.query) + if err != nil { + t.Fatalf("NewQuery(%q) failed: %v", tc.query, err) + } + logFormat := query.EffectiveLogFormat("") + warnings := PlanVariableWarnings(query, logFormat) + + joined := strings.Join(warnings, "\n") + for _, want := range tc.wantVars { + if countVarWarnings(warnings, want) != 1 { + t.Errorf("expected exactly one warning for %q, got warnings:\n%s", + want, joined) + } + } + for _, unwant := range tc.unwantVars { + if countVarWarnings(warnings, unwant) != 0 { + t.Errorf("did not expect a warning for %q, got warnings:\n%s", + unwant, joined) + } + } + }) + } +} + +// TestPlanVariableWarningsText locks the exact warning wording so downstream +// tooling and users can rely on it. +func TestPlanVariableWarningsText(t *testing.T) { + query, err := mapr.NewQuery("from STATS select $service group by $service") + if err != nil { + t.Fatalf("NewQuery failed: %v", err) + } + warnings := PlanVariableWarnings(query, query.EffectiveLogFormat("")) + if len(warnings) != 1 { + t.Fatalf("expected exactly one warning (deduped), got %d: %v", len(warnings), warnings) + } + want := `warning: $service is not a known variable for log format "default"; did you mean bareword service?` + if warnings[0] != want { + t.Errorf("warning text mismatch:\n got: %s\nwant: %s", warnings[0], want) + } +} + +// TestPlanVariableWarningsNonEnumerable ensures that log formats whose variable +// set cannot be determined statically produce no warnings, avoiding false +// positives that would train users to ignore the diagnostic. +func TestPlanVariableWarningsNonEnumerable(t *testing.T) { + query, err := mapr.NewQuery("from STATS select $whatever group by $whatever") + if err != nil { + t.Fatalf("NewQuery failed: %v", err) + } + for _, format := range []string{"mimecast", "custom1", "custom2", "doesnotexist"} { + if warnings := PlanVariableWarnings(query, format); len(warnings) != 0 { + t.Errorf("expected no warnings for non-enumerable format %q, got: %v", + format, warnings) + } + } +} + +// countVarWarnings counts warnings whose SUBJECT is the given variable, i.e. +// "warning: is not a known variable ...". It deliberately matches +// only the subject and not the "did you mean bareword X" suggestion, so that a +// bareword appearing in a suggestion is not mistaken for a warning about that +// bareword. The trailing " is not" also prevents "$service" from matching a +// hypothetical "$services" subject. +func countVarWarnings(warnings []string, variable string) int { + subject := "warning: " + variable + " is not a known variable" + count := 0 + for _, w := range warnings { + if strings.HasPrefix(w, subject) { + count++ + } + } + return count +} -- cgit v1.2.3