diff --git a/cmd/file.d/file.d.go b/cmd/file.d/file.d.go
index 4cb8b58ca..0a25269b2 100644
--- a/cmd/file.d/file.d.go
+++ b/cmd/file.d/file.d.go
@@ -26,6 +26,7 @@ import (
_ "github.com/ozontech/file.d/plugin/action/debug"
_ "github.com/ozontech/file.d/plugin/action/decode"
_ "github.com/ozontech/file.d/plugin/action/discard"
+ _ "github.com/ozontech/file.d/plugin/action/file_multiline"
_ "github.com/ozontech/file.d/plugin/action/flatten"
_ "github.com/ozontech/file.d/plugin/action/hash"
_ "github.com/ozontech/file.d/plugin/action/join"
diff --git a/e2e/start_work_test.go b/e2e/start_work_test.go
index c7c349eb2..a3fceef2c 100644
--- a/e2e/start_work_test.go
+++ b/e2e/start_work_test.go
@@ -32,6 +32,7 @@ import (
_ "github.com/ozontech/file.d/plugin/action/debug"
_ "github.com/ozontech/file.d/plugin/action/decode"
_ "github.com/ozontech/file.d/plugin/action/discard"
+ _ "github.com/ozontech/file.d/plugin/action/file_multiline"
_ "github.com/ozontech/file.d/plugin/action/flatten"
_ "github.com/ozontech/file.d/plugin/action/hash"
_ "github.com/ozontech/file.d/plugin/action/join"
diff --git a/plugin/action/file_multiline/README.md b/plugin/action/file_multiline/README.md
new file mode 100644
index 000000000..d3a35a85a
--- /dev/null
+++ b/plugin/action/file_multiline/README.md
@@ -0,0 +1,46 @@
+# File multiline action
+It joins split stdout/container log chunks into a single event.
+
+Docker/CRI splits long logs into ~16kb chunks. Use this action with [file input plugin](/plugin/input/file/README.md) when reading container logs from files (e.g. `/var/log/containers/*.log`).
+
+> ⚠ Place it **before** `json_decode`, `decode`, and other actions that expect a complete field value. For Docker JSON logs use `field: log`.
+
+**Example:**
+```yaml
+pipelines:
+ example_pipeline:
+ settings:
+ decoder: json
+ input:
+ type: file
+ offsets_file: /data/offsets.yaml
+ watching_dir: /var/log/containers/
+ actions:
+ - type: file_multiline
+ field: log
+ split_event_size: 1000000
+ - type: json_decode
+ field: log
+```
+
+### How it works
+1. Reads the configured field (`log` by default) from each sequential event in the same file stream.
+2. If the value does **not** end with a real newline character `\n`, the chunk is buffered and the plugin waits for the next event (`ActionCollapse`). Further actions are not run yet.
+3. When the next chunk arrives, it is appended to the buffer and the end is checked again.
+4. When the value ends with `\n`, all buffered chunks are merged into one field and a single event is passed downstream (`ActionPass`).
+5. If the joined event exceeds `split_event_size`, it may be split forcibly even without `\n` at the end.
+6. If no continuation arrives within the pipeline `event_timeout` (default `30s`), the buffer is reset and the partial event is discarded.
+
+### Config params
+**`field`** *`cfg.FieldSelector`* *`default=log`*
+
+The event field which will be joined.
+
+
+
+**`split_event_size`** *`int`* *`default=1000000`*
+
+Docker splits long logs by 16kb chunks. The plugin joins them back, but if an event is longer than this value in bytes, it will be split after all.
+> Due to the optimization process it's not a strict rule. Events may be split even if they won't exceed the limit.
+
+
diff --git a/plugin/action/file_multiline/file_multiline.go b/plugin/action/file_multiline/file_multiline.go
new file mode 100644
index 000000000..80eae23aa
--- /dev/null
+++ b/plugin/action/file_multiline/file_multiline.go
@@ -0,0 +1,239 @@
+package file_multiline
+
+import (
+ "slices"
+ "sync"
+
+ "github.com/ozontech/file.d/cfg"
+ "github.com/ozontech/file.d/fd"
+ "github.com/ozontech/file.d/pipeline"
+ "go.uber.org/zap"
+)
+
+/*{ introduction
+It joins split stdout/container log chunks into a single event.
+
+Docker/CRI splits long logs by 16kb chunks. Use this action with [file input plugin](/plugin/input/file/README.md)
+when reading container logs from files.
+
+> ⚠ Place it **before** `json_decode`, `decode`, and other actions that expect a complete field value. For Docker JSON logs use `field: log`.
+
+**Example:**
+```yaml
+pipelines:
+ example_pipeline:
+ settings:
+ decoder: json
+ input:
+ type: file
+ offsets_file: /data/offsets.yaml
+ watching_dir: /var/log/containers/
+ actions:
+ - type: file_multiline
+ field: log
+ split_event_size: 1000000
+ - type: json_decode
+ field: log
+```
+
+### How it works
+1. Reads the configured field (`log` by default) from each sequential event in the same file stream.
+2. If the value does **not** end with a real newline character `\n`, the chunk is buffered and the plugin waits for the next event. Further actions are not run yet.
+3. When the next chunk arrives, it is appended to the buffer and the end is checked again.
+4. When the value ends with `\n`, all buffered chunks are merged into one field and a single event is passed downstream.
+5. If the joined event exceeds `split_event_size`, it may be split forcibly even without `\n` at the end.
+6. If no continuation arrives within the pipeline `event_timeout` (default `30s`), the buffer is reset and the partial event is discarded.
+}*/
+
+type Plugin struct {
+ config *Config
+ field []string
+
+ logger *zap.SugaredLogger
+ controller pipeline.ActionPluginController
+ maxEventSize int
+ sourceNameMetaField string
+ cutOffEventByLimit bool
+ cutOffEventByLimitField string
+
+ eventBuf []byte
+ eventSize int
+ skipNextEvent bool
+ cutOffEvent bool
+}
+
+const (
+ predictionLookahead = 128 * 1024
+ newLine = `\n`
+)
+
+// ! config-params
+// ^ config-params
+type Config struct {
+ // > @3@4@5@6
+ // >
+ // > The event field which will be joined.
+ Field cfg.FieldSelector `json:"field" default:"log" parse:"selector"` // *
+ Field_ []string
+
+ // > @3@4@5@6
+ // >
+ // > Docker splits long logs by 16kb chunks. The plugin joins them back, but if an event is longer than this value in bytes, it will be split after all.
+ // > > Due to the optimization process it's not a strict rule. Events may be split even if they won't exceed the limit.
+ SplitEventSize int `json:"split_event_size" default:"1000000"` // *
+}
+
+func init() {
+ fd.DefaultPluginRegistry.RegisterAction(&pipeline.PluginStaticInfo{
+ Type: "file_multiline",
+ Factory: factory,
+ })
+}
+
+func factory() (pipeline.AnyPlugin, pipeline.AnyConfig) {
+ return &Plugin{}, &Config{}
+}
+
+func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginParams) {
+ p.logger = params.Logger
+ p.controller = params.Controller
+ p.maxEventSize = params.PipelineSettings.MaxEventSize
+ p.sourceNameMetaField = params.PipelineSettings.SourceNameMetaField
+ p.cutOffEventByLimit = params.PipelineSettings.CutOffEventByLimit
+ p.cutOffEventByLimitField = params.PipelineSettings.CutOffEventByLimitField
+
+ p.config = config.(*Config)
+ p.field = p.config.Field_
+ if len(p.field) == 0 {
+ p.field = cfg.ParseFieldSelector("log")
+ }
+
+ p.eventBuf = append(p.eventBuf, '"')
+}
+
+func (p *Plugin) Stop() {
+}
+
+type bytesBuf struct {
+ B []byte
+}
+
+var escapedStringBufPool = sync.Pool{
+ New: func() any {
+ return new(bytesBuf)
+ },
+}
+
+func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult {
+ if event.IsTimeoutKind() {
+ p.logger.Errorf("can't read next sequential event for file stream, source=%s", event.SourceName)
+ p.resetLogBuf()
+ return pipeline.ActionDiscard
+ }
+
+ logNode := event.Root.Dig(p.field...)
+ if logNode == nil {
+ if len(p.eventBuf) > 1 {
+ p.logger.Warnf("field %v is missing while joining split log, source=%s", p.field, event.SourceName)
+ p.resetLogBuf()
+ }
+ return pipeline.ActionPass
+ }
+
+ buf := escapedStringBufPool.Get().(*bytesBuf)
+ defer escapedStringBufPool.Put(buf)
+ buf.B = slices.Grow(buf.B[:0], event.Size)
+
+ buf.B = logNode.AppendEscapedString(buf.B)
+ logFragment := pipeline.ByteToStringUnsafe(buf.B)
+ if logFragment == "" {
+ return pipeline.ActionPass
+ }
+
+ logFragmentLen := len(logFragment)
+ if logFragmentLen < 3 {
+ return pipeline.ActionPass
+ }
+
+ logValue := logNode.AsString()
+
+ p.eventSize += event.Size
+ predictedLen := p.eventSize + predictionLookahead
+ shouldSplit := predictedLen > p.config.SplitEventSize
+ // check the real trailing newline of the log line, not escaped JSON suffix:
+ // chunk may end with literal "\r\n" text and falsely match the old escaped check.
+ isEnd := len(logValue) > 0 && logValue[len(logValue)-1] == '\n'
+ if !isEnd && !shouldSplit {
+ sizeAfterAppend := len(p.eventBuf) + len(logFragment)
+ if p.maxEventSize == 0 || sizeAfterAppend < p.maxEventSize {
+ p.eventBuf = append(p.eventBuf, logFragment[1:logFragmentLen-1]...)
+ } else if !p.skipNextEvent {
+ source := event.SourceName
+ if p.sourceNameMetaField != "" {
+ if val := event.Root.Dig(p.sourceNameMetaField).AsString(); val != "" {
+ source = val
+ }
+ }
+ if p.controller != nil {
+ p.controller.IncMaxEventSizeExceeded(source)
+ }
+
+ p.skipNextEvent = true
+
+ if p.cutOffEventByLimit {
+ offset := sizeAfterAppend - p.maxEventSize
+ p.eventBuf = append(p.eventBuf, logFragment[1:logFragmentLen-1-offset]...)
+ p.cutOffEvent = true
+
+ p.logger.Errorf("event chunk will be cut off due to max_event_size, source_name=%s", event.SourceName)
+ } else {
+ p.logger.Errorf("event chunk will be discarded due to max_event_size, source_name=%s", event.SourceName)
+ }
+ }
+ return pipeline.ActionCollapse
+ }
+
+ if p.skipNextEvent {
+ if !isEnd {
+ return pipeline.ActionCollapse
+ }
+ p.skipNextEvent = false
+
+ if !p.cutOffEvent {
+ p.resetLogBuf()
+ return pipeline.ActionDiscard
+ }
+ }
+
+ if shouldSplit {
+ p.logger.Warnf("too long event found, it'll be split, source=%s consider increase split_event_size, split_event_size=%d, predicted event size=%d", event.SourceName, p.config.SplitEventSize, predictedLen)
+ }
+
+ if len(p.eventBuf) > 1 {
+ if !p.cutOffEvent {
+ p.eventBuf = append(p.eventBuf, logFragment[1:logFragmentLen-1]...)
+ } else {
+ if isEnd {
+ p.eventBuf = append(p.eventBuf, newLine...)
+ }
+
+ if p.cutOffEventByLimitField != "" {
+ event.Root.AddFieldNoAlloc(event.Root, p.cutOffEventByLimitField).MutateToBool(true)
+ }
+ }
+ p.eventBuf = append(p.eventBuf, '"')
+
+ l := len(event.Buf)
+ event.Buf = append(event.Buf, p.eventBuf...)
+ pipeline.CreateNestedField(event.Root, p.field).MutateToEscapedString(pipeline.ByteToStringUnsafe(event.Buf[l:]))
+ }
+ p.resetLogBuf()
+
+ return pipeline.ActionPass
+}
+
+func (p *Plugin) resetLogBuf() {
+ p.eventBuf = p.eventBuf[:1]
+ p.eventSize = 0
+ p.cutOffEvent = false
+}
diff --git a/plugin/action/file_multiline/file_multiline_test.go b/plugin/action/file_multiline/file_multiline_test.go
new file mode 100644
index 000000000..b964c98ad
--- /dev/null
+++ b/plugin/action/file_multiline/file_multiline_test.go
@@ -0,0 +1,154 @@
+package file_multiline
+
+import (
+ "fmt"
+ "testing"
+
+ "github.com/ozontech/file.d/pipeline"
+ "github.com/ozontech/file.d/test"
+ insaneJSON "github.com/ozontech/insane-json"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestPlugin_Do(t *testing.T) {
+ plugin := &Plugin{}
+ config := &Config{
+ SplitEventSize: predictionLookahead * 4,
+ }
+ params := test.NewEmptyActionPluginParams()
+ params.PipelineSettings = &pipeline.Settings{MaxEventSize: 20}
+ plugin.Start(config, params)
+
+ sourceName := "/var/log/containers/app.log"
+
+ tcs := []struct {
+ Name string
+ EventParts []string
+
+ CutOffEventByLimit bool
+ CutOffEventByLimitField string
+
+ ActionResults []pipeline.ActionResult
+ ExpectedRoot string
+ }{
+ {
+ Name: "ok",
+ EventParts: []string{`{"log": "hello"}`, `{"log": " "}`, `{"log": "world\n"}`},
+ ActionResults: []pipeline.ActionResult{pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionPass},
+ ExpectedRoot: `{"log":"hello world\n"}`,
+ },
+ {
+ Name: "continue process events",
+ EventParts: []string{`{"log": "some "}`, `{"log": "other "}`, `{"log": "logs\n"}`},
+ ActionResults: []pipeline.ActionResult{pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionPass},
+ ExpectedRoot: `{"log":"some other logs\n"}`,
+ },
+ {
+ Name: "literal backslash-r-n suffix is not record end",
+ EventParts: []string{
+ `{"log": "chunk1"}`,
+ `{"log": "OAA\\r\\n"}`,
+ `{"log": "chunk2\n"}`,
+ },
+ ActionResults: []pipeline.ActionResult{pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionPass},
+ ExpectedRoot: `{"log":"chunk1OAA\\r\\nchunk2\n"}`,
+ },
+ {
+ Name: "must discard long event",
+ EventParts: []string{`{"log": "some "}`, `{"log": "other long "}`, `{"log":"long long"}`, `{"log": "event\n"}`},
+ ActionResults: []pipeline.ActionResult{pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionDiscard},
+ },
+ {
+ Name: "must cutoff long event",
+ EventParts: []string{`{"log": "some "}`, `{"log": "other long "}`, `{"log":"long long"}`, `{"log": "event\n"}`},
+ CutOffEventByLimit: true,
+ ActionResults: []pipeline.ActionResult{pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionPass},
+ ExpectedRoot: `{"log":"some other long l\n"}`,
+ },
+ {
+ Name: "must cutoff long event with field",
+ EventParts: []string{`{"log": "some "}`, `{"log": "other long "}`, `{"log":"long long"}`, `{"log": "event\n"}`},
+ CutOffEventByLimit: true,
+ CutOffEventByLimitField: "cutoff",
+ ActionResults: []pipeline.ActionResult{pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionCollapse, pipeline.ActionPass},
+ ExpectedRoot: `{"log":"some other long l\n","cutoff":true}`,
+ },
+ }
+
+ root := insaneJSON.Spawn()
+ defer insaneJSON.Release(root)
+
+ for _, tc := range tcs {
+ t.Run(tc.Name, func(t *testing.T) {
+ for i, part := range tc.EventParts {
+ plugin.cutOffEventByLimit = tc.CutOffEventByLimit
+ plugin.cutOffEventByLimitField = tc.CutOffEventByLimitField
+
+ require.NoError(t, root.DecodeString(part))
+ event := &pipeline.Event{Root: root, SourceName: sourceName, Size: len(part)}
+
+ result := plugin.Do(event)
+
+ assert.Equalf(t, tc.ActionResults[i], result, "wrong action result for iteration=%v, part=%s", i, part)
+ }
+ if tc.ActionResults[len(tc.ActionResults)-1] != pipeline.ActionDiscard {
+ assert.Equal(t, tc.ExpectedRoot, root.EncodeToString())
+ }
+ })
+ }
+}
+
+func TestPlugin_Do_shouldSplit(t *testing.T) {
+ plugin := &Plugin{}
+ config := &Config{
+ SplitEventSize: predictionLookahead * 4,
+ }
+ plugin.Start(config, test.NewEmptyActionPluginParams())
+
+ sourceName := "/var/log/containers/app.log"
+
+ tcs := []struct {
+ Name string
+ MaxEventSize int
+ Message string
+ ExpectedLogField string
+ ActionResult pipeline.ActionResult
+ }{
+ {
+ Name: "is last chunk and shouldSplit is false",
+ MaxEventSize: 1,
+ Message: wrapLogContent(`hi\n`),
+ ExpectedLogField: `"hi\n"`,
+ ActionResult: pipeline.ActionPass,
+ },
+ {
+ Name: "isnt last chunk and shouldSplit is false",
+ MaxEventSize: 10,
+ Message: wrapLogContent(`hi`),
+ ExpectedLogField: `"hi"`,
+ ActionResult: pipeline.ActionCollapse,
+ },
+ }
+
+ root := insaneJSON.Spawn()
+ defer insaneJSON.Release(root)
+
+ for _, tc := range tcs {
+ t.Run(tc.Name, func(t *testing.T) {
+ plugin.maxEventSize = tc.MaxEventSize
+ require.NoError(t, root.DecodeString(tc.Message))
+ event := &pipeline.Event{Root: root, SourceName: sourceName, Size: len(tc.Message)}
+
+ result := plugin.Do(event)
+ resultRoot := root.Dig("log").EncodeToString()
+
+ assert.Equalf(t, tc.ActionResult, result, "wrong action result")
+ assert.Equal(t, tc.ExpectedLogField, resultRoot)
+ })
+ }
+}
+
+func wrapLogContent(s string) string {
+ return fmt.Sprintf(`{"log": "%s"}`, s)
+}