diff --git a/encoder/encoder.go b/encoder/encoder.go
new file mode 100644
index 000000000..04d3afe40
--- /dev/null
+++ b/encoder/encoder.go
@@ -0,0 +1,41 @@
+package encoder
+
+import (
+ "encoding/json"
+ "fmt"
+
+ "github.com/ozontech/file.d/pipeline"
+)
+
+const (
+ EncoderTypeJSON = "json"
+ EncoderTypeRaw = "raw"
+)
+
+type Encoder interface {
+ Encode(event *pipeline.Event, buf []byte) ([]byte, error)
+}
+
+type EncodingConfig struct {
+ Type string `json:"type" default:"json" options:"json|raw"`
+ Params json.RawMessage `json:"params"`
+}
+
+func NewEncoder(cfg EncodingConfig) (Encoder, error) {
+ switch cfg.Type {
+ case EncoderTypeJSON, "":
+ return newJSONEncoder(&JSONEncoderParams{}), nil
+
+ case EncoderTypeRaw:
+ var params RawEncoderParams
+ if len(cfg.Params) > 0 {
+ if err := json.Unmarshal(cfg.Params, ¶ms); err != nil {
+ return nil, fmt.Errorf("raw encoder params: %w", err)
+ }
+ }
+ return newRawEncoder(¶ms), nil
+
+ default:
+ return nil, fmt.Errorf("unknown encoding type %q; supported: json, raw", cfg.Type)
+ }
+}
diff --git a/encoder/encoder_test.go b/encoder/encoder_test.go
new file mode 100644
index 000000000..a34b74b56
--- /dev/null
+++ b/encoder/encoder_test.go
@@ -0,0 +1,153 @@
+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, err := enc.Encode(event, nil)
+ require.NoError(t, err)
+ 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, 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
new file mode 100644
index 000000000..e8c2b40a4
--- /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, error) {
+ buf, _ = event.Encode(buf)
+ return buf, nil
+}
diff --git a/encoder/json_test.go b/encoder/json_test.go
new file mode 100644
index 000000000..077b675ec
--- /dev/null
+++ b/encoder/json_test.go
@@ -0,0 +1,67 @@
+package encoder
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+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, 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
new file mode 100644
index 000000000..c4f8d92aa
--- /dev/null
+++ b/encoder/raw.go
@@ -0,0 +1,41 @@
+package encoder
+
+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"`
+}
+
+type RawEncoder struct {
+ field string
+}
+
+func newRawEncoder(params *RawEncoderParams) *RawEncoder {
+ field := params.Field
+ if field == "" {
+ field = defaultRawField
+ }
+ return &RawEncoder{field: field}
+}
+
+func (e *RawEncoder) Encode(event *pipeline.Event, buf []byte) ([]byte, error) {
+ node := event.Root.Dig(e.field)
+ if node == nil {
+ return buf, fmt.Errorf("%w: %q", ErrFieldNotFound, e.field)
+ }
+
+ if node.IsString() {
+ return append(buf, node.AsBytes()...), nil
+ }
+
+ return node.Encode(buf), nil
+}
diff --git a/encoder/raw_test.go b/encoder/raw_test.go
new file mode 100644
index 000000000..e535e292f
--- /dev/null
+++ b/encoder/raw_test.go
@@ -0,0 +1,182 @@
+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
+ wantErr bool
+ }{
+ {
+ 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 and ErrFieldNotFound",
+ field: "message",
+ input: `{"other":"value"}`,
+ expected: "",
+ wantErr: true,
+ },
+ {
+ 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, 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/README.md b/plugin/output/http/README.md
index 7a1763dc8..bd98b0b62 100755
--- a/plugin/output/http/README.md
+++ b/plugin/output/http/README.md
@@ -17,14 +17,28 @@ Content-Type header for HTTP requests.
-**`encoding`** *`EncodingConfig`*
+**`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.
+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/http/encoding.go b/plugin/output/http/encoding.go
deleted file mode 100644
index 78c76707a..000000000
--- a/plugin/output/http/encoding.go
+++ /dev/null
@@ -1,83 +0,0 @@
-package http
-
-import (
- "encoding/json"
- "fmt"
-
- "github.com/ozontech/file.d/pipeline"
-)
-
-const (
- EncoderTypeJSON = "json"
- EncoderTypeRaw = "raw"
-)
-
-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"`
-}
-
-func NewEncoder(cfg EncodingConfig) (Encoder, error) {
- switch cfg.Type {
- case EncoderTypeJSON, "":
- return newJSONEncoder(&JSONEncoderParams{}), nil
-
- case EncoderTypeRaw:
- var params RawEncoderParams
- if len(cfg.Params) > 0 {
- if err := json.Unmarshal(cfg.Params, ¶ms); err != nil {
- return nil, fmt.Errorf("raw encoder params: %w", err)
- }
- }
- return newRawEncoder(¶ms), nil
-
- default:
- return nil, fmt.Errorf("unknown encoding type %q; supported: json, raw", cfg.Type)
- }
-}
diff --git a/plugin/output/http/http.go b/plugin/output/http/http.go
index bae5be023..1cdc321ed 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
@@ -67,12 +68,25 @@ 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.
- Encoding EncodingConfig `json:"encoding" child:"true"` // *
+ // > 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
// >
@@ -234,7 +248,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))
}
@@ -374,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 b6ca5f22d..a616c3074 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,18 +27,20 @@ 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, _ = rawEncoder.Encode(event, data.outBuf)
data.outBuf = append(data.outBuf, '\n')
expected = fmt.Sprintf("%s\n", `[INFO] some event`)
@@ -48,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/README.md b/plugin/output/kafka/README.md
index f9cce8402..3e8401384 100755
--- a/plugin/output/kafka/README.md
+++ b/plugin/output/kafka/README.md
@@ -72,6 +72,31 @@ 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` 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
+```
+
+
+
**`compression`** *`string`* *`default=none`* *`options=none|gzip|snappy|lz4|zstd`*
Compression codec
diff --git a/plugin/output/kafka/kafka.go b/plugin/output/kafka/kafka.go
index e3f24fae2..ec57df0ee 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"
@@ -36,6 +37,7 @@ type Plugin struct {
controller pipeline.OutputPluginController
client KafkaClient
+ encoder encoder.Encoder
batcher *pipeline.RetriableBatcher
router *pipeline.Router
tokenSource xoauth.TokenSource
@@ -113,6 +115,30 @@ 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` 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
// >
// > Compression codec
@@ -269,12 +295,17 @@ 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.Info("starting",
zap.Int("workers_count", p.config.WorkersCount_),
zap.Int("batch_size", p.config.BatchSize_),
)
- var err error
p.tokenSource, err = cfg.GetKafkaClientOAuthTokenSource(p.ctx, p.config)
if err != nil {
p.logger.Fatal("can't create oauth token source", zap.Error(err))
@@ -358,7 +389,12 @@ 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)
+ var err error
+ outBuf, err = p.encoder.Encode(event, outBuf)
+ if err != nil {
+ p.logger.Warn("can't encode event, sending empty value", zap.Error(err))
+ }
topic := p.config.DefaultTopic
if p.config.UseTopicField {