diff --git a/README.md b/README.md index b61d2961..a10af0df 100644 --- a/README.md +++ b/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 diff --git a/docs/profiling.md b/docs/profiling.md index 8e192f4b..796f8563 100644 --- a/docs/profiling.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 8e89db8d..cca44a88 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/splunk_otel/_declarative_profiling.py b/src/splunk_otel/_declarative_profiling.py new file mode 100644 index 00000000..15abe617 --- /dev/null +++ b/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 diff --git a/src/splunk_otel/_profiling_schema.json b/src/splunk_otel/_profiling_schema.json new file mode 100644 index 00000000..d7a4b426 --- /dev/null +++ b/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"} + ] + } + } + } + } + } + } + } +} diff --git a/src/splunk_otel/configurator.py b/src/splunk_otel/configurator.py index 32d103f0..f69144b1 100644 --- a/src/splunk_otel/configurator.py +++ b/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 diff --git a/src/splunk_otel/profile.py b/src/splunk_otel/profile.py index 6fb5995f..b1b663ba 100644 --- a/src/splunk_otel/profile.py +++ b/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,6 +285,7 @@ def start(self): if self.running: return + self._stop_event.clear() self.running = True self.thread.start() @@ -283,19 +293,24 @@ 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() diff --git a/tests/integration/declarative_custom_profile_exporter.py b/tests/integration/declarative_custom_profile_exporter.py new file mode 100644 index 00000000..efe562fe --- /dev/null +++ b/tests/integration/declarative_custom_profile_exporter.py @@ -0,0 +1,77 @@ +import logging +from pathlib import Path + +from lib import project_path, trace_loop, wait_for_server + + +if __name__ == "__main__": + from opentelemetry._logs import LogRecord, get_logger + + get_logger("application").emit(LogRecord(body="application log")) + trace_loop(1) + + +class DeclarativeCustomProfileExporterOtelTest: + def requirements(self): + return (project_path(),) + + def declarative_configuration(self): + config_yaml = Path(project_path(), "docs", "examples", "declarative-config.yaml").read_text() + return ( + config_yaml + + """ +distribution: + splunk: + profiling: + exporter: + otlp_log_http: + endpoint: http://localhost:14318/v1/logs + headers: + - name: x-profile-test + value: yaml-token + timeout: 5000 + always_on: + cpu_profiler: + sampling_interval: 100 +""" + ) + + def environment_variables(self): + return { + "SPLUNK_PROFILER_LOGS_ENDPOINT": "http://localhost:1/unused", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT": "http://localhost:1/unused", + "OTEL_PYTHON_DISABLED_INSTRUMENTATIONS": "system_metrics", + } + + def wrapper_command(self): + return "opentelemetry-instrument" + + def is_http(self): + return False + + def on_start(self): + from oteltest.sink import HttpSink, raise_if_port_in_use + from oteltest.sink.handler import AccumulatingHandler + + raise_if_port_in_use(14318) + self.profile_handler = AccumulatingHandler() + self.profile_sink = HttpSink(self.profile_handler, logging.getLogger(__name__), port=14318) + self.profile_sink.start() + wait_for_server("localhost", 14318) + + def on_stop(self, telemetry, stdout, stderr, returncode): + from oteltest.telemetry import extract_leaves, get_logs + + self.profile_sink.stop() + assert returncode == 0, stderr + assert any(record.body.string_value == "application log" for record in get_logs(telemetry)) + application_scope_logs = extract_leaves(telemetry, "log_requests", "pbreq", "resource_logs", "scope_logs") + assert all(scope.scope.name != "otel.profiling" for scope in application_scope_logs) + + profile_telemetry = self.profile_handler.telemetry + profile_scope_logs = extract_leaves(profile_telemetry, "log_requests", "pbreq", "resource_logs", "scope_logs") + assert profile_scope_logs + assert all(scope.scope.name == "otel.profiling" and scope.log_records for scope in profile_scope_logs) + for request in profile_telemetry.log_requests: + headers = {key.lower(): value for key, value in request.headers.items()} + assert headers["x-profile-test"] == "yaml-token" diff --git a/tests/integration/declarative_default_profile_exporter.py b/tests/integration/declarative_default_profile_exporter.py new file mode 100644 index 00000000..dcdaed2f --- /dev/null +++ b/tests/integration/declarative_default_profile_exporter.py @@ -0,0 +1,88 @@ +import base64 +import gzip +import importlib.util +from pathlib import Path + +from lib import project_path, trace_loop + + +if __name__ == "__main__": + trace_loop(3) + + +class DeclarativeDefaultProfileExporterOtelTest: + def requirements(self): + return (project_path(),) + + def declarative_configuration(self): + return """ +file_format: "1.0" +resource: + attributes: + - name: service.name + value: yaml-profile-service +distribution: + splunk: + profiling: + always_on: + cpu_profiler: + sampling_interval: 250 +""" + + def environment_variables(self): + return { + "OTEL_SERVICE_NAME": "environment-service", + "SPLUNK_PROFILER_ENABLED": "false", + "SPLUNK_PROFILER_CALL_STACK_INTERVAL": "9999", + "SPLUNK_PROFILER_LOGS_ENDPOINT": "http://localhost:1/unused", + "OTEL_EXPORTER_OTLP_LOGS_ENDPOINT": "http://localhost:1/unused", + "OTEL_PYTHON_DISABLED_INSTRUMENTATIONS": "system_metrics", + } + + def wrapper_command(self): + return "opentelemetry-instrument" + + def is_http(self): + return True + + def on_start(self): + pass + + def on_stop(self, telemetry, stdout, stderr, returncode): + from oteltest.telemetry import extract_leaves, get_attribute + + profile_proto_path = Path(project_path(), "src", "splunk_otel", "profile_pb2.py") + spec = importlib.util.spec_from_file_location("profile_pb2", profile_proto_path) + profile_pb2 = importlib.util.module_from_spec(spec) + spec.loader.exec_module(profile_pb2) + + assert returncode == 0, stderr + resource_log_groups = extract_leaves(telemetry, "log_requests", "pbreq", "resource_logs") + profile_records = [] + for resource_logs in resource_log_groups: + for scope_logs in resource_logs.scope_logs: + if scope_logs.scope.name != "otel.profiling": + continue + assert ( + get_attribute(resource_logs.resource.attributes, "service.name").value.string_value + == "yaml-profile-service" + ) + assert ( + get_attribute(resource_logs.resource.attributes, "telemetry.distro.name").value.string_value + == "splunk-opentelemetry" + ) + profile_records.extend(scope_logs.log_records) + + assert profile_records + for record in profile_records: + assert ( + get_attribute(record.attributes, "profiling.instrumentation.source").value.string_value == "continuous" + ) + profile = profile_pb2.Profile() + profile.ParseFromString(gzip.decompress(base64.b64decode(record.body.string_value))) + assert profile.sample + for sample in profile.sample: + (sample_period_millis,) = [ + label.num for label in sample.label if profile.string_table[label.key] == "source.event.period" + ] + assert sample_period_millis == 250 diff --git a/tests/test_profile.py b/tests/test_profile.py index 729403b1..20e70af3 100644 --- a/tests/test_profile.py +++ b/tests/test_profile.py @@ -3,6 +3,7 @@ import json import random import time +from threading import Event, Thread from collections import OrderedDict from os.path import abspath, dirname @@ -11,6 +12,8 @@ from opentelemetry._logs import Logger from opentelemetry.context import attach, detach from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, LogRecordExportResult from opentelemetry.trace import ( NonRecordingSpan, SpanContext, @@ -25,6 +28,7 @@ _pb_profile_to_str, _ProfileScraper, _stacktraces_to_cpu_profile, + _start_continuous_profiling, _StringTable, ) @@ -158,26 +162,196 @@ def test_interval_timer_stop_before_start(): timer.stop() -def test_interval_timer_pause_and_resume(): - # pause_after() should halt ticking; start() should resume it. - ticks = [] - timer = _IntervalTimer(20, lambda: ticks.append(1)) +def test_interval_timer_stop_interrupts_long_sampling_interval(): + sampled = Event() + timer = _IntervalTimer(60000, sampled.set) timer.start() + assert sampled.wait(timeout=2) + + stop_thread = Thread(target=timer.stop, daemon=True) + stop_thread.start() + stop_thread.join(timeout=2) + + assert not stop_thread.is_alive(), "stop must interrupt the sampling wait" + + +@pytest.mark.parametrize("pause_state", ["paused", "entering_pause"]) +@pytest.mark.parametrize("pause_action", ["stop", "resume"]) +def test_interval_timer_stop_or_resume_during_pause(monkeypatch, pause_state, pause_action): + pause_reached = Event() + release_pause = Event() + wakeup_requested = Event() + resume_requested = Event() + sampled_while_paused = Event() + resumed = Event() + + def _fake_sample(): + if pause_reached.is_set(): + if not resume_requested.is_set(): + sampled_while_paused.set() + resumed.set() + else: + timer.pause_after(0) + + timer = _IntervalTimer(0, _fake_sample) + wakeup_clear = timer.wakeup_event.clear + wakeup_wait = timer.wakeup_event.wait + wakeup_set = timer.wakeup_event.set + + def _fake_wakeup_clear(): + if pause_state == "entering_pause": + pause_reached.set() + assert release_pause.wait(timeout=5) + wakeup_clear() + + def _fake_wakeup_wait(): + if pause_state == "paused": + pause_reached.set() + return wakeup_wait() + + def _fake_wakeup_set(): + wakeup_set() + wakeup_requested.set() + + monkeypatch.setattr(timer.wakeup_event, "clear", _fake_wakeup_clear) + monkeypatch.setattr(timer.wakeup_event, "wait", _fake_wakeup_wait) + monkeypatch.setattr(timer.wakeup_event, "set", _fake_wakeup_set) + timer.start() + try: + assert pause_reached.wait(timeout=2) + assert not resumed.is_set(), "sampling must stay paused until resume is requested" + wakeup_requested.clear() + if pause_action == "stop": + stop_thread = Thread(target=timer.stop, daemon=True) + stop_thread.start() + assert wakeup_requested.wait(timeout=2) + release_pause.set() + stop_thread.join(timeout=2) + assert not stop_thread.is_alive(), "stop must wake a paused sampler" + else: + resume_requested.set() + timer.start() + release_pause.set() + assert resumed.wait(timeout=2), "resume must wake a paused sampler" + assert not sampled_while_paused.is_set(), "sampling must stay paused until resume is requested" + finally: + release_pause.set() + wakeup_set() + timer.stop() - time.sleep(0.1) - assert len(ticks) > 0, "timer should have ticked before pause" - timer.pause_after(0) - time.sleep(0.2) # enough for the timer to reach pause_at and block - count_at_pause = len(ticks) - time.sleep(0.1) # confirm no new ticks while paused - assert len(ticks) == count_at_pause, "timer should not tick while paused" +def test_interval_timer_stop_waits_for_current_sample(monkeypatch): + sample_started = Event() + release_sample = Event() + stop_waiting = Event() + lifecycle = [] - timer.start() - time.sleep(0.1) # let it resume - assert len(ticks) > count_at_pause, "timer should tick again after start()" + def _fake_sample(): + sample_started.set() + assert release_sample.wait(timeout=5) + lifecycle.append("sample completed") - timer.stop() + timer = _IntervalTimer(60000, _fake_sample) + sampler_join = timer.thread.join + + def _fake_sampler_join(timeout=None): + stop_waiting.set() + sampler_join(timeout) + + def _stop_timer(): + timer.stop() + lifecycle.append("stop completed") + + monkeypatch.setattr(timer.thread, "join", _fake_sampler_join) + timer.start() + try: + assert sample_started.wait(timeout=2) + stop_thread = Thread(target=_stop_timer, daemon=True) + stop_thread.start() + assert stop_waiting.wait(timeout=2) + assert stop_thread.is_alive() + assert not lifecycle + release_sample.set() + stop_thread.join(timeout=2) + assert not stop_thread.is_alive() + assert lifecycle == ["sample completed", "stop completed"] + finally: + release_sample.set() + timer.stop() + + +def test_exit_flushes_final_profile_after_sampling_stops(monkeypatch): + shutdown_callbacks = [] + sample_started = Event() + release_sample = Event() + stop_waiting = Event() + exported_records = [] + lifecycle = [] + timers = [] + + def _fake_register_shutdown(callback): + shutdown_callbacks.append(callback) + return callback + + def _fake_interval_timer(interval_millis, target): + timer = _IntervalTimer(interval_millis, target) + timers.append(timer) + return timer + + class FakeLogExporter: + def export(self, records): + exported_records.extend(records) + lifecycle.append("sample exported") + return LogRecordExportResult.SUCCESS + + def shutdown(self): + lifecycle.append("exporter shut down") + + monkeypatch.setattr("splunk_otel.profile._atexit.register", _fake_register_shutdown) + monkeypatch.setattr("splunk_otel.profile._IntervalTimer", _fake_interval_timer) + resource = Resource({"service.name": "final-profile-test"}) + provider = LoggerProvider(resource=resource) + provider.add_log_record_processor(BatchLogRecordProcessor(FakeLogExporter(), schedule_delay_millis=60000)) + logger = provider.get_logger("otel.profiling") + + class FakeProfilingLogger: + def emit(self, record): + sample_started.set() + assert release_sample.wait(timeout=5) + logger.emit(record) + lifecycle.append("sample emitted") + + context = _start_continuous_profiling(resource=resource, interval_millis=60000, logger=FakeProfilingLogger()) + (timer,) = timers + sampler_join = timer.thread.join + + def _fake_sampler_join(timeout=None): + stop_waiting.set() + sampler_join(timeout) + + def _release_final_sample(): + if stop_waiting.wait(timeout=5): + release_sample.set() + + monkeypatch.setattr(timer.thread, "join", _fake_sampler_join) + release_thread = Thread(target=_release_final_sample, daemon=True) + release_thread.start() + try: + assert sample_started.wait(timeout=2) + assert not exported_records + for shutdown in reversed(shutdown_callbacks): + shutdown() + release_thread.join(timeout=2) + assert not release_thread.is_alive() + assert lifecycle == ["sample emitted", "sample exported", "exporter shut down"] + (record,) = exported_records + assert record.log_record.attributes["profiling.instrumentation.source"] == "continuous" + assert _pb_profile_from_str(record.log_record.body).sample + finally: + release_sample.set() + context.stop() + provider.shutdown() + release_thread.join(timeout=2) class _FakeLogger(Logger): diff --git a/tests/test_profiling_configuration.py b/tests/test_profiling_configuration.py new file mode 100644 index 00000000..f710a7c7 --- /dev/null +++ b/tests/test_profiling_configuration.py @@ -0,0 +1,202 @@ +import pytest +from opentelemetry._logs import NoOpLoggerProvider +from opentelemetry.configuration import ConfigurationError +from opentelemetry.sdk.resources import Resource + +from splunk_otel.env import Env +from splunk_otel.profile import start_profiling +from splunk_otel._declarative_profiling import _configure_declarative_continuous_profiling + + +@pytest.fixture +def logger_provider_configs(): + return [] + + +@pytest.fixture +def profiling_contexts(monkeypatch, logger_provider_configs): + contexts = [] + + class FakeProfilingContext: + def __init__(self, resource, interval_millis, logger): + self.resource = resource + self.interval_millis = interval_millis + self.logger = logger + self.started = False + contexts.append(self) + + def start(self): + self.started = True + + def stop(self): + self.started = False + + monkeypatch.setattr("splunk_otel.profile.ProfilingContext", FakeProfilingContext) + + def _fake_create_logger_provider(logger_provider_config, _resource): + logger_provider_configs.append(logger_provider_config) + return NoOpLoggerProvider() + + monkeypatch.setattr("splunk_otel._declarative_profiling.create_logger_provider", _fake_create_logger_provider) + return contexts + + +@pytest.mark.parametrize( + ("cpu_profiler_config", "interval_millis"), [({}, 1000), (None, 1000), ({"sampling_interval": 250}, 250)] +) +def test_yaml_starts_continuous_profiling(monkeypatch, profiling_contexts, cpu_profiler_config, interval_millis): + monkeypatch.setenv("SPLUNK_PROFILER_ENABLED", "false") + monkeypatch.setenv("SPLUNK_PROFILER_CALL_STACK_INTERVAL", "9999") + resource = Resource({"service.name": "yaml-service", "deployment.environment.name": "test"}) + + _configure_declarative_continuous_profiling( + {"splunk": {"profiling": {"always_on": {"cpu_profiler": cpu_profiler_config}}}}, resource + ) + + (context,) = profiling_contexts + assert context.started + assert context.interval_millis == interval_millis + assert context.resource is resource + + +def test_start_profiling_uses_supplied_environment(monkeypatch, profiling_contexts): + monkeypatch.setenv("OTEL_SERVICE_NAME", "process-service") + monkeypatch.setenv("SPLUNK_PROFILER_CALL_STACK_INTERVAL", "9999") + environment = Env({"OTEL_SERVICE_NAME": "environment-service", "SPLUNK_PROFILER_CALL_STACK_INTERVAL": "250"}) + + context = start_profiling(environment) + + assert profiling_contexts == [context] + assert context.started + assert context.interval_millis == 250 + assert context.resource.attributes["service.name"] == "environment-service" + + +def test_start_profiling_without_arguments_uses_process_environment(monkeypatch, profiling_contexts): + monkeypatch.setenv("OTEL_SERVICE_NAME", "process-service") + monkeypatch.setenv("SPLUNK_PROFILER_CALL_STACK_INTERVAL", "500") + monkeypatch.setenv("SPLUNK_PROFILER_ENABLED", "false") + + context = start_profiling() + + assert profiling_contexts == [context] + assert context.started + assert context.interval_millis == 500 + assert context.resource.attributes["service.name"] == "process-service" + + +@pytest.mark.parametrize("configuration_source", ["environment", "declarative"]) +def test_continuous_profiling_stops_at_exit(monkeypatch, profiling_contexts, configuration_source): + shutdown_callbacks = [] + + def _fake_register_shutdown(callback): + shutdown_callbacks.append(callback) + + monkeypatch.setattr("splunk_otel.profile._atexit.register", _fake_register_shutdown) + if configuration_source == "environment": + start_profiling(Env({"OTEL_SERVICE_NAME": "environment-service"})) + else: + _configure_declarative_continuous_profiling( + {"splunk": {"profiling": {"always_on": {"cpu_profiler": {}}}}}, Resource({}) + ) + + (context,) = profiling_contexts + assert context.started + (shutdown,) = shutdown_callbacks + shutdown() + assert not context.started + + +@pytest.mark.parametrize( + "distribution_config", + [ + None, + {}, + {"splunk": None}, + {"splunk": {"profiling": None}}, + {"splunk": {"profiling": {}}}, + {"splunk": {"profiling": {"always_on": None}}}, + {"splunk": {"profiling": {"always_on": {}}}}, + {"splunk": {"profiling": {"callgraphs": {}}}}, + {"splunk": {"profiling": {"exporter": {"otlp_log_grpc": {}}}}}, + ], +) +def test_yaml_without_cpu_profiler_does_not_start_profiling(monkeypatch, profiling_contexts, distribution_config): + monkeypatch.setenv("SPLUNK_PROFILER_ENABLED", "true") + + _configure_declarative_continuous_profiling(distribution_config, Resource({})) + + assert not profiling_contexts + + +@pytest.mark.parametrize("http_exporter_config", [None, {}, {"endpoint": None}, {"endpoint": ""}]) +def test_unset_profile_endpoint_uses_sdk_default( + monkeypatch, profiling_contexts, logger_provider_configs, http_exporter_config +): + monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:1/base") + monkeypatch.setenv("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT", "http://localhost:1/logs") + + _configure_declarative_continuous_profiling( + { + "splunk": { + "profiling": { + "always_on": {"cpu_profiler": {}}, + "exporter": {"otlp_log_http": http_exporter_config}, + } + } + }, + Resource({}), + ) + + (logger_provider_config,) = logger_provider_configs + assert logger_provider_config.processors[0].batch.exporter.otlp_http.endpoint == "http://localhost:4318/v1/logs" + assert profiling_contexts[0].started + + +@pytest.mark.parametrize("interval_millis", [0, -1, True, 1.0, 1.5, "1000", None]) +def test_invalid_sampling_interval_fails_before_starting(profiling_contexts, interval_millis): + with pytest.raises( + ConfigurationError, + match=r"distribution\.splunk\.profiling\.always_on\.cpu_profiler\.sampling_interval", + ): + _configure_declarative_continuous_profiling( + {"splunk": {"profiling": {"always_on": {"cpu_profiler": {"sampling_interval": interval_millis}}}}}, + Resource({}), + ) + + assert not profiling_contexts + + +@pytest.mark.parametrize("key", ["splunk", "profiling", "always_on", "cpu_profiler"]) +def test_invalid_profiling_block_fails_before_starting(profiling_contexts, key): + distribution_config = {key: False} + keys = ["splunk", "profiling", "always_on", "cpu_profiler"] + for parent in reversed(keys[: keys.index(key)]): + distribution_config = {parent: distribution_config} + + with pytest.raises(ConfigurationError, match=key): + _configure_declarative_continuous_profiling(distribution_config, Resource({})) + + assert not profiling_contexts + + +@pytest.mark.parametrize( + "exporter_config", + [ + False, + {"otlp_log_grpc": {}}, + {"otlp_log_http": False}, + {"otlp_log_http": {}, "console": {}}, + {"otlp_log_http": {"endpoint": 123}}, + {"otlp_log_http": {"headers": "invalid"}}, + {"otlp_log_http": {"timeout": "invalid"}}, + ], +) +def test_invalid_profile_exporter_fails_before_starting(profiling_contexts, exporter_config): + with pytest.raises(ConfigurationError): + _configure_declarative_continuous_profiling( + {"splunk": {"profiling": {"exporter": exporter_config, "always_on": {"cpu_profiler": {}}}}}, + Resource({}), + ) + + assert not profiling_contexts