Support continuous profiling with declarative configuration

README.md

@@ -81,8 +81,8 @@ which exports traces, metrics, and logs to a local Collector at `http://localhos
 This support configures upstream OpenTelemetry SDK components. Configure
 exporters, including their endpoints and authentication headers, in the YAML file.
 
-Support for the GDI `distribution.splunk` configuration block is limited to
-`instrumentations.http.trace_response_header_enabled`, which defaults to `true`.
+The GDI `distribution.splunk` configuration block supports continuous CPU profiling
+and `instrumentations.http.trace_response_header_enabled`, which defaults to `true`.
 To disable the trace response header, add this block to the YAML file:
 
 ```yaml
@@ -93,8 +93,9 @@ distribution:
         trace_response_header_enabled: false
 ```
 
+To enable continuous profiling, see [profiling configuration](docs/profiling.md#yaml-configuration).
 When `OTEL_CONFIG_FILE` is set, the distribution does not enable
-[continuous or call graph profiling](docs/profiling.md) or
+[call graph profiling](docs/profiling.md#call-graph-profiling) or
 [PostgreSQL trace correlation](docs/postgresql.md).
 
 OpAMP does not report the declarative file as effective configuration and cannot

docs/profiling.md

@@ -5,9 +5,8 @@ profiling** (also called snapshot profiling). Both modes collect CPU stack trace
 export them as `pprof` data via the OTel logs exporter, with trace and span IDs
 embedded so profiles can be correlated with traces in Splunk APM.
 
-Both profiling modes require `OTEL_CONFIG_FILE` to be unset. When it is set, the
-distribution does not start either profiler, even if its environment variable is
-set to `true`.
+Continuous profiling supports environment variables and YAML configuration.
+Call graph profiling requires `OTEL_CONFIG_FILE` to be unset.
 
 ## Continuous profiling
 
@@ -31,6 +30,54 @@ opentelemetry-instrument python app.py
 | `SPLUNK_PROFILER_CALL_STACK_INTERVAL` | `1000`                                      | How often (in milliseconds) to collect a stack sample from all threads.              |
 | `SPLUNK_PROFILER_LOGS_ENDPOINT`       | _(uses `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT`)_ | Override the endpoint where profiling data is sent. Applies to both profiling modes. |
 
+### YAML configuration
+
+When `OTEL_CONFIG_FILE` is set, add the following block to enable continuous CPU
+profiling. An empty `cpu_profiler: {}` block uses the default sampling interval
+of 1000 milliseconds. Omit `cpu_profiler` to disable continuous profiling.
+
+```yaml
+distribution:
+  splunk:
+    profiling:
+      always_on:
+        cpu_profiler:
+          sampling_interval: 1000
+```
+
+`sampling_interval` must be a positive integer in milliseconds. Profiles use the
+file's `resource` attributes and a dedicated OTLP HTTP exporter. When no profiling
+exporter is configured, profiles go to `http://localhost:4318/v1/logs`.
+Profile export does not require a `logger_provider` block.
+
+To configure the profile export endpoint or headers, add `exporter` under
+`distribution.splunk.profiling`:
+
+```yaml
+distribution:
+  splunk:
+    profiling:
+      exporter:
+        otlp_log_http:
+          endpoint: http://localhost:4318/v1/logs
+          headers:
+            - name: X-SF-TOKEN
+              value: ${SPLUNK_ACCESS_TOKEN}
+      always_on:
+        cpu_profiler:
+          sampling_interval: 1000
+```
+
+The dedicated exporter receives only profile logs. Application logs continue
+through `logger_provider`. An empty `otlp_log_http: {}` block uses
+`http://localhost:4318/v1/logs`. The block accepts the standard OTLP HTTP log
+exporter settings, including `headers`, `timeout` (in milliseconds), and
+`compression`. The profiling thread stops and its exporter flushes at process exit.
+
+`SPLUNK_PROFILER_ENABLED`, `SPLUNK_PROFILER_CALL_STACK_INTERVAL`, and
+`SPLUNK_PROFILER_LOGS_ENDPOINT` do not control profiling when `OTEL_CONFIG_FILE`
+is set. To use environment values, reference them explicitly in the YAML file.
+
 ### How it works
 
 A background thread fires at each `SPLUNK_PROFILER_CALL_STACK_INTERVAL` tick, collects

pyproject.toml

@@ -23,6 +23,7 @@ classifiers = [
   "Programming Language :: Python :: Implementation :: PyPy",
 ]
 dependencies = [
+  "jsonschema>=4.0",
   "opentelemetry-api==1.45.1",
   "opentelemetry-sdk==1.45.1",
   "opentelemetry-propagator-b3==1.45.1",

src/splunk_otel/_declarative_profiling.py

@@ -0,0 +1,84 @@
+# Copyright Splunk Inc.
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from importlib.resources import files
+from urllib.parse import urljoin
+
+from jsonschema import Draft202012Validator, ValidationError, validators
+from opentelemetry.configuration import ConfigurationError
+from opentelemetry.configuration._conversion import _dict_to_dataclass
+from opentelemetry.configuration._logger_provider import create_logger_provider
+from opentelemetry.configuration.file._loader import _get_schema
+from opentelemetry.configuration.models import LoggerProvider as LoggerProviderConfig
+from opentelemetry.exporter.otlp.proto.http import _log_exporter as otlp_http_logs
+from opentelemetry.sdk._logs import LoggerProvider
+from opentelemetry.sdk.resources import Resource
+
+from splunk_otel.profile import (
+    _DEFAULT_PROF_CALL_STACK_INTERVAL_MILLIS,
+    _get_profiling_logger,
+    _start_continuous_profiling,
+)
+
+
+def _create_profiling_validator():
+    profiling_schema = json.loads(files("splunk_otel").joinpath("_profiling_schema.json").read_text(encoding="utf-8"))
+    # JSON Schema accepts 1.0 as an integer, but protobuf sampling periods require
+    # Python int values. The exact-type check rejects floats and bools for every
+    # integer field in this validator, including HTTP exporter settings.
+    validator_class = validators.extend(
+        Draft202012Validator,
+        type_checker=Draft202012Validator.TYPE_CHECKER.redefine("integer", lambda _checker, value: type(value) is int),
+    )
+    opentelemetry_definitions = _get_schema()["$defs"]
+    profiling_definitions = profiling_schema["$defs"]
+    combined_definitions = {**opentelemetry_definitions, **profiling_definitions}
+    combined_schema = {**profiling_schema, "$defs": combined_definitions}
+    return validator_class(combined_schema)
+
+
+_PROFILING_VALIDATOR = _create_profiling_validator()
+
+
+def _configure_declarative_continuous_profiling(distribution_config: dict | None, resource: Resource) -> None:
+    _validate_profiling_config(distribution_config)
+    profiling_config = _config_at(distribution_config, ("splunk", "profiling"))
+    always_on_config = _config_at(profiling_config, ("always_on",))
+    if "cpu_profiler" not in always_on_config:
+        return
+
+    cpu_profiler_config = _config_at(always_on_config, ("cpu_profiler",))
+    interval_millis = cpu_profiler_config.get("sampling_interval", _DEFAULT_PROF_CALL_STACK_INTERVAL_MILLIS)
+
+    logger_provider = _create_profiling_logger_provider(profiling_config, resource)
+    logger = _get_profiling_logger(logger_provider)
+    _start_continuous_profiling(resource=resource, interval_millis=interval_millis, logger=logger)
+
+
+def _create_profiling_logger_provider(profiling_config: dict, resource: Resource) -> LoggerProvider:
+    http_exporter_config = _config_at(profiling_config, ("exporter", "otlp_log_http"))
+    default_endpoint = urljoin(otlp_http_logs.DEFAULT_ENDPOINT, otlp_http_logs.DEFAULT_LOGS_EXPORT_PATH)
+    # Missing or null endpoints must use the SDK default rather than environment settings.
+    endpoint = http_exporter_config.get("endpoint") or default_endpoint
+    http_exporter_config = {**http_exporter_config, "endpoint": endpoint}
+    logger_provider_data = {"processors": [{"batch": {"exporter": {"otlp_http": http_exporter_config}}}]}
+    logger_provider_config = _dict_to_dataclass(logger_provider_data, LoggerProviderConfig)
+    return create_logger_provider(logger_provider_config, resource)
+
+
+def _config_at(config: dict | None, path: tuple[str, ...]) -> dict:
+    """Read mapping-valued sections from already validated configuration."""
+    section = config or {}
+    for key in path:
+        section = section.get(key) or {}
+    return section
+
+
+def _validate_profiling_config(distribution_config: dict | None) -> None:
+    try:
+        _PROFILING_VALIDATOR.validate(distribution_config)
+    except ValidationError as error:
+        config_path = ".".join(["distribution", *(str(part) for part in error.absolute_path)])
+        message = f"{config_path}: {error.message}"
+        raise ConfigurationError(message) from error

src/splunk_otel/_profiling_schema.json

@@ -0,0 +1,54 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "type": ["object", "null"],
+  "properties": {
+    "splunk": {
+      "type": ["object", "null"],
+      "properties": {
+        "profiling": {"$ref": "#/$defs/SplunkProfiling"}
+      }
+    }
+  },
+  "$defs": {
+    "SplunkProfiling": {
+      "type": ["object", "null"],
+      "properties": {
+        "always_on": {
+          "type": ["object", "null"],
+          "properties": {
+            "cpu_profiler": {
+              "type": ["object", "null"],
+              "properties": {
+                "sampling_interval": {"type": "integer", "minimum": 1}
+              }
+            }
+          }
+        }
+      },
+      "if": {
+        "$comment": "Exporter settings are used only when continuous CPU profiling is enabled.",
+        "type": "object",
+        "required": ["always_on"],
+        "properties": {
+          "always_on": {"type": "object", "required": ["cpu_profiler"]}
+        }
+      },
+      "then": {
+        "properties": {
+          "exporter": {
+            "type": ["object", "null"],
+            "additionalProperties": false,
+            "properties": {
+              "otlp_log_http": {
+                "anyOf": [
+                  {"type": "null"},
+                  {"$ref": "#/$defs/OtlpHttpExporter"}
+                ]
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}

src/splunk_otel/configurator.py

@@ -64,8 +64,7 @@ def _configure(self, **kwargs):
             set_global_response_propagator(ServerTimingResponsePropagator())
         configure_logging_instrumentation(env)
 
-        # A later change will configure profiling and callgraphs from OTEL_CONFIG_FILE.
-        # For now, start them from environment variables only when no config file is set.
+        # Call graphs and PostgreSQL trace correlation do not support file configuration yet.
         if not config_file:
             configure_postgresql_trace_context_propagation(env)
             _start_profiling_if_enabled()
@@ -79,6 +78,7 @@ def _configure_declarative_sdk(config_file: str, **kwargs) -> OpenTelemetryConfi
     from opentelemetry.configuration._propagator import configure_propagator
     from opentelemetry.configuration._tracer_provider import configure_tracer_provider
     from opentelemetry.configuration.instrumentation import configure_instrumentation
+    from splunk_otel._declarative_profiling import _configure_declarative_continuous_profiling
 
     if kwargs:
         _logger.warning(
@@ -98,6 +98,7 @@ def _configure_declarative_sdk(config_file: str, **kwargs) -> OpenTelemetryConfi
     configure_logger_provider(config.logger_provider, resource)
     configure_propagator(config.propagator)
     configure_instrumentation(config.instrumentation_development)
+    _configure_declarative_continuous_profiling(config.distribution, resource)
     return config
 
 

src/splunk_otel/profile.py

@@ -1,3 +1,4 @@
+import atexit as _atexit
 import base64
 import gzip
 import sys
@@ -12,6 +13,7 @@
 import opentelemetry.context
 import wrapt
 from opentelemetry._logs import Logger, LogRecord, SeverityNumber, get_logger
+from opentelemetry._logs import LoggerProvider as _LoggerProvider
 from opentelemetry.context import Context
 from opentelemetry.instrumentation.version import __version__ as version
 from opentelemetry.sdk._logs import ReadWriteLogRecord
@@ -79,21 +81,27 @@ def _start_profiling_if_enabled(env=None):
 
 
 def start_profiling(env=None):
+    """Start continuous profiling using environment settings."""
     env = env or Env()
     interval_millis = env.getint(SPLUNK_PROFILER_CALL_STACK_INTERVAL, _DEFAULT_PROF_CALL_STACK_INTERVAL_MILLIS)
-    svcname = env.getval(OTEL_SERVICE_NAME)
-
-    ctx = ProfilingContext(
-        _mk_resource(svcname),
-        interval_millis,
-        get_logger(_SCOPE_NAME, _SCOPE_VERSION),
+    service_name = env.getval(OTEL_SERVICE_NAME)
+    return _start_continuous_profiling(
+        resource=_mk_resource(service_name),
+        interval_millis=interval_millis,
+        logger=_get_profiling_logger(),
     )
-    ctx.start()
-    return ctx
 
 
-def _get_profiling_logger() -> Logger:
-    return get_logger(_SCOPE_NAME, _SCOPE_VERSION)
+def _start_continuous_profiling(resource: Resource, interval_millis: int, logger: Logger) -> ProfilingContext:
+    context = ProfilingContext(resource, interval_millis, logger)
+    context.start()
+    # Providers register shutdown first; stop sampling before they flush at exit.
+    _atexit.register(context.stop)
+    return context
+
+
+def _get_profiling_logger(logger_provider: _LoggerProvider | None = None) -> Logger:
+    return get_logger(_SCOPE_NAME, _SCOPE_VERSION, logger_provider=logger_provider)
 
 
 def _mk_resource(service_name) -> Resource:
@@ -268,6 +276,7 @@ def __init__(self, interval_millis, target):
         self.running = False
         self.pause_at = None
         self.wakeup_event = threading.Event()
+        self._stop_event = threading.Event()
 
     def start(self):
         self.pause_at = None
@@ -276,26 +285,32 @@ def start(self):
         if self.running:
             return
 
+        self._stop_event.clear()
         self.running = True
         self.thread.start()
 
     def _loop(self):
         while self.running:
             start_time_seconds = time.monotonic()
 
-            if self.pause_at is not None and start_time_seconds >= self.pause_at:
-                self.wakeup_event.clear()  # clear event so next line will block
-                self.wakeup_event.wait()  # wait for event.set() (either via start() or stop())
+            pause_at = self.pause_at
+            if pause_at is not None and start_time_seconds >= pause_at:
+                self.wakeup_event.clear()
+                # Recheck state because clear() can erase a concurrent stop or resume signal.
+                if not self.running or self.pause_at is None:
+                    continue
+                self.wakeup_event.wait()
                 continue
 
             self.target()
             elapsed_seconds = time.monotonic() - start_time_seconds
             sleep_seconds = max(0, self.interval_seconds - elapsed_seconds)
-            time.sleep(sleep_seconds)
+            self._stop_event.wait(sleep_seconds)
 
     def stop(self):
         self.running = False
         self.pause_at = None
+        self._stop_event.set()
         self.wakeup_event.set()  # unblock _loop() if waiting, so it can return and thread can exit
         if self.thread.is_alive():
             self.thread.join()