Kafka sink metrics to track network traffic

[almarcotte]

Hello!

We're using the Kafka sink to ship logs to Kafka and we're trying to keep track of network traffic this generates. Looking at the documentation there are a couple of different options:

• kafka_produced_messages_bytes_total
• component_sent_event_bytes_total

The values for component_sent_event_bytes_total seem off though. Additionally, when comparing it to component_received_event_bytes_total it seems like the delta between them is constantly growing:

Why is there such a discrepancy between received and sent? I understand adding metadata and everything but that should be fairly constant, no? Looking here https://github.com/vectordotdev/vector/blob/master/src/sinks/kafka/service.rs#L110 but I might be missing something.

Another odd thing we're seeing: the kafka_produced_messages_bytes_total metric doesn't have a component_id label even though the documentation says it should. It's not a problem now since we have a singular use-case to send to Kafka from this particular instance of Vector but that might not always be the case at which point we wouldn't be able to tell the difference.

[jszwedko]

Hey! Can you share the configuration you are running as well as Vector version?

I would expect kafka_produced_messages_bytes_total to not match since that is a librdkafka level metric that measures differently than Vector measures component_sent_event_bytes_total. It should have a comoponent_id though. Would you mind opening a bug report for that?

[almarcotte]

For sure!

$ vector --version
vector 0.28.0 (x86_64-unknown-linux-gnu 971c594 2023-02-27)

Here's our configuration (with removed paths/endpoints/credentials) without some of the source fields which mostly just do labelling.

data_dir = "/path/to/var/lib/vector"

# Log collector
[sources.log_reader]
type = "file"
include = [ "/path/to/logs/**/*.log" ]
read_from = "end"
ignore_older_secs = 600
ignore_not_found = true
fingerprint.strategy = "device_and_inode"

# Make sure logs are properly tagged so they can be filtered
[transforms.log_reader_tagged]
type = "remap"
inputs = [ "log_reader" ]
source = '''
# we do some basic mapping here to make sure the logs are properly tagged
'''

[sinks.publish_to_kafka]
type = "kafka"
inputs = [ "log_reader_tagged" ]
bootstrap_servers = "kafka-bootstrap-servers.example.com:9092"
topic = "our-kafka-topic"
compression = "gzip"
encoding.codec = "json"
sasl.enabled = true
sasl.username = "kafka-username"
sasl.password = "kafka-password"
sasl.mechanism = "SCRAM-SHA-512"
librdkafka_options."message.max.bytes" = "1000000"
librdkafka_options."security.protocol" = "sasl_ssl"

[sources.internal_metrics]
type = "internal_metrics"
namespace = "vector"
scrape_interval_secs = 60

# Keep only the metrics we're interested in, i.e. for our components but also anything with a kafka_ prefix since for
# some reason those don't have a component_id label.
[transforms.logs_metric_clean]
type = "filter"
inputs = [ "internal_metrics" ]
condition = 'includes(["log_reader", "publish_to_kafka"], .tags.component_id) || starts_with(to_string(.tags.name) ?? "default", "kafka_")'

# Add labels to the metrics
[transforms.labelled_metrics]
type = "remap"
inputs = [ "logs_metric_clean" ]
source = """
	.app = "vector"
	# more labeling here specific to this machine
"""

# Sink metrics to Prometheus remote write endpoint
[sinks.log_reader_prometheus_remote_write]
type = "prometheus_remote_write"
inputs = [ "labelled_metrics" ]
tenant_id = "my-tenant-id"
endpoint = "https://prometheus-remote-writer.example.com"
default_namespace = "vector"
buffer.type = "memory"
buffer.when_full= "block"
request.retry_attempts = 3
request.retry_initial_backoff_secs = 5
request.rate_limit_num = 5
tls.ca_file = "/path/to/certs/vector-ca.pem"
tls.crt_file = "/path/to/certs/client.pem"
tls.key_file = "/path/to/certs/client-key.pem"
tls.verify_certificate = false
healthcheck.enabled = false

I'm still curious about the increasing difference between component_sent_event_bytes_total and component_received_event_bytes_total on the sink (especially compared to other components). Is there some explanation to that. Here's a graph of the following:

abs(sum(vector_component_sent_event_bytes_total{component_id="file_reader"}) - sum(vector_component_received_event_bytes_total{component_id="file_reader"}))

and

abs(sum(vector_component_sent_event_bytes_total{component_id="kafka_sink"}) - sum(vector_component_received_event_bytes_total{component_id="kafka_sink"}))

[jszwedko]

Thanks!

The source I would actually expect a difference since it adds fields after "receiving" an event.

The sink is a little more confusing since it shouldn't be adding data before sending. It can remove data if the encoding.(only|except)_fields options are used but you don't seem to be using those. If you are able to get this down to an MRE and open a bug report that would help with tracking it down.