From 9e6f1adb563056b15e5a61283cd4cf26bc1f6ab2 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Thu, 6 Aug 2026 19:47:37 +0300 Subject: [PATCH 1/8] add support json batches consumption --- fd/util.go | 9 ++ pipeline/pipeline.go | 198 ++++++++++++++++++++--------- pipeline/pipeline_whitebox_test.go | 65 ++++++++++ 3 files changed, 211 insertions(+), 61 deletions(-) diff --git a/fd/util.go b/fd/util.go index 9aa5af788..a011d3c7e 100644 --- a/fd/util.go +++ b/fd/util.go @@ -34,6 +34,8 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings { antispamMaintenanceInterval := pipeline.DefaultMaintenanceInterval var antispamExceptions antispam.Exceptions var antispamRules antispam.Rules + var splitJSONArrayField []string + splitJSONArray := pipeline.DefaultSplitJSONArray metricHoldDuration := pipeline.DefaultMetricHoldDuration metricMaxLabelValueLength := pipeline.DefaultMetricMaxLabelValueLength @@ -146,6 +148,11 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings { metricHoldDuration = i } + splitJSONArray = settings.Get("split_json_array").MustBool() + if str := settings.Get("split_json_array_field").MustString(); str != "" { + splitJSONArrayField = cfg.ParseFieldSelector(str) + } + metricMaxLabelValueLength = metrics.Get("max_label_value_length").MustInt() if metricMaxLabelValueLength < 0 { logger.Warn("negative max_label_value_length value, metric label truncation is disabled") @@ -173,6 +180,8 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings { EventTimeout: eventTimeout, StreamField: streamField, IsStrict: isStrict, + SplitJSONArray: splitJSONArray, + SplitJSONArrayField: splitJSONArrayField, Pool: pipeline.PoolType(pool), Metric: &pipeline.MetricSettings{ HoldDuration: metricHoldDuration, diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index e322d35f7..b998b8f92 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "github.com/go-faster/jx" "github.com/ozontech/file.d/decoder" "github.com/ozontech/file.d/logger" "github.com/ozontech/file.d/metric" @@ -42,6 +43,7 @@ const ( DefaultMetricHoldDuration = time.Minute * 30 DefaultMetaCacheSize = 1024 DefaultMetricMaxLabelValueLength = 0 + DefaultSplitJSONArray = false EventSeqIDError = uint64(0) @@ -158,6 +160,8 @@ type Settings struct { CutOffEventByLimitField string StreamField string IsStrict bool + SplitJSONArray bool + SplitJSONArrayField []string Pool PoolType Metric *MetricSettings } @@ -482,81 +486,99 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt } } - p.inputEvents.Inc() - p.inputSize.Add(int64(length)) - - now := time.Now() - event := p.eventPool.get(len(bytes)) - p.eventPoolLatency.Observe(time.Since(now).Seconds()) - - err = nil - if !(dec == decoder.JSON || dec == decoder.PROTOBUF) { - _ = event.Root.DecodeString("{}") - } - switch dec { - case decoder.JSON, decoder.NGINX_ERROR, decoder.PROTOBUF, - decoder.SYSLOG_RFC3164, decoder.SYSLOG_RFC5424, decoder.CSV: - err = p.decoder.DecodeToJson(event.Root, bytes) - case decoder.RAW: - if bytes[len(bytes)-1] == '\n' { - event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, bytes[:len(bytes)-1]) - } else { - event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, bytes) + chunks := [][]byte{bytes} + if p.settings.SplitJSONArray && dec == decoder.JSON { + if elems, split := extractJSONArrElements(bytes, p.settings.SplitJSONArrayField); split { + chunks = elems } - case decoder.CRI: - event.Root.AddFieldNoAlloc(event.Root, "log").MutateToBytesCopy(event.Root, row.Log) - event.Root.AddFieldNoAlloc(event.Root, "time").MutateToBytesCopy(event.Root, row.Time) - event.Root.AddFieldNoAlloc(event.Root, "stream").MutateToBytesCopy(event.Root, row.Stream) - case decoder.POSTGRES: - err = decoder.DecodePostgresToJson(event.Root, bytes) - default: - p.logger.Panic("unknown decoder", zap.Int("decoder", int(dec))) } - if err != nil { - level := zapcore.ErrorLevel - if p.settings.IsStrict { - level = zapcore.FatalLevel + var lastSeqID uint64 + for _, chunk := range chunks { + length := len(chunk) + + p.inputEvents.Inc() + p.inputSize.Add(int64(length)) + + now := time.Now() + event := p.eventPool.get(len(chunk)) + p.eventPoolLatency.Observe(time.Since(now).Seconds()) + + err = nil + if !(dec == decoder.JSON || dec == decoder.PROTOBUF) { + _ = event.Root.DecodeString("{}") } + switch dec { + case decoder.JSON, decoder.NGINX_ERROR, decoder.PROTOBUF, + decoder.SYSLOG_RFC3164, decoder.SYSLOG_RFC5424, decoder.CSV: + err = p.decoder.DecodeToJson(event.Root, chunk) + case decoder.RAW: + if chunk[len(chunk)-1] == '\n' { + event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, chunk[:len(chunk)-1]) + } else { + event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, chunk) + } + case decoder.CRI: + event.Root.AddFieldNoAlloc(event.Root, "log").MutateToBytesCopy(event.Root, row.Log) + event.Root.AddFieldNoAlloc(event.Root, "time").MutateToBytesCopy(event.Root, row.Time) + event.Root.AddFieldNoAlloc(event.Root, "stream").MutateToBytesCopy(event.Root, row.Stream) + case decoder.POSTGRES: + err = decoder.DecodePostgresToJson(event.Root, chunk) + default: + p.logger.Panic("unknown decoder", zap.Int("decoder", int(dec))) + } + + if err != nil { + level := zapcore.ErrorLevel + if p.settings.IsStrict { + level = zapcore.FatalLevel + } - p.logger.Log(level, "wrong log format", zap.Error(err), - zap.Int64("offset", offsets.current), - zap.Int("length", length), - zap.Uint64("source", uint64(sourceID)), - zap.String("source_name", sourceName), - zap.ByteString("log", bytes)) + p.logger.Log(level, "wrong log format", zap.Error(err), + zap.Int64("offset", offsets.current), + zap.Int("length", length), + zap.Uint64("source", uint64(sourceID)), + zap.String("source_name", sourceName), + zap.ByteString("log", chunk)) - // Can't process event, return to pool. - p.eventPool.back(event) - return EventSeqIDError - } + // Can't process event, return to pool. + p.eventPool.back(event) + return EventSeqIDError + } - if len(meta) > 0 { - if event.Root.IsArray() { - nodeArray := event.Root.AsArray() - for _, elem := range nodeArray { - if elem.IsObject() { - for k, v := range meta { - elem.AddField(k).MutateToString(v) + if len(meta) > 0 { + if event.Root.IsArray() { + nodeArray := event.Root.AsArray() + for _, elem := range nodeArray { + if elem.IsObject() { + for k, v := range meta { + elem.AddField(k).MutateToString(v) + } } } - } - } else { - for k, v := range meta { - CreateNestedField(event.Root, []string{k}).MutateToString(v) + } else { + for k, v := range meta { + CreateNestedField(event.Root, []string{k}).MutateToString(v) + } } } - } - if cutoff && p.settings.CutOffEventByLimitField != "" { - event.Root.AddFieldNoAlloc(event.Root, p.settings.CutOffEventByLimitField).MutateToBool(true) + if cutoff && p.settings.CutOffEventByLimitField != "" { + event.Root.AddFieldNoAlloc(event.Root, p.settings.CutOffEventByLimitField).MutateToBool(true) + } + + event.Offset = offsets.current + event.SourceID = sourceID + event.SourceName = sourceName + event.streamName = DefaultStreamName + + lastSeqID = p.streamEvent(event) } - event.Offset = offsets.current - event.SourceID = sourceID - event.SourceName = sourceName - event.streamName = DefaultStreamName + if lastSeqID == 0 { + return EventSeqIDError + } - return p.streamEvent(event) + return lastSeqID } func (p *Pipeline) checkInputBytes(bytes []byte, sourceName string, meta metadata.MetaData) ([]byte, bool, bool) { @@ -1041,6 +1063,60 @@ func (p *Pipeline) serveActionSample(actionIndex int) func(http.ResponseWriter, } } +func extractJSONArrElements(data []byte, fieldPath []string) ([][]byte, bool) { + d := &jx.Decoder{} + d.ResetBytes(data) + + for _, key := range fieldPath { + if d.Next() != jx.Object { + return nil, false + } + objIter, err := d.ObjIter() + if err != nil { + return nil, false + } + + var found bool + for objIter.Next() { + if string(objIter.Key()) == key { + found = true + break + } + if err := d.Skip(); err != nil { + return nil, false + } + } + + if !found { + return nil, false + } + } + + arrIter, err := d.ArrIter() + if err != nil { + return nil, false + } + + var elements [][]byte + for arrIter.Next() { + raw, err := d.Raw() + if err != nil { + return nil, false + } + + elements = append(elements, []byte(raw)) + } + + if err := arrIter.Err(); err != nil { + return nil, false + } + if len(elements) == 0 { + return nil, false + } + + return elements, true +} + func writeErr(w io.Writer, err string) { type ErrResp struct { Error string `json:"error"` diff --git a/pipeline/pipeline_whitebox_test.go b/pipeline/pipeline_whitebox_test.go index d56ffefb1..6ae0a7cde 100644 --- a/pipeline/pipeline_whitebox_test.go +++ b/pipeline/pipeline_whitebox_test.go @@ -303,3 +303,68 @@ func TestSuggestDecoder(t *testing.T) { }) } } + +func TestExtractJSONArrElements(t *testing.T) { + tests := []struct { + name string + input string + fieldPath []string + wantSplit bool + wantElems []string + }{ + { + name: "root array of objects", + input: `[{"message1":"value1"},{"message2":"value2"},{"message3":"value3"}]`, + wantSplit: true, + wantElems: []string{ + `{"message1":"value1"}`, + `{"message2":"value2"}`, + `{"message3":"value3"}`, + }, + }, + { + name: "nested array but non-splited", + input: `{"data":[{"message1":"value1"},{"message2":"value2"},{"message3":"value3"}], "other-field": "other-value"}`, + wantSplit: false, + }, + { + name: "nested array by single-level fieldPath", + input: `{"data":[{"message1":"value1"},{"message2":"value2"},{"message3":"value3"}], "other-field": "other-value"}`, + wantSplit: true, + fieldPath: []string{"data"}, + wantElems: []string{ + `{"message1":"value1"}`, + `{"message2":"value2"}`, + `{"message3":"value3"}`, + }, + }, + { + name: "nested array by multi-level fieldPath", + input: `{"something":{"data":[{"message1":"value1"},{"message2":"value2"},{"message3":"value3"}]}, "other-field": "other-value"}`, + wantSplit: true, + fieldPath: []string{"something", "data"}, + wantElems: []string{ + `{"message1":"value1"}`, + `{"message2":"value2"}`, + `{"message3":"value3"}`, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + elems, split := extractJSONArrElements([]byte(tt.input), tt.fieldPath) + require.Equal(t, tt.wantSplit, split) + if !tt.wantSplit { + return + } + + require.Equal(t, len(tt.wantElems), len(elems)) + for i := range tt.wantElems { + require.JSONEq(t, tt.wantElems[i], string(elems[i])) + } + }) + } +} From 3ab33012c768fc62e53e2583ec64a142c8bf29ba Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Thu, 6 Aug 2026 21:15:23 +0300 Subject: [PATCH 2/8] add docs --- pipeline/README.idoc.md | 12 ++++++++++++ pipeline/README.md | 12 ++++++++++++ pipeline/pipeline.go | 2 +- 3 files changed, 25 insertions(+), 1 deletion(-) diff --git a/pipeline/README.idoc.md b/pipeline/README.idoc.md index 5bf384e30..9b5e9dc36 100644 --- a/pipeline/README.idoc.md +++ b/pipeline/README.idoc.md @@ -108,6 +108,18 @@ Whether to fatal on decoding error.
+**`split_json_array`** *`bool`* *`default=false`* + +Splitting of incoming JSON arrays into separate events before event pool allocation. Each array element becomes its own event. Applies only when the pipeline decoder is `json`. + +
+ +**`split_json_array_field`** *`string`* + +Path to the JSON array of objects. Alternative to the `split` action plugin, applied before event pool allocation. Only used together with `split_json_array: true`. + +
+ **`metric_hold_duration`** *`string`* *`default=30m`* The amount of time the metric can be idle until it is deleted. Used for deleting rarely updated metrics to save metrics storage resources. The value must be passed in format of duration (`(ms|s|m|h)`). diff --git a/pipeline/README.md b/pipeline/README.md index 16072d00d..b8624187d 100755 --- a/pipeline/README.md +++ b/pipeline/README.md @@ -108,6 +108,18 @@ Whether to fatal on decoding error.
+**`split_json_array`** *`bool`* *`default=false`* + +Splitting of incoming JSON arrays into separate events before event pool allocation. Each array element becomes its own event. Applies only when the pipeline decoder is `json`. + +
+ +**`split_json_array_field`** *`string`* + +Path to the JSON array of objects. Alternative to the `split` action plugin, applied before event pool allocation. Only used together with `split_json_array: true`. + +
+ **`metric_hold_duration`** *`string`* *`default=30m`* The amount of time the metric can be idle until it is deleted. Used for deleting rarely updated metrics to save metrics storage resources. The value must be passed in format of duration (`(ms|s|m|h)`). diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index b998b8f92..9118febd3 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -543,7 +543,7 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt // Can't process event, return to pool. p.eventPool.back(event) - return EventSeqIDError + continue } if len(meta) > 0 { From a87b7a3149888f7334a0039aca197879a16492e8 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Fri, 7 Aug 2026 10:46:51 +0300 Subject: [PATCH 3/8] update method of creating decoder --- pipeline/pipeline.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index 9118febd3..88cdfafd7 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -1064,8 +1064,7 @@ func (p *Pipeline) serveActionSample(actionIndex int) func(http.ResponseWriter, } func extractJSONArrElements(data []byte, fieldPath []string) ([][]byte, bool) { - d := &jx.Decoder{} - d.ResetBytes(data) + d := jx.DecodeBytes(data) for _, key := range fieldPath { if d.Next() != jx.Object { From 05d705db97978876c420358b31de6acdecc2520e Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Mon, 10 Aug 2026 14:17:46 +0300 Subject: [PATCH 4/8] update logic in fd/util --- fd/util.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/fd/util.go b/fd/util.go index a011d3c7e..c3d592bf4 100644 --- a/fd/util.go +++ b/fd/util.go @@ -149,7 +149,9 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings { } splitJSONArray = settings.Get("split_json_array").MustBool() - if str := settings.Get("split_json_array_field").MustString(); str != "" { + if arr := settings.Get("split_json_array_field").MustStringArray(); len(arr) > 0 { + splitJSONArrayField = arr + } else if str := settings.Get("split_json_array_field").MustString(); str != "" { splitJSONArrayField = cfg.ParseFieldSelector(str) } From 404d6e88adf76c3e1913a47314af7b6574b1cb6d Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Mon, 10 Aug 2026 15:12:39 +0300 Subject: [PATCH 5/8] remove new logic --- fd/util.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/fd/util.go b/fd/util.go index c3d592bf4..a011d3c7e 100644 --- a/fd/util.go +++ b/fd/util.go @@ -149,9 +149,7 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings { } splitJSONArray = settings.Get("split_json_array").MustBool() - if arr := settings.Get("split_json_array_field").MustStringArray(); len(arr) > 0 { - splitJSONArrayField = arr - } else if str := settings.Get("split_json_array_field").MustString(); str != "" { + if str := settings.Get("split_json_array_field").MustString(); str != "" { splitJSONArrayField = cfg.ParseFieldSelector(str) } From 10c4e15770ebc1e4af164d48f70bbc68e1cea595 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Tue, 11 Aug 2026 18:51:32 +0300 Subject: [PATCH 6/8] add func streamChunk --- pipeline/pipeline.go | 179 +++++++++++++++++++++++++------------------ 1 file changed, 103 insertions(+), 76 deletions(-) diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index 88cdfafd7..27bdebd14 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -178,6 +178,17 @@ type AntispamSettings struct { MaintenanceInterval time.Duration } +// values that stay the same for all chunks of one In call +type chunkParams struct { + dec decoder.Type + row decoder.CRIRow + sourceID SourceID + sourceName string + offsets Offsets + meta metadata.MetaData + cutoff bool +} + type PoolType string const ( @@ -493,85 +504,21 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt } } + params := &chunkParams{ + sourceID: sourceID, + sourceName: sourceName, + offsets: offsets, + meta: meta, + cutoff: cutoff, + dec: dec, + row: row, + } + var lastSeqID uint64 for _, chunk := range chunks { - length := len(chunk) - - p.inputEvents.Inc() - p.inputSize.Add(int64(length)) - - now := time.Now() - event := p.eventPool.get(len(chunk)) - p.eventPoolLatency.Observe(time.Since(now).Seconds()) - - err = nil - if !(dec == decoder.JSON || dec == decoder.PROTOBUF) { - _ = event.Root.DecodeString("{}") + if id := p.streamChunk(params, chunk); id != EventSeqIDError { + lastSeqID = id } - switch dec { - case decoder.JSON, decoder.NGINX_ERROR, decoder.PROTOBUF, - decoder.SYSLOG_RFC3164, decoder.SYSLOG_RFC5424, decoder.CSV: - err = p.decoder.DecodeToJson(event.Root, chunk) - case decoder.RAW: - if chunk[len(chunk)-1] == '\n' { - event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, chunk[:len(chunk)-1]) - } else { - event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, chunk) - } - case decoder.CRI: - event.Root.AddFieldNoAlloc(event.Root, "log").MutateToBytesCopy(event.Root, row.Log) - event.Root.AddFieldNoAlloc(event.Root, "time").MutateToBytesCopy(event.Root, row.Time) - event.Root.AddFieldNoAlloc(event.Root, "stream").MutateToBytesCopy(event.Root, row.Stream) - case decoder.POSTGRES: - err = decoder.DecodePostgresToJson(event.Root, chunk) - default: - p.logger.Panic("unknown decoder", zap.Int("decoder", int(dec))) - } - - if err != nil { - level := zapcore.ErrorLevel - if p.settings.IsStrict { - level = zapcore.FatalLevel - } - - p.logger.Log(level, "wrong log format", zap.Error(err), - zap.Int64("offset", offsets.current), - zap.Int("length", length), - zap.Uint64("source", uint64(sourceID)), - zap.String("source_name", sourceName), - zap.ByteString("log", chunk)) - - // Can't process event, return to pool. - p.eventPool.back(event) - continue - } - - if len(meta) > 0 { - if event.Root.IsArray() { - nodeArray := event.Root.AsArray() - for _, elem := range nodeArray { - if elem.IsObject() { - for k, v := range meta { - elem.AddField(k).MutateToString(v) - } - } - } - } else { - for k, v := range meta { - CreateNestedField(event.Root, []string{k}).MutateToString(v) - } - } - } - if cutoff && p.settings.CutOffEventByLimitField != "" { - event.Root.AddFieldNoAlloc(event.Root, p.settings.CutOffEventByLimitField).MutateToBool(true) - } - - event.Offset = offsets.current - event.SourceID = sourceID - event.SourceName = sourceName - event.streamName = DefaultStreamName - - lastSeqID = p.streamEvent(event) } if lastSeqID == 0 { @@ -610,6 +557,86 @@ func (p *Pipeline) checkInputBytes(bytes []byte, sourceName string, meta metadat return bytes, false, true } +func (p *Pipeline) streamChunk(params *chunkParams, chunk []byte) uint64 { + length := len(chunk) + + p.inputEvents.Inc() + p.inputSize.Add(int64(length)) + + now := time.Now() + event := p.eventPool.get(len(chunk)) + p.eventPoolLatency.Observe(time.Since(now).Seconds()) + + var err error + if !(params.dec == decoder.JSON || params.dec == decoder.PROTOBUF) { + _ = event.Root.DecodeString("{}") + } + switch params.dec { + case decoder.JSON, decoder.NGINX_ERROR, decoder.PROTOBUF, + decoder.SYSLOG_RFC3164, decoder.SYSLOG_RFC5424, decoder.CSV: + err = p.decoder.DecodeToJson(event.Root, chunk) + case decoder.RAW: + if chunk[len(chunk)-1] == '\n' { + event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, chunk[:len(chunk)-1]) + } else { + event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, chunk) + } + case decoder.CRI: + event.Root.AddFieldNoAlloc(event.Root, "log").MutateToBytesCopy(event.Root, params.row.Log) + event.Root.AddFieldNoAlloc(event.Root, "time").MutateToBytesCopy(event.Root, params.row.Time) + event.Root.AddFieldNoAlloc(event.Root, "stream").MutateToBytesCopy(event.Root, params.row.Stream) + case decoder.POSTGRES: + err = decoder.DecodePostgresToJson(event.Root, chunk) + default: + p.logger.Panic("unknown decoder", zap.Int("decoder", int(params.dec))) + } + + if err != nil { + level := zapcore.ErrorLevel + if p.settings.IsStrict { + level = zapcore.FatalLevel + } + + p.logger.Log(level, "wrong log format", zap.Error(err), + zap.Int64("offset", params.offsets.current), + zap.Int("length", length), + zap.Uint64("source", uint64(params.sourceID)), + zap.String("source_name", params.sourceName), + zap.ByteString("log", chunk)) + + // Can't process event, return to pool. + p.eventPool.back(event) + return EventSeqIDError + } + + if len(params.meta) > 0 { + if event.Root.IsArray() { + nodeArray := event.Root.AsArray() + for _, elem := range nodeArray { + if elem.IsObject() { + for k, v := range params.meta { + elem.AddField(k).MutateToString(v) + } + } + } + } else { + for k, v := range params.meta { + CreateNestedField(event.Root, []string{k}).MutateToString(v) + } + } + } + if params.cutoff && p.settings.CutOffEventByLimitField != "" { + event.Root.AddFieldNoAlloc(event.Root, p.settings.CutOffEventByLimitField).MutateToBool(true) + } + + event.Offset = params.offsets.current + event.SourceID = params.sourceID + event.SourceName = params.sourceName + event.streamName = DefaultStreamName + + return p.streamEvent(event) +} + func (p *Pipeline) streamEvent(event *Event) uint64 { streamID := StreamID(event.SourceID) From 71e3d3f9ec06c442295e9e60938f82f08bad4599 Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Fri, 2 Oct 2026 12:59:40 +0300 Subject: [PATCH 7/8] update lodic --- e2e/file_file_split/config.yml | 14 +++ e2e/file_file_split/file_file_split.go | 96 ++++++++++++++++++ e2e/start_work_test.go | 11 +++ pipeline/event.go | 5 + pipeline/pipeline.go | 132 +++++++++++++++---------- pipeline/pipeline_whitebox_test.go | 76 +++++++++++++- 6 files changed, 277 insertions(+), 57 deletions(-) create mode 100644 e2e/file_file_split/config.yml create mode 100644 e2e/file_file_split/file_file_split.go diff --git a/e2e/file_file_split/config.yml b/e2e/file_file_split/config.yml new file mode 100644 index 000000000..71ccbbc2f --- /dev/null +++ b/e2e/file_file_split/config.yml @@ -0,0 +1,14 @@ +pipelines: + file_file_split: + settings: + decoder: json + split_json_array: true + split_json_array_filed: data + input: + type: file + persistence_mode: async + watching_dir: SOME_DIR + offsets_file: SOME_FILE + offsets_op: reset + output: + type: file diff --git a/e2e/file_file_split/file_file_split.go b/e2e/file_file_split/file_file_split.go new file mode 100644 index 000000000..48c3f057f --- /dev/null +++ b/e2e/file_file_split/file_file_split.go @@ -0,0 +1,96 @@ +package file_file_split + +import ( + "fmt" + "log" + "os" + "path" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + uuid "github.com/satori/go.uuid" + + "github.com/ozontech/file.d/cfg" + "github.com/ozontech/file.d/test" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Config for file-file-split plugin e2e test +type Config struct { + FilesDir string + Count int + Lines int + BatchSize int + RetTime string +} + +// Configure sets additional fields for input and output plugins +func (c *Config) Configure(t *testing.T, conf *cfg.Config, pipelineName string) { + c.FilesDir = t.TempDir() + offsetsDir := t.TempDir() + + input := conf.Pipelines[pipelineName].Raw.Get("input") + input.Set("watching_dir", c.FilesDir) + input.Set("filename_pattern", "split-input-*.log") + input.Set("offsets_file", filepath.Join(offsetsDir, "offsets.yaml")) + + output := conf.Pipelines[pipelineName].Raw.Get("output") + output.Set("target_file", path.Join(c.FilesDir, "file-d.log")) + output.Set("retention_interval", c.RetTime) +} + +func (c *Config) Send(t *testing.T) { + wg := &sync.WaitGroup{} + wg.Add(c.Count) + for i := 0; i < c.Count; i++ { + go func() { + defer wg.Done() + u := strings.ReplaceAll(uuid.NewV4().String(), "-", "") + name := path.Join(c.FilesDir, fmt.Sprintf("split-input-%s.log", u)) + file, err := os.Create(name) + if err != nil { + log.Fatalf("failed to create file: %s", err.Error()) + } + + var sb strings.Builder + for j := 0; j < c.Lines; j++ { + sb.WriteString(buildBatch(j, c.BatchSize)) + sb.WriteByte('\n') + } + + if _, err = file.WriteString(sb.String()); err != nil { + log.Fatalf("failed to write to file: %s", err.Error()) + } + if err = file.Close(); err != nil { + log.Fatalf("failed to close file: %s", err.Error()) + } + }() + } + wg.Wait() +} + +func (c *Config) Validate(t *testing.T) { + logFilePattern := path.Join(c.FilesDir, "file-d*.log") + expected := c.Count * c.Lines * c.BatchSize + test.WaitProcessEvents(t, expected, 3*time.Second, 30*time.Second, logFilePattern) + matches := test.GetMatches(t, logFilePattern) + assert.True(t, len(matches) > 0, "no files with processed events") + require.Equal(t, expected, test.CountLines(t, logFilePattern), "wrong number of processed events after split") +} + +func buildBatch(lineIdx, size int) string { + var sb strings.Builder + sb.WriteString(`{"data":[`) + for i := range size { + if i > 0 { + sb.WriteByte(',') + } + fmt.Fprintf(&sb, `{"m":"line-%d-%d"}`, lineIdx, i) + } + sb.WriteString(`]}`) + return sb.String() +} diff --git a/e2e/start_work_test.go b/e2e/start_work_test.go index e751522e8..4a85b7735 100644 --- a/e2e/start_work_test.go +++ b/e2e/start_work_test.go @@ -14,6 +14,7 @@ import ( "github.com/ozontech/file.d/e2e/file_elasticsearch" "github.com/ozontech/file.d/e2e/file_es_split" "github.com/ozontech/file.d/e2e/file_file" + "github.com/ozontech/file.d/e2e/file_file_split" "github.com/ozontech/file.d/e2e/file_loki" "github.com/ozontech/file.d/e2e/file_socket" "github.com/ozontech/file.d/e2e/http_file" @@ -122,6 +123,16 @@ func TestE2EStabilityWorkCase(t *testing.T) { }, cfgPath: "./file_file/config.yml", }, + { + name: "file_file_split", + e2eTest: &file_file_split.Config{ + Count: 5, + Lines: 50, + BatchSize: 10, + RetTime: "1s", + }, + cfgPath: "./file_file_split/config.yml", + }, { name: "http_file", e2eTest: &http_file.Config{ diff --git a/pipeline/event.go b/pipeline/event.go index 31e23217d..1ecb18b01 100644 --- a/pipeline/event.go +++ b/pipeline/event.go @@ -16,6 +16,10 @@ import ( type Event struct { kind Kind + // set to true for non-last chunks of a split batch, so pipeline.finalize() + // skips input.Commit() and Event.Offset is committed only once + skipInputCommit bool + children []*Event Root *insaneJSON.Root @@ -116,6 +120,7 @@ func (e *Event) reset() { e.stream = nil e.children = e.children[:0] e.kind = EventKindRegular + e.skipInputCommit = false } func (e *Event) StreamNameBytes() []byte { diff --git a/pipeline/pipeline.go b/pipeline/pipeline.go index 5343fd6a3..7c7a76182 100644 --- a/pipeline/pipeline.go +++ b/pipeline/pipeline.go @@ -179,18 +179,6 @@ type AntispamSettings struct { Sampler *antispam.Sampler } -// values that stay the same for all chunks of one In call -type chunkParams struct { - dec decoder.Type - row decoder.CRIRow - sourceID SourceID - sourceName string - offsets Offsets - meta metadata.MetaData - cutoff bool - antispamSampled bool -} - type PoolType string const ( @@ -418,9 +406,8 @@ func (p *Pipeline) GetOutput() OutputPlugin { // In decodes message and passes it to event stream. func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, bytes []byte, isNewSource bool, meta metadata.MetaData) (seqID uint64) { var ( - ok bool - cutoff bool - antispamSampled bool + ok bool + cutoff bool ) // don't process mud. bytes, cutoff, ok = p.checkInputBytes(bytes, sourceName, meta) @@ -455,6 +442,12 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt } } + var ( + checkSourceID string + checkSourceName string + eventTime time.Time + ) + // Skip IsSpam for partial logs is necessary to avoid the case // when some parts of a large event have got into the ban, // thereby cutting off a piece of the event. @@ -463,16 +456,13 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt // The event is Partial if it is larger than the driver configuration. // For example, for containerd this setting is called max_container_log_line_size // https://github.com/containerd/containerd/blob/f7f2be732159a411eae46b78bfdb479b133a823b/pkg/cri/config/config.go#L263-L266 - if !row.IsPartial && p.settings.Antispam.Threshold >= 0 { + antispamEnabled := !row.IsPartial && p.settings.Antispam.Threshold >= 0 + if antispamEnabled { streamOffset := offsets.ByStream(string(row.Stream)) - currentOffset := offsets.current - - if streamOffset > 0 && currentOffset < streamOffset { + if streamOffset > 0 && offsets.current < streamOffset { return EventSeqIDError } - var checkSourceID string - var checkSourceName string if p.settings.SourceNameMetaField == "" { checkSourceID = strconv.FormatUint(uint64(sourceID), 10) checkSourceName = sourceName @@ -488,51 +478,53 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt } } - var eventTime time.Time if len(row.Time) > 0 { eventTime, err = time.Parse("2006-01-02T15:04:05.999999999Z", string(row.Time)) if err != nil { p.Error(fmt.Sprintf("cannot parse raw time %s: %v", row.Time, err)) } } - - switch p.antispamer.IsSpam(checkSourceID, checkSourceName, isNewSource, bytes, eventTime, meta) { - case antispam.Dropped: - return EventSeqIDError - case antispam.Sampled: - antispamSampled = true - } } chunks := [][]byte{bytes} if p.settings.SplitJSONArray && dec == decoder.JSON { - if elems, split := extractJSONArrElements(bytes, p.settings.SplitJSONArrayField); split { - chunks = elems + if buf, split := extractJSONArrElements(bytes, p.settings.SplitJSONArrayField); split { + chunks = buf.Elements + defer buf.Reset() } } params := &chunkParams{ - sourceID: sourceID, - sourceName: sourceName, - offsets: offsets, - meta: meta, - cutoff: cutoff, - antispamSampled: antispamSampled, - dec: dec, - row: row, + sourceID: sourceID, + sourceName: sourceName, + offsets: offsets, + meta: meta, + cutoff: cutoff, + dec: dec, + row: row, } + lastIdx := len(chunks) - 1 + var lastSeqID uint64 - for _, chunk := range chunks { + for i, chunk := range chunks { + params.antispamSampled = false + if antispamEnabled { + switch p.antispamer.IsSpam(checkSourceID, checkSourceName, isNewSource, bytes, eventTime, meta) { + case antispam.Dropped: + isNewSource = false + continue + case antispam.Sampled: + params.antispamSampled = true + } + isNewSource = false + } + + params.isLastChunk = i == lastIdx if id := p.streamChunk(params, chunk); id != EventSeqIDError { lastSeqID = id } } - - if lastSeqID == 0 { - return EventSeqIDError - } - return lastSeqID } @@ -565,6 +557,20 @@ func (p *Pipeline) checkInputBytes(bytes []byte, sourceName string, meta metadat return bytes, false, true } +// values shared by all chunks of one In call. +// isLast is the only field that changes between chunks +type chunkParams struct { + dec decoder.Type + row decoder.CRIRow + sourceID SourceID + sourceName string + offsets Offsets + meta metadata.MetaData + cutoff bool + antispamSampled bool + isLastChunk bool +} + func (p *Pipeline) streamChunk(params *chunkParams, chunk []byte) uint64 { length := len(chunk) @@ -661,6 +667,7 @@ func (p *Pipeline) streamChunk(params *chunkParams, chunk []byte) uint64 { event.SourceID = params.sourceID event.SourceName = params.sourceName event.streamName = DefaultStreamName + event.skipInputCommit = !params.isLastChunk return p.streamEvent(event) } @@ -708,7 +715,9 @@ func (p *Pipeline) finalize(event *Event, notifyInput bool, backEvent bool) { } if notifyInput { - p.input.Commit(event) + if !event.skipInputCommit { + p.input.Commit(event) + } p.outputEvents.Inc() p.outputSize.Add(int64(event.Size)) } @@ -1118,13 +1127,25 @@ func (p *Pipeline) serveActionSample(actionIndex int) func(http.ResponseWriter, } } -func extractJSONArrElements(data []byte, fieldPath []string) ([][]byte, bool) { +type chunksBuffer struct { + Elements [][]byte +} + +func (b *chunksBuffer) Reset() { + b.Elements = b.Elements[:0] + chunkBufferPool.Put(b) +} + +var chunkBufferPool = sync.Pool{ + New: func() any { + return &chunksBuffer{Elements: make([][]byte, 0, 128)} // мб побольше или меньше + }, +} + +func extractJSONArrElements(data []byte, fieldPath []string) (*chunksBuffer, bool) { d := jx.DecodeBytes(data) for _, key := range fieldPath { - if d.Next() != jx.Object { - return nil, false - } objIter, err := d.ObjIter() if err != nil { return nil, false @@ -1151,24 +1172,27 @@ func extractJSONArrElements(data []byte, fieldPath []string) ([][]byte, bool) { return nil, false } - var elements [][]byte + buf := chunkBufferPool.Get().(*chunksBuffer) for arrIter.Next() { raw, err := d.Raw() if err != nil { + buf.Reset() return nil, false } - elements = append(elements, []byte(raw)) + buf.Elements = append(buf.Elements, []byte(raw)) } if err := arrIter.Err(); err != nil { + buf.Reset() return nil, false } - if len(elements) == 0 { + if len(buf.Elements) == 0 { + buf.Reset() return nil, false } - return elements, true + return buf, true } func writeErr(w io.Writer, err string) { diff --git a/pipeline/pipeline_whitebox_test.go b/pipeline/pipeline_whitebox_test.go index 6ae0a7cde..1cb11dd70 100644 --- a/pipeline/pipeline_whitebox_test.go +++ b/pipeline/pipeline_whitebox_test.go @@ -1,6 +1,8 @@ package pipeline import ( + "fmt" + "strings" "testing" "github.com/ozontech/file.d/decoder" @@ -355,15 +357,83 @@ func TestExtractJSONArrElements(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - elems, split := extractJSONArrElements([]byte(tt.input), tt.fieldPath) + buf, split := extractJSONArrElements([]byte(tt.input), tt.fieldPath) require.Equal(t, tt.wantSplit, split) if !tt.wantSplit { return } - require.Equal(t, len(tt.wantElems), len(elems)) + require.Equal(t, len(tt.wantElems), len(buf.Elements)) for i := range tt.wantElems { - require.JSONEq(t, tt.wantElems[i], string(elems[i])) + require.JSONEq(t, tt.wantElems[i], string(buf.Elements[i])) + } + }) + } +} + +func BenchmarkExtractJSONArrElements(b *testing.B) { + buildData := func(n int, splitFields ...string) []byte { + var sb strings.Builder + for _, sf := range splitFields { + fmt.Fprintf(&sb, `{%q:`, sf) + } + sb.WriteByte('[') + + for i := range n { + if i > 0 { + sb.WriteByte(',') + } + fmt.Fprintf(&sb, `{"i":%d,"m":"message-%d"}`, i, i) + } + sb.WriteByte(']') + + for range splitFields { + sb.WriteByte('}') + } + return []byte(sb.String()) + } + + tests := []struct { + name string + data []byte + splitFields []string + }{ + { + name: "root/10", + data: buildData(10), + }, + { + name: "root/100", + data: buildData(100), + }, + { + name: "root/1000", + data: buildData(1000), + }, + { + name: "nested/10", + data: buildData(10, "data"), + splitFields: []string{"data"}, + }, + { + name: "nested/100", + data: buildData(100, "data"), + splitFields: []string{"data"}, + }, + { + name: "nested/1000", + data: buildData(1000, "data"), + splitFields: []string{"data"}, + }, + } + + for _, tt := range tests { + b.Run(tt.name, func(b *testing.B) { + for b.Loop() { + buf, split := extractJSONArrElements(tt.data, tt.splitFields) + if split { + buf.Reset() + } } }) } From ba58b55727543a72e1f0f81bda41b697d653833d Mon Sep 17 00:00:00 2001 From: Sergey Lazarenko Date: Fri, 2 Oct 2026 13:08:27 +0300 Subject: [PATCH 8/8] clumsy... --- e2e/file_file_split/config.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/e2e/file_file_split/config.yml b/e2e/file_file_split/config.yml index 71ccbbc2f..9d373e11a 100644 --- a/e2e/file_file_split/config.yml +++ b/e2e/file_file_split/config.yml @@ -3,7 +3,7 @@ pipelines: settings: decoder: json split_json_array: true - split_json_array_filed: data + split_json_array_field: data input: type: file persistence_mode: async