Backend Listener, InfluxDB/Graphite, Grafana, and Real-Time Telemetry: Guided Hands-On Workflow

1. Safety and trust boundary

2. Create the pinned local telemetry stack

Save observability/compose.yaml:

name: p24-telemetry

services:
  influxdb:
    image: influxdb:2.9.1
    container_name: p24-influxdb
    ports:
      - "127.0.0.1:8086:8086"
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: p24admin
      DOCKER_INFLUXDB_INIT_PASSWORD: Prompt24-Local-Only-ChangeMe!
      DOCKER_INFLUXDB_INIT_ORG: devops-academy
      DOCKER_INFLUXDB_INIT_BUCKET: jmeter
      DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: p24-local-token-0123456789-do-not-reuse
    volumes:
      - p24-influxdb-data:/var/lib/influxdb2

  grafana:
    image: grafana/grafana:13.2.1
    container_name: p24-grafana
    ports:
      - "127.0.0.1:3000:3000"
    environment:
      GF_SECURITY_ADMIN_USER: p24admin
      GF_SECURITY_ADMIN_PASSWORD: Prompt24-Local-Only-ChangeMe!
      GF_USERS_ALLOW_SIGN_UP: "false"
    depends_on:
      - influxdb
    volumes:
      - p24-grafana-data:/var/lib/grafana

volumes:
  p24-influxdb-data:
  p24-grafana-data:

Start and inspect:

docker compose -f .\observability\compose.yaml up -d
docker compose -f .\observability\compose.yaml ps
docker inspect p24-influxdb --format "{{.Config.Image}}"
docker inspect p24-grafana --format "{{.Config.Image}}"

Expected images: influxdb:2.9.1 and grafana/grafana:13.2.1. Do not use latest; InfluxData has announced major latest-tag behavior changes, and floating tags break reproducibility.

3. Verify services before JMeter

curl --fail --silent http://127.0.0.1:8086/health
curl --fail --silent http://127.0.0.1:3000/api/health

InfluxDB must be healthy and Grafana must answer before you create a Backend Listener. Record container/native versions and current host time.

4. Create the local SUT fixture

Save fixtures/telemetry_fixture.py:

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse, parse_qs
import argparse
import json
import re
import threading
import time

FIXTURE_VERSION = "prompt24-telemetry-fixture-v1"
SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")

lock = threading.Lock()
event_log = None
state = {
    "mode": "baseline",
    "configured_delay_ms": 40,
    "inflight": 0,
    "work_requests": 0,
    "errors": 0,
    "last_control_ms": 0,
}

def now_ms():
    return int(time.time() * 1000)

def log_event(event):
    if event_log is None:
        return
    with lock:
        with event_log.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, sort_keys=True) + "\n")

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def send_json(self, status, payload):
        raw = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        self.end_headers()
        self.wfile.write(raw)

    def do_GET(self):
        started = now_ms()
        parsed = urlparse(self.path)

        if parsed.path == "/health":
            with lock:
                snapshot = dict(state)
            self.send_json(200, {
                "status": "ok",
                "fixture_version": FIXTURE_VERSION,
                "epoch_ms": now_ms(),
                "state": snapshot,
            })
            return

        if parsed.path == "/stats":
            with lock:
                snapshot = dict(state)
            self.send_json(200, {
                "fixture_version": FIXTURE_VERSION,
                "epoch_ms": now_ms(),
                "state": snapshot,
            })
            return

        if parsed.path != "/work":
            self.send_json(404, {"status": "not_found"})
            return

        q = parse_qs(parsed.query)
        run_id = q.get("run_id", [""])[0]
        thread_id = q.get("thread", [""])[0]
        seq_raw = q.get("seq", [""])[0]

        if not SAFE_TOKEN.fullmatch(run_id) or not SAFE_TOKEN.fullmatch(thread_id):
            self.send_json(400, {"status": "invalid_metadata"})
            return
        try:
            seq = int(seq_raw)
        except ValueError:
            seq = -1
        if not 1 <= seq <= 10000:
            self.send_json(400, {"status": "invalid_seq"})
            return

        with lock:
            state["inflight"] += 1
            mode = state["mode"]
            delay_ms = state["configured_delay_ms"]

        try:
            time.sleep(delay_ms / 1000.0)
            self.send_json(200, {
                "status": "ok",
                "run_id": run_id,
                "thread": thread_id,
                "seq": seq,
                "mode": mode,
                "configured_delay_ms": delay_ms,
            })
        finally:
            ended = now_ms()
            with lock:
                state["inflight"] -= 1
                state["work_requests"] += 1
            log_event({
                "ts_ms": ended,
                "operation": "work",
                "status": 200,
                "run_id": run_id,
                "thread": thread_id,
                "seq": seq,
                "mode": mode,
                "configured_delay_ms": delay_ms,
                "service_wall_ms": ended - started,
            })

    def do_POST(self):
        parsed = urlparse(self.path)
        if parsed.path != "/control":
            self.send_json(404, {"status": "not_found"})
            return

        q = parse_qs(parsed.query)
        mode = q.get("mode", [""])[0]
        run_id = q.get("run_id", ["control"])[0]

        if mode not in {"baseline", "degraded"} or not SAFE_TOKEN.fullmatch(run_id):
            self.send_json(400, {"status": "invalid_control"})
            return

        delay_ms = 40 if mode == "baseline" else 180
        changed = now_ms()
        with lock:
            state["mode"] = mode
            state["configured_delay_ms"] = delay_ms
            state["last_control_ms"] = changed

        log_event({
            "ts_ms": changed,
            "operation": "control",
            "status": 200,
            "run_id": run_id,
            "mode": mode,
            "configured_delay_ms": delay_ms,
        })
        self.send_json(200, {
            "status": "ok",
            "mode": mode,
            "configured_delay_ms": delay_ms,
            "epoch_ms": changed,
        })

    def log_message(self, format, *args):
        return

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8024)
    parser.add_argument("--log", default="results/server-events.jsonl")
    args = parser.parse_args()

    global event_log
    event_log = Path(args.log).resolve()
    event_log.parent.mkdir(parents=True, exist_ok=True)
    event_log.write_text("", encoding="utf-8")

    print(f"fixture_version={FIXTURE_VERSION}")
    print(f"listen=http://{args.host}:{args.port}")
    print(f"event_log={event_log}")
    ThreadingHTTPServer((args.host, args.port), Handler).serve_forever()

if __name__ == "__main__":
    main()

Start it:

python .\fixtures\telemetry_fixture.py `
  --host 127.0.0.1 `
  --port 8024 `
  --log .\results\server-events.jsonl

The fixture starts in baseline mode (40 ms delay). Degraded mode is 180 ms. The /control endpoint is loopback-only lab state; it is not a production failure-injection recommendation.

5. Create the independent SUT metric publisher

Save tools/publish_sut_metric.py:

import argparse
import json
import time
import urllib.parse
import urllib.request

def get_json(url):
    with urllib.request.urlopen(url, timeout=2) as response:
        return json.loads(response.read())

def write_line(url, token, line):
    request = urllib.request.Request(
        url,
        data=line.encode("utf-8"),
        method="POST",
        headers={
            "Authorization": f"Token {token}",
            "Content-Type": "text/plain; charset=utf-8",
        },
    )
    with urllib.request.urlopen(request, timeout=3) as response:
        if response.status not in (200, 204):
            raise RuntimeError(f"unexpected Influx status {response.status}")

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--stats-url", default="http://127.0.0.1:8024/stats")
    parser.add_argument("--influx-url", default="http://127.0.0.1:8086")
    parser.add_argument("--org", default="devops-academy")
    parser.add_argument("--bucket", default="jmeter")
    parser.add_argument("--token", required=True)
    parser.add_argument("--run-id", required=True)
    parser.add_argument("--duration", type=int, default=25)
    parser.add_argument("--interval", type=float, default=1.0)
    args = parser.parse_args()

    write_url = (
        f"{args.influx_url}/api/v2/write?"
        + urllib.parse.urlencode({
            "org": args.org,
            "bucket": args.bucket,
            "precision": "ms",
        })
    )

    deadline = time.monotonic() + args.duration
    sent = 0
    while time.monotonic() < deadline:
        doc = get_json(args.stats_url)
        s = doc["state"]
        ts = int(doc["epoch_ms"])
        # String fields are avoided; mode is encoded as a low-cardinality tag.
        line = (
            f"p24_sut,run_id={args.run_id},mode={s['mode']} "
            f"configured_delay_ms={int(s['configured_delay_ms'])}i,"
            f"inflight={int(s['inflight'])}i,"
            f"work_requests={int(s['work_requests'])}i "
            f"{ts}"
        )
        write_line(write_url, args.token, line)
        sent += 1
        time.sleep(args.interval)

    print(f"published_points={sent}")

if __name__ == "__main__":
    main()

This publisher reads /stats and writes a separate p24_sut measurement. It does not reuse JMeter SampleResults, so it is an independent server-state signal. It writes only numeric fields and low-cardinality run_id/mode tags.

6. Create the reversible control timeline

Save tools/control_timeline.py:

import argparse
import json
import time
import urllib.parse
import urllib.request

def post(base, mode, run_id):
    url = base + "/control?" + urllib.parse.urlencode({"mode": mode, "run_id": run_id})
    request = urllib.request.Request(url, method="POST", data=b"")
    with urllib.request.urlopen(request, timeout=2) as response:
        doc = json.loads(response.read())
    print(json.dumps(doc, sort_keys=True))

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--base", default="http://127.0.0.1:8024")
    parser.add_argument("--run-id", required=True)
    parser.add_argument("--degrade-after", type=float, default=4.0)
    parser.add_argument("--degraded-for", type=float, default=6.0)
    args = parser.parse_args()

    post(args.base, "baseline", args.run_id)
    time.sleep(args.degrade_after)
    post(args.base, "degraded", args.run_id)
    time.sleep(args.degraded_for)
    post(args.base, "baseline", args.run_id)

if __name__ == "__main__":
    main()

For the checkpoint it records baseline, switches to degraded after 4 seconds, remains degraded 6 seconds, then restores baseline. Each control transition is also written to the fixture JSONL with an epoch timestamp.

7. Create JMeter properties

config/local.properties:

# Prompt 24 local JMeter workload/result/Backend Listener baseline
target.host=127.0.0.1
target.port=8024
threads=2
loops=80
pacing.ms=100
connect.timeout.ms=500
response.timeout.ms=2000

# Keep forensic JTL lean and independent from real-time telemetry.
jmeter.save.saveservice.output_format=csv
jmeter.save.saveservice.print_field_names=true
jmeter.save.saveservice.response_data=false
jmeter.save.saveservice.response_data.on_error=false
jmeter.save.saveservice.samplerData=false
jmeter.save.saveservice.responseHeaders=false
jmeter.save.saveservice.requestHeaders=false

# Make the tiny lab visible in near-real time.
backend_influxdb.send_interval=1
backend_influxdb.connection_timeout=1000
backend_influxdb.socket_timeout=3000
backend_influxdb.connection_request_timeout=100

# Backend percentile/window policy.
backend_metrics_window_mode=fixed
backend_metrics_window=100
backend_metrics_percentile_estimator=R_3

The 1-second send interval is deliberately more frequent than JMeter's 5-second Influx default so the short lab has useful time resolution. That change itself adds telemetry traffic and therefore belongs in the overhead note.

8. Build the JMeter plan in GUI

Test Plan
├── HTTP Request Defaults
│   host=${__P(target.host,127.0.0.1)}
│   port=${__P(target.port,8024)}
└── Thread Group
    threads=${__P(threads,1)}
    loops=${__P(loops,1)}
    ├── Counter -> SEQ (per user)
    ├── HTTP Request — Work
    │   GET /work
    │   run_id=${__P(run.id,p24-local)}
    │   thread=T${__threadNum}
    │   seq=${SEQ}
    │   └── Constant Timer ${__P(pacing.ms,100)} ms
    └── Backend Listener — InfluxDB Live Metrics
        implementation=InfluxdbBackendListenerClient
        queue=500

Keep the Backend Listener under the Thread Group or Test Plan scope where its intent is obvious. It consumes SampleResults; it does not generate target requests.

9. Configure the Backend Listener exactly

Backend Listener implementation:
  org.apache.jmeter.visualizers.backend.influxdb.InfluxdbBackendListenerClient

Async Queue size:
  500

Parameters:
  influxdbMetricsSender =
    org.apache.jmeter.visualizers.backend.influxdb.HttpMetricsSender

  influxdbUrl =
    ${__P(influx.url,http://127.0.0.1:8086/api/v2/write?org=devops-academy&bucket=jmeter)}

  influxdbToken =
    ${__P(influx.token,p24-local-token-0123456789-do-not-reuse)}

  application = p24-lab
  measurement = jmeter
  summaryOnly = false
  samplersRegex = ^Work$
  percentiles = 90;95;99
  testTitle = ${__P(run.id,p24-local)}
  eventTags = local
  TAG_run_id = ${__P(run.id,p24-local)}
  TAG_env = local

State effects:

• summaryOnly=false + samplersRegex=^Work$ sends per-Work metrics plus cumulative all metrics.
• TAG_run_id lets Grafana isolate one run, but adds one tag value/series dimension per run; retention must remain bounded.
• TAG_env=local is deliberately low-cardinality.
• The fake token is read from a JMeter property. In real environments, do not expose reusable tokens in source or shell history.

10. Configure Grafana's InfluxDB datasource

Open Grafana locally, sign in with the disposable credentials, add an InfluxDB datasource:

• Product: InfluxDB OSS 2.x
• Query language: Flux
• URL from Grafana container: http://influxdb:8086 (native Grafana: http://127.0.0.1:8086)
• Organization: devops-academy
• Token: p24-local-token-0123456789-do-not-reuse
• Default bucket: jmeter

Click Save & test. Grafana queries the InfluxDB datasource server-side; the container-to-container hostname differs from the browser's localhost URL.

11. Send a tiny telemetry test first

Before the full run, override to 1 thread ×5 loops:

& "$env:JMETER_HOME\bin\jmeter.bat" `
  -n -t .\plans\telemetry-local.jmx `
  -q .\config\local.properties `
  -Jthreads=1 -Jloops=5 `
  -Jrun.id=p24-smoke `
  -Jinflux.token=p24-local-token-0123456789-do-not-reuse `
  -l .\results\p24-smoke\results.jtl `
  -j .\results\p24-smoke\jmeter.log

Wait a couple of send intervals. Check jmeter.log for Backend Listener errors and confirm Grafana Explore can see measurement jmeter with application p24-lab.

12. Query InfluxDB without Grafana

Save tools/query_influx.py:

import argparse
import urllib.parse
import urllib.request

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--url", default="http://127.0.0.1:8086")
    parser.add_argument("--org", default="devops-academy")
    parser.add_argument("--token", required=True)
    parser.add_argument("--flux", required=True)
    args = parser.parse_args()

    url = args.url + "/api/v2/query?" + urllib.parse.urlencode({"org": args.org})
    request = urllib.request.Request(
        url,
        data=args.flux.encode("utf-8"),
        method="POST",
        headers={
            "Authorization": f"Token {args.token}",
            "Content-Type": "application/vnd.flux",
            "Accept": "application/csv",
        },
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        print(response.read().decode("utf-8"))

if __name__ == "__main__":
    main()

Example:

python .\tools\query_influx.py `
  --token p24-local-token-0123456789-do-not-reuse `
  --flux 'from(bucket:"jmeter") |> range(start:-15m) |> filter(fn:(r)=> r._measurement=="jmeter") |> limit(n:20)'

This separates “Influx has no data” from “Grafana query/panel has no data.”

13. Build four simple Grafana panels

Panel A — JMeter Work avg + p95:
from(bucket: "jmeter")
  |> range(start: -30m)
  |> filter(fn: (r) => r._measurement == "jmeter")
  |> filter(fn: (r) => r.application == "p24-lab")
  |> filter(fn: (r) => r.run_id == "p24-check")
  |> filter(fn: (r) => r.transaction == "Work" and r.statut == "all")
  |> filter(fn: (r) => r._field == "avg" or r._field == "pct95.0")

Panel B — JMeter aggregate hits/window:
from(bucket: "jmeter")
  |> range(start: -30m)
  |> filter(fn: (r) => r._measurement == "jmeter")
  |> filter(fn: (r) => r.application == "p24-lab")
  |> filter(fn: (r) => r.run_id == "p24-check")
  |> filter(fn: (r) => r.transaction == "all" and r.statut == "all")
  |> filter(fn: (r) => r._field == "hit")

Panel C — JMeter mean active threads:
from(bucket: "jmeter")
  |> range(start: -30m)
  |> filter(fn: (r) => r._measurement == "jmeter")
  |> filter(fn: (r) => r.application == "p24-lab")
  |> filter(fn: (r) => r.run_id == "p24-check")
  |> filter(fn: (r) => r.transaction == "internal")
  |> filter(fn: (r) => r._field == "meanAT")

Panel D — independent SUT configured delay:
from(bucket: "jmeter")
  |> range(start: -30m)
  |> filter(fn: (r) => r._measurement == "p24_sut")
  |> filter(fn: (r) => r.run_id == "p24-check")
  |> filter(fn: (r) => r._field == "configured_delay_ms" or r._field == "inflight")

Use the same dashboard time range for every panel. Panel D is independent SUT state; Panels A–C are JMeter client/load metrics. Do not label the JMeter avg field as “server CPU” or the SUT configured delay as “JMeter latency.”

14. Run the live correlation experiment

Use three terminals.

Terminal A — SUT publisher:

python .\tools\publish_sut_metric.py `
  --token p24-local-token-0123456789-do-not-reuse `
  --run-id p24-live `
  --duration 25

Terminal B — control timeline:

python .\tools\control_timeline.py `
  --run-id p24-live `
  --degrade-after 4 `
  --degraded-for 6

Terminal C — JMeter CLI:

& "$env:JMETER_HOME\bin\jmeter.bat" `
  -n -t .\plans\telemetry-local.jmx `
  -q .\config\local.properties `
  -Jrun.id=p24-live `
  -Jinflux.token=p24-local-token-0123456789-do-not-reuse `
  -l .\results\p24-live\results.jtl `
  -j .\results\p24-live\jmeter.log

Expected live pattern: Panel D moves configured delay 40→180→40 ms. JMeter Work average/p95 rises shortly after the degraded transition and falls after recovery. Hit/completion rate may dip because this is a closed thread model.

15. Cross-check the timeline against JTL and target events

Save tools/correlate_timeline.py:

import csv
import json
import math
import sys
from pathlib import Path

def p95(values):
    data = sorted(values)
    if not data:
        return 0
    return data[max(0, min(len(data)-1, math.ceil(0.95 * len(data)) - 1))]

if len(sys.argv) != 4:
    raise SystemExit("usage: correlate_timeline.py results.jtl server-events.jsonl RUN_ID")

jtl_path = Path(sys.argv[1])
events_path = Path(sys.argv[2])
run_id = sys.argv[3]

events = [
    json.loads(line)
    for line in events_path.read_text(encoding="utf-8").splitlines()
    if line.strip()
]
controls = [
    e for e in events
    if e.get("operation") == "control" and e.get("run_id") == run_id
]
if len(controls) < 3:
    raise SystemExit(f"need baseline/degraded/recovery control events for {run_id}")

degrade_ts = next(e["ts_ms"] for e in controls if e["mode"] == "degraded")
recovery_ts = [
    e["ts_ms"] for e in controls
    if e["mode"] == "baseline" and e["ts_ms"] > degrade_ts
][0]

def phase(ts):
    if ts < degrade_ts:
        return "baseline_pre"
    if ts < recovery_ts:
        return "degraded"
    return "recovery"

jtl_rows = list(csv.DictReader(jtl_path.open(newline="", encoding="utf-8")))
jtl_phase = {}
for name in ("baseline_pre", "degraded", "recovery"):
    rows = [r for r in jtl_rows if phase(int(r["timeStamp"])) == name]
    elapsed = [int(float(r["elapsed"])) for r in rows]
    failures = sum(r["success"].lower() != "true" for r in rows)
    jtl_phase[name] = {
        "samples": len(rows),
        "failures": failures,
        "avg_ms": round(sum(elapsed)/len(elapsed), 2) if elapsed else 0,
        "p95_nearest_rank_ms": p95(elapsed),
    }

target_phase = {}
work = [
    e for e in events
    if e.get("operation") == "work" and e.get("run_id") == run_id
]
for name in ("baseline_pre", "degraded", "recovery"):
    rows = [e for e in work if phase(int(e["ts_ms"])) == name]
    elapsed = [int(e["service_wall_ms"]) for e in rows]
    target_phase[name] = {
        "events": len(rows),
        "avg_service_wall_ms": round(sum(elapsed)/len(elapsed), 2) if elapsed else 0,
        "p95_service_wall_ms": p95(elapsed),
        "delay_values": sorted({int(e["configured_delay_ms"]) for e in rows}),
    }

print(json.dumps({
    "run_id": run_id,
    "degrade_ts_ms": degrade_ts,
    "recovery_ts_ms": recovery_ts,
    "jtl": jtl_phase,
    "target": target_phase,
}, indent=2))

python tools/correlate_timeline.py   results/p24-live/results.jtl   results/server-events.jsonl   p24-live

Require the degraded phase to show higher raw JTL elapsed and target service_wall_ms, with delay values reflecting the control timeline. Grafana may aggregate into one-second windows; the raw files retain individual events.

16. Quantify Backend Listener overhead

Create telemetry-local-no-backend.jmx as an exact copy with only the Backend Listener disabled. For both variants force target baseline and use 2 threads ×40 loops ×100 ms pacing.

Record for each run:

• JMeter process CPU seconds / working set / heap snapshots;
• wall-clock duration;
• JTL rows and target work events (must both equal 80);
• InfluxDB/Grafana container CPU/memory via docker stats --no-stream;
• Influx point presence/count for the backend-enabled run;
• jmeter.log Backend Listener errors/timeouts;
• host network/disk state if available.

Do not claim a universal overhead percentage. Co-locating JMeter, target, InfluxDB and Grafana on one machine intentionally makes the lab simple but also couples their CPU/disk/network resources.

17. Challenge

Grafana Work p95 rises exactly when configured delay rises, but JTL p95 is stable. Which source should you trust?

Stop the conclusion. Verify run/tag/transaction filters and metric-window semantics first. JTL is the per-sample forensic record; Backend Listener p95 is an interval/window aggregate and may be querying a different run/series. Only after dataset alignment should the difference be interpreted.

Knowledge check

Official references and version notes

• JMeter Component Reference — Backend Listener — asynchronous listener queue, Graphite and InfluxDB clients, parameters, custom tags, raw-client resource warning.
• JMeter User Manual — Real-time Results — built-in InfluxDB/Graphite paths, live metrics, thread/response metrics, InfluxDB v2 setup and Grafana flow.
• JMeter Properties Reference — Backend Listener — send intervals, timeouts, metrics window and percentile estimator.
• InfluxDB OSS v2 — Write API — /api/v2/write, bucket/org/token and line protocol.
• InfluxDB OSS v2 release notes — 2.9.x compatibility baseline.
• Grafana InfluxDB data source — current product/version/query-language support.
• Grafana OSS downloads — current stable Grafana version.
• Apache JMeter downloads — current stable release and Java requirement.