diff --git a/.secrets.baseline b/.secrets.baseline index 671caa03..b7a2d390 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -157,7 +157,7 @@ "filename": "AGENTS.md", "hashed_secret": "9d4e1e23bd5b727046a9e3b4b7db57bd8d6ee684", "is_verified": false, - "line_number": 641 + "line_number": 643 } ], "Jenkinsfile": [ @@ -2317,5 +2317,5 @@ } ] }, - "generated_at": "2026-11-12T20:45:31Z" + "generated_at": "2026-11-12T21:55:04Z" } diff --git a/AGENTS.md b/AGENTS.md index 64b2a500..d2102957 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -53,6 +53,8 @@ Every contract is delivered through an 11-stage pipeline. Each stage is a hard g The `replace` rows describe the native engine. On the OpenTofu engine (aws, gcp, snowflake, confluent) `replace` emits the same module as `amend` and adds the data-loss gate; a table is replaced only where `tofu` plans it, and on GCP its row access policies and table grants are recreated with it (`replace_triggered_by`). A BigQuery table is never recreated to replace its rows: the planner's replace and `fluid rollback` run a `WRITE_TRUNCATE_DATA` query job into the existing table (a `CREATE OR REPLACE TABLE` drops its row access policies and policy tags), and the pre-replace backup is a `CREATE SNAPSHOT TABLE ā€¦ CLONE`. dbt's `table` materialisation still recreates a governed table (known gap; `docs/governance-parity.md`). See `docs/apply.md`. +On Databricks a dbt build with `execution.runtime.platform: databricks` runs as the job `fluid apply` provisions. `amend-and-build` and `replace-and-build` start a run of the job this apply's OpenTofu state holds (found by its `fluid_product` / `fluid_build` tags), wait for it, and fail with the task's error and the end of dbt's output; `replace-and-build` adds `--full-refresh` for that run only. The job's settings, schedule and pause state are never written; `amend` and `create-only` only provision it, and a continuous job is never started. The wait is capped by `FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS` (an hour, or the job's own limits plus 15 minutes); at the cap, on Ctrl-C and on SIGTERM or SIGHUP (unless ignored, as under `nohup`) the run is cancelled, while a run that cannot be read five times in a row is left alone. To build an estate in lineage order, provision every product in parallel with `--mode amend`, then run `--mode amend-and-build` level by level; the second apply plans no change. Code: `build_runners/_databricks_job_run.py`. + ### Plan-Binding (Terraform-Style "Apply Consumes Exact Plan") Stage 6 `fluid plan` emits two cryptographic fields in `plan.json`: diff --git a/CHANGELOG.md b/CHANGELOG.md index 423ff8be..7b3820c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **`fluid apply --mode amend-and-build` runs a Databricks dbt build now.** A dbt build with + `execution.runtime.platform: databricks` runs as the serverless job `fluid apply` provisions. + `amend-and-build` and `replace-and-build` now start a run of that job after provisioning + and wait for it. They print the run page and each state the run enters, with the reason + while it is queued. A failed run fails the apply with the task's error and the end of + dbt's output. `replace-and-build` adds `--full-refresh` for that run only. Before, the apply + only provisioned the job and said how to start it, so a manual job never ran. + - `amend` and `create-only` still only provision, and a continuous job is not started. + - The job's settings, schedule and pause state are never changed. + - `FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS` caps the wait: an hour, or the job's own timeouts + and retries plus 15 minutes. The run is cancelled at the cap and on Ctrl-C, SIGTERM or + SIGHUP. A run that cannot be read five times in a row is left as it is, and the apply + says so. run-now is retried with the same idempotency token, so a retry never starts a + second run. + - A missing `databricks-sdk` (the `[databricks]` extra) is refused before anything is + applied. - **Keys for two or more providers stop the run, even with Ollama running.** With `OPENAI_API_KEY` and `ANTHROPIC_API_KEY` both set (or one provider's key and `OLLAMA_HOST`) and nothing naming a provider, a run without a TTY skipped both keys and used a running local diff --git a/CLAUDE.md b/CLAUDE.md index 4f00c221..9b7affb9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -434,6 +434,12 @@ hand-rolling the protocol. - **The validate hook** (`agent_skills/hooks/validate_contract.py`) reads the edited path from stdin JSON: hook commands get no `${tool_input.*}` substitution, and an `if` rule is one permission rule (`|` is not OR). One file serves the plugin (`python3 "${CLAUDE_PLUGIN_ROOT}/hooks/validate_contract.py"`) and scaffold-ide (` /hooks/validate_contract.py`, written to `.claude/settings.local.json`). **Never `python -m` in a hook or an agent-reachable shim:** it puts the working directory (the agent's project) first on `sys.path`, so a planted `json.py` runs (reproduced in the 2026-11-12 security review); the hook calls the `fluid` console script instead. - **`tests/agent_evals/`** is the eval harness (arms no-skill / skill / forge, deterministic grading). Its workspaces live in the system temp dir, never inside a repository: Claude Code loads every `CLAUDE.md` on the way up from the working directory, and a first run with workspaces under `tests/agent_evals/runs/` fed forge-cli's own CLAUDE.md (33.8k tokens) to every arm. The live test is marker `agent_eval` (deselected in `addopts`) plus `FLUID_AGENT_EVAL=1`; `test_harness.py` grades every scenario's reference solution in the default lane. A local DuckDB build that reads a sibling product's output needs `fluid.workspace.yaml` at the repo root. Under `--mode amend` the provider path never binds `consumes[]` and reads a `parameters.inputs` entry without a `format` as CSV (`providers/local/planner.py` fills in `csv` before the extension is looked at); `--mode amend-and-build` binds each `consumes[]` entry the SQL reads by its `exposeId` (`build_runners/_embedded_sql_io.py`) and infers an input's format from its extension. An input with a `format` works in both. +## Recent architectural changes worth knowing (2026-11-12, Databricks job runs) + +- **`-and-build` runs a Databricks dbt build now.** The seam: `cli/apply.py` passes `applied_state=[]` to `apply_via_opentofu` (only when `_databricks_job_runs` finds a non-continuous Databricks dbt build), the engine fills it with `runner.tofu_state_resources` after `post_apply` binds masks, and `jobs_in_state` (tags `databricks_jobs.PRODUCT_TAG` / `BUILD_TAG`, never the job name) hands `run_builds_from_args(databricks_jobs=...)` the ids. `build_runners/_databricks_job_run.py` does run-now (idempotency token, `queue.enabled`, tried 3 times with the same token), polls `runs/get` (5 failed reads in a row end the wait without a cancel: the SDK does not repeat a 500/502/504), and on failure reads `runs/get-output` per failed task. Ctrl-C, SIGTERM and SIGHUP cancel the run only while it is waited on. An id-less build's job is tagged by its position (`databricks_jobs.job_build_id`). `databricks_jobs=None` keeps the old "runs on its trigger" report. `_gate_databricks_sdk_for_apply` refuses a missing SDK before `tofu`. +- **Live-verified on Free Edition (2026-11-12):** a queued run's `runs/get` carries only `status.state: QUEUED` and an empty `state.state_message` (no `queue_details`), so the reason comes from `runs/list?active_only=true`; a dbt task's stdout is in `get-output.logs` with ANSI codes; `get-output.dbt_output.artifacts_link` is a pre-signed URL and is never printed; the deprecated run-now `dbt_commands` is still accepted and overrides the commands for that run only. +- Pins: `tests/build_runners/test_databricks_job_run.py`, `tests/build_runners/test_databricks_job_builds.py`, `tests/cli/test_apply_databricks_job_runs.py`, and the real-`tofu` `tests/iac/test_iac_databricks_jobs_fake_e2e.py` (`_fake_databricks.py` serves runs). + ## Working style - **Goal-driven execution.** For non-trivial tasks, state success criteria before implementing and loop until verified. Transform imperatives into tests: "add redactor coverage" ā†’ "new pattern is masked by a test assertion"; "fix the bug" ā†’ "reproducing test passes first, then fix makes it green". diff --git a/fluid_build/build_runners/_databricks_job_run.py b/fluid_build/build_runners/_databricks_job_run.py new file mode 100644 index 00000000..5ccfd49e --- /dev/null +++ b/fluid_build/build_runners/_databricks_job_run.py @@ -0,0 +1,729 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""Start a Databricks dbt job now and wait for it: the build phase of ``fluid apply``. + +A dbt build whose ``execution.runtime.platform`` is ``databricks`` runs in the workspace, as +the serverless job ``fluid apply`` provisions (``iac/providers/databricks_jobs.py``). Under +``--mode amend-and-build`` and ``--mode replace-and-build`` the build runner hands such a +build here once ``tofu apply`` has succeeded and the masks and row filters are bound: + +1. The job is the one THIS apply's OpenTofu state holds for the build, found by the tags the + emitter puts on it (``fluid_product``, ``fluid_build``; :func:`jobs_in_state`), never by + name: a job name is not unique in a workspace, and two environments of one contract + applied to one workspace each own a job of that name. +2. One run starts now (``POST /api/2.2/jobs/run-now``). It carries an idempotency token, so a + retried request cannot start a second run, and queueing on, so a run that finds the job + busy (``max_concurrent_runs`` is 1, and a scheduled run may be going) or the workspace at + its limit of concurrent tasks waits instead of being skipped. ``replace-and-build`` runs + the job's own commands with ``--full-refresh`` on ``dbt build`` (the run's + ``dbt_commands``), as the local runner adds it for that mode. Nothing is written to the + job: its settings, schedule and pause state stay as they are, so a scheduled job keeps + its schedule and a job created paused (``DatabricksIacPlugin.plan_notices``) stays paused. +3. The run is polled until it ends (``GET /api/2.2/jobs/runs/get``): its page URL first, then + a line for every state it enters, and a line a minute while nothing changes. A queued + run says why: Databricks' reason when it gives one; when it gives none (the usual case, + seen live), whether it waits behind another run of the job or for the workspace's limit + on runs or tasks at once (``GET /api/2.2/jobs/runs/list``, active runs of the job). +4. A run that does not end in ``SUCCESS`` fails the build, with its state and message and, + for each task that did not succeed, the task's ``error`` and the last lines of its output + (``GET /api/2.2/jobs/runs/get-output``). Every credential value in the environment is + masked in what is printed, and terminal control sequences are removed. + +The wait is capped (:func:`wait_budget`). At the cap, and on Ctrl-C, SIGTERM or SIGHUP while +the run is waited on, the run is cancelled before the build fails, so ``fluid apply`` leaves +nothing running that it started (an interrupted load cancels its INSERT the same way, +``_databricks_load.py``); a cancel that fails is said, with the run page. Failed reads alone +never cancel a run: one that cannot be read five times in a row is left as it is, and the +build fails saying so. run-now is tried again with the same idempotency token, so a retry never starts +a second run. A continuous job is never started here: Databricks keeps it running, and a +run of it does not end. + +Polling follows databricks-sdk-py's run waiter (``JobsAPI.wait_get_run_job_terminated_or_ +skipped``: a poll a second longer each time, at most 10 s; Apache-2.0), and the failed-task +report follows Apache Airflow's Databricks provider (``extract_failed_task_errors``: the +latest attempt of each failed task, its ``get-output`` error; Apache-2.0) and +``databricks bundle run``. The client is the provider's (``databricks_enforce.default_client``) +through ``api_client.do``; the SDK is imported only there (the ``[databricks]`` extra). +""" + +from __future__ import annotations + +import logging +import os +import re +import shlex +import signal +import threading +import time +import uuid +from contextlib import contextmanager +from dataclasses import dataclass +from typing import ( + Any, + Callable, + Dict, + Iterable, + Iterator, + List, + Mapping, + Optional, + Sequence, + Tuple, +) + +from fluid_build._console import cprint +from fluid_build._console import error as console_error +from fluid_build.util.contract import is_continuous_build, is_databricks_job_build + +LOG = logging.getLogger(__name__) + +#: The wait cap for one run, in seconds (:func:`wait_budget`). +TIMEOUT_ENV = "FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS" +DEFAULT_TIMEOUT_SECONDS = 3600 +#: Room past the job's own time limits, for the queue and the serverless start. +SLACK_SECONDS = 900 +#: A progress line at least this often while the run's state does not change. +HEARTBEAT_SECONDS = 60 +#: Lines of a failed task's output that are printed. +TAIL_LINES = 30 +#: Tries of run-now (one idempotency token), and reads of a run in a row that may fail. +_START_ATTEMPTS = 3 +_READ_ATTEMPTS = 5 + +_JOBS = "/api/2.2/jobs" +_REPLACE_MODES = ("replace", "replace-and-build") +#: The dbt commands that take ``--full-refresh``. +_FULL_REFRESH_COMMANDS = ("build", "run", "seed") +#: ECMA-48 escape sequences (as ``llm/coding_agent.py`` strips them), then every other C0 +#: and C1 control character except tab and newline. +_ANSI_RE = re.compile(r"\x1b(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") +_CONTROL_RE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f-\x9f]") +#: The ``HH:MM:SS`` dbt puts before each line it logs. +_DBT_TIMESTAMP_RE = re.compile(r"^\s*\d{2}:\d{2}:\d{2}\s+") + +Say = Callable[[str], None] + + +@dataclass(frozen=True) +class JobRef: + """A Databricks job an apply's OpenTofu state holds for one build.""" + + job_id: int + url: str = "" + + +def started_builds(contract: Mapping[str, Any], build_id: Optional[str]) -> List[Mapping[str, Any]]: + """The builds a ``-and-build`` apply of ``contract`` starts as Databricks job runs. + + An empty ``build_id`` selects every build, as the build runner's own filter does. + """ + return [ + build + for build in contract.get("builds") or [] + if isinstance(build, Mapping) + and is_databricks_job_build(dict(build)) + and not is_continuous_build(dict(build)) + and (not build_id or build.get("id") == build_id) + ] + + +def jobs_in_state(resources: Iterable[Any], product: str) -> Dict[str, List[JobRef]]: + """Build id to the ``databricks_job`` resources ``product`` has in an OpenTofu state. + + ``resources`` are a state's resources as ``runner.tofu_state_resources`` returns them. A + list per build, so that a caller can refuse a build with more than one. The tags are the + emitter's own (``databricks_jobs.PRODUCT_TAG`` / ``BUILD_TAG``). + """ + from fluid_build.iac.providers.databricks_jobs import BUILD_TAG, PRODUCT_TAG + + found: Dict[str, List[JobRef]] = {} + for resource in resources: + if not isinstance(resource, Mapping) or resource.get("type") != "databricks_job": + continue + if resource.get("mode", "managed") != "managed": + continue + values = resource.get("values") if isinstance(resource.get("values"), Mapping) else {} + tags = values.get("tags") if isinstance(values.get("tags"), Mapping) else {} + if tags.get(PRODUCT_TAG) != product or not tags.get(BUILD_TAG): + continue + try: + job_id = int(str(values.get("id"))) + except ValueError: + continue + found.setdefault(str(tags[BUILD_TAG]), []).append( + JobRef(job_id, str(values.get("url") or "")) + ) + return found + + +def wait_budget(settings: Mapping[str, Any], env: Mapping[str, str]) -> int: + """How long ``fluid apply`` waits for one run of a job with ``settings``, in seconds. + + ``FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS`` when it is a positive whole number. Otherwise an + hour, or, when it is longer, as long as the job may run by its own limits (the job's + ``timeout_seconds``, else every task's timeout times its attempts plus the retry + intervals) plus 15 minutes for the queue and the start, so a contract's own + ``runtime.timeout`` is never cut short. A task without a timeout, or one retried without + end, leaves the hour. + """ + override, _raw = _timeout_override(env) + if override is not None: + return override + own = _whole(settings.get("timeout_seconds")) + if not own: + tasks = [task for task in settings.get("tasks") or [] if isinstance(task, Mapping)] + for task in tasks: + timeout, retries = _whole(task.get("timeout_seconds")), _whole(task.get("max_retries")) + if not timeout or retries < 0: + own = 0 + break + interval = _whole(task.get("min_retry_interval_millis")) // 1000 + own += timeout * (retries + 1) + interval * retries + return max(DEFAULT_TIMEOUT_SECONDS, own + SLACK_SECONDS) if own > 0 else DEFAULT_TIMEOUT_SECONDS + + +def _timeout_override(env: Mapping[str, str]) -> Tuple[Optional[int], str]: + """``FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS`` as whole seconds above 0, or ``None``; and + the raw value. ASCII digits only: ``str.isdigit`` also takes ``Ā²``, which ``int`` refuses.""" + raw = str(env.get(TIMEOUT_ENV) or "").strip() + if raw.isascii() and raw.isdigit() and len(raw) <= 9 and int(raw) > 0: + return int(raw), raw + return None, raw + + +def full_refresh_commands(settings: Mapping[str, Any]) -> Optional[List[str]]: + """The job's dbt commands with ``--full-refresh`` on each that takes it, or ``None``.""" + for task in settings.get("tasks") or []: + dbt = task.get("dbt_task") if isinstance(task, Mapping) else None + if not isinstance(dbt, Mapping) or not dbt.get("commands"): + continue + commands = [] + for command in dbt["commands"]: + argv = shlex.split(str(command)) + if ( + argv[:1] == ["dbt"] + and argv[1:2] + and argv[1] in _FULL_REFRESH_COMMANDS + and "--full-refresh" not in argv + ): + argv.append("--full-refresh") + commands.append(shlex.join(argv)) + return commands + return None + + +def run_job_build( + contract: Mapping[str, Any], + build: Mapping[str, Any], + jobs: Mapping[str, Sequence[JobRef]], + *, + dry_run: bool = False, + mode: Optional[str] = None, + client_factory: Optional[Callable[[], Any]] = None, + env: Optional[Mapping[str, str]] = None, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.monotonic, +) -> int: + """Start a run of ``build``'s job now and wait for it to end. Returns 0 or 1. + + ``jobs`` is :func:`jobs_in_state` of the apply that provisioned it. + """ + from fluid_build.iac.providers.databricks_jobs import ( + BUILD_TAG, + PRODUCT_TAG, + job_build_id, + job_product, + ) + + build_id = job_build_id(contract, build) + product = job_product(contract) + name = f"fluid {product} {build_id}" + full_refresh = (mode or "").strip().lower() in _REPLACE_MODES + with_refresh = " with dbt build --full-refresh" if full_refresh else "" + if dry_run: + _say( + f"\nšŸ” [DRY RUN] Build '{build_id}' - would start a run of Databricks job " + f"'{name}'{with_refresh} and wait for it." + ) + return 0 + refs = list(jobs.get(build_id) or []) + if len(refs) != 1: + found = "no Databricks job" if not refs else f"{len(refs)} Databricks jobs" + console_error( + f"Build '{build_id}' - this apply's OpenTofu state holds {found} tagged " + f"{PRODUCT_TAG}={product}, {BUILD_TAG}={build_id}, so no run was started. Either " + "the state could not be read back after the apply (said above), or the contract " + "was not applied to databricks; fix that, then re-run fluid apply." + ) + return 1 + try: + client = (client_factory or _default_client)() + except ImportError: + console_error( + f"Build '{build_id}' - databricks-sdk is not installed, and the run of Databricks job " + f"'{name}' is started through it. Install the extra: " + "pip install 'data-product-forge[databricks]'" + ) + return 1 + except Exception as exc: # noqa: BLE001 ā€” a bad credential must not become a traceback + console_error( + f"Build '{build_id}' - could not create the Databricks client to start job " + f"'{name}': {_failure(exc)}" + ) + return 1 + return start_and_wait( + client, + refs[0], + build_id=build_id, + name=name, + full_refresh=full_refresh, + env=os.environ if env is None else env, + sleep=sleep, + clock=clock, + ) + + +def start_and_wait( + client: Any, + job: JobRef, + *, + build_id: str, + name: str, + full_refresh: bool = False, + env: Mapping[str, str], + say: Optional[Say] = None, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.monotonic, +) -> int: + """Run ``job`` now through ``client`` and wait for the run to end. Returns 0 or 1.""" + say = say or _say + api = client.api_client + label = f"Databricks job '{name}' ({job.job_id})" + try: + settings = _job_settings(api, job.job_id) + except Exception as exc: # noqa: BLE001 + console_error(f"Build '{build_id}' - could not read {label}: {_failure(exc)}") + return 1 + body: Dict[str, Any] = { + "job_id": job.job_id, + "idempotency_token": uuid.uuid4().hex, + "queue": {"enabled": True}, + } + if full_refresh: + commands = full_refresh_commands(settings) + if commands is None: + console_error( + f"Build '{build_id}' - replace-and-build runs dbt with --full-refresh, and " + f"{label} has no dbt commands to add it to, so no run was started. Re-run " + "fluid apply to bring the job back to what the contract declares." + ) + return 1 + body["dbt_commands"] = commands + # Decided before the run starts, so nothing between run-now and the wait can fail. + budget = wait_budget(settings, env) + override, raw = _timeout_override(env) + try: + run_id = _run_now(api, body, sleep) + except Exception as exc: # noqa: BLE001 + console_error(f"Build '{build_id}' - could not start a run of {label}: {_failure(exc)}") + if job.url: + say(f" if Databricks started it anyway, it is on the job's page: {clean(job.url)}") + return 1 + LOG.debug("databricks_job_run_started job_id=%s run_id=%s", job.job_id, run_id) + say(f"\nšŸš€ Build '{build_id}' - started run {run_id} of {label}") + if raw and override is None: + say( + f" {TIMEOUT_ENV}={clean(raw)!r} is not a whole number of seconds above 0: " + f"waiting up to {_duration(budget)}" + ) + start = clock() + with _cancelled_if_interrupted(api, run_id, say): + ended = _wait(api, job.job_id, run_id, build_id, budget, start, say, sleep, clock) + if ended is None: + return 1 + # The run has ended: an interrupt from here on stops the report, never a run. + return _report_end(api, ended.run, run_id, build_id, ended.elapsed, ended.url, say) + + +@dataclass(frozen=True) +class _Ended: + """A run that reached a terminal state, as the last poll saw it.""" + + run: Mapping[str, Any] + elapsed: float + url: str + + +def _run_now(api: Any, body: Mapping[str, Any], sleep: Callable[[float], None]) -> int: + """``POST run-now``; tried again with the same idempotency token when it fails. + + Databricks starts at most one run per token, so a retry after a timeout or a server + error (neither of which the SDK repeats for a POST) gets back the run the first try + may already have started, and never a second one. + """ + for attempt in range(1, _START_ATTEMPTS + 1): + try: + started = api.do("POST", f"{_JOBS}/run-now", body=dict(body)) + return int((started or {})["run_id"]) + except Exception: # noqa: BLE001 + if attempt == _START_ATTEMPTS: + raise + sleep(2 * attempt) + raise AssertionError("unreachable") # pragma: no cover + + +def _wait( + api: Any, + job_id: int, + run_id: int, + build_id: str, + budget: int, + start: float, + say: Say, + sleep: Callable[[float], None], + clock: Callable[[], float], +) -> Optional[_Ended]: + """Poll the run until it ends: what it ended as, or ``None`` once its failure is said.""" + last: Optional[Tuple[str, str]] = None + queued_why: Optional[str] = None + noted = start + url = "" + delay = 0 + failed_reads = 0 + while True: + try: + run = api.do("GET", f"{_JOBS}/runs/get", query={"run_id": run_id}) or {} + except Exception as exc: # noqa: BLE001 + # The SDK repeats only some failures (not a 500, 502 or 504), and one failed read + # says nothing about the run: try again, and never cancel a run that cannot be seen. + failed_reads += 1 + if failed_reads == 1: + say( + f" {_clock(clock() - start)} could not read run {run_id} " + f"({_failure(exc)}); trying again" + ) + if failed_reads >= _READ_ATTEMPTS: + console_error( + f"Build '{build_id}' - could not read run {run_id} {failed_reads} times in " + f"a row: {_failure(exc)}. fluid apply leaves it as it is; it may still be " + "going." + ) + if url: + say(f" run page: {url}") + return None + sleep(5 * failed_reads) + continue + failed_reads = 0 + if not url and run.get("run_page_url"): + url = clean(run["run_page_url"]) + say(f" run page: {url}") + now = clock() + phase, why = _phase(run) + if phase == "TERMINATED": + return _Ended(run, now - start, url) + if phase != "QUEUED": + queued_why = None + elif not why: + # Databricks gives no reason for most queued runs (seen live, 2026-11-12): name + # it here, from one read per queued spell. + if queued_why is None: + queued_why = _queued_behind(api, job_id, run_id) + why = queued_why + if (phase, why) != last: + say(f" {_clock(now - start)} {phase}" + (f" - {why}" if why else "")) + last, noted = (phase, why), now + elif now - noted >= HEARTBEAT_SECONDS: + say(f" {_clock(now - start)} still {phase}") + noted = now + if now - start >= budget: + cancelled = _cancel(api, run_id, say) + console_error( + f"Build '{build_id}' - run {run_id} did not end within {_duration(budget)}" + + (" and was cancelled" if cancelled else "") + + f". Set {TIMEOUT_ENV} (seconds) to wait longer." + ) + if url: + say(f" run page: {url}") + LOG.debug("databricks_job_run_timed_out run_id=%s budget=%s", run_id, budget) + return None + delay = min(delay + 1, 10) + sleep(delay) + + +@contextmanager +def _cancelled_if_interrupted(api: Any, run_id: int, say: Say) -> Iterator[None]: + """Ctrl-C, SIGTERM or SIGHUP (a CI job cancelled, a terminal closed) while the run is + waited on cancels it, then interrupts as Ctrl-C always does.""" + previous = _interrupt_on_terminate() + try: + yield + except KeyboardInterrupt: + _cancel(api, run_id, say) + raise + finally: + for signum, handler in previous.items(): + try: + signal.signal(signum, handler) + except (TypeError, ValueError, OSError): + # Best effort: a handler Python cannot set back (one installed from C reads + # as None) is left as it is; the wait is over, and failing here would hide + # the build's own result. + pass + + +def _interrupt_on_terminate() -> Dict[int, Any]: + """Make SIGTERM and SIGHUP raise ``KeyboardInterrupt``; the handlers they replaced. + + Only the main thread may set a handler, so elsewhere nothing changes, and a signal the + operator chose to ignore stays ignored (``nohup fluid apply`` must outlive its terminal). + """ + if threading.current_thread() is not threading.main_thread(): + return {} + previous: Dict[int, Any] = {} + for name in ("SIGTERM", "SIGHUP"): + signum = getattr(signal, name, None) + if signum is None: + continue + try: + if signal.getsignal(signum) is signal.SIG_IGN: + continue + previous[signum] = signal.signal(signum, _raise_interrupt) + except (ValueError, OSError): + continue + return previous + + +def _raise_interrupt(signum: int, frame: Any) -> None: + raise KeyboardInterrupt + + +def _report_end( + api: Any, + run: Mapping[str, Any], + run_id: int, + build_id: str, + elapsed: float, + url: str, + say: Say, +) -> int: + code, message = _outcome(run) + say(f" {_clock(elapsed)} TERMINATED {code}") + LOG.debug("databricks_job_run_ended run_id=%s result=%s", run_id, code) + if code == "SUCCESS": + for task_key, task in _latest_attempts(run): + summary = _dbt_summary(api, task["run_id"]) + if summary: + say(f" task '{task_key}': {summary}") + say(f"āœ… Build '{build_id}' - Databricks run {run_id} succeeded in {_duration(elapsed)}") + return 0 + console_error( + f"Build '{build_id}' - Databricks run {run_id} ended {code} after {_duration(elapsed)}" + + (f": {message}" if message else "") + ) + for task_key, task in _latest_attempts(run): + if _outcome(task)[0] != "SUCCESS": + _report_task(api, task_key, task["run_id"], say) + if url: + say(f" run page: {url}") + return 1 + + +def _queued_behind(api: Any, job_id: int, run_id: int) -> str: + """Why a run Databricks queued without saying why waits: behind another active run of + the same job (a job runs one at a time), else for the workspace's limit on runs or tasks + at once. ``""`` when the runs cannot be listed: the wait goes on either way.""" + try: + listed = api.do( + "GET", + f"{_JOBS}/runs/list", + query={"job_id": job_id, "active_only": "true", "limit": 25}, + ) + except Exception: # noqa: BLE001 ā€” the reason is a courtesy + return "" + runs = listed.get("runs") if isinstance(listed, Mapping) else None + # Only a run that is going is ahead: another queued run waits as this one does. + others = [ + other + for other, entry in ( + (_whole(r.get("run_id")), r) for r in runs or [] if isinstance(r, Mapping) + ) + if other and other != run_id and _phase(entry)[0] not in ("QUEUED", "TERMINATED") + ] + if others: # newest first: the last one started first + return f"behind run {others[-1]} of this job, which runs one at a time" + return ( + "waiting for the workspace's limit on runs or tasks at once " + "(Databricks Free Edition runs at most 5 job tasks at a time)" + ) + + +def _dbt_summary(api: Any, task_run: Any) -> str: + """dbt's own last ``Done. PASS=...`` line of a task's output, or ``""``: what a run that + succeeded built, at the cost of one read that may fail without failing the build.""" + try: + output = api.do("GET", f"{_JOBS}/runs/get-output", query={"run_id": task_run}) or {} + except Exception: # noqa: BLE001 ā€” the run succeeded; its summary is a courtesy + return "" + done = [line for line in clean(output.get("logs") or "").splitlines() if "Done. PASS=" in line] + return " ".join(_DBT_TIMESTAMP_RE.sub("", done[-1]).split()) if done else "" + + +def _report_task(api: Any, task_key: str, task_run: Any, say: Say) -> None: + try: + output = api.do("GET", f"{_JOBS}/runs/get-output", query={"run_id": task_run}) or {} + except Exception as exc: # noqa: BLE001 + say(f" task '{task_key}': its output could not be read ({_failure(exc)})") + return + # One line: Databricks quotes the failed shell command with its newlines escaped. + error = " ".join(clean(output.get("error") or "").replace("\\n", " ").split()) + say(f" task '{task_key}'" + (f": {error}" if error else " failed")) + logs = clean(output.get("logs") or output.get("error_trace") or "") + lines = [line.rstrip() for line in logs.splitlines() if line.strip()] + if not lines: + return + shown = lines[-TAIL_LINES:] + of = f"last {len(shown)} of {len(lines)} lines" if len(lines) > len(shown) else "output" + truncated = " (Databricks kept only the end)" if output.get("logs_truncated") else "" + say(f" dbt {of}{truncated}:") + for line in shown: + say(f" {line}") + + +def _phase(run: Mapping[str, Any]) -> Tuple[str, str]: + """``(state, why)``: the run's ``status.state`` (its deprecated ``state`` when absent), + ``TERMINATED`` for every state a run ends in, and why it waits while it is queued.""" + status = run.get("status") if isinstance(run.get("status"), Mapping) else {} + legacy = run.get("state") if isinstance(run.get("state"), Mapping) else {} + phase = str(status.get("state") or "").upper() + if not phase: + life = str(legacy.get("life_cycle_state") or "PENDING").upper() + phase = "TERMINATED" if life in ("TERMINATED", "SKIPPED", "INTERNAL_ERROR") else life + why = "" + if phase == "QUEUED": + details = ( + status.get("queue_details") if isinstance(status.get("queue_details"), Mapping) else {} + ) + message = clean(details.get("message") or legacy.get("queue_reason") or "").strip() + code = clean(details.get("code") or "").strip() + why = f"{message} ({code})" if message and code else message or code + return phase, " ".join(why.split()) + + +def _outcome(run: Mapping[str, Any]) -> Tuple[str, str]: + """``(code, message)`` of a run that ended: ``SUCCESS`` only when it succeeded.""" + status = run.get("status") if isinstance(run.get("status"), Mapping) else {} + legacy = run.get("state") if isinstance(run.get("state"), Mapping) else {} + details = ( + status.get("termination_details") + if isinstance(status.get("termination_details"), Mapping) + else {} + ) + code = str( + details.get("code") + or legacy.get("result_state") + or legacy.get("life_cycle_state") + or "UNKNOWN" + ).upper() + message = clean(details.get("message") or legacy.get("state_message") or "") + return clean(code).strip(), " ".join(message.split()) + + +def _latest_attempts(run: Mapping[str, Any]) -> List[Tuple[str, Mapping[str, Any]]]: + """``(task_key, task)`` for the latest attempt of each task of the run.""" + latest: Dict[str, Mapping[str, Any]] = {} + for task in run.get("tasks") or []: + if not isinstance(task, Mapping) or task.get("run_id") is None: + continue + key = str(task.get("task_key") or task.get("run_id")) + if key not in latest or _whole(task.get("attempt_number")) >= _whole( + latest[key].get("attempt_number") + ): + latest[key] = task + return [(clean(key), task) for key, task in latest.items()] + + +def _job_settings(api: Any, job_id: int) -> Mapping[str, Any]: + job = api.do("GET", f"{_JOBS}/get", query={"job_id": job_id}) or {} + settings = job.get("settings") if isinstance(job, Mapping) else None + return settings if isinstance(settings, Mapping) else {} + + +def _cancel(api: Any, run_id: int, say: Say) -> bool: + """Cancel the run; ``False`` (and a line saying so) when the request fails.""" + try: + api.do("POST", f"{_JOBS}/runs/cancel", body={"run_id": run_id}) + except Exception as exc: # noqa: BLE001 + say(f" could not cancel run {run_id}: {_failure(exc)}") + return False + say(f" cancelled run {run_id}") + LOG.debug("databricks_job_run_cancelled run_id=%s", run_id) + return True + + +def clean(text: Any) -> str: + """``text`` safe to print: escape sequences and control characters removed (tab and + newline kept), and every credential value in the environment masked.""" + from fluid_build.iac.providers.databricks_auth import credential_values + from fluid_build.observability.secret_redactor import redact_secret_text + + plain = _CONTROL_RE.sub("", _ANSI_RE.sub("", str(text or ""))) + return redact_secret_text( + plain, extra_secrets=credential_values(os.environ), keep_dotted_names=True + ) + + +def _failure(exc: BaseException) -> str: + from fluid_build.iac.providers.databricks_enforce import scrub + + return f"{type(exc).__name__}: {clean(scrub(exc))}" + + +def _default_client() -> Any: + from fluid_build.iac.providers.databricks_enforce import default_client + + return default_client() + + +def _whole(value: Any) -> int: + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 + + +def _clock(seconds: float) -> str: + """``m:ss`` (``h:mm:ss`` from an hour) since the run started.""" + whole = max(int(seconds), 0) + hours, rest = divmod(whole, 3600) + minutes, secs = divmod(rest, 60) + return f"{hours}:{minutes:02d}:{secs:02d}" if hours else f"{minutes}:{secs:02d}" + + +def _duration(seconds: float) -> str: + """``57s``, ``1m18s``, ``1h00m00s``.""" + whole = max(int(seconds), 0) + hours, rest = divmod(whole, 3600) + minutes, secs = divmod(rest, 60) + if hours: + return f"{hours}h{minutes:02d}m{secs:02d}s" + return f"{minutes}m{secs:02d}s" if minutes else f"{secs}s" + + +def _say(text: str) -> None: + # Unwrapped: a run page URL or a log line must survive a copy from a CI log intact. + cprint(text, markup=False, soft_wrap=True) + + +__all__ = [ + "DEFAULT_TIMEOUT_SECONDS", + "JobRef", + "TIMEOUT_ENV", + "clean", + "full_refresh_commands", + "jobs_in_state", + "run_job_build", + "start_and_wait", + "started_builds", + "wait_budget", +] diff --git a/fluid_build/build_runners/base.py b/fluid_build/build_runners/base.py index 6e34888d..29de220c 100644 --- a/fluid_build/build_runners/base.py +++ b/fluid_build/build_runners/base.py @@ -119,24 +119,65 @@ def _report_externally_run_build( _record_run_elsewhere(build_id, report) -def _report_databricks_job_build(contract: Dict[str, Any], build_id: str, report: Any) -> None: +def _report_databricks_job_build( + contract: Dict[str, Any], build: Dict[str, Any], build_id: str, report: Any +) -> None: """Say that a dbt build runs in its Databricks workspace, as a job; not here. ``fluid apply`` provisions the job through OpenTofu (``iac/providers/databricks_jobs.py``) and Databricks runs dbt on the job's - trigger, so a local run would be a second, unscheduled copy of it. Counted - and recorded as :func:`_report_externally_run_build` counts a Flink job. + trigger, so a local run would be a second, unscheduled copy of it. Said when + no apply in this process provisioned the job, and for a continuous job, which + Databricks keeps running (a ``-and-build`` apply starts any other one: + ``_databricks_job_run.py``). Counted and recorded as + :func:`_report_externally_run_build` counts a Flink job. """ - product = contract.get("id") or contract.get("name") or "product" + from fluid_build.iac.providers.databricks_jobs import job_build_id, job_product + + name = f"fluid {job_product(contract)} {job_build_id(contract, build)}" + if is_continuous_build(build): + how = "continuously; Databricks keeps it running, so fluid apply starts no run of it." + else: + how = ( + "on its trigger (fluid apply provisions it). An apply to databricks with --mode " + "amend-and-build starts a run now and waits for it; so do the job's page and: " + "databricks jobs run-now ." + ) cprint( - f"\nā­ļø Build '{build_id}' - dbt runs in the Databricks workspace as job " - f"'fluid {product} {build_id}' on its trigger (fluid apply provisions it); " - "start a run now from the job's page, or with: databricks jobs run-now .", + f"\nā­ļø Build '{build_id}' - dbt runs in the Databricks workspace as job '{name}' {how}", markup=False, ) _record_run_elsewhere(build_id, report) +def _databricks_job_build( + contract: Dict[str, Any], + build: Dict[str, Any], + report: Any, + databricks_jobs: Optional[Dict[str, Any]], + args: argparse.Namespace, +) -> Optional[int]: + """Run a dbt build that runs as a Databricks job: its exit code, or ``None`` when + Databricks runs it on its own (no apply here provisioned it, or it is continuous). + + A build that writes no Databricks table has no job, and fails. With the jobs the + provisioning apply handed over, a run of the build's job starts now and is waited + for (``_databricks_job_run.run_job_build``). + """ + build_id = build.get("id", "unknown") + if not _writes_a_databricks_table(contract, build): + # No Databricks table: no plugin provisions its job, so nothing would run it. + return _refuse_unprovisioned_databricks_build(build_id) + if databricks_jobs is None or is_continuous_build(build): + _report_databricks_job_build(contract, build, build_id, report) + return None + from ._databricks_job_run import run_job_build + + return run_job_build( + contract, build, databricks_jobs, dry_run=args.dry_run, mode=getattr(args, "mode", None) + ) + + def _writes_a_databricks_table(contract: Dict[str, Any], build: Dict[str, Any]) -> bool: """True when ``build`` writes a Databricks-bound expose, so the Databricks plugin provisions its job (``iac/providers/databricks_jobs.py``).""" @@ -1214,6 +1255,7 @@ def run_builds_from_args( force_run: bool = False, plan_data: Optional[Dict[str, Any]] = None, contract: Optional[Dict[str, Any]] = None, + databricks_jobs: Optional[Dict[str, Any]] = None, ) -> int: """Execute builds from a FLUID contract. @@ -1244,6 +1286,10 @@ def run_builds_from_args( Each build is recorded on the running ``fluid apply``'s Command Center report, when there is one (``observability.apply_run``): its status and the run record it wrote, and why the build phase failed. + + ``databricks_jobs`` (build id to the jobs the OpenTofu state holds) comes from + the ``fluid apply`` that just provisioned them: a dbt build that runs on + Databricks then runs its job now (:func:`_databricks_job_build`). """ # Deferred imports to avoid circular import at module-load time: # base.py -> python.runner -> base.py (for _resolve_env_placeholders). @@ -1458,16 +1504,18 @@ def run_builds_from_args( continue if is_databricks_job_build(build): - if _writes_a_databricks_table(contract, build): - _report_databricks_job_build(contract, build_id, report) + job_result = _databricks_job_build(contract, build, report, databricks_jobs, args) + if job_result is None: total_external += 1 continue - # No Databricks table: no plugin provisions its job, so nothing would run it. - result = _refuse_unprovisioned_databricks_build(build_id) + result = job_result _report_build(report, contract_path.parent, product_id, build_id, result, runs_before) - total_failed += 1 - if args.fail_fast: - break + if result == 0: + total_executed += 1 + else: + total_failed += 1 + if args.fail_fast: + break continue if _runs_outside_forge_continuously(build): diff --git a/fluid_build/cli/_apply_opentofu_engine.py b/fluid_build/cli/_apply_opentofu_engine.py index 122ace51..e35d23e6 100644 --- a/fluid_build/cli/_apply_opentofu_engine.py +++ b/fluid_build/cli/_apply_opentofu_engine.py @@ -55,7 +55,11 @@ def apply_via_opentofu( - args, logger: logging.Logger, contract: Optional[Mapping[str, Any]] = None + args, + logger: logging.Logger, + contract: Optional[Mapping[str, Any]] = None, + *, + applied_state: Optional[List[Dict[str, Any]]] = None, ) -> int: """Run ``fluid apply`` through the OpenTofu engine. Returns an exit code. @@ -65,14 +69,24 @@ def apply_via_opentofu( the gate cannot reach ``tofu`` unvalidated. Without it the contract is loaded from ``args.contract``. + ``applied_state``, when given, receives the resources of the state a + successful apply leaves (``runner.tofu_state_resources``, read once and not + refreshed): the build phase starts the jobs it provisioned by the ids there. + A dry run applies nothing and leaves it empty. + Every compile of the run sees the contract's file (``iac/source.py``), so a plugin that ships a build's local files finds them as the runners do. """ with contract_source(source_contract_file(getattr(args, "contract", None))): - return _apply_via_opentofu(args, logger, contract) + return _apply_via_opentofu(args, logger, contract, applied_state) -def _apply_via_opentofu(args, logger: logging.Logger, contract: Optional[Mapping[str, Any]]) -> int: +def _apply_via_opentofu( + args, + logger: logging.Logger, + contract: Optional[Mapping[str, Any]], + applied_state: Optional[List[Dict[str, Any]]] = None, +) -> int: # Plan-binding verification ā€” must run BEFORE any tofu apply so a # tampered plan.json is rejected before infra changes. The native # apply path (cli/apply.py::_verify_plan_digests) had this gate; the @@ -336,6 +350,15 @@ def _apply_via_opentofu(args, logger: logging.Logger, contract: Optional[Mapping # tables that exist and are protected. _converge_output_ports(plugin, contract) + if applied_state is not None: + # After the bindings: a job the build phase starts writes tables already protected. + applied_state.extend(runner.tofu_state_resources(str(workdir), env=env)) + if not applied_state: # an apply that provisioned a job leaves a state with it + cprint( + f" āš ļø could not read the applied state back (tofu show -json in {workdir}): " + "the build phase cannot find the jobs this apply provisioned." + ) + info(logger, "opentofu_apply_ok", provider=provider, **applied) return 0 diff --git a/fluid_build/cli/apply.py b/fluid_build/cli/apply.py index 9f5f49f8..0797d677 100644 --- a/fluid_build/cli/apply.py +++ b/fluid_build/cli/apply.py @@ -339,6 +339,70 @@ def _gate_build_ids_for_apply(contract: Dict[str, Any]) -> None: ) from exc +def _databricks_job_runs(contract: Dict[str, Any], build_id: Optional[str]) -> bool: + """True when a build phase of ``contract`` starts Databricks job runs + (``build_runners/_databricks_job_run.py``): it needs the apply's state, and the SDK.""" + from fluid_build.build_runners._databricks_job_run import started_builds + + return bool(started_builds(contract, build_id)) + + +def _state_for_job_runs( + contract: Dict[str, Any], mode: Any, build_id: Optional[str] +) -> Dict[str, Any]: + """``apply_via_opentofu`` keyword arguments: ``applied_state`` when this apply's build + phase starts Databricks job runs, so the engine hands back the state it leaves.""" + from fluid_build.forge.core.apply_modes import needs_build + + if needs_build(mode) and _databricks_job_runs(contract, build_id): + return {"applied_state": []} + return {} + + +def _jobs_from_state(engine_kwargs: Dict[str, Any], contract: Dict[str, Any]) -> Dict[str, Any]: + """``run_builds_from_args`` keyword arguments: the Databricks jobs the applied state + holds for ``contract``, by build id (``_databricks_job_run.jobs_in_state``).""" + if "applied_state" not in engine_kwargs: + return {} + from fluid_build.build_runners._databricks_job_run import jobs_in_state + from fluid_build.iac.providers.databricks_jobs import job_product + + jobs = jobs_in_state(engine_kwargs["applied_state"], job_product(contract)) + return {"databricks_jobs": jobs} + + +def _gate_databricks_sdk_for_apply( + contract: Dict[str, Any], build_id: Optional[str], dry_run: bool, engine: str +) -> None: + """Refuse, before anything is applied, a build phase that starts Databricks job runs + without databricks-sdk, which starts them. Otherwise ``tofu apply`` would provision the + jobs and the build would then fail. Only the OpenTofu engine starts them (the native one + provisions no job), and a dry run starts none. + + Raises: + CLIError: ``apply_databricks_sdk_missing``. + """ + if dry_run or engine != "opentofu" or not _databricks_job_runs(contract, build_id): + return + import importlib.util + + try: + found = importlib.util.find_spec("databricks.sdk") is not None + except (ImportError, ValueError): + found = False + if not found: + raise CLIError( + 1, + "apply_databricks_sdk_missing", + { + "error": "a dbt build runs on Databricks, and the build phase starts its job " + "through databricks-sdk, which is not installed; nothing was applied", + "hint": "pip install 'data-product-forge[databricks]', or apply with " + "--mode amend to provision the job without starting it", + }, + ) + + def _gate_selected_build_for_apply(contract: Dict[str, Any], build_id: Optional[str]) -> None: """Refuse, before anything is applied, a ``--build-id`` that names no build. @@ -2059,6 +2123,12 @@ def run(args, logger: logging.Logger) -> int: if needs_build(resolved_mode): _gate_build_ids_for_apply(source_contract) _gate_selected_build_for_apply(source_contract, resolved_build_id) + _gate_databricks_sdk_for_apply( + source_contract, + resolved_build_id, + bool(getattr(args, "dry_run", False)), + apply_engine, + ) except CLIError as exc: _log_apply_failure(logger, start_time, exc) raise @@ -2079,7 +2149,8 @@ def run(args, logger: logging.Logger) -> int: _hook_rc = _run_apply_hooks_for_source(args, source_contract, logger) if _hook_rc != 0: return _hook_rc - rc = apply_via_opentofu(args, logger, contract=source_contract) + engine_kwargs = _state_for_job_runs(source_contract, resolved_mode, resolved_build_id) + rc = apply_via_opentofu(args, logger, contract=source_contract, **engine_kwargs) # The OpenTofu engine provisions infrastructure only. Build-augmented # modes (amend-and-build / replace-and-build) still need their build # phase to run ā€” mirror the native dispatch below so ``--build`` is @@ -2088,7 +2159,8 @@ def run(args, logger: logging.Logger) -> int: args.build_id = resolved_build_id from fluid_build.cli._apply_opentofu_engine import build_after_apply - return build_after_apply(args, logger, source_contract, build_kwargs) + jobs = _jobs_from_state(engine_kwargs, source_contract) + return build_after_apply(args, logger, source_contract, {**build_kwargs, **jobs}) return rc try: diff --git a/fluid_build/cli/doctor.py b/fluid_build/cli/doctor.py index 7d7b8972..1a5222f0 100644 --- a/fluid_build/cli/doctor.py +++ b/fluid_build/cli/doctor.py @@ -163,6 +163,12 @@ "1800s default", "Per-`tofu` invocation wall-clock cap", ), + ( + "FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS", + "3600s, or the job's own limits + 15 min", + "How long `fluid apply --mode amend-and-build` waits for a Databricks job run " + "before it cancels the run", + ), ( "FLUID_TOFU_S3_LOCKFILE", "on (OpenTofu 1.10+)", diff --git a/fluid_build/credentials/protected_env.py b/fluid_build/credentials/protected_env.py index 9e8ba252..539f56e5 100644 --- a/fluid_build/credentials/protected_env.py +++ b/fluid_build/credentials/protected_env.py @@ -638,6 +638,7 @@ "FLUID_COST_LIMIT_USD", "FLUID_COST_LIMIT_USD_PER_PRODUCT", "FLUID_COST_LIMIT_USD_PER_RUN", + "FLUID_DATABRICKS_RUN_TIMEOUT_SECONDS", "FLUID_FEDERATION_TIMEOUT_SECONDS", "FLUID_FORGE_AGENT_TIMEOUT_SECONDS", "FLUID_HOSTED_MCP_TIMEOUT_SECONDS", diff --git a/fluid_build/iac/providers/databricks_jobs.py b/fluid_build/iac/providers/databricks_jobs.py index 91ec4a79..243c56da 100644 --- a/fluid_build/iac/providers/databricks_jobs.py +++ b/fluid_build/iac/providers/databricks_jobs.py @@ -83,6 +83,11 @@ MAX_PROJECT_BYTES = 20 * 1024 * 1024 MAX_PROJECT_FILES = 1000 +#: The tags every job carries. ``fluid apply --mode amend-and-build`` finds a build's job in +#: the apply's state by them (``build_runners/_databricks_job_run.py::jobs_in_state``). +PRODUCT_TAG = "fluid_product" +BUILD_TAG = "fluid_build" + #: The data source that names the deploying identity's home folder. CURRENT_USER = "databricks_current_user" _HOME = tofu_ref(f"data.{CURRENT_USER}.me.home") @@ -742,6 +747,26 @@ class _Source: files: Tuple[Tuple[str, bytes], ...] = () +def job_product(contract: Mapping[str, Any]) -> str: + """The product a job is named (``fluid ``) and tagged (``PRODUCT_TAG``) after.""" + return str(contract.get("id") or contract.get("name") or "product") + + +def job_build_id(contract: Mapping[str, Any], build: Mapping[str, Any]) -> str: + """The build a job is named and tagged (``BUILD_TAG``) after: its ``id``, or, since the + schema does not require one, its position in ``builds[]``.""" + if build.get("id"): + return str(build["id"]) + builds = list(contract.get("builds") or []) + for index, candidate in enumerate(builds): + if candidate is build: + return str(index) + for index, candidate in enumerate(builds): + if candidate == build: + return str(index) + return "unknown" + + def compile_dbt_jobs( contract: Mapping[str, Any], tables: Sequence[Any], @@ -764,11 +789,11 @@ def compile_dbt_jobs( """ compiled = CompiledJobs() builds = [b for b in contract.get("builds") or [] if isinstance(b, Mapping)] - product = str(contract.get("id") or contract.get("name") or "product") + product = job_product(contract) for index, build in enumerate(contract.get("builds") or []): if not isinstance(build, Mapping) or not is_databricks_job_build(build): continue - build_id = str(build.get("id") or index) + build_id = job_build_id(contract, build) item = _Build( build, build_id, f"builds[{index}] ({build_id})", product, warehouse, run_as_allowed ) @@ -904,7 +929,7 @@ def _emit_job( job: Dict[str, Any] = { "name": item.name, "max_concurrent_runs": 1, - "tags": {"fluid_product": item.product, "fluid_build": item.build_id}, + "tags": {PRODUCT_TAG: item.product, BUILD_TAG: item.build_id}, "environment": [ { "environment_key": "dbt", diff --git a/tests/build_runners/test_databricks_job_builds.py b/tests/build_runners/test_databricks_job_builds.py index 3219862e..9877ef92 100644 --- a/tests/build_runners/test_databricks_job_builds.py +++ b/tests/build_runners/test_databricks_job_builds.py @@ -1,13 +1,16 @@ # Copyright 2025 Agentics Transformation Limited # SPDX-License-Identifier: Apache-2.0 -"""A dbt build that runs in a Databricks workspace as a job is not run here. - -``fluid apply`` provisions the job (``iac/providers/databricks_jobs.py``) and -Databricks runs dbt on its trigger, so the build runner reports the build and -neither runs it nor warns about a continuous trigger. A local dbt build is -unchanged. Driven the way ``tests/test_flink_engine.py`` drives an externally -run build; no dbt, no network. +"""A dbt build that runs in a Databricks workspace as a job is never run locally. + +``fluid apply`` provisions the job (``iac/providers/databricks_jobs.py``). When that +apply hands the build runner the jobs it provisioned (``databricks_jobs``), a +``-and-build`` mode starts a run of the job now and waits for it +(``_databricks_job_run.py``, pinned in ``test_databricks_job_run.py``); a continuous +job is left to Databricks. Without them the build runner reports the build as one +Databricks runs on its trigger, and never warns about a continuous trigger. A +local dbt build is unchanged. Driven the way ``tests/test_flink_engine.py`` drives +an externally run build; no dbt, no network. """ from __future__ import annotations @@ -20,6 +23,7 @@ import pytest import yaml +from fluid_build.build_runners._databricks_job_run import JobRef from fluid_build.build_runners.base import ( _batch_runner_for, continuous_trigger_warnings, @@ -104,6 +108,24 @@ def _must_not_run(*_a: Any, **_k: Any) -> int: monkeypatch.setattr("fluid_build.build_runners.python.runner.execute_build", _must_not_run) +@pytest.fixture +def recorded_builds(monkeypatch: pytest.MonkeyPatch) -> list: + """What the build runner records on the running apply's Command Center report.""" + import fluid_build.observability.apply_run as apply_run + + recorded: list = [] + + class _Report: + def record_build(self, **kwargs: Any) -> None: + recorded.append(kwargs) + + def build_failed(self, event: str) -> None: + raise AssertionError(event) + + monkeypatch.setattr(apply_run, "current_apply_run", lambda: _Report()) + return recorded + + def _flat(capsys: pytest.CaptureFixture) -> str: """Captured output with the console's line wrapping undone.""" return " ".join(capsys.readouterr().out.split()) @@ -200,24 +222,12 @@ def test_a_continuous_databricks_job_build_is_reported_the_same_way( assert JOB_LINE in out and "Run outside forge-cli: 1" in out def test_the_run_report_records_it_skipped_for_running_outside_forge( - self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, no_local_run: None + self, tmp_path: Path, recorded_builds: list, no_local_run: None ) -> None: - import fluid_build.observability.apply_run as apply_run - - recorded: list = [] - - class _Report: - def record_build(self, **kwargs: Any) -> None: - recorded.append(kwargs) - - def build_failed(self, event: str) -> None: - raise AssertionError(event) - - monkeypatch.setattr(apply_run, "current_apply_run", lambda: _Report()) path = tmp_path / "contract.fluid.yaml" path.write_text(yaml.safe_dump(_contract(_build())), encoding="utf-8") assert run_builds_from_args(_args(path, dry_run=True), logging.getLogger("t")) == 0 - assert recorded == [ + assert recorded_builds == [ {"build_id": "transform", "status": "skipped", "reason": "runs_outside_forge"} ] @@ -286,6 +296,96 @@ def _fake_dbt(build: Dict[str, Any], *_a: Any, **_k: Any) -> int: assert "Executed: 1" in out +class TestAfterTheApplyThatProvisionedIt: + """``fluid apply`` hands over the jobs its state holds: the build runs now.""" + + JOBS = {"transform": [JobRef(5)], "second": [JobRef(6)]} + + @pytest.fixture + def started(self, monkeypatch: pytest.MonkeyPatch) -> list: + calls: list = [] + + def _run(contract: Any, build: Any, jobs: Any, **kwargs: Any) -> int: + calls.append((build["id"], jobs, kwargs)) + return 1 if build["id"] == "broken" else 0 + + monkeypatch.setattr("fluid_build.build_runners._databricks_job_run.run_job_build", _run) + return calls + + def _apply(self, tmp_path: Path, contract: Dict[str, Any], **kwargs: Any) -> int: + path = tmp_path / "contract.fluid.yaml" + path.write_text(yaml.safe_dump(contract), encoding="utf-8") + args = _args(path, dry_run=kwargs.pop("dry_run", False)) + args.fail_fast = kwargs.pop("fail_fast", False) + args.mode = kwargs.pop("mode", "amend-and-build") + return run_builds_from_args(args, logging.getLogger("t"), databricks_jobs=self.JOBS) + + def test_the_build_runs_now_and_counts_as_executed( + self, tmp_path: Path, capsys: pytest.CaptureFixture, no_local_run: None, started: list + ) -> None: + assert self._apply(tmp_path, _contract(_build(trigger="schedule"))) == 0 + assert [(build_id, kwargs) for build_id, _jobs, kwargs in started] == [ + ("transform", {"dry_run": False, "mode": "amend-and-build"}) + ] + assert started[0][1] is self.JOBS + out = _flat(capsys) + assert "Executed: 1" in out and "Run outside forge-cli" not in out + assert JOB_LINE not in out + + def test_the_mode_and_a_dry_run_reach_it( + self, tmp_path: Path, no_local_run: None, started: list + ) -> None: + assert self._apply(tmp_path, _contract(_build()), mode="replace-and-build") == 0 + assert self._apply(tmp_path, _contract(_build()), dry_run=True) == 0 + assert [kwargs for _b, _j, kwargs in started] == [ + {"dry_run": False, "mode": "replace-and-build"}, + {"dry_run": True, "mode": "amend-and-build"}, + ] + + def test_a_failed_run_fails_the_build_phase_and_fail_fast_stops_there( + self, tmp_path: Path, capsys: pytest.CaptureFixture, no_local_run: None, started: list + ) -> None: + contract = _contract(_build(build_id="broken"), _build(build_id="second")) + assert self._apply(tmp_path, contract) == 1 + assert [build_id for build_id, _j, _k in started] == ["broken", "second"] + assert "Failed: 1" in capsys.readouterr().err # the console prints errors to stderr + + started.clear() + assert self._apply(tmp_path, contract, fail_fast=True) == 1 + assert [build_id for build_id, _j, _k in started] == ["broken"] + + def test_a_continuous_job_is_left_to_databricks( + self, tmp_path: Path, capsys: pytest.CaptureFixture, no_local_run: None, started: list + ) -> None: + assert self._apply(tmp_path, _contract(_build(trigger="continuous"))) == 0 + assert started == [] + out = _flat(capsys) + assert f"{JOB_LINE} continuously; Databricks keeps it running" in out + assert "Run outside forge-cli: 1" in out + + def test_the_run_report_records_how_the_run_ended( + self, tmp_path: Path, recorded_builds: list, no_local_run: None, started: list + ) -> None: + contract = _contract(_build(), _build(build_id="broken")) + assert self._apply(tmp_path, contract) == 1 + assert [(r["build_id"], r["status"]) for r in recorded_builds] == [ + ("transform", "succeeded"), + ("broken", "failed"), + ] + + def test_a_build_that_writes_no_databricks_table_is_still_refused( + self, tmp_path: Path, capsys: pytest.CaptureFixture, no_local_run: None, started: list + ) -> None: + contract = _contract(_build()) + contract["exposes"][0]["binding"] = { + "platform": "aws", + "format": "parquet", + "location": {"bucket": "b", "path": "orders/"}, + } + assert self._apply(tmp_path, contract) == 1 + assert started == [] and "databricks-job-no-target" in _flat(capsys) + + class TestNoSchedulerStartsAJobBuild: """Databricks starts a job build on its own trigger: forge-cli's schedulers give it no task, as they give a Flink job none.""" diff --git a/tests/build_runners/test_databricks_job_run.py b/tests/build_runners/test_databricks_job_run.py new file mode 100644 index 00000000..3acde3bf --- /dev/null +++ b/tests/build_runners/test_databricks_job_run.py @@ -0,0 +1,787 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""``fluid apply --mode amend-and-build`` starts a Databricks dbt job now and waits for it. + +Driven through ``build_runners/_databricks_job_run.py`` with a scripted ``api_client`` +(the shape ``databricks_enforce.default_client`` returns) and a fake clock: no network, +no SDK. Pinned: the run starts with an idempotency token and queueing on, and nothing +but ``run-now`` is ever written to the job; its page URL and every state it enters are +printed, a queue reason once; a run that does not succeed fails the build with the task's +error and the end of its output, credentials masked and escape sequences gone; the wait +is capped, and the cap or Ctrl-C cancels the run; ``replace-and-build`` adds +``--full-refresh`` to the job's own ``dbt build``. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Mapping, Optional + +import pytest + +from fluid_build.build_runners import _databricks_job_run as job_run +from fluid_build.build_runners._databricks_job_run import ( + DEFAULT_TIMEOUT_SECONDS, + TIMEOUT_ENV, + JobRef, + full_refresh_commands, + jobs_in_state, + run_job_build, + started_builds, + wait_budget, +) + +pytestmark = pytest.mark.unit + +URL = "https://dbc-1.cloud.databricks.com/?o=1#job/5/run/77" +TOKEN = "test-only-token-0f3c9a7e5b1d2468ace" # pragma: allowlist secret +SETTINGS: Dict[str, Any] = { + "name": "fluid sales.orders transform", + "tags": {"fluid_product": "sales.orders", "fluid_build": "transform"}, + "tasks": [ + { + "task_key": "dbt", + "dbt_task": {"commands": ["dbt deps", "dbt build --select orders"]}, + } + ], +} + + +def _run( + state: str, + *, + code: Optional[str] = None, + message: str = "", + queue: Optional[Dict[str, str]] = None, + tasks: Optional[List[Dict[str, Any]]] = None, +) -> Dict[str, Any]: + status: Dict[str, Any] = {"state": state} + if queue: + status["queue_details"] = queue + if code: + status["termination_details"] = {"code": code, "message": message} + run: Dict[str, Any] = {"run_id": 77, "run_page_url": URL, "status": status} + if tasks is not None: + run["tasks"] = tasks + return run + + +def _task(code: str, run_id: int = 78, attempt: int = 0) -> Dict[str, Any]: + return { + "task_key": "dbt", + "run_id": run_id, + "attempt_number": attempt, + "status": {"state": "TERMINATED", "termination_details": {"code": code}}, + } + + +class _Api: + """``WorkspaceClient.api_client``: answers each jobs endpoint; ``runs`` in turn.""" + + def __init__( + self, + runs: List[Dict[str, Any]], + *, + settings: Optional[Mapping[str, Any]] = None, + outputs: Optional[Dict[int, Dict[str, Any]]] = None, + fail: Optional[Dict[str, BaseException]] = None, + active: Optional[List[Dict[str, Any]]] = None, + ) -> None: + self.runs = list(runs) + self.active = active or [] + self.settings = SETTINGS if settings is None else settings + self.outputs = outputs or {} + self.fail = fail or {} + self.calls: List[tuple] = [] + + def do(self, method: str, path: str, query: Any = None, body: Any = None, **_: Any) -> Any: + endpoint = path.split("/api/2.2/jobs/", 1)[1] + self.calls.append((method, endpoint, query, body)) + if endpoint in self.fail: + raise self.fail[endpoint] + if endpoint == "get": + return {"job_id": query["job_id"], "settings": self.settings} + if endpoint == "run-now": + return {"run_id": 77, "number_in_job": 77} + if endpoint == "runs/get": + return self.runs.pop(0) if len(self.runs) > 1 else self.runs[0] + if endpoint == "runs/get-output": + return self.outputs.get(query["run_id"], {}) + if endpoint == "runs/cancel": + return {} + if endpoint == "runs/list": + return {"runs": self.active} + raise AssertionError(f"unexpected endpoint {method} {path}") + + def writes(self) -> List[str]: + return [endpoint for method, endpoint, _q, _b in self.calls if method != "GET"] + + def body(self, endpoint: str) -> Dict[str, Any]: + [body] = [b for _m, e, _q, b in self.calls if e == endpoint] + return body + + +class _Client: + def __init__(self, api: _Api) -> None: + self.api_client = api + + +class _Time: + """A clock that moves only when the code under test sleeps.""" + + def __init__(self, interrupt_after: Optional[int] = None) -> None: + self.now = 1000.0 + self.slept: List[float] = [] + self.interrupt_after = interrupt_after + + def clock(self) -> float: + return self.now + + def sleep(self, seconds: float) -> None: + if self.interrupt_after is not None and len(self.slept) >= self.interrupt_after: + raise KeyboardInterrupt + self.slept.append(seconds) + self.now += seconds + + +@pytest.fixture(autouse=True) +def _environment(monkeypatch: pytest.MonkeyPatch) -> None: + for name in (TIMEOUT_ENV, "DATABRICKS_TOKEN", "DATABRICKS_CLIENT_SECRET"): + monkeypatch.delenv(name, raising=False) + + +CONTRACT = {"id": "sales.orders", "builds": []} +BUILD = {"id": "transform", "engine": "dbt"} +JOBS = {"transform": [JobRef(5, "https://dbc-1.cloud.databricks.com/#job/5")]} + + +def _go( + api: _Api, + capsys: pytest.CaptureFixture, + *, + mode: str = "amend-and-build", + time_: Optional[_Time] = None, + env: Optional[Dict[str, str]] = None, +) -> tuple: + time_ = time_ or _Time() + rc = run_job_build( + CONTRACT, + BUILD, + JOBS, + mode=mode, + client_factory=lambda: _Client(api), + env=env or {}, + sleep=time_.sleep, + clock=time_.clock, + ) + captured = capsys.readouterr() + return rc, " ".join((captured.out + captured.err).split()) + + +# ā”€ā”€ the run ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ +class TestStartAndWait: + def test_a_run_starts_now_and_its_success_is_reported(self, capsys) -> None: + api = _Api([_run("PENDING"), _run("RUNNING"), _run("TERMINATED", code="SUCCESS")]) + rc, out = _go(api, capsys) + + assert rc == 0, out + body = api.body("run-now") + assert body["job_id"] == 5 and body["queue"] == {"enabled": True} + assert len(body["idempotency_token"]) == 32 and "dbt_commands" not in body + assert "started run 77 of Databricks job 'fluid sales.orders transform' (5)" in out + assert f"run page: {URL}" in out + assert "0:00 PENDING" in out and "RUNNING" in out and "TERMINATED SUCCESS" in out + assert "Databricks run 77 succeeded in" in out + + def test_nothing_but_the_run_is_ever_written(self, capsys) -> None: + # The job's settings, schedule and pause state are not touched: a scheduled + # job keeps its schedule, and a job created paused stays paused. + api = _Api([_run("TERMINATED", code="SUCCESS")]) + assert _go(api, capsys)[0] == 0 + assert api.writes() == ["run-now"] + + def test_a_queued_run_says_why_once_then_a_line_a_minute(self, capsys) -> None: + reason = { + "code": "ACTIVE_RUN_JOB_TASKS_LIMIT_REACHED", + "message": "The run is queued: the workspace runs its limit of 5 concurrent tasks.", + } + queued = [_run("QUEUED", queue=reason) for _ in range(14)] + api = _Api([*queued, _run("RUNNING"), _run("TERMINATED", code="SUCCESS")]) + rc, out = _go(api, capsys) + + assert rc == 0, out + assert out.count("limit of 5 concurrent tasks") == 1 + assert "QUEUED - The run is queued" in out and "(ACTIVE_RUN_JOB_TASKS_LIMIT_REACHED)" in out + assert "still QUEUED" in out + assert [e for _m, e, _q, _b in api.calls if e == "runs/list"] == [] + + @pytest.mark.parametrize( + "active, said", + [ + pytest.param( + [{"run_id": 77}, {"run_id": 70}, {"run_id": 12}], + "QUEUED - behind run 12 of this job, which runs one at a time", + id="behind-another-run", + ), + pytest.param( + [{"run_id": 77}], + "QUEUED - waiting for the workspace's limit on runs or tasks at once " + "(Databricks Free Edition runs at most 5 job tasks at a time)", + id="workspace-limit", + ), + ], + ) + def test_a_queued_run_without_a_reason_is_given_one(self, capsys, active, said) -> None: + # What the live API answers for a queued run: a state, and no reason. + queued = [_run("QUEUED") for _ in range(5)] + api = _Api([*queued, _run("RUNNING"), _run("TERMINATED", code="SUCCESS")], active=active) + rc, out = _go(api, capsys) + assert rc == 0, out + assert out.count(said) == 1 + listed = [q for _m, e, q, _b in api.calls if e == "runs/list"] + assert listed == [{"job_id": 5, "active_only": "true", "limit": 25}] + + def test_runs_that_cannot_be_listed_leave_the_queued_line_plain(self, capsys) -> None: + api = _Api( + [_run("QUEUED"), _run("TERMINATED", code="SUCCESS")], + fail={"runs/list": ConnectionError("reset")}, + ) + rc, out = _go(api, capsys) + assert rc == 0 and "0:00 QUEUED 0:01" in out + + def test_the_poll_interval_grows_to_ten_seconds(self, capsys) -> None: + time_ = _Time() + api = _Api([*[_run("RUNNING") for _ in range(14)], _run("TERMINATED", code="SUCCESS")]) + assert _go(api, capsys, time_=time_)[0] == 0 + assert time_.slept[:3] == [1, 2, 3] and max(time_.slept) == 10 + + def test_a_failed_run_fails_with_the_task_error_and_the_end_of_its_output(self, capsys) -> None: + logs = "\n".join([f"noise {i}" for i in range(40)] + ["Database Error in model orders"]) + api = _Api( + [ + _run( + "TERMINATED", + code="RUN_EXECUTION_ERROR", + message="Task dbt failed with message: dbt command failed.", + tasks=[_task("RUN_EXECUTION_ERROR")], + ) + ], + outputs={78: {"error": "dbt command failed", "logs": logs}}, + ) + rc, out = _go(api, capsys) + + assert rc == 1 + assert "Databricks run 77 ended RUN_EXECUTION_ERROR after" in out + assert "Task dbt failed with message: dbt command failed." in out + assert "task 'dbt': dbt command failed" in out + assert "dbt last 30 of 41 lines:" in out + assert "Database Error in model orders" in out + assert "noise 10 " not in out and "noise 11 " in out + assert out.count(f"run page: {URL}") == 2 # first, and again after the failure + + def test_only_the_latest_attempt_of_a_failed_task_is_read(self, capsys) -> None: + tasks = [_task("RUN_EXECUTION_ERROR", 78, 0), _task("RUN_EXECUTION_ERROR", 79, 1)] + api = _Api( + [_run("TERMINATED", code="RUN_EXECUTION_ERROR", tasks=tasks)], + outputs={79: {"error": "second attempt failed"}}, + ) + rc, out = _go(api, capsys) + assert rc == 1 and "second attempt failed" in out + read = [q["run_id"] for _m, e, q, _b in api.calls if e == "runs/get-output"] + assert read == [79] + + def test_credentials_and_escape_sequences_never_reach_the_terminal( + self, capsys, monkeypatch + ) -> None: + monkeypatch.setenv("DATABRICKS_TOKEN", TOKEN) + api = _Api( + [_run("TERMINATED", code="FAILED", tasks=[_task("FAILED")])], + outputs={ + 78: { + "error": f"bad token {TOKEN}", + "logs": f"\x1b[31mred line\x1b[0m\x07\nuses {TOKEN} here\r\n\x9b2Jdone", + } + }, + ) + rc, out = _go(api, capsys) + assert rc == 1 + assert TOKEN not in out + assert "red line" in out and "done" in out + assert not any(ch in out for ch in ("\x1b", "\x07", "\x9b", "\r")) + + def test_a_run_that_outlasts_the_wait_is_cancelled(self, capsys) -> None: + api = _Api([_run("RUNNING")]) + rc, out = _go(api, capsys, env={TIMEOUT_ENV: "30"}) + assert rc == 1 + assert api.writes() == ["run-now", "runs/cancel"] + assert api.body("runs/cancel") == {"run_id": 77} + assert "did not end within 30s and was cancelled" in out and URL in out + assert TIMEOUT_ENV in out + + def test_ctrl_c_cancels_the_run_and_still_interrupts(self, capsys) -> None: + api = _Api([_run("RUNNING")]) + with pytest.raises(KeyboardInterrupt): + _go(api, capsys, time_=_Time(interrupt_after=2)) + assert api.writes() == ["run-now", "runs/cancel"] + assert "cancelled run 77" in capsys.readouterr().out + + def test_replace_and_build_adds_full_refresh_to_the_jobs_own_dbt_build(self, capsys) -> None: + api = _Api([_run("TERMINATED", code="SUCCESS")]) + rc, out = _go(api, capsys, mode="replace-and-build") + assert rc == 0, out + assert api.body("run-now")["dbt_commands"] == [ + "dbt deps", + "dbt build --select orders --full-refresh", + ] + + def test_replace_and_build_without_dbt_commands_starts_nothing(self, capsys) -> None: + api = _Api([_run("TERMINATED", code="SUCCESS")], settings={"tasks": []}) + rc, out = _go(api, capsys, mode="replace-and-build") + assert rc == 1 and api.writes() == [] + assert "has no dbt commands" in out + + @pytest.mark.parametrize( + "legacy, code", + [ + ({"life_cycle_state": "SKIPPED", "state_message": "another run is active"}, "SKIPPED"), + ({"life_cycle_state": "INTERNAL_ERROR", "result_state": "FAILED"}, "FAILED"), + ({"life_cycle_state": "TERMINATED", "result_state": "TIMEDOUT"}, "TIMEDOUT"), + ], + ) + def test_a_run_that_ends_any_other_way_fails(self, capsys, legacy, code) -> None: + # The deprecated ``state`` is read when there is no ``status``. + api = _Api([{"run_id": 77, "state": legacy}]) + rc, out = _go(api, capsys) + assert rc == 1 and f"ended {code}" in out + + def test_the_deprecated_state_alone_can_succeed(self, capsys) -> None: + legacy = {"life_cycle_state": "TERMINATED", "result_state": "SUCCESS"} + api = _Api([{"run_id": 77, "state": {"life_cycle_state": "RUNNING"}}, {"state": legacy}]) + assert _go(api, capsys)[0] == 0 + + def test_a_run_now_that_fails_is_reported_without_a_traceback( + self, capsys, monkeypatch + ) -> None: + monkeypatch.setenv("DATABRICKS_TOKEN", TOKEN) + api = _Api([], fail={"run-now": PermissionError(f"PERMISSION_DENIED for {TOKEN}")}) + rc, out = _go(api, capsys) + assert rc == 1 + assert "could not start a run of Databricks job 'fluid sales.orders transform' (5)" in out + assert "PermissionError" in out and TOKEN not in out + # Tried three times, always with the one token: Databricks starts one run at most. + assert api.writes() == ["run-now"] * 3 + tokens = {b["idempotency_token"] for _m, e, _q, b in api.calls if e == "run-now"} + assert len(tokens) == 1 + assert "if Databricks started it anyway, it is on the job's page:" in out + + def test_a_poll_that_fails_says_the_run_may_still_be_going(self, capsys) -> None: + api = _Api([_run("RUNNING")]) + first = api.do + + def flaky(method: str, path: str, query: Any = None, body: Any = None, **kw: Any) -> Any: + if path.endswith("runs/get") and any(e == "runs/get" for _m, e, _q, _b in api.calls): + raise ConnectionError("connection reset") + return first(method, path, query=query, body=body, **kw) + + api.do = flaky # type: ignore[method-assign] + rc, out = _go(api, capsys) + assert rc == 1 + assert "could not read run 77 (ConnectionError: connection reset); trying again" in out + assert "could not read run 77 5 times in a row" in out and "may still be going" in out + # A run that cannot be seen is never cancelled: it may be well under way. + assert api.writes() == ["run-now"] + assert f"run page: {URL}" in out + + def test_one_failed_read_is_tried_again_and_the_run_goes_on(self, capsys) -> None: + # A plain 500, 502 or 504 the SDK does not repeat: one says nothing about the run. + api = _Api([_run("RUNNING"), _run("TERMINATED", code="SUCCESS")]) + first = api.do + failed: list = [] + + def once(method: str, path: str, query: Any = None, body: Any = None, **kw: Any) -> Any: + if path.endswith("runs/get") and not failed: + failed.append(1) + raise RuntimeError("502 Bad Gateway") + return first(method, path, query=query, body=body, **kw) + + api.do = once # type: ignore[method-assign] + rc, out = _go(api, capsys) + assert rc == 0, out + assert "trying again" in out and "succeeded" in out + assert api.writes() == ["run-now"] + + def test_run_now_is_tried_again_with_the_same_token(self, capsys) -> None: + api = _Api([_run("TERMINATED", code="SUCCESS")]) + first = api.do + tries: list = [] + + def slow(method: str, path: str, query: Any = None, body: Any = None, **kw: Any) -> Any: + if path.endswith("run-now"): + tries.append(body["idempotency_token"]) + if len(tries) < 3: + raise TimeoutError("read timed out") + return first(method, path, query=query, body=body, **kw) + + api.do = slow # type: ignore[method-assign] + rc, out = _go(api, capsys) + assert rc == 0, out + assert len(tries) == 3 and len(set(tries)) == 1 + + def test_ctrl_c_once_the_run_has_ended_cancels_nothing(self, capsys) -> None: + api = _Api( + [_run("TERMINATED", code="SUCCESS", tasks=[_task("SUCCESS")])], + fail={"runs/get-output": KeyboardInterrupt()}, + ) + with pytest.raises(KeyboardInterrupt): + _go(api, capsys) + assert api.writes() == ["run-now"] + + def test_sigterm_while_waiting_cancels_the_run(self, capsys) -> None: + import os + import signal + import threading + + if threading.current_thread() is not threading.main_thread(): + pytest.skip("a signal handler needs the main thread") + before = signal.getsignal(signal.SIGTERM) + api = _Api([_run("RUNNING")]) + + class _Kill(_Time): + def sleep(self, seconds: float) -> None: + os.kill(os.getpid(), signal.SIGTERM) # what a cancelled CI job sends + super().sleep(seconds) + + with pytest.raises(KeyboardInterrupt): + _go(api, capsys, time_=_Kill()) + assert api.writes() == ["run-now", "runs/cancel"] + assert signal.getsignal(signal.SIGTERM) is before # put back + + +# ā”€ā”€ choosing what to run ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ +class TestRunJobBuild: + def test_a_dry_run_starts_nothing(self, capsys) -> None: + def boom() -> Any: + raise AssertionError("a dry run created a client") + + rc = run_job_build(CONTRACT, BUILD, {}, dry_run=True, client_factory=boom) + out = " ".join(capsys.readouterr().out.split()) + assert rc == 0 + assert "[DRY RUN] Build 'transform' - would start a run of Databricks job" in out + + @pytest.mark.parametrize("refs, said", [([], "no Databricks job"), ([1, 2], "2 Databricks")]) + def test_a_build_needs_exactly_one_job_in_the_state(self, capsys, refs, said) -> None: + def boom() -> Any: + raise AssertionError("created a client") + + jobs = {"transform": [JobRef(n) for n in refs]} + assert run_job_build(CONTRACT, BUILD, jobs, client_factory=boom) == 1 + err = " ".join(capsys.readouterr().err.split()) + assert f"holds {said}" in err + assert "tagged fluid_product=sales.orders, fluid_build=transform" in err + + def test_without_the_sdk_it_says_which_extra_to_install(self, capsys) -> None: + def missing() -> Any: + raise ImportError("No module named 'databricks'") + + assert run_job_build(CONTRACT, BUILD, JOBS, client_factory=missing) == 1 + assert "pip install 'data-product-forge[databricks]'" in capsys.readouterr().err + + def test_a_client_that_cannot_be_made_is_one_line(self, capsys, monkeypatch) -> None: + monkeypatch.setenv("DATABRICKS_TOKEN", TOKEN) + + def cannot_configure() -> Any: + raise ValueError(f"cannot configure default credentials {TOKEN}") + + assert run_job_build(CONTRACT, BUILD, JOBS, client_factory=cannot_configure) == 1 + err = capsys.readouterr().err + assert "could not create the Databricks client" in err and TOKEN not in err + + def test_started_builds_are_the_non_continuous_databricks_dbt_builds(self) -> None: + def build(build_id: str, platform: str = "databricks", trigger: str = "") -> Dict: + execution: Dict[str, Any] = {"runtime": {"platform": platform}} + if trigger: + execution["trigger"] = {"type": trigger} + return {"id": build_id, "engine": "dbt", "execution": execution} + + contract = { + "builds": [ + build("a"), + build("b", trigger="schedule"), + build("c", trigger="continuous"), + build("d", platform="local"), + ] + } + assert [b["id"] for b in started_builds(contract, None)] == ["a", "b"] + assert [b["id"] for b in started_builds(contract, "")] == ["a", "b"] + assert [b["id"] for b in started_builds(contract, "b")] == ["b"] + assert started_builds(contract, "c") == [] + + +# ā”€ā”€ the state, the commands and the cap ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ā”€ +def test_jobs_are_found_in_the_state_by_their_tags() -> None: + def job(job_id: Any, product: str, build: str, **extra: Any) -> Dict[str, Any]: + values = {"id": job_id, "url": f"u/{job_id}", "tags": {}} + if product: + values["tags"]["fluid_product"] = product + if build: + values["tags"]["fluid_build"] = build + return {"type": "databricks_job", "mode": "managed", "values": values, **extra} + + state = [ + job("11", "sales.orders", "transform"), + job("12", "sales.other", "transform"), # another product's job + job("13", "sales.orders", ""), # not one of ours + job("x", "sales.orders", "bad-id"), + job("14", "sales.orders", "nightly", mode="data"), + {"type": "databricks_sql_table", "values": {"id": "main.sales.orders"}}, + job("15", "sales.orders", "twice"), + job("16", "sales.orders", "twice"), + ] + assert jobs_in_state(state, "sales.orders") == { + "transform": [JobRef(11, "u/11")], + "twice": [JobRef(15, "u/15"), JobRef(16, "u/16")], + } + + +def test_the_job_the_emitter_writes_is_the_one_the_state_reader_finds() -> None: + from fluid_build.iac.providers import databricks as dbx + from fluid_build.iac.providers import databricks_jobs + from fluid_build.iac.source import contract_source + + contract = { + "fluidVersion": "0.7.6", + "kind": "DataProduct", + "id": "sales.orders", + "name": "Orders", + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "databricks", + "format": "delta_table", + "location": {"database": "main", "schema": "sales", "table": "orders"}, + }, + "contract": {"schema": [{"name": "id", "type": "bigint"}]}, + } + ], + "builds": [ + { + "id": "transform", + "engine": "dbt", + "repository": "https://github.com/example-org/orders?tag=v1", + "outputs": ["orders"], + "properties": {"model": "orders"}, + "execution": {"runtime": {"platform": "databricks"}}, + } + ], + } + with contract_source(None): + compiled = dbx._compile(contract, {dbx.WAREHOUSE_ENV: "wh-1"}) + assert not compiled.refusals + ((key, body),) = compiled.resources["databricks_job"].items() + # What ``tofu show -json`` lists for it once applied: the id, and the tags as sent. + state = [ + { + "address": f"databricks_job.{key}", + "type": "databricks_job", + "mode": "managed", + "values": {"id": "42", "tags": body["tags"]}, + } + ] + product = databricks_jobs.job_product(contract) + assert job_run.jobs_in_state(state, product) == {"transform": [JobRef(42)]} + + +@pytest.mark.parametrize( + "commands, expected", + [ + ( + ["dbt deps", "dbt build --select orders+"], + ["dbt deps", "dbt build --select orders+ --full-refresh"], + ), + (["dbt seed", "dbt run"], ["dbt seed --full-refresh", "dbt run --full-refresh"]), + (["dbt build --full-refresh"], ["dbt build --full-refresh"]), + (["dbt test"], ["dbt test"]), + ], +) +def test_full_refresh_goes_on_the_commands_that_take_it(commands, expected) -> None: + settings = {"tasks": [{"task_key": "x"}, {"dbt_task": {"commands": commands}}]} + assert full_refresh_commands(settings) == expected + + +def test_full_refresh_needs_dbt_commands() -> None: + assert full_refresh_commands({"tasks": [{"notebook_task": {}}]}) is None + + +@pytest.mark.parametrize( + "settings, env, expected", + [ + pytest.param({}, {}, DEFAULT_TIMEOUT_SECONDS, id="default-hour"), + pytest.param({}, {TIMEOUT_ENV: "120"}, 120, id="env"), + pytest.param({}, {TIMEOUT_ENV: "0"}, DEFAULT_TIMEOUT_SECONDS, id="env-zero-ignored"), + pytest.param({}, {TIMEOUT_ENV: "2h"}, DEFAULT_TIMEOUT_SECONDS, id="env-not-a-number"), + pytest.param( + {"tasks": [{"timeout_seconds": 600}]}, {}, DEFAULT_TIMEOUT_SECONDS, id="short-job" + ), + pytest.param( + { + "tasks": [ + {"timeout_seconds": 7200, "max_retries": 1, "min_retry_interval_millis": 60000} + ] + }, + {}, + 7200 * 2 + 60 + 900, + id="task-timeout-and-retries", + ), + pytest.param({"timeout_seconds": 5400}, {}, 5400 + 900, id="job-timeout"), + pytest.param( + {"tasks": [{"timeout_seconds": 7200}, {"task_key": "no-timeout"}]}, + {}, + DEFAULT_TIMEOUT_SECONDS, + id="a-task-without-timeout", + ), + pytest.param( + {"tasks": [{"timeout_seconds": 7200, "max_retries": -1}]}, + {}, + DEFAULT_TIMEOUT_SECONDS, + id="endless-retries", + ), + pytest.param( + {"timeout_seconds": 99999}, {TIMEOUT_ENV: "300"}, 300, id="env-wins-over-the-job" + ), + ], +) +def test_the_wait_cap(settings, env, expected) -> None: + assert wait_budget(settings, env) == expected + + +def test_a_success_shows_dbt_s_own_summary(capsys) -> None: + logs = ( + "\x1b[0m19:54:43 \x1b[32mCompleted successfully\x1b[0m\n" + "\x1b[0m19:54:43 Done. PASS=1 WARN=0 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=1\n" + ) + api = _Api( + [_run("TERMINATED", code="SUCCESS", tasks=[_task("SUCCESS")])], + outputs={78: {"logs": logs}}, + ) + rc, out = _go(api, capsys) + assert rc == 0, out + assert "task 'dbt': Done. PASS=1 WARN=0 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=1" in out + + +def test_a_summary_that_cannot_be_read_does_not_fail_the_build(capsys) -> None: + api = _Api( + [_run("TERMINATED", code="SUCCESS", tasks=[_task("SUCCESS")])], + fail={"runs/get-output": ConnectionError("connection reset")}, + ) + rc, out = _go(api, capsys) + assert rc == 0 and "task 'dbt'" not in out and "succeeded" in out + + +def test_the_task_error_is_one_line(capsys) -> None: + # As Databricks words it for a dbt task: the shell command quoted with "\n" escapes. + error = "Command 'dbt build --select orders\\nelse\\n exit 1\\n\\nfi\\n' returned non-zero exit status 1." + api = _Api( + [_run("TERMINATED", code="RUN_EXECUTION_ERROR", tasks=[_task("RUN_EXECUTION_ERROR")])], + outputs={78: {"error": error}}, + ) + rc, out = _go(api, capsys) + assert rc == 1 + assert ( + "task 'dbt': Command 'dbt build --select orders else exit 1 fi ' returned non-zero" in out + ) + + +def test_a_run_id_from_the_list_is_printed_as_a_number_only(capsys) -> None: + api = _Api( + [_run("QUEUED"), _run("TERMINATED", code="SUCCESS")], + active=[{"run_id": "\x1b[2J12"}, {"run_id": 77}], + ) + rc, out = _go(api, capsys) + assert rc == 0 and "\x1b" not in out + assert "waiting for the workspace's limit" in out # nothing but a number counts as a run + + +def test_a_run_that_cannot_be_cancelled_at_the_cap_is_said(capsys) -> None: + api = _Api([_run("RUNNING")], fail={"runs/cancel": ConnectionError("down")}) + rc, out = _go(api, capsys, env={TIMEOUT_ENV: "30"}) + assert rc == 1 + assert "could not cancel run 77: ConnectionError: down" in out + assert "did not end within 30s." in out and f"run page: {URL}" in out + + +@pytest.mark.parametrize("value", ["\u00b2", "9" * 5000]) +def test_a_timeout_int_cannot_read_is_said_before_anything_starts(capsys, value) -> None: + # str.isdigit() takes a superscript two and a 5000-digit number; int() refuses both. + api = _Api([_run("TERMINATED", code="SUCCESS")]) + rc, out = _go(api, capsys, env={TIMEOUT_ENV: value}) + assert rc == 0, out + assert "is not a whole number of seconds above 0" in out + + +def test_a_build_without_an_id_finds_the_job_the_emitter_tagged_by_position(capsys) -> None: + build = {"engine": "dbt", "execution": {"runtime": {"platform": "databricks"}}} + contract = {"id": "sales.orders", "builds": [build]} + api = _Api([_run("TERMINATED", code="SUCCESS")]) + rc = run_job_build( + contract, + build, + {"0": [JobRef(5)]}, + client_factory=lambda: _Client(api), + env={}, + sleep=_Time().sleep, + clock=_Time().clock, + ) + out = " ".join(capsys.readouterr().out.split()) + assert rc == 0, out + assert "started run 77 of Databricks job 'fluid sales.orders 0' (5)" in out + + +def test_a_queued_run_of_the_job_is_not_the_one_ahead(capsys) -> None: + api = _Api( + [_run("QUEUED"), _run("TERMINATED", code="SUCCESS")], + active=[{"run_id": 90, "status": {"state": "QUEUED"}}, {"run_id": 77}], + ) + rc, out = _go(api, capsys) + assert rc == 0 + assert "behind run 90" not in out and "waiting for the workspace's limit" in out + + +@pytest.mark.parametrize("value", ["1h", "0", "-5", "soon"]) +def test_a_timeout_that_is_not_seconds_is_said(capsys, value) -> None: + api = _Api([_run("TERMINATED", code="SUCCESS")]) + rc, out = _go(api, capsys, env={TIMEOUT_ENV: value}) + assert rc == 0 + assert f"{TIMEOUT_ENV}='{value}' is not a whole number of seconds above 0" in out + assert "waiting up to 1h00m00s" in out + + +def test_a_timeout_in_seconds_is_taken_quietly(capsys) -> None: + api = _Api([_run("TERMINATED", code="SUCCESS")]) + rc, out = _go(api, capsys, env={TIMEOUT_ENV: "120"}) + assert rc == 0 and TIMEOUT_ENV not in out + + +def test_a_sighup_the_operator_ignores_stays_ignored(capsys) -> None: + """`nohup fluid apply ...` ignores SIGHUP: a closed terminal must not cancel the run.""" + import signal + import threading + + if threading.current_thread() is not threading.main_thread(): + pytest.skip("a signal handler needs the main thread") + before = signal.signal(signal.SIGHUP, signal.SIG_IGN) + seen: list = [] + try: + + class _Look(_Time): + def sleep(self, seconds: float) -> None: + seen.append(signal.getsignal(signal.SIGHUP)) + super().sleep(seconds) + + api = _Api([_run("RUNNING"), _run("TERMINATED", code="SUCCESS")]) + assert _go(api, capsys, time_=_Look())[0] == 0 + assert seen and all(handler is signal.SIG_IGN for handler in seen) + assert signal.getsignal(signal.SIGHUP) is signal.SIG_IGN + finally: + signal.signal(signal.SIGHUP, before) diff --git a/tests/cli/test_apply_databricks_job_runs.py b/tests/cli/test_apply_databricks_job_runs.py new file mode 100644 index 00000000..39121583 --- /dev/null +++ b/tests/cli/test_apply_databricks_job_runs.py @@ -0,0 +1,265 @@ +# Copyright 2025 Agentics Transformation Limited +# SPDX-License-Identifier: Apache-2.0 + +"""``fluid apply`` hands its build phase the Databricks jobs this apply provisioned. + +On the OpenTofu engine, a ``-and-build`` apply of a contract with a dbt build that runs +on Databricks asks the engine for the state the apply leaves (``applied_state``) and +gives the build runner the jobs there (``databricks_jobs``), so the runner starts the +job this apply provisioned (``build_runners/_databricks_job_run.py``). ``--mode amend``, +a continuous job, and a contract or ``--build-id`` without such a build read no state. +A build phase that would start a job without databricks-sdk is refused before ``tofu`` +runs. Driven through ``run()`` and the real argparse surface; ``tofu``, the hooks and +the build runner are recorders. +""" + +from __future__ import annotations + +import copy +import importlib.util +import logging +from pathlib import Path +from typing import Any, Dict, Iterator, List +from unittest import mock + +import pytest +import yaml + +import fluid_build.cli as cli_mod +from fluid_build.build_runners._databricks_job_run import JobRef +from fluid_build.cli._common import CLIError +from fluid_build.cli.apply import _gate_contract_for_apply +from fluid_build.cli.apply import run as apply_run + +pytestmark = pytest.mark.unit + +LOG = logging.getLogger("test.apply_databricks_job_runs") + +_CONTRACT: Dict[str, Any] = { + "fluidVersion": "0.7.6", + "kind": "DataProduct", + "id": "sales.orders", + "name": "Orders", + "description": "A dbt build that runs on Databricks.", + "metadata": {"owner": {"team": "data-platform"}, "layer": "Silver"}, + "exposes": [ + { + "exposeId": "orders", + "kind": "table", + "binding": { + "platform": "databricks", + "format": "delta_table", + "location": {"database": "main", "schema": "sales", "table": "orders"}, + }, + "contract": {"schema": [{"name": "id", "type": "bigint"}]}, + } + ], + "builds": [ + { + "id": "transform", + "engine": "dbt", + "pattern": "hybrid-reference", + "repository": "https://github.com/example-org/orders?tag=v1", + "outputs": ["orders"], + "properties": {"model": "orders"}, + "execution": {"trigger": {"type": "manual"}, "runtime": {"platform": "databricks"}}, + } + ], +} + +#: What ``tofu show -json`` lists after the apply: this product's job, another's, a table. +_STATE: List[Dict[str, Any]] = [ + { + "type": "databricks_job", + "mode": "managed", + "values": { + "id": "42", + "url": "https://dbc-1.cloud.databricks.com/#job/42", + "tags": {"fluid_product": "sales.orders", "fluid_build": "transform"}, + }, + }, + { + "type": "databricks_job", + "mode": "managed", + "values": {"id": "43", "tags": {"fluid_product": "sales.other", "fluid_build": "x"}}, + }, + {"type": "databricks_sql_table", "mode": "managed", "values": {"id": "main.sales.orders"}}, +] + + +@pytest.fixture +def workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("FLUID_ENV", raising=False) + return tmp_path + + +@pytest.fixture +def tofu() -> Iterator[mock.Mock]: + """The OpenTofu engine, as a recorder that fills ``applied_state`` when asked.""" + + def _apply(args: Any, logger: Any, contract: Any = None, **kwargs: Any) -> int: + if "applied_state" in kwargs: + kwargs["applied_state"].extend(copy.deepcopy(_STATE)) + return 0 + + with ( + mock.patch( + "fluid_build.cli._apply_opentofu_engine.resolve_apply_engine", + return_value="opentofu", + ), + mock.patch( + "fluid_build.cli._apply_opentofu_engine.apply_via_opentofu", side_effect=_apply + ) as engine, + mock.patch("fluid_build.cli.apply._run_apply_hooks", return_value=0), + ): + yield engine + + +@pytest.fixture +def runner() -> Iterator[mock.Mock]: + with mock.patch("fluid_build.build_runners.run_builds_from_args", return_value=0) as spy: + yield spy + + +def _apply(contract: Dict[str, Any], *flags: str) -> int: + Path("contract.fluid.yaml").write_text(yaml.safe_dump(contract, sort_keys=False)) + argv = ["apply", "contract.fluid.yaml", *flags, "--yes"] + return apply_run(cli_mod.build_parser().parse_args(argv), LOG) + + +def _with_build(**changes: Any) -> Dict[str, Any]: + contract = copy.deepcopy(_CONTRACT) + contract["builds"][0].update(changes) + return contract + + +def test_the_contract_passes_the_apply_gate() -> None: + _gate_contract_for_apply(copy.deepcopy(_CONTRACT), LOG) + + +@pytest.fixture +def sdk_installed() -> Iterator[None]: + """databricks-sdk as the apply's gate sees it, whether or not this venv has it (CI's + `.[dev,local]` install does not).""" + from importlib.machinery import ModuleSpec + + real = importlib.util.find_spec + + def _find(name: str, *args: Any, **kwargs: Any) -> Any: + if name == "databricks.sdk": + return ModuleSpec("databricks.sdk", None) + return real(name, *args, **kwargs) + + with mock.patch("importlib.util.find_spec", side_effect=_find): + yield + + +@pytest.mark.usefixtures("workspace", "sdk_installed") +class TestTheBuildPhaseGetsThisApplysJobs: + @pytest.mark.parametrize( + "flags", + [ + pytest.param(("--mode", "amend-and-build"), id="amend-and-build"), + pytest.param( + ("--mode", "replace-and-build", "--allow-data-loss"), id="replace-and-build" + ), + ], + ) + def test_a_build_mode_reads_the_state_and_hands_over_the_jobs( + self, tofu: mock.Mock, runner: mock.Mock, flags: tuple + ) -> None: + assert _apply(copy.deepcopy(_CONTRACT), *flags) == 0 + assert tofu.call_args.kwargs["applied_state"] == _STATE + assert runner.call_args.kwargs["databricks_jobs"] == { + "transform": [JobRef(42, "https://dbc-1.cloud.databricks.com/#job/42")] + } + + def test_amend_provisions_only(self, tofu: mock.Mock, runner: mock.Mock) -> None: + assert _apply(copy.deepcopy(_CONTRACT), "--mode", "amend") == 0 + assert "applied_state" not in tofu.call_args.kwargs + runner.assert_not_called() + + @pytest.mark.parametrize( + "contract, flags", + [ + pytest.param( + _with_build( + execution={ + "trigger": {"type": "continuous"}, + "runtime": {"platform": "databricks"}, + } + ), + (), + id="continuous-job", + ), + pytest.param(_with_build(execution={"trigger": {"type": "manual"}}), (), id="local"), + pytest.param( + { + **copy.deepcopy(_CONTRACT), + "builds": [ + copy.deepcopy(_CONTRACT["builds"][0]), + {**copy.deepcopy(_CONTRACT["builds"][0]), "id": "local", "execution": {}}, + ], + }, + ("--build-id", "local"), + id="build-id-selects-another-build", + ), + ], + ) + def test_no_state_is_read_when_no_job_is_started( + self, tofu: mock.Mock, runner: mock.Mock, contract: Dict[str, Any], flags: tuple + ) -> None: + assert _apply(contract, "--mode", "amend-and-build", *flags) == 0 + assert "applied_state" not in tofu.call_args.kwargs + assert "databricks_jobs" not in runner.call_args.kwargs + + +@pytest.mark.usefixtures("workspace") +class TestDatabricksSdkIsCheckedFirst: + @pytest.fixture + def no_sdk(self) -> Iterator[None]: + real = importlib.util.find_spec + + def _find(name: str, *args: Any, **kwargs: Any) -> Any: + if name == "databricks.sdk": + return None + return real(name, *args, **kwargs) + + with mock.patch("importlib.util.find_spec", side_effect=_find): + yield + + @pytest.mark.usefixtures("no_sdk") + def test_a_build_mode_without_the_sdk_is_refused_before_tofu( + self, tofu: mock.Mock, runner: mock.Mock + ) -> None: + with pytest.raises(CLIError) as exc: + _apply(copy.deepcopy(_CONTRACT), "--mode", "amend-and-build") + assert exc.value.event == "apply_databricks_sdk_missing" + assert "data-product-forge[databricks]" in exc.value.context["hint"] + tofu.assert_not_called() + runner.assert_not_called() + + @pytest.mark.usefixtures("no_sdk") + @pytest.mark.parametrize( + "flags", + [ + pytest.param(("--mode", "amend"), id="amend"), + pytest.param(("--mode", "amend-and-build", "--dry-run"), id="dry-run"), + ], + ) + def test_nothing_that_starts_no_job_needs_it( + self, tofu: mock.Mock, runner: mock.Mock, flags: tuple + ) -> None: + assert _apply(copy.deepcopy(_CONTRACT), *flags) == 0 + tofu.assert_called_once() + + +def test_the_native_engine_never_needs_the_sdk() -> None: + """Only the OpenTofu engine provisions the job and starts its run.""" + from fluid_build.cli.apply import _gate_databricks_sdk_for_apply + + with mock.patch("importlib.util.find_spec", return_value=None): + _gate_databricks_sdk_for_apply(copy.deepcopy(_CONTRACT), None, False, "native") + with pytest.raises(CLIError): + _gate_databricks_sdk_for_apply(copy.deepcopy(_CONTRACT), None, False, "opentofu") diff --git a/tests/iac/_fake_databricks.py b/tests/iac/_fake_databricks.py index 186c0de9..fd63c7b6 100644 --- a/tests/iac/_fake_databricks.py +++ b/tests/iac/_fake_databricks.py @@ -43,7 +43,9 @@ effective-permissions read. Jobs and workspace files are kept as sent (``jobs[id]["settings"]``, ``workspace_files[path]``), so a test asserts on exactly what the provider submitted; ``add_job`` / ``add_workspace_file`` -seed brownfield objects. +seed brownfield objects. A job can be run (``run-now``, ``runs/get``, ``runs/get-output``, +``runs/cancel``): each run plays ``run_states`` and ends as ``run_result`` says, and +``job_runs`` keeps every run-now request as sent. It also serves the Files API on Unity Catalog volumes (``add_volume``): PUT / DELETE of a file, PUT / GET / DELETE of a directory, backed by a temporary directory so the bytes a client uploads @@ -55,8 +57,8 @@ unless ``insert_reports_count`` is False, and ``SELECT COUNT(*) FROM ``. Not proven here: what real Databricks accepts (SQL dialect, warehouse state, the privilege -model, Unity Catalog storage, ``read_files`` itself), whether a job's settings are valid (the fake stores any), job -runs, and any workspace-level setting. +model, Unity Catalog storage, ``read_files`` itself), whether a job's settings are valid (the fake stores any), that +a job run does anything (the fake runs nothing), and any workspace-level setting. The provider is pointed at it with ``host`` and a dummy ``token``. Standalone: ``python _fake_databricks.py --port 8765``. @@ -1050,6 +1052,18 @@ def __init__( self.table_rows: Dict[str, List[Dict[str, Any]]] = {} # False: an INSERT answers with no row count, as a statement that returns none would self.insert_reports_count = True + # run_id -> {job_id, body (the run-now request), polls, cancelled}. A run is served + # one state of ``run_states`` per runs/get, then ends as ``run_result`` says; its one + # task ("dbt", run id + 100000) serves ``run_result``'s error and logs as its output. + self.job_runs: Dict[int, Dict[str, Any]] = {} + self.run_states: List[str] = ["PENDING", "RUNNING"] + self.run_result: Dict[str, str] = { + "code": "SUCCESS", + "message": "", + "error": "", + "logs": "", + } + self._run_seq = 500 # Genie spaces by space_id, kept as the service keeps them (serialized_space is the # string it returns: what was sent, minus every ``false`` flag, as measured on a live # workspace), and the permissions set on each one directly: {(key, name): level}. @@ -1338,6 +1352,10 @@ def _register(self) -> None: a("GET", r"/api/2\.2/jobs/list", self._job_list) a("POST", r"/api/2\.2/jobs/reset", self._job_reset) a("POST", r"/api/2\.2/jobs/delete", self._job_delete) + a("POST", r"/api/2\.2/jobs/run-now", self._run_now) + a("GET", r"/api/2\.2/jobs/runs/get", self._run_get) + a("GET", r"/api/2\.2/jobs/runs/get-output", self._run_output) + a("POST", r"/api/2\.2/jobs/runs/cancel", self._run_cancel) # ---- workspace files a("POST", r"/api/2\.0/workspace/mkdirs", self._ws_mkdirs) a("POST", r"/api/2\.0/workspace/import", self._ws_import) @@ -2043,6 +2061,70 @@ def _job_delete(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: self.jobs.pop(int(b.get("job_id", 0)), None) return 200, {} + # -- job runs + def _run_now(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + jid = int(b.get("job_id", 0)) + if jid not in self.jobs: + return _err(400, "INVALID_PARAMETER_VALUE", f"Job {jid} does not exist.") + token = b.get("idempotency_token") + for rid, run in self.job_runs.items(): # the same token is the same run + if token and run["body"].get("idempotency_token") == token: + return 200, {"run_id": rid, "number_in_job": rid} + self._run_seq += 1 + self.job_runs[self._run_seq] = { + "job_id": jid, + "body": copy.deepcopy(b), + "polls": 0, + "cancelled": False, + } + return 200, {"run_id": self._run_seq, "number_in_job": self._run_seq} + + def _run_get(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + rid = int((q.get("run_id") or ["0"])[0]) + run = self.job_runs.get(rid) + if run is None: + return _err(400, "INVALID_PARAMETER_VALUE", f"Run {rid} does not exist.") + polls, run["polls"] = run["polls"], run["polls"] + 1 + doc: Dict[str, Any] = { + "run_id": rid, + "job_id": run["job_id"], + "run_page_url": f"{self.endpoint}/#job/{run['job_id']}/run/{rid}", + } + if not run["cancelled"] and polls < len(self.run_states): + doc["status"] = {"state": self.run_states[polls]} + return 200, doc + code, message = ( + ("CANCELED", "Run cancelled.") + if run["cancelled"] + else (self.run_result["code"], self.run_result["message"]) + ) + details = {"code": code, "message": message} + doc["status"] = {"state": "TERMINATED", "termination_details": details} + doc["tasks"] = [ + { + "task_key": "dbt", + "run_id": rid + 100000, + "attempt_number": 0, + "status": {"state": "TERMINATED", "termination_details": {"code": code}}, + } + ] + return 200, doc + + def _run_output(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + rid = int((q.get("run_id") or ["0"])[0]) - 100000 + if rid not in self.job_runs: + return _err(400, "INVALID_PARAMETER_VALUE", f"Run {rid + 100000} does not exist.") + return 200, { + "error": self.run_result.get("error") or None, + "logs": self.run_result.get("logs") or None, + } + + def _run_cancel(self, m: Any, q: Any, b: Any) -> Tuple[int, Any]: + rid = int(b.get("run_id", 0)) + if rid in self.job_runs: + self.job_runs[rid]["cancelled"] = True + return 200, {} + # -- workspace files def _put_file(self, path: str, data: bytes) -> None: parent = path.rsplit("/", 1)[0] diff --git a/tests/iac/test_iac_databricks_jobs_fake_e2e.py b/tests/iac/test_iac_databricks_jobs_fake_e2e.py index d7050592..c1cd5872 100644 --- a/tests/iac/test_iac_databricks_jobs_fake_e2e.py +++ b/tests/iac/test_iac_databricks_jobs_fake_e2e.py @@ -24,6 +24,7 @@ from __future__ import annotations +import importlib.util import json from pathlib import Path from typing import Any, Dict, Iterator, Optional @@ -243,6 +244,79 @@ def test_the_table_a_dbt_build_writes_keeps_what_dbt_made_and_the_drift_is_said( assert fake.sql_log[statements:] == [] and fake.tables[TABLE]["marker"] == marker +def test_amend_provisions_the_job_and_starts_no_run(workspace, fake, capfd): + rc, out = apply_contract(capfd, write_contract(workspace, job_contract())) + assert rc == 0, out + assert len(fake.jobs) == 1 and fake.job_runs == {} + + +def _sdk_missing() -> bool: + try: + return importlib.util.find_spec("databricks.sdk") is None + except ModuleNotFoundError: + return True + + +@pytest.fixture +def run_client(monkeypatch, fake: FakeDatabricks) -> None: + """The client the build phase starts the job through. + + With the [databricks] extra installed it is the real SDK, against the stand-in. CI's lane + installs `.[dev,test-emulators]`, which has no SDK; there it is the stand-in's own HTTP + client and the SDK gate is lifted, so the run is still driven end to end. + """ + if _sdk_missing(): + monkeypatch.setattr( + "fluid_build.build_runners._databricks_job_run._default_client", fake.client + ) + monkeypatch.setattr( + "fluid_build.cli.apply._gate_databricks_sdk_for_apply", lambda *_a, **_k: None + ) + + +@pytest.mark.usefixtures("run_client") +def test_amend_and_build_starts_the_job_this_apply_provisioned_and_waits(workspace, fake, capfd): + path = write_contract(workspace, job_contract()) + rc, out = apply_contract(capfd, path, "--mode", "amend-and-build") + assert rc == 0, out + + ((job_id, job),) = fake.jobs.items() + ((run_id, run),) = fake.job_runs.items() + assert run["job_id"] == job_id # the id this apply's state holds for the build + assert run["body"]["queue"] == {"enabled": True} + assert "dbt_commands" not in run["body"] + assert run["polls"] == 3 # PENDING, RUNNING, then the end + name = "fluid dbx.fake.orders dbt_daily" + assert f"started run {run_id} of Databricks job '{name}' ({job_id})" in out + assert f"run page: {fake.endpoint}/#job/{job_id}/run/{run_id}" in out + assert "PENDING" in out and "RUNNING" in out and "TERMINATED SUCCESS" in out + assert f"Databricks run {run_id} succeeded" in out and "Executed: 1" in out + # Running it now wrote nothing to the job: its schedule stays as the apply set it. + assert job["settings"]["schedule"]["pause_status"] == "UNPAUSED" + started = fake.requests.index(("POST", "/api/2.2/jobs/run-now")) + assert not [r for r in fake.requests[started:] if r[0] == "POST" and "jobs/run" not in r[1]] + + +@pytest.mark.usefixtures("run_client") +def test_a_failed_run_fails_amend_and_build_with_dbt_s_output(workspace, fake, capfd): + fake.run_states = [] + fake.run_result = { + "code": "RUN_EXECUTION_ERROR", + "message": "Task dbt failed with message: dbt command failed.", + "error": "dbt command failed", + "logs": "Running with dbt=1.10.0\nCompilation Error in model orders (models/orders.sql)\n" + " [UNRESOLVED_COLUMN] A column with name `nope` cannot be resolved.", + } + rc, out = apply_contract( + capfd, write_contract(workspace, job_contract()), "--mode", "amend-and-build" + ) + assert rc == 1, out + assert "ended RUN_EXECUTION_ERROR" in out + assert "task 'dbt': dbt command failed" in out + assert "A column with name `nope` cannot be resolved." in out + assert "Failed: 1" in out + + def test_a_job_that_writes_a_keyed_table_declares_its_keys_again_after_dbt(workspace, fake, capfd): document = job_contract() document["exposes"][0]["contract"]["schema"][0]["tags"] = ["primary-key"]