diff --git a/e2e/file_file_split/config.yml b/e2e/file_file_split/config.yml
new file mode 100644
index 000000000..9d373e11a
--- /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_field: 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/fd/util.go b/fd/util.go
index 2071c736a..3cde11087 100644
--- a/fd/util.go
+++ b/fd/util.go
@@ -35,6 +35,8 @@ func extractPipelineParams(settings *simplejson.Json) *pipeline.Settings {
var antispamExceptions antispam.Exceptions
var antispamRules antispam.Rules
var antispamSampler *antispam.Sampler
+ var splitJSONArrayField []string
+ splitJSONArray := pipeline.DefaultSplitJSONArray
metricHoldDuration := pipeline.DefaultMetricHoldDuration
metricMaxLabelValueLength := pipeline.DefaultMetricMaxLabelValueLength
@@ -152,6 +154,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")
@@ -180,6 +187,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/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/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 6ba6901f2..7c7a76182 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
}
@@ -402,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)
@@ -439,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.
@@ -447,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
@@ -472,51 +478,131 @@ 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 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,
+ dec: dec,
+ row: row,
+ }
+
+ lastIdx := len(chunks) - 1
+
+ var lastSeqID uint64
+ 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
+ }
+ }
+ return lastSeqID
+}
+
+func (p *Pipeline) checkInputBytes(bytes []byte, sourceName string, meta metadata.MetaData) ([]byte, bool, bool) {
+ length := len(bytes)
+
+ if length == 0 || (bytes[0] == '\n' && length == 1) {
+ return bytes, false, false
+ }
+
+ if p.settings.MaxEventSize != 0 && length > p.settings.MaxEventSize {
+ source := sourceName
+ if val, ok := meta[p.settings.SourceNameMetaField]; ok {
+ source = val
+ }
+ p.IncMaxEventSizeExceeded(source)
+
+ if !p.settings.CutOffEventByLimit {
+ return bytes, false, false
+ }
+
+ wasNewLine := bytes[len(bytes)-1] == '\n'
+ bytes = bytes[:p.settings.MaxEventSize]
+ if wasNewLine {
+ bytes = append(bytes, '\n')
}
+ return bytes, true, true
}
+ 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)
+
p.inputEvents.Inc()
p.inputSize.Add(int64(length))
now := time.Now()
- event := p.eventPool.get(len(bytes))
+ event := p.eventPool.get(len(chunk))
p.eventPoolLatency.Observe(time.Since(now).Seconds())
- err = nil
- if !(dec == decoder.JSON || dec == decoder.PROTOBUF) {
+ var err error
+ if !(params.dec == decoder.JSON || params.dec == decoder.PROTOBUF) {
_ = event.Root.DecodeString("{}")
}
- switch dec {
+ 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, bytes)
+ err = p.decoder.DecodeToJson(event.Root, chunk)
case decoder.RAW:
- if bytes[len(bytes)-1] == '\n' {
- event.Root.AddFieldNoAlloc(event.Root, "message").MutateToBytesCopy(event.Root, bytes[:len(bytes)-1])
+ 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, bytes)
+ 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)
+ 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, bytes)
+ err = decoder.DecodePostgresToJson(event.Root, chunk)
default:
- p.logger.Panic("unknown decoder", zap.Int("decoder", int(dec)))
+ p.logger.Panic("unknown decoder", zap.Int("decoder", int(params.dec)))
}
if err != nil {
@@ -526,37 +612,38 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt
}
p.logger.Log(level, "wrong log format", zap.Error(err),
- zap.Int64("offset", offsets.current),
+ zap.Int64("offset", params.offsets.current),
zap.Int("length", length),
- zap.Uint64("source", uint64(sourceID)),
- zap.String("source_name", sourceName),
- zap.ByteString("log", bytes))
+ 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(meta) > 0 {
+ if len(params.meta) > 0 {
if event.Root.IsArray() {
nodeArray := event.Root.AsArray()
for _, elem := range nodeArray {
if elem.IsObject() {
- for k, v := range meta {
+ for k, v := range params.meta {
elem.AddField(k).MutateToString(v)
}
}
}
} else {
- for k, v := range meta {
+ for k, v := range params.meta {
CreateNestedField(event.Root, []string{k}).MutateToString(v)
}
}
}
- if cutoff && p.settings.CutOffEventByLimitField != "" {
+ if params.cutoff && p.settings.CutOffEventByLimitField != "" {
event.Root.AddFieldNoAlloc(event.Root, p.settings.CutOffEventByLimitField).MutateToBool(true)
}
- if antispamSampled {
+
+ if params.antispamSampled {
// cfg is always non-nil here: antispamSampled is set only when the sampler is configured.
cfg := p.settings.Antispam.Sampler
if cfg.MarkerField != "" {
@@ -576,43 +663,15 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt
}
}
- event.Offset = offsets.current
- event.SourceID = sourceID
- event.SourceName = sourceName
+ event.Offset = params.offsets.current
+ event.SourceID = params.sourceID
+ event.SourceName = params.sourceName
event.streamName = DefaultStreamName
+ event.skipInputCommit = !params.isLastChunk
return p.streamEvent(event)
}
-func (p *Pipeline) checkInputBytes(bytes []byte, sourceName string, meta metadata.MetaData) ([]byte, bool, bool) {
- length := len(bytes)
-
- if length == 0 || (bytes[0] == '\n' && length == 1) {
- return bytes, false, false
- }
-
- if p.settings.MaxEventSize != 0 && length > p.settings.MaxEventSize {
- source := sourceName
- if val, ok := meta[p.settings.SourceNameMetaField]; ok {
- source = val
- }
- p.IncMaxEventSizeExceeded(source)
-
- if !p.settings.CutOffEventByLimit {
- return bytes, false, false
- }
-
- wasNewLine := bytes[len(bytes)-1] == '\n'
- bytes = bytes[:p.settings.MaxEventSize]
- if wasNewLine {
- bytes = append(bytes, '\n')
- }
- return bytes, true, true
- }
-
- return bytes, false, true
-}
-
func (p *Pipeline) streamEvent(event *Event) uint64 {
streamID := StreamID(event.SourceID)
@@ -656,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))
}
@@ -1066,6 +1127,74 @@ func (p *Pipeline) serveActionSample(actionIndex int) func(http.ResponseWriter,
}
}
+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 {
+ 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
+ }
+
+ buf := chunkBufferPool.Get().(*chunksBuffer)
+ for arrIter.Next() {
+ raw, err := d.Raw()
+ if err != nil {
+ buf.Reset()
+ return nil, false
+ }
+
+ buf.Elements = append(buf.Elements, []byte(raw))
+ }
+
+ if err := arrIter.Err(); err != nil {
+ buf.Reset()
+ return nil, false
+ }
+ if len(buf.Elements) == 0 {
+ buf.Reset()
+ return nil, false
+ }
+
+ return buf, 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..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"
@@ -303,3 +305,136 @@ 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()
+
+ 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(buf.Elements))
+ for i := range tt.wantElems {
+ 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()
+ }
+ }
+ })
+ }
+}