Category: Azure

A lightweight CQRS supporting library with Event Store based on Azure Table Storage.

Quick start guide

Install

Install the SuperNova.Storage Nuget package into the project.

Install-Package SuperNova.Storage -Version 1.0.0

The dependencies of the package are:

.NETCoreApp 2.0
Microsoft.Azure.DocumentDB.Core (>= 1.7.1)
Microsoft.Extensions.Logging.Debug (>= 2.0.0)
SuperNova.Shared (>= 1.0.0)
WindowsAzure.Storage (>= 8.5.0)

Implemention guide

Write Side – Event Sourcing

Once the package is installed, we can start sourcing events in an application. For example, let’s start with a canonical example of UserController in a Web API project.

We can use the dependency injection to make EventStore avilable in our controller.

Here’s an example where we register an instance of Event Store with DI framework in our Startup.cs

// Config object encapsulates the table storage connection string
services.AddSingleton(new EventStore( ... provide config ));

Now the controller:

[Produces("application/json")]
[Route("users")]
public class UsersController : Controller
{   
    public UsersController(IEventStore eventStore)
    {
        this.eventStore = eventStore; // Here capture the event store handle
    }   
    
    ... other methods skipped here
}

Aggregate

Implementing event sourcing becomes way much handier, when it’s fostered with Domain Driven Design (aka DDD). We are going to assume that we are familiar with DDD concepts (especially Aggregate Roots).

An aggregate is our consistency boundary (read as transactional boundary) in Event Sourcing. (Technically, Aggregate ID’s are our partition keys on Event Store table – therefore, we can only apply an atomic operation on a single aggregate root level.)

Let’s create an Aggregate for our User domain entity:

using SuperNova.Shared.Messaging.Events.Users;
using SuperNova.Shared.Supports;

public class UserAggregate : AggregateRoot
{
    private string _userName;
    private string _emailAddress;
    private Guid _userId;
    private bool _blocked;

Once we have the aggregate class written, we should come up with the events that are relevant to this aggregate. We can use Event storming to come up with the relevant events.

Here are the events that we will use for our example scenario:

public class UserAggregate : AggregateRoot
{

    ... skipped other codes

    #region Apply events
    private void Apply(UserRegistered e)
    {
        this._userId = e.AggregateId;
        this._userName = e.UserName;
        this._emailAddress = e.Email;            
    }

    private void Apply(UserBlocked e)
    {
        this._blocked = true;
    }

    private void Apply(UserNameChanged e)
    {
        this._userName = e.NewName;
    }
    #endregion

    ... skipped other codes
}

Now that we have our business events defined, we will define our commands for the aggregate:

public class UserAggregate : AggregateRoot
{
    #region Accept commands
    public void RegisterNew(string userName, string emailAddress)
    {
        Ensure.ArgumentNotNullOrWhiteSpace(userName, nameof(userName));
        Ensure.ArgumentNotNullOrWhiteSpace(emailAddress, nameof(emailAddress));

        ApplyChange(new UserRegistered
        {
            AggregateId = Guid.NewGuid(),
            Email = emailAddress,
            UserName = userName                
        });
    }

    public void BlockUser(Guid userId)
    {            
        ApplyChange(new UserBlocked
        {
            AggregateId = userId
        });
    }

    public void RenameUser(Guid userId, string name)
    {
        Ensure.ArgumentNotNullOrWhiteSpace(name, nameof(name));

        ApplyChange(new UserNameChanged
        {
            AggregateId = userId,
            NewName = name
        });
    }
    #endregion


    ... skipped other codes
}

So far so good!

Now we will modify the web api controller to send the correct command to the aggregate.

public class UserPayload 
{  
    public string UserName { get; set; } 
    public string Email { get; set; } 
}

// POST: User
[HttpPost]
public async Task Post(Guid projectId, [FromBody]UserPayload user)
{
    Ensure.ArgumentNotNull(user, nameof(user));

    var userId = Guid.NewGuid();    

    await eventStore.ExecuteNewAsync(
        Tenant, "user_event_stream", userId, async () => {

        var aggregate = new UserAggregate();

        aggregate.RegisterNew(user.UserName, user.Email);

        return await Task.FromResult(aggregate);
    });

    return new JsonResult(new { id = userId });
}

And another API to modify existing users into the system:

//PUT: User
[HttpPut("{userId}")]
public async Task Put(Guid projectId, Guid userId, [FromBody]string name)
{
    Ensure.ArgumentNotNullOrWhiteSpace(name, nameof(name));

    await eventStore.ExecuteEditAsync(
        Tenant, "user_event_stream", userId,
        async (aggregate) =>
        {
            aggregate.RenameUser(userId, name);

            await Task.CompletedTask;
        }).ConfigureAwait(false);

    return new JsonResult(new { id = userId });
}

That’s it! We have our WRITE side completed. The event store is now contains the events for user event stream.

Read Side – Materialized Views

We can consume the events in a seperate console worker process and generate the materialized views for READ side.

The readers (the console application – Azure Web Worker for instance) are like feed processor and have their own lease collection that makes them fault tolerant and resilient. If crashes, it catches up form the last event version that was materialized successfully. It’s doing a polling – instead of a message broker (Service Bus for instance) on purpose, to speed up and avoid latencies during event propagation. Scalabilities are ensured by means of dedicating lease per tenants and event streams – which provides pretty high scalability.

How to listen for events?

In a worker application (typically a console application) we will listen for events:

private static async Task Run()
{
    var eventConsumer = new EventStreamConsumer(        
        ... skipped for simplicity
        "user-event-stream", 
        "user-event-stream-lease");
    
    await eventConsumer.RunAndBlock((evts) =>
    {
        foreach (var @evt in evts)
        {
            if (evt is UserRegistered userAddedEvent)
            {
                readModel.AddUserAsync(new UserDto
                {
                    UserId = userAddedEvent.AggregateId,
                    Name = userAddedEvent.UserName,
                    Email = userAddedEvent.Email
                }, evt.Version);
            }

            else if (evt is UserNameChanged userChangedEvent)
            {
                readModel.UpdateUserAsync(new UserDto
                {
                    UserId = userChangedEvent.AggregateId,
                    Name = userChangedEvent.NewName
                }, evt.Version);
            }
        }

    }, CancellationToken.None);
}

static void Main(string[] args)
{
    Run().Wait();
}

Now we have a document collection (we are using Cosmos Document DB in this example for materialization but it could be any database essentially) that is being updated as we store events in event stream.

Conclusion

The library is very light weight and havily influenced by Greg’s event store model and aggreagate model. Feel free to use/contribute.

Thank you!

	FROM mcr.microsoft.com/dotnet/core/aspnet:3.0-buster-slim AS base
	WORKDIR /app

	# Build the app
	FROM mcr.microsoft.com/dotnet/core/sdk:3.0-buster AS build
	WORKDIR /src
	COPY . .
	WORKDIR "/src/BlazorApp"
	RUN dotnet restore "BlazorApp.csproj"
	RUN dotnet build "BlazorApp.csproj" -c Release -o /app

	# Run the unit tests
	RUN dotnet restore "BlazorAppTest.csproj"
	RUN dotnet build "BlazorAppTest.csproj"
	RUN dotnet test "BlazorAppTest.csproj"


	# Publish the app
	FROM build AS publish
	RUN dotnet publish "BlazorApp.csproj" -c Release -o /app

	FROM microsoft/azure-cli AS final
	ARG AppID
	ARG AppSecret
	ARG TenantID
	ARG StorageAccountName

	WORKDIR /app
	COPY –from=publish /app .
	WORKDIR "/app/BlazorApp/dist"
	RUN az login –service-principal –username $AppID –password $AppSecret –tenant $TenantID
	RUN az storage blob delete-batch –account-name $StorageAccountName –source "\$web"
	RUN az storage blob upload-batch –account-name $StorageAccountName -s . -d "\$web"

	docker build
	-f Blazor.Dockerfile
	–build-arg StorageAccountName=<Storage-Account-Name>
	–build-arg AppID=<APP GUID>
	–build-arg AppSecret=<PASSWORD>
	–build-arg TenantID=<Azure AD TENANT ID>
	-t blazor-app .


	trigger:
	– master

	pool:
	name: Hosted Ubuntu 1604
	steps:
	– task: Docker@0
	displayName: 'Build and release Blazor to Storage Account'
	inputs:
	dockerFile: src/Blazor.Dockerfile
	buildArguments: \|
	StorageAccountName=$(StorageAccountName)
	AppID=$(AppID)
	AppSecret=$(AppSecret)
	TenantID=$(TenantID)



	# This script should be used into the build pipeline to update API changes
	# to Azure API management Service.
	# The steps that are scripted below:
	#
	# 1. Updates the Internal Git Repo with the published API documents in Azure
	# 2. Retrieves the Git Credentails of internal Repo
	# 3. Clones the repo in to a temporary folder
	# 4. Merges changes from VSTS repo to the temporary cloned repository
	# 5. Commit dirty changes
	# 6. Push changes to Azure Internal Repo (master branch)
	# 7. Publishes the API updates from internal Repo to Azure API management service.
	#
	# Example Usage:
	#
	# Parameters:
	# – SubscriptionId: The subscription ID where the API management Service is provisioned
	# – ResourceGroup: Resource group name
	# – ServiceName: API management Service name
	# – SourceDirectory: Directory where the API documents, policies etc. are located
	# – TempDirectory: A temporary directory where the data will be downloaded
	# – UserName: $ENV:RELEASE_DEPLOYMENT_REQUESTEDFOREMAIL – The user name that will be used to commit to Git. You can use Relase triggerd here
	# – UserEmailAddress: $ENV:RELEASE_DEPLOYMENT_REQUESTEDFOR The email address of the user
	# – CommitMessage: $ENV:RELEASE_RELEASEWEBURL The commit message
	#
	# Get-AzureRMApiManagementGitRepo `
	# -SubscriptionId "– Subscription ID — " `
	# -ResourceGroup " — Resource group name — " `
	# -ServiceName " — APIM name — " `
	# -SourceDirectory "..\src\api-management" `
	# -TempDirectory 'C:\Temp\apim' `
	# -UserName "Moim Hossain" `
	# -CommitMessage "Commited from CI"
	# -UserEmailAddress "moim.hossain"


	Function ExecuteGitCommand {
	param
	(
	[System.Object[]]$gitCommandArguments
	)

	$gitExePath = "C:\Program Files\git\bin\git.exe"
	& $gitExePath $gitCommandArguments
	}

	Function Copy-DirectoryContents {
	param
	(
	[System.String]$SrcPath,
	[System.String]$destPath
	)

	ROBOCOPY $SrcPath $destPath /MIR
	}

	Function Get-AzureRMApiManagementGitRepo {

	param
	(
	[System.String]
	$SubscriptionId,
	[System.String]
	$ResourceGroup,
	[System.String]
	$ServiceName,
	[System.String]
	$UserName,
	[System.String]
	$UserEmailAddress,
	[System.String]
	$CommitMessage,
	[System.String]
	$SourceDirectory,
	[System.String]
	$TempDirectory
	)
	Write-Output "Initializing context…"
	$context = New-AzureRmApiManagementContext `
	–ResourceGroupName $ResourceGroup `
	–ServiceName $ServiceName
	Write-Output "Initializing context…Completed"

	Write-Output "Syncing Git Repo with current API management state…"
	Save-AzureRmApiManagementTenantGitConfiguration `
	–Context $context `
	–Branch 'master' `
	–PassThru –Force
	Write-Output "Syncing Git Repo with current API management state…Completed"

	$expiry = (Get-Date) + '1:00:00'
	$parameters = @{
	"keyType" = "primary"
	"expiry" = ('{0:yyyy-MM-ddTHH:mm:ss.000Z}' -f $expiry)
	}

	$resourceId = '/subscriptions/{0}/resourceGroups/{1}/providers/Microsoft.ApiManagement/service/{2}/users/git' -f $SubscriptionId, $ResourceGroup, $ServiceName


	if ((Test-Path –Path $TempDirectory )) {
	Remove-Item $TempDirectory –Force –Recurse –ErrorAction "Stop"
	}

	$gitRemoteSrcPath = Join-Path –Path $TempDirectory –ChildPath 'remote-api-src'

	Write-Output "Retrieving Git Credentials…"
	$gitUsername = 'apim'
	$gitPassword = (Invoke-AzureRmResourceAction `
	–Action 'token' `
	–ResourceId $resourceId `
	–Parameters $parameters `
	–ApiVersion '2016-10-10' `
	–Force).Value
	$escapedGitPassword = [System.Uri]::EscapeDataString($gitPassword)
	Write-Output "Retrieving Git Credentials…Completed"


	$gitRepositoryUrl = 'https://{0}:{1}@{2}.scm.azure-api.net/' -f $gitUsername, $escapedGitPassword, $ServiceName


	Write-Host "Performing Git clone… $gitRemoteSrcPath"
	ExecuteGitCommand –gitCommandArguments @("clone", "$gitRepositoryUrl", "$gitRemoteSrcPath")
	Write-Host "Performing Git clone… Completed"


	# Copy changes from Source Repository (VSTS) to Locally Cloned Repository
	$apiSources = Join-Path –Path $gitRemoteSrcPath –ChildPath 'api-management'
	Copy-DirectoryContents `
	–SrcPath $SourceDirectory `
	–destPath $apiSources


	Set-Location $gitRemoteSrcPath

	ExecuteGitCommand –gitCommandArguments @("config", "user.email", $UserEmailAddress)
	ExecuteGitCommand –gitCommandArguments @("config", "user.name", $UserName)
	ExecuteGitCommand –gitCommandArguments @("add", "–all")
	ExecuteGitCommand –gitCommandArguments @("commit", "-m", $CommitMessage)
	ExecuteGitCommand –gitCommandArguments @("push")

	Publish-AzureRmApiManagementTenantGitConfiguration `
	–Context $context `
	–Branch 'master' `
	–PassThru `
	–Verbose
	}

Service	Resource ID	Status	Date	Assign access
Azure Resource Manager	https://management.azure.com/	Available	September 2017	Azure portal PowerShell Azure CLI
Azure Key Vault	https://vault.azure.net	Available	September 2017
Azure Data Lake	https://datalake.azure.net/	Available	September 2017
Azure SQL	https://database.windows.net/	Available	October 2017
Azure Event Hubs	https://eventhubs.azure.net	Available	December 2017
Azure Service Bus	https://servicebus.azure.net	Available	December 2017
Azure Storage	https://storage.azure.com/	Preview	May 2018

	internal class TokenHelper
	{
	internal async Task<string> GetManagementApiAccessTokenAsync()
	{
	var astp = new AzureServiceTokenProvider();
	var accessToken = await astp
	.GetAccessTokenAsync(Constants.AzureManagementAPI);
	return accessToken;
	}
	}

	internal class StorageAccountHelper
	{
	internal async Task<StorageKeys> GetStorageKeysAsync()
	{
	var token = await new TokenHelper().GetManagementApiAccessTokenAsync();

	return await GetStorageKeysAsync(token);
	}


	internal async Task<StorageKeys> GetStorageKeysAsync(string token)
	{
	var uri = new Uri($"{Constants.AzureManagementAPI}subscriptions/{Constants.SubscriptionID}/resourceGroups/{Constants.ResourceGroupName}/providers/Microsoft.Storage/storageAccounts/{Constants.StorageAccountName}/listKeys?api-version=2016-01-01");
	var content = new StringContent(string.Empty, Encoding.UTF8, "text/html");

	using (var httpClient = new HttpClient())
	{
	httpClient.DefaultRequestHeaders.Authorization
	= new AuthenticationHeaderValue("Bearer", token);

	using (var response = await httpClient.PostAsync(uri, content))
	{
	var responseText = await response.Content.ReadAsStringAsync();
	var keys = JsonConvert.DeserializeObject<StorageKeys>(responseText);
	return keys;
	}
	}
	}
	}

	public class MessageClient
	{
	public MessageClient()
	{

	}

	public virtual async Task SendAsync(string message)
	{
	var cq = await GetQueueClient();
	await cq.AddMessageAsync(new CloudQueueMessage(message));
	}

	private static async Task<CloudQueue> GetQueueClient( )
	{
	var keys = await new StorageAccountHelper().GetStorageKeysAsync();
	var storageCredentials =
	new StorageCredentials(Constants.StorageAccountName, keys.Keys.First().Value);

	var csa = new CloudStorageAccount(storageCredentials, true);
	var cqc = csa.CreateCloudQueueClient();
	var cq = cqc.GetQueueReference(Constants.QueueName);

	await cq.CreateIfNotExistsAsync();
	return cq;
	}
	}

	public static class TokenHelper
	{
	public static Task<String> GetTokenAsync()
	{
	var provider = new AzureServiceTokenProvider();
	return provider.GetAccessTokenAsync("https://database.windows.net/");
	}
	}

	public static class SqlConnectionExtensions
	{
	public async static Task<TPayload> ExecuteScalar<TPayload>(string commandText)
	where TPayload: class
	{
	var connectionString = "connection string without credentails";
	var token = await ADAuthentication.GetSqlTokenAsync();
	using (var conn = new SqlConnection(connectionString))
	{
	conn.AccessToken = token;
	await conn.OpenAsync();

	using (var cmd = new SqlCommand(commandText, conn))
	{
	var result = await cmd.ExecuteScalarAsync();
	return result as TPayload;
	}
	}
	}
	}

	public class WebApp
	{
	public async static void StartUp()
	{
	var userName = await SqlConnectionExtensions
	.ExecuteScalar<string>("SELECT SUSER_SNAME()");
	}
	}

Pre-requisite

Publishing in Docker

Azure Pipeline as code

Creating the pipeline

Azure Pipeline Secret variables

What is API management

The challenge – Continuous Deployment

APIM Configuration Git Repository

Solution design

Building the solution

Building the release pipeline

Final thoughts

The Solution Design

Azure Functions

Prerequisites

Configuration

Hacking the function

Final thoughts!

Add-AzureRmWebAppIPRestrictions

Get-AzureRmWebAppIPRestrictions

Remove-AzureRmWebAppIPRestrictions

Publishing the project

Create a compressed artifact folder

Upload the zip into Azure web app

PowerShell task in Visual Studio Online

That’s it!

Secret management in application

Azure Key Vault as secret store

What is Service Principal?

Managed Service Identity

Using MSI from App service

Step up – activating zero-secret mode

Azure AD Authentication for Azure Services

Example: Accessing Storage Queues with MSI

Another example, this time SQL server

Conclusion

The Problem

How to do that?

How to provision?

Important !

Contribute

Quick start guide

Install

Implemention guide

Write Side – Event Sourcing

Aggregate

Read Side – Materialized Views

How to listen for events?

Conclusion

What is a swarm?

What’s in it?

Thanks!