rewrite EventHubs component #1292

mukundansundar · 2021-11-11T09:29:17Z

Description

Rewrite of eventhubs component.

Added support for partition keys.
Added support for honoring incoming topic names in requests.
Fixed bug on running multiple subscribers for a topic(leaser checkpointer prefix issue).
Added support for authentication via AAD.
Added support for creating eventHubs/consumer groups on the fly as long as a the event hubs namespace is created and with right permissions.

Adding more resiliency and retries if needed on connection will be part of the certification task and is left off for this PR.

Issue reference

We strive to have all PR being opened based on an issue, where the problem or feature have been discussed prior to implementation.

part of #951
closes #964
closes dapr/dapr#3753
closes #1225

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

Code compiles correctly
Created/updated tests
Extended the documentation / Created issue in the https://github.com/dapr/docs/ repo: Update docs for eventhubs component docs#1941

mukundansundar · 2021-11-12T05:51:44Z

authentication/azure/auth_amqp.go

+// 1. Client credentials
+// 2. Client certificate
+// 3. MSI.
+func (s EnvironmentSettings) GetAADTokenProvider() (*aad.TokenProvider, error) {


Same as the one in #1201

mukundansundar · 2021-11-12T05:52:40Z

pubsub/azure/eventhubs/eventhubs.go

+	return processor, nil
+}
+
+func (aeh *AzureEventHubs) createHubManager() error {


HubManager client from azure-event-hubs go sdk for creating/managing an event hub.

mukundansundar · 2021-11-12T05:53:32Z

pubsub/azure/eventhubs/eventhubs.go

+	return aeh.createConsumerGroup(hubName)
+}
+
+func (aeh *AzureEventHubs) getConsumerGroupsClient() (*mgmt.ConsumerGroupsClient, error) {


consumer group client from the event hubs management clients in azure-go-sdk for managing consumer groups.

mukundansundar · 2021-11-12T09:05:53Z

pubsub/azure/eventhubs/eventhubs.go

 	"time"

+	"github.com/Azure/azure-amqp-common-go/v3/aad"
+	"github.com/Azure/azure-amqp-common-go/v3/conn"
 	eventhub "github.com/Azure/azure-event-hubs-go/v3"
 	"github.com/Azure/azure-event-hubs-go/v3/eph"


as right now there does not seem to be any option for batching within event processor host.

(Linking dapr/dapr#2218 (comment) for context on this comment.)

Correct, EPH does not. To implement batch receive, the component would have to go down a level of abstraction to Hub.Receive with the ReceiveWithPrefetchCount option, which also means taking on explicit partition and checkpoint management that EPH otherwise currently handles. There's definitely a cost-benefit analysis to be done there, if/when pubsub batching is pursued as a feature in Dapr.

mukundansundar · 2021-11-12T09:22:34Z

pubsub/azure/eventhubs/eventhubs.go

 	if err != nil {
 		return err
 	}

-	leaserCheckpointer, err := storage.NewStorageLeaserCheckpointer(cred, aeh.metadata.storageAccountName, aeh.metadata.storageContainerName, azure.PublicCloud)
+	// Set topic name, consumerID prefix for partition checkpoint lease blob path.
+	leaserPrefixOpt := storage.WithPrefixInBlobPath(aeh.getStoragePrefixString(req.Topic))


This is needed to support multiple consumers with the same storage container option.

This should be a comment in the code as well for future maintainers.

mukundansundar · 2021-11-12T09:26:01Z

pubsub/azure/eventhubs/eventhubs.go

+		aeh.logger.Info("connecting to Azure EventHubs via AAD. connection established on first publish/subscribe")
+		aeh.logger.Debugf("req.Topic field in incoming requests honored")
+
+		if aeh.metadata.EnableEnitiyManagement {


A couple of more points here:
if entity Management is enabled, it does need permissions at least at the EventHub namespace level to be able to create event hubs and consumer groups. (Manage permission)
Also the other metadata, SubscriptionID and ResourceGroupName is a mandatory requirement when enabling entityManagement as for now, it assumes that the topic will be subscribed to.

If needed resourceGroupName and SubscriptionID can be validated only as part of subscription metadata validation since it will only be required for a subscriber and not for a publisher alone.

wrt comment #1055 (comment) from @CodeMonkeyLeet , entity management for EventHubs through Dapr does step into provisioning infrastructure resources and also does not go into fully managing the lifecycle of the resource. (should not delete)

CodeMonkeyLeet

I'm going to timebox my review on this for now because I've spent a couple of hours looking at this and I'm not done yet. I think this needs a design review first as I'm spending a chunk of time trying to infer if particular behaviors are intended.

In particular, there are a lot of contingent and interacting configuration options as indicated in this comment and I think it would be a good starting point to enumerate the expected valid configuration combinations and the resulting behaviors.

In general, it seems like we could scope down the scenarios we support, for example, supporting connection string auth to EventHubs namespace but without enableEntityManagement where the component supports multiple event hubs that must be predeployed already seems like a something we could just choose not to support, hence design review.

CodeMonkeyLeet · 2021-11-12T20:22:17Z

pubsub/azure/eventhubs/eventhubs.go

 	"time"

+	"github.com/Azure/azure-amqp-common-go/v3/aad"
+	"github.com/Azure/azure-amqp-common-go/v3/conn"
 	eventhub "github.com/Azure/azure-event-hubs-go/v3"
 	"github.com/Azure/azure-event-hubs-go/v3/eph"


(Linking dapr/dapr#2218 (comment) for context on this comment.)

Correct, EPH does not. To implement batch receive, the component would have to go down a level of abstraction to Hub.Receive with the ReceiveWithPrefetchCount option, which also means taking on explicit partition and checkpoint management that EPH otherwise currently handles. There's definitely a cost-benefit analysis to be done there, if/when pubsub batching is pursued as a feature in Dapr.

CodeMonkeyLeet · 2021-11-12T20:29:13Z

pubsub/azure/eventhubs/eventhubs.go

+	defaultMessageRetentionInDays = 1
+	defaultPartitionCount         = 1
+
+	timeoutInSec time.Duration = 5 * time.Second //nolint


Why the //nolint here? Is it because the time.Duration is specified? (It's unnecessary given that it should be inferred from time.Second already?)

This is for the name timeoutInSec, that is throwing a linter error.

Please, use a new name and consider a bigger default as resource provisioning can be slow.

If needed, split into multiple timeout configs for different scenarios.

CodeMonkeyLeet · 2021-11-12T21:38:45Z

pubsub/azure/eventhubs/eventhubs.go

+	missingResourceGroupNameMsg              = "error: missing resourceGroupName attribute required for entityManagement"
+	missingSubscriptionIDMsg                 = "error: missing subscriptionID attribute required for entityManagement"
+	entityManagementConnectionStrMsg         = "error: entity management support is not available with connectionString"
+	differentTopicConnectionStringErrorTmpl  = "error: connection string already contains topic/hubName different from requested topic %s"


hubName is an internal var name, consider using clarifying:

Suggested change

differentTopicConnectionStringErrorTmpl = "error: connection string already contains topic/hubName different from requested topic %s"

differentTopicConnectionStringErrorTmpl = "error: specified topic %s does not match the event hub name in the provided connectionString"

CodeMonkeyLeet · 2021-11-12T21:40:38Z

pubsub/azure/eventhubs/eventhubs.go

-	storageAccountName   string
-	storageAccountKey    string
-	storageContainerName string
+	ConnectionString       string `json:"connectionString,omitempty"`


Consider refactoring into separate source files now that the implementation has grown sufficiently complicated to warrant it. A common split in other components is to deal with metadata in a separate metadata.go file.

CodeMonkeyLeet · 2021-11-12T21:46:52Z

pubsub/azure/eventhubs/eventhubs.go

+	StorageAccountKey      string `json:"storageAccountKey,omitempty"`
+	StorageContainerName   string `json:"storageContainerName,omitempty"`
+	EnableEnitiyManagement bool   `json:"enableEntityManagement,omitempty,string"`
+	MessageRetentionInDays int32  `json:"messageRetentionInDays,omitempty,string"`


For my understanding, is there any reason to annotate int32 members MessageRetentionInDays and PartitionCount as string?

Yes without the annotation as String, it fails when unmarshaling the JSON serialised config.

pubsub/azure/eventhubs/eventhubs.go

CodeMonkeyLeet · 2021-11-12T22:10:42Z

pubsub/azure/eventhubs/eventhubs.go

+	if parsed.HubName != "" {
+		return parsed.HubName, nil
 	}
+	return "", nil


The if check and default return can be condensed down to just return parsed.HubName, nil

CodeMonkeyLeet · 2021-11-12T22:15:20Z

pubsub/azure/eventhubs/eventhubs.go

+	StorageAccountName     string `json:"storageAccountName,omitempty"`
+	StorageAccountKey      string `json:"storageAccountKey,omitempty"`
+	StorageContainerName   string `json:"storageContainerName,omitempty"`
+	EnableEnitiyManagement bool   `json:"enableEntityManagement,omitempty,string"`


Typo:

Suggested change

EnableEnitiyManagement bool `json:"enableEntityManagement,omitempty,string"`

EnableEntityManagement bool `json:"enableEntityManagement,omitempty,string"`

CodeMonkeyLeet · 2021-11-12T22:21:08Z

pubsub/azure/eventhubs/eventhubs.go

 	if err != nil {
 		return err
 	}

-	leaserCheckpointer, err := storage.NewStorageLeaserCheckpointer(cred, aeh.metadata.storageAccountName, aeh.metadata.storageContainerName, azure.PublicCloud)
+	// Set topic name, consumerID prefix for partition checkpoint lease blob path.
+	leaserPrefixOpt := storage.WithPrefixInBlobPath(aeh.getStoragePrefixString(req.Topic))


This should be a comment in the code as well for future maintainers.

CodeMonkeyLeet · 2021-11-12T22:32:41Z

pubsub/azure/eventhubs/eventhubs.go

+			aeh.logger.Infof("connectionString will be used to connect to given hub %q only. connection established on first publish/subscribe", hubName)
+			aeh.logger.Debugf("req.Topic field in incoming requests must be same as the one in the connectionString")


Nit: Capitalization. Also consider clarifying implied impact on first connect instead of having a separate debug message.

Suggested change

aeh.logger.Infof("connectionString will be used to connect to given hub %q only. connection established on first publish/subscribe", hubName)

aeh.logger.Debugf("req.Topic field in incoming requests must be same as the one in the connectionString")

aeh.logger.Infof("connectionString provided is specific to event hub %q. Publishing or subscribing to a topic that does not match this event hub will fail when attempted.", hubName)

mukundansundar · 2021-11-13T05:04:30Z

I'm going to timebox my review on this for now because I've spent a couple of hours looking at this and I'm not done yet. I think this needs a design review first as I'm spending a chunk of time trying to infer if particular behaviors are intended.

In particular, there are a lot of contingent and interacting configuration options as indicated in this comment and I think it would be a good starting point to enumerate the expected valid configuration combinations and the resulting behaviors.

In general, it seems like we could scope down the scenarios we support, for example, supporting connection string auth to EventHubs namespace but without enableEntityManagement where the component supports multiple event hubs that must be predeployed already seems like a something we could just choose not to support, hence design review.

@CodeMonkeyLeet
Added design with valid configurations and more explanation. #951 (comment)

Based on the decision there, I can modify this PR.

artursouza · 2021-12-15T05:37:39Z

pubsub/azure/eventhubs/eventhubs.go

+	defaultMessageRetentionInDays = 1
+	defaultPartitionCount         = 1
+
+	timeoutInSec time.Duration = 5 * time.Second //nolint


Please, use a new name and consider a bigger default as resource provisioning can be slow.

artursouza · 2021-12-15T05:37:57Z

pubsub/azure/eventhubs/eventhubs.go

+	defaultMessageRetentionInDays = 1
+	defaultPartitionCount         = 1
+
+	timeoutInSec time.Duration = 5 * time.Second //nolint


If needed, split into multiple timeout configs for different scenarios.

artursouza · 2021-12-15T05:39:17Z

pubsub/azure/eventhubs/eventhubs.go

+		return false, err
+	}
+	if *g.Name == aeh.metadata.ConsumerGroup {
+		aeh.logger.Infof("consumer group %s esists for the requested topic/eventHub %s", aeh.metadata.ConsumerGroup, hubName)


typo: exists

resolved all three.

artursouza

Overall is good. Just some minor comments.

artursouza · 2021-12-28T00:23:16Z

pubsub/azure/eventhubs/eventhubs.go

@@ -52,6 +62,15 @@ const (
 	sysPropIotHubConnectionModuleID   = "iothub-connection-module-id"
 	sysPropIotHubEnqueuedTime         = "iothub-enqueuedtime"
 	sysPropMessageID                  = "message-id"
+
+	defaultMessageRetentionInDays = 1


Can the default be infinity?

no it cannot be. See https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-quotas for numbers.

artursouza · 2021-12-28T00:24:27Z

pubsub/azure/eventhubs/eventhubs.go

+	timeoutInSec time.Duration = 5 * time.Second //nolint
+
+	// See https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-quotas for numbers.
+	maxMessageRetention = int32(90)


Is this max coming from event hubs or is it arbitrary?

see comment above line.

artursouza · 2021-12-28T00:26:18Z

pubsub/azure/eventhubs/eventhubs.go

-	storageContainerName string
+	ConnectionString        string `json:"connectionString,omitempty"`
+	EventHubNamespace       string `json:"eventHubNamespace,omitempty"`
+	ConsumerGroup           string `json:"consumerID"`


Why JSON is consumerID and not ConsumerGroup? I think these should match, not?

I think consumerID is a field coming from dapr ... It is across all pubsub ... not only eventhub ... And in every pubsub it is mapped to some form of a field called cosumer group or such ...

The metadata should make sense within the component itself. Someone that uses EventHubs expects ConsumerGroup as the terminology. There is no requirement that metadata should match cross components.

Currently the metadata properties are marshalled into json and then un-marshalled into the Go struct here. consumerID is the incoming property name from daprd Here consumerID is automatically set by daprd and not an independent property of this component.
It is called consumerGroup within this component.

OK. I found the code that handles this logic in daprd. That is the logic that automatically sets the ConsumerID attributed based on the app ID. So, this attribute is the same for all components. LGTM.

pubsub/azure/eventhubs/eventhubs.go

Signed-off-by: Mukundan Sundararajan <msundar.ms@outlook.com>

codecov · 2022-01-03T14:11:23Z

Codecov Report

Merging #1292 (aaadbd7) into master (6d9d0bc) will decrease coverage by 0.35%.
The diff coverage is 18.46%.

❗ Current head aaadbd7 differs from pull request most recent head 29e22a0. Consider uploading reports for the commit 29e22a0 to get more accurate results

@@            Coverage Diff             @@
##           master    #1292      +/-   ##
==========================================
- Coverage   34.95%   34.60%   -0.36%     
==========================================
  Files         149      149              
  Lines       12933    13112     +179     
==========================================
+ Hits         4521     4537      +16     
- Misses       7927     8089     +162     
- Partials      485      486       +1

Impacted Files	Coverage Δ
authentication/azure/auth.go	`39.88% <0.00%> (-0.47%)`	⬇️
pubsub/azure/eventhubs/eventhubs.go	`14.84% <18.63%> (-8.08%)`	⬇️
pubsub/gcp/pubsub/pubsub.go	`24.21% <0.00%> (-0.39%)`	⬇️
pubsub/rabbitmq/rabbitmq.go	`60.40% <0.00%> (+0.09%)`	⬆️
pubsub/aws/snssqs/snssqs.go	`21.46% <0.00%> (+0.09%)`	⬆️
bindings/aws/s3/s3.go	`15.17% <0.00%> (+0.10%)`	⬆️
tests/conformance/common.go	`16.73% <0.00%> (+0.12%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 5c9365b...29e22a0. Read the comment docs.

artursouza

LGTM. I just have a comment to use ConsumerGroup for consistency within EventHubs. That wins over having the same name cross components for metadata.

Signed-off-by: Mukundan Sundararajan <msundar.ms@outlook.com>

Signed-off-by: Mukundan Sundararajan <msundar.ms@outlook.com> Signed-off-by: Amit Mor <amit.mor@hotmail.com>

Signed-off-by: Mukundan Sundararajan <msundar.ms@outlook.com> Signed-off-by: jigargandhi <jigarr.gandhi@gmail.com>

mukundansundar force-pushed the eventhub-rewrite branch from 86a324d to 4d96d30 Compare November 11, 2021 09:47

mukundansundar commented Nov 12, 2021

View reviewed changes

mukundansundar marked this pull request as ready for review November 12, 2021 05:54

mukundansundar requested review from a team as code owners November 12, 2021 05:54

mukundansundar commented Nov 12, 2021

View reviewed changes

ItalyPaleAle mentioned this pull request Nov 12, 2021

Update all other Azure components to use the common auth layer (and support Azure AD auth) #1103

Closed

22 tasks

CodeMonkeyLeet reviewed Nov 12, 2021

View reviewed changes

mukundansundar mentioned this pull request Nov 13, 2021

Rewrite Azure Event Hubs Pubsub component #951

Closed

mukundansundar marked this pull request as draft November 13, 2021 07:23

mukundansundar force-pushed the eventhub-rewrite branch from 204b508 to 0c18574 Compare December 13, 2021 18:20

mukundansundar marked this pull request as ready for review December 13, 2021 18:30

mukundansundar requested review from CodeMonkeyLeet and a team December 13, 2021 18:31

mukundansundar force-pushed the eventhub-rewrite branch 2 times, most recently from 967e15d to a8f2f4b Compare December 14, 2021 04:52

mukundansundar mentioned this pull request Dec 14, 2021

Certify Azure Events Hub as Stable in pubsub #1379

Closed

artursouza reviewed Dec 15, 2021

View reviewed changes

artursouza requested changes Dec 28, 2021

View reviewed changes

mukundansundar force-pushed the eventhub-rewrite branch from aaadbd7 to 8d1e3ec Compare January 3, 2022 07:40

rewrite eventhubs component

29e22a0

Signed-off-by: Mukundan Sundararajan <msundar.ms@outlook.com>

mukundansundar force-pushed the eventhub-rewrite branch from 8d1e3ec to 29e22a0 Compare January 3, 2022 14:11

mukundansundar requested review from artursouza and a team January 3, 2022 14:25

artursouza requested changes Jan 3, 2022

View reviewed changes

mukundansundar requested a review from artursouza January 4, 2022 03:29

artursouza approved these changes Jan 4, 2022

View reviewed changes

artursouza merged commit fe6c9e3 into dapr:master Jan 4, 2022

amimimor pushed a commit to amimimor/components-contrib that referenced this pull request Jan 6, 2022

rewrite eventhubs component (dapr#1292)

547ce5f

Signed-off-by: Mukundan Sundararajan <msundar.ms@outlook.com>

amimimor pushed a commit to amimimor/components-contrib that referenced this pull request Jan 6, 2022

rewrite eventhubs component (dapr#1292)

3f9fef8

Signed-off-by: Mukundan Sundararajan <msundar.ms@outlook.com> Signed-off-by: Amit Mor <amit.mor@hotmail.com>

mukundansundar mentioned this pull request Jan 20, 2022

Update Event Hubs pubsub to Beta from Alpha dapr/docs#2121

Merged

7 tasks

jigargandhi pushed a commit to jigargandhi/components-contrib that referenced this pull request Jan 21, 2022

rewrite eventhubs component (dapr#1292)

b8d2885

Signed-off-by: Mukundan Sundararajan <msundar.ms@outlook.com> Signed-off-by: jigargandhi <jigarr.gandhi@gmail.com>

christle mentioned this pull request Aug 6, 2022

binding.eventhub: use same name schema for storage leases like on pubsub.eventhub #1940

Merged

	differentTopicConnectionStringErrorTmpl = "error: connection string already contains topic/hubName different from requested topic %s"
	differentTopicConnectionStringErrorTmpl = "error: specified topic %s does not match the event hub name in the provided connectionString"

	EnableEnitiyManagement bool `json:"enableEntityManagement,omitempty,string"`
	EnableEntityManagement bool `json:"enableEntityManagement,omitempty,string"`

		aeh.logger.Infof("connectionString will be used to connect to given hub %q only. connection established on first publish/subscribe", hubName)
		aeh.logger.Debugf("req.Topic field in incoming requests must be same as the one in the connectionString")

	aeh.logger.Infof("connectionString will be used to connect to given hub %q only. connection established on first publish/subscribe", hubName)
	aeh.logger.Debugf("req.Topic field in incoming requests must be same as the one in the connectionString")
	aeh.logger.Infof("connectionString provided is specific to event hub %q. Publishing or subscribing to a topic that does not match this event hub will fail when attempted.", hubName)

rewrite EventHubs component #1292

rewrite EventHubs component #1292

Conversation

mukundansundar commented Nov 11, 2021 • edited

Description

Issue reference

Checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mukundansundar Nov 12, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mukundansundar Nov 12, 2021 • edited

Choose a reason for hiding this comment

mukundansundar Nov 12, 2021 • edited

Choose a reason for hiding this comment

CodeMonkeyLeet left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mukundansundar commented Nov 13, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artursouza left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Jan 3, 2022

Codecov Report

artursouza left a comment

Choose a reason for hiding this comment

mukundansundar commented Nov 11, 2021 •

edited

mukundansundar Nov 12, 2021 •

edited

mukundansundar Nov 12, 2021 •

edited

mukundansundar Nov 12, 2021 •

edited

mukundansundar commented Nov 13, 2021 •

edited