test: add integration tests for semconv family resolution

A dedicated package runs SigNoz with resolve_semconv_families on and a
second instance with the flag at its default. The fleet has one
identity per state: OLD (old spelling only), NEW (current only), BOTH
(current staging and old production), NEITHER (keyless).

- 36 filter cells: nine operators, both spellings, both contexts. The
  result sets do not depend on the requested spelling; the current
  spelling wins on the conflict row; negative operators keep keyless
  rows, exactly like a single key.
- Group-by merges the fleet into one production group, a staging group,
  and a NULL group, and the group column carries the requested spelling.
- A bare name with the family under two contexts warns and keeps the
  resource side; the family survives the collision as one unit.
- Logs stay literal with the flag on, and everything stays literal with
  the flag off.

Verified: 42 passed against the live stack.

Assisted-by: Claude Fable 5

.github/workflows/integrationci.yaml

@@ -62,6 +62,7 @@ jobs:
           - role
           - rootuser
           - savedview
+          - semconvfamilies
           - serviceaccount
           - spanmapper
           - querier_json_body

tests/conftest.py

@@ -19,6 +19,7 @@ pytest_plugins = [
     "fixtures.traces",
     "fixtures.metrics",
     "fixtures.queriercommon",
+    "fixtures.semconvfamilies",
     "fixtures.metadata",
     "fixtures.meter",
     "fixtures.browser",

tests/fixtures/semconvfamilies.py

@@ -0,0 +1,64 @@
+from collections.abc import Callable, Generator
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from fixtures.logs import Logs
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
+
+PREFIX = "semconv-fam"
+CURRENT_KEY = "deployment.environment.name"
+OLD_KEY = "deployment.environment"
+
+# Row identities. The span name, the log body, and service.name are the identity.
+# Tests compare identity sets filtered by PREFIX, so reruns on a reused stack
+# with leftover rows stay stable.
+OLD = f"{PREFIX}-old"  # only the old spelling, value "production"
+NEW = f"{PREFIX}-new"  # only the current spelling, value "production"
+BOTH = f"{PREFIX}-both"  # current "staging" and old "production" - the conflict row
+NEITHER = f"{PREFIX}-neither"  # no member at all
+
+_ROWS = [
+    (OLD, {OLD_KEY: "production"}, timedelta(seconds=4)),
+    (NEW, {CURRENT_KEY: "production"}, timedelta(seconds=3)),
+    (BOTH, {CURRENT_KEY: "staging", OLD_KEY: "production"}, timedelta(seconds=2)),
+    (NEITHER, {}, timedelta(seconds=1)),
+]
+
+
+@pytest.fixture(name="family_fleet", scope="function")
+def family_fleet(
+    insert_logs: Callable[[list[Logs]], None],
+    insert_traces: Callable[[list[Traces]], None],
+) -> Generator[datetime]:
+    """Yields the base timestamp of the inserted rows."""
+    now = datetime.now(tz=UTC).replace(microsecond=0) - timedelta(minutes=1)
+
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - offset,
+                duration=timedelta(milliseconds=10),
+                trace_id=TraceIdGenerator.trace_id(),
+                span_id=TraceIdGenerator.span_id(),
+                name=identity,
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                resources={"service.name": identity, **family},
+                attributes=dict(family),
+            )
+            for identity, family, offset in _ROWS
+        ]
+    )
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - offset,
+                body=identity,
+                resources={"service.name": identity, **family},
+                attributes=dict(family),
+            )
+            for identity, family, offset in _ROWS
+        ]
+    )
+    yield now

tests/integration/tests/semconvfamilies/01_family_matrix.py

@@ -0,0 +1,218 @@
+from collections.abc import Callable
+from datetime import datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import (
+    RequestType,
+    build_aggregation,
+    build_group_by_field,
+    build_order_by,
+    build_raw_query,
+    build_traces_scalar_query,
+    get_column_data_from_response,
+    make_query_request,
+)
+from fixtures.semconvfamilies import (
+    BOTH,
+    CURRENT_KEY,
+    NEITHER,
+    NEW,
+    OLD,
+    OLD_KEY,
+    PREFIX,
+)
+
+FILTER_MATRIX = [
+    pytest.param("{key} = 'production'", {OLD, NEW}, id="eq_matches_either_spelling"),
+    pytest.param("{key} = 'staging'", {BOTH}, id="eq_current_wins_on_conflict"),
+    pytest.param("{key} != 'production'", {BOTH, NEITHER}, id="neq_keeps_keyless_and_conflict"),
+    pytest.param("{key} IN ['production', 'staging']", {OLD, NEW, BOTH}, id="in_matches_merged_value"),
+    pytest.param("{key} NOT IN ['production']", {BOTH, NEITHER}, id="not_in_keeps_keyless"),
+    pytest.param("{key} LIKE '%prod%'", {OLD, NEW}, id="like_matches_merged_value"),
+    pytest.param("{key} EXISTS", {OLD, NEW, BOTH}, id="exists_is_any_member"),
+    pytest.param("{key} NOT EXISTS", {NEITHER}, id="not_exists_is_no_member"),
+    pytest.param("{key} != 'production' AND {key} EXISTS", {BOTH}, id="neq_composed_with_exists"),
+]
+
+LITERAL_MATRIX = [
+    pytest.param("{key} = 'production'", {NEW}, id="literal_eq_reads_one_spelling"),
+    pytest.param("{key} != 'production'", {OLD, BOTH, NEITHER}, id="literal_neq_reads_one_spelling"),
+]
+
+
+@pytest.mark.parametrize("expression_template,expected", FILTER_MATRIX)
+@pytest.mark.parametrize("requested_key", [CURRENT_KEY, OLD_KEY], ids=["current", "old"])
+@pytest.mark.parametrize("context", ["resource", "attribute"])
+def test_family_filters(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+    context: str,
+    requested_key: str,
+    expression_template: str,
+    expected: set[str],
+) -> None:
+    """The result set is a property of the family, not of the requested spelling."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(key=f"{context}.{requested_key}")
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                "traces",
+                limit=100,
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": "span.name"}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    matched = {name for name in get_column_data_from_response(response.json(), "name") if name.startswith(PREFIX)}
+    assert matched == expected, expression
+
+
+@pytest.mark.parametrize("expression_template,expected", LITERAL_MATRIX)
+def test_flag_off_stays_literal(
+    signoz_families_off: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+    expression_template: str,
+    expected: set[str],
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(key=f"resource.{CURRENT_KEY}")
+    response = make_query_request(
+        signoz_families_off,
+        token,
+        start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                "traces",
+                limit=100,
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": "span.name"}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    matched = {name for name in get_column_data_from_response(response.json(), "name") if name.startswith(PREFIX)}
+    assert matched == expected, expression
+
+
+@pytest.mark.parametrize("expression_template,expected", LITERAL_MATRIX)
+def test_logs_stay_literal_with_flag_on(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+    expression_template: str,
+    expected: set[str],
+) -> None:
+    """Only traces have family support today."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(key=f"resource.{CURRENT_KEY}")
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                "logs",
+                limit=100,
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": "body"}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    matched = {body for body in get_column_data_from_response(response.json(), "body") if body.startswith(PREFIX)}
+    assert matched == expected, expression
+
+
+def test_group_by_merges_and_echoes_requested_spelling(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.SCALAR,
+        queries=[
+            build_traces_scalar_query(
+                [build_aggregation("count()")],
+                filter_expression=f"service.name LIKE '{PREFIX}%'",
+                group_by=[build_group_by_field(CURRENT_KEY, "string", "resource")],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    result = response.json()["data"]["data"]["results"][0]
+    group_column = result["columns"][0]
+    assert group_column["name"] == CURRENT_KEY, group_column
+    assert group_column["columnType"] == "group", group_column
+
+    groups = {row[0] for row in result["data"]}
+    assert {"production", "staging"}.issubset(groups), groups
+    assert None in groups, groups
+
+
+def test_bare_name_prefers_resource_and_warns(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    family_fleet: datetime,
+) -> None:
+    """Both contexts carry the family, so a bare name is ambiguous: resolution
+    warns and keeps the resource side."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                "traces",
+                limit=100,
+                filter_expression=f"{CURRENT_KEY} = 'production'",
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": "span.name"}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    matched = {name for name in get_column_data_from_response(response.json(), "name") if name.startswith(PREFIX)}
+    assert matched == {OLD, NEW}
+
+    warning = response.json()["data"].get("warning") or {}
+    messages = " ".join(entry.get("message", "") for entry in warning.get("warnings", []))
+    assert "ambiguous" in messages.lower(), messages

tests/integration/tests/semconvfamilies/conftest.py

@@ -0,0 +1,55 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.signoz import create_signoz
+
+
+@pytest.fixture(name="signoz", scope="package")
+def signoz_semconv_families(
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-semconv-families",
+        env_overrides={
+            "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_RESOLVE__SEMCONV__FAMILIES": True,
+        },
+    )
+
+
+@pytest.fixture(name="signoz_families_off", scope="package")
+def signoz_families_off(
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    """Shares the sqlstore and clickhouse with the flag-on instance, so the
+    same admin token and seeded rows work."""
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-semconv-families-off",
+        env_overrides={},
+    )