diff --git a/.github/workflows/pr-validation.yml b/.github/workflows/pr-validation.yml index 6602ff107..d623b98c5 100644 --- a/.github/workflows/pr-validation.yml +++ b/.github/workflows/pr-validation.yml @@ -112,6 +112,32 @@ jobs: - name: Run tests run: tox -e ${{ matrix.tox-env }} + agent365-storage: + name: Agent365 storage (${{ matrix.os }}) + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + steps: + - name: Check out repository + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.10" + - name: Install package and pinned storage provider + run: python -m pip install -e ".[test]" "azure-monitor-opentelemetry-exporter==1.0.0b58" + - name: Exercise real filesystem storage and replay + run: >- + python -m pytest + tests/a365/test_persistent_storage.py + tests/a365/test_exporter.py + tests/a365/test_durable_restart.py + tests/a365/test_handler.py + tests/a365/test_replay_coordinator.py + tests/test_distro.py -q + build: name: Build wheel runs-on: ubuntu-latest diff --git a/A365_DOCUMENTATION.md b/A365_DOCUMENTATION.md index 4e4489694..a214fefe6 100644 --- a/A365_DOCUMENTATION.md +++ b/A365_DOCUMENTATION.md @@ -43,7 +43,7 @@ use_microsoft_opentelemetry( | `a365_exporter_timeout_ms` | `int` | `30000` | — | Timeout for a single A365 export operation in milliseconds. | | `a365_max_export_batch_size` | `int` | `512` | — | Maximum batch size for a single A365 export operation. | | `a365_exporter_disable_offline_storage` | `bool` | `False` | — | Disable durable offline storage. When `True`, failed exports are not persisted to disk and at-least-once delivery is not guaranteed. Defaults to `False` (storage enabled). | -| `a365_exporter_storage_directory` | `str` | `None` | — | Custom directory for durable offline storage. When `None`, a platform default path is used. Choose a path that only the current user or service account can read, because stored OTLP payloads may contain prompts or completions when sensitive-data capture is enabled. | +| `a365_exporter_storage_directory` | `str` | `None` | — | Root for an `a365` cache using Azure Monitor's offline storage provider. Configure separate sibling roots for A365 and Azure Monitor; shared or overlapping cache paths are unsupported and are not rejected automatically. `None` uses Azure Monitor's default path resolver with an Agent365 namespace. See [offline storage](#durable-delivery-offline-storage) for security and upgrade guidance. | ### Resource / Service Name @@ -95,18 +95,40 @@ Key defaults and limits: |---|---| | Enabled by default | Yes (when exporter is active) | | Retention window | 2 days | -| Maximum storage size | 50 MB | +| Maximum storage size | 50 MiB (upstream capacity check before each write; one write can cross the limit) | | Storage path | Platform default (see below) | -The default path is a sub-directory of the platform's local app data folder, -derived from the user, executable, and working directory. Processes for the -same application can share the SQLite queue; leases coordinate replay across -those processes. +Agent365 uses the **same `LocalFileStorage` provider as Azure Monitor**, not a +separate SQLite queue. The provider owns atomic JSON-blob writes, rename-based +leases, expiration, capacity checks, maintenance, and platform permissions. +Agent365 only adapts identity envelopes and its authenticated replay protocol. +Processes using the same cache can coordinate replay through those file leases. + +For an explicitly configured root, Agent365 records live in `/a365/`. +Azure Monitor uses its configured storage directory directly. When both +exporters are enabled, configure **separate sibling roots**, as in the example +below, so neither effective cache directory contains the other. Reusing the +same root does not isolate capacity: Azure Monitor recursively counts files +in its directory, including an `a365` child. Configurations that resolve to +the same effective cache directory can also cause either exporter to claim +or delete the other's incompatible records. **Shared or overlapping cache paths +are unsupported and are not detected or rejected automatically.** Customers +must keep their resolved paths separate, including any filesystem aliases. + +With no configured root, Agent365 calls Azure Monitor's +`_get_storage_directory("agent365")`: the resulting cache is +under the system temporary directory, in +`Microsoft-AzureMonitor-/opentelemetry-python-agent365`. The upstream hash +includes the namespace, user, process name, and application directory. Reusing +the same resulting path preserves blob replay across restarts. Temporary-directory +cleanup by the operating system can remove queued records; configure persistent +storage when restart durability must survive that cleanup. **Replay behavior:** - Each queued record stores only its identity (tenant, agent, agentic user, S2S flag) and payload — not a fixed URL. On replay, the exporter reconstructs the export endpoint from the exporter's *current* configuration and re-resolves a fresh bearer token, so records queued before an endpoint or credential change are still delivered correctly, and any endpoint that would not resolve to HTTPS is rejected rather than replayed. - Records that are permanently rejected by the service (e.g. `400 Bad Request`) or that fail internal validation (unsupported schema version, blank tenant/agent id, or blank payload — "poison" records) are discarded from the durable queue instead of being retried forever. +- Unreadable blobs are logged and retained for retry or retention expiry. The provider does not distinguish malformed JSON from transient read errors, so an unreadable blob is not treated as a proven poison record. - `shutdown()` is drain-safe: it signals the background replay loop to stop and blocks until any in-flight replay send actually finishes before closing the durable store and HTTP session, so an in-progress send is never left holding a closed resource and no accepted span is dropped mid-flight. Concurrent `shutdown()` callers all wait for that same cleanup to complete. **To disable durable storage** (no disk writes, best-effort delivery only): @@ -119,17 +141,89 @@ use_microsoft_opentelemetry( ) ``` -**To specify a custom storage directory:** +**To configure both exporters with separate storage roots:** ```python use_microsoft_opentelemetry( enable_a365=True, a365_enable_observability_exporter=True, - a365_exporter_storage_directory="/var/lib/my-agent/telemetry", + enable_azure_monitor=True, + a365_exporter_storage_directory=r"C:\MyApp\Telemetry\Agent365", + azure_monitor_exporter_storage_directory=r"C:\MyApp\Telemetry\AzureMonitor", ) ``` -> **Security note:** Stored OTLP payloads are unencrypted. When sensitive-data capture is enabled (`enable_sensitive_data=True`), payloads may include prompts, completions, or tool arguments. The exporter restricts the queue directory to the current user and administrators on Windows; on POSIX systems it enforces mode `0700` on the directory and `0600` on the SQLite database and sidecar files. +The resulting cache directories are `C:\MyApp\Telemetry\Agent365\a365` and +`C:\MyApp\Telemetry\AzureMonitor`. On Unix, use absolute sibling roots such as +`/var/lib/my-agent/telemetry/Agent365` and +`/var/lib/my-agent/telemetry/AzureMonitor`, provisioned with appropriate +service-account ownership and permissions. Configure the Azure Monitor +connection string or credential as usual. The same independent exporter-root +configuration is supported by the .NET SDK; Python's existing `a365` child +layout and provider permission handling remain unchanged. + +> **Security note:** Stored OTLP payloads are unencrypted. With sensitive-data capture enabled (`enable_sensitive_data=True`), payloads may include prompts, completions, or tool arguments. Do not configure a shared temporary directory such as `/tmp` as a customer storage root: a predictable child in a shared parent can be replaced or precreated. Normally provision a private, service-account-owned `0700` root and review its ACLs, ancestors, and protection against path replacement. Cache permissions do not isolate data from root, administrators, or other processes using the same account. + +The implementation inspected is Azure Monitor Python exporter **1.0.0b58**, pinned +in `uv.lock`, at commit +[`83907dc9ed2eb8c742b353b5f2cf276e039026da`](https://github.com/Azure/azure-sdk-for-python/blob/83907dc9ed2eb8c742b353b5f2cf276e039026da/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_storage.py). +Unlike the .NET provider, this Python provider creates its cache directory and, +on Linux/macOS, opens the final directory without following a symlink, checks +that the owner is the current user or root, and applies `0700` with `fchmod`. +It creates blob temporary files exclusively with mode `0600` before renaming +them. On Windows it grants the current user and administrators access and +removes inherited ACLs using `icacls`. Agent365 neither duplicates nor bypasses +these upstream cache protections. + +On Linux/macOS, Agent365 additionally checks the explicitly configured root +before initializing the provider on its `a365` child. The configured path is +normalized with `os.path.abspath`, matching the provider, before root validation +and legacy-backlog detection. The root must be a real +directory (not a symlink), owned by the current user or root, and not writable +by group or others. Unsafe roots disable offline delivery without permission +repair. A missing root is created with mode `0700` (subject to umask) and checked +after creation as well. Existing root permissions are preserved; upstream +checks and permission enforcement still apply to the existing or newly created +`a365` child. Windows behavior and the implicit default resolver are unchanged. +Creation of missing ancestors follows ordinary filesystem defaults. + +These point-in-time checks do not validate ancestor ownership, ACLs, or +protection against subsequent path replacement. Customers remain responsible +for provisioning and protecting the entire path; the checks are not a +confidentiality guarantee. + +Storage is initialized lazily on first export. Invalid configuration (including +blank paths), provider setup failure, or an initialization exception logs an +error and disables **only offline buffering and replay**; online export continues. +An explicitly configured path never falls back to another location on failure. +Explicit disablement bypasses storage initialization altogether. + +**Breaking change and existing backlogs (all platforms):** the SQLite format, +explicit-root layout, and implicit default location have changed. Existing +`queue.db`, `queue.db-wal`, and `queue.db-shm` files are **not** opened, converted, +deleted, or automatically migrated. **Required before upgrading:** restore connectivity and +drain the old queue using the previous SDK and original configuration, then shut +down all writers cleanly. Verify delivery before retiring the old directory. +This is a customer-managed upgrade prerequisite, not an SDK migration step. +If draining cannot be completed, retain a secured, coherent SQLite backup +(including any necessary WAL/SHM state) for controlled recovery with the previous +SDK; copying or renaming SQLite files into the new blob cache does not migrate +them. A detected `queue.db` in an explicitly configured root produces a warning. +No old implicit temporary/home-directory backlog is discovered automatically. + +To locate a previous implicit queue, inspect the old SDK's service environment. +On Unix, its normal parent was `$XDG_STATE_HOME/a365-durable-queue` when +`XDG_STATE_HOME` was nonempty, otherwise `~/.local/state/a365-durable-queue`. On +Windows, it was `%LOCALAPPDATA%\a365-durable-queue`, with a system-temporary +fallback when `LOCALAPPDATA` was unset. Unix also used a temporary fallback if +the home directory could not be resolved and `XDG_STATE_HOME` was unset. +Each queue was in a child named by the first 16 hexadecimal characters of +`sha256((getpass.getuser() + sys.executable + str(Path.cwd())).encode())`. +Changes to the service account, Python executable, or working directory can +leave multiple old queue directories. Identify them before upgrading and use +the previous SDK in the original environment to drain them; do not assume that +successful replay of new JSON blobs proves recovery of an old SQLite backlog. +Reusing an old directory alone does not convert its SQLite records. ## Auto-Instrumented Libraries diff --git a/CHANGELOG.md b/CHANGELOG.md index 1564ad0fc..0285f1df9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,42 @@ # Release History # Unreleased +### Breaking Changes +- Agent365 offline delivery now reuses Azure Monitor's JSON-blob storage provider, + file leases, Unix permission handling, and Windows ACL handling instead of a + separate SQLite implementation. Explicit roots use an `a365` child; + implicit paths use Azure Monitor's resolver with an Agent365 namespace. + When both exporters are enabled, configure separate sibling storage roots + for independent capacity and replay. Shared or overlapping cache paths are + unsupported and are not detected or rejected automatically. + This changes cache layout, format, and default locations on all platforms, + including Linux and macOS. Drain existing SQLite backlogs with the previous + SDK before upgrading; no automatic migration or legacy replay is performed. + Provision a private service-account-owned root (normally `0700` on Unix), + review ACLs and ancestors, and protect against path replacement. Upstream + cache permission enforcement remains enabled. +- Explicit Agent365 roots on Linux/macOS must not be symlinks, foreign-owned + (other than root), or group/world-writable. Invalid roots disable offline + delivery without changing their permissions. Missing roots are created with + mode `0700` subject to umask and validated before cache initialization. + +### Bugs Fixed +- Invalid Agent365 storage configuration or provider initialization failure logs + a nonfatal error and disables only offline buffering/replay, preserving online + export. Explicit offline disablement bypasses storage initialization. +- Malformed durable records with oversized numeric timestamps are logged and + discarded without interrupting replay of valid records in the same batch. +- Normalize configured Agent365 roots with the storage provider's `abspath` + rules before validation, so symlinks followed by `..` cannot redirect the + checks to a different root. Legacy-backlog warnings use the same root. +- Serialize Agent365 cache enumeration and lease transitions across Windows + threads and processes, including storage maintenance, to prevent concurrent + consumers from claiming the same record. +- Isolate Agent365 storage setup diagnostics from Azure Monitor's process-global + customer SDKStats state while preserving upstream permission enforcement. +- Log blob deletion-check filesystem errors without interrupting replay of + other valid Agent365 records in the same batch. Verify deletion with `stat()` + so Python 3.14 does not silently treat access failures as missing blobs. ### Other Changes - Update the `AZURE_OPENAI_API_VERSION` fallback version in integration tests diff --git a/README.md b/README.md index c227f95d3..0467168f3 100644 --- a/README.md +++ b/README.md @@ -138,7 +138,7 @@ string arguments/results remain supported. See the A365 guide for complete examp | `a365_exporter_timeout_ms` | `int` | `30000` | Timeout for a single A365 export operation (ms). | | `a365_max_export_batch_size` | `int` | `512` | Maximum batch size for a single A365 export operation. | | `a365_exporter_disable_offline_storage` | `bool` | `False` | Disable durable offline storage. When `True`, failed exports are not persisted to disk. | -| `a365_exporter_storage_directory` | `str` | `None` | Custom directory for durable offline storage. Restrict to the service account; payloads may contain prompts or completions when sensitive-data capture is enabled. | +| `a365_exporter_storage_directory` | `str` | `None` | Root for an `a365` cache using Azure Monitor's offline storage provider and permission handling. Use separate sibling roots for A365 and Azure Monitor; shared or overlapping cache paths are unsupported and are not rejected automatically. `None` uses its default path resolver with an Agent365 namespace. Payloads are unencrypted; protect the root and ancestors. Existing SQLite backlogs require draining before upgrade. See [storage security and migration](A365_DOCUMENTATION.md#durable-delivery-offline-storage). | > For A365 token resolver patterns, baggage, and scope classes, see the [A365 guide](https://github.com/microsoft/opentelemetry-distro-python/blob/main/A365_DOCUMENTATION.md). diff --git a/src/microsoft/opentelemetry/_distro.py b/src/microsoft/opentelemetry/_distro.py index 0b56dfddf..33f6f6cf6 100644 --- a/src/microsoft/opentelemetry/_distro.py +++ b/src/microsoft/opentelemetry/_distro.py @@ -209,10 +209,12 @@ def use_microsoft_opentelemetry(**kwargs: object) -> None: # pylint: disable=to failed export payloads are not persisted to disk and at-least-once delivery is not guaranteed. Defaults to ``False`` (storage enabled). :keyword str a365_exporter_storage_directory: - Custom directory for durable offline storage. When ``None``, a - platform default path is used. Stored payloads are unencrypted; choose - a path that only the service account can read, especially when - sensitive-data capture is enabled. + Root for an Agent365-only ``a365`` cache using Azure Monitor file + storage and permission handling. ``None`` selects Azure Monitor's + default path resolver with an Agent365 namespace. Stored payloads + are unencrypted; provision private parents for sensitive telemetry. + Configure separate sibling roots for Agent365 and Azure Monitor; + shared or overlapping cache paths are unsupported and are not rejected. :keyword bool enable_console: Enable console exporter for traces, metrics, and logs (development only). Mirrors ``ExportTarget.Console`` from the .NET distro. @@ -536,9 +538,8 @@ def _append_a365_components( coerce_storage_directory, ) - # Validate/normalize the storage directory up front so an explicitly empty - # value raises ValueError instead of being swallowed by the broad handler - # below (which would silently fall back to the platform default path). + # Defer invalid storage configuration to lazy initialization, so it disables + # offline delivery without preventing online export. storage_directory_path = coerce_storage_directory(storage_directory) _resolve_batch_options( max_queue_size=max_queue_size, diff --git a/src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter.py b/src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter.py index 6cc909ea1..7441d4dfa 100644 --- a/src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter.py +++ b/src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter.py @@ -96,7 +96,7 @@ def __init__( cluster_category: str = "prod", use_s2s_endpoint: bool = False, max_payload_bytes: int = DEFAULT_MAX_PAYLOAD_BYTES, - storage_directory: Optional[Path] = None, + storage_directory: Optional[str | Path] = None, enable_durable_delivery: bool = True, ): if token_resolver is None and contextual_token_resolver is None: diff --git a/src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter_options.py b/src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter_options.py index c6e87bd2e..13af99fef 100644 --- a/src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter_options.py +++ b/src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter_options.py @@ -61,15 +61,13 @@ def __init__( 900_000 (~100 KB headroom for estimator error and JSON envelope overhead). disable_offline_storage: When True, disables durable delivery (no disk writes or replay). Defaults to False (storage enabled). - storage_directory: Custom directory for durable offline storage. When None, a - platform default path is used. An explicitly empty or whitespace-only string - is rejected with ``ValueError``. Defaults to None. - - Raises: - ValueError: If ``storage_directory`` is an empty or whitespace-only string. + storage_directory: Root for an Agent365-only ``a365`` cache subdirectory. + None selects Azure Monitor's default path resolver with an Agent365 + namespace. Invalid configuration is logged during lazy initialization + and disables only offline delivery. Defaults to None. + Use separate sibling roots for Agent365 and Azure Monitor; + shared or overlapping cache paths are unsupported and are not rejected. """ - if storage_directory is not None and not str(storage_directory).strip(): - raise ValueError("storage_directory must be a non-empty path or None") self.cluster_category = cluster_category self.token_resolver = token_resolver self.contextual_token_resolver = contextual_token_resolver diff --git a/src/microsoft/opentelemetry/a365/core/exporters/persistent_storage.py b/src/microsoft/opentelemetry/a365/core/exporters/persistent_storage.py index 56223df88..ccf16055a 100644 --- a/src/microsoft/opentelemetry/a365/core/exporters/persistent_storage.py +++ b/src/microsoft/opentelemetry/a365/core/exporters/persistent_storage.py @@ -1,55 +1,111 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -"""Secure SQLite-backed durable queue for A365 telemetry delivery.""" +"""Agent365 record adapter for Azure Monitor's offline storage provider.""" from __future__ import annotations -import getpass import hashlib +import json import logging -import ntpath +import math import os -import sqlite3 import stat -import subprocess import sys -import tempfile import threading import time -from dataclasses import dataclass, field +from collections.abc import Callable, Generator, Iterator +from contextlib import closing, contextmanager +from dataclasses import asdict, dataclass, field from pathlib import Path +from types import FunctionType, TracebackType + +from azure.monitor.opentelemetry.exporter._storage import LocalFileBlob, LocalFileStorage, StorageExportResult +from azure.monitor.opentelemetry.exporter.export._base import _get_storage_directory from microsoft.opentelemetry.a365.core.exporters.durable_delivery import IdentityKey _logger = logging.getLogger(__name__) +_DEFAULT_CAPACITY_BYTES = 50 * 1024 * 1024 +_DEFAULT_RETENTION_SECONDS = 2 * 24 * 3600 +_SCHEMA_VERSION = 2 -_DEFAULT_CAPACITY_BYTES = 50 * 1024 * 1024 # 50 MB -_DEFAULT_RETENTION_SECONDS = 2 * 24 * 3600 # 2 days -_SCHEMA_VERSION = 2 +@contextmanager +def _storage_lock(directory: str) -> Iterator[bool]: + if sys.platform != "win32": + yield True + return + import ctypes + from ctypes import wintypes + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.CreateMutexW.argtypes = [wintypes.LPVOID, wintypes.BOOL, wintypes.LPCWSTR] + kernel32.CreateMutexW.restype = wintypes.HANDLE + kernel32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD] + kernel32.WaitForSingleObject.restype = wintypes.DWORD + kernel32.ReleaseMutex.argtypes = [wintypes.HANDLE] + kernel32.ReleaseMutex.restype = wintypes.BOOL + kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + kernel32.CloseHandle.restype = wintypes.BOOL + + identity = os.fsencode(os.path.normcase(os.path.realpath(directory))) + name = "Global\\Microsoft.Agent365.Storage." + hashlib.sha256(identity).hexdigest() + handle = kernel32.CreateMutexW(None, False, name) + if not handle: + _logger.error("Agent365 storage lock could not be created: %s", ctypes.WinError(ctypes.get_last_error())) + yield False + return + acquired = False + try: + result = kernel32.WaitForSingleObject(handle, 30_000) + # WAIT_ABANDONED also grants ownership after a consumer process exits. + acquired = result in (0, 0x80) + if not acquired: + _logger.error( + "Agent365 storage lock acquisition failed: result=%s, error=%s", result, ctypes.get_last_error() + ) + yield acquired + finally: + if acquired and not kernel32.ReleaseMutex(handle): + _logger.error("Agent365 storage lock release failed: %s", ctypes.WinError(ctypes.get_last_error())) + if not kernel32.CloseHandle(handle): + _logger.error("Agent365 storage lock close failed: %s", ctypes.WinError(ctypes.get_last_error())) + + +class _Agent365LocalFileStorage(LocalFileStorage): + def _check_and_set_folder_permissions(self) -> bool: + # Rebind only this invocation's diagnostics, never Azure Monitor's globals. + # Execute the upstream permission implementation unchanged. + check = LocalFileStorage._check_and_set_folder_permissions + namespace = dict(check.__globals__) + namespace["set_local_storage_setup_state_exception"] = lambda error: _logger.error( + "Agent365 storage setup failed: %s", error + ) + namespace["set_local_storage_setup_state_readonly"] = lambda: _logger.error( + "Agent365 storage setup failed: read-only filesystem" + ) + isolated_check: Callable[[_Agent365LocalFileStorage], bool] = FunctionType( + check.__code__, namespace, check.__name__, check.__defaults__, check.__closure__ + ) + if not isolated_check(self): # pylint: disable=not-callable + return False + with _storage_lock(self._path) as acquired: + return acquired -def _table_ddl(table_name: str) -> str: - return f""" -CREATE TABLE IF NOT EXISTS {table_name} ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - schema_version INTEGER NOT NULL, - tenant_id TEXT NOT NULL, - agent_id TEXT NOT NULL, - agentic_user_id TEXT, - use_s2s_endpoint INTEGER NOT NULL, - payload TEXT NOT NULL, - created_at REAL NOT NULL, - lease_until REAL, - retry_count INTEGER NOT NULL DEFAULT 0 -); -""" + def gets(self) -> Generator[LocalFileBlob, None, None]: + # Windows can rename the same source through concurrent open handles. + # Hold the mutex across enumeration AND the caller's lease/read, including + # enumeration by the upstream maintenance task. + with _storage_lock(self._path) as acquired: + if acquired: + yield from super().gets() @dataclass(frozen=True) class DurableRecord: - """An envelope for one telemetry payload.""" + """An identity and telemetry payload, never a replay URL or bearer token.""" schema_version: int tenant_id: str @@ -76,371 +132,220 @@ def new(key: IdentityKey, payload: str) -> DurableRecord: ) -def _resolve_default_directory() -> Path: - digest = hashlib.sha256((getpass.getuser() + sys.executable + str(Path.cwd())).encode()).hexdigest()[:16] - if sys.platform == "win32": - base = Path(os.environ.get("LOCALAPPDATA", tempfile.gettempdir())) - else: - xdg = os.environ.get("XDG_STATE_HOME") - if xdg: - base = Path(xdg) - else: - # Prefer ~/.local/state (created on demand) rather than falling back - # to a shared temp dir merely because the path does not exist yet. - # Only fall back when the home directory cannot be resolved at all. - try: - base = Path.home() / ".local" / "state" - except (RuntimeError, OSError): - base = Path(tempfile.gettempdir()) - return base / "a365-durable-queue" / digest - - -def _restrict_windows_directory_permissions(directory: Path) -> None: - """Restrict a Windows queue directory to administrators and the current user.""" - domain = os.environ.get("USERDOMAIN") - username = os.environ.get("USERNAME") - current_user = f"{domain}\\{username}" if domain and username else os.getlogin() - system_root = os.environ.get("SYSTEMROOT", r"C:\Windows") - icacls = ntpath.join(system_root, "System32", "icacls.exe") - commands = [ - [icacls, str(directory), "/reset", "/T"], - [ - icacls, - str(directory), - "/inheritance:r", - "/grant:r", - "*S-1-5-32-544:(OI)(CI)F", - f"{current_user}:(OI)(CI)F", - ], - ] - for command in commands: - result = subprocess.run( - command, - check=False, - stdout=subprocess.DEVNULL, - stderr=subprocess.PIPE, - text=True, - ) - if result.returncode == 0: - continue - detail = result.stderr.strip() or f"icacls exited with code {result.returncode}" - raise PermissionError(f"Could not restrict durable queue permissions: {detail}") - - -def _ensure_private_directory(directory: Path) -> None: - """Create the directory with mode 0700, or validate ownership if it exists.""" - if directory.exists(): - if os.name == "nt": - _restrict_windows_directory_permissions(directory) - return - # Use a non-symlink-following lstat so a symlinked queue directory - # cannot redirect telemetry writes outside a private, caller-owned - # location or defeat the ownership check via its target. - st = os.lstat(directory) - if stat.S_ISLNK(st.st_mode): - raise PermissionError(f"Durable queue directory must not be a symlink: {directory}") - if st.st_uid != getattr(os, "getuid")(): - raise PermissionError(f"Durable queue directory has unsafe ownership: {directory}") - os.chmod(directory, 0o700) - return - # Create parents first (no mode enforcement needed for intermediate dirs), - # then create the final directory with explicit mode so the kernel sets it - # before any child entry can appear. chmod follows to override a restrictive umask. - directory.mkdir(parents=True, exist_ok=True, mode=0o700) - if os.name == "nt": - _restrict_windows_directory_permissions(directory) - return - os.chmod(directory, 0o700) - - class PersistentStorage: - """Thread-safe SQLite-backed durable record queue.""" + """Adapt Agent365 replay to Azure Monitor's files, leases and permissions. + + Each blob holds one identity envelope. Configured roots get an ``a365`` + child. Use separate sibling roots for Azure Monitor and Agent365; + shared or overlapping cache paths are unsupported and are not rejected. + """ def __init__( self, - directory: Path | None = None, + directory: str | Path | None = None, capacity_bytes: int = _DEFAULT_CAPACITY_BYTES, retention_seconds: float = _DEFAULT_RETENTION_SECONDS, ) -> None: - self._directory = Path(directory) if directory is not None else _resolve_default_directory() + if directory is not None and (not isinstance(directory, (str, Path)) or not str(directory).strip()): + raise ValueError("storage_directory must be a non-empty string path, Path, or None") + if directory is None: + self.directory = Path(_get_storage_directory("agent365")) + else: + root = Path(os.path.abspath(directory)) + if sys.platform != "win32": + try: + root_stat = root.lstat() + except FileNotFoundError: + root.mkdir(mode=0o700, parents=True, exist_ok=True) + root_stat = root.lstat() + if not stat.S_ISDIR(root_stat.st_mode): + raise PermissionError(f"Agent365 storage root must be a directory, not a file or symlink: {root}") + if root_stat.st_uid not in (os.getuid(), 0): # pylint: disable=no-member + raise PermissionError(f"Agent365 storage root must be owned by the current user or root: {root}") + if root_stat.st_mode & (stat.S_IWGRP | stat.S_IWOTH): + raise PermissionError(f"Agent365 storage root must not be group- or world-writable: {root}") + self.directory = root / "a365" self._capacity_bytes = capacity_bytes self._retention_seconds = retention_seconds self._lock = threading.RLock() - - _ensure_private_directory(self._directory) - - self.database_path = self._directory / "queue.db" - # isolation_level=None → autocommit; all transactions are explicit. - self._conn = sqlite3.connect(str(self.database_path), check_same_thread=False, isolation_level=None) - if os.name != "nt": - os.chmod(self.database_path, 0o600) - - self._conn.execute("PRAGMA journal_mode=WAL") - self._initialize_schema() - - if os.name != "nt": - # WAL journal initialization creates the -wal/-shm sidecars, which can - # hold the same OTLP payloads as the DB. Lock the DB and any existing - # sidecars to owner-only now; the 0700 directory keeps future sidecars - # private on creation. - self._restrict_file_permissions() - - def _initialize_schema(self) -> None: - """Create the current schema or migrate the legacy url-backed schema.""" - self._conn.execute("BEGIN IMMEDIATE") - try: - columns = self._table_columns("durable_records") - if not columns: - self._conn.execute(_table_ddl("durable_records")) - elif "url" in columns: - self._migrate_legacy_schema() - self._conn.execute("COMMIT") - except sqlite3.Error: - self._conn.execute("ROLLBACK") - raise - - def _table_columns(self, table_name: str) -> set[str]: - rows = self._conn.execute(f"PRAGMA table_info({table_name})").fetchall() - return {row[1] for row in rows} - - def _migrate_legacy_schema(self) -> None: - """Rewrite the v1 url-backed table into the v2 identity-only schema.""" - self._conn.execute("DROP TABLE IF EXISTS durable_records_v2") - self._conn.execute(_table_ddl("durable_records_v2")) - self._conn.execute( - """ - INSERT INTO durable_records_v2 ( - id, schema_version, tenant_id, agent_id, agentic_user_id, - use_s2s_endpoint, payload, created_at, lease_until, retry_count + self._closed = False + self._next_record_id = 0 + self._claimed: dict[int, tuple[LocalFileBlob, float]] = {} + if directory is not None and (self.directory.parent / "queue.db").exists(): + _logger.warning( + "Legacy Agent365 SQLite backlog exists at %s; it is not migrated or replayed by " + "Azure Monitor file storage. Drain it with the previous SDK before upgrading.", + self.directory.parent / "queue.db", ) - SELECT - id, - ?, - tenant_id, - agent_id, - agentic_user_id, - use_s2s_endpoint, - payload, - created_at, - lease_until, - retry_count - FROM durable_records - """, - (_SCHEMA_VERSION,), + self._provider = _Agent365LocalFileStorage( + str(self.directory), + max_size=capacity_bytes, + retention_period=math.ceil(retention_seconds), + name="Agent365Storage", ) - self._conn.execute("DROP TABLE durable_records") - self._conn.execute("ALTER TABLE durable_records_v2 RENAME TO durable_records") - - def _restrict_file_permissions(self) -> None: - """Restrict the DB and any existing WAL/SHM sidecars to owner-only (0600). - - Callers gate this on POSIX; on Windows the durable queue relies on the - private (0700) directory for confidentiality. - """ - for suffix in ("", "-wal", "-shm"): - sidecar = Path(f"{self.database_path}{suffix}") - if sidecar.exists(): - os.chmod(sidecar, 0o600) - - # ------------------------------------------------------------------ - # Public API - # ------------------------------------------------------------------ + # The provider reports permission failures by disabling itself, rather + # than raising. Propagate that state to the exporter's nonfatal handler. + if not self._provider._enabled: # pylint: disable=protected-access + self._provider.close() + raise OSError("Azure Monitor local file storage could not be initialized") def store(self, record: DurableRecord) -> bool: - """Persist a record. Returns False (and logs) if storage fails.""" + """Persist one envelope using the upstream atomic blob writer.""" with self._lock: - try: - now = time.time() - expire_before = now - self._retention_seconds - - self._conn.execute("BEGIN IMMEDIATE") - # Prune expired rows first - self._conn.execute( - "DELETE FROM durable_records WHERE created_at < ?", - (expire_before,), - ) + if self._closed: + _logger.error("PersistentStorage.store failed: storage is closed") + return False + envelope = asdict(record) + del envelope["record_id"] + del envelope["lease_until"] + if len(json.dumps(envelope).encode("utf-8")) > self._capacity_bytes: + _logger.error("PersistentStorage.store failed: record exceeds storage capacity") + return False + result = self._provider.put([envelope], lease_period=0) + if result != StorageExportResult.LOCAL_FILE_BLOB_SUCCESS: + _logger.error("PersistentStorage.store failed: %s", result) + return False + return True - # Capacity check based on live (in-use) pages. SQLite does not - # shrink the file on delete; freed pages move to the freelist and - # would otherwise be counted as "used", permanently wedging the - # queue after a fill -> delete -> refill cycle. Subtracting the - # freelist gives the reclaimable, live footprint. - row = self._conn.execute( - "SELECT (page_count - freelist_count) * page_size " - "FROM pragma_page_count(), pragma_freelist_count(), pragma_page_size()" - ).fetchone() - current_bytes = row[0] if row and row[0] is not None else 0 - current_bytes = max(current_bytes, 0) - if current_bytes + len(record.payload.encode()) > self._capacity_bytes: - self._conn.execute("ROLLBACK") + def claim(self, limit: int, lease_seconds: float) -> list[DurableRecord]: + """Lease blobs with serialized Windows enumeration and lease transitions.""" + with self._lock, closing(self._provider.gets()) as blobs: + if self._closed: + _logger.error("PersistentStorage.claim failed: storage is closed") + return [] + now = time.time() + self._claimed = {key: entry for key, entry in self._claimed.items() if entry[1] > now} + records: list[DurableRecord] = [] + for blob in blobs: + if len(records) >= limit: + break + leased = blob.lease(math.ceil(lease_seconds)) + if leased is None: + continue + data = leased.get() + # Upstream returns None for both corrupt JSON and transient I/O + # failures. Do not discard a potentially valid unreadable blob. + if data is None: _logger.error( - "PersistentStorage: capacity exceeded (%d bytes used, limit %d)", - current_bytes, - self._capacity_bytes, + "PersistentStorage.claim failed to read durable record; retained: %s", leased.fullpath ) - return False - - self._conn.execute( - """INSERT INTO durable_records - (schema_version, tenant_id, agent_id, agentic_user_id, - use_s2s_endpoint, payload, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?)""", - ( - record.schema_version, - record.tenant_id, - record.agent_id, - record.agentic_user_id, - int(record.use_s2s_endpoint), - record.payload, - record.created_at, - ), - ) - self._conn.execute("COMMIT") - return True - except sqlite3.Error as exc: - _logger.error("PersistentStorage.store failed: %s", exc) + if leased.lease(0) is None: + _logger.error("PersistentStorage.release failed for unreadable blob: %s", leased.fullpath) + continue try: - self._conn.execute("ROLLBACK") - except sqlite3.Error: - pass - return False - - def claim(self, limit: int, lease_seconds: float) -> list[DurableRecord]: - """Atomically lease up to *limit* unleased records.""" - with self._lock: - try: - now = time.time() - expire_before = now - self._retention_seconds - lease_until = now + lease_seconds - - self._conn.execute("BEGIN IMMEDIATE") - - # Prune expired rows inside the same transaction - self._conn.execute( - "DELETE FROM durable_records WHERE created_at < ?", - (expire_before,), + if len(data) != 1 or not isinstance(data[0], dict): + raise ValueError("expected one Agent365 identity envelope") + record = self._decode_record(data[0]) + except (ValueError, TypeError, KeyError, OverflowError) as exc: + _logger.warning("PersistentStorage.claim dropped invalid durable record: %s", exc) + self._delete_blob(leased) + continue + if record.created_at < now - self._retention_seconds: + self._delete_blob(leased) + continue + self._next_record_id += 1 + record_id = self._next_record_id + lease_until = now + math.ceil(lease_seconds) + self._claimed[record_id] = (leased, lease_until) + records.append( + DurableRecord( + schema_version=record.schema_version, + tenant_id=record.tenant_id, + agent_id=record.agent_id, + agentic_user_id=record.agentic_user_id, + use_s2s_endpoint=record.use_s2s_endpoint, + payload=record.payload, + created_at=record.created_at, + retry_count=record.retry_count, + lease_until=lease_until, + record_id=record_id, + ) ) + return records - records: list[DurableRecord] = [] - claimed_ids: set[int] = set() - while len(records) < limit: - select_sql = """ - SELECT id, schema_version, tenant_id, agent_id, agentic_user_id, - use_s2s_endpoint, payload, created_at, lease_until, retry_count - FROM durable_records - WHERE (lease_until IS NULL OR lease_until <= ?) - AND created_at >= ? - """ - params: list[object] = [now, expire_before] - if claimed_ids: - placeholders = ", ".join("?" for _ in claimed_ids) - select_sql += f" AND id NOT IN ({placeholders})" - params.extend(claimed_ids) - select_sql += " ORDER BY created_at LIMIT 1" - - row = self._conn.execute(select_sql, tuple(params)).fetchone() - if row is None: - break - - invalid_reason = self._claim_validation_error(row) - if invalid_reason is not None: - self._conn.execute( - "DELETE FROM durable_records WHERE id = ?", - (row[0],), - ) - _logger.warning( - "PersistentStorage.claim dropped durable record %s: %s", - row[0], - invalid_reason, - ) - continue - - self._conn.execute( - "UPDATE durable_records SET lease_until = ? WHERE id = ?", - (lease_until, row[0]), - ) - claimed_ids.add(int(row[0])) - records.append( - DurableRecord( - record_id=row[0], - schema_version=row[1], - tenant_id=row[2], - agent_id=row[3], - agentic_user_id=row[4], - use_s2s_endpoint=bool(row[5]), - payload=row[6], - created_at=row[7], - lease_until=lease_until, - retry_count=row[9], - ) - ) + @staticmethod + def _decode_record(data: dict) -> DurableRecord: + for name in ("tenant_id", "agent_id", "payload"): + if not isinstance(data.get(name), str) or not data[name].strip(): + raise ValueError(f"blank or invalid {name}") + if data.get("schema_version") != _SCHEMA_VERSION: + raise ValueError("unsupported schema_version") + if not isinstance(data.get("use_s2s_endpoint"), bool): + raise ValueError("invalid use_s2s_endpoint") + if data.get("agentic_user_id") is not None and not isinstance(data["agentic_user_id"], str): + raise ValueError("invalid agentic_user_id") + created_at = data["created_at"] + if isinstance(created_at, bool) or not isinstance(created_at, (int, float)) or not math.isfinite(created_at): + raise ValueError("invalid created_at") + retry_count = data.get("retry_count", 0) + if isinstance(retry_count, bool) or not isinstance(retry_count, int) or retry_count < 0: + raise ValueError("invalid retry_count") + return DurableRecord( + schema_version=_SCHEMA_VERSION, + tenant_id=data["tenant_id"], + agent_id=data["agent_id"], + agentic_user_id=data.get("agentic_user_id"), + use_s2s_endpoint=data["use_s2s_endpoint"], + payload=data["payload"], + created_at=created_at, + retry_count=retry_count, + ) - self._conn.execute("COMMIT") - return records - except sqlite3.Error as exc: - _logger.error("PersistentStorage.claim failed: %s", exc) - try: - self._conn.execute("ROLLBACK") - except sqlite3.Error: - pass - return [] + @staticmethod + def _delete_blob(blob: LocalFileBlob) -> bool: + blob.delete() + try: + Path(blob.fullpath).stat() + except FileNotFoundError: + return True + except OSError as exc: + _logger.error("PersistentStorage.delete failed: could not check blob %s: %s", blob.fullpath, exc) + return False + _logger.error("PersistentStorage.delete failed: blob was not removed: %s", blob.fullpath) + return False def delete(self, record_id: int) -> bool: - """Delete a record by id. Returns False if not found or on error.""" - with self._lock: - try: - self._conn.execute("BEGIN IMMEDIATE") - cur = self._conn.execute("DELETE FROM durable_records WHERE id = ?", (record_id,)) - found = cur.rowcount > 0 - self._conn.execute("COMMIT") - return found - except sqlite3.Error as exc: - _logger.error("PersistentStorage.delete failed: %s", exc) - try: - self._conn.execute("ROLLBACK") - except sqlite3.Error: - pass + """Delete a claimed blob, reporting upstream's otherwise-silent failure.""" + with self._lock, _storage_lock(str(self.directory)) as acquired: + if not acquired: return False - - @staticmethod - def _claim_validation_error(row: tuple[object, ...]) -> str | None: - if row[1] != _SCHEMA_VERSION: - return f"unsupported schema_version={row[1]!r}" - if not isinstance(row[2], str) or not row[2].strip(): - return "blank tenant_id" - if not isinstance(row[3], str) or not row[3].strip(): - return "blank agent_id" - if not isinstance(row[6], str) or not row[6].strip(): - return "blank payload" - return None + entry = self._claimed.get(record_id) + if entry is None: + return False + if not self._delete_blob(entry[0]): + return False + del self._claimed[record_id] + return True def release(self, record_id: int) -> bool: - """Release a lease so the record becomes claimable again.""" - with self._lock: - try: - self._conn.execute("BEGIN IMMEDIATE") - cur = self._conn.execute( - "UPDATE durable_records SET lease_until = NULL WHERE id = ?", - (record_id,), - ) - found = cur.rowcount > 0 - self._conn.execute("COMMIT") - return found - except sqlite3.Error as exc: - _logger.error("PersistentStorage.release failed: %s", exc) - try: - self._conn.execute("ROLLBACK") - except sqlite3.Error: - pass + """Expire a claimed blob's lease so another replay pass can claim it.""" + with self._lock, _storage_lock(str(self.directory)) as acquired: + if not acquired: + return False + entry = self._claimed.get(record_id) + if entry is None: return False + if entry[0].lease(0) is None: + _logger.error("PersistentStorage.release failed: could not release blob %s", entry[0].fullpath) + return False + del self._claimed[record_id] + return True def close(self) -> None: - """Close the underlying SQLite connection.""" + """Stop the upstream maintenance task.""" with self._lock: - try: - self._conn.close() - except sqlite3.Error as exc: - _logger.error("PersistentStorage.close failed: %s", exc) + if not self._closed: + self._provider.close() + self._claimed.clear() + self._closed = True + + def __enter__(self) -> PersistentStorage: + return self + + def __exit__( + self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None + ) -> None: + self.close() __all__ = ["DurableRecord", "PersistentStorage"] diff --git a/src/microsoft/opentelemetry/a365/core/exporters/replay_coordinator.py b/src/microsoft/opentelemetry/a365/core/exporters/replay_coordinator.py index 57f3198b7..220866c72 100644 --- a/src/microsoft/opentelemetry/a365/core/exporters/replay_coordinator.py +++ b/src/microsoft/opentelemetry/a365/core/exporters/replay_coordinator.py @@ -25,7 +25,7 @@ _MAX_RECORDS_PER_PASS = 10 # A pass leases all records before sending them sequentially. Keep every lease -# valid for the worst-case full pass so another process sharing the SQLite +# valid for the worst-case full pass so another process sharing the file # queue cannot reclaim a record while this coordinator is still sending it. _LEASE_SECONDS = (_MAX_RECORDS_PER_PASS * A365_HTTP_TIMEOUT_SECONDS) + 5.0 diff --git a/src/microsoft/opentelemetry/a365/core/exporters/utils.py b/src/microsoft/opentelemetry/a365/core/exporters/utils.py index 960effb15..19dd6d57f 100644 --- a/src/microsoft/opentelemetry/a365/core/exporters/utils.py +++ b/src/microsoft/opentelemetry/a365/core/exporters/utils.py @@ -635,22 +635,11 @@ def _env_bool(name: str, default: bool = False) -> bool: return val in ("true", "1", "yes", "on") -def coerce_storage_directory(value: Optional[str | Path]) -> Optional[Path]: - """Normalize a ``storage_directory`` option to a ``Path`` or ``None``. - - ``None`` selects the platform default path. An explicitly empty or - whitespace-only string is rejected with :class:`ValueError` so it is never - silently treated as "use the default". - """ - if value is None: - return None - if isinstance(value, Path): - return value - if isinstance(value, str): - if not value.strip(): - raise ValueError("storage_directory must be a non-empty path or None") +def coerce_storage_directory(value: Optional[str | Path]) -> Optional[str | Path]: + """Convert nonblank paths, deferring validation to nonfatal storage initialization.""" + if isinstance(value, str) and value.strip(): return Path(value) - raise ValueError("storage_directory must be a string path, Path, or None") + return value def create_a365_components( @@ -669,8 +658,11 @@ def create_a365_components( ``token_resolver`` when set. :param disable_offline_storage: When True, disables durable delivery (no disk writes or replay). Defaults to False. - :param storage_directory: Custom directory for durable offline storage. When None, a - platform default path is used. Defaults to None. + :param storage_directory: Root for an Agent365-only ``a365`` cache subdirectory. + When None, Azure Monitor's default path resolver selects an Agent365 namespace. + Invalid configuration disables only offline delivery, with an error log. + Use separate sibling roots for Agent365 and Azure Monitor; + shared or overlapping cache paths are unsupported and are not rejected. All other configuration is read from environment variables: - ``ENABLE_A365_OBSERVABILITY_EXPORTER`` -- must be true for the HTTP exporter diff --git a/tests/a365/test_durable_restart.py b/tests/a365/test_durable_restart.py index 5e6ae2d05..7a078bf72 100644 --- a/tests/a365/test_durable_restart.py +++ b/tests/a365/test_durable_restart.py @@ -91,6 +91,7 @@ def _queue_size(storage) -> int: def test_restart_replays_persisted_record_with_fresh_token(tmp_path): storage_dir = tmp_path / "queue" + storage_dir.mkdir() # --- Instance A: a retryable send persists the payload, then shuts down. --- exporter_a = _Agent365Exporter( @@ -137,6 +138,7 @@ def fake_post_once(url, body, headers): def test_restart_replays_using_current_exporter_endpoint_settings(tmp_path): storage_dir = tmp_path / "queue" + storage_dir.mkdir() exporter_a = _Agent365Exporter( token_resolver=lambda a, t: "stale-token", @@ -179,6 +181,7 @@ def fake_post_once(url, body, headers): def test_restart_replays_record_persisted_after_token_failure(tmp_path): storage_dir = tmp_path / "queue" + storage_dir.mkdir() # --- Instance A: token resolution fails, so the payload is persisted. --- exporter_a = _Agent365Exporter( @@ -220,6 +223,7 @@ def test_shutdown_blocks_until_active_replay_send_completes_then_removes_record( for the next restart instead of cleanly removed. """ storage_dir = tmp_path / "queue" + storage_dir.mkdir() exporter_a = _Agent365Exporter( token_resolver=lambda a, t: "token", diff --git a/tests/a365/test_exporter.py b/tests/a365/test_exporter.py index 18f2f913d..4f028d7d0 100644 --- a/tests/a365/test_exporter.py +++ b/tests/a365/test_exporter.py @@ -384,6 +384,7 @@ class TestAgent365ExporterActiveReplayShutdown(unittest.TestCase): """ def _make_durable_exporter_with_pending_record(self, storage_dir): + storage_dir.mkdir(parents=True, exist_ok=True) exporter = _Agent365Exporter( token_resolver=lambda a, t: "token", storage_directory=storage_dir, @@ -817,21 +818,109 @@ def test_plaintext_replay_endpoint_raises_before_resolving_token_or_sending(self class TestAgent365ExporterStorageDirectory(unittest.TestCase): """End-to-end behavior of the storage_directory option and no-leak checks.""" + @patch.dict(os.environ, {}, clear=True) + def test_invalid_roots_disable_only_offline_delivery(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + file_root = root / "file" + file_root.touch() + for storage_dir in ("", " ", 123, file_root): + with self.subTest(storage_directory=storage_dir): + exporter = _Agent365Exporter( + token_resolver=lambda a, t: "token", + storage_directory=storage_dir, + enable_durable_delivery=True, + ) + exporter._post_once = MagicMock(return_value=_delivered()) # type: ignore[method-assign] + try: + with self.assertLogs(exporter_module.logger, level="ERROR") as logs: + self.assertIs(exporter.export([_make_span()]), SpanExportResult.SUCCESS) + self.assertIn("Durable delivery disabled", "\n".join(logs.output)) + self.assertFalse(exporter._enable_durable_delivery) + self.assertIsNone(exporter._storage) + self.assertIsNone(exporter._replay) + with patch.object(exporter_module, "PersistentStorage") as storage: + self.assertIs(exporter.export([_make_span()]), SpanExportResult.SUCCESS) + storage.assert_not_called() + exporter._post_once.return_value = _retryable(30) + self.assertIs(exporter.export([_make_span()]), SpanExportResult.FAILURE) + finally: + exporter.shutdown() + self.assertEqual([p.name for p in root.iterdir()], ["file"]) + + @patch.dict(os.environ, {}, clear=True) + @unittest.skipIf(os.name == "nt", "Unix configured-root validation") + def test_symlinked_root_disables_only_offline_delivery(self): + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "target" + target.mkdir(mode=0o750) + root = Path(tmp) / "root" + root.symlink_to(target, target_is_directory=True) + before = target.stat().st_mode + exporter = _Agent365Exporter(token_resolver=lambda a, t: "token", storage_directory=root) + exporter._post_once = MagicMock(return_value=_delivered()) # type: ignore[method-assign] + try: + with self.assertLogs(exporter_module.logger, level="ERROR") as logs: + self.assertIs(exporter.export([_make_span()]), SpanExportResult.SUCCESS) + self.assertIn("storage root", "\n".join(logs.output)) + self.assertFalse(exporter._enable_durable_delivery) + self.assertIsNone(exporter._storage) + self.assertIsNone(exporter._replay) + finally: + exporter.shutdown() + self.assertEqual(target.stat().st_mode, before) + self.assertEqual(list(target.iterdir()), []) + + @patch.dict(os.environ, {}, clear=True) + def test_storage_initialization_failure_keeps_online_export_running(self): + exporter = _Agent365Exporter(token_resolver=lambda a, t: "token") + exporter._post_once = MagicMock(return_value=_delivered()) # type: ignore[method-assign] + try: + with ( + patch.object(exporter_module, "PersistentStorage", side_effect=OSError("disk unavailable")) as storage, + self.assertLogs(exporter_module.logger, level="ERROR") as logs, + ): + self.assertIs(exporter.export([_make_span()]), SpanExportResult.SUCCESS) + self.assertIs(exporter.export([_make_span()]), SpanExportResult.SUCCESS) + storage.assert_called_once() + self.assertIn("disk unavailable", "\n".join(logs.output)) + self.assertIsNone(exporter._storage) + self.assertIsNone(exporter._replay) + finally: + exporter.shutdown() + @patch.dict(os.environ, {}, clear=True) def test_storage_directory_is_honored_end_to_end(self): with tempfile.TemporaryDirectory() as tmp: storage_dir = Path(tmp) / "queue" + storage_dir.mkdir() exporter = _Agent365Exporter( token_resolver=lambda a, t: "token", storage_directory=storage_dir, enable_durable_delivery=True, ) exporter._ensure_durable_initialized() - # The durable queue database lives under the requested directory. - self.assertEqual(exporter._storage.database_path.parent, storage_dir) - self.assertTrue((storage_dir / "queue.db").exists()) + self.assertEqual(exporter._storage.directory, storage_dir / "a365") + self.assertTrue((storage_dir / "a365").is_dir()) exporter.shutdown() + def test_explicit_disablement_bypasses_invalid_storage_configuration(self): + for root in (None, "", " ", "relative-root"): + with self.subTest(root=root): + exporter = _Agent365Exporter( + token_resolver=lambda a, t: "token", + storage_directory=root, + enable_durable_delivery=False, + ) + exporter._post_once = MagicMock(return_value=_delivered()) # type: ignore[method-assign] + try: + with patch.object(exporter_module, "PersistentStorage") as storage: + self.assertIs(exporter.export([_make_span()]), SpanExportResult.SUCCESS) + storage.assert_not_called() + self.assertIsNone(exporter._replay) + finally: + exporter.shutdown() + @patch.dict(os.environ, {}, clear=True) def test_no_disk_writes_when_durable_delivery_disabled(self): with tempfile.TemporaryDirectory() as tmp: diff --git a/tests/a365/test_handler.py b/tests/a365/test_handler.py index 40f293973..08b741854 100644 --- a/tests/a365/test_handler.py +++ b/tests/a365/test_handler.py @@ -122,17 +122,17 @@ def test_storage_directory_can_be_set(self): opts = Agent365ExporterOptions(storage_directory="C:\\telemetry") self.assertEqual(opts.storage_directory, "C:\\telemetry") - def test_empty_storage_directory_raises_value_error(self): + def test_empty_storage_directory_is_deferred_until_storage_initialization(self): from microsoft.opentelemetry.a365.core.exporters.agent365_exporter_options import Agent365ExporterOptions - with self.assertRaises(ValueError): - Agent365ExporterOptions(storage_directory="") + opts = Agent365ExporterOptions(storage_directory="") + self.assertEqual(opts.storage_directory, "") - def test_whitespace_storage_directory_raises_value_error(self): + def test_explicit_disablement_accepts_unused_blank_storage_directory(self): from microsoft.opentelemetry.a365.core.exporters.agent365_exporter_options import Agent365ExporterOptions - with self.assertRaises(ValueError): - Agent365ExporterOptions(storage_directory=" ") + opts = Agent365ExporterOptions(disable_offline_storage=True, storage_directory=" ") + self.assertTrue(opts.disable_offline_storage) class TestCreateA365ComponentsOfflineStorage(unittest.TestCase): @@ -173,11 +173,20 @@ def test_storage_directory_none_by_default(self): self.assertIsNone(kwargs["storage_directory"]) @patch.dict(os.environ, {"ENABLE_A365_OBSERVABILITY_EXPORTER": "true"}) - def test_empty_storage_directory_rejected_end_to_end(self): - """An explicitly empty storage directory must raise ValueError rather - than silently defaulting to the platform path.""" - with self.assertRaises(ValueError): - create_a365_components(storage_directory="") + def test_empty_storage_directory_disables_only_buffering_end_to_end(self): + from microsoft.opentelemetry.a365.core.exporters.agent365_exporter import logger + + handlers = create_a365_components(token_resolver=lambda a, t: "token", storage_directory="") + exporter = handlers.span_processors[0].span_exporter + try: + with self.assertLogs(logger, level="ERROR") as logs: + exporter._ensure_durable_initialized() + self.assertIn("Durable delivery disabled", "\n".join(logs.output)) + self.assertFalse(exporter._enable_durable_delivery) + self.assertIsNone(exporter._storage) + finally: + for processor in handlers.span_processors: + processor.shutdown() if __name__ == "__main__": diff --git a/tests/a365/test_persistent_storage.py b/tests/a365/test_persistent_storage.py index 0a0aa1d31..4a281f60b 100644 --- a/tests/a365/test_persistent_storage.py +++ b/tests/a365/test_persistent_storage.py @@ -1,579 +1,624 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -"""Tests for PersistentStorage (Task 2).""" +"""Agent365's adapter to Azure Monitor offline storage.""" from __future__ import annotations -import inspect +import errno +import json +import multiprocessing import os -import sqlite3 import stat -import subprocess +import threading import time -from unittest.mock import call, patch +from concurrent.futures import ThreadPoolExecutor +from dataclasses import asdict, replace +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import patch import pytest +from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage, StorageExportResult +from azure.monitor.opentelemetry.exporter.statsbeat.customer import _state as storage_state +from azure.monitor.opentelemetry.exporter.statsbeat.customer._utils import track_dropped_items_from_storage from microsoft.opentelemetry.a365.core.exporters.durable_delivery import IdentityKey -from microsoft.opentelemetry.a365.core.exporters.persistent_storage import ( - DurableRecord, - PersistentStorage, -) - -KEY = IdentityKey( - tenant_id="t1", - agent_id="a1", - agentic_user_id=None, - use_s2s_endpoint=False, -) -LEGACY_URL = "https://example.test" - - -def _new_record(payload: str, key: IdentityKey = KEY) -> DurableRecord: - return DurableRecord.new(key, payload) - - -def _schema_version_for(columns: set[str]) -> int: - return 1 if "url" in columns else 2 - - -def _table_columns(storage: PersistentStorage) -> set[str]: - rows = storage._conn.execute("PRAGMA table_info(durable_records)").fetchall() - return {row[1] for row in rows} - - -def insert_raw_record( - storage: PersistentStorage, - *, - created_at: float | None = None, - **overrides, -) -> int: - columns = _table_columns(storage) - row = { - "schema_version": _schema_version_for(columns), - "tenant_id": KEY.tenant_id, - "agent_id": KEY.agent_id, - "agentic_user_id": KEY.agentic_user_id, - "use_s2s_endpoint": int(KEY.use_s2s_endpoint), - "payload": '{"raw":true}', - "created_at": time.time() - 1.0 if created_at is None else created_at, - "lease_until": None, - "retry_count": 0, - } - if "url" in columns: - row["url"] = LEGACY_URL - row.update(overrides) - names = list(row) - placeholders = ", ".join("?" for _ in names) - sql = f"INSERT INTO durable_records ({', '.join(names)}) VALUES ({placeholders})" - with storage._lock: - storage._conn.execute("BEGIN IMMEDIATE") - cur = storage._conn.execute(sql, tuple(row[name] for name in names)) - storage._conn.execute("COMMIT") - return int(cur.lastrowid) - - -def raw_record_exists(storage: PersistentStorage, record_id: int) -> bool: - with storage._lock: - row = storage._conn.execute( - "SELECT COUNT(*) FROM durable_records WHERE id = ?", - (record_id,), - ).fetchone() - return bool(row[0]) - - -def _create_legacy_v1_database(database_path): - database_path.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(str(database_path)) - try: - conn.execute(""" - CREATE TABLE durable_records ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - schema_version INTEGER NOT NULL, - tenant_id TEXT NOT NULL, - agent_id TEXT NOT NULL, - agentic_user_id TEXT, - use_s2s_endpoint INTEGER NOT NULL, - url TEXT NOT NULL, - payload TEXT NOT NULL, - created_at REAL NOT NULL, - lease_until REAL, - retry_count INTEGER NOT NULL DEFAULT 0 - ) - """) - conn.execute( - """ - INSERT INTO durable_records ( - schema_version, tenant_id, agent_id, agentic_user_id, - use_s2s_endpoint, url, payload, created_at, lease_until, retry_count - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - 1, - "legacy-tenant", - "legacy-agent", - "legacy-user", - 1, - "https://legacy.example.test", - '{"legacy":true}', - time.time() - 5.0, - None, - 3, - ), - ) - conn.commit() - finally: - conn.close() - - -# --------------------------------------------------------------------------- -# Transaction safety: isolation_level=None (autocommit mode) -# --------------------------------------------------------------------------- - - -def test_connection_is_in_autocommit_mode(tmp_path): - """The connection must use isolation_level=None so transactions are explicit.""" - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=3600) - assert storage._conn.isolation_level is None - storage.close() - - -# --------------------------------------------------------------------------- -# claim() prunes expired rows during the transaction -# --------------------------------------------------------------------------- - - -def test_claim_prunes_expired_rows(tmp_path): - """Expired records stored before a claim call must be deleted by claim itself.""" - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=0) - insert_raw_record(storage, created_at=0.0, payload='{"stale":true}') - - # claim must prune the expired row and return nothing - claimed = storage.claim(limit=10, lease_seconds=30) - assert not claimed - - # Row must actually be gone - with storage._lock: - row = storage._conn.execute("SELECT COUNT(*) FROM durable_records").fetchone() - assert row[0] == 0 - storage.close() - - -def test_initialization_migrates_v1_schema_and_preserves_records(tmp_path): - queue_dir = tmp_path / "queue" - _create_legacy_v1_database(queue_dir / "queue.db") - - storage = PersistentStorage(queue_dir, capacity_bytes=1024 * 1024, retention_seconds=3600) - columns = _table_columns(storage) - - assert "url" not in columns - claimed = storage.claim(limit=10, lease_seconds=30) - assert len(claimed) == 1 - assert claimed[0].schema_version == 2 - assert claimed[0].tenant_id == "legacy-tenant" - assert claimed[0].agent_id == "legacy-agent" - assert claimed[0].agentic_user_id == "legacy-user" - assert claimed[0].use_s2s_endpoint is True - assert claimed[0].payload == '{"legacy":true}' - assert claimed[0].retry_count == 3 - storage.close() - - -@pytest.mark.parametrize( - ("column", "value"), - [ - ("schema_version", 999), - ("tenant_id", ""), - ("agent_id", ""), - ("payload", ""), - ], -) -def test_claim_deletes_invalid_records_and_continues_to_valid_later_record(tmp_path, column, value): - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=3600) - invalid_id = insert_raw_record(storage, **{column: value}) - valid = _new_record('{"resourceSpans":[]}') - - assert storage.store(valid) - - claimed = storage.claim(limit=1, lease_seconds=30) - - assert [record.payload for record in claimed] == [valid.payload] - assert raw_record_exists(storage, invalid_id) is False - storage.close() - - -# --------------------------------------------------------------------------- -# Round-trip -# --------------------------------------------------------------------------- - - -def test_store_claim_delete_round_trip(tmp_path): - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=3600) - record = _new_record('{"resourceSpans":[]}') - assert storage.store(record) - claimed = storage.claim(limit=10, lease_seconds=30) - assert [item.payload for item in claimed] == [record.payload] - assert storage.delete(claimed[0].record_id) - assert not storage.claim(limit=10, lease_seconds=30) - storage.close() - - -# --------------------------------------------------------------------------- -# Lease release -# --------------------------------------------------------------------------- - - -def test_release_makes_record_claimable_again(tmp_path): - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=3600) - record = _new_record('{"payload":1}') - assert storage.store(record) - - claimed = storage.claim(limit=10, lease_seconds=30) - assert len(claimed) == 1 - - # While leased, claim returns nothing - assert not storage.claim(limit=10, lease_seconds=30) - - # After release the record is available again - assert storage.release(claimed[0].record_id) - reclaimed = storage.claim(limit=10, lease_seconds=30) - assert len(reclaimed) == 1 - assert reclaimed[0].payload == record.payload - storage.close() +from microsoft.opentelemetry.a365.core.exporters.persistent_storage import DurableRecord, PersistentStorage +KEY = IdentityKey("tenant", "agent", "user", False) +MODULE = "microsoft.opentelemetry.a365.core.exporters.persistent_storage" -# --------------------------------------------------------------------------- -# Expired-record cleanup -# --------------------------------------------------------------------------- +def _record(payload='{"resourceSpans":[]}'): + return DurableRecord.new(KEY, payload) -def test_expired_records_are_cleaned_up(tmp_path): - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=0) - # Insert the to-be-expired row directly with an unambiguously-past - # created_at (matching the insert_raw_record convention used by - # test_claim_prunes_expired_rows above) instead of two back-to-back - # DurableRecord.new()/store() calls. time.time() has ~15ms resolution on - # Windows, so two calls made microseconds apart can return the identical - # value; with retention_seconds=0 that made "created_at < expire_before" - # a tie (record.created_at == the claim's own now-clock read), leaving - # the "expired" record un-pruned and intermittently failing this test. - insert_raw_record(storage, created_at=0.0, payload='{"payload":2}') - # Store a second, live record to trigger the cleanup path alongside it. - record2 = _new_record('{"payload":3}') - assert storage.store(record2) - - # Expired records must not be returned by claim - claimed = storage.claim(limit=10, lease_seconds=30) - for item in claimed: - assert item.payload != '{"payload":2}' - storage.close() - - -# --------------------------------------------------------------------------- -# Capacity rejection -# --------------------------------------------------------------------------- - - -def test_store_rejects_when_capacity_exceeded(tmp_path): - storage = PersistentStorage(tmp_path, capacity_bytes=1, retention_seconds=3600) - record = _new_record("x" * 100) - # Must return False and not raise - result = storage.store(record) - assert result is False - storage.close() - - -def test_store_reclaims_capacity_after_fill_delete_refill(tmp_path): - """Capacity accounting must use live pages, not the file high-water mark. - - Regression: with ``page_count * page_size`` accounting, filling the queue to - its cap and then claiming/deleting every record left the freed (freelist) - pages counted as "used", because SQLite does not shrink the file on delete. - The queue was therefore permanently wedged and rejected all new records. - Live-page accounting — ``(page_count - freelist_count) * page_size`` — must - reclaim the freed space so new records can be stored again. - """ - storage = PersistentStorage(tmp_path, capacity_bytes=64 * 1024, retention_seconds=3600) - payload = "x" * 4000 - - # Fill until the capacity cap rejects a store. - stored = 0 - while stored < 500 and storage.store(_new_record(payload)): - stored += 1 - # A store was actually rejected (we reached the cap, not the loop guard). - assert 0 < stored < 500 - - # Claim and delete every stored record. - while True: - claimed = storage.claim(limit=100, lease_seconds=300) - if not claimed: - break - for rec in claimed: - assert storage.delete(rec.record_id) - - # The freed space must be reclaimed so new records can be stored again. - assert storage.store(_new_record(payload)) is True - storage.close() - - -# --------------------------------------------------------------------------- -# POSIX permissions -# --------------------------------------------------------------------------- - - -@pytest.mark.skipif(os.name == "nt", reason="POSIX permissions") -def test_storage_permissions_are_private(tmp_path): - storage = PersistentStorage(tmp_path / "queue") - assert stat.S_IMODE((tmp_path / "queue").stat().st_mode) == 0o700 - assert stat.S_IMODE(storage.database_path.stat().st_mode) == 0o600 - storage.close() - - -@pytest.mark.skipif(os.name == "nt", reason="POSIX permissions") -def test_wal_and_shm_sidecars_are_private(tmp_path): - """The DB and its WAL/SHM sidecars must be mode 0600 after journal init. - - WAL mode creates ``queue.db-wal`` and ``queue.db-shm`` sidecars that would - otherwise inherit the process umask; they can contain the same OTLP payloads - as the DB and must be locked to the owner. - """ - storage = PersistentStorage(tmp_path / "queue") +def test_uses_real_azure_monitor_provider_in_isolated_namespace(tmp_path): + storage = PersistentStorage(tmp_path) try: - for name in ("queue.db", "queue.db-wal", "queue.db-shm"): - sidecar = tmp_path / "queue" / name - assert sidecar.exists(), name - assert stat.S_IMODE(sidecar.stat().st_mode) == 0o600, name + assert isinstance(storage._provider, LocalFileStorage) + assert storage.directory == tmp_path / "a365" + assert storage.store(_record()) + files = list(storage.directory.glob("*.blob")) + assert len(files) == 1 + envelope = json.loads(files[0].read_text(encoding="utf-8")) + assert envelope["payload"] == _record().payload + assert envelope["tenant_id"] == "tenant" + assert "url" not in envelope + assert not (tmp_path / "queue.db").exists() finally: storage.close() -def test_restrict_file_permissions_locks_db_and_sidecars(tmp_path, monkeypatch): - """The helper chmods the DB and any existing WAL/SHM sidecars to 0600. - - Runs on Windows (mocks only ``os.chmod``, not ``os.name``, so ``pathlib`` is - unaffected) to make the sidecar hardening verifiable off-POSIX; the real - end-to-end modes are checked by the POSIX-only test above. - """ - import microsoft.opentelemetry.a365.core.exporters.persistent_storage as _mod - - storage = PersistentStorage(tmp_path / "queue") +def test_sibling_roots_keep_exporter_replay_and_deletion_separate(tmp_path): + monitor = LocalFileStorage(str(tmp_path / "AzureMonitor")) try: - # Journal init created queue.db plus its -wal/-shm sidecars. - for name in ("queue.db", "queue.db-wal", "queue.db-shm"): - assert (tmp_path / "queue" / name).exists(), name - - recorded: dict[str, int] = {} - monkeypatch.setattr(_mod.os, "chmod", lambda p, m: recorded.__setitem__(os.path.basename(str(p)), m)) - storage._restrict_file_permissions() - - assert recorded.get("queue.db") == 0o600 - assert recorded.get("queue.db-wal") == 0o600 - assert recorded.get("queue.db-shm") == 0o600 + assert monitor.put([{"azure_monitor": True}], lease_period=0) == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS + with PersistentStorage(tmp_path / "Agent365") as storage: + assert not storage.claim(10, 30) + assert storage.store(_record()) + claimed = storage.claim(10, 30) + assert [record.payload for record in claimed] == [_record().payload] + assert storage.delete(claimed[0].record_id) + monitor_blobs = list(monitor.gets()) + assert [blob.get() for blob in monitor_blobs] == [({"azure_monitor": True},)] + assert storage.store(_record("retained")) + leased = monitor_blobs[0].lease(30) + assert leased is not None + leased.delete() + assert not list(monitor.gets()) + assert [record.payload for record in storage.claim(10, 30)] == ["retained"] finally: - storage.close() - - -# --------------------------------------------------------------------------- -# Windows permissions -# --------------------------------------------------------------------------- - - -def test_windows_directory_permissions_restrict_access_to_admins_and_current_user(tmp_path, monkeypatch): - import microsoft.opentelemetry.a365.core.exporters.persistent_storage as _mod - - monkeypatch.setenv("SYSTEMROOT", r"C:\Windows") - monkeypatch.setenv("USERDOMAIN", "CONTOSO") - monkeypatch.setenv("USERNAME", "alice") - - with patch("subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run: - _mod._restrict_windows_directory_permissions(tmp_path) - - common_options = { - "check": False, - "stdout": subprocess.DEVNULL, - "stderr": subprocess.PIPE, - "text": True, - } - assert run.call_args_list == [ - call( - [r"C:\Windows\System32\icacls.exe", str(tmp_path), "/reset", "/T"], - **common_options, - ), - call( - [ - r"C:\Windows\System32\icacls.exe", - str(tmp_path), - "/inheritance:r", - "/grant:r", - "*S-1-5-32-544:(OI)(CI)F", - r"CONTOSO\alice:(OI)(CI)F", - ], - **common_options, - ), - ] - - -def test_windows_directory_permissions_fail_closed_when_icacls_fails(tmp_path, monkeypatch): - import microsoft.opentelemetry.a365.core.exporters.persistent_storage as _mod - - monkeypatch.setenv("USERDOMAIN", "CONTOSO") - monkeypatch.setenv("USERNAME", "alice") - - results = [ - subprocess.CompletedProcess([], 0), - subprocess.CompletedProcess([], 5, stderr="Access is denied."), - ] - with patch("subprocess.run", side_effect=results): - with pytest.raises(PermissionError, match="Access is denied"): - _mod._restrict_windows_directory_permissions(tmp_path) - - -# --------------------------------------------------------------------------- -# Unsafe ownership rejected -# --------------------------------------------------------------------------- - - -@pytest.mark.skipif(os.name == "nt", reason="POSIX ownership check") -def test_rejects_directory_owned_by_another_uid(tmp_path): - import types - import microsoft.opentelemetry.a365.core.exporters.persistent_storage as _mod - - foreign_uid = getattr(os, "getuid")() + 1 - - real_stat = os.lstat(tmp_path) - mock_result = types.SimpleNamespace( - st_uid=foreign_uid, - st_mode=real_stat.st_mode, - st_size=real_stat.st_size, - st_mtime=real_stat.st_mtime, - ) - - target_dir = tmp_path / "queue_foreign" - target_dir.mkdir(parents=True, exist_ok=True) - - # Ownership is validated with a non-symlink-following lstat. - with patch.object(_mod.os, "lstat", return_value=mock_result): - with pytest.raises(PermissionError, match="unsafe ownership"): - PersistentStorage(target_dir) - - -@pytest.mark.skipif(os.name == "nt", reason="POSIX symlink rejection") -def test_rejects_symlinked_directory(tmp_path): - """A symlinked queue directory must be rejected, not silently followed.""" - real_dir = tmp_path / "real" - real_dir.mkdir() - link_dir = tmp_path / "link" - os.symlink(real_dir, link_dir, target_is_directory=True) - - with pytest.raises(PermissionError, match="symlink"): - PersistentStorage(link_dir) - - -def test_default_directory_prefers_local_state_even_if_missing(tmp_path): - """On POSIX without XDG_STATE_HOME, prefer ~/.local/state (to be created) - rather than falling back to the temp dir merely because it does not exist.""" - import microsoft.opentelemetry.a365.core.exporters.persistent_storage as _mod - - fake_home = tmp_path / "home" - fake_home.mkdir() - local_state = fake_home / ".local" / "state" - assert not local_state.exists() - - env = {k: v for k, v in os.environ.items() if k != "XDG_STATE_HOME"} - with ( - patch.object(_mod.sys, "platform", "linux"), - patch.dict(os.environ, env, clear=True), - patch.object(_mod.Path, "home", return_value=fake_home), - ): - resolved = _mod._resolve_default_directory() + monitor.close() - # The resolved base must be under ~/.local/state, not the temp directory. - assert str(resolved).startswith(str(local_state)) - -def test_default_directory_falls_back_to_tmp_when_home_unusable(tmp_path): - """When the home directory cannot be resolved, fall back to the temp dir.""" - import microsoft.opentelemetry.a365.core.exporters.persistent_storage as _mod - - env = {k: v for k, v in os.environ.items() if k != "XDG_STATE_HOME"} +@pytest.mark.parametrize("full_cache", ["agent365", "azure_monitor"]) +def test_sibling_roots_keep_exporter_capacity_independent(full_cache, tmp_path): + monitor_directory = tmp_path / "AzureMonitor" + monitor = LocalFileStorage(str(monitor_directory), max_size=1024) + try: + with PersistentStorage(tmp_path / "Agent365", capacity_bytes=1024) as storage: + monitor_payload = {"azure_monitor": "x" * 512} + record = _record("x" * 512) + for _ in range(10): + if full_cache == "agent365": + stored = storage.store(record) + else: + stored = ( + monitor.put([monitor_payload], lease_period=0) == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS + ) + if not stored: + break + else: + pytest.fail("cache did not reach its configured capacity") + + full_directory = storage.directory if full_cache == "agent365" else monitor_directory + assert sum(blob.stat().st_size for blob in full_directory.glob("*.blob")) >= 1024 + if full_cache == "agent365": + assert len(storage.claim(10, 30)) > 0 + assert monitor.put([monitor_payload], lease_period=0) == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS + assert [blob.get() for blob in monitor.gets()] == [(monitor_payload,)] + else: + assert len(list(monitor.gets())) > 0 + assert storage.store(record) + assert [claimed.payload for claimed in storage.claim(10, 30)] == [record.payload] + finally: + monitor.close() + + +def test_restart_replays_same_identity_and_payload(tmp_path): + original = _record() + with PersistentStorage(tmp_path) as storage: + assert storage.store(original) + with PersistentStorage(tmp_path) as restarted: + records = restarted.claim(10, 30) + assert len(records) == 1 + assert replace(records[0], record_id=None, lease_until=None) == original + assert restarted.delete(records[0].record_id) + assert not restarted.claim(10, 30) + + +def test_leases_coordinate_two_instances_and_release(tmp_path): + with PersistentStorage(tmp_path) as first, PersistentStorage(tmp_path) as second: + assert first.store(_record()) + claimed = first.claim(10, 30) + assert len(claimed) == 1 + assert not second.claim(10, 30) + assert first.release(claimed[0].record_id) + reclaimed = second.claim(10, 30) + assert len(reclaimed) == 1 + assert second.delete(reclaimed[0].record_id) + assert not first.claim(10, 30) + + +def _claim_concurrently(directory, barrier): + with PersistentStorage(directory) as storage: + barrier.wait(timeout=60) + return [record.payload for record in storage.claim(1000, 120)] + + +def _claim_in_process(directory, barrier, results): + results.put(_claim_concurrently(directory, barrier)) + + +@pytest.mark.parametrize("record_count", [1, 200]) +def test_concurrent_threads_claim_each_blob_once(tmp_path, record_count, caplog): + expected = [str(index) for index in range(record_count)] + with PersistentStorage(tmp_path) as storage: + for payload in expected: + assert storage.store(_record(payload)) + + barrier = threading.Barrier(8) + with ThreadPoolExecutor(max_workers=8) as executor: + futures = [executor.submit(_claim_concurrently, tmp_path, barrier) for _ in range(8)] + claimed = [payload for future in futures for payload in future.result(timeout=90)] + + assert sorted(claimed) == sorted(expected) + assert "failed to read durable record" not in caplog.text + with PersistentStorage(tmp_path) as storage: + assert not storage.claim(1000, 120) + + +@pytest.mark.parametrize("record_count", [1, 200]) +def test_concurrent_processes_claim_each_blob_once(tmp_path, record_count): + expected = [str(index) for index in range(record_count)] + with PersistentStorage(tmp_path) as storage: + for payload in expected: + assert storage.store(_record(payload)) + + context = multiprocessing.get_context("spawn") + barrier = context.Barrier(8) + results = context.Queue() + processes = [context.Process(target=_claim_in_process, args=(tmp_path, barrier, results)) for _ in range(8)] + try: + for process in processes: + process.start() + claimed = [payload for _ in processes for payload in results.get(timeout=90)] + for process in processes: + process.join(timeout=30) + assert process.exitcode == 0 + assert sorted(claimed) == sorted(expected) + finally: + for process in processes: + if process.is_alive(): + process.terminate() + process.join(timeout=30) + results.close() + results.join_thread() + with PersistentStorage(tmp_path) as storage: + assert not storage.claim(1000, 120) + + +def test_claim_limit_releases_coordination_for_another_thread(tmp_path): + with PersistentStorage(tmp_path) as first, PersistentStorage(tmp_path) as second: + assert first.store(_record("first")) + assert first.store(_record("second")) + claimed = first.claim(1, 120) + with ThreadPoolExecutor(max_workers=1) as executor: + other = executor.submit(second.claim, 1, 120).result(timeout=10) + assert sorted(record.payload for record in claimed + other) == ["first", "second"] + + +@pytest.mark.skipif(os.name != "nt", reason="Windows claim coordination") +def test_windows_lock_failure_disables_storage_without_polluting_monitor_state(tmp_path, caplog): with ( - patch.object(_mod.sys, "platform", "linux"), - patch.dict(os.environ, env, clear=True), - patch.object(_mod.Path, "home", side_effect=RuntimeError("no home")), + patch("ctypes.WinDLL") as kernel, + patch.dict(storage_state._LOCAL_STORAGE_SETUP_STATE, {"READONLY": False, "EXCEPTION_OCCURRED": ""}), ): - resolved = _mod._resolve_default_directory() + kernel.return_value.CreateMutexW.return_value = None + with pytest.raises(OSError, match="Azure Monitor"): + PersistentStorage(tmp_path) + assert storage_state.get_local_storage_setup_state_exception() == "" + assert "storage lock could not be created" in caplog.text - assert str(resolved).startswith(str(_mod.tempfile.gettempdir())) +@pytest.mark.parametrize("error_number", [errno.EACCES, errno.EROFS]) +@pytest.mark.parametrize("monitor_error", ["", "Azure Monitor setup failed"]) +def test_agent365_setup_failure_does_not_change_azure_monitor_sdkstats(tmp_path, error_number, monitor_error): + monitor = LocalFileStorage(str(tmp_path / "AzureMonitor")) + try: + with patch.dict( + storage_state._LOCAL_STORAGE_SETUP_STATE, + {"READONLY": False, "EXCEPTION_OCCURRED": monitor_error}, + ): + with patch( + "azure.monitor.opentelemetry.exporter._storage.os.makedirs", + side_effect=OSError(error_number, "Agent365 setup failed"), + ): + with pytest.raises(OSError, match="Azure Monitor"): + PersistentStorage(tmp_path) + result = monitor.put([{"azure_monitor": True}], lease_period=0) + assert result == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS + with patch("azure.monitor.opentelemetry.exporter.statsbeat.customer._utils.track_dropped_items") as dropped: + track_dropped_items_from_storage(result, [object()]) + assert storage_state.get_local_storage_setup_state_exception() == monitor_error + assert not storage_state.get_local_storage_setup_state_readonly() + assert dropped.call_count == (1 if monitor_error else 0) + finally: + monitor.close() -def test_default_directory_honors_xdg_state_home(tmp_path): - """XDG_STATE_HOME, when set, takes precedence over ~/.local/state.""" - import microsoft.opentelemetry.a365.core.exporters.persistent_storage as _mod - xdg = tmp_path / "xdg" - with patch.object(_mod.sys, "platform", "linux"), patch.dict(os.environ, {"XDG_STATE_HOME": str(xdg)}, clear=False): - resolved = _mod._resolve_default_directory() +def test_expired_lease_survives_restart(tmp_path): + with PersistentStorage(tmp_path) as storage: + assert storage.store(_record()) + assert len(storage.claim(1, 0)) == 1 + with PersistentStorage(tmp_path) as restarted: + assert len(restarted.claim(1, 30)) == 1 - assert str(resolved).startswith(str(xdg)) +def test_claim_limit_and_unknown_handles(tmp_path): + with PersistentStorage(tmp_path) as storage: + for i in range(5): + assert storage.store(_record(str(i))) + assert len(storage.claim(3, 30)) == 3 + assert len(storage.claim(3, 30)) == 2 + assert not storage.delete(-1) + assert not storage.release(-1) -# --------------------------------------------------------------------------- -# Multiple claims respect limit -# --------------------------------------------------------------------------- +def test_retention_discards_stale_records(tmp_path): + with PersistentStorage(tmp_path, retention_seconds=10) as storage: + assert storage.store(replace(_record("stale"), created_at=time.time() - 100)) + assert storage.store(_record("live")) + assert [r.payload for r in storage.claim(10, 30)] == ["live"] -def test_claim_respects_limit(tmp_path): - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=3600) - for i in range(5): - storage.store(_new_record(f'{{"i":{i}}}')) - claimed = storage.claim(limit=3, lease_seconds=30) - assert len(claimed) == 3 - storage.close() +@pytest.mark.parametrize( + "overrides", + [ + {"schema_version": 999}, + {"tenant_id": ""}, + {"agent_id": " "}, + {"payload": ""}, + {"created_at": "invalid"}, + {"created_at": float("inf")}, + {"use_s2s_endpoint": "false"}, + {"agentic_user_id": {}}, + ], +) +def test_invalid_envelope_is_logged_and_dropped_without_blocking_valid_records(tmp_path, caplog, overrides): + with PersistentStorage(tmp_path) as storage: + envelope = asdict(_record()) + envelope.update(overrides) + # Force invalid-first ordering even with a coarse-resolution Windows clock. + with ( + patch("azure.monitor.opentelemetry.exporter._storage._now", return_value=datetime.now(timezone.utc)), + patch("azure.monitor.opentelemetry.exporter._storage.random.getrandbits", side_effect=[0, 1]), + ): + assert storage._provider.put([envelope], lease_period=0) == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS + assert storage.store(_record("valid")) + assert [r.payload for r in storage.claim(1, 30)] == ["valid"] + assert "dropped invalid durable record" in caplog.text + assert not storage.claim(1, 30) + + +@pytest.mark.parametrize("created_at", [10**400, -(10**400)], ids=["positive", "negative"]) +def test_oversized_timestamp_does_not_interrupt_replay_batch(tmp_path, caplog, created_at): + with PersistentStorage(tmp_path) as storage: + envelope = asdict(_record()) + envelope["created_at"] = created_at + with ( + patch("azure.monitor.opentelemetry.exporter._storage._now", return_value=datetime.now(timezone.utc)), + patch("azure.monitor.opentelemetry.exporter._storage.random.getrandbits", side_effect=[0, 1, 2]), + ): + assert storage.store(_record("before")) + assert storage._provider.put([envelope], lease_period=0) == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS + assert storage.store(_record("after")) + claimed = storage.claim(2, 30) + assert [record.payload for record in claimed] == ["before", "after"] + assert "dropped invalid durable record" in caplog.text + for record in claimed: + assert storage.delete(record.record_id) + assert not storage.claim(2, 30) + assert not list(storage.directory.iterdir()) + + +def test_unreadable_blob_is_logged_and_retained_until_expiry(tmp_path, caplog): + with PersistentStorage(tmp_path) as storage: + assert storage.store(_record()) + blob = next(storage.directory.glob("*.blob")) + blob.write_text("not json", encoding="utf-8") + assert not storage.claim(1, 30) + assert "failed to read durable record" in caplog.text + assert len(list(storage.directory.iterdir())) == 1 + + +def test_transient_read_failure_never_deletes_undelivered_telemetry(tmp_path, caplog): + with PersistentStorage(tmp_path) as storage: + assert storage.store(_record("retained")) + with patch("builtins.open", side_effect=PermissionError("temporary read failure")): + assert not storage.claim(1, 30) + assert "failed to read durable record" in caplog.text + assert len(list(storage.directory.iterdir())) == 1 + assert [r.payload for r in storage.claim(1, 30)] == ["retained"] + + +def test_capacity_is_reclaimed_after_delete(tmp_path, caplog): + with PersistentStorage(tmp_path, capacity_bytes=1024) as storage: + count = 0 + while count < 100 and storage.store(_record("x" * 400)): + count += 1 + assert 0 < count < 100 + assert "PersistentStorage.store failed" in caplog.text + for record in storage.claim(100, 30): + assert storage.delete(record.record_id) + assert storage.store(_record("x" * 400)) + + +def test_oversized_record_is_rejected(tmp_path, caplog): + with PersistentStorage(tmp_path, capacity_bytes=1) as storage: + assert not storage.store(_record()) + assert "capacity" in caplog.text + assert not list(storage.directory.iterdir()) + + +@pytest.mark.parametrize("result", [StorageExportResult.CLIENT_STORAGE_DISABLED, "disk full"]) +def test_upstream_write_failure_is_logged(tmp_path, caplog, result): + with PersistentStorage(tmp_path) as storage: + with patch.object(storage._provider, "put", return_value=result): + assert not storage.store(_record()) + assert "PersistentStorage.store failed" in caplog.text + + +def test_silent_upstream_delete_failure_is_reported(tmp_path, caplog): + with PersistentStorage(tmp_path) as storage: + assert storage.store(_record()) + claimed = storage.claim(1, 30) + with patch("azure.monitor.opentelemetry.exporter._storage.os.remove", side_effect=PermissionError("denied")): + assert not storage.delete(claimed[0].record_id) + assert "PersistentStorage.delete failed" in caplog.text + assert storage.delete(claimed[0].record_id) + + +@pytest.mark.parametrize("error_type", [PermissionError, NotADirectoryError, OSError]) +def test_delete_stat_failure_is_logged_and_reported(tmp_path, caplog, error_type): + with PersistentStorage(tmp_path) as storage: + assert storage.store(_record()) + claimed = storage.claim(1, 30) + with patch.object(Path, "stat", side_effect=error_type("directory access changed")): + assert not storage.delete(claimed[0].record_id) + assert "PersistentStorage.delete failed" in caplog.text + assert "directory access changed" in caplog.text + assert storage.delete(claimed[0].record_id) + + +@pytest.mark.parametrize("error_type", [PermissionError, OSError]) +def test_delete_stat_failure_does_not_interrupt_replay_batch(tmp_path, caplog, error_type): + with PersistentStorage(tmp_path) as storage: + envelope = asdict(_record("invalid")) + envelope["tenant_id"] = "" + with ( + patch("azure.monitor.opentelemetry.exporter._storage._now", return_value=datetime.now(timezone.utc)), + patch("azure.monitor.opentelemetry.exporter._storage.random.getrandbits", side_effect=[0, 1, 2]), + ): + assert storage.store(_record("before")) + assert storage._provider.put([envelope], lease_period=0) == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS + assert storage.store(_record("after")) + with patch.object(Path, "stat", side_effect=error_type("directory access changed")): + claimed = storage.claim(2, 30) + assert [record.payload for record in claimed] == ["before", "after"] + assert "PersistentStorage.delete failed" in caplog.text + assert "directory access changed" in caplog.text + for record in claimed: + assert storage.delete(record.record_id) + assert not storage.claim(2, 30) + + +def test_provider_setup_failure_raises_for_exporter_nonfatal_disable(tmp_path): + with patch( + f"{MODULE}._Agent365LocalFileStorage._check_and_set_folder_permissions", + return_value=False, + ): + with pytest.raises(OSError, match="Azure Monitor"): + PersistentStorage(tmp_path) -# --------------------------------------------------------------------------- -# DurableRecord.new round-trips IdentityKey fields -# --------------------------------------------------------------------------- +@pytest.mark.parametrize("root", ["", " ", 123]) +def test_invalid_configuration_is_rejected_at_storage_initialization(root): + with pytest.raises(ValueError, match="storage_directory"): + PersistentStorage(root) -def test_durable_record_new_fields(): - key = IdentityKey( - tenant_id="myTenant", - agent_id="myAgent", - agentic_user_id="user42", - use_s2s_endpoint=True, - ) - assert list(inspect.signature(DurableRecord.new).parameters) == ["key", "payload"] - rec = DurableRecord.new(key, '{"data":1}') - assert rec.tenant_id == "myTenant" - assert rec.agent_id == "myAgent" - assert rec.agentic_user_id == "user42" - assert rec.use_s2s_endpoint is True - assert not hasattr(rec, "url") - assert rec.payload == '{"data":1}' - assert rec.record_id is None # not yet persisted +def test_file_root_fails_without_fallback(tmp_path): + root = tmp_path / "file" + root.write_text("unchanged", encoding="utf-8") + with pytest.raises(OSError, match="Azure Monitor|storage root"): + PersistentStorage(root) + assert root.read_text(encoding="utf-8") == "unchanged" -# --------------------------------------------------------------------------- -# delete returns False for unknown id -# --------------------------------------------------------------------------- +def test_default_directory_uses_azure_monitor_resolver(tmp_path): + with patch(f"{MODULE}._get_storage_directory", return_value=str(tmp_path / "default")) as resolve: + with PersistentStorage() as storage: + assert storage.directory == tmp_path / "default" + assert storage.store(_record()) + resolve.assert_called_once_with("agent365") -def test_delete_unknown_record_id(tmp_path): - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=3600) - assert storage.delete(99999) is False - storage.close() +def test_configured_root_uses_provider_normalization(tmp_path): + (tmp_path / "unused").mkdir() + root = tmp_path / "unused" / ".." / "root" + with PersistentStorage(root) as storage: + assert storage.directory == tmp_path / "root" / "a365" + assert storage.store(_record()) + assert [record.payload for record in storage.claim(1, 30)] == [_record().payload] -# --------------------------------------------------------------------------- -# release returns False for unknown id -# --------------------------------------------------------------------------- +def test_does_not_open_or_migrate_sqlite_backlogs(tmp_path, caplog): + database = tmp_path / "queue.db" + database.write_bytes(b"old SQLite backlog") + with PersistentStorage(tmp_path) as storage: + assert not storage.claim(10, 30) + assert storage.store(_record()) + assert database.read_bytes() == b"old SQLite backlog" + assert "Legacy Agent365 SQLite backlog" in caplog.text -def test_release_unknown_record_id(tmp_path): - storage = PersistentStorage(tmp_path, capacity_bytes=1024 * 1024, retention_seconds=3600) - assert storage.release(99999) is False - storage.close() +@pytest.mark.skipif(os.name == "nt", reason="Unix filesystem permissions") +@pytest.mark.parametrize("mask", [0o000, 0o022, 0o077]) +def test_real_unix_permissions_come_from_provider_without_changing_parent(tmp_path, mask): + parent_mode = stat.S_IMODE(tmp_path.stat().st_mode) + previous = os.umask(mask) + try: + with PersistentStorage(tmp_path) as storage: + assert storage.store(_record()) + assert stat.S_IMODE(storage.directory.stat().st_mode) == 0o700 + assert stat.S_IMODE(next(storage.directory.glob("*.blob")).stat().st_mode) == 0o600 + assert stat.S_IMODE(tmp_path.stat().st_mode) == parent_mode + finally: + os.umask(previous) + + +@pytest.mark.skipif(os.name == "nt", reason="Unix provider permissions") +def test_provider_repairs_existing_owned_cache_but_not_configured_parent(tmp_path): + tmp_path.chmod(0o750) + cache = tmp_path / "a365" + cache.mkdir() + cache.chmod(0o755) + with PersistentStorage(tmp_path) as storage: + assert storage.store(_record()) + assert stat.S_IMODE(cache.stat().st_mode) == 0o700 + assert stat.S_IMODE(tmp_path.stat().st_mode) == 0o750 + + +@pytest.mark.skipif(os.name == "nt", reason="Unix configured-root validation") +def test_configured_root_symlink_is_rejected_before_creating_cache(tmp_path): + target = tmp_path / "target" + target.mkdir(mode=0o750) + root = tmp_path / "root" + root.symlink_to(target, target_is_directory=True) + before = target.stat().st_mode + with pytest.raises(PermissionError, match="storage root"): + with PersistentStorage(root): + pass + assert root.is_symlink() + assert target.stat().st_mode == before + assert not list(target.iterdir()) + + +@pytest.mark.skipif(os.name == "nt", reason="Unix configured-root validation") +@pytest.mark.parametrize("normalized_mode", [0o700, 0o777], ids=["private", "writable"]) +def test_symlink_dot_dot_validates_provider_root(tmp_path, caplog, normalized_mode): + base = tmp_path / "base" + base.mkdir() + normalized_root = base / "root" + normalized_root.mkdir() + normalized_root.chmod(normalized_mode) + target = tmp_path / "target" + target.mkdir() + (target / "child").mkdir() + traversed_root = target / "root" + traversed_root.mkdir() + traversed_root.chmod(0o777 if normalized_mode == 0o700 else 0o700) + (base / "link").symlink_to(target / "child", target_is_directory=True) + configured_root = base / "link" / ".." / "root" + + if normalized_mode == 0o777: + with pytest.raises(PermissionError, match="storage root"): + with PersistentStorage(configured_root): + pass + assert not (normalized_root / "a365").exists() + else: + legacy_database = normalized_root / "queue.db" + legacy_database.write_bytes(b"old SQLite backlog") + with PersistentStorage(configured_root) as storage: + assert storage.directory == normalized_root / "a365" + assert storage.store(_record("persisted")) + assert "Legacy Agent365 SQLite backlog" in caplog.text + assert str(legacy_database) in caplog.text + assert legacy_database.read_bytes() == b"old SQLite backlog" + with PersistentStorage(normalized_root) as restarted: + assert [record.payload for record in restarted.claim(1, 30)] == ["persisted"] + + assert not (traversed_root / "a365").exists() + assert stat.S_IMODE(normalized_root.stat().st_mode) == normalized_mode + assert stat.S_IMODE(traversed_root.stat().st_mode) == (0o777 if normalized_mode == 0o700 else 0o700) + + +@pytest.mark.skipif(os.name == "nt", reason="Unix configured-root ownership") +@pytest.mark.parametrize( + ("existing", "mask"), + [(False, 0o022), (True, 0o022), (True, 0o077)], + ids=["created", "existing-022", "existing-077"], +) +def test_foreign_owned_configured_root_is_rejected_before_creating_cache(tmp_path, existing, mask): + root = tmp_path / "root" + if existing: + previous = os.umask(mask) + try: + root.mkdir(mode=0o750) + finally: + os.umask(previous) + expected_mode = stat.S_IMODE(root.stat().st_mode) if existing else 0o700 + real_lstat = Path.lstat + + def foreign_owner(path): + info = real_lstat(path) + if path == root: + fields = list(info) + fields[4] = os.getuid() + 1 # pylint: disable=no-member + return os.stat_result(fields) + return info + + with patch.object(Path, "lstat", new=foreign_owner): + with pytest.raises(PermissionError, match="storage root"): + with PersistentStorage(root): + pass + assert not (root / "a365").exists() + assert stat.S_IMODE(root.stat().st_mode) == expected_mode + + +@pytest.mark.skipif(os.name == "nt", reason="Unix configured-root permissions") +@pytest.mark.parametrize("mode", [0o770, 0o707]) +def test_writable_configured_root_is_rejected_without_permission_repair(tmp_path, mode): + tmp_path.chmod(mode) + with pytest.raises(PermissionError, match="storage root"): + with PersistentStorage(tmp_path): + pass + assert stat.S_IMODE(tmp_path.stat().st_mode) == mode + assert not list(tmp_path.iterdir()) + + +@pytest.mark.skipif(os.name == "nt", reason="Unix root and cache creation") +@pytest.mark.parametrize("mask", [0o000, 0o022, 0o077]) +def test_new_configured_root_and_cache_are_private_and_survive_restart(tmp_path, mask): + root = tmp_path / "new" + parent_mode = tmp_path.stat().st_mode + previous = os.umask(mask) + try: + with PersistentStorage(root) as storage: + assert stat.S_IMODE(root.stat().st_mode) == 0o700 + assert stat.S_IMODE(storage.directory.stat().st_mode) == 0o700 + assert storage.store(_record("persisted")) + assert stat.S_IMODE(next(storage.directory.glob("*.blob")).stat().st_mode) == 0o600 + with PersistentStorage(root) as restarted: + assert [record.payload for record in restarted.claim(1, 30)] == ["persisted"] + finally: + os.umask(previous) + assert tmp_path.stat().st_mode == parent_mode + + +@pytest.mark.skipif(os.name == "nt", reason="Unix provider symlink protection") +def test_provider_rejects_symlinked_cache_without_changing_target(tmp_path): + target = tmp_path / "target" + target.mkdir(mode=0o755) + (tmp_path / "a365").symlink_to(target, target_is_directory=True) + before = target.stat().st_mode + with pytest.raises(OSError, match="Azure Monitor"): + PersistentStorage(tmp_path) + assert target.stat().st_mode == before + assert not list(target.iterdir()) + + +@pytest.mark.skipif(os.name != "nt", reason="Windows provider ACLs") +def test_windows_uses_upstream_acl_handling(tmp_path): + with patch("azure.monitor.opentelemetry.exporter._storage.subprocess.run") as run: + run.return_value.returncode = 0 + with PersistentStorage(tmp_path) as storage: + assert storage.store(_record()) + args = run.call_args.args[0] + assert Path(args[0]).name == "icacls.exe" + assert args[1] == str(tmp_path / "a365") + assert "/inheritance:r" in args + assert "/grant" in args diff --git a/tests/test_distro.py b/tests/test_distro.py index f290d3267..65ed45fd5 100644 --- a/tests/test_distro.py +++ b/tests/test_distro.py @@ -1101,19 +1101,27 @@ def test_storage_directory_none_by_default(self, default_resolver_mock): self.assertIsNone(exporter_kwargs["storage_directory"]) @patch("microsoft.opentelemetry.a365.core.exporters.utils._create_default_token_resolver") - def test_empty_storage_directory_raises_value_error(self, default_resolver_mock): - """An explicitly empty storage_directory must raise ValueError rather - than silently defaulting to the platform path.""" + def test_empty_storage_directory_disables_only_buffering(self, default_resolver_mock): + from microsoft.opentelemetry.a365.core.exporters.agent365_exporter import logger + default_resolver_mock.return_value = lambda aid, tid: "token" - with patch("microsoft.opentelemetry.a365.core.exporters.agent365_exporter._Agent365Exporter"): - otel_kwargs = {"span_processors": []} - with self.assertRaises(ValueError): - _append_a365_components( - True, - otel_kwargs, - enable_observability_exporter=True, - storage_directory="", - ) + otel_kwargs = {"span_processors": []} + _append_a365_components( + True, + otel_kwargs, + enable_observability_exporter=True, + storage_directory="", + ) + exporter = otel_kwargs["span_processors"][-1].span_exporter + try: + with self.assertLogs(logger, level="ERROR") as logs: + exporter._ensure_durable_initialized() + self.assertIn("Durable delivery disabled", "\n".join(logs.output)) + self.assertFalse(exporter._enable_durable_delivery) + self.assertIsNone(exporter._storage) + finally: + for processor in otel_kwargs["span_processors"]: + processor.shutdown() class TestA365Components(unittest.TestCase):