diff --git a/acceptance/bin/dms_resources.py b/acceptance/bin/dms_resources.py new file mode 100644 index 00000000000..25e469c5bad --- /dev/null +++ b/acceptance/bin/dms_resources.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +""" +Read resource ids and state from the deployment metadata service. + +While a bundle records deployment history the service owns the resource set, so the state file is +not where ids and state come from. The service is asked instead, which takes two lookups: the CLI +keeps no deployment id locally, and the id is the object id of the workspace node the service +registers under /resources.deployment.json (see libs/dms/resolve.go). +""" + +import functools +import glob +import json +import os +import posixpath +import subprocess +import sys + +sys.path.insert(0, os.path.dirname(__file__)) +from print_state import get_state_file + +CLI = os.environ.get("CLI", "databricks") + +# Must match dms.DeploymentNodeName. +DEPLOYMENT_NODE_NAME = "resources.deployment.json" + + +def run_json(cmd, allow_failure=False): + """Run cmd and parse its stdout, or return None if it fails and allow_failure is set. stderr is + captured rather than inherited: these lookups are plumbing, and a CLI warning like "no files to + sync" would otherwise land in the test output.""" + result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8") + if result.returncode != 0: + if allow_failure: + return None + raise SystemExit(f"{cmd} failed with code {result.returncode}\n{result.stdout}{result.stderr}".strip()) + return json.loads(result.stdout) + + +def records_deployment_history(): + """Whether this run records deployment history, so the service is what to ask.""" + return os.environ.get("DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY") == "true" + + +def get_remote_state_path(target): + """The bundle's remote state directory. + + Preferred source is the sync snapshot, because it needs no CLI call: re-running the config + load would need whatever --var and flags the test deployed with, which a helper cannot know. + A bundle with no files to sync writes no snapshot, so fall back to asking the CLI - those + bundles are the ones with nothing to parameterize.""" + target_dir = os.path.dirname(get_state_file(target, False)) + snapshots = glob.glob(f"{target_dir}/sync-snapshots/*.json") + if snapshots: + # One snapshot per remote path, so a test that moved its root leaves several: the newest + # is the one the last deploy used. + newest = max(snapshots, key=os.path.getmtime) + remote_path = json.loads(open(newest).read())["remote_path"] + # state and files are siblings under the bundle root. + return posixpath.join(posixpath.dirname(remote_path), "state") + + args = [CLI, "bundle", "validate", "--output", "json"] + if target: + args += ["-t", target] + return run_json(args)["workspace"]["state_path"] + + +@functools.cache +def get_resources(target): + """Map every recorded resource key ("jobs.foo") to its {"id", "state"}. + + Empty when the bundle has no deployment recorded yet. Cached because a lookup costs three + round trips and a script asks for one resource at a time. + """ + state_path = get_remote_state_path(target) + if not state_path: + return {} + + # No node means nothing has been recorded, the conclusion dms.resolveDeploymentID also draws + # from a 404 - the deployment is gone once the bundle is destroyed. + node = run_json([CLI, "workspace", "get-status", f"{state_path}/{DEPLOYMENT_NODE_NAME}"], allow_failure=True) + if not node or not node.get("object_id"): + return {} + deployment_id = node["object_id"] + + result = {} + # The service pages at 50 resources; the local fake returns everything at once. + page_token = None + while True: + url = f"/api/2.0/bundle/deployments/{deployment_id}/resources" + if page_token: + url += f"?page_token={page_token}" + listed = run_json([CLI, "api", "get", url]) + for resource in listed.get("resources") or []: + # The service stores state as the opaque envelope the CLI wrote (dstate.RecordedState), + # so unwrap it to the resource state itself. + envelope = json.loads(resource["state"]) if resource.get("state") else {} + result[resource["resource_key"]] = { + "id": resource.get("resource_id"), + "state": envelope.get("state") or {}, + "depends_on": envelope.get("depends_on") or [], + } + page_token = listed.get("next_page_token") + if not page_token: + return result diff --git a/acceptance/bin/print_state.py b/acceptance/bin/print_state.py index 7e82f7b3d16..2959a76f81b 100755 --- a/acceptance/bin/print_state.py +++ b/acceptance/bin/print_state.py @@ -8,9 +8,14 @@ import argparse import glob +import json import os +def records_deployment_history(): + return os.environ.get("DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY") == "true" + + def print_file(filename): data = open(filename).read() print(data, end="") @@ -53,6 +58,26 @@ def get_state_file(target, backup): return filtered[0] if filtered else result[0] +def print_recorded_state(filename, target): + """Print the state file with its resources filled in from the deployment metadata service. + + While recording, the file itself carries only the header - the service holds the resources - so + printing it raw would show an empty state and differ from the same test's non-recording run. + """ + # Imported here rather than at module level: dms_resources reads get_state_file from this module. + from dms_resources import get_resources + + data = json.loads(open(filename).read()) + state = {} + for key, value in sorted(get_resources(target).items()): + entry = {"__id__": value["id"], "state": value["state"]} + if value["depends_on"]: + entry["depends_on"] = value["depends_on"] + state[f"resources.{key}"] = entry + data["state"] = state + print(json.dumps(data, indent=1)) + + def main(): parser = argparse.ArgumentParser() parser.add_argument("-t", "--target") @@ -60,7 +85,11 @@ def main(): args = parser.parse_args() for filename in get_state_files(args.target, args.backup): - if os.path.exists(filename): + if not os.path.exists(filename): + continue + if filename.endswith("resources.json") and records_deployment_history(): + print_recorded_state(filename, args.target) + else: print_file(filename) diff --git a/acceptance/bin/read_id.py b/acceptance/bin/read_id.py index 87cd2954bdf..06e7b69281e 100755 --- a/acceptance/bin/read_id.py +++ b/acceptance/bin/read_id.py @@ -15,6 +15,7 @@ sys.path.insert(0, str(Path(__file__).parent)) from add_repl import add_repl +from dms_resources import get_resources, records_deployment_history from print_state import get_state_file @@ -34,6 +35,15 @@ def get_id_terraform(filename, name): print(f"Cannot find resource with {name=}. Available: {available}", file=sys.stderr) +def get_id_recorded(target, name): + resources = get_resources(target) + for key, value in resources.items(): + if key.split(".")[1] == name: + return value["id"] + + print(f"Cannot find recorded resource with {name=}. Available: {list(resources)}", file=sys.stderr) + + def get_id_direct(filename, name): raw = open(filename).read() data = json.loads(raw) @@ -53,11 +63,14 @@ def main(): parser.add_argument("name") args = parser.parse_args() - filename = get_state_file(args.target, args.backup) - if filename.endswith(".tfstate"): - id = get_id_terraform(filename, args.name) + if records_deployment_history(): + id = get_id_recorded(args.target, args.name) else: - id = get_id_direct(filename, args.name) + filename = get_state_file(args.target, args.backup) + if filename.endswith(".tfstate"): + id = get_id_terraform(filename, args.name) + else: + id = get_id_direct(filename, args.name) if id: print(id) diff --git a/acceptance/bin/read_state.py b/acceptance/bin/read_state.py index 0166bf9abbb..4d9bf84186e 100755 --- a/acceptance/bin/read_state.py +++ b/acceptance/bin/read_state.py @@ -9,6 +9,9 @@ import os import sys +sys.path.insert(0, os.path.dirname(__file__)) +from dms_resources import get_resources, records_deployment_history + def print_resource_terraform(group, name, *attrs): resource_type = "databricks_" + group[:-1] @@ -50,7 +53,21 @@ def print_resource_direct(group, name, *attrs): print(group, name, " ".join(values)) -if os.environ.get("DATABRICKS_BUNDLE_ENGINE", "").startswith("direct"): +def print_resource_recorded(group, name, *attrs): + result = get_resources(None).get(f"{group}.{name}") + if result is None: + print(f"State not found for {group}.{name}") + return + + state = dict(result["state"]) + state.setdefault("id", result["id"]) + values = [f"{x}={state.get(x)!r}" for x in attrs] + print(group, name, " ".join(values)) + + +if records_deployment_history(): + print_resource_recorded(*sys.argv[1:]) +elif os.environ.get("DATABRICKS_BUNDLE_ENGINE", "").startswith("direct"): print_resource_direct(*sys.argv[1:]) else: print_resource_terraform(*sys.argv[1:]) diff --git a/acceptance/bin/replace_ids.py b/acceptance/bin/replace_ids.py index e0ebf1cd0e6..2f5d8165032 100755 --- a/acceptance/bin/replace_ids.py +++ b/acceptance/bin/replace_ids.py @@ -10,6 +10,7 @@ sys.path.insert(0, str(Path(__file__).parent)) from add_repl import add_repl +from dms_resources import get_resources, records_deployment_history from print_state import get_state_file @@ -26,6 +27,12 @@ def iter_ids_terraform(filename): yield r_name, id +def iter_ids_recorded(target): + for key, value in get_resources(target).items(): + if value["id"]: + yield key.split(".")[1], value["id"] + + def iter_ids_direct(filename): raw = open(filename).read() data = json.loads(raw) @@ -44,11 +51,14 @@ def main(): parser.add_argument("--backup", action="store_true") args = parser.parse_args() - filename = get_state_file(args.target, args.backup) - if filename.endswith(".tfstate"): - it = iter_ids_terraform(filename) + if records_deployment_history(): + it = iter_ids_recorded(args.target) else: - it = iter_ids_direct(filename) + filename = get_state_file(args.target, args.backup) + if filename.endswith(".tfstate"): + it = iter_ids_terraform(filename) + else: + it = iter_ids_direct(filename) for name, id in it: add_repl(id, name.upper() + "_ID") diff --git a/acceptance/bundle/deploy/files/out-of-band-delete/test.toml b/acceptance/bundle/deploy/files/out-of-band-delete/test.toml index 8c3f2408d5c..f8f11b0a174 100644 --- a/acceptance/bundle/deploy/files/out-of-band-delete/test.toml +++ b/acceptance/bundle/deploy/files/out-of-band-delete/test.toml @@ -1,6 +1,8 @@ -# Recording needs a bundle it has seen from the start. This test seeds a state file, -# so recording refuses it. TODO(DMS): drop this once existing state can be -# handed over to the service (see the TODO in dstate.Open). +# The out-of-band delete takes the whole bundle root, including the workspace node the +# service registers the deployment under. The state file survives and is accepted (it records +# record_deployment_history), so the redeploy re-stamps the job with the new deployment's id +# and reports it as changed where the non-recording run reports it unchanged. Both are right, +# but the two variants share one golden. EnvMatrix.DMS = [""] Badness = "After the remote bundle files are deleted out-of-band, the next deploy does not re-upload them until the local sync snapshot is removed." diff --git a/acceptance/bundle/deploy/readplan/test.toml b/acceptance/bundle/deploy/readplan/test.toml index e6ef3fcdcfb..576ff1ab001 100644 --- a/acceptance/bundle/deploy/readplan/test.toml +++ b/acceptance/bundle/deploy/readplan/test.toml @@ -1,3 +1,3 @@ -# Saved plans don't carry the deployment stamp. Applying one plans an update on the next run. -# See dms_no_readplan in acceptance/bundle/test.toml. +# Dumps post-deploy state/plan, which carry the DMS deployment stamp under recording (applied at +# deploy, not saved in the plan). deploy --plan recording is covered by bundle/dms, so opt out here. EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/deploy/wal/header-only-wal/output.txt b/acceptance/bundle/deploy/wal/header-only-wal/output.txt index 807fa926044..da8c1bdc1f3 100644 --- a/acceptance/bundle/deploy/wal/header-only-wal/output.txt +++ b/acceptance/bundle/deploy/wal/header-only-wal/output.txt @@ -28,7 +28,7 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged >>> errcode assert_not_exists.py .databricks/bundle/default/resources.json.wal ->>> errcode cat .databricks/bundle/default/resources.json +>>> errcode print_state.py { "serial": 1, "state_keys": [ diff --git a/acceptance/bundle/deploy/wal/header-only-wal/script b/acceptance/bundle/deploy/wal/header-only-wal/script index 89da7e1a277..557fc7d4f47 100644 --- a/acceptance/bundle/deploy/wal/header-only-wal/script +++ b/acceptance/bundle/deploy/wal/header-only-wal/script @@ -19,4 +19,4 @@ title "Third deploy (must recover and succeed, not blocked by the leftover WAL)" trace errcode $CLI bundle deploy --force-lock trace errcode assert_not_exists.py .databricks/bundle/default/resources.json.wal -trace errcode cat .databricks/bundle/default/resources.json | jq -S '{serial: .serial, state_keys: (.state | keys)}' +trace errcode print_state.py | jq -S '{serial: .serial, state_keys: (.state | keys)}' diff --git a/acceptance/bundle/dms/existing-state/output.txt b/acceptance/bundle/dms/existing-state/output.txt index 4fa2048e8a8..68843cf8624 100644 --- a/acceptance/bundle/dms/existing-state/output.txt +++ b/acceptance/bundle/dms/existing-state/output.txt @@ -8,6 +8,12 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged >>> print_requests.py --dms //api/2.0/bundle --oneline +>>> jq {state_version, features} .databricks/bundle/default/resources.json +{ + "state_version": 2, + "features": null +} + === Turning recording on afterwards is an error: those resources were never recorded, so treating DMS as authoritative would deploy them a second time >>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true [CLI] bundle deploy Error: cannot record deployment history for a bundle that already has deployed resources tracked in [TEST_TMP_DIR]/.databricks/bundle/default/resources.json: only new deployments can be recorded @@ -57,3 +63,27 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} {"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} {"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} + +=== Recording moves the state to the feature version and records the feature, which is what says its resources are the ones the service holds +>>> jq {state_version, features} .databricks/bundle/default/resources.json +{ + "state_version": 3, + "features": { + "record_deployment_history": {} + } +} + +=== So it is accepted even with the deployment gone. The file's state is a no-op while recording, so the service having nothing means the job is created again under a fresh deployment +>>> MSYS_NO_PATHCONV=1 [CLI] workspace delete /Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/resources.deployment.json + +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files... +Created jobs.one +Files: 2 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +>>> print_requests.py --dms //api/2.0/bundle --oneline +{"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-existing-state-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default"}}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.one", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.one", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-existing-state-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"one\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} diff --git a/acceptance/bundle/dms/existing-state/script b/acceptance/bundle/dms/existing-state/script index df29c005a6d..b63bed146cc 100644 --- a/acceptance/bundle/dms/existing-state/script +++ b/acceptance/bundle/dms/existing-state/script @@ -3,6 +3,7 @@ envsubst < databricks.yml.tmpl > databricks.yml title "Deploy without recording: the bundle gets ordinary direct-engine state, unknown to DMS" trace $CLI bundle deploy trace print_requests.py --dms //api/2.0/bundle --oneline +trace jq '{state_version, features}' .databricks/bundle/default/resources.json title "Turning recording on afterwards is an error: those resources were never recorded, so treating DMS as authoritative would deploy them a second time" musterr trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true $CLI bundle deploy @@ -19,3 +20,11 @@ title "Destroy clears the tracked resources, so recording can be enabled afterwa trace $CLI bundle destroy --auto-approve trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true $CLI bundle deploy trace print_requests.py --dms //api/2.0/bundle --oneline + +title "Recording moves the state to the feature version and records the feature, which is what says its resources are the ones the service holds" +trace jq '{state_version, features}' .databricks/bundle/default/resources.json + +title "So it is accepted even with the deployment gone. The file's state is a no-op while recording, so the service having nothing means the job is created again under a fresh deployment" +trace MSYS_NO_PATHCONV=1 $CLI workspace delete "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-existing-state-${UNIQUE_NAME}/default/state/resources.deployment.json" +trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true $CLI bundle deploy +trace print_requests.py --dms //api/2.0/bundle --oneline diff --git a/acceptance/bundle/dms/failed-delete/output.txt b/acceptance/bundle/dms/failed-delete/output.txt index f9b27e6c5dc..46d6d4a9c2e 100644 --- a/acceptance/bundle/dms/failed-delete/output.txt +++ b/acceptance/bundle/dms/failed-delete/output.txt @@ -119,10 +119,13 @@ API message: Fault injected by test. >>> print_state.py { - "state_version": 2, + "state_version": 3, "cli_version": "[CLI_VERSION]", "lineage": "[UUID]", "serial": 1, + "features": { + "record_deployment_history": {} + }, "state": { "resources.jobs.stuck": { "__id__": "[NUMID]", diff --git a/acceptance/bundle/dms/failed-delete/script b/acceptance/bundle/dms/failed-delete/script index 93b7146c08c..fff56460c21 100644 --- a/acceptance/bundle/dms/failed-delete/script +++ b/acceptance/bundle/dms/failed-delete/script @@ -10,3 +10,7 @@ trace fault.py "POST /api/2.2/jobs/delete" 400 0 1 INVALID_PARAMETER_VALUE trace musterr $CLI bundle destroy --auto-approve trace print_requests.py --dms //api/2.0/bundle trace print_state.py + +# print_state.py asks the service for the resources while recording, so drain those reads instead +# of leaving them behind for the harness to diff. +print_requests.py //api/2.0 > /dev/null diff --git a/acceptance/bundle/dms/readplan/databricks.yml.tmpl b/acceptance/bundle/dms/readplan/databricks.yml.tmpl new file mode 100644 index 00000000000..d11621ab1e5 --- /dev/null +++ b/acceptance/bundle/dms/readplan/databricks.yml.tmpl @@ -0,0 +1,19 @@ +bundle: + name: dms-readplan-$UNIQUE_NAME +experimental: + record_deployment_history: true + +resources: + jobs: + foo: + name: foo + pipelines: + bar: + # Unique: the Pipelines API refuses a duplicate name across overlapping runs. + name: bar-$UNIQUE_NAME + catalog: main + schema: default + serverless: true + libraries: + - file: + path: ./transform.py diff --git a/acceptance/bundle/dms/readplan/out.test.toml b/acceptance/bundle/dms/readplan/out.test.toml new file mode 100644 index 00000000000..23c07f70dca --- /dev/null +++ b/acceptance/bundle/dms/readplan/out.test.toml @@ -0,0 +1,3 @@ +Cloud = true +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.DMS = ["true"] diff --git a/acceptance/bundle/dms/readplan/output.txt b/acceptance/bundle/dms/readplan/output.txt new file mode 100644 index 00000000000..371855dbcc0 --- /dev/null +++ b/acceptance/bundle/dms/readplan/output.txt @@ -0,0 +1,22 @@ + +=== Save a plan, then deploy from it: the deploy stamps the id and version onto the job and pipeline the plan predates +>>> [CLI] bundle plan -o json + +>>> [CLI] bundle deploy --plan tmp.plan.json +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-readplan-[UNIQUE_NAME]/default/files... +Created jobs.foo +Created pipelines.bar +Files: 7 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged + +=== Re-plan: no drift, the deployed job and pipeline carry the deployment stamp +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged + +=== Both recorded operations carry deployment_id and version_id (lineage), not just kind +>>> print_requests.py --dms //api/2.0/bundle --sort --oneline +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.foo", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-readplan-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/pipelines.bar", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[UUID]", "sequence_id": "0", "state": "{\"state\":{\"catalog\":\"main\",\"channel\":\"CURRENT\",\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-readplan-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edition\":\"ADVANCED\",\"libraries\":[{\"file\":{\"path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-readplan-[UNIQUE_NAME]/default/files/transform.py\"}}],\"name\":\"bar-[UNIQUE_NAME]\",\"schema\":\"default\",\"serverless\":true}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-readplan-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-readplan-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-readplan-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-readplan-[UNIQUE_NAME]/default"}}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_CREATE"}, {"resource_key": "pipelines.bar", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} diff --git a/acceptance/bundle/dms/readplan/script b/acceptance/bundle/dms/readplan/script new file mode 100644 index 00000000000..40bd110bf00 --- /dev/null +++ b/acceptance/bundle/dms/readplan/script @@ -0,0 +1,13 @@ +envsubst < databricks.yml.tmpl > databricks.yml + +title "Save a plan, then deploy from it: the deploy stamps the id and version onto the job and pipeline the plan predates" +trace $CLI bundle plan -o json > tmp.plan.json +trace $CLI bundle deploy --plan tmp.plan.json + +title "Re-plan: no drift, the deployed job and pipeline carry the deployment stamp" +trace $CLI bundle plan + +title "Both recorded operations carry deployment_id and version_id (lineage), not just kind" +trace print_requests.py --dms //api/2.0/bundle --sort --oneline | contains.py deployment_id version_id + +rm -f tmp.plan.json diff --git a/acceptance/bundle/dms/readplan/transform.py b/acceptance/bundle/dms/readplan/transform.py new file mode 100644 index 00000000000..c07007592a6 --- /dev/null +++ b/acceptance/bundle/dms/readplan/transform.py @@ -0,0 +1 @@ +# Minimal source so the pipeline has a library, which the API requires. diff --git a/acceptance/bundle/dms/record/output.txt b/acceptance/bundle/dms/record/output.txt index 008731c0126..beb068424a4 100644 --- a/acceptance/bundle/dms/record/output.txt +++ b/acceptance/bundle/dms/record/output.txt @@ -69,6 +69,26 @@ Resources: 1 created, 0 changed, 0 deleted, 0 unchanged >>> jq has("deployment_id") .databricks/bundle/default/resources.json false +=== The state file holds no resources at all - the service does. All it carries is the feature, which a CLI that does not know it refuses rather than deploying over the deployment +>>> jq {state_version, features, resources: (.state | keys)} .databricks/bundle/default/resources.json +{ + "state_version": 3, + "features": { + "record_deployment_history": {} + }, + "resources": [] +} + +=== The same header-only state is uploaded to the workspace: that upload is where a redeploy from a clean cache reads the feature back from +>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/dms-record-[UNIQUE_NAME]/default/state/resources.json +{ + "state_version": 3, + "features": { + "record_deployment_history": {} + }, + "resources": [] +} + === Redeploy after deleting the local cache: the deployment ID is resolved from that node, the same deployment is reused, and the version increments (no new CreateDeployment) >>> [CLI] bundle deploy Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-record-[UNIQUE_NAME]/default/files... diff --git a/acceptance/bundle/dms/record/script b/acceptance/bundle/dms/record/script index ad08fdf1a7f..970edaabf9f 100644 --- a/acceptance/bundle/dms/record/script +++ b/acceptance/bundle/dms/record/script @@ -12,6 +12,12 @@ title "The deployment ID is the ID of the workspace node the service registered trace MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-record-${UNIQUE_NAME}/default/state/resources.deployment.json" | jq '{object_type,path}' trace jq 'has("deployment_id")' .databricks/bundle/default/resources.json +title "The state file holds no resources at all - the service does. All it carries is the feature, which a CLI that does not know it refuses rather than deploying over the deployment" +trace jq '{state_version, features, resources: (.state | keys)}' .databricks/bundle/default/resources.json + +title "The same header-only state is uploaded to the workspace: that upload is where a redeploy from a clean cache reads the feature back from" +trace MSYS_NO_PATHCONV=1 $CLI workspace export "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-record-${UNIQUE_NAME}/default/state/resources.json" | jq '{state_version, features, resources: (.state | keys)}' + title "Redeploy after deleting the local cache: the deployment ID is resolved from that node, the same deployment is reused, and the version increments (no new CreateDeployment)" rm -rf .databricks trace $CLI bundle deploy diff --git a/acceptance/bundle/dms/requires-recording/databricks.yml.tmpl b/acceptance/bundle/dms/requires-recording/databricks.yml.tmpl new file mode 100644 index 00000000000..ae8d964d03a --- /dev/null +++ b/acceptance/bundle/dms/requires-recording/databricks.yml.tmpl @@ -0,0 +1,9 @@ +bundle: + name: dms-requires-recording-$UNIQUE_NAME +experimental: + record_deployment_history: true + +resources: + jobs: + one: + name: one diff --git a/acceptance/bundle/dms/requires-recording/out.test.toml b/acceptance/bundle/dms/requires-recording/out.test.toml new file mode 100644 index 00000000000..23c07f70dca --- /dev/null +++ b/acceptance/bundle/dms/requires-recording/out.test.toml @@ -0,0 +1,3 @@ +Cloud = true +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.DMS = ["true"] diff --git a/acceptance/bundle/dms/requires-recording/output.txt b/acceptance/bundle/dms/requires-recording/output.txt new file mode 100644 index 00000000000..a432a70137b --- /dev/null +++ b/acceptance/bundle/dms/requires-recording/output.txt @@ -0,0 +1,98 @@ + +=== Deploy with recording on: the state records the feature +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-requires-recording-[UNIQUE_NAME]/default/files... +Created jobs.one +Files: 5 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +>>> jq .features .databricks/bundle/default/resources.json +{ + "record_deployment_history": {} +} + +=== Turning recording off is refused: the service holds this deployment, so deploying without recording would leave it describing resources that have moved on +>>> update_file.py databricks.yml record_deployment_history: true record_deployment_history: false + +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= [CLI] bundle deploy +Error: unsetting experimental.record_deployment_history is not supported + +This deployment's resources are recorded with the deployment metadata service. Set experimental.record_deployment_history: true to deploy or destroy this bundle + + +=== Destroy is refused for the same reason, which is why the error says to put the setting back +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= [CLI] bundle destroy --auto-approve +Error: unsetting experimental.record_deployment_history is not supported + +This deployment's resources are recorded with the deployment metadata service. Set experimental.record_deployment_history: true to deploy or destroy this bundle + + +=== With recording back on, deploy and destroy work again +>>> update_file.py databricks.yml record_deployment_history: false record_deployment_history: true + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-requires-recording-[UNIQUE_NAME]/default/files... +Files: 2 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 1 unchanged + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.one + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-requires-recording-[UNIQUE_NAME]/default + +Destroy: 1 deleted + +=== Destroy leaves the state file behind, but drops the marker: nothing is recorded any more, so it has nothing left to protect +>>> jq {state_version, features, resources: (.state | keys)} .databricks/bundle/default/resources.json +{ + "state_version": 3, + "features": null, + "resources": [] +} + +=== So recording can be turned off afterwards, rather than the destroyed bundle being stuck with it forever +>>> update_file.py databricks.yml record_deployment_history: true record_deployment_history: false + +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-requires-recording-[UNIQUE_NAME]/default/files... +Created jobs.one +Files: 5 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.one + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-requires-recording-[UNIQUE_NAME]/default + +Destroy: 1 deleted + +=== bind and unbind are refused on a recorded state even with the setting dropped: their writes would be discarded +>>> update_file.py databricks.yml record_deployment_history: false record_deployment_history: true + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-requires-recording-[UNIQUE_NAME]/default/files... +Created jobs.one +Files: 5 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +>>> update_file.py databricks.yml record_deployment_history: true record_deployment_history: false + +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= [CLI] bundle deployment unbind one +Error: unbind is not supported for a bundle that records deployment history + + +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= [CLI] bundle deployment bind one [ONE_ID] --auto-approve +Error: bind is not supported for a bundle that records deployment history + + +>>> update_file.py databricks.yml record_deployment_history: false record_deployment_history: true + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.one + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-requires-recording-[UNIQUE_NAME]/default + +Destroy: 1 deleted diff --git a/acceptance/bundle/dms/requires-recording/script b/acceptance/bundle/dms/requires-recording/script new file mode 100644 index 00000000000..bca90e750d6 --- /dev/null +++ b/acceptance/bundle/dms/requires-recording/script @@ -0,0 +1,35 @@ +envsubst < databricks.yml.tmpl > databricks.yml + +title "Deploy with recording on: the state records the feature" +trace $CLI bundle deploy +trace jq '.features' .databricks/bundle/default/resources.json + +title "Turning recording off is refused: the service holds this deployment, so deploying without recording would leave it describing resources that have moved on" +trace update_file.py databricks.yml "record_deployment_history: true" "record_deployment_history: false" +musterr trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= $CLI bundle deploy + +title "Destroy is refused for the same reason, which is why the error says to put the setting back" +musterr trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= $CLI bundle destroy --auto-approve + +title "With recording back on, deploy and destroy work again" +trace update_file.py databricks.yml "record_deployment_history: false" "record_deployment_history: true" +trace $CLI bundle deploy +trace $CLI bundle destroy --auto-approve + +title "Destroy leaves the state file behind, but drops the marker: nothing is recorded any more, so it has nothing left to protect" +trace jq '{state_version, features, resources: (.state | keys)}' .databricks/bundle/default/resources.json + +title "So recording can be turned off afterwards, rather than the destroyed bundle being stuck with it forever" +trace update_file.py databricks.yml "record_deployment_history: true" "record_deployment_history: false" +trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= $CLI bundle deploy +trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= $CLI bundle destroy --auto-approve + +title "bind and unbind are refused on a recorded state even with the setting dropped: their writes would be discarded" +trace update_file.py databricks.yml "record_deployment_history: false" "record_deployment_history: true" +trace $CLI bundle deploy +job_id=$(read_id.py one) +trace update_file.py databricks.yml "record_deployment_history: true" "record_deployment_history: false" +musterr trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= $CLI bundle deployment unbind one +musterr trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY= $CLI bundle deployment bind one "$job_id" --auto-approve +trace update_file.py databricks.yml "record_deployment_history: false" "record_deployment_history: true" +trace $CLI bundle destroy --auto-approve diff --git a/acceptance/bundle/dms/requires-recording/test.toml b/acceptance/bundle/dms/requires-recording/test.toml new file mode 100644 index 00000000000..cb125e34d6b --- /dev/null +++ b/acceptance/bundle/dms/requires-recording/test.toml @@ -0,0 +1,3 @@ +# This test asserts the CLI's own output, not the calls behind it; bundle/dms/record is where +# the call budget is pinned. +RecordRequests = false diff --git a/acceptance/bundle/dms/stale-deployment/empty.yml.tmpl b/acceptance/bundle/dms/stale-deployment/empty.yml.tmpl new file mode 100644 index 00000000000..4998e0e4b9c --- /dev/null +++ b/acceptance/bundle/dms/stale-deployment/empty.yml.tmpl @@ -0,0 +1,2 @@ +bundle: + name: dms-stale-deployment-$UNIQUE_NAME diff --git a/acceptance/bundle/dms/stale-deployment/out.test.toml b/acceptance/bundle/dms/stale-deployment/out.test.toml new file mode 100644 index 00000000000..d73c45e3119 --- /dev/null +++ b/acceptance/bundle/dms/stale-deployment/out.test.toml @@ -0,0 +1,3 @@ +Cloud = true +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/dms/stale-deployment/output.txt b/acceptance/bundle/dms/stale-deployment/output.txt new file mode 100644 index 00000000000..ba9ba06ac1f --- /dev/null +++ b/acceptance/bundle/dms/stale-deployment/output.txt @@ -0,0 +1,61 @@ + +=== Record a bundle that has no resources yet. The deployment is registered before anything is planned, so it exists even though nothing was deployed and no state file was written +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-stale-deployment-[UNIQUE_NAME]/default/files... +Files: 7 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 0 unchanged + +>>> print_requests.py --dms //api/2.0/bundle --oneline +{"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-stale-deployment-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-deployment-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-deployment-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-deployment-[UNIQUE_NAME]/default"}}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} + +>>> find.py resources.json --expect 0 + +=== Add a job and deploy with recording off. The job is tracked in an ordinary state file the service knows nothing about, while the deployment from the first step is still there +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-stale-deployment-[UNIQUE_NAME]/default/files... +Created jobs.one +Files: 3 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +>>> jq {state_version, features, resources: (.state | keys)} .databricks/bundle/default/resources.json +{ + "state_version": 2, + "features": null, + "resources": [ + "resources.jobs.one" + ] +} + +=== Turning recording back on is refused. The deployment does resolve, but this state was never recorded, so letting the service be authoritative would create the job a second time +>>> DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true [CLI] bundle deploy +Error: cannot record deployment history for a bundle that already has deployed resources tracked in [TEST_TMP_DIR]/.databricks/bundle/default/resources.json: only new deployments can be recorded + +To record this bundle's history, start it over as a new deployment: + 1. remove experimental.record_deployment_history from your bundle configuration + 2. run "databricks bundle destroy" to delete the existing resources + 3. add experimental.record_deployment_history back and deploy again + +To keep the existing resources instead, unset experimental.record_deployment_history + + +=== The job is still tracked by the state file, and nothing was recorded against the stale deployment +>>> jq {state_version, features, resources: (.state | keys)} .databricks/bundle/default/resources.json +{ + "state_version": 2, + "features": null, + "resources": [ + "resources.jobs.one" + ] +} + +>>> print_requests.py --dms //api/2.0/bundle --oneline + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.one + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-stale-deployment-[UNIQUE_NAME]/default + +Destroy: 1 deleted diff --git a/acceptance/bundle/dms/stale-deployment/script b/acceptance/bundle/dms/stale-deployment/script new file mode 100755 index 00000000000..7ba6b19ddb4 --- /dev/null +++ b/acceptance/bundle/dms/stale-deployment/script @@ -0,0 +1,24 @@ +envsubst < empty.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +title "Record a bundle that has no resources yet. The deployment is registered before anything is planned, so it exists even though nothing was deployed and no state file was written" +trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true $CLI bundle deploy +trace print_requests.py --dms //api/2.0/bundle --oneline +trace find.py 'resources.json' --expect 0 + +title "Add a job and deploy with recording off. The job is tracked in an ordinary state file the service knows nothing about, while the deployment from the first step is still there" +envsubst < with-job.yml.tmpl > databricks.yml +trace $CLI bundle deploy +trace jq '{state_version, features, resources: (.state | keys)}' .databricks/bundle/default/resources.json + +title "Turning recording back on is refused. The deployment does resolve, but this state was never recorded, so letting the service be authoritative would create the job a second time" +musterr trace DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true $CLI bundle deploy + +title "The job is still tracked by the state file, and nothing was recorded against the stale deployment" +trace jq '{state_version, features, resources: (.state | keys)}' .databricks/bundle/default/resources.json +trace print_requests.py --dms //api/2.0/bundle --oneline diff --git a/acceptance/bundle/dms/stale-deployment/test.toml b/acceptance/bundle/dms/stale-deployment/test.toml new file mode 100644 index 00000000000..3227096ed19 --- /dev/null +++ b/acceptance/bundle/dms/stale-deployment/test.toml @@ -0,0 +1,3 @@ +# This test turns recording on and off per command, so it opts out of the parent's DMS=true +# and drives DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY inline instead. +EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/dms/stale-deployment/with-job.yml.tmpl b/acceptance/bundle/dms/stale-deployment/with-job.yml.tmpl new file mode 100644 index 00000000000..d06d9594561 --- /dev/null +++ b/acceptance/bundle/dms/stale-deployment/with-job.yml.tmpl @@ -0,0 +1,7 @@ +bundle: + name: dms-stale-deployment-$UNIQUE_NAME + +resources: + jobs: + one: + name: one diff --git a/acceptance/bundle/dms/stale-plan/databricks.yml.tmpl b/acceptance/bundle/dms/stale-plan/databricks.yml.tmpl new file mode 100644 index 00000000000..dea8fbd9a8d --- /dev/null +++ b/acceptance/bundle/dms/stale-plan/databricks.yml.tmpl @@ -0,0 +1,9 @@ +bundle: + name: dms-stale-plan-$UNIQUE_NAME +experimental: + record_deployment_history: true + +resources: + jobs: + foo: + name: foo diff --git a/acceptance/bundle/dms/stale-plan/out.test.toml b/acceptance/bundle/dms/stale-plan/out.test.toml new file mode 100644 index 00000000000..23c07f70dca --- /dev/null +++ b/acceptance/bundle/dms/stale-plan/out.test.toml @@ -0,0 +1,3 @@ +Cloud = true +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.DMS = ["true"] diff --git a/acceptance/bundle/dms/stale-plan/output.txt b/acceptance/bundle/dms/stale-plan/output.txt new file mode 100644 index 00000000000..2ce3b598d1c --- /dev/null +++ b/acceptance/bundle/dms/stale-plan/output.txt @@ -0,0 +1,21 @@ + +=== Save a plan, then deploy it: records version 1 +>>> [CLI] bundle plan -o json + +>>> [CLI] bundle deploy --plan tmp.plan.json +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/files... +Created jobs.foo +Files: 6 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +=== Replay: re-deploying the same plan is rejected - it predates the recorded version +>>> musterr [CLI] bundle deploy --plan tmp.plan.json +Error: this plan predates the deployment's current version 1; run 'bundle plan' again + + +=== Only the first deploy recorded a version; the rejected replay created none +>>> print_requests.py --dms //api/2.0/bundle --oneline +{"method": "POST", "path": "/api/2.0/bundle/deployments", "body": {"display_name": "dms-stale-plan-[UNIQUE_NAME]", "initial_parent_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/state", "target_name": "default", "workspace_info": {"file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/files", "root_path": "/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default"}}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions", "q": {"version_id": "1"}, "body": {"cli_version": "[CLI_VERSION]", "version_type": "VERSION_TYPE_DEPLOY", "operations": [{"resource_key": "jobs.foo", "action_type": "OPERATION_ACTION_TYPE_CREATE"}]}} +{"method": "PATCH", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/operations/jobs.foo", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[NUMID]", "sequence_id": "0", "state": "{\"state\":{\"deployment\":{\"deployment_id\":\"[NUMID]\",\"kind\":\"BUNDLE\",\"metadata_file_path\":\"/Workspace/Users/[USERNAME]/.bundle/dms-stale-plan-[UNIQUE_NAME]/default/state/metadata.json\",\"version_id\":\"1\"},\"edit_mode\":\"UI_LOCKED\",\"format\":\"MULTI_TASK\",\"max_concurrent_runs\":1,\"name\":\"foo\",\"queue\":{\"enabled\":true}}}", "status": "OPERATION_STATUS_SUCCEEDED"}} +{"method": "POST", "path": "/api/2.0/bundle/deployments/[NUMID]/versions/1/complete", "body": {"completion_reason": "VERSION_COMPLETE_SUCCESS"}} diff --git a/acceptance/bundle/dms/stale-plan/script b/acceptance/bundle/dms/stale-plan/script new file mode 100644 index 00000000000..1f9e792f30f --- /dev/null +++ b/acceptance/bundle/dms/stale-plan/script @@ -0,0 +1,13 @@ +envsubst < databricks.yml.tmpl > databricks.yml + +title "Save a plan, then deploy it: records version 1" +trace $CLI bundle plan -o json > tmp.plan.json +trace $CLI bundle deploy --plan tmp.plan.json + +title "Replay: re-deploying the same plan is rejected - it predates the recorded version" +trace musterr $CLI bundle deploy --plan tmp.plan.json + +title "Only the first deploy recorded a version; the rejected replay created none" +trace print_requests.py --dms //api/2.0/bundle --oneline + +rm -f tmp.plan.json diff --git a/acceptance/bundle/dms/state-from-service/databricks.yml.tmpl b/acceptance/bundle/dms/state-from-service/databricks.yml.tmpl new file mode 100644 index 00000000000..7f526744d53 --- /dev/null +++ b/acceptance/bundle/dms/state-from-service/databricks.yml.tmpl @@ -0,0 +1,26 @@ +bundle: + name: dms-state-from-service-$UNIQUE_NAME + +experimental: + record_deployment_history: true + +resources: + jobs: + # source is stamped with the deployment (jobs and pipelines are), and dependent references + # its id, so the pair covers a dependency edge surviving the round trip through the service. + source: + name: source + tags: + source_id: placeholder + dependent: + name: dependent + tags: + upstream: ${resources.jobs.source.id} + + # A secret scope carries no deployment stamp, so it is the case where nothing about the resource + # changes between deploys: after a cache wipe it can only come back as unchanged if the service + # is really the state. + secret_scopes: + scope: + name: scope-$UNIQUE_NAME + backend_type: DATABRICKS diff --git a/acceptance/bundle/dms/state-from-service/out.test.toml b/acceptance/bundle/dms/state-from-service/out.test.toml new file mode 100644 index 00000000000..23c07f70dca --- /dev/null +++ b/acceptance/bundle/dms/state-from-service/out.test.toml @@ -0,0 +1,3 @@ +Cloud = true +EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] +EnvMatrix.DMS = ["true"] diff --git a/acceptance/bundle/dms/state-from-service/output.txt b/acceptance/bundle/dms/state-from-service/output.txt new file mode 100644 index 00000000000..c207b025371 --- /dev/null +++ b/acceptance/bundle/dms/state-from-service/output.txt @@ -0,0 +1,232 @@ + +=== Deploy: the service records all three resources, including the dependency edge +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/files... +Created jobs.dependent +Created jobs.source +Created secret_scopes.scope +Created secret_scopes.scope.permissions +Files: 5 uploaded, 0 deleted +Resources: 4 created, 0 changed, 0 deleted, 0 unchanged + +>>> print_state.py +{ + "state_version": 3, + "cli_version": "[CLI_VERSION]", + "lineage": "[UUID]", + "serial": 1, + "features": { + "record_deployment_history": {} + }, + "state": { + "resources.jobs.dependent": { + "__id__": "[NUMID]", + "state": { + "deployment": { + "deployment_id": "[NUMID]", + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/state/metadata.json", + "version_id": "1" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "max_concurrent_runs": 1, + "name": "dependent", + "queue": { + "enabled": true + }, + "tags": { + "upstream": "[NUMID]" + } + }, + "depends_on": [ + { + "node": "resources.jobs.source", + "label": "${resources.jobs.source.id}" + } + ] + }, + "resources.jobs.source": { + "__id__": "[NUMID]", + "state": { + "deployment": { + "deployment_id": "[NUMID]", + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/state/metadata.json", + "version_id": "1" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "max_concurrent_runs": 1, + "name": "source", + "queue": { + "enabled": true + }, + "tags": { + "source_id": "placeholder" + } + } + }, + "resources.secret_scopes.scope": { + "__id__": "scope-[UNIQUE_NAME]", + "state": { + "scope": "scope-[UNIQUE_NAME]", + "scope_backend_type": "DATABRICKS" + } + }, + "resources.secret_scopes.scope.permissions": { + "__id__": "scope-[UNIQUE_NAME]", + "state": { + "scope_name": "scope-[UNIQUE_NAME]", + "acls": [ + { + "permission": "MANAGE", + "principal": "[USERNAME]" + } + ] + }, + "depends_on": [ + { + "node": "resources.secret_scopes.scope", + "label": "${resources.secret_scopes.scope.name}" + } + ] + } + } +} + +=== Throw away every local trace of the deployment. The state file the deploy left is only a header, so anything the next deploy knows has to come from the service +>>> MSYS_NO_PATHCONV=1 [CLI] workspace export /Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/state/resources.json +{ + "state_version": 3, + "features": { + "record_deployment_history": {} + }, + "resources": [] +} + +=== Redeploy: nothing is created a second time, and nothing even reads as changed - what the service holds matches the config, stamp included +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/files... +Files: 5 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 4 unchanged + +=== And the recovered state has the ids and the dependency edge back, from the service +>>> print_state.py +{ + "state_version": 3, + "cli_version": "[CLI_VERSION]", + "lineage": "[UUID]", + "serial": 1, + "features": { + "record_deployment_history": {} + }, + "state": { + "resources.jobs.dependent": { + "__id__": "[NUMID]", + "state": { + "deployment": { + "deployment_id": "[NUMID]", + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/state/metadata.json", + "version_id": "1" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "max_concurrent_runs": 1, + "name": "dependent", + "queue": { + "enabled": true + }, + "tags": { + "upstream": "[NUMID]" + } + }, + "depends_on": [ + { + "node": "resources.jobs.source", + "label": "${resources.jobs.source.id}" + } + ] + }, + "resources.jobs.source": { + "__id__": "[NUMID]", + "state": { + "deployment": { + "deployment_id": "[NUMID]", + "kind": "BUNDLE", + "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/state/metadata.json", + "version_id": "1" + }, + "edit_mode": "UI_LOCKED", + "format": "MULTI_TASK", + "max_concurrent_runs": 1, + "name": "source", + "queue": { + "enabled": true + }, + "tags": { + "source_id": "placeholder" + } + } + }, + "resources.secret_scopes.scope": { + "__id__": "scope-[UNIQUE_NAME]", + "state": { + "scope": "scope-[UNIQUE_NAME]", + "scope_backend_type": "DATABRICKS" + } + }, + "resources.secret_scopes.scope.permissions": { + "__id__": "scope-[UNIQUE_NAME]", + "state": { + "scope_name": "scope-[UNIQUE_NAME]", + "acls": [ + { + "permission": "MANAGE", + "principal": "[USERNAME]" + } + ] + }, + "depends_on": [ + { + "node": "resources.secret_scopes.scope", + "label": "${resources.secret_scopes.scope.name}" + } + ] + } + } +} + +=== Strip the marker from the local state, keeping its lineage. The resources still come from the service, so the deploy is a no-op - and because it writes nothing, the file keeps the stripped header until some later deploy does write +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/files... +Files: 1 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 4 unchanged + +>>> jq {state_version, features} .databricks/bundle/default/resources.json +{ + "state_version": 3, + "features": null +} + +=== Now delete the remote state file too and wipe the local cache again. No state file exists anywhere, so the service is the only record of the deployment left +>>> MSYS_NO_PATHCONV=1 [CLI] workspace delete /Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/state/resources.json + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default/files... +Files: 5 uploaded, 0 deleted +Resources: 0 created, 0 changed, 0 deleted, 4 unchanged + +=== And no state file comes back: the deploy changed nothing, so it wrote nothing. Under recording the file is optional - the service is the state +>>> find.py resources.json --expect 0 + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.jobs.dependent + delete resources.jobs.source + delete resources.secret_scopes.scope + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-state-from-service-[UNIQUE_NAME]/default + +Destroy: 3 deleted diff --git a/acceptance/bundle/dms/state-from-service/script b/acceptance/bundle/dms/state-from-service/script new file mode 100644 index 00000000000..efd431d0add --- /dev/null +++ b/acceptance/bundle/dms/state-from-service/script @@ -0,0 +1,34 @@ +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve + rm -f out.requests.txt +} +trap cleanup EXIT + +title "Deploy: the service records all three resources, including the dependency edge" +trace $CLI bundle deploy +trace print_state.py + +title "Throw away every local trace of the deployment. The state file the deploy left is only a header, so anything the next deploy knows has to come from the service" +rm -rf .databricks +trace MSYS_NO_PATHCONV=1 $CLI workspace export "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-state-from-service-${UNIQUE_NAME}/default/state/resources.json" | jq '{state_version, features, resources: (.state | keys)}' + +title "Redeploy: nothing is created a second time, and nothing even reads as changed - what the service holds matches the config, stamp included" +trace $CLI bundle deploy + +title "And the recovered state has the ids and the dependency edge back, from the service" +trace print_state.py + +title "Strip the marker from the local state, keeping its lineage. The resources still come from the service, so the deploy is a no-op - and because it writes nothing, the file keeps the stripped header until some later deploy does write" +jq 'del(.features)' .databricks/bundle/default/resources.json > tmp.json && mv tmp.json .databricks/bundle/default/resources.json +trace $CLI bundle deploy +trace jq '{state_version, features}' .databricks/bundle/default/resources.json + +title "Now delete the remote state file too and wipe the local cache again. No state file exists anywhere, so the service is the only record of the deployment left" +trace MSYS_NO_PATHCONV=1 $CLI workspace delete "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-state-from-service-${UNIQUE_NAME}/default/state/resources.json" +rm -rf .databricks +trace $CLI bundle deploy + +title "And no state file comes back: the deploy changed nothing, so it wrote nothing. Under recording the file is optional - the service is the state" +trace find.py 'resources.json' --expect 0 diff --git a/acceptance/bundle/dms/state-from-service/test.toml b/acceptance/bundle/dms/state-from-service/test.toml new file mode 100644 index 00000000000..c4010cbd3f0 --- /dev/null +++ b/acceptance/bundle/dms/state-from-service/test.toml @@ -0,0 +1,2 @@ +# This test asserts what the state holds, not the calls behind it. +RecordRequests = false diff --git a/acceptance/bundle/escaped_refs/output.txt b/acceptance/bundle/escaped_refs/output.txt index 6e12dbfd8fc..342535d0037 100644 --- a/acceptance/bundle/escaped_refs/output.txt +++ b/acceptance/bundle/escaped_refs/output.txt @@ -15,7 +15,7 @@ Created jobs.example_ingestion_job Files: 6 uploaded, 0 deleted Resources: 1 created, 0 changed, 0 deleted, 0 unchanged ->>> print_requests.py //jobs +>>> print_requests.py //jobs --nostamp { "method": "POST", "path": "/api/2.2/jobs/create", diff --git a/acceptance/bundle/escaped_refs/script b/acceptance/bundle/escaped_refs/script index 34640492656..5ff97b09bcb 100644 --- a/acceptance/bundle/escaped_refs/script +++ b/acceptance/bundle/escaped_refs/script @@ -18,4 +18,6 @@ trace $CLI bundle plan -o json > out.plan.$DATABRICKS_BUNDLE_ENGINE.json # that the payload below is identical either way. title "Deploy: payload sent to the jobs API\n" $CLI bundle deploy $(readplanarg out.plan.$DATABRICKS_BUNDLE_ENGINE.json) -trace print_requests.py //jobs +# --nostamp: under deployment-history recording the job payload carries the DMS deployment +# stamp; drop it so the assertion is the same whether or not recording is on. +trace print_requests.py //jobs --nostamp diff --git a/acceptance/bundle/invariant/delete_idempotent/out.test.toml b/acceptance/bundle/invariant/delete_idempotent/out.test.toml index 59018deb1ce..8439713320e 100644 --- a/acceptance/bundle/invariant/delete_idempotent/out.test.toml +++ b/acceptance/bundle/invariant/delete_idempotent/out.test.toml @@ -1,6 +1,6 @@ Cloud = true EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] -EnvMatrix.DMS = [""] +EnvMatrix.DMS = ["", "true"] EnvMatrix.INPUT_CONFIG = [ "alert.yml.tmpl", "app.yml.tmpl", diff --git a/acceptance/bundle/invariant/delete_idempotent/test.toml b/acceptance/bundle/invariant/delete_idempotent/test.toml index f876726848d..c9ff20b59b4 100644 --- a/acceptance/bundle/invariant/delete_idempotent/test.toml +++ b/acceptance/bundle/invariant/delete_idempotent/test.toml @@ -1,9 +1,10 @@ -# Recording needs a bundle from the start. This test rewinds state and wipes the -# deployment record path, so recording refuses it. -EnvMatrix.DMS = [""] - EnvMatrix.READPLAN = ["", "1"] +# A 1000-task job serializes to ~110 KB, over the 64 KB per-operation state limit the +# deployment metadata service accepts, so recording it fails the deploy. Raising the limit +# or splitting the state is a service-side decision. +EnvMatrixExclude.dms_state_too_large = ["DMS=true", "INPUT_CONFIG=job_pydabs_1000_tasks.yml.tmpl"] + # Snapshot of pre-delete state used to re-run the delete on state that still # references the (now-gone) resources; may linger if the test fails mid-run. Ignore = [".databricks.backup"] diff --git a/acceptance/bundle/invariant/destroy_idempotent/out.test.toml b/acceptance/bundle/invariant/destroy_idempotent/out.test.toml index 59018deb1ce..8439713320e 100644 --- a/acceptance/bundle/invariant/destroy_idempotent/out.test.toml +++ b/acceptance/bundle/invariant/destroy_idempotent/out.test.toml @@ -1,6 +1,6 @@ Cloud = true EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"] -EnvMatrix.DMS = [""] +EnvMatrix.DMS = ["", "true"] EnvMatrix.INPUT_CONFIG = [ "alert.yml.tmpl", "app.yml.tmpl", diff --git a/acceptance/bundle/invariant/destroy_idempotent/test.toml b/acceptance/bundle/invariant/destroy_idempotent/test.toml index d86b5366f20..d44e1cf47eb 100644 --- a/acceptance/bundle/invariant/destroy_idempotent/test.toml +++ b/acceptance/bundle/invariant/destroy_idempotent/test.toml @@ -1,9 +1,10 @@ -# Recording needs a bundle from the start. This test rewinds state and wipes the -# deployment record path, so recording refuses it. -EnvMatrix.DMS = [""] - EnvMatrix.READPLAN = ["", "1"] +# A 1000-task job serializes to ~110 KB, over the 64 KB per-operation state limit the +# deployment metadata service accepts, so recording it fails the deploy. Raising the limit +# or splitting the state is a service-side decision. +EnvMatrixExclude.dms_state_too_large = ["DMS=true", "INPUT_CONFIG=job_pydabs_1000_tasks.yml.tmpl"] + # Snapshot of pre-destroy state used to re-run destroy on state that still # references the (now-gone) resources; may linger if the test fails mid-run. Ignore = [".databricks.backup"] diff --git a/acceptance/bundle/resource_deps/escaped_ref/output.txt b/acceptance/bundle/resource_deps/escaped_ref/output.txt index 09857524d69..3089e5c6995 100644 --- a/acceptance/bundle/resource_deps/escaped_ref/output.txt +++ b/acceptance/bundle/resource_deps/escaped_ref/output.txt @@ -18,7 +18,7 @@ Created jobs.foo Files: 6 uploaded, 0 deleted Resources: 2 created, 0 changed, 0 deleted, 0 unchanged ->>> print_requests.py //jobs --sort +>>> print_requests.py //jobs --sort --nostamp { "method": "POST", "path": "/api/2.2/jobs/create", diff --git a/acceptance/bundle/resource_deps/escaped_ref/script b/acceptance/bundle/resource_deps/escaped_ref/script index 413394d8b7d..c53b9b85671 100644 --- a/acceptance/bundle/resource_deps/escaped_ref/script +++ b/acceptance/bundle/resource_deps/escaped_ref/script @@ -12,4 +12,6 @@ trace jq '.plan | map_values(.depends_on)' plan.json # that the payload below is identical either way. title "Payload: escaped stays literal, real is resolved\n" $CLI bundle deploy $(readplanarg plan.json) -trace print_requests.py //jobs --sort +# --nostamp: under deployment-history recording the job payload carries the DMS deployment +# stamp; drop it so the assertion is the same whether or not recording is on. +trace print_requests.py //jobs --sort --nostamp diff --git a/acceptance/bundle/resources/jobs/big_id/test.toml b/acceptance/bundle/resources/jobs/big_id/test.toml index b15d8f9e757..8957ed9ce06 100644 --- a/acceptance/bundle/resources/jobs/big_id/test.toml +++ b/acceptance/bundle/resources/jobs/big_id/test.toml @@ -3,8 +3,8 @@ EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ['direct'] EnvMatrix.READPLAN = ["", "1"] -# `bundle deploy --plan` applies pre-saved state; the stamp isn't written into the plan, -# so the next plan reports it as a change. Recording skipped until stamp reaches saved plans. +# Dumps raw state/plan, which carry the DMS deployment stamp under recording (applied at deploy, +# not saved in the plan). deploy --plan recording is covered by bundle/dms, so opt out here. EnvMatrix.DMS = [""] [[Repls]] diff --git a/acceptance/bundle/resources/jobs/delete_task/test.toml b/acceptance/bundle/resources/jobs/delete_task/test.toml index 46a4d91b848..04d6ea1e5b3 100644 --- a/acceptance/bundle/resources/jobs/delete_task/test.toml +++ b/acceptance/bundle/resources/jobs/delete_task/test.toml @@ -1,5 +1,5 @@ RecordRequests = false EnvMatrix.READPLAN = ["", "1"] -# `bundle deploy --plan` applies pre-saved state; the stamp isn't written into the plan, -# so the next plan reports it as a change. Recording skipped until stamp reaches saved plans. +# Dumps raw state/plan, which carry the DMS deployment stamp under recording (applied at deploy, +# not saved in the plan). deploy --plan recording is covered by bundle/dms, so opt out here. EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/resources/jobs/remote_delete/deploy/test.toml b/acceptance/bundle/resources/jobs/remote_delete/deploy/test.toml index 46a4d91b848..04d6ea1e5b3 100644 --- a/acceptance/bundle/resources/jobs/remote_delete/deploy/test.toml +++ b/acceptance/bundle/resources/jobs/remote_delete/deploy/test.toml @@ -1,5 +1,5 @@ RecordRequests = false EnvMatrix.READPLAN = ["", "1"] -# `bundle deploy --plan` applies pre-saved state; the stamp isn't written into the plan, -# so the next plan reports it as a change. Recording skipped until stamp reaches saved plans. +# Dumps raw state/plan, which carry the DMS deployment stamp under recording (applied at deploy, +# not saved in the plan). deploy --plan recording is covered by bundle/dms, so opt out here. EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/resources/jobs/update/test.toml b/acceptance/bundle/resources/jobs/update/test.toml index 20c8733641d..a8a43da097a 100644 --- a/acceptance/bundle/resources/jobs/update/test.toml +++ b/acceptance/bundle/resources/jobs/update/test.toml @@ -1,4 +1,4 @@ EnvMatrix.READPLAN = ["", "1"] -# `bundle deploy --plan` applies pre-saved state; the stamp isn't written into the plan, -# so the next plan reports it as a change. Recording skipped until stamp reaches saved plans. +# Dumps raw state/plan, which carry the DMS deployment stamp under recording (applied at deploy, +# not saved in the plan). deploy --plan recording is covered by bundle/dms, so opt out here. EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/resources/permissions/jobs/destroy_without_mgmtperms/test.toml b/acceptance/bundle/resources/permissions/jobs/destroy_without_mgmtperms/test.toml index 921045dfd09..4ca61be3f36 100644 --- a/acceptance/bundle/resources/permissions/jobs/destroy_without_mgmtperms/test.toml +++ b/acceptance/bundle/resources/permissions/jobs/destroy_without_mgmtperms/test.toml @@ -1,4 +1,3 @@ -# Recording needs a bundle it has seen from the start. This test seeds a state file, -# so recording refuses it. TODO(DMS): drop this once existing state can be -# handed over to the service (see the TODO in dstate.Open). +# Deploy runs as the service principal, later commands as the user - who resolves no deployment of +# its own, so bundle summary reads the job id back null. Recording needs state staged for the reader. EnvMatrix.DMS = [""] diff --git a/acceptance/bundle/resources/postgres_branches/purge_on_delete_transitions/script b/acceptance/bundle/resources/postgres_branches/purge_on_delete_transitions/script index 919fe6e4f4f..f743dc45177 100644 --- a/acceptance/bundle/resources/postgres_branches/purge_on_delete_transitions/script +++ b/acceptance/bundle/resources/postgres_branches/purge_on_delete_transitions/script @@ -16,7 +16,7 @@ trap cleanup EXIT # resources.json. The field is omitted from JSON when unset (omitempty + # ForceSendFields tracking), so an absent value falls back to "(unset)". get_purge() { - gron.py < .databricks/bundle/default/resources.json | grep 'postgres_branches.branch.state.purge_on_delete' || echo '(unset)' + print_state.py | gron.py | grep 'postgres_branches.branch.state.purge_on_delete' || echo '(unset)' } title "Step 1: deploy with purge_on_delete unset" diff --git a/acceptance/bundle/resources/postgres_projects/purge_on_delete_transitions/script b/acceptance/bundle/resources/postgres_projects/purge_on_delete_transitions/script index 1bdc077e6e1..958d4e2d980 100644 --- a/acceptance/bundle/resources/postgres_projects/purge_on_delete_transitions/script +++ b/acceptance/bundle/resources/postgres_projects/purge_on_delete_transitions/script @@ -12,7 +12,7 @@ trap cleanup EXIT # resources.json. The field is omitted from JSON when unset (omitempty + # ForceSendFields tracking), so an absent value falls back to "(unset)". get_purge() { - gron.py < .databricks/bundle/default/resources.json | grep purge_on_delete || echo '(unset)' + print_state.py | gron.py | grep purge_on_delete || echo '(unset)' } title "Step 1: deploy with purge_on_delete unset" diff --git a/acceptance/bundle/resources/secrets/update-value/output.txt b/acceptance/bundle/resources/secrets/update-value/output.txt index 9e94921e2f3..3a6267216c8 100644 --- a/acceptance/bundle/resources/secrets/update-value/output.txt +++ b/acceptance/bundle/resources/secrets/update-value/output.txt @@ -117,20 +117,14 @@ secrets secret1 catalog_name='main' schema_name='default' name='test_secret' com === Verify state does not contain actual secret value >>> print_state.py { - "state_version": 2, - "cli_version": "[CLI_VERSION]", - "lineage": "[UUID]", - "serial": 2, - "state": { "resources.secrets.secret1": { - "__id__": "main.default.test_secret", - "state": { - "catalog_name": "main", - "comment": "Test secret", - "name": "test_secret", - "schema_name": "default", - "value": "" - } + "__id__": "main.default.test_secret", + "state": { + "catalog_name": "main", + "comment": "Test secret", + "name": "test_secret", + "schema_name": "default", + "value": "" + } } - } } diff --git a/acceptance/bundle/resources/secrets/update-value/script b/acceptance/bundle/resources/secrets/update-value/script index a5c1596e29c..977c2f22378 100755 --- a/acceptance/bundle/resources/secrets/update-value/script +++ b/acceptance/bundle/resources/secrets/update-value/script @@ -11,6 +11,6 @@ trace print_requests.py //unity-catalog read_state.py secrets secret1 catalog_name schema_name name comment value title "Verify state does not contain actual secret value" -{ trace print_state.py | contains.py "!initial-secret-value" "!updated-secret-value"; } || true +{ trace print_state.py | jq .state | contains.py "!initial-secret-value" "!updated-secret-value"; } || true rm -f out.requests.txt diff --git a/acceptance/bundle/state/feature_flags/output.txt b/acceptance/bundle/state/feature_flags/output.txt index df55e7c6cee..90326dfe2cd 100644 --- a/acceptance/bundle/state/feature_flags/output.txt +++ b/acceptance/bundle/state/feature_flags/output.txt @@ -1,5 +1,5 @@ -=== a version-3 state recording a feature is rejected (this CLI records no features yet) +=== a version-3 state recording a feature this CLI does not write is rejected >>> errcode [CLI] bundle plan Error: migrating state [TEST_TMP_DIR]/.databricks/bundle/default/resources.json: the deployment state requires features this CLI does not support: future_feature; upgrade to the latest CLI version and see https://docs.databricks.com/aws/en/dev-tools/bundles/state-features#state-features for more information diff --git a/acceptance/bundle/state/feature_flags/script b/acceptance/bundle/state/feature_flags/script index e70b562f2ff..0212a4102be 100644 --- a/acceptance/bundle/state/feature_flags/script +++ b/acceptance/bundle/state/feature_flags/script @@ -1,6 +1,6 @@ mkdir -p .databricks/bundle/default -title "a version-3 state recording a feature is rejected (this CLI records no features yet)" +title "a version-3 state recording a feature this CLI does not write is rejected" cp resources.with_feature.json .databricks/bundle/default/resources.json trace errcode $CLI bundle plan 2>&1 | contains.py "requires features this CLI does not support: future_feature" "upgrade to the latest CLI version" "https://docs.databricks.com/aws/en/dev-tools/bundles/state-features#state-features" diff --git a/acceptance/bundle/test.toml b/acceptance/bundle/test.toml index d8582b4b61d..2db98f4b429 100644 --- a/acceptance/bundle/test.toml +++ b/acceptance/bundle/test.toml @@ -10,10 +10,6 @@ EnvMatrix.DMS = ["", "true"] # DMS recording is only supported by the direct engine; it is a no-op on terraform. EnvMatrixExclude.dms_needs_direct = ["DMS=true", "DATABRICKS_BUNDLE_ENGINE=terraform"] -# Saved plans don't carry the deployment stamp. A first plan writes the deployment record, -# so `deploy --plan` creates resources without it and reports drift on the next plan. -EnvMatrixExclude.dms_no_readplan = ["DMS=true", "READPLAN=1"] - # The DMS run asserts the same golden files as the engine runs. EnvRepl.DMS = false diff --git a/bundle/deployplan/plan.go b/bundle/deployplan/plan.go index c9f5bec3516..dac0efef687 100644 --- a/bundle/deployplan/plan.go +++ b/bundle/deployplan/plan.go @@ -16,11 +16,12 @@ import ( const currentPlanVersion = 2 type Plan struct { - PlanVersion int `json:"plan_version,omitempty"` - CLIVersion string `json:"cli_version,omitempty"` - Lineage string `json:"lineage,omitempty"` - Serial int `json:"serial,omitempty"` - Plan map[string]*PlanEntry `json:"plan,omitzero"` + PlanVersion int `json:"plan_version,omitempty"` + CLIVersion string `json:"cli_version,omitempty"` + Lineage string `json:"lineage,omitempty"` + Serial int `json:"serial,omitempty"` + + Plan map[string]*PlanEntry `json:"plan,omitzero"` // NotSelected is the number of resources removed by FilterToSelected via the // --select flag. Serialized so the summary survives a deploy from a plan file diff --git a/bundle/direct/bind.go b/bundle/direct/bind.go index 798096f6c22..0a3ad2592f5 100644 --- a/bundle/direct/bind.go +++ b/bundle/direct/bind.go @@ -2,6 +2,7 @@ package direct import ( "context" + "errors" "fmt" "os" "strings" @@ -65,10 +66,17 @@ func (b *DeploymentBundle) Bind(ctx context.Context, client *databricks.Workspac // phases.Bind and phases.Unbind refuse to run at all when recording is enabled. var checkStateDB dstate.DeploymentState if err := checkStateDB.Open(ctx, statePath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, ""); err == nil { + // The setting can be dropped from the config while the state still records the feature, + // which the phases-level refusal misses. The writes below would then be discarded and + // the command would report a bind that did not happen. + recorded := checkStateDB.RequiresDeploymentHistory() existingID := checkStateDB.GetResourceID(resourceKey) if _, err := checkStateDB.Finalize(ctx); err != nil { log.Warnf(ctx, "failed to finalize state: %v", err) } + if recorded { + return nil, errors.New("bind is not supported for a bundle that records deployment history") + } if existingID != "" { return nil, ErrResourceAlreadyBound{ ResourceKey: resourceKey, @@ -222,6 +230,12 @@ func (b *DeploymentBundle) Unbind(ctx context.Context, statePath, resourceKey st return err } + // See the note in Bind: the state's marker outlives the config setting the phases-level + // refusal reads, and the deletes below would be discarded. + if b.StateDB.RequiresDeploymentHistory() { + return errors.New("unbind is not supported for a bundle that records deployment history") + } + // Delete the main resource err = b.StateDB.DeleteState(ctx, resourceKey, false) if err != nil { diff --git a/bundle/direct/bundle_plan.go b/bundle/direct/bundle_plan.go index 739ca3bd8e7..ff5e99239ad 100644 --- a/bundle/direct/bundle_plan.go +++ b/bundle/direct/bundle_plan.go @@ -25,6 +25,7 @@ import ( "github.com/databricks/cli/libs/structs/structvar" "github.com/databricks/databricks-sdk-go" "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/service/jobs" ) var errDelayed = errors.New("must be resolved after apply") @@ -58,9 +59,10 @@ func ValidatePlanAgainstState(stateDB *dstate.DeploymentState, plan *deployplan. return nil } -// InitForApply initializes the DeploymentBundle for applying a pre-computed plan. +// InitForApply initializes the DeploymentBundle for applying a pre-computed plan. A non-empty +// deploymentID/versionID (recording) is stamped onto each job/pipeline here, not into the saved plan. // StateDB must already be open for write before calling this function. -func (b *DeploymentBundle) InitForApply(ctx context.Context, client *databricks.WorkspaceClient, plan *deployplan.Plan) error { +func (b *DeploymentBundle) InitForApply(ctx context.Context, client *databricks.WorkspaceClient, plan *deployplan.Plan, deploymentID, versionID string) error { b.StateDB.AssertOpenedForWrite() err := b.init(client) @@ -91,6 +93,25 @@ func (b *DeploymentBundle) InitForApply(ctx context.Context, client *databricks. if err != nil { return fmt.Errorf("loading plan entry %s: %w", resourceKey, err) } + // Stamp the DMS id and version here (see InitForApply doc), not into the saved plan. + if deploymentID != "" { + var stamped bool + switch v := sv.Value.(type) { + case *jobs.JobSettings: + v.Deployment.DeploymentId = deploymentID + v.Deployment.VersionId = versionID + stamped = true + case *dresources.PipelineState: + v.Deployment.DeploymentId = deploymentID + v.Deployment.VersionId = versionID + stamped = true + } + if stamped { + if err := sv.SyncToJSON(entry.NewState); err != nil { + return fmt.Errorf("%s: stamping deployment into loaded plan: %w", resourceKey, err) + } + } + } b.StateCache.Store(resourceKey, sv) } diff --git a/bundle/direct/dstate/dms.go b/bundle/direct/dstate/dms.go index 96eaf75513f..c0d4ffb2a7f 100644 --- a/bundle/direct/dstate/dms.go +++ b/bundle/direct/dstate/dms.go @@ -16,9 +16,10 @@ type RecordedState struct { DependsOn []deployplan.DependsOnEntry `json:"depends_on,omitempty"` } -// applyDMSState replaces the file-derived resource state with what DMS recorded. Recording is -// only enabled for net-new deployments, so DMS owns the resource set outright: an empty set -// means a successful deploy of nothing, not missing data. The caller holds db.mu. +// applyDMSState replaces the file-derived resource state with what DMS recorded. DMS owns the +// resource set outright, so this runs on every recorded open and the file's copy is never read +// back as the truth: an empty set means the service tracks nothing, whether because the deploy +// created nothing or because the deployment is gone. The caller holds db.mu. func (db *DeploymentState) applyDMSState(recorded []dms.Resource) error { // Built first and assigned together, so a malformed envelope leaves the state as it was // rather than half replaced. diff --git a/bundle/direct/dstate/migrate.go b/bundle/direct/dstate/migrate.go index e4d21a7054a..422ff0e60a1 100644 --- a/bundle/direct/dstate/migrate.go +++ b/bundle/direct/dstate/migrate.go @@ -15,19 +15,25 @@ import ( // migrateState runs all necessary migrations on the database. // It is called after loading state from disk. func migrateState(db *Database) error { - // featureStateVersion states carry a feature list this CLI does not yet write or - // understand (see the featureStateVersion doc comment). A featureStateVersion + // featureStateVersion states carry a feature list this CLI may not recognize + // (see the featureStateVersion doc comment). A featureStateVersion // state with no features is equivalent to currentStateVersion, so accept it and // return without running the migrations below, leaving the on-disk version at - // featureStateVersion rather than flipping it down. One that records any feature - // depends on capabilities this CLI lacks, so refuse it and tell the user to upgrade. + // featureStateVersion rather than flipping it down. One that records an + // unrecognized feature depends on capabilities this CLI lacks, so refuse it and + // tell the user to upgrade. if db.StateVersion == featureStateVersion { if len(db.Features) == 0 { return nil } features := make([]string, 0, len(db.Features)) for name := range db.Features { - features = append(features, name) + if _, ok := recognizedFeatures[name]; !ok { + features = append(features, name) + } + } + if len(features) == 0 { + return nil } slices.Sort(features) return fmt.Errorf("the deployment state requires features this CLI does not support: %s; upgrade to the latest CLI version and see %s for more information", strings.Join(features, ", "), featuresDocURL) diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go index 10a57dafebc..2263a6d02e0 100644 --- a/bundle/direct/dstate/state.go +++ b/bundle/direct/dstate/state.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io/fs" + "maps" "os" "path/filepath" "sync" @@ -29,12 +30,13 @@ const ( maxWalEntrySize = 10 * 1024 * 1024 walSuffix = ".wal" - // featureStateVersion is the schema version a future CLI will write once it - // records deployment state "feature flags" (see Header.Features). This CLI does - // not write it and records no features; it exists now only so this CLI reads - // such states correctly (see migrateState): + // featureStateVersion is the schema version a CLI writes once it + // records deployment state "feature flags" (see Header.Features). This CLI writes + // it for a state that records a feature, and reads such states as follows + // (see migrateState): // - featureStateVersion with no features -> accept and leave the version as-is - // - featureStateVersion with any feature -> refuse, tell the user to upgrade + // - featureStateVersion with recognized features -> accept and leave the version as-is + // - featureStateVersion with any other feature -> refuse, tell the user to upgrade // // A featureStateVersion state with no features is equivalent to // currentStateVersion, but we deliberately do not flip the on-disk version down @@ -48,11 +50,27 @@ const ( // supportedStateVersion is the highest schema version this CLI can read. It is // normally equal to currentStateVersion — the version this CLI reads is the // version it writes — and exceeds it only during a two-phase version bump like - // the current feature-flag scaffolding, where this CLI reads (but does not - // write) featureStateVersion. A state newer than this is rejected as too new. + // the current feature-flag scaffolding, where this CLI writes featureStateVersion + // only for a state that records a feature. A state newer than this is rejected as too new. supportedStateVersion = featureStateVersion ) +// featureRecordDeploymentHistory marks a state whose resources are also recorded with the +// deployment metadata service. Both stores are kept in step, so the marker is what tells a +// reader the two already agree. A CLI that does not know the name refuses the state rather +// than deploying over a deployment it would leave the service out of step with. +// +// The marker is sticky: once a deployment is recorded, the service holds resources that a +// CLI which is not recording must not touch. So turning recording off does not clear it, and +// deploying such a state without recording is refused (see RequiresDeploymentHistory). +const featureRecordDeploymentHistory = "record_deployment_history" + +// recognizedFeatures are the state features this CLI understands. A state recording anything +// outside this set is refused (see migrateState). +var recognizedFeatures = map[string]struct{}{ + featureRecordDeploymentHistory: {}, +} + // featuresDocURL is the single documentation page describing deployment state // feature flags. It is shown when a state records a feature this CLI does not // support; it is a fixed link for all features. The #state-features anchor points @@ -89,10 +107,9 @@ type Header struct { Serial int `json:"serial"` // Features maps each feature flag this state depends on to a (currently empty) - // value. This CLI writes no features; it only reads the field to detect a state - // that depends on features it lacks and refuse it (see migrateState). It is a - // map so a future CLI can attach per-feature data without reshaping the state. - // Empty/omitted for states that use no features. + // value. It is read to detect a state that depends on features this CLI lacks and + // refuse it (see migrateState). It is a map so a future CLI can attach per-feature + // data without reshaping the state. Empty/omitted for states that use no features. Features map[string]struct{} `json:"features,omitempty"` } @@ -285,6 +302,18 @@ func (db *DeploymentState) StateCLIVersion() string { return db.Data.CLIVersion } +// RequiresDeploymentHistory reports whether the state depends on the deployment metadata +// service recording it. Deploying such a state without recording would leave the service +// holding a deployment that no longer matches, so the caller refuses instead. +func (db *DeploymentState) RequiresDeploymentHistory() bool { + db.AssertOpenedForReadOrWrite() + db.mu.Lock() + defer db.mu.Unlock() + + _, ok := db.Data.Features[featureRecordDeploymentHistory] + return ok +} + // GetOrInitLineage returns the deployment lineage, generating and storing a new // one if the state does not have one yet. It is the single place the lineage is // initialized, shared so the direct deployment engine (when it writes state, via @@ -369,6 +398,12 @@ func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, withRe db.stateIDs[key] = entry.ID } + // Read off the committed file, before the WAL replay below changes it. Resources the WAL adds + // come from a deploy that was already recording, so they are not resources the service never + // saw; see the guard in the dmsClient block below. + _, fileRecorded := db.Data.Features[featureRecordDeploymentHistory] + fileHasResources := len(db.Data.State) > 0 + walPath := db.Path + walSuffix _, err = os.Stat(walPath) switch { @@ -394,7 +429,13 @@ func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, withRe // Only empty bundles can be recorded. Once DMS owns the deployment, pre-existing // resources it never saw would be created again. TODO: support migration via state // upgrade with feature flag and per-resource tombstones. - if dmsDeploymentID == "" && len(db.Data.State) > 0 { + // + // The resources have to be ones the committed file tracked without recording them, and + // still tracks after recovery: a WAL that deletes them leaves nothing to clash with. The + // check cannot key off the deployment resolving instead - a deployment can exist while + // the file tracks resources it never recorded (an empty first recorded deploy creates one + // and writes no state), and applyDMSState below would then silently drop them. + if !fileRecorded && fileHasResources && len(db.Data.State) > 0 { // The remedy is ordered deliberately: this error also blocks destroy, so the // setting has to come out first or there is no way to tear the bundle down. return fmt.Errorf(`cannot record deployment history for a bundle that already has deployed resources tracked in %s: only new deployments can be recorded @@ -406,6 +447,16 @@ To record this bundle's history, start it over as a new deployment: To keep the existing resources instead, unset experimental.record_deployment_history`, path) } + + // Mark the state as depending on the service: a CLI that does not recognize the feature + // refuses it (see migrateState) instead of deploying over the deployment and leaving the + // service behind. unlockedSave then writes the header alone. + db.Data.StateVersion = featureStateVersion + if db.Data.Features == nil { + db.Data.Features = make(map[string]struct{}, 1) + } + db.Data.Features[featureRecordDeploymentHistory] = struct{}{} + if dmsDeploymentID != "" { recorded, err := dmsClient.ListResources(ctx, dmsDeploymentID) if err != nil { @@ -521,6 +572,7 @@ func (db *DeploymentState) mergeWalIntoState(ctx context.Context) (bool, error) if header.Serial > expectedSerial { return false, fmt.Errorf("WAL serial (%d) is ahead of expected (%d), state may be corrupted", header.Serial, expectedSerial) } + newSerial = header.Serial newCLIVersion = header.CLIVersion } else { @@ -720,7 +772,7 @@ func (db *DeploymentState) ExportState(ctx context.Context) resourcestate.Export // the WAL. A torn write would therefore leave a state file that Open rejects // next to an intact WAL it never reads. func (db *DeploymentState) unlockedSave() error { - data, err := json.MarshalIndent(db.Data, "", " ") + data, err := json.MarshalIndent(db.dataToPersist(), "", " ") if err != nil { return err } @@ -756,6 +808,28 @@ func (db *DeploymentState) unlockedSave() error { return nil } +// dataToPersist returns what the state file should hold: db.Data itself, unless the deployment +// records its history, in which case it is the header alone. The service holds those resources and +// is where they are read back from, so writing them here too would be a second, never-read copy. +// Returns a copy, leaving the in-memory state - what the rest of the deploy works from - untouched. +func (db *DeploymentState) dataToPersist() Database { + data := db.Data + if _, recorded := data.Features[featureRecordDeploymentHistory]; !recorded { + return data + } + + // A destroy leaves nothing recorded, so the marker has nothing left to protect. Keeping it + // would refuse every later deploy that does not record, with no way back. + if len(data.State) == 0 { + data.Features = maps.Clone(data.Features) + delete(data.Features, featureRecordDeploymentHistory) + return data + } + + data.State = map[string]ResourceEntry{} + return data +} + func appendJSONLine(file *os.File, obj any) error { data, err := json.Marshal(obj) if err != nil { diff --git a/bundle/direct/dstate/state_test.go b/bundle/direct/dstate/state_test.go index edaba875974..c54370d5536 100644 --- a/bundle/direct/dstate/state_test.go +++ b/bundle/direct/dstate/state_test.go @@ -185,7 +185,7 @@ func TestHeaderOnlyWALRecoveryDoesNotAdvanceSerial(t *testing.T) { // TestEmptyFeatureStateAcceptedWithoutFlippingVersion pins the special case that a // featureStateVersion state with no features is accepted as-is — the on-disk version // is left at featureStateVersion, not flipped down to currentStateVersion — and that -// a featureStateVersion state recording any feature is refused. This is scaffolding +// a featureStateVersion state recording an unrecognized feature is refused. This is scaffolding // for the deferred version bump, special-cased to featureStateVersion only (see the // featureStateVersion doc comment). // @@ -201,7 +201,7 @@ func TestEmptyFeatureStateAcceptedWithoutFlippingVersion(t *testing.T) { require.NoError(t, migrateState(empty)) assert.Equal(t, featureStateVersion, empty.StateVersion, "v3 + no features keeps its on-disk version, not flipped to v2") - // v3 that records a feature is refused: this CLI does not understand features. + // v3 that records an unrecognized feature is refused. withFeature := &Database{Header: Header{ StateVersion: featureStateVersion, Features: map[string]struct{}{"future_feature": {}}, @@ -214,6 +214,68 @@ func TestEmptyFeatureStateAcceptedWithoutFlippingVersion(t *testing.T) { assert.Contains(t, err.Error(), featuresDocURL) } +// TestSupportedFeatureAcceptedUnknownOneNamed covers the features this CLI does write: +// a state recording only those loads, and an unsupported feature alongside one of them +// is still refused — naming only the feature the user has to upgrade for. +func TestSupportedFeatureAcceptedUnknownOneNamed(t *testing.T) { + supported := &Database{Header: Header{ + StateVersion: featureStateVersion, + Features: map[string]struct{}{featureRecordDeploymentHistory: {}}, + }} + require.NoError(t, migrateState(supported)) + assert.Equal(t, featureStateVersion, supported.StateVersion) + assert.Contains(t, supported.Features, featureRecordDeploymentHistory, "a supported feature is left on the state, not stripped") + + mixed := &Database{Header: Header{ + StateVersion: featureStateVersion, + Features: map[string]struct{}{featureRecordDeploymentHistory: {}, "future_feature": {}}, + }} + err := migrateState(mixed) + require.Error(t, err) + assert.Contains(t, err.Error(), "future_feature") + assert.NotContains(t, err.Error(), featureRecordDeploymentHistory) +} + +// TestDataToPersistOnlyStripsRecordedState pins that the header-only write is scoped to a +// recorded deployment: an ordinary one persists its resources untouched. It also pins that the +// in-memory state survives, since the rest of the deploy reads from it. +func TestDataToPersistOnlyStripsRecordedState(t *testing.T) { + entries := map[string]ResourceEntry{ + "resources.jobs.my_job": {ID: "123", State: json.RawMessage(`{"name":"n"}`)}, + } + + var plain DeploymentState + plain.Data = NewDatabase("test-lineage", 1) + plain.Data.State = entries + assert.Equal(t, entries, plain.dataToPersist().State, "an unrecorded deployment persists its resources") + + var recorded DeploymentState + recorded.Data = NewDatabase("test-lineage", 1) + recorded.Data.State = entries + recorded.Data.StateVersion = featureStateVersion + recorded.Data.Features = map[string]struct{}{featureRecordDeploymentHistory: {}} + + persisted := recorded.dataToPersist() + assert.Empty(t, persisted.State, "a recorded deployment persists the header alone") + assert.Equal(t, featureStateVersion, persisted.StateVersion) + assert.Contains(t, persisted.Features, featureRecordDeploymentHistory) + assert.Equal(t, entries, recorded.Data.State, "the in-memory state is what the deploy reads, so it must not be cleared") +} + +// TestDataToPersistDropsMarkerWhenNothingIsRecorded pins that a recorded state which has lost its +// last resource - what a destroy leaves behind - stops recording the feature. Keeping it would +// refuse every later deploy that does not record, with nothing left for the marker to protect. +func TestDataToPersistDropsMarkerWhenNothingIsRecorded(t *testing.T) { + var db DeploymentState + db.Data = NewDatabase("test-lineage", 1) + db.Data.StateVersion = featureStateVersion + db.Data.Features = map[string]struct{}{featureRecordDeploymentHistory: {}} + + persisted := db.dataToPersist() + assert.NotContains(t, persisted.Features, featureRecordDeploymentHistory) + assert.Contains(t, db.Data.Features, featureRecordDeploymentHistory, "the in-memory copy is untouched") +} + func TestDeleteState(t *testing.T) { path := filepath.Join(t.TempDir(), "state.json") diff --git a/bundle/phases/deploy.go b/bundle/phases/deploy.go index 197a6bac823..e089c976419 100644 --- a/bundle/phases/deploy.go +++ b/bundle/phases/deploy.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "slices" + "strconv" "strings" "github.com/databricks/cli/bundle" @@ -271,27 +272,24 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand return } + planFromFile := plan != nil if b.DeploymentBundle.DmsApiClient != nil { - // Create the deployment and settle the version before planning: Plan snapshots the config, - // and the version itself is created after approval. The deployment has to exist to be stamped. + // Create the deployment before planning so it exists to stamp. createOrUpdateDeployment(ctx, b, dmsDeployment) if logdiag.HasError(ctx) { return } - // The deployment ID is stamped earlier, when the state is opened; only the - // version is new here. A first deploy has no ID until now, so stamp both. - deploymentID, versionID := recordedDeployment(b) - bundle.ApplySeqContext( - ctx, b, - metadata.AnnotateDeployment(deploymentID), - metadata.AnnotateDeploymentVersion(versionID), - ) - if logdiag.HasError(ctx) { - return + // A normal deploy stamps id + version onto the config, off the plan, so RunPlan carries them + // into the applied plan; deploy --plan stamps the loaded plan in InitForApply below instead. + if !planFromFile { + deploymentID, version := recordedDeployment(b) + bundle.ApplySeqContext(ctx, b, metadata.AnnotateDeployment(deploymentID), metadata.AnnotateDeploymentVersion(version)) + if logdiag.HasError(ctx) { + return + } } } - planFromFile := plan != nil if plan == nil { // State is already open for read by process.go (for direct engine) plan = RunPlan(ctx, b, stateEngine) @@ -313,8 +311,10 @@ func Deploy(ctx context.Context, b *bundle.Bundle, outputHandler sync.OutputHand } if planFromFile { - // Initialize DeploymentBundle for applying the loaded plan - err := b.DeploymentBundle.InitForApply(ctx, b.WorkspaceClient(ctx), plan) + // Stamp the deployment id and version onto the loaded plan here, not into the saved plan. + // Non-recording deploys pass "" and stamp nothing. + deploymentID, version := recordedDeployment(b) + err := b.DeploymentBundle.InitForApply(ctx, b.WorkspaceClient(ctx), plan, deploymentID, strconv.FormatInt(version, 10)) if err != nil { logdiag.LogError(ctx, err) return diff --git a/bundle/phases/dms.go b/bundle/phases/dms.go index 21dd15d6acf..2dc12a0da96 100644 --- a/bundle/phases/dms.go +++ b/bundle/phases/dms.go @@ -173,7 +173,6 @@ func drainOperationsAndCompleteVersion(ctx context.Context, b *bundle.Bundle, su if buf.Drain() != nil { success = false } - buf.Stop() reason := bundledeployments.VersionCompleteVersionCompleteSuccess if !success { diff --git a/cmd/bundle/generate/dashboard.go b/cmd/bundle/generate/dashboard.go index 83612d37c16..48ceebd8908 100644 --- a/cmd/bundle/generate/dashboard.go +++ b/cmd/bundle/generate/dashboard.go @@ -404,7 +404,13 @@ func (d *dashboard) runForResource(ctx context.Context, b *bundle.Bundle) { var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, ""); err != nil { + // While recording, the resources come from the service and not the state file. + dmsClient, dmsDeploymentID, err := utils.DmsStateSource(ctx, b) + if err != nil { + logdiag.LogError(ctx, err) + return + } + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dmsClient, dmsDeploymentID); err != nil { logdiag.LogError(ctx, err) return } diff --git a/cmd/bundle/generate/genie_space.go b/cmd/bundle/generate/genie_space.go index f0b3072c01c..82ed6708b20 100644 --- a/cmd/bundle/generate/genie_space.go +++ b/cmd/bundle/generate/genie_space.go @@ -322,7 +322,13 @@ func (g *genieSpace) runForResource(ctx context.Context, b *bundle.Bundle) { var state statemgmt.ExportedResourcesMap if stateDesc.Engine.IsDirect() { _, localPath := b.StateFilenameDirect(ctx) - if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), nil, ""); err != nil { + // While recording, the resources come from the service and not the state file. + dmsClient, dmsDeploymentID, err := utils.DmsStateSource(ctx, b) + if err != nil { + logdiag.LogError(ctx, err) + return + } + if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dmsClient, dmsDeploymentID); err != nil { logdiag.LogError(ctx, err) return } diff --git a/cmd/bundle/utils/process.go b/cmd/bundle/utils/process.go index 3e10d90df25..0e7c28b4ea4 100644 --- a/cmd/bundle/utils/process.go +++ b/cmd/bundle/utils/process.go @@ -197,9 +197,10 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle return b, nil, err } - // The deployment record the recording read, if any; passed to phases.Deploy for the metadata - // diff. Nil for a non-recording bundle or a first deploy. + // The current deployment read from the service (nil, id "" if there is none yet). Used for the + // metadata diff and to reject a saved plan that predates the deployment's recorded version. var dmsDeployment *bundledeployments.Deployment + var dmsDeploymentID string shouldReadState := opts.ReadState || opts.AlwaysPull || opts.InitIDs || opts.ErrorOnEmptyState || opts.PreDeployChecks || opts.Deploy || opts.ReadPlanPath != "" @@ -246,7 +247,6 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle _, localPath := b.StateFilenameDirect(ctx) var dmsClient *dms.Client - var dmsDeploymentID string if b.RecordsDeploymentHistory(ctx) { deploymentID, deployment, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) if err != nil { @@ -263,6 +263,8 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle dmsDeployment = deployment b.DeploymentBundle.DmsApiClient = dmsClient + // A resolved deployment id goes into history and is stamped onto resources to avoid drift + // (empty on a first deploy; the deploy phase stamps the created id; version_id is DMS-managed). if deploymentID != "" { bundle.ApplyFuncContext(ctx, b, func(_ context.Context, b *bundle.Bundle) { b.Config.Bundle.Deployment.History = &config.DeploymentHistory{ @@ -270,14 +272,10 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle LatestVersionID: deployment.LastVersionId, } }) - } - - // Stamp the deployment before anything diffs the resources: the workspace - // has it, so leaving it unset would report drift on an untouched resource. - // The deploy phase stamps the version, once it claims one. - bundle.ApplyContext(ctx, b, metadata.AnnotateDeployment(deploymentID)) - if logdiag.HasError(ctx) { - return b, stateDesc, root.ErrAlreadyPrinted + bundle.ApplyContext(ctx, b, metadata.AnnotateDeployment(deploymentID)) + if logdiag.HasError(ctx) { + return b, stateDesc, root.ErrAlreadyPrinted + } } } if err := b.DeploymentBundle.StateDB.Open(ctx, localPath, dstate.WithRecovery(true), dstate.WithWrite(false), dmsClient, dmsDeploymentID); err != nil { @@ -285,6 +283,15 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle return b, stateDesc, root.ErrAlreadyPrinted } + // The service holds this deployment, so it has to keep being recorded: deploying + // without recording would leave it describing resources that have moved on. + if !b.RecordsDeploymentHistory(ctx) && b.DeploymentBundle.StateDB.RequiresDeploymentHistory() { + logdiag.LogError(ctx, errors.New(`unsetting experimental.record_deployment_history is not supported + +This deployment's resources are recorded with the deployment metadata service. Set experimental.record_deployment_history: true to deploy or destroy this bundle`)) + return b, stateDesc, root.ErrAlreadyPrinted + } + // Warn when the state was last written by a newer CLI than the one // running now. The state schema version is a hard gate (dstate.Open // rejects a too-new state_version), but a state can be written by a @@ -356,6 +363,13 @@ func ProcessBundleRet(cmd *cobra.Command, opts ProcessOptions) (b *bundle.Bundle logdiag.LogError(ctx, err) return b, stateDesc, root.ErrAlreadyPrinted } + + // A first-deploy plan has empty lineage, so ValidatePlanAgainstState skips it; if the deployment + // has since recorded a version, reject the replay (a non-empty lineage is covered above). + if plan.Lineage == "" && dmsDeployment != nil && dmsDeployment.LastVersionId != "" { + logdiag.LogError(ctx, fmt.Errorf("this plan predates the deployment's current version %s; run 'bundle plan' again", dmsDeployment.LastVersionId)) + return b, stateDesc, root.ErrAlreadyPrinted + } } else if opts.Deploy { opts.Build = true opts.PreDeployChecks = true @@ -476,6 +490,27 @@ func ResolveEngineSetting(ctx context.Context, b *bundle.Bundle) (engine.EngineS return engine.EngineSetting{Type: engine.Default, Source: engine.SourceDefault, IsDefault: true}, nil } +// DmsStateSource returns the client and deployment id dstate.Open needs to read a recorded +// bundle's resources from the deployment metadata service. Both are zero when the bundle does not +// record: while recording the state file holds only the marker, so a command that opens state +// without these sees no resources at all. +func DmsStateSource(ctx context.Context, b *bundle.Bundle) (*dms.Client, string, error) { + if !b.RecordsDeploymentHistory(ctx) { + return nil, "", nil + } + + deploymentID, _, err := fetchDeploymentFromStatePath(ctx, b.WorkspaceClient(ctx), b.Config.Workspace.StatePath) + if err != nil { + return nil, "", err + } + + client, err := dms.NewClient(b.WorkspaceClient(ctx)) + if err != nil { + return nil, "", err + } + return client, deploymentID, nil +} + // Lookup and return the deployment object from ${workspace.state_path}/resources.deployment.json func fetchDeploymentFromStatePath(ctx context.Context, w *databricks.WorkspaceClient, statePath string) (string, *bundledeployments.Deployment, error) { nodePath := path.Join(statePath, dms.DeploymentNodeName) diff --git a/libs/dms/client.go b/libs/dms/client.go index 70a310311f0..8dbf9915399 100644 --- a/libs/dms/client.go +++ b/libs/dms/client.go @@ -94,14 +94,6 @@ func (c *Client) CompleteVersion(ctx context.Context, deploymentID string, versi return err } -// Heartbeat renews the version's lease. -func (c *Client) Heartbeat(ctx context.Context, deploymentID string, version int64) error { - _, err := c.Service.Heartbeat(ctx, bundledeployments.HeartbeatRequest{ - Name: versionName(deploymentID, version), - }) - return err -} - // UpdateOperation fills in one operation the version staged, and returns the sequence id the // next update for that resource must send. func (c *Client) UpdateOperation(ctx context.Context, deploymentID string, version int64, stateKey, sequenceID string, update OperationUpdate) (string, error) { diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go index 39a89fd8606..bc58ff5285e 100644 --- a/libs/dms/operation_buffer.go +++ b/libs/dms/operation_buffer.go @@ -3,20 +3,10 @@ package dms import ( "context" "encoding/json" - "errors" "fmt" - "net/http" "sync" - "time" - - "github.com/databricks/cli/libs/log" - "github.com/databricks/databricks-sdk-go/apierr" ) -// The server expires a version's lease if it does not receive a heartbeat -// within a 2-minute TTL; we heartbeat well inside that window. -const defaultHeartbeatInterval = 30 * time.Second - // bufferedOperations caps how far ahead of the service a deploy may get; DMS is what the next // plan reads. const bufferedOperations = 10 @@ -26,16 +16,13 @@ const bufferedOperations = 10 const stagedSequenceID = "0" // OperationBuffer records each state write with DMS for one deployment version, off the apply -// path: writes are queued and sent on one background goroutine, and a heartbeat keeps the -// version's lease alive until Stop. It exists only while a bundle records deployment history; -// callers hold a nil buffer otherwise and must not call it. +// path: writes are queued and sent on one background goroutine. It exists only while a bundle +// records deployment history; callers hold a nil buffer otherwise and must not call it. type OperationBuffer struct { client *Client deploymentID string versionNum int64 - stopHeartbeat context.CancelFunc - // The buffer. queue holds bundle state keys, and pending the newest update per key, so a // second write for a resource replaces the first. queue chan string @@ -56,9 +43,8 @@ type OperationBuffer struct { err error } -// StartOperationBuffer opens the buffer for the version the caller just created, and starts the -// heartbeat that keeps its lease alive. The version must already exist: operations record under -// it, and nothing here creates it. +// StartOperationBuffer opens the buffer for the version the caller just created. The version +// must already exist: operations record under it, and nothing here creates it. func StartOperationBuffer(ctx context.Context, client *Client, deploymentID string, versionNum int64) *OperationBuffer { b := &OperationBuffer{ client: client, @@ -70,7 +56,6 @@ func StartOperationBuffer(ctx context.Context, client *Client, deploymentID stri sequenceIDs: make(map[string]string), } b.stopQueue = sync.OnceFunc(func() { close(b.queue) }) - b.stopHeartbeat = startHeartbeat(ctx, client, deploymentID, versionNum) go b.run(ctx) return b } @@ -167,11 +152,6 @@ func (b *OperationBuffer) Drain() error { return b.Err() } -// Stop stops the heartbeat. Call it once the version is completed. -func (b *OperationBuffer) Stop() { - b.stopHeartbeat() -} - // setErr keeps the first error; one failure is enough to fail the deploy. func (b *OperationBuffer) setErr(err error) { b.mu.Lock() @@ -189,43 +169,3 @@ func (b *OperationBuffer) Err() error { return b.err } - -// startHeartbeat starts a background goroutine that sends heartbeats to keep -// the deployment version's lease alive. Returns a cancel function to stop it. -func startHeartbeat(ctx context.Context, client *Client, deploymentID string, version int64) context.CancelFunc { - ctx, cancel := context.WithCancel(ctx) - - go func() { - ticker := time.NewTicker(defaultHeartbeatInterval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - err := client.Heartbeat(ctx, deploymentID, version) - if err != nil { - // A 409 Conflict is expected if the version was completed - // between the ticker firing and the heartbeat. - if isAbortedErr(err) { - log.Debugf(ctx, "Heartbeat stopped: version already completed") - return - } - log.Warnf(ctx, "Failed to send deployment heartbeat: %v", err) - } else { - log.Debugf(ctx, "Deployment heartbeat sent: deployment=%s version=%d", deploymentID, version) - } - } - } - }() - - return cancel -} - -// isAbortedErr reports whether err is an HTTP 409 Conflict from the DMS API, whose error -// code is ABORTED. -func isAbortedErr(err error) bool { - apiErr, ok := errors.AsType[*apierr.APIError](err) - return ok && apiErr.StatusCode == http.StatusConflict && apiErr.ErrorCode == "ABORTED" -}