Category: .NET

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!

	public class TokenProvider
	{
	private readonly string tenantId;
	private readonly string clientId;
	private readonly string secret;
	private readonly string scopeUri;
	private const string IdentityEndpoint = "https://login.microsoftonline.com";
	private const string DEFAULT_SCOPE = "https://management.azure.com/";
	private const string MEDIATYPE = "application/x-www-form-urlencoded";

	public OAuthTokenProvider(string tenantId, string clientId, string secret, string scopeUri = DEFAULT_SCOPE)
	{
	this.tenantId = tenantId;
	this.clientId = WebUtility.UrlEncode(clientId);
	this.secret = WebUtility.UrlEncode(secret);
	this.scopeUri = WebUtility.UrlEncode(scopeUri);
	}

	public async Task<Token> GetAccessTokenV2EndpointAsync()
	{
	var url = $"{IdentityEndpoint}/{this.tenantId}/oauth2/v2.0/token";
	var Http = Statics.Http;
	Http.DefaultRequestHeaders.Accept.Clear();
	Http.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue(MEDIATYPE));
	var body = $"grant_type=client_credentials&client_id={clientId}&client_secret={secret}&scope={scopeUri}";
	var response = await Http.PostAsync(url, new StringContent(body, Encoding.UTF8, MEDIATYPE));
	if (response.IsSuccessStatusCode)
	{
	var tokenResponse = await response.Content.ReadAsStringAsync();
	return JsonConvert.DeserializeObject<Token>(tokenResponse);
	}
	return default(Token);
	}

	public class Token
	{
	public string access_token { get; set; }
	public string token_type { get; set; }
	public int expires_in { get; set; }
	public int ext_expires_in { get; set; }
	}
	}

	public class FileSystemApi
	{
	private readonly string storageAccountName;
	private readonly OAuthTokenProvider tokenProvider;
	private readonly Uri baseUri;
	private const string ACK_HEADER_NAME = "x-ms-acl";
	private const string API_VERSION_HEADER_NAME = "x-ms-version";
	private const string API_VERSION_HEADER_VALUE = "2018-11-09";
	private int Timeout = 100;

	public FileSystemApi(string storageAccountName, OAuthTokenProvider tokenProvider)
	{
	this.storageAccountName = storageAccountName;
	this.tokenProvider = tokenProvider;
	this.baseUri = new Uri($"https://{this.storageAccountName}.dfs.core.windows.net");
	}

	public enum AclType
	{
	User,
	Group,
	Other,
	Mask
	}
	public enum AclScope
	{
	Access,
	Default
	}

	[FlagsAttribute]
	public enum GrantType : short
	{
	None = 0,
	Read = 1,
	Write = 2,
	Execute = 4
	};

	public class AclEntry
	{
	public AclEntry(AclScope scope, AclType type, string upnOrObjectId, GrantType grant)
	{
	Scope = scope;
	AclType = type;
	UpnOrObjectId = upnOrObjectId;
	Grant = grant;
	}

	public AclScope Scope { get; private set; }
	public AclType AclType { get; private set; }
	public string UpnOrObjectId { get; private set; }
	public GrantType Grant { get; private set; }

	public string GetGrantPosixFormat()
	{
	return $"{(this.Grant.HasFlag(GrantType.Read) ? 'r' : '-')}{(this.Grant.HasFlag(GrantType.Write) ? 'w' : '-')}{(this.Grant.HasFlag(GrantType.Execute) ? 'x' : '-')}";
	}

	public override string ToString()
	{
	return $"{(this.Scope == AclScope.Default ? "default:" : string.Empty)}{this.AclType.ToString().ToLowerInvariant()}:{this.UpnOrObjectId}:{GetGrantPosixFormat()}";
	}
	}

	public async Task<bool> CreateFileSystemAsync(
	string fileSystemName)
	{
	var tokenInfo = await tokenProvider.GetAccessTokenV2EndpointAsync();

	var jsonContent = new StringContent(string.Empty);
	var headers = Statics.Http.DefaultRequestHeaders;
	headers.Clear();
	headers.Add("Authorization", $"Bearer {tokenInfo.access_token}");
	headers.Add(API_VERSION_HEADER_NAME, API_VERSION_HEADER_VALUE);
	var response = await Statics.Http.PutAsync($"{baseUri}{WebUtility.UrlEncode(fileSystemName)}?resource=filesystem", jsonContent);

	return response.IsSuccessStatusCode;
	}

	public async Task<bool> CreateDirectoryAsync(string fileSystemName, string fullPath)
	{
	var tokenInfo = await tokenProvider.GetAccessTokenV2EndpointAsync();

	var jsonContent = new StringContent(string.Empty);
	var headers = Statics.Http.DefaultRequestHeaders;
	headers.Clear();
	headers.Add("Authorization", $"Bearer {tokenInfo.access_token}");
	headers.Add(API_VERSION_HEADER_NAME, API_VERSION_HEADER_VALUE);
	var response = await Statics.Http.PutAsync($"{baseUri}{WebUtility.UrlEncode(fileSystemName)}{fullPath}?resource=directory", jsonContent);

	return response.IsSuccessStatusCode;
	}

	public async Task<bool> CreateEmptyFileAsync(string fileSystemName, string path, string fileName)
	{
	var tokenInfo = await tokenProvider.GetAccessTokenV2EndpointAsync();

	var jsonContent = new StringContent(string.Empty);
	var headers = Statics.Http.DefaultRequestHeaders;
	headers.Clear();
	headers.Add("Authorization", $"Bearer {tokenInfo.access_token}");
	headers.Add(API_VERSION_HEADER_NAME, API_VERSION_HEADER_VALUE);
	var response = await Statics.Http.PutAsync($"{baseUri}{WebUtility.UrlEncode(fileSystemName)}{path}{fileName}?resource=file", jsonContent);

	return response.IsSuccessStatusCode;
	}

	public async Task<bool> CreateFileAsync(string filesystem, string path,
	string fileName, Stream stream)
	{
	var operationResult = await this.CreateEmptyFileAsync(filesystem, path, fileName);
	if (operationResult)
	{
	var tokenInfo = await tokenProvider.GetAccessTokenV2EndpointAsync();
	var headers = Statics.Http.DefaultRequestHeaders;
	headers.Clear();
	headers.Add("Authorization", $"Bearer {tokenInfo.access_token}");
	headers.Add(API_VERSION_HEADER_NAME, API_VERSION_HEADER_VALUE);

	using (var streamContent = new StreamContent(stream))
	{
	var resourceUrl = $"{baseUri}{filesystem}{path}{fileName}?action=append&timeout={this.Timeout}&position=0";
	var msg = new HttpRequestMessage(new HttpMethod("PATCH"), resourceUrl);
	msg.Content = streamContent;
	var response = await Statics.Http.SendAsync(msg);

	//flush the buffer to commit the file
	var flushUrl = $"{baseUri}{filesystem}{path}{fileName}?action=flush&timeout={this.Timeout}&position={msg.Content.Headers.ContentLength}";
	var flushMsg = new HttpRequestMessage(new HttpMethod("PATCH"), flushUrl);
	response = await Statics.Http.SendAsync(flushMsg);

	return response.IsSuccessStatusCode;
	}
	}
	return false;
	}

	public async Task<bool> SetAccessControlAsync(string fileSystemName, string path, AclEntry[] acls)
	{
	var targetPath = $"{WebUtility.UrlEncode(fileSystemName)}{path}";
	var tokenInfo = await tokenProvider.GetAccessTokenV2EndpointAsync();

	var jsonContent = new StringContent(string.Empty);
	var headers = Statics.Http.DefaultRequestHeaders;
	headers.Clear();
	headers.Add("Authorization", $"Bearer {tokenInfo.access_token}");
	headers.Add(API_VERSION_HEADER_NAME, API_VERSION_HEADER_VALUE);
	headers.Add(ACK_HEADER_NAME, string.Join(',', acls.Select(a => a.ToString()).ToArray()));
	var response = await Statics.Http.PatchAsync($"{baseUri}{targetPath}?action=setAccessControl", jsonContent);

	return response.IsSuccessStatusCode;
	}

	var tokenProvider = new OAuthTokenProvider(tenantId, clientId, secret, scope);
	var hdfs = new FileSystemApi(storageAccountName, tokenProvider);

	var response = hdfs.CreateFileSystemAsync(fileSystemName).Result;

	hdfs.CreateDirectoryAsync(fileSystemName, "/demo").Wait();
	hdfs.CreateEmptyFileAsync(fileSystemName, "/demo/", "example.txt").Wait();

	var stream = new FileStream(@"C:\temp.txt", FileMode.Open, FileAccess.Read);
	hdfs.CreateFileAsync(fileSystemName, "/demo/", "mytest.txt", stream).Wait();

	var acls = new AclEntry[]
	{
	new AclEntry(
	AclScope.Access,
	AclType.Group,
	"2dec2374-3c51-4743-b247-ad6f80ce4f0b",
	(GrantType.Read \| GrantType.Execute)),
	new AclEntry(
	AclScope.Access,
	AclType.Group,
	"62049695-0418-428e-a5e4-64600d6d68d8",
	(GrantType.Read \| GrantType.Write \| GrantType.Execute)),
	new AclEntry(
	AclScope.Default,
	AclType.Group,
	"62049695-0418-428e-a5e4-64600d6d68d8",
	(GrantType.Read \| GrantType.Write \| GrantType.Execute))
	};
	hdfs.SetAccessControlAsync(fileSystemName, "/", acls).Wait();

	private static void Server()
	{
	try
	{
	Console.Clear();
	// use WcfService.Tcp for NetTcp binding or WcfService.Http for WSHttpBinding

	var hosts = WcfService.DefaultFactory.CreateServers(
	new List<Type> { typeof(MyService) },
	(t) => { return t.Name; },
	(t) => { return typeof(IWcf); },
	"WcfServices",
	port,
	(sender, exception) => { Trace.Write(exception); },
	(msg) => { Trace.Write(msg); },
	(msg) => { Trace.Write(msg); },
	(msg) => { Trace.Write(msg); });

	Console.WriteLine($"Server started …. {NetworkUtils.GetLocalIPAddress()}:{port}");
	}
	catch (Exception ex)
	{
	Console.WriteLine(ex);
	}
	Console.ReadLine();
	}

	using (var wcf =
	WcfService.DefaultFactory.CreateChannel<IWcf>(IP, port, (t) => { return "MyService"; }, "WcfServices"))
	{
	var count = 100;
	var runAgain = false;
	do
	{
	Console.Clear();
	Console.WriteLine("How many iterations you want to perform?");
	if (!Int32.TryParse(Console.ReadLine(), out count))
	{
	count = 10;
	}

	perfCount = count;
	instance = wcf;

	//BenchmarkDotNet.Running.BenchmarkRunner.Run<Program>();
	NetworkThroughputBenchmark.Perform(new Action(new Program().Execute), perfCount);

	Console.ResetColor();
	Console.WriteLine("Press [A] to run again…");
	runAgain = (Console.ReadKey(intercept: true).Key == ConsoleKey.A);

	} while (runAgain);
	}

	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)

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()");
	}
	}

Background

About Azure Data Lake Store Gen 2

Create a Service Principal

Add required permissions

Access Token from Azure Active Directory

Creating ADLS Gen 2 REST client

Data Lake ACLs and POSIX permissions

Conclusion

Background

Environment and Hardware specs

Windows Container: Quick refresh

Measure/benchmark

WCF TCP/IP Channel

WCF server

WCF Client

gRPC

Web Sockets

Unix Domain Sockets

Summary

TL;DR:

Background

Blazor meets Terraform

Resource Discovery in Terraform Providers

Building Infrastructure as code

Project

Blueprint

Environments

Deployments

Terraform State

Terraform plan

Terraform apply

Final thoughts

Pre-requisite

Publishing in Docker

Azure Pipeline as code

Creating the pipeline

Azure Pipeline Secret variables

The Solution Design

Azure Functions

Prerequisites

Configuration

Hacking the function

Final thoughts!

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

Quick start guide

Install

Implemention guide

Write Side – Event Sourcing

Aggregate

Read Side – Materialized Views

How to listen for events?

Conclusion

Background

Spin up VMs on Azure

Install RabbitMQ

Install Management portal for RabbitMQ

Configuring the cluster

Conclusion