test(rabbitmq): add rabbitmq-full evalya metric-coverage fixture

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:<ver>-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://<DB_HOST>: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
+```

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()

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

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"]

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