From a23bcab1747a13c54ad145d3716637e0a5b4bff6 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Fri, 9 Oct 2026 13:19:56 +0530 Subject: [PATCH 1/2] fix(logs): keep the agent's pattern.hash, read Datadog span ids as decimal, throttle the freopen scan From the #380 review: - A JSON field named pattern.hash no longer duplicates or replaces the agent's attribute. - dd.span_id is decimal (Datadog's log injection); a 16-digit value was read as hex. A span id without a trace id stays an attribute. - The freopen fallback reads a process's fds at most every 10s: a process that opens its /var/log file for every write triggered it on every write. - A tail reader waiting for its file to appear logs that once. - The pattern extraction limit is per log source, not per container: say so. --- containers/container.go | 24 +++++++++++++++++++++- containers/process.go | 15 ++++++++------ flags/flags.go | 2 +- logs/otel.go | 36 +++++++++++++++++++++++++++++---- logs/otel_test.go | 44 +++++++++++++++++++++++++++++++++++++++++ logs/tail_reader.go | 4 +++- 6 files changed, 112 insertions(+), 13 deletions(-) create mode 100644 logs/otel_test.go diff --git a/containers/container.go b/containers/container.go index 07925ea..d25a4a5 100644 --- a/containers/container.go +++ b/containers/container.go @@ -715,7 +715,9 @@ func (c *Container) onFileOpen(pid uint32, fd uint64, mnt uint64, log bool) { case log && (info == nil || !strings.HasPrefix(info.Dest, "/var/log/")): // The kernel saw a /var/log/ file opened, but the fd now points // elsewhere (freopen): find the log files the process holds. - logPaths = findLogFiles(pid) + if c.logFilesScanDue(pid) { + logPaths = findLogFiles(pid) + } } for _, logPath := range logPaths { c.lock.Lock() @@ -2509,6 +2511,26 @@ func sampleString(v interface{}) string { return "" } +// logFilesScanInterval bounds how often findLogFiles runs for a process: one +// that opens its log for every write (PHP's error_log, a shell's >>) would +// otherwise have all its fds read on every write. +const logFilesScanInterval = 10 * time.Second + +func (c *Container) logFilesScanDue(pid uint32) bool { + c.lock.Lock() + defer c.lock.Unlock() + p := c.processes[pid] + if p == nil { + return true + } + now := time.Now() + if now.Sub(p.logFilesScannedAt) < logFilesScanInterval { + return false + } + p.logFilesScannedAt = now + return true +} + func findLogFiles(pid uint32) []string { fds, err := proc.ReadFds(pid) if err != nil { diff --git a/containers/process.go b/containers/process.go index 824736d..931f47f 100644 --- a/containers/process.go +++ b/containers/process.go @@ -53,12 +53,15 @@ type Process struct { // TLS attach state, only touched by the registry's event loop. goTlsUprobesChecked bool openSslUprobesChecked bool - openSslChecks int - openSslLastCheck time.Time - tlsAttached bool - tlsExe exeIdentity - tlsExeName string - tlsExeCheckedAt time.Time + + // last findLogFiles scan, only touched by the registry's event loop + logFilesScannedAt time.Time + openSslChecks int + openSslLastCheck time.Time + tlsAttached bool + tlsExe exeIdentity + tlsExeName string + tlsExeCheckedAt time.Time // Only touched by the instrument goroutine. pythonGilChecked bool diff --git a/flags/flags.go b/flags/flags.go index 1d04212..f5ac32d 100644 --- a/flags/flags.go +++ b/flags/flags.go @@ -54,7 +54,7 @@ var ( LogPerSecond = kingpin.Flag("log-per-second", "The number of logs per second").Default("10.0").Envar("LOG_PER_SECOND").Float64() LogBurst = kingpin.Flag("log-burst", "The maximum number of tokens that can be consumed in a single call to allow").Default("100").Envar("LOG_BURST").Int() LogPatternsPerContainer = kingpin.Flag("log-patterns-per-container", "Max unique log patterns per container per level").Default("256").Envar("LOG_PATTERNS_PER_CONTAINER").Int() - LogPatternExtractionLimit = kingpin.Flag("log-pattern-extraction-limit", "Max warning/error log messages per second per container for which patterns are extracted. Over-limit messages are still exported and counted, under a dedicated 'event was sampled' pattern (0 - unlimited)").Default("0").Envar("LOG_PATTERN_EXTRACTION_LIMIT").Float64() + LogPatternExtractionLimit = kingpin.Flag("log-pattern-extraction-limit", "Max warning/error log messages per second per log source (a file or a container's stdout/stderr) for which patterns are extracted. Over-limit messages are still exported and counted, under a dedicated 'event was sampled' pattern (0 - unlimited)").Default("0").Envar("LOG_PATTERN_EXTRACTION_LIMIT").Float64() MaxFQDNsPerContainer = kingpin.Flag("max-fqdns-per-container", "Max unique FQDN values per container, extras are bucketed under '~other'").Default("50").Envar("MAX_FQDNS_PER_CONTAINER").Int() MaxLabelLength = kingpin.Flag("max-label-length", "Maximum length of a metric label value").Default("4096").Envar("MAX_LABEL_LENGTH").Int() diff --git a/logs/otel.go b/logs/otel.go index 5614ade..d3475f9 100644 --- a/logs/otel.go +++ b/logs/otel.go @@ -2,6 +2,8 @@ package logs import ( "context" + "encoding/binary" + "strconv" "strings" "time" @@ -23,7 +25,7 @@ import ( var otelLogger otelLogs.Logger -// PatternExtractionRateLimiter caps, per container, how many warning and +// PatternExtractionRateLimiter caps, per log source, how many warning and // error messages per second get a log pattern extracted (nil: unlimited). func PatternExtractionRateLimiter() *rate.Limiter { limit := *flags.LogPatternExtractionLimit @@ -109,12 +111,18 @@ func normalizeTraceContextKey(k string) string { return "" } +const patternHashKey = "pattern.hash" + func logRecordAttrs(patternHash string, attributes map[string]string) ([]attribute.KeyValue, *trace.TraceID, *trace.SpanID) { var traceId *trace.TraceID var spanId *trace.SpanID + var spanIdAttr string attrs := make([]attribute.KeyValue, 0, len(attributes)+1) - attrs = append(attrs, attribute.Key("pattern.hash").String(patternHash)) + attrs = append(attrs, attribute.Key(patternHashKey).String(patternHash)) for k, v := range attributes { + if k == patternHashKey { // the agent's own; a duplicate key would hide it + continue + } switch normalizeTraceContextKey(k) { case traceIdKey: if traceId == nil { @@ -125,17 +133,37 @@ func logRecordAttrs(patternHash string, attributes map[string]string) ([]attribu } case spanIdKey: if spanId == nil { - if id, err := trace.SpanIDFromHex(strings.ToLower(v)); err == nil { - spanId = &id + if id, ok := parseSpanId(k, v); ok { + spanId, spanIdAttr = &id, k continue } } } attrs = append(attrs, attribute.Key(k).String(v)) } + if spanId != nil && traceId == nil { // a span id means nothing without its trace + attrs = append(attrs, attribute.Key(spanIdAttr).String(attributes[spanIdAttr])) + spanId = nil + } return attrs, traceId, spanId } +// parseSpanId parses a span id, which Datadog's log injection (dd.span_id) +// writes as a decimal number and everything else as hex. +func parseSpanId(k, v string) (trace.SpanID, bool) { + if strings.HasPrefix(strings.ToLower(k), "dd.") { + n, err := strconv.ParseUint(v, 10, 64) + if err != nil || n == 0 { + return trace.SpanID{}, false + } + var id trace.SpanID + binary.BigEndian.PutUint64(id[:], n) + return id, true + } + id, err := trace.SpanIDFromHex(strings.ToLower(v)) + return id, err == nil +} + func OtelLogEmitter(containerId string) logparser.OnMsgCallbackF { if otelLogger == nil { return nil diff --git a/logs/otel_test.go b/logs/otel_test.go new file mode 100644 index 0000000..2d68a22 --- /dev/null +++ b/logs/otel_test.go @@ -0,0 +1,44 @@ +package logs + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "go.opentelemetry.io/otel/attribute" +) + +func attrMap(attrs []attribute.KeyValue) map[string][]string { + m := map[string][]string{} + for _, a := range attrs { + m[string(a.Key)] = append(m[string(a.Key)], a.Value.AsString()) + } + return m +} + +func TestLogRecordAttrs(t *testing.T) { + // a JSON field named like the agent's attribute neither duplicates nor replaces it + attrs, _, _ := logRecordAttrs("1234", map[string]string{"pattern.hash": "spoofed", "user": "alice"}) + assert.Equal(t, map[string][]string{"pattern.hash": {"1234"}, "user": {"alice"}}, attrMap(attrs)) + + // hex trace context + attrs, traceId, spanId := logRecordAttrs("1", map[string]string{ + "trace_id": "4BF92F3577B34DA6A3CE929D0E0E4736", + "span_id": "00f067aa0ba902b7", + }) + assert.Equal(t, "4bf92f3577b34da6a3ce929d0e0e4736", traceId.String()) + assert.Equal(t, "00f067aa0ba902b7", spanId.String()) + assert.Len(t, attrs, 1) + + // Datadog writes span ids in decimal: 16 digits must not be read as hex + _, _, spanId = logRecordAttrs("1", map[string]string{ + "dd.trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", + "dd.span_id": "1234567890123456", + }) + assert.Equal(t, "000462d53c8abac0", spanId.String()) + + // a span id without a trace id stays an attribute + attrs, traceId, spanId = logRecordAttrs("1", map[string]string{"span_id": "00f067aa0ba902b7"}) + assert.Nil(t, traceId) + assert.Nil(t, spanId) + assert.Equal(t, []string{"00f067aa0ba902b7"}, attrMap(attrs)["span_id"]) +} diff --git a/logs/tail_reader.go b/logs/tail_reader.go index ac0fb38..2721b22 100644 --- a/logs/tail_reader.go +++ b/logs/tail_reader.go @@ -40,7 +40,9 @@ func NewTailReader(fileName string, ch chan<- logparser.LogEntry) (*TailReader, if r.file, err = os.Open(fileName); err != nil && !os.IsNotExist(err) { return nil, err } - if r.file != nil { // the file may not exist yet, poll() waits for it to appear + if r.file == nil { + klog.Infof("%s does not exist yet, waiting for it to appear", fileName) + } else { if r.info, err = r.file.Stat(); err != nil { _ = r.file.Close() return nil, err From 3cce5116c5749865743ae91c7b150f690e218927 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Fri, 9 Oct 2026 22:00:24 +0530 Subject: [PATCH 2/2] fix(logs): read Datadog's decimal dd.trace_id as the lower half of the trace id Datadog's log injection writes 64-bit trace ids in decimal. They were not parsed, so with this branch's decimal span ids a record still got no trace context: a span id without a trace id stays an attribute (#390 review). --- logs/otel.go | 23 +++++++++++++++++++++-- logs/otel_test.go | 14 ++++++++++++-- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/logs/otel.go b/logs/otel.go index d3475f9..78e3f74 100644 --- a/logs/otel.go +++ b/logs/otel.go @@ -126,7 +126,7 @@ func logRecordAttrs(patternHash string, attributes map[string]string) ([]attribu switch normalizeTraceContextKey(k) { case traceIdKey: if traceId == nil { - if id, err := trace.TraceIDFromHex(strings.ToLower(v)); err == nil { + if id, ok := parseTraceId(k, v); ok { traceId = &id continue } @@ -148,10 +148,25 @@ func logRecordAttrs(patternHash string, attributes map[string]string) ([]attribu return attrs, traceId, spanId } +// parseTraceId parses a trace id. Datadog's log injection writes dd.trace_id +// as a decimal 64-bit number unless 128-bit ids are logged, as 32 hex +// characters; the decimal form is the lower half of the OpenTelemetry id. +func parseTraceId(k, v string) (trace.TraceID, bool) { + if isDatadogKey(k) { + if n, err := strconv.ParseUint(v, 10, 64); err == nil { + var id trace.TraceID + binary.BigEndian.PutUint64(id[8:], n) + return id, n != 0 + } + } + id, err := trace.TraceIDFromHex(strings.ToLower(v)) + return id, err == nil +} + // parseSpanId parses a span id, which Datadog's log injection (dd.span_id) // writes as a decimal number and everything else as hex. func parseSpanId(k, v string) (trace.SpanID, bool) { - if strings.HasPrefix(strings.ToLower(k), "dd.") { + if isDatadogKey(k) { n, err := strconv.ParseUint(v, 10, 64) if err != nil || n == 0 { return trace.SpanID{}, false @@ -164,6 +179,10 @@ func parseSpanId(k, v string) (trace.SpanID, bool) { return id, err == nil } +func isDatadogKey(k string) bool { + return strings.HasPrefix(strings.ToLower(k), "dd.") +} + func OtelLogEmitter(containerId string) logparser.OnMsgCallbackF { if otelLogger == nil { return nil diff --git a/logs/otel_test.go b/logs/otel_test.go index 2d68a22..ecbd3ee 100644 --- a/logs/otel_test.go +++ b/logs/otel_test.go @@ -30,12 +30,22 @@ func TestLogRecordAttrs(t *testing.T) { assert.Len(t, attrs, 1) // Datadog writes span ids in decimal: 16 digits must not be read as hex - _, _, spanId = logRecordAttrs("1", map[string]string{ - "dd.trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", + _, traceId, spanId = logRecordAttrs("1", map[string]string{ + "dd.trace_id": "4bf92f3577b34da6a3ce929d0e0e4736", // 128-bit, logged as hex "dd.span_id": "1234567890123456", }) + assert.Equal(t, "4bf92f3577b34da6a3ce929d0e0e4736", traceId.String()) assert.Equal(t, "000462d53c8abac0", spanId.String()) + // ...and 64-bit trace ids in decimal, the lower half of the OpenTelemetry id + attrs, traceId, spanId = logRecordAttrs("1", map[string]string{ + "dd.trace_id": "1234567890123456", + "dd.span_id": "987654321", + }) + assert.Equal(t, "0000000000000000000462d53c8abac0", traceId.String()) + assert.Equal(t, "000000003ade68b1", spanId.String()) + assert.Len(t, attrs, 1) + // a span id without a trace id stays an attribute attrs, traceId, spanId = logRecordAttrs("1", map[string]string{"span_id": "00f067aa0ba902b7"}) assert.Nil(t, traceId)