DEV Community: Deyan Petrov

Claude and Kusto or MCP and CQRS+

Deyan Petrov — Sat, 28 Mar 2026 00:25:01 +0000

TLDR; Connecting Claude with Azure Data Explorer (Kusto) gives you lightweight natural language query/reporting capabilities. The Remote MCP Server gets exposed as another protocol on the existing CQRS+ based microservices.

Introduction

The thirst for data and analytics in a company is insatiable, and many non-technical users try to learn SQL or specific dashboarding tools just to be able to answer their own data-related questions without waiting for a specialist to find time. Ideally though the business users would like to ask those questions in natural language, the same way how they are already shooting questions to an AI/LLM.

This post is all about satisfying data-hungry non-technical users by connecting Claude Desktop with a database called Azure Data Explorer by the means of a Remote MCP Server:

Diagram 1: Claude Desktop + Kusto Remote MCP Server + Azure Data Explorer (Kusto)

Additionally, the post discusses the architectural decision where to place this so-called MCP Tool within a CQRS+ Microservice Architecture.

Claude

Claude has turned nowadays (2026) into one of the best AI tools for coders (Claude Code) and non-coders (Claude Desktop/Cowork). Even though many organizations are still held up in the ecosystems of their existing vendors (e.g. Copilot in the Microsoft Office 365/Azure world or Amazon Q in the AWS world) more and more companies reach out to the better and more ergonomic Claude. For example, Claude Desktop has this amazing split-screen view where query remains on the left-hand side, and a visualization ("Interactive Artefact") is displayed on the right-hand side.

Azure Data Explorer (aka Kusto or ADX)

Kusto/ADX is an amazing, lean and inexpensive cloud-hosted (only Azure) column store database, designed for very fast reads, that nobody knows of. It reminds me a bit of ClickHouse, however it does not use SQL, but KQL or Kusto Query Language, which looks very attractive for functional developers like me (F#) as it is similar to code pipelines:

Transactions
| where CreatedOn >= last3_start and CreatedOn <= last3_end
| project TransactionId, Amount, Type
| order by CreatedOn desc

Kusto is the engine powering all Microsoft Azure logging systems and even serves as the core of many other products like Azure AI Foundry, Microsoft Sentinel, Azure Application Insights, etc.

Natural Language Query

Of course it is possible to write SQL or in case of Kusto KQL to get data, however that requires getting acquainted with some quirky keywords and symbols. Asking questions in natural language and getting the answers also in natural language + visualization is what the non-technical people actually want. LLMs like Claude are amazing at generating code as well as SQL/KQL queries based on English questions, but to copy and paste those queries into a database query editor is so 2024-ish ;)

MCP Server

MCP Servers come to the rescue, as they make a connection between the AI Tool and many systems like your database (e.g. Kusto). MCP Servers come in 2 kinds:

Local MCP Server (aka stdio) - requires installing some software/program (e.g. python app) on your laptop, and the AI Tool is invoking this program locally. Your local config/permissions/logged in account are being used
Remote MCP Server (over HTTP) - comes ready-to-use out-of-the-box, after some initial authentication/authorization via oAuth.

Ideally the database vendor would offer an integrated Remote MCP Server. As a matter of fact I did have a chat some time ago with a Product Manager in the ADX/Kusto team (as part of a support ticket discussion, I have no connections to MS) who told me that they are working on such, however it seems that one would be more focused on bigger/enterprise products like Azure AI Foundry, instead of the relatively low-level Kusto.

Alternatively, such a connector can be built in a way similar to building a REST Web Api. The MCP Protocol is based on JSON RPC and is relatively straightforward calling functions/methods over HTTP(s), the only complication is related to oAuth.

CQRS+, Microservices, and how MCP fits in

In my older post I explain CQRS and CQRS+ in the context of microservices. In a nutshell, a software system is split into "services" based on functional areas or domains with their own context/language/terminology (aka bounded contexts), and those are subdivided into physical "microservices" (= standalone running processes) based on technical aspects like command vs query handling vs event publishing vs event handling, etc. So the question that pops up immediately is where does MCP fit in this picture?

The first idea that comes to mind is to create a separate MCP microservice, responsible for implementing the MCP protocol.

Diagram 2: Xyz Service and its microservices, including a standalone MCP microservice

However, MCP is nothing more than another protocol for exposing the same command-handling or query-handling logic that you already have in your existing microservices. So the better choice seems to be that MCP co-exists with REST within the same microservices. This means that in the example below where a Remote MCP Server for Kusto is created for the sake of querying only the server would be co-hosted in the same microservice responsible for Query Handling within the Reporting service/bounded context:

Diagram 3: Xyz Service and its microservices, with a MCP server integrated into the existing microservices

Sample Implementation

The sample below is using .NET and F#, my favorite combo, but can be obviously written in any tech stack. Actually Microsoft's Local MCP Server for Fabric RTI written in Python "served as inspiration" to Claude Code with which I wrote 75% of the below code ;)

This is what the main function looks like:

/// Application entry point: Key Vault decryption, host builder composition, and startup.
module Program

open System.Threading
open Microsoft.Extensions.Hosting
open Framework.AzureKeyVault.Environment
open Framework.Hosting.HostBuilder
open Api.Wiring

module McpDI = MCP.DependencyInjection

// this has to be first, otherwise modules initialize with env vars which have not been decrypted yet!
Environment.overwriteEnvironmentVariablesFromKVRef () |> Async.RunSynchronously

let webApis =
    [ WebApi.health

      WebApi.OAuth.register
      WebApi.OAuth.authorize
      WebApi.OAuth.token
      WebApi.OAuth.wellKnownProtectedResource
      WebApi.OAuth.wellKnownAuthServer ]

let mcpTools = [ McpTools.executeKustoQuery ]

[<EntryPoint>]
let main argv =
    let builder =
        createDefaultBuilder argv BackgroundServiceExceptionBehavior.StopHost
        |> McpDI.configureMcpServices mcpTools
        |> configureWebHost webApis McpDI.OAuth.requireBearerToken (Framework.Mcp.Hosting.mapMcpEndpoints "")

    use tokenSource = new CancellationTokenSource()
    use host = builder.Build()

    printfn "MCP Server running at %s" EnvVars.baseUrl

    host.RunAsync(tokenSource.Token) |> Async.AwaitTask |> Async.RunSynchronously

    0

The actual definition of the MCP Server methods/functions called tools is in Api.Wiring.fs:

module McpTools =
    let executeKustoQuery: McpServerToolDef =
        { Name = "execute_kusto_query"
          Description =
            "Executes a KQL query against the configured Azure Data Explorer (Kusto) cluster and database. Returns query results as a JSON array. The query runs under the authenticated user's identity."
          ReadOnly = true
          Destructive = false
          ExecuteOperation = Func<string, Task<string>>(McpDI.McpTools.executeKustoQuery) }

How MCP and REST co-exist in the same Microservice

The key insight is that MCP is literally just another set of routes alongside REST — like GraphQL or gRPC. Rather than a single monolithic setup function, the codebase separates concerns into composable pipeline stages.

Step 1 — Register MCP tools in the DI container (`Mcp.Hosting.fs`):

  let configureMcpServices (mcpTools: McpServerToolDef list) (builder: IHostBuilder) : IHostBuilder
  =                                                                                                 
      builder.ConfigureServices(fun _ services ->                                                
          services.AddHttpContextAccessor() |> ignore                                               

          let tools =
              mcpTools                                                                              
              |> List.map (fun toolDef ->                                                        
                  let options =                                                                     
                      McpServerToolCreateOptions(
                          Name = toolDef.Name,
                          Description = toolDef.Description,
                          ReadOnly = Nullable toolDef.ReadOnly,
                          Destructive = Nullable toolDef.Destructive                                
                      )

                  McpServerTool.Create(toolDef.ExecuteOperation, options))                       
              |> Array.ofList                                                                       

          services.AddMcpServer().WithHttpTransport().WithTools(tools) |> ignore)

Step 2 — Configure the web host generically (`Hosting.HostBuilder.fs`). This function knows nothing about MCP — it just wires up REST endpoints, a middleware hook, and an additional-endpoints hook:

  let configureWebHost                                                                           
      (webApiDefs: WebApiDef list)                                                                  
      (configureMiddleware: IApplicationBuilder -> unit)
      (configureAdditionalEndpoints: IEndpointRouteBuilder -> unit)                                 
      (builder: IHostBuilder)
      : IHostBuilder =                                                                              
      builder.ConfigureWebHostDefaults(fun webBuilder ->                                         
          webBuilder.Configure(fun context app ->                                                   
              app.UseRouting() |> ignore                                                            

              configureMiddleware app                                                               

              app.UseEndpoints(fun endpoints ->                                                     
                  webApiDefs
                  |> List.iter (fun webApiDef ->
                      endpoints
                          .MapMethods(webApiDef.Path, [ webApiDef.Method.ToString() ],
  webApiDef.ExecuteOperation)                                                                       
                          .AllowAnonymous()
                      |> ignore)                                                                    

                  configureAdditionalEndpoints endpoints)                                           
              |> ignore)

          |> ignore)

Step 3 — Map MCP endpoints via a one-liner that plugs into the `configureAdditionalEndpoints` hook:

  let mapMcpEndpoints (basePath: string) (endpoints: IEndpointRouteBuilder) =                       
      endpoints.MapGroup($"%s{basePath}/mcp").MapMcp() |> ignore

Composition in Program.fs — everything comes together as a pipeline:

  let builder =                                                                                     
      createDefaultBuilder argv BackgroundServiceExceptionBehavior.StopHost                         
      |> McpDI.configureMcpServices mcpTools                                                        
      |> configureWebHost webApis McpDI.OAuth.requireBearerToken
  (Framework.Mcp.Hosting.mapMcpEndpoints "")

The single host now serves:

GET /health — REST health check
GET /.well-known/*, /oauth/* — REST OAuth endpoints
POST /mcp — MCP JSON-RPC transport

Notice that configureWebHost has no dependency on MCP at all. MCP plugs in through the same configureAdditionalEndpoints hook that any other protocol (GraphQL, gRPC, SignalR) would use. Adding MCP to an existing service is just two pipeline stages — one for DI, one for routing — not a new deployment.

The OAuth 2.1 Proxy — The Hard Part

I mentioned above that the only complication with building a Remote MCP Server is "related to oAuth." Let me unpack that because it turned out to be a bigger than expected engineering challenge in this project.

The problem is a three-way mismatch between what Claude expects, what Azure Entra ID supports, and what is actually needed:

Claude expects Dynamic Client Registration (DCR) per RFC 7591 — it wants to call a /oauth/register endpoint and get back a client_id and client_secret. Azure Entra ID does not support DCR. App registrations are created in the Azure Portal or via scripts, not at runtime.
Claude authenticates to the MCP server — so the OAuth scope it requests is scoped to the MCP server itself. But what is needed is the resulting token to be scoped to Kusto, not to the MCP server, because the user's own token will be passed directly to Azure Data Explorer.
Claude sends a dummy client_secret — the one it received during DCR. But Entra ID needs the real client_secret from the app registration.

The solution is an OAuth Proxy — our MCP server impersonates an OAuth authorization server by implementing five endpoints that intercept, rewrite, and forward Claude's OAuth requests to Entra ID.

Discovery: "I am your OAuth server"

When Claude first connects and gets a 401, it looks for .well-known metadata. 2 standard RFC endpoints are served that point Claude back at the Kusto MCP Server:

let wellKnownAuthServer (logInfo: string -> unit) (ctx: HttpContext) =                                                   
  async {                                                                                                                
      let response =
          {| issuer = sprintf "https://login.microsoftonline.com/%s/v2.0" EnvVars.tenantId                               
             authorization_endpoint = sprintf "%s/oauth/authorize" EnvVars.baseUrl                                       
             token_endpoint = sprintf "%s/oauth/token" EnvVars.baseUrl                                                 
             registration_endpoint = sprintf "%s/oauth/register" EnvVars.baseUrl                                         
             scopes_supported =                                                                                          
              [| EnvVars.ADX.connectionString + "/.default"                                                            
                 "openid"; "profile"; "offline_access" |]                                                                
             response_types_supported = [| "code" |]                                                                     
             grant_types_supported = [| "authorization_code"; "refresh_token" |]                                         
             code_challenge_methods_supported = [| "S256" |] |}                                                          

      ctx.Response.StatusCode <- 200                                                                                     
      do! ctx.Response.WriteAsJsonAsync(response) |> Async.AwaitTask                                                   
  }

Claude now thinks our server is the authorization server itself — all OAuth requests will come to us.

Mock DCR: "Sure, here's your client_id"

Claude calls POST /oauth/register expecting DCR. The Kusto MCP Server accepts its request, echoes back its redirect_uris (per RFC 7591), and returns the pre-registered Entra ID client_id along with a dummy client_secret that Claude will use later — but which will be thrown away:

let register (logInfo: string -> unit) (ctx: HttpContext) =
  async {
      let! body = ctx.Request.ReadFromJsonAsync<JsonElement>().AsTask() |> Async.AwaitTask

      let redirectUris =                                                                                                 
          match body.TryGetProperty("redirect_uris") with                                                                
          | true, uris -> uris.EnumerateArray() |> Seq.map (fun u -> u.GetString()) |> Array.ofSeq                       
          | _ -> [||]                                                                                                    

      let dummySecret = Guid.NewGuid().ToString("N")                                                                     

      let response =
          {| client_id = EnvVars.OAuth.clientId
             client_secret = dummySecret                                                                                 
             grant_types = [| "authorization_code"; "refresh_token" |]
             redirect_uris = redirectUris |}                                                                             

      ctx.Response.StatusCode <- 201
      do! ctx.Response.WriteAsJsonAsync(response) |> Async.AwaitTask                                                     
  }

Scope Rewriting: "You think you're authenticating to me, but you're actually authenticating to Kusto"

When Claude calls GET /oauth/authorize, the call is intercepted and the scope parameter gets rewritten before redirecting to Entra ID. Claude asked for a scope targeting our MCP server, but that is replaced with the Kusto cluster scope. This is the key trick — the access token that comes back will have the Kusto cluster as its audience, not the MCP server:

let authorize (logInfo: string -> unit) (ctx: HttpContext) =
  async {
      let query = ctx.Request.Query

      // The magic: rewrite scope to target Kusto
      let scope =                                                                                                        
          sprintf "%s/.default openid profile offline_access" EnvVars.ADX.connectionString                               

      let entraAuthorizeUrl =                                                                                            
          sprintf "https://login.microsoftonline.com/%s/oauth2/v2.0/authorize" EnvVars.tenantId                        

      // Pass through PKCE (code_challenge), state, and other params                                                     
      let queryParts =
          [ "client_id", EnvVars.OAuth.clientId                                                                          
            "response_type", "code"                                                                                    
            "redirect_uri", query.["redirect_uri"].ToString()                                                            
            "scope", scope                                                                                             
            "state", query.["state"].ToString()                                                                          
            "response_mode", "query" ]                                                                                 
          @ (if String.IsNullOrEmpty(query.["code_challenge"].ToString()) then []                                        
             else [ "code_challenge", query.["code_challenge"].ToString()
                    "code_challenge_method", "S256" ])                                                                   
          |> List.map (fun (k, v) -> sprintf "%s=%s" (Uri.EscapeDataString k) (Uri.EscapeDataString v))                
          |> String.concat "&"                                                                                           

      ctx.Response.Redirect(sprintf "%s?%s" entraAuthorizeUrl queryParts)                                                
  }

The user now sees the familiar Microsoft login screen. After authenticating, Entra ID redirects back to Claude's callback with an authorization code.

Credential Injection: "Let me fix that secret for you"

Claude exchanges the auth code by calling POST /oauth/token — but it sends the dummy client_secret from the mock DCR. The MCP Server strips Claude's credentials, injects the real ones from our Entra ID app registration, and forwards the request to Entra ID:

let token (logInfo: string -> unit)                                                                                      
        (callOverHttp: string -> IDictionary<string, string> -> Async<int * string>)                                     
        (ctx: HttpContext) =                                                                                             
  async {                                                                                                                
      let! form = ctx.Request.ReadFormAsync() |> Async.AwaitTask                                                         
      let entraTokenUrl =                                                                                                
          sprintf "https://login.microsoftonline.com/%s/oauth2/v2.0/token" EnvVars.tenantId

      let formData = Dictionary<string, string>() :> IDictionary<string, string>                                       

      // Copy everything EXCEPT client credentials                                                                       
      for kvp in form do
          if kvp.Key <> "client_secret" && kvp.Key <> "client_id" then                                                   
              formData.[kvp.Key] <- kvp.Value.ToString()                                                                 

      // Inject real credentials                                                                                         
      formData.["client_id"] <- EnvVars.OAuth.clientId                                                                 
      formData.["client_secret"] <- EnvVars.OAuth.clientSecret                                                           

      let! statusCode, responseBody = callOverHttp entraTokenUrl formData

      ctx.Response.StatusCode <- statusCode                                                                            
      ctx.Response.ContentType <- "application/json"
      do! ctx.Response.WriteAsync(responseBody) |> Async.AwaitTask                                                       
  }

Entra ID returns an access_token scoped to the Kusto cluster. Claude stores it and sends it as a Bearer token on every subsequent MCP request. Our MCP server then passes that same token directly to Azure Data Explorer — it never sees or stores the user's credentials.

Per-User Token Passthrough

Some tutorials and examples for MCP servers that connect to databases use a shared service principal — a single identity with broad access that executes all queries on behalf of all users. This is the easy path, but it has serious drawbacks:

every user sees the same data regardless of their actual permissions,
audit logs show a generic service account instead of the real person, and
a bug or prompt injection in the LLM could expose data the user should never have access to.

Here a different approach is taken. The user's own OAuth token — the one they obtained by logging into Microsoft Entra ID through the OAuth proxy flow described above — is passed directly to Azure Data Explorer:

let executeKustoQuery                                                                                                    
  (logError: string -> exn -> unit)
  (serviceUrl: string)                                                                                                   
  (getBearerToken: unit -> string)
  (query: string)                                                                                                        
  : Async<string> =                                                                                                    
  async {                                                                                                                
      match QueryValidation.validate query with
      | Error msg -> return sprintf "Query rejected: %s" msg                                                             
      | Ok() ->                                                                                                          
          try
              let userToken = getBearerToken ()                                                                          

              let csb =                                                                                                  
                  KustoConnectionStringBuilder(serviceUrl)
                      .WithAadUserTokenAuthentication(userToken)                                                         

              use client = KustoClientFactory.CreateCslQueryProvider(csb)                                                
              let! results = executeQuery<JsonObject> client (TimeSpan.FromSeconds 30L) query []                         
              return JsonSerializer.Serialize(results)                                                                   
          with ex ->                                                                                                   
              logError "Error executing Kusto query" ex                                                                  
              return "Error executing query"                                                                           
  }

The critical line is .WithAadUserTokenAuthentication(userToken). This is not a shared service principal — it is the actual user's token. Azure Data Explorer sees the real caller. This means:

Database roles apply per-user. If a user only has viewer access to the Sales database, they cannot query Engineering. This is enforced directly by Kusto.
Row-level security policies work. If the company has RLS policies that restrict sales reps to seeing only their own region's data, those policies apply. The LLM cannot bypass them because the token identifies the actual user.
Audit logs show who really queried. When compliance or security wants to know who ran a particular query, the answer is "Alice from Finance" — not "mcp-server-service-principal."

The trade-off is that tokens expire (typically after one hour), so Claude needs to handle token refresh. The OAuth proxy handles refresh_token grants through the same credential-injection pattern as the initial token exchange, so this is transparent to the user.

The Full Flow

Here is the complete OAuth dance visualized:

sequenceDiagram
      participant C as Claude Desktop
      participant M as MCP Server
      participant E as Entra ID
      participant K as Kusto                                                                                                 

      C->>M: POST /mcp                                                                                                       
      M-->>C: 401 (WWW-Authenticate → /.well-known/*)                                                                     
      C->>M: GET /.well-known/oauth-*                                                                                        
      M-->>C: {endpoints point to self}
      C->>M: POST /oauth/register                                                                                            
      M-->>C: 201 {client_id, dummy secret}                                                                               
      C->>M: GET /oauth/authorize                                                                                            
      M->>E: 302 (scope rewritten to kusto/.default)                                                                      
      E-->>C: 302 + auth code (user logs in)                                                                                 
      C->>M: POST /oauth/token (dummy secret)                                                                                
      M->>E: POST /token (real secret injected)                                                                              
      E-->>M: access_token (scoped to Kusto)                                                                                 
      M-->>C: access_token                                                                                                   
      C->>M: POST /mcp (Bearer token)                                                                                        
      M->>K: KQL query (user's own token)                                                                                    
      K-->>M: results                                                                                                        
      M-->>C: query results

This flow is reusable for any scenario where you need Claude to authenticate against an identity provider that does not support DCR, while obtaining tokens scoped to a backend service different from the MCP server itself.

Security — Multi-Layer Defense

When an LLM generates database queries, you have to assume it will occasionally generate something you do not want executed. The solution is defense in depth — multiple independent validation layers, each catching different classes of problems.

Layer 1: Token Pre-Validation (Middleware)

Before any MCP tool sees the request, middleware validates the JWT token's structure, expiration, and issuer. This is not cryptographic verification — Azure Data Explorer does that later. This is an early-rejection layer that avoids wasting Kusto resources on obviously invalid tokens:

let validateBearerToken (expectedTenantId: string) (token: string) : Result<unit, string> =
  let parts = token.Split('.')                                                                                           

  if parts.Length <> 3 then                                                                                              
      Error "Malformed token"                                                                                          
  else                                                                                                                   
      let payloadBytes = Base64UrlEncoder.DecodeBytes(parts.[1])                                                       
      let payload = Encoding.UTF8.GetString(payloadBytes)                                                                
      use doc = JsonDocument.Parse(payload)
      let root = doc.RootElement                                                                                         

      // Check expiration (with 5-minute clock skew)                                                                     
      // Check issuer contains expected tenant ID                                                                        
      // → Ok() or Error "reason"

Expired token? Rejected at the door. Wrong tenant? Never reaches Kusto. No network call needed.

Layer 2: Whitelist-Based Query Validation

This is the most important layer for LLM-generated queries. Instead of blacklisting dangerous patterns (a losing game), the MCP Server whitelists exactly which KQL operators and plugins are allowed.

Only 44 safe, read-only tabular operators are permitted (where, project, summarize, join, extend, render, etc.). For evaluate plugins, only 17 safe ones are allowed — critically blocking python, r, sql_request, http_request, and anything else that could reach outside the Kusto cluster.

Several classes of queries are blocked outright:

let private blockedSourcePatterns =
    [ Regex(@"\bexternaldata\b", RegexOptions.IgnoreCase ||| RegexOptions.Compiled), "externaldata is not allowed"
      Regex(@"\bexternal_table\s*\(", RegexOptions.IgnoreCase ||| RegexOptions.Compiled),
      "external_table() is not allowed"
      Regex(@"\bcluster\s*\(", RegexOptions.IgnoreCase ||| RegexOptions.Compiled),
      "cross-cluster queries are not allowed"
      Regex(@"\bdatabase\s*\(", RegexOptions.IgnoreCase ||| RegexOptions.Compiled),
      "cross-database queries are not allowed" ]

And any statement starting with . (a Kusto management command) is rejected immediately. When a query is rejected, the tool returns the validation error directly to Claude as text — no query ever reaches Kusto. Claude then typically understands what went wrong and rewrites the query using only allowed operators.

Layer 3: Azure Data Explorer Server-Side Enforcement

Even after passing both previous layers, Kusto enforces its own security: full cryptographic token validation, database role enforcement, row-level security policies, and audit logging with the real user's identity.

No single layer is sufficient on its own. The middleware catches bad tokens cheaply. The query validation catches dangerous queries before they leave our server. And Kusto enforces the actual data access permissions. A failure in any one layer is caught by the others.

Functional Dependency Injection

To the surprise of e.g. C# developers this F# project has no IoC container, no constructor injection, no interfaces. Dependencies are wired through plain functions and partial application.

Each module in DependencyInjection.fs creates partially applied functions where dependencies are "baked in" at startup. For example, the executeKustoQuery implementation accepts 4 parameters — log, getAuthorizationHeader, createAdxClient, and query — but the DI module partially applies the first 3, leaving only query: string -> Task<string>, exactly what the MCP framework needs:

  module McpTools =
      let private getAuthorizationHeader = Framework.Http.getAuthorizationHeader accessor

      let private createAdxClient (userToken: string) : ICslQueryProvider =                         
          KustoConnectionStringBuilder(EnvVars.ADX.serviceUrl)
              .WithAadUserTokenAuthentication(userToken)                                            
          |> KustoClientFactory.CreateCslQueryProvider                                           

      let executeKustoQuery (query: string) : Task<string> =                                        
          MCP.Api.Functions.McpTools.executeKustoQuery log getAuthorizationHeader createAdxClient
  query                                                                                             
          |> Async.StartAsTask

This is the functional equivalent of constructor injection, but without the ceremony. Testing benefits similarly: you call the underlying function with fake loggers, test URLs, and stub token providers without requiring mocking frameworks.

A notable detail: getAuthorizationHeader is passed as a function (unit -> string) rather than invoked eagerly. It captures HttpContextAccessor, reading the current request's Authorization header only at query execution time — after validation passes. Similarly, createAdxClient is a function (string -> ICslQueryProvider), not a pre-built client — each query gets a fresh client authenticated with the current user's token. This approach avoids threading infrastructure types through domain logic.

The same pattern scales to the OAuth handlers — each one partially applies config (tenant ID, client ID, secrets, scopes) and leaves just HttpContext -> Async<unit>.

Local Development with Cloudflare Tunnel

Remote MCP servers require public URLs for OAuth callbacks. During local development, https://localhost:5001 is not reachable from the internet. While ngrok was traditionally used, ngrok's free tier now shows a browser interstitial page that breaks OAuth redirects.

Cloudflare Tunnel offers a superior alternative: free, no signup required, and no interstitial blocking. The project includes two scripts that automate the setup.

cloudflared_tunnel.sh is a reusable helper that starts a tunnel and returns the public URL. It auto-installs cloudflared if needed (via Homebrew on macOS, direct download on Linux), handles HTTPS local targets with --no-tls-verify, and outputs the tunnel URL and PID on stdout for the caller to parse and clean up:

  output=$(./cloudflared_tunnel.sh --url https://localhost:5001)                                    
  TUNNEL_URL=$(echo "$output" | head -1)
  CLOUDFLARED_PID=$(echo "$output" | tail -1)                                                       
  trap "kill $CLOUDFLARED_PID 2>/dev/null" EXIT

dotnet_run.sh orchestrates local startup. With the --mcp flag, it:

Loads environment variables from launchSettings.json
Reads the listen URL from the LocalDev launch profile
Starts a Cloudflare Tunnel via cloudflared_tunnel.sh and exports MCP_BASE_URL with the tunnel address
Launches the .NET server

  ./dotnet_run.sh --mcp true
  # Loading environment variables from launchSettings.json...                                       
  # Starting cloudflared tunnel to https://localhost:5001...
  # cloudflared tunnel: https://random-words.trycloudflare.com -> https://localhost:5001            
  # MCP endpoint: https://random-words.trycloudflare.com/mcp

Without --mcp, it simply loads the env vars and runs the server — useful when working on REST-only features.

The script prints the MCP endpoint URL to add as a connector in Claude Desktop:

  MCP endpoint: https://random-words.trycloudflare.com/mcp

The server differentiates between listen address (MCP_LISTEN_URL on localhost, defaulting to https://localhost:5001) and public address (MCP_BASE_URL via tunnel, defaulting to MCP_LISTEN_URL if unset). OAuth endpoints advertise the tunnel URL in .well-known metadata while actually listening locally. Claude communicates with the tunnel URL, which routes to localhost, which redirects to Entra ID, completing the OAuth flow.

Tunnel URLs regenerate on restart — this is acceptable since Claude rediscovers endpoints via .well-known on each connection.

Conclusion

What started as a simple idea — "let business users ask questions about their data in natural language" — turned into an interesting exercise in protocol bridging, security layering, and architectural decisions.

The key takeaways:

MCP is just another protocol. It sits alongside REST in the same microservice, shares the same middleware and authentication, and does not deserve its own deployment boundary. In a CQRS+ architecture, MCP tools belong in the microservice that already handles the relevant queries or commands.
The OAuth proxy pattern is reusable. If you need Claude (or any RFC-compliant OAuth client) to authenticate against Azure Entra ID — or any identity provider that does not support Dynamic Client Registration — the proxy approach of mock DCR + scope rewriting + credential injection works generically. Swap out the Kusto scope for a Microsoft Graph scope, and you have a different integration with the same plumbing.
Per-user token passthrough is worth the effort. Passing the user's own token to the database instead of using a shared service principal means that existing access controls, row-level security, and audit logging work without any additional code. The database already knows how to enforce permissions — let it.
Whitelist, do not blacklist, LLM-generated queries. When an LLM is writing your database queries, you cannot anticipate every dangerous pattern it might produce. Whitelisting allowed operators is a safer default — anything not explicitly permitted is rejected.
F# is a natural fit for this kind of work. Pipeline-style KQL queries, pipeline-style F# code, partial application for dependency injection, discriminated unions for validation results — the language aligns well with both the problem domain and the implementation patterns.

The complete source code is available on GitHub.

If you are building something similar for a different database or identity provider, the OAuth proxy and query validation patterns should translate directly — only the query language and connection builder change.

References

Azure Devops Pipeline with SonarQube Community in Docker and Playwright

Deyan Petrov — Sat, 08 Feb 2025 12:10:16 +0000

TLDR; Scanning projects with SonarQube Community and retrieving a couple of metrics and reports can be done easily with the below scripts.

Introduction

SonarQube is one of the most well-known static code analysis tools which is basically scanning source code for security issues. It does come with a "self-managed" Community version, which unfortunately does not contain out of the box any easy option to generate reports with the scan results. The "self-managed" Enterprise Edition, or the cloud Enterprise Edition do support reporting, however they come with the (IMHO pretty funny for 2025) "Talk to sales" label ;)
So the challenge I set for myself is how to set up (as a prototype, to get comfortable with SonarQube)

an automated AzureDevops pipeline
a manual bash script

which not only scan some typescript projects, but also scrape a couple of reports/screenshots, and all this without maintaining a fixed, always running server instance of SonarQube ..

Important Note: The (makefile.sh)[https://github.com/gitricko/sonarless/blob/main/makefile.sh] file from (Sonarless)[https://github.com/gitricko/sonarless] served as a basis for the below script/pipeline.

Run SonarQube Server in Docker

The first step is to run SonarQube server in Docker. The official SonarQube Docker images can be found in [Docker Hub)[https://hub.docker.com/_/sonarqube]. Also a dedicated docker network is created:

DOCKER_NETWORK_NAME="sonarqube"
DOCKER_SONAR_SERVER_INSTANCE_NAME="sonar-server"
DOCKER_SONAR_SERVER_INSTANCE_PORT="9234"
DOCKER_SONAR_SERVER_IMAGE="sonarqube:latest"

docker network create "${DOCKER_NETWORK_NAME}"

docker run -d --name "${DOCKER_SONAR_SERVER_INSTANCE_NAME}" \
    -p "${DOCKER_SONAR_SERVER_INSTANCE_PORT}:9000" \
    --network "${DOCKER_NETWORK_NAME}" \ 
    "${DOCKER_SONAR_SERVER_IMAGE}"

So the server has been started, but we need to await some time (< 1 minute) for the instance to really start. This is done by first trying to make an HTTP GET to the root URL, and then to the system status endpoint:

curl -k -s -o /dev/null -I -w "%{http_code}" -H 'User-Agent: Mozilla/6.0' "http://localhost:$DOCKER_SONAR_SERVER_INSTANCE_PORT"

curl -s "http://localhost:$DOCKER_SONAR_SERVER_INSTANCE_PORT/api/system/status" 2>/dev/null | jq -r '.status'

The above commands are wrapped in loops until successful result is received.

Reset admin password

Resetting the default admin password for the SonarQube server instance running in Docker seems to be important as otherwise the automated calls to the Web API or most importantly Web UI would require changing that password ...
The new password must "be at least 12 characters long","must contain at least one uppercase character" and "must contain at least one special character".
The password change is performed by invoking the /users/change_password operation.

curl -s -X POST -u "$USERNAME:$OLD_PASSWORD" \
    -d "login=$USERNAME&previousPassword=$OLD_PASSWORD&password=$PASSWORD" \
    "http://localhost:${DOCKER_SONAR_SERVER_INSTANCE_PORT}/api/users/change_password"

Create Project

Then a default project must be created. This is done by calling the /projects/create and /users/set_homepage APIs:

curl -s -u "$CREDENTIALS" -X POST "http://localhost:$DOCKER_SONAR_SERVER_INSTANCE_PORT/api/projects/create?name=$PROJECT_NAME&project=$PROJECT_NAME"

curl -s -u "$CREDENTIALS" -X POST "http://localhost:$DOCKER_SONAR_SERVER_INSTANCE_PORT/api/users/set_homepage?type=PROJECT&component=$PROJECT_NAME"

Start the Scanner using NPX

Before the scanner can be started, a token needs to be created so that the SonarQube scanner can connect to the server:

SONAR_TOKEN=$(curl -s -X POST -u "$CREDENTIALS" "http://localhost:$DOCKER_SONAR_SERVER_INSTANCE_PORT/api/user_tokens/generate?name=$(date +%s%N)" | jq -r .token)

Then running the scanner is just an npx command away, but note that it has to be run in the source code project folder:

npx -y sonarqube-scanner -Dsonar.host.url="http://localhost:$DOCKER_SONAR_SERVER_INSTANCE_PORT" -Dsonar.token="$SONAR_TOKEN" -Dsonar.projectKey="$PROJECT_NAME"

Get some metrics via Web API

After the scan has finished, we can collect a couple of numeric metrics from the standard/official Web API of the SonarQube (without any funky scraping):

curl -s -u "$CREDENTIALS" "http://localhost:${DOCKER_SONAR_SERVER_INSTANCE_PORT}/api/measures/component?component=${PROJECT_NAME}&metricKeys=bugs,vulnerabilities,code_smells,quality_gate_details,violations,duplicated_lines_density,ncloc,coverage,reliability_rating,security_rating,security_review_rating,sqale_rating,security_hotspots,open_issues" \
    | jq -r > $SONAR_METRICS_PATH

The output looks like this:


{
  "component": {
    "key": "PROJECT_NAME",
    "name": "PROJECT_NAME",
    "qualifier": "TRK",
    "measures": [
      {
        "metric": "coverage",
        "value": "x.0",
        "bestValue": false
      },
      {
        "metric": "reliability_rating",
        "value": "x.0",
        "bestValue": false
      },
      {
        "metric": "code_smells",
        "value": "12345",
        "bestValue": false
      },
      {
        "metric": "duplicated_lines_density",
        "value": "x.0",
        "bestValue": false
      },
      {
        "metric": "security_rating",
        "value": "x.0",
        "bestValue": true
      },
      {
        "metric": "violations",
        "value": "12345",
        "bestValue": false
      },
      {
        "metric": "quality_gate_details",
        "value": "{\"level\":\"OK\",\"conditions\":[],\"ignoredConditions\":false}"
      },
      {
        "metric": "security_hotspots",
        "value": "x",
        "bestValue": false
      },
      {
        "metric": "security_review_rating",
        "value": "x.0",
        "bestValue": false
      },
      {
        "metric": "sqale_rating",
        "value": "x.0",
        "bestValue": true
      },
      {
        "metric": "bugs",
        "value": "12345",
        "bestValue": false
      },
      {
        "metric": "ncloc",
        "value": "12345"
      },
      {
        "metric": "vulnerabilities",
        "value": "12345",
        "bestValue": true
      },
      {
        "metric": "open_issues",
        "value": "12345",
        "bestValue": false
      }
    ]
  }
}

The above is anonymized sample, so the real one has PROJECT_NAME, x and 12345 replaced by real values.

Capture screenshots as reports via Web UI with Playwright

In addition, a simple Playwright script is executed against the SonarQube server instance running in Docker to capture some screenshots as reports from the UI.

For that purpose the following script is executed:

tsc save_mhtml.ts
node save_mhtml.js --projectName $PROJECT_NAME --outputFolder $SONAR_OUTPUT_FOLDER

The playwright script itself goes through the following steps:

Start a browser, etc:

    const browser = await chromium.launch({ headless: true, slowMo: 0 }); // Set headless to false to show the browser window + slowMo to 5000 for example
    const context = await browser.newContext();
    const page = await context.newPage();

Navigate to SonarQube Server Web UI:

    await page.goto("http://localhost:9234/");

    await page.fill('input[name="login"]', 'admin');
    await page.fill('input[name="password"]', password);
    await page.getByRole('button', { name: 'Log in' }).click();

Click through some really nasty buttons which just stay in the way:

    await clickButtonIfVisible(page, 'I understand the risk');
    await clickButtonIfVisible(page, 'Later');
    await clickButtonIfVisible(page, 'Got it');
    await clickButtonIfVisible(page, 'Dismiss');

Finally scrape some web pages and store them as *.mhtml report files:

await gotoPageAndMakeMhtml(page, context, `http://localhost:9234/dashboard?branch=main&id=${projectName}&codeScope=overall`, `${outputFolder}/${projectName}_overview.mhtml`);

    await gotoPageAndMakeMhtml(page, context, `http://localhost:9234/project/issues?id=${projectName}&issueStatuses=OPEN%2CCONFIRMED`, `${outputFolder}/${projectName}_issues.mhtml`);

    await gotoPageAndMakeMhtml(page, context, `http://localhost:9234/security_hotspots?id=${projectName}`, `${outputFolder}/${projectName}_security_hotspots.mhtml`);

Check if there are security vulnerabilities

Finally, the script can check some of the metric values captured before, and fail if a threshold has been exceeded, e.g.:

vulnerabilities=$(cat $SONAR_METRICS_PATH | jq -r '.component.measures[] | select(.metric == "vulnerabilities") | .value')
if [[ "$vulnerabilities" -gt 0 ]]; then
    echo "Vulnerabilities found: $vulnerabilities"
    exit 1
else
    echo "No vulnerabilities found"
fi

There are of course a few waitForLoadStates and waitForTimeouts scattered in between, so the script does take about 1 minute in total.

Azure DevOps Pipeline, and the final scripts

The steps described above are wrapped up in 2 scripts:

Bash script for manual scanning a project from the terminal
Azure DevOps Pipeline for automatic scanning of a project, e.g. on a nightly schedule

The Azure DevOps pipeline performs some additional steps, but that depends on what is already pre-installed or not on your VM image, e.g.:

Install Docker
Install jq
Install NodeJS

The scripts and pipelines can be found at https://github.com/deyanp/azuredevops-sonarqube

References

https://github.com/gitricko/sonarless

Running and Approving Azure DevOps Pipelines from the terminal

Deyan Petrov — Sat, 08 Feb 2025 00:31:50 +0000

TLDR; This bash script will allow you to start and approve (multiple) Azure DevOps pipelines from the terminal.

Introduction

If you are using Azure DevOps Pipelines (similar to Github Actions) you might be going to the browser, navigating to the list of pipelines, selecting one, and clicking on the "Run pipeline" + "Run" buttons. If you want to start 10 pipelines (e.g. for 10 microservices) at once, you need to do the above clicking 10x2 times.

Once the builds have reached a stage guarded by an approval, you have to go into the build, click the "Review" + "Approve" buttons ... which would be another 10x2 clicks.

Wouldn't it be better to trigger the above from the terminal/CLI? You are happy to find the az pipelines run az cli extension command, however things become very tricky when it comes to the approvals ...

The rest of this post is referring to a pipeline with 3 Stages - Build, Deploy to test and Deploy to prod

The `az rest .. --resource ..` command

Before we get to solving the approval problem, there is one very useful command that will help us immensely - az rest. This command allows us to invoke various Azure DevOps (and probably other) APIs without any personal access tokens (PATs) or similar. The important attribute to pass is --resource "https://management.core.windows.net/", which reuses our Azure DevOps session.

For example, a much faster alternative to az pipelines run is the following command:

az rest --uri "https://dev.azure.com/$org/$project/_apis/pipelines?api-version=7.1" --resource "https://management.core.windows.net/"

Pretty neat solution to the generally very nasty authentication topic, which usually requires some awkward token handling!

Resource Model and Relations, or how to find the Approval

In order to implement the approval action, one has to first understand how to get to it, which requires navigation through a couple of Azure DevOps resources, namely:

Pipeline
Last Build
Build Timeline and 3 specific records
Approval

Pipeline

So first we find the pipelineId by pipeline name:

pipelineId=$(az rest --uri "$adoBaseUrl/pipelines?api-version=7.1" --resource $resource | jq -c ".value.[] | select( .name == \"$pipeline\" ).id")

where adoBaseUrl=https://dev.azure.com/$org/$project/_apis/ and resource=https://management.core.windows.net/

Last Build

Next we find the last build for this pipeline:

build=$(az rest --uri "$adoBaseUrl/build/builds?api-version=7.1&definitions=$pipelineId&queryOrder=startTimeDescending&\$top=1" --resource $resource)

and from it we extract the timeline url:

timelineUrl=$(echo $build | jq -r '.value[0]._links.timeline.href')

Build Timeline

Getting the contents of the pipeline is easy:

timeline=$(az rest --uri "$timelineUrl" --resource $resource)

however it contains tons of records - depending on the pipeline definition, each one having parent record. We need to find the record corresponding to the approval stage, and for that we need to traverse (at least in my particular case) 3 records:

deploy Stage record
Checkpoint record
Checkpoint.Approval record

This is done in the following way:

deployStageRecordId=$(echo $timeline | jq -r ".records[] | select( .identifier == \"$deployStageRecordIdentifier\" and .type == \"Stage\" ).id")

checkpointRecordId=$(echo $timeline | jq -r ".records[] | select( .parentId == \"$deployStageRecordId\" and .type == \"Checkpoint\" ).id")

approvalId=$(echo $timeline | jq -r ".records[] | select( .parentId == \"$checkpointRecordId\" and .type == \"Checkpoint.Approval\" ).id")

The last one gives us the approvalId!

Approval

Once we have the approvalId we can get for example its status, so that we approve only if the status is pending:

approvalStatus=$(az rest --uri "$adoBaseUrl/pipelines/approvals/$approvalId?api-version=7.1" --resource $resource | jq -r ".status")

Finally, approve!

Once the approvalId is found, the following POST command must be issued to Review/Approve it:

az rest --method PATCH --uri "$adoBaseUrl/pipelines/approvals?api-version=7.1" --resource $resource --body "[ { \"approvalId\":\"$approvalId\", \"comment\":\"cli\", \"status\":\"approved\" } ]"

The full script can be found at https://github.com/deyanp/adopipelines/blob/main/pipeplines.sh

References

F# App Stub for AKS hosting v2 (without Azure WebJobs SDK fluff)

Deyan Petrov — Mon, 25 Mar 2024 18:43:43 +0000

TLDR; Use a standard .NET 8+ host for your microservices running for example in AKS without any Azure WebJobs SDK syntactic sugar or similar fluff/magic¹.

Introduction

We started originally now almost 5 years ago with .NET/F# Azure Functions apps, and we had pretty bad experiences with that setup, the biggest issues being low node app density, high costs and too much bloat from MS frameworks.

Soon after that we migrated to AKS (managed Kubernetes on Azure similar to EKS and GKE), but to keep migration efforts reasonable we re-used the WebJobs SDK, which is a building stone for the Azure Functions SDK. More than 2 years ago I wrote an article about the app stub we were using.

Everything was running relatively fine until recently when we had to improve our local development experience, and we needed a way to change the low-level workings of some of our code related to Azure Event Hubs and similar. That was the trigger for re-visiting the usage of WebJobs SDK and actually migrating away from, which is the topic of this article.

When it comes to what Azure WebJobs SDK triggers we were using, I believe we are nothing special and like many others we have the following in place:

Pub-Sub message bus, in our case Azure Event Hubs by means of EventHubTrigger
Queues, in our case Azure Storage Queues by means of QueueTrigger
Time-triggered jobs by means of TimerTrigger
Websockets, in our case Azure SignalR by means of SignalRTrigger

Additionally, we use various other IHostedService/BackgroundService processes, which get started upon app start and are running in the background, doing some other work - e.g. listening to notifications from the database (change streams), aggregating something every x seconds, etc.

All of the above is using configuration based on simple environment variables, some of them pointing to a secrets store (in our case Azure Key Vault). There is of course some logging and telemetry sent to a cloud service (in our case Azure Application Insights), as well as some web/REST/HTTP API (in our case using barebone ASP.NET Core SDK).

All XyzTriggers above have been migrated away from the Azure WebJobs SDK to pretty small, sweet and (almost) fully in our control implementations using directly the Azure SDK, which will be explain in the sections below. But before we do that let's look at how the app stub looks like now.

App Stub v2

// Program.fs
module TestService.XyzHandling.Program

open System.Threading
open Microsoft.Extensions.Hosting
open Framework.Hosting
open Framework.AzureKeyVault.Environment
open TestService.XyzHandling.Api.Wiring

Environment.overwriteEnvironmentVariablesFromKVRef () |> Async.RunSynchronously

[<EntryPoint>]
let main argv =

    let builder =
        HostBuilder.createDefaultBuilder argv BackgroundServiceExceptionBehavior.StopHost
        |> HostBuilder.configureLogging
        |> HostBuilder.configureAppInsights
        |> HostBuilder.configureEventHubProcessors
            EnvVars.appName
            [ EventHubProcessors.eventHubProcessor1; EventHubProcessors.eventHubProcessor2 ]
        |> HostBuilder.configureQueueProcessors
            EnvVars.appName
            ([ Some QueueProcessors.queueProcessor1 ] @ [ QueueProcessors.queueProcessor2 ]
             |> List.choose id)
        |> HostBuilder.configureBackgroundServices
            EnvVars.appName
            [ BackgroundServices.eventHubPublisherTest
              BackgroundServices.queuePublisherTest ]
        |> HostBuilder.configureTimers EnvVars.appName [ Timers.timerProcessor1; Timers.timerProcessor2 ]
        |> HostBuilder.configureStartup Startup.startupFunctions
        |> HostBuilder.configureWebHost
            EnvVars.appName
            [ WebApi.checkHealth; WebApi.checkReadiness; WebApi.Entity.getById ]

    use tokenSource = new CancellationTokenSource()
    use host = builder.Build()
    host.RunAsync(tokenSource.Token) |> Async.AwaitTask |> Async.RunSynchronously

    0 // return an integer exit code

Notes:

The above code spawns in total 9 IHosted/Background services (incl. the standard web host)
Event Hub Processors listen to & process Azure Event Hub events
Queue Processors listen to & process Azure Storage queue messages
Timer Processors run stuff every x seconds, minutes, hours etc.
There are even some generic Background Services started for writing to an event hub and a queue
One can see at a glance all the running processes and even web api endpoints from the Program.fs file

Trigger Implementations

The trigger implementations below are all based on BackgroundService/IHostedService, which means a background task is spawned and is running the whole time the host itself is running.

Compared to the original implementations in Azure Webjobs SDK the below ones may have less functionality (no auto-scaling for Azure Functions or similar) but at the same time are very lean and much easier to understand/maintain.

The source code of a working sample application will be eventually made available here.

Event Hub Processor

Old code:


// Api.Wiring.fs, handle the event
type WebJobs(...) =
    [<FunctionName("HandleXyzEvent")>]
    member _.HandleXyzEvent
        (
            [<EventHubTrigger("",
                              Connection = EnvVars.EventHubs.Xyz.connectionStringKey,
                              ConsumerGroup = EnvVars.EventHubs.consumerGroup)>] msg: EventData,
            enqueuedTimeUtc: DateTime,
            sequenceNumber: Int64,
            offset: string,
            logger: ILogger
        )
        =
        // handle the event

// Program.fs, configure the host with web jobs
    let configureWebJobs (builder:IHostBuilder) = 
        builder.ConfigureWebJobs(fun b ->
          b.AddAzureStorageCoreServices() |> ignore
          b.AddEventHubs() |> ignore)

Notes:

There is some magic going on, because in Program.fs you do not say what you actually want to listen to, you just say "I want to enable Event Hubs Handling", and then you decorate some method of a class with attributes, which indicate that you want to listen to an event hub
The EventHubTrigger insists on getting a key to a connection string environment variable, and fetch the value when it decides ...
You seem to need to specify also a FunctionName in addition to the method name
Any configuration settings for the EventHubTrigger are hidden away, you need to know what environment variables to configure, which are picked "automatigally" by the framework ;)

New code:


// Api.Functions.fs, handle the event
    let processEvent log partitionId (cancellationToken: CancellationToken) event =
        // handle the event

        () |> Async.retn

// Api.Wiring.fs, instantiate the processor
    let eventHubProcessor1 =
        EventHubProcessorDef.create
            "EventHubProcessor1"
            EnvVars.appName
            EnvVars.EventHubs.checkpointStorageConnectionString
            EnvVars.EventHubs.eventHubConnectionString
            EnvVars.EventHubs.consumerGroup
            EnvVars.EventHubs.eventBatchMaximumCount
            EnvVars.EventHubs.assignedPartitionIds
            EnvVars.EventHubs.defaultStartingPosition
            EventHubHandlers.processEvent

// Program.fs, configure the host with the processors
        |> HostBuilder.configureEventHubProcessors
            EnvVars.appName
            [ EventHubProcessors.eventHubProcessor1; EventHubProcessors.eventHubProcessor2 ]

Event Hub Processor Implementation:

module Framework.AzureEventHubs.EventProcessing

open System
open System.Threading.Tasks
open System.Collections.Generic
open System.Threading
open Azure.Messaging.EventHubs.Consumer
open Azure.Storage.Blobs
open Azure.Messaging.EventHubs
open Azure.Messaging.EventHubs.Primitives
open Framework
open Framework.AzureEventHubs.LogEvents
open Framework.Logging.StructuredLog

// https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/eventhub/Azure.Messaging.EventHubs/samples/Sample08_CustomEventProcessor.md
// Event Processor which considers assigned partitions
type AssignablePartitionProcessor
    (
        log: Log,
        name: string,
        storageClient: BlobContainerClient,
        assignedPartitions: string[] option,
        eventBatchMaximumCount: int,
        consumerGroup: string,
        connectionString: string,
        clientOptions: EventProcessorOptions,
        processEvent: Log -> string -> CancellationToken -> EventData -> Async<unit>,
        processError: Log -> string option -> string -> CancellationToken -> Exception -> Async<unit>
    )
    =
    inherit
        PluggableCheckpointStoreEventProcessor<EventProcessorPartition>(
            BlobCheckpointStore(storageClient),
            eventBatchMaximumCount,
            consumerGroup,
            connectionString,
            clientOptions
        )

    // Workaround, see https://github.com/dotnet/fsharp/issues/12448 ...
    member this.BaseListPartitionIdsAsync
        (
            connection: EventHubConnection,
            cancellationToken: CancellationToken
        )
        : Task<string[]>
        =
        base.ListPartitionIdsAsync(connection, cancellationToken)

    override this.ListPartitionIdsAsync
        (
            connection: EventHubConnection,
            cancellationToken: CancellationToken
        )
        : Task<string[]>
        =
        match assignedPartitions with
        | Some assignedPartitions -> assignedPartitions |> Task.FromResult
        | None -> this.BaseListPartitionIdsAsync(connection, cancellationToken)

    // Workaround, see https://github.com/dotnet/fsharp/issues/12448 ...
    member this.BaseListOwnershipAsync
        (cancellationToken: CancellationToken)
        : Task<IEnumerable<EventProcessorPartitionOwnership>>
        =
        base.ListOwnershipAsync(cancellationToken)

    override this.ListOwnershipAsync
        (cancellationToken: CancellationToken)
        : Task<IEnumerable<EventProcessorPartitionOwnership>>
        =
        match assignedPartitions with
        | Some assignedPartitions ->
            assignedPartitions
            |> Seq.map (fun partition ->
                EventProcessorPartitionOwnership(
                    FullyQualifiedNamespace = this.FullyQualifiedNamespace,
                    EventHubName = this.EventHubName,
                    ConsumerGroup = this.ConsumerGroup,
                    PartitionId = partition,
                    OwnerIdentifier = this.Identifier,
                    LastModifiedTime = DateTimeOffset.UtcNow
                ))
            |> Task.FromResult
        | None -> this.BaseListOwnershipAsync(cancellationToken)

    // Workaround, see https://github.com/dotnet/fsharp/issues/12448 ...
    member this.BaseClaimOwnershipAsync
        (
            desiredOwnership: IEnumerable<EventProcessorPartitionOwnership>,
            cancellationToken: CancellationToken
        )
        : Task<IEnumerable<EventProcessorPartitionOwnership>>
        =
        base.ClaimOwnershipAsync(desiredOwnership, cancellationToken)

    override this.ClaimOwnershipAsync
        (
            desiredOwnership: IEnumerable<EventProcessorPartitionOwnership>,
            cancellationToken: CancellationToken
        )
        : Task<IEnumerable<EventProcessorPartitionOwnership>>
        =
        // Warning: if the match is removed, and only the code in the Some part is left => High CPU utilization if no assignedPartitions defined!
        // for more info see https://github.com/Azure/azure-sdk-for-net/issues/39603
        match assignedPartitions with
        | Some _ ->
            desiredOwnership
            |> Seq.iter (fun ownership -> ownership.LastModifiedTime <- DateTimeOffset.UtcNow)

            desiredOwnership |> Task.FromResult
        | None -> this.BaseClaimOwnershipAsync(desiredOwnership, cancellationToken)

    // Workaround, see https://github.com/dotnet/fsharp/issues/12448 ...
    member this.BaseUpdateCheckpointAsync
        (
            partitionId: string,
            offset: int64,
            sequenceNumber: Nullable<int64>,
            cancellationToken: CancellationToken
        )
        =
        base.UpdateCheckpointAsync(partitionId, offset, sequenceNumber, cancellationToken)

    // used in the OnProcessingEventBatchAsync member, calculate once
    member private this.EventHubFullName =
        Subscribing.createEventHubFullPath this.FullyQualifiedNamespace this.EventHubName this.ConsumerGroup

    // https://learn.microsoft.com/en-us/dotnet/api/azure.messaging.eventhubs.eventprocessorclient.onprocessingeventbatchasync?view=azure-dotnet#remarks
    override this.OnProcessingEventBatchAsync
        (
            events: IEnumerable<EventData>,
            partition: EventProcessorPartition,
            cancellationToken: CancellationToken
        )
        : Task
        =
        task {
            try
                if not (isNull events || events |> Seq.isEmpty) then
                    do!
                        events
                        |> Seq.map (fun event ->
                            async {
                                Subscribing.checkEventEnDequeueTime log this.EventHubFullName event

                                do! processEvent log partition.PartitionId cancellationToken event
                            })
                        |> Async.Sequential
                        |> Async.Ignore
                        |> Async.StartAsTask
                        :> Task

                    let lastEvent = events |> Seq.last

                    do!
                        this.BaseUpdateCheckpointAsync(
                            partition.PartitionId,
                            lastEvent.Offset,
                            lastEvent.SequenceNumber,
                            cancellationToken
                        )
            with ex ->
                // It is very important that you always guard against exceptions in
                // your handler code; the processor does not have enough
                // understanding of your code to determine the correct action to take.
                // Any exceptions from your handlers go uncaught by the processor and
                // will NOT be redirected to the error handler.
                //
                // In this case, the partition processing task will fault and be restarted
                // from the last recorded checkpoint.

                log.Exception
                    (int EventId.EventProcessorError, string EventId.EventProcessorError)
                    "OnProcessingEventBatchAsync: Exception while processing events: {ex}"
                    ex
                    [| ex |]

        // bubble up, which will kill the background service and result in Health Check alert
        // alternative is to log and "swallow" the exception here, but then the processor will go in infinite loop ...
        // NOT a good idea, raising exception here invokes OnProcessingErrorAsync, which causes the host to get restarted automatically, and the health check does not detect this ..
        // ex.Reraise()
        }

    // https://learn.microsoft.com/en-us/dotnet/api/azure.messaging.eventhubs.eventprocessorclient.onprocessingerrorasync?view=azure-dotnet#remarks
    override this.OnProcessingErrorAsync
        (
            ex: Exception,
            partition: EventProcessorPartition,
            operationDescription: string,
            cancellationToken: CancellationToken
        )
        : Task
        =
        task {
            try
                let partitionId = partition |> Option.ofObj |> Option.map _.PartitionId

                do!
                    processError log partitionId operationDescription cancellationToken ex
                    |> Async.StartAsTask

            with wex ->
                // It is very important that you always guard against exceptions
                // in your handler code; the processor does not have enough
                // understanding of your code to determine the correct action to
                // take. Any exceptions from your handlers go uncaught by the
                // processor and will NOT be handled in any way.
                //
                // In this case, unhandled exceptions will not impact the processor
                // operation but will go unobserved, hiding potential application problems.

                log.Exception
                    (int EventId.EventProcessorError, string EventId.EventProcessorError)
                    "OnProcessingErrorAsync: Exception occurred while processing events: {wex}. Original exception: {ex}."
                    wex
                    [| wex; ex |]

        // do! this.StopProcessingAsync(cancellationToken)

        // bubble up, which will kill the background service and result in Health Check alert
        // alternative is to log and "swallow" the exception here, but then the processor will go in infinite loop ...
        // NOT a good idea, the host gets restarted automatically, and the health check does not detect this ..
        // ex.Reraise()
        }

/// Starts the Event Processor
let startConsumeEvents
    (name: string)
    (checkpointStorageConnectionString: string)
    (checkpointBlobContainerName: string)
    (eventHubConnectionString: string)
    (consumerGroup: string)
    (eventBatchMaximumCount: int)
    (assignedPartitionIds: string[] option)
    (defaultStartingPosition: EventPosition)
    (processEvent: Log -> string -> CancellationToken -> EventData -> Async<unit>)
    (processError: Log -> string option -> string -> CancellationToken -> Exception -> Async<unit>)
    (started: ManualResetEvent)
    (log: Log)
    : (IDictionary<string, obj> -> CancellationToken -> Async<unit>)
    =
    fun state cancellationToken ->
        async {
            let blobContainerClient =
                BlobContainerClient(checkpointStorageConnectionString, checkpointBlobContainerName)
            // automatically create container if it does not exist
            do! blobContainerClient.CreateIfNotExistsAsync() |> Async.AwaitTask |> Async.Ignore

            let options = EventProcessorOptions() // TODO: Customize some of them?

            options.DefaultStartingPosition <- defaultStartingPosition

            let processor =
                AssignablePartitionProcessor(
                    log,
                    name,
                    blobContainerClient,
                    assignedPartitionIds,
                    eventBatchMaximumCount,
                    consumerGroup,
                    eventHubConnectionString,
                    options,
                    processEvent,
                    processError
                )

            state.Add("processor", processor)

            log.Info
                (int EventId.EventProcessorStarted, string EventId.EventProcessorStarted)
                "Starting with config:\n\
                    \tEventHubNamespace/Name/ConsumerGroup = {eventHubNamespace}/{eventHubName}/{consumerGroup}\n\
                    \tBlobContainerClient.Uri = {blobContainerClientUri}\n\
                    \tAssignedPartitionIds = {assignedPartitionIds}\n\
                    \tEventBatchMaximumCount = {eventBatchMaximumCount}\n\
                    \tOptions.PrefetchCount = {prefetchCount}\n\
                    \tOptions.PrefetchSizeInBytes = {prefetchSizeInBytes}\n\
                    \tOptions.MaximumWaitTime = {maximumWaitTime}\n\
                    \tOptions.TrackLastEnqueuedEventProperties = {trackLastEnqueuedEventProperties}\n\
                    \tOptions.DefaultStartingPosition = {defaultStartingPosition}\n\
                    \tOptions.LoadBalancingStrategy = {loadBalancingStrategy}\n\
                    \tOptions.LoadBalancingUpdateInterval = {loadBalancingUpdateInterval}\n\
                    \tOptions.PartitionOwnershipExpirationInterval = {partitionOwnershipExpirationInterval}\n\
                    \tOptions.RetryOptions.Mode = {retryOptionsMode}\n\
                    \tOptions.RetryOptions.Delay = {retryOptionsDelay}\n\
                    \tOptions.RetryOptions.MaximumDelay = {retryOptionsMaximumDelay}\n\
                    \tOptions.RetryOptions.MaximumRetries = {retryOptionsMaximumRetries}\n\
                    \tOptions.RetryOptions.TryTimeout = {retryOptionsTryTimeout}\n\
                    \tOptions.RetryOptions.CustomRetryPolicy = {retryOptionsCustomRetryPolicy}\n\
                    "
                [|
                    (processor.FullyQualifiedNamespace |> String.replace ".servicebus.windows.net" "")
                    processor.EventHubName
                    processor.ConsumerGroup
                    blobContainerClient.Uri
                    $"%A{assignedPartitionIds}"
                    eventBatchMaximumCount
                    options.PrefetchCount
                    options.PrefetchSizeInBytes
                    options.MaximumWaitTime
                    options.TrackLastEnqueuedEventProperties
                    options.DefaultStartingPosition
                    options.LoadBalancingStrategy
                    options.LoadBalancingUpdateInterval
                    options.PartitionOwnershipExpirationInterval
                    options.RetryOptions.Mode
                    options.RetryOptions.Delay
                    options.RetryOptions.MaximumDelay
                    options.RetryOptions.MaximumRetries
                    options.RetryOptions.TryTimeout
                    options.RetryOptions.CustomRetryPolicy
                |]

            do! processor.StartProcessingAsync(cancellationToken) |> Async.AwaitTask

            started.Set() |> ignore
        }

/// Stops the Event Processor
let stopConsumeEvents
    name
    (started: ManualResetEvent)
    (log: Log)
    (state: IDictionary<string, obj>)
    : (CancellationToken -> Async<unit>)
    =
    fun cancellationToken ->
        async {
            let processor = state["processor"] :?> AssignablePartitionProcessor
            do! processor.StopProcessingAsync(cancellationToken) |> Async.AwaitTask

            log.Info
                (int EventId.EventProcessorStopped, string EventId.EventProcessorStopped)
                "Event Hub Processor was stopped"
                [||]

            started.Reset() |> ignore
        }

Notes:

You define a function, then instantiate processor(s) with the function and a bunch of configuration values, which you can fetch by yourself, and then you tell the HostBuilder to configure your processor(s) - pretty straightforward, no reflection, no magic
The implementation of the EventHubProcessor is using EventProcessorClient in the background, which does everything required, including the same checkpointing in blob storage as done by WebJobs SDK. The whole implementation is less than 350 LOCs ..

Queue Processor

Old code:

type WebJobs(...) =
    [<FunctionName("RetryHandleXyzEvent")>]
    member _.RetryHandleXyzEvent
        ([<QueueTrigger("xyz-events-retry-queue", Connection = "StorageQueueConnectionStringKey")>] msg: string)
        (logger: ILogger)
        =
        // handle the message

// Program.fs, configure the host with web jobs
    let configureWebJobs (builder:IHostBuilder) = 
        builder.ConfigureWebJobs(fun b ->
          b.AddAzureStorageCoreServices() |> ignore
          b.AddAzureStorageQueues() |> ignore)

New code:

// Api.Functions.fs, handle the event
    let processMessage log (cancellationToken: CancellationToken) (msg: QueueMessage) : Async<unit> =
        // handle the queue message

        () |> Async.retn

// Api.Wiring.fs, instantiate the processor
    let queueProcessor1 =
        QueueProcessorDef.create
            "QueueProcessor1"
            EnvVars.appName
            EnvVars.Queues.queueStorageConnectionString
            EnvVars.Queues.queueName
            EnvVars.Queues.messageBatchMaximumCount
            EnvVars.Queues.visibilityTimeout
            EnvVars.Queues.maxPollingInterval
            EnvVars.Queues.maxDequeueCount
            EnvVars.Queues.defaultBackOffIntervalMs
            QueueHandlers.processMessage

// Program.fs, configure the host with the processors
        |> HostBuilder.configureQueueProcessors
            EnvVars.appName
            [ QueueProcessors.queueProcessor1 ]

Queue Processor Implementation:

module Framework.AzureStorageQueues.QueueProcessing

open System
open System.Threading
open Azure.Storage.Queues
open Azure.Storage.Queues.Models
open Framework.AzureStorageQueues.BasicOperations
open Framework.AzureStorageQueues.LogEvents
open Framework.Logging.StructuredLog
open Framework.ExceptionHandling

let private doProcessMessage
    (log: Log)
    (name: string)
    (cancellationToken: CancellationToken)
    (maxDequeueCount: int)
    (queueClient: QueueClient)
    (poisonQueueClient: QueueClient)
    (processMessage: Log -> CancellationToken -> QueueMessage -> Async<unit>)
    (msg: QueueMessage)
    =
    async {
        if msg.DequeueCount > maxDequeueCount then
            // message has been retried too many times => move it the the poison queue
            do!
                poisonQueueClient.SendMessageAsync(msg.Body, cancellationToken = cancellationToken)
                |> Async.AwaitTask
                |> Async.Ignore

            log.Debug
                (int EventId.QueueMessageProcessingError, string EventId.QueueMessageProcessingError)
                "Message dequeue count = {messageDequeueCount} > maxDequeueCount = {maxDequeueCount} => message moved to poison queue {poisonQueueUri}"
                [| msg.DequeueCount; maxDequeueCount; poisonQueueClient.Uri |]

            do!
                queueClient.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, cancellationToken)
                |> Async.AwaitTask
                |> Async.Ignore

            log.Debug
                (int EventId.QueueMessageProcessingError, string EventId.QueueMessageProcessingError)
                "Message deleted from queue {queueUri}"
                [| queueClient.Uri |]
        else
            // normal processing
            // NOTE: try-with is used instead of Async.Catch to catch also cases when processMessage throws exception outside of an Async block ...
            try
                do! processMessage log cancellationToken msg

                do!
                    queueClient.DeleteMessageAsync(msg.MessageId, msg.PopReceipt, cancellationToken)
                    |> Async.AwaitTask
                    |> Async.Ignore

                log.Debug
                    (int EventId.QueueMessageBeingProcessed, string EventId.QueueMessageBeingProcessed)
                    "Deleted message with id = {messageId} and dequeue count = {messageDequeueCount} in queue {queueUri} after successful processing"
                    [| msg.MessageId; msg.DequeueCount; queueClient.Uri |]
            with ex ->
                // message remains in the queue and will become again visible after visibilityTimeout
                log.Exception
                    (int EventId.QueueMessageProcessingError, string EventId.QueueMessageProcessingError)
                    "Queue message processing failed. Retrying {remainingDequeueCount} more times, then the message will be moved to poison queue. Queue: {queueUri}; Message Body: {messageBody}"
                    ex
                    [|
                        (int64 maxDequeueCount - msg.DequeueCount)
                        queueClient.Uri
                        msg |> QueueMessage.toString
                    |]
    }
    |> Async.Catch
    |> Async.map (function
        | Choice1Of2 _ -> ()
        | Choice2Of2 ex ->
            // log the message body
            log.Exception
                (int EventId.QueueMessageProcessingError, string EventId.QueueMessageProcessingError)
                "Error occurred while performing auxiliary queue message processing (e.g. DeleteMessageAsync). Message Body: {messageBody}"
                ex
                [| msg |> QueueMessage.toString |]
            // and propagate up the exception
            ex.Reraise())

/// Starts an infinite loop for receiving queue messages
let consumeMessages
    (name: string)
    (storageConnectionString: string)
    (queueName: string)
    (poisonQueueName: string)
    (messageBatchMaximumCount: int)
    (visibilityTimeout: TimeSpan)
    (maxPollingInterval: TimeSpan)
    (maxDequeueCount: int)
    (defaultBackOffIntervalMs: int)
    (processMessage: Log -> CancellationToken -> QueueMessage -> Async<unit>)
    (started: ManualResetEvent)
    (log: Log)
    : (CancellationToken -> Async<unit>)
    =
    fun cancellationToken ->
        async {
            let options = QueueClientOptions() // TODO: Configure options?
            let queueClient = QueueClient(storageConnectionString, queueName, options)
            do! queueClient.CreateIfNotExistsAsync() |> Async.AwaitTask |> Async.Ignore

            let poisonQueueClient =
                QueueClient(storageConnectionString, poisonQueueName, options)

            do! poisonQueueClient.CreateIfNotExistsAsync() |> Async.AwaitTask |> Async.Ignore

            let mutable backoffTimeMs = defaultBackOffIntervalMs

            started.Set() |> ignore

            log.Info
                (int EventId.QueueProcessorStarting, string EventId.QueueProcessorStarting)
                "Starting with config:\n\
                    \tUri = {queueClientUri}\n\
                    \tMessageBatchMaximumCount = {messageBatchMaximumCount}\n\
                    \tVisibilityTimeout = {visibilityTimeout}\n\
                    \tMaxPollingInterval = {maxPollingInterval}\n\
                    \tMaxDequeueCount = {maxDequeueCount}\n\
                    \tOptions.MessageEncoding = {messageEncoding}\n\
                    \tOptions.Retry.Mode = {retryMode}\n\
                    \tOptions.Retry.Delay = {retryDelay}\n\
                    \tOptions.Retry.MaxDelay = {retryMaxDelay}\n\
                    \tOptions.Retry.MaxRetries = {retryMaxRetries}\n\
                    \tOptions.Retry.NetworkTimeout = {retryNetworkTimeout}\n\
                    \tPoisonQueueName = {poisonQueueName}\n\
                    "
                [|
                    queueClient.Uri
                    messageBatchMaximumCount
                    visibilityTimeout
                    maxPollingInterval
                    maxDequeueCount
                    options.MessageEncoding
                    options.Retry.Mode
                    options.Retry.Delay
                    options.Retry.MaxDelay
                    options.Retry.MaxRetries
                    options.Retry.NetworkTimeout
                    poisonQueueName
                |]

            while not cancellationToken.IsCancellationRequested do
                try
                    // any exception in this block will be caught and logged
                    // because if propagated up they will stop the queue processor/background service => health check alert,
                    // but no automatic restart is currently possible ...

                    let! response =
                        queueClient.ReceiveMessagesAsync(
                            maxMessages = messageBatchMaximumCount,
                            visibilityTimeout = visibilityTimeout,
                            cancellationToken = cancellationToken
                        )
                        |> Async.AwaitTask

                    if response.HasValue && response.Value.Length > 0 then
                        do!
                            response.Value
                            |> Seq.map (fun msg ->
                                async {
                                    // central check for endequeTime
                                    // TODO: Enable this once a proper config/solution is found for the invisibility period, which is *not* exposed as a QueueMessage property (only InsertedOn and NextVisibleOn, but VisibleOn is needed ). See fore more info https://github.com/Azure/azure-sdk-for-net/issues/40147
                                    // do checkMessageEnDequeueTime log queueClient.Name backoffTimeMs msg

                                    return!
                                        doProcessMessage
                                            log
                                            name
                                            cancellationToken
                                            maxDequeueCount
                                            queueClient
                                            poisonQueueClient
                                            processMessage
                                            msg
                                })
                            |> Async.Sequential
                            |> Async.Ignore

                        backoffTimeMs <- defaultBackOffIntervalMs

                        log.Debug
                            (int EventId.WaitingForQueueMessages, string EventId.WaitingForQueueMessages)
                            "{messageCount} messages successfully processed from queue {queueUri}. Waiting for {backoffTimeMs} milliseconds before checking again ..."
                            [| response.Value.Length; queueClient.Uri; backoffTimeMs |]
                    else
                        backoffTimeMs <- Math.Min(backoffTimeMs * 2, int maxPollingInterval.TotalMilliseconds)

                        log.Debug
                            (int EventId.WaitingForQueueMessages, string EventId.WaitingForQueueMessages)
                            "No messages found in queue {queueUri}. Waiting for {backoffTimeMs} milliseconds before checking again ..."
                            [| queueClient.Uri; backoffTimeMs |]

                    do! Async.Sleep(backoffTimeMs)

                with ex ->
                    log.Exception
                        (int EventId.QueueMessageProcessingError, string EventId.QueueMessageProcessingError)
                        "Exception while processing messages: {ex}"
                        ex
                        [| ex |]
        }

Notes:

The implementation is based on the default QueueClient.ReceiveMessagesAsync approach of handling queue messages with the Azure SDK, invoked in an infinite loop with some Thread.Sleep sprinkled in it ... The whole implementation is about 200 LOCs.

Timer Processor

Old code:

type WebJobs(...) =
    [<FunctionName("DoSomethingRegularly")>]
    member this.ExpireCustomerDocuments
        (
            [<TimerTrigger("%DoSomethingRegularlyCrontab%")>] timer: TimerInfo,
            logger: ILogger
        )
        =
        // do something

// Program.fs, configure the host with web jobs
    let configureWebJobs (builder:IHostBuilder) = 
        builder.ConfigureWebJobs(fun b ->
          b.AddAzureStorageCoreServices() |> ignore
          b.AddTimers() |> ignore)

Notes:

Some exotic placeholder format of the crontab placeholder ..

New code:

// Api.Functions.fs, handle the event
    let processTimer1 log (cancellationToken: CancellationToken) (toProcessOn: DateTime) : Async<unit> =
        // do something

        () |> Async.retn
// Api.Wiring.fs, instantiate the processor
    let timerProcessor1 =
        TimerProcessorDef.create
            "TimerProcessor1"
            EnvVars.appName
            EnvVars.Timers.timerProcessorQueueStorageConnectionString
            (TimeSpan.FromSeconds(10) |> Some)
            "* * * * *"
            TimerHandlers.processTimer1

// Program.fs, configure the host with the processors
        |> HostBuilder.configureTimers EnvVars.appName [ Timers.timerProcessor1; Timers.timerProcessor2 ]

Timer Processor Implementation:

module Framework.AzureStorageQueues.TimerProcessing

open System
open System.Text
open System.Threading
open Azure.Storage.Queues
open NCrontab
open Framework
open Framework.AzureStorageQueues.LogEvents
open Framework.Logging.StructuredLog

type TimerMessage = { ToProcessOn: DateTime }

let private maxTimeoutPeriod = TimeSpan.FromDays(3) // could be up to 7 (messages are deleted after 7 days from the queue), but calculating possible downtimes/system recovery

let private calculateNextCheckOn (toProcessOn: DateTime) (now: DateTime) =
    if toProcessOn - now < maxTimeoutPeriod then
        toProcessOn
    else
        now + maxTimeoutPeriod

let private createAndSendNextTimerMessage (queueClient: QueueClient) (crontab: CrontabSchedule) =
    async {
        let toProcessOn = crontab.GetNextOccurrence(DateTime.UtcNow)

        let msg = { ToProcessOn = toProcessOn }
        let encodedMessage = msg |> Json.serialize |> Encoding.base64Encode Encoding.UTF8

        let nextCheckOn = calculateNextCheckOn toProcessOn DateTime.UtcNow
        let visibilityTimeout = nextCheckOn - DateTime.UtcNow // message should become visible always slightly after ToProcessOn because sending the msg to the queues takes some ms

        do!
            queueClient.SendMessageAsync(encodedMessage, visibilityTimeout, TimeSpan.FromSeconds(-1)) // -1 second indicates "infinite" message TTL (i.e. 7 days)
            |> Async.AwaitTask
            |> Async.Ignore

        return msg
    }

/// Starts an infinite loop for receiving queue timer messages
let private consumeMessages
    (log: Log)
    (name: string)
    (queueClient: QueueClient)
    (cancellationToken: CancellationToken)
    (crontab: CrontabSchedule)
    processMessage
    (messageBatchMaximumCount: int)
    (visibilityTimeout: TimeSpan)
    =
    async {
        let mutable firstRun = true

        while not cancellationToken.IsCancellationRequested do
            try
                // any exception in this block will be caught and logged
                // because if propagated up they will stop the timer processor/background service => health check alert,
                // but no automatic restart is currently possible ...
                let! response =
                    queueClient.ReceiveMessagesAsync(
                        maxMessages = messageBatchMaximumCount,
                        visibilityTimeout = visibilityTimeout,
                        cancellationToken = cancellationToken
                    )
                    |> Async.AwaitTask

                let! nextCheckOn =
                    if response.HasValue && (response.Value |> Seq.tryHead |> Option.isSome) then // the message became visible, so it needs to be processed or rescheduled
                        async {
                            let msg = response.Value |> Seq.head // we expect/process only 1 timer message per timer queue!

                            let decodedMessage =
                                msg.Body.ToString() |> Encoding.base64Decode Encoding.UTF8 |> Json.deserialize

                            log.Debug
                                (int EventId.TimerMessageBeingProcessed, string EventId.TimerMessageBeingProcessed)
                                "Timer message received in queue {queueUri} with ToProcessOn = {toProcessOn}."
                                [| queueClient.Uri; decodedMessage.ToProcessOn |> DateTime.toStringIso |]

                            let! toProcessOn =
                                // if ToProcessOn in the past => ready to process!
                                if decodedMessage.ToProcessOn <= DateTime.UtcNow then
                                    async {
                                        do! processMessage log cancellationToken decodedMessage.ToProcessOn

                                        log.Debug
                                            (int EventId.TimerMessageBeingProcessed,
                                             string EventId.TimerMessageBeingProcessed)
                                            "Timer message in queue {queueUri} successfully processed."
                                            [| queueClient.Uri |]

                                        // delete this (all) message(s)
                                        do! queueClient.ClearMessagesAsync() |> Async.AwaitTask |> Async.Ignore

                                        log.Debug
                                            (int EventId.TimerMessageBeingProcessed,
                                             string EventId.TimerMessageBeingProcessed)
                                            "Deleted timer message in queue {queueUri}."
                                            [| queueClient.Uri |]

                                        // schedule next execution in a new message
                                        let! newTimerMessage = createAndSendNextTimerMessage queueClient crontab

                                        log.Debug
                                            (int EventId.TimerMessageBeingProcessed,
                                             string EventId.TimerMessageBeingProcessed)
                                            "New timer message with ToProcessOn = {toProcessOn} sent to queue {queueUri}."
                                            [| newTimerMessage.ToProcessOn; queueClient.Uri |]

                                        return newTimerMessage.ToProcessOn
                                    }
                                else // message visible before ToProcessOn .. make invisible again
                                    async {
                                        let nextCheckOn =
                                            calculateNextCheckOn decodedMessage.ToProcessOn DateTime.UtcNow // message should become visible always slightly after ToProcessOn because sending the msg to the queues takes some ms

                                        let visibilityTimeout = nextCheckOn - DateTime.UtcNow

                                        if visibilityTimeout > TimeSpan.Zero then
                                            do!
                                                queueClient.UpdateMessageAsync(
                                                    msg.MessageId,
                                                    msg.PopReceipt,
                                                    visibilityTimeout = visibilityTimeout
                                                )
                                                |> Async.AwaitTask
                                                |> Async.Ignore

                                            log.Debug
                                                (int EventId.TimerMessageBeingProcessed,
                                                 string EventId.TimerMessageBeingProcessed)
                                                "Timer message in queue {queueUri} with ToProcessOn = {toProcessOn} became visible before ToProcessOn. The message's invisibility was extended."
                                                [| queueClient.Uri; decodedMessage.ToProcessOn |> DateTime.toStringIso |]

                                        return decodedMessage.ToProcessOn
                                    }

                            return calculateNextCheckOn toProcessOn DateTime.UtcNow
                        }
                    elif firstRun then
                        log.Debug
                            (int EventId.WaitingForTimerMessage, string EventId.WaitingForTimerMessage)
                            "No visible message in queue {queueUri}, but first run, so await schedule ..."
                            [| queueClient.Uri |]

                        let toProcessOn = crontab.GetNextOccurrence(DateTime.UtcNow)
                        calculateNextCheckOn toProcessOn DateTime.UtcNow |> Async.retn
                    else
                        log.Debug
                            (int EventId.WaitingForTimerMessage, string EventId.WaitingForTimerMessage)
                            "No visible message in queue {queueUri} after sleep but there should have been one, maybe delayed, so check frequently ..."
                            [| queueClient.Uri |]

                        DateTime.UtcNow.AddMinutes(1) |> Async.retn

                let sleepTimeSpan = (nextCheckOn - DateTime.UtcNow) + TimeSpan.FromSeconds(1) // add 1 second to make 100% sure sleep is over only after the message has already become visible

                log.Debug
                    (int EventId.WaitingForTimerMessage, string EventId.WaitingForTimerMessage)
                    "Sleeping {sleepTimeSpan} before checking again in queue {queueUri}"
                    [| sleepTimeSpan; queueClient.Uri |]

                if sleepTimeSpan > TimeSpan.Zero then
                    do! Async.Sleep(sleepTimeSpan)

            with ex ->
                log.Exception
                    (int EventId.TimerMessageProcessingError, string EventId.TimerMessageProcessingError)
                    "Exception while processing messages: {ex}"
                    ex
                    [| ex |]

            firstRun <- false
    }

/// Creates timer queue messages if missing and starts an infinite loop for receiving queue timer messages
let createAndConsumeMessages
    (name: string)
    (crontab: CrontabSchedule)
    (storageConnectionString: string)
    (queueName: string)
    (visibilityTimeout: TimeSpan)
    (processMessage: Log -> CancellationToken -> DateTime -> Async<unit>)
    (started: ManualResetEvent)
    (log: Log)
    : (CancellationToken -> Async<unit>)
    =
    fun cancellationToken ->
        async {
            let options = QueueClientOptions() // TODO: Configure options?

            let queueClient = QueueClient(storageConnectionString, queueName, options)
            let! response = queueClient.CreateIfNotExistsAsync() |> Async.AwaitTask

            if not (isNull response) then // queue was just created
                let! timerMessage = createAndSendNextTimerMessage queueClient crontab

                log.Info
                    (int EventId.TimerProcessorStarting, string EventId.TimerProcessorStarting)
                    "New queue {queueUri} was created and a new timer message with ToProcessOn = {toProcessOn} was sent to it."
                    [| queueClient.Uri; timerMessage.ToProcessOn |> DateTime.toStringIso |]
            else // queue was already existing
                let! response = queueClient.GetPropertiesAsync() |> Async.AwaitTask

                if response.HasValue && response.Value.ApproximateMessagesCount = 0 then // ApproximateMessagesCount, even though not exact, is guaranteed to have a value > 0 if there are messages. Additionally "Approximate messages count will give you an approximate count of total messages in a queue and will include both visible and invisible messages."
                    let! timerMessage = createAndSendNextTimerMessage queueClient crontab

                    log.Info
                        (int EventId.TimerProcessorStarting, string EventId.TimerProcessorStarting)
                        "Queue {queueUri} exists, but no timer message found in it. Created timer message with ToProcessOn = {toProcessOn}."
                        [| queueClient.Uri; timerMessage.ToProcessOn |> DateTime.toStringIso; queueClient.Uri |]
                else
                    log.Info
                        (int EventId.TimerProcessorStarting, string EventId.TimerProcessorStarting)
                        "Existing queue {queueUri} with timer message found."
                        [| queueClient.Uri |]

            let messageBatchMaximumCount = 1 // 1 queue per timer processor, with 1 message per queue only

            started.Set() |> ignore

            log.Info
                (int EventId.TimerProcessorStarting, string EventId.TimerProcessorStarting)
                "Starting with config:\n\
                    \tUri = {queueClientUri}\n\
                    \tMessageBatchMaximumCount = {messageBatchMaximumCount}\n\
                    \tVisibilityTimeout = {visibilityTimeout}\n\
                    \tOptions.MessageEncoding = {messageEncoding}\n\
                    \tOptions.Retry.Mode = {retryMode}\n\
                    \tOptions.Retry.Delay = {retryDelay}\n\
                    \tOptions.Retry.MaxDelay = {retryMaxDelay}\n\
                    \tOptions.Retry.MaxRetries = {retryMaxRetries}\n\
                    \tOptions.Retry.NetworkTimeout = {retryNetworkTimeout}\n\
                    "
                [|
                    queueClient.Uri
                    messageBatchMaximumCount
                    visibilityTimeout
                    options.MessageEncoding
                    options.Retry.Mode
                    options.Retry.Delay
                    options.Retry.MaxDelay
                    options.Retry.MaxRetries
                    options.Retry.NetworkTimeout
                |]

            do!
                consumeMessages
                    log
                    name
                    queueClient
                    cancellationToken
                    crontab
                    processMessage
                    messageBatchMaximumCount
                    visibilityTimeout
        }

Notes:

The implementation is based on an Azure Storage Queue with a single message inside, which is made "invisible" for a certain period of time, which allows for surviving a process crash, having multiple instances running etc. The whole implementation is about 250 LOCs.

SignalR

Old code:

type WebJobs(...) =   
    [<FunctionName("HandleAndPushToClient")>]
    member _.HandleAndPushToClient
        (
            [<EventHubTrigger("",
                              Connection = EnvVars.EventHubs.Xyz.connectionStringKey,
                              ConsumerGroup =
                                  DependencyInjection.EventHubs.Xyz.consumerGroupForWebSocketNotification)>] msg:
                EventData,
            enqueuedTimeUtc: DateTime,
            sequenceNumber: Int64,
            offset: string,
            [<SignalR(HubName = DependencyInjection.SignalR.hubName,
                      ConnectionStringSetting = "AzureSignalRConnectionString")>] signalRMessages:
                IAsyncCollector<SignalRMessage>,
            logger: ILogger
        )
        =
        // transform some internal event to external
        // publish to SignalR using signalRMessages.AddAsync

Notes:

The output triggers are generally a killer feature, which is really killing you - instead of invoking a very simple Azure SDK client method, you deal with IAsyncCollector and SignalRTrigger magic, which is completely unnecessary. Go figure out how to send message to a specific user or to all ...

New code (same as for EventHubProcessor above:

    let serviceManager =
        SignalRClient.getServiceManager EnvVars.SignalR.connectionString

    let hub =
        SignalRClient.getHubContext serviceManager EnvVars.SignalR.azureSignalRHubName
        |> Async.RunSynchronously // TODO: Find a way to get rid of this

    let sendToUser = SignalRClient.sendToUser hub

// Api.Functions.fs, handle the event
    let processEvent log sendToUser partitionId (cancellationToken: CancellationToken) event = async {
        // handle the event
        do! sendToUser "SomeTarget" "Some Message" "SomeUserId"

    }

Notes:

The Azure SDK for sending SignalR Messages is very very straightforward, when you use it directly ...

Conclusion

Removing a layer of indirection has always generated great satisfaction in me. Not only does it make the whole application easier to understand, but you gain also a lot more control, and get to know the inner workings of the technology, without someone deciding something for you, or translating stuff like configuration for you.

MS seems to always try to make things easier for the developer (patronizing him?) by providing a magical and abstract framework which achieves exactly the opposite. My recommendation to MS would be to try make everything look like a stupid console application instead, with full control of the client developer, who is just using a bunch of simple "helper" functions (or class methods in OOP) from MS, and nothing more.

Hopefully someone can save some time doing something similar based on the ideas and code in this article!

Funny enough in the meantime MS seems to be trying to do something similar, by integrating WebJobs SDK into .NET 9's HostBuilder. ↩

Isolated Local Dev Env using k3d, multi-tenant Azure services, docker containers, and mirrord

Deyan Petrov — Sun, 12 Nov 2023 01:00:53 +0000

TLDR; Use k3d, multi-tenant Azure Services, local docker containers and mirrord for an amazing local development experience.

Introduction

Being a part of a small team (< 10) of full-stack developers with end-to-end responsibility (from requirements till running systems) I have been primarily focusing in the past couple of years more or less on having stable test and production environments, utilizing Kubernetes (AKS) and having a set of microservices whereby individual microservices (pods) can be deployed on a daily basis in a controlled way.

Our tech stack consists of .NET/F#, Kubernetes (AKS¹ on Azure), MongoDB (Atlas on Azure) as main OLTP database and Azure Data Explorer² as a DWH, and many additional Azure services like Key Vault, Event Hubs³, Queues⁴, Blob Storage⁵, Azure SignalR for websockets, etc.

But what about the development environment? Well, we have shared Azure services for it - e.g. Key Vault, Event Hubs, Queues and Blob Storage, as well as the DWH, but we do run MongoDB locally as a docker container, and we do run the applications themselves locally, usually started using JetBrains Rider's Run/Debug configuration. And this was working relatively fine as usually a developer needs to open 1 microservice, implement some additional API, write automated tests, then manually test it via UI out of which only those pages are invoked which call the microservice in question.

The above under-invested setup had its issues and we needed something better ..

Disclaimer: Even though this article demonstrates a very specific implementation of a local dev environment based on a perhaps unique combination of technologies which probably nobody else uses, the underlying principles/approaches may be portable also to other tech stacks/setups.

Problem Statement

The problem statement is straightforward: how to set up a development environment for every developer with a "complete" isolation from the other team members, and how to do this in a cost-efficient way.

First, when we started developing more and more our web UIs and native mobile apps, the need arose more and more frequently to be able to perform some manual local testing of end-to-end business processes, which requires firing up a number of microservices, pub-sub of multiple events, and even querying eventually-consistent Read Models.

Second, we do have some integrated authentication and other Traefik middleware functionality in Kubernetes, which cannot be tested locally unless run in a Kubernetes cluster.

And last but not least, some if not most of the Azure services which we use do not have good developer experience, i.e. there are no easy-to-run docker images or similar.

Solution

The solution consists of several approaches:

install an emulator on the developer machine as a docker container or
run the service in the cloud (Azure in our case), but in a special/custom "multi-tenant mode". The latter means that many developers can use the Azure services but in a complete isolation from each other.

The sections below contain detailed explanations of the technologies/components and approaches involved.

k3d

k3d (wrapper around k3s) is an excellent and very lightweight Kubernetes running as a docker container. We are speaking about setup time of 20-30 seconds, and a couple of minutes to compile and deploy all our microservices to it. Cluster start/stop is supported, but full cluster deletion and re-creation is almost as easy ;)

k3d registry create $registryName -p 5050
k3d cluster create $clusterName --registry-use $registryName:5050 --k3s-arg "--disable=traefik@server:*" --subnet '172.18.0.0/16'

The local docker registry is useful for pushing to it all locally compiled docker images, to be pulled by Kubernetes upon deployment/pod creation.

K3d comes with traefik out-of-the-box, but as you see above it is disabled in this case, as we need a more customized version of it.

Traefik and all other pods we need get deployed by simply kubectl apply -f a bunch of yamls (we are not using Helm).

There is one nasty problem with k3d whereby after the developer laptop goes to sleep and gets woken up the DNS mapping host.k3d.internal does not work anymore. There is a workaround though, namely to specify subnet e.g. 172.18.0.0/16 upon creation and use 172.18.0.1 instead.

Stopping the cluster is done with:

k3d cluster stop cluster-name

starting with:

k3d cluster start cluster-name

and deletion with

k3d cluster delete cluster-name
k3d registry delete registry-name.localhost

Simple and fast!

Multi-tenant Azure services

As mentioned above, ideally all cloud services used should have emulators which can be run locally in docker containers. Unfortunately it seems that Microsoft does not care consistently about developer UX and does not provide emulators running in docker containers for all its services, especially for such central ones like Azure Event Hubs in our case (our messaging backbone).

The other problem with emulators in general is that there is no 100% guarantee that they would behave the same as the "real" cloud service.

And last but not least, some cloud (in particular Azure) services are integrated with each other - e.g. Azure Data Explorer ingests data from Azure Event Hubs, something which would be generally missing from the emulators.

Per-Developer Azure Event Hubs Partitions

The general idea for making Azure Events Hubs "multi-tenant" is simple - write a small wrapper around Azure SDK for Event Hubs so that publishing to event hubs (~topics) and subscribing to them via consumer groups is limited to a specific partition. Every Azure Event Hub in the standard SKU can have up to 32 partitions, which means there can be up to 32 isolated clients (1 partition each) - be it local developer environments, shared environments or CI/CD pipelines.

Every environment has an environment variable e.g. LOCAL_DEV_EVENTHUBS_ASSIGNEDPARTITIONIDS=0,1 which specifies which partitions should be used. If the env var is missing then all available partitions are used, which is the case for the shared test and production environments (using their own dedicated sets of Azure Event Hubs).

How "easy" is it to create a wrapper over the standard SDK? Well, that was relatively easy, based on a nice sample from Microsoft. Unfortunately it required migration away from the WebJobs SDK which was being used so far, and which did not give enough control to the underlying implementation.⁶

Per-Developer Azure Storage Accounts (incl. Blob Containers and Storage Queues)

When it comes to Azure Storage Accounts - that was easy, every developer got her/his own storage account (1 per every developer) where the naming uses a suffix, e.g. company1stacc4xy (xy being the developer initials). All blob containers and storage queues created inside are used exclusively be the developer in question. Storage accounts are very cheap.

Note: There is an open-source emulator for Azure Storage called Azurite, which can be run as a docker container, and even though it might support all our requirements, it was simply easier (and very cheap if not free) to provision storage accounts directly in the cloud.

Per-Developer Azure Data Explorer (Logical) Databases

Azure Data Explorer (ADX) is a Column Store database, which means it stores the data in columns instead of rows, which makes it extremely fast for queries on large datasets. Of course, there are some limitations when it comes to ingestion of data (to be batched) and updates/deletion of data (not easily possible). It powers all Azure logging/monitoring technologies (Azure Monitor, Azure Application Insights) and can be relatively inexpensive for a database (dev/test SKU around 100 EUR/month, and production SKUs starting from 500 EUR/month).

In the system in question the data is ingested to ADX via Event Hubs (no direct ETL from OLTP -> DWH!) in near real-time - by default in batches every 10 minutes, however for some tables also much shorter intervals can be configured.

The challenge here was how to isolate developers from each other when it comes to ADX.

The solution consists of 2 parts:

Every developer gets her/his own logical database (in the same physical database cluster), with a database name having a suffix db4ab for example where "ab" are the initials of the developer. A script can re-create the database in a matter of a few minutes.
Implement routing of events to an alternative database based by setting a "Database" property of every event upon publishing. The document for this built-in feature is here

Note: Microsoft offers an ADX emulator running as a docker container(https://learn.microsoft.com/en-us/azure/data-explorer/kusto-emulator-install), which is perfect in general, however there is no support for the data ingestion from the Azure Event Hubs which are running in the cloud.

Per-Developer Azure SignalR Notification Hubs

Azure SignalR service is WebSocket implementation which is used for realtime server <> client communication used in our cose for some realtime dashboards and UI flows.

Azure SignalR works with the named hubs to which clients connect by first requesting the hub url and access token, and then establishing a connection. So the solution in this case is to create a per-developer hub based on an environment variable which is set to a different value on every developer machine, but to the same value for the test/production environment. As the hub name/url/accessToken is fully controlled by the server app (k8s pod) there is no change required in the client application (SPA in our case).

Docker Containers running locally

In case there are high-quality local emulators available, then these are of course preferentially used, as then there is out-of-the-box isolation from other developers. We are running local docker containers for MongoDB and SFTP Server for example.

The install/run script can be as simple as a 1-liner:

docker run --name mongodb --restart unless-stopped -v mongodata:/data/db -d -p 27017:27017 mongo --oplogSize 50 --replSet rs0

docker run --name sftp --restart unless-stopped -p 22:22 -d atmoz/sftp foo:pass:::upload,upload/processed

Note the --restart unless-stopped policy, which makes sure that upon restart of the dev machine the docker containers are started automatically.

What is important when running local docker containers are the scripts for setting them up, including all configurations to be applied - e.g. in case of MongoDB DDL scripts for the setup all local databases used by the apps.

mirrord

mirrord is an amazing and magic piece of technology, which allows you to debug a running container in a Kubernetes cluster - any cluster, including the local k3d cluster for example. More importantly, from developer UX point of view the debugging runs:

within the context of the container, which means all environment variables for example with which the container was deployed(!), as well as inbound/outbound network connectivity
within your own locally-running IDE - e.g. Jetbrains Rider(!)

What happens in reality: mirrord installs a pod in the Kubernetes cluster which redirects (or duplicates) all calls to the pod to the additional local process which it starts on the local dev machine.

Invoking mirrord can be done in 2 ways:

Terminal:
- find the running pod name:

- start mirrord targetting the pod and at the same time your local app

```bash
mirrord exec --target pod/test-qryh-test-xyzh-68f79c7c7b-jdp4v 
dotnet bin/Debug/net7.0/TestService.XyzHandling.dll
```

- Attach your IDE to the process

Directly from the IDE (e.g. Jetbrains Rider) - run debug configuration + mirrord:

Once mirrord is enabled by clicking the round button to the left, you get a confirmation:

and can click on the Debug button for the app in question:

after which the app is started, and any set breakpoints will be hit:

The best thing is - the environment variables and network connections that are made from the application being debugged are coming from the Kubernetes cluster and going through the cluster to any other running pod or even outside application ... pure magic ;)

I need to mention also the great support of the mirrord creator I got for an issue I had with the Rider plugin. Not only did Aviram Hassan jump on a discord call with me on the same evening I reported the issue(wow!!), but he also shared his thoughts that with a bigger team size a shared development environment (in the cloud) becomes a necessity.

Some Non-Functional Considerations

Scalability: The described setup has a limitation of 32 Azure Event Hubs partitions available, which means up to 30-31 developers if every developer gets 1 partition, or 15 if everybody gets 2 partitions. There is however no problem create additional Azure Event Hub Namespaces (~ Kafka clusters) which can support another 15 developers and so on. All the other cloud services used - Azure Storage Accounts, Azure Data Explorer, Azure SignalR etc. can scale out/up with the team size ... Never thought though what happens when the team is > 100 developers, but this is a non-issue in the context I am in currently ;)
Network Connectivity/Offline usage: Offline usage is not supported, as there are cloud services involved. I am yet to experience a case where I have to developer/run/debug something without Internet connection though - 4G/5G network coverage is pretty good everywhere (incl. in public transport) nowadays.
Docker containers vs multi-tenant/per-developer cloud services: as mentioned above, this is just a tradeoff, can go more to the one or the other direction. Emulators are not the real thing or may not be available, but implementing multi-tenancy for cloud services requires substantial changes in the software layer responsible for infrastructure, which may also not be that easy.
mirrord on Windows/WSL: Even though everything in the above setup should and does work on Linux, MacOS and Windows, mirrord is yet to be tested whether it works well on Windows with WSL, especially the part with Jetbrains Rider connecting to the process to debug, or spawning it using mirrord plugin ..

Conclusion

A couple of approaches were combined to create a pretty easy to setup, convenient to use and cost-efficient development environment, allowing every developer to run/debug applications in full isolation:

Local Kubernetes cluster with k3d
Multi-tenant Azure cloud services based on some service sub-entity like partitions, logical databases or hubs
Locally running emulators in docker containers
Mirrord for pod debugging

Even though there was some effort setting this up, and even though the setup sounds a bit too specific and perhaps even fragile, it is doing what it is required to do, namely allowing developers to run and debug applications in isolation from each other.

Hopefully someone can save some time improving her/his development environment based on the ideas in this article!

Similar to EKS on AWS and GKE on GCP ↩
Similar to ClickHouse ↩
Similar to AWS Kinesis or GCP Pub Sub (Lite) ↩
Similar to SQS on AWS ↩
Similar to S3 on AWS ↩
A dedicated article about this migration will be published in the near future ↩

CQRS+

Deyan Petrov — Sat, 29 Jul 2023 14:44:01 +0000

TLDR; Splitting your "services" into several "microservices" based on technical responsibility simplifies development and code management and gives you deployment flexibility and isolation.

Introduction

This post is intended to demonstrate one concrete way of implementing microservices by using CQRS but going beyond the the standard Command/Query separation. It is a follow-up to a previous post about Current State + Last Event as an alternative to Event Sourcing, at the end of which I outlined this architectural pattern.
Note that even though some examples may refer to F#/.NET/Azure, the CQRS+ approach can be used with any tech stack and programming language.

Problem Statement

You have split your system into several "bounded contexts", each one consisting of a single or a few relatively encapsulated "services", e.g. Customer Registration, Order Processing, Notification Service, etc.), where every "service" contains a few (1-5) "domain entities" (e.g. Notification Service contains an entity called "Notification"). For each "service" you have created a .NET/F# solution, but now you are wondering how to organize internally the functionality inside each solution - do you need a single project (+1 test project) only?

On the one hand, you have numerous requirements for every "service":

Getting some user input via the UI, wrapping it as "command" which results in the creation/update of some "entities" or simply data records, validating some business rules or maintaining some invariants along the way
Displaying the existing data to the user in a UI, or returning it to an external system
Handling events raised by other internal/external systems, based on which you potentially means you trigger internal commands (=> point 1.)
Maybe you need to publish some events yourself, so other internal systems can react to them?
Maybe you need to publish some events externally via webhooks, so other external systems can react to them?
Write some audit trail entries for some critical entities, or calculate some balance or Read Model asynchronously.

On the other hand, you have heard about "microservices" with just a "few" LOCs, and you have been wondering how the hell do the "wise guys" manage to produce such microservices, while yours always end up being fat services with thousands of LOCs (or you have a single monolith).

Ideally you would modularize/split your "service" in such a way that you end up with small microservices which can be written/re-written/compiled/tested/deployed/scaled/monitored/restarted individually.

Solution

The simple structure I follow is the following:

Your company's overall software
- Bounded Context 1
  - Service 1
    - Microservice 1
    - Microservice 2

Diagram 1: Bounded Context/Service/Microservice Hierarchy

Note: it could well be that you have only a single "service" in a bounded context.

The important point is that a service is split into microservices based on technical responsibilities - a microservice responsible for command processing, another one for query processing, a third one for event handling, and so on.

So a CQRS+ solution could look like this:

Diagram 2: Xyz Service and its microservices

You would create a solution for the "service" with multiple projects (one per technical microservice), and multiple test projects (again, one per technical microservice). Your solution structure would then look like this:

XyzService.sln
  src/
    XyzService.CommandHandling.fsproj
    XyzService.QueryHandling.fsproj
    XyzService.EventHandling.fsproj
    XyzService.EventPublishing.fsproj
    XyzService.ExternalEventPublishing.fsproj
    XyzService.ExternalEventHandling.fsproj
    XyzService.ChangeHandling.fsproj
    XyzService.Shared.fsproj
  tests/
    XyzService.CommandHandling.Tests.fsproj
    XyzService.QueryHandling.Tests.fsproj
    XyzService.EventHandling.Tests.fsproj
    XyzService.EventPublishing.Tests.fsproj
    XyzService.ExternalEventPublishing.Tests.fsproj
    XyzService.ExternalEventHandling.Tests.fsproj
    XyzService.ChangeHandling.Tests.fsproj

Additionally, as you see from the above, a XyzService.Shared.fsproj project is needed as well, containing the common Persistence, Api Dtos and even Domain Types used by several projects - e.g. CommandHandling (write) and QueryHandling (read).

The deployed applications (e.g. pods in K8s) would look like this:

xyz-cmdh-aaaaaaaaaa-bbbbb
xyz-cmdh-aaaaaaaaaa-ccccc
xyz-qryh-bbbbbbbbbb-ddddd
xyz-qryh-bbbbbbbbbb-eeeee
xyz-qryh-bbbbbbbbbb-fffff
xyz-evh-cccccccccc-ggggg
xyz-evp-dddddddddd-hhhhh
xyz-extevp-eeeeeeeeee-kkkkk
xyz-extevh-ffffffffff-lllll
xyz-chgh-gggggggggg-mmmmm

Note that cmdh and qryh replica sets above have multiple replicas, whereas the rest could have a single replica configured.

K8s is ideal for deploying such fine-grained applications, as it allows for a very high density of pods per node. .NET is not that ideal though, as the memory used per pod is usually 100Mb+, so the nodes need to have more memory (Rust/Go would be better in this regard).

Benefits

Clarity how to break down bounded contexts/services into microservices. A "service" is more or less a bounded context or a big part of such, and a "microservice" is a part of the "service" focused on a particular technical responsibility
Possibility to have small microservices of a couple of hundred LOCs => easy to grasp and "cache" in your brain
The microservices are standalone processes/apps/pods, can be deployed, scaled, restarted, monitored individually
The place of the domain logic is clear - only in the Command Handling microservice. All the other microservices contain no business logic.

Drawbacks

Too much unnecessary segregation => accidental complexity?
- Pays off once you start looking after memory leaks, or quickly fix a bug by deploying only a small microservice without affecting the rest
- Is also offset by the standardization of the microservices, and the smaller context each microservice represents => simplicity
The (technical) microservices are sharing some data model or API contract, hence several of them may need to be deployed together
- Data models and API contracts are generally difficult to change (due to so many other reasons), and backwards compatibility must be highly respected anyway.

Implementation Details

The structure and contents of each "microservice" as well as the Shared project will be described below.

XyzService.CommandHandling (CmdH)

CmdH microservice is responsible for processing all commands which arrive in the form of PUT, POST, PATCH and DELETE requests. Automatic Jobs (e.g. cron jobs/Timer Triggers) are also hosted in this application, as usually these invoke some business logic.

The standard command processing pipeline consists of the following steps:

Deserialize request into Api.Requests.XxxxCommandDto and furthemore into XyzCommand record type, in the process of which the fields are validated (incl. enum value lookup, boundary checks, etc.)
Fetch something from the database if needed (especially in case of executing an action on an existing entity)
Invoke the domain logic for creating a new entity, or updating an existing one, considering all business rules (e.g. Blocked Customer can be Closed or put back to Active, but cannot be set to PendingKYC for example).
After the business logic has been executed and a result has been produced then the domain entity is mapped to a Persistence Dto and the latter is stored in the database, and then again mapped to an Api DTO, and the latter is serialized to JSON and returned by the API.

CmdH is the beefiest project/app of all, as it contains the domain logic, e.g. validation, calculation, etc. More often than not the domain entities are defined as state machines, whereby every state transition is validated, some flags are set, etc.

CmdH is focused on writing stuff to a Write Data Model (optimized for writing), always respecting entity (aggregate root) consistency boundaries. NoSQL/Document databases are a good fit for a write model and remove the need for an ORM, or any additional work splitting a domain entity across several tables (not a responsibility of CmdH!).

XyzService.QueryHandling (QryH)

QryH microservice is responsible for handling GET queries for fetching REST resources.

The query processing pipeline usually consists of:

Deserialize JSON into Api.Requests.XyzQueryDto
Validate some of the properties (query parameters) by trying to parse them into domain types (this is why some domain types have to be in Shared.Domain.Types!)
Execute database query
Map Persistence.Dtos.XyzDto to Api.Responses.XyzDto and return to caller

QryH does not care about domain entities in general, it is happy to serve also flattened/"joined" data of several entities, or subset of the data of an entity. It relies either on the same Data Model used for writing (in case of fetching of a single entity by id for example, or any filtering based on simple indexes) or requires its own Read Model for optimized searching/filtering by a random combination of filters, where the Read Model is usually filled asynchronously => eventually consistent. Column store databases can be a pretty good fit for a dedicated high performance Read Model.

XyzService.EventPublishing (EvP)

EvP is responsible for publishing events to a message bus, so that internal subscribers (e.g. xyz2-evh, or a DWH) can receive them.

The standard event publishing pipeline looks like this (in case database Change Streams/Feed technology is used):

Receive database change stream event, deserialize it to a Persistence.Dtos.XyzDto
Map the persistence dto to a Api.Dtos.XyzDto (event schema is part of the public service contract!)
Publish dto to a message bus topic

EvP exposes also API for subscribing and unsubscribing to specific event types. Additionally, EvP also allows (again via a dedicated API) for replay of events.

XyzService.EventHandling (EvH)

EvH microservice is handling events received from other internal microservices (published by some other xyz2-evp, see above), received usually from a message bus like Kafka, Azure Event Hubs, etc.

The standard event processing pipelines looks like this:

Deserialize received event JSON into Api.Dtos.XyzDto
Handle the parsed event by e.g. creating a command and sending it to CmdH

EvH is taking care of cases when a certain event cannot be processed, due to some internal event handling issues, or inability to connect to CmdH for example. In that case EvH has the choice to either stop/restart the processing of all events for the specific consumer group of the topic, or "park" the problematic event on a message queue for further retrying later on, which comes with the downside of losing strict ordering of the consumer group-based event processing.

XyzService.ExternalEventPublishing (ExtEvp)

ExtEvp is responsible for pushing events via webhooks/HTTPs to external subscribers (including retry in case the subscriber is down).

The standard external event publishing pipeline looks like this (in case database Change Streams/Feed technology is used):

Receive database change stream event, deserialize it to Persistence.Dtos.XyzDto
Map the persistence dto to Api.Dtos.XyzDto (event schema is part of the public service contract!)
Send it to a collection of (subscribed in advance) HTTP endpoints

ExtEvp must take care of temporary unavailability of some (out of all) subscribers. Parking messages in "retry message queues" with invisibility period is usually a good approach of handling this problem.

XyzService.ExternalEventHandling (ExtEvh)

ExtEvh is responsible for handling events received from external systems, usually via webhooks => public listener endpoints.

The standard external event handling pipeline looks like this:

Receive e.g. HTTPs request, map it to Api.Dtos.XyzDto (= initial schema validation)
Convert the request to an internal command to CmdH, and invoke CmdH.
Respond with "200 OK" back to the webhook call

ExtEvh's endpoints must be secured, so that they can be invoked only by trusted external callers.

XyzService.ChangeHandling (ChgH)

ChgH is responsible for handling database Change Streams/Feed changes and calculating/creating something. It can be regarded as a more general version of EvP/ExtEvP, and originally it was used only for auditing or creating/calculating (e.g. diff between 2 versions of an entity) audit trail entries and writing them to the database, but some other calculation logic can be put here as well (e.g. maintaining balance entities or similar).

The standard change handling pipeline looks like this (in case database Change Streams/Feed technology is used):

Receive database change stream event, deserialize it to Persistence.Dtos.XyzDto
Do some processing, e.g. calculate a diff between 2 entity states and store an audit trail entry, or trigger a command to CmdH, etc.

ChgH, together with Evp and ExtEvp is used to offload CmdH from any additional duties. CmdH focuses on performing business logic and storing data into the write model. All the other services act in a near-real-time on the data change and perform their additional duties.

It is possible though that some of the other microservices also directly subscribe to database change streams - such is the case for example when an in-memory (= in-process) cache has to be built in QryH for example.

Additional Solution Files

The solution contains the following central files:

Open API (Swagger) yaml file - combines all PUT/POST/PATCH/DELETE/GET operations from all microservices in a single API

Additional Project Files

Every fsproj contains the following deployment-related files:

Dockerfile
Kubernetes deployment yaml file
Azure DevOps deployment yaml file + bash script
Bicep templates + bash script (in case of using Azure, similar for AWS/GCP)
DB DDL/DML Scripts for setting up and upgrading the database

Solution containing reusable code

What if you have 10 bounded contexts with 1 service per each, and then you have to split those 10 services into microservices? What to do with some repetitive code for e.g. writing or reading messages from message bus and writing to a message queue in case something goes wrong?

In case all microservices are using the same technology, then some generic/reusable functionality used in all microservices of the same type (e.g. EventHandling) can be extracted into a separate solution/project (e.g. Framework.Services.EventHandling).

You may easily end up with 2 solutions containing reusable "Framework"/helper code:

Framework - containing projects/code reusable across all types of microservices, e.g. string manipulation, date helpers etc
Framework.Services - containing projects/code reusable across certain class of microservices, e.g. *.EventHandling

Diagram 3: Framework, Framework.Service and Services hierarchy

Conclusion

Splitting services into microservices based on CQRS+ clearly explains how to break down your code into manageable components. This approach allows you to focus on a single technical responsibility in each microservice, as well as write, compile, deploy, scale and monitor the latter individually. Starting a new service and breaking it down into microservices is then a straightforward story which can also be done by less-experienced developers.

Use Change Streams instead of Traditional Outbox or Distributed Transactions

Deyan Petrov — Mon, 26 Sep 2022 15:05:06 +0000

TLDR; Use Change Streams instead of traditional outbox pattern with local database transaction, or even worse - distributed transaction across database and message bus.

Introduction

In the context of Microservice Architecture (MSA) and Event-Driven Architecture (EDA) one of the core questions is how to save an entity's state to the database and reliably (= with 100% guarantee) push corresponding event(s) to a message bus, so that other services can eventually consume the published event(s) and do their job in an eventually consistent manner.

To my surprise there is still not enough awareness (e.g. judging by answers I get in interviews) of what is the most efficient solution and furthermore, experts with 20+ years of experience are still referring to older patterns, which are already superseded by newer, more efficient ones.

In detail, assuming we have these 2 operations:

An entity's state is written to a database table (regardless if insert/update of last state, or insert of event in the event-sourced persistence approach)
An event message is written to a message bus,

the problem to solve is the following: the process could crash in between the 2 steps which leaves us with an inserted/updated entity in the database but with no event sent to other microservices, so some side-effects (e.g. sending welcome email) are not performed.

Note: This post does not deal with the topic how to create Domain Events/External Events, for more info on one way of doing that please read the (older) post Current State + Last Event as an alternative to Event Sourcing.

Traditional Solutions

Outbox Pattern with a local database transaction

The idea here is to take advantage of local database transactions in order to guarantee 100% that an event message is also written in addition to writing the entity, and then process all event messages in a reliable/retriable manner (again with 100% guarantee that every message will be processed at least once).

The following 2 steps need to be implemented:

Bundle together the insert/update of the entity's state with an insert of a "message/event" in another database table (in the same database), within a local database transaction:

BEGIN TRANSACTION
INSERT INTO/UPDATE Customers ...
INSERT INTO CustomerEvents ....
COMMIT TRANSACTION

Then have a "Reader" process which goes over the CustomerEvents sequentially, reads every record and publishes the event onto a message bus, and marks the event as "published". In case of an error (process crashes), then start again from last not-published event. As it is possible to experience a crash in between

e.g. every X seconds

SELECT * FROM CustomerEvents 
WHERE IsPublished = False 
ORDER BY CreatedOn ASC

messageBusClient.publish(event)

UPDATE CustomerEvents 
SET IsPublished = True 
WHERE CustomerEventId = eventId

"Issues" with this approach:

Need to create/maintain the CustomerEvents table
Upon updating the Customer entity it is necessary to take care also of inserting into the CustomerEvents table, and this in a local database transaction (imagine there are many microservices with many entities to store) ... Ideally one would need to only care about inserting/updating Customers table, without anything else ...
The local database transaction creates short-lived page locks, which reduce concurrency

Distributed Transactions

This approach spans a distributed transaction between 2 different technologies - database and message bus. Used to be quite popular 15-20 years ago, but since then many sources discourage from using it due to vendor lock-in, complexity, concurrency issues.

Pseudo code would be:

distributedTransactionCoordinator.StartTransaction()
try 
    databaseClient.insert/update(customer)
    messageBusClient.publish(event)
    distributedTransactionCoordinator.CommitTransaction()
with _ ->
    distributedTransactionCoordinator.RollbackTransaction()

I have not used distributed transactions in the past 15 years myself so treat the above pseudo code as a very rough approximation conveying only the general idea.

"Issues" with this approach:

Distributed transactions are not really supported across all different technologies, such should be carefully chosen ..
Distributed transactions have their own complexity and implications in terms of rollbacks, locks, etc.

Note: Explaining how distributed transaction coordinators are implemented using Two-Phase Commit (2PC) is beyond the scope of this post, as this approach is anyway not used/recommended.

Additional Considerations

Message Delivery, Idempotency, Duplicate Detection

Having in mind the 3 message delivery guarantees:

At most once
At least once
Exactly once,

the most practical approach is to settle on "At least once", which means that the event message will be published at least once, but it is possible that it is published 2+ times (e.g. in case the publishing process crashes, or a timeout is received when calling the message bus API). Such duplicate event messages can be handled in the following ways:

Idempotent processing or event message deduplication by the subscriber
Event message deduplication by the Message Bus middleware

Idempotent processing or event message deduplication by the subscriber in simple terms means that if the same event is received by a subscriber twice, the subscriber is able to detect that this event has already been processed, do nothing and return OK (as if the event was processed successfully). This can be achieved in different ways, e.g.:

A local ProcessedEvents table in the subscriber's database which contains the (unique) event id of those events which were successfully processed. Upon receiving a new event first a lookup in that table is done, and if nothing found only then the event is processed, otherwise return OK directly
The event id is used as a column in another table, and there is a unique constraint on it. Upon inserting a new row to this other table the unique constraint generates an error in case of duplicate processing, which can be caught and converted to OK result
The primary key of another table can be the same as the event id, so that again upon duplicate insert the default unique constraint generates an error in case of duplicate processing, which can be caught and converted to OK result

Event message deduplication by the message bus middleware itself is something that I usually do not use, as I prefer not to be tightly coupled to specific middleware with exclusive functionality, but instead be able to use any message bus having just basic pub-sub (at least once) functionality ...

Checkpointing

What happens if the subscriber process consuming the change stream dies/crashes? Upon restarting that process has several options:

Start re-processing all change stream events from the beginning - this will result in a lot of duplicate events published, and requires idempotent/deduplicating downstream subscribers
Start from the end of the change streams (only new change stream events will be processed) - in this case some events in between will never be processed/published.
Start from a checkpoint which indicates last successfully processed change stream entry.

Obviously option 3 is the best but it requires a little bit of extension to the change stream "watching" code:

Upon startup the watcher reads the last checkpoint from a custom table and starts watching change stream events from that one onwards
After processing every 1 (or N) change stream events the watcher must update the checkpoint with the last successfully processed event id/timestamp (in case of MongoDB every change stream event has a ResumeToken property). If checkpoints are saved upon every processed change stream event then max 1 change stream events will be retried. In case the checkpoint is saved less frequently, e.g. after every 5th event, then max 5 events will be retried again in case of a process crash or similar. In both cases the "at least once" message delivery guarantee is still valid, and if downstream processor are implemented with this in mind there should be no problem.

Multiple subscribers (Competing Consumers)

By default the change stream event handling application is a single instance/single thread application, which gets all change stream events and publishes them to a message bus as event messages. This is a very fast operation and for "normal" applications a single instance is perfectly fine, as long as it is hosted in such a way (e.g. Kubernetes) that if it dies then automatically a new instance is spawned.

But what if there are so many events per second that multiple instances are required to run in parallel?

This is something I have not needed so far, but in such a case I would probably have e.g. 3 application instances listening to the same change stream (Customers) and skipping every first, second or third change stream event. Every change stream "watcher" will have its own checkpoint.

Event Replay

Sometimes there is need for replaying all events. We have 2 "buffers" to consider:

The Change Stream database storage - in case of MongoDB is configurable but usually does not hold all changes since the creation of the database
The Message Bus - also stores messages only for a limited period of time, e.g. in case of standard Azure Event Hubs for 7 days

In the best case only change stream events or event messages are replayed which in one of the 2 buffers.

If events must be replayed all the way from source table (e.g. Customers) then we have 3 options:

All changes to Customers are additionally stored in an CustomerAuditTrailEntries collection, so the latter can be read sequentially and for each entry a custom change stream event can be created and published to the message bus directly.
Customers table actually does not exist, Event Sourcing is used and every single Customer event is inserted as a separate table row/document. Then iterating over these rows/documents will be sufficient for generating the event messages and publishing them to the message bus
No audit trail is stored, no Event Sourcing is used, updates are done in place. In this case the best that can be done is to generate and publish to the message bus events for the last state of each customer entity.

It is worth mentioning that the Event Replay itself can also be exposed to external webhook subscribers who can trigger it over a REST API (e.g. POST /replays with from,to,eventType and subscriberId properties in the request body).

Furthermore, the event replay should ideally publish event message to a separate topic on the message bus, to which only interested in replay subscribers should be listening, otherwise all existing subscribers can be flooded with a lot of events.

Last but not list, sometimes direct database manipulations might be required (e.g. database refactoring/migrations). Such migration should or should not trigger change stream event processing. In case of the latter, a simple filtering scheme on some entity property (e.g. LastModifiedOn) should be introducing in the change stream "watching" application so that such changes can be skipped.

Error Handling

While processing the change stream events (= looping over the open cursor polling the change stream collection) an error/exception may happen in relation to:

Individual document - e.g. cannot be deserialized, or publishing to message bus fails due to transient error
The whole change stream - e.g. network/database outage

Individual document processing error can be handled by simply "parking" the document into a retry queue, which is handled by another thread automatically with a small delay, and if reprocessing again fails the message can be parked in a poison queue, to be looked at manually.

In case of a general error the change stream can be either restarted from the last saved checkpoint after a small delay (= catch exception, re-open cursor), or the whole process may be let to crash, which would invoke its automatic restart by e.g. Kubernetes.

Remember, what we are speaking about here is just iterating over a database collection containing changes (= change stream) while keeping a clientside "index" how far down the road the watcher is. In case of a single watcher (enough in most cases) you can always stop for a while, wait, restart from last index or even from a previous position on the stream. This is the same simple and scalable approach used in many other technologies based on the append-only "Log".

Real-life performance comparison of MongoDB Atlas on Azure/GCP/AWS

Deyan Petrov — Wed, 05 Jan 2022 00:44:45 +0000

TLDR; MongoDB Atlas M20-M40 on Azure, GCP and AWS perform pretty similarly, with slightly varying levels of performance (shorter durations) and stability (outliers, variability) across the different cluster/storage sizes:

M20: AWS > Azure > GCP (sometimes Azure and AWS have higher variability)
M30: AWS > Azure > GCP (higher variability for GCP)
M40: AWS > Azure ~> GCP (higher variability for GCP and AWS).

Introduction

The intention of this post is to compare the real-life performance of MongoDB M20-M40 running on Azure, GCP and AWS. Read further if you want to get the condensed results of otherwise a multi-day/week exercise¹.
The trigger for the load tests were survarious disk-related performance issues on Azure and repeated statements that MongoDB on AWS and GCP is running much better.

Important Notes:

This post is discussing only small dedicated instances - M20-M40 with 128 and 256 GB storage only (no need for/experience with bigger ones yet)
The Reader should only pay attention to the relative aspect of the results, i.e. how different clouds compare to each other, or different instance/storage sizes improve the performance. The absolute request durations in milliseconds etc. are to be ignored, as they are only valid for the specific custom code under test.
This is not a "scientific" or properly done benchmark (I am an amateur performance/load tester). If you see the approach is wrong or/and can be improved - please let me know in the comments what I can do better (appreciated).
The architecture details of the software under test is out of scope for this article (I may or may not write another one on that). Same for the load test automation tooling, which is basically implemented using Azure DevOps Pipelines and a substantial number of bash scripts.
Even though many load tests with different requests per second were performed, only the result of a subset of the tests - 50 requests per second - will be used in this article, as these are enough to illustrate the differences in performance of MongoDB Atlas on Azure/GCP/AWS.

Setup

The software under test is a set of 5 microservices responsible for synchronously² processing a business operation/transaction ³. The overall interaction between the services looks like this (anonymized):

Note that all database operations are single-document ones where the document size is less than 1 KB, and fall into 3 categories:
1) find a single document by UUID id (primary key)
2) insert a single document
3) replace a single document

The load test infrastructure involves the following tools:

Azure DevOps Pipelines - for running bash scripts in sequence, responsible for both setup, running the load tests and tear down of the load test environment (both Azure and GCP)
NBomber - a load test client, generating in this case 50 requests per second using InjectPerSec(rate = 50, during = seconds 600) Load Simulation
"Client" Kubernetes cluster (AKS, GKE or EKS) with just 1 node (4 vCPUs, 8Gb RAM) for hosting the Azure DevOps self-hosted pool with 1 agent running NBomber
"Server" Kubernetes cluster (AKS, GKE or EKS) with 2 nodes (2 vCPUs, 16 GB RAM) for running the microservices (3/5 microservices configured with 2 replicas, 2/5 with 1 replica only). Traefik is running as well in the "server" cluster to act as an (HTTPs) API Gateway.
Both K8s clusters are in the same region, Vnet/VPC Network, different subnets, however the client clusters uses HTTPs with the public DNS hostname of the server for making the calls (so certain traversal of the network infrastructure of the cloud provider takes place, however we are speaking about single-digit ms here).

The MongoDB instances used are summarized in the following table:

TABLE MongoDB Instances, including IOPS and $$$

Cloud	Instance Type	Storage Size (GB)	vCPUs	RAM	IOPS	$$$/h	$$$/month
Azure	M20	128	2⁴	4	500	0.34	245
GCP	M20	128	1	3.75	7680⁵	0.33	238
AWS	M20	128	2⁴	4	2000	0.30	238
Azure	M20	256	2⁴	4	1100	0.45	324
GCP	M20	256	1	3.75	15000⁵	0.45	324
AWS	M20	256	2⁴	4	2000	0.38	274
Azure	M30	128	2	8	500	0.80	576
GCP	M30	128	2	7.5	7680⁵	0.60	432
AWS	M30	128	2	8	3000⁶	0.67	482
Azure	M30	256	2	8	1100	0.91	655
GCP	M30	256	2	7.5	15000⁵	0.72	518
AWS	M30	256	2	8	3000⁶	0.75	540
Azure	M40	128	4	16	500	1.46	1051
GCP	M40	128	4	15	7680⁵	1.06	763
AWS	M40	128	4	16	3000⁶	1.23	886

Worth noting:

Azure M20 has 2 CPUs vs 1 vCPU for GCP M20, however in case of Azure the vCPU are burstable, whereas in GCP the single vCPU can be utilized to 100% all the time.
There is a huge difference in IOPS (on paper) with GCP offering 14-30x more IOPS than Azure for the same storage size!
With bigger instance sizes Azure becomes noticeably more expensive.

The following sections will present the load testing results from different perspectives.

Azure vs GCP vs AWS Performance M20 with 128Gb disk

50 rps, 5 minutes duration, 3 executions

Interpretation:

Azure M20 is clearly faster than GCP M20, and when it comes to mean durations at the same level as AWS M20.
AWS M20 has very high 99 percentile and max durations though (=> a lot variability)

50 rps, 1 hour duration, 3 executions

Interpretation:

AWS is faster than Azure M20 which is faster than GCP when it comes to average durations,
AWS has better 99th percentiles as well
Maximal durations for Azure are worse than GCP and AWS.

Azure vs GCP vs AWS Performance M20 with 256Gb disk

50 rps, 5 minutes duration, 3 executions

Interpretation:

Azure and AWS are roughly at the same level, faster than GCP
AWS shows max duration outliers in 2 out of the 3 executions

1 hour duration, 3 executions

Interpretation:

AWS M20 is faster than Azure which is faster than GCP when it comes to average durations and 99th percentile
All 3 have a relatively high max request duration

Azure vs GCP vs AWS Performance M30 with 128Gb disk

50 rps, 5 minutes duration, 3 executions

Interpretation:

AWS M30 is significantly faster than Azure M30, which is faster than GCP M30
For all 3 the maximal durations are more consistent (less variability) compared to M20 (especially for AWS)

50 rps, 1 hour duration, 3 executions

Interpretation:

AWS is faster than Azure & GCP
Azure M30 is at the same level as GCP when it comes to average durations, 99th percentiles are almost equal, but the maximal durations of GCP are the highest (could be also individual outliers only).

Azure vs GCP Performance M30 with 256Gb disk

50 rps, 1 hour duration, 3 executions

Interpretation:

In the 1-hour test executions Azure M30 is at the same level as GCP when it comes to average durations, 99th percentiles are almost equal, with the maximal durations for Azure better than GCP's.

Azure vs GCP vs AWS Performance M40 with 128Gb disk

50 rps, 5 minutes duration, 3 executions

Interpretation:

AWS M40 is slightly faster than Azure M40 and GCP M40 (however Azure M40 is substantially more expensive, see table above)

50 rps, 1 hour duration, 3 executions

Interpretation:

AWS M40 is slightly faster than Azure M40 and GCP M40
GCP and AWS have higher variability (the 3rd AWS test execution resulted even in 5 timeouts out of 180k requests ..)

Azure Performance Different Instance/Storage Sizes

50 rps, 5 minute duration

Interpretation:

Average durations do not improve substantially with the increase of the instance/storage size (even increase from M20 -> M30??)
99 percentile and max durations are higher with M20/128 and more stable with M20/128, M30 and M40

50 rps, 1 hour duration

Interpretation:

Average durations do not improve substantially with the increase of the instance/storage size
99 percentile and max durations improve significantly with the increase of the instance/storage size (variability decreases, less outliers)

GCP Performance Different Instance/Storage Sizes

50 rps, 5 minute duration

Interpretation:

Averages improve slightly with the instance/storage size. 2. Same is valid for 99 percentile and max durations, which are pretty stable

50 rps, 1 hour duration

Interpretation:

Averages improve slightly with the instance/storage size. 2. Same is not valid for 99 percentile and max durations ... there are strange peaks in case of M40/128 ...

AWS Performance Different Instance/Storage Sizes

50 rps, 5 minute duration

Interpretation:

Averages are better with M30, but almost no difference with M40 for this load
High 99 percentile and max durations in case of M20, stable for M30 and M40

50 rps, 1 hour duration

Interpretation:

Averages are better with M30, but almost no difference with M40 for this load
High 99 percentile and max durations in case of M20, stable for M30 and M40 (1 execution only has max duration outlier)

MongoDB Metrics Comparison

Max Disk Write Latency

M20, 128Gb, Azure:

M20, 128Gb, GCP:

M20, 128Gb, AWS:

Notes:

Blue line is the Max Disk Write Latency
Green line is Average Disk Write Latency

Interpretation:

AWS has 10-20x lower Max Disk Write Latency (1-2ms stable, only 1-2 spikes within 1h up to 6ms ) compared to Azure (40-60ms, 1 spike within 1h up to 80 ms) and GCP (30-40ms, 2-3 spikes within 1h up to 50 ms). This is a gigantic difference!

Factors possibly affecting the results

There are a few hypotheses as to why the results could be incorrect:

Hypothesis 1: Burstability of M20 on Azure might lead to good results in shorter test durations, and worse results in longer ones
Hypothesis 2: There is another bottleneck in the load test infrastructure (K8s node utilization checked)
Hypothesis 3: Even though all setups use Network Peering between the K8s Cluster and MongoDB Atlas cluster it could be (due to routing issues) that the messages take another less-efficient route.

Conclusion

Overall AWS has a better performance compared to Azure and GCP.

Strangely enough, even though the IOPS numbers for GCP are the highest, it seems to offer the lowest performance.

Azure for sure has its problems with variability/outliers⁷, but performs relatively well. Having in mind that Azure becomes considerably more expensive with bigger instance/storage sizes, it seems that AWS and GCP are the winner in the M30/M40 range.

The real interesting finding IMHO is the far superior Max Write Disk Latency of AWS ... wondering if MongoDB Atlas is using direct attached storage in AWS vs. network-attached storage in Azure/GCP ...

P.S. I would appreciate any comments pointing out an obvious mistake leading to wrong results!

Including the load test automation/setup itself ↩
Synchronous = Client calls via HTTPS/REST Service1, Service1 acts as an orchestrator and calls synchronously Service2, Service3 etc. and awaits (async) their responses. Some of the inter-microservice calls are done in sequence, and some in parallel. ↩
"Transaction" does not mean database transaction or distributed database transaction. No database transactions are used (no database was harmed ;) ↩
Burstable, which means from 2 x 100% = 200% only 40% are provisioned, but accumulated credits allow bursting up to the full 200% from time to time ↩
Half of the IOPS are read IOPS and half are write IOPS ↩
Non-provisioned, the provisioned IOPS on AWS are very expensive ↩
See this post about various disk-related performance issues on Azure ↩

MongoDB Atlas & Azure - a forced marriage?

Deyan Petrov — Mon, 09 Aug 2021 09:59:41 +0000

TLDR; MongoDB Atlas on Azure (smaller instances with smaller storage) works but comes with a number of pitfalls you should be aware of. You'd save a lot of headaches hosting MongoDB Atlas on AWS/GCP instead. UPDATE 4th Nov 2024: MongoDB Atlas finally implemented Azure Premium SSD v2 and most performance issues have been solved!!!

Introduction

The intention of this post is to point out a number of issues we have experienced over the past 2 years using MongoDB Atlas on Azure, having 2 main objectives in mind ¹:

Shorten your/others' troubleshooting path if you happen to go the same way
Gather inputs from other customers of MongoDB Atlas on Azure

But you might ask immediately - why MongoDB Atlas on Azure instead of using the native CosmosDB? The reasons for us were:

Bloated document size + charging based on non-compressed data 0,25$/Gb/month - during our testing with CosmosDB a simple 180 byte json document was somehow taking 981 bytes in the end storage wise
Missing real atomic updates (findOneAndUpdate, supported by the MongoDB API for Cosmos DB) in the standard SQL API, in particular in stored procedures - see this and this for more information.
Partition-First is mandatory - every table must be partitioned, with the partition key usually unable to satisfy both good distribution and queries by another attribute. MongoDB allows you to start with single and unlimited partitions and you can add sharding later on.
Partitioning Limitations - 10k RUs and 20GB per partition only
Limited database transaction support - only within a single partition.

Important Notes:

This post is discussing only small instances - M10-M30 (no need/experience yet for/with bigger ones)!
Write Concern = Majority is assumed in below discussions. In our case written data has to mean "durable on at least 2 out of 3 nodes", as we cannot afford losing "written/ACK-ed" data upon node failover.

The following issues will be discussed in detail below:

Limited IOPS for small storage sizes when setting up new MongoDB Atlas cluster on Azure (UPDATE 4th Nov 2024 - resolved!)
Sudden Disk Latencies of up to 15 seconds, with missing/misleading metrics (UPDATE 4th Nov 2024 - resolved!)
Burstable CPU with Missing CPU Steal Metric
Oplog deletion upon storage downgrade
Different Operation Processing Time with different primary nodes
CPU Spikes every 15 minutes
Downtime during cluster upgrade (resolved)
Random node failovers (resolved)
Node "stalled" (froze) for 45 seconds
Sudden but long-lasting Disk Utilizition Percent/Disk Latencies

Limited IOPS for small storage sizes on new clusters (mostly resolved)

Inadequate storage IOPS when using small storage sizes (< 512Gb) is the elephant in the room.

When starting with MongoDB Atlas on Azure one of the first important questions I had was - is the pricing similar to that for AWS/GCP, or is it higher (I have experienced quite a few Azure services incl. basic infrastructure ones are in reality more expensive than AWS/GCP ...). At a first glance I thought it's all relatively similar:

MongoDB Atlas Pricing on Azure Netherlands (westeurope), small instances:

MongoDB Atlas Pricing on AWS Ireland (eu-west-1), small instances:

MongoDB Atlas Pricing on GCP Belgium (europe-west1), small instances:

Yes, Azure is still the most expensive hosting for MongoDB, but the difference does not seem that big at a first glance ... This is compounded though by the fact that performance is not the same - expand the details for M20 for example and check the IOPS value in there:

Note: IOPS for Azure have been increased from 120 to 1280 after MongoDB Atlas implemented Azure Premium SSD v2 in Q4 2024!

Did you notice the difference? Azure gives you only 1280 IOPS for starters vs. 2000+ IOPS on AWS and GCP.

Sudden Disk Latencies of up to 15 seconds, Missing/Misleading Metrics (resolved)

The story about low disk IOPS with small storage sizes does not end here though. The problem is made worse by having no way to see when the IOPS are throttled. This is how the Disk IOPS metric looks for the a M20 primary node:

Doesn't the above graph give you the impression that not 120, but even 20 IOPS will be enough? Yes ... but actually not, and you may see in db logs or in the MongoDB Atlas Profiler (visual) the following or worse:

(UPDATE 8th Oct 2021) MongoDB have added Max Values for all hardware metrics (thanks @MongoDB Atlas Team!), so now you can see more clearly how the IOPS are fully utilized from time to time (for some yet
unknown reason):

Note 1: The peak coincides with the regular every 15-minutes "chef scripts" run by MongoDB Atlas, see below for more info)
Note 2: Dev.to corrupts the resolution of the uploaded images ... otherwise you would see the higher blue-green line on the CPU metrics being iowait.

Important: Even with 128Gb (500 IOPS) and normal (=low) opcounters (no peaks!) we have experienced within 1 week 2 cases of single small document insert taking up to 7-15 seconds (yes, seconds!, instead of 6-8 ms in average) when something happens to the IOPS ..

After several km-long support cases based on scarce comments and internal metrics provided by MongoDB Support my conclusion is that IOPS limits are hit by the primary or secondary nodes when the write operation must be confirmed by primary + at least 1 secondary, and the secondary cannot confirm because the oplog has not been synced yet due to IOPS throttling.

(UPDATE 8th Oct 2021) Turns out all Azure Premium SSDs support bursting up to 3500 IOPS for up to 30 minutes (even confirmed by MongoDB Atlas Support Engineer!), but the $1mln question is then why, oh why, do we experience still these 7-15 seconds (not milliseconds) disk latency (insert of single small document, no db trx or anything) then??

Trying to find someone @MongoDB Atlas to access the underlying Azure VM Disk Metrics and check the values of the following ones, which can throw some light if burstable credits are exhausted from time to time (but I don't really believe that this is happening in our case ..):

Data Disk Used Burst IO Credits Percentage (Max)
OS Disk Used Burst IO Credits Percentage (Max)
Data Disk IOPS Consumed Percentage (Max)
OS Disk IOPS Consumed Percentage (Max)

(UPDATE 11th Oct 2021) Even after storage upgrade to 256Gb / 1100 IOPS still getting randomly hit by single-document insert/replace operations taking 100-200x more than usual, e.g. 1600+ ms instead of 6-8ms ... Happens couple of times per day when the load is relatively low - single-digit business operations per second, every business operation = about 10 single-document read/write db operations.

(UPDATE 4th Jan 2022) The last statement about the root cause of the intermittent disk latency issues is that the regular 15-minute Ansible monitoring/management scripts hit hard the OS Disk of the Standard_B2s VM (in case of M20) which causes a delayed (after about 7 minutes??) disk throttling affecting also the MongoDB process ...

(UPDATE 22th Feb 2022) After many months of investigations we are back at square 1: the Ansible monitoring/management process running every 15 minutes is not the reason for the slowdowns, root cause is not stated (=> unknown?), we should upgrade to M30 or change the cloud provider ...

(UPDATE 4th March 2022) Problem still occurring after upgrade to M30, contrary to statement that the problem would be solved by upgrade to M30 ...

(UPDATE 31st Oct 2022) Problem can NOT be resolved, issue lies with Azure Infrastructure (Disks) and not with MongoDB Atlas itself. Only options are move to AWS/GCP or wait for new generation Azure disks ...

(UPDATE 4th Nov 2024) MongoDB Support upgraded our existing clusters to Azure Premium SSD v2, and we finally see much more stable operation durations! Current IOPS Comparison is also a bit better for Azure, but still lagging from AWS/GCP:

Attribute	Azure	AWS (eu-west1)	GCP
M10 vCPUs	1	2	0.5
M10 RAM Gb	2	2	1.7
M10 Disk Size Gb	8-128	10-128	10-128
M10 IOPS	640	1000	600 (300 read + 300 write) - 7680 (3840 read + 3840 write)
M10 Custom IOPS?	no	no	no
M10 Max Connections	1500	1500	1500
M10 Network Performance	Low network performance	Up to 5Gb	Low to Moderate network performance
M20 vCPUs	2	2	1
M20 RAM Gb	4	4	3.75
M20 Disk Size	16-256	20-256	20-256
M20 IOPS	1280	2000	1200(600 read + 600 write) – 15000 (7680 read + 7680 write)
M20 Custom IOPS?	no	no	no
M20 Max Connections	3000	3000	3000
M20 Network Performance	Moderate network performance	Up to 5Gb	Moderate network performance
M30 vCPUs	2	2	2
M30 RAM Gb	8	8	8
M30 Disk Size	32-512	40-512	40-512
M30 IOPS	3200	3000	2400(1200 read + 1200 write) – 15000 (7680 read + 7680 write)
M30 Custom IOPS?	no	Yes, 10-3600	no
M30 Max Connections	3000	3000	3000
M30 Network Performance	High network performance	Up to 10Gb	High network performance

Burstable CPU with Missing CPU Steal Metric

M10 and M20 instances are using B-series Azure VMs (e.g. M20 is using Standard_B2s). These are burstable VMs where you have for Standard_B2s 40% CPU baseline performance guaranteed (40% = 2 vCPUs * 20% CPU utilization each). If you use less than 40% you accumulate credits and every credit gives you the right to burst above the 40% (e.g. to 100% = 1 vCPU fully utilized or 200% = both vCPUs fully utilized) for certain period of time until credits reach 0.

There is a CPU Steal % metric in MongoDB Atlas which should be 0 if all good, and should start increasing in case you need more CPU but you cannot get it because you are throttled to your baseline performance (= no available credits for bursting). An alert can be configured once this metric reaches certain threshold (e.g. above 0 for several minutes) ...

That is all fine and good, but the problem is that MongoDB Atlas seems to have implemented the CPU Steal % metric/alert only for AWS ... so in Azure there is no way to detect and alert such an important situation ...

Oplog deletion upon storage downgrade

While testing different combinations of instance size (e.g. M20, M30) and storage size (16, 64, 128Gb) we had to downgrade the storage a couple of times (e.g. from 128 back to 16Gb). Usually upgrade operations worked flawlessly - MongoDB Atlas takes one node after the other offline, replaces, syncs with primary and puts back in cluster, no data loss. In case of storage downgrade we lost all the data in the Oplog, which is of critical importance for our application built on top of MongoDB Change Streams for all the async event publishing functionality. This means our K8s pods waiting for change events lost their resume tokens (saved checkpoints) as the latter were pointing to non-existent oplog positions ...

According to MongoDB this happens when

"An Azure machine is migrated to a new instance family"
"A user requests a disk decrease on Azure instances"

and I have the feeling the above is again Azure-specific. My request for changing that behavior on Azure has not been honoured yet.

Different Operation Processing Time with different primary nodes

While testing with different instance and storage sizes I noticed that in a standard 3-node cluster I get different average business operation (1 business operation contains 2-3 single document find and 7-8 modify operations) processing times depending on which node is primary:

2nd Node is Primary:

3rd Node is Primary:

You see 15-20ms difference in average times, meaning up to 25%, which is huge! My suspicion is that this is because of some networking overhead due to putting the different nodes in different Availability Zones, however my client applications are running on 3 Kubernetes nodes split in the same way to the 3 different Azure Availability as the MongoDB nodes ... Wouldn't be surprised if this is another Azure idiosyncrasy ...

CPU Spikes every 15 minutes

This is how the CPU looks like on all of our M20 clusters:

What are these regular CPU spikes every 15 minutes you might ask - is there some heavy regular application activity? The answer is no, our application is not doing anything, however there is some MongoDB Atlas Monitoring cron job (aka "Chef scripts") which is doing some heavy work every 15 minutes. Remember: M20 is a burstable instance with baseline of 40%, so every 15 minutes this monitoring process is "stealing" 1-2 minutes of CPU, for which you have paid ...
I had a couple of tickets with MongoDB Support on this topic suspecting that this is the reason for the randomly slow operation processing times, not clear if it is correlated or not (several disk latency occurrences happened at the same time, but a few also not), however it still feels like unoptimized admin intervention ...

Downtime during cluster upgrade (resolved)

With MongoDB Azure we were suffering for more than 6-months from downtimes (applications could not connect anymore to the cluster and needed to be manually restarted!) during cluster maintenances. This happened exactly at the point of time when the 3rd node was getting replaced (all fine with 1st and 2nd).
It took tons of discussions (incl. with an Account Manager) and paying for Professional Services until this problem was fixed in the .NET Driver, but only after another big customer complained.

Random node failovers

For a week now we are experiencing random node failovers. Activity Feed in Atlas Portal shows only this:

The current explanation by MongoDB Support (as far as I have understood it) is that there are network issues between the nodes and a node triggers re-election. But how come one of the most stable cloud components (network) is having so many outages, and what is the resolution? No answer yet ...

(Update 8 Oct 2021): After MongoDB Atlas Support talked to Azure Support network reconfiguration was performed. Since then we have not experienced additional unexpected failovers, so (assuming this has been integrated in the setup scripts for new MongoDB Atlas clusters on Azure) the problem can be treated as resolved.

(Update 22 Feb 2022): We had another such case 2 weeks ago, which caused some of our business operations to take 25 seconds instead of 70 ms (still were successful though). Turns out the network configuration performed previously was not persistent (after VM reboot gone), which I was not told at the time. Now MongoDB Support has repeated the network config but in a persistent manner, let's see ...

Node "stalled" (froze) for 45 seconds

Another case from past week - the primary node just "stalled" or froze for 45 seconds, and then continued working. Of course, during that time all business operations were affected, and some of them timed out. According to MongoDB Support something happened to the node's disk - it just physically stopped working. What can be done so that this does not happen again? Nothing ... if it happens again the node can be replaced ...

And another case of 3 seconds disk "stalling" or whatever - disk queue went up, db operations got 100 slower:

Was this due to exceeding Disk IOPS and being throttled by Azure Storage - if the metric is to be believed (and I don't) - No ..:

I understand, that cloud VMs may go away or freeze in case the host crashes, but somehow too many strange things happen lately with our MongoDB Atlas clusters on Azure - can Azure be that unstable??

Sudden but long-lasting Disk Utilizition Percent/Disk Latencies

A picture is worth a thousand words:

Disk Utilization % - from all processes running on the node

Disk Latency

Opcounters - absolutely no load ...

Happens already for the 2nd time in the past few days, the solution last time was to fail over manually to another node, and wait 30-45 minutes (!??!)

Conclusion

The astute reader may have already concluded from the above that the performance and stability of the M20 MongoDB Atlas instance on Azure is a joke, and this due to disk-related issues.

As recommended multiple times by MongoDB Support and Account Management, MongoDB Atlas should rather be used in conjunction with hosting on AWS/GCP where it seems to be much more stable, fast and also cheap.

This is something that was not crystal-clear to me when I decided for Azure and implemented the rest of the infrastructure there. Also a quick look at the different MongoDB Atlas hosting options did not explicitly warn me that one gets much less IOPS and a bunch of additional problems on Azure.

If CosmosDB didn't have some of the issues (bloated/expensive storage, lack of atomic update, enforced partitioning from the start, etc.) we would have moved to it long ago. Migration to GCP/AWS is something I will be actively investigating, however there are some goodies we are using (AKS, App Insights, Azure Data Explorer/Kusto) which need more work.

I have read that Azure has a weak offering of low-level IaaS, but why it is so lagging behind AWS/GCP when it comes to IOPS for smaller disks is beyond my understanding (read about the software layer they have put on top of it, but I don't care). Or rather - I cannot wrap my head around the question why Azure is not working day and night to fix this big gap in such an important fundamental/enabling service.

I wish Azure could fix this and additionally expose CPU Steal % and other metrics, so that MongoDB Atlas could level up its Azure hosting. I wish additionally that MongoDB Atlas could invest a little bit more in its Azure hosting (15 minutes heavy cron jobs can be optimized, additional metrics can be added even with the current Azure API I guess). But I have learnt the hard way that such wishes usually end up sitting in glorious Product Feedback Lists for ages ...

Have you experienced similar issues like the above? Or have you found other solutions? Would be happy to get such input from other Azure MongoDB Atlas customers!

P.S. Please vote for the following MongoDB feedback ideas:

OK, you got me, additionally I have a secret hope that if someone from MongoDB Atlas and Azure reads this (s)he might trigger some internal improvement ... but realistically this never works that way, we all know that :( ↩

F# App Stub for AKS hosting (with WebJobs but without Azure Functions fluff)

Deyan Petrov — Tue, 01 Jun 2021 17:01:58 +0000

TLDR; Use a standard .NET 5 host which gives you full control (incl. upgradability!) and still allows for WebJobs syntactic sugar, instead of the fluffy and "magical" Azure Functions host.

Introduction

In a previous post I listed a number of reasons why you'd better migrate away from Azure Functions and host your .NET (F# recommended!) apps in AKS. In this post I will try to give you as much code as you need to be able to start with this alternative approach in a matter of minutes instead of hours/days.

I will create a fully-blown stub including all the functionality you may need - WebJobs (EventHubTrigger, QueueTrigger, TimerTrigger), Web Server, custom Background Services, App Insights integration, Console Log Formatting, so it will look as close to Azure Functions as possible, but with full control.

Note: Azure Functions support for .NET 5 (the so-called Isolated Host) has actually taken a similar approach, but there is still too much fluff/magic for my liking ..

Design Approach/Principles

Mimic Azure Functions setup - including the usage of WebJobs (used by Azure Functions themselves), Application Insights integration, even the same or slightly better console log formatting!
"Close to the metal" or as basic as possible - just use Asp.Net Core, no framework like Saturn, Giraffe etc. required for just a few PUT/POST/PATCH/GET operations.
Utilize standard HostBuilder functionality like running additional HostedServices or BackgroundServices.

Program.fs

If you don't have prior knowledge about HostBuilder then in a nutshell it is just a configurable way of spawning several threads/application services, e.g. a web server, some background services used by the WebJobs SDK for listening to Azure Storage Queues or Event Hubs for example, some other custom background processes of yours (e.g. for watching a MongoDB collection) or even some startup code which runs when the whole application starts (e.g. for filling an in-memory cache from a db table).

The main bootstrapping code is in the main function and may look like this:

[<EntryPoint>]
let main argv =
    let builder =
        Host.CreateDefaultBuilder(argv)
        |> Framework.Hosting.HostBuilder.configureLogging
        |> Framework.Hosting.HostBuilder.configureAppInsights
        |> configureWorker1
        |> configureWorker2
        |> Framework.Hosting.HostBuilder.configureWebHost configureEndpoints
        |> configureWebJobs

    use tokenSource = new CancellationTokenSource()    
    use host = builder.Build()
    host.RunAsync(tokenSource.Token) |> Async.AwaitTask |> Async.RunSynchronously

    0 // return an integer exit code

As you see from above, we are configuring logging, App Insights integration, then 2 background workers (IHostedServices), the web server with 3 "HttpTriggers", and web jobs which actually means EventHubTriggers, QueueTriggers, TimerTriggers etc.

WebJobs Configuration

WebJobs configuration is extremely straightforward:

let configureWebJobs (builder:IHostBuilder) = 
    builder.ConfigureWebJobs(fun b ->
        b.AddAzureStorageCoreServices() |> ignore
        b.AddEventHubs() |> ignore
        b.AddAzureStorageQueues() |> ignore
        b.AddTimers() |> ignore)

Additional services can be configured as well, e.g. b.AddSignalR() |> ignore, etc.

Web Server Configuration

Standard Kestrel configuration ... basic mapping of GET/PUT/POST/PATCH to functions, with manual injection of dependencies (e.g. logger) as a function parameter (even partial application not required):

let configureEndpoints (app:IApplicationBuilder) (endpoints:IEndpointRouteBuilder) =
    let hostedServices = Framework.BackgroundService.Hosted.findAll app.ApplicationServices

    let worker1 = Framework.BackgroundService.Hosted.find hostedServices "Worker1"
    endpoints.MapGet("api/v1/background-services/worker1/status", fun context ->
        worker1.GetProcessingStatusHttp()
        |> MappedHttpResponse.toHttpResponse context
        :> Task) |> ignore

    let worker2 = Framework.BackgroundService.Hosted.find hostedServices "Worker2"
    endpoints.MapGet("api/v1/background-services/worker2/status", fun context ->
        worker2.GetProcessingStatusHttp()
        |> MappedHttpResponse.toHttpResponse context
        :> Task) |> ignore

    endpoints.MapGet("api/v1/webFunction1", fun context ->
        let logger = app.ApplicationServices.GetService<ILogger<Object>>()

        WebApi.webFunction1 logger context.Request
        |> Async.StartAsTask
        |> Task.bind (MappedHttpResponse.toHttpResponse context)
        :> Task) |> ignore

let configureWebHost configureEndpoints (builder:IHostBuilder) : IHostBuilder =
    builder.ConfigureServices(fun context services ->
        if context.HostingEnvironment.IsDevelopment() then
            services.AddCors(fun options ->
                options.AddDefaultPolicy(fun b ->
                    b
                        .AllowCredentials()
                        .WithOrigins("http://localhost:8080")
                        .WithHeaders("authorization","content-type","if-match","etag","content-disposition","x-requested-with")
                        .WithExposedHeaders("authorization","content-type","if-match","etag","content-disposition")
                        .WithMethods("GET","POST","PUT","PATCH")
                    |> ignore
                )
            ) |> ignore
        ) |> ignore

    builder.ConfigureWebHostDefaults(fun b ->
        b.Configure(fun context app ->
            app.UseRouting() |> ignore

            if context.HostingEnvironment.IsDevelopment() then        
                app.UseCors() |> ignore

            app.UseEndpoints (fun endpoints -> configureEndpoints app endpoints) |> ignore

        ) |> ignore
    )

Bonus: As you see above there are 2 status web functions which report the status of the threads started by the background services (see next section).

Custom Background Services Configuration

You can do anything in a HostedService/BackgroundService, even get rid of the WebJobs SDK and simply listen to event hubs using EventHubProcessor for example by yourself (I have always regarded XxxTrigger as a non-mandatory syntactic sugar ...):

let startedEvent1 = new ManualResetEvent(false) 

let configureWorker1 (builder:IHostBuilder) : IHostBuilder =
    builder.ConfigureServices(fun context services ->
        services.AddSingleton<IHostedService>(
            fun serviceProvider -> 
                let logger = serviceProvider.GetService<ILogger<Object>>()
                Framework.BackgroundService.Hosted.create
                    "Worker1"
                    (BackgroundServices.worker1 logger startedEvent1) :> IHostedService) |> ignore
        )

let configureWorker2 (builder:IHostBuilder) : IHostBuilder =
    builder.ConfigureServices(fun context services ->
        services.AddSingleton<IHostedService>(
            fun serviceProvider -> 
                let logger = serviceProvider.GetService<ILogger<Object>>()
                Framework.BackgroundService.Hosted.create
                    "Worker2"
                    (BackgroundServices.worker2 logger startedEvent1) :> IHostedService) |> ignore
        )

Logging Configuration, including Console Log Formatting

To simulate the Azure Functions logger formatting I had to unfortunately implement a full-blown ConsoleFormatter ... but also made the log output a bit more-concise on the way ;)

let configureLogging (builder:IHostBuilder) : IHostBuilder =
    builder.ConfigureLogging(fun context b ->
//        b.ClearProviders() |> ignore  // commented out because otherwise Application Insights does not collect logger.Information ...
        b.AddConsole(fun options -> options.FormatterName <- "CustomConsoleFormatter")
            .AddConsoleFormatter<Framework.Logging.Console.CustomConsoleFormatter, Framework.Logging.Console.CustomConsoleFormatterOptions>(fun options ->
                options.Excludes <- Framework.Logging.Console.defaultExcludes
                options.SingleLinePerCategory <- ["Microsoft.Azure.WebJobs.Hosting.OptionsLoggingService", false; "Host.Triggers.Timer", false] |> dict
                options.SingleLine <- true
                options.ColorBehavior <- LoggerColorBehavior.Enabled)
        |> ignore
    )

Application Insights Integration

Application Insights works here as well, note that I am injecting a custom initializer and processor to suppress some http status codes etc.

let configureAppInsights (builder:IHostBuilder) : IHostBuilder =    
    builder.ConfigureServices(fun context services ->
        let aiOptions = ApplicationInsightsServiceOptions(EnableAdaptiveSampling = false)   // disable sampling (turned on by default, see https://docs.microsoft.com/en-us/azure/azure-monitor/app/sampling#brief-summary)
        services.AddApplicationInsightsTelemetry(aiOptions) |> ignore
        services.AddSingleton<ITelemetryInitializer>(
            Framework.Logging.Telemetry.Configuration.CustomTelemetryInitializer(
                Environment.GetEnvironmentVariable("WEBSITE_SITE_NAME", true),
                Framework.Logging.Telemetry.Configuration.defaultSuppressedStatusCodes))
        |> ignore

        services.AddSingleton<Framework.Logging.Telemetry.Configuration.CustomTelemetryProcessorConfig>(Framework.Logging.Telemetry.Configuration.defaultCustomTelemetryProcessorConfig) |> ignore
        services.AddApplicationInsightsTelemetryProcessor<Framework.Logging.Telemetry.Configuration.CustomTelemetryProcessor>() |> ignore
    )

Custom Functions

And the last bit are the user-defined functions (="Azure Functions):



module BackgroundServices =
    let worker1 (logger:ILogger) (startedEvent: ManualResetEvent) (token:CancellationToken) : Async<unit> =
        async {
            startedEvent.Set() |> ignore
            while true do
                do logger.LogInformation("Hello from worker1")
                Thread.Sleep(5000)
        }        
    let worker2 (logger:ILogger) (startedEvent: ManualResetEvent) (token:CancellationToken) : Async<unit> =
        async {
            startedEvent.Set() |> ignore
            while true do
                do logger.LogInformation("Hello from worker2")
                Thread.Sleep(5000)
        }        

module WebApi =
    let webFunction1 (logger:ILogger) (req: HttpRequest) : Async<MappedHttpResponse> =
        async {
            do logger.LogInformation("webFunction1")
            return {
                StatusCode = 200
                Content = "Some response" |> MappedHttpResponseContent.Json
                Headers = List.empty
            }
        }

type WebJobs(telemetryConfiguration: TelemetryConfiguration) =
    let telemetryClient = TelemetryClient(telemetryConfiguration)

    [<FunctionName("HandleEventHubMessage")>]
    member _.HandleEventHubMessage
        ([<EventHubTrigger("", Connection = "EventHubConnectionString", ConsumerGroup = "test-cg")>]    // path to event hub is in the connection string
        msg: EventData,
        enqueuedTimeUtc: DateTime,
        sequenceNumber: Int64,
        offset: string,
        logger: ILogger)
        =
        async {
            do logger.LogInformation($"HandleEventHubMessage: {Encoding.UTF8.GetString(msg.Body.ToArray())}")
        } |> Async.StartAsTask

    [<FunctionName("HandleQueueMessage")>]
    member _.HandleQueueMessage
        ([<QueueTrigger("test-queue", Connection = "StorageQueueConnectionString")>] msg: string,
        logger:ILogger)
        =
        async {
            do logger.LogInformation($"HandleQueueMessage: {msg}")
        } |> Async.StartAsTask

    [<FunctionName("HandleTimerEvent")>]
    member this.HandleTimerEvent
        ([<TimerTrigger("0 0 0 1 * *", RunOnStartup = true)>] timer: TimerInfo,
        logger: ILogger)
        =
        async {
            do logger.LogInformation("HandleTimerEvent")
        } |> Async.StartAsTask

As you see, not much difference compared to "real" Azure Functions ...

Migration from Azure Functions

Add Program.fs if not already existing, and copy over the host builder pipeline (including the required WebJobs AddXxxx services)
Configure WebJobs (no change required to corresponding XxxTriggers)
Implement configureEndpoints, and remove HttpTrigger attributes from the http-based azure functions (no change required for the non-http azure functions, attributes remain)

Conclusion

With greater control comes bigger responsibility, however you can see from the above that the required code is not only relatively easy to grasp and maintain, but it offers much more! For example, background services can be injected nicely, and a lot of additional HostBuilder functionality can be used - there are tons of Q&As and documentations online. Last but not least, after migrating from Azure Functions v3 to the above "custom .NET host" we experienced a 50% drop in memory consumption of the AKS pods which I believe is attributable to the removed Azure Functions bloat, so that was another bonus for us.

P.S. Working project can be found at https://github.com/deyanp/FSharpAKSAppStub.

BDD-style Testing in F# with Xunit.Gherkin, GherkinProvider and TickSpec

Deyan Petrov — Mon, 31 May 2021 13:56:24 +0000

TLDR: Use TickSpec in F# for BDD-style tests, and utilize the classless or module/function-based approach with a Context record type being passed between the individual step functions.

Note: The below requires basic knowledge of Behavior-Driven Development/ Specification By Example / Gherkin.

Over the years BDD has been helping me immensely to document the software systems I build "for free" while writing functional tests for them. Naturally, after moving to F#, one of the first topics for me was what is the most suitable BDD framework for F#?
I had prior experience with SpecFlow in C#, and even though SpecFlow is an amazing tool with excellent IDE integration, sometimes it was getting in my way due to its code-generation approach, and its general "heaviness". So not only did I want to find a BDD framework in F#, but I wished it were a bit more lightweight than SpecFlow.

This blog post contains a short analysis of the following 3 approaches, but starts with a plain-old Xunit just for the sake of basic introduction:

1) Xunit.GherkinQuick
2) GherkinProvider
3) TickSpec

But before we start - what are we testing? I have selected a set of very simple functions based on NCrontab:

module Framework.Crontab

open System
open NCrontab

/// Parses 5 (standard) or 6 (with seconds) component crontab expressions
let parse (crontabExpression:string) : CrontabSchedule =
    let crontabComponentCount = crontabExpression.Split(" ", StringSplitOptions.RemoveEmptyEntries).Length
    CrontabSchedule.Parse (crontabExpression, CrontabSchedule.ParseOptions ( IncludingSeconds = (crontabComponentCount = 6) ))

/// For start and stop crontabs, e.g. start every day in the morning and stop every day in the evening
let isRunning
    startScheduleCrontab
    stopScheduleCrontab
    maxStartedDurationHours
    maxStoppedDurationHours
    (pointInTime:DateTime)
    =
    let startSchedule = parse startScheduleCrontab
    let stopSchedule = parse stopScheduleCrontab

    let lastStart = startSchedule.GetNextOccurrences(pointInTime.AddHours(-1.0 * maxStartedDurationHours), pointInTime) |> Seq.tryLast
    let lastStop = stopSchedule.GetNextOccurrences(pointInTime.AddHours(-1.0 * maxStoppedDurationHours), pointInTime) |> Seq.tryLast
    //let nextStart = startSchedule.GetNextOccurrences(pointInTime, pointInTime.AddHours(maxStoppedDurationHours)) |> Seq.tryHead
    //let nextStop = stopSchedule.GetNextOccurrences(pointInTime, pointInTime.AddHours(maxStartedDurationHours)) |> Seq.tryHead

    match lastStart, lastStop with
    | Some lastStart, Some lastStop ->
        lastStart > lastStop 
    | None, _ ->
        failwithf "LastStart could not be found using startScheduleCrontab %s and maxStartedDurationHours %f" startScheduleCrontab maxStartedDurationHours
    | _, None ->
        failwithf "LastStop could not be found using stopScheduleCrontab %s and maxStoppedDurationHours %f" stopScheduleCrontab maxStoppedDurationHours

/// For single execution crontabs, e.g. once daily
let getLastRunOn
    runScheduleCrontab
    maxRunIntervalHours
    (pointInTime:DateTime)
    : DateTime    
    =
    let runSchedule = parse runScheduleCrontab
    let lastRunOn = runSchedule.GetNextOccurrences(pointInTime.AddHours(-1.0 * maxRunIntervalHours), pointInTime) |> Seq.tryHead
    match lastRunOn with
    | Some lastRunOn ->
        lastRunOn 
    | None ->
        failwithf "LastRun could not be found using runScheduleCrontab %s and maxRunIntervalHours %f" runScheduleCrontab maxRunIntervalHours

In a nutshell, given a crontab expression for a job with start and stop schedules the function isRunning is telling you whether the job is running for a given point in time. The function getLastRun gives you the last time a single execution job has run calculated from a given point in time.

Plain-old Xunit

Below is how this could be implemented without BDD-style feature file with Given/When/Thens, just with code:

module Framework.Crontab.Tests.Xunit

open System
open Xunit
open Framework.SimpleTypes.TypeExtensions

let parseExamples =
    [|
        [| "0 0 2 * *" |]
        [| "0 0 2 * * *"  |]
    |]

[<Theory; MemberData("parseExamples")>]
let ``Crontab Parsing`` (crontabExpression:string) =
    //When crontab expression <CrontabExpression> is parsed
    let crontabSchedule = Framework.Crontab.parse crontabExpression

    //Then the CrontabSchedule has the same string representation
    Assert.True(crontabSchedule.ToString() = crontabExpression, "Parsed crontabSchedule.ToString() not equal to original crontab expression")

let isRunningExamples =
    [|
        [| DateTime.ParseUtcIso("2020-10-21T00:00:00Z") :> obj; false :> obj |]
        [| DateTime.ParseUtcIso("2020-10-21T08:00:00Z"); true |]
    |]

[<Theory; MemberData("isRunningExamples")>]
let ``Running/Not Running Interval is identified successfully`` (pointInTime:DateTime) (isRunningExpected:bool) =
    //When isRunning is called at <PointInTime> with the following parameters
    let isRunningActual = 
        Framework.Crontab.isRunning
            "0 7 * * 1-5"
            "0 16 * * 1-5"
            72.0
            72.0
            pointInTime

    //Then the isRunning result is <Result>
    Assert.Equal(isRunningExpected, isRunningActual)

let getLastRunOnExamples =
    [|
        [| DateTime.ParseUtcIso("2020-10-21T00:00:00Z") :> obj; DateTime.ParseUtcIso("2020-10-20T02:00:00Z") :> obj |]
        [| DateTime.ParseUtcIso("2020-10-21T08:00:00Z"); DateTime.ParseUtcIso("2020-10-21T02:00:00Z") |]
    |]

[<Theory; MemberData("getLastRunOnExamples")>]
let ``Last Run DateTime is retrieved successfully`` (pointInTime:DateTime) (lastRunOnExpected:DateTime) =
    //When getLastRunOn is called at <PointInTime> with the following parameters
    let lastRunOnActual =
        Framework.Crontab.getLastRunOn
            "0 0 2 * * *"
            24.0
            pointInTime

    //Then the getLastRunOn result is <Result>
    Assert.Equal(lastRunOnExpected, lastRunOnActual)

The points of interest here are:

Theory with MemberData is used to feed the test function with multiple inputs and expected outputs
Comments are used to identify the Given/When/Then (or Arrange/Act/Assert) parts of the test, as all of these are bundled together in the same F# function
The xunit test (=function) names cannot generally accommodate Given/When/Then, as then they would become too long ...

Before I start listing the BDD/Gherkin testing options, here is the feature file containing the Gherkin language-based scenarios:

Feature: Crontab Tests

  Scenario Outline: Crontab Parsing
    When crontab expression <CrontabExpression> is parsed
    Then the CrontabSchedule has the same string representation

    Examples:
      | CrontabExpression |
      | 0 0 2 * *         |
      | 0 0 2 * * *       |

  Scenario Outline: Running/Not Running Interval is identified successfully
    When isRunning is called at <PointInTime> with the following parameters
      | Parameter               | Value        |
      | startScheduleCrontab    | 0 7 * * 1-5  |
      | stopScheduleCrontab     | 0 16 * * 1-5 |
      | maxStartedDurationHours | 72           |
      | maxStoppedDurationHours | 72           |
    Then the isRunning result is <Result>

    Examples:
      | PointInTime          | Result |
      | 2020-10-21T00:00:00Z | false  |
      | 2020-10-21T08:00:00Z | true   |

  Scenario Outline: Last Run DateTime is retrieved successfully
    When getLastRunOn is called at <PointInTime> with the following parameters
      | Parameter           | Value       |
      | runScheduleCrontab  | 0 0 2 * * * |
      | maxRunIntervalHours | 24          |
    Then the getLastRunOn result is <Result>

    Examples:
      | PointInTime          | Result               |
      | 2020-10-21T00:00:00Z | 2020-10-20T02:00:00Z |
      | 2020-10-21T08:00:00Z | 2020-10-21T02:00:00Z |

Xunit.Gherkin.Quick

There is a hidden gem for Xunit called Xunit.Gherkin.Quick which allows you to create standard feature files using the Gherkin language, and automate these with Xunit-based tests.

As usual a nuget package reference is required:

<PackageReference Include="Gherkin.TypeProvider" Version="0.1.10" />

As usual a nuget package reference is required:

<PackageReference Include="Xunit.Gherkin.Quick" Version="4.1.0" />

The feature file must be in some folder (e.g. in the project folder), but it does not need to be added to the fproj file in a special way (Content/EmbeddedResource).

Here is how the code based on Xunit.Gherkin.Quick could look like:

module Framework.Crontab.Tests.XunitGherkinQuick

open System
open Gherkin.Ast
open NCrontab
open Xunit
open Xunit.Gherkin.Quick
open Framework.SimpleTypes.TypeExtensions

module Table =
    let getValueByKey key (dt:DataTable) : string =
        dt.Rows
        |> Seq.find (fun x ->
            let cells = x.Cells |> List.ofSeq
            cells.[0].Value = key)
        |> (fun r -> r.Cells)
        |> Seq.last
        |> (fun c -> c.Value)

[<Xunit.Gherkin.Quick.FeatureFile("./CrontabTests.feature")>]
type CrontabTests() =
    inherit Feature()

    let mutable ctxCrontabExpression = None
    let mutable ctxCrontabSchedule : CrontabSchedule option = None

    let mutable isRunningActual: bool option = None

    let mutable lastRunOnActual: DateTime option = None

    [<When("crontab expression (.+) is parsed")>]
    member this.``When crontab expression <CrontabExpression> is parsed`` (crontabExpression:string) =
        ctxCrontabExpression <- crontabExpression |> Some
        ctxCrontabSchedule <- Framework.Crontab.parse crontabExpression |> Some

    [<Then("the CrontabSchedule has the same string representation")>]
    member this.``Then the CrontabSchedule has the same string representation`` () =
        Assert.True(ctxCrontabSchedule.Value.ToString() = ctxCrontabExpression.Value, "Parsed crontabSchedule.ToString() not equal to original crontab expression")

    [<When("isRunning is called at (.*) with the following parameters")>]
    member this.``When isRunning is called at <PointInTime> with the following parameters`` (pointInTime:string) (datatable:DataTable) =
        isRunningActual <- 
            Framework.Crontab.isRunning
                (Table.getValueByKey "startScheduleCrontab" datatable |> string)
                (Table.getValueByKey "stopScheduleCrontab" datatable |> string)
                (Table.getValueByKey "maxStartedDurationHours" datatable |> Convert.ToDouble)
                (Table.getValueByKey "maxStoppedDurationHours" datatable |> Convert.ToDouble)
                (pointInTime |> DateTime.ParseUtcIso)
            |> Some

    [<Then("the isRunning result is (.*)")>]
    member this.``Then the isRunning result is <Result>`` (isRunningExpected:bool) =
        Assert.Equal(isRunningExpected, isRunningActual.Value)

    [<When("getLastRunOn is called at (.*) with the following parameters")>]
    member this.``When getLastRunOn is called at <PointInTime> with the following parameters`` (pointInTime:string) (datatable:DataTable) =
        lastRunOnActual <-
            Framework.Crontab.getLastRunOn
                (Table.getValueByKey "runScheduleCrontab" datatable |> string)
                (Table.getValueByKey "maxRunIntervalHours" datatable |> Convert.ToDouble)
                (pointInTime |> DateTime.ParseUtcIso)
            |> Some

    [<Then("the getLastRunOn result is (.*)")>]
    member this.``Then the getLastRunOn restheult is <Result>`` (lastRunOnExpected:string) =
        Assert.Equal(lastRunOnExpected |> DateTime.ParseUtcIso, lastRunOnActual.Value)

The points of interest here are:

The feature file is referenced using a class attribute, which automatically means you must have 1 class per feature file (which turns out to be a good thing)
The function names can be anything, here I have just used 1:1 the step text, because the actual mapping between a step (= line in the feature file) and a method in the implementation is done by using method-level attributes. This is a bit of a hassle, because you wonder how to name your methods ...
Mutable variables are used to transfer state from Given->When->Then step functions

GherkinProvider

Gherkin Provider is a F# Type Provider, which means it automatically generates types based on the feature file while you are typing in your IDE - magic! ;) I was (and still am) very excited when I first saw it, as it has the promise to bring the best of 2 worlds:

Scenarios + Examples extracted into a very readable feature file
Code for Given/When/Then residing in the same function!

As usual a nuget package reference is required:

<PackageReference Include="Gherkin.TypeProvider" Version="0.1.10" />

The feature file must be added to the project as Content, e.g.:

<Content Include="CrontabTests.feature" />

Below an implementation with it:

module Framework.Crontab.Tests.GherkinProvider

open System
open Xunit
open FSharp.Data.Gherkin
open GherkinProvider.Validation
open Xunit.Extensions.Ordering
open Framework.SimpleTypes.TypeExtensions

type TestFeature = GherkinProvider<const(__SOURCE_DIRECTORY__ + "/CrontabTests.feature")>
let feature = TestFeature.CreateFeature()

let assertScenarioInSync (scenario:TestFeature.TestFeature_ScenarioBase) =
    Assert.True(scenario.Visited)
    scenario.Steps |> Array.iter (fun x -> Assert.True(x.Visited))

let parseExamples =
    feature.Scenarios.``Crontab Parsing``.Examples
    |> Array.map (fun x -> [|x.CrontabExpression.Value|])

[<Theory; MemberData("parseExamples")>]
let ``Parsing of Crontabs works`` (crontabExpression:string) =
    let scenario = feature.Scenarios.``Crontab Parsing``

    scenario.``0 When crontab expression <CrontabExpression> is parsed`` |> ignore
    let crontabSchedule = Framework.Crontab.parse crontabExpression

    scenario.``1 Then the CrontabSchedule has the same string representation`` |> ignore
    Assert.True(crontabSchedule.ToString() = crontabExpression, "Parsed crontabSchedule.ToString() not equal to original crontab expression")

    assertScenarioInSync scenario

let isRunningExamples =
    feature.Scenarios.``Running_Not Running Interval is identified successfully``.Examples
    |> Array.map (fun x -> [|x.PointInTime.Value; x.Result.Value|])

[<Theory; MemberData("isRunningExamples")>]
let ``Running/Not Running Interval is identified successfully`` (pointInTime:string) (isRunningExpected:string) =
    let scenario = feature.Scenarios.``Running_Not Running Interval is identified successfully``

    scenario.``0 When isRunning is called at <PointInTime> with the following parameters`` |> ignore
    let getValueByKey key =
        scenario.``0 When isRunning is called at <PointInTime> with the following parameters``.Argument
        |> Array.filter (fun i -> i.Parameter.Value = key)
        |> Array.head
        |> (fun x -> x.Value.Value)

    let isRunningActual = 
        Framework.Crontab.isRunning
            (getValueByKey "startScheduleCrontab" |> string)
            (getValueByKey "stopScheduleCrontab" |> string)
            (getValueByKey "maxStartedDurationHours" |> Convert.ToDouble)
            (getValueByKey "maxStoppedDurationHours" |> Convert.ToDouble)
            (pointInTime |> DateTime.ParseUtcIso)

    scenario.``1 Then the isRunning result is <Result>`` |> ignore
    Assert.Equal(isRunningExpected |> bool.Parse, isRunningActual)

    assertScenarioInSync scenario

let getLastRunOnExamples =
    feature.Scenarios.``Last Run DateTime is retrieved successfully``.Examples
    |> Array.map (fun x -> [|x.PointInTime.Value; x.Result.Value|])

[<Theory; MemberData("getLastRunOnExamples")>]
let ``Last Run DateTime is retrieved successfully`` (pointInTime:string) (lastRunOnExpected:string) =
    let scenario = feature.Scenarios.``Last Run DateTime is retrieved successfully``

    scenario.``0 When getLastRunOn is called at <PointInTime> with the following parameters`` |> ignore
    let getValueByKey key =
        scenario.``0 When getLastRunOn is called at <PointInTime> with the following parameters``.Argument
        |> Array.filter (fun i -> i.Parameter.Value = key)
        |> Array.head
        |> (fun x -> x.Value.Value)    

    let lastRunOnActual =
        Framework.Crontab.getLastRunOn
            (getValueByKey "runScheduleCrontab" |> string)
            (getValueByKey "maxRunIntervalHours" |> Convert.ToDouble)
            (pointInTime |> DateTime.ParseUtcIso)
        |> DateTime.ToStringIso

    scenario.``1 Then the getLastRunOn result is <Result>`` |> ignore
    Assert.Equal(lastRunOnExpected |> DateTime.ParseUtcIso, lastRunOnActual |> DateTime.ParseUtcIso)

    assertScenarioInSync scenario

[<Fact; Order(Int32.MaxValue)>]
let validateFeatureVisited () =
    let validator = FeatureValidator()
    match validator.Validate feature with
    | None -> ()
    | Some report -> failwith(report.Summary)

Points of interest:

When you load the feature file it autogenerates a type with a number of neested types (line 10)
The examples are passed to the functions by using standard Xunit Theory + Member Data, very low-level, no magic
Because the sub-types are autogenerated, you cannot easily create functions operating on different subtypes
There are strange statements inside the test functions like scenario. \1 Then the getLastRunOn result is <Result>\|> ignore and there is the validateFeatureVisited invoked at the end (that's why Order(Int32.MaxValue) attribute[^1]) to make sure that all features are visited. I have nothing against the former, as they serve as strongly-typed comments (compare with plain-old Xunit sample at the beginning), but validateFeatureVisited is a bit too much for me ...

TickSpec

TickSpec is the dinasour in the room (dating back to 2010!!) but remarkably straightforward (compared to SpecFlow). It does not use code-generation magic, but it has a little magic related to how steps (in feature files) are mapped to functions and how step parameters (e.g. tables) are mapped to function parameters, and how parameters are passed between Given/When/Then functions.

As usual a nuget package reference is required:

<PackageReference Include="TickSpec" Version="2.0.0"/>

The feature file must be added to the project as Content, e.g.:

<EmbeddedResource Include="CrontabTests.feature" />

An implementation of automated tests for the Crontab functions looks like this:

Features.fs

namespace Framework.Crontab.Tests.TickSpec

open System.Diagnostics
open Framework.Crontab.Tests.TickSpec
open Xunit
open Framework.Testing.TickSpecXunitWiring.Version2

type Features() =
    static let source = AssemblyStepDefinitionsSource(System.Reflection.Assembly.GetExecutingAssembly())
    static let scenarios resourceName = source.ScenariosFromEmbeddedResource resourceName |> MemberData.ofScenarios

    [<Theory; MemberData("scenarios", "Framework.Crontab.Tests.CrontabTests.feature")>]
    member this.CrontabTests (scenario : XunitSerializableScenario) = 
        source.ScenarioAction(scenario).Invoke()

CrontabTests.TickSpec.fs

module Framework.Crontab.Tests.CrontabTests.TickSpec

open System
open NCrontab
open Xunit
open TickSpec
open Framework.Testing.Table
open Framework.SimpleTypes.TypeExtensions

type ParseContext = {
    CrontabExpression : string
    ParsedCrontabSchedule : CrontabSchedule
} 

let [<When>] ``crontab expression (.*) is parsed`` (crontabExpression:string) =
    let crontabSchedule = Framework.Crontab.parse crontabExpression

    {
        CrontabExpression = crontabExpression
        ParsedCrontabSchedule = crontabSchedule
    }

let [<Then>] ``the CrontabSchedule has the same string representation`` (ctx:ParseContext) =
    Assert.True(ctx.ParsedCrontabSchedule.ToString() = ctx.CrontabExpression, "Parsed crontabSchedule.ToString() not equal to original crontab expression")

let [<When>] ``isRunning is called at (.*) with the following parameters`` pointInTime (parameters:Table) =
    Framework.Crontab.isRunning
        (VTable.getValueByKey "startScheduleCrontab" parameters.Rows)
        (VTable.getValueByKey "stopScheduleCrontab" parameters.Rows)
        (VTable.getValueByKey "maxStartedDurationHours" parameters.Rows |> float)
        (VTable.getValueByKey "maxStoppedDurationHours" parameters.Rows |> float)
        (pointInTime |> DateTime.ParseUtcIso)

let [<Then>] ``the isRunning result is (.*)`` (isRunningExpected:string) (isRunningActual:bool) =
    Assert.Equal(isRunningExpected |> bool.Parse, isRunningActual)

let [<When>] ``getLastRunOn is called at (.*) with the following parameters`` pointInTime (parameters:Table) =
    Framework.Crontab.getLastRunOn
        (VTable.getValueByKey "runScheduleCrontab" parameters.Rows)
        (VTable.getValueByKey "maxRunIntervalHours" parameters.Rows |> float)
        (pointInTime |> DateTime.ParseUtcIso)

let [<Then>] ``the getLastRunOn result is (.*)`` (lastRunOnExpected:string) (lastRunOnActual:DateTime) =
    Assert.Equal(lastRunOnExpected |> DateTime.ParseUtcIso, lastRunOnActual)

Points of interest:

There is no need for a class + member methods, just a plain module with functions can be used!
Functions are mapped to steps by name - no need for attributes!
Function arguments are automatically captured by TickSpec based on simple regular expressions
Examples are handled by TickSpec after doing some Xunit wiring involving the usual suspects - Theory and MemberData, however the code is minimal and TickSpec does some function argument magic again

Note the following potential pitfalls related to TickSpec function/argument mapping magic:

Initially we did experience some strange error messages related to arguments not mapped correctly, but as soon as we started using a Context record type per feature all these were gone. That one could look like this:

type Context = {
    Value1: int option
    Value2: string option
}

let [<Given>] ``some given step`` () =
    { 
      Value1 = Some 1
      Value2 = None
    }

let [<When>] ``some when step`` () =
    let result = //...
    { ctx with Value2 = Some result }

let [<Then>] ``some then step (.*)`` (someArgOfThisStep: string) (ctx:Context) =
    Assert.Equal(someArgOfThisStep, ctx.Value2.Value)

Additionally, quite a few times TickSpec was reporting missing step implementation error, and it turned out we have forgotten the () of a function which has no arguments ;)

In the end we decided to focus on using TickSpec for all our automated tests (yes, including unit tests ... [^2]) due to its elegant approach of mapping steps/examples with functions and their arguments.

P.S. Project can be found at: https://github.com/deyanp/FSharpBDDComparison

[^1] Which requires [<assembly: Xunit.TestCaseOrderer("Xunit.Extensions.Ordering.TestCaseOrderer", "Xunit.Extensions.Ordering")>] in AssemblyInfo.fs or similar
[^2] I/we do not care too much about the question if BDD-style test automation should or should not be used for unit tests. See Behavior Driven Development vs Unit Testing
for a discussion I am not sure I agree 100% with ..

Use Azure Kubernetes Service (AKS) + Traefik instead of Azure Functions hosting + Azure API Management

Deyan Petrov — Tue, 20 Apr 2021 11:31:36 +0000

TLDR; You'd better use AKS (incl. Traefik or similar ingress controller) instead of Azure Functions App Service/Premium/Consumption Plans + Azure API Management for hosting your microservices.

Disclaimer: The context of this blog post are systems with a fair number of API/transaction requests per day (e.g. 50k+) with API calls (almost) every second for the majority of the day, and standard non-functional requirements like response times must be way below 1 second, top-notch security, etc. This is not addressing hobby projects with public access to everything and occasional 1-2 API calls per day and/or a single monthly peak of 1000 API calls/hour!

Introduction

Being aware of the latest serverless trend I started a project with the "fully serverless" approach in mind, which in MS/Azure world means Azure Functions framework for building our .NET Core applications, and using Azure Functions Hosting Plans on Azure. The Azure Functions were "hidden" behind Azure API Management (APIM), so that JWT Tokens (from a web-based backoffice UI) and api keys (from B2B integrations) could be centrally validated.

Fast-forward to today and we are currently using none of these:

Azure Functions framework
Azure Functions hosting in Consumption/Premium/App Service plan
APIM

Instead we are pretty happy with these:

Standard .NET 5 apps (using WebJobs SDK only as a syntactic sugar for some non-HTTP triggers)
AKS
Traefik (running in AKS)

Why did we change our mind? There were numerous reasons for that, and I will try to explain each one below. But first a short overview of Azure Functions and APIM.

Azure Functions Hosting

An Azure Functions App can be hosted in multiple ways:

Consumption plan
Premium plan
Dedicated App Service plan (one of the oldest Azure offerings)
App Service Environment (ASE)
Azure Kubernetes Service (AKS)

When I read the list I was thinking - wow, I can't go wrong with Azure Functions, even if one hosting option turns out to be sub-optimal there are so many others! Upon a second look though 2 of the options should be removed straight away:

Consumption plan has unsurmountable cold start issues, and no VNET integration
a. Yes, we did implement a health-check calling every single function app (50+) every 5 minutes, plus integrated this in Azure DevOps pipelines, however this keeps warm 1 existing instance only, not any new instance created due to scale out
b. Yes, we did implement WarmupTrigger in our function apps, however we still experienced cold starts
c. Without VNET integration, you cannot effectively hide all function apps from the Internet and connect them with any other hidden infrastructure like Azure Key Vaults, Event Hubs etc, unless you start playing with source IP restrictions, and this becomes very quickly unmaintainable, latest when you start whitelisting whole Azure datacenters/regions ...
App Service Environment (ASE) is a super expensive dinosaur which seems to be intended for enterprise customers. ASEv2 requires a monthly fee of more than EUR 1000 for "flat stamping" (whatever that means). ASEv3 (preview) reduces that to USD 280/month for merely 2 vCPUs and 8 GB of RAM of processing capacity ... Not to mention the complexity of network configurations, etc.

So you are effectively left with options 2, 3 and 5.

Azure Functions Application Framework

Azure Functions as Application Framework with .NET Core 3.1 takes your "functions" decorated with a FunctionName attribute, and runs them in a magical way ¹ within the Azure Functions WebScript Host (provided by MS). You do not have a Program.cs/fs with main or similar, if you want to wire up some start up code you have to implement a class and use an assembly-level FunctionsStartup attribute so that the Azure Functions runtime can find it (awkwardness to the max ;)

Azure Functions with .NET 5 lets you run your code in a separate process and you have a Program.cs/fs with main, however all calls to your code will be passing first the Azure Functions WebScript host and then the latter will invoke your code via GRPC ² ...

There is no GRPC support on the horizon for Azure Functions to my knowledge, which means that your inter-service calls cannot use GRPC ... FYI, everybody is using GRPC for inter-service calls due to significant performance benefits, and also .NET Core 3.1/.NET 5.0 have very good GRPC support already.

Azure API Management (APIM)

APIM is a pretty sophisticated service which wants to intimately know your APIs. You import your Open API / Swagger yamls for example, and it enumerates all operations, so it understands your APIs. Based on that APIM can also render a Web UI where you can list your APIs, test them, etc. Defining Products and Subscriptions comes on top, as well as various Policy which can be injected at different levels.

APIM comes in several tiers, but note that even though most of them may support your requests/second requirements, only 2 of them have VNET support: Developer (non-production use) and Premium (at staggering EUR 2350/month).

Does that sound like a real enterprise offering? Well, if this is not an enterprise offering, I don't know what is ...

Reasons for migrating away from Azure Functions + APIM

Costs

Azure Functions Premium Plan Costs

There are 3 instances sizes:

and the billing is

which means more than 100 EUR/month for a single instance of the smallest 1 vCpu and 3.5GB memory!!
Azure Functions App Service Plan Costs

Premium v3 (VNET Integration)

In contrast, AKS node billing is actually the standard Azure Virtual Machine billing, so you can get a standard VM with 2 vCPUs + 8 Gb RAM for as low 70-80 EUR/month excluding any discounts for 1-3y reservations, or spot pricing, etc.! Traefik is an open-source project with a pretty sufficient for our uses free community edition ...

Performance

We experienced serious problems running 5 different function apps only on the smallest SKUs of Azure Functions Premium (EP1) or App Service Plan (P1V2) ... Not only the applications were slow at processing a very low number of requests, but the memory utilization was very high (due to only 3,5Gb of RAM, much less available though).

In contrast, we are easily running almost 50 apps in AKS per node (every node with 2 vCPU + 16GB RAM, costing around 90 EUR/month), with less than 20% max CPU utilization and less than 55% RAM utilization ...

Security

Imagine you want to secure your fairly simple APIM + Azure Functions + some underlying Azure services "modern" architecture:

Diagram: "Modern" serverless architecture based on Azure Functions

You are using the "serverless/consumption" versions of Azure Functions and APIM ... you will be surprised by quite a few imho unsurmountable security challenges:

VNET Integration

I am not getting why in 2021 Azure allows you to create resources without a VNET (VPC) ... For hobby projects I understand, but any serious enterprise-grade system cannot survive without vnets ... It is also beyond my understanding why the Consumption Plan has no VNET Integration (besides internal technical limitations of Microsoft's implementation, or marketing/pricing agendas) ... I see talented people wasting their time trying to find Source IP Restriction-based approaches to security their Consumption Plan hosted Azure Functions and I am wondering why they are doing this to themselves ...
Network security for the underlying Storage Account - some people are still not able in 2021 to secure the required by Azure Functions runtime storage account ... this has been a major oustanding issue for the past several years, and even though MS have finally fixed that I am not 100% sure if it has been rolled out everywhere ...
With Premium and App Service Plans VNET Integration exists, however with the former 1 of my function apps once lost its VNET Integration (= full downtime). The answer to my MS Support ticket (120051322002711) was to host my function app in other region in addition, and put azure front door or similar on top ... I am not saying that this incident can/will repeat for you, I just have the feeling that the VNET Integration was bolted on the Azure Functions Premium Plan more of as an afterthought rather than a mandatory underlying infrastructure principle ...
Function Apps on Consumption Plan do not have reliable outboundIPAddresses/possibleOutboundIPAddresses, all IP ranges of the whole Azure region must be whitelisted (> 127 IP ranges), however the firewall rules in Key Vault allows for max 127 rules (no, this is not a joke)
If function apps and storage accounts are in the same region, then connectivity function apps -> storage accounts goes via private Azure network (e.g. 10.150.*), however whitelisting of public ip ranges is only allowed in storage account firewall (or vnets, no private ip ranges – no, this is not a joke either).
It turns out that Azure API Management on Consumption Plan does not have a dedicated IP, so whitelisting of the whole Azure West Europe datacenter IP addresses must be performed, which is clashes with 127 rules max limit for firewalls in Azure. Azure Support is suggesting additionally to use the 40k json file with all Azure IP ranges, however that requires heavy investment in setting up a function app to parse the file on a weekly basis and to update the firewall rules in all ip-restricted services.
Vnet Integration for APIM is only possible in Developer (no SLA, gets restarted with downtime on a monthly basis) or Premium Tier (costs EUR 2500/month). Numerous ideas in the feedback center, no result.

Slow deployment + 100% CPU

On both App Service Linux and Windows Plans we experienced very high CPU peak every time we deployed (incl. slot swap). We had only a few (max 5) function apps hosted on the smallest (1 vCore) instance, with no application load whatsoever (no requests). Microsoft Support's answer was: "It is normal to have some CPU increase during deployment for a short period of time and it should not impact the overall availability of the functions.". Of course, it did affect the performance of the functions dramatically ...

With the App Service Plan the deployment is handled by the Kudu management container where it is using the local storage for deployment which is also slower and besides this there are some additional intermediary steps in the deployment.

Additionally, we experienced from time to time failed deployments with errors like "Bad Gateway 502", or sometimes 409.

I think there is a general issue with the performance of function apps on the App Service Linux plan, and that may be the root cause for the deployment issues …

In contrast, deploying to AKS is very fast (less than 50 seconds) and has no measurable impact on the CPU utilization of the node (2 vCPU, 16 GB RAM).

Inter-service (Service-to-Service) API calls

In the context of Microservice Architecture sooner or later you will need to have synchronous (REST or GRPC) calls from one app to another. Yes, you should try not to have such, yes, you should try to make everything based on events/messages via message bus ... however, the reality is that you will definitely reach some consistency limitations which will force you to have such inter-service calls.

In case you are using Azure Functions Subscription Plan you will be charged for every inter-service call ... imagine you have an orchestration service1, which is making a GET to service2, then making POST to service3 and service4 ... then you will be charged for additional 3 calls for every API call to the orchestration service.

In case of Premium or App Service Plan, your inter-service calls will traverse the Azure network stack, which is much slower than having service calls on the same node as is the case for AKS. We were even forced to deploy several microservices as 1 deployment unit with in-process calls instead (using project references, which was causing a big deployment mess) ...

Other Azure Functions peculiarities

Strange limitations of length of app name

Update of Azure Function's App Application Settings from the UI takes 2x longer for App Service Linux hosted function app compared to Premium Windows hosted function app.
It seems that if we have a long function name (e.g. 59 characters) everything is slower, including updating Application Settings in the UI.
MS Support recommends keeping function app names under 40 characters especially when working with Deployment Slots because there is a limit set in for the length of the hostname that is generated from the function name ( app_name.azurewebsites.net ), and as the name of the deployment slot is added to the host name it can get truncated and could cause issues with deployment or slot swapping.

In contrast, K8s defines name limits of up to 253 characters ...
Strange tooling (Kudu)

While investigating an issue with a function app I got the request from MS Support to "please login to the kudu site" .. of the function app "... or select the functionapp-> platform features-> AdvancedTools(kudu)". Kudu seems to a very old and custom web UI for App Service ... with 100% guarantee that any knowledge gained is not transferrable to any other cloud provider or similar ;)
Strange settings

There are a number of "magic" settings to be considered like Always On so that the app runs correctly. On an App Service plan, the functions runtime goes idle after a few minutes of inactivity, so only HTTP triggers will 'wake up' your functions.
Whenever you are doing run from packages it is recommended to have the WEBSITE_RUN_FROM_PACKAGE setting in the app settings.
For VNET Integration of Premium Plan-hosted function apps for example you should set WEBSITE_VNET_ROUTE_ALL=1 and WEBSITE_DNS_SERVER=168.63.129.16 ...

Other APIM-related hassles

APIM setup takes extremely long time

Extremely long means more than half an hour. The "serverless" tier is the only exception and takes much less, but compare the features before you decide for it (remember - no VNET Integration!)

In contrast Traefik takes seconds to install in AKS/K8s cluster (we are using simple yaml deployment files, no helm charts).
Same server resources seem to be shared between the Developer Portal and API requests

Can it be that someone browsing your API Web Portal has an impact on your REST API response time SLA?

In contrast AKS/K8s has extensive requests/limits capabilities.
CPU utilization for a prolonged time

We were paying 125 EUR/month for 1 unit which should allow us 1000 rps, but even with only 20-30 requests per 5 minutes we were seeing 50% capacity utilization of our APIM instance.

Diagram: 2 APIM instances with same minimal load, the one one the right showing high utilization for no reason

After some useless roundtrips with the MS Support person (no, we have absolutely no load on the system currently, no complex polices, nothing ...) he indicated, that such capacity utilization is to be ignored, and might happen due to updates of a freshly set up APIM instance for a duration of a couple of days ...
Cumbersome APIM deployment scripts

Creating an APIM instance and configuring it requires some heavy scripting, as you need to additionally:
1. Define Products
2. Define Policies
3. Define Subscriptions
4. Import APIs from swagger yamls for example
5. Create Revisions of the APIs
Policies alone can be defined on different levels - Product, all APIs, single API (all operations), single operation.

Until recently the full functionality of APIM could only be configured with Powershell, as Azure CLI did not have the support ...

The creation of the deployment scripts took us tons of time (still fragile), and their execution was also pretty lengthy.

Azure Functions Runtime/Framework issues

The fact that Azure Functions Team "missed" the .NET 5 launch by 4-5 months (Nov 2021 - March 2022) is pretty well known. I think MS Team got surprised that not all of their customers are "enterprise developers" lagging 1-2 years behind latest technologies due to big companies upgrading slowly ... The promise is that this will not repeat with .NET 6+ ...

2: Now we have 2 parallel runtimes - in-process (.NET Core 3.1) and out-of-process (.NET 5) and it seems this duality will continue for a couple of years ... Not sure how often you have to switch in-process -> out-of-process -> in-process ...

With the out-of-process model (.NET 5) the Azure Functions Runtime calls your functions using GRPC ... as mentioned above, not sure what is the performance impact of that vs. fully in process ². I am not sure I want to use GRPC inside my application process just because MS wants to have out-of-process now ...
Azure Functions Runtime seems to consume 2 times more memory
GRPC is not supported for inter-service communications, so a microservice based on the Azure Functions runtime cannot expose GRPC interface, which is easily possible with non-Azure Functions Runtime hosted .NET 5 app

Summary and Recommendation

Long story short, my personal recommendation is: do not waste your time with Azure Functions hosting (or even application framework) and APIM for any serious project with standard security/performance requirements, and with the goal of having competitive pricing for hosting ... Or in other words, use Azure Functions only for quick-and-dirty hobby projects, or one-off jobs (maybe some glue infrastructure code), or in case performance, security and costs are absolutely no factors.
Additionally, value your time and better invest in learning something standard like Kubernetes where you can use your knowledge across clouds and on premise, instead of learning all Azure Functions peculiarities, which IMHO may be a result of Microsoft's trying to reuse existing (but with lots of heritage) Azure App Service/Web Script platform/framework for Azure Function ...

In detail:

Migrate from Azure Functions Hosting (Consumption/Premium/App Service Plan) to AKS³ for much lower costs, better security, and well-known container management
Migrate from Azure API Management to Traefik or similar K8s Ingress Controller for reducing costs, simplifying management and deployment.
Migrate from Azure Functions .NET Core 3 (in-process) or .NET 5 (out-of-process) runtime to standard .NET 5 application using WebJobs SDK directly for independence, faster upgrades, lower memory consumption, GRPC and many other possibilities.

If there is interest I could create another post showing how to do the above ...

For example the mandatory Microsoft.NET.Sdk.Functions nuget generates automatically function.json file upon compilation .. ↩
Had an open github issue on the .NET 5 worker preview project asking what about the performance impact of out-of-process/grpc between the host and my code, however the repo got archived and my issue got dropped unanswered ... ↩
I think AKS is one of the best Azure products currently, very competitive when compared to EKS or GKE etc. ↩

DEV Community: Deyan Petrov

Claude and Kusto or MCP and CQRS+

Introduction

Claude

Azure Data Explorer (aka Kusto or ADX)

Natural Language Query

MCP Server

CQRS+, Microservices, and how MCP fits in

Sample Implementation

How MCP and REST co-exist in the same Microservice

Step 1 — Register MCP tools in the DI container (Mcp.Hosting.fs):

Step 2 — Configure the web host generically (Hosting.HostBuilder.fs). This function knows nothing about MCP — it just wires up REST endpoints, a middleware hook, and an additional-endpoints hook:

Step 3 — Map MCP endpoints via a one-liner that plugs into the configureAdditionalEndpoints hook:

The OAuth 2.1 Proxy — The Hard Part

Discovery: "I am your OAuth server"

Mock DCR: "Sure, here's your client_id"

Scope Rewriting: "You think you're authenticating to me, but you're actually authenticating to Kusto"

Credential Injection: "Let me fix that secret for you"

Per-User Token Passthrough

The Full Flow

Security — Multi-Layer Defense

Layer 1: Token Pre-Validation (Middleware)

Layer 2: Whitelist-Based Query Validation

Layer 3: Azure Data Explorer Server-Side Enforcement

Functional Dependency Injection

Local Development with Cloudflare Tunnel

Conclusion

References

Azure Devops Pipeline with SonarQube Community in Docker and Playwright

Introduction

Run SonarQube Server in Docker

Reset admin password

Create Project

Start the Scanner using NPX

Get some metrics via Web API

Capture screenshots as reports via Web UI with Playwright

Check if there are security vulnerabilities

Azure DevOps Pipeline, and the final scripts

References

Running and Approving Azure DevOps Pipelines from the terminal

Introduction

The az rest .. --resource .. command

Resource Model and Relations, or how to find the Approval

Pipeline

Last Build

Build Timeline

Approval

Finally, approve!

References

F# App Stub for AKS hosting v2 (without Azure WebJobs SDK fluff)

Introduction

App Stub v2

Trigger Implementations

Event Hub Processor

Queue Processor

Timer Processor

SignalR

Conclusion

Isolated Local Dev Env using k3d, multi-tenant Azure services, docker containers, and mirrord

Introduction

Problem Statement

Solution

k3d

Multi-tenant Azure services

Per-Developer Azure Event Hubs Partitions

Per-Developer Azure Storage Accounts (incl. Blob Containers and Storage Queues)

Per-Developer Azure Data Explorer (Logical) Databases

Per-Developer Azure SignalR Notification Hubs

Docker Containers running locally

mirrord

Some Non-Functional Considerations

Conclusion

CQRS+

Introduction

Problem Statement

Solution

Benefits

Drawbacks

Implementation Details

XyzService.CommandHandling (CmdH)

Step 1 — Register MCP tools in the DI container (`Mcp.Hosting.fs`):

Step 2 — Configure the web host generically (`Hosting.HostBuilder.fs`). This function knows nothing about MCP — it just wires up REST endpoints, a middleware hook, and an additional-endpoints hook:

Step 3 — Map MCP endpoints via a one-liner that plugs into the `configureAdditionalEndpoints` hook:

The `az rest .. --resource ..` command