Service Bus

Azure Service Bus is a fully managed enterprise message broker with message queues and publish-subscribe topics. Service Bus is used to decouple applications and services from each other, providing the following benefits:

Load-balancing work across competing workers
Safely routing and transferring data and control across service and application boundaries
Coordinating transactional work that requires a high-degree of reliability

Microsoft Documentation

Basics

Service Bus is managed by terraform scripts in Github Terrraform repo

Namespace

It is main part of Service Bus. Treat it as instance of Service Bus that is going to transport all of the messages and handle security

resource "azurerm_servicebus_namespace" "main" {
  resource_group_name = data.azurerm_resource_group.main.name
  location            = data.azurerm_resource_group.main.location
  name                = local.namings.namespace
  sku                 = local.sku[var.internal_environment]
  zone_redundant      = false
  local_auth_enabled  = false
  tags                = local.tags
}

provider "azurerm" {
  features {}
}

resource "azurerm_resource_group" "example" {
  name     = "terraform-servicebus"
  location = "West Europe"
}

resource "azurerm_servicebus_namespace" "example" {
  name                = "tfex-servicebus-namespace"
  location            = azurerm_resource_group.example.location
  resource_group_name = azurerm_resource_group.example.name
  sku                 = "Standard"

  tags = {
    source = "terraform"
  }
}

Topic

Destination for messages to be published. Topic distribute messages between Subscriptions using filters

resource "azurerm_servicebus_topic" "incoming" {
  name                                    = "incoming-requests"
  namespace_id                            = azurerm_servicebus_namespace.main.id
  duplicate_detection_history_time_window = "PT1M"
  requires_duplicate_detection            = true
  enable_batched_operations               = false
  enable_partitioning                     = false
}

resource "azurerm_resource_group" "example" {
  name     = "tfex-servicebus-topic"
  location = "West Europe"
}

resource "azurerm_servicebus_namespace" "example" {
  name                = "tfex-servicebus-namespace"
  location            = azurerm_resource_group.example.location
  resource_group_name = azurerm_resource_group.example.name
  sku                 = "Standard"

  tags = {
    source = "terraform"
  }
}

resource "azurerm_servicebus_topic" "example" {
  name         = "tfex_servicebus_topic"
  namespace_id = azurerm_servicebus_namespace.example.id

  enable_partitioning = true
}

📘 Topic vs Queue

Queues offer First In, First Out (FIFO) message delivery to one or more competing consumers. That is, receivers typically receive and process messages in the order in which they were added to the queue. And, only one message consumer receives and processes each message.

Topics and subscriptions provide a one-to-many form of communication in a publish and subscribe pattern. It's useful for scaling to large numbers of recipients. Each published message is made available to each subscription registered with the topic. Publisher sends a message to a topic and one or more subscribers receive a copy of the message.

Subscription

Pushes messages to multiple different subscribers and manages message filtering. Also has own dead letter which can be processed based on product needs.

resource "azurerm_servicebus_subscription" "incoming" {
  name               = "in-req-sub"
  topic_id           = azurerm_servicebus_topic.incoming.id
  max_delivery_count = 1
  lock_duration      = "P0DT0H1M0S"
}

resource "azurerm_servicebus_subscription_rule" "marketplace_requests" {
  name            = "mp-req-filter"
  subscription_id = azurerm_servicebus_subscription.incoming.id
  filter_type     = "CorrelationFilter"
  correlation_filter {
    label = "MarketplaceWebhookReceived"
  }
}

resource "azurerm_resource_group" "example" {
  name     = "tfex-servicebus-subscription-rule-sql"
  location = "West Europe"
}

resource "azurerm_servicebus_namespace" "example" {
  name                = "tfex-servicebus-namespace"
  location            = azurerm_resource_group.example.location
  resource_group_name = azurerm_resource_group.example.name
  sku                 = "Standard"

  tags = {
    source = "terraform"
  }
}

resource "azurerm_servicebus_topic" "example" {
  name         = "tfex_servicebus_topic"
  namespace_id = azurerm_servicebus_namespace.example.id

  enable_partitioning = true
}

resource "azurerm_servicebus_subscription" "example" {
  name               = "tfex_servicebus_subscription"
  topic_id           = azurerm_servicebus_topic.example.id
  max_delivery_count = 1
}

resource "azurerm_servicebus_subscription_rule" "example" {
  name            = "tfex_servicebus_rule"
  subscription_id = azurerm_servicebus_subscription.example.id
  filter_type     = "SqlFilter"
  sql_filter      = "colour = 'red'"
}

resource "azurerm_resource_group" "example" {
  name     = "tfex-servicebus-subscription-rule-cor"
  location = "West Europe"
}

resource "azurerm_servicebus_namespace" "example" {
  name                = "tfex-servicebus-namespace"
  location            = azurerm_resource_group.example.location
  resource_group_name = azurerm_resource_group.example.name
  sku                 = "Standard"

  tags = {
    source = "terraform"
  }
}

resource "azurerm_servicebus_topic" "example" {
  name         = "tfex_servicebus_topic"
  namespace_id = azurerm_servicebus_namespace.example.id

  enable_partitioning = true
}

resource "azurerm_servicebus_subscription" "example" {
  name               = "tfex_servicebus_subscription"
  topic_id           = azurerm_servicebus_topic.example.id
  max_delivery_count = 1
}

resource "azurerm_servicebus_subscription_rule" "example" {
  name            = "tfex_servicebus_rule"
  subscription_id = azurerm_servicebus_subscription.example.id
  filter_type     = "CorrelationFilter"

  correlation_filter {
    correlation_id = "high"
    label          = "red"
    properties = {
      customProperty = "value"
    }
  }
}

RBAC

To execute any kind of operations an identity must have one these built-in roles depending on type of operation

Built-in role	Description	ID
Azure Service Bus Data Receiver	Allows for receive access to Azure Service Bus resources	4f6d3b9b-027b-4f4c-9142-0e5a2a2247e0
Azure Service Bus Data Sender	Allows for send access to Azure Service Bus resources	69a216fc-b8fb-44d8-bc22-1f3c2cd27a39
Azure Service Bus Data Owner	Allows for full access to Azure Service Bus resources.	090c5cfd-751d-490a-894a-3ce6f1109419

Message processing

The most observable and resilient way to handle events is in/outbox architectural pattern. Example Reference: https://www.baeldung.com/cs/outbox-pattern-microservices

Publishing

Each service that produces any kind of events will save them to Outbound table in database instead of direct publishing. After that a scheduled background job picks up all unpublished events, publishes to configured topic and updated records in database with sent date.

Service Bus Message properties that are set by publisher:

MessageId - retrieved from database record id to be unique and correlatable
Subject - basically this is type of message that is shared as contract and is part of subscription filters
TimeToLive is set only if it required for message to expire in if not processed. Not used at the moment

Consuming

Each service that consumes any kind of events will have 2 parts:

Message Processor that consumes Service Bus messages from configured subscription and saves message to Inbound table in database. During message processing the message is locked for a duration until it is set to Completed state or Abandoned, thus no other consumer can pick that message up
Message Consumer is background job that polls database Inbound table to retrieve not processed events, serializes into commands and routes to respective command handlers

📘 Maximum delivery count

There's a limit on number of attempts to deliver messages for Service Bus queues and subscriptions. The default value is 10. Whenever a message has been delivered under a peek-lock, but has been either explicitly abandoned or the lock has expired, the delivery count on the message is incremented. When the delivery count exceeds the limit, the message is moved to the DLQ. The dead-letter reason for the message in DLQ is set to: MaxDeliveryCountExceeded. This behavior can't be disabled, but you can set the max delivery count to a large number.

Service Bus Message properties that are consumed by processor:

MessageId - retrieved from message and set to database record before saving
Subject - basically this is type of message that is shared as contract and is part of subscription filters
Body - actual event content that is JSON (can be changed to full object in future) in binary format

How to implement

It has already been implemented as pluggable dotnet library that can be used both in long-running console application in Docker container or as background service of WebAPI. If any python service/application needs to consume/produce messages it will need to implement database polling/saving events. Other part will be covered by instance(s) of dotnet apps.

Configuration of those Service Bus workers can be set up either by environment variables or in Azure AppConfiguration + Key Vault for sensitive settings like connection strings

Azure authentication

Using DefaultAzureCredential or EnvironmentAzureCredential allows application to access Azure resources. But requires set of pre-defined environment variables

Variable	Description
AZURE_TENANT_ID	The Microsoft Entra tenant (directory) ID.
AZURE_CLIENT_ID	The client (application) ID of an App Registration in the tenant.
AZURE_CLIENT_SECRET	A client secret that was generated for the App Registration.

Application configuration

Mandatory if Azure AppConfiguration is used

{
  "APP_CONFIG":  {
    "ENABLED": true,
    "KEY_VAULT_NAME": "aichatkvlocal", // Key Vault for sensitive data is referenced by AppConfig
    "NAME": "aichat-cfg-local",
    "LABELS": "mphook,shared,events"
  }
}

APP_CONFIG__KEY_VAULT_NAME = "aichatkvlocal"
APP_CONFIG__NAME           = "aichat-cfg-local"
APP_CONFIG__LABELS         = "shared,mphook,events"
APP_CONFIG__ENABLED        = true

Events configurations

{
  "EVENTS": {
    "RandomMetric" : {
      "PRODUCE": {
        "MP_METRIC": {
          "TICK_DURATION": 3.0,
          "NAMESPACE": "https://empcentralbus-dv.servicebus.windows.net/",
          "TOPIC_OUT": "incoming-requests",
          "DATABASE": "Server=localhost;Initial Catalog=marketplace;Persist Security Info=False;User ID=sa;Password=passw!0rd;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=True"
        }
      }
    },
    "MarketplaceWebhookReceived": {
      "PRODUCE": {
        "MP_HOOK": {
          "TICK_DURATION": 3.0,
          "NAMESPACE": "https://empcentralbus-dv.servicebus.windows.net/",
          "TOPIC_OUT": "incoming-requests",
          "DATABASE": "Server=localhost;Initial Catalog=marketplace;Persist Security Info=False;User ID=sa;Password=passw!0rd;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=True"
        }
      },
      "CONSUME": {
        "MP_PROC": {
          "NAMESPACE": "https://empcentralbus-dv.servicebus.windows.net/",
          "TOPIC_IN": "incoming-requests",
          "SUBSCRIPTION_IN": "in-req-sub",
          "DATABASE": "Server=localhost;Initial Catalog=marketplace;Persist Security Info=False;User ID=sa;Password=passw!0rd;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=True"
        },
        "METRIC": {
          "NAMESPACE": "https://empcentralbus-dv.servicebus.windows.net/",
          "TOPIC_IN": "incoming-requests",
          "SUBSCRIPTION_IN": "all-req-sub",
          "DATABASE": "Server=localhost;Initial Catalog=metrics;Persist Security Info=False;User ID=sa;Password=passw!0rd;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=True"
        }
      }
    }
  }
}

EVENTS__RandomMetric__PRODUCE__MP_METRIC__TICK_DURATION = 3.0,
EVENTS__RandomMetric__PRODUCE__MP_METRIC__NAMESPACE = "https://empcentralbus-dv.servicebus.windows.net/",
EVENTS__RandomMetric__PRODUCE__MP_METRIC__TOPIC_OUT = "incoming-requests",
EVENTS__RandomMetric__PRODUCE__MP_METRIC__DATABASE = "Server=localhost;Initial Catalog=marketplace;Persist Security Info=False;User ID=sa;Password=passw!0rd;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=True"

EVENTS__MarketplaceWebhookReceived__PRODUCE__MP_HOOK__TICK_DURATION = 3.0,
EVENTS__MarketplaceWebhookReceived__PRODUCE__MP_HOOK__NAMESPACE = "https://empcentralbus-dv.servicebus.windows.net/",
EVENTS__MarketplaceWebhookReceived__PRODUCE__MP_HOOK__TOPIC_OUT = "incoming-requests",
EVENTS__MarketplaceWebhookReceived__PRODUCE__MP_HOOK__DATABASE = "Server=localhost;Initial Catalog=marketplace;Persist Security Info=False;User ID=sa;Password=passw!0rd;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=True"

EVENTS__MarketplaceWebhookReceived__CONSUME__MP_PROC__NAMESPACE = "https://empcentralbus-dv.servicebus.windows.net/",
EVENTS__MarketplaceWebhookReceived__CONSUME__MP_PROC__TOPIC_IN = "incoming-requests",
EVENTS__MarketplaceWebhookReceived__CONSUME__MP_PROC__SUBSCRIPTION_IN = "in-req-sub",
EVENTS__MarketplaceWebhookReceived__CONSUME__MP_PROC__DATABASE = "Server=localhost;Initial Catalog=marketplace;Persist Security Info=False;User ID=sa;Password=passw!0rd;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=True"

EVENTS__MarketplaceWebhookReceived__CONSUME__METRIC__NAMESPACE = "https://empcentralbus-dv.servicebus.windows.net/",
EVENTS__MarketplaceWebhookReceived__CONSUME__METRIC__TOPIC_IN = "incoming-requests",
EVENTS__MarketplaceWebhookReceived__CONSUME__METRIC__SUBSCRIPTION_IN = "all-req-sub",
EVENTS__MarketplaceWebhookReceived__CONSUME__METRIC__DATABASE = "Server=localhost;Initial Catalog=metrics;Persist Security Info=False;User ID=sa;Password=passw!0rd;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=True"

[
  {
    "key": "EVENTS:RandomMetric:PRODUCE:MP_METRIC:TICK_DURATION",
    "label": "events",
    "value": "3.0"
  },
  {
    "key": "EVENTS:RandomMetric:PRODUCE:MP_METRIC:NAMESPACE",
    "label": "events",
    "value": "https://empcentralbus-dv.servicebus.windows.net/"
  },
  {
    "key": "EVENTS:RandomMetric:PRODUCE:MP_METRIC:TOPIC_OUT",
    "label": "events",
    "value": "incoming-requests"
  },
  {
    "key": "EVENTS:RandomMetric:PRODUCE:MP_METRIC:DATABASE",
    "label": "events",
    "value": "{\"uri\":\"https://aichatkvlocal.vault.azure.net/secrets/localsql\"}",
    "content_type": "application/vnd.microsoft.appconfig.keyvaultref+json;charset=utf-8"
  },
  
  {
    "key": "EVENTS:MarketplaceWebhookReceived:PRODUCE:MP_HOOK:TICK_DURATION",
    "label": "events",
    "value": "3.0"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:PRODUCE:MP_HOOK:NAMESPACE",
    "label": "events",
    "value": "https://empcentralbus-dv.servicebus.windows.net/"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:PRODUCE:MP_HOOK:TOPIC_OUT",
    "label": "events",
    "value": "incoming-requests"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:PRODUCE:MP_HOOK:DATABASE",
    "label": "events",
    "value": "{\"uri\":\"https://aichatkvlocal.vault.azure.net/secrets/localsql\"}",
    "content_type": "application/vnd.microsoft.appconfig.keyvaultref+json;charset=utf-8"
  },
  
  {
    "key": "EVENTS:MarketplaceWebhookReceived:CONSUME:MP_PROC:NAMESPACE",
    "label": "events",
    "value": "https://empcentralbus-dv.servicebus.windows.net/"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:CONSUME:MP_PROC:TOPIC_IN",
    "label": "events",
    "value": "incoming-requests"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:CONSUME:MP_PROC:SUBSCRIPTION_IN",
    "label": "events",
    "value": "in-req-sub"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:CONSUME:MP_PROC:DATABASE",
    "label": "events",
    "value": "{\"uri\":\"https://aichatkvlocal.vault.azure.net/secrets/localsql\"}",
    "content_type": "application/vnd.microsoft.appconfig.keyvaultref+json;charset=utf-8"
  },
  
  {
    "key": "EVENTS:MarketplaceWebhookReceived:CONSUME:METRIC:NAMESPACE",
    "label": "events",
    "value": "https://empcentralbus-dv.servicebus.windows.net/"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:CONSUME:METRIC:TOPIC_IN",
    "label": "events",
    "value": "incoming-requests"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:CONSUME:METRIC:SUBSCRIPTION_IN",
    "label": "events",
    "value": "all-req-sub"
  },
  {
    "key": "EVENTS:MarketplaceWebhookReceived:CONSUME:METRIC:DATABASE",
    "label": "events",
    "value": "{\"uri\":\"https://aichatkvlocal.vault.azure.net/secrets/localsql\"}",
    "content_type": "application/vnd.microsoft.appconfig.keyvaultref+json;charset=utf-8"
  }
]

Events configuration structure

EVENTS - base configuration section that stores all consumer/producer settings. If absent or empty, then no workers will be created
- {event_name} - completely dynamic name of event that should be processed by underlying consumers/producers. Must match event name that has been defined in Correlation filter and saved into database
  - PRODUCE - section for producers configurations
    - {component_name} - unique key of a producer. If duplicated then they will be using same instances of database contexts, which might lead to race conditions
      - TICK_DURATION - frequency of database polling and publishing messages to Service Bus topic
      - NAMESPACE - url to Service Bus namespace (instance)
      - TOPIC_OUT - topic name for publishing messages
      - DATABASE - SQL Server connection string to read event that should be published
  - CONSUME - section for event consumers configurations
    - {component_name} - unique key of a consumer/producer. If duplicated then they will be using same instances of database contexts, which might lead to race conditions
      - NAMESPACE - url to Service Bus namespace (instance)
      - TOPIC_IN - topic name where messages should be consumed from
      - SUBSCRIPTION_IN - subscription in the topic where messages are pushed and filtered
      - DATABASE - SQL Server connection string to save consumed events

Configuration is subject for discussions and might be changed after feedbacks

🚧 Azure AppConfiguration

Should be configured and managed only by Terraform.

Manual changes are allowed only for test instances.

Events

👍 To be extended

This list is not complete by far and more events will be added in future. I you have anything to add, suggest an edit

From now on there will be several basic kinds of events:

Metrics
- DatabricksExecutionCompleted
Azure Marketplace Webhook events
- MarketplaceWebhookReceived
- MarketplaceWebhookProcessed
Infrastructure events
- Current:
  - InfrastructureEvent
- Future decomposing InfrastructureEvent into more specific types
  - KeycloakRealmCreated
  - KeycloakRealmUpdated
  - ServiceDeployed
  - ServiceDeploymentFailed
  - AzureComponentCreated
  - AzureComponentUpdated

🚧 Contracts

should be generated and stored in packages with multiple languages support (python, dotnet etc.)

Creating new events

Define event contract
Define topic
1. Create new topic if it doesn't fit into existing
Define subscription
1. Extend existing subscription filter if it should be consumed by existing processor
2. Create new if new processor is required
Create PR in terraform repo with:
1. topic
2. subscription
  1. database if subscription is new
3. subscription filter
4. permissions
5. Azure AppConfiguration settings to add consumers and subscribers dynamically

PoC example

The configuration from Events configuration has been applied to single (dockerized) console app / web api. Starting the application both databases (marketplace and metrics) have been created automatically, migrations applied to ensure tables are there, generated 2 producers and 2 consumers.

Inserting MarketplaceWebhookReceived event into marketplace.messages.Outbound triggered producer MP_HOOK to publish message to Service Bus incoming-requests topic and set event's Processed to current UTC datetime. Then Service Bus has duplicated message to 2 subscriptions (all-req-sub and in-req-sub), because in-req-sub filter accepts only MarketplaceWebhookReceived and all-req-sub accepts everything. Once messages got into subscriptions, Service Bus pushes them to subscribers, which are METRIC and MP_PROC consumers. They received messages and saved to own configured databases metrics.messages.Inbound and marketplace.messages.Inbound and set event type from messages' Subject/Label

Inserting RandomMetric event into marketplace.messages.Outbound triggered producer MP_METRIC to publish message to Service Bus incoming-requests topic and set event's Processed to current UTC datetime. Then Service Bus has moved message to all-req-sub subscription, because it has no filter and gets all messages from incoming-requests topic. Once message got into subscription, Service Bus pushes it to subscriber, which is METRIC consumer. It got the message and saved it to metrics.messages.Inbound and set event type from message's Subject/Label

Basics​

Namespace​

Topic​

Subscription​

RBAC​

Message processing​

Publishing​

Consuming​

How to implement​

Azure authentication​

Application configuration​

Mandatory if Azure AppConfiguration is used​

Events configurations​

Events configuration structure​

Events​

Creating new events​

PoC example​