Fix: Align exporter labels, image tags, and build process

- Helm Chart:
  - Added `app.kubernetes.io/component: exporter` to the exporter pod
    labels via `values.yaml` to match the service selector.
  - Updated image tag defaulting in `exporter-controller.yaml` to use
    `Chart.appVersion` directly (e.g., "0.1.0" instead of "v0.1.0").

- Build Process (`.github/workflows/release.yml`):
  - Configured `docker/metadata-action` to ensure image tags are generated
    without a 'v' prefix (e.g., "0.1.0" from Git tag "v0.1.0").
    This aligns the published image tags with the Helm chart's
    updated image tag references.

- Repository:
  - Added `rendered-manifests.yaml` and `rendered-manifests-updated.yaml`
    to `.gitignore`.

.github/workflows/ci.yaml

@@ -9,26 +9,6 @@ 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,19 +17,6 @@ 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:

charts/iperf3-monitor/Chart.lock

@@ -1,9 +1,12 @@
 dependencies:
 - name: kube-prometheus-stack
   repository: https://prometheus-community.github.io/helm-charts
-  version: 91.5.3
+  version: 75.7.0
+- 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:a3a063101484521564d3e8922c1c1fe7ae8040fe28c2ae246a40d0c00b860e65
-generated: "2026-09-26T09:52:53.951178644+05:30"
+digest: sha256:68485b4e158a6a405073e9c59966d251b62971846cdc9871e41fde46f19aabfe
+generated: "2025-07-01T20:32:00.061995907Z"

charts/iperf3-monitor/Chart.yaml

@@ -21,20 +21,17 @@ maintainers:
 icon: https://raw.githubusercontent.com/malarinv/iperf3-monitor/main/icon.png # Optional icon URL
 annotations:
   artifacthub.io/changes: |
-    - 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
+    - Add initial Helm chart structure.
   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"
+    condition: "dependencies.install, serviceMonitor.enabled, !dependencies.useTrueChartsPrometheusOperator"
+  - 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,11 +43,7 @@ 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"))) -}}
-    {{- $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) -}}
+    {{- $_ := set $baseExporterEnv "IPERF_TEST_TIMEOUT" ($exporterControllerConfig.appConfig.testTimeout | default "10" | 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 -}}

charts/iperf3-monitor/values.yaml

@@ -27,14 +27,8 @@ 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 (kept for backward compatibility).
-      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
+      # -- Timeout in seconds for a single iperf3 test run.
+      testTimeout: 10
       # -- Protocol to use for testing (tcp or udp).
       testProtocol: tcp
       # -- iperf3 server port to connect to. Should match the server's listening port.
@@ -210,3 +204,5 @@ 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

@@ -0,0 +1,14 @@
+{
+  "$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

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

exporter/Dockerfile

@@ -1,24 +1,34 @@
 # Stage 1: Build stage with dependencies
-FROM python:3.12-slim AS builder
+FROM python:3.9-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 --prefix=/install -r requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
 
 # Stage 2: Final runtime stage
-FROM python:3.12-slim
+FROM python:3.9-slim
 
 WORKDIR /app
 
-# Install iperf3
+# Install iperf3 and its runtime dependency libsctp1 directly in the final stage.
+# 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 && \
+    apt-get install -y --no-install-recommends iperf3 libsctp1 && \
     rm -rf /var/lib/apt/lists/*
 
 # Copy installed Python packages from the builder stage
-COPY --from=builder /install /usr/local
+COPY --from=builder /usr/local/lib/python3.9/site-packages /usr/local/lib/python3.9/site-packages
 
 # Copy the exporter application code
 COPY exporter.py .
@@ -26,9 +36,5 @@ 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,20 +4,24 @@ 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 subprocess
+import sys
 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',
@@ -25,16 +29,19 @@ parser.add_argument(
     choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
     help='Set the logging level. Overrides LOG_LEVEL environment variable. (Default: INFO)'
 )
-args, _ = parser.parse_known_args()
+args = parser.parse_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)',
@@ -60,61 +67,24 @@ 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']
-)
 
-# --- 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):
+def discover_iperf_servers():
     """
     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()
+        config.load_incluster_config() # Assumes running inside a Kubernetes pod
         v1 = client.CoreV1Api()
 
         namespace = os.getenv('IPERF_SERVER_NAMESPACE', 'default')
@@ -122,317 +92,208 @@ def discover_iperf_servers(source_node_name):
 
         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_name': item.spec.node_name # Node where the iperf server pod is running
                 })
         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 [] # Return empty list on error to avoid crashing the main loop
 
-# --- iperf3 Subprocess Execution with Stall Detection ---
-def execute_iperf_test(server_ip, server_port, protocol, duration, connect_timeout_ms, stall_threshold_sec):
+def run_iperf_test(server_ip, server_port, protocol, source_node_name, dest_node_name):
     """
-    Executes the iperf3 CLI in a subprocess with JSON output and active stall detection.
+    Runs a single iperf3 test against a specified server and publishes metrics.
+
+    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).
     """
-    global active_subprocess
+    logging.info(f"Running iperf3 {protocol.upper()} test from {source_node_name} to {dest_node_name} ({server_ip}:{server_port})")
 
-    cmd = [
-        "iperf3",
-        "-c", server_ip,
-        "-p", str(server_port),
-        "-t", str(duration),
-        "--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
+    iperf_client = iperf3.Client()
+    iperf_client.server_hostname = server_ip
+    iperf_client.port = server_port
+    iperf_client.protocol = protocol
+    iperf_client.duration = int(os.getenv('IPERF_TEST_DURATION', 5)) # Test duration in seconds
+    iperf_client.json_output = True # Enables easy parsing of results
 
     try:
-        while proc.poll() is None:
-            if shutdown_requested:
-                proc.kill()
-                proc.wait()
-                raise KeyboardInterrupt("Shutdown requested")
-
-            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,
-    ):
+        result = iperf_client.run()
+        parse_and_publish_metrics(result, source_node_name, dest_node_name, protocol)
+    except Exception as e:
+        # Catch unexpected errors during client.run() or parsing
+        logging.error(f"Exception during iperf3 test or metric parsing for {dest_node_name}: {e}")
+        labels = {'source_node': source_node_name, 'destination_node': dest_node_name, 'protocol': protocol}
+        IPERF_TEST_SUCCESS.labels(**labels).set(0)
         try:
-            g.labels(**labels).set(0)
+            IPERF_BANDWIDTH_MBPS.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:
-            pass
+            logging.debug(f"KeyError setting failure metrics for {labels} after client.run() exception.")
 
-# --- Metric Parsing ---
-def parse_and_publish_metrics(data, source_node, dest_node, protocol):
+
+def parse_and_publish_metrics(result, source_node, dest_node, protocol):
     """
-    Parses iperf3 JSON result and updates Prometheus gauges.
+    Parses the iperf3 test 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}
 
-    if not data or not isinstance(data, dict):
-        logging.warning(f"Test from {source_node} to {dest_node} ({protocol.upper()}) failed: empty or invalid result")
-        reset_failure_metrics(labels)
+    # Handle failed tests (e.g., server unreachable) or missing result object
+    if not result or result.error:
+        error_message = result.error if result and result.error else "No result object from iperf3 client"
+        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 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
+    # If we reach here, the test itself was successful in execution
     IPERF_TEST_SUCCESS.labels(**labels).set(1)
 
-    sum_received = end.get("sum_received") or {}
-    sum_sent = end.get("sum_sent") or {}
-    sum_udp = end.get("sum") or {}
-
-    bandwidth_mbps = 0.0
-
-    if protocol == 'tcp':
-        if "bits_per_second" in sum_received and sum_received["bits_per_second"] is not None:
-            bandwidth_mbps = sum_received["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
-
-        # TCP Retransmits
-        retransmits = sum_sent.get("retransmits", 0)
-        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)
-
-        # 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)
+    # Determine bandwidth:
+    # Order of preference: received_Mbps, sent_Mbps, Mbps, then JSON fallbacks.
+    # received_Mbps is often most relevant for TCP client perspective.
+    # 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)
 
-# --- 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
-
+    # UDP specific metrics
+    if protocol == 'udp':
+        # These attributes are specific to UDP tests in iperf3
+        IPERF_JITTER_MS.labels(**labels).set(getattr(result, 'jitter_ms', 0) if result.jitter_ms is not None else 0)
+        IPERF_PACKETS_TOTAL.labels(**labels).set(getattr(result, 'packets', 0) if result.packets is not None else 0)
+        IPERF_LOST_PACKETS.labels(**labels).set(getattr(result, 'lost_packets', 0) if result.lost_packets is not None else 0)
+    else:
+        # For TCP tests, ensure UDP-specific metrics are set to 0
         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
+             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
 
-        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()
+    protocol = os.getenv('IPERF_TEST_PROTOCOL', 'tcp').lower() # Ensure lowercase
     source_node_name = os.getenv('SOURCE_NODE_NAME')
 
-    # 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))
-
+    # SOURCE_NODE_NAME is crucial for labeling metrics correctly.
     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, Duration: {duration}s, Server Port: {server_port}, Protocol: {protocol.upper()}, "
-        f"Connect Timeout: {connect_timeout_ms}ms, Stall Threshold: {stall_threshold_sec}s"
+        f"Test Interval: {test_interval}s, Server Port: {server_port}, Protocol: {protocol.upper()}"
     )
 
-    while not shutdown_requested:
+    while True:
         logging.info("Starting new iperf test cycle...")
-        servers = discover_iperf_servers(source_node_name)
+        servers = discover_iperf_servers()
 
         if not servers:
             logging.warning("No iperf servers discovered in this cycle. Check K8s setup and RBAC permissions.")
         else:
             for server in servers:
-                if shutdown_requested:
-                    break
-
-                dest_node_name = server.get('node_name', 'unknown_destination_node')
+                dest_node_name = server.get('node_name', 'unknown_destination_node') # Default if key missing
                 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_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,
-                )
+                run_iperf_test(server_ip, server_port, protocol, source_node_name, dest_node_name)
 
-        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.")
-            slept = 0
-            while slept < test_interval and not shutdown_requested:
-                time.sleep(1)
-                slept += 1
-
-    logging.info("Exporter main loop stopped gracefully.")
+        logging.info(f"Test cycle completed. Sleeping for {test_interval} seconds.")
+        time.sleep(test_interval)
 
 if __name__ == '__main__':
-    signal.signal(signal.SIGTERM, signal_handler)
-    signal.signal(signal.SIGINT, signal_handler)
+    # Initial logging (like log level) is configured globally at the start of the script.
 
+    # 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)
+        sys.exit(1) # Exit if the metrics server cannot start
 
+    # 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,2 +1,3 @@
-prometheus-client>=0.20.0
-kubernetes>=29.0.0
+prometheus-client
+iperf3
+kubernetes

exporter/test_exporter.py

@@ -1,111 +0,0 @@
-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

@@ -1,17 +0,0 @@
-[tools]
-python = "3.12"
-uv = "latest"
-helm = "3.17.3"
-yq = "latest"
-
-[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"