test: pin the semconv family matrix on a live stack

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

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,75 @@
+"""Seed data for the semconv family matrix tests.
+
+Four identities cover every fleet state of the deployment.environment(.name)
+family. The tests assert which identities a filter returns, so BOTH (a row
+that carries the two spellings with different values) and NEITHER (a keyless
+row) are the point of most cases.
+
+Each row carries its family pairs in the resource attributes and in the span
+attributes, so one fleet serves the resource-context and attribute-context
+matrices. The same rows exist as logs for the logs literalness guard.
+"""
+
+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.
+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]:
+    """Inserts one span and one log per identity and yields the base
+    timestamp."""
+    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,220 @@
+"""The phase-1 matrix for semantic-convention family resolution.
+
+The package runs SigNoz with resolve_semconv_families on. The fleet in
+fixtures/semconvfamilies.py has one identity per state: OLD (old spelling
+only), NEW (current only), BOTH (current "staging" and old "production"),
+NEITHER (keyless). Each case asserts which identities a filter returns, with
+either spelling as the requested name and for both contexts.
+
+The pinned facts:
+  - Both spellings resolve to the same merged field; the result sets do not
+    depend on the requested spelling.
+  - The current spelling wins on a row that carries both (BOTH reads
+    "staging", never "production").
+  - Negative operators keep keyless rows (NEITHER), exactly like a single
+    key; presence stays an explicit EXISTS opt-in.
+  - Logs stay literal: only traces have family support today.
+  - With the flag off, everything stays literal.
+"""
+
+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"),
+]
+
+
+def _trace_identities(
+    signoz: types.SigNoz,
+    token: str,
+    base: datetime,
+    expression: str,
+    signal: str = "traces",
+) -> set[str]:
+    identity_field = "span.name" if signal == "traces" else "body"
+    identity_column = "name" if signal == "traces" else "body"
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((base - timedelta(minutes=2)).timestamp() * 1000),
+        end_ms=int((base + timedelta(minutes=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            build_raw_query(
+                "A",
+                signal,
+                limit=100,
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "asc")],
+                select_fields=[{"name": identity_field}],
+            )
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    # Sets keep the assertion stable when the shared stack is reused and older
+    # rows with the same identities remain.
+    return {name for name in get_column_data_from_response(response.json(), identity_column) if name.startswith(PREFIX)}
+
+
+@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:
+    """One matrix cell: a filter on one spelling, in one context. 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}")
+    assert _trace_identities(signoz, token, family_fleet, expression) == 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:
+    """The same fleet through an instance with the flag at its default: the
+    current spelling reads only rows that carry the current spelling."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(key=f"resource.{CURRENT_KEY}")
+    assert _trace_identities(signoz_families_off, token, family_fleet, expression) == 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. The same filters on the logs copy of
+    the fleet behave literally even with the flag on."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    expression = expression_template.format(key=f"resource.{CURRENT_KEY}")
+    assert _trace_identities(signoz, token, family_fleet, expression, signal="logs") == 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:
+    """Group by the current spelling over the fleet: OLD and NEW land in one
+    "production" group, BOTH lands in "staging", and the group column carries
+    the requested spelling."""
+    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:
+    """The fleet stores the family under the resource and the attribute
+    contexts, so a bare name is ambiguous. Resolution warns and keeps the
+    resource side; the family survives the collision as one unit."""
+    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,56 @@
+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:
+    """Package-scoped SigNoz with resolve_semconv_families on."""
+    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:
+    """A second instance with the flag at its default (off). It shares the
+    sqlstore and clickhouse, 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={},
+    )