Simple Aggregate Logs with Benthos
Tags: datascienc, programming, developer
Published: 2023-06-26 04:37:31

## Motivation

Recently, I had a simple task of creating sample data to simulate multiple users sending requests to the server. I wrote a simple program, which loops and creates records in the database. Every second, it prints the log of the number of records saved within that second (op/s); it was easy to know the total number of ops when running locally with only one instance. When deploying to k8s, I wanted to increase the number of pods to create more stress, but it was difficult to count the total_ops from all these pods.

What I have are logs like this:

```json
{
  "node": "xxx",
  "ops_per_second": 10,
  "avg_latency_ns": 2
}
{
  "node": "yyy",
  "ops_per_second": 10,
  "avg_latency_ns": 4
}
```

What I want:

```json
{
  "total_ops": 20
}
```

## Solution

To get such a result, I need to stream, parse the logs and store their values. Writing a script wouldn't be too hard, but in the process of researching data processing, I came across Bentos.

## What is Benthos?


![what is benthos](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/hu11kjyoqwwg0mvhw8jf.png)

Benthos is a data processor that can receive data from various sources, has flexible and powerful data processing capabilities with bloblang, and can output to various destinations. It is incredibly easy to install and use and can run locally with a simple command or automatically scale on k8s or cloud (AWS Lambda).

I will have a detailed introduction to Bentos later. In this post, I will only use it in a simple use case to get familiar with: Using stdin as input, and returning metrics in Prometheus format.

## Installation

The simplest way is to use a script:

```bash
curl -Lsf https://sh.benthos.dev | bash
```

or Docker:

```bash
docker pull jeffail/benthos
docker run --rm -v /path/to/your/config.yaml:/benthos.yaml jeffail/benthos
```

or Homebrew:

```bash
brew install benthos
```

## Configuration

We need to define two components:

1. Input
2. Processor

### Input

Our input consists of logs from pods on k8s, using kubeclt to stream logs from pods with the specified label.

```bash
kubectl logs -l app=my-app -f
```

We can use a pipeline to provide this input to benthos:

```bash
kubectl logs -l app=my-app -f | benthos -c config.yaml
```

But we can also run the kubectl command itself in benthos using the input type: [subprocess](https://www.benthos.dev/docs/components/processors/subprocess/):

```yaml
input:
  subprocess:
    name: "kubectl"
    args:
      - "logs"
      - "-l"
      - "app=my-app-app"
      - "-f"
      - "--max-log-requests"
      - "100"
```

After this step, you can run the following command to get the log stream:

```bash
benthos -c benthos.yaml
```

Example output:

```bash
INFO Running main config from specified file       @service=benthos path=./benthos.yaml
INFO Listening for HTTP requests at: http://0.0.0.0:4195  @service=benthos
INFO Launching a benthos instance, use CTRL+C to close  @service=benthos
{"node": "my-app-7b9bd88d65-xrvfc" ,"ops_per_second": 0, "avg_latency_ms": 0}
{"node": "my-app-7b9bd88d65-flqx7" ,"ops_per_second": 1, "avg_latency_ms": 1000}
{"node": "my-app-7b9bd88d65-6gt9t" ,"ops_per_second": 7, "avg_latency_ms": 142}
{"node": "my-app-7b9bd88d65-hjtcj" ,"ops_per_second": 0, "avg_latency_ms": 0}
```

### Aggregate

Now we need to collect the fields `ops_per_second` and `avg_latency_ms` as well as count `total_op`. Fortunately, Benthos supports [components](https://www.benthos.dev/docs/components/metrics/about) for collecting metrics:

```yaml
pipeline:
  processors:
    - metric:
        name: total_ops
        type: counter_by
        value: ${!json("ops_per_second")}
```

`counter_by` will increase the value of the counter based on the value obtained from `value`.

```bash
benthos -c benthos.yaml
```

Metrics won't show up in the output but will be displayed on the page: http://0.0.0.0:4195/metrics

```bash
...
total_ops{label="",path="root.pipeline.processors.0"} 348
...
```

So we have the total number of ops created.

## Conclusion

However, this method only provides an approximate value, as it only pulls log messages from when the benthos runs, not when the pod is created. Nevertheless, it meets my simple needs, although it could be improved by deploying benthos as a service, running continuously, caching metrics in a database, and automatically refreshing when pods change. We may return to this issue in the future.