Metrics and Observability

Overview

Lucille uses the Codahale (Dropwizard) Metrics library for runtime observability. Every component — Publisher, Worker, Pipeline, Stage, and Indexer — reports metrics through a shared registry. These metrics are logged periodically during execution and summarized at the end of each run.

The Shared MetricRegistry

All components share a single registry:

MetricRegistry metrics = SharedMetricRegistries.getOrCreate(LogUtils.METRICS_REG);

Where LogUtils.METRICS_REG is the constant "default". Using SharedMetricRegistries means any component in the JVM can access the same registry without passing references around.

The Naming Convention

Metrics are namespaced using a metricsPrefix that encodes the run context:

String metricsPrefix = runId + "." + connector.getName() + "." + connector.getPipelineName();

This ensures metrics are collected separately for each connector/pipeline pair within a run. Individual components append their own suffix:

Stages use their own naming scheme within the Pipeline (see below).

Publisher Metrics

The Publisher tracks the rate and cadence of document publishing:

this.timer = SharedMetricRegistries.getOrCreate(LogUtils.METRICS_REG)
    .timer(metricsPrefix + ".timeBetweenPublishCalls");

In publish(), the timer measures the gap between consecutive calls:

if (timerContext.get() != null) {
    timerContext.get().stop();  // Stop timing the gap since last publish
}
try {
    publishInternal(document);
} finally {
    timerContext.set(timer.time());  // Start timing gap to next publish
}

This gives two insights:

• Mean rate — how fast the connector produces documents (docs/sec)
• Mean duration — average time between publish calls (ms/doc), indicating connector latency

The timer is ThreadLocal so multiple publishing threads don’t interfere with each other.

Periodic logging in waitForCompletion():

log.info(String.format(
    "%d docs published. One minute rate: %.2f docs/sec. Mean connector latency: %.2f ms/doc. Waiting on %d docs.",
    numReceived.get(), timer.getOneMinuteRate(), timer.getSnapshot().getMean() / 1000000, numPending()));

Worker Metrics

The Worker measures document processing time (pipeline latency):

Timer timer = metrics.timer(metricsPrefix + Worker.METRICS_SUFFIX);
// METRICS_SUFFIX = ".worker.docProcessingTme"

// In the processing loop:
Timer.Context context = timer.time();
Iterator<Document> results = pipeline.processDocument(doc);
// ... emit results ...
context.stop();

This timer captures the full pipeline execution time per document, including all stages.

Stage Metrics

Each Stage tracks its own processing time and counts. Metrics are initialized when the stage is added to a pipeline via stage.initialize(position, metricsPrefix):

The Stage base class provides:

• A Timer for per-document processing time
• An error Counter for documents that fail in this stage
• A child Counter for child documents generated by this stage

The logMetrics() method on Pipeline iterates all stages and logs their individual metrics.

Indexer Metrics

The Indexer tracks throughput and backend latency:

this.meter = metrics.meter(metricsPrefix + ".indexer.docsIndexed");
this.histogram = metrics.histogram(metricsPrefix + ".indexer.batchTimeOverSize");

After each batch is sent to the search engine:

stopWatch.reset();
stopWatch.start();
Set<Pair<Document, String>> failedDocPairs = sendToIndex(batchedDocs);
stopWatch.stop();

histogram.update(stopWatch.getNanoTime() / batchedDocs.size());  // Per-doc latency
meter.mark(batchedDocs.size());                                   // Throughput

The histogram records time per document (total batch time / batch size), giving a normalized view of backend performance regardless of batch size.

Periodic logging:

log.info(String.format(
    "%d docs indexed. One minute rate: %.2f docs/sec. Mean backend latency: %.2f ms/doc.",
    meter.getCount(), meter.getOneMinuteRate(), histogram.getSnapshot().getMean() / 1000000));

The WorkerPool Watcher Thread

The WorkerPool starts a scheduled watcher that runs every 500ms:

private ScheduledExecutorService startWatcher(List<Worker> workers, int maxProcessingSecs) {
    TimerTask watcher = new TimerTask() {
        private final Timer timer = metrics.timer(metricsPrefix + Worker.METRICS_SUFFIX);
        private Instant lastLogInstant = null;

        public void run() {
            // Periodic stats logging
            if (Duration.between(lastLogInstant, Instant.now()).getSeconds() >= logSeconds) {
                log.info(String.format(
                    "%d docs processed. One minute rate: %.2f docs/sec. Mean pipeline latency: %.2f ms/doc.",
                    timer.getCount(), timer.getOneMinuteRate(), timer.getSnapshot().getMean() / 1000000));

                // Heartbeat
                if (enableHeartbeat) {
                    heartbeatLog.info("Issuing heartbeat");
                }
            }

            // Stuck worker detection
            for (Worker worker : workers) {
                if (Duration.between(worker.getPreviousPollInstant().get(), Instant.now()).getSeconds() > maxProcessingSecs) {
                    log.error("Worker has not polled in " + maxProcessingSecs + " seconds.");
                    if (exitOnTimeout) {
                        System.exit(1);
                    }
                }
            }
        }
    };
}

The watcher serves three purposes:

1. Periodic Statistics Logging

Every logSeconds (default: 30, configurable via log.seconds), it logs pipeline throughput and latency. This gives operators real-time visibility into processing speed.

2. Stuck Worker Detection

Each Worker updates an AtomicReference<Instant> every time it polls for a new document:

// In Worker.run():
pollInstant.set(Instant.now());
doc = messenger.pollDocToProcess();

The watcher checks if any worker hasn’t polled within maxProcessingSecs (default: 600 seconds / 10 minutes, configurable via worker.maxProcessingSecs). A worker that hasn’t polled is likely stuck processing a single document.

3. The exitOnTimeout Mechanism

When worker.exitOnTimeout is true and a stuck worker is detected, the JVM exits immediately with System.exit(1). This is designed for containerized deployments where a hard restart (via Kubernetes pod restart) is preferable to a hung process.

The Heartbeat Mechanism

public static final String HEARTBEAT_LOG_NAME = "com.kmwllc.lucille.core.Heartbeat";
private static final Logger heartbeatLog = LoggerFactory.getLogger(HEARTBEAT_LOG_NAME);

if (enableHeartbeat) {
    heartbeatLog.info("Issuing heartbeat");
    if (heartbeatLog.isDebugEnabled()) {
        heartbeatLog.debug("Thread Dump:\n{}", 
            Arrays.toString(ManagementFactory.getThreadMXBean().dumpAllThreads(true, true)));
    }
}

When worker.enableHeartbeat is true, the watcher writes to a dedicated heartbeat logger. This can be configured (via logback) to write to a specific file that a Kubernetes liveness probe checks. If the file stops being updated, the probe fails and the pod is restarted.

At DEBUG level, it also dumps all thread stacks — useful for diagnosing what a stuck worker is doing.

End-of-Run Metrics Reporting

After all connectors complete, the Runner logs all collected metrics via Slf4jReporter:

Slf4jReporter.forRegistry(SharedMetricRegistries.getOrCreate(LogUtils.METRICS_REG))
    .outputTo(log)
    .withLoggingLevel(getMetricsLoggingLevel(config))
    .build()
    .report();

The logging level is configurable via runner.metricsLoggingLevel (default: DEBUG). This dumps every timer, meter, histogram, and counter in the registry.

The log.seconds Configuration

Controls how frequently periodic stats are logged:

this.logSeconds = ConfigUtils.getOrDefault(config, "log.seconds", LogUtils.DEFAULT_LOG_SECONDS);
// DEFAULT_LOG_SECONDS = 30

Used by:

• Publisher (in waitForCompletion)
• WorkerPool watcher
• Indexer (in sendToIndexWithAccounting)

Setting this lower gives more frequent visibility; setting it higher reduces log noise.

MDC (Mapped Diagnostic Context) Usage

Lucille uses SLF4J’s MDC to attach contextual information to every log line:

run_id

Set at the start of each thread’s work:

MDC.put("run_id", runId);  // In ConnectorThread
MDC.put(RUNID_FIELD, localRunId);  // In Worker
MDC.pushByKey(RUNID_FIELD, localRunId);  // In Indexer (stack-based for multi-run)

This allows log aggregation tools to filter all log lines for a specific run.

doc_id

Set when processing a specific document:

MDC.put(Document.ID_FIELD, document.getId());  // In Publisher.publish()

try (MDC.MDCCloseable docIdMDC = MDC.putCloseable(ID_FIELD, doc.getId())) {
    // In Worker and Indexer — auto-removed when block exits
    docLogger.info("Worker is processing document {}.", doc.getId());
}

The DocLogger (logger name com.kmwllc.lucille.core.DocLogger) is a dedicated logger for document lifecycle events. Combined with MDC, you can trace a single document’s journey through the entire system.

Indexer MDC Stack

The Indexer uses pushByKey/popByKey for run_id because in Kafka distributed mode, documents from different runs might be interleaved:

if (d.getRunId() != null) {
    MDC.pushByKey(RUNID_FIELD, d.getRunId());
}
// ... send event ...
if (d.getRunId() != null) {
    MDC.popByKey(RUNID_FIELD);
}

Summary of Configurable Observability Settings