DEV Community: Jane Radetska

Setup Knative Eventing with Kafka from scratch, scale based on events volume, and monitor

Jane Radetska — Thu, 04 Jan 2024 17:46:10 +0000

I am going to describe how to create new Kubernetes cluster and install Knative eventing, Kafka flavor, in it. I am actually going to create two Kafka clusters with mirroring enabled, to be able to perform some experiments later on.

I am also going to describe steps one can follow to ensure Knative scales well enough when messages volume increases. And I am going to point to the resources on how to install monitoring for such cluster.

Kubernetes cluster with Knative eventing should fit in Google Cloud trial quotas, but monitoring and scaling workload on top of that might not.

Cluster creation

Create new Kubernetes cluster, one zone, 4-6 nodes, node is Standard compute-optimized (c2-standard-4 at least), 100 Gb disk (best if pd-ssd, but can be pd-standard or pd-balanced). Trial quota is 4 nodes c2-standard-4.

Installing Kafka and Knative

Create namespace knative-eventing.

Follow Strimzi quickstart to install kafka in knative-eventing namespace, but use different Kafka cluster definition, see below. Knative workloads are expecting to be run in knative-eventing namespace, otherwise issues arise. And it's easier to keep Knative and Kafka in one namespace.
Use kafka-cluster.yaml as kafka cluster resource instead of the one used in Strimzi quickstart (kafka-single-persistent.yaml). If you're not limited on disk, best to set storage: size: 50Gi or 100Gb in kafka-cluster yaml, and at least 25Gb for zookeeper storage. For trial quota, you're limited to 20Gb and 10Gb for zookeeper (if we're doing 2 Kafka clusters, if one - can be more).

Follow knative docs to install Knative eventing. Install all Kafka components too: Kafka sink, Kafka broker, Kafka event source. Use this publication to configure broker config to be Kafka broker class (replication: 1).

Also make sure to install Kafka source. kafka-source-dispatcher will have 0 pods until some Kafka sources are created.

Autoscaling Knative

For trial quota GCP, you'll likely won't have space for Keda controller or upscaled Knative workloads. Otherwise,

Follow this blog to configure HA for Knative workloads. I would set HA to 6 though, and keep an eye on memory/CPU consumption of the workloads in case you're got significant events traffic going through the system. Otherwise there's going to be slowdown in events delivery.

Install scaling controller for Kafka sources - Keda autoscaler. HPA parameters are controlled by annotations on the Kafka source yaml definition:

metadata:
  annotations:
    autoscaling.knative.dev/class: keda.autoscaling.knative.dev
    autoscaling.knative.dev/minScale: "0"
    autoscaling.knative.dev/maxScale: "5"
    keda.autoscaling.knative.dev/pollingInterval: "30"
    keda.autoscaling.knative.dev/cooldownPeriod: "30"
    keda.autoscaling.knative.dev/kafkaLagThreshold: "10"

Kafka of course has it's own parallelism mechanism - creating more brokers, which enables higher partitions amount for a given topic.

Monitoring Knative and Kafka

Follow this publication to setup Prometeus monitoring for Kafka cluster. DataDog has a nice description of what those metrics mean.

Knative has a tutorial on how to setup monitoring. However I ended up creating Service and ServiceMonitor by hand for Knative workloads to be able to monitor them.

Here's example Service and ServiceMon for kafka-sink-receiver:

apiVersion: v1
kind: Service
metadata:
  name: knative-sink-service
  labels:
    app: knative-sink-service
spec:
  selector:
    app: kafka-sink-receiver
  ports:
    - name: http-metrics
      protocol: TCP
      port: 9090
      target-port: http-metrics
---
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: knative-sink-service-monitor
  labels:
    app: knative-sink-service-mon
spec:
  selector:
    matchLabels:
      app: knative-sink-service
  endpoints:
  - port: http-metrics

Knative exposes a couple of it's own metrics (like processing delays) and also exposes a huge amount of Kafka metrics for it's consumers/producers. I ended up curl-ing Knative Services on the metrics port, and scripting a tool that would help to create primitive Grafana dashboard for the list of metric names and uid of datasource. See readme on how to use the tool. Or can replace datasource uid in the dashboard-*.json with your datasource uid, and make sure job selectors in the dashboard JSON match the service name that sends metrics.

Knative dashboards together with Kafka's dashboards it sheds light on almost any aspect of what's going on in the system.

More tuning

Some useful production-grade considerations for Knative could be found here

Knative exposes consumer and producer configs for brokers and other workloads as configmap. I had more luck with setting

auto.offset.reset=latest
enable.auto.commit=true
commit interval to be about 1.5 seconds, heartbeat interval/2

for Knative sink-receiver config.

More on Kafka consumer and producer tuning

https://strimzi.io/blog/2021/01/07/consumer-tuning/
https://strimzi.io/blog/2020/10/15/producer-tuning/

Make sure it works

You can create a Kafka topic which messages are transferred to another topic using Knative machinery:

input-topic -> knative source -> knative broker -> knative trigger (opt: filter by message headers) -> knative sink -> output-topic

Example definitions to use are below. Apply topics and broker, make sure they've got status Ready (kubectl get kafkatopic -n knative-eventing, kubectl get broker -n knative-eventing). Then apply sink and source, also make sure they're ready. Last apply trigger.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: input-topic
  namespace: knative-eventing
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824
---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: output-topic
  namespace: knative-eventing
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824
---
apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
  name: my-broker
  namespace: knative-eventing
  annotations:
    eventing.knative.dev/broker.class: Kafka
spec: {}
---
apiVersion: sources.knative.dev/v1beta1
kind: KafkaSource
metadata:
  name: input-topic-source
  namespace: knative-eventing
# keda autoscaler annotations here if using keda
# see Autoscaling section of blog, above
spec:
  consumerGroup: input-topic-source-group
  bootstrapServers:
  - my-cluster-kafka-bootstrap.knative-eventing:9092
  topics:
  - input-topic
  sink:
    ref:
      apiVersion: eventing.knative.dev/v1
      kind: Broker
      name: my-broker
---
apiVersion: eventing.knative.dev/v1alpha1
kind: KafkaSink
metadata:
  name: output-topic-sink
  namespace: knative-eventing
spec:
  topic: output-topic
  bootstrapServers:
   - my-cluster-kafka-bootstrap.knative-eventing:9092
---
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: output-trigger
  namespace: knative-eventing
spec:
  broker: my-broker
  # can define a filter for messages based on header, input Kafka headers get `kafkaheader` prefix. So if message was sent on `input-topic` with header `Ce-my-header: my-value`, it's filter here will be `kafkaheadercemyheader: my-value`
  # filter:
  #  attributes:
  #    kafkaheadercemyheader: my-value
  subscriber:
    ref:
      apiVersion: eventing.knative.dev/v1alpha1
      kind: KafkaSink
      name: output-topic-sink

Here's primitive Python web app that simply logs message upon arrival. Can use echo app as destination sink instead of second topic. Deployment for web app echo should be in namespace knative-eventing, and expose ClusterIP type Service that maps port 80 map to 8083. If you're not familiar with how to create deployment and service for it, use k8s docs or use Google Console "new deployment button" (gotta upload image to dockerhub or another artifact registry first though).

Let's send some messages.

Launch listener for output-topic:

kubectl -n knative-eventing run kafka-consumer -ti --image=quay.io/strimzi/kafka:0.37.0-kafka-3.5.1 --rm=true --restart=Never -- bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic output-topic --from-beginning --property print.headers=true

In other tab, launch client for input-topic:

kubectl -n knative-eventing run kafka-producer -ti --image=quay.io/strimzi/kafka:0.37.0-kafka-3.5.1 --rm=true --restart=Never -- bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic input-topic --property parse.headers=true  --property headers.delimiter=\t --property headers.separator=, --property headers.key.separator=:

And post following payload to input-topic:

Ce-my-header:my-value\t{"msg":"content"}

The same message should arrive to output-topic, with original headers having kafkaheader prefix:

ce_specversion:1.0,ce_id:...,ce_source:...,content-type:application/json; charset=utf-8,kafkaheadercemyheader:my-value {"msg":"content"}

Performance testing Strimzi Kafka in the k8s cluster using xk6-kafka

Jane Radetska — Mon, 01 Jan 2024 20:25:53 +0000

I'm going to describe how to performance test reading/writing from Kafka topic with multiple partitions using xk6-kafka plugin for k6. All resources mentioned here are available in the repo.

Topic to test

Here's topic definition, using Strimzi Kafka:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: my-topic
  namespace: kafka
  labels:
    strimzi.io/cluster: cluster-1
spec:
  partitions: 3
  replicas: 1
  config:
   ...

Test scenario

This topic has three partitions so it makes sense to test it with three virtual users, each reading from a separate partition.
Test scenario is going to execute like this:

Virtual user 1: create producer and consumer, produce 1000 messages to partitions 0 1 2, read 333 messages from partition 0, teardown producer and consumer
Virtual user 2: create producer and consumer, produce 1000 messages to partitions 0 1 2, read 333 messages from partition 1, teardown producer and consumer
Virtual user 3: create producer and consumer, produce 1000 messages to partitions 0 1 2, read 333 messages from partition 2, teardown producer and consumer

Here's the code for the scenario script. It has debug prints that can be helpful to inspect how messages are consumed by virtual users. For more logs, set environment variable LOG_LEVEL=debug, and pass param connectLogger: true to Writer and Reader constructor.

An important aspect is that it is important to set groupID, groupTopics and groupBalancers when using Kafka bootstrap server. ReaderConfig has param topic which doesn't quite work with bootstrap server, it works with Kafka broker's address directly and with explicit partition number set.

Another important aspect is that consumer is instantiated in test code (default function) - meaning each virtual user will use it's own consumer object. All consumers should belong to same consumer group though (groupID param). It is important to close consumer at the end of the function.

import {
    Writer,
    Reader,
    SCHEMA_TYPE_STRING,
    SchemaRegistry,
    GROUP_BALANCER_ROUND_ROBIN,
    SECONDS
} from "k6/x/kafka";
import { check } from "k6";

const bootstrapServers = [
  'localhost:9091',
];

export const options = {
    vus: 3,
    duration: "3h",
    thresholds: {
        kafka_writer_error_count: ["count == 0"],
        kafka_reader_error_count: ["count == 0"],
    },
};

const topicName = "topic1";
const schemaRegistry = new SchemaRegistry();

export default function () {

  let messageAmount = 9;
  let batchSize = 10;

    const producer = new Writer({
      brokers: bootstrapServers,
      topic: topicName,,
      balancer: "balancer_roundrobin",
      requiredAcks: 1,
      batchSize: batchSize,
      maxAttempts: 3,
      connectLogger: true,
    });

    console.log('VU 1, writing messages. Iter ' + __ITER);
    let firstMessageContent = null;
    let lastMessageContent = null;
    for (let index = 0; index < messageAmount; index++) { 
        let msgContent = "test-value-string-" + index + "-vu-" + __VU + "-iter-" + __ITER;
        if (index == 0) {
          firstMessageContent = msgContent;
        }
        if (index == messageAmount - 1) {
          lastMessageContent = msgContent;
        }
        let messages = [];
        for (let i = 0; i < batchSize; i++) {
          messages.push({
            value: schemaRegistry.serialize({
              data: msgContent,
              schemaType: SCHEMA_TYPE_STRING,
            }),
          });
        }
        producer.produce({ messages: messages });

    }
    producer.close();
    console.log("First published msg: " + firstMessageContent);
    console.log("Last published msg: " + lastMessageContent);

    const consumer = new Reader({
      brokers: bootstrapServers,
      groupID: topicName + "-group",
      groupTopics: [topicName],
      groupBalancers: [GROUP_BALANCER_ROUND_ROBIN],
      maxAttempts: 3,
      connectLogger: true,
      commitInterval: 1.2 * SECONDS,
      heartbeatInterval: 3.5 * SECONDS,
    });

    let messages = consumer.consume({ limit: messageAmount * batchSize});

    console.log("Amount of msgs received: " + messages.length + ", VU " + __VU + ", iter " + __ITER);

    if (messages.length) {

      check(messages[0], {
        "Topic equals to": (msg) => msg["topic"] == topicName
      });

    } else {
      console.log("No messages received");
    }
    consumer.close();
}

Here's output this scenario produces. Offset for the first message of each consumer is not zero because topic had prior messages in it, and same consumer group has already read those. So offset is 33 to begin with.

          /\      |‾‾| /‾‾/   /‾‾/   
     /\  /  \     |  |/  /   /  /    
    /  \/    \    |     (   /   ‾‾\  
   /          \   |  |\  \ |  (‾)  | 
  / __________ \  |__| \__\ \_____/ .io

  execution: local
     script: /var/test-scenario/test-scenario.js
     output: -

  scenarios: (100.00%) 1 scenario, 3 max VUs, 10m30s max duration (incl. graceful stop):
           * default: 3 iterations shared among 3 VUs (maxDuration: 10m0s, gracefulStop: 30s)

time="2024-01-01T18:50:56Z" level=info msg="VU 1, writing messages. Iter 0" source=console

...

time="2024-01-01T18:51:01Z" level=info msg="Amount of msgs received: 333, VU 3, iter 0" source=console
time="2024-01-01T18:51:01Z" level=info msg="First msg value test-value-string-99-vu-1-iter-0, offset33, partition 0, VU 3, iter 0" source=console
time="2024-01-01T18:51:01Z" level=info msg="Last msg value test-value-string-993-vu-1-iter-0, offset365, partition 0, VU 3, iter 0" source=console

...

time="2024-01-01T18:51:01Z" level=info msg="Amount of msgs received: 333, VU 1, iter 0" source=console
time="2024-01-01T18:51:01Z" level=info msg="First msg value test-value-string-2-vu-1-iter-0, offset33, partition 2, VU 1, iter 0" source=console
time="2024-01-01T18:51:01Z" level=info msg="Last msg value test-value-string-998-vu-1-iter-0, offset365, partition 2, VU 1, iter 0" source=console
time="2024-01-01T18:51:01Z" level=info msg="Amount of msgs received: 333, VU 2, iter 0" source=console
time="2024-01-01T18:51:01Z" level=info msg="First msg value test-value-string-1-vu-1-iter-0, offset33, partition 1, VU 2, iter 0" source=console
time="2024-01-01T18:51:01Z" level=info msg="Last msg value test-value-string-997-vu-1-iter-0, offset365, partition 1, VU 2, iter 0" source=console

...


     ✓ all messages returned
     ✓ Topic equals to

     █ teardown

     checks.............................: 100.00%     ✓ 6             ✗ 0            
     ...  
     iterations.........................: 3           
     kafka_reader_dial_count............: 3           
     ... 
   ✓ kafka_reader_error_count...........: 0           0/s
     kafka_reader_fetch_bytes...........: 66 kB              
     kafka_reader_fetches_count.........: 6           
     kafka_reader_lag...................: 0           min=0           max=0          
     kafka_reader_message_bytes.........: 33 kB       
     kafka_reader_message_count.........: 1001        
     kafka_reader_offset................: 366         min=366         max=368        
     ...  
     kafka_reader_rebalance_count.......: 3           
     kafka_reader_timeouts_count........: 0           
     ...                  
     kafka_writer_batch_bytes...........: 56 kB       
     kafka_writer_batch_max.............: 1           min=1           max=1          
     ... 
     kafka_writer_batch_size............: 1000        
     ... 
   ✓ kafka_writer_error_count...........: 0           0/s
     kafka_writer_message_bytes.........: 56 kB       
     kafka_writer_message_count.........: 1000        
     ...      
     kafka_writer_write_count...........: 1000        
     ...
     vus................................: 3           min=3           max=3          
     vus_max............................: 3           min=3           max=3          


running, 0/3 VUs, 3 complete and 0 interrupted iterations
default ✓ [ 100% ] 3 VUs  3/3 shared iters

Test results to watch out for:

kafka_reader_error_count - should be zero or low
kafka_writer_error_count - should be zero or low
kafka_writer_message_count and kafka_reader_message_count should match

There could be intermittent issues and error counts might be not zero. Yet they shouldn't be higher than like 5 out of 1000, and of course depend on how do you set SLO for your system. Reader and Writer are instantiated with maxAttempts: 3 so they'll retry writing/reading.
If reader receives no messages this iteration, it won't fail any checks. It will just get those messages in the next test iteration. Main thing is to have total number match kafka_writer_message_count == kafka_reader_message_count.

Pod used to run test scenario, and command to run test in the k8s cluster

Here's pod definition that can be used to run the script in the k8s cluster:

apiVersion: v1
kind: Pod
metadata:
  creationTimestamp: null
  labels:
    app: test-xk6-loadtest
  name: test-xk6-loadtest-1
  namespace: loadtest
spec:
  containers:
  - args:
    - run
    - '/var/test-scenario/test-scenario.js'
    image: mostafamoradian/xk6-kafka:latest
    name: loadtest-xk6
    env:
    - name: LOG_LEVEL
      value: debug
    resources: {}
    volumeMounts:
    - mountPath: /var/test-scenario
      name: test-scenario
  dnsPolicy: ClusterFirst
  volumes:
    - name: test-scenario
      configMap:
        name: kx6-test-scenario
  restartPolicy: Never
status: {}

Here's commands to run the scenario in your k8s cluster:

kubectl create --namespace kafka topic.yaml <-- Strimzi definition of my-topic, see above
kubectl create --namespace loadtest configmap kx6-test-scenario --from-file=test-scenario.js <-- JS file with test scenario, see above
kubectl apply -f test-pod.yml  <-- Pod definition, see above

See test results using:

kubectl logs test-xk6-loadtest-1 -n loadtest -f

In case your kafka cluster has TLS or other auth options enabled, xk6-kafka repo has useful examples on how to setup those. Can mount server cert in the pod using volumes and volumeMounts.

Happy New Year!

KubeCon CloudNativeCon Europe 2023 homepage made usable

Jane Radetska — Fri, 21 Apr 2023 19:38:53 +0000

Especially for virtual attendees on laptop or workstation.

The page in question: https://kubecon-cloudnativecon-europe.com/home-full/

Homepage before:

Homepage after:

What I am unhappy about in the initial homepage

tini-tiny table with talks list
lots of useless elements that occupy screen space
scroll-in-a-scroll

Sched is not cutting it for me - there's no video of the talk in there, not even a link to videostream.
I just need a long list of talk videos... right on the homepage.

I found this page https://kubecon-cloudnativecon-europe.com/agenda/ too late. "Co-located Events + Sessions" just doesn't sound like what I am looking for. And it too has useless right column...

How to add JS to make homepage pretty

Option 1, with autoloading of the script that adjusts KubeCon homepage. Using Tampermonkey or similar Chrome plugin.

Chrome, 112.0 version

Install Tampermonkey plugin into Chrome
pin Tampermonkey plugin in browser header

Click Tampermonkey plugin icon in browser header, select "Create a new script..."

add fix-homepage.js file contents to the script body. Can examine the JS - nothing fancy in there, find element-set style.

Select tab "Settings" right above script edit area. Section "Includes/Excludes", "User matches" box, click "Add..." button - put https://kubecon-cloudnativecon-europe.com/home-full/ in the pop-up window

!! Click "Save" !!

Reload KubeCon homepage
It should autoload the script every time kubecon homepage loads

Option 2. Use content snippets. No plugins. No autoload.

Chrome browser

Load webpage https://kubecon-cloudnativecon-europe.com/home-full/ , right click on page content, click "Inspect"
In developer tools panel, select "Sources" tab, in that select "Snippets" tab
Click "+ New snippet"

Put contents of fix-homepage.js file into the snippet body window. Save snippet (Ctrl+S).
Run snippet - click on run button

You don't have to create snippet each time, but you have to run it each time

Manual (artistic) process of fixing the homepage

Delete the buttons on the right to the schedule

Who needs these buttons here?! They are also in left-side menu and that's where I would look for them

Make left column full-width

Remove the useless "Community in Bloom" header

Remove equally useless "Hello, Jane" header

Seriously? You just put "Hello user" there?! Did you make sysadmin write this website for you? :)

Find container with schedule and remove fixed height to get rid of, oh gosh, scroll-in-a-scroll

P.S. What about Mobile?

Are they hiding the schedule at all for mobile devices? LOL

Histogram of request time in Grafana with Telegraf

Jane Radetska — Fri, 18 Dec 2020 14:44:44 +0000

This is a writing about a cool tool useful for analyzing backend call time. Code that does backend calls and monitoring setup described in previous post.

Grafana panel can not only plot line graphs, but also:

show last reading of metric
show table of metric values
show bar plots
show heatmaps (histogram over time)

Heatmap is helpful for quickly getting understanding what is distribution of backend response time: it can be the case that most requests complete in under 50 msec, but some requests are slow and complete in >500 msec. Average request time doesn't show this information. In previous examples, we're plotting just the average.

We can easily add a heatmat for request execution time:

Need to add new panel, pick measurement details, and select "Heatmap" in "Visualization" collapsible in the right column.
Every 10 seconds, a new set of bricks appears on the panel. Brick color represents how much measurements fall into that bucket (e.g. 5 fall in the 10 msec - 20 msec range, hence that brick is pink). Set a fixed bucket size or fix the number of buckets, or let default values do their magic.

In case Telegraf sends all metrics data to InfluxDB, that's a real heatmap. Telegraf is often configured to send only aggregated values to database (min, avg, max) calculated over short period of time (10sec) in order to reduce metrics reporting traffic. Heatmap based on such aggregated value is not a real heatmap.

It is possible to configure histogram aggregate in Telegraf config (full Telegraf config with histogram aggregator):



[[aggregators.histogram]]
  period = "30s"
  drop_original = false
  reset = true
  cumulative = false

  [[aggregators.histogram.config]]
    buckets = [1.0, 10.0, 12.0, 14.0, 16.0, 18.0, 20.0, 30.0, 40.0]
    measurement_name = "aiohttp-request-exec-time"
    fields = ["value"]

I set reset=true and cumulative=false which will cause buckets values to be calculated anew for each 30 second period. Need to set value ranges (buckets) manually, as well as specify correct measurement_name. If fields is not specified, histogram buckets are computed for all fields of measurement. Here's how bucket values appear in InfluxDB:

The amount of request execution times that falls in a bucket is saved under "value_bucket" field name, "gt" ("greater than") and "le" ("less than or equals to") are bucket edge values that appear as tags.

Let's plot these values using "Bar gauge" panel visualization type:

Let's create 2 separate panels, one for python.org stats and one for mozilla.org (add 'where domain = python.org' in query edit).

Now we can at a glance compare last 30 sec request execution time distribution for python.org and for mozilla.org:

Monitoring sync and async network calls in Python using TIG stack

Jane Radetska — Fri, 18 Dec 2020 14:34:16 +0000

Republished by author. First appeared in Web Performance Calendar 2020.

Web applications and API endpoints are known to perform backend calls. Often that is all application does: fetches data from a couple of backends, combines it, and produces response.

Monitoring how much time fetching data took is essential. There are plenty production-ready buy-and-snap-on solutions that provide such monitoring, but they might be not good fit for some cases. And I think it's fun to dig deeper into things to get more understanding of how it all works.

Let's look at code examples that use popular Python networking libraries and are instrumented to report HTTP request execution time.

What I'm going to explore in this post

I'm going to compare how request timings look for fetching HTML pages using requests library and for asyncronously fetching same HTML pages using aiohttp library. I aim to visualize the difference in timings, and to introduce tools that can be used for such monitoring.

To be fair, requests library has plugins that enable asyncronous IO and there are so many other ways to achieve this in Python... I picked aiohttp as it provides neat request timing tracing opportunities, and I use this library a lot in the wild.

To monitor request timings we will use Telegraf, InfluxDB and Grafana stack. These tools are very easy to setup locally, open source, free for personal usage, and could be used in production environment.

Running code examples section describes in detail how to run example code and setup monitoring infrastructure (Telegraf, InfluxDB, Grafana).

All code from this writing is available in repo.

Example 0: monitor `requests` request time

Let's dive into first Python code example. Here's what it does:

in forever loop, executes two HTTP requests using requests Python library
reports request time and request exceptions to Telegraf

Here's request execution time plotted on the dashboard:

Full code of Example 0 can be found in example-0-requests-send-stats.py.

High-level execution flow can be followed from main part of the program:

if __name__ == '__main__':
    while True:
        result = call_python_and_mozilla_using_requests()
        print(result)
        time.sleep(3)

Inside call_python_and_mozilla_using_requests two simple HTTP requests are performed one by one, and their response text used to compose result:

def call_python_and_mozilla_using_requests():
    py_response = get_response_text('https://www.python.org/')
    moz_response = get_response_text('https://www.mozilla.org/en-US/')
    return (
        f'Py response piece: {py_response[:60].strip()}... ,\n'
        f'Moz response piece: {moz_response[:60].strip()}...'
    )

get_response_text function executes HTTP request for a given URL with primitive exception handling, and hooks to report request execution time:

def profile_request(start_time, response, *args, **kwargs):
    elapsed_time = round((
        time.perf_counter() - start_time
    ) * 1000)
    send_stats(
        'requests_request_exec_time',
        elapsed_time,
        {'domain': URL(response.url).raw_host}
    )


def get_response_text(url):
    try:
        request_complete_callback = partial(
            profile_request,
            time.perf_counter()
        )
        response = requests.get(
            url,
            hooks={'response': request_complete_callback}
        )
        response.raise_for_status()
        return response.content.decode()
    except RequestException as e:
        send_stats(
            'requests_request_exception',
            1,
            {'domain': URL(url).raw_host, 'exception_class': e.__class__.__name__}
        )
        return f'Exception occured: {e}'

This code uses requests library (docs). Basic usage to get text content from URL is as follows:

response = requests.get(url).content.decode()

requests.get accepts optional hooks argument, where function to be called after request is completed is specified - request_complete_callback.

This callback function may look funny if you're not familiar with functional programming. partial(profile_request, time.perf_counter()) is itself a function. It's same function as profile_request but the first argument is already filled in - time.perf_counter() was passed as start_time argument. This trick is used to supply correct start_time for each request, as request_complete_callback function is constructed anew for each request, while code for sending request execution time is isolated in another function profile_request. We can rewrite that as follows:

def get_response_text(url):
    try:
        start_time = time.perf_counter()

        def profile_request(response, *args, **kwargs):
            elapsed_time = round((time.perf_counter() - start_time) * 1000)
            send_stats('requests_request_exec_time', elapsed_time, ...)

        response = requests.get(url, hooks={'response': profile_request})

And it's going to work alright. Now there's a function defined inside a function, and get_response_text is bloated with profiling stuff, which is not something I like.

You can read more about partial functions and Python functools.

time.perf_counter() is used to measure execution time in Python (docs). time.perf_counter() returns microseconds that are converted to milliseconds using * 1000.

Sending stats

send_stats function is used to report measurements to Telegraf: metric name is 'requests_request_exec_time', metric value is time request execution took, tags include additional useful information (domain of URL).
get_response_text also invokes send_stats when exception occurs, passing different metric name this time - 'requests_request_exception'.

I have another post that describes ways to send stats from Python program to Telegraf.

In short, send_stats accepts metric name, metric value and tags dictionary. Those are converted to one string and sent to the socket on which Telegraf listens for measurement data. Telegraf sends received metrics to a database (InfluxDB). Grafana dashboard queries the database to put a dot on graph for each metric value reported.

`profile` decorator

A piece of code which is a decorator suitable for any function (async, sync, method of class or pure function) is adapted here to measure execution time of function that is decorated.
profile decorator is used to profile total execition time of functions call_python_and_mozilla_using_requests and call_python_and_mozilla_using_aiohttp (see the following examples).
Don't confuse with another useful tool - line_profiler - that also provides profile decorator.

`requests` execution time on dashboard

Let's run this example and set up all the monitoring tools. See Running code examples on how to run example code and set up monitoring infrastructure.

We can configure a panel that shows request execution time:

Blue dots of total execution time roughly correspond to the sum of time request to python.org and request to mozilla.org took (green and yellow dots). They measure at approximately 150 msec on average.

Need more exceptions

If we change 'www.python.org' to 'www.python1.org' in function call_python_and_mozilla_using_requests, exceptions appear in terminal output, and exception metrics are sent to Telegraf:

    Reported stats: aiohttp_request_exception=1, tags={'domain': 'www.python1.org', 'exception_class': 'ClientConnectorError'}
    'Py response piece: ...Exception occured: Cannot conn...

Configure a separate Grafana panel to see exceptions on the dashboard:

Exception class is sent as tag along with metric value. This gives us the ability to plot different lines for exceptions of different classes. To achieve this, pick 'group by - tag(exception_class)' when editing request exceptions panel.

Example 0 improved: reuse connection

Code of example 0 can be improved to reuse the same connection for all calls performed in that forever running while loop - here's an improved version.

The only significant code change is this:

...
session = requests.Session()
while True:
    result = call_python_and_mozilla_using_requests(session)
...

Connection creation is moved out of the while loop. Now, the connection is established once and for all.

Let's compare how much time request execution takes when a connection is reused:

The dots on the left are measurements for for original version of Example 0, and ones on the right came from the improved version. We can definitely notice how total execution time get lower, below 100 msec on average.

Example 1: monitor `aiohttp` request time

Let's dive into the next code example. Here's what it does:

in forever loop, executes two asyncronous HTTP requests using aiohttp
hooks into aiohttp request lifecycle signals
reports request time and request exceptions to Telegraf

Full code of Example 1 can be found in example-1-aiohttp-send-stats-basic.py.

High-level execution flow is similar to the Example 0, the way content is fetched from URLs differs.

The tale of two HTTP requests

Let's start with the function call_python_and_mozilla_using_aiohttp that executes two asyncronous HTTP requests and returns pieces of response content. It is the sister of call_python_and_mozilla_using_requests from Example 0:

async def get_response_text(url):
    try:
        async with ClientSession(trace_configs=[Profiler()]) as session:
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.text()
    except ClientError as e:
        return f'Exception occured: {e}'

@profile
async def call_python_and_mozilla_using_aiohttp():
        py_response, moz_response = await asyncio.gather(
            get_response_text('https://www.python.org/'),
            get_response_text('https://www.mozilla.org/en-US/')
        )
        return (
            f'Py response piece: {py_response[:60].strip()}... ,\n'
            f'Moz response piece: {moz_response[:60].strip()}...'
        )

Here, aiohttp library's ClientSession is used to execute the request (docs). Basic usage to get text content from URL is as follows:

async with ClientSession() as session:
    async with session.get(url) as response:
        return await response.text()

which is basically what happens in get_response_text. get_response_text also calls response.raise_for_status(), which raises exception when response status code is error code or timeout occurs . Exception is silenced in get_response_text, so get_response_text always returns str, either with response content or with exception message.

call_python_and_mozilla_using_aiohttp takes care of callings two URLs using asyncio.gather. Execution order for call_python_and_mozilla_using_aiohttp is on the right:

await asyncio.gather returns the result after both of these requests are complete. Total execution time is approximately the time of the longest request out of these two. You're probably aware that this is called non-blocking IO. Instead of blocking, this kind of IO operation frees execution thread until it needs it again.

Synchronous, blocking IO, like in Example 0, has different following execution order (see chart above, on the left). Total execution time is approximately the sum of both requests execution time. For positive integers, it's always true that A + B > MAX(A, B). Hence, asynchronous execution takes less time than synchronous one, provided unlimited CPU was available in both cases.

On the panel that shows requests execution time and their total execution time, it's possible to notice that total execution time call_python_and_mozilla_using_aiohttp_exec_time almost matches the longer-executing request time:

The total execution time for both requests is 75-100 msec.

Next, we're going to look at how execution time of each aiohttp request is reported.

`aiohttp` requests signals

aiohttp provides a way to execute a custom function when HTTP request execution progresses through lifecycle stages: before request is sent, when connection is established, after response chunk is received, etc. For that, object-tracer is passed to aiohttp.ClientSession - trace_configs:

class Profiler(TraceConfig):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.on_request_start.append(on_request_start)
        self.on_request_end.append(on_request_end)
        self.on_request_exception.append(on_request_exception)

...
async with ClientSession(trace_configs=[Profiler()]) as session:
...

Profiler is a subclass of aiohttp.TraceConfig. It "hooks up" functions that are going to be executed when request starts (on_request_start), when it ends (on_request_end) and when request exception is encountered (on_request_exception):

async def on_request_start(session, trace_config_ctx, params):
    trace_config_ctx.request_start = asyncio.get_event_loop().time()

async def on_request_end(session, trace_config_ctx, params):
    elapsed_time = round((
        asyncio.get_event_loop().time() - trace_config_ctx.request_start
    ) * 1000)
    send_stats(
        'aiohttp_request_exec_time',
        elapsed_time,
        {'domain': params.url.raw_host}
    )

async def on_request_exception(session, trace_config_ctx, params):
    send_stats(
        'aiohttp_request_exception',
        1,
        {'domain': params.url.raw_host, 'exception_class': params.exception.__class__.__name__}
    )

Notice how the timestamp is computed:

asyncio.get_event_loop().time()

It is recommended to use event loop’s internal monotonic clock to compute time delta in asyncronous code.

Function-hooks have arguments session, trace_config_ctx, params. Let's look at what they are.

session is an instance of aiohttp.ClientSession.

trace_config_ctx is context that is passed through callbacks. Custom values call be added to it when request is made:

await session.get(url, trace_request_ctx={'flag': 'red'})
...

async def on_request_end(session, trace_config_ctx, params):
    if trace_config_ctx.trace_request_ctx['flag'] == 'red':
        ....

This way function-hook can be programmed to behave differently for different request calls or to report additional data.

Request end hook uses trace_config_ctx.request_start value to compute total time request took. trace_config_ctx.request_start is set in request start hook.

params argument in on_request_end is aiohttp.TraceRequestEndParams and as such has url property. url property is of yarl.URL type. params.url.raw_host returns the domain of the URL that was requested. Domain is sent as a tag for metric, and this makes it possible to plot separate lines for different URLs.

Calling asyncronous code from synchronous

To call async function in sync execution context, special tooling is used, which is adapted from another publication. I'm not going to dive into Python's asyncronous ways in this post. Read more about Python's asyncio, it's pretty cool.

Compare results for Example 0 and 1

Connection is not reused for both cases here. Execution time for async version is lower, as expected.

Example 2: more, more stats

aiohttp provides hooks to measure more than just request execution time and request exceptions.

It's possible to report stats for:

DNS resolution time
DNS cache hit/miss
waiting for available connection time
connection establishing time
connection being reused
redirect happening
response content chunk received
request chunk sent

Impressive, isn't it? Documentation on tracing in aiohttp is here.

Let's add more request lifecycle hooks:

class Profiler(TraceConfig):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.on_request_start.append(on_request_start)
        self.on_request_end.append(on_request_end)
        self.on_request_redirect.append(on_request_redirect)
        self.on_request_exception.append(on_request_exception)
        self.on_connection_queued_start.append(on_connection_queued_start)
        self.on_connection_queued_end.append(on_connection_queued_end)
        self.on_connection_create_start.append(on_connection_create_start)
        self.on_connection_create_end.append(on_connection_create_end)
        self.on_dns_resolvehost_start.append(on_dns_resolvehost_start)
        self.on_dns_resolvehost_end.append(on_dns_resolvehost_end)
        self.on_response_chunk_received.append(on_response_chunk_received)
        self.on_connection_reuseconn.append(on_connection_reuseconn)
        self.on_dns_cache_hit.append(on_dns_cache_hit)
        self.on_dns_cache_miss.append(on_dns_cache_miss)

I won't bore you with code for each function like on_dns_resolvehost_end, it's quite similar to on_request_end. Full code of Example 2 is here.

Reported stats on dashboard for example 2:

We can see that DNS resolution takes couple of milliseconds and happens for every call, and the connection establishing takes 30-40 msec and happens for every call. Also, that DNS cache is not hit, DNS is resolved for every call.

We can definitely improve on that - in Example 3.

Example 3: `aiohttp` reuse session

Let's modify Example 2 code so that ClientSession is created once, outside while loop:

async def main_async():
    async with ClientSession(trace_configs=[Profiler()]) as session:
        while True:
            result = await call_python_and_mozilla_using_aiohttp(session)
            print(result)
            await asyncio.sleep(3)

And check out how stats look now:

There's only one dot for connection establishing, and one per DNS resoltion per domain. There's plenty of dots for connection reuse event.
Total execution time is below 50 msec. Cool.

Full source code of Example 3 is here.

Compare sync and async URL fetch, with and without reusing connection

Total time for both requests (very approximate):

	Connection not reused	Connection reused
Sync	150 msec	80 msec
Async	80 msec	40 msec

Reporting Measurements from Python Code in Real Time: a Beginner-Friendly Tutorial

Jane Radetska — Tue, 01 Dec 2020 14:55:26 +0000

Reporting measurements from Python code in real time

A simple example of how to send measurements from Python code to the real-time monitoring solution (Telegraf/InfluxDB/Grafana).

Code-reported measurements can be:

price of an order user just submitted
amount of free beds in the hospital
how long did a backend call take
percent of file that is already processed, and percent that's left
...
any number of which the program is aware and which might be useful to track

I don't think I need to make a lot of arguments in favor of real-time monitoring: it's a blessing in time of turmoil (outage). Data collected (good times data, outages data) can be analyzed later for various purposes: notice weird pattern in performance over time, notice significant features of traffic that can be leveraged, notice what happens right before outage, ... .

We will start with simple examples of Python programs that report measurements data. But first we need to configure things that are going to listen, record, and display these measurements.

Tutorial materials

All files mentioned are available in the repo CheViana/python-send-stats.

Looking for a quick, ready, robust solution?

Setup Grafana, InfluxDB, Telegraf and use Example 1 code snippet / Telegraf config.

Setup Grafana, InfluxDB, Telegraf

In short, install Grafana, InfluxDB, Telegraf:

Visit https://portal.influxdata.com/downloads/ for information on how to install InfluxDB and Telegraf
Visit https://grafana.com/grafana/download for information on how to install Grafana

Launch Grafana and InfluxDB with default configs:



> cd grafana-7.1.0
> bin/grafana-server

In other terminal tab:



> influxd -config /usr/local/etc/influxdb.conf

Example 1. The simplest example of how to send stats from Python code in 6 lines, and of suitable Telegraf config

First, we're going to make Telegraf listen on the Internet datagram socket for JSON-formatted measurements that Python code will send. Telegraf will write received measurements to database.

https://github.com/CheViana/python-send-stats/blob/master/telegraf-1-stats-simple-datagram-json.conf:



...

[[outputs.influxdb]]
  urls = ["http://127.0.0.1:8086"]
  database = "socket-stats"

[[inputs.socket_listener]]
  service_address = "udp://:8094"
  data_format = "json"
  json_name_key = "metric_name"

Launch Telegraf with this config:



> telegraf -config telegraf-1-stats-simple-datagram-json.conf

More info on telegraf plugin that enables listening for data on socket: socket_listener docs.

1-stats-simple-datagram-json.py is simple Python program that sends measurements to UDP socket. Measurements are sent in Telegraf JSON format every 2 seconds.

1-stats-simple-datagram-json.py:



import time
import socket
import json
import random


while True:
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.sendto(
            json.dumps({'metric_name': 'good_metric_name', 'value1': 10, 'value2': random.randint(1, 10)}).encode(),
            ('localhost', 8094)
        )
        print('Sending sample data...')
        sock.close()
    except socket.error as e:
        print(f'Got error: {e}')

    time.sleep(2)

Start the program that sends stats to socket:



> python3 1-stats-simple-datagram-json.py

This is a complete working example. A tiny piece of code that does what you want it to do - report measurements:



good_metric_name,value1=10,value2=7
good_metric_name,value1=10,value2=2
good_metric_name,value1=10,value2=5
...

In this example, measurement name is not tied to Telegraf config - Telegraf uses measurement name found under key 'metric_name' in JSON that is sent to it. More about this below.

Metric name gotchas

Metric name (also tag name, tag value, any string value reported) should not contain ':', '|', ',', '='. Better to use '-', '_' or '.' as delimiter in metric name. Special characters in reported string values could cause errors during measurement parsing in Telegraf or in InfluxDB, and these errors are easy to miss.

Grafana Dashboard

Add source for InfluxDB database "socket-stats".
Create new dashboard, add panel which will display measurements sent to Telegraf client.

Provided all 4 processes are running (Grafana, InfluxDB, Telegraf and Python program that sends stats), you should see measurements appear on dashboard in real time. Exciting, isn't it?

Example 2. JSON measurements over TCP socket (UNIX domain)

For UDP sockets there's no need to keep connection open, because of how protocol works. However, it might be not possible to use UDP sockets in some network setups, or it's possible but rate of dropped packets is too big: most measurement readings are lost.
Alternative is to use TCP sockets (also called Stream socket). For TCP sockets it's an overhead to open and close connection each time measurement is sent, which could be around 10 times per second. Opening and closing connections is a CPU-expensive operation.
TCP socket can be UNIX domain or INTERNET domain. UNIX domain are better suited for processes that run on same network host, but can't be used when communicating processes are running on different network hosts. Better suited because low-level code that handles UNIX domain socket communication skips some checks that would be needed for INTERNET socket.
For our Python snippets code difference for UNIX domain / INTERNET domain is just socket address and socket type value. See Example 3 for INTERNET domain example.

There are resources on socket types mentioned below.

Program that uses a TCP socket (UNIX domain) in such a way that the socket connection is established when the program starts, and the connection is closed when the program exits is available in 2-stats-json.py:



import time
import socket
import json
import random
import atexit


def format_measurement_data_json(data):
    data['format'] = 'json'
    return json.dumps(data) + '\n'


class StatsReporter:
    def __init__(
        self,
        socket_type,
        socket_address,
        encoding='utf-8',
        formatter=None
    ):
        self._socket_type = socket_type
        self._socket_address = socket_address
        self._encoding = encoding
        self._formatter = formatter if formatter else lambda d: str(d)
        self.create_socket()

    def create_socket(self):
        try:
            sock = socket.socket(*self._socket_type)
            sock.connect(self._socket_address)
            self._sock = sock
            print('Created socket')
        except socket.error as e:
            print(f'Got error while creating socket: {e}')

    def close_socket(self):
        try:
            self._sock.close()
            print('Closed socket')
        except (AttributeError, socket.error) as e:
            print(f'Got error while closing socket: {e}')

    def send_data(self, data):
        try:
            sent = self._sock.send(
                self._formatter(data).encode(self._encoding)
            )
            print(f'Sending sample data... {sent}')
        except (AttributeError, socket.error) as e:
            print(f'Got error while sending data on socket: {e}')

            # attempt to recreate socket on error
            self.close_socket()
            self.create_socket()


reporter = StatsReporter(
    (socket.AF_UNIX, ),
    '/tmp/telegraf.sock',
    formatter=format_measurement_data_json
)
atexit.register(reporter.close_socket)


while True:
    reporter.send_data({'value1': 10, 'value2': random.randint(1, 10)})
    time.sleep(1)

This program opens the connection once and sends measurement over it every second. If the send fails, connection is reestablished. When program exits, the socket is closed using atexit. Even better way would be to reestablish connection once in a while, say every one minute.

StatsReporter class encapsulates operations with socket:
creating, sending data, closing; it also keeps reference to open socket as a field which all those methods can use.

Formatting of measurement data from Python dict into string sent over wire is performed in format_measurement_data_json function. This function is passed as an argument to StatsReporter class, so it will be easy to change data format in future examples.
A tag which corresponds to data format is added in order to distinguish between measurements reported in a different example, and just as an example of a tag.

\n at the end of string that is sent is crucial, this is how Telegraf recognizes the end of a measurement. Without \n at the end of measurement string one can encounter errors like:



  2020-11-10T14:42:17Z E! [inputs.socket_listener] Unable to parse incoming line: invalid character '{' after top-level value
```

Stop Example 1 Python program and Telegraf, and run Example 2 Python program [2-stats-json.py](https://github.com/CheViana/python-send-stats/blob/master/2-stats-json.py) and launch Telegraf for it with config [telegraf-2-stats-json.conf](https://github.com/CheViana/python-send-stats/blob/master/telegraf-2-stats-json.conf):
```
> python3 2-stats-json.py

In other terminal tab
> telegraf -config telegraf-2-stats-json.conf
```

 You should see measurements in real time on dashboard:

![Example 2 Grafana dashboard config and results](https://dev-to-uploads.s3.amazonaws.com/i/ubu8v7s51j5jvrlgxda1.png)


[telegraf-2-stats-json.conf](https://github.com/CheViana/python-send-stats/blob/master/telegraf-2-stats-json.conf#L658) specifies field {% raw %}`name_override = "good_metric_name"`, which is used as measurement name in database records:

```
[[inputs.socket_listener]]
  service_address = "unix:///tmp/telegraf.sock"
  data_format = "json"
  name_override = "good_metric_name"
  tag_keys = ["format"]
```

 Default measurement name would be a non-descriptive input plugin name (e.g. `socket_listener`). It is also possible to specify the key `json_name_key` in Telegraf config to store a measurement in the database with a custom name:

```
[[inputs.socket_listener]]
  service_address = "unix:///tmp/telegraf.sock"
  data_format = "json"
  json_name_key = "metric_name"
```

Then when Telegraf receives the following measurement data:

```
{"metric_name": "speed", "value": 10}
```

The measurement named `speed` with `value=10` will be saved to DB.
This way is more flexible and avoids the need to update config when measurement name varies.

 See more in [JSON Telegraf format docs](https://github.com/influxdata/telegraf/tree/master/plugins/parsers/json).

 Example 2 telegraf config also specifies `tag_keys = ["format"]` - meaning from measurement data dictionary `{'value': 1, 'format': 'json'}` `format` will be used as a tag for measurement (consult [InfluxDB docs](https://docs.influxdata.com/influxdb/v2.0/reference/key-concepts/) if that doesn't mean much to you).


## Example 3. Wavefront (VMWare) Telegraf data format over TCP socket (INTERNET domain)

Python code to send measurement in wavefront format [3-stats-wavefront.py](https://github.com/CheViana/python-send-stats/blob/master/3-stats-wavefront.py), telegraf config [telegraf-3-stats-wavefront.conf](https://github.com/CheViana/python-send-stats/blob/master/telegraf-3-stats-wavefront.conf). Stop other examples and run this one:

```
> python3 3-stats-wavefront.py

In other terminal tab
> telegraf -config telegraf-3-stats-wavefront.conf
```

[3-stats-wavefront.py](https://github.com/CheViana/python-send-stats/blob/master/3-stats-wavefront.py) code differs from Example 2 in couple of lines - formatting function and socket type/address:

```
...
import math

...

def format_measurement_data_wavefront(data):
    lines = []
    for key, value in data.items():
        line = (
            f'prefix_metric_name.{key} {value} '
            f'{math.floor(time.time())} '
            f'source=localhost format="wavefront"\n'
        )
        lines.append(line)
    return ''.join(lines)

...

reporter = StatsReporter(
    (socket.AF_INET, socket.SOCK_STREAM),
    ('127.0.0.1', 8094),
    formatter=format_measurement_data_wavefront
)

...

```

Wavefront format uses timestamp in seconds, so timestamp is set in Python code using `time.time()` without decimal fraction. Omitting timestamp didn't work out for me.
`\n` at the end of str that is sent is quite crucial (same as for Example 2, or any code snippet using TCP socket). Wavefront format also requires `source` tag. `format="wavefront"` part of string is example of how measurement tags should be added.
More about Wavefront data format - in [wavefront docs](https://docs.wavefront.com/wavefront_data_format.html).

Wavefront code piece is using TCP socket, INTERNET domain. This code snippet is suitable when program that sends metrics and Telegraf process run on different hosts. Generally, this code snippet should work in any network configuration, so it can be called more universal than previous examples. TCP connection is reused in similar fashion as in Example 2 for Unix stream socket.

Wavefront Example also has different names of measurements. It can only do single field value per measurement, whereas JSON and Influx Line formats can do measurements with multiple fields - [more about multiple fields measurements](https://stackoverflow.com/questions/45368535/influxdb-single-or-multiple-measurement). So will have to update dashboard or make new panel to see results:

![Example 3 Grafana dashboard config and results](https://dev-to-uploads.s3.amazonaws.com/i/vxje435q0gnun5p1d1wx.png)


## Example 4. Influx Line format over UDP socket

Python code to send measurement in Influx Line format: [4-stats-influx-line.py](https://github.com/CheViana/python-send-stats/blob/master/4-stats-influx-line.py), telegraf config [telegraf-4-stats-influx-line.conf](https://github.com/CheViana/python-send-stats/blob/master/telegraf-4-stats-influx-line.conf). Stop other examples and run this one:

```
> python3 4-stats-influx-line.py

In other terminal tab
> telegraf -config telegraf-4-stats-influx-line.conf
```

Grafana config is same as for Example 2 so you should be able to see real-time results on dashboard:

![Example 4 Grafana dashboard config and results](https://dev-to-uploads.s3.amazonaws.com/i/vdjjiow5r69i16nqcdbe.png)

[4-stats-influx-line.py](https://github.com/CheViana/python-send-stats/blob/master/4-stats-influx-line.py) code differs from Example 2 and 3 in couple of lines - formatting function and UDP socket related things:

```
...
def format_measurement_to_str_influxline(data):
    measurement_name = 'good_metric_name'

    fields = []
    for key, value in data.items():
        fields.append(f'{key}={value}')
    fields_str = ','.join(fields)

    tags = {'format': 'influxline'}
    tags_strs = []
    for tag_key, tag_value in tags.items():
        tags_strs.append(f'{tag_key}={tag_value}')
    tags_str = (',' + ','.join(tags_strs)) if tags else ''

    return f'{measurement_name}{tags_str} {fields_str}\n'

...

def create_socket(self):
    try:
        sock = socket.socket(*self._socket_type)
        # no sock.connect
        self._sock = sock

...

def send_data(self, data):
    try:
        sent = self._sock.sendto(  # sendto not send
            self._formatter(data).encode(self._encoding),
            self._socket_address  # socket address
        )

...

reporter = StatsReporter(
    (socket.AF_INET, socket.SOCK_DGRAM),
    ('localhost', 8094),
    formatter=format_measurement_to_str_influxline
)

...
```

Influx Line data format is string of form `'{measurement_name}{tags_str} {fields_str}'`.
More about Influx Line data format in [it's docs](https://docs.influxdata.com/influxdb/v1.8/write_protocols/line_protocol_tutorial/).

Influx Line example code piece uses UDP socket (Internet type datagram socket).
Notice the difference of networking code for UDP socket code compared to Examples 2 and 3: no need to connect to socket (no `socket.connect` call). Datagram is just send over to specified network address. No need to keep established connection, no need to recreate connection once in a while. Which is rather convenient for sending stats, less socket management code. Downside is UDP doesn't guarantee datagrams delivery, like TCP does for packets of one data transmission sent over established connection. UDP communication might not be good option for every network setup - need to measure how much packets are lost before using it.

I am not covering UNIX type datagram socket config in this tutorial, but if Telegraf config will have:
```
  service_address = "unixgram:///tmp/telegraf.sock"
```
and code of Example 4 will have:
```
  reporter = StatsReporter(
      (socket.AF_UNIX, socket.SOCK_DGRAM),
      '/tmp/telegraf.sock',
      ...
  )
```
that should do it. I haven't tried though.



## More about sockets

If curious to learn more about sockets, suggested reading is this - https://pymotw.com/2/socket/index.html (and "see also" list on that page). Code is for Python 2 so method names might be outdated, but concepts are valid (and older than Python itself).

I'm providing code snippets that send measurements to UNIX stream socket (Example 2), Internet stream socket (Example 3) and Internet datagram socket (Examples 1 and 4). Can just use those if you're not interested in technical details of network communications. If unsure which one is best for you, I suggest to use code and config from Example 1 or Example 4.

You can check out how socket Telegraf process uses look using command `lsof -p [pid of Telegraf process]`. To get `pid` (process id) of Telegraf process, can use `ps aux | grep telegraf` command. `lsof` will show stuff like device name which is associated with Telegraf's socket, socket type, other curiosities. 


## Troubleshooting

If data doesn't appear on dashboards, can launch Telegraf with `--debug` option, to make it print out more information about errors in processing of received data.

When Telegraf successfully receives and write to InfluxDB measurements, it should produce console output similar to:

![telegraf output](https://dev-to-uploads.s3.amazonaws.com/i/gojd6qzhjon4dy5bce0h.png)


You can see it also says that buffer is not full. Means all incoming metrics are making it to database, no dropped readings on Telegraf's side. In real setup, some metrics could be lost in network before they got to Telegraf, but this is not likely when everything runs on same machine.

Also good idea is to check in case of issues:
- InfluxDB is launched
- InfluxDB address in Telegraf config matches the one in InfluxDB config
- Grafana dashboard configuration - address of InfluxDB and database name, measurement names
- Python code sends data to correct socket address, the one Telegraf listens on (specified in Telegraf config)

### InfluxDB data investigation

To debug what's being written to InfluxDB, can use [Influx CLI](https://docs.influxdata.com/influxdb/v1.8/query_language/explore-data/) or [influx flux query language](https://docs.influxdata.com/influxdb/v2.0/query-data/get-started/). I've used Influx CLI and `SELECT` statements, as this is something I'm more familiar with.
Launch Influx CLI with command `influx`. To show list of available databases, use command `show databases`. Switch to database Telegraf sends data to using `use "socket-stats"` command. Show all measurement names using `show measurements`. To see what's going on in particular measurement, can use `select *::field from "value1"` - it will show all fields and all data for measurement called "value1". `select *::field from "value1" limit 3` will show 3 oldest data points, `select last(*::field) from "value1"` will show newest data point.

![Influx CLI example](https://dev-to-uploads.s3.amazonaws.com/i/zc29oeqbm04bf7ctnl8q.png)
![Influx CLI latest measurement](https://dev-to-uploads.s3.amazonaws.com/i/kefuyxl4t0hjtumx5mak.png)

These screenshots show my trouble: `value2` timestamp value is not correct, it's millisecond-precision Unix time whereas data format requires nanosecond-precision Unix time (like "test.value2" timestamp). So `value2` timestamp is interpreted as way older timestamp than it should be (it has late 60s vibe), and won't show up on "last 5 min" Grafana dashboard.

![Readings from the past](https://dev-to-uploads.s3.amazonaws.com/i/xhpiqxduv3u4uh2g4j0i.png)


### Measurement timestamp

It is possible to report timestamp of measurement from Python code, or leave it up to InfluxDB to record timestamp of when reading arrives. Delay between two event is usually negligible: on same machine - real tiny, over network - depends on network, but like couple milliseconds, maybe hundred milliseconds. My suggestion is to leave it up to InfluxDB, to avoid issues when reported time from Python is not correct due to bugs, or different machines have different clock time. If exact time of reading with nanosecond precision is important to you, add timestamp field in Python code. 
Anyway, if reporting program and InfluxDB run on different machines, make sure [Network Time Protocol (NTP)|http://www.ntp.org/] is utilized to keep clocks in sync.

### Dashboard issues

In case you're having difficulties configuring Grafana dashboards, complete JSON that could be used to export dashboard configuration is in [grafana-dashboard-complete.json](https://github.com/CheViana/python-send-stats/blob/master/grafana-dashboard-complete.json) file. Can try to export it in new dashboard or compare it's panels JSON with your panels.


## What I might write about in next post:

- overloading TCP socket (Unix socket, UDP socket) with metrics, and checking out what happens; looking into `read_buffer_size` in Telegraf config and system socket listen queue size; techniques to measure dropped readings rate
- reporting stats of backend calls (`aiohttp` and `requests`)
- optimal uWSGI configurations, for best performance when all is good, and backend failure-resistant configurations
- uWSGI serving Django with aiohttp communications
- babel 7 configurations for less JS in bundle
- running python tests in parallel, and tests coverage

uWSGI stats monitoring from scratch using Telegraf InfluxDB and Grafana

Jane Radetska — Tue, 25 Aug 2020 00:35:00 +0000

At the end of this tutorial, you'll end up with dashboard like this. Each panel shows some uWSGI metric as time series.

Watching over uWSGI-reported statistics like worker busyness is super helpful for investigating uWSGI configurations. Also useful in production environment - it's possible to extend described approach to monitor real-life uWSGI web applications, but this post doesn't aim to cover that.

In future, I plan to add posts about how different uWSGI options influence behavior of web server, and how to monitor Python web app aspects, such as networking, using same tools.

Tutorial roadmap

Here's what will be described:

setup and run simple uWSGI webserver
load that webserver using wrk2
install and run InfluxDB
install, configure and run Telegraf
install and run Grafana
create dashboard in Grafana to show uWSGI metrics
each monitored uWSGI metric dive-in

Code and configs mentioned in here can be found in the repo.

The diagram

uWSGI stats

uWSGI can expose stats on a separate socket.

The simplest way to see these metrics is to use uwsgitop - https://github.com/xrmx/uwsgitop. However, uwsgitop only shows current metrics readings, not history data, just like Linux top command.

Some time ago I created a fork of uwsgitop because of an encoding-related bug in its output. I had fun a time trying to figure out why uwsgitop won't work. That bug seems to be fixed now though.

I think it's nicer to be able to see uWSGI stats metrics over a continuous period of time as that provides more information than just this moment readings.

uwsgitop is like a speedometer readings - change every second. ⌚

Monitoring dashboard is like a cardiogram - recorded readings over time. 📈

Choosing monitoring tools

One option for continuous monitoring is to use Prometheus with exporter for uWSGI stats; here, I'll describe other option - TIG stack.

TIG stack differs from Prometheus mainly in the idea of relying on some other tool to push metrics to time series DB, instead of pulling model which Prometheus uses. Actually, both of the stacks can do push and pull, and there's discussion going on about which method works better for which type of thing you want to watch over.

For the kind of local comparative experiments I'm planning to run on uWSGI web servers in following posts, I don't think there's much difference which monitoring solution to use.

I picked TIG for this post because I have some experience with it and Prometheus approach is described elsewhere.

One little pebble thrown in the direction of Prometheus: its uwsgi-exporter seems to not support reporting of uWSGI worker status which is indeed very useful metric. Probably support for that will be added in time.

List of what Telegraf uWSGI plugin can monitor is rather impressive.

Simple uWSGI web server

Here's code of "Hello World" uWSGI app in uwsgi-hello-world.py:

import time


def application(env, start_response):
    start_response('200 OK', [('Content-Type','text/html')])
    time.sleep(0.25)  # sleep 250 msec
    return [b'Hello World']

And basic uWSGI configs in uwsgi-hello-world-configs.ini:

[uwsgi]
http-socket = 127.0.0.1:9090
wsgi-file = uwsgi-hello-world.py
master = true
processes = 4
threads = 2 
stats = 127.0.0.1:9191
stats-http = true

This code doesn't do anything useful. Worker sleeps for 250 msec and returns a string.
Can use wsgi.py of more interesting Django or Flask server or any other Python web app (change wsgi = path/to/Django-app/wsgi.py in options).

uWSGI can be installed as Python package. Let's assume you have Python installed, know how to create and activate Python virtual environment (if not, please consult virtualenv docs and virtualenvwrapper is handy).

To install uWSGI Python package:

> mkvirtualenv --python=python3 uwsgi-playground
> pip install uWSGI

To run uWSGI server:

> uwsgi --ini uwsgi-hello-world-configs.ini

That last command should produce output similar to:

...
uwsgi socket 0 bound to TCP address 127.0.0.1:9090
*** Operational MODE: preforking+threaded ***
WSGI app 0 ... ready in 0 seconds ...
*** uWSGI is running in multiple interpreter mode ***
spawned uWSGI master process (pid: ...)
spawned uWSGI worker 1 (pid: ..., cores: 2)
spawned uWSGI worker 2 (pid: ..., cores: 2)
spawned uWSGI worker 3 (pid: ..., cores: 2)
spawned uWSGI worker 4 (pid: ..., cores: 2)
*** Stats server enabled on 127.0.0.1:9191 fd: ... ***

This means that the web server has launched and it's listening on port 9090.

Can visit hello-world server in local web browser - on http://127.0.0.1:9090/:

uWSGI process outputs to console that it processed requests after we visited hello-world page:

[pid: ...|app: 0|req: 1/1] 127.0.0.1 () {42 vars in 783 bytes} [Sun Jul 19 20:21:38 2020] GET / => generated 11 bytes in 250 msecs (HTTP/1.1 200) 1 headers in 44 bytes (2 switches on core 0)
[pid: ...|app: 0|req: 1/2] 127.0.0.1 () {38 vars in 670 bytes} [Sun Jul 19 20:21:38 2020] GET /favicon.ico => generated 11 bytes in 143 msecs (HTTP/1.1 200) 1 headers in 44 bytes (2 switches on core 0)

uWSGI can also be configured to log those lines to a file (see uWSGI logging docs).

Can also visit stats endpoint in local web browser - on http://127.0.0.1:9191/:

...
"workers": [
    {
        "id":2,
        "pid":...,
        "accepting":1,
        "requests":1,
        "delta_requests":1,
        "avg_rt":71928
        ....
]
...

That JSON contains "speedometer" readings of what's going on inside uWSGI webserver.

Loading web server with benchmark tool wrk2

To keep uWSGI web server busy (so that stats are not just all zero), let's load uWSGI hello-world server using wrk2 benchmarking tool. Can use any other artificial load generator. I picked wrk2 because I don't need complicated test scenario for this post. Some load testing tools can be easily configured to write to InfluxDB, which makes observing results of load test real handy, right next to web server stats and stats reported from web app code.

To install wrk2:

> git clone https://github.com/giltene/wrk2
> cd wrk2
> make

If make doesn't work for you, can use wrk tool - which provides install wiki pages for all platforms.

To run wrk2 or wrk:

> ./wrk -t2 -c2 -d1200s -R1 http://127.0.0.1:9090/

This command creates 2 threads that try to load webserver with 1 RPS, keeping 2 HTTP connections open.
It will create load for 1200 sec which is 20 min. Feel free to adjust.

Tools for real-time monitoring: Graphana, Telegraf, InfluxDB (also known as TIG stack)

To install all tools locally on MacOS:

> brew install influxdb  <-- Database for metrics
> brew install telegraf  <-- agent-collector of metrics
> brew install graphana  <-- UI for metrics exploration and plotting

To download all tools binaries locally on Linux:

> wget https://dl.influxdata.com/influxdb/releases/influxdb-1.8.2_linux_amd64.tar.gz
> tar xvfz influxdb-1.8.2_linux_amd64.tar.gz
> wget https://dl.influxdata.com/telegraf/releases/telegraf-1.15.2_linux_amd64.tar.gz
> tar xf telegraf-1.15.2_linux_amd64.tar.gz
> wget https://dl.grafana.com/oss/release/grafana-7.1.4.linux-amd64.tar.gz
> tar -zxvf grafana-7.1.4.linux-amd64.tar.gz

For other platforms:

go to https://portal.influxdata.com/downloads/ for InfluxDB and Telegraf
for Grafana visit https://grafana.com/grafana/download

Docker containers are available in this post, or:

> docker pull influxdb
> docker pull telegraf
> docker run -d --name=grafana -p 3000:3000 grafana/grafana

Run time series database InfluxDB

Launch InfluxDB:

> influxd -config /usr/local/etc/influxdb.conf

This starts up DB process.
Create database for uWSGI metrics (in other shell tab):

> influx -precision rfc3339  <-- opens CLI
Connected to http://localhost:8086 version v1.8.1
InfluxDB shell version: v1.8.1
> CREATE DATABASE localmetrics  <-- creates our DB
> SHOW DATABASES  <-- shows DB list
name: databases
name
----
_internal
localmetrics
>

It is now possible to run "INSERT ..." command using CLI, which will add metric reading to the database.

Run Telegraf - stats pull/push agent

We want to have uWSGI stats sent to database automatically every N seconds, and in a format that InfluxDB will understand.
uWSGI passively exposes stats data on socket 9191.
Need something that will query uWSGI stats endpoint and send metrics data to InfluxDB database.

This is where Telegraf comes into light. Telegraf has ability to retrieve metrics data, transform it to format InfluxDB understands using plugins and send it over to InfluxDB database.
Telegraf has a bunch of input plugins: to watch over CPU consumption levels, to read and parse log file tail, to listen to messages on socket and various web servers integrations, including uWSGI.

It's a matter of adding a few lines to Telegraf config to enable uWSGI stats reporting. Resulting telegraf.conf is available in the repo.
To understand Telegraf tool better, let's use telegraf sample-config utility to compose configs with which telegraf can consume uWSGI metrics:

> telegraf -sample-config --input-filter uwsgi --output-filter influxdb > telegraf.conf
> cat telegraf.conf
...
# Read uWSGI metrics.
[[inputs.uwsgi]]
## List with urls of uWSGI Stats servers. URL must match pattern:
## scheme://address[:port]
##
## For example:
## servers = ["tcp://localhost:5050", "http://localhost:1717", "unix:///tmp/statsock"]
servers = ["tcp://127.0.0.1:1717"]

## General connection timeout
# timeout = "5s"

tcp://127.0.0.1:1717 part doesn't match where our uWSGI exposes stats on. It's http://127.0.0.1:9191. Need to update that in telegraf.conf file.

Another thing that needs to be updated is the address of where to send stats:

# Configuration for sending metrics to InfluxDB
[[outputs.influxdb]]
## The full HTTP or UDP URL for your InfluxDB instance.
##
## Multiple URLs can be specified for a single cluster, only ONE of the
## urls will be written to each interval.
# urls = ["unix:///var/run/influxdb.sock"]
# urls = ["udp://127.0.0.1:8089"]
urls = ["http://127.0.0.1:8086"]

## The target database for metrics; will be created as needed.
## For UDP url endpoint database needs to be configured on server side.
database = "uWSGI"

I uncommented urls = ["http://127.0.0.1:8086"] line and added database = "uWSGI" so metrics from uWSGI stats server will flow in separate database.

Resulting telegraf.conf is in the repo.

To run telegraf:

> telegraf -config telegraf.conf

Looking in InfluxDB console output, can see:

2020-07-20T01:18:26.727302Z info    Executing query {"log_id": "0O6K1AQG000", "service": "query", "query": "CREATE DATABASE uWSGI"}
[httpd] 127.0.0.1 - - [19/Jul/2020:21:18:26 -0400] "POST /query HTTP/1.1" 200 57 "-" "Telegraf/1.14.5" e9bd1368-ca26-11ea-8005-88e9fe853b3a 428
[httpd] 127.0.0.1 - - [19/Jul/2020:21:18:40 -0400] "POST /write?db=uWSGI HTTP/1.1" 204 0 "-" "Telegraf/1.14.5" f1a77b04-ca26-11ea-8006-88e9fe853b3a 137748
[httpd] 127.0.0.1 - - [19/Jul/2020:21:18:50 -0400] "POST /write?db=uWSGI HTTP/1.1" 204 0 "-" "Telegraf/1.14.5" f79d5380-ca26-11ea-8007-88e9fe853b3a 7695
[httpd] 127.0.0.1 - - [19/Jul/2020:21:19:00 -0400] "POST /write?db=uWSGI HTTP/1.1" 204 0 "-" "Telegraf/1.14.5" fd928134-ca26-11ea-8008-88e9fe853b3a 8534
[httpd] 127.0.0.1 - - [19/Jul/2020:21:19:10 -0400] "POST /write?db=uWSGI HTTP/1.1" 204 0 "-" "Telegraf/1.14.5" 0388c512-ca27-11ea-8009-88e9fe853b3a 9345

This means Telegraf is already busy writing uWSGI metrics to database, but we can't see them yet.

Run Grafana - UI for metrics monitoring

Launch grafana web server:

> cd path/to/dir/with/installed/graphana
> bin/grafana-server

Navigating to http://localhost:3000/ in browser opens window with Grafana UI. Login using "admin" "admin" and create new pass as it asks to.

I'm providing screenshots of how to deal with Grafana UI as it was confusing to me. Here Grafana version v7.1.0 is featured on screenshots, UI might look different for other version.

What's left to setup:

create data source in Grafana for InfluxDB
add panels that will show metrics readings over time - can do "import it all" way, can do manually

Create InfluxDB data source in Grafana

Pick "Configuration -> Data Sources -> Add data source". Select InfluxDB.

Put "http://127.0.0.1:8086" in HTTP/URL input and database name "uWSGI" in "InfluxDB Details/Database" input.

Click on "Save and Test" button in the bottom of the screen, green noty reading "Data source is working" should appear.

Visualizing uWSGI metrics in Grafana

Now Grafana can read metrics from database, let's visualize them - add graph panel for each metric.

Grafana dashboard consists of panels, panel can show how particular metric changed in time. There are lots of types of panels but we will only deal with Time Series panel in this tutorial.

Here's dashboard snapshot with measurements of hello-world app with 4 workers being loaded by artificial users. That dashboard is monitoring following uwsgi metrics:

harakiri count
worker status (busy/idle/cheap/total)
listen queue size
workers amount
worker requests
in-request sum
respawn count
worker avg request time
worker running time
load sum

There are more things that can be monitored: amount of transmitted data for worker, amount of exceptions for worker, etc. I didn't need these metrics or prefer to monitor them in other ways.

Can also configure uWSGI memory reporting - how much memory uWSGI consumes. I prefer to watch memory consumption from system monitoring though, along with CPU consumption.

uWSGI even allows to expose your own metrics.

Cheaper subsystem plugin, and other plugins, have their own metrics too.

I am providing instructions on how to setup all panels at once, using dashboard JSON, or how to manually add panels one by one, and learn a bit more in the process.

Manual dashboard setup: add panels one by one

Might be beneficial to read this first How to add a panel in Grafana.

How to create new panel in new dashboard:

Go to "Dashboards -> Manage dashboards", click on "New dashboard"
click "New panel" or "Add panel" in top right corner
on dropdown next to new panel title pick "Edit"

How to populate panel with avg worker request time data:
In panel edit mode, select Query source - InfluxDB.
Modify query builder inputs:

From default uwsgi_workers
Select field(avg_rt) mean()

It is important to tell the panel that the metric it displays is measured in microseconds: in the right column, expand "Axes", "Left Y", "Unit" - "microseconds".

Also, it's nice to configure points thickness so that you can see them clearly ("Display" - "Line width" in the right column).
Give a meaningful name to that panel (in top of right column), save panel (click "Apply" in top right corner).

Congrats, now you have a panel that shows avg request time of all uwsgi workers in real time. Make sure that the refresh frequency selector is set to like 10 sec (tiny dropdown in top right corner of dashboard page) and that the webserver, wrk2, telegraf, and InfluxDB are all still running.

Similar process should be repeat for rest of panel - find queries to use below in this post, and can paste queries in panel "edit query" input.

Automatic dashboard setup: import uWSGI stats dashboard JSON

To setup uWSGI stats monitoring dashboard, can use JSON in uwsgi-dashboard-model.json, and export that to new dashboard:

Go to "Dashboards -> Manage dashboards", click on "New dashboard"
Go to Dashboard settings ("wheel" button at the top right of the page with new dashboard)
In the left menu, pick "JSON Model"
In JSON, find the "panels" field (it should be empty) and paste there contents of field "panels" from uwsgi-dashboard-model.json. Pasting all JSON doesn't work for me for some reason.
Click "Save changes" at the bottom of Dashboard settings page.

Congrats, now you have dashboard that shows uWSGI stats in real time. Make sure refresh frequency selector is set to like 10 sec, tiny dropdown in top right corner of dashboard page, and that the webserver, wrk2, telegraf, and InfluxDB are all still running.

Grafana specifics: time filter and time interval

$timeFilter, seen in panel query builder, stands for time range picked in Grafana UI, the period of time for which you want to see metrics.

$__timeInterval, seen in panel query builder, stands for time interval. Time interval depends on what's the time range you're looking at in dashboard, meaning the time interval between the two nearest dots on series. Looking at a 1 hour range in dashboard (put from=now-1h&to=now in URL or use the time range picker in the upper right corner of dashboard), I see a dot at 11:01:02. The next dot is 11:01:06, so the time interval is 4 seconds.

The time interval is important to use so that the dashboard for a large range (e.g. 30 days) loads in a reasonable amount of time.

Telegraf is configured to send data from uWSGI stats server to InfluxDB once every 10 seconds. If the time interval is bigger than 10 seconds, one dot in the series corresponds to avg/sum/... (set in query which one) of multiple uWSGI stats readings (those that got into time interval).

uWSGI stats - by metric

Harakiri count

uWSGI has a useful feature, to kill worker if worker executes request for longer than defined time (e.g. 10 sec). It's called "harakiri". To configure it, set uwsgi option harakiri=10 (in uwsgi.ini), where 10 is the time of the longest allowed request in seconds.
This option can be used to protect from DDoS attacks that exploit long-executing requests. Such attacks flood a website with long-executing requests to such an extend that the website doesn't have the capacity (free workers) to serve regular user traffic, since all the workers are busy executing attacker's requests.
Setting low harakiri can bite you if the web server is expected to serve long-executing requests in some rare cases. One needs to analyze what's the longest valid request time. Consider refactoring code to avoid long-executing requests in worker processes. There are lots of options depending on the specifics of the problem: uWSGI spooler to send emails, uWSGI mules, or some kind of task queue for long-executing requests that's separate from the webserver.

Harakiri panel query is as follows:

SELECT sum("harakiri_count") FROM "uwsgi_workers" WHERE $timeFilter GROUP BY time($__interval) fill(null)

This will shows sum of harakiri event for time interval.

Worker status

uWSGI worker can be in few states:

idle (not working on a request)
busy (working on requests)
cheap (see uWSGI cheaper subsystem docs)

I configured Worker status panel to show idle, busy, cheap and total amounts of workers.

Query:

SELECT count("status") FROM "uwsgi_workers" WHERE  $timeFilter and "status"='busy' GROUP BY time($__interval) fill(null)

for "busy" series, for other series replace "busy" with "idle" or "cheap", for total - omit status clause (... WHERE $timeFilter GROUP BY ...).

Can configure this panel in different way - to show worker busyness in percent for previous period of time. I thought this approach more useful for uwsgitop and more confusing when metrics are monitored in time by separate monitoring stack.

Listen queue size

If all uWSGI workers are busy working on requests while new requests are arriving, those new ones are first put in queue (socket listen queue) to wait for next free worker.
The size of the listen queue is configurable using the option listen=64 but max allowed value depends on system max socket listen queue size, so you might need to increase system value first.

Panel query is:

SELECT sum("listen_queue") FROM "uwsgi_overview" WHERE $timeFilter GROUP BY time($interval)

The number of workers

The number of workers uWSGI is currently running. Makes more sense when cheaper subsystem is in use, or when cluster has multiple web server instances with scaling (all reporting to same DB) - then this count changes.

The query to get the number of workers with IDs 1 - 4:

SELECT count("avg_rt") FROM "uwsgi_workers" WHERE $timeFilter AND ("worker_id"='1' OR "worker_id"='2' OR "worker_id"='3' OR "worker_id"='4') GROUP BY time($__interval) fill(null)

This is an indirect metric that counts how many times avg_rt metric was reported for "worker_id"='1' during the time interval. It the time interval is bigger than 10 sec (telegraf queries uWSGI stats once every 10 sec) - e.g. when time range you're looking at is 6 hours, this actually shows incorrect data.
A question to figure out if you're curious and don't mind digging into Grafana docs: how does one make it show correct data regardless of chosen time range? Comment below if you find out.

Worker requests

How much requests one worker has executed since worker process was started. When webserver serves requests, this measurement rises smoothly. When a worker restarts, it falls to zero.

Query is:

SELECT mean("requests") FROM "uwsgi_workers" WHERE $timeFilter GROUP BY time($interval), "worker_id"

In-request sum

How many requests uWSGI is working on right now, across all workers, sum for time interval.

Query:

SELECT sum("in_request") FROM "uwsgi_cores" WHERE $timeFilter GROUP BY time($interval)

Respawn count

Query:

SELECT mean("respawn_count") FROM "uwsgi_workers" WHERE $timeFilter GROUP BY time($interval), "uwsgi_host"

How many times workers were respawned since uWSGI start.

For production system it's recommended to gracefully respawn uwsgi workers after some time, to avoid excessive memory consumption of long-living processes, etc.

To limit amount of executed requests after which to respawn, add in uWSGI options:

max-requests=10000
max-requests-delta=1000

This will respawn worker after 10000+(1000*worker_id) requests. Purpose of delta is not to respawn all workers at the same time.

To limit the amount of time passed after which to respawn, add in options:

max-worker-lifetime=36000
max-worker-lifetime-delta=3600

Lifetime values are in sec. This will respawn a worker if 10h+(1h*worker_id) of time has passed since last respawn.

One can use both limits together; whichever one occurs first for particular worker will take effect.

Worker avg request time

Query:

SELECT mean("avg_rt") FROM "uwsgi_workers" WHERE $timeFilter GROUP BY time($__interval), "worker_id" fill(null)

Amount of time worker spends on request, on average, per each worker.

Worker running time

Query:

SELECT mean("running_time") FROM "uwsgi_workers" WHERE  $timeFilter GROUP BY time($interval), "worker_id"

How long is worker running; time since the last respawn.

Load sum

Query:

SELECT mean("load") FROM "uwsgi_overview" WHERE $timeFilter GROUP BY time($interval)

I am 100% sure what's this exactly. It seems to me this is something like "load average" for CPU top, but a "sum". This metric rises dramatically when web server is overloading.

Conclusions

At this point, we have a basic uWSGI web server and a monitored playground to watch over it while we experiment with uWSGI configurations.

I encourage you to try out the effects of changing uWSGI options on the web server. Start by setting max-requests=10 in uwsgi.ini and compare how that changes stats for web server under load. You can also check how it affects the results seen in load tool (wrk2) summary.

DEV Community: Jane Radetska

Setup Knative Eventing with Kafka from scratch, scale based on events volume, and monitor

Cluster creation

Installing Kafka and Knative

Autoscaling Knative

Monitoring Knative and Kafka

More tuning

Make sure it works

Performance testing Strimzi Kafka in the k8s cluster using xk6-kafka

Topic to test

Test scenario

Pod used to run test scenario, and command to run test in the k8s cluster

KubeCon CloudNativeCon Europe 2023 homepage made usable

What I am unhappy about in the initial homepage

How to add JS to make homepage pretty

Option 1, with autoloading of the script that adjusts KubeCon homepage. Using Tampermonkey or similar Chrome plugin.

Option 2. Use content snippets. No plugins. No autoload.

Manual (artistic) process of fixing the homepage

Delete the buttons on the right to the schedule

Make left column full-width

Remove the useless "Community in Bloom" header

Remove equally useless "Hello, Jane" header

Find container with schedule and remove fixed height to get rid of, oh gosh, scroll-in-a-scroll

P.S. What about Mobile?

Histogram of request time in Grafana with Telegraf

Monitoring sync and async network calls in Python using TIG stack

What I'm going to explore in this post

Example 0: monitor requests request time

Sending stats

profile decorator

requests execution time on dashboard

Need more exceptions

Example 0 improved: reuse connection

Example 1: monitor aiohttp request time

The tale of two HTTP requests

aiohttp requests signals

Calling asyncronous code from synchronous

Compare results for Example 0 and 1

Example 2: more, more stats

Example 3: aiohttp reuse session

Compare sync and async URL fetch, with and without reusing connection

Reporting Measurements from Python Code in Real Time: a Beginner-Friendly Tutorial

Reporting measurements from Python code in real time

Tutorial materials

Looking for a quick, ready, robust solution?

Setup Grafana, InfluxDB, Telegraf

Example 1. The simplest example of how to send stats from Python code in 6 lines, and of suitable Telegraf config

Metric name gotchas

Grafana Dashboard

Example 2. JSON measurements over TCP socket (UNIX domain)

uWSGI stats monitoring from scratch using Telegraf InfluxDB and Grafana

Tutorial roadmap

The diagram

uWSGI stats

Choosing monitoring tools

Simple uWSGI web server

Loading web server with benchmark tool wrk2

Tools for real-time monitoring: Graphana, Telegraf, InfluxDB (also known as TIG stack)

Run time series database InfluxDB

Run Telegraf - stats pull/push agent

Run Grafana - UI for metrics monitoring

Create InfluxDB data source in Grafana

Visualizing uWSGI metrics in Grafana

Manual dashboard setup: add panels one by one

Automatic dashboard setup: import uWSGI stats dashboard JSON

Grafana specifics: time filter and time interval

uWSGI stats - by metric

Harakiri count

Worker status

Listen queue size

The number of workers

Worker requests

In-request sum

Respawn count

Worker avg request time

Worker running time

Load sum

Conclusions

Example 0: monitor `requests` request time

`profile` decorator

`requests` execution time on dashboard

Example 1: monitor `aiohttp` request time

`aiohttp` requests signals

Example 3: `aiohttp` reuse session