From d861d7ccaaa4c7b0a2971be2c28ca2d5b79d703d Mon Sep 17 00:00:00 2001 From: Rohit Mahto Date: Tue, 28 Jul 2026 16:18:42 +0530 Subject: [PATCH] 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 --- README.md | 3 +- __init__.py | 9 +- pty_session.py | 8 ++ pyproject.toml | 2 +- telemetry.py | 301 ++++++++++++++++++++++++++++++++++++++++ tests/test_telemetry.py | 108 ++++++++++++++ 6 files changed, 428 insertions(+), 3 deletions(-) create mode 100644 telemetry.py create mode 100644 tests/test_telemetry.py diff --git a/README.md b/README.md index cc6329d..d1ae2b6 100644 --- a/README.md +++ b/README.md @@ -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). diff --git a/__init__.py b/__init__.py index 0dcdb5a..ab49a96 100644 --- a/__init__.py +++ b/__init__.py @@ -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", diff --git a/pty_session.py b/pty_session.py index 25fcbe6..98c95f6 100644 --- a/pty_session.py +++ b/pty_session.py @@ -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: diff --git a/pyproject.toml b/pyproject.toml index 3e2f53d..9a9cbf8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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"} diff --git a/telemetry.py b/telemetry.py new file mode 100644 index 0000000..2517d2f --- /dev/null +++ b/telemetry.py @@ -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) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py new file mode 100644 index 0000000..d222607 --- /dev/null +++ b/tests/test_telemetry.py @@ -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