Forem: Reza Alipour

A simple local development environment with a complete CI/CD pipeline

Reza Alipour — Sun, 01 Oct 2023 11:48:15 +0000

Preface

Modern DevOps methodologies concentrate on being seamlessly reproducible and declarative. The intention is to go through the application development once and be able to run the same application in various environments without the need to do additional work such as installing dependencies.

Moreover, it is preferred to have the deployment process as a written description that can be audited and versioned. Such a description could be considered a desired state in which the application is supposed to work. Automation tools can now use the description to narrow the gap between the actual state and the desired state, without the need for a human operator to go through the deployment process imperatively. A complete DevOps environment with a declarative mindset is referred to as GitOps.

The state-of-the-art architecture to implement a GitOps environment is predicated on containerized applications with a shared language known as CRI and OCI specifications, together with a set of tools based on Kubernetes.

In this article, a local DevOps environment is devised which knows those shared languages. The outcome of the development would be cloud-friendly and can be deployed in many production environments with little to no effort. We use Docker Desktop, Gitlab tools, and Kubernetes on macOS to implement the proposed environment.

TL;DR: The project addressed in this article is accessible from this repository.

Setup

First, you need to create an account on Gitlab's official website, the instructions for which are available on the Internet. Then, install Docker Desktop and enable Kubernetes from Settings -> Kubernetes. It will take time to download relevant images and setup a local K8s instance on Docker, so be patient. Also, make sure you have internet access to Docker Hub and Docker Desktop is able to download K8s images.

The generated K8s instance must be accessible through the kubectl CLI. Running the following command should return docker-desktop in the list of contexts:



kubectl config get-contexts

You can also use other single-node variants of a K8s cluster such as minikube.

Next, you have to install Helm client to facilitate complex deployments on K8s. Kubernetes resources provide a rich toolset for application deployment, but it doesn't provide a built-in packaging out of the box. Helm is a community-driven tool to aggregate all K8s resource manifests(descriptions) for an application. The Helm packages are called charts and instances of charts deployed on a K9s cluster are called releases. We will create a Helm chart and create a release for our own application throughout this article. Helm client can be installed on a macOS machine with homebrew using the following command:



brew install kubernetes-helm

Continuous Integration

Suppose you have a simple Go web server on port 8080 with a single /ping endpoint. You wish to have a local CI/CD pipeline for this application. The application will sleep for a short random time and then respond with a pong! string.



package main

import (
    "fmt"
    "math/rand"
    "net/http"
    "time"
)

func Handler(w http.ResponseWriter, req *http.Request) {

    time.Sleep(time.Duration(rand.Intn(800)+200) * time.Millisecond)

    w.Write([]byte("pong!"))
}

func main() {

    rand.Seed(time.Now().Unix())

    http.HandleFunc("/ping", Handler)

    if err := http.ListenAndServe(":8080", nil); err != nil {
        fmt.Println(err)
    }
}

To run a Gitlab CI script, you need to provide a runner for your project. If you have a valid credit card, you can use Gitlab's shared runners for free. Otherwise, you need to setup a runner on your local machine and attach it to your project on Gitlab. We are going to run a gitlab-runner instance on a local K8s cluster using official gitlab helm charts. First, you need to add Gitlab's helm repo:



helm repo add gitlab https://charts.gitlab.io

Then update the repo:



helm repo update gitlab

Now create a values.yaml file and put the following lines inside it:



gitlabUrl: https://gitlab.com/
runnerRegistrationToken: <REGISTRATION_TOKEN>

rbac:
  create: true
  rules:
    - apiGroups: [""]
      resources: ["pods"]
      verbs: ["list", "get", "watch", "create", "delete"]
    - apiGroups: [""]
      resources: ["pods/exec"]
      verbs: ["create"]
    - apiGroups: [""]
      resources: ["pods/log"]
      verbs: ["get"]
    - apiGroups: [""]
      resources: ["pods/attach"]
      verbs: ["list", "get", "create", "delete", "update"]
    - apiGroups: [""]
      resources: ["secrets"]
      verbs: ["list", "get", "create", "delete", "update"]      
    - apiGroups: [""]
      resources: ["configmaps"]
      verbs: ["list", "get", "create", "delete", "update"]      

runners:
  config: |
    [[runners]]
      [runners.kubernetes]
        privileged = true
      [[runners.kubernetes.volumes.empty_dir]]
        name = "docker-certs"
        mount_path = "/certs/client"
        medium = "Memory"

You have to replace <REGISTRATION_TOKEN> with your own Gitlab registration token. To obtain the token go to Gitlab's page for your project. Go to Settings -> CI/CD and then expand the Runners section. Under this section, click on New project runner and after providing some details of the runner you want to create, click on create runner. Gitlab will provide a token that you can replace inside your values.yaml file.

Note that you have deployed the runner in the privileged: true setting to enable Docker-in-Docker mode. Enabling a pod in a privileged mode poses security risks and is not recommended. We talk about an alternative approach later.

We will put all gitlab related deployments in a dedicated namespace:



kubectl create namespace gitlab

Now, run the following command inside the directory in which you have created your values.yaml file:



helm install --namespace gitlab gitlab-runner -f ./values.yaml gitlab/gitlab-runner

If installation succeeds, you should be able to see your runner's state as online on Gitlab. At this point, you are able to run a CI script on your local runner. The runner you have created is a pod on your local Kubernetes cluster, which doesn't have access to your Docker daemon to build the image. The first solution is to use Docker-in-Docker technique to run a Docker daemon container or dind to build the image. The CI script to build the application image and then pushing it to the Gitlab registry using this approach is:



build_docker:
  image: 
    name: docker:20.10.16
  variables:
    DOCKER_HOST: tcp://docker:2376
    DOCKER_TLS_CERTDIR: "/certs"
    DOCKER_TLS_VERIFY: 1
    DOCKER_CERT_PATH: "$DOCKER_TLS_CERTDIR/client"
  services:
    - docker:20.10.6-dind
  stage: build 
  before_script:
    - docker login $CI_REGISTRY -u $CI_REGISTRY_USER -p $CI_REGISTRY_PASSWORD
  script:
    - docker build -t "${CI_REGISTRY_IMAGE}:${CI_COMMIT_TAG}" .
    - docker push $CI_REGISTRY_IMAGE:$CI_COMMIT_TAG
  after_script:
    - docker rmi $CI_REGISTRY_IMAGE:$CI_COMMIT_TAG

  rules:
    - if: $CI_COMMIT_TAG

To start a pipleline, you need to create a tag from your main branch. Then you can see jobs of the pipeline start to run.

As we discussed earlier this approach has its caveats pertaining security concerns. A better solution is to use Google's Kaniko base image to run the pipeline and build the image. Kaniko doesn't need privileged access and builds the image itself. Therefore, you can remove the runners section from the runner's values.yaml file and redeploy your gitlab runner. The corresponding CI script would be something like this:



build:
  stage: build
  image:
    name: gcr.io/kaniko-project/executor:v1.9.0-debug
    entrypoint: [""]
  script:
    - /kaniko/executor
      --context "${CI_PROJECT_DIR}"
      --dockerfile "${CI_PROJECT_DIR}/Dockerfile"
      --destination "${CI_REGISTRY_IMAGE}:${CI_COMMIT_TAG}"
  rules:
    - if: $CI_COMMIT_TAG

Now that you have built and pushed your application's image to Gitlab's container registry, you need to write a deployment description for your image and deploy it on your Kubernetes cluster using helm. First, you need to make sure Kubernetes has access to your container registry on Gitlab. To do so, first you have to generate a deploy token from Settings -> Repository and expanding the Deploy token section. Give your token a name and read_registry access and then click on create. Use the generated token's username and password and run the following command on your machine:



kubectl create secret docker-registry regcred --docker-server=registry.gitlab.com --docker-username=<token_username> --docker-password=<token>

This command will generate a secret named regcred, which we instruct Kubernetes to use for pulling images from container registry. Then, create a helm chart under /deployment directory of your project. Inside the /deployment directory run the following command to create a boilerplate for your helm chart:



helm create simple-api

It is recommended to have your helm chart inside a separate repository, but here we have create the files inside our application repository for the sake of simplicity. Delete the contents of the template directory and replace values.yaml file with the following content:



# Default values for simple-api.
# This is a YAML-formatted file.
# Declare variables to be passed into your templates.

replicaCount: 2

image:
  repository: <path_to_your_repository>
  pullPolicy: IfNotPresent
  # Overrides the image tag whose default is the chart appVersion.
  tag: <image_tag>

Replace the image's repository and tag accordingly. Then, Setup a deployment and a service resources inside templates folder:



apiVersion: apps/v1
kind: Deployment

metadata:
    name: simple-api

spec:
  replicas: {{ .Values.replicaCount }}
  selector:
    matchLabels:
      app: simple-api
  template:
    metadata:
      labels:
        app: simple-api
    spec:
      containers:
      - name: simple-app
        image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
        ports:
        - name: http
          containerPort: 8080
          protocol: TCP
      imagePullSecrets:
      - name: regcred

Note the imagePullSecrets section in the manifest.



apiVersion: v1
kind: Service 

metadata:
    name: simple-api

spec:
    selector:
        app: simple-api
    ports:
    - port: 8000
      targetPort: 8080
    type: LoadBalancer

Create a release from your chart by running the following command from your deployment directory:



helm install -f simple-api/values.yaml simple-api ./simple-api

You should be able to access your web server on port 8000.

Continuous Deployment using ArgoCD

For the next step, you can install ArgoCD on your k8s cluster. First, you have to create a dedicate namespace for it:



kubectl create namespace argocd

Then install ArgoCD resources via the following command:



kubectl apply -n argocd -f https://raw.githubusercontent.com/argoproj/argo-cd/stable/manifests/install.yaml

After successful installation, port forward the ArgoCD service to be accessible:



kubectl port-forward svc/argocd-server -n argocd 8080:443

The service will be accessible on port 8080. Alternatively, you can modify ArgoCD service to make it a LoadBalancer type which is accessible on port 443.



kubectl patch svc argocd-server -n argocd -p '{"spec": {"type": "LoadBalancer"}}'

Get the admin password using the following command and login to the UI using admin and the generated password.



kubectl -n argocd get secret argocd-initial-admin-secret -o jsonpath="{.data.password}" | base64 -d; echo

To deploy your helm chart using ArgoCD you need to connect ArgoCD to your repository. To do that you have to create a deploy token with read access rights just like you created for Kubernetes. After that, go to the Settings -> Repositories section on ArgoCD UI and connect to your repository using HTTPS mode and your deploy token's username and password.

Then under Applications section click on New App and define your application. After editing, the yaml file of your app should look like below:



apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: simple-api
spec:
  destination:
    name: ''
    namespace: default
    server: 'https://kubernetes.default.svc'
  source:
    path: deployment/simple-api
    repoURL: <your_repository_url>
    targetRevision: HEAD
    helm:
      valueFiles:
        - values.yaml
  sources: []
  project: default
  syncPolicy:
    automated:
      prune: false
      selfHeal: false

Click on Create and wait for the application to deploy. After deployment your application must be Healthy and Synced. At this point, changing your deployment is a matter of changing your helm chart in your git repository.

Summary

We have introduced a very basic GitOps flow using Docker, Kubernetes, Helm, Gitlab, and ArgoCD. An aggregated CI/CD flow in this article can be displayed as follow:

A possible improvement to this setup would be to use a local registry to save your images.

Why is Go so darn fast?

Reza Alipour — Mon, 26 Dec 2022 12:41:33 +0000

Go is famous for being fast. In fact, Go's outstanding performance manifests the best when used in concurrent tasks. While many facets enable Go to perform well in concurrent tasks, there is one particular ingenious in Go that is of great importance. Go helps the underlying OS and takes responsibility for some of the heavy lifting.

Disclaimer: This post is not meant to be literally true with respect to every detail, nor does it explain everything relevant to the topic. It's just an attempt to simplify a thorough concept.

To understand the extent of this help, a short context is in order. There are quite a few hardware-level CPU cores on a single PC, e.g. 8 logical CPU cores on an Intel Core-i7 processor. However, when you boot up a PC, there are probably hundreds of processes that run simultaneously. How is that possible? Every prominent OS has a task scheduler is tasked to allocate the processor's resources to threads. A thread is a path of execution containing an ordered set of machine instructions and each process has one or more of them.

Generally, a thread can be in three different states: It's in Executing when it's currently running on a CPU core. It's in Runnable state when it's ready to be executed, and it's in aa Wating state when it is stopped, waiting for something to happen, e.g. an I/O operation, and cannot be run. Threads switch between these states and when they are ready the scheduler designates them a time slot on a CPU core to be executed. A naive scheduler would simply split the core's slot uniformly and give each runnable thread a similar share. For example, if the core's slot is 1s and there are 100 runnable threads, each thread would get 10ms of the core's computation time.

If there are threads to be run, the scheduler makes sure none of the cores would remain idle. The procedure in which the scheduler takes a thread off a CPU core and replaces another one to be executed is called a context switch. The point is, context switching is tremendously expensive. It's been said that each context switch takes the equivalent time of running a few thousand machine instructions.

Two types of threads exist. Some threads never encounter a point where they should wait for some system call or a request for network request to accomplish before they can go further and they are always runnable. The instruction sequence in these threads can push forward seamlessly and the only limit to how the fact they can execute is the computation power of the CPU. Therefore, they are called CPU-Bound threads. Naturally, there is another type of thread that is occasionally stopped for some IO tasks and cannot continue until those IO tasks are complete. That is to say, they get into the waiting state along the way and when they do, they are switched by the OS scheduler. These are called IO-Bound threads.

Normally, when performing concurrent tasks via a single process, employing multithreading would prompt the programming language's runtime to split the tasks into threads and let the OS manage the way they are executed. If the threads are IO-Bound, the scheduler's context switches would let them run almost in parallel and therefore there is a performance surge compared to a naive approach where the core would remain idle when a thread is waiting for an IO task. However, If this process has multiple threads over each core and all those threads are CPU-Bound, the ensuing context switches would result in the process's performance dropping drastically. That is why running simple CPU-Bound computations concurrently over a multitude of threads, say a million, is not a good idea.

The fact is, when the OS scheduler receives the threads, the context switching overhead is there regardless of the types of threads running. At the application level, there is a trade-off to be made between the context switching overhead and program's idle time through the number of threads. What makes context switching cost-effective is that the payoff, avoiding idle cores, is magnificent when IO-Bound threads are involved. More importantly, context switching would make different processes look like they are running in parallel. These make the overhead inevitable at the OS level.

But, what if we could mitigate such an overhead at the application level? Runtime has more control over the threads of an application. If one could exploit the privileges of runtime over its threads to perform context switches more efficiently, the overhead would drop and the performance would improve. The Go's runtime scheduler is apparently designed with the same mentality.

Go has introduced a new abstract layer over threads in its runtime to decouple ordered tasks, execution threads, from the OS threads. Go's conceptual model calls a logical CPU core a processor or a (P) and a thread a machine or an (M). Threads of execution are called Goroutine or (G) and they are different than (M)s. Go assigns every (P) an initial (M) and (G)s are distributed between (P)s for execution. In this model, as the OS scheduler performs context switching of threads over a CPU core, the Go scheduler performs context switching of goroutines over a host thread a.k.a (M).

A lot is going on about what tricks the Go scheduler uses to perform context switching such as networking polling and asynchronous system calls and work stealing, which we are not going to discuss. But, the outcome is that the context switching of IO-Bound goroutines over a thread makes that thread look like CPU-Bound for the OS scheduler. From the OS's perspective, the program's threads, (M)s, always have work to do, even though these works are actually related to different goroutines. Since the Go scheduler's context switching is way cheaper than that of the OS, Go runtime has managed to minimize the overhead of the context Switch while making sure the program experiences the minimum possible idle time on each of the CPU cores. Consequently, Go gains performance by optimizing on a trade-off explained above.

Handling variable-length Kafka tasks using Python

Reza Alipour — Fri, 16 Dec 2022 14:12:46 +0000

Let's say you have a service that takes customers' requests, distributes them as events amongst several workers, waits for the workers to perform a long-running process for each request, and gathers the results to send back to customers. If you intend to use Apache Kafka as the medium to populate events and you are not careful, chances are you are going to encounter a certain predicament: Your requests will be processed more than once by different workers!

TL;DR: Handling Kafka event consumption when processing events needs a long variable-length time is particularly challenging. This post is a discussion on this challenge and possible solutions for it. A simple implementation of a chosen solution, which is accessible via this link, is also described.

Introduction

Kafka is a prominent event-streaming infrastructure. In Kafka's world, events (or requests in our simple scenario) are generated by a multitude of entities called producers. Kafka classifies these events into a predefined set of semantically close groups called topics. Producers decide to which topic their event belongs. Consumers are entities that read these events and do something over them accordingly. Each topic can have multiple producers and consumers.

Consumers can cooperate and form a group to share a topic and read from it collectively. Producers and consumers are meant to be fully decoupled as they can write (push) or read (pull) events whenever they want. Kafka is responsible to manage and retain those events. Since topics are accessible to read from and write to through more than one interface called brokers, Kafka partitions events of a topic once more to avoid any inconsistencies between the topic's consumers connected to different brokers. Each topic-partition is assigned to a unique consumer. Consumers can have more than one topic-partition. Upon event receipt, each consumer of a group must send a delivery acknowledgment (commit) to the broker so Kafka doesn't send it again to the group.

Note that this design doesn't permit consumers of a topic to be more than partitions. Partitions are where Kafka respects the order of events and will serve events to the consumer in a FIFO¹ manner. Ordering is not preserved at the topic level in Kafka.

The problem

While you are the one in charge of defining topics, the number of partitions, and the number of consumers up to the number of partitions, Kafka decides which topic-parition is assigned to which consumer from the consumer group. It constantly checks for live consumers and designates a topic-partition. If a consumer disconnects from a broker, Kafka reassigns its topic-partition to another consumer, a procedure called Rebalancing.

The point is, Kafka needs to know which of the consumers are alive and which of them have disconnected from the broker to be able to perform rebalancing. Therefore, Kafka mandates every consumer in the consumer group to send a heartbeat at least every max.poll.interval.ms milliseconds to affirm their presence. This is also the interval in which the consumer will request new events from the Kafka broker.

So far so good. If your consumers receive an event, perform a quick process on it, commit, and are ready to take the next event in a time much shorter than max.poll.interval.ms, congratulation! You are all set. The problem arises when your single-threaded application takes more than max.poll.interval.ms to process an event and consequently misses the heartbeat deadline.

Let's see why you don't want this to happen. Suppose events A is in topic t and topic-partition t-p and two single-threaded consumers alpha and beta have formed a group to read from topic t. Also, suppose Kafka assigns t-p to alpha and it consumes event A immediately. If alpha takes more than max.poll.interval.ms to send the next heartbeat, Kafka assumes alpha has been disconnected from the group and reassigns t-p to the beta at the next Rebalancing. Now since there is no commit to exclude A, Kafka will resend the A to beta. If the processing time for A is long enough to repeat the pattern for B as well, your service will be stuck in a livelock. It's important to mention that if alpha tries to send commits during or after rebalancing, it will fail.

How to mitigate the situation

The trivial solution would be to set max.poll.interval.ms big enough to avoid such a situation altogether. There are two downfalls to this quick patch. First, if the time to process your tasks varies significantly, as is indeed the case for many machine learning tasks where computation time depends on the size of the dataset or algorithm, your chances of finding a threshold that guarantees all your tasks will fall below it are slim. In addition, as discussed here, if you want to introduce new consumers to the group, or you need a rebalance due to failed consumers, you will have to wait a significant amount of time.

Another somewhat obvious workaround would be to commit right after event receipt and before processing it, but that exacts reducing processing acknowledgment to just a delivery acknowledgment. If the process fails for a reason, the event is lost. Some might argue having another pipeline from the consumer to the producer will notify the producer of the lost event. That will not be of much help since you won't get any response pertaining to the failed event. That means you have to specify a duration threshold to distinguish between in-process events from failed events and resend failed events to the broker, which is
not easy when your tasks are variable-length.

The third solution expands on the previous one and assumes the application has a dedicated main thread for dealing with the Kafka broker and a bunch of worker threads to process the incoming events. The main thread receives events and caches them in a local queue, from which it can feed the workers. The main
thread distributes events while sending constant heartbeats to the Kafka broker. It then gathers the results from workers, commits accomplished tasks, and reassigns failed tasks.

This is particularly tricky to implement for Kafka as Kafka doesn't accept commits of individual events. What you have to commit is just an offset that points to the last place where events have been processed. As a result, the main thread must ensure every event up to a certain point is processed before sending a commit of that point. Therefore, if the main thread crashes before all consecutive events up to a point are processed, those events will all be processed again after rebalance.

A workaround for the above problem would be to transform every worker into a unique consumer and assign at least one topic-partition to each worker discretely instead of having them obtain events from the main thread. Every worker would have two threads: one responsible to accept events and sending constant heartbeats and another to process events as they arrive. Events will be immediately acknowledged as their process completes and also the consumers won't occupy themselves with queue management and
separation of concerns will follow. This is our solution of choice in this post.

Next, we are going to talk about how to implement this solution in Python.

Python Implementation

This implementation is not meant to be bug-free and production-ready and is just a glimpse into what the proposed solution might look like. Additionally, we assume the producer application is already in place and produces events. The consumer application receives events as they are produced, process them, commits them on Kafka, and sends a new event containing the results to a separate topic to inform the producer of the result of the process. We also use Protocol Buffers to serialize our communication.

What every worker is about to do is cpu-bound. Therefore, separating workers as threads is not ideal as threads are meant for io-bound tasks rather than cpu-bound tasks. We need a module to provide a platform to spawn multiple worker instances as processes and execute them. The following module would do the job:

from pathos.multiprocessing import Pool
from tools.log.logger import Logger
from typing import Callable
import os
import signal


# to ignore events being sent to process children. The parent must
# take care of the event handling, e.g., keyboard interrupts
def init_worker():
    signal.signal(signal.SIGINT, signal.SIG_IGN)



class WorkerPool(Logger):

    '''
    worker pool gets a worker routine function and spawns a certain 
    number of processes with that routine.

    worker_routine: the worker function to be spawn and executed
    number_of_workers: number of workers to be created and executed

    after instantiation call the execute method and pass along the argument
    you want to send to the worker. 
    '''

    def __init__(self, worker_routine: Callable, number_of_workers: int=os.cpu_count()) -> None:

        # instantiate a process-safe logger
        Logger.__init__(self, "worker_pool")

        # validate the inputs before proceeding
        if not isinstance(number_of_workers, int):
            raise TypeError("number of workers is not correct")

        if not callable(worker_routine):
            raise TypeError("no worker specified")

        # limit the number of worker to the number of cpus 
        self.__number_of_workers = min(number_of_workers, os.cpu_count())
        self.__worker_routine = worker_routine

        if(self.__number_of_workers < number_of_workers):
            self.logger.warning(f"number of cpus limit dictated {self.__number_of_workers} workers instead of {number_of_workers} you provided")



    def execute(self, *args, **kwargs):

        self.logger.info(f"starting worker pool with {os.cpu_count()} cpus and {self.__number_of_workers} workers")

        # augment worker's arguments with its id
        def worker_routine_wrapper(worker_number):
            kwargs["worker_number"] = worker_number
            self.__worker_routine(*args, **kwargs)


        # Use initializer to ignore SIGINT in child processes
        with Pool(initializer=init_worker) as mp_pool:

            try:

                mp_pool.map(
                        worker_routine_wrapper,
                        range(self.__number_of_workers)
                    )

            except KeyboardInterrupt:
                self.logger.info("worker pool shutting down")
                return

Note that we have used the pathos fork to create the process pool. You can follow this discussion
to understand the reasoning behind it. If we were to use the native multiprocessing, we had to define the worker routine inside the module instead of getting it as an argument.

Next, we have to specify our worker routine. Therefore, we define a function that takes the Kafka configuration object, populates a template function object with the configuration, and returns the resultant function, which could be used as the routine for our WorkerPool. The config for our setup includes Kafka's broker endpoint, the topic where events will be sent, the consumer group's id, and a response topic to which consumers will write process results. We use pydantic to get this information from environment variables:

from pydantic import BaseSettings

class KafkaConfig(BaseSettings):
    kafka_bootstrap_server: str
    kafka_request_topic: str
    kafka_ack_topic: str
    kafka_consumer_group_id: str

    class Config:
        env_prefix = ''
        case_sensitive = False
        allow_mutation = False

In addition, the Protocol Buffers to enact the communication is as follows:

syntax = "proto3";
package communication;

message Request {
    int32 sequence = 1;
    string payload = 2;
}

message Response {
    int32 sequence = 1;
    bool isprocessed = 2;
}

We use the confluent_kafka library, a prominent Python client to connect to Kafka's broker. The worker routine creates a consumer and a producer to send acknowledgments. It loops to poll events from Kafka. When one is received, a service thread is spawned and given the event for processing. It then constantly sends heartbeats to the Kafka broker while waiting for service results. After receiving the result from the service thread, it resumes getting new messages. The handler function that returns the worker routine function seems like this:

def KafkaHandler(config: config.KafkaConfig) -> Callable:


    # the worker pool provides a worker_number to each worker 
    # for logging purposes. 
    def routine(worker_number, *args, **kwargs):

        # callback to be called when worker thread has done its work
        def callback(future):
            logger.info(f"{future.result()}")

        try:

            # instantiate a logger
            logger = Logger(f"kafka_handler[{worker_number}]").getLogger()

            # each worker is supposed to act as a unique consumer 
            # in the consumer group with one or more topic-partitions 
            # assigned to it 
            c = Consumer({
                    'bootstrap.servers': config.kafka_bootstrap_server,
                    'group.id': config.kafka_consumer_group_id,
                    'auto.offset.reset': 'earliest',
                    'enable.auto.commit': False,
                })

            # each worker will send acknowledgement events to producers 
            # to inform them of the process result. 
            p = Producer({
                'bootstrap.servers': config.kafka_bootstrap_server,
            })

            logger.info(f"starting kafka handler")

            # subscribe to the requst topic to receive events
            c.subscribe([config.kafka_request_topic])

            # iterate indefinitely to poll new events and send heartbeats
            # if the worker is through a process the polling must pause 
            # message reception and merely send heartbeats to the broker. 
            while True:

                msg = c.poll(2.0)

                if msg is not None: 

                    if msg.error():
                        logger.error(f"consumer error: {msg.error()}")
                        continue

                    # generate a thread for every event and get a future
                    # to check. 
                    executor = ThreadPoolExecutor(max_workers= 1)

                    # service is a Callabe containing the business logic 
                    # to process events. service will receive the event itself,
                    # the consumer, the producer, config, and every argument that 
                    # is passed to the worker routine at execution time. 
                    future = executor.submit(
                        service, 
                        msg=msg, 
                        consumer=c,
                        producer=p,
                        config=config,
                        logger= logger,
                        *args,
                        **kwargs
                        )

                    future.add_done_callback(callback)


        # graceful shutdown
        finally:
            p.flush()
            c.close()  

    return routine

We define service in KafkaHandler in the same module. The service receives the event alongside the consumer and producer. First, it pauses event reception at the worker, then it proceeds to parse the incoming protobuf message and process it. Upon successful completion of the process, it tries to send the result in the ack topic and waits for the Kafka broker to confirm that the ack has been registered at the broker. If the ack is sent without error, the service thread commits to the request topic,
otherwise, it just terminates the service. The service function is as follows:

def service(msg, 
            consumer: Consumer, 
            producer: Producer, 
            config: config.KafkaConfig, 
            logger: Logger,
            *args, 
            **kwargs):

    # when worker is done processing the event, it has to send 
    # an acknowledgement event to another topic to inform producers
    # of the process result. The worker will commit the event only when
    # it is assured that this ack has been set in Kafka, otherwise the event
    # is not considered complete and worker won't commit. 
    # this is a callback function for the producer that sends the ack
    def acked(error, message):

        if error is None:

            # deserialize ack event
            parsedData = communication_pb2.Response()
            parsedData.ParseFromString(message.value())


            consumer.commit(asynchronous=False)
            logger.info(f"ack sent: {parsedData.sequence}")
        else:
            logger.error(f"error sending ack: {error}, will not commit")

    # main thread can pass arbitrary arguments to workers at
    # execution time to be used in workers. Note that these 
    # arguments are not worker specfic and each worker receives
    # them
    ### process defined arguments
    dummy = kwargs["dummy"]

    try:

        # pause event consumption until the process is complete
        consumer.pause(consumer.assignment())

        # deserialize incoming event 
        parsedData = communication_pb2.Request()
        parsedData.ParseFromString(msg.value())

        ### process data here 


        ### send the ack result to the kafka
        ack = communication_pb2.Response()
        ack.sequence = parsedData.sequence
        ack.isprocessed = True

        # each worker is also a producer that sends ack events 
        # to a response topic. Producers consume these events
        # to get a sense of their events. 
        producer.produce(
            config.kafka_ack_topic, 
            key=msg.key(), 
            value=ack.SerializeToString(),
            callback=acked
            )

        # wait a certain amount of time for the producer
        # to send the ack. if the operation fails somehow 
        # worker won't commit and event will be processed again 
        producer.poll(1)
        result = f"event has been processed: {parsedData.sequence}"

    except Exception as ex:
        result = f"error processing the event: {ex}"

    # resume reception of events
    finally:
        consumer.resume(consumer.assignment())

    return result

Now we are ready to fire up our consumer in the main module:

from controller.kafka.kafka_handler import KafkaHandler

from tools.pool.worker_pool import WorkerPool
from tools.kafka.config import KafkaConfig



def runner():

    config = KafkaConfig()

    handler = KafkaHandler(config)

    pool = WorkerPool(handler)

    # every argument for the service functin needs to be passed here 
    pool.execute(dummy="dummy")

if __name__ == "__main__":

    runner()

Thanks for reading.

First-In First-Out ↩

Recursive Matrix Determinant Calculation Using Go

Reza Alipour — Thu, 15 Dec 2022 08:16:17 +0000

Some time ago while I was trying to learn the Go programming language,
I thought of documenting some of the funnier aspects of my journey. Now here I am, with my very first
post on Medium and it’s about using Go’s packaging mentality to implement a recursive method to calculate
Square Matrices’ determinant.

TL;DR: The project discussed in this post is accessible via this GitHub repository.

Determinants are only defined over square matrices and can be derived under a recursive algorithm, like the one discussed in this lecture, which we will employ here.

We are not going to walk through the details of this algorithm here and focus only on the implementation itself. However, being familiar with it is a prerequisite to what we discuss here.

Before going after a matrix’s determinant, first, we need to have a notion of a matrix itself.

Go is a strictly-typed language, and we can exploit this trait to modularize our code. We intend to decouple the implementation of the matrix concept from the main package.

Therefore, we need to initialize a Go module. initialize a new module with the following command in the project root directory:

go mod init "example/det"

We define a new type named Matrix and put it inside a matrix package alongside our main package, under the following structure:

├── main.go
├── matrix
│   ├── matrix.go
├── go.mod

Our matrix would be a two-dimensional slice of float numbers. The matrix.go would look like as follows:

package matrix

type Matrix [][]float64

Next, we are going to define some methods for this type. Go permits the definition of operations over a type, a pattern that resembles OOP programming languages. Since a determinant is only defined for square matrices, we have to have a way to check if a matrix is square. Let’s define functions to get the number of rows and columns of a Matrix instance:

func (m Matrix)Rows() int {
  return len(m)
}

// we assume the Matrix have the same number of elements in each row and
// is not empy, which is true for all matrices
func (m Matrix) Columns() int {
 return len(m[0])
}

func (m Matrix) IsSquare() bool {
 return m.Columns() == m.Rows()
}

Besides, our definition of the matrix at this point doesn’t enforce an Matrix instance to actually be a matrix. One can define a Matrix instance in a way that different rows have a different number of elements; a structure that is definitely not a matrix! We will discuss a solution to address this problem
later, but for now, we would need a method to confirm a Matrix is indeed a matrix! To check for a Matrix instance to be a matrix is to check for it not to be empty and have the same number of elements in each row:

func (m Matrix) IsMatrix() bool {

 if m.Rows() == 0 {
  return false
 }

 for _, row := range m {

  if len(m[0]) != len(row) {
   return false
  }

 }

 return true
}

Now we are ready to start writing a method for determinant calculation. First, we check whether the Matrix instance is a valid square matrix. Subsequently, if a matrix is 2×2 in size, the determinant is simply calculated using the following expression:

det(\begin{bmatrix} a & b \\ c & d \end{bmatrix}) = ac - bd

Our method at this point would be:

func (m Matrix) Det() (float64, error) {

 if !m.IsMatrix() || !m.IsSquare() {
  return -1, 
  fmt.Errorf(
    "determinant is not defined for the input [Matrix: %t][Square: %t]",
    m.IsMatrix(), m.IsSquare())
 }

 if m.Rows() == 2 {
   return m[0][0]*m[1][1] - m[0][1]*m[1][0], nil
 }

  //TODO: calculate the determinant otherwise
  return 0, nil
}

For bigger matrices, determinants could be calculated recursively. For example, the determinant of a 3×3 matrix is derived through the following expression:

det(\begin{bmatrix} a & b & c \\ d & e & f \\ g & h & i \end{bmatrix}) = a \times det(\begin{bmatrix} e & f \\ h & i \end{bmatrix}) - b \times det(\begin{bmatrix} d & f \\ g & i \end{bmatrix}) + c \times det(\begin{bmatrix} d & e \\ g & h \end{bmatrix}) \\ \\ A = \begin{bmatrix} e & f \\ h & i \end{bmatrix} \quad B = \begin{bmatrix} d & f \\ g & i \end{bmatrix} \quad C = \begin{bmatrix} d & e \\ g & h \end{bmatrix}

As you see, to implement this algorithm, we need to compute the submatrices A, B, and C. Upon observing these submatrices you realize they have been created by removing a particular row and a column from the original matrix. Such an iterative approach can be generalized to n×n matrices, where determinant expression at each iteration breaks into the determinants of a set of submatrices. Since the algorithm works iteratively, it's adequate to remove one row and one column at each iteration. The following function will remove a column from the matrix and return the consequent matrix.

func InBetween(i, min, max int) bool {
 if (i >= min) && (i <= max) {
  return true
 } else {
  return false
 }
}

func (m Matrix) ExcludeColumn(col_index int) (Matrix, error) {

 if !InBetween(col_index, 1, m.Columns()) {
  return Matrix{}, fmt.Errorf("input not in range")
 }

 result := make(Matrix, m.Rows())
 for i, row := range m {
  for j, el := range row {
   if j == col_index-1 {
    continue
   }
   result[i] = append(result[i], el)
  }
 }
 return result, nil
}

Note the function InBetween that is used to ensure the removing column's index lies inside the valid range. We take each row of the input matrix and copy the elements into another matrix except the ones in col_index column. Excluding a row would be implemented with the same mentality:

func (m Matrix) ExcludeRow(row_index int) (Matrix, error) {
 if !InBetween(row_index, 1, m.Rows()) {
  return Matrix{}, fmt.Errorf("input not in range")
 }

 var result Matrix
 for i, r := range m {
  if i == row_index-1 {
   continue
  }
  result = append(result, r)
 }
 return result, nil
}

Again, we take each row of the input matrix and append the whole row to the new matrix, except the row_index row. Now we are ready to complete our Det function.

func (m Matrix) Det() (float64, error) {

 if !m.isMatrix() || !m.isSquare() {
  return -1, fmt.Errorf("determinant is not defined for the input [Matrix: %t][Square: %t]",
   m.isMatrix(), m.isSquare())
 }

 if m.Rows() == 2 {
  return m[0][0]*m[1][1] - m[0][1]*m[1][0], nil
 }

 // if rows are more than 2
 // exclude the first row accoring to the algorithm
 partial_matrix, err := m.ExcludeRow(1)
 if err != nil {
  return -1, err
 }

 var temp float64 = 0

 // iterate over the elements of the first row
 for i, el := range m[0] {

  // get the corresponding submatrix
  reduced_matrix, err := partial_matrix.ExcludeColumn(i + 1)
  if err != nil {
   return -1, err
  }

  // get the dterminant of the submatrix, recursively
  partial_det, err := reduced_matrix.Det()
  if err != nil {
   return -1, err
  }

  // determinant would be the sum of the determinant of 
  //the sumatrices multiplied by the right coefficients
  temp = temp + partial_det*el*math.Pow(-1, float64(i))
 }

 return temp, nil
}

Note the use of standard math package to compute the sign of each multiplication coefficient in our algorithm. We can use our newly curated package to compute the determinant of a matrix in the main package:

package main

import (
 "example/det/matrix"
 "fmt"
 "log"
)

func main() {

// sample matrix
 matrix := matrix.Matrix{
  {3, 6, 2, 4},
  {7, 1, 5, 3},
  {9, 9, 1, 2},
  {4, 6, 3, 2}}

 det, err := matrix.Det()
 if err != nil {
  log.Fatalf("Error in calculating the determinant: %v", err)
 }

 fmt.Printf("The determinant is: %f", det)

}

The output would be the true determinant of the sample matrix:

The determinant is: -543.000000

Matrix Validation

Our initial implementation allowed the initialization of Matrix instances that were not a matrix mathematically. This is an implementation concern rather than a mathematical one. A typical solution would be to alter Matrix into an unexported type that will only be accessible through an instantiation fiction that you implement and will have adequate validations in place. I am not going into details of the implementation, but the structure would be something like this:

package matrix 

import "fmt"

type matrix [][]float64

func GenerateNewMarix(input [][]float64) (matrix, error){

  // do some validations on input argument and see if it passes the checks
  passed := true

  if !passed {
    return matrix{}, fmt.Errorf("error, not passed") 
  }


  return matrix(input), nil
}

Thank for reading.