Chip batch emit

pkg/beholder/chip_ingress_emitter.go

@@ -26,23 +26,31 @@ func (c *ChipIngressEmitter) Close() error {
 }
 
 func (c *ChipIngressEmitter) Emit(ctx context.Context, body []byte, attrKVs ...any) error {
+	return c.BatchEmit(ctx, NewMessage(body, attrKVs...))
+}
 
-	sourceDomain, entityType, err := ExtractSourceAndType(attrKVs...)
-	if err != nil {
-		return err
-	}
+func (c *ChipIngressEmitter) BatchEmit(ctx context.Context, messages ...Message) error {
+	events := make([]chipingress.CloudEvent, len(messages))
+	for i, msg := range messages {
+		sourceDomain, entityType, err := ExtractSourceAndType(msg.Attrs)
+		if err != nil {
+			return err
+		}
 
-	event, err := chipingress.NewEvent(sourceDomain, entityType, body, newAttributes(attrKVs...))
-	if err != nil {
-		return err
+		event, err := chipingress.NewEvent(sourceDomain, entityType, msg.Body, msg.Attrs)
+		if err != nil {
+			return err
+		}
+
+		events[i] = event
 	}
 
-	eventPb, err := chipingress.EventToProto(event)
+	eventPb, err := chipingress.EventsToBatch(events)
 	if err != nil {
 		return fmt.Errorf("failed to convert event to proto: %w", err)
 	}
 
-	_, err = c.client.Publish(ctx, eventPb)
+	_, err = c.client.PublishBatch(ctx, eventPb)
 	if err != nil {
 		return err
 	}
@@ -51,10 +59,7 @@ func (c *ChipIngressEmitter) Emit(ctx context.Context, body []byte, attrKVs ...a
 }
 
 // ExtractSourceAndType extracts source domain and entity from the attributes
-func ExtractSourceAndType(attrKVs ...any) (string, string, error) {
-
-	attributes := newAttributes(attrKVs...)
-
+func ExtractSourceAndType(attributes Attributes) (string, string, error) {
 	var sourceDomain string
 	var entityType string
 

pkg/beholder/chip_ingress_emitter_test.go

@@ -190,7 +190,7 @@ func TestExtractSourceAndType(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			domain, entity, err := beholder.ExtractSourceAndType(tt.attrs...)
+			domain, entity, err := beholder.ExtractSourceAndType(beholder.ExtractAttributes(tt.attrs))
 
 			if tt.wantErr {
 				if err == nil {

pkg/beholder/client.go

@@ -31,8 +31,9 @@ import (
 const defaultGRPCCompressor = "gzip"
 
 type Emitter interface {
-	// Sends message with bytes and attributes to OTel Collector
+	// Emit Sends message with bytes and attributes to OTel Collector
 	Emit(ctx context.Context, body []byte, attrKVs ...any) error
+	BatchEmit(ctx context.Context, messages ...Message) error
 	io.Closer
 }
 

pkg/beholder/dual_source_emitter.go

@@ -56,9 +56,12 @@ func (d *DualSourceEmitter) Close() error {
 }
 
 func (d *DualSourceEmitter) Emit(ctx context.Context, body []byte, attrKVs ...any) error {
+	return d.BatchEmit(ctx, NewMessage(body, attrKVs...))
+}
 
+func (d *DualSourceEmitter) BatchEmit(ctx context.Context, messages ...Message) error {
 	// Emit via OTLP first
-	if err := d.otelCollectorEmitter.Emit(ctx, body, attrKVs...); err != nil {
+	if err := d.otelCollectorEmitter.BatchEmit(ctx, messages...); err != nil {
 		return err
 	}
 
@@ -72,7 +75,7 @@ func (d *DualSourceEmitter) Emit(ctx context.Context, body []byte, attrKVs ...an
 		ctx, cancel = d.stopCh.Ctx(ctx)
 		defer cancel()
 
-		if err := d.chipIngressEmitter.Emit(ctx, body, attrKVs...); err != nil {
+		if err := d.chipIngressEmitter.BatchEmit(ctx, messages...); err != nil {
 			// If the chip ingress emitter fails, we ONLY log the error
 			// because we still want to send the data to the OTLP collector and not cause disruption
 			d.log.Infof("failed to emit to chip ingress: %v", err)

pkg/beholder/message_emitter.go

@@ -22,10 +22,15 @@ func (e messageEmitter) Close() error { return nil }
 // Emits logs the message, but does not wait for the message to be processed.
 // Open question: what are pros/cons for using use map[]any vs use otellog.KeyValue
 func (e messageEmitter) Emit(ctx context.Context, body []byte, attrKVs ...any) error {
-	message := NewMessage(body, attrKVs...)
-	if err := message.Validate(); err != nil {
-		return err
+	return e.BatchEmit(ctx, NewMessage(body, attrKVs...))
+}
+
+func (e messageEmitter) BatchEmit(ctx context.Context, messages ...Message) error {
+	for _, message := range messages {
+		if err := message.Validate(); err != nil {
+			return err
+		}
+		e.messageLogger.Emit(ctx, message.OtelRecord())
 	}
-	e.messageLogger.Emit(ctx, message.OtelRecord())
 	return nil
 }

pkg/beholder/noop.go

@@ -108,6 +108,11 @@ func (e noopMessageEmitter) Close() error { return nil }
 func (noopMessageEmitter) Emit(ctx context.Context, body []byte, attrKVs ...any) error {
 	return nil
 }
+
+func (noopMessageEmitter) BatchEmit(ctx context.Context, messages ...Message) error {
+	return nil
+}
+
 func (noopMessageEmitter) EmitMessage(ctx context.Context, message Message) error {
 	return nil
 }