diff --git a/docs/cli/configuration.mdx b/docs/cli/configuration.mdx index 957b675793..b6a3b45376 100644 --- a/docs/cli/configuration.mdx +++ b/docs/cli/configuration.mdx @@ -93,6 +93,7 @@ Environment variables override configuration file settings. This is useful for C | `NMP_OUTPUT_FORMAT` | Output format (table, json, yaml, csv, markdown) | | `NMP_TIMESTAMP_FORMAT` | Timestamp format (relative, iso8601) | | `NMP_COLOR_OUTPUT` | Enable/disable colored output (true/false) | +| `NEMO_TELEMETRY_ENABLED` | Enable CLI telemetry only when set to `true`; any other explicit value disables it | Example: @@ -115,6 +116,27 @@ Settings are resolved in this order (highest priority first): This means you can set defaults in your config file and override them as needed with environment variables or flags. +## CLI Telemetry + +The NeMo CLI sends anonymous usage telemetry by default to help improve setup, command reliability, and product workflows. The first CLI invocation with telemetry enabled prints a notice to stderr and creates a `telemetry-notice-shown` marker next to the CLI config file. Stdout is not changed, so scripts that parse command output remain stable. + +Telemetry events include the command or workflow category, task status, duration, client version, a random session ID that rotates every 30 days, deployment type, and whether the command appears to be running in CI. Telemetry does not include prompts, model inputs or outputs, datasets, secrets, access tokens, configuration file contents, file contents, local paths, usernames, email addresses, or hostnames. + +To disable telemetry for a single command or shell session: + +```bash +NEMO_TELEMETRY_ENABLED=false nemo models list +export NEMO_TELEMETRY_ENABLED=false +``` + +To disable telemetry persistently, add this top-level field to your CLI config file: + +```yaml +telemetry_enabled: false +``` + +If `NEMO_TELEMETRY_ENABLED` is set, only `true` keeps the environment layer enabled. Any other explicit value disables telemetry for that process. A persisted `telemetry_enabled: false` value disables telemetry even when the environment variable is set to `true`. + ## Shell Completion The NeMo CLI supports tab completion for Bash, Zsh, and Fish shells. Enable it with: diff --git a/packages/nemo_platform_ext/pyproject.toml b/packages/nemo_platform_ext/pyproject.toml index 4211b40e6b..2e23517fef 100644 --- a/packages/nemo_platform_ext/pyproject.toml +++ b/packages/nemo_platform_ext/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "ngcsdk>=4.8.2", "nvidia-ml-py>=13.0.0", "psutil>=5.9.0", + "httpx>=0.23.0,<1", ] requires-python = ">=3.11,<3.15" readme = "README.md" diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/__init__.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/__init__.py new file mode 100644 index 0000000000..39a3df4fea --- /dev/null +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/__init__.py @@ -0,0 +1,14 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from nemo_platform_ext.cli.telemetry.handler import ( + QueuedEvent, + TelemetryHandler, + build_payload, +) + +__all__ = [ + "QueuedEvent", + "TelemetryHandler", + "build_payload", +] diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/emit.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/emit.py new file mode 100644 index 0000000000..a40685a7f6 --- /dev/null +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/emit.py @@ -0,0 +1,99 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Fire-and-flush emission with three opt-out layers and the first-run notice.""" + +from __future__ import annotations + +import logging +import sys +from pathlib import Path + +from nemo_platform_ext.cli.telemetry.events import PlatformTelemetryEvent +from nemo_platform_ext.cli.telemetry.handler import TelemetryHandler, _telemetry_enabled +from nemo_platform_ext.cli.telemetry.session import get_session_id + +logger = logging.getLogger(__name__) + +_invocation_opt_out = False + +_NOTICE_TEXT = ( + "NeMo Platform CLI telemetry is on by default and sends anonymous usage data to improve the product. " + "It does not send prompts, model inputs or outputs, datasets, secrets, file contents, or personal identifiers. " + "Turn it off with NEMO_TELEMETRY_ENABLED=false or telemetry_enabled: false in the CLI config. " + "Run nemo docs cli/configuration for details.\n" +) + + +def set_invocation_opt_out(value: bool) -> None: + """Per-invocation opt-out (e.g. a --no-telemetry flag on the current command).""" + global _invocation_opt_out + _invocation_opt_out = value + + +def _config_opted_out() -> bool: + """True when the persisted config file sets ``telemetry_enabled: false``.""" + try: + from nemo_platform_ext.config.config import Config + + cfg = Config.load() + return cfg.get_config_file().telemetry_enabled is False + except Exception: + # A privacy control must fail closed: if we cannot read the config to confirm + # the user is opted in, treat them as opted out and do not send. + logger.debug("Could not read telemetry opt-out config; failing closed (opted out)", exc_info=True) + return True + + +def telemetry_opted_in() -> bool: + """Opted in only when all three layers agree: per-invocation, env, and config.""" + if _invocation_opt_out: + return False + if not _telemetry_enabled(): + return False + return not _config_opted_out() + + +def _client_version() -> str: + try: + import nemo_platform + + return nemo_platform.__version__ + except Exception: + logger.debug("Could not resolve client version for telemetry", exc_info=True) + return "undefined" + + +def emit_event(event: PlatformTelemetryEvent) -> None: + """Best effort. Telemetry must never break a user command.""" + try: + if not telemetry_opted_in(): + return + # No retries on the CLI exit path: a synchronous send blocks the user's command, + # so cap the worst case at one bounded send (SEND_TIMEOUT_SECONDS) rather than + # retrying against a slow or unreachable endpoint while the user waits. + handler = TelemetryHandler(source_client_version=_client_version(), session_id=get_session_id(), max_retries=0) + handler.enqueue(event) + handler.stop() + except Exception: + logger.debug("Failed to emit telemetry event", exc_info=True) + + +def _notice_marker_path() -> Path: + from nemo_platform_ext.config.config import Config + + return Config.get_default_config_path().parent / "telemetry-notice-shown" + + +def maybe_print_first_run_notice() -> None: + """Print the first-run notice to stderr once. Stdout stays machine-clean.""" + try: + if not telemetry_opted_in(): + return + marker = _notice_marker_path() + if marker.exists(): + return + marker.parent.mkdir(parents=True, exist_ok=True) + marker.touch() + sys.stderr.write(_NOTICE_TEXT) + except Exception: + logger.debug("Failed to print telemetry notice", exc_info=True) diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/events.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/events.py new file mode 100644 index 0000000000..bbdb247126 --- /dev/null +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/events.py @@ -0,0 +1,97 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Platform usage-telemetry event models. + +Field names and aliases follow the shared NeMo telemetry schema +(aire/microservices/nemo-telemetry, schemas/anonymous_events.json, v1.9). +""" + +from __future__ import annotations + +import os +from enum import Enum +from typing import ClassVar + +from pydantic import BaseModel, ConfigDict, Field + +_CI_ENV_VARS = ( + "CI", + "GITLAB_CI", + "GITHUB_ACTIONS", + "BUILDKITE", + "CIRCLECI", + "JENKINS_URL", + "TEAMCITY_VERSION", + "TF_BUILD", + "TRAVIS", +) +_FALSEY = ("", "0", "false", "no", "off") + + +def is_ci_environment() -> bool: + return any(os.getenv(v, "").lower() not in _FALSEY for v in _CI_ENV_VARS) + + +class TaskStatusEnum(str, Enum): + COMPLETED = "completed" + ERROR = "error" + CANCELED = "canceled" + UNDEFINED = "undefined" + + +class DeploymentTypeEnum(str, Enum): + CLI = "cli" + SDK = "sdk" + NVIDIA_INTERNAL = "nvidia-internal" + UNDEFINED = "undefined" + + +def _deployment_type() -> DeploymentTypeEnum: + raw = os.getenv("NEMO_DEPLOYMENT_TYPE", "cli").lower() + try: + return DeploymentTypeEnum(raw) + except ValueError: + return DeploymentTypeEnum.UNDEFINED + + +class PlatformTelemetryEvent(BaseModel): + """Base for all platform events. extra="forbid" is a privacy guard.""" + + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + _event_name: ClassVar[str] = "undefined" + _schema_version: ClassVar[str] = "1.9" + + nemo_source: str = Field(default="platform", serialization_alias="nemoSource") + task_status: TaskStatusEnum = Field(serialization_alias="taskStatus") + deployment_type: DeploymentTypeEnum = Field(default_factory=_deployment_type, serialization_alias="deploymentType") + is_ci: bool = Field(default_factory=is_ci_environment, serialization_alias="isCi") + + +class OnboardingStepEvent(PlatformTelemetryEvent): + _event_name: ClassVar[str] = "onboarding_step" + + step: str + provider_type: str = Field(default="undefined", serialization_alias="providerType") + models_discovered_bucket: str = Field(default="undefined", serialization_alias="modelsDiscoveredBucket") + skills_target: str = Field(default="undefined", serialization_alias="skillsTarget") + agent_deployed: bool = Field(default=False, serialization_alias="agentDeployed") + + +class CommandInvokedEvent(PlatformTelemetryEvent): + _event_name: ClassVar[str] = "command_invoked" + + command: str + duration_sec: float = Field(serialization_alias="durationSec") + agent_mode: bool = Field(default=False, serialization_alias="agentMode") + + +class JobRunEvent(PlatformTelemetryEvent): + _event_name: ClassVar[str] = "job_run" + + job_type: str = Field(serialization_alias="jobType") + duration_sec: float = Field(default=-1.0, serialization_alias="durationSec") + plugins: list[str] = Field(default_factory=list) + model: str = "undefined" + input_tokens: int = Field(default=-1, serialization_alias="inputTokens") + output_tokens: int = Field(default=-1, serialization_alias="outputTokens") diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/handler.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/handler.py new file mode 100644 index 0000000000..48119b8305 --- /dev/null +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/handler.py @@ -0,0 +1,440 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Telemetry handler for NeMo products. + +Environment variables: +- NEMO_TELEMETRY_ENABLED: Whether telemetry is enabled. +- NEMO_DEPLOYMENT_TYPE: The deployment type the event came from. +- NEMO_TELEMETRY_ENDPOINT: The endpoint to send the telemetry events to. +- NEMO_SESSION_PREFIX: Optional prefix to add to session IDs. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import platform +import threading +from collections.abc import Coroutine +from concurrent.futures import Future, ThreadPoolExecutor +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +import httpx + +from nemo_platform_ext.cli.telemetry.events import PlatformTelemetryEvent + +CLIENT_ID = "184482118588404" +NEMO_TELEMETRY_VERSION = "nemo-telemetry/1.0" +DEFAULT_ENDPOINT = "https://events.telemetry.data.nvidia.com/v1.1/events/json" +MAX_RETRIES = 3 +# Tight explicit timeout so a hung or black-holed endpoint can never block a command +# at exit for longer than this. httpx's default is ~5s, which is too long for a +# best-effort flush that runs synchronously on the command's exit path. +SEND_TIMEOUT_SECONDS = 2.0 +logger = logging.getLogger(__name__) + + +def _telemetry_enabled() -> bool: + value = os.getenv("NEMO_TELEMETRY_ENABLED") + if value is None: + return True + return value.strip().lower() == "true" + + +def _cpu_architecture() -> str: + return platform.machine() or "undefined" + + +def _telemetry_endpoint() -> str: + return os.getenv("NEMO_TELEMETRY_ENDPOINT", DEFAULT_ENDPOINT) + + +def _redact_endpoint(endpoint: str) -> str: + """Redact query parameters before logging telemetry endpoints.""" + try: + parsed = urlsplit(endpoint) + except ValueError: + return "" + query = "" if parsed.query else "" + return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, query, parsed.fragment)) + + +def _session_prefix() -> str | None: + return os.getenv("NEMO_SESSION_PREFIX") + + +@dataclass +class QueuedEvent: + event: PlatformTelemetryEvent + timestamp: datetime + retry_count: int = 0 + + +def _get_iso_timestamp(dt: datetime | None = None) -> str: + if dt is None: + dt = datetime.now(timezone.utc) + return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}Z" + + +def build_payload( + events: list[QueuedEvent], *, source_client_version: str, session_id: str = "undefined" +) -> dict[str, Any]: + if not events: + raise ValueError("build_payload requires at least one event") + return { + "browserType": "undefined", + "clientId": CLIENT_ID, + "clientType": "Native", + "clientVariant": "Release", + "clientVer": source_client_version, + "cpuArchitecture": _cpu_architecture(), + "deviceGdprBehOptIn": "None", + "deviceGdprFuncOptIn": "None", + "deviceGdprTechOptIn": "None", + "deviceId": "undefined", + "deviceMake": "undefined", + "deviceModel": "undefined", + "deviceOS": "undefined", + "deviceOSVersion": "undefined", + "deviceType": "undefined", + "eventProtocol": "1.6", + "eventSchemaVer": events[0].event._schema_version, + "eventSysVer": NEMO_TELEMETRY_VERSION, + "externalUserId": "undefined", + "gdprBehOptIn": "None", + "gdprFuncOptIn": "None", + "gdprTechOptIn": "None", + "idpId": "undefined", + "integrationId": "undefined", + "productName": "undefined", + "productVersion": "undefined", + "sentTs": _get_iso_timestamp(), + "sessionId": session_id, + "userId": "undefined", + "events": [ + { + "ts": _get_iso_timestamp(queued.timestamp), + "parameters": queued.event.model_dump(by_alias=True, mode="json"), + "name": queued.event._event_name, + } + for queued in events + ], + } + + +class TelemetryHandler: + """ + Handles telemetry event batching, flushing, and retry logic for NeMo products. + + Supports two usage patterns: + + - **Background mode**: call ``start()`` (or use ``with handler:``) to spawn + a daemon thread with its own event loop that drives periodic flushing. + ``stop()`` schedules a final flush, then stops the loop and joins the thread. + - **Fire-and-flush mode**: skip ``start()``, ``enqueue()`` events, then call + ``stop()`` to flush once. No background thread is created unless the caller + already has a running event loop, in which case the one-shot flush is + offloaded to a worker thread with its own loop. + + Args: + flush_interval_seconds (float): The interval in seconds to flush the events. + max_queue_size (int): The maximum number of events to queue before flushing. + max_retries (int): The maximum number of times to retry sending an event. + source_client_version (str): The version of the source client. This should be the version of + the actual NeMo product that is sending the events, typically the same as the version of + a PyPi package that a user would install. + session_id (str): An optional session ID to associate with the events. + This should be a unique identifier for the session, such as a UUID. + It is used to group events together. + """ + + def __init__( + self, + flush_interval_seconds: float = 120.0, + max_queue_size: int = 50, + max_retries: int = MAX_RETRIES, + source_client_version: str = "undefined", + session_id: str = "undefined", + ): + self._flush_interval = flush_interval_seconds + self._max_queue_size = max_queue_size + self._max_retries = max_retries + self._events: list[QueuedEvent] = [] + self._dlq: list[QueuedEvent] = [] + self._queue_lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._flush_signal: asyncio.Event | None = None + self._timer_task: asyncio.Task | None = None + self._running = False + self._source_client_version = source_client_version + prefix = _session_prefix() + self._session_id = f"{prefix}{session_id}" if prefix else session_id + + # -- Async API ----------------------------------------------------------- + + async def astart(self) -> None: + """Start the background timer task on the current event loop.""" + if self._running: + return + self._loop = asyncio.get_running_loop() + self._flush_signal = asyncio.Event() + self._running = True + self._timer_task = asyncio.create_task(self._timer_loop()) + + async def astop(self) -> None: + """Cancel the timer task and flush any remaining events.""" + if not self._running: + await self._flush_events() + return + self._running = False + if self._flush_signal is not None: + self._flush_signal.set() + if self._timer_task is not None: + self._timer_task.cancel() + try: + await self._timer_task + except asyncio.CancelledError: + pass + self._timer_task = None + await self._flush_events() + self._loop = None + self._flush_signal = None + + async def aflush(self) -> None: + """Flush all queued events immediately and await completion.""" + await self._flush_events() + + # -- Sync API ------------------------------------------------------------ + + def start(self) -> None: + """Spawn a daemon thread with a persistent event loop for periodic flushing.""" + if self._running: + return + ready = threading.Event() + startup_error: list[BaseException] = [] + + def _run() -> None: + try: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + self._loop = loop + self._flush_signal = asyncio.Event() + self._timer_task = loop.create_task(self._timer_loop()) + self._running = True + except BaseException as exc: # noqa: BLE001 + startup_error.append(exc) + ready.set() + return + ready.set() + try: + loop.run_forever() + finally: + loop.close() + + self._thread = threading.Thread(target=_run, name="nemo-telemetry", daemon=True) + self._thread.start() + ready.wait() + if startup_error: + self._thread = None + raise startup_error[0] + + def stop(self) -> None: + """Flush pending events. If a background thread is running, shut it down and join.""" + if self._running and self._loop is not None and self._thread is not None: + loop = self._loop + future = asyncio.run_coroutine_threadsafe(self._astop_inner(), loop) + try: + future.result(timeout=30) + except Exception: # noqa: BLE001 + pass + loop.call_soon_threadsafe(loop.stop) + self._thread.join(timeout=5) + self._thread = None + self._loop = None + self._flush_signal = None + self._timer_task = None + self._running = False + return + if self._events or self._dlq: + try: + self._run_sync(self._flush_events()) + except Exception: # noqa: BLE001 + logger.debug("Telemetry stop flush failed", exc_info=True) + + def flush(self) -> None: + """Flush all queued events immediately and wait for completion.""" + if self._running and self._loop is not None and self._thread is not None: + future: Future[None] = asyncio.run_coroutine_threadsafe(self._flush_events(), self._loop) + try: + future.result(timeout=30) + except Exception: # noqa: BLE001 + pass + return + if self._events or self._dlq: + try: + self._run_sync(self._flush_events()) + except Exception: # noqa: BLE001 + logger.debug("Telemetry flush failed", exc_info=True) + + @staticmethod + def _run_sync(coro: Coroutine[Any, Any, Any]) -> Any: + """Run a coroutine synchronously from sync or async caller contexts. + + ``asyncio.run`` raises when called from a thread that already has a + running event loop, such as a notebook kernel or an async SDK caller. In + that case, run the coroutine in a worker thread so telemetry still gets + a fresh event loop while remaining synchronous to the caller. + """ + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + + if loop is not None: + with ThreadPoolExecutor(max_workers=1) as pool: + future = pool.submit(asyncio.run, coro) + return future.result(timeout=30) + return asyncio.run(coro) + + async def _astop_inner(self) -> None: + """Async shutdown body run on the background loop.""" + self._running = False + if self._flush_signal is not None: + self._flush_signal.set() + if self._timer_task is not None: + self._timer_task.cancel() + try: + await self._timer_task + except asyncio.CancelledError: + pass + self._timer_task = None + await self._flush_events() + + # -- Enqueue / signalling ------------------------------------------------ + + def enqueue(self, event: object) -> None: + if not _telemetry_enabled(): + return + if not isinstance(event, PlatformTelemetryEvent): + return + queued = QueuedEvent(event=event, timestamp=datetime.now(timezone.utc)) + with self._queue_lock: + self._events.append(queued) + should_signal = len(self._events) >= self._max_queue_size + if should_signal: + self._signal_flush() + + def _signal_flush(self) -> None: + """Set the flush signal, threadsafe across the background-loop boundary.""" + loop = self._loop + signal = self._flush_signal + if loop is None or signal is None: + return + try: + loop.call_soon_threadsafe(signal.set) + except RuntimeError: + pass + + # -- Context managers ---------------------------------------------------- + + def __enter__(self) -> TelemetryHandler: + self.start() + return self + + def __exit__(self, *_: object) -> None: + self.stop() + + async def __aenter__(self) -> TelemetryHandler: + await self.astart() + return self + + async def __aexit__(self, *_: object) -> None: + await self.astop() + + # -- Internal loop ------------------------------------------------------- + + async def _timer_loop(self) -> None: + assert self._flush_signal is not None + while self._running: + try: + await asyncio.wait_for( + self._flush_signal.wait(), + timeout=self._flush_interval, + ) + except asyncio.TimeoutError: + pass + self._flush_signal.clear() + await self._flush_events() + + async def _flush_events(self) -> None: + with self._queue_lock: + dlq_events, self._dlq = self._dlq, [] + new_events, self._events = self._events, [] + events_to_send = dlq_events + new_events + if events_to_send: + await self._send_events(events_to_send) + + async def _send_events(self, events: list[QueuedEvent]) -> None: + try: + async with httpx.AsyncClient(timeout=SEND_TIMEOUT_SECONDS) as client: + await self._send_events_with_client(client, events) + except Exception: # noqa: BLE001 + logger.debug("Telemetry send failed; routing events to DLQ", exc_info=True) + self._add_to_dlq(events) + + async def _send_events_with_client(self, client: httpx.AsyncClient, events: list[QueuedEvent]) -> None: + if not events: + return + + payload = build_payload(events, source_client_version=self._source_client_version, session_id=self._session_id) + endpoint = _telemetry_endpoint() + logger.debug( + "Sending telemetry events", + extra={ + "ctx": { + "endpoint": _redact_endpoint(endpoint), + "event_count": len(events), + "events": [ + { + "name": queued.event._event_name, + "task": getattr(queued.event, "task", "unknown"), + "task_status": getattr(queued.event, "task_status", "unknown"), + "deployment_type": getattr(queued.event, "deployment_type", "unknown"), + "retry_count": queued.retry_count, + } + for queued in events + ], + } + }, + ) + try: + response = await client.post(endpoint, json=payload) + if response.status_code in (400, 422) or response.is_success: + return + if response.status_code == 413: + if len(events) == 1: + return + mid = len(events) // 2 + await self._send_events_with_client(client, events[:mid]) + await self._send_events_with_client(client, events[mid:]) + return + if response.status_code in (408, 429) or response.status_code >= 500: + self._add_to_dlq(events) + except Exception: # noqa: BLE001 + logger.debug("Telemetry POST failed; routing events to DLQ", exc_info=True) + self._add_to_dlq(events) + + def _add_to_dlq(self, events: list[QueuedEvent]) -> None: + with self._queue_lock: + for queued in events: + queued.retry_count += 1 + if queued.retry_count > self._max_retries: + continue + self._dlq.append(queued) diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/session.py b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/session.py new file mode 100644 index 0000000000..5ebf2e5a74 --- /dev/null +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/cli/telemetry/session.py @@ -0,0 +1,157 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Local telemetry session state with time-bounded identifier rotation.""" + +from __future__ import annotations + +import json +import logging +import os +import stat +import uuid +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +_SESSION_STATE_FILENAME = "telemetry-state.json" +_SESSION_ROTATION_DAYS = 30 +_SESSION_ROTATION_INTERVAL = timedelta(days=_SESSION_ROTATION_DAYS) +_SESSION_ID_KEY = "session_id" +_CREATED_AT_KEY = "created_at" +_STATE_FILE_MODE = stat.S_IRUSR | stat.S_IWUSR +_STATE_DIR_MODE = stat.S_IRWXU + + +@dataclass(frozen=True) +class _SessionState: + session_id: str + created_at: datetime + + +_cached_state: _SessionState | None = None + + +def _now_utc() -> datetime: + return datetime.now(timezone.utc) + + +def _default_state_root() -> Path: + xdg_state_home = os.environ.get("XDG_STATE_HOME") + if xdg_state_home: + return Path(xdg_state_home).expanduser() / "nmp" + return Path.home() / ".local" / "state" / "nmp" + + +def _session_state_path() -> Path: + return _default_state_root() / _SESSION_STATE_FILENAME + + +def _format_created_at(dt: datetime) -> str: + return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _parse_created_at(value: Any) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + try: + parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _read_session_state(path: Path) -> _SessionState | None: + if not path.exists(): + return None + try: + with path.open(encoding="utf-8") as f: + data = json.load(f) + except json.JSONDecodeError: + return None + if not isinstance(data, dict): + return None + + session_id = data.get(_SESSION_ID_KEY) + created_at = _parse_created_at(data.get(_CREATED_AT_KEY)) + if not isinstance(session_id, str) or not session_id.strip() or created_at is None: + return None + return _SessionState(session_id=session_id, created_at=created_at) + + +def _requires_rotation(state: _SessionState, now: datetime) -> bool: + if state.created_at > now: + return True + return now - state.created_at >= _SESSION_ROTATION_INTERVAL + + +def _new_session_state(now: datetime) -> _SessionState: + return _SessionState(session_id=uuid.uuid4().hex, created_at=now) + + +def _write_session_state(path: Path, state: _SessionState) -> None: + created_parent = not path.parent.exists() + path.parent.mkdir(parents=True, exist_ok=True) + if created_parent: + try: + os.chmod(path.parent, _STATE_DIR_MODE) + except OSError: + pass + + payload = { + _SESSION_ID_KEY: state.session_id, + _CREATED_AT_KEY: _format_created_at(state.created_at), + } + tmp_path = path.with_name(f".{path.name}.{os.getpid()}.tmp") + fd = os.open(tmp_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, _STATE_FILE_MODE) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + fd = -1 + json.dump(payload, f, sort_keys=True) + f.write("\n") + os.replace(tmp_path, path) + os.chmod(path, _STATE_FILE_MODE) + finally: + if fd >= 0: + os.close(fd) + try: + tmp_path.unlink() + except FileNotFoundError: + pass + + +def _load_or_rotate_session_state(path: Path, now: datetime) -> _SessionState: + state = _read_session_state(path) + if state is not None and not _requires_rotation(state, now): + return state + + state = _new_session_state(now) + _write_session_state(path, state) + return state + + +def get_session_id() -> str: + """Return the current random telemetry session identifier, rotating every 30 days. + + A missing, corrupt, dateless, future-dated, or expired state file is replaced + with a new random identifier and fresh creation timestamp. If local state cannot + be read or written, fall back to a process-local random identifier so telemetry + remains best-effort and never blocks command execution. + """ + global _cached_state + + now = _now_utc() + if _cached_state is not None and not _requires_rotation(_cached_state, now): + return _cached_state.session_id + + try: + _cached_state = _load_or_rotate_session_state(_session_state_path(), now) + return _cached_state.session_id + except Exception: + logger.debug("Failed to resolve telemetry session state; using an ephemeral identifier", exc_info=True) + _cached_state = _new_session_state(now) + return _cached_state.session_id diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/config/models.py b/packages/nemo_platform_ext/src/nemo_platform_ext/config/models.py index 246b1f0eca..d114ef2d4f 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/config/models.py +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/config/models.py @@ -221,6 +221,10 @@ class ConfigFile(BaseModel): default=None, description="User-selected paths for local services (set by `nemo setup`).", ) + telemetry_enabled: bool = Field( + default=True, + description="Whether anonymous usage telemetry is enabled. Set to false to opt out.", + ) def ensure_context( self, diff --git a/packages/nemo_platform_ext/tests/cli/telemetry/conftest.py b/packages/nemo_platform_ext/tests/cli/telemetry/conftest.py new file mode 100644 index 0000000000..c7a4b48bd6 --- /dev/null +++ b/packages/nemo_platform_ext/tests/cli/telemetry/conftest.py @@ -0,0 +1,21 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from collections.abc import Iterator + +import pytest +from nemo_platform_ext.cli.telemetry.events import _CI_ENV_VARS + + +@pytest.fixture(autouse=True) +def _clear_ci_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + """Clear CI markers so telemetry tests are deterministic on developer machines and in CI. + + Without this, tests that assert the default ``is_ci`` value pass locally but fail when the + suite runs under GitHub Actions (which sets ``CI`` and ``GITHUB_ACTIONS``). Tests that + exercise CI detection set these variables explicitly, which overrides this fixture. + """ + for var in _CI_ENV_VARS: + monkeypatch.delenv(var, raising=False) + monkeypatch.delenv("NEMO_DEPLOYMENT_TYPE", raising=False) + yield diff --git a/packages/nemo_platform_ext/tests/cli/telemetry/test_emit.py b/packages/nemo_platform_ext/tests/cli/telemetry/test_emit.py new file mode 100644 index 0000000000..2134b8d1a8 --- /dev/null +++ b/packages/nemo_platform_ext/tests/cli/telemetry/test_emit.py @@ -0,0 +1,208 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +import json +import uuid +from collections.abc import Iterator +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest +from nemo_platform_ext.cli.telemetry import emit as emit_mod +from nemo_platform_ext.cli.telemetry import session as session_mod +from nemo_platform_ext.cli.telemetry.events import CommandInvokedEvent, TaskStatusEnum + + +@pytest.fixture(autouse=True) +def _isolate_local_telemetry_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Iterator[None]: + session_mod._cached_state = None + monkeypatch.delenv("NEMO_TELEMETRY_ENABLED", raising=False) + monkeypatch.setenv("NMP_CONFIG_FILE", str(tmp_path / "config.yaml")) + monkeypatch.setattr(session_mod, "_session_state_path", lambda: tmp_path / "telemetry-state.json") + yield + session_mod._cached_state = None + + +def _event(): + return CommandInvokedEvent(command="docs", task_status=TaskStatusEnum.COMPLETED, duration_sec=0.1) + + +class TestOptOutLayers: + def test_env_layer(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + assert emit_mod.telemetry_opted_in() is False + + def test_config_layer(self, monkeypatch, tmp_path): + cfg = tmp_path / "config.yaml" + cfg.write_text("telemetry_enabled: false\n") + monkeypatch.setenv("NMP_CONFIG_FILE", str(cfg)) + assert emit_mod.telemetry_opted_in() is False + + def test_invocation_flag_layer(self): + emit_mod.set_invocation_opt_out(True) + try: + assert emit_mod.telemetry_opted_in() is False + finally: + emit_mod.set_invocation_opt_out(False) + + def test_default_is_on(self): + assert emit_mod.telemetry_opted_in() is True + + def test_config_load_error_fails_closed(self, monkeypatch): + """A broken/parse-error config must fail closed (opted out), not default to on.""" + from nemo_platform_ext.config import config as config_mod + + def boom(*_args: object, **_kwargs: object) -> None: + raise RuntimeError("broken config") + + monkeypatch.setattr(config_mod.Config, "load", boom) + assert emit_mod.telemetry_opted_in() is False + + +class TestEmitEvent: + @patch.object(emit_mod, "TelemetryHandler") + def test_emit_enqueues_and_stops(self, handler_cls): + instance = Mock() + handler_cls.return_value = instance + emit_mod.emit_event(_event()) + instance.enqueue.assert_called_once() + instance.stop.assert_called_once() + + @patch.object(emit_mod, "TelemetryHandler") + def test_emit_skipped_when_opted_out(self, handler_cls, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "0") + emit_mod.emit_event(_event()) + handler_cls.assert_not_called() + + @patch.object(emit_mod, "TelemetryHandler", side_effect=RuntimeError("boom")) + def test_emit_never_raises(self, _handler_cls): + emit_mod.emit_event(_event()) # must not raise + + @patch.object(emit_mod, "TelemetryHandler") + def test_session_id_is_stable_across_calls(self, handler_cls): + """Every event in one process shares the cached session id.""" + session_mod._cached_state = None + emit_mod.emit_event(_event()) + emit_mod.emit_event(_event()) + session_ids = [call.kwargs["session_id"] for call in handler_cls.call_args_list] + assert len(session_ids) == 2 + assert session_ids[0] == session_ids[1] + + +class TestTelemetrySessionState: + def _use_state_path(self, monkeypatch, tmp_path): + session_mod._cached_state = None + path = tmp_path / "telemetry-state.json" + monkeypatch.setattr(session_mod, "_session_state_path", lambda: path) + return path + + def _set_now(self, monkeypatch, now): + monkeypatch.setattr(session_mod, "_now_utc", lambda: now) + + def _set_next_uuid(self, monkeypatch, value: str): + monkeypatch.setattr(session_mod.uuid, "uuid4", lambda: uuid.UUID(value)) + + def test_missing_state_creates_session_id_with_creation_date(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + now = datetime(2026, 7, 27, 12, 0, tzinfo=timezone.utc) + self._set_now(monkeypatch, now) + self._set_next_uuid(monkeypatch, "11111111-1111-4111-8111-111111111111") + + session_id = session_mod.get_session_id() + + assert session_id == "11111111111141118111111111111111" + assert json.loads(path.read_text()) == { + "created_at": "2026-07-27T12:00:00Z", + "session_id": session_id, + } + + def test_recent_state_is_reused(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + path.write_text( + json.dumps({"session_id": "existing", "created_at": "2026-07-01T00:00:00Z"}), + encoding="utf-8", + ) + self._set_now(monkeypatch, datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc)) + + assert session_mod.get_session_id() == "existing" + assert json.loads(path.read_text())["session_id"] == "existing" + + def test_state_rotates_after_thirty_days(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + now = datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc) + path.write_text( + json.dumps( + { + "session_id": "expired", + "created_at": session_mod._format_created_at(now - timedelta(days=30)), + } + ), + encoding="utf-8", + ) + self._set_now(monkeypatch, now) + self._set_next_uuid(monkeypatch, "22222222-2222-4222-8222-222222222222") + + session_id = session_mod.get_session_id() + + assert session_id == "22222222222242228222222222222222" + state = json.loads(path.read_text()) + assert state["session_id"] == session_id + assert state["created_at"] == "2026-07-27T00:00:00Z" + + def test_state_without_creation_date_rotates(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + now = datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc) + path.write_text(json.dumps({"session_id": "dateless"}), encoding="utf-8") + self._set_now(monkeypatch, now) + self._set_next_uuid(monkeypatch, "33333333-3333-4333-8333-333333333333") + + assert session_mod.get_session_id() == "33333333333343338333333333333333" + + def test_malformed_state_rotates(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + path.write_text("not json", encoding="utf-8") + self._set_now(monkeypatch, datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc)) + self._set_next_uuid(monkeypatch, "55555555-5555-4555-8555-555555555555") + + session_id = session_mod.get_session_id() + + assert session_id == "55555555555545558555555555555555" + assert json.loads(path.read_text())["session_id"] == session_id + + def test_future_dated_state_rotates(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + now = datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc) + path.write_text( + json.dumps({"session_id": "future", "created_at": "2026-08-01T00:00:00Z"}), + encoding="utf-8", + ) + self._set_now(monkeypatch, now) + self._set_next_uuid(monkeypatch, "44444444-4444-4444-8444-444444444444") + + assert session_mod.get_session_id() == "44444444444444448444444444444444" + + +class TestFirstRunNotice: + def test_notice_printed_once_to_stderr(self, capsys, tmp_path, monkeypatch): + monkeypatch.setattr(emit_mod, "_notice_marker_path", lambda: tmp_path / "telemetry-notice-shown") + emit_mod.maybe_print_first_run_notice() + first = capsys.readouterr() + assert "anonymous usage data" in first.err + assert first.out == "" # stderr only; stdout stays machine-clean + emit_mod.maybe_print_first_run_notice() + assert capsys.readouterr().err == "" + + def test_no_notice_when_opted_out(self, capsys, tmp_path, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + monkeypatch.setattr(emit_mod, "_notice_marker_path", lambda: tmp_path / "telemetry-notice-shown") + emit_mod.maybe_print_first_run_notice() + assert capsys.readouterr().err == "" + + def test_notice_not_printed_when_marker_write_fails(self, capsys, tmp_path, monkeypatch): + marker_parent = tmp_path / "not-a-directory" + marker_parent.write_text("already a file") + monkeypatch.setattr(emit_mod, "_notice_marker_path", lambda: marker_parent / "telemetry-notice-shown") + + emit_mod.maybe_print_first_run_notice() + + assert capsys.readouterr().err == "" diff --git a/packages/nemo_platform_ext/tests/cli/telemetry/test_events.py b/packages/nemo_platform_ext/tests/cli/telemetry/test_events.py new file mode 100644 index 0000000000..2f696e2302 --- /dev/null +++ b/packages/nemo_platform_ext/tests/cli/telemetry/test_events.py @@ -0,0 +1,75 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +import pytest +from nemo_platform_ext.cli.telemetry.events import ( + CommandInvokedEvent, + JobRunEvent, + OnboardingStepEvent, + TaskStatusEnum, + is_ci_environment, +) +from pydantic import ValidationError + + +class TestCommandInvokedEvent: + def test_defaults_and_aliases(self): + e = CommandInvokedEvent(command="jobs create", task_status=TaskStatusEnum.COMPLETED, duration_sec=1.25) + d = e.model_dump(by_alias=True, mode="json") + assert d["nemoSource"] == "platform" + assert d["command"] == "jobs create" + assert d["taskStatus"] == "completed" + assert d["durationSec"] == 1.25 + assert d["agentMode"] is False + assert d["isCi"] is False + assert e._event_name == "command_invoked" + assert e._schema_version == "1.9" + + def test_no_free_text_fields_beyond_known(self): + # privacy guard: the event cannot carry arbitrary payloads + with pytest.raises(ValidationError): + CommandInvokedEvent(command="x", task_status=TaskStatusEnum.COMPLETED, duration_sec=0, prompt="secret") + + +class TestOnboardingStepEvent: + def test_fields(self): + e = OnboardingStepEvent(step="provider_connected", task_status=TaskStatusEnum.COMPLETED, provider_type="openai") + d = e.model_dump(by_alias=True, mode="json") + assert d["step"] == "provider_connected" + assert d["providerType"] == "openai" + assert d["modelsDiscoveredBucket"] == "undefined" + assert d["skillsTarget"] == "undefined" + assert d["agentDeployed"] is False + assert e._event_name == "onboarding_step" + + +class TestJobRunEvent: + def test_token_defaults_are_minus_one(self): + e = JobRunEvent(job_type="auditor.audit", task_status=TaskStatusEnum.ERROR, duration_sec=10.0) + d = e.model_dump(by_alias=True, mode="json") + assert d["jobType"] == "auditor.audit" + assert d["inputTokens"] == -1 + assert d["outputTokens"] == -1 + assert d["model"] == "undefined" + assert d["plugins"] == [] + assert e._event_name == "job_run" + + +class TestCiDetection: + @pytest.mark.parametrize("var", ["CI", "GITLAB_CI", "GITHUB_ACTIONS", "BUILDKITE", "JENKINS_URL"]) + def test_ci_env_detected(self, monkeypatch, var): + monkeypatch.setenv(var, "true") + assert is_ci_environment() is True + + def test_ci_env_presence_value_detected(self, monkeypatch): + monkeypatch.setenv("TEAMCITY_VERSION", "2026.07") + assert is_ci_environment() is True + + @pytest.mark.parametrize("value", ["", "0", "false", "no", "off"]) + def test_ci_env_falsey_value_is_not_detected(self, monkeypatch, value): + monkeypatch.setenv("CI", value) + assert is_ci_environment() is False + + def test_no_ci_env(self, monkeypatch): + for var in ("CI", "GITLAB_CI", "GITHUB_ACTIONS", "BUILDKITE", "JENKINS_URL"): + monkeypatch.delenv(var, raising=False) + assert is_ci_environment() is False diff --git a/packages/nemo_platform_ext/tests/cli/telemetry/test_handler.py b/packages/nemo_platform_ext/tests/cli/telemetry/test_handler.py new file mode 100644 index 0000000000..8e02e4b7cc --- /dev/null +++ b/packages/nemo_platform_ext/tests/cli/telemetry/test_handler.py @@ -0,0 +1,599 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import asyncio +import importlib +import threading +from datetime import datetime, timezone +from typing import ClassVar +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from nemo_platform_ext.cli.telemetry.events import DeploymentTypeEnum, PlatformTelemetryEvent, _deployment_type +from nemo_platform_ext.cli.telemetry.handler import ( + QueuedEvent, + TelemetryHandler, + _telemetry_enabled, + _telemetry_endpoint, + build_payload, +) +from pydantic import Field + +telemetry_module = importlib.import_module("nemo_platform_ext.cli.telemetry.handler") + + +# ============================================================================= +# Stub Event Model for Testing +# ============================================================================= + + +class _StubEvent(PlatformTelemetryEvent): + """Minimal concrete event for testing, subclassing the real PlatformTelemetryEvent. + + ``task_status`` and ``deployment_type`` are redeclared as plain strings to shed + the base's serialization aliases, so the handler tests keep asserting the + snake_case keys the Task 1 handler emitted. + """ + + _event_name: ClassVar[str] = "stub_event" + task: str = Field(default="test_task") + task_status: str = Field(default="completed") + deployment_type: str = Field(default="sdk") + + +# ============================================================================= +# Env-var helpers +# ============================================================================= + + +class TestEnvHelpers: + def test_telemetry_enabled_default(self, monkeypatch): + monkeypatch.delenv("NEMO_TELEMETRY_ENABLED", raising=False) + assert _telemetry_enabled() is True + + def test_telemetry_enabled_disabled(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + assert _telemetry_enabled() is False + + def test_telemetry_enabled_true_when_explicitly_true(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + assert _telemetry_enabled() is True + + @pytest.mark.parametrize("value", ["", "0", "1", "yes", "yep", "for sure"]) + def test_telemetry_enabled_non_true_values_disable(self, monkeypatch, value): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", value) + assert _telemetry_enabled() is False + + def test_telemetry_endpoint_preserves_case(self, monkeypatch): + custom = "https://Events.Telemetry.example.COM/v1/Events?Token=AbC" + monkeypatch.setenv("NEMO_TELEMETRY_ENDPOINT", custom) + assert _telemetry_endpoint() == custom + + def test_deployment_type_default(self, monkeypatch): + monkeypatch.delenv("NEMO_DEPLOYMENT_TYPE", raising=False) + assert _deployment_type() is DeploymentTypeEnum.CLI + + def test_deployment_type_invalid_falls_back_to_undefined(self, monkeypatch): + monkeypatch.setenv("NEMO_DEPLOYMENT_TYPE", "definitely-not-real") + assert _deployment_type() is DeploymentTypeEnum.UNDEFINED + + def test_deployment_type_nvidia_internal(self, monkeypatch): + monkeypatch.setenv("NEMO_DEPLOYMENT_TYPE", "nvidia-internal") + assert _deployment_type() is DeploymentTypeEnum.NVIDIA_INTERNAL + + +# ============================================================================= +# build_payload +# ============================================================================= + + +class TestBuildPayload: + def _make_queued(self, task: str = "generate", status: str = "completed") -> QueuedEvent: + event = _StubEvent(task=task, task_status=status) + return QueuedEvent(event=event, timestamp=datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc)) + + def test_structure(self): + queued = self._make_queued() + payload = build_payload([queued], source_client_version="1.2.3", session_id="test-session") + assert payload["clientId"] == "184482118588404" + assert payload["clientVer"] == "1.2.3" + assert payload["sessionId"] == "test-session" + assert len(payload["events"]) == 1 + + def test_event_fields_serialize_as_strings(self, monkeypatch): + monkeypatch.delenv("NEMO_DEPLOYMENT_TYPE", raising=False) + + queued = self._make_queued(task="train", status="error") + payload = build_payload([queued], source_client_version="0.0.1") + event_entry = payload["events"][0] + assert event_entry["name"] == "stub_event" + assert event_entry["ts"] == "2025-01-01T12:00:00.000Z" + params = event_entry["parameters"] + assert params["task"] == "train" + assert params["task_status"] == "error" + + def test_payload_is_json_serializable(self): + """Regression: the full payload must be encodable by the stdlib JSON encoder.""" + import json + + queued = self._make_queued(status="error") + payload = build_payload([queued], source_client_version="1.0.0") + json.dumps(payload) + + def test_multiple_events(self): + events = [self._make_queued(task=t) for t in ("train", "generate", "evaluate")] + payload = build_payload(events, source_client_version="1.0.0") + assert len(payload["events"]) == 3 + tasks = [e["parameters"]["task"] for e in payload["events"]] + assert tasks == ["train", "generate", "evaluate"] + + def test_default_session_id(self): + queued = self._make_queued() + payload = build_payload([queued], source_client_version="1.0.0") + assert payload["sessionId"] == "undefined" + + def test_cpu_architecture_uses_platform_machine(self, monkeypatch): + monkeypatch.setattr(telemetry_module.platform, "machine", lambda: "arm64") + payload = build_payload([self._make_queued()], source_client_version="1.0.0") + assert payload["cpuArchitecture"] == "arm64" + + def test_cpu_architecture_empty_value_falls_back_to_undefined(self, monkeypatch): + monkeypatch.setattr(telemetry_module.platform, "machine", lambda: "") + payload = build_payload([self._make_queued()], source_client_version="1.0.0") + assert payload["cpuArchitecture"] == "undefined" + + def test_empty_events_raises(self): + with pytest.raises(ValueError): + build_payload([], source_client_version="1.0.0") + + +# ============================================================================= +# TelemetryHandler — telemetry disabled +# ============================================================================= + + +class TestTelemetryDisabled: + def test_enqueue_noop_when_disabled(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + handler = TelemetryHandler() + event = _StubEvent(task="generate") + handler.enqueue(event) + assert handler._events == [] + + def test_enqueue_noop_for_non_event(self, monkeypatch): + """Silently ignores non-TelemetryEvent objects regardless of env.""" + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler() + handler.enqueue("not an event") + assert handler._events == [] + + +# ============================================================================= +# TelemetryHandler — enqueue and flush +# ============================================================================= + + +class TestTelemetryHandlerEnqueue: + def test_enqueue_adds_event(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler() + event = _StubEvent(task="generate") + handler.enqueue(event) + assert len(handler._events) == 1 + assert handler._events[0].event is event + + def test_enqueue_does_not_add_event_when_telemetry_is_disabled(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + handler = TelemetryHandler() + event = _StubEvent(task="generate") + handler.enqueue(event) + assert len(handler._events) == 0 + + def test_enqueue_at_max_queue_size_signals_flush_when_running(self, monkeypatch): + """When a background loop is up, hitting max_queue_size should signal a flush.""" + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(max_queue_size=3, flush_interval_seconds=60.0) + flushed = threading.Event() + + async def fake_send(_events): + flushed.set() + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.start() + try: + for _ in range(3): + handler.enqueue(_StubEvent(task="run")) + assert flushed.wait(timeout=2.0), "max_queue_size flush signal did not fire" + finally: + handler.stop() + + +# ============================================================================= +# TelemetryHandler — _flush_events queue clearing and DLQ +# ============================================================================= + + +class TestFlushEventsQueueClearing: + async def test_flush_events_clears_queue(self): + """_flush_events() must drain _events even when the underlying send succeeds.""" + handler = TelemetryHandler(source_client_version="1.0.0") + event = _StubEvent(task="generate") + handler._events.append(QueuedEvent(event=event, timestamp=datetime.now(timezone.utc))) + + async def fake_send(events): + assert len(events) == 1 + + with patch.object(handler, "_send_events", side_effect=fake_send): + await handler._flush_events() + + assert handler._events == [] + assert handler._dlq == [] + + async def test_flush_events_includes_dlq(self): + handler = TelemetryHandler(source_client_version="1.0.0") + event = _StubEvent(task="generate") + handler._dlq.append(QueuedEvent(event=event, timestamp=datetime.now(timezone.utc), retry_count=1)) + handler._events.append(QueuedEvent(event=event, timestamp=datetime.now(timezone.utc))) + + sent: list[list[QueuedEvent]] = [] + + async def fake_send(events): + sent.append(list(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + await handler._flush_events() + + assert handler._events == [] + assert handler._dlq == [] + assert len(sent) == 1 + assert len(sent[0]) == 2 + + +# ============================================================================= +# TelemetryHandler — send and retry +# ============================================================================= + + +class TestTelemetryHandlerSend: + def _make_handler(self) -> TelemetryHandler: + return TelemetryHandler(source_client_version="1.0.0", session_id="s1") + + def _make_queued(self) -> QueuedEvent: + event = _StubEvent(task="generate") + return QueuedEvent(event=event, timestamp=datetime.now(timezone.utc)) + + async def test_debug_logs_event_metadata_before_post(self, monkeypatch): + handler = self._make_handler() + queued = self._make_queued() + events_seen: list[str] = [] + debug_calls = [] + + monkeypatch.setenv("NEMO_TELEMETRY_ENDPOINT", "https://Events.Telemetry.example.COM/v1/Events?Token=AbC") + + def fake_debug(message, *, extra): + events_seen.append("debug") + debug_calls.append((message, extra)) + + async def fake_post(*_args, **_kwargs): + events_seen.append("post") + return MagicMock(status_code=200, is_success=True) + + monkeypatch.setattr(telemetry_module.logger, "debug", fake_debug) + mock_client = AsyncMock() + mock_client.post.side_effect = fake_post + + await handler._send_events_with_client(mock_client, [queued]) + + assert events_seen == ["debug", "post"] + message, extra = debug_calls[0] + assert message == "Sending telemetry events" + ctx = extra["ctx"] + assert ctx["endpoint"] == "https://Events.Telemetry.example.COM/v1/Events?" + assert ctx["event_count"] == 1 + assert ctx["events"] == [ + { + "name": "stub_event", + "task": "generate", + "task_status": "completed", + "deployment_type": "sdk", + "retry_count": 0, + } + ] + + async def test_successful_send_does_not_dlq(self): + handler = self._make_handler() + queued = self._make_queued() + + mock_response = MagicMock(status_code=200, is_success=True) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + + await handler._send_events_with_client(mock_client, [queued]) + mock_client.post.assert_awaited_once() + assert handler._dlq == [] + + async def test_500_adds_to_dlq(self): + handler = self._make_handler() + queued = self._make_queued() + + mock_response = MagicMock(status_code=500, is_success=False) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + + await handler._send_events_with_client(mock_client, [queued]) + assert len(handler._dlq) == 1 + assert handler._dlq[0].retry_count == 1 + + async def test_429_adds_to_dlq(self): + handler = self._make_handler() + queued = self._make_queued() + + mock_response = MagicMock(status_code=429, is_success=False) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + + await handler._send_events_with_client(mock_client, [queued]) + assert len(handler._dlq) == 1 + assert handler._dlq[0].retry_count == 1 + + async def test_exceeds_max_retries_dropped(self): + handler = self._make_handler() + queued = self._make_queued() + queued.retry_count = handler._max_retries + + mock_response = MagicMock(status_code=500, is_success=False) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + + await handler._send_events_with_client(mock_client, [queued]) + assert handler._dlq == [] + + async def test_413_splits_and_retries(self): + handler = self._make_handler() + event = _StubEvent(task="generate") + events = [ + QueuedEvent(event=event, timestamp=datetime.now(timezone.utc)), + QueuedEvent(event=event, timestamp=datetime.now(timezone.utc)), + ] + + success_response = MagicMock(status_code=200, is_success=True) + too_large_response = MagicMock(status_code=413, is_success=False) + mock_client = AsyncMock() + mock_client.post.side_effect = [too_large_response, success_response, success_response] + + await handler._send_events_with_client(mock_client, events) + assert mock_client.post.await_count == 3 + + async def test_send_events_routes_to_dlq_on_client_setup_failure(self): + """If httpx client creation raises, events must land in DLQ rather than vanish.""" + handler = self._make_handler() + queued = self._make_queued() + + with patch("httpx.AsyncClient", side_effect=RuntimeError("boom")): + await handler._send_events([queued]) + + assert len(handler._dlq) == 1 + + def test_session_prefix_applied(self, monkeypatch): + monkeypatch.setenv("NEMO_SESSION_PREFIX", "pfx-") + handler = TelemetryHandler(session_id="abc") + assert handler._session_id == "pfx-abc" + + +# ============================================================================= +# TelemetryHandler — aflush awaits a real flush +# ============================================================================= + + +class TestAflushAwaits: + async def test_aflush_actually_flushes(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0") + event = _StubEvent(task="generate") + handler.enqueue(event) + assert len(handler._events) == 1 + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + await handler.aflush() + + assert handler._events == [] + assert sent == [1] + + +# ============================================================================= +# TelemetryHandler — sync flush from async caller contexts +# ============================================================================= + + +class TestFlushFromRunningLoop: + """Regression coverage for notebooks and async SDK callers.""" + + def test_flush_runs_to_completion_when_loop_is_running(self) -> None: + sent: list[int] = [] + + async def fake_flush(self) -> None: # noqa: ARG001 + sent.append(1) + + async def driver() -> None: + handler = TelemetryHandler(source_client_version="1.0.0") + handler._events.append( + QueuedEvent( + event=_StubEvent(task="run"), + timestamp=datetime.now(timezone.utc), + ) + ) + with patch.object(TelemetryHandler, "_flush_events", new=fake_flush): + handler.flush() + + asyncio.run(driver()) + assert sent == [1] + + def test_stop_flushes_fire_and_flush_path_when_loop_is_running(self, monkeypatch) -> None: + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + async def driver() -> None: + handler = TelemetryHandler(source_client_version="1.0.0") + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.enqueue(_StubEvent(task="run")) + handler.stop() + assert handler._events == [] + + asyncio.run(driver()) + assert sent == [1] + + def test_run_sync_propagates_exception_to_flush_boundary(self) -> None: + async def boom() -> None: + raise RuntimeError("kaboom") + + async def driver() -> None: + with pytest.raises(RuntimeError, match="kaboom"): + TelemetryHandler._run_sync(boom()) + + asyncio.run(driver()) + + +# ============================================================================= +# TelemetryHandler — sync lifecycle and context manager +# ============================================================================= + + +class TestSyncLifecycle: + def test_fire_and_flush_without_start(self, monkeypatch): + """Pattern used by the SDK: construct, enqueue, stop. No start() call. stop() must flush.""" + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0") + event = _StubEvent(task="generate") + handler.enqueue(event) + assert len(handler._events) == 1 + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.stop() + + assert handler._events == [] + assert sent == [1] + assert handler._thread is None + + def test_start_spawns_thread_and_stop_flushes(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0) + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.start() + assert handler._thread is not None + assert handler._thread.is_alive() + + handler.enqueue(_StubEvent(task="run")) + handler.stop() + + assert handler._thread is None + assert handler._loop is None + assert sent == [1] + + def test_sync_context_manager(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(TelemetryHandler, "_send_events", side_effect=fake_send, autospec=False): + with TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0) as handler: + handler.enqueue(_StubEvent(task="run")) + + assert sent == [1] + + def test_sync_flush_during_background_run(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0) + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.start() + try: + handler.enqueue(_StubEvent(task="run")) + handler.flush() + assert sent == [1] + assert handler._events == [] + finally: + handler.stop() + + def test_timer_driven_flush(self, monkeypatch): + """With a short flush interval, the background timer should drive a flush without explicit calls.""" + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=0.05) + + flushed = threading.Event() + + async def fake_send(_events): + flushed.set() + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.start() + try: + handler.enqueue(_StubEvent(task="run")) + assert flushed.wait(timeout=2.0), "timer-driven flush did not fire" + finally: + handler.stop() + + +# ============================================================================= +# TelemetryHandler — async lifecycle +# ============================================================================= + + +class TestAsyncLifecycle: + async def test_async_context_manager_flushes_on_exit(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + async with TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0) as handler: + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.enqueue(_StubEvent(task="run")) + await handler.astop() + + assert sent == [1] + + async def test_enqueue_at_max_size_signals_flush_in_async_mode(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + + flushed = asyncio.Event() + + async def fake_send(_events): + flushed.set() + + handler = TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0, max_queue_size=2) + with patch.object(handler, "_send_events", side_effect=fake_send): + await handler.astart() + try: + handler.enqueue(_StubEvent(task="run")) + handler.enqueue(_StubEvent(task="run")) + await asyncio.wait_for(flushed.wait(), timeout=2.0) + finally: + await handler.astop() diff --git a/packages/nemo_platform_ext/tests/cli/telemetry/test_wire_contract_smoke.py b/packages/nemo_platform_ext/tests/cli/telemetry/test_wire_contract_smoke.py new file mode 100644 index 0000000000..4bb5daf8c3 --- /dev/null +++ b/packages/nemo_platform_ext/tests/cli/telemetry/test_wire_contract_smoke.py @@ -0,0 +1,179 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""CI-safe wire-contract smoke test for telemetry events. + +This module verifies that each telemetry event serializes to the wire contract +without making network calls. Each test constructs a valid event, wraps it in +QueuedEvent, calls build_payload, and asserts JSON serialization and envelope +structure. + +Live UAT validation (POST against the telemetry endpoint) is a manual runbook: +set NEMO_TELEMETRY_ENDPOINT to the UAT URL, NEMO_TELEMETRY_ENABLED=true, and +run one real command; live validation is pending nemoSource schema registration. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone + +from nemo_platform_ext.cli.telemetry.events import ( + CommandInvokedEvent, + JobRunEvent, + OnboardingStepEvent, + TaskStatusEnum, +) +from nemo_platform_ext.cli.telemetry.handler import QueuedEvent, build_payload + +EXPECTED_ENVELOPE_KEYS = { + "browserType", + "clientId", + "clientType", + "clientVariant", + "clientVer", + "cpuArchitecture", + "deviceGdprBehOptIn", + "deviceGdprFuncOptIn", + "deviceGdprTechOptIn", + "deviceId", + "deviceMake", + "deviceModel", + "deviceOS", + "deviceOSVersion", + "deviceType", + "eventProtocol", + "eventSchemaVer", + "eventSysVer", + "externalUserId", + "gdprBehOptIn", + "gdprFuncOptIn", + "gdprTechOptIn", + "idpId", + "integrationId", + "productName", + "productVersion", + "sentTs", + "sessionId", + "userId", + "events", +} + + +class TestWireContractSmoke: + """Verify each event type serializes to the wire contract envelope.""" + + def test_onboarding_step_event_contract(self): + """OnboardingStepEvent must serialize with camelCase aliases and correct envelope.""" + event = OnboardingStepEvent( + task_status=TaskStatusEnum.COMPLETED, + step="provider_discovery", + provider_type="huggingface", + models_discovered_bucket="10-100", + skills_target="audit_content", + agent_deployed=True, + ) + queued = QueuedEvent(event=event, timestamp=datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)) + payload = build_payload([queued], source_client_version="1.0.0", session_id="test") + + # Envelope keys must match exactly. + assert set(payload.keys()) == EXPECTED_ENVELOPE_KEYS, ( + f"Envelope keys mismatch: {set(payload.keys()) ^ EXPECTED_ENVELOPE_KEYS}" + ) + + # Schema version must be 1.9. + assert payload["eventSchemaVer"] == "1.9" + + # Must be JSON serializable. + json_str = json.dumps(payload) + assert isinstance(json_str, str) + + # Event parameters must include camelCase aliases. + event_entry = payload["events"][0] + params = event_entry["parameters"] + assert params["nemoSource"] == "platform" + assert params["taskStatus"] == "completed" + assert params["deploymentType"] == "cli" + assert params["isCi"] is False + assert params["step"] == "provider_discovery" + assert params["providerType"] == "huggingface" + assert params["modelsDiscoveredBucket"] == "10-100" + assert params["skillsTarget"] == "audit_content" + assert params["agentDeployed"] is True + assert event_entry["name"] == "onboarding_step" + + def test_command_invoked_event_contract(self): + """CommandInvokedEvent must serialize with camelCase aliases and correct envelope.""" + event = CommandInvokedEvent( + task_status=TaskStatusEnum.COMPLETED, + command="nemo platform apply", + duration_sec=2.5, + agent_mode=True, + ) + queued = QueuedEvent(event=event, timestamp=datetime(2026, 1, 1, 13, 0, 0, tzinfo=timezone.utc)) + payload = build_payload([queued], source_client_version="1.2.3", session_id="cmd-test") + + # Envelope keys must match exactly. + assert set(payload.keys()) == EXPECTED_ENVELOPE_KEYS, ( + f"Envelope keys mismatch: {set(payload.keys()) ^ EXPECTED_ENVELOPE_KEYS}" + ) + + # Schema version must be 1.9. + assert payload["eventSchemaVer"] == "1.9" + + # Must be JSON serializable. + json_str = json.dumps(payload) + assert isinstance(json_str, str) + + # Event parameters must include camelCase aliases. + event_entry = payload["events"][0] + params = event_entry["parameters"] + assert params["nemoSource"] == "platform" + assert params["taskStatus"] == "completed" + assert params["deploymentType"] == "cli" + assert params["isCi"] is False + assert params["command"] == "nemo platform apply" + assert params["durationSec"] == 2.5 + assert params["agentMode"] is True + assert event_entry["name"] == "command_invoked" + + def test_job_run_event_contract(self): + """JobRunEvent must serialize with camelCase aliases and correct envelope.""" + event = JobRunEvent( + task_status=TaskStatusEnum.COMPLETED, + job_type="evaluate", + duration_sec=15.75, + plugins=["garak", "llm_judge"], + model="nemotron-4-8b", + input_tokens=2048, + output_tokens=512, + ) + queued = QueuedEvent(event=event, timestamp=datetime(2026, 1, 1, 14, 0, 0, tzinfo=timezone.utc)) + payload = build_payload([queued], source_client_version="2.0.0", session_id="job-test") + + # Envelope keys must match exactly. + assert set(payload.keys()) == EXPECTED_ENVELOPE_KEYS, ( + f"Envelope keys mismatch: {set(payload.keys()) ^ EXPECTED_ENVELOPE_KEYS}" + ) + + # Schema version must be 1.9. + assert payload["eventSchemaVer"] == "1.9" + + # Must be JSON serializable. + json_str = json.dumps(payload) + assert isinstance(json_str, str) + + # Event parameters must include camelCase aliases. + event_entry = payload["events"][0] + params = event_entry["parameters"] + assert params["nemoSource"] == "platform" + assert params["taskStatus"] == "completed" + assert params["deploymentType"] == "cli" + assert params["isCi"] is False + assert params["jobType"] == "evaluate" + assert params["durationSec"] == 15.75 + assert params["plugins"] == ["garak", "llm_judge"] + assert params["model"] == "nemotron-4-8b" + assert params["inputTokens"] == 2048 + assert params["outputTokens"] == 512 + assert event_entry["name"] == "job_run" diff --git a/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/__init__.py b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/__init__.py new file mode 100644 index 0000000000..56939c5380 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/__init__.py @@ -0,0 +1,14 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from nemo_platform.cli.telemetry.handler import ( + QueuedEvent, + TelemetryHandler, + build_payload, +) + +__all__ = [ + "QueuedEvent", + "TelemetryHandler", + "build_payload", +] diff --git a/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/emit.py b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/emit.py new file mode 100644 index 0000000000..5cfeb4707c --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/emit.py @@ -0,0 +1,99 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Fire-and-flush emission with three opt-out layers and the first-run notice.""" + +from __future__ import annotations + +import logging +import sys +from pathlib import Path + +from nemo_platform.cli.telemetry.events import PlatformTelemetryEvent +from nemo_platform.cli.telemetry.handler import TelemetryHandler, _telemetry_enabled +from nemo_platform.cli.telemetry.session import get_session_id + +logger = logging.getLogger(__name__) + +_invocation_opt_out = False + +_NOTICE_TEXT = ( + "NeMo Platform CLI telemetry is on by default and sends anonymous usage data to improve the product. " + "It does not send prompts, model inputs or outputs, datasets, secrets, file contents, or personal identifiers. " + "Turn it off with NEMO_TELEMETRY_ENABLED=false or telemetry_enabled: false in the CLI config. " + "Run nemo docs cli/configuration for details.\n" +) + + +def set_invocation_opt_out(value: bool) -> None: + """Per-invocation opt-out (e.g. a --no-telemetry flag on the current command).""" + global _invocation_opt_out + _invocation_opt_out = value + + +def _config_opted_out() -> bool: + """True when the persisted config file sets ``telemetry_enabled: false``.""" + try: + from nemo_platform.config.config import Config + + cfg = Config.load() + return cfg.get_config_file().telemetry_enabled is False + except Exception: + # A privacy control must fail closed: if we cannot read the config to confirm + # the user is opted in, treat them as opted out and do not send. + logger.debug("Could not read telemetry opt-out config; failing closed (opted out)", exc_info=True) + return True + + +def telemetry_opted_in() -> bool: + """Opted in only when all three layers agree: per-invocation, env, and config.""" + if _invocation_opt_out: + return False + if not _telemetry_enabled(): + return False + return not _config_opted_out() + + +def _client_version() -> str: + try: + import nemo_platform + + return nemo_platform.__version__ + except Exception: + logger.debug("Could not resolve client version for telemetry", exc_info=True) + return "undefined" + + +def emit_event(event: PlatformTelemetryEvent) -> None: + """Best effort. Telemetry must never break a user command.""" + try: + if not telemetry_opted_in(): + return + # No retries on the CLI exit path: a synchronous send blocks the user's command, + # so cap the worst case at one bounded send (SEND_TIMEOUT_SECONDS) rather than + # retrying against a slow or unreachable endpoint while the user waits. + handler = TelemetryHandler(source_client_version=_client_version(), session_id=get_session_id(), max_retries=0) + handler.enqueue(event) + handler.stop() + except Exception: + logger.debug("Failed to emit telemetry event", exc_info=True) + + +def _notice_marker_path() -> Path: + from nemo_platform.config.config import Config + + return Config.get_default_config_path().parent / "telemetry-notice-shown" + + +def maybe_print_first_run_notice() -> None: + """Print the first-run notice to stderr once. Stdout stays machine-clean.""" + try: + if not telemetry_opted_in(): + return + marker = _notice_marker_path() + if marker.exists(): + return + marker.parent.mkdir(parents=True, exist_ok=True) + marker.touch() + sys.stderr.write(_NOTICE_TEXT) + except Exception: + logger.debug("Failed to print telemetry notice", exc_info=True) diff --git a/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/events.py b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/events.py new file mode 100644 index 0000000000..bbdb247126 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/events.py @@ -0,0 +1,97 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Platform usage-telemetry event models. + +Field names and aliases follow the shared NeMo telemetry schema +(aire/microservices/nemo-telemetry, schemas/anonymous_events.json, v1.9). +""" + +from __future__ import annotations + +import os +from enum import Enum +from typing import ClassVar + +from pydantic import BaseModel, ConfigDict, Field + +_CI_ENV_VARS = ( + "CI", + "GITLAB_CI", + "GITHUB_ACTIONS", + "BUILDKITE", + "CIRCLECI", + "JENKINS_URL", + "TEAMCITY_VERSION", + "TF_BUILD", + "TRAVIS", +) +_FALSEY = ("", "0", "false", "no", "off") + + +def is_ci_environment() -> bool: + return any(os.getenv(v, "").lower() not in _FALSEY for v in _CI_ENV_VARS) + + +class TaskStatusEnum(str, Enum): + COMPLETED = "completed" + ERROR = "error" + CANCELED = "canceled" + UNDEFINED = "undefined" + + +class DeploymentTypeEnum(str, Enum): + CLI = "cli" + SDK = "sdk" + NVIDIA_INTERNAL = "nvidia-internal" + UNDEFINED = "undefined" + + +def _deployment_type() -> DeploymentTypeEnum: + raw = os.getenv("NEMO_DEPLOYMENT_TYPE", "cli").lower() + try: + return DeploymentTypeEnum(raw) + except ValueError: + return DeploymentTypeEnum.UNDEFINED + + +class PlatformTelemetryEvent(BaseModel): + """Base for all platform events. extra="forbid" is a privacy guard.""" + + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + _event_name: ClassVar[str] = "undefined" + _schema_version: ClassVar[str] = "1.9" + + nemo_source: str = Field(default="platform", serialization_alias="nemoSource") + task_status: TaskStatusEnum = Field(serialization_alias="taskStatus") + deployment_type: DeploymentTypeEnum = Field(default_factory=_deployment_type, serialization_alias="deploymentType") + is_ci: bool = Field(default_factory=is_ci_environment, serialization_alias="isCi") + + +class OnboardingStepEvent(PlatformTelemetryEvent): + _event_name: ClassVar[str] = "onboarding_step" + + step: str + provider_type: str = Field(default="undefined", serialization_alias="providerType") + models_discovered_bucket: str = Field(default="undefined", serialization_alias="modelsDiscoveredBucket") + skills_target: str = Field(default="undefined", serialization_alias="skillsTarget") + agent_deployed: bool = Field(default=False, serialization_alias="agentDeployed") + + +class CommandInvokedEvent(PlatformTelemetryEvent): + _event_name: ClassVar[str] = "command_invoked" + + command: str + duration_sec: float = Field(serialization_alias="durationSec") + agent_mode: bool = Field(default=False, serialization_alias="agentMode") + + +class JobRunEvent(PlatformTelemetryEvent): + _event_name: ClassVar[str] = "job_run" + + job_type: str = Field(serialization_alias="jobType") + duration_sec: float = Field(default=-1.0, serialization_alias="durationSec") + plugins: list[str] = Field(default_factory=list) + model: str = "undefined" + input_tokens: int = Field(default=-1, serialization_alias="inputTokens") + output_tokens: int = Field(default=-1, serialization_alias="outputTokens") diff --git a/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/handler.py b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/handler.py new file mode 100644 index 0000000000..4e0a7b0f0a --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/handler.py @@ -0,0 +1,440 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Telemetry handler for NeMo products. + +Environment variables: +- NEMO_TELEMETRY_ENABLED: Whether telemetry is enabled. +- NEMO_DEPLOYMENT_TYPE: The deployment type the event came from. +- NEMO_TELEMETRY_ENDPOINT: The endpoint to send the telemetry events to. +- NEMO_SESSION_PREFIX: Optional prefix to add to session IDs. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import platform +import threading +from collections.abc import Coroutine +from concurrent.futures import Future, ThreadPoolExecutor +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +import httpx + +from nemo_platform.cli.telemetry.events import PlatformTelemetryEvent + +CLIENT_ID = "184482118588404" +NEMO_TELEMETRY_VERSION = "nemo-telemetry/1.0" +DEFAULT_ENDPOINT = "https://events.telemetry.data.nvidia.com/v1.1/events/json" +MAX_RETRIES = 3 +# Tight explicit timeout so a hung or black-holed endpoint can never block a command +# at exit for longer than this. httpx's default is ~5s, which is too long for a +# best-effort flush that runs synchronously on the command's exit path. +SEND_TIMEOUT_SECONDS = 2.0 +logger = logging.getLogger(__name__) + + +def _telemetry_enabled() -> bool: + value = os.getenv("NEMO_TELEMETRY_ENABLED") + if value is None: + return True + return value.strip().lower() == "true" + + +def _cpu_architecture() -> str: + return platform.machine() or "undefined" + + +def _telemetry_endpoint() -> str: + return os.getenv("NEMO_TELEMETRY_ENDPOINT", DEFAULT_ENDPOINT) + + +def _redact_endpoint(endpoint: str) -> str: + """Redact query parameters before logging telemetry endpoints.""" + try: + parsed = urlsplit(endpoint) + except ValueError: + return "" + query = "" if parsed.query else "" + return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, query, parsed.fragment)) + + +def _session_prefix() -> str | None: + return os.getenv("NEMO_SESSION_PREFIX") + + +@dataclass +class QueuedEvent: + event: PlatformTelemetryEvent + timestamp: datetime + retry_count: int = 0 + + +def _get_iso_timestamp(dt: datetime | None = None) -> str: + if dt is None: + dt = datetime.now(timezone.utc) + return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}Z" + + +def build_payload( + events: list[QueuedEvent], *, source_client_version: str, session_id: str = "undefined" +) -> dict[str, Any]: + if not events: + raise ValueError("build_payload requires at least one event") + return { + "browserType": "undefined", + "clientId": CLIENT_ID, + "clientType": "Native", + "clientVariant": "Release", + "clientVer": source_client_version, + "cpuArchitecture": _cpu_architecture(), + "deviceGdprBehOptIn": "None", + "deviceGdprFuncOptIn": "None", + "deviceGdprTechOptIn": "None", + "deviceId": "undefined", + "deviceMake": "undefined", + "deviceModel": "undefined", + "deviceOS": "undefined", + "deviceOSVersion": "undefined", + "deviceType": "undefined", + "eventProtocol": "1.6", + "eventSchemaVer": events[0].event._schema_version, + "eventSysVer": NEMO_TELEMETRY_VERSION, + "externalUserId": "undefined", + "gdprBehOptIn": "None", + "gdprFuncOptIn": "None", + "gdprTechOptIn": "None", + "idpId": "undefined", + "integrationId": "undefined", + "productName": "undefined", + "productVersion": "undefined", + "sentTs": _get_iso_timestamp(), + "sessionId": session_id, + "userId": "undefined", + "events": [ + { + "ts": _get_iso_timestamp(queued.timestamp), + "parameters": queued.event.model_dump(by_alias=True, mode="json"), + "name": queued.event._event_name, + } + for queued in events + ], + } + + +class TelemetryHandler: + """ + Handles telemetry event batching, flushing, and retry logic for NeMo products. + + Supports two usage patterns: + + - **Background mode**: call ``start()`` (or use ``with handler:``) to spawn + a daemon thread with its own event loop that drives periodic flushing. + ``stop()`` schedules a final flush, then stops the loop and joins the thread. + - **Fire-and-flush mode**: skip ``start()``, ``enqueue()`` events, then call + ``stop()`` to flush once. No background thread is created unless the caller + already has a running event loop, in which case the one-shot flush is + offloaded to a worker thread with its own loop. + + Args: + flush_interval_seconds (float): The interval in seconds to flush the events. + max_queue_size (int): The maximum number of events to queue before flushing. + max_retries (int): The maximum number of times to retry sending an event. + source_client_version (str): The version of the source client. This should be the version of + the actual NeMo product that is sending the events, typically the same as the version of + a PyPi package that a user would install. + session_id (str): An optional session ID to associate with the events. + This should be a unique identifier for the session, such as a UUID. + It is used to group events together. + """ + + def __init__( + self, + flush_interval_seconds: float = 120.0, + max_queue_size: int = 50, + max_retries: int = MAX_RETRIES, + source_client_version: str = "undefined", + session_id: str = "undefined", + ): + self._flush_interval = flush_interval_seconds + self._max_queue_size = max_queue_size + self._max_retries = max_retries + self._events: list[QueuedEvent] = [] + self._dlq: list[QueuedEvent] = [] + self._queue_lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._flush_signal: asyncio.Event | None = None + self._timer_task: asyncio.Task | None = None + self._running = False + self._source_client_version = source_client_version + prefix = _session_prefix() + self._session_id = f"{prefix}{session_id}" if prefix else session_id + + # -- Async API ----------------------------------------------------------- + + async def astart(self) -> None: + """Start the background timer task on the current event loop.""" + if self._running: + return + self._loop = asyncio.get_running_loop() + self._flush_signal = asyncio.Event() + self._running = True + self._timer_task = asyncio.create_task(self._timer_loop()) + + async def astop(self) -> None: + """Cancel the timer task and flush any remaining events.""" + if not self._running: + await self._flush_events() + return + self._running = False + if self._flush_signal is not None: + self._flush_signal.set() + if self._timer_task is not None: + self._timer_task.cancel() + try: + await self._timer_task + except asyncio.CancelledError: + pass + self._timer_task = None + await self._flush_events() + self._loop = None + self._flush_signal = None + + async def aflush(self) -> None: + """Flush all queued events immediately and await completion.""" + await self._flush_events() + + # -- Sync API ------------------------------------------------------------ + + def start(self) -> None: + """Spawn a daemon thread with a persistent event loop for periodic flushing.""" + if self._running: + return + ready = threading.Event() + startup_error: list[BaseException] = [] + + def _run() -> None: + try: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + self._loop = loop + self._flush_signal = asyncio.Event() + self._timer_task = loop.create_task(self._timer_loop()) + self._running = True + except BaseException as exc: # noqa: BLE001 + startup_error.append(exc) + ready.set() + return + ready.set() + try: + loop.run_forever() + finally: + loop.close() + + self._thread = threading.Thread(target=_run, name="nemo-telemetry", daemon=True) + self._thread.start() + ready.wait() + if startup_error: + self._thread = None + raise startup_error[0] + + def stop(self) -> None: + """Flush pending events. If a background thread is running, shut it down and join.""" + if self._running and self._loop is not None and self._thread is not None: + loop = self._loop + future = asyncio.run_coroutine_threadsafe(self._astop_inner(), loop) + try: + future.result(timeout=30) + except Exception: # noqa: BLE001 + pass + loop.call_soon_threadsafe(loop.stop) + self._thread.join(timeout=5) + self._thread = None + self._loop = None + self._flush_signal = None + self._timer_task = None + self._running = False + return + if self._events or self._dlq: + try: + self._run_sync(self._flush_events()) + except Exception: # noqa: BLE001 + logger.debug("Telemetry stop flush failed", exc_info=True) + + def flush(self) -> None: + """Flush all queued events immediately and wait for completion.""" + if self._running and self._loop is not None and self._thread is not None: + future: Future[None] = asyncio.run_coroutine_threadsafe(self._flush_events(), self._loop) + try: + future.result(timeout=30) + except Exception: # noqa: BLE001 + pass + return + if self._events or self._dlq: + try: + self._run_sync(self._flush_events()) + except Exception: # noqa: BLE001 + logger.debug("Telemetry flush failed", exc_info=True) + + @staticmethod + def _run_sync(coro: Coroutine[Any, Any, Any]) -> Any: + """Run a coroutine synchronously from sync or async caller contexts. + + ``asyncio.run`` raises when called from a thread that already has a + running event loop, such as a notebook kernel or an async SDK caller. In + that case, run the coroutine in a worker thread so telemetry still gets + a fresh event loop while remaining synchronous to the caller. + """ + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + + if loop is not None: + with ThreadPoolExecutor(max_workers=1) as pool: + future = pool.submit(asyncio.run, coro) + return future.result(timeout=30) + return asyncio.run(coro) + + async def _astop_inner(self) -> None: + """Async shutdown body run on the background loop.""" + self._running = False + if self._flush_signal is not None: + self._flush_signal.set() + if self._timer_task is not None: + self._timer_task.cancel() + try: + await self._timer_task + except asyncio.CancelledError: + pass + self._timer_task = None + await self._flush_events() + + # -- Enqueue / signalling ------------------------------------------------ + + def enqueue(self, event: object) -> None: + if not _telemetry_enabled(): + return + if not isinstance(event, PlatformTelemetryEvent): + return + queued = QueuedEvent(event=event, timestamp=datetime.now(timezone.utc)) + with self._queue_lock: + self._events.append(queued) + should_signal = len(self._events) >= self._max_queue_size + if should_signal: + self._signal_flush() + + def _signal_flush(self) -> None: + """Set the flush signal, threadsafe across the background-loop boundary.""" + loop = self._loop + signal = self._flush_signal + if loop is None or signal is None: + return + try: + loop.call_soon_threadsafe(signal.set) + except RuntimeError: + pass + + # -- Context managers ---------------------------------------------------- + + def __enter__(self) -> TelemetryHandler: + self.start() + return self + + def __exit__(self, *_: object) -> None: + self.stop() + + async def __aenter__(self) -> TelemetryHandler: + await self.astart() + return self + + async def __aexit__(self, *_: object) -> None: + await self.astop() + + # -- Internal loop ------------------------------------------------------- + + async def _timer_loop(self) -> None: + assert self._flush_signal is not None + while self._running: + try: + await asyncio.wait_for( + self._flush_signal.wait(), + timeout=self._flush_interval, + ) + except asyncio.TimeoutError: + pass + self._flush_signal.clear() + await self._flush_events() + + async def _flush_events(self) -> None: + with self._queue_lock: + dlq_events, self._dlq = self._dlq, [] + new_events, self._events = self._events, [] + events_to_send = dlq_events + new_events + if events_to_send: + await self._send_events(events_to_send) + + async def _send_events(self, events: list[QueuedEvent]) -> None: + try: + async with httpx.AsyncClient(timeout=SEND_TIMEOUT_SECONDS) as client: + await self._send_events_with_client(client, events) + except Exception: # noqa: BLE001 + logger.debug("Telemetry send failed; routing events to DLQ", exc_info=True) + self._add_to_dlq(events) + + async def _send_events_with_client(self, client: httpx.AsyncClient, events: list[QueuedEvent]) -> None: + if not events: + return + + payload = build_payload(events, source_client_version=self._source_client_version, session_id=self._session_id) + endpoint = _telemetry_endpoint() + logger.debug( + "Sending telemetry events", + extra={ + "ctx": { + "endpoint": _redact_endpoint(endpoint), + "event_count": len(events), + "events": [ + { + "name": queued.event._event_name, + "task": getattr(queued.event, "task", "unknown"), + "task_status": getattr(queued.event, "task_status", "unknown"), + "deployment_type": getattr(queued.event, "deployment_type", "unknown"), + "retry_count": queued.retry_count, + } + for queued in events + ], + } + }, + ) + try: + response = await client.post(endpoint, json=payload) + if response.status_code in (400, 422) or response.is_success: + return + if response.status_code == 413: + if len(events) == 1: + return + mid = len(events) // 2 + await self._send_events_with_client(client, events[:mid]) + await self._send_events_with_client(client, events[mid:]) + return + if response.status_code in (408, 429) or response.status_code >= 500: + self._add_to_dlq(events) + except Exception: # noqa: BLE001 + logger.debug("Telemetry POST failed; routing events to DLQ", exc_info=True) + self._add_to_dlq(events) + + def _add_to_dlq(self, events: list[QueuedEvent]) -> None: + with self._queue_lock: + for queued in events: + queued.retry_count += 1 + if queued.retry_count > self._max_retries: + continue + self._dlq.append(queued) diff --git a/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/session.py b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/session.py new file mode 100644 index 0000000000..5ebf2e5a74 --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/cli/telemetry/session.py @@ -0,0 +1,157 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Local telemetry session state with time-bounded identifier rotation.""" + +from __future__ import annotations + +import json +import logging +import os +import stat +import uuid +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +_SESSION_STATE_FILENAME = "telemetry-state.json" +_SESSION_ROTATION_DAYS = 30 +_SESSION_ROTATION_INTERVAL = timedelta(days=_SESSION_ROTATION_DAYS) +_SESSION_ID_KEY = "session_id" +_CREATED_AT_KEY = "created_at" +_STATE_FILE_MODE = stat.S_IRUSR | stat.S_IWUSR +_STATE_DIR_MODE = stat.S_IRWXU + + +@dataclass(frozen=True) +class _SessionState: + session_id: str + created_at: datetime + + +_cached_state: _SessionState | None = None + + +def _now_utc() -> datetime: + return datetime.now(timezone.utc) + + +def _default_state_root() -> Path: + xdg_state_home = os.environ.get("XDG_STATE_HOME") + if xdg_state_home: + return Path(xdg_state_home).expanduser() / "nmp" + return Path.home() / ".local" / "state" / "nmp" + + +def _session_state_path() -> Path: + return _default_state_root() / _SESSION_STATE_FILENAME + + +def _format_created_at(dt: datetime) -> str: + return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _parse_created_at(value: Any) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + try: + parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _read_session_state(path: Path) -> _SessionState | None: + if not path.exists(): + return None + try: + with path.open(encoding="utf-8") as f: + data = json.load(f) + except json.JSONDecodeError: + return None + if not isinstance(data, dict): + return None + + session_id = data.get(_SESSION_ID_KEY) + created_at = _parse_created_at(data.get(_CREATED_AT_KEY)) + if not isinstance(session_id, str) or not session_id.strip() or created_at is None: + return None + return _SessionState(session_id=session_id, created_at=created_at) + + +def _requires_rotation(state: _SessionState, now: datetime) -> bool: + if state.created_at > now: + return True + return now - state.created_at >= _SESSION_ROTATION_INTERVAL + + +def _new_session_state(now: datetime) -> _SessionState: + return _SessionState(session_id=uuid.uuid4().hex, created_at=now) + + +def _write_session_state(path: Path, state: _SessionState) -> None: + created_parent = not path.parent.exists() + path.parent.mkdir(parents=True, exist_ok=True) + if created_parent: + try: + os.chmod(path.parent, _STATE_DIR_MODE) + except OSError: + pass + + payload = { + _SESSION_ID_KEY: state.session_id, + _CREATED_AT_KEY: _format_created_at(state.created_at), + } + tmp_path = path.with_name(f".{path.name}.{os.getpid()}.tmp") + fd = os.open(tmp_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, _STATE_FILE_MODE) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + fd = -1 + json.dump(payload, f, sort_keys=True) + f.write("\n") + os.replace(tmp_path, path) + os.chmod(path, _STATE_FILE_MODE) + finally: + if fd >= 0: + os.close(fd) + try: + tmp_path.unlink() + except FileNotFoundError: + pass + + +def _load_or_rotate_session_state(path: Path, now: datetime) -> _SessionState: + state = _read_session_state(path) + if state is not None and not _requires_rotation(state, now): + return state + + state = _new_session_state(now) + _write_session_state(path, state) + return state + + +def get_session_id() -> str: + """Return the current random telemetry session identifier, rotating every 30 days. + + A missing, corrupt, dateless, future-dated, or expired state file is replaced + with a new random identifier and fresh creation timestamp. If local state cannot + be read or written, fall back to a process-local random identifier so telemetry + remains best-effort and never blocks command execution. + """ + global _cached_state + + now = _now_utc() + if _cached_state is not None and not _requires_rotation(_cached_state, now): + return _cached_state.session_id + + try: + _cached_state = _load_or_rotate_session_state(_session_state_path(), now) + return _cached_state.session_id + except Exception: + logger.debug("Failed to resolve telemetry session state; using an ephemeral identifier", exc_info=True) + _cached_state = _new_session_state(now) + return _cached_state.session_id diff --git a/sdk/python/nemo-platform/src/nemo_platform/config/models.py b/sdk/python/nemo-platform/src/nemo_platform/config/models.py index 3a13e80916..f9a6500686 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/config/models.py +++ b/sdk/python/nemo-platform/src/nemo_platform/config/models.py @@ -221,6 +221,10 @@ class ConfigFile(BaseModel): default=None, description="User-selected paths for local services (set by `nemo setup`).", ) + telemetry_enabled: bool = Field( + default=True, + description="Whether anonymous usage telemetry is enabled. Set to false to opt out.", + ) def ensure_context( self, diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/__init__.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/__init__.py new file mode 100644 index 0000000000..1275d78dff --- /dev/null +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/__init__.py @@ -0,0 +1,15 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/conftest.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/conftest.py new file mode 100644 index 0000000000..1388482745 --- /dev/null +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/conftest.py @@ -0,0 +1,21 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from collections.abc import Iterator + +import pytest +from nemo_platform.cli.telemetry.events import _CI_ENV_VARS + + +@pytest.fixture(autouse=True) +def _clear_ci_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + """Clear CI markers so telemetry tests are deterministic on developer machines and in CI. + + Without this, tests that assert the default ``is_ci`` value pass locally but fail when the + suite runs under GitHub Actions (which sets ``CI`` and ``GITHUB_ACTIONS``). Tests that + exercise CI detection set these variables explicitly, which overrides this fixture. + """ + for var in _CI_ENV_VARS: + monkeypatch.delenv(var, raising=False) + monkeypatch.delenv("NEMO_DEPLOYMENT_TYPE", raising=False) + yield diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_emit.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_emit.py new file mode 100644 index 0000000000..699b61301a --- /dev/null +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_emit.py @@ -0,0 +1,208 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +import json +import uuid +from collections.abc import Iterator +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest +from nemo_platform.cli.telemetry import emit as emit_mod +from nemo_platform.cli.telemetry import session as session_mod +from nemo_platform.cli.telemetry.events import CommandInvokedEvent, TaskStatusEnum + + +@pytest.fixture(autouse=True) +def _isolate_local_telemetry_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Iterator[None]: + session_mod._cached_state = None + monkeypatch.delenv("NEMO_TELEMETRY_ENABLED", raising=False) + monkeypatch.setenv("NMP_CONFIG_FILE", str(tmp_path / "config.yaml")) + monkeypatch.setattr(session_mod, "_session_state_path", lambda: tmp_path / "telemetry-state.json") + yield + session_mod._cached_state = None + + +def _event(): + return CommandInvokedEvent(command="docs", task_status=TaskStatusEnum.COMPLETED, duration_sec=0.1) + + +class TestOptOutLayers: + def test_env_layer(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + assert emit_mod.telemetry_opted_in() is False + + def test_config_layer(self, monkeypatch, tmp_path): + cfg = tmp_path / "config.yaml" + cfg.write_text("telemetry_enabled: false\n") + monkeypatch.setenv("NMP_CONFIG_FILE", str(cfg)) + assert emit_mod.telemetry_opted_in() is False + + def test_invocation_flag_layer(self): + emit_mod.set_invocation_opt_out(True) + try: + assert emit_mod.telemetry_opted_in() is False + finally: + emit_mod.set_invocation_opt_out(False) + + def test_default_is_on(self): + assert emit_mod.telemetry_opted_in() is True + + def test_config_load_error_fails_closed(self, monkeypatch): + """A broken/parse-error config must fail closed (opted out), not default to on.""" + from nemo_platform.config import config as config_mod + + def boom(*_args: object, **_kwargs: object) -> None: + raise RuntimeError("broken config") + + monkeypatch.setattr(config_mod.Config, "load", boom) + assert emit_mod.telemetry_opted_in() is False + + +class TestEmitEvent: + @patch.object(emit_mod, "TelemetryHandler") + def test_emit_enqueues_and_stops(self, handler_cls): + instance = Mock() + handler_cls.return_value = instance + emit_mod.emit_event(_event()) + instance.enqueue.assert_called_once() + instance.stop.assert_called_once() + + @patch.object(emit_mod, "TelemetryHandler") + def test_emit_skipped_when_opted_out(self, handler_cls, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "0") + emit_mod.emit_event(_event()) + handler_cls.assert_not_called() + + @patch.object(emit_mod, "TelemetryHandler", side_effect=RuntimeError("boom")) + def test_emit_never_raises(self, _handler_cls): + emit_mod.emit_event(_event()) # must not raise + + @patch.object(emit_mod, "TelemetryHandler") + def test_session_id_is_stable_across_calls(self, handler_cls): + """Every event in one process shares the cached session id.""" + session_mod._cached_state = None + emit_mod.emit_event(_event()) + emit_mod.emit_event(_event()) + session_ids = [call.kwargs["session_id"] for call in handler_cls.call_args_list] + assert len(session_ids) == 2 + assert session_ids[0] == session_ids[1] + + +class TestTelemetrySessionState: + def _use_state_path(self, monkeypatch, tmp_path): + session_mod._cached_state = None + path = tmp_path / "telemetry-state.json" + monkeypatch.setattr(session_mod, "_session_state_path", lambda: path) + return path + + def _set_now(self, monkeypatch, now): + monkeypatch.setattr(session_mod, "_now_utc", lambda: now) + + def _set_next_uuid(self, monkeypatch, value: str): + monkeypatch.setattr(session_mod.uuid, "uuid4", lambda: uuid.UUID(value)) + + def test_missing_state_creates_session_id_with_creation_date(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + now = datetime(2026, 7, 27, 12, 0, tzinfo=timezone.utc) + self._set_now(monkeypatch, now) + self._set_next_uuid(monkeypatch, "11111111-1111-4111-8111-111111111111") + + session_id = session_mod.get_session_id() + + assert session_id == "11111111111141118111111111111111" + assert json.loads(path.read_text()) == { + "created_at": "2026-07-27T12:00:00Z", + "session_id": session_id, + } + + def test_recent_state_is_reused(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + path.write_text( + json.dumps({"session_id": "existing", "created_at": "2026-07-01T00:00:00Z"}), + encoding="utf-8", + ) + self._set_now(monkeypatch, datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc)) + + assert session_mod.get_session_id() == "existing" + assert json.loads(path.read_text())["session_id"] == "existing" + + def test_state_rotates_after_thirty_days(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + now = datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc) + path.write_text( + json.dumps( + { + "session_id": "expired", + "created_at": session_mod._format_created_at(now - timedelta(days=30)), + } + ), + encoding="utf-8", + ) + self._set_now(monkeypatch, now) + self._set_next_uuid(monkeypatch, "22222222-2222-4222-8222-222222222222") + + session_id = session_mod.get_session_id() + + assert session_id == "22222222222242228222222222222222" + state = json.loads(path.read_text()) + assert state["session_id"] == session_id + assert state["created_at"] == "2026-07-27T00:00:00Z" + + def test_state_without_creation_date_rotates(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + now = datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc) + path.write_text(json.dumps({"session_id": "dateless"}), encoding="utf-8") + self._set_now(monkeypatch, now) + self._set_next_uuid(monkeypatch, "33333333-3333-4333-8333-333333333333") + + assert session_mod.get_session_id() == "33333333333343338333333333333333" + + def test_malformed_state_rotates(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + path.write_text("not json", encoding="utf-8") + self._set_now(monkeypatch, datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc)) + self._set_next_uuid(monkeypatch, "55555555-5555-4555-8555-555555555555") + + session_id = session_mod.get_session_id() + + assert session_id == "55555555555545558555555555555555" + assert json.loads(path.read_text())["session_id"] == session_id + + def test_future_dated_state_rotates(self, monkeypatch, tmp_path): + path = self._use_state_path(monkeypatch, tmp_path) + now = datetime(2026, 7, 27, 0, 0, tzinfo=timezone.utc) + path.write_text( + json.dumps({"session_id": "future", "created_at": "2026-08-01T00:00:00Z"}), + encoding="utf-8", + ) + self._set_now(monkeypatch, now) + self._set_next_uuid(monkeypatch, "44444444-4444-4444-8444-444444444444") + + assert session_mod.get_session_id() == "44444444444444448444444444444444" + + +class TestFirstRunNotice: + def test_notice_printed_once_to_stderr(self, capsys, tmp_path, monkeypatch): + monkeypatch.setattr(emit_mod, "_notice_marker_path", lambda: tmp_path / "telemetry-notice-shown") + emit_mod.maybe_print_first_run_notice() + first = capsys.readouterr() + assert "anonymous usage data" in first.err + assert first.out == "" # stderr only; stdout stays machine-clean + emit_mod.maybe_print_first_run_notice() + assert capsys.readouterr().err == "" + + def test_no_notice_when_opted_out(self, capsys, tmp_path, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + monkeypatch.setattr(emit_mod, "_notice_marker_path", lambda: tmp_path / "telemetry-notice-shown") + emit_mod.maybe_print_first_run_notice() + assert capsys.readouterr().err == "" + + def test_notice_not_printed_when_marker_write_fails(self, capsys, tmp_path, monkeypatch): + marker_parent = tmp_path / "not-a-directory" + marker_parent.write_text("already a file") + monkeypatch.setattr(emit_mod, "_notice_marker_path", lambda: marker_parent / "telemetry-notice-shown") + + emit_mod.maybe_print_first_run_notice() + + assert capsys.readouterr().err == "" diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_events.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_events.py new file mode 100644 index 0000000000..5cc7cf0e82 --- /dev/null +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_events.py @@ -0,0 +1,75 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +import pytest +from nemo_platform.cli.telemetry.events import ( + CommandInvokedEvent, + JobRunEvent, + OnboardingStepEvent, + TaskStatusEnum, + is_ci_environment, +) +from pydantic import ValidationError + + +class TestCommandInvokedEvent: + def test_defaults_and_aliases(self): + e = CommandInvokedEvent(command="jobs create", task_status=TaskStatusEnum.COMPLETED, duration_sec=1.25) + d = e.model_dump(by_alias=True, mode="json") + assert d["nemoSource"] == "platform" + assert d["command"] == "jobs create" + assert d["taskStatus"] == "completed" + assert d["durationSec"] == 1.25 + assert d["agentMode"] is False + assert d["isCi"] is False + assert e._event_name == "command_invoked" + assert e._schema_version == "1.9" + + def test_no_free_text_fields_beyond_known(self): + # privacy guard: the event cannot carry arbitrary payloads + with pytest.raises(ValidationError): + CommandInvokedEvent(command="x", task_status=TaskStatusEnum.COMPLETED, duration_sec=0, prompt="secret") + + +class TestOnboardingStepEvent: + def test_fields(self): + e = OnboardingStepEvent(step="provider_connected", task_status=TaskStatusEnum.COMPLETED, provider_type="openai") + d = e.model_dump(by_alias=True, mode="json") + assert d["step"] == "provider_connected" + assert d["providerType"] == "openai" + assert d["modelsDiscoveredBucket"] == "undefined" + assert d["skillsTarget"] == "undefined" + assert d["agentDeployed"] is False + assert e._event_name == "onboarding_step" + + +class TestJobRunEvent: + def test_token_defaults_are_minus_one(self): + e = JobRunEvent(job_type="auditor.audit", task_status=TaskStatusEnum.ERROR, duration_sec=10.0) + d = e.model_dump(by_alias=True, mode="json") + assert d["jobType"] == "auditor.audit" + assert d["inputTokens"] == -1 + assert d["outputTokens"] == -1 + assert d["model"] == "undefined" + assert d["plugins"] == [] + assert e._event_name == "job_run" + + +class TestCiDetection: + @pytest.mark.parametrize("var", ["CI", "GITLAB_CI", "GITHUB_ACTIONS", "BUILDKITE", "JENKINS_URL"]) + def test_ci_env_detected(self, monkeypatch, var): + monkeypatch.setenv(var, "true") + assert is_ci_environment() is True + + def test_ci_env_presence_value_detected(self, monkeypatch): + monkeypatch.setenv("TEAMCITY_VERSION", "2026.07") + assert is_ci_environment() is True + + @pytest.mark.parametrize("value", ["", "0", "false", "no", "off"]) + def test_ci_env_falsey_value_is_not_detected(self, monkeypatch, value): + monkeypatch.setenv("CI", value) + assert is_ci_environment() is False + + def test_no_ci_env(self, monkeypatch): + for var in ("CI", "GITLAB_CI", "GITHUB_ACTIONS", "BUILDKITE", "JENKINS_URL"): + monkeypatch.delenv(var, raising=False) + assert is_ci_environment() is False diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_handler.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_handler.py new file mode 100644 index 0000000000..d4c6f47dae --- /dev/null +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_handler.py @@ -0,0 +1,599 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import asyncio +import importlib +import threading +from datetime import datetime, timezone +from typing import ClassVar +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from nemo_platform.cli.telemetry.events import DeploymentTypeEnum, PlatformTelemetryEvent, _deployment_type +from nemo_platform.cli.telemetry.handler import ( + QueuedEvent, + TelemetryHandler, + _telemetry_enabled, + _telemetry_endpoint, + build_payload, +) +from pydantic import Field + +telemetry_module = importlib.import_module("nemo_platform.cli.telemetry.handler") + + +# ============================================================================= +# Stub Event Model for Testing +# ============================================================================= + + +class _StubEvent(PlatformTelemetryEvent): + """Minimal concrete event for testing, subclassing the real PlatformTelemetryEvent. + + ``task_status`` and ``deployment_type`` are redeclared as plain strings to shed + the base's serialization aliases, so the handler tests keep asserting the + snake_case keys the Task 1 handler emitted. + """ + + _event_name: ClassVar[str] = "stub_event" + task: str = Field(default="test_task") + task_status: str = Field(default="completed") + deployment_type: str = Field(default="sdk") + + +# ============================================================================= +# Env-var helpers +# ============================================================================= + + +class TestEnvHelpers: + def test_telemetry_enabled_default(self, monkeypatch): + monkeypatch.delenv("NEMO_TELEMETRY_ENABLED", raising=False) + assert _telemetry_enabled() is True + + def test_telemetry_enabled_disabled(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + assert _telemetry_enabled() is False + + def test_telemetry_enabled_true_when_explicitly_true(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + assert _telemetry_enabled() is True + + @pytest.mark.parametrize("value", ["", "0", "1", "yes", "yep", "for sure"]) + def test_telemetry_enabled_non_true_values_disable(self, monkeypatch, value): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", value) + assert _telemetry_enabled() is False + + def test_telemetry_endpoint_preserves_case(self, monkeypatch): + custom = "https://Events.Telemetry.example.COM/v1/Events?Token=AbC" + monkeypatch.setenv("NEMO_TELEMETRY_ENDPOINT", custom) + assert _telemetry_endpoint() == custom + + def test_deployment_type_default(self, monkeypatch): + monkeypatch.delenv("NEMO_DEPLOYMENT_TYPE", raising=False) + assert _deployment_type() is DeploymentTypeEnum.CLI + + def test_deployment_type_invalid_falls_back_to_undefined(self, monkeypatch): + monkeypatch.setenv("NEMO_DEPLOYMENT_TYPE", "definitely-not-real") + assert _deployment_type() is DeploymentTypeEnum.UNDEFINED + + def test_deployment_type_nvidia_internal(self, monkeypatch): + monkeypatch.setenv("NEMO_DEPLOYMENT_TYPE", "nvidia-internal") + assert _deployment_type() is DeploymentTypeEnum.NVIDIA_INTERNAL + + +# ============================================================================= +# build_payload +# ============================================================================= + + +class TestBuildPayload: + def _make_queued(self, task: str = "generate", status: str = "completed") -> QueuedEvent: + event = _StubEvent(task=task, task_status=status) + return QueuedEvent(event=event, timestamp=datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc)) + + def test_structure(self): + queued = self._make_queued() + payload = build_payload([queued], source_client_version="1.2.3", session_id="test-session") + assert payload["clientId"] == "184482118588404" + assert payload["clientVer"] == "1.2.3" + assert payload["sessionId"] == "test-session" + assert len(payload["events"]) == 1 + + def test_event_fields_serialize_as_strings(self, monkeypatch): + monkeypatch.delenv("NEMO_DEPLOYMENT_TYPE", raising=False) + + queued = self._make_queued(task="train", status="error") + payload = build_payload([queued], source_client_version="0.0.1") + event_entry = payload["events"][0] + assert event_entry["name"] == "stub_event" + assert event_entry["ts"] == "2025-01-01T12:00:00.000Z" + params = event_entry["parameters"] + assert params["task"] == "train" + assert params["task_status"] == "error" + + def test_payload_is_json_serializable(self): + """Regression: the full payload must be encodable by the stdlib JSON encoder.""" + import json + + queued = self._make_queued(status="error") + payload = build_payload([queued], source_client_version="1.0.0") + json.dumps(payload) + + def test_multiple_events(self): + events = [self._make_queued(task=t) for t in ("train", "generate", "evaluate")] + payload = build_payload(events, source_client_version="1.0.0") + assert len(payload["events"]) == 3 + tasks = [e["parameters"]["task"] for e in payload["events"]] + assert tasks == ["train", "generate", "evaluate"] + + def test_default_session_id(self): + queued = self._make_queued() + payload = build_payload([queued], source_client_version="1.0.0") + assert payload["sessionId"] == "undefined" + + def test_cpu_architecture_uses_platform_machine(self, monkeypatch): + monkeypatch.setattr(telemetry_module.platform, "machine", lambda: "arm64") + payload = build_payload([self._make_queued()], source_client_version="1.0.0") + assert payload["cpuArchitecture"] == "arm64" + + def test_cpu_architecture_empty_value_falls_back_to_undefined(self, monkeypatch): + monkeypatch.setattr(telemetry_module.platform, "machine", lambda: "") + payload = build_payload([self._make_queued()], source_client_version="1.0.0") + assert payload["cpuArchitecture"] == "undefined" + + def test_empty_events_raises(self): + with pytest.raises(ValueError): + build_payload([], source_client_version="1.0.0") + + +# ============================================================================= +# TelemetryHandler — telemetry disabled +# ============================================================================= + + +class TestTelemetryDisabled: + def test_enqueue_noop_when_disabled(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + handler = TelemetryHandler() + event = _StubEvent(task="generate") + handler.enqueue(event) + assert handler._events == [] + + def test_enqueue_noop_for_non_event(self, monkeypatch): + """Silently ignores non-TelemetryEvent objects regardless of env.""" + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler() + handler.enqueue("not an event") + assert handler._events == [] + + +# ============================================================================= +# TelemetryHandler — enqueue and flush +# ============================================================================= + + +class TestTelemetryHandlerEnqueue: + def test_enqueue_adds_event(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler() + event = _StubEvent(task="generate") + handler.enqueue(event) + assert len(handler._events) == 1 + assert handler._events[0].event is event + + def test_enqueue_does_not_add_event_when_telemetry_is_disabled(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "false") + handler = TelemetryHandler() + event = _StubEvent(task="generate") + handler.enqueue(event) + assert len(handler._events) == 0 + + def test_enqueue_at_max_queue_size_signals_flush_when_running(self, monkeypatch): + """When a background loop is up, hitting max_queue_size should signal a flush.""" + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(max_queue_size=3, flush_interval_seconds=60.0) + flushed = threading.Event() + + async def fake_send(_events): + flushed.set() + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.start() + try: + for _ in range(3): + handler.enqueue(_StubEvent(task="run")) + assert flushed.wait(timeout=2.0), "max_queue_size flush signal did not fire" + finally: + handler.stop() + + +# ============================================================================= +# TelemetryHandler — _flush_events queue clearing and DLQ +# ============================================================================= + + +class TestFlushEventsQueueClearing: + async def test_flush_events_clears_queue(self): + """_flush_events() must drain _events even when the underlying send succeeds.""" + handler = TelemetryHandler(source_client_version="1.0.0") + event = _StubEvent(task="generate") + handler._events.append(QueuedEvent(event=event, timestamp=datetime.now(timezone.utc))) + + async def fake_send(events): + assert len(events) == 1 + + with patch.object(handler, "_send_events", side_effect=fake_send): + await handler._flush_events() + + assert handler._events == [] + assert handler._dlq == [] + + async def test_flush_events_includes_dlq(self): + handler = TelemetryHandler(source_client_version="1.0.0") + event = _StubEvent(task="generate") + handler._dlq.append(QueuedEvent(event=event, timestamp=datetime.now(timezone.utc), retry_count=1)) + handler._events.append(QueuedEvent(event=event, timestamp=datetime.now(timezone.utc))) + + sent: list[list[QueuedEvent]] = [] + + async def fake_send(events): + sent.append(list(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + await handler._flush_events() + + assert handler._events == [] + assert handler._dlq == [] + assert len(sent) == 1 + assert len(sent[0]) == 2 + + +# ============================================================================= +# TelemetryHandler — send and retry +# ============================================================================= + + +class TestTelemetryHandlerSend: + def _make_handler(self) -> TelemetryHandler: + return TelemetryHandler(source_client_version="1.0.0", session_id="s1") + + def _make_queued(self) -> QueuedEvent: + event = _StubEvent(task="generate") + return QueuedEvent(event=event, timestamp=datetime.now(timezone.utc)) + + async def test_debug_logs_event_metadata_before_post(self, monkeypatch): + handler = self._make_handler() + queued = self._make_queued() + events_seen: list[str] = [] + debug_calls = [] + + monkeypatch.setenv("NEMO_TELEMETRY_ENDPOINT", "https://Events.Telemetry.example.COM/v1/Events?Token=AbC") + + def fake_debug(message, *, extra): + events_seen.append("debug") + debug_calls.append((message, extra)) + + async def fake_post(*_args, **_kwargs): + events_seen.append("post") + return MagicMock(status_code=200, is_success=True) + + monkeypatch.setattr(telemetry_module.logger, "debug", fake_debug) + mock_client = AsyncMock() + mock_client.post.side_effect = fake_post + + await handler._send_events_with_client(mock_client, [queued]) + + assert events_seen == ["debug", "post"] + message, extra = debug_calls[0] + assert message == "Sending telemetry events" + ctx = extra["ctx"] + assert ctx["endpoint"] == "https://Events.Telemetry.example.COM/v1/Events?" + assert ctx["event_count"] == 1 + assert ctx["events"] == [ + { + "name": "stub_event", + "task": "generate", + "task_status": "completed", + "deployment_type": "sdk", + "retry_count": 0, + } + ] + + async def test_successful_send_does_not_dlq(self): + handler = self._make_handler() + queued = self._make_queued() + + mock_response = MagicMock(status_code=200, is_success=True) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + + await handler._send_events_with_client(mock_client, [queued]) + mock_client.post.assert_awaited_once() + assert handler._dlq == [] + + async def test_500_adds_to_dlq(self): + handler = self._make_handler() + queued = self._make_queued() + + mock_response = MagicMock(status_code=500, is_success=False) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + + await handler._send_events_with_client(mock_client, [queued]) + assert len(handler._dlq) == 1 + assert handler._dlq[0].retry_count == 1 + + async def test_429_adds_to_dlq(self): + handler = self._make_handler() + queued = self._make_queued() + + mock_response = MagicMock(status_code=429, is_success=False) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + + await handler._send_events_with_client(mock_client, [queued]) + assert len(handler._dlq) == 1 + assert handler._dlq[0].retry_count == 1 + + async def test_exceeds_max_retries_dropped(self): + handler = self._make_handler() + queued = self._make_queued() + queued.retry_count = handler._max_retries + + mock_response = MagicMock(status_code=500, is_success=False) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + + await handler._send_events_with_client(mock_client, [queued]) + assert handler._dlq == [] + + async def test_413_splits_and_retries(self): + handler = self._make_handler() + event = _StubEvent(task="generate") + events = [ + QueuedEvent(event=event, timestamp=datetime.now(timezone.utc)), + QueuedEvent(event=event, timestamp=datetime.now(timezone.utc)), + ] + + success_response = MagicMock(status_code=200, is_success=True) + too_large_response = MagicMock(status_code=413, is_success=False) + mock_client = AsyncMock() + mock_client.post.side_effect = [too_large_response, success_response, success_response] + + await handler._send_events_with_client(mock_client, events) + assert mock_client.post.await_count == 3 + + async def test_send_events_routes_to_dlq_on_client_setup_failure(self): + """If httpx client creation raises, events must land in DLQ rather than vanish.""" + handler = self._make_handler() + queued = self._make_queued() + + with patch("httpx.AsyncClient", side_effect=RuntimeError("boom")): + await handler._send_events([queued]) + + assert len(handler._dlq) == 1 + + def test_session_prefix_applied(self, monkeypatch): + monkeypatch.setenv("NEMO_SESSION_PREFIX", "pfx-") + handler = TelemetryHandler(session_id="abc") + assert handler._session_id == "pfx-abc" + + +# ============================================================================= +# TelemetryHandler — aflush awaits a real flush +# ============================================================================= + + +class TestAflushAwaits: + async def test_aflush_actually_flushes(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0") + event = _StubEvent(task="generate") + handler.enqueue(event) + assert len(handler._events) == 1 + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + await handler.aflush() + + assert handler._events == [] + assert sent == [1] + + +# ============================================================================= +# TelemetryHandler — sync flush from async caller contexts +# ============================================================================= + + +class TestFlushFromRunningLoop: + """Regression coverage for notebooks and async SDK callers.""" + + def test_flush_runs_to_completion_when_loop_is_running(self) -> None: + sent: list[int] = [] + + async def fake_flush(self) -> None: # noqa: ARG001 + sent.append(1) + + async def driver() -> None: + handler = TelemetryHandler(source_client_version="1.0.0") + handler._events.append( + QueuedEvent( + event=_StubEvent(task="run"), + timestamp=datetime.now(timezone.utc), + ) + ) + with patch.object(TelemetryHandler, "_flush_events", new=fake_flush): + handler.flush() + + asyncio.run(driver()) + assert sent == [1] + + def test_stop_flushes_fire_and_flush_path_when_loop_is_running(self, monkeypatch) -> None: + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + async def driver() -> None: + handler = TelemetryHandler(source_client_version="1.0.0") + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.enqueue(_StubEvent(task="run")) + handler.stop() + assert handler._events == [] + + asyncio.run(driver()) + assert sent == [1] + + def test_run_sync_propagates_exception_to_flush_boundary(self) -> None: + async def boom() -> None: + raise RuntimeError("kaboom") + + async def driver() -> None: + with pytest.raises(RuntimeError, match="kaboom"): + TelemetryHandler._run_sync(boom()) + + asyncio.run(driver()) + + +# ============================================================================= +# TelemetryHandler — sync lifecycle and context manager +# ============================================================================= + + +class TestSyncLifecycle: + def test_fire_and_flush_without_start(self, monkeypatch): + """Pattern used by the SDK: construct, enqueue, stop. No start() call. stop() must flush.""" + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0") + event = _StubEvent(task="generate") + handler.enqueue(event) + assert len(handler._events) == 1 + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.stop() + + assert handler._events == [] + assert sent == [1] + assert handler._thread is None + + def test_start_spawns_thread_and_stop_flushes(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0) + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.start() + assert handler._thread is not None + assert handler._thread.is_alive() + + handler.enqueue(_StubEvent(task="run")) + handler.stop() + + assert handler._thread is None + assert handler._loop is None + assert sent == [1] + + def test_sync_context_manager(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(TelemetryHandler, "_send_events", side_effect=fake_send, autospec=False): + with TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0) as handler: + handler.enqueue(_StubEvent(task="run")) + + assert sent == [1] + + def test_sync_flush_during_background_run(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0) + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.start() + try: + handler.enqueue(_StubEvent(task="run")) + handler.flush() + assert sent == [1] + assert handler._events == [] + finally: + handler.stop() + + def test_timer_driven_flush(self, monkeypatch): + """With a short flush interval, the background timer should drive a flush without explicit calls.""" + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + handler = TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=0.05) + + flushed = threading.Event() + + async def fake_send(_events): + flushed.set() + + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.start() + try: + handler.enqueue(_StubEvent(task="run")) + assert flushed.wait(timeout=2.0), "timer-driven flush did not fire" + finally: + handler.stop() + + +# ============================================================================= +# TelemetryHandler — async lifecycle +# ============================================================================= + + +class TestAsyncLifecycle: + async def test_async_context_manager_flushes_on_exit(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + + sent: list[int] = [] + + async def fake_send(events): + sent.append(len(events)) + + async with TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0) as handler: + with patch.object(handler, "_send_events", side_effect=fake_send): + handler.enqueue(_StubEvent(task="run")) + await handler.astop() + + assert sent == [1] + + async def test_enqueue_at_max_size_signals_flush_in_async_mode(self, monkeypatch): + monkeypatch.setenv("NEMO_TELEMETRY_ENABLED", "true") + + flushed = asyncio.Event() + + async def fake_send(_events): + flushed.set() + + handler = TelemetryHandler(source_client_version="1.0.0", flush_interval_seconds=60.0, max_queue_size=2) + with patch.object(handler, "_send_events", side_effect=fake_send): + await handler.astart() + try: + handler.enqueue(_StubEvent(task="run")) + handler.enqueue(_StubEvent(task="run")) + await asyncio.wait_for(flushed.wait(), timeout=2.0) + finally: + await handler.astop() diff --git a/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_wire_contract_smoke.py b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_wire_contract_smoke.py new file mode 100644 index 0000000000..ed20d891cd --- /dev/null +++ b/sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/telemetry/test_wire_contract_smoke.py @@ -0,0 +1,179 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""CI-safe wire-contract smoke test for telemetry events. + +This module verifies that each telemetry event serializes to the wire contract +without making network calls. Each test constructs a valid event, wraps it in +QueuedEvent, calls build_payload, and asserts JSON serialization and envelope +structure. + +Live UAT validation (POST against the telemetry endpoint) is a manual runbook: +set NEMO_TELEMETRY_ENDPOINT to the UAT URL, NEMO_TELEMETRY_ENABLED=true, and +run one real command; live validation is pending nemoSource schema registration. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone + +from nemo_platform.cli.telemetry.events import ( + CommandInvokedEvent, + JobRunEvent, + OnboardingStepEvent, + TaskStatusEnum, +) +from nemo_platform.cli.telemetry.handler import QueuedEvent, build_payload + +EXPECTED_ENVELOPE_KEYS = { + "browserType", + "clientId", + "clientType", + "clientVariant", + "clientVer", + "cpuArchitecture", + "deviceGdprBehOptIn", + "deviceGdprFuncOptIn", + "deviceGdprTechOptIn", + "deviceId", + "deviceMake", + "deviceModel", + "deviceOS", + "deviceOSVersion", + "deviceType", + "eventProtocol", + "eventSchemaVer", + "eventSysVer", + "externalUserId", + "gdprBehOptIn", + "gdprFuncOptIn", + "gdprTechOptIn", + "idpId", + "integrationId", + "productName", + "productVersion", + "sentTs", + "sessionId", + "userId", + "events", +} + + +class TestWireContractSmoke: + """Verify each event type serializes to the wire contract envelope.""" + + def test_onboarding_step_event_contract(self): + """OnboardingStepEvent must serialize with camelCase aliases and correct envelope.""" + event = OnboardingStepEvent( + task_status=TaskStatusEnum.COMPLETED, + step="provider_discovery", + provider_type="huggingface", + models_discovered_bucket="10-100", + skills_target="audit_content", + agent_deployed=True, + ) + queued = QueuedEvent(event=event, timestamp=datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)) + payload = build_payload([queued], source_client_version="1.0.0", session_id="test") + + # Envelope keys must match exactly. + assert set(payload.keys()) == EXPECTED_ENVELOPE_KEYS, ( + f"Envelope keys mismatch: {set(payload.keys()) ^ EXPECTED_ENVELOPE_KEYS}" + ) + + # Schema version must be 1.9. + assert payload["eventSchemaVer"] == "1.9" + + # Must be JSON serializable. + json_str = json.dumps(payload) + assert isinstance(json_str, str) + + # Event parameters must include camelCase aliases. + event_entry = payload["events"][0] + params = event_entry["parameters"] + assert params["nemoSource"] == "platform" + assert params["taskStatus"] == "completed" + assert params["deploymentType"] == "cli" + assert params["isCi"] is False + assert params["step"] == "provider_discovery" + assert params["providerType"] == "huggingface" + assert params["modelsDiscoveredBucket"] == "10-100" + assert params["skillsTarget"] == "audit_content" + assert params["agentDeployed"] is True + assert event_entry["name"] == "onboarding_step" + + def test_command_invoked_event_contract(self): + """CommandInvokedEvent must serialize with camelCase aliases and correct envelope.""" + event = CommandInvokedEvent( + task_status=TaskStatusEnum.COMPLETED, + command="nemo platform apply", + duration_sec=2.5, + agent_mode=True, + ) + queued = QueuedEvent(event=event, timestamp=datetime(2026, 1, 1, 13, 0, 0, tzinfo=timezone.utc)) + payload = build_payload([queued], source_client_version="1.2.3", session_id="cmd-test") + + # Envelope keys must match exactly. + assert set(payload.keys()) == EXPECTED_ENVELOPE_KEYS, ( + f"Envelope keys mismatch: {set(payload.keys()) ^ EXPECTED_ENVELOPE_KEYS}" + ) + + # Schema version must be 1.9. + assert payload["eventSchemaVer"] == "1.9" + + # Must be JSON serializable. + json_str = json.dumps(payload) + assert isinstance(json_str, str) + + # Event parameters must include camelCase aliases. + event_entry = payload["events"][0] + params = event_entry["parameters"] + assert params["nemoSource"] == "platform" + assert params["taskStatus"] == "completed" + assert params["deploymentType"] == "cli" + assert params["isCi"] is False + assert params["command"] == "nemo platform apply" + assert params["durationSec"] == 2.5 + assert params["agentMode"] is True + assert event_entry["name"] == "command_invoked" + + def test_job_run_event_contract(self): + """JobRunEvent must serialize with camelCase aliases and correct envelope.""" + event = JobRunEvent( + task_status=TaskStatusEnum.COMPLETED, + job_type="evaluate", + duration_sec=15.75, + plugins=["garak", "llm_judge"], + model="nemotron-4-8b", + input_tokens=2048, + output_tokens=512, + ) + queued = QueuedEvent(event=event, timestamp=datetime(2026, 1, 1, 14, 0, 0, tzinfo=timezone.utc)) + payload = build_payload([queued], source_client_version="2.0.0", session_id="job-test") + + # Envelope keys must match exactly. + assert set(payload.keys()) == EXPECTED_ENVELOPE_KEYS, ( + f"Envelope keys mismatch: {set(payload.keys()) ^ EXPECTED_ENVELOPE_KEYS}" + ) + + # Schema version must be 1.9. + assert payload["eventSchemaVer"] == "1.9" + + # Must be JSON serializable. + json_str = json.dumps(payload) + assert isinstance(json_str, str) + + # Event parameters must include camelCase aliases. + event_entry = payload["events"][0] + params = event_entry["parameters"] + assert params["nemoSource"] == "platform" + assert params["taskStatus"] == "completed" + assert params["deploymentType"] == "cli" + assert params["isCi"] is False + assert params["jobType"] == "evaluate" + assert params["durationSec"] == 15.75 + assert params["plugins"] == ["garak", "llm_judge"] + assert params["model"] == "nemotron-4-8b" + assert params["inputTokens"] == 2048 + assert params["outputTokens"] == 512 + assert event_entry["name"] == "job_run" diff --git a/uv.lock b/uv.lock index 42abc6acc5..373267702e 100644 --- a/uv.lock +++ b/uv.lock @@ -5459,6 +5459,7 @@ version = "0.0.0" source = { editable = "packages/nemo_platform_ext" } dependencies = [ { name = "docker", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "httpx", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-sdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "ngcsdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -5485,6 +5486,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "docker", specifier = ">=7.0.0" }, + { name = "httpx", specifier = ">=0.23.0,<1" }, { name = "nemo-platform-plugin", editable = "packages/nemo_platform_plugin" }, { name = "nemo-platform-sdk", editable = "sdk/python/nemo-platform" }, { name = "ngcsdk", specifier = ">=4.8.2" },