AWS CloudWatch metric sink not cleaning up buffer files consistently

[johannesfloriangeiger]

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

Preamble: To reproduce the below I suggest to edit lib/vector-buffers/src/variants/disk_v2/common.rs and set DEFAULT_MAX_DATA_FILE_SIZE to 10 KB (10 * 1024) instead of 128 MB to reduce test times.

Steps to reproduce:

Obtain an AWS account and set up local access

Consider the following vector.toml:

data_dir = "."

[sources.static_metrics]
type = "static_metrics"
metrics = [
  { name = "test_metric_0", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_1", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_2", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_3", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_4", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_5", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_6", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_7", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_8", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_9", kind = "absolute", tags = { }, value.gauge.value = 42 }
]

[sinks.console]
type = "console"
inputs = ["static_metrics"]
encoding.codec = "json"
buffer.type = "disk"
buffer.max_size = 2097184

[sinks.aws_cloudwatch_metrics]
type = "aws_cloudwatch_metrics"
inputs = ["static_metrics"]
default_namespace = "default"
healthcheck.enabled = true
buffer.type = "disk"
buffer.max_size = 2097184

Run vector with ./target/release/vector --config ./vector.toml and observe that the buffer-files (buffer-data-x.dat) in buffer/v2/aws_cloudwatch_metrics rotate every few seconds, i.e. once the buffer is full and flushed the file gets deleted.

Now consider the following, slightly different vector.toml that also includes an aggregated histogram metric :

data_dir = "."

[sources.static_metrics]
type = "static_metrics"
metrics = [
  { name = "test_metric_0", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_1", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_2", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_3", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_4", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_5", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_6", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_7", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_8", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_9", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_10", kind = "absolute", tags = { }, value.aggregated_histogram.buckets = [{ le = 10, count = 1, upper_limit = 42.0 }], value.aggregated_histogram.count = 42, value.aggregated_histogram.sum = 42 },
]

[sinks.console]
type = "console"
inputs = ["static_metrics"]
encoding.codec = "json"
buffer.type = "disk"
buffer.max_size = 2097184

[sinks.aws_cloudwatch_metrics]
type = "aws_cloudwatch_metrics"
inputs = ["static_metrics"]
default_namespace = "default"
healthcheck.enabled = true
buffer.type = "disk"
buffer.max_size = 2097184

and when running Vector you will notice that the buffer files are rotating but not cleaned up anymore.

A workaround is to filter for the Metric types that the AWS CloudWatch metric sink supports, e.g. such as the following vector.toml:

data_dir = "."

[sources.static_metrics]
type = "static_metrics"
metrics = [
  { name = "test_metric_0", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_1", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_2", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_3", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_4", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_5", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_6", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_7", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_8", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_9", kind = "absolute", tags = { }, value.gauge.value = 42 },
  { name = "test_metric_10", kind = "absolute", tags = { }, value.aggregated_histogram.buckets = [{ le = 10, count = 1, upper_limit = 42.0 }], value.aggregated_histogram.count = 42, value.aggregated_histogram.sum = 42 },
]

[sinks.console]
type = "console"
inputs = ["static_metrics"]
encoding.codec = "json"
buffer.type = "disk"
buffer.max_size = 2097184

[transforms.metrics_filter]
inputs = ["static_metrics"]
type = "filter"
condition = '.type == "gauge"'

[sinks.aws_cloudwatch_metrics]
type = "aws_cloudwatch_metrics"
inputs = ["metrics_filter"]
default_namespace = "default"
healthcheck.enabled = true
buffer.type = "disk"
buffer.max_size = 2097184

Configuration

Version

vector 0.54.0 (aarch64-apple-darwin)

Debug Output

Example Data

No response

Additional Context

No response

References

No response

[johannesfloriangeiger]

A similar behavior can be observed when processing Counter metrics:

data_dir = "."

[sources.static_metrics]
type = "static_metrics"
metrics = [
    { name = "test_metric_0", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_1", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_2", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_3", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_4", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_5", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_6", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_7", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_8", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_9", kind = "absolute", tags = { }, value.counter.value = 42 },
]

[sinks.aws_cloudwatch_metrics]
type = "aws_cloudwatch_metrics"
inputs = ["static_metrics"]
default_namespace = "default"
healthcheck.enabled = true
buffer.type = "disk"
buffer.max_size = 2097184

I assume as Vector creates a new metric based on the relative increments and ignores the original metric it's not acknowledged to be Delivered and thus kept in the buffer.

Ideally as per https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/ContainerInsights-Prometheus-metrics-conversion.html Vector should do the same as the CW agent: "The Prometheus metrics with the unsupported histogram metric type are dropped by the CloudWatch agent." and also treat the original metric as being delivered.

Filtering for incremental Counter metrics works again, the buffer is rotating (as I assume the incremental counters are passed through to the sink as they are):

data_dir = "."

[sources.static_metrics]
type = "static_metrics"
metrics = [
    { name = "test_metric_0", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_1", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_2", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_3", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_4", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_5", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_6", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_7", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_8", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_9", kind = "absolute", tags = { }, value.counter.value = 42 },
    { name = "test_metric_10", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_11", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_12", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_13", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_14", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_15", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_16", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_17", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_18", kind = "incremental", tags = { }, value.counter.value = 42 },
    { name = "test_metric_19", kind = "incremental", tags = { }, value.counter.value = 42 },
]

[transforms.metrics_filter]
inputs = ["static_metrics"]
type = "filter"
condition = '.type == "counter" && .kind == "incremental"'

[sinks.aws_cloudwatch_metrics]
type = "aws_cloudwatch_metrics"
inputs = ["metrics_filter"]
default_namespace = "default"
healthcheck.enabled = true
buffer.type = "disk"
buffer.max_size = 2097184

[johannesfloriangeiger]

After some more investigations I have to correct myself - the reason the "aggregated histogram" metric causes the buffer not to be cleared seems not to stem from the fact that it's an unsupported metric type but from the fact that it's an "absolute" metric, which gets transformed into an "incremental" metric in the normalizer.