Will vector cause disk IO to be full?

[FriskKiddo]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

I ingested log data from machine files into kafka through vector. My kafka cluster was unavailable for a period of time. After my kafka cluster was deployed successfully and became available, my machine disk read IO suddenly increased.
Is there any way to avoid this problem with vector?

Configuration

[sources.${sourceName}]
type = "file"
include = ["/home/admin/xxxx/logs/biz/**/application.log*"]
data_dir="${offsetFilePath}"
read_from="end"

[transforms.${transformName}]
type = "remap"
inputs = ["${sourceName}"]
source = '''
del(.timestamp)
del(.source_type)
.__file__=del(.file)
.__host__=del(.host)
.__raw_log__ = del(.message)
.__project__ = "hammal-image-scan"
.__dt__ = format_timestamp!(now(),"%Y-%m-%d %H:%M:%S","local")
.__level__ = "info"
."string.values", err = split(.__raw_log__, "|", limit: 100)
."string.keys" = ["log_time","sync","region","request_id","uid","request_from","tenant_id","client_ip","sdk_version","cfg_version","user_type","user_id","user_nick","avatar","imei","imsi","umid","ip","os","channel","host_app_name","host_package","host_version","biz_type","scenes","task_id","data_id","time","url","new_url","code","msg","total_time","download_time","algo_time","results","algo_results","hit_custom_keywords","hit_system_keywords","hit_custom_similar_texts","hit_system_similar_texts","hit_custom_images","hit_system_images","task_biz_type","channel_uid","profile_uid","business_name","first_product_name","ai_service_codes","hit_rule_ids","user_data","event_id","event_code","column_code","src_id","gray_flag_4_green","aigc_biz"]
del(.parseMsg)
.parse_log = {}
for_each(."string.keys") -> |_index, value| {
    single_value, err = get(."string.values", path: [to_int(_index)])
    .parse_log = set!(value:.parse_log, [value], single_value)
}
. |=.parse_log
del(.parse_log)
del(."string.keys")
del(."string.values")
'''

[sinks.${sinkName}]
type = "kafka"
inputs = ["${transformName}"]
bootstrap_servers = "${kafkaServers}"
topic = "${kafkaTopic}"
encoding.codec = "json"
buffer.type="disk"
buffer.max_size=268435488
buffer.when_full="block"
acknowledgements.enabled=true

Version

vector 0.31.0 (x86_64-unknown-linux-gnu 0f13b22 2023-07-06 13:52:34.591204470)

Debug Output

No response

Example Data

No response

Additional Context

No response

References

No response

[jszwedko]

Hi @FriskKiddo ,

If Kafka was offline, then disk buffer you defined for the kafka sink would fill up. Once Kafka recovers, it is expected that you would see a large number of reads to process the buffered data.

If the disk buffer became full, then the file source would have also paused, and then start reading again once it recovered. This could also cause a large number of reads.

Why do you want to avoid this?

[FriskKiddo]

Hi @FriskKiddo ,

If Kafka was offline, then disk buffer you defined for the kafka sink would fill up. Once Kafka recovers, it is expected that you would see a large number of reads to process the buffered data.

If the disk buffer became full, then the file source would have also paused, and then start reading again once it recovered. This could also cause a large number of reads.

Why do you want to avoid this?

Thank you @jszwedko
This behavior caused my machine's disk read IO usage to be very high, which further caused my machine's disk write problems. At the same time, the load value of this machine was also very high. The reason for this phenomenon may be that the log files I read have very large writes.