diff --git a/tests/collector/helpers/test_fargate_docker_helper.py b/tests/collector/helpers/test_fargate_docker_helper.py new file mode 100644 index 00000000..5b1fe99b --- /dev/null +++ b/tests/collector/helpers/test_fargate_docker_helper.py @@ -0,0 +1,230 @@ +# (c) Copyright IBM Corp. 2026 + +from collections.abc import Generator +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from instana.collector.helpers.fargate.docker import DockerHelper + +DOCKER_ID = "docker-id" + + +def make_container_stats() -> dict[str, Any]: + return { + "networks": { + "eth0": { + "tx_bytes": 10, + "tx_dropped": 1, + "tx_errors": 2, + "tx_packets": 3, + "rx_bytes": 20, + "rx_dropped": 4, + "rx_errors": 5, + "rx_packets": 6, + }, + "eth1": {"tx_bytes": 5, "rx_bytes": 5}, + "lo": {"tx_bytes": 1000, "rx_bytes": 1000}, + }, + "cpu_stats": { + "cpu_usage": { + "total_usage": 50, + "usage_in_usermode": 30, + "usage_in_kernelmode": 20, + }, + "system_cpu_usage": 100, + "online_cpus": 2, + "throttling_data": {"periods": 8, "throttled_time": 9}, + }, + "memory_stats": { + "usage": 100, + "max_usage": 200, + "limit": 300, + "stats": { + "active_anon": 1, + "active_file": 2, + "inactive_anon": 3, + "inactive_file": 4, + "total_cache": 5, + "total_rss": 6, + }, + }, + "blkio_stats": { + "io_service_bytes_recursive": [ + {"op": "Read", "value": 100}, + {"op": "Write", "value": 200}, + {"op": "Total", "value": 300}, + ] + }, + } + + +def make_plugin_data() -> dict[str, Any]: + from instana.util import DictionaryOfStan + + return {"name": "com.instana.plugin.docker", "data": DictionaryOfStan()} + + +class TestDockerHelper: + @pytest.fixture(autouse=True) + def _resource(self) -> Generator[None, None, None]: + self.collector = MagicMock() + self.collector.task_metadata = { + "Containers": [ + { + "DockerId": DOCKER_ID, + "Name": "my-container", + "CreatedAt": "created", + "StartedAt": "started", + "Image": "image", + "Labels": {"com.amazonaws.ecs.task-arn": "task-arn"}, + "Ports": [{"ContainerPort": 80}], + "Networks": [{"NetworkMode": "awsvpc"}], + } + ] + } + self.collector.task_stats_metadata = {DOCKER_ID: make_container_stats()} + self.helper = DockerHelper(self.collector) + yield + self.helper = None + + def test_collect_metrics_without_metadata(self) -> None: + self.collector.task_metadata = None + assert self.helper.collect_metrics() == [] + + def test_collect_metrics_without_snapshot(self) -> None: + plugins = self.helper.collect_metrics() + assert len(plugins) == 1 + plugin = plugins[0] + assert plugin["name"] == "com.instana.plugin.docker" + assert plugin["entityId"] == "task-arn::my-container" + assert plugin["data"]["Id"] == DOCKER_ID + assert "Created" not in plugin["data"] + assert plugin["data"]["tx"]["bytes"] == 15 + assert plugin["data"]["tx"]["dropped"] == 1 + assert plugin["data"]["tx"]["errors"] == 2 + assert plugin["data"]["tx"]["packets"] == 3 + assert plugin["data"]["rx"]["bytes"] == 25 + assert plugin["data"]["rx"]["dropped"] == 4 + assert plugin["data"]["rx"]["errors"] == 5 + assert plugin["data"]["rx"]["packets"] == 6 + + def test_collect_metrics_cpu(self) -> None: + cpu = self.helper.collect_metrics()[0]["data"]["cpu"] + assert cpu["total_usage"] == 1.0 + assert cpu["user_usage"] == 0.6 + assert cpu["system_usage"] == 0.4 + assert cpu["throttling_count"] == 8 + assert cpu["throttling_time"] == 9 + + def test_collect_metrics_memory(self) -> None: + memory = self.helper.collect_metrics()[0]["data"]["memory"] + assert memory["usage"] == 100 + assert memory["max_usage"] == 200 + assert memory["limit"] == 300 + assert memory["active_anon"] == 1 + assert memory["active_file"] == 2 + assert memory["inactive_anon"] == 3 + assert memory["inactive_file"] == 4 + assert memory["total_cache"] == 5 + assert memory["total_rss"] == 6 + + def test_collect_metrics_blkio(self) -> None: + blkio = self.helper.collect_metrics()[0]["data"]["blkio"] + assert blkio["blk_read"] == 100 + assert blkio["blk_write"] == 200 + assert self.helper.previous_blkio[DOCKER_ID]["blk_read"] == 100 + assert self.helper.previous_blkio[DOCKER_ID]["blk_write"] == 200 + + def test_collect_metrics_blkio_reports_delta_on_second_run(self) -> None: + self.helper.collect_metrics() + stats = self.collector.task_stats_metadata[DOCKER_ID] + stats["blkio_stats"]["io_service_bytes_recursive"] = [ + {"op": "Read", "value": 150}, + {"op": "Write", "value": 200}, + ] + blkio = self.helper.collect_metrics()[0]["data"]["blkio"] + assert blkio["blk_read"] == 50 + assert blkio["blk_write"] == 0 + assert self.helper.previous_blkio[DOCKER_ID]["blk_write"] == 200 + assert self.helper.previous_blkio[DOCKER_ID]["blk_read"] == 150 + + def test_collect_metrics_unchanged_values_are_not_resent(self) -> None: + self.helper.collect_metrics() + plugin = self.helper.collect_metrics()[0] + assert not plugin["data"]["tx"] + assert not plugin["data"]["rx"] + assert not plugin["data"]["memory"] + assert not plugin["data"]["cpu"] + + def test_collect_metrics_with_snapshot_resends_values(self) -> None: + self.helper.collect_metrics() + plugin = self.helper.collect_metrics(with_snapshot=True)[0] + assert plugin["data"]["tx"]["bytes"] == 15 + assert plugin["data"]["memory"]["usage"] == 100 + assert plugin["data"]["Created"] == "created" + assert plugin["data"]["Started"] == "started" + assert plugin["data"]["Image"] == "image" + assert plugin["data"]["Labels"] == {"com.amazonaws.ecs.task-arn": "task-arn"} + assert plugin["data"]["Ports"] == [{"ContainerPort": 80}] + assert plugin["data"]["NetworkMode"] == "awsvpc" + + def test_collect_metrics_snapshot_without_networks(self) -> None: + self.collector.task_metadata["Containers"][0]["Networks"] = [] + plugin = self.helper.collect_metrics(with_snapshot=True)[0] + assert "NetworkMode" not in plugin["data"] + + def test_collect_metrics_unknown_container_stats(self) -> None: + self.collector.task_stats_metadata = {} + plugin = self.helper.collect_metrics()[0] + assert plugin["data"]["Id"] == DOCKER_ID + assert "tx" not in plugin["data"] + + def test_collect_metrics_missing_optional_sections(self) -> None: + self.collector.task_stats_metadata = { + DOCKER_ID: {"cpu_stats": {}, "blkio_stats": {}} + } + plugin = self.helper.collect_metrics()[0] + assert "tx" not in plugin["data"] + assert "cpu" not in plugin["data"] + assert "blkio" not in plugin["data"] + + def test_collect_metrics_outer_exception( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + self.collector.task_metadata = {"Containers": [5]} + assert self.helper.collect_metrics() == [] + assert "DockerHelper.collect_metrics" in caplog.text + + def test_collect_container_snapshot_exception( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + plugin_data = make_plugin_data() + self.helper._collect_container_snapshot(plugin_data, {"Networks": None}) + assert "_collect_container_snapshot" in caplog.text + + @pytest.mark.parametrize( + "method, message", + [ + ("_collect_network_metrics", "_collect_network_metrics"), + ("_collect_cpu_metrics", "_collect_cpu_metrics"), + ("_collect_memory_metrics", "_collect_memory_metrics"), + ("_collect_blkio_metrics", "_collect_blkio_metrics"), + ], + ) + def test_collect_metrics_helpers_log_exceptions( + self, method: str, message: str, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + plugin_data = make_plugin_data() + container = { + "networks": {"eth0": "not-a-dict"}, + "cpu_stats": {"cpu_usage": {}, "system_cpu_usage": 1}, + "memory_stats": "not-a-dict", + "blkio_stats": {"io_service_bytes_recursive": [{}]}, + } + getattr(self.helper, method)(container, plugin_data, DOCKER_ID, False) + assert message in caplog.text diff --git a/tests/collector/helpers/test_fargate_task_container_helpers.py b/tests/collector/helpers/test_fargate_task_container_helpers.py new file mode 100644 index 00000000..991f6bbc --- /dev/null +++ b/tests/collector/helpers/test_fargate_task_container_helpers.py @@ -0,0 +1,185 @@ +# (c) Copyright IBM Corp. 2026 + +from collections.abc import Generator +from unittest.mock import MagicMock, PropertyMock + +import pytest + +from instana.collector.helpers.fargate.container import ContainerHelper +from instana.collector.helpers.fargate.task import TaskHelper + +TASK_ARN = "arn:aws:ecs:us-east-2:123:task/abc" + +CONTAINER = { + "DockerId": "docker-id", + "Name": "my-container", + "DockerName": "ecs-my-container", + "Image": "my/image:latest", + "ImageID": "sha256:abc", + "DesiredStatus": "RUNNING", + "KnownStatus": "RUNNING", + "Ports": [{"ContainerPort": 80}], + "CreatedAt": "2025-01-01T00:00:00Z", + "StartedAt": "2025-01-01T00:00:01Z", + "Type": "NORMAL", + "Limits": {"CPU": 2, "Memory": 512}, + "Labels": { + "com.amazonaws.ecs.task-arn": TASK_ARN, + "com.amazonaws.ecs.task-definition-family": "family", + "com.amazonaws.ecs.task-definition-version": "7", + "com.amazonaws.ecs.cluster": "cluster-arn", + }, +} + +TASK_METADATA = { + "TaskARN": TASK_ARN, + "Cluster": "cluster-arn", + "Family": "family", + "Revision": "7", + "AvailabilityZone": "us-east-2a", + "DesiredStatus": "RUNNING", + "KnownStatus": "RUNNING", + "PullStartedAt": "start", + "PullStoppeddAt": "stop", + "Limits": {"CPU": 1, "Memory": 1024}, + "Containers": [CONTAINER], +} + + +class TestTaskHelper: + @pytest.fixture(autouse=True) + def _resource(self) -> Generator[None, None, None]: + self.collector = MagicMock() + self.collector.task_metadata = dict(TASK_METADATA) + self.collector.agent.options.zone = None + self.collector.agent.options.tags = None + self.helper = TaskHelper(self.collector) + yield + self.helper = None + + def test_collect_metrics_without_metadata(self) -> None: + self.collector.task_metadata = None + assert self.helper.collect_metrics() == [] + + def test_collect_metrics_without_snapshot(self) -> None: + plugins = self.helper.collect_metrics() + assert len(plugins) == 1 + plugin = plugins[0] + assert plugin["name"] == "com.instana.plugin.aws.ecs.task" + assert plugin["entityId"] == TASK_ARN + assert plugin["data"]["taskArn"] == TASK_ARN + assert plugin["data"]["clusterArn"] == "cluster-arn" + assert plugin["data"]["taskDefinition"] == "family" + assert plugin["data"]["taskDefinitionVersion"] == "7" + assert plugin["data"]["availabilityZone"] == "us-east-2a" + assert "desiredStatus" not in plugin["data"] + + def test_collect_metrics_with_snapshot(self) -> None: + plugin = self.helper.collect_metrics(with_snapshot=True)[0] + assert plugin["data"]["desiredStatus"] == "RUNNING" + assert plugin["data"]["knownStatus"] == "RUNNING" + assert plugin["data"]["pullStartedAt"] == "start" + assert plugin["data"]["pullStoppedAt"] == "stop" + assert plugin["data"]["limits"]["cpu"] == 1 + assert plugin["data"]["limits"]["memory"] == 1024 + assert "instanaZone" not in plugin["data"] + assert "tags" not in plugin["data"] + + def test_collect_metrics_with_snapshot_zone_and_tags(self) -> None: + self.collector.agent.options.zone = "zone-a" + self.collector.agent.options.tags = {"env": "test"} + plugin = self.helper.collect_metrics(with_snapshot=True)[0] + assert plugin["data"]["instanaZone"] == "zone-a" + assert plugin["data"]["tags"] == {"env": "test"} + + def test_collect_metrics_inner_exception_still_appends_plugin( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + self.collector.task_metadata = {"Limits": None} + plugins = self.helper.collect_metrics(with_snapshot=True) + assert len(plugins) == 1 + assert plugins[0]["name"] == "com.instana.plugin.aws.ecs.task" + assert "collect_task_metrics" in caplog.text + + def test_collect_metrics_outer_exception( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + type(self.collector).task_metadata = PropertyMock( + side_effect=RuntimeError("boom") + ) + assert self.helper.collect_metrics() == [] + assert "collect_task_metrics" in caplog.text + + +class TestContainerHelper: + @pytest.fixture(autouse=True) + def _resource(self) -> Generator[None, None, None]: + self.collector = MagicMock() + self.collector.task_metadata = dict(TASK_METADATA) + self.collector.root_metadata = {"Name": "my-container"} + self.helper = ContainerHelper(self.collector) + yield + self.helper = None + + def test_collect_metrics_without_metadata(self) -> None: + self.collector.task_metadata = None + assert self.helper.collect_metrics() == [] + + def test_collect_metrics_without_containers(self) -> None: + self.collector.task_metadata = {} + assert self.helper.collect_metrics() == [] + + def test_collect_metrics_without_snapshot(self) -> None: + plugins = self.helper.collect_metrics() + assert len(plugins) == 1 + plugin = plugins[0] + assert plugin["name"] == "com.instana.plugin.aws.ecs.container" + assert plugin["entityId"] == f"{TASK_ARN}::my-container" + assert plugin["data"]["instrumented"] is True + assert plugin["data"]["dockerId"] == "docker-id" + assert plugin["data"]["taskArn"] == TASK_ARN + assert "runtime" not in plugin["data"] + + def test_collect_metrics_not_instrumented_container(self) -> None: + self.collector.root_metadata = {"Name": "other-container"} + plugin = self.helper.collect_metrics()[0] + assert "instrumented" not in plugin["data"] + + def test_collect_metrics_with_snapshot(self) -> None: + data = self.helper.collect_metrics(with_snapshot=True)[0]["data"] + assert data["runtime"] == "python" + assert data["dockerName"] == "ecs-my-container" + assert data["containerName"] == "my-container" + assert data["image"] == "my/image:latest" + assert data["imageId"] == "sha256:abc" + assert data["taskDefinition"] == "family" + assert data["taskDefinitionVersion"] == "7" + assert data["clusterArn"] == "cluster-arn" + assert data["desiredStatus"] == "RUNNING" + assert data["knownStatus"] == "RUNNING" + assert data["ports"] == [{"ContainerPort": 80}] + assert data["createdAt"] == "2025-01-01T00:00:00Z" + assert data["startedAt"] == "2025-01-01T00:00:01Z" + assert data["type"] == "NORMAL" + assert data["limits"]["cpu"] == 2 + assert data["limits"]["memory"] == 512 + + def test_collect_metrics_inner_exception_still_appends_plugin( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + self.collector.root_metadata = {} + plugins = self.helper.collect_metrics() + assert len(plugins) == 1 + assert plugins[0]["name"] == "com.instana.plugin.aws.ecs.container" + assert "_collect_container_snapshots" in caplog.text + + def test_collect_metrics_outer_exception( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + self.collector.task_metadata = {"Containers": 5} + assert self.helper.collect_metrics() == [] + assert "collect_container_metrics" in caplog.text diff --git a/tests/collector/helpers/test_process_helper.py b/tests/collector/helpers/test_process_helper.py new file mode 100644 index 00000000..a0e121fb --- /dev/null +++ b/tests/collector/helpers/test_process_helper.py @@ -0,0 +1,114 @@ +# (c) Copyright IBM Corp. 2026 + +import os +from collections.abc import Generator +from unittest.mock import MagicMock + +import pytest + +from instana.collector.helpers.process import ProcessHelper + + +class TestProcessHelper: + @pytest.fixture(autouse=True) + def _resource(self) -> Generator[None, None, None]: + self.collector = MagicMock() + self.collector.agent.options.secrets_matcher = "contains" + self.collector.agent.options.secrets_list = ["KEY", "PASSWORD", "SECRET"] + self.collector.fetching_start_time = 1234 + self.helper = ProcessHelper(self.collector) + os.environ["INSTANA_TEST_PLAIN_VAR"] = "visible" + os.environ["INSTANA_TEST_PASSWORD"] = "hidden" + yield + os.environ.pop("INSTANA_TEST_PLAIN_VAR", None) + os.environ.pop("INSTANA_TEST_PASSWORD", None) + self.helper = None + + def test_collect_metrics_without_snapshot(self) -> None: + plugin_data = self.helper.collect_metrics() + assert plugin_data["name"] == "com.instana.plugin.process" + assert plugin_data["entityId"] == str(os.getpid()) + assert plugin_data["data"]["pid"] == os.getpid() + assert "env" not in plugin_data["data"] + + def test_collect_metrics_with_snapshot(self) -> None: + plugin_data = self.helper.collect_metrics(with_snapshot=True) + data = plugin_data["data"] + assert data["pid"] == os.getpid() + assert data["env"]["INSTANA_TEST_PLAIN_VAR"] == "visible" + assert data["env"]["INSTANA_TEST_PASSWORD"] == "" + assert data["start"] == 1234 + assert isinstance(data["args"], list) + + def test_collect_metrics_exception( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + monkeypatch.setattr( + "instana.collector.helpers.process.DictionaryOfStan", + MagicMock(side_effect=RuntimeError("boom")), + ) + plugin_data = self.helper.collect_metrics() + assert "data" not in plugin_data + assert "ProcessHelper.collect_metrics" in caplog.text + + def test_collect_process_snapshot_exec_path_missing( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + monkeypatch.setattr( + "instana.collector.helpers.process.os.path.isfile", + lambda path: False, + ) + plugin_data = self.helper.collect_metrics(with_snapshot=True) + assert "exec" not in plugin_data["data"] + assert "Can't access /proc/self/exe" in caplog.text + + def test_collect_process_snapshot_exec_path_present( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + "instana.collector.helpers.process.os.path.isfile", + lambda path: True, + ) + monkeypatch.setattr( + "instana.collector.helpers.process.os.readlink", + lambda path: "/usr/bin/python", + ) + plugin_data = self.helper.collect_metrics(with_snapshot=True) + assert plugin_data["data"]["exec"] == "/usr/bin/python" + + def test_collect_process_snapshot_drops_exe_from_cmdline( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + "instana.collector.helpers.process.get_proc_cmdline", + lambda: ["python", "app.py", "--flag"], + ) + plugin_data = self.helper.collect_metrics(with_snapshot=True) + assert plugin_data["data"]["args"] == ["app.py", "--flag"] + + def test_collect_process_snapshot_user_lookup_error( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + monkeypatch.setattr( + "instana.collector.helpers.process.pwd.getpwuid", + MagicMock(side_effect=KeyError("no user")), + ) + plugin_data = self.helper.collect_metrics(with_snapshot=True) + assert "user" not in plugin_data["data"] + assert plugin_data["data"]["start"] == 1234 + assert "euid/egid detection" in caplog.text + + def test_collect_process_snapshot_exception( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + monkeypatch.setattr( + "instana.collector.helpers.process.contains_secret", + MagicMock(side_effect=RuntimeError("boom")), + ) + plugin_data = self.helper.collect_metrics(with_snapshot=True) + assert "env" not in plugin_data["data"] + assert "ProcessHelper._collect_process_snapshot" in caplog.text diff --git a/tests/collector/test_serverless_collectors.py b/tests/collector/test_serverless_collectors.py new file mode 100644 index 00000000..343486c8 --- /dev/null +++ b/tests/collector/test_serverless_collectors.py @@ -0,0 +1,433 @@ +# (c) Copyright IBM Corp. 2026 + +import json +from collections.abc import Generator +from unittest.mock import MagicMock + +import pytest + +from instana.collector.aws_eks_fargate import EKSFargateCollector +from instana.collector.aws_fargate import AWSFargateCollector +from instana.collector.google_cloud_run import GCRCollector +from instana.collector.helpers.google_cloud_run.instance_entity import ( + InstanceEntityHelper, +) + +ECMU = "http://169.254.170.2/v3/abc" + + +class FakeHelper: + def __init__(self, plugins=None, error=None) -> None: + self.plugins = plugins if plugins is not None else [] + self.error = error + self.kwargs = None + + def collect_metrics(self, **kwargs: object) -> list[dict[str, str]]: + self.kwargs = kwargs + if self.error: + raise self.error + return self.plugins + + +def make_response(payload) -> MagicMock: + response = MagicMock() + response.content = json.dumps(payload) + response.json.return_value = payload + return response + + +class TestAWSFargateCollector: + @pytest.fixture(autouse=True) + def _resource(self, monkeypatch: pytest.MonkeyPatch) -> Generator[None, None, None]: + monkeypatch.setenv("ECS_CONTAINER_METADATA_URI", ECMU) + self.collector = AWSFargateCollector(MagicMock()) + yield + self.collector = None + + def test_init_valid_metadata_uri_sets_urls_and_helpers(self) -> None: + assert self.collector.ready_to_start is True + assert self.collector.ecmu_url_root == ECMU + assert self.collector.ecmu_url_task == f"{ECMU}/task" + assert self.collector.ecmu_url_stats == f"{ECMU}/stats" + assert self.collector.ecmu_url_task_stats == f"{ECMU}/task/stats" + assert len(self.collector.helpers) == 5 + + @pytest.mark.parametrize("value", ["", "not a url"]) + def test_init_invalid_metadata_uri( + self, value: str, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("ECS_CONTAINER_METADATA_URI", value) + collector = AWSFargateCollector(MagicMock()) + assert collector.ready_to_start is False + + def test_start_not_ready(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ECS_CONTAINER_METADATA_URI", "") + collector = AWSFargateCollector(MagicMock()) + collector.start() + assert collector.started is False + assert collector.reporting_thread is None + + def test_start_ready(self, monkeypatch: pytest.MonkeyPatch) -> None: + start = MagicMock() + monkeypatch.setattr("instana.collector.base.BaseCollector.start", start) + self.collector.start() + start.assert_called_once() + + def test_get_ecs_metadata_full_fetch(self) -> None: + responses = { + self.collector.ecmu_url_root: {"Name": "root"}, + self.collector.ecmu_url_task: {"TaskARN": "task"}, + self.collector.ecmu_url_stats: {"stats": 1}, + self.collector.ecmu_url_task_stats: {"task_stats": 2}, + } + self.collector.http_client = MagicMock() + self.collector.http_client.get.side_effect = lambda url, timeout: make_response( + responses[url] + ) + self.collector.get_ecs_metadata() + assert self.collector.root_metadata == {"Name": "root"} + assert self.collector.task_metadata == {"TaskARN": "task"} + assert self.collector.stats_metadata == {"stats": 1} + assert self.collector.task_stats_metadata == {"task_stats": 2} + assert self.collector.http_client.get.call_count == 4 + assert self.collector.last_ecmu_full_fetch > 0 + + def test_get_ecs_metadata_partial_fetch(self) -> None: + self.collector.last_ecmu_full_fetch = 2**40 + self.collector.http_client = MagicMock() + self.collector.http_client.get.return_value = make_response({"k": "v"}) + self.collector.get_ecs_metadata() + assert self.collector.root_metadata is None + assert self.collector.task_metadata is None + assert self.collector.stats_metadata == {"k": "v"} + assert self.collector.task_stats_metadata == {"k": "v"} + assert self.collector.http_client.get.call_count == 2 + + def test_get_ecs_metadata_exception(self, caplog: pytest.LogCaptureFixture) -> None: + caplog.set_level("DEBUG", logger="instana") + self.collector.http_client = MagicMock() + self.collector.http_client.get.side_effect = RuntimeError("boom") + self.collector.get_ecs_metadata() + assert self.collector.root_metadata is None + assert "AWSFargateCollector.get_ecs_metadata" in caplog.text + + def test_should_send_snapshot_data_by_interval(self) -> None: + self.collector.snapshot_data_last_sent = 0 + assert self.collector.should_send_snapshot_data() is True + self.collector.snapshot_data_last_sent = 2**40 + assert self.collector.should_send_snapshot_data() is False + + def test_prepare_payload_with_snapshot( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(self.collector, "get_ecs_metadata", MagicMock()) + helper = FakeHelper(plugins=[{"name": "plugin"}]) + self.collector.helpers = [helper, helper] + payload = self.collector.prepare_payload() + assert payload["spans"] == [] + assert payload["metrics"]["plugins"] == [{"name": "plugin"}] * 2 + assert helper.kwargs == {"with_snapshot": True} + assert self.collector.snapshot_data_last_sent > 0 + self.collector.get_ecs_metadata.assert_called_once() + + def test_prepare_payload_without_snapshot( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(self.collector, "get_ecs_metadata", MagicMock()) + self.collector.snapshot_data_last_sent = 2**40 + helper = FakeHelper() + self.collector.helpers = [helper] + payload = self.collector.prepare_payload() + assert helper.kwargs == {"with_snapshot": False} + assert payload["metrics"]["plugins"] == [] + assert self.collector.snapshot_data_last_sent == 2**40 + + def test_prepare_payload_with_spans(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(self.collector, "get_ecs_metadata", MagicMock()) + monkeypatch.setattr( + "instana.collector.aws_fargate.format_span", lambda spans: ["formatted"] + ) + monkeypatch.setattr(self.collector, "queued_spans", lambda: ["span"]) + self.collector.span_queue.put("span") + self.collector.helpers = [] + payload = self.collector.prepare_payload() + assert payload["spans"] == ["formatted"] + + def test_prepare_payload_exception( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + monkeypatch.setattr(self.collector, "get_ecs_metadata", MagicMock()) + self.collector.helpers = [FakeHelper(error=RuntimeError("boom"))] + payload = self.collector.prepare_payload() + assert payload["metrics"]["plugins"] == [] + assert "collect_snapshot error" in caplog.text + + def test_get_fq_arn_without_metadata(self) -> None: + assert self.collector.get_fq_arn() == "Missing ECMU metadata" + + def test_get_fq_arn_with_metadata(self) -> None: + self.collector.root_metadata = { + "Name": "container", + "Labels": {"com.amazonaws.ecs.task-arn": "arn:task"}, + } + assert self.collector.get_fq_arn() == "arn:task::container" + self.collector.root_metadata = None + assert self.collector.get_fq_arn() == "arn:task::container" + + def test_get_fq_arn_without_labels(self) -> None: + self.collector.root_metadata = {"Name": "container"} + assert self.collector.get_fq_arn() == "::container" + + +class TestEKSFargateCollector: + @pytest.fixture(autouse=True) + def _resource(self) -> Generator[None, None, None]: + agent = MagicMock() + agent.podname = "my-pod" + self.collector = EKSFargateCollector(agent) + yield + self.collector = None + + def test_init_sets_podname_and_helpers(self) -> None: + assert self.collector.podname == "my-pod" + assert self.collector.snapshot_data_sent is False + assert len(self.collector.helpers) == 2 + + def test_should_send_snapshot_data_by_interval(self) -> None: + self.collector.snapshot_data_last_sent = 0 + assert self.collector.should_send_snapshot_data() is True + self.collector.snapshot_data_last_sent = 2**40 + assert self.collector.should_send_snapshot_data() is False + + def test_prepare_payload_with_snapshot(self) -> None: + helper = FakeHelper(plugins=[{"name": "plugin"}]) + self.collector.helpers = [helper] + payload = self.collector.prepare_payload() + assert payload["metrics"]["plugins"] == [{"name": "plugin"}] + assert helper.kwargs == {"with_snapshot": True} + assert self.collector.snapshot_data_last_sent > 0 + + def test_prepare_payload_without_snapshot(self) -> None: + self.collector.snapshot_data_last_sent = 2**40 + helper = FakeHelper() + self.collector.helpers = [helper] + self.collector.prepare_payload() + assert helper.kwargs == {"with_snapshot": False} + assert self.collector.snapshot_data_last_sent == 2**40 + + def test_prepare_payload_with_spans(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "instana.collector.aws_eks_fargate.format_span", + lambda spans: ["formatted"], + ) + monkeypatch.setattr(self.collector, "queued_spans", lambda: ["span"]) + self.collector.span_queue.put("span") + self.collector.helpers = [] + assert self.collector.prepare_payload()["spans"] == ["formatted"] + + def test_prepare_payload_exception(self, caplog: pytest.LogCaptureFixture) -> None: + caplog.set_level("DEBUG", logger="instana") + self.collector.helpers = [FakeHelper(error=RuntimeError("boom"))] + payload = self.collector.prepare_payload() + assert payload["metrics"]["plugins"] == [] + assert "prepare_payload error" in caplog.text + + +class TestGCRCollector: + @pytest.fixture(autouse=True) + def _resource(self, monkeypatch: pytest.MonkeyPatch) -> Generator[None, None, None]: + monkeypatch.delenv("GOOGLE_CLOUD_RUN_METADATA_ENDPOINT", raising=False) + self.collector = GCRCollector(MagicMock(), "service", "config", "rev") + yield + self.collector = None + + def test_init_default_endpoint_sets_urls_and_helpers(self) -> None: + assert self.collector.ready_to_start is True + assert self.collector.service == "service" + assert self.collector.configuration == "config" + assert self.collector.revision == "rev" + assert self.collector._gcr_md_project_uri == ( + "http://metadata.google.internal/computeMetadata/v1/project/?recursive=true" + ) + assert self.collector._gcr_md_instance_uri == ( + "http://metadata.google.internal/computeMetadata/v1/instance/?recursive=true" + ) + assert len(self.collector.helpers) == 2 + + @pytest.mark.parametrize("value", ["", "not a url"]) + def test_init_invalid_metadata_endpoint( + self, value: str, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("GOOGLE_CLOUD_RUN_METADATA_ENDPOINT", value) + collector = GCRCollector(MagicMock(), "service", "config", "rev") + assert collector.ready_to_start is False + + def test_start_not_ready(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GOOGLE_CLOUD_RUN_METADATA_ENDPOINT", "") + collector = GCRCollector(MagicMock(), "service", "config", "rev") + collector.start() + assert collector.started is False + + def test_start_ready(self, monkeypatch: pytest.MonkeyPatch) -> None: + start = MagicMock() + monkeypatch.setattr("instana.collector.base.BaseCollector.start", start) + self.collector.start() + start.assert_called_once() + + def test_get_project_instance_metadata_stores_responses(self) -> None: + responses = { + self.collector._gcr_md_project_uri: {"projectId": "project"}, + self.collector._gcr_md_instance_uri: {"id": "instance"}, + } + self.collector._http_client = MagicMock() + self.collector._http_client.get.side_effect = lambda url, **kwargs: ( + make_response(responses[url]) + ) + self.collector._GCRCollector__get_project_instance_metadata() + assert self.collector.project_metadata == {"projectId": "project"} + assert self.collector.instance_metadata == {"id": "instance"} + assert self.collector._GCRCollector__last_gcr_md_full_fetch > 0 + for call in self.collector._http_client.get.call_args_list: + assert call.kwargs["headers"] == {"Metadata-Flavor": "Google"} + + def test_get_project_instance_metadata_exception( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + self.collector._http_client = MagicMock() + self.collector._http_client.get.side_effect = RuntimeError("boom") + self.collector._GCRCollector__get_project_instance_metadata() + assert self.collector.project_metadata is None + assert "get_project_instance_metadata" in caplog.text + + def test_should_send_snapshot_data_by_interval(self) -> None: + self.collector.snapshot_data_last_sent = 0 + assert self.collector.should_send_snapshot_data() is True + self.collector.snapshot_data_last_sent = 2**40 + assert self.collector.should_send_snapshot_data() is False + + def prepare(self, monkeypatch: pytest.MonkeyPatch, instance, project) -> FakeHelper: + def fake_fetch() -> None: + self.collector.instance_metadata = instance + self.collector.project_metadata = project + + monkeypatch.setattr( + self.collector, "_GCRCollector__get_project_instance_metadata", fake_fetch + ) + helper = FakeHelper(plugins=[{"name": "plugin"}]) + self.collector.helpers = [helper] + return helper + + def test_prepare_payload_with_snapshot( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + helper = self.prepare(monkeypatch, {"id": "i"}, {"projectId": "p"}) + payload = self.collector.prepare_payload() + assert payload["metrics"]["plugins"] == [{"name": "plugin"}] + assert helper.kwargs == { + "with_snapshot": True, + "instance_metadata": {"id": "i"}, + "project_metadata": {"projectId": "p"}, + } + assert self.collector.snapshot_data_last_sent > 0 + + def test_prepare_payload_without_snapshot( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + helper = self.prepare(monkeypatch, {"id": "i"}, None) + self.collector.snapshot_data_last_sent = 2**40 + self.collector.prepare_payload() + assert helper.kwargs["with_snapshot"] is False + assert self.collector.snapshot_data_last_sent == 2**40 + + def test_prepare_payload_without_metadata( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + helper = self.prepare(monkeypatch, None, None) + payload = self.collector.prepare_payload() + assert payload["metrics"]["plugins"] == [] + assert helper.kwargs is None + + def test_prepare_payload_skipped_within_fetch_interval( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + helper = self.prepare(monkeypatch, {"id": "i"}, {}) + self.collector._GCRCollector__last_gcr_md_full_fetch = 2**40 + # fetching_start_time is set to "now", so delta is negative (< interval) + payload = self.collector.prepare_payload() + assert payload["metrics"]["plugins"] == [] + assert helper.kwargs is None + + def test_prepare_payload_with_spans(self, monkeypatch: pytest.MonkeyPatch) -> None: + self.prepare(monkeypatch, {"id": "i"}, {}) + monkeypatch.setattr( + "instana.collector.google_cloud_run.format_span", + lambda spans: ["formatted"], + ) + monkeypatch.setattr(self.collector, "queued_spans", lambda: ["span"]) + self.collector.span_queue.put("span") + assert self.collector.prepare_payload()["spans"] == ["formatted"] + + def test_prepare_payload_exception( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + self.prepare(monkeypatch, {"id": "i"}, {}) + self.collector.helpers = [FakeHelper(error=RuntimeError("boom"))] + payload = self.collector.prepare_payload() + assert payload["metrics"]["plugins"] == [] + assert "collect_snapshot error" in caplog.text + + def test_get_instance_id_with_and_without_metadata(self) -> None: + assert self.collector.get_instance_id() is None + self.collector.instance_metadata = {"id": "instance-id"} + assert self.collector.get_instance_id() == "instance-id" + + def test_get_instance_id_exception(self, caplog: pytest.LogCaptureFixture) -> None: + caplog.set_level("DEBUG", logger="instana") + self.collector.instance_metadata = ["not-a-dict"] + assert self.collector.get_instance_id() is None + assert "get_instance_id error" in caplog.text + + +class TestInstanceEntityHelper: + @pytest.fixture(autouse=True) + def _resource(self, monkeypatch: pytest.MonkeyPatch) -> Generator[None, None, None]: + monkeypatch.setenv("PORT", "8080") + self.collector = MagicMock() + self.collector.service = "service" + self.collector.configuration = "config" + self.collector.revision = "rev" + self.helper = InstanceEntityHelper(self.collector) + yield + self.helper = None + + def test_collect_metrics_builds_instance_entity(self) -> None: + plugins = self.helper.collect_metrics( + instance_metadata={"id": "instance", "region": "projects/1/regions/eu"}, + project_metadata={"numericProjectId": 1, "projectId": "project"}, + ) + assert len(plugins) == 1 + plugin = plugins[0] + assert plugin["name"] == "com.instana.plugin.gcp.run.revision.instance" + assert plugin["entityId"] == "instance" + data = plugin["data"] + assert data["runtime"] == "python" + assert data["region"] == "eu" + assert data["service"] == "service" + assert data["configuration"] == "config" + assert data["revision"] == "rev" + assert data["instanceId"] == "instance" + assert data["port"] == "8080" + assert data["numericProjectId"] == 1 + assert data["projectId"] == "project" + + def test_collect_metrics_exception_still_appends_plugin( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + plugins = self.helper.collect_metrics() + assert len(plugins) == 1 + assert plugins[0]["name"] == "com.instana.plugin.gcp.run.revision.instance" + assert "collect_service_revision_entity_metrics" in caplog.text diff --git a/tests/instrumentation/test_aws_triggers.py b/tests/instrumentation/test_aws_triggers.py new file mode 100644 index 00000000..87cd5869 --- /dev/null +++ b/tests/instrumentation/test_aws_triggers.py @@ -0,0 +1,407 @@ +# (c) Copyright IBM Corp. 2026 + +import base64 +import gzip +import json +from collections.abc import Callable, Generator +from typing import Any, Optional, Union +from unittest.mock import MagicMock + +import pytest + +from instana.instrumentation.aws.triggers import ( + capture_extra_headers, + enrich_lambda_span, + get_context, + is_api_gateway_proxy_trigger, + is_api_gateway_v2_proxy_trigger, + is_application_load_balancer_trigger, + is_cloudwatch_logs_trigger, + is_cloudwatch_trigger, + is_s3_trigger, + is_sqs_trigger, + read_http_query_params, +) +from instana.propagators.format import Format + +API_GATEWAY_EVENT = { + "resource": "/{proxy+}", + "path": "/path/to/resource", + "httpMethod": "POST", + "headers": {"X-Custom": "custom-value"}, + "queryStringParameters": {"foo": "bar"}, +} + +API_GATEWAY_V2_EVENT = { + "version": "2.0", + "routeKey": "GET /items/{id}", + "requestContext": { + "apiId": "api", + "stage": "$default", + "http": {"method": "GET", "path": "/items/42"}, + }, + "headers": {"X-Custom": "custom-value"}, + "queryStringParameters": {"a": "b"}, +} + +ALB_EVENT = { + "requestContext": {"elb": {"targetGroupArn": "arn"}}, + "httpMethod": "GET", + "path": "/alb", + "headers": {"X-Custom": "custom-value"}, + "queryStringParameters": {"x": "y"}, +} + +CLOUDWATCH_EVENT = { + "id": "event-id", + "source": "aws.events", + "detail-type": "Scheduled Event", + "resources": ["arn:1", "arn:2"], +} + +S3_EVENT = { + "Records": [ + { + "eventSource": "aws:s3", + "eventName": "ObjectCreated:Put", + "s3": {"bucket": {"name": "my-bucket"}, "object": {"key": "my-key"}}, + } + ] +} + +SQS_EVENT = { + "Records": [{"eventSource": "aws:sqs", "eventSourceARN": "arn:aws:sqs:queue"}] +} + + +def make_logs_event(log_data: dict[str, Any]) -> dict[str, Any]: + compressed = gzip.compress(json.dumps(log_data).encode("utf-8")) + return {"awslogs": {"data": base64.b64encode(compressed).decode("utf-8")}} + + +def get_attributes(span: MagicMock) -> dict[str, Any]: + return {call.args[0]: call.args[1] for call in span.set_attribute.call_args_list} + + +class TestTriggerDetection: + def test_is_api_gateway_proxy_trigger(self) -> None: + assert is_api_gateway_proxy_trigger(API_GATEWAY_EVENT) + assert not is_api_gateway_proxy_trigger({"resource": "/", "path": "/"}) + + @pytest.mark.parametrize( + "event, expected", + [ + (API_GATEWAY_V2_EVENT, True), + ({"requestContext": {}}, False), + ({"version": "1.0", "requestContext": {}}, False), + ({"version": "2.0", "requestContext": {"apiId": "a"}}, False), + ], + ) + def test_is_api_gateway_v2_proxy_trigger( + self, event: dict[str, Any], expected: bool + ) -> None: + assert is_api_gateway_v2_proxy_trigger(event) == expected + + def test_is_application_load_balancer_trigger(self) -> None: + assert is_application_load_balancer_trigger(ALB_EVENT) + assert not is_application_load_balancer_trigger({"requestContext": {}}) + assert not is_application_load_balancer_trigger({}) + + @pytest.mark.parametrize( + "event, expected", + [ + (CLOUDWATCH_EVENT, True), + ({"source": "aws.events"}, False), + ({"source": "other", "detail-type": "Scheduled Event"}, False), + ({"source": "aws.events", "detail-type": "Other"}, False), + ], + ) + def test_is_cloudwatch_trigger(self, event: dict[str, Any], expected: bool) -> None: + assert is_cloudwatch_trigger(event) == expected + + def test_is_cloudwatch_logs_trigger(self) -> None: + assert is_cloudwatch_logs_trigger({"awslogs": {"data": "x"}}) + assert not is_cloudwatch_logs_trigger({}) + assert not is_cloudwatch_logs_trigger("not-a-dict") + + @pytest.mark.parametrize( + "checker, event, expected", + [ + (is_s3_trigger, S3_EVENT, True), + (is_s3_trigger, SQS_EVENT, False), + (is_s3_trigger, {"Records": []}, False), + (is_s3_trigger, {}, False), + (is_sqs_trigger, SQS_EVENT, True), + (is_sqs_trigger, S3_EVENT, False), + (is_sqs_trigger, {"Records": []}, False), + (is_sqs_trigger, {}, False), + ], + ) + def test_is_records_trigger( + self, + checker: Callable[[dict[str, Any]], bool], + event: dict[str, Any], + expected: bool, + ) -> None: + assert checker(event) == expected + + +class TestGetContext: + @pytest.mark.parametrize( + "event, expected_carrier", + [ + (API_GATEWAY_EVENT, API_GATEWAY_EVENT["headers"]), + (API_GATEWAY_V2_EVENT, API_GATEWAY_V2_EVENT["headers"]), + (ALB_EVENT, ALB_EVENT["headers"]), + ], + ) + def test_get_context_proxy_event_uses_headers( + self, event: dict[str, Any], expected_carrier: dict[str, Any] + ) -> None: + tracer = MagicMock() + result = get_context(tracer, event) + assert result == tracer.extract.return_value + tracer.extract.assert_called_once_with( + Format.HTTP_HEADERS, expected_carrier, disable_w3c_trace_context=True + ) + + def test_get_context_non_proxy_event_uses_event(self) -> None: + tracer = MagicMock() + event = {"custom": "event"} + result = get_context(tracer, event) + assert result == tracer.extract.return_value + tracer.extract.assert_called_once_with( + Format.HTTP_HEADERS, event, disable_w3c_trace_context=True + ) + + +class TestReadHttpQueryParams: + @pytest.mark.parametrize( + "event, expected", + [ + (None, ""), + ("string", ""), + ({}, ""), + ({"queryStringParameters": {"a": "b", "c": "d"}}, "a=b&c=d"), + ({"multiValueQueryStringParameters": {"a": ["1", "2"]}}, "a=['1', '2']"), + ( + { + "multiValueQueryStringParameters": {"m": "1"}, + "queryStringParameters": {"q": "2"}, + }, + "m=1", + ), + ], + ) + def test_read_http_query_params( + self, event: Optional[Union[dict[str, Any], str]], expected: str + ) -> None: + assert read_http_query_params(event) == expected + + def test_read_http_query_params_exception( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + + class BadValue: + def __str__(self) -> str: + raise RuntimeError("boom") + + event = {"queryStringParameters": {"a": BadValue()}} + assert read_http_query_params(event) == "" + assert "read_http_query_params error" in caplog.text + + +class TestCaptureExtraHeaders: + def test_capture_extra_headers_case_insensitive(self) -> None: + span = MagicMock() + event = {"headers": {"x-custom": "value", "Other": "ignored"}} + capture_extra_headers(event, span, ["X-Custom"]) + span.set_attribute.assert_called_once_with("http.header.X-Custom", "value") + + def test_capture_extra_headers_no_headers(self) -> None: + span = MagicMock() + capture_extra_headers({}, span, ["X-Custom"]) + span.set_attribute.assert_not_called() + + def test_capture_extra_headers_exception( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + span = MagicMock() + capture_extra_headers(None, span, ["X-Custom"]) + span.set_attribute.assert_not_called() + assert "capture_extra_headers error" in caplog.text + + +class TestEnrichLambdaSpan: + @pytest.fixture(autouse=True) + def _resource(self) -> Generator[None, None, None]: + self.agent = MagicMock() + self.agent.collector.get_fq_arn.return_value = "arn:aws:lambda:fn" + self.agent.options.extra_http_headers = ["X-Custom"] + self.span = MagicMock() + self.context = MagicMock() + self.context.function_name = "my-function" + self.context.function_version = "$LATEST" + yield + self.span = None + + def enrich( + self, event: Optional[Union[dict[str, Any], str, list[str]]] + ) -> dict[str, Any]: + enrich_lambda_span(self.agent, self.span, event, self.context) + return get_attributes(self.span) + + def test_enrich_lambda_span_unknown_trigger_sets_base_attributes(self) -> None: + attributes = self.enrich({"unknown": "event"}) + assert attributes["lambda.arn"] == "arn:aws:lambda:fn" + assert attributes["lambda.name"] == "my-function" + assert attributes["lambda.version"] == "$LATEST" + assert attributes["lambda.trigger"] == "unknown" + + @pytest.mark.parametrize("event", [None, {}, "string", ["list"]]) + def test_enrich_lambda_span_bad_event( + self, event: Optional[Union[dict[str, Any], str, list[str]]] + ) -> None: + attributes = self.enrich(event) + assert "lambda.trigger" not in attributes + assert attributes["lambda.name"] == "my-function" + + def test_enrich_lambda_span_api_gateway(self) -> None: + attributes = self.enrich(API_GATEWAY_EVENT) + assert attributes["lambda.trigger"] == "aws:api.gateway" + assert attributes["http.method"] == "POST" + assert attributes["http.url"] == "/path/to/resource" + assert attributes["http.path_tpl"] == "/{proxy+}" + assert attributes["http.params"] == "foo=bar" + assert attributes["http.header.X-Custom"] == "custom-value" + + def test_enrich_lambda_span_api_gateway_no_extra_headers(self) -> None: + self.agent.options.extra_http_headers = None + attributes = self.enrich(API_GATEWAY_EVENT) + assert "http.header.X-Custom" not in attributes + + def test_enrich_lambda_span_api_gateway_v2(self) -> None: + attributes = self.enrich(API_GATEWAY_V2_EVENT) + assert attributes["lambda.trigger"] == "aws:api.gateway" + assert attributes["http.method"] == "GET" + assert attributes["http.url"] == "/items/42" + assert attributes["http.path_tpl"] == "/items/{id}" + assert attributes["http.params"] == "a=b" + assert attributes["http.header.X-Custom"] == "custom-value" + + def test_enrich_lambda_span_api_gateway_v2_no_extra_headers(self) -> None: + self.agent.options.extra_http_headers = None + attributes = self.enrich(API_GATEWAY_V2_EVENT) + assert "http.header.X-Custom" not in attributes + + def test_enrich_lambda_span_application_load_balancer(self) -> None: + attributes = self.enrich(ALB_EVENT) + assert attributes["lambda.trigger"] == "aws:application.load.balancer" + assert attributes["http.method"] == "GET" + assert attributes["http.url"] == "/alb" + assert attributes["http.params"] == "x=y" + assert attributes["http.header.X-Custom"] == "custom-value" + + def test_enrich_lambda_span_application_load_balancer_no_extra_headers( + self, + ) -> None: + self.agent.options.extra_http_headers = None + attributes = self.enrich(ALB_EVENT) + assert "http.header.X-Custom" not in attributes + + def test_enrich_lambda_span_cloudwatch_events(self) -> None: + attributes = self.enrich(CLOUDWATCH_EVENT) + assert attributes["lambda.trigger"] == "aws:cloudwatch.events" + assert attributes["data.lambda.cw.events.id"] == "event-id" + assert attributes["lambda.cw.events.more"] is False + assert attributes["lambda.cw.events.resources"] == ["arn:1", "arn:2"] + + def test_enrich_lambda_span_cloudwatch_events_truncation(self) -> None: + event = dict(CLOUDWATCH_EVENT) + event["resources"] = ["a" * 250, "b", "c", "d"] + attributes = self.enrich(event) + assert attributes["lambda.cw.events.more"] is True + assert attributes["lambda.cw.events.resources"] == ["a" * 200, "b", "c"] + + def test_enrich_lambda_span_cloudwatch_logs(self) -> None: + log_data = { + "logGroup": "group", + "logStream": "stream", + "logEvents": [{"message": "m1"}, {"message": "x" * 250}], + } + attributes = self.enrich(make_logs_event(log_data)) + assert attributes["lambda.trigger"] == "aws:cloudwatch.logs" + assert attributes["lambda.cw.logs.group"] == "group" + assert attributes["lambda.cw.logs.stream"] == "stream" + assert "lambda.cw.logs.more" not in attributes + assert attributes["lambda.cw.logs.events"] == ["m1", "x" * 200] + + def test_enrich_lambda_span_cloudwatch_logs_more_than_three_events(self) -> None: + log_data = { + "logGroup": "group", + "logStream": "stream", + "logEvents": [{"message": f"m{i}"} for i in range(5)], + } + attributes = self.enrich(make_logs_event(log_data)) + assert attributes["lambda.cw.logs.more"] is True + assert attributes["lambda.cw.logs.events"] == ["m0", "m1", "m2"] + + def test_enrich_lambda_span_cloudwatch_logs_without_data(self) -> None: + attributes = self.enrich({"awslogs": {}}) + assert attributes["lambda.trigger"] == "aws:cloudwatch.logs" + assert "lambda.cw.logs.events" not in attributes + + def test_enrich_lambda_span_cloudwatch_logs_decoding_error(self) -> None: + attributes = self.enrich({"awslogs": {"data": "not-base64-gzip"}}) + assert attributes["lambda.trigger"] == "aws:cloudwatch.logs" + assert "lambda.cw.logs.decodingError" in attributes + + def test_enrich_lambda_span_s3(self) -> None: + attributes = self.enrich(S3_EVENT) + assert attributes["lambda.trigger"] == "aws:s3" + assert attributes["lambda.s3.events"] == [ + { + "event": "ObjectCreated:Put", + "bucket": "my-bucket", + "object": "my-key", + } + ] + + def test_enrich_lambda_span_s3_missing_details_and_long_key(self) -> None: + event = { + "Records": [ + {"eventSource": "aws:s3", "eventName": "Created"}, + { + "eventSource": "aws:s3", + "eventName": "Removed", + "s3": {"bucket": {"name": "b"}, "object": {"key": "k" * 250}}, + }, + { + "eventSource": "aws:s3", + "eventName": "Other", + "s3": {"bucket": {"name": "b"}, "object": {}}, + }, + ] + } + attributes = self.enrich(event) + assert attributes["lambda.s3.events"] == [ + {"event": "Created", "bucket": "Unknown", "object": ""}, + {"event": "Removed", "bucket": "b", "object": "k" * 200}, + {"event": "Other", "bucket": "b", "object": "Unknown"}, + ] + + def test_enrich_lambda_span_sqs(self) -> None: + attributes = self.enrich(SQS_EVENT) + assert attributes["lambda.trigger"] == "aws:sqs" + assert attributes["lambda.sqs.messages"] == [{"queue": "arn:aws:sqs:queue"}] + + def test_enrich_lambda_span_exception( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level("DEBUG", logger="instana") + self.agent.collector.get_fq_arn.side_effect = RuntimeError("boom") + enrich_lambda_span(self.agent, self.span, {}, self.context) + self.span.set_attribute.assert_not_called() + assert "enrich_lambda_span error" in caplog.text