[cappl-86] feat(workflows/wasm): emit msgs to beholder

pkg/capabilities/events/events.go

@@ -14,14 +14,14 @@ import (
 
 const (
 	// Duplicates the attributes in beholder/message.go::Metadata
-	labelWorkflowOwner             = "workflow_owner_address"
-	labelWorkflowID                = "workflow_id"
-	labelWorkflowExecutionID       = "workflow_execution_id"
-	labelWorkflowName              = "workflow_name"
-	labelCapabilityContractAddress = "capability_contract_address"
-	labelCapabilityID              = "capability_id"
-	labelCapabilityVersion         = "capability_version"
-	labelCapabilityName            = "capability_name"
+	LabelWorkflowOwner             = "workflow_owner_address"
+	LabelWorkflowID                = "workflow_id"
+	LabelWorkflowExecutionID       = "workflow_execution_id"
+	LabelWorkflowName              = "workflow_name"
+	LabelCapabilityContractAddress = "capability_contract_address"
+	LabelCapabilityID              = "capability_id"
+	LabelCapabilityVersion         = "capability_version"
+	LabelCapabilityName            = "capability_name"
 )
 
 type EmitMetadata struct {
@@ -93,35 +93,35 @@ func (e EmitMetadata) attrs() []any {
 	a := []any{}
 
 	if e.WorkflowOwner != "" {
-		a = append(a, labelWorkflowOwner, e.WorkflowOwner)
+		a = append(a, LabelWorkflowOwner, e.WorkflowOwner)
 	}
 
 	if e.WorkflowID != "" {
-		a = append(a, labelWorkflowID, e.WorkflowID)
+		a = append(a, LabelWorkflowID, e.WorkflowID)
 	}
 
 	if e.WorkflowExecutionID != "" {
-		a = append(a, labelWorkflowExecutionID, e.WorkflowExecutionID)
+		a = append(a, LabelWorkflowExecutionID, e.WorkflowExecutionID)
 	}
 
 	if e.WorkflowName != "" {
-		a = append(a, labelWorkflowName, e.WorkflowName)
+		a = append(a, LabelWorkflowName, e.WorkflowName)
 	}
 
 	if e.CapabilityContractAddress != "" {
-		a = append(a, labelCapabilityContractAddress, e.CapabilityContractAddress)
+		a = append(a, LabelCapabilityContractAddress, e.CapabilityContractAddress)
 	}
 
 	if e.CapabilityID != "" {
-		a = append(a, labelCapabilityID, e.CapabilityID)
+		a = append(a, LabelCapabilityID, e.CapabilityID)
 	}
 
 	if e.CapabilityVersion != "" {
-		a = append(a, labelCapabilityVersion, e.CapabilityVersion)
+		a = append(a, LabelCapabilityVersion, e.CapabilityVersion)
 	}
 
 	if e.CapabilityName != "" {
-		a = append(a, labelCapabilityName, e.CapabilityName)
+		a = append(a, LabelCapabilityName, e.CapabilityName)
 	}
 
 	return a

pkg/custmsg/custom_message.go

@@ -19,17 +19,35 @@ func NewLabeler() Labeler {
 	return Labeler{labels: make(map[string]string)}
 }
 
+// WithMapLabels adds multiple key-value pairs to the CustomMessageLabeler for transmission
+// With SendLogAsCustomMessage
+func (l Labeler) WithMapLabels(labels map[string]string) Labeler {
+	newCustomMessageLabeler := NewLabeler()
+
+	// Copy existing labels from the current agent
+	for k, v := range l.labels {
+		newCustomMessageLabeler.labels[k] = v
+	}
+
+	// Add new key-value pairs
+	for k, v := range labels {
+		newCustomMessageLabeler.labels[k] = v
+	}
+
+	return newCustomMessageLabeler
+}
+
 // With adds multiple key-value pairs to the CustomMessageLabeler for transmission With SendLogAsCustomMessage
-func (c Labeler) With(keyValues ...string) Labeler {
+func (l Labeler) With(keyValues ...string) Labeler {
 	newCustomMessageLabeler := NewLabeler()
 
 	if len(keyValues)%2 != 0 {
 		// If an odd number of key-value arguments is passed, return the original CustomMessageLabeler unchanged
-		return c
+		return l
 	}
 
 	// Copy existing labels from the current agent
-	for k, v := range c.labels {
+	for k, v := range l.labels {
 		newCustomMessageLabeler.labels[k] = v
 	}
 
@@ -43,10 +61,22 @@ func (c Labeler) With(keyValues ...string) Labeler {
 	return newCustomMessageLabeler
 }
 
+func (l Labeler) Emit(msg string) error {
+	return sendLogAsCustomMessageW(msg, l.labels)
+}
+
+func (l Labeler) Labels() map[string]string {
+	copied := make(map[string]string, len(l.labels))
+	for k, v := range l.labels {
+		copied[k] = v
+	}
+	return copied
+}
+
 // SendLogAsCustomMessage emits a BaseMessage With msg and labels as data.
 // any key in labels that is not part of orderedLabelKeys will not be transmitted
-func (c Labeler) SendLogAsCustomMessage(msg string) error {
-	return sendLogAsCustomMessageW(msg, c.labels)
+func (l Labeler) SendLogAsCustomMessage(msg string) error {
+	return sendLogAsCustomMessageW(msg, l.labels)
 }
 
 func sendLogAsCustomMessageW(msg string, labels map[string]string) error {

pkg/custmsg/custom_message_test.go

@@ -12,5 +12,19 @@ func Test_CustomMessageAgent(t *testing.T) {
 	cma1 := cma.With("key1", "value1")
 	cma2 := cma1.With("key2", "value2")
 
-	assert.NotEqual(t, cma1.labels, cma2.labels)
+	assert.NotEqual(t, cma1.Labels(), cma2.Labels())
+}
+
+func Test_CustomMessageAgent_With(t *testing.T) {
+	cma := NewLabeler()
+	cma = cma.With("key1", "value1")
+
+	assert.Equal(t, cma.Labels(), map[string]string{"key1": "value1"})
+}
+
+func Test_CustomMessageAgent_WithMapLabels(t *testing.T) {
+	cma := NewLabeler()
+	cma = cma.WithMapLabels(map[string]string{"key1": "value1"})
+
+	assert.Equal(t, cma.Labels(), map[string]string{"key1": "value1"})
 }

pkg/values/map.go

@@ -20,7 +20,7 @@ func EmptyMap() *Map {
 	}
 }
 
-func NewMap(m map[string]any) (*Map, error) {
+func NewMap[T any](m map[string]T) (*Map, error) {
 	mv := map[string]Value{}
 	for k, v := range m {
 		val, err := Wrap(v)

pkg/workflows/sdk/runtime.go

@@ -7,9 +7,22 @@ import (
 
 var BreakErr = capabilities.ErrStopExecution
 
+type MessageEmitter interface {
+	// Emit sends a message to the labeler's destination.
+	Emit(string) error
+
+	// With sets the labels for the message to be emitted.  Labels are passed as key-value pairs
+	// and are cumulative.
+	With(kvs ...string) MessageEmitter
+}
+
+// Guest interface
 type Runtime interface {
 	Logger() logger.Logger
 	Fetch(req FetchRequest) (FetchResponse, error)
+
+	// Emitter sends the given message and labels to the configured collector.
+	Emitter() MessageEmitter
 }
 
 type FetchRequest struct {

pkg/workflows/sdk/testutils/runtime.go

@@ -17,3 +17,7 @@ func (nr *NoopRuntime) Logger() logger.Logger {
 	l, _ := logger.New()
 	return l
 }
+
+func (nr *NoopRuntime) Emitter() sdk.MessageEmitter {
+	return nil
+}