From 41759f75efb097eb1175ba69f0504b32bcc41933 Mon Sep 17 00:00:00 2001 From: timggggggg Date: Wed, 24 Jun 2026 17:55:02 +0300 Subject: [PATCH 1/5] Kafka output encode parameter --- .../http/encoding.go => encoder/encoder.go | 44 +------------------ encoder/json.go | 16 +++++++ encoder/raw.go | 32 ++++++++++++++ plugin/output/http/http.go | 7 +-- plugin/output/http/http_test.go | 13 +++--- plugin/output/kafka/kafka.go | 22 +++++++++- 6 files changed, 82 insertions(+), 52 deletions(-) rename plugin/output/http/encoding.go => encoder/encoder.go (52%) create mode 100644 encoder/json.go create mode 100644 encoder/raw.go diff --git a/plugin/output/http/encoding.go b/encoder/encoder.go similarity index 52% rename from plugin/output/http/encoding.go rename to encoder/encoder.go index 78c76707a..3dde7ebec 100644 --- a/plugin/output/http/encoding.go +++ b/encoder/encoder.go @@ -1,4 +1,4 @@ -package http +package encoder import ( "encoding/json" @@ -16,48 +16,6 @@ type Encoder interface { Encode(event *pipeline.Event, buf []byte) []byte } -type JSONEncoderParams struct{} - -type JSONEncoder struct{} - -func newJSONEncoder(_ *JSONEncoderParams) *JSONEncoder { - return &JSONEncoder{} -} - -func (e *JSONEncoder) Encode(event *pipeline.Event, buf []byte) []byte { - buf, _ = event.Encode(buf) - return buf -} - -type RawEncoderParams struct { - Field string `json:"field" default:"message"` -} - -type RawEncoder struct { - field string -} - -func newRawEncoder(params *RawEncoderParams) *RawEncoder { - field := params.Field - if field == "" { - field = "message" - } - return &RawEncoder{field: field} -} - -func (e *RawEncoder) Encode(event *pipeline.Event, buf []byte) []byte { - node := event.Root.Dig(e.field) - if node == nil { - return buf[:0] - } - - if node.IsString() { - return append(buf, node.AsBytes()...) - } - - return node.Encode(buf) -} - type EncodingConfig struct { Type string `json:"type" default:"json" options:"json|raw"` Params json.RawMessage `json:"params"` diff --git a/encoder/json.go b/encoder/json.go new file mode 100644 index 000000000..5449b589a --- /dev/null +++ b/encoder/json.go @@ -0,0 +1,16 @@ +package encoder + +import "github.com/ozontech/file.d/pipeline" + +type JSONEncoderParams struct{} + +type JSONEncoder struct{} + +func newJSONEncoder(_ *JSONEncoderParams) *JSONEncoder { + return &JSONEncoder{} +} + +func (e *JSONEncoder) Encode(event *pipeline.Event, buf []byte) []byte { + buf, _ = event.Encode(buf) + return buf +} diff --git a/encoder/raw.go b/encoder/raw.go new file mode 100644 index 000000000..e8d5996df --- /dev/null +++ b/encoder/raw.go @@ -0,0 +1,32 @@ +package encoder + +import "github.com/ozontech/file.d/pipeline" + +type RawEncoderParams struct { + Field string `json:"field" default:"message"` +} + +type RawEncoder struct { + field string +} + +func newRawEncoder(params *RawEncoderParams) *RawEncoder { + field := params.Field + if field == "" { + field = "message" + } + return &RawEncoder{field: field} +} + +func (e *RawEncoder) Encode(event *pipeline.Event, buf []byte) []byte { + node := event.Root.Dig(e.field) + if node == nil { + return buf[:0] + } + + if node.IsString() { + return append(buf, node.AsBytes()...) + } + + return node.Encode(buf) +} diff --git a/plugin/output/http/http.go b/plugin/output/http/http.go index bae5be023..d7a143f82 100644 --- a/plugin/output/http/http.go +++ b/plugin/output/http/http.go @@ -9,6 +9,7 @@ import ( "time" "github.com/ozontech/file.d/cfg" + "github.com/ozontech/file.d/encoder" "github.com/ozontech/file.d/fd" "github.com/ozontech/file.d/metric" "github.com/ozontech/file.d/pipeline" @@ -32,7 +33,7 @@ type Plugin struct { config *Config client *xhttp.Client - encoder Encoder + encoder encoder.Encoder logger *zap.Logger controller pipeline.OutputPluginController @@ -72,7 +73,7 @@ type Config struct { // > * `raw` - extracts a single field and sends its value as-is. // > By default `json` is used. // > 2) Params - Encoder parameters. - Encoding EncodingConfig `json:"encoding" child:"true"` // * + Encoding encoder.EncodingConfig `json:"encoding" child:"true"` // * // > @3@4@5@6 // > @@ -234,7 +235,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP } var err error - p.encoder, err = NewEncoder(p.config.Encoding) + p.encoder, err = encoder.NewEncoder(p.config.Encoding) if err != nil { p.logger.Fatal("can't create encoder", zap.Error(err)) } diff --git a/plugin/output/http/http_test.go b/plugin/output/http/http_test.go index b6ca5f22d..d6c1753f2 100644 --- a/plugin/output/http/http_test.go +++ b/plugin/output/http/http_test.go @@ -4,6 +4,7 @@ import ( "fmt" "testing" + "github.com/ozontech/file.d/encoder" "github.com/ozontech/file.d/pipeline" "github.com/ozontech/file.d/test" insaneJSON "github.com/ozontech/insane-json" @@ -26,16 +27,18 @@ func TestEncoding(t *testing.T) { data := data{} event := &pipeline.Event{Root: root} - encoder := newJSONEncoder(&JSONEncoderParams{}) - data.outBuf = encoder.Encode(event, data.outBuf) + jsonEncoder, _ := encoder.NewEncoder(encoder.EncodingConfig{ + Type: encoder.EncoderTypeJSON, + }) + data.outBuf = jsonEncoder.Encode(event, data.outBuf) data.outBuf = append(data.outBuf, '\n') expected := fmt.Sprintf("%s\n", `{"message":"[INFO] some event","field_a":"AAAA","field_b":"BBBB"}`) assert.Equal(t, expected, string(data.outBuf), "wrong request content") - var params RawEncoderParams - rawEncoder := newRawEncoder(¶ms) - + rawEncoder, _ := encoder.NewEncoder(encoder.EncodingConfig{ + Type: encoder.EncoderTypeRaw, + }) data.outBuf = data.outBuf[:0] data.outBuf = rawEncoder.Encode(event, data.outBuf) data.outBuf = append(data.outBuf, '\n') diff --git a/plugin/output/kafka/kafka.go b/plugin/output/kafka/kafka.go index c43829fba..6bea4add1 100644 --- a/plugin/output/kafka/kafka.go +++ b/plugin/output/kafka/kafka.go @@ -5,6 +5,7 @@ import ( "time" "github.com/ozontech/file.d/cfg" + "github.com/ozontech/file.d/encoder" "github.com/ozontech/file.d/fd" "github.com/ozontech/file.d/metric" "github.com/ozontech/file.d/pipeline" @@ -35,6 +36,7 @@ type Plugin struct { controller pipeline.OutputPluginController client KafkaClient + encoder encoder.Encoder batcher *pipeline.RetriableBatcher ctx context.Context cancelFunc context.CancelFunc @@ -110,6 +112,17 @@ type Config struct { MaxMessageBytes cfg.Expression `json:"max_message_bytes" default:"1000000" parse:"expression"` // * MaxMessageBytes_ int + // > @3@4@5@6 + // > + // > Configure event serialization before sending. + // > Includes: + // > 1) Type - codec to use for serializing events: + // > * `json` - serializes the full event as a JSON object (default). + // > * `raw` - extracts a single field and sends its value as-is. + // > By default `json` is used. + // > 2) Params - Encoder parameters. + Encoding encoder.EncodingConfig `json:"encoding" child:"true"` // * + // > @3@4@5@6 // > // > Compression codec @@ -250,6 +263,12 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP p.logger.Fatal("'retention' can't be <1") } + var err error + p.encoder, err = encoder.NewEncoder(p.config.Encoding) + if err != nil { + p.logger.Fatal("can't create encoder", zap.Error(err)) + } + p.logger.Infof("workers count=%d, batch size=%d", p.config.WorkersCount_, p.config.BatchSize_) p.client = NewClient(p.config, p.logger.Desugar()) @@ -330,7 +349,8 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err defer cancel() batch.ForEach(func(event *pipeline.Event) { - outBuf, start = event.Encode(outBuf) + start = len(outBuf) + outBuf = p.encoder.Encode(event, outBuf) topic := p.config.DefaultTopic if p.config.UseTopicField { From 0d08468948f49deba68bed078ebbc764ee104b86 Mon Sep 17 00:00:00 2001 From: timggggggg Date: Wed, 24 Jun 2026 18:02:25 +0300 Subject: [PATCH 2/5] fix doc --- plugin/output/http/README.md | 2 +- plugin/output/kafka/README.md | 11 +++++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/plugin/output/http/README.md b/plugin/output/http/README.md index 7a1763dc8..990630527 100755 --- a/plugin/output/http/README.md +++ b/plugin/output/http/README.md @@ -17,7 +17,7 @@ Content-Type header for HTTP requests.
-**`encoding`** *`EncodingConfig`* +**`encoding`** *`encoder.EncodingConfig`* Configure event serialization before sending. Includes: diff --git a/plugin/output/kafka/README.md b/plugin/output/kafka/README.md index bb4dc3c28..b5072d8ac 100755 --- a/plugin/output/kafka/README.md +++ b/plugin/output/kafka/README.md @@ -72,6 +72,17 @@ Should be set equal to or smaller than the broker's `message.max.bytes`.
+**`encoding`** *`encoder.EncodingConfig`* + +Configure event serialization before sending. +Includes: +1) Type - codec to use for serializing events: +* `json` - serializes the full event as a JSON object (default). +* `raw` - extracts a single field and sends its value as-is. +2) Params - Encoder parameters. + +
+ **`compression`** *`string`* *`default=none`* *`options=none|gzip|snappy|lz4|zstd`* Compression codec From cfdffc8e1911638c3f05cb8e4fdfaa52dea0ec6d Mon Sep 17 00:00:00 2001 From: timggggggg Date: Fri, 26 Jun 2026 16:59:52 +0300 Subject: [PATCH 3/5] add encoder tests --- encoder/encoder_test.go | 151 ++++++++++++++++++++++++++++++++++++++++ encoder/json_test.go | 65 +++++++++++++++++ encoder/raw_test.go | 128 ++++++++++++++++++++++++++++++++++ 3 files changed, 344 insertions(+) create mode 100644 encoder/encoder_test.go create mode 100644 encoder/json_test.go create mode 100644 encoder/raw_test.go diff --git a/encoder/encoder_test.go b/encoder/encoder_test.go new file mode 100644 index 000000000..ef0b61153 --- /dev/null +++ b/encoder/encoder_test.go @@ -0,0 +1,151 @@ +package encoder + +import ( + "encoding/json" + "testing" + + "github.com/ozontech/file.d/pipeline" + insaneJSON "github.com/ozontech/insane-json" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newTestEvent(t testing.TB, raw string) *pipeline.Event { + t.Helper() + + root, err := insaneJSON.DecodeString(raw) + require.NoError(t, err) + + t.Cleanup(func() { + insaneJSON.Release(root) + }) + + return &pipeline.Event{Root: root} +} + +func TestNewEncoder(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + cfg EncodingConfig + wantErr bool + errContains string + assertType func(t *testing.T, enc Encoder) + }{ + { + name: "explicit json type", + cfg: EncodingConfig{Type: EncoderTypeJSON}, + assertType: func(t *testing.T, enc Encoder) { + assert.IsType(t, &JSONEncoder{}, enc) + }, + }, + { + name: "empty type defaults to json", + cfg: EncodingConfig{Type: ""}, + assertType: func(t *testing.T, enc Encoder) { + assert.IsType(t, &JSONEncoder{}, enc) + }, + }, + { + name: "raw type without params uses default field", + cfg: EncodingConfig{Type: EncoderTypeRaw}, + assertType: func(t *testing.T, enc Encoder) { + raw, ok := enc.(*RawEncoder) + require.True(t, ok) + assert.Equal(t, "message", raw.field) + }, + }, + { + name: "raw type with empty params object uses default field", + cfg: EncodingConfig{Type: EncoderTypeRaw, Params: json.RawMessage(`{}`)}, + assertType: func(t *testing.T, enc Encoder) { + raw, ok := enc.(*RawEncoder) + require.True(t, ok) + assert.Equal(t, "message", raw.field) + }, + }, + { + name: "raw type with custom field", + cfg: EncodingConfig{Type: EncoderTypeRaw, Params: json.RawMessage(`{"field":"data"}`)}, + assertType: func(t *testing.T, enc Encoder) { + raw, ok := enc.(*RawEncoder) + require.True(t, ok) + assert.Equal(t, "data", raw.field) + }, + }, + { + name: "raw type with empty field falls back to message", + cfg: EncodingConfig{Type: EncoderTypeRaw, Params: json.RawMessage(`{"field":""}`)}, + assertType: func(t *testing.T, enc Encoder) { + raw, ok := enc.(*RawEncoder) + require.True(t, ok) + assert.Equal(t, "message", raw.field) + }, + }, + { + name: "raw type with invalid params", + cfg: EncodingConfig{Type: EncoderTypeRaw, Params: json.RawMessage(`{"field":`)}, + wantErr: true, + errContains: "raw encoder params", + }, + { + name: "unknown type", + cfg: EncodingConfig{Type: "yaml"}, + wantErr: true, + errContains: `unknown encoding type "yaml"`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + enc, err := NewEncoder(tt.cfg) + + if tt.wantErr { + require.Error(t, err) + assert.Nil(t, enc) + if tt.errContains != "" { + assert.Contains(t, err.Error(), tt.errContains) + } + return + } + + require.NoError(t, err) + require.NotNil(t, enc) + if tt.assertType != nil { + tt.assertType(t, enc) + } + }) + } +} + +func TestEncode(t *testing.T) { + t.Parallel() + + t.Run("json", func(t *testing.T) { + t.Parallel() + + enc, err := NewEncoder(EncodingConfig{Type: EncoderTypeJSON}) + require.NoError(t, err) + + event := newTestEvent(t, `{"message":"hi"}`) + out := enc.Encode(event, nil) + assert.JSONEq(t, `{"message":"hi"}`, string(out)) + }) + + t.Run("raw", func(t *testing.T) { + t.Parallel() + + enc, err := NewEncoder(EncodingConfig{ + Type: EncoderTypeRaw, + Params: json.RawMessage(`{"field":"message"}`), + }) + require.NoError(t, err) + + event := newTestEvent(t, `{"message":"hi"}`) + out := enc.Encode(event, nil) + assert.Equal(t, "hi", string(out)) + }) +} diff --git a/encoder/json_test.go b/encoder/json_test.go new file mode 100644 index 000000000..1764e0b52 --- /dev/null +++ b/encoder/json_test.go @@ -0,0 +1,65 @@ +package encoder + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestJSONEncode(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + input string + expected string + }{ + { + name: "simple object", + input: `{"message":"hello"}`, + expected: `{"message":"hello"}`, + }, + { + name: "nested object", + input: `{"level":"info","data":{"a":1,"b":[1,2,3]}}`, + expected: `{"level":"info","data":{"a":1,"b":[1,2,3]}}`, + }, + { + name: "array root", + input: `[1,2,3]`, + expected: `[1,2,3]`, + }, + { + name: "scalar string root", + input: `"just a string"`, + expected: `"just a string"`, + }, + { + name: "number root", + input: `42`, + expected: `42`, + }, + { + name: "empty object", + input: `{}`, + expected: `{}`, + }, + { + name: "special characters escaped", + input: `{"msg":"line1\nline2\t\"quoted\""}`, + expected: `{"msg":"line1\nline2\t\"quoted\""}`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + enc := newJSONEncoder(&JSONEncoderParams{}) + event := newTestEvent(t, tt.input) + + out := enc.Encode(event, nil) + assert.JSONEq(t, tt.expected, string(out)) + }) + } +} diff --git a/encoder/raw_test.go b/encoder/raw_test.go new file mode 100644 index 000000000..6d9c19354 --- /dev/null +++ b/encoder/raw_test.go @@ -0,0 +1,128 @@ +package encoder + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewRawEncoder(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + params *RawEncoderParams + wantField string + }{ + { + name: "custom field", + params: &RawEncoderParams{Field: "data"}, + wantField: "data", + }, + { + name: "empty field falls back to message", + params: &RawEncoderParams{Field: ""}, + wantField: "message", + }, + { + name: "nested field path", + params: &RawEncoderParams{Field: "log.message"}, + wantField: "log.message", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + enc := newRawEncoder(tt.params) + require.NotNil(t, enc) + assert.Equal(t, tt.wantField, enc.field) + }) + } +} + +func TestRawEncode(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + field string + input string + expected string + }{ + { + name: "string field returns raw value without quotes", + field: "message", + input: `{"message":"hello world"}`, + expected: "hello world", + }, + { + name: "string field with special characters keeps them raw", + field: "message", + input: `{"message":"line1\nline2"}`, + expected: "line1\nline2", + }, + { + name: "missing field returns empty", + field: "message", + input: `{"other":"value"}`, + expected: "", + }, + { + name: "number field returns encoded representation", + field: "code", + input: `{"code":200}`, + expected: "200", + }, + { + name: "bool field returns encoded representation", + field: "ok", + input: `{"ok":true}`, + expected: "true", + }, + { + name: "object field returns encoded JSON", + field: "data", + input: `{"data":{"a":1}}`, + expected: `{"a":1}`, + }, + { + name: "array field returns encoded JSON", + field: "items", + input: `{"items":[1,2,3]}`, + expected: `[1,2,3]`, + }, + { + name: "null field returns encoded null", + field: "value", + input: `{"value":null}`, + expected: "null", + }, + { + name: "empty string field returns empty", + field: "message", + input: `{"message":""}`, + expected: "", + }, + { + name: "unescape string field", + field: "message", + input: `{"message":"{\"log\":\"[INFO] some event\"}","field_a":"AAAA","field_b":"BBBB"}`, + expected: `{"log":"[INFO] some event"}`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + enc := newRawEncoder(&RawEncoderParams{Field: tt.field}) + event := newTestEvent(t, tt.input) + + out := enc.Encode(event, nil) + assert.Equal(t, tt.expected, string(out)) + }) + } +} From 6d49e612e838ce788ef0c3750381ab26e6050236 Mon Sep 17 00:00:00 2001 From: timggggggg Date: Wed, 15 Jul 2026 15:34:02 +0300 Subject: [PATCH 4/5] improve doc --- plugin/output/http/README.md | 22 ++++++++++++++++++---- plugin/output/kafka/README.md | 22 ++++++++++++++++++---- 2 files changed, 36 insertions(+), 8 deletions(-) diff --git a/plugin/output/http/README.md b/plugin/output/http/README.md index 990630527..bd98b0b62 100755 --- a/plugin/output/http/README.md +++ b/plugin/output/http/README.md @@ -21,10 +21,24 @@ Content-Type header for HTTP requests. Configure event serialization before sending. Includes: -1) Type - codec to use for serializing events: -* `json` - serializes the full event as a JSON object (default). -* `raw` - extracts a single field and sends its value as-is. -2) Params - Encoder parameters. +1) `type` - codec to use for serializing events (`json` by default): +* `json` - serializes the full event as a JSON object. +* `raw` - extracts a single field and sends its value as-is (unquoted + for string fields, encoded JSON otherwise). If the field is missing an + empty value is sent and a warning is logged. +2) `params` - encoder parameters, keyed by encoder type: +* `json` - none. +* `raw`: + * `field` - event field to extract (default `message`); supports + nested paths such as `log.message`. + +Example sending only the `message` field as a raw value: +```yaml +encoding: + type: raw + params: + field: message +```
diff --git a/plugin/output/kafka/README.md b/plugin/output/kafka/README.md index b5072d8ac..5ebea04b1 100755 --- a/plugin/output/kafka/README.md +++ b/plugin/output/kafka/README.md @@ -76,10 +76,24 @@ Should be set equal to or smaller than the broker's `message.max.bytes`. Configure event serialization before sending. Includes: -1) Type - codec to use for serializing events: -* `json` - serializes the full event as a JSON object (default). -* `raw` - extracts a single field and sends its value as-is. -2) Params - Encoder parameters. +1) `type` - codec to use for serializing events (`json` by default): +* `json` - serializes the full event as a JSON object. +* `raw` - extracts a single field and sends its value as-is (unquoted + for string fields, encoded JSON otherwise). If the field is missing an + empty value is sent and a warning is logged. +2) `params` - encoder parameters, keyed by encoder type: +* `json` - none. +* `raw`: + * `field` - event field to extract (default `message`); supports + nested paths such as `log.message`. + +Example sending only the `message` field as a raw value: +```yaml +encoding: + type: raw + params: + field: message +```
From ac6578acd1ec5f13374802eeaa7605b04a132d1c Mon Sep 17 00:00:00 2001 From: timggggggg Date: Wed, 15 Jul 2026 15:36:27 +0300 Subject: [PATCH 5/5] error when rawEncoder field is not found --- encoder/encoder.go | 2 +- encoder/encoder_test.go | 6 ++-- encoder/json.go | 4 +-- encoder/json_test.go | 4 ++- encoder/raw.go | 23 +++++++++---- encoder/raw_test.go | 58 +++++++++++++++++++++++++++++++-- plugin/output/http/http.go | 29 +++++++++++++---- plugin/output/http/http_test.go | 6 ++-- plugin/output/kafka/kafka.go | 29 +++++++++++++---- 9 files changed, 131 insertions(+), 30 deletions(-) diff --git a/encoder/encoder.go b/encoder/encoder.go index 3dde7ebec..04d3afe40 100644 --- a/encoder/encoder.go +++ b/encoder/encoder.go @@ -13,7 +13,7 @@ const ( ) type Encoder interface { - Encode(event *pipeline.Event, buf []byte) []byte + Encode(event *pipeline.Event, buf []byte) ([]byte, error) } type EncodingConfig struct { diff --git a/encoder/encoder_test.go b/encoder/encoder_test.go index ef0b61153..a34b74b56 100644 --- a/encoder/encoder_test.go +++ b/encoder/encoder_test.go @@ -131,7 +131,8 @@ func TestEncode(t *testing.T) { require.NoError(t, err) event := newTestEvent(t, `{"message":"hi"}`) - out := enc.Encode(event, nil) + out, err := enc.Encode(event, nil) + require.NoError(t, err) assert.JSONEq(t, `{"message":"hi"}`, string(out)) }) @@ -145,7 +146,8 @@ func TestEncode(t *testing.T) { require.NoError(t, err) event := newTestEvent(t, `{"message":"hi"}`) - out := enc.Encode(event, nil) + out, err := enc.Encode(event, nil) + require.NoError(t, err) assert.Equal(t, "hi", string(out)) }) } diff --git a/encoder/json.go b/encoder/json.go index 5449b589a..e8c2b40a4 100644 --- a/encoder/json.go +++ b/encoder/json.go @@ -10,7 +10,7 @@ func newJSONEncoder(_ *JSONEncoderParams) *JSONEncoder { return &JSONEncoder{} } -func (e *JSONEncoder) Encode(event *pipeline.Event, buf []byte) []byte { +func (e *JSONEncoder) Encode(event *pipeline.Event, buf []byte) ([]byte, error) { buf, _ = event.Encode(buf) - return buf + return buf, nil } diff --git a/encoder/json_test.go b/encoder/json_test.go index 1764e0b52..077b675ec 100644 --- a/encoder/json_test.go +++ b/encoder/json_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestJSONEncode(t *testing.T) { @@ -58,7 +59,8 @@ func TestJSONEncode(t *testing.T) { enc := newJSONEncoder(&JSONEncoderParams{}) event := newTestEvent(t, tt.input) - out := enc.Encode(event, nil) + out, err := enc.Encode(event, nil) + require.NoError(t, err) assert.JSONEq(t, tt.expected, string(out)) }) } diff --git a/encoder/raw.go b/encoder/raw.go index e8d5996df..c4f8d92aa 100644 --- a/encoder/raw.go +++ b/encoder/raw.go @@ -1,9 +1,18 @@ package encoder -import "github.com/ozontech/file.d/pipeline" +import ( + "errors" + "fmt" + + "github.com/ozontech/file.d/pipeline" +) + +const defaultRawField = "message" + +var ErrFieldNotFound = errors.New("field not found") type RawEncoderParams struct { - Field string `json:"field" default:"message"` + Field string `json:"field"` } type RawEncoder struct { @@ -13,20 +22,20 @@ type RawEncoder struct { func newRawEncoder(params *RawEncoderParams) *RawEncoder { field := params.Field if field == "" { - field = "message" + field = defaultRawField } return &RawEncoder{field: field} } -func (e *RawEncoder) Encode(event *pipeline.Event, buf []byte) []byte { +func (e *RawEncoder) Encode(event *pipeline.Event, buf []byte) ([]byte, error) { node := event.Root.Dig(e.field) if node == nil { - return buf[:0] + return buf, fmt.Errorf("%w: %q", ErrFieldNotFound, e.field) } if node.IsString() { - return append(buf, node.AsBytes()...) + return append(buf, node.AsBytes()...), nil } - return node.Encode(buf) + return node.Encode(buf), nil } diff --git a/encoder/raw_test.go b/encoder/raw_test.go index 6d9c19354..e535e292f 100644 --- a/encoder/raw_test.go +++ b/encoder/raw_test.go @@ -51,6 +51,7 @@ func TestRawEncode(t *testing.T) { field string input string expected string + wantErr bool }{ { name: "string field returns raw value without quotes", @@ -65,10 +66,11 @@ func TestRawEncode(t *testing.T) { expected: "line1\nline2", }, { - name: "missing field returns empty", + name: "missing field returns empty and ErrFieldNotFound", field: "message", input: `{"other":"value"}`, expected: "", + wantErr: true, }, { name: "number field returns encoded representation", @@ -121,7 +123,59 @@ func TestRawEncode(t *testing.T) { enc := newRawEncoder(&RawEncoderParams{Field: tt.field}) event := newTestEvent(t, tt.input) - out := enc.Encode(event, nil) + out, err := enc.Encode(event, nil) + if tt.wantErr { + require.ErrorIs(t, err, ErrFieldNotFound) + } else { + require.NoError(t, err) + } + assert.Equal(t, tt.expected, string(out)) + }) + } +} + +func TestRawEncodeAppendsToBuffer(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + input string + field string + expected string + wantErr bool + }{ + { + name: "present field appends after prefix", + input: `{"message":"hi"}`, + field: "message", + expected: "PREFIX/hi", + }, + { + name: "missing field keeps prefix intact", + input: `{"other":"x"}`, + field: "message", + expected: "PREFIX/", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + enc := newRawEncoder(&RawEncoderParams{Field: tt.field}) + event := newTestEvent(t, tt.input) + + buf := []byte("PREFIX/") + start := len(buf) + out, err := enc.Encode(event, buf) + if tt.wantErr { + require.ErrorIs(t, err, ErrFieldNotFound) + } else { + require.NoError(t, err) + } + + require.GreaterOrEqual(t, len(out), start) assert.Equal(t, tt.expected, string(out)) }) } diff --git a/plugin/output/http/http.go b/plugin/output/http/http.go index d7a143f82..1cdc321ed 100644 --- a/plugin/output/http/http.go +++ b/plugin/output/http/http.go @@ -68,11 +68,24 @@ type Config struct { // > // > Configure event serialization before sending. // > Includes: - // > 1) Type - codec to use for serializing events: - // > * `json` - serializes the full event as a JSON object (default). - // > * `raw` - extracts a single field and sends its value as-is. - // > By default `json` is used. - // > 2) Params - Encoder parameters. + // > 1) `type` - codec to use for serializing events (`json` by default): + // > * `json` - serializes the full event as a JSON object. + // > * `raw` - extracts a single field and sends its value as-is (unquoted + // > for string fields, encoded JSON otherwise). If the field is missing an + // > empty value is sent and a warning is logged. + // > 2) `params` - encoder parameters, keyed by encoder type: + // > * `json` - none. + // > * `raw`: + // > * `field` - event field to extract (default `message`); supports + // > nested paths such as `log.message`. + // > + // > Example sending only the `message` field as a raw value: + // > ```yaml + // > encoding: + // > type: raw + // > params: + // > field: message + // > ``` Encoding encoder.EncodingConfig `json:"encoding" child:"true"` // * // > @3@4@5@6 @@ -375,7 +388,11 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err batch.ForEach(func(event *pipeline.Event) { eventsCount++ data.begin = append(data.begin, len(data.outBuf)) - data.outBuf = p.encoder.Encode(event, data.outBuf) + var err error + data.outBuf, err = p.encoder.Encode(event, data.outBuf) + if err != nil { + p.logger.Warn("can't encode event, sending empty value", zap.Error(err)) + } data.outBuf = append(data.outBuf, '\n') }) data.begin = append(data.begin, len(data.outBuf)) diff --git a/plugin/output/http/http_test.go b/plugin/output/http/http_test.go index d6c1753f2..a616c3074 100644 --- a/plugin/output/http/http_test.go +++ b/plugin/output/http/http_test.go @@ -30,7 +30,7 @@ func TestEncoding(t *testing.T) { jsonEncoder, _ := encoder.NewEncoder(encoder.EncodingConfig{ Type: encoder.EncoderTypeJSON, }) - data.outBuf = jsonEncoder.Encode(event, data.outBuf) + data.outBuf, _ = jsonEncoder.Encode(event, data.outBuf) data.outBuf = append(data.outBuf, '\n') expected := fmt.Sprintf("%s\n", `{"message":"[INFO] some event","field_a":"AAAA","field_b":"BBBB"}`) @@ -40,7 +40,7 @@ func TestEncoding(t *testing.T) { Type: encoder.EncoderTypeRaw, }) data.outBuf = data.outBuf[:0] - data.outBuf = rawEncoder.Encode(event, data.outBuf) + data.outBuf, _ = rawEncoder.Encode(event, data.outBuf) data.outBuf = append(data.outBuf, '\n') expected = fmt.Sprintf("%s\n", `[INFO] some event`) @@ -51,7 +51,7 @@ func TestEncoding(t *testing.T) { event.Root = root2 data.outBuf = data.outBuf[:0] - data.outBuf = rawEncoder.Encode(event, data.outBuf) + data.outBuf, _ = rawEncoder.Encode(event, data.outBuf) data.outBuf = append(data.outBuf, '\n') expected = fmt.Sprintf("%s\n", `{"log":"[INFO] some event"}`) diff --git a/plugin/output/kafka/kafka.go b/plugin/output/kafka/kafka.go index 6bea4add1..6fea7702b 100644 --- a/plugin/output/kafka/kafka.go +++ b/plugin/output/kafka/kafka.go @@ -116,11 +116,24 @@ type Config struct { // > // > Configure event serialization before sending. // > Includes: - // > 1) Type - codec to use for serializing events: - // > * `json` - serializes the full event as a JSON object (default). - // > * `raw` - extracts a single field and sends its value as-is. - // > By default `json` is used. - // > 2) Params - Encoder parameters. + // > 1) `type` - codec to use for serializing events (`json` by default): + // > * `json` - serializes the full event as a JSON object. + // > * `raw` - extracts a single field and sends its value as-is (unquoted + // > for string fields, encoded JSON otherwise). If the field is missing an + // > empty value is sent and a warning is logged. + // > 2) `params` - encoder parameters, keyed by encoder type: + // > * `json` - none. + // > * `raw`: + // > * `field` - event field to extract (default `message`); supports + // > nested paths such as `log.message`. + // > + // > Example sending only the `message` field as a raw value: + // > ```yaml + // > encoding: + // > type: raw + // > params: + // > field: message + // > ``` Encoding encoder.EncodingConfig `json:"encoding" child:"true"` // * // > @3@4@5@6 @@ -350,7 +363,11 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err batch.ForEach(func(event *pipeline.Event) { start = len(outBuf) - outBuf = p.encoder.Encode(event, outBuf) + var err error + outBuf, err = p.encoder.Encode(event, outBuf) + if err != nil { + p.logger.Warnf("can't encode event, sending empty value: %v", err) + } topic := p.config.DefaultTopic if p.config.UseTopicField {