[extension/encoding] Add marshaling support to avrologencodingextension

.chloggen/avrologencoding-add-marshaler.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: avrologencodingextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for marshaling logs to AVRO.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33006]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

extension/encoding/avrologencodingextension/README.md

@@ -11,14 +11,19 @@
 [development]: https://github.com/open-telemetry/opentelemetry-collector#development
 <!-- end autogenerated section -->
 
-The `avrolog` encoding extension is used to unmarshal AVRO and insert it into the body of a log record. Marshalling is not supported.
+This extension allows to marshal and unmarshal AVRO encoded data from/into the body of a log record.
+
+The extension supports the following configuration options:
+
+- `schema_id` (optional): A schema ID used as prefix for the serialized data. If not provided or set to 0, the schema ID will not be included in the serialized data.
+- `schema` (required): The AVRO schema used to encode/decode the data.
 
-The extension accepts a configuration option to specify the Avro schema to use to read the log record body.
 
 Example:
 ```yaml
 extensions:
   avro_log_encoding:
+    schema_id: 1
     schema: |
       {
         "type" : "record",
@@ -29,4 +34,4 @@ extensions:
           { "name" : "Value" , "type" : "int" }
         ]
       }
-```
+```

extension/encoding/avrologencodingextension/avro.go

@@ -4,31 +4,45 @@
 package avrologencodingextension // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/avrologencodingextension"
 
 import (
+	"encoding/binary"
 	"fmt"
 
 	"github.com/linkedin/goavro/v2"
 )
 
-type avroDeserializer interface {
+type avroSerDe interface {
+	Serialize(map[string]any, uint32) ([]byte, error)
 	Deserialize([]byte) (map[string]any, error)
 }
 
-type avroStaticSchemaDeserializer struct {
+type avroStaticSchemaSerDe struct {
 	codec *goavro.Codec
 }
 
-func newAVROStaticSchemaDeserializer(schema string) (avroDeserializer, error) {
+func newAVROStaticSchemaSerDe(schema string) (avroSerDe, error) {
 	codec, err := goavro.NewCodec(schema)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create avro codec: %w", err)
 	}
 
-	return &avroStaticSchemaDeserializer{
-		codec: codec,
-	}, nil
+	return &avroStaticSchemaSerDe{codec: codec}, nil
 }
 
-func (d *avroStaticSchemaDeserializer) Deserialize(data []byte) (map[string]any, error) {
+func (d *avroStaticSchemaSerDe) Serialize(data map[string]any, schemaID uint32) ([]byte, error) {
+	logMsgBinary, err := d.codec.BinaryFromNative(nil, data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to serialize avro record: %w", err)
+	}
+
+	if schemaID != 0 {
+		schemaIDPrefix := binary.BigEndian.AppendUint32([]byte{0x0}, schemaID)
+		logMsgBinary = append(schemaIDPrefix, logMsgBinary...)
+	}
+
+	return logMsgBinary, nil
+}
+
+func (d *avroStaticSchemaSerDe) Deserialize(data []byte) (map[string]any, error) {
 	native, _, err := d.codec.NativeFromBinary(data)
 	if err != nil {
 		return nil, fmt.Errorf("failed to deserialize avro record: %w", err)

extension/encoding/avrologencodingextension/avro_test.go

@@ -4,16 +4,87 @@
 package avrologencodingextension
 
 import (
+	"encoding/binary"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 )
 
-func TestNewAvroLogsUnmarshaler(t *testing.T) {
+func TestNewAvroLogsMarshaler(t *testing.T) {
+	schema, jsonMap := createMapTestData(t)
+
+	avroEncoder, err := newAVROStaticSchemaSerDe(schema)
+	if err != nil {
+		t.Errorf("Did not expect an error, got %q", err.Error())
+	}
+
+	avroData, err := avroEncoder.Serialize(jsonMap, 0)
+	if err != nil {
+		t.Fatalf("Did not expect an error, got %q", err.Error())
+	}
+
+	logMap, err := avroEncoder.Deserialize(avroData)
+	if err != nil {
+		t.Fatalf("Did not expect an error, got %q", err.Error())
+	}
+
+	assert.Equal(t, int64(1697187201488000000), logMap["timestamp"].(time.Time).UnixNano())
+	assert.Equal(t, "host1", logMap["hostname"])
+	assert.Equal(t, int64(12), logMap["nestedRecord"].(map[string]any)["field1"])
+
+	props := logMap["properties"].([]any)
+	propsStr := make([]string, len(props))
+	for i, prop := range props {
+		propsStr[i] = prop.(string)
+	}
+
+	assert.Equal(t, []string{"prop1", "prop2"}, propsStr)
+}
+
+func TestNewAvroLogsMarshalerInvalidData(t *testing.T) {
+	schema, _ := createMapTestData(t)
+
+	avroEncoder, err := newAVROStaticSchemaSerDe(schema)
+	if err != nil {
+		t.Errorf("Did not expect an error, got %q", err.Error())
+	}
+
+	_, err = avroEncoder.Serialize(map[string]any{"invalid": "data"}, 0)
+	assert.Error(t, err)
+}
+
+func TestSchemaID(t *testing.T) {
+	schema, jsonMap := createMapTestData(t)
+
+	avroEncoder, err := newAVROStaticSchemaSerDe(schema)
+	if err != nil {
+		t.Errorf("Did not expect an error, got %q", err.Error())
+	}
+
+	schemaID := uint32(4294967295) // This number is 32 bits of all 1s
+
+	avroData, err := avroEncoder.Serialize(jsonMap, schemaID)
+	if err != nil {
+		t.Fatalf("Did not expect an error, got %q", err.Error())
+	}
+
+	binarySchemaID := []byte{0, 0, 0, 0}
+	binary.BigEndian.PutUint32(binarySchemaID, schemaID)
+
+	assert.Equal(t, avroData[0], uint8(0x0))
+	assert.Equal(t, avroData[1:5], binarySchemaID)
+
+	_, err = avroEncoder.Deserialize(avroData[5:])
+	if err != nil {
+		t.Fatalf("Did not expect an error, got %q", err.Error())
+	}
+}
+
+func TestNewAvroLogsSerDeDeserialize(t *testing.T) {
 	schema, data := createAVROTestData(t)
 
-	deserializer, err := newAVROStaticSchemaDeserializer(schema)
+	deserializer, err := newAVROStaticSchemaSerDe(schema)
 	if err != nil {
 		t.Errorf("Did not expect an error, got %q", err.Error())
 	}
@@ -36,7 +107,7 @@ func TestNewAvroLogsUnmarshaler(t *testing.T) {
 	assert.Equal(t, []string{"prop1", "prop2"}, propsStr)
 }
 
-func TestNewAvroLogsUnmarshalerInvalidSchema(t *testing.T) {
-	_, err := newAVROStaticSchemaDeserializer("invalid schema")
+func TestNewAvroLogsSerDeInvalidSchema(t *testing.T) {
+	_, err := newAVROStaticSchemaSerDe("invalid schema")
 	assert.Error(t, err)
 }

extension/encoding/avrologencodingextension/config.go

@@ -8,7 +8,8 @@ import "errors"
 var errNoSchema = errors.New("no schema provided")
 
 type Config struct {
-	Schema string `mapstructure:"schema"`
+	Schema   string `mapstructure:"schema"`
+	SchemaID uint32 `mapstructure:"schema_id"`
 }
 
 func (c *Config) Validate() error {

extension/encoding/avrologencodingextension/extension.go

@@ -16,26 +16,51 @@ import (
 )
 
 var (
+	_ encoding.LogsMarshalerExtension   = (*avroLogExtension)(nil)
 	_ encoding.LogsUnmarshalerExtension = (*avroLogExtension)(nil)
 )
 
 type avroLogExtension struct {
-	deserializer avroDeserializer
+	config  *Config
+	encoder avroSerDe
 }
 
 func newExtension(config *Config) (*avroLogExtension, error) {
-	deserializer, err := newAVROStaticSchemaDeserializer(config.Schema)
+	encoder, err := newAVROStaticSchemaSerDe(config.Schema)
 	if err != nil {
 		return nil, err
 	}
 
-	return &avroLogExtension{deserializer: deserializer}, nil
+	return &avroLogExtension{
+		config:  config,
+		encoder: encoder,
+	}, nil
+}
+
+func (e *avroLogExtension) MarshalLogs(logs plog.Logs) ([]byte, error) {
+	body := logs.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Body()
+
+	var raw map[string]any
+
+	switch body.Type() {
+	case pcommon.ValueTypeMap:
+		raw = body.Map().AsRaw()
+	default:
+		return nil, fmt.Errorf("marshal: body expected to be of type 'Map' found '%v'", body.Type().String())
+	}
+
+	buf, err := e.encoder.Serialize(raw, e.config.SchemaID)
+	if err != nil {
+		return nil, err
+	}
+
+	return buf, nil
 }
 
 func (e *avroLogExtension) UnmarshalLogs(buf []byte) (plog.Logs, error) {
 	p := plog.NewLogs()
 
-	avroLog, err := e.deserializer.Deserialize(buf)
+	avroLog, err := e.encoder.Deserialize(buf)
 	if err != nil {
 		return p, fmt.Errorf("failed to deserialize avro log: %w", err)
 	}

extension/encoding/avrologencodingextension/extension_test.go

@@ -10,8 +10,11 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/pdata/plog"
 )
 
+const testJSONBody = "{\"count\":5,\"hostname\":\"host1\",\"level\":\"warn\",\"levelEnum\":\"INFO\",\"mapField\":{},\"message\":\"log message\",\"nestedRecord\":{\"field1\":12,\"field2\":\"val2\"},\"properties\":[\"prop1\",\"prop2\"],\"severity\":1,\"timestamp\":1697187201488000000}"
+
 func TestExtension_Start_Shutdown(t *testing.T) {
 	avroExtention := &avroLogExtension{}
 
@@ -22,6 +25,40 @@ func TestExtension_Start_Shutdown(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestMarshal(t *testing.T) {
+	t.Parallel()
+
+	schema, jsonMap := createMapTestData(t)
+
+	e, err := newExtension(&Config{Schema: schema})
+	assert.NoError(t, err)
+
+	ld := plog.NewLogs()
+	err = ld.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords().AppendEmpty().Body().FromRaw(jsonMap)
+	assert.NoError(t, err)
+
+	_, err = e.MarshalLogs(ld)
+	assert.NoError(t, err)
+}
+
+func TestInvalidMarshal(t *testing.T) {
+	t.Parallel()
+
+	schema, err := loadAVROSchemaFromFile()
+	if err != nil {
+		t.Fatalf("Failed to read avro schema file: %q", err.Error())
+	}
+
+	e, err := newExtension(&Config{Schema: string(schema)})
+	assert.NoError(t, err)
+
+	ld := plog.NewLogs()
+	ld.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords().AppendEmpty().Body().SetStr("INVALID")
+
+	_, err = e.MarshalLogs(ld)
+	assert.Error(t, err)
+}
+
 func TestUnmarshal(t *testing.T) {
 	t.Parallel()
 
@@ -34,13 +71,13 @@ func TestUnmarshal(t *testing.T) {
 	logRecord := logs.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0)
 
 	assert.NoError(t, err)
-	assert.Equal(t, "{\"count\":5,\"hostname\":\"host1\",\"level\":\"warn\",\"levelEnum\":\"INFO\",\"mapField\":{},\"message\":\"log message\",\"nestedRecord\":{\"field1\":12,\"field2\":\"val2\"},\"properties\":[\"prop1\",\"prop2\"],\"severity\":1,\"timestamp\":1697187201488000000}", logRecord.Body().AsString())
+	assert.Equal(t, testJSONBody, logRecord.Body().AsString())
 }
 
 func TestInvalidUnmarshal(t *testing.T) {
 	t.Parallel()
 
-	schema, err := loadAVROSchemaFromFile("testdata/schema1.avro")
+	schema, err := loadAVROSchemaFromFile()
 	if err != nil {
 		t.Fatalf("Failed to read avro schema file: %q", err.Error())
 	}