feat(ingestion): per-connector CLI version matrix + resolution stamp …

[puneetagarwal-datahub]

feat(ingestion): per-connector CLI version matrix + audit stamp + test-connection path

Adds per-connector CLI version resolution so a Snowflake fix can ship without
forcing a bump on every other connector. A JSON matrix maps server release ×
connector → version, with optional per-deployment cohort allowlists for canary
rollouts.

Matrix key conventions: Server-version keys are git describe --tags truncated at the first - (so v1.1.0-cloud → v1.1.0); CLI versions (_default, cohort version) are forwarded verbatim to pip install acryl-datahub==… and must use PyPI format (no v prefix).

Matrix schema:

{
  "v1.1.0": {                              // server key: GitHub tag with `-cloud` stripped
    "snowflake": {
      "_default": "1.6.0",                 // CLI: PyPI version, no `v`
      "cohorts": [
        { "version": "1.5.0.13.post3",     // CLI: PyPI version, no `v`
          "deployments": ["acme-corp"] }   // matches DataHubIngestionSourceConfig.deploymentId
      ]
    },
    "bigquery": { "_default": "1.6.0" }
  },
  "v1.0.2": {
    "snowflake": { "_default": "1.5.0.19" }
  }
}

deployments entries are matched against ingestion.deploymentId (bound to
the DATAHUB_EXECUTOR_CUSTOMER_ID env var injected by the Acryl Cloud Helm
chart from the K8s namespace). The leading underscore on _default marks it
as a sentinel key so it can't collide with a connector name.

Resolution priority (top wins):

1. Per-source explicit config.version (SOURCE_CONFIG_OVERRIDE)
2. matrix[serverVersion][connectorType].cohorts (MATRIX_COHORT)
   first cohort whose deployments list contains this
   deployment's id
3. matrix[serverVersion][connectorType]._default (MATRIX_CONNECTOR_DEFAULT)
4. defaultCliVersion from application.yaml (APPLICATION_DEFAULT)

Storage is pluggable. The consumer (IngestionCliVersionMatrixService) is
decoupled from where the matrix lives via the IngestionCliVersionMatrixSource
interface, selected by ingestion.cliVersionMatrix.source:

• source: "http" (default) + http.url set
  → HttpUrlIngestionCliVersionMatrixSource (periodic HTTP GET)
• source: "http" + http.url empty (OSS default)
  → NoOpIngestionCliVersionMatrixSource (empty matrix, never any null checks)
• source: "none"
  → NoOpIngestionCliVersionMatrixSource (explicit kill-switch, wins over URL)

Future backends (GMS aspect on a globalSettings entity, AppConfig / Consul /
etcd, …) add their own discriminator value here — the resolver and the audit
stamp don't change.

Three execution entry points are covered:

• CreateIngestionExecutionRequestResolver (manual triggers)
• IngestionScheduler.ExecutionRequestRunnable (scheduled triggers)
• CreateTestConnectionRequestResolver (test-connection from UI)

Each one stamps a structured CliVersionAudit record on the resulting
ExecutionRequestInput aspect, alongside the resolved version string in
args.version. Post-hoc forensics queries both with one SQL pass rather than
iterating the generic args map:

args.version : "1.3.1.5"
cliVersionAudit : { source: MATRIX_COHORT, serverVersion: "v1.6.0rc1" }

Each call site also emits a structured DEBUG log (and WARN when the resolved
version is empty — i.e. every tier including defaultCliVersion fell through)
under the caller's class so operators can grep all three trigger paths with
grep "Resolved ingestion CLI version".

Test connections also now honor defaultCliVersion when no explicit version is
supplied — previously they silently omitted version, causing the executor to
use whatever bundled CLI it shipped with rather than the configured default.

Feature is off by default — when no http.url is set the factory wires a
no-op source and every resolveVersion() returns empty, preserving existing
behavior exactly. In single-tenant deployments without a deployment id set,
cohort matching never fires and only the per-connector _default applies.

Also: treat empty / whitespace-only version strings as unset (bootstrap YAML
can render version: " " when the source has no version pin; forwarding
that blank to the executor would silently fall back to the bundled CLI).

Config (nested under ingestion.cliVersionMatrix):

INGESTION_VERSION_MATRIX_SOURCE (default: "http"; "none" disables)
INGESTION_VERSION_MATRIX_URL (default: empty → no-op)
INGESTION_VERSION_MATRIX_REFRESH_SECONDS (default: 600)
INGESTION_VERSION_MATRIX_AUTH_TOKEN (optional; sent as Authorization
header verbatim — supports a
private gist / S3 / CDN endpoint)
DATAHUB_EXECUTOR_CUSTOMER_ID (deployment id for cohort matching)

[github-actions]

Linear: CAT-2046

Thanks for your contribution! We have created an internal ticket to track this PR. A member of the core DataHub team will be assigned to review it within the next few business days - you will get a follow-up comment once a reviewer is assigned.

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
22669	1	22668	184

View the top 1 failed test(s) by shortest run time

tests.integration.test_plugin::test_airflow_plugin[v2_sqlite_operator]

Stack Traces | 36.6s run time

tmp_path = PosixPath('.../pytest-of-runner/pytest-0/test_airflow_plugin_v2_sqlite_0')
golden_filename = 'v2_sqlite_operator'
test_case = DagTestCase(dag_id='sqlite_operator', success=True, multiple_connections=False, platform_instance='myairflow', enable_datajob_lineage=True, cluster=None, test_variant=None)

    @pytest.mark.parametrize(
        ["golden_filename", "test_case"],
        [
            pytest.param(
                f"v2_{test_case.dag_test_id}",
                test_case,
                id=f"v2_{test_case.dag_test_id}",
            )
            for test_case in test_cases
        ],
    )
    def test_airflow_plugin(
        tmp_path: pathlib.Path,
        golden_filename: str,
        test_case: DagTestCase,
    ) -> None:
        # This test:
        # - Configures the plugin.
        # - Starts a local airflow instance in a subprocess.
        # - Runs a DAG that uses an operator supported by the extractor.
        # - Waits for the DAG to complete.
        # - Validates the metadata generated against a golden file.
    
        golden_path = GOLDENS_FOLDER / f"{golden_filename}.json"
    
        dag_id = test_case.dag_id
    
        with _run_airflow(
            tmp_path,
            dags_folder=DAGS_FOLDER,
            multiple_connections=test_case.multiple_connections,
            platform_instance=test_case.platform_instance,
            enable_datajob_lineage=test_case.enable_datajob_lineage,
            cluster=test_case.cluster,
        ) as airflow_instance:
            print(f"Running DAG {dag_id}...")
            _wait_for_dag_to_load(airflow_instance, dag_id)
    
            trigger_cmd = [
                str(airflow_instance.airflow_executable),
                "dags",
                "trigger",
                "--logical-date",
                "2023-09-27T21:34:38+00:00",
                "-r",
                "manual_run_test",
                dag_id,
            ]
    
            subprocess.check_call(
                trigger_cmd,
                env=airflow_instance.env_vars,
            )
    
            print("Waiting for DAG to finish...")
>           _wait_for_dag_finish(
                airflow_instance, dag_id, require_success=test_case.success
            )

tests/integration/test_plugin.py:670: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
venv/lib/python3.11.............../site-packages/tenacity/__init__.py:338: in wrapped_f
    return copy(f, *args, **kw)
           ^^^^^^^^^^^^^^^^^^^^
venv/lib/python3.11.............../site-packages/tenacity/__init__.py:477: in __call__
    do = self.iter(retry_state=retry_state)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
venv/lib/python3.11.............../site-packages/tenacity/__init__.py:378: in iter
    result = action(retry_state)
             ^^^^^^^^^^^^^^^^^^^
venv/lib/python3.11.............../site-packages/tenacity/__init__.py:400: in <lambda>
    self._add_action_func(lambda rs: rs.outcome.result())
                                     ^^^^^^^^^^^^^^^^^^^
.../hostedtoolcache/Python/3.11.15....../x64/lib/python3.11....../concurrent/futures/_base.py:449: in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
.../hostedtoolcache/Python/3.11.15....../x64/lib/python3.11....../concurrent/futures/_base.py:401: in __get_result
    raise self._exception
venv/lib/python3.11.............../site-packages/tenacity/__init__.py:480: in __call__
    result = fn(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

airflow_instance = AirflowInstance(airflow_home=PosixPath('.../pytest-of-runner/pytest-0/test_airflow_plugin_v2_sqlite_0/airflow_home'), airflow_port=10071, pid=7360, env_vars={'PATH': '.../airflow-plugin/venv/bin:.../airflow-plugin/venv/bin:.../hostedtoolcache/Python/3.11.15/x64/bin:.../hostedtoolcache/Python/3.11.15/x64:.../hostedtoolcache/Java_Zulu_jdk/21.0.11-10/x64/bin:/snap/bin:.../home/runner/.local/bin:/opt/pipx_bin:.../home/runner/.cargo/bin:.../home/runner/.config.../composer/vendor/bin:.../usr/local/.ghcup/bin:.../home/runner/.dotnet/tools:.../usr/local/sbin:.../usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:.../usr/local/games:/snap/bin', 'AIRFLOW_HOME': '.../pytest-of-runner/pytest-0/test_airflow_plugin_v2_sqlite_0/airflow_home', 'no_proxy': '*', 'PYTHONFAULTHANDLER': '1', 'PYTHONPATH': '/tmp', 'AIRFLOW__API__PORT': '10071', 'AIRFLOW__API__BASE_URL': 'http://airflow.example.com', 'AIRFLOW__CORE__LOAD_EXAMPLES': 'False', 'AIRFLOW__CORE__DAGS_FOLDER': '.../tests/integration/dags', 'AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION': 'False', 'AIRFLOW__CORE__LAZY_LOAD_PLUGINS': 'True', 'AIRFLOW__DATAHUB__CONN_ID': 'datahub_file_default', 'AIRFLOW__DATAHUB__DAG_FILTER_STR': '{ "deny": ["dag_to_skip"] }', 'AIRFLOW_CONN_DATAHUB_FILE_DEFAULT': 'datahub-file://%2Ftmp%2Fpytest-of-runner%2Fpytest-0%2Ftest_airflow_plugin_v2_sqlite_0%2Fdatahub_metadata.json', 'AIRFLOW_CONN_MY_SNOWFLAKE': 'snowflake://fake_username:fake_password@/DATAHUB_TEST_SCHEMA?account=fake_account&database=DATAHUB_TEST_DATABASE&warehouse=fake_warehouse&role=fake_role&insecure_mode=true', 'AIRFLOW_CONN_MY_AWS': 'aws:///?region_name=us-east-1&aws_access_key_id=AKIAIOSFODNN7EXAMPLE&aws_secret_access_key=wJalrXUtnFEMI%2FK7MDENG%2FbPxRfiCYEXAMPLEKEY', 'AIRFLOW_CONN_MY_BIGQUERY': 'google-cloud-platform:///?project=test_project&key_path=%2Fdev%2Fnull', 'AIRFLOW_CONN_MY_SQLITE': 'sqlite://%2Ftmp%2Fpytest-of-runner%2Fpytest-0%2Ftest_airflow_plugin_v2_sqlite_0%2Fmy_sqlite.db', 'AIRFLOW_CONN_MY_TERADATA': 'teradata://fake_username:fake_password@fake_teradata_host/?tmode=ANSI', 'DATAHUB_AIRFLOW_PLUGIN_RUN_IN_THREAD_TIMEOUT': '30', 'AIRFLOW__DATAHUB__LOG_LEVEL': 'DEBUG', 'AIRFLOW__DATAHUB__DEBUG_EMITTER': 'True', 'SQLALCHEMY_SILENCE_UBER_WARNING': '1', 'AIRFLOW__DATAHUB__ENABLE_DATAJOB_LINEAGE': 'true', 'AIRFLOW__API__AUTH_BACKENDS': 'airflow.api.auth.backend.anonymous', 'AIRFLOW__OPENLINEAGE__TRANSPORT': '{"type": "console"}', 'AIRFLOW__EXECUTION_API__JWT_EXPIRATION_TIME': '300', 'AIRFLOW__API_AUTH__JWT_SECRET': 'test-secret-key-for-jwt-signing-in-tests', 'AIRFLOW__DATAHUB__PLATFORM_INSTANCE': 'myairflow'}, airflow_executable=PosixPath('.../airflow-plugin/venv/bin/airflow'), username='admin', password='X5993NhDHfBc5ykr', metadata_file=PosixPath('.../pytest-of-runner/pytest-0/test_airflow_plugin_v2_sqlite_0/datahub_metadata.json'), metadata_file2=PosixPath('.../pytest-of-runner/pytest-0/test_airflow_plugin_v2_sqlite_0/datahub_metadata_2.json'))
dag_id = 'sqlite_operator', require_success = True

    @tenacity.retry(
        reraise=True,
        wait=tenacity.wait_fixed(1),
        stop=tenacity.stop_after_delay(90),
        retry=tenacity.retry_if_exception_type(NotReadyError),
    )
    def _wait_for_dag_finish(
        airflow_instance: AirflowInstance, dag_id: str, require_success: bool
    ) -> None:
        print("Checking if DAG is finished")
        res = _make_api_request(
            airflow_instance.session,
            f"{airflow_instance.airflow_url}.../api/v2/dags/{dag_id}/dagRuns",
        )
    
        dag_runs = res.json()["dag_runs"]
        if not dag_runs:
            raise NotReadyError("No DAG runs found")
    
        dag_run = dag_runs[0]
        if dag_run["state"] == "failed":
            if require_success:
>               raise ValueError("DAG failed")
E               ValueError: DAG failed

tests/integration/test_plugin.py:143: ValueError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[david-leifker]

Overall this looks reasonable. A few corner cases and gaps that need addressing.

[david-leifker]

This leads me to ask how this will work in airgapped environments. I had assumed that the configuration was from GMS. We need to support systems with no outside access.

[sgomezvillamor]

Good point @david-leifker !

The matrix storage is pluggable and configurable. HttpMatrixSource is just the first implementation. AFIK the plan is to serve it from S3 in a follow-up iteration, by implementing a new matrix source.

Neither HTTP nor S3 may work for air-gapped environments, but that's just a matter of providing a matrix source that fits their requirements, whether fetching from GMS itself or somewhere else (e.g. a ConfigMap in k8s?).

Is air-gapped support a blocking requirement for this first iteration?

[puneetagarwal-datahub]

Hi @david-leifker we have already mentioned this gap in RFC for the same . https://www.notion.so/acryldata/RFC-Independent-version-deployment-of-connectors-by-customer-and-connector-type-74afc6a6427783508d768150ae08be46?source=copy_link

On a high level thought , We can plugin the matrix inside the codebase which is shipped , start a local server and server json via that or we can think of something else .

[david-leifker]

As long as there is no regression in the current design, per source configuration with bundled and a common default configured via GMS's config endpoint, not a blocker.

[david-leifker]

Ideally there is a way to source this matrix from GMS itself, while a centralized external source is useful for Saas, there are on-prem scenarios where an individual operator may want to have control of this matrix without relying on an external system.

[sgomezvillamor]

We're certainly the most interested in this feature, managing a fleet of instances. It's true, it can be useful even for smaller onprem setups (dev, pre, prod).

Either way, this ties into my comment here: #17436 (comment) — the key point being that air-gapped support doesn't need to be solved now, as long as we know it can be addressed later with different matrix source implementations.

[david-leifker]

Approving, I would have liked to see a comprehensive implementation with GMS' config endpoint being able to serve the matrix configuration, passed through via environment variable containing a json configuration option. It would have been a simple and straightforward addition to serving it from a remote location.

[puneetagarwal-datahub]

Approving, I would have liked to see a comprehensive implementation with GMS' config endpoint being able to serve the matrix configuration, passed through via environment variable containing a json configuration option. It would have been a simple and straightforward addition to serving it from a remote location.

Good call — the source layer is pluggable by design .We kept the HTTP source as the default because the primary goal is updating per-connector versions fleet-wide without redeploying GMS ( edit the matrix once, all instances refresh in ~10min). Inline can't give that, but it's a great fit for self-contained/air-gapped deployments and is cheap to add alongside.

I will work on it once this is shipped and working .