direct: retry 504 errors (#5349)

## Changes
Retry resource methods that return an error that has http_code equal to
504 but that has not been retried by SDK.

This affects all DoRead, DoUpdate(WithID), DoDelete. This affects
DoCreate for grants and permissions.

Note, for DoCreate the retry functionality is opt-in - implementations
need to wrap error with retrySafe(). For other methods the retry is
always enabled.

## Why
We've seen reports where deploy fails with 
> Error: cannot create resources.pipelines.<pipeline>.permissions: The
service at /api/2.0/permissions/pipelines/<pipeline_id> is taking too
long to process your request. (504 TEMPORARILY_UNAVAILABLE)

We also saw that terraform does custom retries for 504/GET
databricks/terraform-provider-databricks#4355

Note, the two cases are different - the first one is "cannot create" so
it refers to PUT.

## Tests
New testserver feature that allows injecting expiring faults in a given
endpoint. See fault.py.
New acceptance tests make use of fault.py to check failures in
plan/create/update for permissions.

Author: denik

NEXT_CHANGELOG.md

@@ -8,5 +8,6 @@
 * `experimental open` now opens every DABs resource type that has a workspace URL, picking up `catalogs`, `schemas`, `volumes`, `database_instances`, `database_catalogs`, `synced_database_tables`, `postgres_catalogs`, `postgres_synced_tables`, `quality_monitors`, `vector_search_endpoints`, and `vector_search_indexes` ([#5346](https://github.com/databricks/cli/pull/5346)).
 
 ### Bundles
+* Retry transient HTTP 504 Gateway Timeout errors in direct deployment engine ([#5349](https://github.com/databricks/cli/pull/5349)).
 
 ### Dependency updates

acceptance/bin/fault.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""Set up a fault rule on the testserver for the current test token.
+
+Usage: fault.py PATTERN STATUS_CODE OFFSET TIMES
+
+  PATTERN     HTTP method and path, supports trailing * wildcard,
+              e.g. "PUT /api/2.0/permissions/pipelines/*"
+  STATUS_CODE HTTP status code to return, e.g. 504
+  OFFSET      number of requests to let through before fault starts
+  TIMES       number of times to return the fault response
+
+The rule is scoped to the current DATABRICKS_TOKEN so it only affects
+the test that registers it, even when tests share a server.
+"""
+
+import json
+import os
+import sys
+import urllib.request
+
+host = os.environ.get("DATABRICKS_HOST", "")
+token = os.environ.get("DATABRICKS_TOKEN", "")
+
+if not host:
+    print("DATABRICKS_HOST not set", file=sys.stderr)
+    sys.exit(1)
+
+if len(sys.argv) != 5:
+    print(f"usage: {sys.argv[0]} PATTERN STATUS_CODE OFFSET TIMES", file=sys.stderr)
+    sys.exit(1)
+
+pattern, status_code, offset, times = sys.argv[1], int(sys.argv[2]), int(sys.argv[3]), int(sys.argv[4])
+body = '{"error_code":"INJECTED","message":"Fault injected by test."}'
+
+data = json.dumps(
+    {
+        "pattern": pattern,
+        "status_code": status_code,
+        "body": body,
+        "offset": offset,
+        "times": times,
+    }
+).encode()
+
+req = urllib.request.Request(
+    f"{host}/__testserver/fault",
+    data=data,
+    headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
+    method="POST",
+)
+urllib.request.urlopen(req)

acceptance/bundle/resources/permissions/pipelines/504/create/databricks.yml

@@ -0,0 +1,10 @@
+bundle:
+  name: test-bundle
+
+resources:
+  pipelines:
+    foo:
+      name: foo
+      permissions:
+        - level: CAN_VIEW
+          user_name: viewer@example.com

acceptance/bundle/resources/permissions/pipelines/504/create/out.test.toml

@@ -0,0 +1,9 @@
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Warn: deploying resources.pipelines.foo.permissions: retrying after 504 Gateway Timeout from PUT /api/2.0/permissions/pipelines/[UUID]
+Updating deployment state...
+Deployment complete!
+
+>>> print_requests.py //api/2.0/permissions/pipelines
+"PUT /api/2.0/permissions/pipelines/[UUID]"
+"PUT /api/2.0/permissions/pipelines/[UUID]"

acceptance/bundle/resources/permissions/pipelines/504/create/output.txt

@@ -0,0 +1,8 @@
+# Inject a single 504 on the first permissions PUT to simulate a transient error.
+# Permissions Set is idempotent, so DoCreate opts in via retrySafe and the deploy succeeds.
+fault.py "PUT /api/2.0/permissions/pipelines/*" 504 0 1
+
+$CLI bundle deploy
+
+# Two PUT requests should appear: the initial 504 and the successful retry.
+trace print_requests.py //api/2.0/permissions/pipelines | jq '.method + " " + .path'

acceptance/bundle/resources/permissions/pipelines/504/create/script

@@ -0,0 +1 @@
+RecordRequests = true

acceptance/bundle/resources/permissions/pipelines/504/create/test.toml

@@ -0,0 +1,10 @@
+bundle:
+  name: test-bundle
+
+resources:
+  pipelines:
+    foo:
+      name: foo
+      permissions:
+        - level: CAN_VIEW
+          user_name: viewer@example.com

acceptance/bundle/resources/permissions/pipelines/504/plan/databricks.yml

@@ -0,0 +1,31 @@
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-bundle/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle plan
+Warn: planning resources.pipelines.foo.permissions: retrying after 504 Gateway Timeout from GET /api/2.0/permissions/pipelines/[UUID]
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+
+>>> print_requests.py //api/2.0/permissions/pipelines --get --oneline
+2 {"method": "GET", "path": "/api/2.0/permissions/pipelines/[UUID]"}
+
+>>> [CLI] bundle plan
+Warn: planning resources.pipelines.foo.permissions: retrying after 504 Gateway Timeout from GET /api/2.0/permissions/pipelines/[UUID]
+Warn: planning resources.pipelines.foo.permissions: retrying after 504 Gateway Timeout from GET /api/2.0/permissions/pipelines/[UUID]
+Plan: 0 to add, 0 to change, 0 to delete, 2 unchanged
+3 {"method": "GET", "path": "/api/2.0/permissions/pipelines/[UUID]"}
+
+>>> musterr [CLI] bundle plan
+Warn: planning resources.pipelines.foo.permissions: retrying after 504 Gateway Timeout from GET /api/2.0/permissions/pipelines/[UUID]
+Warn: planning resources.pipelines.foo.permissions: retrying after 504 Gateway Timeout from GET /api/2.0/permissions/pipelines/[UUID]
+Error: cannot plan resources.pipelines.foo.permissions: reading id="/pipelines/[UUID]": Fault injected by test. (504 INJECTED)
+
+Endpoint: GET [DATABRICKS_URL]/api/2.0/permissions/pipelines/[UUID]?
+HTTP Status: 504 Gateway Timeout
+API error_code: INJECTED
+API message: Fault injected by test.
+
+Error: planning failed
+
+3 {"method": "GET", "path": "/api/2.0/permissions/pipelines/[UUID]"}