diff --git a/internal/component/loki/process/stages/docker.go b/internal/component/loki/process/stages/docker.go index c3bb290d8cc..1404d7c4fdd 100644 --- a/internal/component/loki/process/stages/docker.go +++ b/internal/component/loki/process/stages/docker.go @@ -38,28 +38,36 @@ const ( ) func (d *DockerStage) Process(labels model.LabelSet, extracted map[string]any, t *time.Time, entry *string) { - var log DockerLog - - if err := json.Unmarshal([]byte(*entry), &log); err != nil { + var parsed DockerLog + if err := json.Unmarshal([]byte(*entry), &parsed); err != nil { if Debug { level.Debug(d.logger).Log("msg", "failed to parse docker log", "err", err) } return } + // NOTE: json.Unmarshal will happily parse any JSON and produce a zero-value struct. + // To protect against incorrect usage, validate that the log field is present. + if parsed.Log == "" { + if Debug { + level.Debug(d.logger).Log("msg", "not valid docker format") + } + return + } + // NOTE: Previous implementation used a "sub-pipeline" // to parse docker logs where the json stage added these fields // as "extracted" values so the other stages could operate on them. // We don't need this anymore but it would be a breaking change to // no longer set these. - extracted[dockerOutput] = log.Log - extracted[dockerStream] = log.Stream - extracted[dockerTimestamp] = log.Time + extracted[dockerOutput] = parsed.Log + extracted[dockerStream] = parsed.Stream + extracted[dockerTimestamp] = parsed.Time - *entry = log.Log - labels["stream"] = model.LabelValue(log.Stream) + *entry = parsed.Log + labels["stream"] = model.LabelValue(parsed.Stream) - ts, err := time.Parse(time.RFC3339Nano, log.Time) + ts, err := time.Parse(time.RFC3339Nano, parsed.Time) if err == nil { *t = ts } diff --git a/internal/component/loki/process/stages/docker_test.go b/internal/component/loki/process/stages/docker_test.go index ffd597a4b07..37b9544fd3a 100644 --- a/internal/component/loki/process/stages/docker_test.go +++ b/internal/component/loki/process/stages/docker_test.go @@ -7,9 +7,11 @@ import ( "github.com/go-kit/log" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/common/model" - "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/grafana/alloy/internal/component/common/loki" "github.com/grafana/alloy/internal/featuregate" + "github.com/grafana/loki/pkg/push" ) var ( @@ -22,63 +24,91 @@ var ( ) func TestDocker(t *testing.T) { - loc, err := time.LoadLocation("UTC") - if err != nil { - t.Fatal("could not parse timezone", err) + type testCase struct { + name string + input loki.Entry + expected loki.Entry } - tests := map[string]struct { - entry string - expectedEntry string - t time.Time - expectedT time.Time - labels map[string]string - expectedLabels map[string]string - }{ - "happy path": { - dockerRaw, - dockerProcessed, - time.Now(), - time.Date(2019, 4, 30, 02, 12, 41, 844351500, loc), - map[string]string{}, - map[string]string{ - "stream": "stderr", + tests := []testCase{ + { + name: "happy path", + input: loki.Entry{ + Entry: push.Entry{ + Line: dockerRaw, + Timestamp: time.Now(), + }, + }, + expected: loki.Entry{ + Labels: model.LabelSet{"stream": "stderr"}, + Entry: push.Entry{ + Line: dockerProcessed, + Timestamp: time.Date(2019, 4, 30, 02, 12, 41, 844351500, time.UTC), + }, + }, + }, + { + name: "invalid timestamp", + input: loki.Entry{ + Entry: push.Entry{ + Line: dockerInvalidTimestampRaw, + Timestamp: dockerTestTimeNow, + }, + }, + expected: loki.Entry{ + Labels: model.LabelSet{"stream": "stderr"}, + Entry: push.Entry{ + Line: "log message\n", + Timestamp: dockerTestTimeNow, + }, }, }, - "invalid timestamp": { - dockerInvalidTimestampRaw, - "log message\n", - dockerTestTimeNow, - dockerTestTimeNow, - map[string]string{}, - map[string]string{ - "stream": "stderr", + { + name: "not json", + input: loki.Entry{ + Entry: push.Entry{ + Line: "i'm not json!", + Timestamp: dockerTestTimeNow, + }, + }, + expected: loki.Entry{ + Labels: model.LabelSet{}, + Entry: push.Entry{ + Line: "i'm not json!", + Timestamp: dockerTestTimeNow, + }, }, }, - "invalid json": { - "i'm not json!", - "i'm not json!", - dockerTestTimeNow, - dockerTestTimeNow, - map[string]string{}, - map[string]string{}, + { + name: "json but not docker format", + input: loki.Entry{ + Entry: push.Entry{ + Line: `{"msg": "test"}`, + Timestamp: dockerTestTimeNow, + }, + }, + expected: loki.Entry{ + Labels: model.LabelSet{}, + Entry: push.Entry{ + Line: `{"msg": "test"}`, + Timestamp: dockerTestTimeNow, + }, + }, }, } - for tName, tt := range tests { - t.Run(tName, func(t *testing.T) { + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { t.Parallel() p, err := NewDocker(log.NewNopLogger(), prometheus.DefaultRegisterer, featuregate.StabilityGenerallyAvailable) if err != nil { t.Fatalf("failed to create Docker parser: %s", err) } - out := processEntries(p, newEntry(nil, toLabelSet(tt.labels), tt.entry, tt.t))[0] + out := processEntries(p, newEntry(nil, tt.input.Labels, tt.input.Line, tt.input.Timestamp))[0] - assertLabels(t, tt.expectedLabels, out.Labels) - assert.Equal(t, tt.expectedEntry, out.Line, "did not receive expected log entry") - if out.Timestamp.Unix() != tt.expectedT.Unix() { - t.Fatalf("mismatch ts want: %s got:%s", tt.expectedT, tt.t) - } + require.EqualValues(t, tt.expected.Labels, out.Entry.Labels) + require.Equal(t, tt.expected.Entry.Line, out.Entry.Line) + require.Equal(t, tt.expected.Entry.Timestamp, out.Entry.Timestamp) }) } }