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..78e3f74 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,33 +111,78 @@ 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 { - if id, err := trace.TraceIDFromHex(strings.ToLower(v)); err == nil { + if id, ok := parseTraceId(k, v); ok { traceId = &id continue } } 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 } +// 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 isDatadogKey(k) { + 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 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 new file mode 100644 index 0000000..ecbd3ee --- /dev/null +++ b/logs/otel_test.go @@ -0,0 +1,54 @@ +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 + _, 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) + 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