"""Test the telemetry that should be emitted from the library."""

import base64

import json

import time

import uuid

import pytest

from .conftest import StableConfigWriter

from utils.telemetry_utils import TelemetryUtils

from utils import context, scenarios, rfc, features, logger

from utils.docker_fixtures import TestAgentAPI

from .conftest import APMLibrary

telemetry_name_mapping: dict[str, dict[str, str | list[str]]] = {

"instrumentation_source": {

"java": "DD_INSTRUMENTATION_SOURCE",

"nodejs": "instrumentationSource",

},

"ssi_injection_enabled": {

"python": "DD_INJECTION_ENABLED",

"java": "DD_INJECTION_ENABLED",

"ruby": "DD_INJECTION_ENABLED",

"nodejs": "DD_INJECTION_ENABLED",

"golang": ["DD_INJECTION_ENABLED", "injection_enabled"],

},

"ssi_forced_injection_enabled": {

"python": "DD_INJECT_FORCE",

"ruby": "DD_INJECT_FORCE",

"java": "DD_INJECT_FORCE",

"nodejs": "DD_INJECT_FORCE",

"golang": ["DD_INJECT_FORCE", "inject_force"],

},

"trace_sample_rate": {

"dotnet": "DD_TRACE_SAMPLE_RATE",

"java": "DD_TRACE_SAMPLE_RATE",

"nodejs": "DD_TRACE_SAMPLE_RATE",

"python": "DD_TRACE_SAMPLE_RATE",

"ruby": "DD_TRACE_SAMPLE_RATE",

"golang": ["DD_TRACE_SAMPLE_RATE", "trace_sample_rate"],

},

"logs_injection_enabled": {

"dotnet": "DD_LOGS_INJECTION",

"nodejs": "DD_LOGS_INJECTION",

"python": "DD_LOGS_INJECTION",

"php": "DD_TRACE_LOGS_ENABLED",

"ruby": "DD_LOGS_INJECTION",

"golang": ["DD_LOGS_INJECTION", "trace.logs_enabled"],

"java": "DD_LOGS_INJECTION_ENABLED",

},

"trace_header_tags": {

"dotnet": "DD_TRACE_HEADER_TAGS",

"nodejs": "DD_TRACE_HEADER_TAGS",

"python": "DD_TRACE_HEADER_TAGS",

"golang": ["DD_TRACE_HEADER_TAGS", "trace_header_tags"],

"java": "DD_TRACE_HEADER_TAGS",

"ruby": "DD_TRACE_HEADER_TAGS",

},

"trace_tags": {

"dotnet": "DD_TAGS",

"java": "DD_TRACE_TAGS",

"nodejs": "DD_TAGS",

"python": "DD_TAGS",

"golang": ["DD_TAGS", "trace_tags"],

"ruby": "DD_TAGS",

},

"trace_enabled": {

"dotnet": "DD_TRACE_ENABLED",

"java": "DD_TRACE_ENABLED",

"nodejs": "DD_TRACE_ENABLED",

"python": "DD_TRACE_ENABLED",

"ruby": "DD_TRACE_ENABLED",

"golang": ["DD_TRACE_ENABLED", "trace_enabled"],

},

"profiling_enabled": {

"dotnet": "DD_PROFILING_ENABLED",

"nodejs": "DD_PROFILING_ENABLED",

"python": "DD_PROFILING_ENABLED",

"ruby": "DD_PROFILING_ENABLED",

"golang": ["DD_PROFILING_ENABLED", "profiling_enabled"],

"java": "DD_PROFILING_ENABLED",

},

"appsec_enabled": {

"dotnet": "DD_APPSEC_ENABLED",

"nodejs": "DD_APPSEC_ENABLED",

"python": "DD_APPSEC_ENABLED",

"ruby": "DD_APPSEC_ENABLED",

"golang": ["DD_APPSEC_ENABLED", "appsec_enabled"],

"java": "DD_APPSEC_ENABLED",

},

"data_streams_enabled": {

"dotnet": "DD_DATA_STREAMS_ENABLED",

"nodejs": "DD_DATA_STREAMS_ENABLED",

"python": "DD_DATA_STREAMS_ENABLED",

"java": "DD_DATA_STREAMS_ENABLED",

"golang": ["DD_DATA_STREAMS_ENABLED", "data_streams_enabled"],

"ruby": "DD_DATA_STREAMS_ENABLED",

},

"runtime_metrics_enabled": {

"java": "DD_RUNTIME_METRICS_ENABLED",

"dotnet": "DD_RUNTIME_METRICS_ENABLED",

"nodejs": "DD_RUNTIME_METRICS_ENABLED",

"python": "DD_RUNTIME_METRICS_ENABLED",

"ruby": "DD_RUNTIME_METRICS_ENABLED",

"golang": ["DD_RUNTIME_METRICS_ENABLED", "runtime_metrics_enabled"],

},

"dynamic_instrumentation_enabled": {

"java": "DD_DYNAMIC_INSTRUMENTATION_ENABLED",

"dotnet": "DD_DYNAMIC_INSTRUMENTATION_ENABLED",

"nodejs": "DD_DYNAMIC_INSTRUMENTATION_ENABLED",

"python": "DD_DYNAMIC_INSTRUMENTATION_ENABLED",

"php": "DD_DYNAMIC_INSTRUMENTATION_ENABLED",

"ruby": "DD_DYNAMIC_INSTRUMENTATION_ENABLED",

"golang": ["DD_DYNAMIC_INSTRUMENTATION_ENABLED", "dynamic_instrumentation_enabled"],

},

"trace_debug_enabled": {

"php": "DD_TRACE_DEBUG",

"java": "DD_TRACE_DEBUG",

"ruby": "DD_TRACE_DEBUG",

"python": "DD_TRACE_DEBUG",

"golang": ["trace_debug_enabled", "DD_TRACE_DEBUG"],

},

"tags": {

"java": "DD_TRACE_TAGS",

"dotnet": "DD_TAGS",

"python": "DD_TAGS",

"nodejs": "DD_TAGS",

"golang": ["DD_TAGS", "trace_tags"],

"ruby": "DD_TAGS",

},

"trace_propagation_style": {

"java": "DD_TRACE_PROPAGATION_STYLE",

"dotnet": "DD_TRACE_PROPAGATION_STYLE",

"php": "DD_TRACE_PROPAGATION_STYLE",

"golang": ["DD_TRACE_PROPAGATION_STYLE", "trace.propagation_style"],

"ruby": "DD_TRACE_PROPAGATION_STYLE",

},

}

def _mapped_telemetry_name(apm_telemetry_name: str) -> list[str]:

if apm_telemetry_name in telemetry_name_mapping:

lang_mapping = telemetry_name_mapping[apm_telemetry_name]

mapped_name = lang_mapping.get(context.library.name)

if mapped_name is not None:

if isinstance(mapped_name, list):

return mapped_name

return [mapped_name]

return [apm_telemetry_name]

def _check_propagation_style_with_inject_and_extract(

test_agent: TestAgentAPI,

configuration_by_name: dict,

expected_origin: str,

library_name: str,

*,

allow_calculated_origin: bool = False,

) -> None:

"""Check both inject and extract propagation style keys for languages that report them separately.

Some libraries report propagation style using separate inject and extract keys

instead of a single combined key. This function validates that both keys exist with the

expected origin and have non-empty values.

When a library derives the inject/extract values from DD_TRACE_PROPAGATION_STYLE,

the reported origin may be "calculated" instead of the original source origin.

In that case, allow_calculated_origin can be used to accept either origin.

"""

allowed_origins = [expected_origin]

if allow_calculated_origin:

allowed_origins.append("calculated")

if library_name in ("python", "ruby", "nodejs"):

keys = ["DD_TRACE_PROPAGATION_STYLE_INJECT", "DD_TRACE_PROPAGATION_STYLE_EXTRACT"]

else:

raise ValueError(f"Unsupported library for inject/extract propagation style: {library_name}")

def _assert_config_with_allowed_origins(config_name: str, origins: list[str]) -> None:

config_item = next(

(

item

for origin in origins

if (item := test_agent.get_telemetry_config_by_origin(configuration_by_name, config_name, origin))

is not None

),

None,

)

assert config_item is not None, (

f"No configuration found for '{config_name}' with any origin in {origins}. Full configuration_by_name: {configuration_by_name}"

)

assert isinstance(config_item, dict)

assert config_item["origin"] in origins, (

f"Origin mismatch for {config_item}. Expected one of {origins}, Actual origin: '{config_item.get('origin', '<missing>')}'"

)

assert config_item["value"], f"Expected non-empty value for '{config_name}'"

# The combined key is not asserted here: libraries that derive inject/extract

# may still report it with origin "default" (their built-in default value),

# which is not a wrong origin to flag, just a different reporting shape.

for key in keys:

_assert_config_with_allowed_origins(key, allowed_origins)

@scenarios.parametric

@rfc("https://docs.google.com/document/d/1In4TfVBbKEztLzYg4g0si5H56uzAbYB3OfqzRGP2xhg/edit")

@features.telemetry_app_started_event

class Test_Defaults:

"""Clients should use and report the same default values for features."""

@pytest.mark.parametrize(

"library_env",

[

{

# Decrease the heartbeat/poll intervals to speed up the tests

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.1",

}

],

)

def test_library_settings(self, test_agent: TestAgentAPI, test_library: APMLibrary):

with test_library.dd_start_span("test"):

pass

configuration_by_name = test_agent.wait_for_telemetry_configurations()

# DSM is enabled by default in .NET, but not in other languages

# see https://github.com/DataDog/dd-trace-dotnet/pull/7244 for more details

if context.library >= "[email protected]":

data_streams_enabled = ("true", True)

else:

data_streams_enabled = ("false", False)

for apm_telemetry_name, value in [

("trace_sample_rate", (1.0, None, "1.0")),

("logs_injection_enabled", ("false", False, "true", True, "structured")),

("trace_header_tags", ""),

("trace_tags", ("_dd.svc_src:m", "")),

("trace_enabled", ("true", True)),

("profiling_enabled", ("false", False, None)),

("appsec_enabled", ("false", False, "inactive", None)),

("data_streams_enabled", data_streams_enabled),

]:

# The Go tracer does not support logs injection.

if context.library == "golang" and apm_telemetry_name in ("logs_injection_enabled",):

continue

if context.library == "cpp" and apm_telemetry_name in (

"logs_injection_enabled",

"trace_header_tags",

"profiling_enabled",

"appsec_enabled",

"data_streams_enabled",

"trace_sample_rate",

):

continue

if context.library == "python" and apm_telemetry_name in ("trace_sample_rate",):

# DD_TRACE_SAMPLE_RATE is not supported in ddtrace>=3.x

continue

mapped_apm_telemetry_names = _mapped_telemetry_name(apm_telemetry_name)

cfg_item = None

matched_name = None

for mapped_name in mapped_apm_telemetry_names:

cfg_item = test_agent.get_telemetry_config_by_origin(configuration_by_name, mapped_name, "default")

if cfg_item is not None:

matched_name = mapped_name

break

assert cfg_item is not None, (

f"No configuration found for any of {' or '.join(mapped_apm_telemetry_names)} with origin 'default'"

)

assert isinstance(cfg_item, dict)

if isinstance(value, tuple):

assert cfg_item.get("value") in value, f"Unexpected value for '{matched_name}' ('{context.library}')"

else:

assert cfg_item.get("value") == value, f"Unexpected value for '{matched_name}'"

assert cfg_item.get("origin") == "default", f"Unexpected origin for '{matched_name}'"

@scenarios.parametric

@rfc("https://docs.google.com/document/d/1kI-gTAKghfcwI7YzKhqRv2ExUstcHqADIWA4-TZ387o")

# To pass this test, ensure the lang you are testing has the necessary mapping in its config_rules.json file: https://github.com/DataDog/dd-go/tree/prod/trace/apps/tracer-telemetry-intake/telemetry-payload/static

# And replace the `missing_feature` marker under the lang's manifest file, for Test_Consistent_Configs

@features.telemetry_configurations_collected

class Test_Consistent_Configs:

"""Clients should report modifications to features."""

@pytest.mark.parametrize(

"library_env",

[

{

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.1", # Decrease the heartbeat/poll intervals to speed up the tests

"DD_ENV": "dev",

"DD_SERVICE": "service_test",

"DD_VERSION": "5.2.0",

"DD_TRACE_RATE_LIMIT": 10,

"DD_TRACE_HEADER_TAGS": "User-Agent:my-user-agent,Content-Type.",

"DD_TRACE_ENABLED": "true",

"DD_TRACE_OBFUSCATION_QUERY_STRING_REGEXP": r"\d{3}-\d{2}-\d{4}",

"DD_TRACE_LOG_DIRECTORY": "/some/temporary/directory",

"DD_TRACE_CLIENT_IP_HEADER": "random-header-name",

"DD_TRACE_HTTP_CLIENT_ERROR_STATUSES": "200-250",

"DD_TRACE_HTTP_SERVER_ERROR_STATUSES": "250-200",

"DD_TRACE_HTTP_CLIENT_TAG_QUERY_STRING": "false",

# "DD_TRACE_AGENT_URL": "http://localhost:8126", # Don't want to configure this, since we need tracer <> agent connection to run these tests!

# "DD_TRACE_<INTEGRATION>_ENABLED": "N/A", # Skipping because it is blocked by the telemetry intake & this information is already collected through other (non-config) telemetry.

}

],

)

def test_library_settings(self, test_agent: TestAgentAPI, test_library: APMLibrary):

with test_library.dd_start_span("test"):

pass

configuration_by_name = test_agent.wait_for_telemetry_configurations()

# # Check that the tags name match the expected value

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_ENV", "env_var", return_value_only=True

)

== "dev"

)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_SERVICE", "env_var", return_value_only=True

)

== "service_test"

)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_VERSION", "env_var", return_value_only=True

)

== "5.2.0"

)

assert test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_RATE_LIMIT", "env_var", return_value_only=True

) in ("10", 10)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_HEADER_TAGS", "env_var", return_value_only=True

)

== "User-Agent:my-user-agent,Content-Type."

)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_ENABLED", "env_var", return_value_only=True

)

is True

)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_OBFUSCATION_QUERY_STRING_REGEXP", "env_var", return_value_only=True

)

== r"\d{3}-\d{2}-\d{4}"

)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_CLIENT_IP_HEADER", "env_var", return_value_only=True

)

== "random-header-name"

)

@pytest.mark.parametrize(

"library_env",

[

{

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.1", # Decrease the heartbeat/poll intervals to speed up the tests

"DD_TRACE_LOG_DIRECTORY": "/some/temporary/directory",

"DD_TRACE_HTTP_CLIENT_ERROR_STATUSES": "200-250",

"DD_TRACE_HTTP_SERVER_ERROR_STATUSES": "250-200",

"DD_TRACE_HTTP_CLIENT_TAG_QUERY_STRING": "false",

}

],

)

def test_library_settings_2(self, test_agent: TestAgentAPI, test_library: APMLibrary):

with test_library.dd_start_span("test"):

pass

configuration_by_name = test_agent.wait_for_telemetry_configurations()

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_LOG_DIRECTORY", "env_var", return_value_only=True

)

== "/some/temporary/directory"

)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_HTTP_CLIENT_ERROR_STATUSES", "env_var", return_value_only=True

)

== "200-250"

)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_HTTP_SERVER_ERROR_STATUSES", "env_var", return_value_only=True

)

== "250-200"

)

assert (

test_agent.get_telemetry_config_by_origin(

configuration_by_name, "DD_TRACE_HTTP_CLIENT_TAG_QUERY_STRING", "env_var", return_value_only=True

)

is False

)

@scenarios.parametric

@rfc("https://docs.google.com/document/d/1In4TfVBbKEztLzYg4g0si5H56uzAbYB3OfqzRGP2xhg/edit")

@features.telemetry_app_started_event

class Test_Environment:

"""Clients should use and report the same environment values for features."""

@pytest.mark.parametrize(

"library_env",

[

{

# Decrease the heartbeat/poll intervals to speed up the tests

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.1",

"DD_TRACE_SAMPLE_RATE": "0.3",

"DD_LOGS_INJECTION": "true",

"DD_TRACE_HEADER_TAGS": "X-Header-Tag-1:header_tag_1,X-Header-Tag-2:header_tag_2",

"DD_TAGS": "team:apm,component:web",

"DD_TRACE_ENABLED": "true",

# node.js DD_TRACING_ENABLED is equivalent to DD_TRACE_ENABLED in other libraries

"DD_TRACING_ENABLED": "true",

"DD_PROFILING_ENABLED": "false",

"DD_APPSEC_ENABLED": "false",

"DD_DATA_STREAMS_ENABLED": "false",

}

],

)

def test_library_settings(self, test_agent: TestAgentAPI, test_library: APMLibrary):

with test_library.dd_start_span("test"):

pass

configuration_by_name = test_agent.wait_for_telemetry_configurations()

for apm_telemetry_name, environment_value in [

("trace_sample_rate", ("0.3", 0.3)),

("logs_injection_enabled", ("true", True)),

(

"trace_header_tags",

(

"X-Header-Tag-1:header_tag_1,X-Header-Tag-2:header_tag_2",

"x-header-tag-1:header_tag_1,x-header-tag-2:header_tag_2",

),

),

("trace_tags", ("team:apm,component:web", "component:web,team:apm")),

("trace_enabled", ("true", True)),

("profiling_enabled", ("false", False)),

("appsec_enabled", ("false", False)),

("data_streams_enabled", ("false", False)),

]:

# The Go tracer does not support logs injection.

if context.library == "golang" and apm_telemetry_name in ("logs_injection_enabled",):

continue

if context.library == "cpp" and apm_telemetry_name in (

"logs_injection_enabled",

"trace_header_tags",

"profiling_enabled",

"appsec_enabled",

"data_streams_enabled",

):

continue

if context.library == "python" and apm_telemetry_name in ("trace_sample_rate",):

# DD_TRACE_SAMPLE_RATE is not supported in ddtrace>=3.x

continue

mapped_apm_telemetry_names = _mapped_telemetry_name(apm_telemetry_name)

cfg_item = None

matched_name = None

for mapped_name in mapped_apm_telemetry_names:

cfg_item = test_agent.get_telemetry_config_by_origin(configuration_by_name, mapped_name, "env_var")

if cfg_item is not None:

matched_name = mapped_name

break

assert cfg_item is not None, (

f"Missing telemetry config item for any of {' or '.join(mapped_apm_telemetry_names)}"

)

assert isinstance(cfg_item, dict)

if isinstance(environment_value, tuple):

assert cfg_item.get("value") in environment_value, f"Unexpected value for '{matched_name}'"

else:

assert cfg_item.get("value") == environment_value, f"Unexpected value for '{matched_name}'"

assert cfg_item.get("origin") == "env_var", f"Unexpected origin for '{matched_name}'"

@pytest.mark.parametrize(

"library_env",

[

{

"DD_TRACE_AGENT_PORT": "agent.port",

"DD_TRACE_OTEL_ENABLED": 1,

"DD_TELEMETRY_HEARTBEAT_INTERVAL": 1,

"TIMEOUT": 1500,

"DD_SERVICE": "service",

"OTEL_SERVICE_NAME": "otel_service",

"DD_TRACE_LOG_LEVEL": "error",

"DD_LOG_LEVEL": "error",

"OTEL_LOG_LEVEL": "debug",

# python tracer supports DD_TRACE_SAMPLING_RULES not DD_TRACE_SAMPLE_RATE

"DD_TRACE_SAMPLING_RULES": '[{"sample_rate":0.5}]',

"DD_TRACE_SAMPLE_RATE": "0.5",

"OTEL_TRACES_SAMPLER": "traceidratio",

"OTEL_TRACES_SAMPLER_ARG": "0.1",

"DD_TRACE_ENABLED": "true",

"OTEL_TRACES_EXPORTER": "none",

"DD_RUNTIME_METRICS_ENABLED": "true",

"OTEL_METRICS_EXPORTER": "none",

"DD_TAGS": "foo:bar,baz:qux",

"OTEL_RESOURCE_ATTRIBUTES": "foo=bar1,baz=qux1",

"DD_TRACE_PROPAGATION_STYLE": "datadog",

"OTEL_PROPAGATORS": "datadog,tracecontext",

"OTEL_LOGS_EXPORTER": "none",

"OTEL_SDK_DISABLED": "false",

}

],

)

def test_telemetry_otel_env_hiding(self, test_agent: TestAgentAPI, test_library: APMLibrary):

with test_library.dd_start_span("test"):

pass

event = test_agent.wait_for_telemetry_event("generate-metrics", wait_loops=400)

payload = event["payload"]

assert event["request_type"] == "generate-metrics"

metrics = payload["series"]

assert payload["namespace"] == "tracers"

otel_hiding = [s for s in metrics if s["metric"] == "otel.env.hiding"]

assert not [s for s in metrics if s["metric"] == "otel.env.invalid"]

if context.library == "nodejs":

ddlog_config = "dd_trace_log_level"

elif context.library == "python":

ddlog_config = "dd_trace_debug"

else:

ddlog_config = "dd_log_level"

if context.library == "python":

otelsampler_config = "otel_traces_sampler"

else:

otelsampler_config = "otel_traces_sampler_arg"

if context.library == "python":

ddsampling_config = "dd_trace_sampling_rules"

else:

ddsampling_config = "dd_trace_sample_rate"

dd_to_otel_mapping: list[list[str | None]] = [

["dd_trace_propagation_style", "otel_propagators"],

["dd_service", "otel_service_name"],

[ddsampling_config, "otel_traces_sampler"],

["dd_trace_enabled", "otel_traces_exporter"],

["dd_runtime_metrics_enabled", "otel_metrics_exporter"],

["dd_tags", "otel_resource_attributes"],

["dd_trace_otel_enabled", "otel_sdk_disabled"],

[ddlog_config, "otel_log_level"],

[ddsampling_config, otelsampler_config],

]

for dd_config, otel_config in dd_to_otel_mapping:

for metric in otel_hiding:

if (

f"config_datadog:{dd_config}" in metric["tags"]

and f"config_opentelemetry:{otel_config}" in metric["tags"]

):

assert metric["points"][0][1] == 1

break

else:

pytest.fail(

f"Could not find a metric with {dd_config} and {otel_config} in otelHiding metrics: {otel_hiding}"

)

@pytest.mark.parametrize(

"library_env",

[

{

"DD_TRACE_AGENT_PORT": "agent.port",

"DD_TELEMETRY_HEARTBEAT_INTERVAL": 1,

"TIMEOUT": 1500,

"OTEL_SERVICE_NAME": "otel_service",

"OTEL_LOG_LEVEL": "foo",

"OTEL_TRACES_SAMPLER": "foo",

"OTEL_TRACES_SAMPLER_ARG": "foo",

"OTEL_TRACES_EXPORTER": "foo",

"OTEL_METRICS_EXPORTER": "foo",

"OTEL_RESOURCE_ATTRIBUTES": "foo",

"OTEL_PROPAGATORS": "foo",

"OTEL_LOGS_EXPORTER": "foo",

"DD_TRACE_OTEL_ENABLED": None,

"DD_TRACE_DEBUG": None,

"OTEL_SDK_DISABLED": "foo",

}

],

)

def test_telemetry_otel_env_invalid(self, test_agent: TestAgentAPI, test_library: APMLibrary):

with test_library.dd_start_span("test"):

pass

event = test_agent.wait_for_telemetry_event("generate-metrics", wait_loops=400)

payload = event["payload"]

assert event["request_type"] == "generate-metrics"

metrics = payload["series"]

assert payload["namespace"] == "tracers"

otel_invalid = [s for s in metrics if s["metric"] == "otel.env.invalid"]

if context.library == "nodejs":

ddlog_config = "dd_trace_log_level"

elif context.library == "python":

ddlog_config = "dd_trace_debug"

else:

ddlog_config = "dd_log_level"

if context.library == "python":

otelsampler_config = "otel_traces_sampler"

else:

otelsampler_config = "otel_traces_sampler_arg"

if context.library == "python":

ddsampling_config = "dd_trace_sampling_rules"

else:

ddsampling_config = "dd_trace_sample_rate"

dd_to_otel_mapping: list[list[str | None]] = [

["dd_trace_propagation_style", "otel_propagators"],

[ddsampling_config, "otel_traces_sampler"],

["dd_trace_enabled", "otel_traces_exporter"],

["dd_runtime_metrics_enabled", "otel_metrics_exporter"],

["dd_tags", "otel_resource_attributes"],

["dd_trace_otel_enabled", "otel_sdk_disabled"],

[ddlog_config, "otel_log_level"],

[ddsampling_config, otelsampler_config],

[None, "otel_logs_exporter"],

]

for dd_config, otel_config in dd_to_otel_mapping:

for metric in otel_invalid:

if (

dd_config is None or f"config_datadog:{dd_config}" in metric["tags"]

) and f"config_opentelemetry:{otel_config}" in metric["tags"]:

assert metric["points"][0][1] == 1

break

else:

pytest.fail(

f"Could not find a metric with {dd_config} and {otel_config} in otel_invalid metrics: {otel_invalid}"

)

@scenarios.parametric

@features.stable_configuration_support

@rfc("https://docs.google.com/document/d/1MNI5d3g6R8uU3FEWf2e08aAsFcJDVhweCPMjQatEb0o")

class Test_Stable_Configuration_Origin(StableConfigWriter):

"""Clients should report origin of configurations set by stable configuration faithfully"""

@pytest.mark.parametrize(

("local_cfg", "library_env", "fleet_cfg", "expected_origins"),

[

(

{

"DD_LOGS_INJECTION": True,

"DD_RUNTIME_METRICS_ENABLED": True,

"DD_DYNAMIC_INSTRUMENTATION_ENABLED": True,

},

{

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.1", # Decrease the heartbeat/poll intervals to speed up the tests

"DD_RUNTIME_METRICS_ENABLED": True,

},

{"DD_LOGS_INJECTION": True},

{

"logs_injection_enabled": "fleet_stable_config",

# Reporting for other origins than stable config is not completely implemented

# "runtime_metrics_enabled": "env_var",

"dynamic_instrumentation_enabled": "local_stable_config",

},

)

],

)

def test_stable_configuration_origin(

self,

local_cfg: dict[str, bool],

fleet_cfg: dict[str, bool],

test_agent: TestAgentAPI,

test_library: APMLibrary,

expected_origins: dict[str, str],

):

with test_library:

self.write_stable_config(

{

"apm_configuration_default": local_cfg,

},

"/etc/datadog-agent/application_monitoring.yaml",

test_library,

)

self.write_stable_config(

{

"apm_configuration_default": fleet_cfg,

},

"/etc/datadog-agent/managed/datadog-agent/stable/application_monitoring.yaml",

test_library,

)

# Sleep between telemetry events to ensure they are recorded with different timestamps, to later reorder them.

# seq_id can't be used to sort because payloads are sent from different tracer sessions.

time.sleep(1)

test_library.container_restart()

test_library.dd_start_span("test")

configuration_by_name = test_agent.wait_for_telemetry_configurations()

for cfg_name, expected_origin in expected_origins.items():

# The Go tracer does not support logs injection.

if context.library == "golang" and cfg_name == "logs_injection_enabled":

continue

apm_telemetry_names = _mapped_telemetry_name(cfg_name)

telemetry_item = None

for apm_name in apm_telemetry_names:

telemetry_item = test_agent.get_telemetry_config_by_origin(

configuration_by_name, apm_name, expected_origin

)

if telemetry_item is not None:

break

assert telemetry_item is not None, (

f"No configuration found for any of {' or '.join(apm_telemetry_names)} with origin '{expected_origin}'"

)

assert isinstance(telemetry_item, dict)

assert telemetry_item["origin"] == expected_origin, f"wrong origin for {telemetry_item}"

assert telemetry_item["value"]

@pytest.mark.parametrize(

("local_cfg", "library_env", "fleet_cfg", "fleet_config_id"),

[

(

{"DD_DYNAMIC_INSTRUMENTATION_ENABLED": True},

{

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.1", # Decrease the heartbeat/poll intervals to speed up the tests

},

{

"DD_TRACE_DEBUG": True,

},

"1231231231231",

)

],

)

# This is the specific test that's currently failing, but any test that calls _mapped_telemetry_name for debug mode, for golang, would fail.

def test_stable_configuration_config_id(

self,

local_cfg: dict[str, bool],

fleet_cfg: dict[str, bool],

test_agent: TestAgentAPI,

test_library: APMLibrary,

fleet_config_id: str,

):

with test_library:

self.write_stable_config(

{

"apm_configuration_default": local_cfg,

},

"/etc/datadog-agent/application_monitoring.yaml",

test_library,

)

self.write_stable_config(

{

"apm_configuration_default": fleet_cfg,

"config_id": fleet_config_id,

},

"/etc/datadog-agent/managed/datadog-agent/stable/application_monitoring.yaml",

test_library,

)

# Sleep to ensure the telemetry events are sent with different timestamps

time.sleep(1)

test_library.container_restart()

test_library.dd_start_span("test")

configuration_by_name = test_agent.wait_for_telemetry_configurations()

# Configuration set via fleet config should have the config_id set

apm_telemetry_names = _mapped_telemetry_name("trace_debug_enabled")

telemetry_item = None

for apm_name in apm_telemetry_names:

telemetry_item = test_agent.get_telemetry_config_by_origin(

configuration_by_name, apm_name, "fleet_stable_config"

)

if telemetry_item is not None:

break

assert telemetry_item is not None, (

f"No configuration found for any of {' or '.join(apm_telemetry_names)} with origin 'fleet_stable_config'"

)

assert isinstance(telemetry_item, dict)

assert telemetry_item["origin"] == "fleet_stable_config"

assert telemetry_item["config_id"] == fleet_config_id

# Configuration set via local config should not have the config_id set

apm_telemetry_names = _mapped_telemetry_name("dynamic_instrumentation_enabled")

telemetry_item = None

for apm_name in apm_telemetry_names:

telemetry_item = test_agent.get_telemetry_config_by_origin(

configuration_by_name, apm_name, "local_stable_config"

)

if telemetry_item is not None:

break

assert telemetry_item is not None, (

f"No configuration found for any of {' or '.join(apm_telemetry_names)} with origin 'local_stable_config'"

)

assert isinstance(telemetry_item, dict)

assert telemetry_item["origin"] == "local_stable_config"

assert "config_id" not in telemetry_item or telemetry_item["config_id"] is None

@pytest.mark.parametrize(

("local_cfg", "library_env", "fleet_cfg", "expected_origins"),

[

(

{

"DD_TRACE_PROPAGATION_STYLE": "tracecontext",

"DD_TAGS": "tag1:value1,tag2:value2",

},

{

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.1", # Decrease the heartbeat/poll intervals to speed up the tests

},

{

"DD_TRACE_PROPAGATION_STYLE": "datadog",

},

{

"tags": "local_stable_config",

"trace_propagation_style": "fleet_stable_config",

},

)

],

)

def test_stable_configuration_origin_extended_configs_good_use_case(

self,

local_cfg: dict[str, str],

fleet_cfg: dict[str, str],

test_agent: TestAgentAPI,

test_library: APMLibrary,

expected_origins: dict[str, str],

):

"""Test that extended configuration options (tags, propagation style) report their origin correctly.

This test verifies that complex configuration values like tag arrays and propagation

styles are properly tracked with their configuration origin (local vs fleet stable config).

"""

with test_library:

self.write_stable_config(

{

"apm_configuration_default": local_cfg,

},

"/etc/datadog-agent/application_monitoring.yaml",

test_library,

)

self.write_stable_config(

{

"apm_configuration_default": fleet_cfg,

},

"/etc/datadog-agent/managed/datadog-agent/stable/application_monitoring.yaml",

test_library,

)

# Sleep between telemetry events to ensure they are recorded with different timestamps, to later reorder them.

# seq_id can't be used to sort because payloads are sent from different tracer sessions.

time.sleep(1)

test_library.container_restart()

test_library.dd_start_span("test")

configuration_by_name = test_agent.wait_for_telemetry_configurations()

for cfg_name, expected_origin in expected_origins.items():

apm_telemetry_names = _mapped_telemetry_name(cfg_name)

telemetry_item = None

for apm_name in apm_telemetry_names:

telemetry_item = test_agent.get_telemetry_config_by_origin(

configuration_by_name, apm_name, expected_origin

)

if telemetry_item is not None:

break

assert telemetry_item is not None, (

f"No configuration found for any of {' or '.join(apm_telemetry_names)} with origin '{expected_origin}'. Full configuration_by_name: {configuration_by_name}"

)

assert isinstance(telemetry_item, dict)

actual_origin = telemetry_item.get("origin", "<missing>")

assert isinstance(telemetry_item, dict)

assert telemetry_item["origin"] == expected_origin, (

f"Origin mismatch for {telemetry_item}. Expected origin: '{expected_origin}', Actual origin: '{actual_origin}'"

)

assert telemetry_item["value"]

@pytest.mark.parametrize(

("local_cfg", "library_env", "fleet_cfg", "expected_origins"),

[

(

{

"DD_TRACE_PROPAGATION_STYLE": "tracecontext",

"DD_TAGS": "tag1:value1,tag2:value2",

},

{

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.1", # Decrease the heartbeat/poll intervals to speed up the tests

},

{

"DD_TRACE_PROPAGATION_STYLE": "datadog",

},

{

"tags": "local_stable_config",

"trace_propagation_style": "fleet_stable_config",

},

)

],

)

def test_stable_configuration_origin_extended_configs_temporary_use_case(

self,

local_cfg: dict[str, str],

fleet_cfg: dict[str, str],

test_agent: TestAgentAPI,

test_library: APMLibrary,

expected_origins: dict[str, str],

):

"""Test that extended configuration options (tags, propagation style) report their origin correctly.

This test verifies that complex configuration values like tag arrays and propagation

styles are properly tracked with their configuration origin (local vs fleet stable config).

"""

with test_library:

self.write_stable_config(

{

"apm_configuration_default": local_cfg,

},

"/etc/datadog-agent/application_monitoring.yaml",

test_library,

)

self.write_stable_config(

{

"apm_configuration_default": fleet_cfg,

},

"/etc/datadog-agent/managed/datadog-agent/stable/application_monitoring.yaml",

test_library,

)

# Sleep between telemetry events to ensure they are recorded with different timestamps, to later reorder them.

# seq_id can't be used to sort because payloads are sent from different tracer sessions.

time.sleep(1)

test_library.container_restart()

test_library.dd_start_span("test")

configuration_by_name = test_agent.wait_for_telemetry_configurations()

for cfg_name, expected_origin in expected_origins.items():

# Python, Ruby and Node.js only report inject and extract keys for trace_propagation_style

if cfg_name == "trace_propagation_style" and context.library.name in ["python", "ruby", "nodejs"]:

_check_propagation_style_with_inject_and_extract(

test_agent,

configuration_by_name,

expected_origin,

context.library.name,

allow_calculated_origin=True,

)

elif cfg_name == "tags" and context.library.name in ["ruby"]:

continue

else:

apm_telemetry_names = _mapped_telemetry_name(cfg_name)

telemetry_item = None

for apm_name in apm_telemetry_names:

telemetry_item = test_agent.get_telemetry_config_by_origin(

configuration_by_name, apm_name, expected_origin

)

if telemetry_item is not None:

break

assert telemetry_item is not None, (

f"No configuration found for any of {' or '.join(apm_telemetry_names)} with origin '{expected_origin}'. Full configuration_by_name: {configuration_by_name}"

)

assert isinstance(telemetry_item, dict)

actual_origin = telemetry_item.get("origin", "<missing>")

assert telemetry_item["origin"] == expected_origin, (

f"Origin mismatch for {telemetry_item}. Expected origin: '{expected_origin}', Actual origin: '{actual_origin}'"

)

assert telemetry_item["value"]

DEFAULT_ENVVARS = {

# Decrease the heartbeat/poll intervals to speed up the tests

"DD_TELEMETRY_HEARTBEAT_INTERVAL": "0.2",

}

@rfc("https://docs.google.com/document/d/14vsrCbnAKnXmJAkacX9I6jKPGKmxsq0PKUb3dfiZpWE/edit")

@scenarios.parametric

@features.telemetry_app_started_event

class Test_TelemetryInstallSignature:

"""This telemetry provides insights into how a library was installed."""

@pytest.mark.parametrize(

"library_env",

[

{

**DEFAULT_ENVVARS,

"DD_INSTRUMENTATION_INSTALL_TIME": str(int(time.time())),

"DD_INSTRUMENTATION_INSTALL_TYPE": "k8s_single_step",

"DD_INSTRUMENTATION_INSTALL_ID": str(uuid.uuid4()),

},

],

)

def test_telemetry_event_propagated(

self, library_env: dict[str, str], test_agent: TestAgentAPI, test_library: APMLibrary

):

"""Ensure the installation ID is included in the app-started telemetry event.

The installation ID is generated as soon as possible in the APM installation process. It is propagated

to the APM library via an environment variable. It is used to correlate telemetry events to help determine

where installation issues are occurring.

"""

# Some libraries require a first span for telemetry to be emitted.

with test_library.dd_start_span("first_span"):

pass