Updated ocr3 Metadata type to include Encoding and Decoding and updated common beholder code

pkg/beholder/attributes.go

@@ -0,0 +1,8 @@
+package beholder
+
+const (
+	AttrKeyDataSchema = "beholder_data_schema"
+	AttrKeyEntity     = "beholder_entity"
+	AttrKeyDomain     = "beholder_domain"
+	AttrKeyDataType   = "beholder_data_type"
+)

pkg/beholder/chip_ingress_emitter.go

@@ -51,12 +51,12 @@ func ExtractSourceAndType(attrKVs ...any) (string, string, error) {
 	for key, value := range attributes {
 
 		// Retrieve source and type using either ChIP or legacy attribute names, prioritizing source/type
-		if key == "source" || (key == "beholder_domain" && sourceDomain == "") {
+		if key == "source" || (key == AttrKeyDomain && sourceDomain == "") {
 			if val, ok := value.(string); ok {
 				sourceDomain = val
 			}
 		}
-		if key == "type" || (key == "beholder_entity" && entityType == "") {
+		if key == "type" || (key == AttrKeyEntity && entityType == "") {
 			if val, ok := value.(string); ok {
 				entityType = val
 			}

pkg/beholder/chip_ingress_emitter_test.go

@@ -42,7 +42,7 @@ func TestChipIngressEmit(t *testing.T) {
 		emitter, err := beholder.NewChipIngressEmitter(clientMock)
 		require.NoError(t, err)
 
-		err = emitter.Emit(t.Context(), body, "beholder_domain", domain, "beholder_entity", entity)
+		err = emitter.Emit(t.Context(), body, beholder.AttrKeyDomain, domain, beholder.AttrKeyEntity, entity)
 		require.NoError(t, err)
 
 		clientMock.AssertExpectations(t)
@@ -59,7 +59,7 @@ func TestChipIngressEmit(t *testing.T) {
 		emitter, err := beholder.NewChipIngressEmitter(clientMock)
 		require.NoError(t, err)
 
-		err = emitter.Emit(t.Context(), body, "beholder_domain", domain)
+		err = emitter.Emit(t.Context(), body, beholder.AttrKeyDomain, domain)
 		assert.Error(t, err)
 	})
 
@@ -74,7 +74,7 @@ func TestChipIngressEmit(t *testing.T) {
 		emitter, err := beholder.NewChipIngressEmitter(clientMock)
 		require.NoError(t, err)
 
-		err = emitter.Emit(t.Context(), body, "beholder_domain", domain, "beholder_entity", entity)
+		err = emitter.Emit(t.Context(), body, beholder.AttrKeyDomain, domain, beholder.AttrKeyEntity, entity)
 		require.Error(t, err)
 
 		clientMock.AssertExpectations(t)
@@ -92,7 +92,7 @@ func TestExtractSourceAndType(t *testing.T) {
 	}{
 		{
 			name:       "happy path - domain and entity exist",
-			attrs:      []any{map[string]any{"beholder_domain": "test-domain", "beholder_entity": "test-entity"}},
+			attrs:      []any{map[string]any{beholder.AttrKeyDomain: "test-domain", beholder.AttrKeyEntity: "test-entity"}},
 			wantDomain: "test-domain",
 			wantEntity: "test-entity",
 			wantErr:    false,
@@ -106,22 +106,22 @@ func TestExtractSourceAndType(t *testing.T) {
 		},
 		{
 			name:       "happy path - domain and entity exist - uses source/type",
-			attrs:      []any{map[string]any{"source": "other-domain", "beholder_domain": "test-domain", "beholder_entity": "test-entity", "type": "other-entity"}},
+			attrs:      []any{map[string]any{"source": "other-domain", beholder.AttrKeyDomain: "test-domain", beholder.AttrKeyEntity: "test-entity", "type": "other-entity"}},
 			wantDomain: "other-domain",
 			wantEntity: "other-entity",
 			wantErr:    false,
 		},
 		{
 			name:          "missing domain/source",
-			attrs:         []any{map[string]any{"beholder_entity": "test-entity"}},
+			attrs:         []any{map[string]any{beholder.AttrKeyEntity: "test-entity"}},
 			wantDomain:    "",
 			wantEntity:    "",
 			wantErr:       true,
 			expectedError: "source/beholder_domain not found in provided key/value attributes",
 		},
 		{
 			name:          "missing entity/type",
-			attrs:         []any{map[string]any{"beholder_domain": "test-domain"}},
+			attrs:         []any{map[string]any{beholder.AttrKeyDomain: "test-domain"}},
 			wantDomain:    "",
 			wantEntity:    "",
 			wantErr:       true,
@@ -146,10 +146,10 @@ func TestExtractSourceAndType(t *testing.T) {
 		{
 			name: "domain and entity with additional attributes",
 			attrs: []any{map[string]any{
-				"other_key":       "other_value",
-				"beholder_domain": "test-domain",
-				"beholder_entity": "test-entity",
-				"something_else":  123,
+				"other_key":            "other_value",
+				beholder.AttrKeyDomain: "test-domain",
+				beholder.AttrKeyEntity: "test-entity",
+				"something_else":       123,
 			}},
 			wantDomain: "test-domain",
 			wantEntity: "test-entity",
@@ -158,9 +158,9 @@ func TestExtractSourceAndType(t *testing.T) {
 		{
 			name: "non-string keys ignored",
 			attrs: []any{map[string]any{
-				"other_value":     "value",
-				"beholder_domain": "test-domain",
-				"beholder_entity": "test-entity",
+				"other_value":          "value",
+				beholder.AttrKeyDomain: "test-domain",
+				beholder.AttrKeyEntity: "test-entity",
 			}, 123, "other_key"},
 			wantDomain: "test-domain",
 			wantEntity: "test-entity",
@@ -169,9 +169,9 @@ func TestExtractSourceAndType(t *testing.T) {
 		{
 			name: "non-string values handled",
 			attrs: []any{map[string]any{
-				"other_key":       123,
-				"beholder_domain": "test-domain",
-				"beholder_entity": "test-entity",
+				"other_key":            123,
+				beholder.AttrKeyDomain: "test-domain",
+				beholder.AttrKeyEntity: "test-entity",
 			}},
 			wantDomain: "test-domain",
 			wantEntity: "test-entity",

pkg/beholder/client.go

@@ -135,7 +135,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 	}
 
 	loggerAttributes := []attribute.KeyValue{
-		attribute.String("beholder_data_type", "zap_log_message"),
+		attribute.String(AttrKeyDataType, "zap_log_message"),
 	}
 	loggerResource, err := sdkresource.Merge(
 		sdkresource.NewSchemaless(loggerAttributes...),
@@ -195,7 +195,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 	}
 
 	messageAttributes := []attribute.KeyValue{
-		attribute.String("beholder_data_type", "custom_message"),
+		attribute.String(AttrKeyDataType, "custom_message"),
 	}
 	messageLoggerResource, err := sdkresource.Merge(
 		sdkresource.NewSchemaless(messageAttributes...),

pkg/beholder/client_test.go

@@ -43,18 +43,18 @@ func (m *MockExporter) ForceFlush(ctx context.Context) error {
 func TestClient(t *testing.T) {
 	defaultCustomAttributes := func() map[string]any {
 		return map[string]any{
-			"int_key_1":            123,
-			"int64_key_1":          int64(123),
-			"int32_key_1":          int32(123),
-			"str_key_1":            "str_val_1",
-			"bool_key_1":           true,
-			"float_key_1":          123.456,
-			"byte_key_1":           []byte("byte_val_1"),
-			"str_slice_key_1":      []string{"str_val_1", "str_val_2"},
-			"nil_key_1":            nil,
-			"beholder_domain":      "TestDomain",        // Required field
-			"beholder_entity":      "TestEntity",        // Required field
-			"beholder_data_schema": "/schemas/ids/1001", // Required field, URI
+			"int_key_1":                123,
+			"int64_key_1":              int64(123),
+			"int32_key_1":              int32(123),
+			"str_key_1":                "str_val_1",
+			"bool_key_1":               true,
+			"float_key_1":              123.456,
+			"byte_key_1":               []byte("byte_val_1"),
+			"str_slice_key_1":          []string{"str_val_1", "str_val_2"},
+			"nil_key_1":                nil,
+			beholder.AttrKeyDomain:     "TestDomain",        // Required field
+			beholder.AttrKeyEntity:     "TestEntity",        // Required field
+			beholder.AttrKeyDataSchema: "/schemas/ids/1001", // Required field, URI
 		}
 	}
 	defaultMessageBody := []byte("body bytes")

pkg/beholder/example_test.go

@@ -44,10 +44,10 @@ func ExampleNewClient() {
 	fmt.Println("Emit custom messages")
 	for range 10 {
 		err := beholder.GetEmitter().Emit(context.Background(), payloadBytes,
-			"beholder_data_schema", "/custom-message/versions/1", // required
-			"beholder_domain", "ExampleDomain", // required
-			"beholder_entity", "ExampleEntity", // required
-			"beholder_data_type", "custom_message",
+			beholder.AttrKeyDataSchema, "/custom-message/versions/1", // required
+			beholder.AttrKeyDomain, "ExampleDomain", // required
+			beholder.AttrKeyEntity, "ExampleEntity", // required
+			beholder.AttrKeyDataType, "custom_message",
 			"foo", "bar",
 		)
 		if err != nil {
@@ -106,9 +106,9 @@ func ExampleNewNoopClient() {
 	fmt.Println("Emitting custom message via noop otel client")
 
 	err := beholder.GetEmitter().Emit(context.Background(), []byte("test message"),
-		"beholder_data_schema", "/custom-message/versions/1", // required
-		"beholder_domain", "ExampleDomain", // required
-		"beholder_entity", "ExampleEntity", // required
+		beholder.AttrKeyDataSchema, "/custom-message/versions/1", // required
+		beholder.AttrKeyDomain, "ExampleDomain", // required
+		beholder.AttrKeyEntity, "ExampleEntity", // required
 	)
 	if err != nil {
 		log.Printf("Error emitting message: %v", err)

pkg/beholder/httpclient.go

@@ -100,7 +100,7 @@ func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro
 		loggerProcessor = sdklog.NewSimpleProcessor(sharedLogExporter)
 	}
 	loggerAttributes := []attribute.KeyValue{
-		attribute.String("beholder_data_type", "zap_log_message"),
+		attribute.String(AttrKeyDataType, "zap_log_message"),
 	}
 	loggerResource, err := sdkresource.Merge(
 		sdkresource.NewSchemaless(loggerAttributes...),
@@ -160,7 +160,7 @@ func NewHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro
 	}
 
 	messageAttributes := []attribute.KeyValue{
-		attribute.String("beholder_data_type", "custom_message"),
+		attribute.String(AttrKeyDataType, "custom_message"),
 	}
 	messageLoggerResource, err := sdkresource.Merge(
 		sdkresource.NewSchemaless(messageAttributes...),

pkg/beholder/message.go

@@ -60,9 +60,9 @@ func (m Metadata) Attributes() Attributes {
 		"workflow_owner_address":      m.WorkflowOwnerAddress,
 		"workflow_spec_id":            m.WorkflowSpecID,
 		"workflow_execution_id":       m.WorkflowExecutionID,
-		"beholder_domain":             m.BeholderDomain,
-		"beholder_entity":             m.BeholderEntity,
-		"beholder_data_schema":        m.BeholderDataSchema,
+		AttrKeyDomain:                 m.BeholderDomain,
+		AttrKeyEntity:                 m.BeholderEntity,
+		AttrKeyDataSchema:             m.BeholderDataSchema,
 		"capability_contract_address": m.CapabilityContractAddress,
 		"capability_id":               m.CapabilityID,
 		"capability_version":          m.CapabilityVersion,
@@ -206,11 +206,11 @@ func (m *Metadata) FromAttributes(attrs Attributes) *Metadata {
 			m.WorkflowSpecID = v.(string)
 		case "workflow_execution_id":
 			m.WorkflowExecutionID = v.(string)
-		case "beholder_domain":
+		case AttrKeyDomain:
 			m.BeholderDomain = v.(string)
-		case "beholder_entity":
+		case AttrKeyEntity:
 			m.BeholderEntity = v.(string)
-		case "beholder_data_schema":
+		case AttrKeyDataSchema:
 			m.BeholderDataSchema = v.(string)
 		case "capability_contract_address":
 			m.CapabilityContractAddress = v.(string)

pkg/beholder/message_emitter_test.go

@@ -44,49 +44,49 @@ func TestEmitterMessageValidation(t *testing.T) {
 		{
 			name: "Invalid URI not causing error",
 			attrs: beholder.Attributes{
-				"beholder_domain":      "TestDomain",
-				"beholder_entity":      "TestEntity",
-				"beholder_data_schema": "example-schema",
+				beholder.AttrKeyDomain:     "TestDomain",
+				beholder.AttrKeyEntity:     "TestEntity",
+				beholder.AttrKeyDataSchema: "example-schema",
 			},
 			exporterCalledTimes: 1,
 			expectedError:       "",
 		},
 		{
 			name: "Invalid Beholder domain (double underscore)",
 			attrs: beholder.Attributes{
-				"beholder_data_schema": "/example-schema/versions/1",
-				"beholder_entity":      "TestEntity",
-				"beholder_domain":      "Test__Domain",
+				beholder.AttrKeyDataSchema: "/example-schema/versions/1",
+				beholder.AttrKeyEntity:     "TestEntity",
+				beholder.AttrKeyDomain:     "Test__Domain",
 			},
 			exporterCalledTimes: 0,
 			expectedError:       "'Metadata.BeholderDomain' Error:Field validation for 'BeholderDomain' failed on the 'domain_entity' tag",
 		},
 		{
 			name: "Invalid Beholder domain (special characters)",
 			attrs: beholder.Attributes{
-				"beholder_data_schema": "/example-schema/versions/1",
-				"beholder_entity":      "TestEntity",
-				"beholder_domain":      "TestDomain*$",
+				beholder.AttrKeyDataSchema: "/example-schema/versions/1",
+				beholder.AttrKeyEntity:     "TestEntity",
+				beholder.AttrKeyDomain:     "TestDomain*$",
 			},
 			exporterCalledTimes: 0,
 			expectedError:       "'Metadata.BeholderDomain' Error:Field validation for 'BeholderDomain' failed on the 'domain_entity' tag",
 		},
 		{
 			name: "Invalid Beholder entity (double underscore)",
 			attrs: beholder.Attributes{
-				"beholder_data_schema": "/example-schema/versions/1",
-				"beholder_entity":      "Test__Entity",
-				"beholder_domain":      "TestDomain",
+				beholder.AttrKeyDataSchema: "/example-schema/versions/1",
+				beholder.AttrKeyEntity:     "Test__Entity",
+				beholder.AttrKeyDomain:     "TestDomain",
 			},
 			exporterCalledTimes: 0,
 			expectedError:       "'Metadata.BeholderEntity' Error:Field validation for 'BeholderEntity' failed on the 'domain_entity' tag",
 		},
 		{
 			name: "Invalid Beholder entity (special characters)",
 			attrs: beholder.Attributes{
-				"beholder_data_schema": "/example-schema/versions/1",
-				"beholder_entity":      "TestEntity*$",
-				"beholder_domain":      "TestDomain",
+				beholder.AttrKeyDataSchema: "/example-schema/versions/1",
+				beholder.AttrKeyEntity:     "TestEntity*$",
+				beholder.AttrKeyDomain:     "TestDomain",
 			},
 			exporterCalledTimes: 0,
 			expectedError:       "'Metadata.BeholderEntity' Error:Field validation for 'BeholderEntity' failed on the 'domain_entity' tag",
@@ -95,19 +95,19 @@ func TestEmitterMessageValidation(t *testing.T) {
 			name:                "Valid Attributes",
 			exporterCalledTimes: 1,
 			attrs: beholder.Attributes{
-				"beholder_domain":      "TestDomain",
-				"beholder_entity":      "TestEntity",
-				"beholder_data_schema": "/example-schema/versions/1",
+				beholder.AttrKeyDomain:     "TestDomain",
+				beholder.AttrKeyEntity:     "TestEntity",
+				beholder.AttrKeyDataSchema: "/example-schema/versions/1",
 			},
 			expectedError: "",
 		},
 		{
 			name:                "Valid Attributes (special characters)",
 			exporterCalledTimes: 1,
 			attrs: beholder.Attributes{
-				"beholder_domain":      "Test.Domain_42-1",
-				"beholder_entity":      "Test.Entity_42-1",
-				"beholder_data_schema": "/example-schema/versions/1",
+				beholder.AttrKeyDomain:     "Test.Domain_42-1",
+				beholder.AttrKeyEntity:     "Test.Entity_42-1",
+				beholder.AttrKeyDataSchema: "/example-schema/versions/1",
 			},
 			expectedError: "",
 		},