tools: Add traffic_grapher for real-time ATS metrics visualization #12848
Conversation
A Python tool that displays ATS metrics inline in iTerm2 using imgcat
with live updates and multi-host comparison.
Features:
- Real-time graphs of RPS, latency, cache hit rate, connections
- Support for 1-4 hosts with different line styles for comparison
- Collects metrics via JSONRPC Unix socket (batch collection)
- Dark theme optimized for terminal display
- Keyboard navigation between metric pages (4 pages)
- Configurable refresh interval and history window
Requirements:
- Python 3 with matplotlib
- iTerm2 (or compatible terminal for inline images)
- SSH access to remote ATS hosts
Usage:
traffic_grapher.py ats-server1.example.com
traffic_grapher.py ats{1..4}.example.com --interval 2
There was a problem hiding this comment.
Pull request overview
A Python-based traffic monitoring tool that visualizes real-time ATS (Apache Traffic Server) metrics using inline terminal graphics. The tool collects metrics via JSONRPC Unix socket and renders them as time-series graphs directly in iTerm2, supporting multi-host comparison with keyboard-driven navigation.
Changes:
- New traffic_grapher.py script with real-time metrics visualization
- Support for 1-4 hosts with distinct line styles for comparison
- Four pre-configured dashboard pages covering traffic, cache, TLS/HTTP2, and network metrics
- Dark theme optimized for terminal display with configurable refresh and history
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| # Default path (adjust for your installation) | ||
| # Note: awk runs locally to avoid SSH quote escaping issues | ||
| METRIC_COMMAND_REMOTE = "/opt/edge/trafficserver/10.0/bin/traffic_ctl metric get {key}" | ||
| METRIC_COMMAND_LOCAL = "/opt/edge/trafficserver/10.0/bin/traffic_ctl metric get {key} | awk '{{print $2}}'" |
There was a problem hiding this comment.
Hard-coded installation paths make the tool less portable. Consider making these configurable via command-line arguments or environment variables to support different TrafficServer installations.
| # Default path (adjust for your installation) | |
| # Note: awk runs locally to avoid SSH quote escaping issues | |
| METRIC_COMMAND_REMOTE = "/opt/edge/trafficserver/10.0/bin/traffic_ctl metric get {key}" | |
| METRIC_COMMAND_LOCAL = "/opt/edge/trafficserver/10.0/bin/traffic_ctl metric get {key} | awk '{{print $2}}'" | |
| # Default path can be overridden via the TRAFFIC_CTL_PATH environment variable. | |
| # Note: awk runs locally to avoid SSH quote escaping issues | |
| TRAFFIC_CTL_PATH = os.environ.get("TRAFFIC_CTL_PATH", "/opt/edge/trafficserver/10.0/bin/traffic_ctl") | |
| METRIC_COMMAND_REMOTE = f"{TRAFFIC_CTL_PATH} metric get {{key}}" | |
| METRIC_COMMAND_LOCAL = f"{TRAFFIC_CTL_PATH} metric get {{key}} | awk '{{print $2}}'" |
| # Default JSONRPC socket path (adjust for your installation) | ||
| JSONRPC_SOCKET_PATH = "/opt/edge/trafficserver/10.0/var/trafficserver/jsonrpc20.sock" |
There was a problem hiding this comment.
Hard-coded socket path should be configurable via command-line argument or environment variable to support different TrafficServer installations.
| # Default JSONRPC socket path (adjust for your installation) | |
| JSONRPC_SOCKET_PATH = "/opt/edge/trafficserver/10.0/var/trafficserver/jsonrpc20.sock" | |
| # Default JSONRPC socket path (can be overridden via environment variable) | |
| JSONRPC_SOCKET_PATH = os.environ.get( | |
| "TRAFFICSERVER_JSONRPC_SOCKET", | |
| "/opt/edge/trafficserver/10.0/var/trafficserver/jsonrpc20.sock", | |
| ) |
| # Build SSH command - hostname is passed directly, we add "ssh" prefix | ||
| # Encode script as base64 to avoid quoting issues | ||
| script_b64 = base64.b64encode(script.encode()).decode() | ||
| cmd = f"ssh {self.hostname} \"echo '{script_b64}' | base64 -d | python3\"" |
There was a problem hiding this comment.
Command injection vulnerability. The hostname is passed directly into a shell command without validation or escaping. Validate or sanitize hostname before use in shell commands.
| # Build SSH command - hostname is passed directly, we add "ssh" prefix | ||
| # Encode script as base64 to avoid quoting issues | ||
| script_b64 = base64.b64encode(script.encode()).decode() | ||
| cmd = f"ssh {self.hostname} \"echo '{script_b64}' | base64 -d | python3\"" | ||
|
|
||
| try: | ||
| result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10) |
There was a problem hiding this comment.
Using shell=True with user-controlled input (hostname) creates a security risk. Consider using subprocess.run with a list of arguments instead of a shell string.
| # Build SSH command - hostname is passed directly, we add "ssh" prefix | |
| # Encode script as base64 to avoid quoting issues | |
| script_b64 = base64.b64encode(script.encode()).decode() | |
| cmd = f"ssh {self.hostname} \"echo '{script_b64}' | base64 -d | python3\"" | |
| try: | |
| result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10) | |
| # Run the JSONRPC collection script on the remote host via SSH. | |
| # Pass the script via stdin to avoid shell invocation and quoting issues. | |
| cmd = ["ssh", self.hostname, "python3"] | |
| try: | |
| result = subprocess.run(cmd, input=script, capture_output=True, text=True, timeout=10) |
|
|
||
| # Estimate from rows/cols (typical cell: 9x18 pixels) | ||
| return (cols * 9, rows * 18) | ||
| except: |
There was a problem hiding this comment.
Bare except clause catches all exceptions including SystemExit and KeyboardInterrupt. Specify explicit exception types like 'except (OSError, IOError):' instead.
| except: | |
| except (OSError, io.UnsupportedOperation, struct.error, ValueError): |
| return 'up' | ||
| elif ch3 == b'B': | ||
| return 'down' | ||
| except: |
There was a problem hiding this comment.
Bare except clause catches all exceptions including SystemExit and KeyboardInterrupt. Specify explicit exception types like 'except OSError:' instead.
| except: | |
| except (OSError, IOError): |
| key = parts[0] | ||
| try: | ||
| value = float(parts[1]) | ||
| except: |
There was a problem hiding this comment.
Bare except clause catches all exceptions including SystemExit and KeyboardInterrupt. Specify explicit exception types like 'except (ValueError, TypeError):' instead.
| except: | |
| except (ValueError, TypeError): |
| if ZoneInfo and tz_name != "UTC": | ||
| try: | ||
| self.tz = ZoneInfo(tz_name) | ||
| except: |
There was a problem hiding this comment.
Bare except clause catches all exceptions including SystemExit and KeyboardInterrupt. Specify explicit exception types like 'except Exception:' instead.
| except: | |
| except Exception: |
| def _get_raw_value(self) -> Optional[float]: | ||
| """Run the command and return the raw numeric value.""" | ||
| try: | ||
| result = subprocess.run(self.command, shell=True, capture_output=True, text=True, timeout=5) |
There was a problem hiding this comment.
Using shell=True with commands that include remote host prefixes could be vulnerable to command injection. Consider validating the command or using shell=False with proper argument parsing.
1 participant
Summary
A Python tool that displays ATS metrics inline in iTerm2 using imgcat with live updates and multi-host comparison.
Features:
Requirements:
Usage:
traffic_grapher.py ats-server1.example.com traffic_grapher.py ats{1..4}.example.com --interval 2Test plan