If buffer event in memory or disk will write to sink before k8s pod shutdown?

[linnaname]

If buffer event in memory or disk will write to sink before k8s pod shutdown?

[jszwedko]

Hi @linnaname !

When using in-memory buffers, Vector will attempt to flush them before shutting down. There is a grace period of 60 seconds. After that, Vector will shutdown and drop any events that were still in-memory.

When using disk buffers, events are buffered to disk as the sink receives them. When shutting down, all events in-flight (in memory) from the source to the sink should be written to the disk buffer. The disk buffer will also attempt to flush events to the sink for up to 60 seconds before fulling shutting down.

Hope this helps! Let me know if you have any additional questions.

[linnaname]

I'm using vector in production environment. Kafka as source and ElasticSearch as sink, more 40 million Kafka message every day.

After vector deploy we found vector consumer Kafka message too fast,then write to ElasticSearch too fast. It caused ElasticSearch on the brink of collapse. I try to set options config for Kafka but not work well.

I wonder if there is a way to store some event in vector k8s pod in disk(or memory) to buff it and slow down the speed write to ElasticSearch?

I want maximize the use of memory and the disk for vector k8s pod, not ElasticSearch resource

[sources.kafka_log]
type = "kafka"
bootstrap_servers = "xxxx"
group_id = "xxx-vector"
auto_offset_reset = "largest"
topics = ["xxxxxx"]
librdkafka_options."fetch.max.bytes" = "1048576"


some transforms

[sinks.events_to_es]
type = "elasticsearch"
inputs = [ "xxx" ]
endpoint = "$xxxx"
index = "xxxx"
mode = "normal"
bulk_action = "index"
doc_type = "logs"
compression = "none"
batch.max_events = 100
batch.timeout_secs = 3
buffer.type = "memory"
buffer.when_full = "block"
encoding.except_fields = [xxx]
request.timeout_secs=10

[linnaname]

@jszwedko

[jszwedko]

Hi @linnaname ! You could try limiting the number of concurrent requests in the Elasticsearch sink by setting request.concurrency = X. This will have Vector limit itself to the specified number of concurrent requests and apply back-pressure when saturated.