Integrate deployment metadata service for server-side locking and state

Add client integration with the deployment metadata service API for
server-side deployment locking and resource state tracking. Gated behind
DATABRICKS_BUNDLE_DEPLOYMENT_SERVICE=true environment variable.

Co-authored-by: Isaac

Author: shreyas-goenka

bundle/deploy/metadata/service/client.go

@@ -0,0 +1,183 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"errors"
+
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/apierr"
+	"github.com/databricks/databricks-sdk-go/client"
+)
+
+const basePath = "/api/2.0/bundle"
+
+// Client wraps the Databricks API client for the deployment metadata service.
+type Client struct {
+	api *client.DatabricksClient
+}
+
+// NewClient creates a new deployment metadata service client from a workspace client.
+func NewClient(w *databricks.WorkspaceClient) (*Client, error) {
+	apiClient, err := client.New(w.Config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create deployment metadata API client: %w", err)
+	}
+	return &Client{api: apiClient}, nil
+}
+
+// CreateDeployment creates a new deployment.
+func (c *Client) CreateDeployment(ctx context.Context, deploymentID string, deployment *Deployment) (*Deployment, error) {
+	resp := &Deployment{}
+	path := fmt.Sprintf("%s/deployments", basePath)
+	err := c.api.Do(ctx, http.MethodPost, path, nil, nil, CreateDeploymentRequest{
+		DeploymentID: deploymentID,
+		Deployment:   deployment,
+	}, resp)
+	if err != nil {
+		return nil, mapError("create deployment", err)
+	}
+	return resp, nil
+}
+
+// GetDeployment retrieves a deployment by ID.
+func (c *Client) GetDeployment(ctx context.Context, deploymentID string) (*Deployment, error) {
+	resp := &Deployment{}
+	path := fmt.Sprintf("%s/deployments/%s", basePath, deploymentID)
+	err := c.api.Do(ctx, http.MethodGet, path, nil, nil, nil, resp)
+	if err != nil {
+		return nil, mapError("get deployment", err)
+	}
+	return resp, nil
+}
+
+// DeleteDeployment soft-deletes a deployment.
+func (c *Client) DeleteDeployment(ctx context.Context, deploymentID string) error {
+	path := fmt.Sprintf("%s/deployments/%s", basePath, deploymentID)
+	err := c.api.Do(ctx, http.MethodDelete, path, nil, nil, nil, nil)
+	if err != nil {
+		return mapError("delete deployment", err)
+	}
+	return nil
+}
+
+// CreateVersion creates a new version (acquires the deployment lock).
+func (c *Client) CreateVersion(ctx context.Context, deploymentID string, versionID string, version *Version) (*Version, error) {
+	resp := &Version{}
+	path := fmt.Sprintf("%s/deployments/%s/versions", basePath, deploymentID)
+	err := c.api.Do(ctx, http.MethodPost, path, nil, nil, CreateVersionRequest{
+		Parent:    fmt.Sprintf("deployments/%s", deploymentID),
+		Version:   version,
+		VersionID: versionID,
+	}, resp)
+	if err != nil {
+		return nil, mapError("create version", err)
+	}
+	return resp, nil
+}
+
+// GetVersion retrieves a version.
+func (c *Client) GetVersion(ctx context.Context, deploymentID, versionID string) (*Version, error) {
+	resp := &Version{}
+	path := fmt.Sprintf("%s/deployments/%s/versions/%s", basePath, deploymentID, versionID)
+	err := c.api.Do(ctx, http.MethodGet, path, nil, nil, nil, resp)
+	if err != nil {
+		return nil, mapError("get version", err)
+	}
+	return resp, nil
+}
+
+// Heartbeat renews the lock lease for an in-progress version.
+func (c *Client) Heartbeat(ctx context.Context, deploymentID, versionID string) (*HeartbeatResponse, error) {
+	resp := &HeartbeatResponse{}
+	path := fmt.Sprintf("%s/deployments/%s/versions/%s/heartbeat", basePath, deploymentID, versionID)
+	err := c.api.Do(ctx, http.MethodPost, path, nil, nil, struct{}{}, resp)
+	if err != nil {
+		return nil, mapError("heartbeat", err)
+	}
+	return resp, nil
+}
+
+// CompleteVersion marks a version as completed (releases the deployment lock).
+func (c *Client) CompleteVersion(ctx context.Context, deploymentID, versionID string, reason VersionComplete, force bool) (*Version, error) {
+	resp := &Version{}
+	path := fmt.Sprintf("%s/deployments/%s/versions/%s/complete", basePath, deploymentID, versionID)
+	err := c.api.Do(ctx, http.MethodPost, path, nil, nil, CompleteVersionRequest{
+		Name:             fmt.Sprintf("deployments/%s/versions/%s", deploymentID, versionID),
+		CompletionReason: reason,
+		Force:            force,
+	}, resp)
+	if err != nil {
+		return nil, mapError("complete version", err)
+	}
+	return resp, nil
+}
+
+// CreateOperation records a resource operation for a version.
+func (c *Client) CreateOperation(ctx context.Context, deploymentID, versionID, resourceKey string, operation *Operation) (*Operation, error) {
+	resp := &Operation{}
+	path := fmt.Sprintf("%s/deployments/%s/versions/%s/operations", basePath, deploymentID, versionID)
+	err := c.api.Do(ctx, http.MethodPost, path, nil, nil, CreateOperationRequest{
+		Parent:      fmt.Sprintf("deployments/%s/versions/%s", deploymentID, versionID),
+		ResourceKey: resourceKey,
+		Operation:   operation,
+	}, resp)
+	if err != nil {
+		return nil, mapError("create operation", err)
+	}
+	return resp, nil
+}
+
+// ListResources lists all resources for a deployment.
+func (c *Client) ListResources(ctx context.Context, deploymentID string) ([]Resource, error) {
+	var allResources []Resource
+	pageToken := ""
+
+	for {
+		resp := &ListResourcesResponse{}
+		path := fmt.Sprintf("%s/deployments/%s/resources", basePath, deploymentID)
+
+		q := map[string]any{
+			"parent":    fmt.Sprintf("deployments/%s", deploymentID),
+			"page_size": 1000,
+		}
+		if pageToken != "" {
+			q["page_token"] = pageToken
+		}
+
+		err := c.api.Do(ctx, http.MethodGet, path, nil, q, nil, resp)
+		if err != nil {
+			return nil, mapError("list resources", err)
+		}
+
+		allResources = append(allResources, resp.Resources...)
+		if resp.NextPageToken == "" {
+			break
+		}
+		pageToken = resp.NextPageToken
+	}
+
+	return allResources, nil
+}
+
+// mapError translates API errors into user-friendly messages.
+func mapError(operation string, err error) error {
+	var apiErr *apierr.APIError
+	if !errors.As(err, &apiErr) {
+		return fmt.Errorf("%s: %w", operation, err)
+	}
+
+	switch apiErr.StatusCode {
+	case http.StatusConflict:
+		return fmt.Errorf("%s: deployment is locked by another active deployment. "+
+			"Use --force-lock to override", operation)
+	case http.StatusNotFound:
+		return fmt.Errorf("%s: resource not found: %w", operation, err)
+	case http.StatusBadRequest:
+		return fmt.Errorf("%s: bad request: %s", operation, apiErr.Message)
+	default:
+		return fmt.Errorf("%s: %w", operation, err)
+	}
+}

bundle/deploy/metadata/service/heartbeat.go

@@ -0,0 +1,37 @@
+package service
+
+import (
+	"context"
+	"time"
+
+	"github.com/databricks/cli/libs/log"
+)
+
+const DefaultHeartbeatInterval = 2 * time.Minute
+
+// StartHeartbeat starts a background goroutine that sends heartbeats to keep
+// the deployment lock alive. Returns a cancel function to stop the heartbeat.
+func StartHeartbeat(ctx context.Context, client *Client, deploymentID, versionID string, interval time.Duration) context.CancelFunc {
+	ctx, cancel := context.WithCancel(ctx)
+
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				_, err := client.Heartbeat(ctx, deploymentID, versionID)
+				if err != nil {
+					log.Warnf(ctx, "Failed to send deployment heartbeat: %v", err)
+				} else {
+					log.Debugf(ctx, "Deployment heartbeat sent for deployment=%s version=%s", deploymentID, versionID)
+				}
+			}
+		}
+	}()
+
+	return cancel
+}