A log pipeline can look healthy for as long as its input stays polite. The design is tested when a burst of input meets a slower Elasticsearch destination. Queued records then start consuming memory.

The queue measures pressure. A growing backlog shows demand outpacing processing capacity. If the destination is already saturated, adding consumers can increase contention while the backlog keeps growing. My default uses a bounded queue with a fixed worker count. The overload policy is explicit. The original example below does something more aggressive, which makes it useful to examine.

This walkthrough uses a reader thread, a queue, and asynchronous workers to move container logs into Elasticsearch. I like it as a teaching example because the failure modes are visible in the design. For production, I would replace its unbounded queue and worker controller.

The original environment used Python 3.8 and Elasticsearch 7.6.2. The code excerpts omit setup and helper functions, and the companion repository is currently unavailable. This is a design walkthrough. A complete runnable project and a maintained Logstash replacement are outside its scope. The sample logs come from Elastic’s public tutorial dataset, linked below. Familiarity with Python coroutines, Docker Compose, and HTTP APIs will help you follow the example.

Set up the original example

The configuration mounts the Docker socket and host storage into the consumer and exposes Elasticsearch on port 9200. Those choices grant substantial access and belong in an isolated learning environment. Review access controls and version compatibility before adapting the configuration.

The producer and consumer share this Dockerfile.

Dockerfile:

FROM python:3.8-slim

WORKDIR /code

RUN pip install -U uvloop aiohttp urllib3 docker

COPY . .

Docker Compose sets up the multi-node Elasticsearch cluster and the producer and consumer services.

docker-compose.yml:

version: '3.8'
services:
  producer:
    image: base
    build: .
    container_name: prod1
    volumes:
      - .:/code
    command: python producer.py
    environment:
      - PYTHONUNBUFFERED=1
    networks:
      - elastic
  consumer:
    image: base
    volumes:
      - .:/code
      - /var/run/docker.sock:/var/run/docker.sock
      - /var/lib/docker:/var/lib/docker
    command: python consumer.py
    environment:
      - PYTHONUNBUFFERED=1
      - SOURCE_CONTAINER=prod1
    networks:
      - elastic
  es01:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.6.2
    container_name: es01
    environment:
      - node.name=es01
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es02,es03
      - cluster.initial_master_nodes=es01,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - data01:/usr/share/elasticsearch/data
    ports:
      - 9200:9200
    networks:
      - elastic
  es02:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.6.2
    container_name: es02
    environment:
      - node.name=es02
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es01,es03
      - cluster.initial_master_nodes=es01,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - data02:/usr/share/elasticsearch/data
    networks:
      - elastic
  es03:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.6.2
    container_name: es03
    environment:
      - node.name=es03
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es01,es02
      - cluster.initial_master_nodes=es01,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - data03:/usr/share/elasticsearch/data
    networks:
      - elastic
volumes:
  data01:
    driver: local
  data02:
    driver: local
  data03:
    driver: local

networks:
  elastic:
    driver: bridge

The Compose configuration defines a producer, a consumer, and a three-node Elasticsearch cluster. The producer reads Elastic’s public sample log dataset from a local file named logstash-tutorial.log.

producer.py:

import time
from pathlib import Path

path = "logstash-tutorial.log"
text = Path(path).read_text().split("\n")
while True:
    for line in text:
        print(line)
        time.sleep(0.1)

The producer loads the sample once and repeatedly writes its lines to standard output. The following output comes from the original run and is abbreviated.

$ python producer.py
83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"
... (original output abbreviated)

Read logs without blocking the event loop

Docker captures container output through its configured logging driver. Its default json-file driver stores logs as JSON. The selected driver determines storage behavior. Check the Docker logging documentation before relying on a host path or retention behavior.

The consumer parses the sample log format with the following regular expressions. A format-specific parser needs a policy for unmatched records.

INT = '(?:[+-]?(?:[0-9]+))'
IP = '(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)'
USER = '[a-zA-Z0-9._-]+'
MONTHDAY =  '(?:(?:0[1-9])|(?:[12][0-9])|(?:3[01])|[1-9])'
MONTH = '(?:Jan(?:uary)?|Feb(?:ruary)?|Mar(?:ch)?|Apr(?:il)?|May|Jun(?:e)?|Jul(?:y)?|Aug(?:ust)?|Sep(?:tember)?|Oct(?:ober)?|Nov(?:ember)?|Dec(?:ember)?)'
YEAR = '(?:\d\d){1,2}'
HOUR = '(?:2[0123]|[01]?[0-9])'
MINUTE = '(?:[0-5][0-9])'
SECOND = '(?:(?:[0-5]?[0-9]|60)(?:[:.,][0-9]+)?)'
TIME = f'(?!<[0-9]){HOUR}:{MINUTE}(?::{SECOND})(?![0-9])'
HTTPDATE = f'{MONTHDAY}/{MONTH}/{YEAR}:{TIME} {INT}'
VERSION = '[0-9]+(?:(?:\.[0-9])+)?'
REQUEST = f'\\"(?:\w+ \S+(?: HTTP/{VERSION})?|.*?)\\"'
TIMESTAMP = f"\[{HTTPDATE}\]"
QUOTE = f'(?:\\".+\\")'
COMMONLOG = r" ".join([f'(?P<ip>{IP})',f'(?P<ident>{USER})',f'(?P<auth>{USER})',f'(?P<timestamp>{TIMESTAMP})',f'(?P<request>{REQUEST})',"(?P<status>\d+)", "(?P<bytes>\d+|-)", f'(?P<referrer>{QUOTE})', f'(?P<agent>{QUOTE})'])

The Docker client’s blocking log iterator runs in a separate reader thread. That keeps a wait for the next log record from blocking the event loop. A thread-safe queue.Queue passes records to asynchronous workers, which use nonblocking queue reads and await HTTP requests.

The excerpt below preserves the original structure. The queue is unbounded, and a record leaves it before Elasticsearch confirms receipt. Those choices leave memory growth and delivery failures unresolved.

consumer.py:


pattern = re.compile(COMMONLOG)
queue = queue.Queue()

async def worker(name, client):
    log = logging.getLogger(name)
    while True:
        try:
            line = queue.get_nowait()
        except Empty:
            log.info('Queue is empty.')
            await asyncio.sleep(1)
        else:
            line = line.decode('utf-8')
            match = pattern.match(line)
            if match is None:
                log.info(f"No match found for {line.strip()}")
                await asyncio.sleep(0.001)
            else:
                async with client.post("http://es01:9200/logs/_doc/",
                               data=json.dumps(match.groupdict()).encode('utf-8'),
                              headers={'Content-Type': 'application/json'}
                          ) as resp:
                    if resp.status != 201:
                         err = await resp.text()
                         log.info(f"{resp.status}: {err}")
                    else:
                        log.info("Upload successful.")


def reader(container):
    log = logging.getLogger('reader')
    stream = container.logs(stream=True)
    while True:
        try:
            queue.put_nowait(next(stream))
        except StopIteration:
            log.debug("No more logs")
            break

async def main():
    c_name = os.getenv('SOURCE_CONTAINER')
    if not c_name:
        print("You must specify a source container name.")
        sys.exit(1)
    container = get_container(c_name)
    threading.Thread(target=reader, args = (container,), daemon=True).start()
    async with aiohttp.ClientSession() as client:
        done, pending = await asyncio.wait([asyncio.create_task(worker(f"worker{i}", client)) for i in range(5)])


if __name__ == '__main__':
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(threadName)s %(name)s: %(message)s',
        stream=sys.stderr,
    )
    asyncio.run(main())

The reader exits when its iterator is exhausted. Reconnecting after a container restart would need separate handling. The following excerpt comes from the original example with five worker coroutines.

consumer_1  | MainThread worker1: Upload successful.
consumer_1  | MainThread worker0: Upload successful.
consumer_1  | MainThread worker3: Upload successful.
... (original output abbreviated)

Observe a growing queue

The initial producer emits approximately ten records per second. A burstier producer can fill the queue faster than the workers drain it. The original controller below responds by adding a worker when the observed queue size is above a threshold and growing.

I think this controller is the wrong production response to the signal it observes. It has no upper bound on workers or scale-down policy. Queue size is approximate, and a growing backlog can indicate a saturated destination. Adding workers in that situation can increase contention and failures. The snippet also lacks explicit worker supervision and cancellation.

async def controller(max_size=100):
    log = logging.getLogger('controller')
    async with aiohttp.ClientSession() as client:
        # start with 5 workers
        [asyncio.create_task(worker(f"worker{i}", client)) for i in range(5)]
        num = 5
        delay = 1
        curr_size = 0
        prev_size = 0
        while True:
            curr_size = queue.qsize()
            if curr_size > max_size and curr_size > prev_size:
                asyncio.create_task(worker(f"worker{num}", client))
                await asyncio.sleep(delay/100)
                delay += 1
                num += 1
            else:
                await asyncio.sleep(1)
                delay = 1
            prev_size = curr_size
            log.debug(f"Currently running {len(asyncio.all_tasks()) - 2} workers. Queue size: {queue.qsize()}")

async def main():
    c_name = os.getenv('SOURCE_CONTAINER')
    if not c_name:
        print("You must specify a source container name.")
        sys.exit(1)
    container = get_container(c_name)
    threading.Thread(target=reader, args = (container,), daemon=True).start()
    done, pending = await asyncio.wait([asyncio.create_task(controller())])

The following replacement producer loop varies the interval between sample records. It assumes the same loaded text list and an import of random.

def main():
    while True:
        for line in text:
            print(line)
            time.sleep(random.random() * 0.01)

Output:

consumer_1  | MainThread controller: Currently running 6 workers. Queue size: 325
consumer_1  | MainThread controller: Currently running 7 workers. Queue size: 325
consumer_1  | MainThread controller: Currently running 66 workers. Queue size: 1692
consumer_1  | MainThread worker54: Queue is empty.
consumer_1  | MainThread worker56: Queue is empty.
... (original output abbreviated)

The output shows a reported queue size of 1,692 records while the controller is adding workers. That number measures the backlog. Worker count and throughput need separate measurements. A later empty-queue message shows that no record was available at that moment.

Define the missing operational behavior

I would replace the controller with a bounded queue and a fixed number of supervised workers, then load-test that boundary before adding adaptive concurrency. Before adapting any part of the design, make the following decisions explicit:

Concern Limitation in the example Behavior to define
Backpressure Queue and worker growth are unbounded Bound pending work and concurrency; choose what happens at capacity
Delivery Records are removed before a successful write Decide when to acknowledge, retry, or persist a record
Duplicates Retried writes can create additional documents Use stable record identities when duplicate suppression is required
Worker failures Request errors can terminate a worker Supervise workers and expose failures as metrics
Shutdown Reader and workers have no coordinated drain Stop intake, finish or persist pending work, and close clients
Parsing Unmatched records are only logged Count rejected records and define retention for inspection

Batching writes may reduce request overhead. It also introduces partial failures and additional buffering. Measure queue age and write latency as well as queue length before changing concurrency or batch size.

For a smaller runnable introduction to scheduling and bounded concurrency, start with Understanding concurrency with asyncio. For application-level usage events, the Bedrock token-usage example explores another telemetry delivery boundary.