Clickhouse exporter dropping metrics under pressure

[ksharinarayanan]

Component(s)

exporter/clickhouse

What happened?

Description

Hi,

We are currently using pushing only metrics to the clickhouse exporter which inserts to clickhouse cloud every 10 seconds or every 100k records. The exporter at source exports every 10 seconds.

This works fine until we generate 7K metric data points per second from the source, but once it reaches around 8.7K metric data points per second, almost 30-40% of the metrics are dropped. The source is managed AWS flink and unfortunately it doesn't give us logs related to opentelemetry SDK. So, I have completely no idea why they are getting dropped. The reason I'm concluding that the collector is dropping is because I expect 6 inserts to happen in a minute, but under this load on average an insert is skipped in a minute consistently which I am able to see through the clickhouse cloud query insights, attaching the image below, looking at the image, you can notice that at 12:23:51, an insert should have occured which is missing,

I am just thinking that I might be doing something wrong with the collector config. I've tried a bunch of different configurations related to the sending_queue and batch processor but nothing seems to be working.

Collector version

v0.120.0

Environment information

Environment

OS: Amazon linux

OpenTelemetry Collector configuration

receivers:
  otlp:
    protocols:
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    send_batch_size: 80000
    timeout: 10s
    send_batch_max_size: 100000
  filter:
    error_mode: ignore
    metrics:
      datapoint:
        - 'attributes["http.request.method"] == "GET"'

exporters:
  clickhouse:
    # credentials 
    metrics_tables:
      sum: 
        name: "otel_metrics_sum_null"
    async_insert: true
    compress: zstd
    timeout: 60s
    database: otel
    retry_on_failure:
      enabled: true
      initial_interval: 1s
      max_interval: 30s
      max_elapsed_time: 300s
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 1000000

service:
  pipelines:
    metrics:
      receivers: [otlp]
      processors: [filter, batch]
      exporters: [clickhouse]

Log output

There were no logs when the metrics were getting dropped

Additional context

No response

[github-actions]

Pinging code owners:

• exporter/clickhouse: @hanjm @dmitryax @Frapschen @SpencerTorres

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[ksharinarayanan]

I also see on the ALB that it is responding back with 4xx for some of the requests, which increases my suspicion that collector is rejecting the requests.

[SpencerTorres]

Hello!

I've been doing some performance testing with the exporter recently, and I'm surprised to hear you're struggling with 7k points per second. While testing logs I am able to push upwards of ~800k/s depending on the config.

I have a few suggestions:

1. LOWER YOUR BATCH SIZE! 80k is huge! Try something between 5 - 15k, I think internally we aim for around 8k per batch. Max size I suppose could be in this range as well. The timeout could also be lowered to 5s. This could be why you're seeing 4xxs, perhaps a payload size is being exceeded. The numbers I suggested are based on our experience with logs, so perhaps you can go higher if the metrics payload is smaller, but I think your config is uncomfortably high for the exporter. If you're using async inserts you don't need to worry about doing batches that large (see my next point)
2. I see you have async_insert enabled, but make sure it's not being disabled in your DSN url (the endpoint param). Async inserts will allow the ClickHouse server to buffer these, which is great for when you have multiple otelcol processes. While each process batches, ClickHouse server can batch those batches via async insert which should help performance.
3. Consider using wait_for_async_insert=0. This will change the inserts to be non-blocking; fire and forget. How important are your metrics points? ClickHouse is quite durable, but you should understand the risk of this. If the server process dies before it gets persisted then that batch could be lost. Depending on your metrics data this could be a reasonable risk in exchange for throughput. I think we have this enabled for our internal logging since we found that the error rate is incredibly small (and logs are arguably more important than individual metrics points depending on what you're logging)
4. You may also want to lower your consumer queue_size. These are batches, so it's 1000000 * batch_size that will persist in memory. I feel like your process would crash long before this would be hit (depending on your machine). Yes batches could be discarded, but this may be better than outright crashing the process.
5. Any reason to change the compress algorithm? This is likely negligible but I'm curious if you had any specific reason for this. I believe it defaults to lz4

For more info on how we tuned our OTel processes for logging, see here: https://clickhouse.com/blog/building-a-logging-platform-with-clickhouse-and-saving-millions-over-datadog#tuning-gateways-and-handling-back-pressure

Let me know if you have any more questions 👌

[ksharinarayanan]

Hi @SpencerTorres ,

I've figured out the cause why collector is throwing 4xxs. It is because the default request limit size is 20 MB which I am exceeding very quickly. To get around this, I've increase the limit to 200 MB. Even though that solves the problem, I would like to get more insights into our config related to the points that you've shared,

Just to be clear, when you say 800K/s, is it that so many data points are generated at your source every second? If yes, would it be fair to say that if your export interval from your agent / opentelemetry exporter running at source is 10 seconds, you will accumulate 8M data points?

My case is that I've two services in my pipeline pushing datapoints. Assuming we are dealing with 20K RPS, the first service generate 20K datapoints per second, and the second 40K points per second which results to 60K datapoints per second accumulating in a second. My export interval being 10 seconds, it amounts to 600K datapoints pushed in each cycle to the collector. Once it reaches the collector, it immediately inserts to clickhouse since 600K breaches the batch processor threshold. But, I don't see inserts happening for 600K records, but it happens for around 100K-120K records in a batch, I am not sure on the reason though.

Now, coming to your points:-

1. Since we are generating 60K metric points per second just at the source, I am not sure if we will be able to do inserts of 5-15k records. The only possible way would be to have the export interval at the source to be 250 ms which I feel is way too low. Do let me what you think about this. Just to be clear, clickhouse isn't throwing 4xx, but the collector is outright rejecting the request due to the payload being too large.

2. Yes, I'll also add that in the DSN.

3. For my use case, I would want the metrics data loss to be minimal, I am okay to trade that with lesser throughput.

4. Agreed, I'll lower my queue size to 1000. Increased it just for experimenting.

5. There's no specific reason to it. I just got confused with codec compression in the blog post. Do you suggest I use the default lz4 ?

[SpencerTorres]

It is because the default request limit size is 20 MB which I am exceeding very quickly. To get around this, I've increase the limit to 200 MB.

Interesting. Where did you change this? I'm not familiar with such a limit, is this a config option in otel or somewhere on ClickHouse?

Just to be clear, when you say 800K/s, is it that so many data points are generated at your source every second?

The 800k/s metric was done on my local machine with 16-cores / 64GB RAM as a small stress test. I also had the clickhouse.sending_queue.num_consumers set to something like 30 to make sure there were plenty of goroutines to take the batches as they were ready. So yes the source was emitting this number of data points, but not from one process. I ran multiple processes that were using the file receiver component to spam ~1GB of data on a loop. And to be clear they were log rows not metrics, but I expect performance to be somewhat similar. I did eventually run into a wall with this, after 800k/s batches started dropping on the exporter. Not sure if this was because my ClickHouse server or because the exporter was out of consumers. I wasn't using ClickHouse Cloud though, so it didn't auto scale, it was just one node.

If yes, would it be fair to say that if your export interval from your agent / opentelemetry exporter running at source is 10 seconds, you will accumulate 8M data points?

Yes, in total end-to-end it generated 8M data points in a span of 10 seconds. This is after tuning the num_consumers and send_batch_size and send_max_batch_size. Batch size was around 9k and consumers was set to 30, but I'm not sure if 30 is a reasonable number (I was just trying to see how it affected throughput). I think if you have the cores for it, go for it. I did notice that changing batch size from 5k to 10k didn't have much effect on the throughput. I think it's all about the concurrency and how efficiently the exporter can schedule/push batches on each thread.

My case is that I've two services in my pipeline pushing datapoints. Assuming we are dealing with 20K RPS, the first service generate 20K datapoints per second, and the second 40K points per second which results to 60K datapoints per second accumulating in a second. My export interval being 10 seconds, it amounts to 600K datapoints pushed in each cycle to the collector. Once it reaches the collector, it immediately inserts to clickhouse since 600K breaches the batch processor threshold.

I think at some point you'll need to scale your collector since the concurrency tuning is not as simple as matching your send_batch_size to your throughput. Based on the IO limitations, I believe it would be better to send many smaller batches than few massive batches. The batches are still batches, but I'm not convinced that it should match your throughput 1:1. Especially with the default num_consumers. How much of that time is spent waiting for the server? I think it's a balance, and you'll have to tune it to find the best throughput.

But, I don't see inserts happening for 600K records, but it happens for around 100K-120K records in a batch, I am not sure on the reason though.

This makes sense based on my understanding of the batch processor. send_batch_size is the threshold for sending the batch, send_batch_max_size is the absolute limit a single batch can be, and timeout is when the batch will send if send_batch_size isn't reached. With steady throughput, I doubt your timeout is ever reached, if anything I think your send_batch_max_size is always being enforced which would result in batches showing with 100K records.

  batch:
    send_batch_size: 80000
    timeout: 10s
    send_batch_max_size: 100000

Since we are generating 60K metric points per second just at the source, I am not sure if we will be able to do inserts of 5-15k records. The only possible way would be to have the export interval at the source to be 250 ms which I feel is way too low. Do let me what you think about this. Just to be clear, clickhouse isn't throwing 4xx, but the collector is outright rejecting the request due to the payload being too large.

This is a good point, perhaps you can change the rate from the source as part of the tuning process. I think an important note for this is that the num_consumers affects the pipeline. Data isn't being transferred 1:1, it's being split between goroutines. This makes it faster, but also harder to balance. Even with the default num_consumers set to 10, it doesn't always behave like 10 actual consumers since some are waiting for batches, some are waiting to finish sending batches, etc.

I would definitely consider increasing the number of batches from the source. It doesn't need to be 250ms, but perhaps 10s is also unrealistic (you know your source better than I do though)

There's no specific reason to [changing the compression]. I just got confused with codec compression in the blog post. Do you suggest I use the default lz4 ?

LZ4 is the default, I think with compression_level set to 3. To verify this you can add it to your DSN. &compression=lz4&compression_level=3. We have a page for this on our docs, but ultimately it's based on your preferred tradeoff for CPU/network IO. I agree this is confusing since we also have column compression, which can be different per column on a table. But in this case this setting refers to transfer compression between client/server.

Summary

To summarize all of this, I would say that concurrency is the solution to this throughput problem. It's easy to think of it as batch_size = total_data_points_per_second, but when you factor in IO limitations and running multiple goroutines, it gets harder to balance these things intuitively. I think you solved your main problem which was the 4xxs, but I think there's room for optimization in the otel config to make things more efficient. With all of the variables (not only in the config, but in hardware + latency) it's hard to prescribe what exact config you need to use. The best I can suggest is where to look and what to consider.

Hopefully my responses have supplied enough information to begin tuning further. Let me know if I need to clarify any point further, or if you have more questions. I'm always interested in learning more about the exporter performance from users 👍

[ksharinarayanan]

Interesting. Where did you change this? I'm not familiar with such a limit, is this a config option in otel or somewhere on ClickHouse?

Yes, the limit is on the request that comes to collector itself https://github.com/open-telemetry/opentelemetry-collector/tree/main/config/confighttp#:~:text=max_request_body_size
It took some time for me as well to figure out that there is such a limit since there is no logging happening when the request is rejected due to high payload size.

it generated 8M data points in a span of 10 seconds

In your case, how do you then manage inserts of 5-15k batch size because just in a second, you're crossing it very easily. Even if you have 10 collector instances running, each of it would get 80K datapoints per second which is way higher than your threshold. What would be your average count of datapoints generated at your source and your expert interval because I believe at your rate, unless you have some large number of collector instances running, you'll never be able to insert at 15k batch size?

I believe it would be better to send many smaller batches than few massive batches

Agreed, but the problem is that I am using AWS managed flink which generally gives me a big single machine instead of multiple smaller machines and this results in the batch being generated from a single machine.

I'm not convinced that it should match your throughput 1:1

Absolutely, but since my batch is filling up fast in a single request, I've no other way to go. If I wait for 3-4 batches (30-40s), I will easily accumulate 2M records to be inserted to clickhouse in one shot. If I had the freedom of having multiple smaller machines instead of one big machine in AWS managed flink, I could have got around this problem by horizontally scaling the collector.

this setting refers to transfer compression between client/server

Understood, thanks for clarifying

I think my main problem comes from the fact that I'm unable to control the size of batches that I transmit from the managed flink service. Please let me know what you would suggest here in this case and also curious to understand on how you're keeping your batch size within 15k records.

[SpencerTorres]

In your case, how do you then manage inserts of 5-15k batch size because just in a second, you're crossing it very easily. Even if you have 10 collector instances running, each of it would get 80K datapoints per second which is way higher than your threshold. What would be your average count of datapoints generated at your source and your expert interval because I believe at your rate, unless you have some large number of collector instances running, you'll never be able to insert at 15k batch size?

The batches do max out at whatever your max batch size is, but it can do multiple batches per second. If each batch is ~10k then 800k records per second would be 80 batches per second, multiple batches are happening within a second. There's no need to balance the config so that 1 request is happening per second. The goal is to balance the CPU processing with the batch duration/latency so that every resource is being used to its fullest. ClickHouse Cloud can handle all of these batches perfectly fine, especially with async inserts

Agreed, but the problem is that I am using AWS managed flink which generally gives me a big single machine instead of multiple smaller machines and this results in the batch being generated from a single machine.

Absolutely, but since my batch is filling up fast in a single request, I've no other way to go. If I wait for 3-4 batches (30-40s), I will easily accumulate 2M records to be inserted to clickhouse in one shot. If I had the freedom of having multiple smaller machines instead of one big machine in AWS managed flink, I could have got around this problem by horizontally scaling the collector.

Not too familiar with the manged Flink config, but you're saying there's no way to do multiple consumers? Have you looked into the OTel Gateway architecture? https://opentelemetry.io/docs/collector/deployment/gateway

Perhaps it's more lightweight to load balance these requests then to send them directly to the exporter. Hypothetically you could receive the big batches from Flink and then load balance them to multiple collectors/exporters. This then lets you make tuning decisions based on your OTel Gateway's throughput vs your source's throughput

I think my main problem comes from the fact that I'm unable to control the size of batches that I transmit from the managed flink service. Please let me know what you would suggest here in this case and also curious to understand on how you're keeping your batch size within 15k records.

Definitely look into using the gateway/multi-collector approach. To clarify again, the batch limit is almost always hit, but it's okay to have multiple batches running at the same time. In simpler figures, if you have 100k data points per second with a max batch size of 10k, you should see 10 batches per second. If you factor in your concurrency configuration you may see even more than this depending on whether those batches are full or partially full (some may be 5k, some may be 10k).

[ksharinarayanan]

Thanks a lot for all the insights, will get back with more questions if I have any.