Where Were the Logs? Building a Log Ingestion Engine That Never Silently Drops a Log

1. The problem

There are twelve services, each producing bespoke log files onto individual disks. There is no central search, no live dashboard, and no audit trail.

It can take an hour to determine which service (or machine), or which file had the issue. By using grep on a laptop, it can be found in seconds. However, using grep on all twelve different formats takes an entire afternoon.

This resulted in three requirements:

• A central search of all log files.
• Signals in minutes about which service has an issue and how big the issue is.
• An audit trail that is persistent and no data is lost when restarting the service.

The log ingestion engine has been built to provide some of the searchable requirements (see section 7).

Three hours were lost in this example scenario. Two hours were spent on root cause analysis (where an engineer would log onto each server, open up various log files, and search manually).

There was no error reported from any of the applications on the monitoring dashboard, meaning that no panels highlighted the service that was failing. There was also no clear indication of which service was failing the most. The compliance team wanted to see audit trails, but there were none to provide.

The brief is the running example I built this project from, and I later published the project as a repository.

TL;DR: Go service which receives batched JSON on /logs, validates at field level, enriches and routes to three SQLite destinations via RabbitMQ with a bounded in-memory fallback. Admission is durable via a write-ahead spool fsynced before 202; unacked spool entries are replayed on startup. Batch writer with bounded retries and rotating dead-letter file, idempotent writes per log_id. Load test sustained 1,000 logs/s acceptance with 12 pipeline workers, batch 100, on a laptop; default 4 workers do not sustain the target.

2. The goal and the ground rules

Overview

A centralised logging platform to receive logs using an API, validate and enrich log data, route by service, store it reliably, and provide signals so we can see issues occur before our customers do.

Constraints

The brief says we must limit our libraries. I made a few changes here:

• Used the rate-limiting library golang.org/x/time/rate rather than implementing my own token bucket. The interesting engineering was implementing per-IP keys, request shaping, and Retry-After.
• The frontend uses React with Vite, Tailwind CSS as a utility CSS framework, Space Grotesk from the Google Fonts CDN, and Chart.js for charts, instead of hand-written HTML/CSS.
• Storage is kept within the brief: Three databases using SQLite. One file per destination.
• The configuration is simple with only .env files, integer values for duration, and ROUTING_RULES.

The cost of using SQLite on real logs is discussed in section 7.

3. Architecture at a glance

This diagram represents the architecture shipped with the project:

flowchart TB
client(["Clients"])
subgraph ingress["HTTP ingress"]
direction TB
rateLimit["Rate limiter<br/>per IP, in-memory"]
logsEndpoint["POST /logs"]
validate["Validate batch<br/>body size + per-entry errors"]
end
spool[("Write-ahead spool<br/>fsync before ack")]
rawQueue[("raw_logs<br/>bounded channel")]
subgraph processing["Pipeline worker pool"]
direction TB
enrich["Enrich<br/>received_at · source_ip · environment"]
route["Route<br/>service → destination rules"]
end
subgraph publishing["Publisher"]
direction TB
rabbitPublisher["RabbitMQ publisher<br/>persistent + confirms<br/>5s publish timeout<br/>auto-reconnect with backoff"]
fallback[("Bounded in-memory fallback<br/>used while disconnected")]
end
broker[("RabbitMQ<br/>exchange_service1..3<br/>queue_service1..3<br/>routing key: log.write")]
subgraph delivery["Consumers and storage"]
direction TB
supervisor["Supervisor<br/>one connection<br/>declares topology, reconnects"]
amqpConsumer["AMQP consumers<br/>one connection per queue<br/>ack after stored"]
fallbackConsumer["Fallback consumer"]
batchWriter["Batch writer<br/>per-destination buffers<br/>retry with backoff"]
sqlite[("SQLite<br/>logs_service1.db<br/>logs_service2.db<br/>logs_service3.db")]
deadLetters[("Dead-letter NDJSON<br/>one shared file<br/>rotated by size")]
replay["cmd/replay"]
end
client --> rateLimit --> logsEndpoint --> validate --> spool --> rawQueue
rawQueue --> enrich --> route --> rabbitPublisher
rabbitPublisher -->|"connected"| broker
rabbitPublisher -->|"publish fails"| fallback
fallback -.->|"fallback full: dead letter"| deadLetters
broker -->|"deliver"| supervisor
supervisor --> amqpConsumer
fallback -->|"stored directly, never re-published"| fallbackConsumer
amqpConsumer --> batchWriter
fallbackConsumer --> batchWriter
batchWriter -->|"stored"| sqlite
batchWriter -->|"retries exhausted"| deadLetters
deadLetters -.-> replay
replay -.->|"replay"| sqlite

There are 6 stages within the pipeline:

flowchart LR
post["1 Post<br/>rate limit,<br/>accept the batch"]
validate["2 Validate and spool<br/>per-entry errors,<br/>fsync admission"]
buffer["3 Buffer<br/>bounded raw_logs<br/>channel (10,000)<br/>202 or 503"]
enrichRoute["4 Enrich and route<br/>stamp fields,<br/>pick destination"]
publish["5 Publish<br/>RabbitMQ, or the<br/>in-memory fallback"]
store["6 Consume and store<br/>batched SQLite writes,<br/>dead letters on failure"]
post --> validate --> buffer --> enrichRoute --> publish --> store

• Post
• Validate
• Buffer
• Enrich/Route
• Publish/Fallback
• Consume/Store

Each stage handles one specific type of failure/slowdown. The rate limiter handles abusive clients. raw_logs handles separating quick HTTP traffic from slow enrichment. The publisher fallback handles outages of the broker.

The logs are each written to storage once, as persistence is idempotent on a per log_id basis. They are never ping-ponged between the broker and memory. The batch writer spreads the cost of inserting into SQLite. The dead letter file handles unrecoverable failure.

The fallback is consumed by a dedicated consumer and written directly to storage. No re-publishing occurs. As consumers only acknowledge once data has been stored, it provides at-least-once delivery. Persistence within the database is idempotent on a per log_id basis, so duplicate data can be handled.

4. The stages

4.1 Ingestion and validation

This will validate the Content-Type header, ensure the body is a certain size using http.MaxBytesReader, parse the JSON, and ensure that there are four fields for each log entry. These are:

• Timestamp (RFC3339)
• Service (at most 100 characters)
• Level (INFO/WARN/ERROR/DEBUG)
• Message (at most 10,000 characters)

Validation runs first. Valid entries are appended to the write-ahead spool and fsynced to disk. The handler then attempts to enqueue the batch onto the bounded raw_logs channel. If the enqueue succeeds within CHANNEL_SEND_TIMEOUT, the handler returns 202 with the batch ID and per-entry rejection details; if the channel is full or the send times out, the handler returns 503.

The 202 therefore means the batch is durably admitted for replay and now being processed, not yet in SQLite. The spool guarantees the batch can be replayed on restart. In the 503 case, the spool entry is retained so it can be replayed later; the client may retry and create duplicates, which receive new log_ids and are therefore stored as separate rows (idempotency does not collapse them).

Design decision: I decided to accept partial batches. This will mean that if a log has an invalid time, it will only lose one log instead of one hundred.

Each valid log entry will have a UUID added as the log_id field. This means they can be safely resent and played back.

Design decision: The spool fsync is performed inline on the request path to make a 202 durable. This breaks the “no blocking” ideal and adds latency on the critical path, but it guarantees admission durability. The fsync is small per batch, and the cost is accepted for durability; compaction of the spool log and .acked file occurs only when every spooled ID has been acknowledged.

func ValidateLogEntries(logEntries []model.LogEntry) (validEntries []model.LogEntry, rejectedEntries []model.RejectedEntry) {
  for entryIndex, logEntry := range logEntries {
    fieldErrors := validateLogEntry(logEntry)
    if len(fieldErrors) > 0 {
      rejectedEntries = append(rejectedEntries, model.RejectedEntry{
        Index: entryIndex,
        Errors: fieldErrors,
      })
    } else {
      validEntries = append(validEntries, logEntry)
    }
  }
  return validEntries, rejectedEntries
}

Valid and invalid log entries will be separated by ValidateLogEntries whilst retaining their original index. Every log entry needs to have a time, service, level, and message.

POST /logs can return 202, 400, 413, 415, 503, or a 429 from the limiter.

4.2 Protecting the boundary

Rate limiting middleware using a lazy in-memory token bucket, keyed on the source IP address. Token buckets are purged explicitly when idle. Rejected requests will not consume a token.

Design decision: The AllowWithoutConsumingCapacity method is used. This method checks how much delay is required. If the delay is positive, the reservation is cancelled, and the request is rejected before any work is done, returning 429 with a Retry-After header in whole seconds. The rejected request keeps its token.

reservation := entry.limiter.Reserve()
delay := reservation.Delay()
if delay > 0 {
  reservation.Cancel()
  return delay, false
}

Rate limiting middleware uses a token bucket for each IP address. 429 status code is returned with a delay in whole seconds. No token is consumed on rejection.

X-Forwarded-For is trusted only when a trusted proxy is configured; with no proxy list, the socket address is used for identity.

4.3 The raw_logs buffer

I have a bounded channel that defaults to 10,000 entries for RAW_LOGS_BUFFER. Validations happen quickly, but enrichments happen at their own pace.

Design decision: Producers wait for a bounded 100 ms rather than fail instantly. PushBatch rejects oversized batches immediately, then takes the producer lock and waits up to CHANNEL_SEND_TIMEOUT 100 ms for room for the entire batch. This is blocking with a bounded timeout, providing backpressure.

for len(rawLogs) > cap(rawLogsQueue.channel) - len(rawLogsQueue.channel) {
  select {
  case <- capacityCheck.C:
  case <- sendContext.Done():
    return ErrQueueOverloaded
  }
}

Writing straight to SQL would be on the slow path. Using the bounded channel means they are separated. I use back pressure by returning a 503.

4.4 Enrichment

Adds received_at, source_ip, and environment (defaulting to production) from config. Also adds log_id, batch_id, and retry_count (starting at 0) to every log. No I/O operation, so no failure case.

Design decision: All data is captured at the edge as it is based on request context. This means I only have access to the socket address, forwarded chain, and received time. These are stamped and then passed through. If any of these fields are missing, a sensible default is used.

return model.EnrichedLog{
  LogEntry: rawLog.LogEntry,
  LogID: logEnricher.logIDOrDefault(rawLog.LogID),
  BatchID: rawLog.BatchID,
  ReceivedAt: receivedAt,
  SourceIP: sourceIP,
  Environment: logEnricher.applicationEnvironment,
  RetryCount: 0,
}

received_at uses the handler clock; source_ip uses the socket address or the first valid X-Forwarded-For header from behind a trusted proxy. All of these are captured at the edge.

4.5 Routing

The destination is stamped based on a validated mapping from the .env file using the ROUTING_RULES=claims=service2,billing=service3 format. If an unknown service is provided, it will default to service1. The config is loaded using only godotenv with no merging.

Design decision: This was a design decision to keep the routing rules in the .env file. I accept unknown services, as I can have new services logging and then add a new rule. All destination names are checked on startup and will fail if an unknown value is found. This means any typo will cause a crash on startup rather than the service being incorrectly routed.

destination, exists := serviceRouter.rules[enrichedLog.Service]
if !exists {
  slog.Warn("service has no routing rule, using default destination", "service", enrichedLog.Service, "default_destination", string(serviceRouter.defaultDestination))
  enrichedLog.Destination = serviceRouter.defaultDestination
}

A validated map of ROUTING_RULES has been created to route to either of the 3 destinations based on the service. The default destination has been set in the config and will show a warning if there is no rule to match.

4.6 Queuing with RabbitMQ

3 direct exchanges (exchange_service1..3) are bound to durable queues with a routing key log.write. Messages are published using amqp.Persistent, with a 5-second timeout (PUBLISH_TIMEOUT) and confirm mode. The publish will only return nil once the broker has acknowledged. The publisher will retry with backoff.

Design decision: Each destination uses an individual exchange and queue so that messages do not queue up in one queue, starving the others. We always get confirms and cannot allow fire-and-forget.

if publishError := publisher.publishOperation(enrichedLog); publishError != nil {
  publisher.markDisconnected()
  slog.Warn("publishing to rabbitmq failed, using in-memory fallback", "error", publishError, "destination", enrichedLog.Destination, "batch_id", enrichedLog.BatchID)
  return publisher.publishToFallbackAndRecord(enrichedLog)
}
publisher.recordPublishedMessage()
return nil

The broker is a persistent buffer, so it can handle bursts of activity and restarts. If it goes offline, messages will be sent to an in-memory store with a maximum size. These messages will be written to the store and will never be republished. The publisher will retry with backoff.

Timeouts on confirms do not necessarily mean the broker has failed. In ambiguous situations, the publisher will fall back to usin