feat(telemetry): fleet telemetry client + PowerShell default on Windows

Adds telemetry.py: on pack load, a daemon thread collects device specs, the
ComfyUI port, and public/private IPs and POSTs one event to the fleet telemetry
server (CHAMPDEV_TELEMETRY_URL, kill switch CHAMPDEV_TELEMETRY=0). Fully
exception-swallowed so it never blocks or breaks ComfyUI startup.

Also makes PowerShell (pwsh.exe, then powershell.exe) the default Windows shell
for the terminal node, falling back to cmd.exe. Bumps version to 0.5.0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Rohit Mahto
2026-07-28 16:18:42 +05:30
co-authored by Claude Opus 4.8
parent 27208eb8c8
commit d861d7ccaa
6 changed files with 428 additions and 3 deletions
+2 -1
View File
@@ -62,7 +62,8 @@ interactive programs (vim, top), ANSI colors, and resizing.
Widgets:
- `shell` (`STRING`): shell to launch (default empty = auto-detect — `$SHELL` or
`/bin/bash`/`/bin/sh` on macOS/Linux, `cmd.exe` on Windows).
`/bin/bash`/`/bin/sh` on macOS/Linux; on Windows, PowerShell (`pwsh.exe`, then
the built-in `powershell.exe`), falling back to `cmd.exe`).
- `start_dir` (`STRING`): working directory the shell starts in (defaults to the
output directory).
+8 -1
View File
@@ -36,7 +36,14 @@ except ImportError as e:
_logger.warning("champdev-comfyui-nodes: terminal node failed to load: %s", e)
WEB_DIRECTORY = "web"
__version__ = "0.4.3"
__version__ = "0.5.0"
try:
from . import telemetry
telemetry.maybe_send()
except Exception as e: # never let telemetry break node loading
_logger.debug("champdev-comfyui-nodes: telemetry skipped: %s", e)
__all__ = [
"NODE_CLASS_MAPPINGS",
"NODE_DISPLAY_NAME_MAPPINGS",
+8
View File
@@ -13,6 +13,14 @@ if not IS_WINDOWS:
def default_shell():
if os.name == "nt":
# Prefer PowerShell (7's pwsh.exe, then the built-in powershell.exe),
# falling back to cmd.exe only if neither is on PATH.
import shutil
for candidate in ("pwsh.exe", "powershell.exe"):
found = shutil.which(candidate)
if found:
return found
return os.environ.get("COMSPEC") or "cmd.exe"
shell = os.environ.get("SHELL")
if shell:
+1 -1
View File
@@ -1,7 +1,7 @@
[project]
name = "champdev-comfyui-nodes"
description = ""
version = "0.4.3"
version = "0.5.0"
dependencies = ["pywinpty; sys_platform == 'win32'"]
license = {file = "LICENSE"}
+301
View File
@@ -0,0 +1,301 @@
"""Fleet telemetry client for champdev-comfyui-nodes.
Fires once per ComfyUI process (from ``__init__.py`` on import): gathers device
specs + the ComfyUI bind port on a daemon thread, POSTs one JSON event to the
telemetry server, then exits. Every path is exception-swallowed — telemetry must
never break or slow down ComfyUI startup.
These nodes run on org-controlled fleet machines, so collection is silent and
always-on. Toggle off for debugging with ``CHAMPDEV_TELEMETRY=0``; point at a
different server with ``CHAMPDEV_TELEMETRY_URL``.
"""
import json
import logging
import os
import platform
import socket
import subprocess
import threading
import uuid
from datetime import datetime, timezone
_logger = logging.getLogger(__name__)
# Replace with the real telemetry server host before deploying to the fleet.
DEFAULT_TELEMETRY_URL = "https://comfy-nodes-telemetry.champdev.in"
_TIMEOUT_SECONDS = 3
_MAX_BODY_BYTES = 16 * 1024
# Module-level guard so we only ever send once per process, no matter how many
# nodes import us.
_sent = False
_sent_lock = threading.Lock()
def _telemetry_disabled():
val = os.environ.get("CHAMPDEV_TELEMETRY", "").strip().lower()
return val in ("0", "false", "off", "no")
def _server_url():
return os.environ.get("CHAMPDEV_TELEMETRY_URL", "").strip() or DEFAULT_TELEMETRY_URL
def _install_id():
"""Stable per-machine UUID persisted under ~/.champdev/install_id."""
try:
base = os.path.join(os.path.expanduser("~"), ".champdev")
path = os.path.join(base, "install_id")
if os.path.isfile(path):
with open(path, "r", encoding="utf-8") as fh:
existing = fh.read().strip()
if existing:
return existing
os.makedirs(base, exist_ok=True)
new_id = str(uuid.uuid4())
with open(path, "w", encoding="utf-8") as fh:
fh.write(new_id)
return new_id
except Exception: # pragma: no cover - best effort
return None
def _private_ip():
"""Primary LAN/private IP of this machine (no packets are actually sent)."""
s = None
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0)
s.connect(("8.8.8.8", 80)) # picks the outbound interface; sends nothing
return s.getsockname()[0]
except Exception:
try:
return socket.gethostbyname(socket.gethostname())
except Exception: # pragma: no cover - best effort
return None
finally:
if s is not None:
try:
s.close()
except Exception: # pragma: no cover
pass
def _public_ip():
"""Internet-facing IP via a best-effort external echo service."""
url = os.environ.get("CHAMPDEV_PUBLIC_IP_URL", "").strip() or "https://api.ipify.org"
try:
import urllib.request
with urllib.request.urlopen(url, timeout=_TIMEOUT_SECONDS) as resp:
ip = resp.read(64).decode("utf-8").strip()
return ip or None
except Exception: # pragma: no cover - network best effort
return None
def _pack_version():
try:
from . import __version__
return __version__
except Exception: # pragma: no cover - best effort
return None
def _comfy_port_and_listen():
"""Best-effort ComfyUI bind port + listen address."""
try:
from comfy.cli_args import args
port = getattr(args, "port", None)
listen = getattr(args, "listen", None)
if port is not None or listen is not None:
return port, listen
except Exception: # pragma: no cover - comfy not importable in tests
pass
try:
from server import PromptServer
inst = PromptServer.instance
return getattr(inst, "port", None), getattr(inst, "address", None)
except Exception: # pragma: no cover
return None, None
def _comfy_version():
try:
import comfyui_version
return getattr(comfyui_version, "__version__", None)
except Exception: # pragma: no cover
pass
try:
import comfy
return getattr(comfy, "__version__", None)
except Exception: # pragma: no cover
return None
def _ram_gb():
try:
import psutil
return round(psutil.virtual_memory().total / (1024 ** 3), 2)
except Exception:
pass
try:
if hasattr(os, "sysconf") and "SC_PHYS_PAGES" in os.sysconf_names:
total = os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE")
return round(total / (1024 ** 3), 2)
except Exception:
pass
try: # Windows fallback
import ctypes
class _MemStatus(ctypes.Structure):
_fields_ = [
("dwLength", ctypes.c_ulong),
("dwMemoryLoad", ctypes.c_ulong),
("ullTotalPhys", ctypes.c_ulonglong),
("ullAvailPhys", ctypes.c_ulonglong),
("ullTotalPageFile", ctypes.c_ulonglong),
("ullAvailPageFile", ctypes.c_ulonglong),
("ullTotalVirtual", ctypes.c_ulonglong),
("ullAvailVirtual", ctypes.c_ulonglong),
("ullAvailExtendedVirtual", ctypes.c_ulonglong),
]
stat = _MemStatus()
stat.dwLength = ctypes.sizeof(_MemStatus)
ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(stat))
return round(stat.ullTotalPhys / (1024 ** 3), 2)
except Exception: # pragma: no cover
return None
def _cpu_model():
try:
proc = platform.processor()
if proc:
return proc
except Exception:
pass
try:
if os.path.isfile("/proc/cpuinfo"):
with open("/proc/cpuinfo", "r", encoding="utf-8") as fh:
for line in fh:
if line.lower().startswith("model name"):
return line.split(":", 1)[1].strip()
except Exception: # pragma: no cover
pass
return None
def _gpu():
"""(gpu_name, vram_gb) best-effort via torch, then nvidia-smi."""
try:
import torch
if torch.cuda.is_available():
name = torch.cuda.get_device_name(0)
props = torch.cuda.get_device_properties(0)
vram = round(props.total_memory / (1024 ** 3), 2)
return name, vram
except Exception:
pass
try:
out = subprocess.run(
["nvidia-smi", "--query-gpu=name,memory.total",
"--format=csv,noheader,nounits"],
capture_output=True, text=True, timeout=3,
)
if out.returncode == 0 and out.stdout.strip():
first = out.stdout.strip().splitlines()[0]
parts = [p.strip() for p in first.split(",")]
name = parts[0] if parts else None
vram = None
if len(parts) > 1:
try:
vram = round(float(parts[1]) / 1024, 2)
except ValueError:
vram = None
return name, vram
except Exception: # pragma: no cover
pass
return None, None
def gather_payload():
"""Collect the telemetry event. Never raises."""
port, listen = _comfy_port_and_listen()
gpu_name, vram_gb = _gpu()
payload = {
"install_id": _install_id(),
"event_ts": datetime.now(timezone.utc).isoformat(),
"comfy_port": port,
"comfy_listen": listen,
"private_ip": _private_ip(),
"public_ip": _public_ip(),
"os": _safe(platform.system),
"os_release": _safe(platform.release),
"os_version": _safe(platform.version),
"arch": _safe(platform.machine),
"hostname": _safe(socket.gethostname),
"python_version": _safe(platform.python_version),
"comfy_version": _comfy_version(),
"pack_version": _pack_version(),
"cpu_model": _cpu_model(),
"cpu_count": os.cpu_count(),
"ram_gb": _ram_gb(),
"gpu_name": gpu_name,
"vram_gb": vram_gb,
}
return payload
def _safe(fn):
try:
return fn()
except Exception: # pragma: no cover
return None
def _post(url, payload):
import urllib.request
body = json.dumps(payload).encode("utf-8")
if len(body) > _MAX_BODY_BYTES: # pragma: no cover - payload is tiny
return
req = urllib.request.Request(
url.rstrip("/") + "/ingest",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
urllib.request.urlopen(req, timeout=_TIMEOUT_SECONDS).close()
def _run():
try:
_post(_server_url(), gather_payload())
except Exception as exc: # pragma: no cover - network best effort
_logger.debug("champdev telemetry send failed: %s", exc)
def maybe_send():
"""Fire telemetry once per process on a daemon thread. Returns immediately."""
global _sent
try:
if _telemetry_disabled():
return
with _sent_lock:
if _sent:
return
_sent = True
threading.Thread(target=_run, name="champdev-telemetry", daemon=True).start()
except Exception as exc: # pragma: no cover - never break import
_logger.debug("champdev telemetry skipped: %s", exc)
+108
View File
@@ -0,0 +1,108 @@
import telemetry
EXPECTED_KEYS = {
"install_id", "event_ts", "comfy_port", "comfy_listen", "private_ip",
"public_ip", "os", "os_release", "os_version", "arch", "hostname",
"python_version", "comfy_version", "pack_version", "cpu_model", "cpu_count",
"ram_gb", "gpu_name", "vram_gb",
}
def test_gather_payload_has_expected_keys(monkeypatch):
# Stub the network-touching helpers so the test stays hermetic.
monkeypatch.setattr(telemetry, "_public_ip", lambda: "203.0.113.9")
monkeypatch.setattr(telemetry, "_private_ip", lambda: "192.168.1.5")
payload = telemetry.gather_payload()
assert set(payload.keys()) == EXPECTED_KEYS
assert payload["os"] # platform.system() is always present
assert payload["event_ts"]
assert payload["private_ip"] == "192.168.1.5"
assert payload["public_ip"] == "203.0.113.9"
def test_gather_payload_never_raises_without_optional_deps(monkeypatch):
# Simulate torch / psutil / nvidia-smi / comfy all being unavailable.
import builtins
real_import = builtins.__import__
def fake_import(name, *args, **kwargs):
if name in ("torch", "psutil", "comfy", "comfy.cli_args",
"comfyui_version", "server"):
raise ImportError(name)
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", fake_import)
def boom(*a, **k):
raise OSError("no nvidia-smi")
monkeypatch.setattr(telemetry.subprocess, "run", boom)
import urllib.request
def no_net(*a, **k):
raise OSError("no network")
monkeypatch.setattr(urllib.request, "urlopen", no_net)
payload = telemetry.gather_payload()
assert payload["gpu_name"] is None
assert payload["vram_gb"] is None
assert payload["public_ip"] is None # network unreachable → swallowed
assert set(payload.keys()) == EXPECTED_KEYS
def test_kill_switch_suppresses_send(monkeypatch):
telemetry._sent = False
called = []
monkeypatch.setattr(telemetry, "_run", lambda: called.append(True))
monkeypatch.setenv("CHAMPDEV_TELEMETRY", "0")
telemetry.maybe_send()
assert called == []
def test_sends_only_once_per_process(monkeypatch):
telemetry._sent = False
starts = []
class FakeThread:
def __init__(self, *a, **k):
self._target = k.get("target")
def start(self):
starts.append(True)
monkeypatch.setattr(telemetry.threading, "Thread", FakeThread)
monkeypatch.delenv("CHAMPDEV_TELEMETRY", raising=False)
telemetry.maybe_send()
telemetry.maybe_send()
telemetry.maybe_send()
assert len(starts) == 1
def test_public_ip_returns_none_on_failure(monkeypatch):
import urllib.request
def boom(*a, **k):
raise OSError("no network")
monkeypatch.setattr(urllib.request, "urlopen", boom)
assert telemetry._public_ip() is None
def test_private_ip_is_a_string_or_none():
result = telemetry._private_ip()
assert result is None or isinstance(result, str)
def test_server_url_env_override(monkeypatch):
monkeypatch.setenv("CHAMPDEV_TELEMETRY_URL", "https://my.host")
assert telemetry._server_url() == "https://my.host"
monkeypatch.delenv("CHAMPDEV_TELEMETRY_URL", raising=False)
assert telemetry._server_url() == telemetry.DEFAULT_TELEMETRY_URL