DEV Community: Gabriel Sadaka

React Native Expo automated deployment using fastlane and CircleCI

Gabriel Sadaka — Wed, 02 Nov 2022 06:36:48 +0000

Intro

As hybrid mobile application frameworks mature, React Native has become a popular choice for building iOS and Android mobile apps using a shared codebase. When kicking off a React Native mobile app, developers are presented with a choice between using Expo Go or the React Native CLI. While React Native provides you with an engine that can be used to build a hybrid mobile app, Expo Go provides you with the complete car ready to drive away from the dealer. It provides you with a streamlined developer experience, with all the tooling required to build, deploy and test your mobile apps. The Expo SDK provides access to device and system functionality using shared code, enabling developers to focus on building their app without worrying about platform specific native code.

Expo also provides a cloud based build and deployment service, EAS, that makes it easy to deploy an app to the store and keep it up to date. While this may be suitable for some organisations, there may be a need to build and deploy the app using an organisation's own infrastructure or existing CI/CD platform. This post will walk through an approach to building, testing and deploying an app using fastlane and CircleCI. While CircleCI is used for this example, this could be substituted with a CI/CD platform of choice that has support for Linux and MacOS build agents. CircleCI was chosen because its free tier is sufficient for building, testing and deploying Android and iOS apps with little configuration.

This post assumes prior knowledge in deploying apps to the Google Play Store and Apple App Store and will not cover the details of setting up the app listings. Please refer to the many resources already available online on how to do this, if you have not done this before.

Project setup

Prerequisites

Xcode
Android Studio
NodeJS
Google Play Console Account and knowledge on how to setup a Play Store app listing.
Apple Developer Account and knowledge on how to setup an App Store app listing.

Initialise Expo app

Lets start off by creating a new Expo app using the create-expo-app npx package, by running the following command.

npx create-expo-app --template

I chose the Navigation (Typescript) template, but this example should work with any template. Then provide the name of the application and hit enter.

While the Expo SDK provides access to many native capabilities it is likely you will encounter the need to reference third party React Native packages that have their own custom native code or you might want to customise the build. Doing so will require an Expo development build, which will require referencing the expo-dev-client package, which you can so by running the following command.

npx expo install expo-dev-client

Now that we have an Expo app with the development client package referenced, we are ready to start the app in an emulator or an a device. To do this we can run either of the following commands depending on the platform we want to run it on. While the Expo CLI can automatically start the iOS Simulator, if you would like to run it in an Android Emulator you will need to start one up before running the command. Note that while the initial build can be slow, the subsequent builds will be much quicker.

npx expo run:android #run on Android Emulator
npx expo run:ios #run on iOS Simulator
npx expo run:android -d #run on Android Device
npx expo run:ios -d #run on iOS Device

Before we setup fastlane and CircleCI, we will need to run the Expo prebuild command, which will generate the iOS and Android native code and project files. This will enable us to use standard build tools for the respective platforms to build and deploy the apps to the stores, thus avoiding the need to rely on EAS, by running the following command.

npx expo prebuild

Fastlane

The next step is setting up the fastlane workflow, which will take care of building, signing and deploying the Expo React Native mobile app. fastlane is being used as it automates many tedious tasks that can be tricky to get right using the platform provided CLI tooling. Since fastlane is a Ruby package Bundler will be used to define the dependency, to make it easy for other developers to run it and to enable consistency with the CI pipeline. Refer to the fastlane documentation on steps to get Ruby installed on your machine. Once Ruby is installed run the following commands to install fastlane in the ios and android directories.

gem install bundler

cd ios
bundler init
bundler add fastlane
bundle install

cd android
bundler init
bundler add fastlane
bundle install

Now that fastlane is installed, we can initialise fastlane in the ios and android directories, using the following command. When prompted, choose the Manual Setup option.

cd ios
fastlane init

At the time of writing the fastline init command failed when running within the Android folder, so if you also receive an error similar to no iOS project in the current directory, then use the following commands to copy the iOS generated config files. We will be replacing the contents of the Appfile and Fastfile next, so need to worry about iOS specific commands in there.

cd android
mkdir fastlane
cp ../ios/fastlane/Appfile fastlane/Appfile
cp ../ios/fastlane/Fastfile fastlane/Fastfile

Android fastlane workflow

Now that the base fastlane configuration is ready to go, we can define the workflow to build, sign and deploy the Android app.

To enable automatic incrementing of the build version number, a third party package will need to be installed using the following command. When prompted to update the gemfile, enter y.

cd android
fastlane add_plugin increment_version_code

To begin with, update android/fastlane/Appfile to define the package name that will be used. This will need to match the identifier that you specified when setting up Expo.

package_name("org.sample.mobileapp")

Next, we will need to configure a fastlane lane that will increment the build number, build the app and then upload it to the Play Store, by updating android/fastlane/Fastfile with the following contents.

default_platform(:android)

platform :android do
    desc "Deploy a new version to the Google Play Beta track"
    lane :beta do
        # Grab the latest build number from the Play Store
        previous_build_number = google_play_track_version_codes(
            track: "internal",
            json_key: "#{Dir.pwd}/play-store-credentials.json"
        )[0]

        # Increment the build number
        increment_version_code(
            gradle_file_path: "app/build.gradle",
            version_code: previous_build_number + 1
        )

        # Build a release version of the app
        gradle(
            task: "clean assembleRelease",
            print_command: false,
            properties: {
                "android.injected.signing.store.file" => "#{Dir.pwd}/example-app.keystore",
                "android.injected.signing.store.password" => ENV["ANDROID_SIGNING_KEY_PASSWORD"],
                # replace alias with your signing key alias
                "android.injected.signing.key.alias" => "[example-app]",
                "android.injected.signing.key.password" => ENV["ANDROID_SIGNING_KEY_PASSWORD"],
            }
        )

        # Upload the app to the Play Store
        upload_to_play_store(
            track: "internal",
            json_key: "#{Dir.pwd}/play-store-credentials.json"
        )
    end
end

With the fastlane configuration complete, the next step is to create a shell script which will retrieve the Android Signing Key and the Play Store Credentials from the CI environment variables and store them as files that can be accessed by fastlane, then run the fastlane beta lane. Create a shell script android/fastlane/deploy.sh, with the following contents. Then run chmod +x deploy.sh to give it execution permission.

#!/bin/sh

echo ${ANDROID_SIGNING_KEY} | base64 -d > fastlane/example-app.keystore
echo ${ANDROID_PLAY_STORE_CREDENTIALS} | base64 -d > fastlane/play-store-credentials.json

fastlane beta

iOS fastlane Workflow

Configuring the iOS fastlane workflow is a similar process to the Android one.

To start with update ios/fastlane/Appfile to define the app_identifier, which will need to match the bundle identifier you chose during the Expo setup. It will also define the itc_team_id and team_id configuration variables.

app_identifier("org.sample.mobileapp") # The bundle identifier of your app

itc_team_id("111111") # App Store Connect Team ID
team_id("111111") # Developer Portal Team ID

Next, we will need to configure a fastlane lane that will increment the build number, build the app and then upload it to the App Store. The following workflow relies on a certificate that has already been created and is pulled into the build machine via an environment variable. There is an alternative approach using the get_certificates, which will pull the certificate from Apple or create one if it doesn't exist already. Update ios/fastlane/Fastfile with the following contents.

default_platform(:ios)

platform :ios do
    before_all do
        setup_circle_ci
    end

    desc "Push a new beta build to TestFlight"
    lane :beta do
        # Retrieve the api key from key file
        api_key = app_store_connect_api_key(
            # Replace key_id and issuer_id with your api key key and issuer IDs
            key_id: "[key_id]",
            issuer_id: "[issuer_id]",
            key_filepath: "AppStoreConnectApiKey.p8",
            duration: 1000,
            in_house: false
        )

        # Retrieve signing certificate from file
        import_certificate(
            certificate_path: "PrivateKey.p12",
            keychain_name: ENV["MATCH_KEYCHAIN_NAME"]
        )

        # Import mobile provisioning profile from app store
        get_provisioning_profile(
            filename: "distribution.mobileprovision",
            # Replace provisioning_name with your provisioning profile name
            provisioning_name: "[provisioning_name]",
            ignore_profiles_with_different_name: true,
            readonly: true,
            api_key: api_key
        )

        # Get latest build number
        previous_build_number = latest_testflight_build_number(
            api_key: api_key
        )

        # Increment build number
        increment_build_number(
            # Replace xcodeproj
            xcodeproj: "AppName.xcodeproj",
            build_number: previous_build_number + 1
        )

        # Disable automatic code signing, so the signing certificate on the filesystem can be used
        update_code_signing_settings(
            use_automatic_signing: false,
            # Replace xcodeproj, bundle_identifier and profile_name
            path: "AppName.xcodeproj",
            bundle_identifier: "org.sample.mobileapp",
            profile_name: "AppName",
        )

        # Build the iOS app
        build_app(
            # replace workspace, scheme provisioningProfiles[0] and codesigning_identity
            workspace: "AppName.xcworkspace",
            scheme: "AppName",
            skip_profile_detection: true,
            export_method: "app-store",
            export_options: {
                provisioningProfiles: {
                    "org.sample.mobileapp" => "AppName",
                }
            },
            codesigning_identity: "iPhone Distribution: Org (Team ID)"
        )

        # Upload app to TestFlight, skipping waiting for build processing to reduce CI credit waste
        upload_to_testflight(
            skip_waiting_for_build_processing: true,
            api_key: api_key
        )
    end
end

With the fastlane configuration complete, the next step is to create a shell script which will retrieve the Signing Certificate Private Key and the App Store Connect Api Key from the CI environment variables and store them as files that can be accessed by fastlane, then run the fastlane beta lane. Create a shell script ios/fastlane/deploy.sh with the following contents. Then run chmod +x deploy.sh to give it execution permission.

#!/bin/sh

echo ${IOS_PRIVATE_KEY} | base64 -d > PrivateKey.p12
echo ${APP_STORE_CONNECT_API_KEY} | base64 -d > AppStoreConnectApiKey.p8

fastlane beta

CircleCI Pipeline

Finally we can configure the CircleCI pipeline to build, test and deploy the latest changes to the Play Store and App Store when a new commit is pushed. To connect CircleCI to your repository, so that it can be triggered when a new commit is pushed and access the code, go to the CircleCI documentation relevant to your VCS.

Create a CircleCI config.yml fine in a .circleci folder in your repository, with the following contents, which will checkout the code, restore packages while storing them in a cache to speed up subsequent builds, run tests, compile TypeScript, build, sign and deploy your app. The app store deployment steps require manual approval because they take a long time and can quickly consume all available build credits, so only run them when you want a new build published to the store.

version: 2.1
jobs:
  react-native-test:
    docker:
      - image: cimg/node:18.8

    steps:
      - checkout

      - restore_cache:
          key: yarn-v1-{{ checksum "yarn.lock" }}-{{ arch }}

      - restore_cache:
          key: node-v1-{{ checksum "package.json" }}-{{ arch }}

      - run: yarn install
      - run: yarn test
      - run: yarn tsc

      - save_cache:
          key: yarn-v1-{{ checksum "yarn.lock" }}-{{ arch }}
          paths:
            - ~/.cache/yarn

      - save_cache:
          key: node-v1-{{ checksum "package.json" }}-{{ arch }}
          paths:
            - node_modules

  macos-build-and-deploy:
    macos:
      xcode: 13.4.1
    environment:
      FL_OUTPUT_DIR: output

    steps:
      - checkout
      - restore_cache:
          key: yarn-v1-{{ checksum "yarn.lock" }}-{{ arch }}

      - restore_cache:
          key: node-v1-{{ checksum "package.json" }}-{{ arch }}

      - run: yarn install

      - save_cache:
          key: yarn-v1-{{ checksum "yarn.lock" }}-{{ arch }}
          paths:
            - ~/.cache/yarn

      - save_cache:
          key: node-v1-{{ checksum "package.json" }}-{{ arch }}
          paths:
            - node_modules

      - restore_cache:
          key: bundle-v1-{{ checksum "ios/Gemfile.lock" }}-{{ arch }}

      - restore_cache:
          key: pods-v1-{{ checksum "ios/Podfile.lock" }}-{{ arch }}

      - run:
          command: pod install
          working_directory: ios

      - run:
          command: bundle install
          working_directory: ios

      - save_cache:
          key: bundle-v1-{{ checksum "ios/Gemfile.lock" }}-{{ arch }}
          paths:
            - vendor/bundle

      - save_cache:
          key: pods-v1-{{ checksum "ios/Podfile.lock" }}-{{ arch }}
          paths:
            - ios/Pods

      - run:
          name: Fastlane deploy TestFlight
          command: ./fastlane/deploy.sh
          working_directory: ios

# replace example-app with repository name
  android-build-and-deploy:
    working_directory: ~/example-app
    docker:
      - image: cimg/android:2022.08-node
    steps:
      - checkout:
          path: ~/example-app

      - attach_workspace:
          at: ~/example-app

      - restore_cache:
          key: yarn-v1-{{ checksum "yarn.lock" }}-{{ arch }}

      - restore_cache:
          key: node-v1-{{ checksum "package.json" }}-{{ arch }}

      - run: yarn install

      - save_cache:
          key: yarn-v1-{{ checksum "yarn.lock" }}-{{ arch }}
          paths:
            - ~/.cache/yarn

      - save_cache:
          key: node-v1-{{ checksum "package.json" }}-{{ arch }}
          paths:
            - node_modules

      - restore_cache:
          key: bundle-v1-{{ checksum "android/Gemfile.lock" }}-{{ arch }}

      - run:
          command: bundle install
          working_directory: android

      - save_cache:
          key: bundle-v1-{{ checksum "android/Gemfile.lock" }}-{{ arch }}
          paths:
            - vendor/bundle

      - run:
          name: Fastlane deploy beta
          working_directory: android
          command: ./fastlane/deploy.sh

workflows:
  react-native-android-ios:
    jobs:
      - react-native-test
      - approve_deploy_to_apple_app_store:
          type: approval
          requires:
            - react-native-test
      - approve_deploy_to_google_play_store:
          type: approval
          requires:
            - react-native-test
      - android-build-and-deploy:
          requires:
            - approve_deploy_to_google_play_store
      - macos-build-and-deploy:
          requires:
            - approve_deploy_to_apple_app_store

To configure the environment variables that are referenced by the iOS & Android deploy.sh bash scripts, follow the CircleCI guide to set them up for your project.

The following environment variables will need to be configured:

APP_STORE_CONNECT_API_KEY: API Key to authenticate with the Apple App Store. Follow the Apple documentation to generate an API Key, then base64 encode the file and set it as the environment variable value.
IOS_PRIVATE_KEY: Private key used to sign the iOS builds. Follow this post to generate a distribution certificate and export the private key from it, then base64 encode the file and set it as the environment variable value.
ANDROID_PLAY_STORE_CREDENTIALS: Google Play Store Service Account credentials to authenticate with the Play Store. Follow the fastlane setup documentation to create a service account and generate credentials for it, then base64 encode the file and set it as the environment variable value.
ANDROID_SIGNING_KEY: Signing key used to sign the Android builds. Follow the Google documentation to Generate an upload key then enrol into Play App Signing using that key, then base64 encode the file and set it as the environment variable value.
ANDROID_SIGNING_KEY_PASSWORD: Password for signing key used to sign the Android builds.

For those following this guide on MacOS you can base64 encode a file using the included terminal application with the following command:

base64 -i file.ext

Once all of the changes are pushed to your repository, CircleCI should test & build every commit, while providing the option to either deploy your app to the Google Play Store or Apple App Store.

Wrap up

An alternative approach to storing the app signing secrets as environment variables in the pipeline is to use fastlane match which encrypts them and stores them in a shared git repository. It will make it easier to share the secrets with other developers on the team and help avoid storing secrets in the pipeline environment variables, which might not be ideal for all teams.

This guide only demonstrates how to build and deploy apps to Apple TestFlight and Google Play Store Internal track. It does not demonstrate promoting builds to the release tracks, nor does it show how to automatically generate screenshots and release notes. To further expand the pipeline to cover those steps, please refer to the excellent fastlane documentation.

You can see an example of this approach in this repo.

Please let me know what you think and any questions/feedback you may have.

Contributions

Thanks to George Ostrobrod, Acer Zhou & Luke Kugelman for their excellent work on the project that we leveraged Expo & fastlane to build a React Native mobile app that was automatically deployed and built using CircleCI.

Deploying a .NET Core API to Google Cloud Run with Cloud SQL Postgres using Pulumi and Batect

Gabriel Sadaka — Mon, 07 Dec 2020 12:38:41 +0000

Motivation

Having never worked with GCP or Pulumi, I thought it would be fun to explore the technologies by building a .NET Core API following as many best practices as possible for production ready cloud native services. I also wanted to explore more about Batect, which is a tool that I have used in past projects but have never utilized in a greenfield project. The intention was to develop an API that implemented every best practice I could think of for developing cloud native services and then share what I learnt throughout the process. Rather than wait until I have completed the journey I have decided to share my learnings so far.

What is it?

The API is a simple .NET Core Web API that creates and retrieves weather forecasts. The forecasts are stored in a GCP Cloud SQL Postgres database while the API is deployed as a Docker container running on GCP Cloud Run. It has tests written at every level of the test pyramid relevant to the component, starting at unit tests for the business logic, controllers and repository layer, leading to integration tests that run against a Postgres docker container.

GCP Secret Manager is used to store the database password which is accessed by the .NET Core API and the database schema migration tool. The database schema migrations are applied using Flyway running in a docker container, defined as a task in Batect. They are applied during deployment and every time the integration tests are run to ensure a clean database is available for each test run.

The source code is hosted on GitHub using GitHub actions to automatically build and test the application then deploy it and its infrastructure to GCP using Pulumi. It uses Batect to define the tasks that are run during the CI/CD pipeline to enable them to run consistently during local development and in the GitHub actions.

There are a few tools used for static analysis and for security/dependency scanning to ensure the application has no known vulnerabilities. The ASP.NET Core API uses SonarAnalyzer.CSharp and Microsoft.CodeAnalysis.NetAnalyzers both of which are static analysis libraries for C# that detect code patterns that introduce security vulnerabilities or other flaws such as memory leaks. GitHub CodeQL is another static analysis tool that is used as a GitHub action. Dependabot is used for scanning the .NET dependencies to ensure they are up to date. Trivy is used to scan the .NET Core Docker image for known vulnerabilities while Dockle is a Dockerfile linter for detecting security flaws.

What does it look like?

The Build Pipeline

Although the code is hosted on GitHub and uses GitHub actions to automate the building and deployment, it makes use of little GitHub actions specific functionality. Instead it uses Batect to define what happens at each step, enabling the developer to validate that it will work locally before pushing the code and waiting to see if it will succeed. It also ensures that the tasks can be run consistently on any machine with little setup required, removing the concern that arises from "doesn't run on my machine".

To build the application the standard .NET Core SDK base image is defined as a container in Batect, along with volumes to map the code, the nuget cache and the obj folder to optimise the performance of subsequent builds by caching intermediate artefacts. A task is also defined that will use that container to build the API.

containers:
---
build-env:
  image: mcr.microsoft.com/dotnet/core/sdk:3.1
  volumes:
    - local: .
      container: /code
      options: cached
    - type: cache
      name: nuget-cache
      container: /root/.nuget/packages
    - type: cache
      name: weatherApi-obj
      container: /code/src/WeatherApi/obj
    - type: cache
      name: weatherApi-tests-obj
      container: /code/src/WeatherApi.Tests/obj
  working_directory: /code
---
tasks:
---
build:
  description: Build Weather API
  run:
    container: build-env
    command: dotnet build

To run the integration tests the same build-env Docker container is used along with a Docker container for Postgres which will run the schema migrations using Flyway as a setup command. A task is defined that will run the tests that depends on the Postgres container becoming healthy which will only happen once the migrations are run. The integration tests can be run locally the exact same way they are run in the build pipeline to enable the developer to be confident that the changes they have made won't result in a broken build.

containers:
---
postgres:
  build_directory: db/tests
  ports:
    - local: 5432
      container: 5432
  setup_commands:
    - command: /scripts/run-migrations-integration-tests.sh
  volumes:
    - local: db/migrations
      container: /migrations
      options: cached
    - local: db/scripts
      container: /scripts
      options: cached
  environment:
    POSTGRES_DB: weather_db
    POSTGRES_USER: weather_user
    POSTGRES_PASSWORD: $POSTGRES_PASSWORD
    POSTGRES_HOST: postgres
    POSTGRES_PORT: "5432"
    POSTGRES_SCHEMA_NAME: public
    POSTGRES_MIGRATIONS_LOCATION: filesystem:/migrations
---
tasks:
---
test:
  description: Test Weather API
  run:
    container: build-env
    command: dotnet test
    environment:
      ASPNETCORE_ENVIRONMENT: test
      WEATHERDB__PASSWORD: $POSTGRES_PASSWORD
  dependencies:
    - postgres

The GitHub action now only needs to run the Batect task called test which will trigger Postgres to start, the Flyway migrations to run and then the API unit and integration tests to run. It uses the default Ubuntu image, first checking out the code then calling Batect to run the task.

jobs:
---
build-test:
  runs-on: ubuntu-latest
  steps:
    - name: Checkout code
      uses: actions/checkout@v2

    - name: Build and test application
      run: ./batect test
      env:
        POSTGRES_PASSWORD: ${{ secrets.POSTGRES_PASSWORD }}
---

The Infrastructure

Deploying the infrastructure is performed in two steps by two different GCP accounts. There is an account that has the permissions to make IAM changes and enable GCP APIs named, iam-svc. The IAM account is created manually using gcloud CLI commands prior to running the first deployment of the application, there are instructions in the readme of the repository to create this account. The IAM account then creates a CI account named, ci-svc, with only the permissions it requires to deploy the infrastructure and deploy the application to GCP Cloud Run, along with an account named, weather-api-cloud-run, that is the service account the Cloud Run app runs as. The Cloud Run service account has only the permissions it needs to run the application, access the database password secret and the database itself.

The deployment of the IAM changes, infrastructure and the app itself are all performed by Pulumi. The tasks performed in the GitHub Actions like the build and test task are all defined as containers/tasks in Batect.

containers:
---
pulumi:
  image: pulumi/pulumi:v2.12.1
  working_directory: /app
  volumes:
    - local: infra
      container: /app
      options: cached
    - local: /var/run/docker.sock
      container: /var/run/docker.sock
  environment:
    APP_NAME: $APP_NAME
    GOOGLE_CREDENTIALS: $GOOGLE_CREDENTIALS
    GOOGLE_SERVICE_ACCOUNT: $GOOGLE_SERVICE_ACCOUNT
    GOOGLE_PROJECT: $GOOGLE_PROJECT
    GOOGLE_REGION: $GOOGLE_REGION
    PULUMI_ACCESS_TOKEN: $PULUMI_ACCESS_TOKEN
    GITHUB_SHA: $GITHUB_SHA
  entrypoint: bash

tasks:
---
deploy-iam:
  description: Deploy IAM using Pulumi
  run:
    container: pulumi
    command: deploy-iam.sh
deploy-infra:
  description: Deploy infra using Pulumi
  run:
    container: pulumi
    command: deploy-infra.sh
    environment:
      DB_NAME: $DB_NAME
      DB_USERNAME: $DB_USERNAME

The two GCP accounts, the IAM changes and the enabling of the GCP APIs are declared using the GCP Pulumi library, defined in the deploy-iam task. Like Terraform, Pulumi maintains the latest state of the deployment, comparing that to the declared state in the latest commit, applying only the changes that were made.

...
const ciServiceAccount = new gcp.serviceaccount.Account(`ci-svc`, {
    accountId: `ci-svc`,
    description: `CI Service Account`,
    displayName: `CI Service Account`
}, {dependsOn: enableGcpApis.enableIamApi});

const storageAdminIamBinding = new gcp.projects.IAMBinding(`ci-svc-storage-admin`, {
    members: [ciServiceAccountEmail],
    role: "roles/storage.admin"
}, {parent: ciServiceAccount, dependsOn: enableGcpApis.enableIamApi});

...

export const enableCloudRunApi= new gcp.projects.Service("EnableCloudRunApi", {
    service: "run.googleapis.com",
});

The only infrastructure deployed for this application is the container registry that stores the build Docker images that GCP Cloud Run depends on and the GCP Cloud SQL Postgres instance that the application writes and reads weather forecasts from. They are both defined using Pulumi, defined in the deploy-infra task.

const registry = new gcp.container.Registry("weather-registry");

...

export const databaseInstance = new gcp.sql.DatabaseInstance(`${config.appName}-db`, {
    name: `${config.appName}-db`,
    databaseVersion: "POSTGRES_12",
    settings: {
        tier: "db-f1-micro",
        ipConfiguration: {
            ipv4Enabled: true,
            requireSsl: true
        }
    },
});

export const database = new gcp.sql.Database(`${config.appName}-db`, {
    name: config.dbName,
    instance: databaseInstance.id
});

The Database

The database schema migrations are performed by Flyway running in a Docker container, defined as a Batect container/task. The task first calls GCP Secrets Manager to retrieve the database password then connects to the Cloud SQL Postgres instance using the Cloud SQL Proxy and then performs the schema migrations.

containers:
---
flyway-migrator:
  build_directory: db
  volumes:
    - local: db/migrations
      container: /migrations
      options: cached
    - local: db/scripts
      container: /scripts
      options: cached
  environment:
    DB_PORT: "5432"
    DB_SCHEMA_NAME: public
    DB_MIGRATIONS_LOCATION: filesystem:/migrations
---
tasks:
---
migrate-db:
  description: Run database migrations
  run:
    container: flyway-migrator
    entrypoint: /scripts/run-migrations-gcloud.sh
    environment:
      GOOGLE_PROJECT: $GOOGLE_PROJECT
      GOOGLE_REGION: $GOOGLE_REGION
      GOOGLE_CREDENTIALS: $GOOGLE_CREDENTIALS
      DB_INSTANCE: $DB_INSTANCE
      DB_HOST: localhost
      DB_NAME: $DB_NAME
      DB_USERNAME: $DB_USERNAME
      DB_PASSWORD_SECRET_ID: $DB_PASSWORD_SECRET_ID
      DB_PASSWORD_SECRET_VERSION: $DB_PASSWORD_SECRET_VERSION

The Deployment

Build, scan and publish image

To run the application in GCP Cloud Run a Docker image needs to be built and deployed to a container registry that is accessible by the service. I chose GCP Container Registry since it was the easiest to get working with Cloud Run. Just like the previous tasks the build, scan and publish image step is defined as a Batect container/task using the GCP SDK Alpine image.

containers:
---
gcloud-sdk:
  image: google/cloud-sdk:314.0.0-alpine
  working_directory: /app
  volumes:
    - local: .
      container: /app
      options: cached
    - local: /var/run/docker.sock
      container: /var/run/docker.sock
  environment:
    APP_NAME: $APP_NAME
    GOOGLE_CREDENTIALS: $GOOGLE_CREDENTIALS
    GOOGLE_PROJECT: $GOOGLE_PROJECT
    GITHUB_SHA: $GITHUB_SHA
---
tasks:
---
build-scan-image:
  description: Build and scan docker image for vulnerabilities
  run:
    container: gcloud-sdk
    command: infra/build-scan-image.sh
push-image:
  description: Push docker image to GCR
  run:
    container: gcloud-sdk
    command: infra/push-image.sh

After the image is built, it is scanned for vulnerabilities using Trivy and Dockle to ensure that there are no dependencies with known vulnerabilities and that the Dockerfile follows the best security practices. One of the vulnerabilities Dockle identified for me was that I forgot to change the running user at the end of the Dockerfile so that the application didn't run as the root user.

IMAGE_TAG=gcr.io/"$GOOGLE_PROJECT"/"$APP_NAME":"$GITHUB_SHA"

docker build . -t "$IMAGE_TAG"

docker run --rm -v /var/run/docker.sock:/var/run/docker.sock aquasec/trivy:0.12.0 \
    --exit-code 1 --no-progress --severity CRITICAL "$IMAGE_TAG"

docker run --rm  --rm -v /var/run/docker.sock:/var/run/docker.sock -i goodwithtech/dockle:v0.3.1 \
     --exit-code 1 --exit-level warn "$IMAGE_TAG"

---
FROM mcr.microsoft.com/dotnet/core/aspnet:3.1 AS runtime
WORKDIR /app

COPY --from=build /app/out ./

RUN useradd -m -s /bin/bash dotnet-user
USER dotnet-user

ENV ASPNETCORE_URLS=http://*:8080

ENTRYPOINT ["dotnet", "WeatherApi.dll"]

Deploy app to GCP Cloud Run

Once the API is built and published to the Container Registry it is then deployed to GCP Cloud Run using Pulumi, which is again defined as a Batect container/task. It uses the same Pulumi container definition declared above for the IAM and infrastructure tasks.

tasks:
---
deploy:
  description: Deploy Weather API to GCP Cloud Run using Pulumi
  run:
    container: pulumi
    command: deploy-app.sh
    environment:
      GOOGLE_RUN_SERVICE_ACCOUNT: $GOOGLE_RUN_SERVICE_ACCOUNT
      DB_INSTANCE: $DB_INSTANCE
      ENVIRONMENT: $ENVIRONMENT

The GCP Cloud Run service is defined to use the container port 8080, as the default port of 80 cannot be bound to by unprivileged users. It is also defined with a dependency on the GCP Cloud SQL Postgres instance, enabling it to access it via the Cloud SQL Proxy. It is set to be visible to all users and is therefore public by default, this can be changed to only allow authenticated GCP users if needed.

const weatherApi = new gcp.cloudrun.Service(appName, {
  location,
  name: appName,
  template: {
    spec: {
      containers: [
        {
          image: `gcr.io/${gcp.config.project}/${appName}:${gitSha}`,
          envs: [
            {
              name: "GOOGLE_PROJECT",
              value: gcp.config.project,
            },
            {
              name: "ASPNETCORE_ENVIRONMENT",
              value: environment,
            },
          ],
          ports: [
            {
              containerPort: 8080,
            },
          ],
        },
      ],
      serviceAccountName: googleCloudRunServiceAccount,
    },
    metadata: {
      annotations: {
        "autoscaling.knative.dev/maxScale": "2",
        "run.googleapis.com/cloudsql-instances": cloudSqlInstance,
      },
    },
  },
});

// Open the service to public unrestricted access
const iamWeatherApi = new gcp.cloudrun.IamMember(`${appName}-everyone`, {
  service: weatherApi.name,
  location,
  role: "roles/run.invoker",
  member: "allUsers",
});

The API

The API consists of a single controller that exposes two simple endpoints, one to create weather forecasts and another to retrieve those forecasts. It uses MediatR to implement a CQRS style architecture where each query and command has separate classes to represent the data transfer objects and the handlers.

[HttpGet]
public async Task<IActionResult> Get([FromQuery] GetWeatherForecastQuery query, CancellationToken ct = default)
{
    var response = await _mediator.Send(query, ct);
    return Ok(response);
}

[HttpPost]
public async Task<IActionResult> Post(AddWeatherForecastCommand command, CancellationToken ct = default)
{
    var response = await _mediator.Send(command, ct);
    return Ok(response);
}

The add weather forecast command handler creates a new instance of the WeatherForecast entity and calls the repository to store in the database.

public async Task<AddWeatherForecastResponse> Handle(AddWeatherForecastCommand request,
    CancellationToken cancellationToken)
{
    var id = Guid.NewGuid();

    await _weatherForecastsRepository.AddWeatherForecast(new WeatherForecast(id,
        request.City, request.ForecastDate, request.Forecast), cancellationToken);

    return new AddWeatherForecastResponse(id);
}

While the get weather forecast query handler retrieves the forecast from the database by calling the repository, calling a NotFoundException when it doesn't exist in the database.

public async Task<GetWeatherForecastResponse> Handle(GetWeatherForecastQuery request,
    CancellationToken cancellationToken)
{
    var weatherForecast = await _weatherForecastsRepository.GetWeatherForecast(request.City, request.ForecastDate, cancellationToken);

    if (weatherForecast == null) throw new NotFoundException();

    return new GetWeatherForecastResponse(weatherForecast.Id, weatherForecast.City, weatherForecast.ForecastDate.UtcDateTime, weatherForecast.Forecast);
}

The NotFoundException is mapped to a problem details HTTP 404 error using the .NET library Hellang.Middleware.ProblemDetails.

services.AddProblemDetails(opts =>
{
    var showExceptionDetails = Configuration["Settings:ShowExceptionDetails"].Equals("true", StringComparison.InvariantCultureIgnoreCase);
    opts.ShouldLogUnhandledException = (ctx, ex, pb) => showExceptionDetails;
    opts.IncludeExceptionDetails = (ctx, ex) => showExceptionDetails;
    opts.MapToStatusCode<NotFoundException>(StatusCodes.Status404NotFound);
});

Entity Framework Core is used for data access within the API to read and write data to the Postgres database. While running locally and in integration tests it uses a Postgres Docker container to host the database, while using GCP Cloud SQL when it is running in GCP Cloud Run. GCP Secrets Manager is used to store the database password and is therefore retrieved when the application starts if it is running in Cloud Run.

private static void ConfigureDbContext(IServiceCollection services, IConfiguration configuration)
{
    var dbSettings = configuration.GetSection("WeatherDb");
    var dbSocketDir = dbSettings["SocketPath"];
    var instanceConnectionName = dbSettings["InstanceConnectionName"];
    var databasePasswordSecret = GetDatabasePasswordSecret(dbSettings);
    var connectionString = new NpgsqlConnectionStringBuilder
    {
        Host = !string.IsNullOrEmpty(dbSocketDir)
            ? $"{dbSocketDir}/{instanceConnectionName}"
            : dbSettings["Host"],
        Username = dbSettings["User"],
        Password = databasePasswordSecret,
        Database = dbSettings["Name"],
        SslMode = SslMode.Disable,
        Pooling = true
    };

    services.AddDbContext<WeatherContext>(options =>
        options
            .UseNpgsql(connectionString.ToString())
            .UseSnakeCaseNamingConvention());
}

private static string GetDatabasePasswordSecret(IConfiguration dbSettings)
{
    var googleProject = Environment.GetEnvironmentVariable("GOOGLE_PROJECT");

    if (string.IsNullOrEmpty(googleProject)) return dbSettings["Password"];

    var dbPasswordSecretId = dbSettings["PasswordSecretId"];
    var dbPasswordSecretVersion = dbSettings["PasswordSecretVersion"];

    var client = SecretManagerServiceClient.Create();

    var secretVersionName = new SecretVersionName(googleProject, dbPasswordSecretId, dbPasswordSecretVersion);

    var result = client.AccessSecretVersion(secretVersionName);

    return result.Payload.Data.ToStringUtf8();
}

To test the API all application code is covered by unit tests, with higher level integration tests designed to ensure the API functions correctly when it is running. The integration tests utilise the ASP.NET testing library to run the API in an in-memory server communicating with a Postgres Docker container to store and retrieve the weather forecasts.

protected override void ConfigureWebHost(IWebHostBuilder builder)
{
    if (builder == null) throw new ArgumentNullException(nameof(builder));
    builder.ConfigureServices(services =>
    {
        var sp = services.BuildServiceProvider();

        using var scope = sp.CreateScope();

        var scopedServices = scope.ServiceProvider;
        var db = scopedServices.GetRequiredService<WeatherContext>();
        var logger = scopedServices.GetRequiredService<ILogger<WeatherApiWebApplicationFactory<TStartup>>>();

        db.Database.EnsureCreated();

        InitializeDbForTests(db);
    });
}

private static void InitializeDbForTests(WeatherContext db)
{
    db.WeatherForecasts.RemoveRange(db.WeatherForecasts);

    db.SaveChanges();

    db.WeatherForecasts.Add(new WeatherForecast(Guid.NewGuid(), "Australia/Melbourne",
        new DateTimeOffset(2020, 01, 02, 0, 0, 0, TimeSpan.Zero), 23.35m));

    db.SaveChanges();
}

[Fact]
public async Task GetReturnsCorrectWeather()
{
    const string city = "Australia/Melbourne";
    const string forecastDate = "2020-01-02T00:00:00+00:00";
    var url = QueryHelpers.AddQueryString("/api/weather-forecasts", new Dictionary<string, string>
    {
        {"city", city},
        {"forecastDate", forecastDate}
    });

    using var client = _factory.CreateClient();
    using var response = await client.GetAsync(new Uri(url, UriKind.Relative));
    var responseContent = await response.Content.ReadAsStringAsync();
    var responseObj = JsonSerializer.Deserialize<object>(responseContent) as JsonElement?;

    Assert.Equal(city, responseObj?.GetProperty("city").ToString());
    Assert.Equal(forecastDate, responseObj?.GetProperty("forecastDate").ToString());
    Assert.Equal(23.35m,
        decimal.Parse(responseObj?.GetProperty("forecast").ToString()!, CultureInfo.InvariantCulture));
}

What could come next?

There were several other things that I planned to implement that would generally be required to deploy an application to production safely and ensure it continued to run correctly.

Currently the application is only deployed to a single environment, ideally there should be at least one other environment that it is deployed to, ensuring it works there before deploying to production.

While GCP Cloud Run aggregates the logs from the running container, it does not automatically enable distributed tracing and alerting which will be needed to ensure the application is running as expected in production.

There is currently no automated linting tool used for the C# code, however ReSharper code clean up was used for each commit manually to ensure consistent formatting. The ReSharper CLI or dotnet format could be used as a pre-commit hook to automatically lint the modified code.

Once the API is deployed automated functional tests can be written to ensure that it continues to function correctly in a deployed environment communicating with a real database. Automated security scanning can also be performed against the running API using tools such as OWASP ZAP, to detect any potential runtime vulnerabilities.

The API allows any user to retrieve and create weather forecasts which is far from ideal and also why I manually turn off the database when I am not using it. The API should authenticate users and then ensure they are authorized to either retrieve or create weather forecasts.

The database should be configured to automatically backup on a regular cadence to enable recovery of the data if data is lost. There should also be a separate user for deploying schema migrations and for accessing the database via the API to ensure a compromised application does not result in unintended schema changes.

The deployment pipeline takes roughly 10 minutes to build, test and deploy the application which can definitely be reduced by a number of optimisations such as; caching third party dependencies between steps and pre-building slim purpose built Docker images for each step.

Although the API code and the Dockerfile currently has security scanning, the bash scripts that are used to automate the various steps don't, so a tool such as Shellshock could be used to perform that scanning. The Pulumi TypeScript code also does not have any linting or security scanning.

The API request don't currently perform any validation other than what is built into ASP.NET Core and could therefore use a library such as FluentValidation to perform that validation to ensure the requests are as expected.

I am sure there are more ways I could improve this application to make it more suitable for production workloads and I intend to continue to iterate on it, sharing my learnings as I continue down this road. If anyone reading this post has any ideas to improve this further I am keen to hear them!

Try it yourself

All of the code in this post is part of a sample application that I have hosted as a public repository on GitHub which you can access here. The readme contains steps on how to deploy it to your own GCP project if you would like to try it out. Look forward to hearing your thoughts/feedback in the comments below.

A simpler approach to building event driven software using .NET Core and Kafka

Gabriel Sadaka — Wed, 23 Sep 2020 07:00:10 +0000

Introduction

Writing software using event driven architecture enables high scalability, fault tolerance and flexibility by decoupling services. However, it can introduce significant complexity into the system. Workflows that could be contained within a single system, would now involve having multiple services to achieve the same outcome. In addition, the system used to stream the events between individual components will need to be considered when building each of these components.

I'd like to share an approach for standardizing the processing of events consumed from Kafka in .NET Core applications. This approach enables developers to write the business logic which reacts to events without needing to actively consider message transport specific concerns. This means the developer can choose what they want to focus on based on the work they are currently doing. It is not intended to remove all considerations of the message transport from developers writing event driven systems.

When building event driven systems multiple services can react to a single event with each service focusing on its own specific requirements utilizing the infrastructure it owns. For example when a customer registers an account, one service may record the customer details in its own datastore, while another may send out an email to welcome the customer. The two services involved in this business process handle separate concerns using different infrastructure to process the event. What is common between them though is that they both react to the same event. Since this is a common concern between the two services, handling this aspect consistently is important to ensure it is done correctly and to allow the developers of each service to not have to consider the needs of processing events when writing business logic that reacts to them.

The solution proposed does not cover complex event streaming applications, instead it is designed to support applications that receive events one at a time in order and then perform particular actions as reactions to that event.

It will be built as a shared library to ensure that both the producer and the consumer uses a consistent approach to publishing and consuming messages. Providing a shared library within an organisation that gives developers a consistent approach to react to events allows all of the cross cutting concerns to be handled consistently. It improves developer productivity by allowing them to focus on the business logic required to handle that particular event. It also enables the reactions to the events to be handled independently making it easier to test in isolation, therefore increasing confidence in the quality of the code.

Adding new reactions to events then becomes straightforward, due to the consistent approach, independence of event handlers and automatic registration of the handlers. By automatically registering the handlers of events it allows a developer to focus purely on the code that will be reacting to the event without having to focus on the concerns of consuming that event. It makes it easier to add more services that listen to events and to also add more reactions within a given service for particular events.

The approach enables the consumer to be built using a vertical slice architecture where each event is handled by an independent handler with it's own dependencies injected within an isolated scope. This makes it easier to adopt a CQRS approach and the ability to test the handling of each event in isolation. Similar to ASP.NET Core where all HTTP requests are executed within their own dependency injection scope, each event will be processed within their own scope to ensure nothing is shared between events.

Message Consumer Sequence Diagram

The sequence diagram below outlines the interactions between the objects within the consumer for the proposed solution.

Message Consumer

In the proposed solution to consume a message that is published to a Kafka topic a MediatR.INotificationHandler<TNotification> will be needed that subscribes to a MessageNotification<TMessage> type where TMessage is the concrete type of the message to be consumed. This handler will be automatically found and registered with a matching Kafka consumer that will listen to the topic it is published to.

This allows developers to easily add event handlers without worrying about messaging infrastructure concerns. It also makes writing handlers using TDD straightforward as all the tests will be purely focused on the business logic without any messaging specific concerns leaking into them.

public class SampleMessageLoggerHandler : INotificationHandler<MessageNotification<SampleMessage>>
{
    private readonly ILogger<SampleMessageLoggerHandler> _logger;

    public SampleMessageLoggerHandler(ILogger<SampleMessageLoggerHandler> logger)
    {
        _logger = logger;
    }

    public Task Handle(MessageNotification<SampleMessage> notification, CancellationToken cancellationToken)
    {
        var message = notification.Message;

        _logger.LogInformation(
            $"Sample message received with key: {message.Key} and value: {message.SomeProperty}");

        return Task.CompletedTask;
    }
}

The type of the message has the properties that will be found on the serialised message and the topic that the message will be published to. The MessageTopicAttribute attribute will be used to define which topic the message is published to so the worker knows which topic to subscribe to.

[MessageTopic("sample-messages")]
public class SampleMessage : IMessage
{
    public SampleMessage(string key, string someProperty)
    {
        Key = key;
        SomeProperty = someProperty;
    }

    public string Key { get; }

    public string SomeProperty { get; }
}

[AttributeUsage(AttributeTargets.Class)]
public class MessageTopicAttribute : Attribute
{
    public MessageTopicAttribute(string topic)
    {
        Topic = topic;
    }

    public string Topic { get; }
}

The MessageNotification<TMessage> class is a wrapper that encapsulates the deserialized consumed message along with any other properties that could be relevant to processing that message e.g. key, headers, correlation ID. It implements the MediatR.INotification to allow MediatR.INotificationHandler<TNotification> handlers to be written that subscribe to that notification.

public class MessageNotification<TMessage> : INotification
    where TMessage : IMessage
{
    public MessageNotification(TMessage message)
    {
        Message = message;
    }

    public TMessage Message { get; }
}

What happens behind the scenes to make a handler work?

The core of the solution to the consumer worker is the KafkaMessageConsumerManager which is responsible for using reflection to scan the registered message notification handlers.

private static IEnumerable<string> GetTopicsWithNotificationHandlers(IServiceCollection services)
{
    var messageTypesWithNotificationHandlers = services
        .Where(s => s.ServiceType.IsGenericType &&
                    s.ServiceType.GetGenericTypeDefinition() == typeof(INotificationHandler<>))
        .Select(s => s.ServiceType.GetGenericArguments()[0])
        .Where(s => s.IsGenericType &&
                    s.GetGenericTypeDefinition() == typeof(MessageNotification<>))
        .Select(s => s.GetGenericArguments()[0])
        .Where(s => typeof(IMessage).IsAssignableFrom(s))
        .Distinct();

    return messageTypesWithNotificationHandlers
        .SelectMany(t => Attribute.GetCustomAttributes(t))
        .OfType<MessageTopicAttribute>()
        .Select(t => t.Topic)
        .Distinct()
        .ToList();
}

It uses the results of the scan to determine what topics have registered handlers so it can then start a new thread with a new KafkaTopicMessageConsumer for each of those topics. This ensures each topic has a dedicated consumer running in an independent thread to ensure multiple topics can be consumed concurrently.

The CancellationToken passed into the method is the token that the worker receives to represent a request to stop the application which could happen as a request by the user in an interactive terminal using CTRL+C or it can be due to the scheduler running the application requesting it to shut down e.g. Kubernetes stopping the running pod.

public void StartConsumers(CancellationToken cancellationToken)
{
    var topicsWithNotificationHandlers = GetTopicsWithNotificationHandlers(_services);

    foreach (var topic in topicsWithNotificationHandlers)
    {
        var kafkaTopicMessageConsumer = _serviceProvider.GetRequiredService<IKafkaTopicMessageConsumer>();

        new Thread(() => kafkaTopicMessageConsumer.StartConsuming(topic, cancellationToken))
            .Start();
    }
}

The KafkaTopicMessageConsumer is responsible for building a new consumer and then subscribing to the topic with that consumer. It will then continuously request to consume a message from the Kafka topic until the cancellation token requests a cancellation.

For each message it receives it will read the message-type header which is a custom header set by the producer to determine the C# type of the serialized message in the body. Once it determines the type it deserializes it using JSON.NET and then creates a new MessageNotification<TMessage> instance using reflection which is a wrapper class that encapsulates the deserialized message. This allows multiple message types to be consumed from the same topic.

A new dependency injection scope is then created to ensure all scoped dependencies that are used in processing the message are not shared between the processing of other messages. Using this new scope it requests an instance of MediatR which it then uses to publish the MessageNotification<TMessage> instance to all registered notification handlers.

It publishes the message to MediatR synchronously waiting for it to finish running all of the handlers for that particular message to ensure that each message is fully processed before the next message in the topic is processed.

public void StartConsuming(string topic, CancellationToken cancellationToken)
{
    using (var consumer = _kafkaConsumerBuilder.Build())
    {
        _logger.LogInformation($"Starting consumer for {topic}");
        consumer.Subscribe(topic);

        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                var consumeResult = consumer.Consume(cancellationToken);

                var messageTypeEncoded = consumeResult.Message.Headers.GetLastBytes("message-type");
                var messageTypeHeader = Encoding.UTF8.GetString(messageTypeEncoded);
                var messageType = Type.GetType(messageTypeHeader);

                var message = JsonConvert.DeserializeObject(consumeResult.Message.Value, messageType);
                var messageNotificationType = typeof(MessageNotification<>).MakeGenericType(messageType);
                var messageNotification = Activator.CreateInstance(messageNotificationType, message);

                using (var scope = _serviceProvider.CreateScope())
                {
                    var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
                    mediator.Publish(messageNotification, cancellationToken).GetAwaiter().GetResult();
                }
            }
        }
        catch (OperationCanceledException)
        {
            // do nothing on cancellation
        }
        finally
        {
            consumer.Close();
        }
    }
}

Message Producer

The KafkaMessageProducer class in the proposed solution is a simple singleton class that takes the instance of the message to be published, determines the topic it will be published to, serializes it then publishes it to the Kafka topic. It is responsible for ensuring a single instance of the Kafka producer is built using the Lazy pattern as it is expensive to build them and it is safe to use a single shared instance for the lifetime of the application.

The message-type header is set with the name of the message type to allow the consumer on the other end to determine how to deserialize and consume it. The MessageTopicAttribute is used to determine the topic of the message to be published. This restricts a particular message type to be published to a single topic, while allowing multiple message types to be published to a single topic. It is possible to move the configuration for a message type's topic outside of the class to allow it to be published to multiple topics if required.

public class KafkaMessageProducer : IMessageProducer, IDisposable
{
    private readonly Lazy<IProducer<string, string>> _cachedProducer;

    public KafkaMessageProducer(IKafkaProducerBuilder kafkaProducerBuilder)
    {
        _cachedProducer = new Lazy<IProducer<string, string>>(() => kafkaProducerBuilder.Build());
    }

    public void Dispose()
    {
        if (_cachedProducer.IsValueCreated) _cachedProducer.Value.Dispose();
    }

    public async Task ProduceAsync(string key, IMessage message, CancellationToken cancellationToken)
    {
        var serialisedMessage = JsonConvert.SerializeObject(message);
        var topic = Attribute.GetCustomAttributes(message.GetType())
            .OfType<MessageTopicAttribute>()
            .Single()
            .Topic;

        var messageType = message.GetType().AssemblyQualifiedName;
        var producedMessage = new Message<string, string>
        {
            Key = key,
            Value = serialisedMessage,
            Headers = new Headers
            {
                {"message-type", Encoding.UTF8.GetBytes(messageType)}
            }
        };

        await _cachedProducer.Value.ProduceAsync(topic, producedMessage, cancellationToken);
    }
}

Sample Application

In the sample application there is a common library that contains the code to communicate to Kafka for both consumers and producers. This enables all producers and consumers to use a consistent approach to ensure all messages can be published and then consumed with confidence. It also allows new services to be developed without needing to worry about the Kafka specific concerns. Ideally in an organisation there will be a platform team that is responsible for maintaining the Kafka producer and consumer shared libraries so there is a standard approach across the entire organisation.

The common library has a single entry point for both consumers and producers that register all of the necessary dependencies using the Microsoft Dependency Injection abstractions. It also uses a common class that represents the configuration options that are required to connect to Kafka.

There is a .NET Core worker for the Consumer which leverages the shared DI setup from the Common.Kafka library. It has 3 notification handlers, one for each message published by the producer.

public static IHostBuilder CreateHostBuilder(string[] args)
{
    return Host.CreateDefaultBuilder(args)
        .ConfigureServices((hostContext, services) =>
        {
            services.AddHostedService<Worker>();

            services.AddOptions<KafkaOptions>()
                .Bind(hostContext.Configuration.GetSection("Kafka"));

            services.AddKafkaConsumer(typeof(Program));
        });
}

There is a .NET Core worker for the Producer which leverages the shared DI setup from the Common.Kafka library. It publishes 3 separate messages, two of which are published to the same topic every second. This shows that multiple message types can be published to the same topic while being consumed by independent handlers.

public static IHostBuilder CreateHostBuilder(string[] args)
{
    return Host.CreateDefaultBuilder(args)
        .ConfigureServices((hostContext, services) =>
        {
            services.AddHostedService<Worker>();

            services.AddOptions<KafkaOptions>()
                .Bind(hostContext.Configuration.GetSection("Kafka"));

            services.AddKafkaProducer();
        });
}

Each producer has its own library that has the types of the messages that will be published allowing consumers to reference these libraries without having to depend directly on the producer. An alternative approach is to use Confluent Schema Registry to store the message schemas which provides a central place for producers and consumers to ensure they are adhering to the latest schema. Using the Confluent Schema Registry will require a schema type that supports multiple message types per topic in .NET which at time of writing is only Avro.

Both the Producer worker and the Consumer worker each have a basic Dockerfile that builds and starts them. They are then connected to each other along with a single Kafka broker and ZooKeeper instance using Docker Compose which enables the sample to be run locally without any external dependencies to be setup. It uses automatic topic creation to simplify the setup, however in a production environment this is not recommended.

Consistent handling of cross cutting concerns

Having a consistent approach to producing and consuming messages enables cross cutting concerns to also be handled consistently. The sample application does not implement this, however it can be extended to support the Interceptor pattern which will allow different steps of the message processing to be intercepted to handle various cross cutting concerns.

To handle common production support scenarios interceptors can be developed to provide context about the events that are processed. A logging interceptor could be developed that produces a log message with the type and key of every message consumed, or a metrics interceptor that records the time each event takes to process. To enable distributed tracing a correlation ID propagator interceptor could be developed to parse the incoming correlation ID and store it in a scoped dependency which can then be used to propagate to downstream API calls or further events raised.

The sample application does not currently handle exceptions that are raised while processing messages, so an exception handler can be developed as an interceptor to solve this problem. Depending on the specific application, one could simply log and skip the message being processed if an error occurs or could halt the entire worker or publish the message to a dead letter queue.

An interceptor can be developed to handle authorisation by blocking the processing of messages that fail authorisation checks. An example is an interceptor that confirms that the incoming message has an authorisation token with the claims that are required to perform the particular action.

Libraries

Confluent Kafka .NET Client

The Confluent Kafka .NET Client is used to communicate with Kafka for both the producer and the consumer code. It is an open source library that is supported by Confluent which is a well recognized commercial provider supporting Kafka. They also provide the docker images that are used within the sample application that enable it to be run consistently on any machine that has docker compose installed.

MediatR

In the sample application the MediatR .NET library is used to implement the mediator pattern which is responsible for passing the MessageNotification<TMessage> instance to the correct INotificationHandler<TNotification>. It is a simple open source battle tested library that is commonly used in the .NET ecosystem with support for all major dependency injection frameworks.

Since it is a simple implementation it does not support complex scenarios, such as filtering handlers based on attributes which could enable more use cases if necessary. Therefore it is recommended to implement the mediator pattern from scratch if MediatR does not the use cases required.

Alternative Approaches

There are a number of alternative approaches to the one outlined in the sample application that can be adopted depending on the needs of the applications that will be producing and consuming events to and from Kafka.

Attribute on handle method

An alternative approach to defining the topic for the message is to apply the attribute to the Handle method of the handler that is handling the message rather than the message class itself. This allows a single message type to be published to multiple topics and therefore be consumed from multiple topics.

Using this approach within the sample application will require either building a custom mediator or extending MediatR to support filtering the handler based on the attributes of the Handle method for the particular message type. The Spring Kafka framework for Java uses this approach where the method is annotated with a @KafkaListener annotation that has a variety of possible configuration options. This will increase the complexity of KafkaMessageConsumerManager and KafkaTopicMessageConsumer as the manager will need to know how to find attributes on the Handle method while the consumer will need to selectively call handlers based on the parameters of the attribute.

Conclusion

There are a number of ways to evolve the proposed solution to handle more complex scenarios or to enable more robust error handling. I will be exploring ways to evolve the solution myself however it would be great to hear from others about what ideas they may have.

Building event driven applications is a complex endeavour so it is important to ensure that the core aspects that are relevant to the systems being built are handled in a consistent manner. This approach provides a starting point to build a framework that addresses those concerns.

Please refer to my Github repository for the sample application.