mirror of
https://github.com/malarinv/iperf3-monitor.git
synced 2026-11-06 14:50:28 +00:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ec0a16353d | |||
| 0d93c9ea67 | |||
| 587290f1fb | |||
| 24904ef084 |
20
.github/workflows/ci.yaml
vendored
20
.github/workflows/ci.yaml
vendored
@@ -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
|
||||
|
||||
18
.github/workflows/release.yml
vendored
18
.github/workflows/release.yml
vendored
@@ -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
|
||||
|
||||
4
.gitignore
vendored
4
.gitignore
vendored
@@ -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
|
||||
|
||||
@@ -1,12 +1,9 @@
|
||||
dependencies:
|
||||
- name: kube-prometheus-stack
|
||||
repository: https://prometheus-community.github.io/helm-charts
|
||||
version: 75.7.0
|
||||
- name: prometheus-operator
|
||||
repository: oci://tccr.io/truecharts
|
||||
version: 11.5.1
|
||||
version: 91.5.3
|
||||
- name: common
|
||||
repository: https://bjw-s-labs.github.io/helm-charts/
|
||||
version: 4.1.2
|
||||
digest: sha256:68485b4e158a6a405073e9c59966d251b62971846cdc9871e41fde46f19aabfe
|
||||
generated: "2025-07-01T20:32:00.061995907Z"
|
||||
digest: sha256:a3a063101484521564d3e8922c1c1fe7ae8040fe28c2ae246a40d0c00b860e65
|
||||
generated: "2026-09-26T09:52:53.951178644+05:30"
|
||||
|
||||
@@ -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"
|
||||
- name: prometheus-operator
|
||||
version: ">=8.11.1"
|
||||
repository: "oci://tccr.io/truecharts"
|
||||
condition: "dependencies.install, serviceMonitor.enabled, dependencies.useTrueChartsPrometheusOperator"
|
||||
condition: "dependencies.install, serviceMonitor.enabled"
|
||||
- name: common
|
||||
version: "4.1.2"
|
||||
repository: "https://bjw-s-labs.github.io/helm-charts/"
|
||||
|
||||
194
charts/iperf3-monitor/grafana/iperf3-dashboard.json
Normal file
194
charts/iperf3-monitor/grafana/iperf3-dashboard.json
Normal file
@@ -0,0 +1,194 @@
|
||||
{
|
||||
"__inputs": [],
|
||||
"__requires": [
|
||||
{
|
||||
"type": "grafana",
|
||||
"id": "grafana",
|
||||
"name": "Grafana",
|
||||
"version": "8.0.0"
|
||||
},
|
||||
{
|
||||
"type": "datasource",
|
||||
"id": "prometheus",
|
||||
"name": "Prometheus",
|
||||
"version": "1.0.0"
|
||||
}
|
||||
],
|
||||
"annotations": {
|
||||
"list": [
|
||||
{
|
||||
"builtIn": 1,
|
||||
"datasource": {
|
||||
"type": "grafana",
|
||||
"uid": "-- Grafana --"
|
||||
},
|
||||
"enable": true,
|
||||
"hide": true,
|
||||
"iconColor": "rgba(0, 211, 255, 1)",
|
||||
"name": "Annotations & Alerts",
|
||||
"type": "dashboard"
|
||||
}
|
||||
]
|
||||
},
|
||||
"editable": true,
|
||||
"fiscalYearStartMonth": 0,
|
||||
"gnetId": null,
|
||||
"graphTooltip": 0,
|
||||
"id": null,
|
||||
"links": [],
|
||||
"panels": [
|
||||
{
|
||||
"datasource": {
|
||||
"type": "prometheus",
|
||||
"uid": "prometheus"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 9,
|
||||
"w": 24,
|
||||
"x": 0,
|
||||
"y": 0
|
||||
},
|
||||
"id": 2,
|
||||
"targets": [
|
||||
{
|
||||
"expr": "avg(iperf_network_bandwidth_mbps) by (source_node, destination_node)",
|
||||
"format": "heatmap",
|
||||
"legendFormat": "{{source_node}} -> {{destination_node}}",
|
||||
"refId": "A"
|
||||
}
|
||||
],
|
||||
"cards": { "cardPadding": null, "cardRound": null },
|
||||
"color": {
|
||||
"mode": "spectrum",
|
||||
"scheme": "red-yellow-green",
|
||||
"exponent": 0.5,
|
||||
"reverse": false
|
||||
},
|
||||
"dataFormat": "tsbuckets",
|
||||
"yAxis": { "show": true, "format": "short" },
|
||||
"xAxis": { "show": true }
|
||||
},
|
||||
{
|
||||
"title": "Bandwidth Over Time (Source: $source_node, Dest: $destination_node)",
|
||||
"type": "timeseries",
|
||||
"datasource": {
|
||||
"type": "prometheus",
|
||||
"uid": "prometheus"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 8,
|
||||
"w": 12,
|
||||
"x": 0,
|
||||
"y": 9
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
"expr": "iperf_network_bandwidth_mbps{source_node=~\"^$source_node$\", destination_node=~\"^$destination_node$\", protocol=~\"^$protocol$\"}",
|
||||
"legendFormat": "Bandwidth",
|
||||
"refId": "A"
|
||||
}
|
||||
],
|
||||
"fieldConfig": {
|
||||
"defaults": {
|
||||
"unit": "mbps"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"title": "Jitter Over Time (Source: $source_node, Dest: $destination_node)",
|
||||
"type": "timeseries",
|
||||
"datasource": {
|
||||
"type": "prometheus",
|
||||
"uid": "prometheus"
|
||||
},
|
||||
"gridPos": {
|
||||
"h": 8,
|
||||
"w": 12,
|
||||
"x": 12,
|
||||
"y": 9
|
||||
},
|
||||
"targets": [
|
||||
{
|
||||
"expr": "iperf_network_jitter_ms{source_node=~\"^$source_node$\", destination_node=~\"^$destination_node$\", protocol=\"udp\"}",
|
||||
"legendFormat": "Jitter",
|
||||
"refId": "A"
|
||||
}
|
||||
],
|
||||
"fieldConfig": {
|
||||
"defaults": {
|
||||
"unit": "ms"
|
||||
}
|
||||
}
|
||||
}
|
||||
],
|
||||
"refresh": "30s",
|
||||
"schemaVersion": 36,
|
||||
"style": "dark",
|
||||
"tags": ["iperf3", "network", "kubernetes"],
|
||||
"templating": {
|
||||
"list": [
|
||||
{
|
||||
"current": {},
|
||||
"datasource": {
|
||||
"type": "prometheus",
|
||||
"uid": "prometheus"
|
||||
},
|
||||
"definition": "label_values(iperf_network_bandwidth_mbps, source_node)",
|
||||
"hide": 0,
|
||||
"includeAll": false,
|
||||
"multi": false,
|
||||
"name": "source_node",
|
||||
"options": [],
|
||||
"query": "label_values(iperf_network_bandwidth_mbps, source_node)",
|
||||
"refresh": 1,
|
||||
"regex": "",
|
||||
"skipUrlSync": false,
|
||||
"sort": 1,
|
||||
"type": "query"
|
||||
},
|
||||
{
|
||||
"current": {},
|
||||
"datasource": {
|
||||
"type": "prometheus",
|
||||
"uid": "prometheus"
|
||||
},
|
||||
"definition": "label_values(iperf_network_bandwidth_mbps{source_node=~\"^$source_node$\"}, destination_node)",
|
||||
"hide": 0,
|
||||
"includeAll": false,
|
||||
"multi": false,
|
||||
"name": "destination_node",
|
||||
"options": [],
|
||||
"query": "label_values(iperf_network_bandwidth_mbps{source_node=~\"^$source_node$\"}, destination_node)",
|
||||
"refresh": 1,
|
||||
"regex": "",
|
||||
"skipUrlSync": false,
|
||||
"sort": 1,
|
||||
"type": "query"
|
||||
},
|
||||
{
|
||||
"current": { "selected": true, "text": "tcp", "value": "tcp" },
|
||||
"hide": 0,
|
||||
"includeAll": false,
|
||||
"multi": false,
|
||||
"name": "protocol",
|
||||
"options": [
|
||||
{ "selected": true, "text": "tcp", "value": "tcp" },
|
||||
{ "selected": false, "text": "udp", "value": "udp" }
|
||||
],
|
||||
"query": "tcp,udp",
|
||||
"skipUrlSync": false,
|
||||
"type": "custom"
|
||||
}
|
||||
]
|
||||
},
|
||||
"time": {
|
||||
"from": "now-1h",
|
||||
"to": "now"
|
||||
},
|
||||
"timepicker": {},
|
||||
"timezone": "browser",
|
||||
"title": "Kubernetes iperf3 Network Performance",
|
||||
"uid": "k8s-iperf3-dashboard",
|
||||
"version": 1,
|
||||
"weekStart": ""
|
||||
}
|
||||
@@ -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 -}}
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
apiVersion: v1
|
||||
kind: ConfigMap
|
||||
metadata:
|
||||
name: {{ .Release.Name }}-grafana-dashboard
|
||||
labels:
|
||||
grafana_dashboard: "1"
|
||||
app.kubernetes.io/name: {{ include "iperf3-monitor.name" . }}
|
||||
helm.sh/chart: {{ .Chart.Name }}-{{ .Chart.Version | replace "+" "_" }}
|
||||
app.kubernetes.io/instance: {{ .Release.Name }}
|
||||
app.kubernetes.io/managed-by: {{ .Release.Service }}
|
||||
data:
|
||||
iperf3-dashboard.json: |
|
||||
{{ .Files.Get "grafana/iperf3-dashboard.json" | nindent 4 }}
|
||||
@@ -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.
|
||||
testTimeout: 10
|
||||
# -- 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
|
||||
# -- 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
|
||||
|
||||
14
devbox.json
14
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"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,4 +0,0 @@
|
||||
{
|
||||
"lockfile_version": "1",
|
||||
"packages": {}
|
||||
}
|
||||
@@ -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.
|
||||
# This simplifies the Dockerfile by removing the need to copy iperf3 components from the builder.
|
||||
# Install iperf3
|
||||
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"]
|
||||
|
||||
|
||||
@@ -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,206 +122,317 @@ def discover_iperf_servers():
|
||||
|
||||
logging.info(f"Discovering iperf3 servers with label '{label_selector}' in namespace '{namespace}'")
|
||||
|
||||
ret = v1.list_pod_for_all_namespaces(label_selector=label_selector, watch=False)
|
||||
ret = v1.list_namespaced_pod(namespace=namespace, label_selector=label_selector, watch=False)
|
||||
|
||||
servers = []
|
||||
for item in ret.items:
|
||||
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.")
|
||||
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.
|
||||
|
||||
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).
|
||||
Executes the iperf3 CLI in a subprocess with JSON output and active stall detection.
|
||||
"""
|
||||
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()
|
||||
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
|
||||
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
|
||||
|
||||
try:
|
||||
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)
|
||||
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,
|
||||
):
|
||||
try:
|
||||
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)
|
||||
g.labels(**labels).set(0)
|
||||
except KeyError:
|
||||
logging.debug(f"KeyError setting failure metrics for {labels} after client.run() exception.")
|
||||
pass
|
||||
|
||||
|
||||
def parse_and_publish_metrics(result, source_node, dest_node, protocol):
|
||||
# --- Metric Parsing ---
|
||||
def parse_and_publish_metrics(data, source_node, dest_node, protocol):
|
||||
"""
|
||||
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').
|
||||
Parses iperf3 JSON result and updates Prometheus gauges.
|
||||
"""
|
||||
labels = {'source_node': source_node, 'destination_node': dest_node, 'protocol': protocol}
|
||||
|
||||
# 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.")
|
||||
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)
|
||||
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:
|
||||
# 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
|
||||
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)
|
||||
|
||||
IPERF_BANDWIDTH_MBPS.labels(**labels).set(bandwidth_mbps)
|
||||
|
||||
# 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:
|
||||
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 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,
|
||||
)
|
||||
|
||||
logging.info(f"Test cycle completed. Sleeping for {test_interval} seconds.")
|
||||
time.sleep(test_interval)
|
||||
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.")
|
||||
|
||||
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()
|
||||
|
||||
@@ -1,3 +1,2 @@
|
||||
prometheus-client
|
||||
iperf3
|
||||
kubernetes
|
||||
prometheus-client>=0.20.0
|
||||
kubernetes>=29.0.0
|
||||
111
exporter/test_exporter.py
Normal file
111
exporter/test_exporter.py
Normal file
@@ -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()
|
||||
17
mise.toml
Normal file
17
mise.toml
Normal file
@@ -0,0 +1,17 @@
|
||||
[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"
|
||||
Reference in New Issue
Block a user