feat: Azure Databricks SDK-compat handler

databricks/databricks.go

@@ -0,0 +1,166 @@
+// Package databricks provides a portable analytics-workspace API with
+// cross-cutting concerns. It wraps a driver.Databricks with recording,
+// metrics, rate limiting, error injection, and latency simulation.
+package databricks
+
+import (
+	"context"
+	"time"
+
+	"github.com/stackshy/cloudemu/databricks/driver"
+	"github.com/stackshy/cloudemu/inject"
+	"github.com/stackshy/cloudemu/metrics"
+	"github.com/stackshy/cloudemu/ratelimit"
+	"github.com/stackshy/cloudemu/recorder"
+)
+
+// Databricks is the portable workspace type wrapping a driver with
+// cross-cutting concerns.
+type Databricks struct {
+	driver   driver.Databricks
+	recorder *recorder.Recorder
+	metrics  *metrics.Collector
+	limiter  *ratelimit.Limiter
+	injector *inject.Injector
+	latency  time.Duration
+}
+
+// NewDatabricks creates a new portable Databricks wrapping the given driver.
+func NewDatabricks(d driver.Databricks, opts ...Option) *Databricks {
+	b := &Databricks{driver: d}
+	for _, opt := range opts {
+		opt(b)
+	}
+
+	return b
+}
+
+// Option configures a portable Databricks.
+type Option func(*Databricks)
+
+// WithRecorder sets the recorder.
+func WithRecorder(r *recorder.Recorder) Option { return func(b *Databricks) { b.recorder = r } }
+
+// WithMetrics sets the metrics collector.
+func WithMetrics(m *metrics.Collector) Option { return func(b *Databricks) { b.metrics = m } }
+
+// WithRateLimiter sets the rate limiter.
+func WithRateLimiter(l *ratelimit.Limiter) Option { return func(b *Databricks) { b.limiter = l } }
+
+// WithErrorInjection sets the error injector.
+func WithErrorInjection(i *inject.Injector) Option { return func(b *Databricks) { b.injector = i } }
+
+// WithLatency sets simulated latency.
+func WithLatency(d time.Duration) Option { return func(b *Databricks) { b.latency = d } }
+
+func (b *Databricks) do(_ context.Context, op string, input any, fn func() (any, error)) (any, error) {
+	start := time.Now()
+
+	if b.injector != nil {
+		if err := b.injector.Check("databricks", op); err != nil {
+			b.rec(op, input, nil, err, time.Since(start))
+			return nil, err
+		}
+	}
+
+	if b.limiter != nil {
+		if err := b.limiter.Allow(); err != nil {
+			b.rec(op, input, nil, err, time.Since(start))
+			return nil, err
+		}
+	}
+
+	if b.latency > 0 {
+		time.Sleep(b.latency)
+	}
+
+	out, err := fn()
+	dur := time.Since(start)
+
+	if b.metrics != nil {
+		labels := map[string]string{"service": "databricks", "operation": op}
+		b.metrics.Counter("calls_total", 1, labels)
+		b.metrics.Histogram("call_duration", dur, labels)
+
+		if err != nil {
+			b.metrics.Counter("errors_total", 1, labels)
+		}
+	}
+
+	b.rec(op, input, out, err, dur)
+
+	return out, err
+}
+
+func (b *Databricks) rec(op string, input, output any, err error, dur time.Duration) {
+	if b.recorder != nil {
+		b.recorder.Record("databricks", op, input, output, err, dur)
+	}
+}
+
+// CreateWorkspace creates a new managed workspace.
+//
+//nolint:gocritic // cfg matches the driver interface signature; copied once on entry.
+func (b *Databricks) CreateWorkspace(ctx context.Context, cfg driver.WorkspaceConfig) (*driver.Workspace, error) {
+	out, err := b.do(ctx, "CreateWorkspace", cfg, func() (any, error) { return b.driver.CreateWorkspace(ctx, cfg) })
+	if err != nil {
+		return nil, err
+	}
+
+	return out.(*driver.Workspace), nil
+}
+
+// GetWorkspace retrieves a workspace by resource group and name.
+func (b *Databricks) GetWorkspace(ctx context.Context, resourceGroup, name string) (*driver.Workspace, error) {
+	out, err := b.do(ctx, "GetWorkspace", name, func() (any, error) { return b.driver.GetWorkspace(ctx, resourceGroup, name) })
+	if err != nil {
+		return nil, err
+	}
+
+	return out.(*driver.Workspace), nil
+}
+
+// DeleteWorkspace deletes a workspace by resource group and name.
+func (b *Databricks) DeleteWorkspace(ctx context.Context, resourceGroup, name string) error {
+	_, err := b.do(ctx, "DeleteWorkspace", name, func() (any, error) {
+		return nil, b.driver.DeleteWorkspace(ctx, resourceGroup, name)
+	})
+
+	return err
+}
+
+// UpdateWorkspaceTags replaces a workspace's tags.
+func (b *Databricks) UpdateWorkspaceTags(
+	ctx context.Context, resourceGroup, name string, tags map[string]string,
+) (*driver.Workspace, error) {
+	out, err := b.do(ctx, "UpdateWorkspaceTags", name, func() (any, error) {
+		return b.driver.UpdateWorkspaceTags(ctx, resourceGroup, name, tags)
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return out.(*driver.Workspace), nil
+}
+
+// ListWorkspacesByResourceGroup lists workspaces in a resource group.
+func (b *Databricks) ListWorkspacesByResourceGroup(ctx context.Context, resourceGroup string) ([]driver.Workspace, error) {
+	out, err := b.do(ctx, "ListWorkspacesByResourceGroup", resourceGroup, func() (any, error) {
+		return b.driver.ListWorkspacesByResourceGroup(ctx, resourceGroup)
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return out.([]driver.Workspace), nil
+}
+
+// ListWorkspaces lists all workspaces in the subscription.
+func (b *Databricks) ListWorkspaces(ctx context.Context) ([]driver.Workspace, error) {
+	out, err := b.do(ctx, "ListWorkspaces", nil, func() (any, error) { return b.driver.ListWorkspaces(ctx) })
+	if err != nil {
+		return nil, err
+	}
+
+	return out.([]driver.Workspace), nil
+}

databricks/databricks_test.go

@@ -0,0 +1,120 @@
+package databricks
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stackshy/cloudemu/config"
+	"github.com/stackshy/cloudemu/databricks/driver"
+	"github.com/stackshy/cloudemu/inject"
+	"github.com/stackshy/cloudemu/metrics"
+	azuredbx "github.com/stackshy/cloudemu/providers/azure/databricks"
+	"github.com/stackshy/cloudemu/ratelimit"
+	"github.com/stackshy/cloudemu/recorder"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newTestDatabricks(opts ...Option) *Databricks {
+	fc := config.NewFakeClock(time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC))
+	o := config.NewOptions(config.WithClock(fc), config.WithRegion("eastus"), config.WithAccountID("sub-1"))
+
+	return NewDatabricks(azuredbx.New(o), opts...)
+}
+
+func validConfig() driver.WorkspaceConfig {
+	return driver.WorkspaceConfig{
+		Name:                   "ws-1",
+		ResourceGroup:          "rg-1",
+		Location:               "eastus",
+		ManagedResourceGroupID: "/subscriptions/sub-1/resourceGroups/managed",
+	}
+}
+
+func TestLifecycle(t *testing.T) {
+	b := newTestDatabricks()
+	ctx := context.Background()
+
+	ws, err := b.CreateWorkspace(ctx, validConfig())
+	require.NoError(t, err)
+	assert.Equal(t, driver.StateSucceeded, ws.ProvisioningState)
+
+	got, err := b.GetWorkspace(ctx, "rg-1", "ws-1")
+	require.NoError(t, err)
+	assert.Equal(t, "ws-1", got.Name)
+
+	updated, err := b.UpdateWorkspaceTags(ctx, "rg-1", "ws-1", map[string]string{"env": "prod"})
+	require.NoError(t, err)
+	assert.Equal(t, "prod", updated.Tags["env"])
+
+	byRG, err := b.ListWorkspacesByResourceGroup(ctx, "rg-1")
+	require.NoError(t, err)
+	assert.Len(t, byRG, 1)
+
+	all, err := b.ListWorkspaces(ctx)
+	require.NoError(t, err)
+	assert.Len(t, all, 1)
+
+	require.NoError(t, b.DeleteWorkspace(ctx, "rg-1", "ws-1"))
+
+	_, err = b.GetWorkspace(ctx, "rg-1", "ws-1")
+	require.Error(t, err)
+}
+
+func TestWithRecorder(t *testing.T) {
+	rec := recorder.New()
+	b := newTestDatabricks(WithRecorder(rec))
+
+	_, err := b.CreateWorkspace(context.Background(), validConfig())
+	require.NoError(t, err)
+
+	calls := rec.Calls()
+	require.GreaterOrEqual(t, len(calls), 1)
+	assert.Equal(t, "databricks", calls[0].Service)
+	assert.Equal(t, "CreateWorkspace", calls[0].Operation)
+}
+
+func TestWithMetrics(t *testing.T) {
+	mc := metrics.NewCollector()
+	b := newTestDatabricks(WithMetrics(mc))
+
+	_, err := b.ListWorkspaces(context.Background())
+	require.NoError(t, err)
+
+	q := metrics.NewQuery(mc)
+	assert.GreaterOrEqual(t, q.ByName("calls_total").Count(), 1)
+}
+
+func TestWithErrorInjection(t *testing.T) {
+	inj := inject.NewInjector()
+	b := newTestDatabricks(WithErrorInjection(inj))
+
+	inj.Set("databricks", "ListWorkspaces", fmt.Errorf("injected failure"), inject.Always{})
+
+	_, err := b.ListWorkspaces(context.Background())
+	require.Error(t, err)
+}
+
+func TestWithLatency(t *testing.T) {
+	b := newTestDatabricks(WithLatency(time.Millisecond))
+
+	start := time.Now()
+	_, err := b.ListWorkspaces(context.Background())
+	require.NoError(t, err)
+	assert.GreaterOrEqual(t, time.Since(start), time.Millisecond)
+}
+
+func TestWithRateLimiter(t *testing.T) {
+	fc := config.NewFakeClock(time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC))
+	o := config.NewOptions(config.WithClock(fc), config.WithRegion("eastus"))
+	lim := ratelimit.New(1, 1, fc)
+	b := NewDatabricks(azuredbx.New(o), WithRateLimiter(lim))
+
+	_, err := b.ListWorkspaces(context.Background())
+	require.NoError(t, err)
+
+	_, err = b.ListWorkspaces(context.Background())
+	require.Error(t, err)
+}

databricks/driver/driver.go

@@ -0,0 +1,51 @@
+// Package driver defines the interface for Databricks-style analytics
+// workspace services: lifecycle management of managed workspaces.
+package driver
+
+import "context"
+
+// Provisioning state values for a workspace.
+const (
+	StateSucceeded = "Succeeded"
+	StateCreating  = "Creating"
+	StateDeleting  = "Deleting"
+	StateFailed    = "Failed"
+)
+
+// WorkspaceConfig describes a workspace to create.
+type WorkspaceConfig struct {
+	Name                   string
+	ResourceGroup          string
+	Location               string
+	SKUName                string
+	SKUTier                string
+	ManagedResourceGroupID string
+	Tags                   map[string]string
+}
+
+// Workspace describes a managed analytics workspace.
+type Workspace struct {
+	ID                     string
+	Name                   string
+	ResourceGroup          string
+	Location               string
+	SKUName                string
+	SKUTier                string
+	ManagedResourceGroupID string
+	WorkspaceURL           string
+	WorkspaceID            string
+	ProvisioningState      string
+	Tags                   map[string]string
+	CreatedAt              string
+}
+
+// Databricks is the interface that workspace service implementations must
+// satisfy.
+type Databricks interface {
+	CreateWorkspace(ctx context.Context, cfg WorkspaceConfig) (*Workspace, error)
+	GetWorkspace(ctx context.Context, resourceGroup, name string) (*Workspace, error)
+	DeleteWorkspace(ctx context.Context, resourceGroup, name string) error
+	UpdateWorkspaceTags(ctx context.Context, resourceGroup, name string, tags map[string]string) (*Workspace, error)
+	ListWorkspacesByResourceGroup(ctx context.Context, resourceGroup string) ([]Workspace, error)
+	ListWorkspaces(ctx context.Context) ([]Workspace, error)
+}

go.mod

@@ -9,8 +9,10 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.1
 	github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos v1.4.2
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/appservice/armappservice/v3 v3.0.0
+	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/authorization/armauthorization/v3 v3.0.0-beta.3
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/compute/armcompute/v5 v5.7.0
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/containerservice/armcontainerservice/v6 v6.6.0
+	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/databricks/armdatabricks v1.1.0
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/mysql/armmysqlflexibleservers v1.2.0
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/network/armnetwork v1.1.0
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/postgresql/armpostgresqlflexibleservers v1.1.0
@@ -27,6 +29,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.1
 	github.com/aws/aws-sdk-go-v2/service/ec2 v1.297.1
 	github.com/aws/aws-sdk-go-v2/service/eks v1.83.0
+	github.com/aws/aws-sdk-go-v2/service/iam v1.53.10
 	github.com/aws/aws-sdk-go-v2/service/lambda v1.90.1
 	github.com/aws/aws-sdk-go-v2/service/neptune v1.44.5
 	github.com/aws/aws-sdk-go-v2/service/rds v1.118.2
@@ -54,7 +57,6 @@ require (
 	cloud.google.com/go/monitoring v1.27.0 // indirect
 	github.com/Azure/azure-sdk-for-go v68.0.0+incompatible // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.12.0 // indirect
-	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/authorization/armauthorization/v3 v3.0.0-beta.3 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.31.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.55.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0 // indirect
@@ -64,7 +66,6 @@ require (
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22 // indirect
-	github.com/aws/aws-sdk-go-v2/service/iam v1.53.10 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 // indirect
 	github.com/aws/aws-sdk-go-v2/service/internal/endpoint-discovery v1.11.21 // indirect