support for consuming JSON batches

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,

pipeline/README.idoc.md

@@ -108,6 +108,18 @@ Whether to fatal on decoding error.
 
 <br>
 
+**`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`.
+
+<br>
+
+**`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`.
+
+<br>
+
 **`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 (`<number>(ms|s|m|h)`).

pipeline/README.md

@@ -108,6 +108,18 @@ Whether to fatal on decoding error.
 
 <br>
 
+**`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`.
+
+<br>
+
+**`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`.
+
+<br>
+
 **`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 (`<number>(ms|s|m|h)`).

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
 }
@@ -174,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 (
@@ -482,35 +497,98 @@ func (p *Pipeline) In(sourceID SourceID, sourceName string, offsets Offsets, byt
 		}
 	}
 
+	chunks := [][]byte{bytes}
+	if p.settings.SplitJSONArray && dec == decoder.JSON {
+		if elems, split := extractJSONArrElements(bytes, p.settings.SplitJSONArrayField); split {
+			chunks = elems
+		}
+	}
+
+	params := &chunkParams{
+		sourceID:   sourceID,
+		sourceName: sourceName,
+		offsets:    offsets,
+		meta:       meta,
+		cutoff:     cutoff,
+		dec:        dec,
+		row:        row,
+	}
+
+	var lastSeqID uint64
+	for _, chunk := range chunks {
+		if id := p.streamChunk(params, chunk); id != EventSeqIDError {
+			lastSeqID = id
+		}
+	}
+
+	if lastSeqID == 0 {
+		return EventSeqIDError
+	}
+
+	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
+}
+
+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 {
@@ -520,74 +598,45 @@ 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)
 	}
 
-	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
 
 	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)
 
@@ -1041,6 +1090,59 @@ func (p *Pipeline) serveActionSample(actionIndex int) func(http.ResponseWriter,
 	}
 }
 
+func extractJSONArrElements(data []byte, fieldPath []string) ([][]byte, 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
+		}
+
+		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"`