The Great Cat Escape: A hands-on guide to OpenTelemetry, Elastic APM, and Kibana alerting

Elastic speaks OpenTelemetry natively. Send traces, logs, and metrics over OTLP straight into Elasticsearch, no proprietary agents required. See how it fits together, try it for free in the cloud, or run it locally.

In this article, you instrument a Python game, The Great Cat Escape, with OpenTelemetry (OTel) and then ship the telemetry to Elastic APM. Next, you write an Elasticsearch Query Language (ES|QL) rule in Kibana that fires before any cat reaches 10 escape attempts, as each cat picks up speed at its own pace. Swap cat.name for user.id and escape attempts for failed logins, and the same rule catches credential stuffing.

How The Great Cat Escape game works

In this detective game, the rules that you eventually create in your Elastic cluster govern whether you win or lose against these cats. Each cat creeps toward the fence at its own accelerating pace. If any cat hits 10 escape attempts, that cat is free and you lose your detective license. Your job, gumshoe, should you choose to accept it, is to configure an Elastic Observability alert that fires before any cat hits 10. The earlier it fires, the higher your score.

 Requirements: Elastic Observability, Python, and a tunnel

• Your own fork/branch of The Great Cat Escape

• Elastic Observability v. 9.3.0 or higher (get a seven-day free trial of Elastic Cloud Serverless), 

• Python v. 3.11.0 or higher (latest version)

• ngrok or a similar tunnel, since Kibana's webhook needs a way to reach your laptop

The Great Cat Escape game is governed through python-dotenv with credentials from an environment (.env) file. Change these to match your Elastic cluster, and make sure that the connection is through Elastic APM. You can input your secret token through your agent by going to Fleet → Agent Policies → click on the policy you want to change → Elastic APM → Settings → and then scroll down until you see "Agent Authorization."

To populate your environment, take a copy of the .env.example and rename it .env. Your .env file needs two values:

ELASTIC_APM_HOST=your-host.apm.us-central1.gcp.cloud.es.io
ELASTIC_SECRET_TOKEN=your_token
NUM_CATS=3

NUM_CATS is optional (default 1, max 6). More cats means more chaos, but more chaos means a potentially higher score for you, detective.

Instrumenting a Python app with OpenTelemetry for Elastic APM

Catching rogue cats is hard. What should we trigger on? What should be measured? Why? To answer these questions, we need to know what our cats are doing at any given time. That’s where instrumentation comes in. In observability, instrumentation is the process of adding code or configurations to your application that can generate and collect telemetry data, as well as export it. This telemetry data can include metrics and logs, along with traces. So let’s get into the code that will help your investigation.

To instrument our application, we’ll use OTel, which is a standardized framework for generating, collecting, and exporting telemetry data. It’s the industry standard for observability. Before we get started, we need to define a resource in our project. In OTel instrumentation, a resource represents the entity that’s producing telemetry data, which, in this case, is our “escape-room-cat” service.

SERVICE_NAME = "escape-room-cat"
resource = Resource.create({"service.name": SERVICE_NAME})

The “service.name” key is a standard semantic convention from OTel. Semantic conventions are standardized and vendor-neutral naming rules used throughout OTel to structure your data. You’ll see more of these conventions as we set up our signals in the sections that follow.

Setting up traces, metrics, and logs with OTLP exporters

We set up our three pillars and our logs, metrics, and traces through OTel in the following code. Let’s go through the code for each signal.

trace_provider = TracerProvider(resource=resource)
trace_provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint=f"{OTLP_BASE_URL}/v1/traces", headers=_AUTH_HEADER)
    )
)
trace.set_tracer_provider(trace_provider)
tracer = trace.get_tracer(SERVICE_NAME)

With this code, we set up the TracerProvider to start collecting traces, which are then exported in batches through the OTLPSpanExporter via the OpenTelemetry Protocol (OTLP). A trace is a collection of spans, and spans can be enriched with span attributes. We go over the span "cat.escape_attempt" later in this article.

meter_provider = MeterProvider(
    resource=resource,
    metric_readers=[
        PeriodicExportingMetricReader(
            OTLPMetricExporter(endpoint=f"{OTLP_BASE_URL}/v1/metrics", headers=_AUTH_HEADER),
            export_interval_millis=5000,
        )
    ],
)
metrics.set_meter_provider(meter_provider)
meter = metrics.get_meter(SERVICE_NAME)

Next, we set up our metrics through a MeterProvider and a PeriodicExportingMetricReader. Metrics are polled on a timer according to the reader’s interval and then exported in OTLP format through OTLPMetricExporter, rather than being pushed when events arrive, like traces and logs. The MeterProvider is still necessary despite the PeriodicExportingMetricReader because it’s the configuration layer from the OTel SDK that actually processes and collects your telemetry and exports it to your Elastic Observability project. This is true for the TracerProvider and for the LoggerProvider (see below):

logger_provider = LoggerProvider(resource=resource)
logger_provider.add_log_record_processor(
    BatchLogRecordProcessor(
        OTLPLogExporter(endpoint=f"{OTLP_BASE_URL}/v1/logs", headers=_AUTH_HEADER)
    )
)
set_logger_provider(logger_provider)
logging.getLogger().addHandler(
    LoggingHandler(level=logging.INFO, logger_provider=logger_provider)
)
logging.getLogger().setLevel(logging.INFO)
logger = logging.getLogger(SERVICE_NAME)

LoggerProvider works slightly differently than the rest of the signals. The code here bridges Python’s built-in logging module, LoggingHandler, with OTel, instead of creating a new logging system. This handler routes any logging.info call in the Python application into the OTel pipelines, which means that your structured log records get exported alongside the rest of your signals.

Counters, gauges, and spans: The signals this game emits

Now that we know how OTel works to instrument our game, let’s get into the details of the specific signal configurations that we’ll use in this application.

First, let’s go over the most important counter (a data point that only goes up):

escape_attempts_counter = meter.create_counter(
    "cat.escape.attempts", description="Number of escape attempts by each cat", unit="1"
)

This metric is important to the game because it determines whether you win or lose. If the ”cat.escape.attempts” counter for any cat hits 10, you lose and the cat gets free. 

The next metric is an observable gauge governed by a function from our game:

meter.create_observable_gauge(
    "cat.suspicion.level",
    callbacks=[_suspicion_callback],
    description="How suspicious each cat is acting (0-100)",
    unit="percent",
)
------------------------------------------------------------
cat.suspicion = min(100, cat.suspicion + random.uniform(8, 15))

This observable gauge meter indicates that, at each collection interval, a cat’s current suspicion level is recorded. “cat.suspicion.level” is governed by a function that picks a random floating number between 8 and 15 and appends that to the current suspicion level. An observable gauge represents a value that can go up and down, unlike a counter that only goes up. We’re using an observable gauge instead of a counter because we’re trying to determine How suspicious is this cat right now? as opposed to How much suspicion has accumulated? In this game, cat suspicion is another meter that you might be able to create a rule on.

Below is the central event, a span that fires every time a cat attempts to break out. This span identifies which cat it was, the escape attempt number, the cat’s position, whether the attempt was successful, and the cat’s suspicion level. The next span is also collected whenever the cat moves:

with tracer.start_as_current_span("cat.escape_attempt") as span:
    span.set_attribute("cat.name", cat.name)
    span.set_attribute("cat.position.x", cat.x)
    span.set_attribute("cat.position.y", cat.y)
    span.set_attribute("escape.attempt_number", attempt_num)
    span.set_attribute("escape.success", attempt_num >= MAX_ATTEMPTS)
    span.set_attribute("cat.suspicion.level", cat.suspicion)
    span.set_attribute("game.phase", game.phase.value)
    escape_attempts_counter.add(1, {"cat.name": cat.name})
----------------------------------------------------------------------------------
with tracer.start_as_current_span("cat.move") as span:
        span.set_attribute("cat.name", cat.name)
        span.set_attribute("cat.position.x", cat.x)
        span.set_attribute("cat.position.y", cat.y)
        span.set_attribute("cat.distance.moved", dist)
        span.set_attribute("game.phase", game.phase.value)
        distance_traveled.add(dist, {"cat.name": cat.name})

When you view the traces data in Kibana, the trace and its span attributes appear as follows:

The fields translated from our game are prepended with “numeric_labels” or “labels.” This is the result of our data being translated from OTel into the Elastic Common Schema (ECS). However, you can rename these in Kibana, if you choose, to shorten the name or make searching easier for you. 

Emitting structured logs with the Python logging module

Logs are the most human-readable of the three signals, and this game emits them through semi-structured logging:

logger.warning(
f"🚨 {cat.name} escape attempt #{attempt_num}/{MAX_ATTEMPTS} "
f"at ({cat.x:.1f}, {cat.y:.1f}) near {edge_str} "
f"— suspicion={cat.suspicion:.0f}%"
       )


logger.info(f"🎮 {cat.name} begins escape plotting")


logger.error(f"🚪 {cat.name} ESCAPED! Game over.")


logger.info(
        f"✅ Alert fired! Worst cat at {worst}/{MAX_ATTEMPTS}. "
        f"Score: {game.score}. All cats caught!"
)

The game sends out several logs and communicates them via multiple methods. When it comes to logging, it’s generally good practice to keep your logs clear and concise so that they’re easily recognizable by anyone debugging, especially for errors. In this Flask application, we used the logging module by Python. The logs sent to Kibana are “logger.info”, “logger.warning”, and “logger.error”, and they indicate the log severity level. logger.info is used for generic information from your application, and logger.warning indicates unexpected events that the system has recovered from. logger.error notes a failure or critical error. In The Great Cat Escape, the only error is when you lose the game. These logs are emitted through the standard Python logger and exported via the OTLP log exporter, so they show structured severity_text and resource attributes, along with trace correlation in Kibana:

🚨 Mittens escape attempt #2/10 at (42.2, 62.1) near inner yard — suspicion=24%

The next logging system is _log, a custom function written for in-game use:

def _log(msg: str):
    ts = time.strftime("%H:%M:%S")
    with game_lock:
        game.log_lines.insert(0, f"[{ts}] {msg}")
        if len(game.log_lines) > 60:
            game.log_lines.pop()
----------------------------------------------------------------------------------
_log(
f"🚨 {cat.emoji} {cat.name} attempt #{attempt_num}/{MAX_ATTEMPTS} "
f"near {edge_str} ({cat.suspicion:.0f}%)"
)
----------------------------------------------------------------------------------
_log(f"🎉 ALERT FIRED! Worst cat at {worst}/{MAX_ATTEMPTS}. Score: {game.score}")
    return jsonify({"ok": True, "score": game.score, "worst_count": worst})
----------------------------------------------------------------------------------
_log(f"🚪💨 {cat.emoji} {cat.name} ESCAPED! YOU LOSE!")
            return
----------------------------------------------------------------------------------
_log(f"{cat.emoji} {cat.name} starts pacing...")

This log function shows the player the same information as Kibana receives by grabbing the current time and by using game_lock through game_lock = threading.Lock(). In this game, since there are requests from multiple threads (cats), the game_lock helps by preventing data loss if two threads reach out to the same resource at the same time, which can lead to data corruption. 

Now that you understand the instrumentation, the final step before you write your rules and play the game is to understand the alert_fired API and wire the needed webhook so that your alert can be received by your URL. 

Wiring a webhook connector for your alert

To send out a webhook alert, you use ngrok or a similar tunnel to expose your port to Kibana, enabling it to POST to that port over the internet. You front your own port with a tunnel, and that public URL is the <your-tunnel> you point Kibana at. Add a connection of type “Webhook” in Kibana with these settings:

Method: POST
URL: http://<your-ip>:5003/api/alert_fired
Headers: Content-Type: application/json
----------------------------------------------------------------------------------
# You configure this with your rule under "Actions" later
Body:
{
  "cat_name": "{{context.group}}",
  "current_count": {{context.value}},
  "rule_name": "{{rule.name}}",
  "fired_at": "{{context.date}}"
}

In Kibana 9.4, the connector lives under Stack Management → Connectors. Save it once, and you can reuse it across all of your rules.

The {{context.group}} field shows which cat triggered the rule, and the fired_at {{context date}} field tells you when the alert happened. These are purely informational parameters, but the winning endpoint, /api/alert_fired takes more precaution with “current_count” and “rule_name.” The game's /api/alert_fired endpoint, shown in the following code, has several checkpoints based on the parameters above that help with specificity in this game. These checkpoints are also important to real alerting in systems to ensure that the correct source, system, or alert was sent via your alerting webhook.

@app.route("/api/alert_fired", methods=["POST"])
def api_alert_fired():
    # Checkpoint 1
    payload = request.get_json(silent=True) or {}


    # Checkpoint 2
    rule_name = payload.get("rule_name")
    if rule_name != EXPECTED_RULE_NAME:
        return jsonify({"ok": False, "msg": "Unexpected or missing rule_name."}), 400


    # Checkpoint 3
    count = payload.get("current_count")
    if not isinstance(count, (int, float)) or count < MIN_COUNT:
        return jsonify({"ok": False, "msg": "Missing or invalid current_count."}), 400

The endpoint runs three checkpoints before it accepts a win, matching the numbered comments in the code above:

1. Parse the request body. request.get_json takes the body sent by the webhook and parses it into a Python dict. Passing silent=True means an empty or malformed body won't raise an error; the endpoint keeps working and fills in unknown values with {}.

2. Check the rule name. "Cat Escape Alert" is the enforced rule_name, so a stray request from another source can't trigger a win. Use this name exactly, or change it in the code. (See EXPECTED_RULE_NAME = "Cat Escape Alert".)

3. Check the attempt count. This gate stops an alert from counting as a win when no cat has actually moved. It's set to one, so at least one escape attempt must have registered. If you want a higher bar, change it in the code. (See MIN_COUNT = 1.)

How to create a Kibana alert rule with ES|QL

This game ends one of two ways. You win if your created rule alerts properly; and you lose if any cat reaches 10 escape attempts. 

To create your rule and play the game, follow these instructions. Lettered subsections give you a full breakdown of how to create a winning rule.

1. Go to Kibana → Stack Management → Rules.

2. Create new rule → Elasticsearch query.

3. Create a rule to alert on logs or metrics. You can use Kibana Query Language (KQL) or Lucene, Query DSL, or ES|QL.

• To create a winning rule, choose ES|QL. Enter the code block below into the box for creating a query. The service for this game is “escape-room-cat,” and “cat.escape.attempts” IS NOT NULL discards every row that isn’t an escape.attempts measurement. Since the cat.escape.attempts is an OTel counter, it only climbs and doesn’t reset when you restart your game. To circumvent this counter, we take the change in the counter over the window (EVAL attempts = hi - lo) instead of its absolute cumulative value.

• ES|QL query: 

FROM metrics-apm*
| WHERE service.name == "escape-room-cat" AND cat.escape.attempts IS NOT NULL
| STATS hi = MAX(cat.escape.attempts), lo = MIN(cat.escape.attempts) BY cat = labels.cat_name
| EVAL attempts = hi - lo
| WHERE attempts >= 1
| SORT attempts DESC
| KEEP cat, attempts

• Change the rule schedule for every 30–45 seconds.

4. Add an action with your configured webhook, and use the configuration from the ″A webhook connector for your alert″” section of this post.

{
"cat_name": "{{context.group}}",
"current_count": {{context.value}},
"rule_name": "{{rule.name}}",
"fired_at": "{{context.date}}"
}

5. Click Start Game, and wait. 

6. If your alert signals in time, you’ve won, detective. You sleuthed out the right cat and caught it in time. If your rule doesn’t alert by the tenth escape attempt, you lose. Try configuring a new rule and checking that it matches the correct rule_name for the checkpoint (including whitespace and casing) or changing your rule’s threshold or timing.

From cat escapes to credential stuffing detection

Beneath this fun, interactive cat-catching game is an application that demonstrates simple and correctly wired observability architecture. The pieces assembled from this application are the building blocks of a production observability stack. They include OTLP/HTTP exporters shipping to an Elastic Observability endpoint and signals stamped with a service.name by a resource, along with the separate MeterProvider, LoggerProvider, and TracerProvider pipelines. They also include a counter that’s emitted alongside the span which describes the same event, an observable gauge sample through a callback, ES|QL queries transformed for alerting on delta values, and more. And, since the Elastic 9.4.0 also allows you to natively use Prometheus Query Language (PromQL) to query time series data, you have even more query possibilities.

If you replace “escape attempt” with “failed authentication” and cat.name with user.id, the same rule that you created to catch a fence-jumping tabby can become a credential-stuffing detector. Try out different rules and types of alerting, along with other query rates, and see how creative you can be with Elastic Observability in this game. 

If you need any help with The Great Cat Escape, feel free to contact the author.

Daylight streamed through the windows. With a robust rule in hand and an alert in the other, the detective sighed with relief. Nights of grepping had turned into quick decisions and high scores for the seasoned private eye. This case had proved difficult, but no cats had escaped the watchful observability back end. Although this investigation was successful, there would be more cases. And more cats. But now the detective was ready for them.