diff --git a/rabbitmq/tests/README.md b/rabbitmq/tests/README.md new file mode 100644 index 0000000000000..656741555325b --- /dev/null +++ b/rabbitmq/tests/README.md @@ -0,0 +1,69 @@ +# rabbitmq test fixtures + +The pytest suite plus the Compose environments used by the tests and by the +evalya fixtures in `evalya.yaml`. + +## Compose environments + +| File | What it is | +|---|---| +| `compose/docker-compose.yaml` | Single management-enabled broker (the pytest default). | +| `compose/full-coverage.compose` | Full metric-coverage environment (see below). | + +Two are published as reusable evalya fixtures: `rabbitmq-standalone` (the plain +broker) and `rabbitmq-full` (the full-coverage environment). Both present the +management HTTP API on `:15672` with the `guest`/`guest` credentials. + +## The full-coverage fixture + +`rabbitmq-full` makes a single fixture emit non-trivial values for the entire +metric surface the rabbitmq check and the OTel `rabbitmqreceiver` collect from a +real broker. An idle broker reports zero for most rate/counter metrics, so two +workloads run against it. Four services: + +- **rabbitmq** — real `rabbitmq:-management` broker, the fixture entrypoint + (published as `rabbitmq-full`). A small `rabbitmq.conf` lifts the loopback-only + restriction on `guest` so the workload containers and the scrapers can connect + over the Compose network. +- **perf-test** — the official RabbitMQ load tool (`pivotalrabbitmq/perf-test`), + running continuous publish/consume/ack with manual acks, so + `message.published` / `message.acknowledged` / `message.delivered`, + `consumer.count` and the `queue.*` rate/count metrics keep moving. +- **activity-gen** — a `pika` sidecar (`activity-gen.py`) for what perf-test does + not exercise: durable queues + persistent messages (`node.msg_store_*`, + `node.queue_index_*`), queue/exchange declare+delete churn + (`node.queue_created/declared/deleted`), connection/channel open+close churn + (`node.connection_*`, `node.channel_*`), unroutable publishes + (`message.dropped`), and messages left both ready and unacknowledged + (`message.current` with the `ready` / `unacknowledged` states). + +The `node.*` and `erlang.*` gauges come naturally from the running broker. + +No host port is published, to avoid clashing with a local broker; reach it over +the Compose network via the `DB_HOST` label, or add `--publish 15672:15672` to +inspect by hand. Set `ACTIVITY_GEN=0` (host env) for a quiescent `rabbitmq-full`: +the sidecar container stays up but generates no traffic. + +## Enabling the OTel node.* metrics + +> **Important:** the OTel `rabbitmqreceiver` ships with **every `rabbitmq.node.*` +> metric disabled by default** — only the six message/consumer metrics +> (`rabbitmq.message.*`, `rabbitmq.consumer.count`) are on out of the box. This +> fixture makes the broker *emit* the underlying data, but a collector scraping +> `rabbitmq-full` will not report the `node.*` series (roughly half of the +> semantic-core mappings) unless they are explicitly enabled in the receiver +> config: + +```yaml +receivers: + rabbitmq: + endpoint: http://:15672 + username: guest + password: guest + collection_interval: 10s + metrics: + rabbitmq.node.disk_free: {enabled: true} + rabbitmq.node.mem_used: {enabled: true} + rabbitmq.node.fd_used: {enabled: true} + # ... enable the remaining rabbitmq.node.* metrics the mappings need +``` diff --git a/rabbitmq/tests/activity-gen.py b/rabbitmq/tests/activity-gen.py new file mode 100644 index 0000000000000..c3d4188fb86ab --- /dev/null +++ b/rabbitmq/tests/activity-gen.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +"""Drive rabbitmq-full so the rabbitmq check and the OTel rabbitmqreceiver report +their full metric surface with non-trivial values. + +perf-test (a sibling container) already drives the raw publish/consume/ack rates. +This sidecar covers what perf-test does not: + + * durable queues + persistent messages -> node.msg_store_*, node.queue_index_* + * queue/exchange declare + delete churn -> node.queue_created/declared/deleted + * connection + channel open/close churn -> node.connection_*, node.channel_* + * unroutable publishes -> message.dropped / *unroutable* + * messages left ready and left unacked -> message.current {ready, unacknowledged} + +Consumes DB_HOST / AMQP_PORT / RABBITMQ_USER / RABBITMQ_PASSWORD from the fixture. +Set ACTIVITY_GEN=0 to keep the container up but idle (a quiescent rabbitmq-full). +""" +import os +import sys +import time +import threading + +import pika + +HOST = os.environ.get("DB_HOST", "rabbitmq") +PORT = int(os.environ.get("AMQP_PORT", "5672")) +USER = os.environ.get("RABBITMQ_USER", "guest") +PASSWORD = os.environ.get("RABBITMQ_PASSWORD", "guest") +INTERVAL = float(os.environ.get("ACTIVITY_INTERVAL", "5")) # seconds between cycles +WINDOW = int(os.environ.get("ACTIVITY_QUEUE_WINDOW", "5")) # durable queues kept alive +UNACKED_PREFETCH = int(os.environ.get("ACTIVITY_UNACKED", "20")) + +EXCHANGE = "activity.direct" +ROUTING_KEY = "activity.rk" +UNROUTABLE_KEY = "activity.no-binding" +UNACKED_QUEUE = "activity.unacked" + +PERSISTENT = pika.BasicProperties(delivery_mode=2) + + +def log(msg): + print(f"activity-gen: {msg}", flush=True) + + +def conn_params(): + return pika.ConnectionParameters( + host=HOST, + port=PORT, + credentials=pika.PlainCredentials(USER, PASSWORD), + heartbeat=30, + blocked_connection_timeout=30, + connection_attempts=1, + socket_timeout=10, + ) + + +def wait_for_broker(timeout=180): + """Block until the broker accepts a connection, or exit non-zero.""" + deadline = time.time() + timeout + while time.time() < deadline: + try: + pika.BlockingConnection(conn_params()).close() + log("broker is up") + return + except Exception as exc: # noqa: BLE001 - startup probe, any failure means "not ready yet" + log(f"waiting for broker ({exc.__class__.__name__})...") + time.sleep(3) + log("broker never became reachable; exiting") + sys.exit(1) + + +def unacked_holder(): + """Keep a pool of messages permanently unacknowledged so + message.current{state=unacknowledged} stays non-zero. Consumes with a + prefetch window and never acks; auto-reconnects on failure.""" + while True: + try: + conn = pika.BlockingConnection(conn_params()) + ch = conn.channel() + ch.queue_declare(queue=UNACKED_QUEUE, durable=True) + ch.basic_qos(prefetch_count=UNACKED_PREFETCH) + # Seed the queue so there is something to hold unacked. + for n in range(UNACKED_PREFETCH * 2): + ch.basic_publish("", UNACKED_QUEUE, f"unacked-{n}".encode(), PERSISTENT) + ch.basic_consume(UNACKED_QUEUE, on_message_callback=lambda *a: None, auto_ack=False) + log("unacked holder attached") + ch.start_consuming() # blocks, holding delivered messages unacked + except Exception as exc: # noqa: BLE001 + log(f"unacked holder reconnecting ({exc.__class__.__name__})") + time.sleep(5) + + +def run_cycle(i, live_queues): + """One churn cycle on its own short-lived connection/channel.""" + conn = pika.BlockingConnection(conn_params()) # connection_created + ch = conn.channel() # channel_created + ch.confirm_delivery() + + ch.exchange_declare(EXCHANGE, exchange_type="direct", durable=True) + + # Durable queue with persistent, partly-consumed messages: exercises + # msg_store / queue_index and leaves some messages "ready". + qname = f"activity.durable.{i}" + ch.queue_declare(queue=qname, durable=True) # queue_declared/created + ch.queue_bind(qname, EXCHANGE, routing_key=ROUTING_KEY) + for n in range(40): + ch.basic_publish(EXCHANGE, ROUTING_KEY, f"msg-{i}-{n}".encode(), PERSISTENT) + # Consume+ack half (drives acknowledged + msg_store reads); leave the rest ready. + for _ in range(20): + method, _props, body = ch.basic_get(qname, auto_ack=False) + if method is None: + break + ch.basic_ack(method.delivery_tag) + live_queues.append(qname) + + # Unroutable publishes: no binding for this key -> dropped (message.dropped / + # *unroutable*). mandatory=False so the broker drops rather than returns. + for n in range(10): + ch.basic_publish(EXCHANGE, UNROUTABLE_KEY, f"drop-{i}-{n}".encode(), mandatory=False) + + # Ephemeral queue declared and immediately deleted -> queue_created + deleted. + eph = f"activity.ephemeral.{i}" + ch.queue_declare(queue=eph, durable=False, auto_delete=False) + ch.queue_delete(eph) # queue_deleted + + # Bound the number of live durable queues; deleting the oldest drives + # queue_deleted and keeps the broker from growing without limit. + while len(live_queues) > WINDOW: + old = live_queues.pop(0) + try: + ch.queue_delete(old) + except Exception: # noqa: BLE001 - queue may already be gone + pass + + ch.close() # channel_closed + conn.close() # connection_closed + + +def main(): + if os.environ.get("ACTIVITY_GEN", "1") == "0": + log("disabled via ACTIVITY_GEN=0; idling") + while True: + time.sleep(3600) + + wait_for_broker() + + holder = threading.Thread(target=unacked_holder, daemon=True) + holder.start() + + live_queues = [] + i = 0 + while True: + i += 1 + try: + run_cycle(i, live_queues) + if i % 10 == 0: + log(f"completed {i} cycles ({len(live_queues)} durable queues live)") + except Exception as exc: # noqa: BLE001 - keep the workload alive across broker blips + log(f"cycle {i} failed ({exc.__class__.__name__}: {exc}); retrying") + time.sleep(5) + time.sleep(INTERVAL) + + +if __name__ == "__main__": + main() diff --git a/rabbitmq/tests/compose/config/rabbitmq.conf b/rabbitmq/tests/compose/config/rabbitmq.conf new file mode 100644 index 0000000000000..cef0161f0c36e --- /dev/null +++ b/rabbitmq/tests/compose/config/rabbitmq.conf @@ -0,0 +1,10 @@ +# Full-coverage fixture broker config. +# +# Allow the guest user to authenticate from non-loopback addresses so the +# workload containers and the scrapers can connect over the compose network. +# Safe here: this is a throwaway test fixture on an internal network. +loopback_users.guest = false + +# Per-object Prometheus metrics, in case a consumer scrapes the openmetrics +# endpoint (:15692) instead of the management API. +prometheus.return_per_object_metrics = true diff --git a/rabbitmq/tests/compose/full-coverage.compose b/rabbitmq/tests/compose/full-coverage.compose new file mode 100644 index 0000000000000..5447d005fea4b --- /dev/null +++ b/rabbitmq/tests/compose/full-coverage.compose @@ -0,0 +1,76 @@ +# Full metric-coverage environment for the rabbitmq check and the OTel +# rabbitmqreceiver. Both scrape the management HTTP API (:15672) of a real +# broker. An idle broker reports zero for most rate/counter metrics, so this +# environment keeps two workloads running against it: +# +# - perf-test continuous publish/consume/ack (the official RabbitMQ load +# tool) -> message.published/acknowledged/delivered, +# consumer.count and the queue.* rate/count metrics. +# - activity-gen a churn sidecar for what perf-test does not exercise: durable +# queues + persistent messages (msg_store / queue_index), +# queue/exchange declare+delete, connection/channel churn, and +# unroutable (dropped) messages. See ../activity-gen.py. +# +# The node.* and erlang.* metrics come naturally from the running broker. +# +# NOTE: the OTel rabbitmqreceiver ships with every rabbitmq.node.* metric +# DISABLED by default (only the 6 message/consumer metrics are on). To verify +# the node.* mappings against this fixture, the scraping collector must enable +# them in its receiver config -- see ../README.md. +services: + rabbitmq: + image: "rabbitmq:${RABBITMQ_VERSION:-4.0}-management" + # Stable node name so the node.* time series are stable across restarts. + hostname: rabbitmq-full + environment: + RABBITMQ_DEFAULT_USER: guest + RABBITMQ_DEFAULT_PASS: guest + volumes: + # Lift the loopback-only restriction on the guest user so the workload + # containers and the scrapers can authenticate over the compose network + # (guest is loopback-only by default and would otherwise be rejected). + - ./config/rabbitmq.conf:/etc/rabbitmq/conf.d/20-full-coverage.conf:ro + # No host-port publish: the consumers (perf-test, activity-gen, and the + # evalya rabbitmq-full task's scraper) reach the broker over the managed + # network via the DB_HOST label. Publishing 15672/5672 to the host only + # invites conflicts with a local broker. Add `--publish` by hand to inspect. + healthcheck: + test: ["CMD-SHELL", "rabbitmq-diagnostics -q check_running && rabbitmq-diagnostics -q check_port_connectivity"] + interval: 5s + timeout: 10s + retries: 20 + start_period: 30s + + # Continuous publish/consume/ack via the official PerfTest tool. Manual acks + # (no -a) so message.acknowledged / queue.messages.ack move; a bounded rate so + # the fixture stays light. + perf-test: + image: "pivotalrabbitmq/perf-test:latest" + depends_on: + rabbitmq: + condition: service_healthy + restart: unless-stopped + command: + - "--uri=amqp://guest:guest@rabbitmq:5672" + - "--producers=2" + - "--consumers=2" + - "--queue=perf-test-full" + - "--rate=50" + + # Churn sidecar: everything perf-test leaves at zero. pika is pip-installed at + # start (small, no custom image to maintain). Honors ACTIVITY_GEN=0 to idle. + activity-gen: + image: "python:3.12-slim" + depends_on: + rabbitmq: + condition: service_healthy + restart: unless-stopped + environment: + DB_HOST: rabbitmq + AMQP_PORT: "5672" + RABBITMQ_USER: guest + RABBITMQ_PASSWORD: guest + ACTIVITY_GEN: "${ACTIVITY_GEN:-1}" + volumes: + - ../activity-gen.py:/opt/activity-gen.py:ro + command: ["sh", "-c", "pip install --quiet --no-cache-dir pika==1.3.2 && exec python -u /opt/activity-gen.py"] diff --git a/rabbitmq/tests/evalya.yaml b/rabbitmq/tests/evalya.yaml new file mode 100644 index 0000000000000..188cf028e960d --- /dev/null +++ b/rabbitmq/tests/evalya.yaml @@ -0,0 +1,47 @@ +version: "1" + +# RabbitMQ fixtures for the rabbitmq check and the OTel rabbitmqreceiver. Both +# scrape the management HTTP API (:15672) of a real broker. +tasks: + # Plain management-enabled broker (mirrors the existing docker-compose.yaml). + - id: rabbitmq-standalone + task: ./compose/docker-compose.yaml@rabbitmq + labels: + evalya.io/publish: "true" + evalya.io/provides.DB_HOST: "{{ .hostname }}" + evalya.io/provides.DB_PORT: "15672" + evalya.io/provides.RABBITMQ_USER: guest + evalya.io/provides.RABBITMQ_PASSWORD: guest + env: + - name: RABBITMQ_VERSION + value: "4.0" + healthcheck: + test: ["CMD-SHELL", "rabbitmq-diagnostics -q check_running"] + interval: 10s + timeout: 10s + retries: 15 + start_period: 30s + + # Full metric-coverage fixture: a real management-enabled broker driven by a + # continuous PerfTest workload (publish/consume/ack) plus a churn sidecar + # (durable queue/exchange declare+delete, connection/channel churn, unroutable + # messages), so every node.* and message.* metric the OTel<->DD mappings need + # reports a non-trivial value. Presents on 15672 like the standalone broker. + # See ./compose/full-coverage.compose and ./activity-gen.py. + - id: rabbitmq-full + task: ./compose/full-coverage.compose@rabbitmq + labels: + evalya.io/publish: "true" + evalya.io/provides.DB_HOST: "{{ .hostname }}" + evalya.io/provides.DB_PORT: "15672" + evalya.io/provides.RABBITMQ_USER: guest + evalya.io/provides.RABBITMQ_PASSWORD: guest + env: + - name: RABBITMQ_VERSION + value: "4.0" + healthcheck: + test: ["CMD-SHELL", "rabbitmq-diagnostics -q check_running"] + interval: 10s + timeout: 10s + retries: 15 + start_period: 30s