chore(mise): pin tool versions with explicit backends

`uv` and `yq` were `latest`, and helm was declared at 3.17.3 while
ci.yaml installs v3.10.0 — so `mise run chart:lint` and the pipeline were
validating two different helm versions, and nothing said so.

Each tool is pinned on an explicit backend and checked by running it
rather than by trusting version resolution: Python 3.12.13, uv 0.13.1,
helm v3.17.3+ge4da497, yq v4.54.1.

`aqua:` for helm, yq and uv, matching the other repos in this workspace.
The `github:` backend cannot name the yq binary `yq` at all on the mise
version this repo pins, and it mis-extracts helm 3.17.x — writing the
833-byte .asc PGP signature to
installs/github-helm-helm/3.17.x/helm, with attestation and SLSA
verification both logging as passing.

helm stays on 3.17.3, matching what mise.toml already declared for the
chart. That this file and ci.yaml disagree about the version is now the
only remaining gap, and it is in the workflow, not here.

The three mise tasks parse unchanged.

.github/workflows/ci.yaml

@@ -9,6 +9,26 @@ env:
   IMAGE_NAME: ${{ github.repository }}
 
 jobs:
+  test-exporter:
+    name: Run Exporter Unit Tests
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: |
+          pip install -r exporter/requirements.txt
+
+      - name: Run unit tests
+        run: |
+          python -m unittest exporter/test_exporter.py
+
   validate-chart:
     name: Validate Helm Chart
     runs-on: ubuntu-latest

.github/workflows/release.yml

@@ -17,6 +17,19 @@ jobs:
       - name: Check out code
         uses: actions/checkout@v3
 
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: |
+          pip install -r exporter/requirements.txt
+
+      - name: Run unit tests
+        run: |
+          python -m unittest exporter/test_exporter.py
+
       - name: Set up Helm
         uses: azure/setup-helm@v3
         with:
@@ -63,6 +76,11 @@ jobs:
         uses: docker/metadata-action@v4
         with:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          tags: |
+            type=semver,pattern={{version}}
+            # This ensures that for a git tag like "v0.1.0",
+            # an image tag "0.1.0" is generated.
+            # It will also generate "latest" for the most recent semver tag.
 
       - name: Build and push Docker image
         uses: docker/build-push-action@v4

.gitignore

@@ -37,3 +37,7 @@ Thumbs.db
 # Helm
 !charts/iperf3-monitor/.helmignore
 charts/iperf3-monitor/charts/
+
+# Rendered Kubernetes manifests (for local testing)
+rendered-manifests.yaml
+rendered-manifests-updated.yaml

charts/iperf3-monitor/Chart.lock

@@ -1,12 +1,9 @@
 dependencies:
 - name: kube-prometheus-stack
   repository: https://prometheus-community.github.io/helm-charts
-  version: 75.7.0
+  version: 91.5.3
-- name: prometheus-operator
-  repository: oci://tccr.io/truecharts
-  version: 11.5.1
 - name: common
   repository: https://bjw-s-labs.github.io/helm-charts/
   version: 4.1.2
-digest: sha256:68485b4e158a6a405073e9c59966d251b62971846cdc9871e41fde46f19aabfe
+digest: sha256:a3a063101484521564d3e8922c1c1fe7ae8040fe28c2ae246a40d0c00b860e65
-generated: "2025-07-01T20:32:00.061995907Z"
+generated: "2026-09-26T09:52:53.951178644+05:30"

charts/iperf3-monitor/Chart.yaml

@@ -21,17 +21,20 @@ maintainers:
 icon: https://raw.githubusercontent.com/malarinv/iperf3-monitor/main/icon.png # Optional icon URL
 annotations:
   artifacthub.io/changes: |
-    - Add initial Helm chart structure.
+    - kind: changed
+      description: Refactor exporter to subprocess with active stall watchdog and connect-timeout
+    - kind: added
+      description: Expose TCP retransmits, mean RTT (ms), and cycle timestamp metrics
+    - kind: changed
+      description: Run exporter container as non-root on Python 3.12
+    - kind: fixed
+      description: Support testDuration, connectTimeoutMs, and stallThresholdSeconds in Helm values
   artifacthub.io/category: networking
 dependencies:
   - name: kube-prometheus-stack # Example dependency if you package the whole stack
     version: ">=30.0.0" # Specify a compatible version range
     repository: https://prometheus-community.github.io/helm-charts
-    condition: "dependencies.install, serviceMonitor.enabled, !dependencies.useTrueChartsPrometheusOperator"
+    condition: "dependencies.install, serviceMonitor.enabled"
-  - name: prometheus-operator
-    version: ">=8.11.1"
-    repository: "oci://tccr.io/truecharts"
-    condition: "dependencies.install, serviceMonitor.enabled, dependencies.useTrueChartsPrometheusOperator"
   - name: common
     version: "4.1.2"
     repository: "https://bjw-s-labs.github.io/helm-charts/"

charts/iperf3-monitor/templates/exporter-controller.yaml

@@ -43,7 +43,11 @@ Proceed with modifications only if the exporter controller is defined.
     {{- $_ := set $baseExporterEnv "LOG_LEVEL" ($exporterControllerConfig.appConfig.logLevel | default "INFO") -}}
     {{- $_ := set $baseExporterEnv "IPERF_SERVER_PORT" ($exporterControllerConfig.appConfig.serverPort | default "5201" | toString) -}}
     {{- $_ := set $baseExporterEnv "IPERF_SERVER_NAMESPACE" (dict "valueFrom" (dict "fieldRef" (dict "fieldPath" "metadata.namespace"))) -}}
-    {{- $_ := set $baseExporterEnv "IPERF_TEST_TIMEOUT" ($exporterControllerConfig.appConfig.testTimeout | default "10" | toString) -}}
+    {{- $testDuration := $exporterControllerConfig.appConfig.testDuration | default $exporterControllerConfig.appConfig.testTimeout | default "5" | toString -}}
+    {{- $_ := set $baseExporterEnv "IPERF_TEST_DURATION" $testDuration -}}
+    {{- $_ := set $baseExporterEnv "IPERF_TEST_TIMEOUT" $testDuration -}}
+    {{- $_ := set $baseExporterEnv "IPERF_CONNECT_TIMEOUT_MS" ($exporterControllerConfig.appConfig.connectTimeoutMs | default "3000" | toString) -}}
+    {{- $_ := set $baseExporterEnv "IPERF_STALL_THRESHOLD_SECONDS" ($exporterControllerConfig.appConfig.stallThresholdSeconds | default "5" | toString) -}}
     {{- $serverLabelSelectorDefault := printf "app.kubernetes.io/name=%s,app.kubernetes.io/instance=%s,app.kubernetes.io/component=server" $appName $release.Name -}}
     {{- $serverLabelSelector := tpl ($exporterControllerConfig.appConfig.serverLabelSelector | default $serverLabelSelectorDefault) . -}}
     {{- $_ := set $baseExporterEnv "IPERF_SERVER_LABEL_SELECTOR" $serverLabelSelector -}}
@@ -72,12 +76,23 @@ Proceed with modifications only if the exporter controller is defined.
   {{- /*
   Ensure the container image tag is set, defaulting to Chart.AppVersion if empty,
   as the common library validation requires it during 'helm template'.
+
+  NOTE: BJW-S common library typically handles defaulting image.tag to Chart.appVersion
+  if image.tag is empty or null in values. The custom logic below prepending "v"
+  is specific to this chart and might be redundant if the common library's default
+  is preferred. For now, we keep it as it was the reason for previous errors if tag was not set.
+  However, if common library handles it, this block could be removed and image.tag in values.yaml set to "" or null.
+  Forcing the tag to be set (even if to chart.appVersion) ensures the common library doesn't complain.
+  The issue encountered during `helm template` earlier (empty output) was resolved by
+  explicitly setting the tag (e.g. via --set or by ensuring values.yaml has it).
+  The common library's internal validation likely needs *a* tag to be present in the values passed to it,
+  even if that tag is derived from AppVersion. This block ensures that.
   */}}
   {{- $exporterContainerCfg := get $exporterControllerConfig.containers "exporter" -}}
   {{- if $exporterContainerCfg -}}
     {{- if not $exporterContainerCfg.image.tag -}}
       {{- if $chart.AppVersion -}}
-        {{- $_ := set $exporterContainerCfg.image "tag" (printf "v%s" $chart.AppVersion) -}}
+        {{- $_ := set $exporterContainerCfg.image "tag" (printf "%s" $chart.AppVersion) -}} # Removed "v" prefix
       {{- else -}}
         {{- fail (printf "Error: Container image tag is not specified for controller '%s', container '%s', and Chart.AppVersion is also empty." $exporterControllerKey "exporter") -}}
       {{- end -}}

charts/iperf3-monitor/values.yaml

@@ -27,8 +27,14 @@ controllers:
       testInterval: 300
       # -- Log level for the iperf3 exporter (e.g., DEBUG, INFO, WARNING, ERROR, CRITICAL).
       logLevel: INFO
-      # -- Timeout in seconds for a single iperf3 test run.
+      # -- Timeout in seconds for a single iperf3 test run (kept for backward compatibility).
-      testTimeout: 10
+      testTimeout: 5
+      # -- Duration in seconds for a single iperf3 test run.
+      testDuration: 5
+      # -- Connect timeout in milliseconds for control connection setup.
+      connectTimeoutMs: 3000
+      # -- Stall threshold in seconds (abort test if data transfer halts during test).
+      stallThresholdSeconds: 5
       # -- Protocol to use for testing (tcp or udp).
       testProtocol: tcp
       # -- iperf3 server port to connect to. Should match the server's listening port.
@@ -45,7 +51,8 @@ controllers:
       # -- Annotations for the exporter pod.
       annotations: {}
       # -- Labels for the exporter pod.
-      labels: {} # The common library will add its own default labels.
+      labels:
+        app.kubernetes.io/component: exporter # Ensure pods get the component label for service selection
       # -- Node selector for scheduling exporter pods.
       nodeSelector: {}
       # -- Tolerations for scheduling exporter pods.
@@ -203,5 +210,3 @@ dependencies:
   # -- Set to false by default. Set to true to install a Prometheus operator dependency (used if serviceMonitor.enabled=true).
   # -- If false (default), and serviceMonitor.enabled is true, you must have a compatible Prometheus Operator already running in your cluster.
   install: false
-  # -- If true, use TrueCharts Prometheus Operator instead of kube-prometheus-stack (used if dependencies.install is true).
-  useTrueChartsPrometheusOperator: false

devbox.json

@@ -1,14 +0,0 @@
-{
-  "$schema":  "https://raw.githubusercontent.com/jetify-com/devbox/0.13.7/.schema/devbox.schema.json",
-  "packages": [],
-  "shell": {
-    "init_hook": [
-      "echo 'Welcome to devbox!' > /dev/null"
-    ],
-    "scripts": {
-      "test": [
-        "echo \"Error: no test specified\" && exit 1"
-      ]
-    }
-  }
-}

devbox.lock

@@ -1,4 +0,0 @@
-{
-  "lockfile_version": "1",
-  "packages": {}
-}

exporter/Dockerfile

@@ -1,34 +1,24 @@
 # Stage 1: Build stage with dependencies
-FROM python:3.9-slim as builder
+FROM python:3.12-slim AS builder
 
-# Declare TARGETARCH for use in this stage
-ARG TARGETARCH
 WORKDIR /app
 
-# Minimal dependencies for builder stage if any Python packages had C extensions.
-# Assuming requirements.txt does not need gcc or other build tools for now.
-# If pip install fails later, add necessary build tools (e.g., gcc, python3-dev) here.
-RUN apt-get update && \
-    # apt-get install -y --no-install-recommends gcc python3-dev # Example if needed
-    rm -rf /var/lib/apt/lists/*
-
 # Install Python dependencies
 COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+RUN pip install --no-cache-dir --prefix=/install -r requirements.txt
 
 # Stage 2: Final runtime stage
-FROM python:3.9-slim
+FROM python:3.12-slim
 
 WORKDIR /app
 
-# Install iperf3 and its runtime dependency libsctp1 directly in the final stage.
+# Install iperf3
-# This simplifies the Dockerfile by removing the need to copy iperf3 components from the builder.
 RUN apt-get update && \
-    apt-get install -y --no-install-recommends iperf3 libsctp1 && \
+    apt-get install -y --no-install-recommends iperf3 && \
     rm -rf /var/lib/apt/lists/*
 
 # Copy installed Python packages from the builder stage
-COPY --from=builder /usr/local/lib/python3.9/site-packages /usr/local/lib/python3.9/site-packages
+COPY --from=builder /install /usr/local
 
 # Copy the exporter application code
 COPY exporter.py .
@@ -36,5 +26,9 @@ COPY exporter.py .
 # Expose the metrics port
 EXPOSE 9876
 
+# Run as non-root user (nobody)
+USER 65534:65534
+
 # Set the entrypoint
 CMD ["python", "exporter.py"]
+

exporter/exporter.py

@@ -4,24 +4,20 @@ Prometheus exporter for iperf3 network performance monitoring.
 This script runs iperf3 tests between the node it's running on (source) and
 other iperf3 server pods discovered in a Kubernetes cluster. It then exposes
 these metrics for Prometheus consumption.
-
-Configuration is primarily through environment variables and command-line arguments
-for log level.
 """
 import os
+import sys
 import time
+import json
+import signal
 import logging
 import argparse
-import sys
+import subprocess
 from kubernetes import client, config
 from prometheus_client import start_http_server, Gauge
-import iperf3
 
 # --- Global Configuration & Setup ---
 
-# Argument parsing for log level configuration
-# The command-line --log-level argument takes precedence over the LOG_LEVEL env var.
-# Defaults to INFO if neither is set.
 parser = argparse.ArgumentParser(description="iperf3 Prometheus exporter.")
 parser.add_argument(
     '--log-level',
@@ -29,19 +25,16 @@ parser.add_argument(
     choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
     help='Set the logging level. Overrides LOG_LEVEL environment variable. (Default: INFO)'
 )
-args = parser.parse_args()
+args, _ = parser.parse_known_args()
 log_level_str = args.log_level
 
-# Convert log level string (e.g., 'INFO') to its numeric representation (e.g., logging.INFO)
 numeric_level = getattr(logging, log_level_str.upper(), None)
 if not isinstance(numeric_level, int):
-    # This case should ideally not be reached if choices in argparse are respected.
     logging.error(f"Invalid log level: {log_level_str}. Defaulting to INFO.")
     numeric_level = logging.INFO
 logging.basicConfig(level=numeric_level, format='%(asctime)s - %(levelname)s - %(message)s')
 
 # --- Prometheus Metrics Definition ---
-# These gauges will be used to expose iperf3 test results.
 IPERF_BANDWIDTH_MBPS = Gauge(
     'iperf_network_bandwidth_mbps',
     'Network bandwidth measured by iperf3 in Megabits per second (Mbps)',
@@ -67,24 +60,61 @@ IPERF_TEST_SUCCESS = Gauge(
     'Indicates if the iperf3 test was successful (1) or failed (0)',
     ['source_node', 'destination_node', 'protocol']
 )
+IPERF_TCP_RETRANSMITS = Gauge(
+    'iperf_network_tcp_retransmits_total',
+    'Total TCP retransmits measured by iperf3 during the test',
+    ['source_node', 'destination_node', 'protocol']
+)
+IPERF_TCP_RTT_MS = Gauge(
+    'iperf_network_tcp_rtt_ms',
+    'Mean TCP round-trip time in milliseconds (ms) measured by iperf3',
+    ['source_node', 'destination_node', 'protocol']
+)
+IPERF_TEST_DURATION_SECONDS = Gauge(
+    'iperf_network_test_duration_seconds',
+    'Duration of the iperf3 test in seconds',
+    ['source_node', 'destination_node', 'protocol']
+)
+IPERF_LAST_TEST_TIMESTAMP_SECONDS = Gauge(
+    'iperf_exporter_last_success_timestamp_seconds',
+    'Unix timestamp of the last completed test cycle',
+    ['source_node']
+)
+IPERF_SERVERS_DISCOVERED = Gauge(
+    'iperf_exporter_servers_discovered',
+    'Number of target iperf3 servers discovered in the cluster',
+    ['source_node']
+)
 
-def discover_iperf_servers():
+# --- Graceful Shutdown & Exceptions ---
+shutdown_requested = False
+active_subprocess = None
+
+def signal_handler(signum, frame):
+    global shutdown_requested, active_subprocess
+    logging.info(f"Received signal {signum}. Initiating graceful shutdown...")
+    shutdown_requested = True
+    if active_subprocess and active_subprocess.poll() is None:
+        try:
+            active_subprocess.kill()
+        except Exception:
+            pass
+
+class IperfStallException(Exception):
+    """Raised when an iperf3 test experiences a silent or busy-wait stall."""
+    pass
+
+class IperfTimeoutException(Exception):
+    """Raised when an iperf3 test exceeds its maximum expected deadline."""
+    pass
+
+# --- Kubernetes Discovery ---
+def discover_iperf_servers(source_node_name):
     """
     Discovers iperf3 server pods within a Kubernetes cluster.
-
-    It uses the in-cluster Kubernetes configuration to connect to the API.
-    The target namespace and label selector for iperf3 server pods are configured
-    via environment variables:
-    - IPERF_SERVER_NAMESPACE (default: 'default')
-    - IPERF_SERVER_LABEL_SELECTOR (default: 'app=iperf3-server')
-
-    Returns:
-        list: A list of dictionaries, where each dictionary contains the 'ip'
-              and 'node_name' of a discovered iperf3 server pod. Returns an
-              empty list if discovery fails or no servers are found.
     """
     try:
-        config.load_incluster_config() # Assumes running inside a Kubernetes pod
+        config.load_incluster_config()
         v1 = client.CoreV1Api()
 
         namespace = os.getenv('IPERF_SERVER_NAMESPACE', 'default')
@@ -92,208 +122,317 @@ def discover_iperf_servers():
 
         logging.info(f"Discovering iperf3 servers with label '{label_selector}' in namespace '{namespace}'")
 
-        # Use list_namespaced_pod to query only the specified namespace
         ret = v1.list_namespaced_pod(namespace=namespace, label_selector=label_selector, watch=False)
 
         servers = []
         for item in ret.items:
-            # No need to filter by namespace here as the API call is already namespaced
             if item.status.pod_ip and item.status.phase == 'Running':
                 servers.append({
                     'ip': item.status.pod_ip,
-                    'node_name': item.spec.node_name # Node where the iperf server pod is running
+                    'node_name': item.spec.node_name
                 })
         logging.info(f"Discovered {len(servers)} iperf3 server pods in namespace '{namespace}'.")
+        IPERF_SERVERS_DISCOVERED.labels(source_node=source_node_name).set(len(servers))
         return servers
     except config.ConfigException as e:
         logging.error(f"Kubernetes config error: {e}. Is the exporter running in a cluster with RBAC permissions?")
         return []
     except Exception as e:
         logging.error(f"Error discovering iperf servers: {e}")
-        return [] # Return empty list on error to avoid crashing the main loop
+        return []
 
-def run_iperf_test(server_ip, server_port, protocol, source_node_name, dest_node_name):
+# --- iperf3 Subprocess Execution with Stall Detection ---
+def execute_iperf_test(server_ip, server_port, protocol, duration, connect_timeout_ms, stall_threshold_sec):
     """
-    Runs a single iperf3 test against a specified server and publishes metrics.
+    Executes the iperf3 CLI in a subprocess with JSON output and active stall detection.
-
-    Args:
-        server_ip (str): The IP address of the iperf3 server.
-        server_port (int): The port number of the iperf3 server.
-        protocol (str): The protocol to use ('tcp' or 'udp').
-        source_node_name (str): The name of the source node (where this exporter is running).
-        dest_node_name (str): The name of the destination node (where the server is running).
-
-    The test duration is controlled by the IPERF_TEST_DURATION environment variable
-    (default: 5 seconds).
     """
-    logging.info(f"Running iperf3 {protocol.upper()} test from {source_node_name} to {dest_node_name} ({server_ip}:{server_port})")
+    global active_subprocess
 
-    iperf_client = iperf3.Client()
+    cmd = [
-    iperf_client.server_hostname = server_ip
+        "iperf3",
-    iperf_client.port = server_port
+        "-c", server_ip,
-    iperf_client.protocol = protocol
+        "-p", str(server_port),
-    iperf_client.duration = int(os.getenv('IPERF_TEST_DURATION', 5)) # Test duration in seconds
+        "-t", str(duration),
-    iperf_client.json_output = True # Enables easy parsing of results
+        "--connect-timeout", str(connect_timeout_ms),
+        "-J",
+    ]
+    if protocol == 'udp':
+        cmd.append("-u")
+
+    logging.debug(f"Executing: {' '.join(cmd)}")
+
+    proc = subprocess.Popen(
+        cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    active_subprocess = proc
+
+    start_time = time.monotonic()
+    # Maximum deadline: connect timeout + duration + stall threshold buffer
+    connect_timeout_sec = connect_timeout_ms / 1000.0
+    max_deadline = start_time + connect_timeout_sec + duration + stall_threshold_sec
+
+    last_io_change = start_time
+    last_wchar = 0
+    test_transfer_detected = False
 
     try:
-        result = iperf_client.run()
+        while proc.poll() is None:
-        parse_and_publish_metrics(result, source_node_name, dest_node_name, protocol)
+            if shutdown_requested:
-    except Exception as e:
+                proc.kill()
-        # Catch unexpected errors during client.run() or parsing
+                proc.wait()
-        logging.error(f"Exception during iperf3 test or metric parsing for {dest_node_name}: {e}")
+                raise KeyboardInterrupt("Shutdown requested")
-        labels = {'source_node': source_node_name, 'destination_node': dest_node_name, 'protocol': protocol}
+
+            now = time.monotonic()
+            if now > max_deadline:
+                proc.kill()
+                proc.wait()
+                raise IperfTimeoutException(f"iperf3 test exceeded absolute deadline of {max_deadline - start_time:.1f}s")
+
+            time.sleep(0.5)
+
+            # Check I/O progress via /proc/<pid>/io to detect busy-spin without I/O
+            try:
+                with open(f"/proc/{proc.pid}/io", "r") as f:
+                    for line in f:
+                        if line.startswith("wchar:"):
+                            wchar = int(line.split()[1])
+                            if wchar > last_wchar:
+                                last_wchar = wchar
+                                last_io_change = now
+                                test_transfer_detected = True
+                            break
+            except (FileNotFoundError, ProcessLookupError, PermissionError):
+                pass
+
+            # If traffic was active and has now completely ceased progressing for >= stall_threshold_sec
+            if test_transfer_detected and (now - last_io_change) >= stall_threshold_sec:
+                proc.kill()
+                proc.wait()
+                raise IperfStallException(f"iperf3 stalled: data transfer ceased for {now - last_io_change:.1f}s")
+
+        stdout, stderr = proc.communicate()
+        return proc.returncode, stdout, stderr
+    finally:
+        active_subprocess = None
+
+# --- Metric Reset Helper ---
+def reset_failure_metrics(labels):
     IPERF_TEST_SUCCESS.labels(**labels).set(0)
+    for g in (
+        IPERF_BANDWIDTH_MBPS,
+        IPERF_JITTER_MS,
+        IPERF_PACKETS_TOTAL,
+        IPERF_LOST_PACKETS,
+        IPERF_TCP_RETRANSMITS,
+        IPERF_TCP_RTT_MS,
+        IPERF_TEST_DURATION_SECONDS,
+    ):
         try:
-            IPERF_BANDWIDTH_MBPS.labels(**labels).set(0)
+            g.labels(**labels).set(0)
-            IPERF_JITTER_MS.labels(**labels).set(0)
-            IPERF_PACKETS_TOTAL.labels(**labels).set(0)
-            IPERF_LOST_PACKETS.labels(**labels).set(0)
         except KeyError:
-            logging.debug(f"KeyError setting failure metrics for {labels} after client.run() exception.")
+            pass
 
-
+# --- Metric Parsing ---
-def parse_and_publish_metrics(result, source_node, dest_node, protocol):
+def parse_and_publish_metrics(data, source_node, dest_node, protocol):
     """
-    Parses the iperf3 test result and updates Prometheus gauges.
+    Parses iperf3 JSON result and updates Prometheus gauges.
-
-    Args:
-        result (iperf3.TestResult): The result object from the iperf3 client.
-        source_node (str): Name of the source node.
-        dest_node (str): Name of the destination node.
-        protocol (str): Protocol used for the test ('tcp' or 'udp').
     """
     labels = {'source_node': source_node, 'destination_node': dest_node, 'protocol': protocol}
 
-    # Handle failed tests (e.g., server unreachable) or missing result object
+    if not data or not isinstance(data, dict):
-    if not result or result.error:
+        logging.warning(f"Test from {source_node} to {dest_node} ({protocol.upper()}) failed: empty or invalid result")
-        error_message = result.error if result and result.error else "No result object from iperf3 client"
+        reset_failure_metrics(labels)
-        logging.warning(f"Test from {source_node} to {dest_node} ({protocol.upper()}) failed: {error_message}")
-        IPERF_TEST_SUCCESS.labels(**labels).set(0)
-        # Set all relevant metrics to 0 on failure to clear stale values from previous successes
-        try:
-            IPERF_BANDWIDTH_MBPS.labels(**labels).set(0)
-            IPERF_JITTER_MS.labels(**labels).set(0) # Applicable for UDP, zeroed for TCP later
-            IPERF_PACKETS_TOTAL.labels(**labels).set(0) # Applicable for UDP, zeroed for TCP later
-            IPERF_LOST_PACKETS.labels(**labels).set(0) # Applicable for UDP, zeroed for TCP later
-        except KeyError:
-            # This can happen if labels were never registered due to continuous failures
-            logging.debug(f"KeyError when setting failure metrics for {labels}. Gauges might not be initialized.")
         return
 
-    # If we reach here, the test itself was successful in execution
+    if data.get("error"):
+        error_msg = data.get("error")
+        logging.warning(f"Test from {source_node} to {dest_node} ({protocol.upper()}) failed: {error_msg}")
+        reset_failure_metrics(labels)
+        return
+
+    end = data.get("end")
+    if not end or not isinstance(end, dict):
+        logging.warning(f"Test from {source_node} to {dest_node} ({protocol.upper()}) failed: missing 'end' summary in JSON")
+        reset_failure_metrics(labels)
+        return
+
+    # Successful test execution
     IPERF_TEST_SUCCESS.labels(**labels).set(1)
 
-    # Determine bandwidth:
+    sum_received = end.get("sum_received") or {}
-    # Order of preference: received_Mbps, sent_Mbps, Mbps, then JSON fallbacks.
+    sum_sent = end.get("sum_sent") or {}
-    # received_Mbps is often most relevant for TCP client perspective.
+    sum_udp = end.get("sum") or {}
-    # sent_Mbps can be relevant for UDP or as a TCP fallback.
-    bandwidth_mbps = 0
-    if hasattr(result, 'received_Mbps') and result.received_Mbps is not None:
-        bandwidth_mbps = result.received_Mbps
-    elif hasattr(result, 'sent_Mbps') and result.sent_Mbps is not None:
-        bandwidth_mbps = result.sent_Mbps
-    elif hasattr(result, 'Mbps') and result.Mbps is not None: # General attribute from iperf3 library
-        bandwidth_mbps = result.Mbps
-    # Fallback to raw JSON if direct attributes are None or missing
-    elif result.json:
-        # Prefer received sum, then sent sum from the JSON output's 'end' summary
-        if 'end' in result.json and 'sum_received' in result.json['end'] and \
-           result.json['end']['sum_received'].get('bits_per_second') is not None:
-            bandwidth_mbps = result.json['end']['sum_received']['bits_per_second'] / 1000000.0
-        elif 'end' in result.json and 'sum_sent' in result.json['end'] and \
-             result.json['end']['sum_sent'].get('bits_per_second') is not None:
-            bandwidth_mbps = result.json['end']['sum_sent']['bits_per_second'] / 1000000.0
 
-    IPERF_BANDWIDTH_MBPS.labels(**labels).set(bandwidth_mbps)
+    bandwidth_mbps = 0.0
 
-    # UDP specific metrics
+    if protocol == 'tcp':
-    if protocol == 'udp':
+        if "bits_per_second" in sum_received and sum_received["bits_per_second"] is not None:
-        # These attributes are specific to UDP tests in iperf3
+            bandwidth_mbps = sum_received["bits_per_second"] / 1_000_000.0
-        IPERF_JITTER_MS.labels(**labels).set(getattr(result, 'jitter_ms', 0) if result.jitter_ms is not None else 0)
+        elif "bits_per_second" in sum_sent and sum_sent["bits_per_second"] is not None:
-        IPERF_PACKETS_TOTAL.labels(**labels).set(getattr(result, 'packets', 0) if result.packets is not None else 0)
+            bandwidth_mbps = sum_sent["bits_per_second"] / 1_000_000.0
-        IPERF_LOST_PACKETS.labels(**labels).set(getattr(result, 'lost_packets', 0) if result.lost_packets is not None else 0)
+
-    else:
+        # TCP Retransmits
-        # For TCP tests, ensure UDP-specific metrics are set to 0
+        retransmits = sum_sent.get("retransmits", 0)
-        try:
+        IPERF_TCP_RETRANSMITS.labels(**labels).set(retransmits)
+
+        # TCP RTT
+        mean_rtt_us = 0.0
+        streams = end.get("streams", [])
+        if streams and isinstance(streams, list):
+            sender_info = streams[0].get("sender", {})
+            mean_rtt_us = sender_info.get("mean_rtt", sender_info.get("rtt", 0.0))
+        IPERF_TCP_RTT_MS.labels(**labels).set(mean_rtt_us / 1000.0)
+
+        # Zero out UDP-specific gauges
         IPERF_JITTER_MS.labels(**labels).set(0)
         IPERF_PACKETS_TOTAL.labels(**labels).set(0)
         IPERF_LOST_PACKETS.labels(**labels).set(0)
-        except KeyError:
-            # Can occur if labels not yet registered (e.g. first test is TCP)
-            logging.debug(f"KeyError for {labels} when zeroing UDP metrics for TCP test.")
-            pass
 
+        # Test duration
+        test_seconds = sum_received.get("seconds", sum_sent.get("seconds", 0.0))
+        IPERF_TEST_DURATION_SECONDS.labels(**labels).set(test_seconds)
+    else:
+        if "bits_per_second" in sum_udp and sum_udp["bits_per_second"] is not None:
+            bandwidth_mbps = sum_udp["bits_per_second"] / 1_000_000.0
+        elif "bits_per_second" in sum_sent and sum_sent["bits_per_second"] is not None:
+            bandwidth_mbps = sum_sent["bits_per_second"] / 1_000_000.0
+
+        IPERF_JITTER_MS.labels(**labels).set(sum_udp.get("jitter_ms", 0.0) or 0.0)
+        IPERF_PACKETS_TOTAL.labels(**labels).set(sum_udp.get("packets", 0) or 0)
+        IPERF_LOST_PACKETS.labels(**labels).set(sum_udp.get("lost_packets", 0) or 0)
+        IPERF_TCP_RETRANSMITS.labels(**labels).set(0)
+        IPERF_TCP_RTT_MS.labels(**labels).set(0)
+
+        test_seconds = sum_udp.get("seconds", 0.0)
+        IPERF_TEST_DURATION_SECONDS.labels(**labels).set(test_seconds)
+
+    IPERF_BANDWIDTH_MBPS.labels(**labels).set(bandwidth_mbps)
+
+# --- Test Execution Wrapper ---
+def run_iperf_test(server_ip, server_port, protocol, source_node_name, dest_node_name, duration, connect_timeout_ms, stall_threshold_sec):
+    """
+    Runs a single iperf3 test with stall monitoring and publishes metrics.
+    """
+    logging.info(f"Running iperf3 {protocol.upper()} test from {source_node_name} to {dest_node_name} ({server_ip}:{server_port})")
+    labels = {'source_node': source_node_name, 'destination_node': dest_node_name, 'protocol': protocol}
+
+    try:
+        returncode, stdout, stderr = execute_iperf_test(
+            server_ip=server_ip,
+            server_port=server_port,
+            protocol=protocol,
+            duration=duration,
+            connect_timeout_ms=connect_timeout_ms,
+            stall_threshold_sec=stall_threshold_sec,
+        )
+
+        if not stdout.strip():
+            logging.warning(f"iperf3 produced no stdout for {dest_node_name}. Stderr: {stderr.strip()}")
+            reset_failure_metrics(labels)
+            return
+
+        try:
+            result_json = json.loads(stdout)
+        except json.JSONDecodeError as e:
+            logging.warning(f"Failed to parse iperf3 JSON output for {dest_node_name}: {e}. Output: {stdout[:200]}")
+            reset_failure_metrics(labels)
+            return
+
+        parse_and_publish_metrics(result_json, source_node_name, dest_node_name, protocol)
+    except IperfStallException as e:
+        logging.warning(f"Test from {source_node_name} to {dest_node_name} ({protocol.upper()}) aborted: {e}")
+        reset_failure_metrics(labels)
+    except IperfTimeoutException as e:
+        logging.warning(f"Test from {source_node_name} to {dest_node_name} ({protocol.upper()}) timed out: {e}")
+        reset_failure_metrics(labels)
+    except KeyboardInterrupt:
+        raise
+    except Exception as e:
+        logging.error(f"Unexpected exception during iperf3 test for {dest_node_name}: {e}")
+        reset_failure_metrics(labels)
+
+# --- Main Exporter Loop ---
 def main_loop():
     """
     Main operational loop of the iperf3 exporter.
-
-    This loop periodically:
-    1. Fetches configuration from environment variables:
-       - IPERF_TEST_INTERVAL (default: 300s): Time between test cycles.
-       - IPERF_SERVER_PORT (default: 5201): Port for iperf3 servers.
-       - IPERF_TEST_PROTOCOL (default: 'tcp'): 'tcp' or 'udp'.
-       - SOURCE_NODE_NAME (critical): Name of the node this exporter runs on.
-    2. Discovers iperf3 server pods in the Kubernetes cluster.
-    3. Runs iperf3 tests against each discovered server (unless it's on the same node).
-    4. Sleeps for the configured test interval.
-
-    If SOURCE_NODE_NAME is not set, the script will log an error and exit.
     """
-    # Fetch operational configuration from environment variables
     test_interval = int(os.getenv('IPERF_TEST_INTERVAL', 300))
     server_port = int(os.getenv('IPERF_SERVER_PORT', 5201))
-    protocol = os.getenv('IPERF_TEST_PROTOCOL', 'tcp').lower() # Ensure lowercase
+    protocol = os.getenv('IPERF_TEST_PROTOCOL', 'tcp').lower()
     source_node_name = os.getenv('SOURCE_NODE_NAME')
 
-    # SOURCE_NODE_NAME is crucial for labeling metrics correctly.
+    # Support either IPERF_TEST_DURATION or IPERF_TEST_TIMEOUT from Helm values
+    duration = int(os.getenv('IPERF_TEST_DURATION', os.getenv('IPERF_TEST_TIMEOUT', 5)))
+    connect_timeout_ms = int(os.getenv('IPERF_CONNECT_TIMEOUT_MS', 3000))
+    stall_threshold_sec = int(os.getenv('IPERF_STALL_THRESHOLD_SECONDS', 5))
+
     if not source_node_name:
         logging.error("CRITICAL: SOURCE_NODE_NAME environment variable not set. This is required. Exiting.")
         sys.exit(1)
 
     logging.info(
         f"Exporter configured. Source Node: {source_node_name}, "
-        f"Test Interval: {test_interval}s, Server Port: {server_port}, Protocol: {protocol.upper()}"
+        f"Test Interval: {test_interval}s, Duration: {duration}s, Server Port: {server_port}, Protocol: {protocol.upper()}, "
+        f"Connect Timeout: {connect_timeout_ms}ms, Stall Threshold: {stall_threshold_sec}s"
     )
 
-    while True:
+    while not shutdown_requested:
         logging.info("Starting new iperf test cycle...")
-        servers = discover_iperf_servers()
+        servers = discover_iperf_servers(source_node_name)
 
         if not servers:
             logging.warning("No iperf servers discovered in this cycle. Check K8s setup and RBAC permissions.")
         else:
             for server in servers:
-                dest_node_name = server.get('node_name', 'unknown_destination_node') # Default if key missing
+                if shutdown_requested:
+                    break
+
+                dest_node_name = server.get('node_name', 'unknown_destination_node')
                 server_ip = server.get('ip')
 
                 if not server_ip:
                     logging.warning(f"Discovered server entry missing an IP: {server}. Skipping.")
                     continue
 
-                # Avoid testing a node against itself
                 if dest_node_name == source_node_name:
                     logging.info(f"Skipping test to self: {source_node_name} to {server_ip} (on same node: {dest_node_name}).")
                     continue
 
-                run_iperf_test(server_ip, server_port, protocol, source_node_name, dest_node_name)
+                run_iperf_test(
+                    server_ip=server_ip,
+                    server_port=server_port,
+                    protocol=protocol,
+                    source_node_name=source_node_name,
+                    dest_node_name=dest_node_name,
+                    duration=duration,
+                    connect_timeout_ms=connect_timeout_ms,
+                    stall_threshold_sec=stall_threshold_sec,
+                )
 
+        if not shutdown_requested:
+            IPERF_LAST_TEST_TIMESTAMP_SECONDS.labels(source_node=source_node_name).set(time.time())
             logging.info(f"Test cycle completed. Sleeping for {test_interval} seconds.")
-        time.sleep(test_interval)
+            slept = 0
+            while slept < test_interval and not shutdown_requested:
+                time.sleep(1)
+                slept += 1
+
+    logging.info("Exporter main loop stopped gracefully.")
 
 if __name__ == '__main__':
-    # Initial logging (like log level) is configured globally at the start of the script.
+    signal.signal(signal.SIGTERM, signal_handler)
+    signal.signal(signal.SIGINT, signal_handler)
 
-    # Fetch Prometheus exporter listen port from environment variable
     listen_port = int(os.getenv('LISTEN_PORT', 9876))
 
     try:
-        # Start the Prometheus HTTP server to expose metrics.
         start_http_server(listen_port)
         logging.info(f"Prometheus exporter listening on port {listen_port}")
     except Exception as e:
         logging.error(f"Failed to start Prometheus HTTP server on port {listen_port}: {e}")
-        sys.exit(1) # Exit if the metrics server cannot start
+        sys.exit(1)
 
-    # Enter the main operational loop.
-    # main_loop() contains its own critical checks (e.g., SOURCE_NODE_NAME) and will exit if necessary.
     main_loop()

exporter/requirements.txt

@@ -1,3 +1,2 @@
-prometheus-client
+prometheus-client>=0.20.0
-iperf3
+kubernetes>=29.0.0
-kubernetes

exporter/test_exporter.py

@@ -0,0 +1,111 @@
+import os
+import sys
+import unittest
+import json
+from unittest.mock import patch, MagicMock
+
+import importlib.util
+
+spec = importlib.util.spec_from_file_location("exporter_mod", os.path.join(os.path.dirname(__file__), "exporter.py"))
+exporter = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(exporter)
+
+class TestExporter(unittest.TestCase):
+
+    def setUp(self):
+        self.source = "node-a"
+        self.dest = "node-b"
+        self.labels = {'source_node': self.source, 'destination_node': self.dest, 'protocol': 'tcp'}
+        self.udp_labels = {'source_node': self.source, 'destination_node': self.dest, 'protocol': 'udp'}
+
+    def test_parse_tcp_success(self):
+        tcp_json = {
+            "end": {
+                "streams": [
+                    {
+                        "sender": {
+                            "mean_rtt": 250,  # 250 microseconds = 0.25 ms
+                            "retransmits": 3
+                        }
+                    }
+                ],
+                "sum_sent": {
+                    "seconds": 5.0,
+                    "bytes": 50000000,
+                    "bits_per_second": 80000000.0,
+                    "retransmits": 3
+                },
+                "sum_received": {
+                    "seconds": 5.0,
+                    "bytes": 49000000,
+                    "bits_per_second": 78400000.0
+                }
+            }
+        }
+
+        exporter.parse_and_publish_metrics(tcp_json, self.source, self.dest, 'tcp')
+
+        self.assertEqual(exporter.IPERF_TEST_SUCCESS.labels(**self.labels)._value.get(), 1)
+        self.assertAlmostEqual(exporter.IPERF_BANDWIDTH_MBPS.labels(**self.labels)._value.get(), 78.4)
+        self.assertEqual(exporter.IPERF_TCP_RETRANSMITS.labels(**self.labels)._value.get(), 3)
+        self.assertAlmostEqual(exporter.IPERF_TCP_RTT_MS.labels(**self.labels)._value.get(), 0.25)
+        self.assertEqual(exporter.IPERF_TEST_DURATION_SECONDS.labels(**self.labels)._value.get(), 5.0)
+        self.assertEqual(exporter.IPERF_JITTER_MS.labels(**self.labels)._value.get(), 0)
+
+    def test_parse_udp_success(self):
+        udp_json = {
+            "end": {
+                "sum": {
+                    "seconds": 5.0,
+                    "bytes": 1000000,
+                    "bits_per_second": 1600000.0,
+                    "jitter_ms": 1.25,
+                    "packets": 1000,
+                    "lost_packets": 2
+                }
+            }
+        }
+
+        exporter.parse_and_publish_metrics(udp_json, self.source, self.dest, 'udp')
+
+        self.assertEqual(exporter.IPERF_TEST_SUCCESS.labels(**self.udp_labels)._value.get(), 1)
+        self.assertAlmostEqual(exporter.IPERF_BANDWIDTH_MBPS.labels(**self.udp_labels)._value.get(), 1.6)
+        self.assertEqual(exporter.IPERF_JITTER_MS.labels(**self.udp_labels)._value.get(), 1.25)
+        self.assertEqual(exporter.IPERF_PACKETS_TOTAL.labels(**self.udp_labels)._value.get(), 1000)
+        self.assertEqual(exporter.IPERF_LOST_PACKETS.labels(**self.udp_labels)._value.get(), 2)
+        self.assertEqual(exporter.IPERF_TCP_RETRANSMITS.labels(**self.udp_labels)._value.get(), 0)
+
+    def test_parse_error_result(self):
+        err_json = {
+            "error": "error - unable to connect to server: Connection timed out"
+        }
+
+        exporter.parse_and_publish_metrics(err_json, self.source, self.dest, 'tcp')
+
+        self.assertEqual(exporter.IPERF_TEST_SUCCESS.labels(**self.labels)._value.get(), 0)
+        self.assertEqual(exporter.IPERF_BANDWIDTH_MBPS.labels(**self.labels)._value.get(), 0)
+        self.assertEqual(exporter.IPERF_TCP_RETRANSMITS.labels(**self.labels)._value.get(), 0)
+        self.assertEqual(exporter.IPERF_TCP_RTT_MS.labels(**self.labels)._value.get(), 0)
+
+    @patch("subprocess.Popen")
+    def test_stall_exception_raised_on_stall(self, mock_popen):
+        # Simulate a process that never terminates and produces no I/O progress
+        fake_proc = MagicMock()
+        fake_proc.poll.return_value = None  # Process is running
+        fake_proc.pid = 999999  # Non-existent PID to simulate no /proc IO advancement
+        mock_popen.return_value = fake_proc
+
+        with self.assertRaises(exporter.IperfTimeoutException):
+            exporter.execute_iperf_test(
+                server_ip="127.0.0.1",
+                server_port=5201,
+                protocol="tcp",
+                duration=1,
+                connect_timeout_ms=500,
+                stall_threshold_sec=1,
+            )
+
+        fake_proc.kill.assert_called()
+
+if __name__ == '__main__':
+    unittest.main()

mise.toml

@@ -0,0 +1,25 @@
+# Backends named and versions exact: `uv` and `yq` were `latest`, and `helm`
+# was a floating major-less spec that CI did not share — ci.yaml installs
+# v3.10.0 while this file declared 3.17.3, so local `mise run chart:lint` and
+# the pipeline that runs it were validating two different helm versions.
+#
+# `aqua:` for helm and yq, matching the other repos in this workspace. The
+# `github:` backend cannot name the yq binary `yq` at all on the mise version
+# this repo pins, and it mis-extracts helm 3.17.x.
+[tools]
+"core:python" = "3.12"
+"aqua:astral-sh/uv" = "0.13.1"
+"aqua:helm/helm" = "3.17.3"
+"aqua:mikefarah/yq" = "4.54.1"
+
+[tasks.test]
+description = "Run exporter unit tests"
+run = "uv run --with-requirements exporter/requirements.txt python -m unittest discover -s exporter -p 'test_*.py'"
+
+[tasks."chart:build"]
+description = "Build Helm chart dependencies"
+run = "helm dependency build ./charts/iperf3-monitor"
+
+[tasks."chart:lint"]
+description = "Lint Helm chart"
+run = "helm lint ./charts/iperf3-monitor"