Merge branch 'nslaughter/azureeventhubreceiver-20240611' into nslaughter/migrate-eventhub-sdk

Author: nslaughter

.github/workflows/build.yml

@@ -22,7 +22,7 @@ jobs:
       - name: Install builder
         working-directory: ./collector
         run: |
-            go install go.opentelemetry.io/collector/cmd/builder@v0.97.0     
+            go install go.opentelemetry.io/collector/cmd/builder@v0.102.1     
       - name: Build
         working-directory: ./collector
         run: make build-linux

collector/Makefile

@@ -38,7 +38,7 @@ install-tools: install-builder
 
 .PHONY: install-builder - Install the builder
 install-builder:
-	go install "go.opentelemetry.io/collector/cmd/builder@v0.98.0"
+	go install "go.opentelemetry.io/collector/cmd/builder@v0.102.1"
 
 .PHONY: docker - Build docker image
 docker:

collector/components/azureeventhubreceiver/azureeventprocessor.go

@@ -0,0 +1,190 @@
+package azureeventhubreceiver
+
+// https://github.com/Azure/azure-sdk-for-go/blob/main/sdk/messaging/azeventhubs/processor.go
+// https://github.com/Azure/azure-sdk-for-go/blob/main/sdk/messaging/azeventhubs/processor_partition_client.go
+
+/*
+>> https://github.com/Azure/azure-sdk-for-go/blob/main/sdk/messaging/azeventhubs/example_consuming_with_checkpoints_test.go
+	- get a processor
+	- dispatchPartitionClients
+	- processor.Run
+
+
+
+>> https://github.com/Azure/azure-sdk-for-go/blob/main/sdk/messaging/azeventhubs/example_consuming_events_test.go
+ 	- ReceiveEvents(ctx, count int, options *ReceiveEventsOptions) ([]*ReceivedEventData, error)
+	- call cancel()
+	- panic if there's an error that isn't context.DeadlineExceeded
+	- process events
+		--> put them into the entity thingy
+*/
+
+// import (
+// 	"context"
+// 	"errors"
+// 	"fmt"
+// 	"time"
+
+// 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+// 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/checkpoints"
+// 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
+// )
+
+// // Assuming there's a struct managing the processor setup
+// // type EventHubProcessor struct {
+// //     Processor *azeventhubs.Processor
+// // }
+
+// // Updated initialization function using the new SDK components
+// func NewEventHubProcessor(ehConn, ehName, storageConn, storageCnt string) (*EventHubProcessor, error) {
+// 	checkpointingProcessor, err := newCheckpointingProcessor(ehConn, ehName, storageConn, storageCnt)
+// 	if err != nil {
+// 		return nil, fmt.Errorf("failed to create checkpointing processor: %w", err)
+// 	}
+
+// 	// Start processing events
+// 	return &EventHubProcessor{
+// 		Processor: checkpointingProcessor,
+// 	}, nil
+// }
+
+// // Assume there's a function to start processing events
+// func (e *EventHubProcessor) StartProcessing(ctx context.Context) error {
+// 	// Start the processor
+// 	if err := e.Processor.Run(ctx); err != nil {
+// 		return fmt.Errorf("error running processor: %w", err)
+// 	}
+// 	return nil
+// }
+
+// // Assuming there's a struct managing the processor setup
+// type EventHubProcessor struct {
+// 	Processor *azeventhubs.Processor
+// }
+
+// // These are config values the processor factory can use to create processors:
+// //
+// //	(a) EventHubConnectionString
+// //	(b) EventHubName
+// //	(c) StorageConnectionString
+// //	(d) StorageContainerName
+// //
+// // You always need the EventHub variable values.
+// // And you need all 4 of these to checkpoint.
+// //
+// // I think the config values should be managed in the factory struct.
+// /*
+// func (pf *processorFactory) CreateProcessor() (*azeventhubs.Processor, error) {
+// 	// Create the consumer client
+// 	consumerClient, err := azeventhubs.NewConsumerClientFromConnectionString(pf.EventHubConnectionString, pf.EventHubName, azeventhubs.DefaultConsumerGroup, nil)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	// Create the blob container client for the checkpoint store
+// 	blobContainerClient, err := container.NewClientFromConnectionString(pf.StorageConnectionString, pf.StorageContainerName, nil)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	// Create the checkpoint store using the blob container client
+// 	checkpointStore, err := azeventhubs.NewBlobCheckpointStore(blobContainerClient, nil)
+// 	// checkpointStore, err := azeventhubs.NewBlobCheckpointStore(blobContainerClient, nil)
+// 	// if err != nil {
+// 	// 	return nil, err
+// 	// }
+
+// 	// Create the processor with checkpointing
+// 	processor, err := azeventhubs.NewProcessor(consumerClient, checkpointStore, nil)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	return processor, nil
+// }
+// */
+
+// // checkpointing processor should be auth aware
+
+// func newCheckpointingProcessor(eventHubConnectionString, eventHubName, storageConnectionString, storageContainerName string) (*azeventhubs.Processor, error) {
+// 	blobContainerClient, err := container.NewClientFromConnectionString(storageConnectionString, storageContainerName, nil)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+// 	checkpointStore, err := checkpoints.NewBlobStore(blobContainerClient, nil)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	consumerClient, err := azeventhubs.NewConsumerClientFromConnectionString(eventHubConnectionString, eventHubName, azeventhubs.DefaultConsumerGroup, nil)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	return azeventhubs.NewProcessor(consumerClient, checkpointStore, nil)
+// }
+/*
+func dispatchPartitionClients(processor *azeventhubs.Processor) {
+	for {
+		processorPartitionClient := processor.NextPartitionClient(context.TODO())
+		if processorPartitionClient == nil {
+			break
+		}
+
+		go func() {
+			if err := processEventsForPartition(processorPartitionClient); err != nil {
+				panic(err)
+			}
+		}()
+	}
+}
+
+func processEventsForPartition(partitionClient *azeventhubs.ProcessorPartitionClient) error {
+	defer shutdownPartitionResources(partitionClient)
+	if err := initializePartitionResources(partitionClient.PartitionID()); err != nil {
+		return err
+	}
+
+	for {
+		receiveCtx, cancelReceive := context.WithTimeout(context.TODO(), time.Minute)
+		events, err := partitionClient.ReceiveEvents(receiveCtx, 100, nil)
+		cancelReceive()
+
+		if err != nil && !errors.Is(err, context.DeadlineExceeded) {
+			return err
+		}
+		if len(events) == 0 {
+			continue
+		}
+
+		if err := processEvents(events, partitionClient); err != nil {
+			return err
+		}
+
+		if err := partitionClient.UpdateCheckpoint(context.TODO(), events[len(events)-1], nil); err != nil {
+			return err
+		}
+	}
+}
+
+func shutdownPartitionResources(partitionClient *azeventhubs.ProcessorPartitionClient) {
+	if err := partitionClient.Close(context.TODO()); err != nil {
+		panic(err)
+	}
+}
+
+func initializePartitionResources(partitionID string) error {
+	fmt.Printf("Initializing resources for partition %s\n", partitionID)
+	return nil
+}
+
+// This is very much like the old processEvents function
+func processEvents(events []*azeventhubs.ReceivedEventData, partitionClient *azeventhubs.ProcessorPartitionClient) error {
+	for _, event := range events {
+
+
+		// fmt.Printf("Processing event: %v\n", event.EventData())
+	}
+	return nil
+}
+*/

collector/components/azureeventhubreceiver/azureresourcelogs_unmarshaler.go

@@ -4,11 +4,11 @@
 package azureeventhubreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/azureeventhubreceiver"
 
 import (
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.uber.org/zap"
 
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/azure"
 )
 

collector/components/azureeventhubreceiver/eventhubhandler.go

@@ -12,15 +12,13 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/checkpoints"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/receiver"
 	"go.uber.org/zap"
 )
 
-const (
-	batchCount = 100
-)
-
 type eventHandler interface {
 	run(ctx context.Context, host component.Host) error
 	close(ctx context.Context) error
@@ -40,11 +38,7 @@ type consumerClientWrapperImpl struct {
 }
 
 func newConsumerClientWrapperImplementation(cfg *Config) (*consumerClientWrapperImpl, error) {
-	splits := strings.Split(cfg.Connection, "/")
-	eventhubName := splits[len(splits)-1]
-	// if that didn't work it's ok as the SDK will try to parse it to create the client
-
-	consumerClient, err := azeventhubs.NewConsumerClientFromConnectionString(cfg.Connection, eventhubName, cfg.ConsumerGroup, nil)
+	consumerClient, err := azeventhubs.NewConsumerClientFromConnectionString(cfg.Connection, cfg.EventHubName, cfg.ConsumerGroup, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -61,7 +55,7 @@ func (c *consumerClientWrapperImpl) GetPartitionProperties(ctx context.Context,
 	return c.consumerClient.GetPartitionProperties(ctx, partitionID, options)
 }
 
-func (c *consumerClientWrapperImpl) NewConsumer(_ context.Context, _ *azeventhubs.ConsumerClientOptions) (*azeventhubs.ConsumerClient, error) {
+func (c *consumerClientWrapperImpl) NewConsumer(ctx context.Context, options *azeventhubs.ConsumerClientOptions) (*azeventhubs.ConsumerClient, error) {
 	return c.consumerClient, nil
 }
 
@@ -74,6 +68,7 @@ func (c *consumerClientWrapperImpl) Close(ctx context.Context) error {
 }
 
 type eventhubHandler struct {
+	processor      *azeventhubs.Processor
 	consumerClient consumerClientWrapper
 	dataConsumer   dataConsumer
 	config         *Config
@@ -94,7 +89,7 @@ func newEventhubHandler(config *Config, settings receiver.CreateSettings) *event
 	return &eventhubHandler{
 		config:       config,
 		settings:     settings,
-		useProcessor: true,
+		useProcessor: false,
 	}
 }
 
@@ -109,33 +104,25 @@ func (h *eventhubHandler) init(ctx context.Context) error {
 }
 
 func (h *eventhubHandler) run(ctx context.Context, host component.Host) error {
-	ctx, h.cancel = context.WithCancel(ctx)
 	if h.useProcessor {
-		return h.runWithProcessor(ctx, host)
+		return h.runWithProcessor(ctx)
 	}
 	return h.runWithConsumerClient(ctx, host)
 }
-func (h *eventhubHandler) runWithProcessor(ctx context.Context, host component.Host) error {
-	checkpointStore, err := createCheckpointStore(ctx, host, h.config, h.settings)
+
+func (h *eventhubHandler) runWithProcessor(ctx context.Context) error {
+	checkpointStore, err := createCheckpointStore(h.config.StorageConnection, h.config.StorageContainer)
 	if err != nil {
-		h.settings.Logger.Debug("Error creating CheckpointStore", zap.Error(err))
 		return err
 	}
 
-	consumerClientImpl, ok := h.consumerClient.(*consumerClientWrapperImpl)
-	if !ok {
-		// we're in a testing environment
-		return nil
-	}
-
-	processor, err := azeventhubs.NewProcessor(consumerClientImpl.consumerClient, checkpointStore, nil)
+	processor, err := azeventhubs.NewProcessor(h.consumerClient.(*consumerClientWrapperImpl).consumerClient, checkpointStore, nil)
 	if err != nil {
-		h.settings.Logger.Debug("Error creating Processor", zap.Error(err))
 		return err
 	}
 
-	processorCtx, processorCancel := context.WithCancel(ctx)
 	go h.dispatchPartitionClients(processor)
+	processorCtx, processorCancel := context.WithCancel(ctx)
 	defer processorCancel()
 
 	return processor.Run(processorCtx)
@@ -166,7 +153,7 @@ func (h *eventhubHandler) processEventsForPartition(partitionClient *azeventhubs
 
 	for {
 		receiveCtx, cancelReceive := context.WithTimeout(context.TODO(), time.Minute)
-		events, err := partitionClient.ReceiveEvents(receiveCtx, 100, nil)
+		events, err := partitionClient.ReceiveEvents(receiveCtx, h.config.BatchCount, nil)
 		cancelReceive()
 
 		if err != nil && !errors.Is(err, context.DeadlineExceeded) {
@@ -193,7 +180,7 @@ func (h *eventhubHandler) processEventsForPartition(partitionClient *azeventhubs
 	}
 }
 
-func (h *eventhubHandler) runWithConsumerClient(ctx context.Context, _ component.Host) error {
+func (h *eventhubHandler) runWithConsumerClient(ctx context.Context, host component.Host) error {
 	if h.consumerClient == nil {
 		if err := h.init(ctx); err != nil {
 			return err
@@ -231,7 +218,11 @@ func (h *eventhubHandler) setupPartition(ctx context.Context, partitionID string
 	if cc == nil {
 		return errors.New("failed to initialize consumer client")
 	}
-	defer cc.Close(ctx)
+	defer func() {
+		if cc != nil {
+			cc.Close(ctx)
+		}
+	}()
 
 	pcOpts := &azeventhubs.PartitionClientOptions{
 		StartPosition: azeventhubs.StartPosition{
@@ -258,12 +249,15 @@ func (h *eventhubHandler) setupPartition(ctx context.Context, partitionID string
 }
 
 func (h *eventhubHandler) receivePartitionEvents(ctx context.Context, pc *azeventhubs.PartitionClient) {
+	var wait = 1
 	for {
-		rcvCtx, rcvCtxCancel := context.WithTimeout(context.TODO(), time.Second*10)
-		events, err := pc.ReceiveEvents(rcvCtx, batchCount, nil)
-		rcvCtxCancel()
-		if err != nil && !errors.Is(err, context.DeadlineExceeded) {
-			h.settings.Logger.Error("Error receiving events", zap.Error(err))
+		rcvCtx, _ := context.WithTimeout(context.TODO(), time.Second*10)
+		events, err := pc.ReceiveEvents(rcvCtx, h.config.BatchCount, nil)
+		if err != nil {
+			h.settings.Logger.Error("Error receiving event", zap.Error(err))
+			time.Sleep(time.Duration(wait) * time.Second)
+			wait *= 2
+			continue
 		}
 
 		for _, event := range events {
@@ -297,3 +291,11 @@ func (h *eventhubHandler) close(ctx context.Context) error {
 func (h *eventhubHandler) setDataConsumer(dataConsumer dataConsumer) {
 	h.dataConsumer = dataConsumer
 }
+
+func createCheckpointStore(storageConnectionString, containerName string) (azeventhubs.CheckpointStore, error) {
+	azBlobContainerClient, err := container.NewClientFromConnectionString(storageConnectionString, containerName, nil)
+	if err != nil {
+		return nil, err
+	}
+	return checkpoints.NewBlobStore(azBlobContainerClient, nil)
+}