fix(logs): keep the agent's pattern.hash, read Datadog span ids as decimal, throttle the freopen scan

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 {

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

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()

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

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"])
+}

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