Add dapr user agent for Azure Components (#1152)

* Add dapr useragent for Service Bus and EventHubs

* Add useragent for tablestorage

* send dapr useragent

* more user agents

* formatting

bindings/azure/blobstorage/blobstorage.go

@@ -119,7 +119,12 @@ func (a *AzureBlobStorage) Init(metadata bindings.Metadata) error {
 	if err != nil {
 		return fmt.Errorf("invalid credentials with error: %w", err)
 	}
-	p := azblob.NewPipeline(credential, azblob.PipelineOptions{})
+
+	userAgent := "dapr-" + logger.DaprVersion
+	options := azblob.PipelineOptions{
+		Telemetry: azblob.TelemetryOptions{Value: userAgent},
+	}
+	p := azblob.NewPipeline(credential, options)
 
 	containerName := a.metadata.Container
 	URL, _ := url.Parse(

bindings/azure/eventgrid/eventgrid.go

@@ -23,8 +23,9 @@ import (
 
 // AzureEventGrid allows sending/receiving Azure Event Grid events.
 type AzureEventGrid struct {
-	metadata *azureEventGridMetadata
-	logger   logger.Logger
+	metadata  *azureEventGridMetadata
+	logger    logger.Logger
+	userAgent string
 }
 
 type azureEventGridMetadata struct {
@@ -55,6 +56,7 @@ func NewAzureEventGrid(logger logger.Logger) *AzureEventGrid {
 
 // Init performs metadata init.
 func (a *AzureEventGrid) Init(metadata bindings.Metadata) error {
+	a.userAgent = "dapr-" + logger.DaprVersion
 	m, err := a.parseMetadata(metadata)
 	if err != nil {
 		return err
@@ -126,6 +128,7 @@ func (a *AzureEventGrid) Invoke(req *bindings.InvokeRequest) (*bindings.InvokeRe
 	request.Header.SetMethod(fasthttp.MethodPost)
 	request.Header.Set("Content-Type", "application/cloudevents+json")
 	request.Header.Set("aeg-sas-key", a.metadata.AccessKey)
+	request.Header.Set("User-Agent", a.userAgent)
 	request.SetRequestURI(a.metadata.TopicEndpoint)
 	request.SetBody(req.Data)
 
@@ -222,6 +225,7 @@ func (a *AzureEventGrid) createSubscription() error {
 	clientCredentialsConfig := auth.NewClientCredentialsConfig(a.metadata.ClientID, a.metadata.ClientSecret, a.metadata.TenantID)
 
 	subscriptionClient := eventgrid.NewEventSubscriptionsClient(a.metadata.SubscriptionID)
+	subscriptionClient.AddToUserAgent(a.userAgent)
 	authorizer, err := clientCredentialsConfig.Authorizer()
 	if err != nil {
 		return err

bindings/azure/eventhubs/eventhubs.go

@@ -132,13 +132,18 @@ func (a *AzureEventHubs) Init(metadata bindings.Metadata) error {
 	if err != nil {
 		return err
 	}
+	userAgent := "dapr-" + logger.DaprVersion
 	a.metadata = m
-	hub, err := eventhub.NewHubFromConnectionString(a.metadata.connectionString)
+	hub, err := eventhub.NewHubFromConnectionString(a.metadata.connectionString,
+		eventhub.HubWithUserAgent(userAgent),
+	)
 
 	// Create partitioned sender if the partitionID is configured
 	if a.metadata.partitioned() {
 		hub, err = eventhub.NewHubFromConnectionString(a.metadata.connectionString,
-			eventhub.HubWithPartitionedSender(a.metadata.partitionID))
+			eventhub.HubWithPartitionedSender(a.metadata.partitionID),
+			eventhub.HubWithUserAgent(userAgent),
+		)
 	}
 
 	if err != nil {

bindings/azure/servicebusqueues/servicebusqueues.go

@@ -51,9 +51,11 @@ func (a *AzureServiceBusQueues) Init(metadata bindings.Metadata) error {
 	if err != nil {
 		return err
 	}
+	userAgent := "dapr-" + logger.DaprVersion
 	a.metadata = meta
 
-	ns, err := servicebus.NewNamespace(servicebus.NamespaceWithConnectionString(a.metadata.ConnectionString))
+	ns, err := servicebus.NewNamespace(servicebus.NamespaceWithConnectionString(a.metadata.ConnectionString),
+		servicebus.NamespaceWithUserAgent(userAgent))
 	if err != nil {
 		return err
 	}

bindings/azure/signalr/signalr.go

@@ -50,6 +50,7 @@ type SignalR struct {
 	accessKey  string
 	version    string
 	hub        string
+	userAgent  string
 	tokens     map[string]signalrCachedToken
 	httpClient *http.Client
 
@@ -58,6 +59,8 @@ type SignalR struct {
 
 // Init is responsible for initializing the SignalR output based on the metadata.
 func (s *SignalR) Init(metadata bindings.Metadata) error {
+	s.userAgent = "dapr-" + logger.DaprVersion
+
 	connectionString, ok := metadata.Properties[connectionStringKey]
 	if !ok || connectionString == "" {
 		return fmt.Errorf("missing connection string")
@@ -128,6 +131,7 @@ func (s *SignalR) sendMessageToSignalR(url string, token string, data []byte) er
 
 	httpReq.Header.Set("Authorization", "Bearer "+token)
 	httpReq.Header.Set("Content-Type", "application/json")
+	httpReq.Header.Set("User-Agent", s.userAgent)
 
 	resp, err := s.httpClient.Do(httpReq)
 	if err != nil {

bindings/azure/storagequeues/storagequeues.go

@@ -58,7 +58,13 @@ func (d *AzureQueueHelper) Init(accountName string, accountKey string, queueName
 	d.credential = credential
 	d.decodeBase64 = decodeBase64
 	u, _ := url.Parse(fmt.Sprintf(d.reqURI, accountName, queueName))
-	d.queueURL = azqueue.NewQueueURL(*u, azqueue.NewPipeline(credential, azqueue.PipelineOptions{}))
+	userAgent := "dapr-" + logger.DaprVersion
+	pipelineOptions := azqueue.PipelineOptions{
+		Telemetry: azqueue.TelemetryOptions{
+			Value: userAgent,
+		},
+	}
+	d.queueURL = azqueue.NewQueueURL(*u, azqueue.NewPipeline(credential, pipelineOptions))
 	ctx := context.TODO()
 	_, err = d.queueURL.Create(ctx, azqueue.Metadata{})
 	if err != nil {

pubsub/azure/eventhubs/eventhubs.go

@@ -157,8 +157,10 @@ func (aeh *AzureEventHubs) Init(metadata pubsub.Metadata) error {
 	if err != nil {
 		return err
 	}
+	userAgent := "dapr-" + logger.DaprVersion
 	aeh.metadata = m
-	hub, err := eventhub.NewHubFromConnectionString(aeh.metadata.connectionString)
+	hub, err := eventhub.NewHubFromConnectionString(aeh.metadata.connectionString,
+		eventhub.HubWithUserAgent(userAgent))
 	if err != nil {
 		return fmt.Errorf("unable to connect to azure event hubs: %v", err)
 	}

pubsub/azure/servicebus/servicebus.go

@@ -257,8 +257,12 @@ func (a *azureServiceBus) Init(metadata pubsub.Metadata) error {
 		return err
 	}
 
+	userAgent := "dapr-" + logger.DaprVersion
 	a.metadata = m
-	a.namespace, err = azservicebus.NewNamespace(azservicebus.NamespaceWithConnectionString(a.metadata.ConnectionString))
+	a.namespace, err = azservicebus.NewNamespace(
+		azservicebus.NamespaceWithConnectionString(a.metadata.ConnectionString),
+		azservicebus.NamespaceWithUserAgent(userAgent))
+
 	if err != nil {
 		return err
 	}

secretstores/azure/keyvault/keyvault.go

@@ -77,6 +77,7 @@ func (k *keyvaultSecretStore) Init(metadata secretstores.Metadata) error {
 	authorizer, err := settings.GetAuthorizer()
 	if err == nil {
 		k.vaultClient.Authorizer = authorizer
+		k.vaultClient.UserAgent = "dapr-" + logger.DaprVersion
 	}
 
 	k.vaultName = settings.Values[componentVaultName]

state/azure/blobstorage/blobstorage.go

@@ -84,7 +84,11 @@ func (r *StateStore) Init(metadata state.Metadata) error {
 		return fmt.Errorf("invalid credentials with error: %s", err.Error())
 	}
 
-	p := azblob.NewPipeline(credential, azblob.PipelineOptions{})
+	userAgent := "dapr-" + logger.DaprVersion
+	options := azblob.PipelineOptions{
+		Telemetry: azblob.TelemetryOptions{Value: userAgent},
+	}
+	p := azblob.NewPipeline(credential, options)
 
 	var containerURL azblob.ContainerURL
 	customEndpoint, ok := metadata.Properties[endpointKey]

state/azure/tablestorage/tablestorage.go

@@ -76,6 +76,8 @@ func (r *StateStore) Init(metadata state.Metadata) error {
 
 	client, _ := storage.NewBasicClient(meta.accountName, meta.accountKey)
 	tables := client.GetTableService()
+	userAgent := "dapr-" + logger.DaprVersion
+	client.AddToUserAgent(userAgent)
 	r.table = tables.GetTableReference(meta.tableName)
 
 	// check table exists

state/mongodb/mongodb.go

@@ -313,6 +313,13 @@ func getMongoDBClient(metadata *mongoDBMetadata) (*mongo.Client, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), metadata.operationTimeout)
 	defer cancel()
 
+	daprUserAgent := "dapr-" + logger.DaprVersion
+	if clientOptions.AppName != nil {
+		clientOptions.SetAppName(daprUserAgent + ":" + *clientOptions.AppName)
+	} else {
+		clientOptions.SetAppName(daprUserAgent)
+	}
+
 	client, err := mongo.Connect(ctx, clientOptions)
 	if err != nil {
 		return nil, err