fix(telemetry): make metrics best-effort so telemetry never crashes storage ops

.release_notes/.unreleased.md

@@ -5,3 +5,5 @@
 ## New Features
 
 ## Bug Fixes
+
+- Telemetry is now best-effort and never crashes a storage operation. Exporter/instrument failures — including OpenTelemetry version skew that dropped `add`/`set` from the cross-process `Counter`/`Gauge` proxies — are logged and disable telemetry for the affected client instead of raising. The metric instrument proxies now declare their mutating method explicitly (`exposed=["add"]`/`["set"]`) so proxy generation no longer depends on the installed OpenTelemetry's instrument shape. Telemetry can also be disabled up front via `MSC_TELEMETRY_DISABLED` or `OTEL_SDK_DISABLED`.

multi-storage-client-docs/src/user_guide/telemetry.rst

@@ -93,6 +93,22 @@ If the default telemetry provider creation doesn't behave as desired, you can ma
    # Use an MSC shortcut to create a storage client for a profile and open an object/file.
    multistorageclient.open("msc://data/file.txt")
 
+*******************
+Disabling Telemetry
+*******************
+
+Telemetry is best-effort observability and never crashes a storage operation. Any telemetry failure (e.g. an exporter that fails to construct, missing optional dependencies, or an unreachable telemetry process) is logged and metrics are disabled for the affected storage client; the storage operation always proceeds.
+
+Telemetry can also be disabled explicitly, before any telemetry process, exporter, or IPC setup, by setting either of these environment variables to a truthy value (``1``, ``true``, or ``yes``):
+
+.. code-block:: shell
+
+   # MSC-specific switch.
+   export MSC_TELEMETRY_DISABLED=true
+
+   # OpenTelemetry SDK standard switch (also honored).
+   export OTEL_SDK_DISABLED=true
+
 **************
 Authentication
 **************

multi-storage-client/src/multistorageclient/providers/base.py

@@ -243,6 +243,10 @@ def __init__(
         self._metrics_dropped_count = 0
         self._metrics_dropped_count_lock = threading.Lock()
 
+        # Latched off after an unrecoverable metric-record failure so telemetry never
+        # crashes a storage operation and failures aren't logged per-operation.
+        self._metrics_disabled = False
+
     def __str__(self) -> str:
         return self._provider_name
 
@@ -254,6 +258,26 @@ def __del__(self) -> None:
             except Exception as e:
                 logger.warning(f"Failed to shutdown async telemetry: {e}", exc_info=True)
 
+    @staticmethod
+    def _telemetry_disabled_via_env() -> bool:
+        """Whether telemetry is explicitly disabled via environment variable."""
+        for name in ("MSC_TELEMETRY_DISABLED", "OTEL_SDK_DISABLED"):
+            value = os.environ.get(name)
+            if value is not None and value.strip().lower() in ("1", "true", "yes"):
+                return True
+        return False
+
+    @staticmethod
+    def _usable_instrument(instrument: Any, record_method: str) -> bool:
+        """
+        Whether a metric instrument can record via ``record_method``.
+
+        A disabled instrument is ``None`` in local mode or, in manager mode, an auto-proxy
+        wrapping ``None`` that lacks the record method (a non-instrument proxy). Both must
+        be treated as disabled so recording never crashes.
+        """
+        return instrument is not None and callable(getattr(instrument, record_method, None))
+
     def _init_metrics(self) -> None:
         """
         Initialize metrics.
@@ -266,6 +290,10 @@ def _init_metrics(self) -> None:
         """
         with self._metric_init_lock:
             if not self._metric_init_event.is_set():
+                if self._telemetry_disabled_via_env():
+                    logger.debug("Telemetry disabled via environment variable; skipping metrics initialization.")
+                    self._metric_init_event.set()
+                    return
                 if self._config_dict is not None and self._telemetry_provider is not None:
                     opentelemetry_config: Optional[dict[str, Any]] = self._config_dict.get("opentelemetry")
                     if opentelemetry_config is not None:
@@ -276,9 +304,13 @@ def _init_metrics(self) -> None:
 
                             if metrics_config is not None:
                                 for name in Telemetry.GaugeName:
-                                    self._metric_gauges[name] = telemetry.gauge(config=metrics_config, name=name)
+                                    gauge = telemetry.gauge(config=metrics_config, name=name)
+                                    self._metric_gauges[name] = gauge if self._usable_instrument(gauge, "set") else None
                                 for name in Telemetry.CounterName:
-                                    self._metric_counters[name] = telemetry.counter(config=metrics_config, name=name)
+                                    counter = telemetry.counter(config=metrics_config, name=name)
+                                    self._metric_counters[name] = (
+                                        counter if self._usable_instrument(counter, "add") else None
+                                    )
 
                                 attributes_provider_configs: Optional[list[dict[str, Any]]] = metrics_config.get(
                                     "attributes"
@@ -571,7 +603,14 @@ def _dispatch_metrics(
         Unlike :meth:`_emit_metrics` which wraps a callable, this method accepts
         pre-computed metric values. Used by :meth:`_emit_metrics_sync`, :meth:`_emit_metrics_async`,
         and directly when the operation is performed outside the standard wrapper (e.g. async Rust downloads).
+
+        Telemetry is best-effort: a failing synchronous record (e.g. a dead telemetry manager
+        or a disabled instrument) must never crash the storage operation. Any failure latches
+        metrics off so subsequent operations silently no-op instead of logging per-operation.
         """
+        if self._metrics_disabled:
+            return
+
         if self._async_metrics_enabled and self._metrics_queue is not None:
             metric_data = {
                 "operation": operation,
@@ -585,7 +624,14 @@ def _dispatch_metrics(
                 with self._metrics_dropped_count_lock:
                     self._metrics_dropped_count += 1
         else:
-            self._record_metrics(operation, latency, data_size, error_type)
+            try:
+                self._record_metrics(operation, latency, data_size, error_type)
+            except (EOFError, BrokenPipeError, ConnectionError):
+                self._metrics_disabled = True
+                logger.warning("Telemetry manager connection closed; disabling metrics.")
+            except Exception:
+                self._metrics_disabled = True
+                logger.warning("Failed to record metrics; disabling metrics.", exc_info=True)
 
     def _append_delimiter(self, s: str, delimiter: str = "/") -> str:
         if not s.endswith(delimiter):

multi-storage-client/src/multistorageclient/telemetry/__init__.py

@@ -114,7 +114,8 @@ class CounterName(enum.Enum):
     }
 
     # Map of config as a sorted JSON string (since dictionaries can't be hashed) to meter provider.
-    _meter_provider_cache: dict[str, api_metrics.MeterProvider]
+    # A cached ``None`` means the config is disabled; it prevents re-running (failed) exporter construction.
+    _meter_provider_cache: dict[str, api_metrics.MeterProvider | None]
     _meter_provider_cache_lock: threading.Lock
     # Map of config as a sorted JSON string (since dictionaries can't be hashed) to meter.
     _meter_cache: dict[str, api_metrics.Meter]
@@ -126,7 +127,8 @@ class CounterName(enum.Enum):
     _counter_cache: dict[str, dict[CounterName, api_metrics.Counter]]
     _counter_cache_lock: threading.Lock
     # Map of config as a sorted JSON string (since dictionaries can't be hashed) to tracer provider.
-    _tracer_provider_cache: dict[str, api_trace.TracerProvider]
+    # A cached ``None`` means the config is disabled; it prevents re-running (failed) exporter construction.
+    _tracer_provider_cache: dict[str, api_trace.TracerProvider | None]
     _tracer_provider_cache_lock: threading.Lock
     # Map of config as a sorted JSON string (since dictionaries can't be hashed) to tracer.
     _tracer_cache: dict[str, api_trace.Tracer]
@@ -194,15 +196,17 @@ def meter_provider(self, config: dict[str, Any]) -> Optional[api_metrics.MeterPr
                         return self._meter_provider_cache.setdefault(
                             config_json, sdk_metrics.MeterProvider(metric_readers=[reader])
                         )
-                    except (AttributeError, ImportError):
+                    except Exception:
                         logger.error(
-                            "Failed to import OpenTelemetry Python SDK or exporter! Disabling metrics.", exc_info=True
+                            "Failed to initialize the OpenTelemetry meter provider or exporter! Disabling metrics.",
+                            exc_info=True,
                         )
-                        return None
+                        # Cache the disabled state so (possibly expensive) exporter construction isn't retried.
+                        return self._meter_provider_cache.setdefault(config_json, None)
                 else:
                     # Don't return a no-op meter provider to avoid unnecessary overhead.
                     logger.error("No exporter configured! Disabling metrics.")
-                    return None
+                    return self._meter_provider_cache.setdefault(config_json, None)
 
     def meter(self, config: dict[str, Any]) -> Optional[api_metrics.Meter]:
         """
@@ -301,14 +305,16 @@ def tracer_provider(self, config: dict[str, Any]) -> Optional[api_trace.TracerPr
                             config_json,
                             sdk_trace.TracerProvider(active_span_processor=processor, sampler=sampler),
                         )
-                    except (AttributeError, ImportError):
+                    except Exception:
                         logger.error(
-                            "Failed to import OpenTelemetry Python SDK or exporter! Disabling traces.", exc_info=True
+                            "Failed to initialize the OpenTelemetry tracer provider or exporter! Disabling traces.",
+                            exc_info=True,
                         )
-                        return None
+                        # Cache the disabled state so (possibly expensive) exporter construction isn't retried.
+                        return self._tracer_provider_cache.setdefault(config_json, None)
                 else:
                     logger.error("No exporter configured! Disabling traces.")
-                    return None
+                    return self._tracer_provider_cache.setdefault(config_json, None)
 
     def tracer(self, config: dict[str, Any]) -> Optional[api_trace.Tracer]:
         """
@@ -429,8 +435,13 @@ def _fully_qualified_name(c: type[Any]) -> str:
 
 
 # Metrics proxy object setup.
-TelemetryManager.register(typeid=_fully_qualified_name(api_metrics._Gauge))
-TelemetryManager.register(typeid=_fully_qualified_name(api_metrics.Counter))
+#
+# ``exposed`` is set explicitly rather than relying on ``multiprocessing``'s default
+# discovery (``public_methods`` via ``dir()``): an OpenTelemetry build that delegates an
+# instrument's mutating method isn't surfaced by ``dir()``, so the ``AutoProxy`` would drop
+# it and raise ``AttributeError: 'AutoProxy[...]' object has no attribute 'add'`` on record.
+TelemetryManager.register(typeid=_fully_qualified_name(api_metrics._Gauge), exposed=["set"])
+TelemetryManager.register(typeid=_fully_qualified_name(api_metrics.Counter), exposed=["add"])
 TelemetryManager.register(
     typeid=_fully_qualified_name(api_metrics.Meter),
     method_to_typeid={

multi-storage-client/src/multistorageclient/telemetry/metrics/exporters/otlp_mtls_vault.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import inspect
 import logging
 from typing import Any
 
@@ -67,7 +68,21 @@ def __init__(
             - key_key: Key name for client key (default: "key")
             - ca_key: Key name for CA certificate (default: "ca")
         :param exporter: OTLP metric exporter config dictionary (passed through to OTLPMetricExporter).
+        :raises RuntimeError: If the installed OTLP exporter does not support mTLS client certificate options.
         """
+        # mTLS client certificate options were added in opentelemetry-exporter-otlp-proto-http 1.32.1.
+        # Fail fast with a clear error (instead of an opaque TypeError from the parent) on older versions.
+        parameters = inspect.signature(OTLPMetricExporter.__init__).parameters
+        supports_client_certificate = "client_certificate_file" in parameters or any(
+            parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters.values()
+        )
+        if not supports_client_certificate:
+            raise RuntimeError(
+                "The installed opentelemetry-exporter-otlp-proto-http does not support the mTLS client "
+                "certificate options (client_certificate_file / client_key_file). Upgrade to "
+                "opentelemetry-exporter-otlp-proto-http >= 1.32.1 to use the _otlp_mtls_vault exporter."
+            )
+
         provider = VaultCertificateProvider(**auth)
         cert_paths = provider.get_certificates()