Fresh logs first, even during a log spike

When an indexer falls far behind its Kafka partition, KalDB doesn’t make the newest logs wait behind the backlog. The indexer jumps to the newest data, and a separate pool of recovery nodes backfills the skipped range in parallel.

How KalDB keeps fresh logs searchable when an indexer falls behindLogstash sends logs to the preprocessor, which appends them to a Kafka partition. During a log spike the partition builds a backlog. A new indexer skips to the head of the partition, writes a recovery task for the skipped offsets, and a recovery node backfills those offsets into S3 as snapshots. Cache nodes download the snapshots, and a user searching from OpenSearch Dashboards sees every offset again.Logstashor any clientPreprocessorvalidate · route10× log spikeKafka partition 1 · the write-ahead loghead12…99100upload snapshotupload snapshotsdownloadsearchRecovery nodetask assignedindexing 2–99done · free againCluster manager+ ZooKeeperIndexerNew indexerindexing offset 2crashedchecking its lag…indexing offset 100S312–4950–99Cache12–4950–99QueryOpenSearchDashboardsdurable offset: 1 · head: 100lag 99 > maxOffsetDelayMessagesRecovery taskpartition 1 · offsets 2–99
Freshness Newest logs are searchable in near real time. Falling behind: the newest logs are stuck at the back of the backlog. Still behind: replaying from offset 2 would index the whole backlog first. Newest logs are searchable again. Offsets 2–99 are queued for backfill. Newest logs are searchable while offsets 2–99 are backfilled. Every offset is searchable again.
  1. 1. Steady state. Logstash (or any client) sends logs to the preprocessor, which appends them to a Kafka partition. The indexer pulls them from the partition, indexes them into Lucene, and uploads each finished chunk to S3 as a snapshot. Cache nodes download snapshots to serve older data, and searches from OpenSearch Dashboards reach both through the query node.
  2. 2. A log spike hits. Log volume from clients jumps 10× and the indexer crashes or is restarted. Kafka keeps accepting writes, so nothing is lost, but a backlog builds up in front of the newest logs.
  3. 3. A new indexer checks its lag. On startup it compares the highest offset already durable in S3 (1) with the head of the partition (100). The gap is larger than maxOffsetDelayMessages.
  4. 4. Skip to the head. Instead of replaying the backlog, the indexer writes a recovery task for offsets 2–99 to ZooKeeper and starts indexing at offset 100. Fresh logs are searchable again right away.
  5. 5. Hand off the backlog. The cluster manager assigns the recovery task to a free recovery node. Recovery nodes are a separate pool that scales on its own.
  6. 6. Backfill in parallel. The recovery node reads offsets 2–99 from Kafka in parallel, builds Lucene chunks, and uploads them to S3 as snapshots. It doesn't serve queries, so all of its capacity goes to catching up.
  7. 7. Gap closed. Cache nodes download the new snapshots from S3, and searches cover every offset again. The recovery task is removed and the node returns to the free pool.

The problem

Log volume is spiky. A bad deploy, a retry storm, or a stalled ingest pipeline can push 10× normal volume for minutes at a time, and those are exactly the moments when engineers are searching logs hardest.

Most search clusters ingest first-in, first-out. When indexing falls behind, the backlog drains in order, so the newest logs, the ones you need during an incident, are the last to become searchable. Teams work around this with rate limits, sampling, and capacity provisioned for peak traffic. These reduce the damage, but they still lose data or add lag, and the peak capacity sits idle the rest of the time.

How it works

Each KalDB indexer reads one Kafka partition. Kafka keeps unindexed data durable and S3 keeps indexed data durable, so an indexer can choose where to start reading without losing anything.

  1. The indexer finds its durable position. When it starts, the indexer looks up the highest offset for its partition that is already uploaded to S3 or already claimed by a pending recovery task.
  2. It measures the gap to the head of the partition. If the gap is within maxOffsetDelayMessages, the indexer resumes from the next offset as usual.
  3. If the gap is larger, it skips to the head. The indexer writes recovery tasks for the skipped range to ZooKeeper and starts indexing at the current head. Large ranges are split into several tasks of up to maxMessagesPerChunk messages each, so backfilled chunks are the same size as normal ones.
  4. The cluster manager assigns the tasks. Recovery nodes wait in a free pool. The manager hands them tasks, oldest first.
  5. Recovery nodes backfill the range. Each node reads its offsets from Kafka in parallel, builds Lucene chunks, uploads them to S3, and publishes their metadata. When the upload succeeds, it deletes the task and returns to the free pool. Recovery nodes don’t serve queries, so all of their capacity goes to catching up.
  6. Cache nodes serve the backfilled data. They load the new chunks from S3, and queries cover the full range again.

The same mechanism covers several situations:

  • An indexer crashes or is redeployed and comes back far behind.
  • An indexer falls behind during a log spike and is restarted.
  • A new cluster starts reading from the latest offset and backfills data already in Kafka (see createRecoveryTasksOnStart).

Why it matters

  • Fresh data during incidents. The newest logs are indexed first, when they matter most.
  • No provisioning for peak. Indexers are sized for normal traffic. Extra capacity is only needed to catch up, and recovery nodes supply it on demand.
  • No dropped data to protect freshness. The backlog stays in Kafka and is indexed later instead of being sampled away or dropped.
  • One code path for failures and spikes. Crash recovery and spike handling use the same mechanism, so there is one thing to operate and monitor instead of two.

Trade-offs

  • The skipped range becomes searchable later. Until recovery finishes and cache nodes load the new chunks, searches over that time range don’t include the skipped logs yet.
  • Kafka retention has to cover the backlog. Recovery nodes read the skipped offsets from Kafka. Offsets that Kafka has already deleted can’t be recovered. Recovery nodes count them in records_no_longer_available_total.
  • The check runs when an indexer starts. A running indexer that falls behind keeps consuming in order. To skip ahead, restart it.
  • Recovery needs spare capacity. Keep at least one recovery node running at all times, and autoscale the pool on CPU utilization. See Recovery for deployment guidance.

Configuration 

indexerConfig:
  # Lag (in messages) beyond which a starting indexer skips to the head
  # and creates recovery tasks for the gap.
  maxOffsetDelayMessages: 300000
  # Also caps the number of messages in each recovery task.
  maxMessagesPerChunk: 100000
  # New clusters only: start at the head and backfill what is already in Kafka.
  readFromLocationOnStart: LATEST
  createRecoveryTasksOnStart: true

To watch recovery in production, track these metrics from the metrics reference:

  • recovery_tasks_created_total: how often indexers skip ahead.
  • recovery_tasks_assigned_total: tasks handed to recovery nodes.
  • recovery_tasks_insufficient_capacity_total: tasks waiting because no recovery node is free. If this keeps growing, add recovery nodes.

Related