Home

Awesome

Docker Repository on Quay Build Status Go Report Card codecov

Konsumerator

Konsumerator is a Kubernetes operator intended to automate management and resource allocations for kafka consumers.

Branchdescription
master (v1)Current stable version (only bugfix)
v2Semi-stable v2 branch (new features go there)

Operator creates and manages Consumer CRD, for this it requires cluster-wide permissions.

apiVersion: konsumerator.lwolf.org/v1
kind: Consumer
metadata:
  name: consumer-sample
spec:
  numPartitions: 100
  numPartitionsPerInstance: 1
  name: "test-consumer"
  namespace: "default"
  autoscaler:
    # only one provider is supported yet - prometheus.
    # `prometheus` is configured using prometheus provider and user specific metrics
    mode: "prometheus"
    prometheus:
      # minimum allowed period to query prometheus for the lag
      # information to avoid DDoS of that service
      minSyncPeriod: "1m"
      # do not scale up if the lag is less than 5 minutes
      tolerableLag: "5m"
      # approximate consumption rate per CPU
      ratePerCore: 20000
      # approximate memory requirements per CPU
      # if ratePerCore is 10k ops, this value is amount of
      # RAM needed during the processing of this 10k ops
      ramPerCore: "100M"
      # criticalLag is some value close to the SLO.
      # if lag has reached this point, autoscaler will
      # give maximum allowed resource to that deployment
      criticalLag: "60m"
      # preferable recovery time. During lag, Consumer will try to allocate 
      # as much resources as possible to recover from lag during this period
      recoveryTime: "30m"
      # prometheus addresses to query
      address:
        - "http://prometheus-operator-prometheus.monitoring.svc.cluster.local:9090"
      # Offset query should return number of messages that is not
      # processed yet a.k.a lag per partitionLabel
      offset:
        query: "sum(rate(kafka_messages_last_offset{topic=''}[5m])) by (partition)"
        partitionLabel: "partition"
      # Production query should return number of messages is being
      # produced per partitionLabel per unit of time (second)
      production:
        query: "sum(rate(kafka_messages_produced_total{topic=''}[5m])) by (partition)"
        partitionLabel: "partition"
      # Consumption query should return number of messages is being
      # consumed per partitionLabel per unit of time (second)
      consumption:
        query: "sum(rate(kafka_messages_consumed_total{topic=''}[5m])) by (partition)"
        partitionLabel: "partition"
  # partitionEnvKey - the name of the environment variable
  # containing partition number the deployment is responsible for
  partitionEnvKey: "PARTITION"
  # DeploymentSpec to run the consumer
  deploymentTemplate:
    replicas: 1
    strategy:
      type: Recreate
    selector:
      matchLabels:
        app: my-dep
    template:
      metadata:
        labels:
          app: my-dep
      spec:
        containers:
          - image: busybox
            name: busybox-sidecar
            command: ["/bin/sh", "-ec", "env && sleep 3000"]
          - image: busybox
            name: busybox
            command: ["/bin/sh", "-ec", "sleep 2000"]
  # resource boundaries, this policy protects
  # the consumer from scaling to 0 or infinity in case
  # of incidents
  resourcePolicy:
    containerPolicies:
    - containerName: busybox
      minAllowed:
        cpu: "100m"
        memory: "100M"
      maxAllowed:
        cpu: "1"
        memory: "1G"
    - containerName: busybox-sidecar
      minAllowed:
        cpu: "100m"
        memory: "100M"
      maxAllowed:
        cpu: "100m"
        memory: "100M"

When such Consumer are being created, operator will create .spec.numPartitions unique deployments.

$ kubectl get consumers

  NAME              EXPECTED   RUNNING   PAUSED   MISSING   LAGGING   OUTDATED   AUTOSCALER   AGE
  consumer-sample   100        100       0        0         6         0          prometheus   6d2h


Deployment specifics:

Metrics Providers

At the moment only one metrics provider is implemented - Prometheus.

Here is a critical settings for the provider:

spec:
  ...
  autoscaler:
    prometheus:
      # minimum allowed period to query prometheus for the lag
      # information to avoid DDoS of that service
      minSyncPeriod: "1m"
      address:
        - "http://prometheus-operator-prometheus.monitoring.svc.cluster.local:9090"
      # Offset query should return number of messages that is not
      # processed yet a.k.a lag per partitionLabel
      offset:
        query: "sum(rate(kafka_messages_last_offset{topic=''}[5m])) by (partition)"
        partitionLabel: "partition"
      # Production query should return number of messages is being
      # produced per partitionLabel per unit of time (second)
      production:
        query: "sum(rate(kafka_messages_produced_total{topic=''}[5m])) by (partition)"
        partitionLabel: "partition"
      # Consumption query should return number of messages is being
      # consumed per partitionLabel per unit of time (second)
      consumption:
        query: "sum(rate(kafka_messages_consumed_total{topic=''}[5m])) by (partition)"
        partitionLabel: "partition"

Resource Predictors

At the moment only one resource predictor is implemented which is tightly coupled with Prometheus metrics provider. NaivePredictor operates using following settings provided by a user in .spec.autoscaler.prometheus

Guest Mode (no cluster wide permissions)

Sometimes you don't have permissions to create CRDs in the cluster, for such cases Konsumerator supports a so-called guest-mode. When running in guest-mode operator uses configmaps instead of CRDs and it is limited to a single namespace. Guest-mode could be activated by setting namespace argument konsumerator --namespace=default.

To create an instance of the consumer, you need to create a ConfigMap with konsumerator.lwolf.org/managed annotation and consumerSpec inside the body.

apiVersion: v1
kind: ConfigMap
metadata:
  name: consumer-sample
  namespace: default
  annotations:
    konsumerator.lwolf.org/managed: "true"
data:
  consumer.yaml: |
    numPartitions: 100
    name: "test-consumer"
    namespace: "default"
    autoscaler:
      ...

Installation

coming soon...

Operations

How to stop/start consumer

At the moment, the only way to stop the consumer is to set .spec.deploymentTemplate.replicas to 0. This will trigger reconciliation, operator will notice that the deployment spec was changed and apply the change, in this case change the number of replicas to 0.

How to pause auto scaling

It is possible to temporary disable autoscaling of the managed deployments. To do so, add the following annotation to the consumer object with any value:

annotations:
  konsumerator.lwolf.org/disable-autoscaler: "true"

In case of disabled autoscaling, operator will allocate minimum resources from the resourcePolicy for this container. If there is no such policy set, no resource request/limit will be set.

Development

Requirements

Dev k8s cluster

To spin up dev k8s cluster run the following:

make kind-create

This will create 2 node cluster (1 master and 1 node) with pre-installed Prometheus and Grafana.

Access cluster resources (Linux):

KIND is configured to expose 2 ports. To get IP address of the KIND worker node, run:

kubectl get nodes konsumerator-worker -o jsonpath='{ $.status.addresses[?(@.type=="InternalIP")].address }'

then, you can access grafana and prometheus on the following ports of that IP address:

make build

make test

make install

make run