test: add semantic convention phase one closure gate

Makefile

@@ -220,6 +220,10 @@ py-test-teardown: ## Tear down the shared SigNoz backend
 py-test: ## Runs integration tests
 	@cd tests && uv run pytest --basetemp=./tmp/ -vv --capture=no integration/tests/
 
+.PHONY: py-test-semconv-phase1
+py-test-semconv-phase1: py-test-setup ## Rebuild the shared stack and run the semantic-convention Phase 1 matrix
+	@cd tests && uv run pytest --basetemp=./tmp/ -vv --reuse --capture=no integration/tests/queriertraces/13_semconv_evolution.py
+
 .PHONY: py-clean
 py-clean: ## Clear all pycache and pytest cache from tests directory recursively
 	@echo ">> cleaning python cache files from tests directory"

tests/integration/tests/queriertraces/13_semconv_evolution.py

@@ -0,0 +1,239 @@
+"""Phase 1 end-to-end checks for semantic-convention name evolution.
+
+The fixture models a fleet split across SDK generations and deliberately includes
+a dual-emitting conflict. Both request spellings must address one logical field,
+with the current spelling winning when a row contains both.
+"""
+
+from collections.abc import Callable, Generator
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+from typing import Any
+
+import pytest
+import requests
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import Aggregation, BuilderQuery, OrderBy, RequestType, TelemetryFieldKey, make_query_request
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
+
+CURRENT = "deployment.environment.name"
+OLD = "deployment.environment"
+PREFIX = "semconv-phase1"
+
+PRODUCTION_SPANS = {
+    f"{PREFIX}-old",
+    f"{PREFIX}-current",
+    f"{PREFIX}-both",
+    f"{PREFIX}-conflict",
+}
+STAGING_SPANS = {f"{PREFIX}-staging"}
+MISSING_SPANS = {f"{PREFIX}-missing"}
+
+
+def _span(timestamp: datetime, suffix: str, environment: dict[str, str]) -> Traces:
+    service = f"{PREFIX}-{suffix}"
+    return Traces(
+        timestamp=timestamp,
+        duration=timedelta(milliseconds=10),
+        trace_id=TraceIdGenerator.trace_id(),
+        span_id=TraceIdGenerator.span_id(),
+        name=service,
+        kind=TracesKind.SPAN_KIND_SERVER,
+        status_code=TracesStatusCode.STATUS_CODE_OK,
+        resources={"service.name": service, **environment},
+        attributes=dict(environment),
+    )
+
+
+@pytest.fixture(name="semconv_phase1_data")
+def semconv_phase1_data(
+    insert_traces: Callable[[list[Traces]], None],
+    clickhouse: types.TestContainerClickhouse,
+) -> Generator[datetime]:
+    now = datetime.now(tz=UTC).replace(microsecond=0) - timedelta(minutes=2)
+    insert_traces(
+        [
+            _span(now - timedelta(seconds=5), "old", {OLD: "production"}),
+            _span(now - timedelta(seconds=4), "current", {CURRENT: "production"}),
+            _span(now - timedelta(seconds=3), "both", {OLD: "production", CURRENT: "production"}),
+            _span(now - timedelta(seconds=2), "conflict", {OLD: "staging", CURRENT: "production"}),
+            _span(now - timedelta(seconds=1), "staging", {OLD: "staging"}),
+            _span(now, "missing", {}),
+        ]
+    )
+
+    # Service-map rows are derived by the collector in production. Seed the
+    # derived table directly here so the backend alias allowlist is tested in
+    # isolation; the collector repository owns its write-path integration test.
+    for environment, suffix in (("production", "production"), ("staging", "staging")):
+        clickhouse.conn.command(
+            f"""
+            INSERT INTO signoz_traces.distributed_dependency_graph_minutes_v2
+                (src, dest, duration_quantiles_state, error_count, total_count, timestamp,
+                 deployment_environment, k8s_cluster_name, k8s_namespace_name)
+            SELECT
+                '{PREFIX}-map-{suffix}', '{PREFIX}-map-child',
+                quantilesState(0.5, 0.75, 0.9, 0.95, 0.99)(toFloat64(1000000)),
+                toUInt64(0), toUInt64(1), toDateTime({int(now.timestamp())}),
+                '{environment}', '', ''
+            """
+        )
+
+    yield now
+
+    cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
+    clickhouse.conn.command(
+        f"ALTER TABLE signoz_traces.dependency_graph_minutes_v2 ON CLUSTER '{cluster}' "
+        f"DELETE WHERE startsWith(src, '{PREFIX}-map-') SETTINGS mutations_sync = 1"
+    )
+
+
+def _result(response: requests.Response) -> dict[str, Any]:
+    assert response.status_code == HTTPStatus.OK, response.text
+    results = response.json()["data"]["data"]["results"]
+    assert len(results) == 1
+    return results[0]
+
+
+def _raw_names(
+    signoz: types.SigNoz,
+    token: str,
+    now: datetime,
+    expression: str,
+) -> set[str]:
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((now + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            BuilderQuery(
+                signal="traces",
+                name="A",
+                limit=100,
+                filter_expression=expression,
+                select_fields=[TelemetryFieldKey("span.name")],
+                order=[OrderBy(TelemetryFieldKey("timestamp"), "asc")],
+            ).to_dict()
+        ],
+    )
+    return {row["data"]["name"] for row in (_result(response).get("rows") or [])}
+
+
+def _metadata_values(signoz: types.SigNoz, token: str, name: str, context: str) -> set[str]:
+    response = requests.get(
+        signoz.self.host_configs["8080"].get("/api/v1/fields/values"),
+        timeout=5,
+        headers={"authorization": f"Bearer {token}"},
+        params={
+            "signal": "traces",
+            "name": name,
+            "fieldContext": context,
+            "fieldDataType": "string",
+        },
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    return set(response.json()["data"]["values"].get("stringValues") or [])
+
+
+def test_semconv_phase1_mixed_sdk_generations(  # pylint: disable=too-many-statements
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    semconv_phase1_data: datetime,
+) -> None:
+    now = semconv_phase1_data
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    # Resource and span-attribute paths share the same matrix. Run every
+    # operator with both the saved-query (old) and current request spellings.
+    for context in ("resource", "attribute"):
+        for requested in (CURRENT, OLD):
+            field = f"{context}.{requested}"
+            assert _raw_names(signoz, token, now, f"{field} = 'production'") == PRODUCTION_SPANS
+            assert _raw_names(signoz, token, now, f"{field} = 'staging'") == STAGING_SPANS
+            assert _raw_names(signoz, token, now, f"{field} != 'production'") == STAGING_SPANS
+            assert _raw_names(signoz, token, now, f"{field} EXISTS") == PRODUCTION_SPANS | STAGING_SPANS
+            assert _raw_names(signoz, token, now, f"{field} NOT EXISTS") == MISSING_SPANS
+
+            grouped = make_query_request(
+                signoz,
+                token,
+                start_ms=int((now - timedelta(minutes=2)).timestamp() * 1000),
+                end_ms=int((now + timedelta(minutes=1)).timestamp() * 1000),
+                request_type=RequestType.SCALAR,
+                queries=[
+                    BuilderQuery(
+                        signal="traces",
+                        name="A",
+                        filter_expression=f"{field} EXISTS",
+                        aggregations=[Aggregation("count()")],
+                        group_by=[TelemetryFieldKey(requested, "string", context)],
+                        order=[OrderBy(TelemetryFieldKey(requested, "string", context), "asc")],
+                    ).to_dict()
+                ],
+            )
+            result = _result(grouped)
+            assert result["columns"][0]["name"] == requested, "response identity must match the request spelling"
+            assert result["data"] == [["production", 4], ["staging", 1]]
+
+        assert _metadata_values(signoz, token, CURRENT, context) == {"production", "staging"}
+        assert _metadata_values(signoz, token, OLD, context) == {"production", "staging"}
+
+    keys_response = requests.get(
+        signoz.self.host_configs["8080"].get("/api/v1/fields/keys"),
+        timeout=5,
+        headers={"authorization": f"Bearer {token}"},
+        params={"signal": "traces", "searchText": OLD},
+    )
+    assert keys_response.status_code == HTTPStatus.OK, keys_response.text
+    keys = keys_response.json()["data"]["keys"]
+    assert CURRENT in keys
+    assert OLD not in keys
+
+    start_ns = str(int((now - timedelta(minutes=2)).timestamp() * 1_000_000_000))
+    end_ns = str(int((now + timedelta(minutes=1)).timestamp() * 1_000_000_000))
+    for requested in (CURRENT, OLD):
+        services_response = requests.post(
+            signoz.self.host_configs["8080"].get("/api/v2/services"),
+            timeout=30,
+            headers={"authorization": f"Bearer {token}"},
+            json={
+                "start": start_ns,
+                "end": end_ns,
+                "tags": [
+                    {
+                        "Key": requested,
+                        "Operator": "In",
+                        "StringValues": ["production"],
+                        "TagType": "ResourceAttribute",
+                    }
+                ],
+            },
+        )
+        assert services_response.status_code == HTTPStatus.OK, services_response.text
+        services = {item["serviceName"] for item in services_response.json()["data"]}
+        assert services == PRODUCTION_SPANS
+
+        map_response = requests.post(
+            signoz.self.host_configs["8080"].get("/api/v1/dependency_graph"),
+            timeout=30,
+            headers={"authorization": f"Bearer {token}"},
+            json={
+                "start": start_ns,
+                "end": end_ns,
+                "tags": [
+                    {
+                        "key": requested,
+                        "operator": "In",
+                        "stringValues": ["production"],
+                        "tagType": "ResourceAttribute",
+                    }
+                ],
+            },
+        )
+        assert map_response.status_code == HTTPStatus.OK, map_response.text
+        assert {edge["parent"] for edge in map_response.json()} == {f"{PREFIX}-map-production"}