How to monitor the number of lines in logs per minute?

[mbork]

Question

Hi all,
I want to start using Vector. For the first thing I decided to monitor the number of log lines per minute (as a proxy for the general server activity – the idea is that if it drops or goes up suddenly, it most probably means trouble, like a DoS attack or unexpected downtime). Since my app is managed by pm2, this means pm2 logs. (This means I just want to know how many new lines appear in /home/me/.pm2/logs/my-app-out.log every minute.) With some help from an LLM, I came up with the following vector.toml (below).

Now I have two questions about it.

1. Is it a good approach? I have a gut feeling that it might be a bit overengineered – ~50 SLOC for a seemingly simple task.
2. Much more importantly, when I look in my database, it turns out that the events with the amount of log lines "per minute" are not really inserted once per minute. On average, they seem to be (more or less), but sometimes I get two rows in 30 seconds and sometimes no row for 90 seconds. I noticed that the distribution is fairly uniform (one row every ~60 seconds) when the load on the app is uniform, too (either very low or very high). What can be the reason?

Vector Config

[sources.real_logs]
type = "file"
include = ["/home/me/.pm2/logs/my-app-out.log"]
read_from = "end"

# Artificial events so that something will be recorded even in
# intervals with no log lines.
[sources.heartbeat]
type = "demo_logs"
format = "json"
interval = 30.0

# Wipe all data (. = {}) to ensure both events look identical (same
# tags) so they fall into the same aggregation bucket.
[transforms.normalize_counts]
type = "remap"
inputs = ["real_logs", "heartbeat"]
source = '''
  # Check if it came from a file before we wipe the data
  is_real_log = exists(.file)

  # Save the original timestamp
  ts = .timestamp

  # WIPE EVERYTHING. Removes random fields from demo_logs and file paths.
  . = {}

  # Restore the timestamp
  .timestamp = ts

  # Set the value: 1 for real logs, 0 for heartbeats
  .amount = if is_real_log { 1 } else { 0 }
'''

[transforms.count_lines]
type = "log_to_metric"
inputs = ["normalize_counts"]

  [[transforms.count_lines.metrics]]
  type = "counter"
  field = "amount"
  name = "pm2_logs_account"

  # IMPORTANT: This tells Vector to read the number (0 or 1)
  # instead of just counting +1 because the field exists.
  increment_by_value = true

[transforms.sum_per_minute]
type = "aggregate"
inputs = ["count_lines"]
# 60000ms = 1 minute
interval_ms = 60000

# This is needed because the `adapt_to_schema` transform would not
# work on metric events.
[transforms.metric_to_log]
type = "metric_to_log"
inputs = ["sum_per_minute"]

[transforms.adapt_to_schema]
type = "remap"
inputs = ["metric_to_log"]
source = '''
  .server = get_env_var("HOSTNAME") ?? "unknown"
  .value = .counter.value
  del(.kind)
  del(.counter)
'''

[sinks.database]
type = "postgres"
inputs = ["adapt_to_schema"]
endpoint = "[REDACTED]"
table = "[REDACTED]"

[sinks.console]
type = "console"
inputs = ["sum_per_minute"]
target = "stdout" 

  [sinks.console.encoding]
  codec = "json" # Outputs clear JSON structure

Vector Logs

No response