How to correlate container logs with APM in django?

Kibana version:
7.14.0
Elasticsearch version:
7.14.0
APM Server version:
7.14.0
Filebeat version:
7.14.0
APM Agent language and version:
Python Django - elastic-apm 6.3.3

Description:
The APM server and Python agents are working as expected just like the Filebeat. They're collecting logs and metrics, but I don't know how to correlate them:

Live streaming from Logs section is working and I can filter apm instances with container.id.

1. APM, Elasticsearch, Kibana configs

docker-compose.yml:

version: '3.3'

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.14.0
    hostname: elasticsearch
    environment:
      - ES_JAVA_OPTS=-Xms512m -Xmx512m
      - ELASTIC_PASSWORD=password
    ports:
      - 192.168.100.100:9200:9200
    volumes:
      - ./data:/usr/share/elasticsearch/data
      - ./elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml
    networks:
      - elk

  kibana:
    image: docker.elastic.co/kibana/kibana:7.14.0
    hostname: kibana
    restart: always
    ports:
      - 192.168.100.100:5601:5601
    volumes:
      - ./kibana.yml:/usr/share/kibana/config/kibana.yml:ro
    networks:
      - elk

  apm:
    image: docker.elastic.co/apm/apm-server:7.14.0
    hostname: apm
    command: --strict.perms=false
    depends_on:
      - elasticsearch
    cap_add: ["CHOWN", "DAC_OVERRIDE", "SETGID", "SETUID"]
    cap_drop: ["ALL"]
    volumes:
      - ./apm-server.yml:/usr/share/apm-server/apm-server.yml
    ports:
      - 192.168.100.100:8200:8200
    networks:
      - elk

networks:
  elk:
    driver: bridge

apm-server.yml:

apm-server:
  host: "apm:8200"
  secret_token: token
  rum:
    enabled: true
  kibana:
    enabled: true
    host: "kibana:5601"
    protocol: "http"
    username: "elastic"
    password: "password"
setup.template.enabled: true
setup.template.name: "apm-%{[observer.version]}"
setup.template.pattern: "apm-%{[observer.version]}-*"
setup.template.fields: "${path.config}/fields.yml"
setup.template.overwrite: false
setup.template.settings:
  index:
    number_of_shards: 1
    number_of_replicas: 0
    codec: best_compression
    number_of_routing_shards: 30
    mapping.total_fields.limit: 2000
output.elasticsearch:
  hosts: ["elasticsearch:9200"]
  username: elastic
  password: password
  index: "apm-%{[observer.version]}-%{+yyyy.MM.dd}"
  indices:
    - index: "apm-%{[observer.version]}-sourcemap"
      when.contains:
        processor.event: "sourcemap"
    - index: "apm-%{[observer.version]}-error-%{+yyyy.MM.dd}"
      when.contains:
        processor.event: "error"
    - index: "apm-%{[observer.version]}-transaction-%{+yyyy.MM.dd}"
      when.contains:
        processor.event: "transaction"
    - index: "apm-%{[observer.version]}-span-%{+yyyy.MM.dd}"
      when.contains:
        processor.event: "span"
    - index: "apm-%{[observer.version]}-metric-%{+yyyy.MM.dd}"
      when.contains:
        processor.event: "metric"
    - index: "apm-%{[observer.version]}-onboarding-%{+yyyy.MM.dd}"
      when.contains:
        processor.event: "onboarding"
monitoring.enabled: true
monitoring.elasticsearch:
  username: "elastic"
  password: "password"
  hosts: ["elasticsearch:9200"]

elasticsearch.yml:

---
cluster.name: "docker-cluster"
network.host: 0.0.0.0
discovery.type: single-node
xpack.license.self_generated.type: basic
xpack.security.enabled: true
xpack.monitoring.collection.enabled: true

kibana.yml:

---
server.name: kibana
server.host: "0"
server.publicBaseUrl: https://kibana.domain.com
elasticsearch.hosts: [ "http://elasticsearch:9200" ]
monitoring.ui.container.elasticsearch.enabled: true
elasticsearch.username: elastic
elasticsearch.password: password

2. Python agent config:

base.py:

(...)
import elasticapm
import logging
from elasticapm.handlers.logging import Formatter
from elasticapm.handlers.logging import LoggingFilter
console = logging.StreamHandler()
console.addFilter(LoggingFilter())
logging.getLogger("").addHandler(console)
fh = logging.FileHandler("spam.log")
formatter = Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
fh.setFormatter(formatter)
formatstring = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
formatstring = (
    formatstring + " | elasticapm "
    "transaction.id=%(elasticapm_transaction_id)s "
    "trace.id=%(elasticapm_trace_id)s "
    "span.id=%(elasticapm_span_id)s"
)
ELASTIC_APM = {
    "SERVICE_NAME": "app",
    "SECRET_TOKEN": "secret",
    "SERVER_URL": "http://192.168.100.100:8200",
    "ENVIRONMENT": "test",
}
(...)
MIDDLEWARE = [
    "elasticapm.contrib.django.middleware.TracingMiddleware",
    "elasticapm.contrib.django.middleware.Catch404Middleware",
    (...)
]
INSTALLED_APPS = [
    "elasticapm.contrib.django",
    (...)
]
(...)

3. Filebeat agent configs

/etc/filebeat/filebeat.yml:

filebeat.inputs:
- type: container
  paths: 
    - '/var/lib/docker/containers/*/*.log'
processors:
- add_docker_metadata:
    host: "unix:///var/run/docker.sock"
- decode_json_fields:
    fields: ["message"]
    target: "json"
    overwrite_keys: true
output.elasticsearch:
  hosts: ["192.168.100.100:9200"]
  indices:
    - index: "docker-%{[agent.version]}-%{+yyyy.MM.dd}"
  username: "elastic"
  password: "password"
logging.json: true
logging.metrics.enabled: false

/usr/share/filebeat/module/elasticsearch/server/ingest/pipeline-json.yml:

(...)
- grok:
    field: message
    patterns: %{GREEDYDATA:msg} | elasticapm transaction.id=%{DATA:transaction.id} trace.id=%{DATA:trace.id} span.id=%{DATA:span.id}
(...)

How to make it work? What am I doing wrong?

Have you been able to find log records in Elasticsearch that have a transaction.id that matches? Just want to make sure the log records are getting there, and the grok pattern is correctly populating the transaction.id in the ES documents.

Okay, I realized that there's no transaction.id and in the last few days I was trying to make it appear.

Now I'm just saving logs to a log file and If it'll work I'll use FIlebeat to send logs to Kibana.

I tried to reformat logs with ecs (docs), but without luck when it comes to transaction.id.

Whole configuration here:

setting/loggers/__init__.py:

import os
import ecs_logging
LOG_LEVEL = os.environ.get("LOG_LEVEL") or "INFO"
BASIC_LOGGER = {
    "handlers": ["console"],
    "level": LOG_LEVEL,
    "propagate": False,
}
LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "ecs_logging": {"()": ecs_logging.StdlibFormatter}
    },
    "handlers": {
        "file": {
            'level': LOG_LEVEL,
            'class': 'logging.FileHandler',
            'filename': 'file.log',
            "formatter": "ecs_logging"
        },
    },
    "loggers": {
        **{logger: BASIC_LOGGER for logger in LOGGERS_LIST},
        "django": {"handlers": ["file"], 'level': LOG_LEVEL, "propagate": False}
    },
}

base.py:

(...)
import elasticapm
import logging
import ecs_logging

ELASTIC_APM = {
    "SERVICE_NAME": "app",
    "SECRET_TOKEN": "secret",
    "SERVER_URL": "http://192.168.100.100:8200",
    "ENVIRONMENT": "test",
    "LOG_ECS_REFORMATTING": "override", #tried with off - no difference
}

logger = logging.getLogger("django")

(...)
MIDDLEWARE = [
    "elasticapm.contrib.django.middleware.TracingMiddleware",
    "elasticapm.contrib.django.middleware.Catch404Middleware",
    (...)
]
INSTALLED_APPS = [
    "elasticapm.contrib.django",
    (...)
]
(...)

file.log:

{  "@timestamp":"2021-09-06T13:19:20.773Z",
   "log.level":"error",
   "message":"\"GET / HTTP/1.1\" 500 222222",
   "ecs":{
      "version":"1.6.0" },
   "event":{
      "dataset":"app.log" },
   "log":{
      "logger":"django.server",
      "origin":{
         "file":{
            "line":161,
            "name":"whttp.py" },
         "function":"log_message" },
      "original":"\"GET / HTTP/1.1\" 500 222222" },
   "process":{
      "name":"MainProcess",
      "pid":256,
      "thread":{
         "id":845453411344867,
         "name":"Thread-45" } },
   "request":"<socket.socket fd=3, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('172.0.0.4', 9000), raddr=('172.0.0.88', 51198)>",
   "server_time":"06/Sep/2021 13:19:20",
   "service":{
      "name":"app" },
   "status_code":500 }

{  "@timestamp":"2021-09-06T13:19:21.088Z",
   "log.level":"error",
   "message":"Internal Server Error: /favicon.ico",
   "ecs":{
      "version":"1.6.0" },
   "error":{
      "message":"Error -3 connecting to redis:6379. Temporary failure in name resolution.",
      "stack_trace":"  File \"/usr/local/lib/python/site-packages/django/core/handlers/exception.py\", (...whole trace, but no transaction or span id...)",
      "type":"ConnectionError" },
   "event":{
      "dataset":"app.log" },
   "log":{
      "logger":"django.request",
      "origin":{
         "file":{
            "line":224,
            "name":"log.py" },
         "function":"log_response" },
      "original":"Internal Server Error: /favicon.ico" },
   "process":{
      "name":"MainProcess",
      "pid":222,
      "thread":{
         "id":948373627243366,
         "name":"Thread-46" } },
   "request":"<WSGIRequest: GET '/favicon.ico'>",
   "service":{
      "name":"app" },
   "status_code":500 }


{  "@timestamp":"2021-09-06T13:21:32.883Z",
   "log.level":"info",
   "message":"Watching for file changes with StatReloader",
   "ecs":{
      "version":"1.6.0" },
   "event":{
      "dataset":"app.log" },
   "log":{
      "logger":"django.utils.autoreload",
      "origin":{
         "file":{
            "line":555,
            "name":"autoreload.py" },
         "function":"run_with_reloader" },
      "original":"Watching for file changes with StatReloader" },
   "process":{
      "name":"MainProcess",
      "pid":223,
      "thread":{
         "id":576784135494040,
         "name":"MainThread" } },
   "service":{
      "name":"app" } }

As you can see the ecs is working, but why it doesn't show the transaction.id and the span.id?

Just as an FYI, you don't need "LOG_ECS_REFORMATTING": "override", because you're using the ecs_logging formatter directly.

I'm quite confused as to why you're not seeing a transaction ID or span ID. The only thing I can think is that these errors are happening outside of an actual transaction/span. Are any of these errors happening inside of a request to a view? If so, they should have the transaction/span ID via our log record factory.