From 78d0ac5c6afcafe9d03c52acd043f2a0235c2afa Mon Sep 17 00:00:00 2001 From: "Lennart Kats (databricks)" Date: Mon, 19 Aug 2024 20:18:50 +0200 Subject: [PATCH] Add configurable presets for name prefixes, tags, etc. (#1490) ## Changes This adds configurable transformations based on the transformations currently seen in `mode: development`. Example databricks.yml showcasing how some transformations: ``` bundle: name: my_bundle targets: dev: presets: prefix: "myprefix_" # prefix all resource names with myprefix_ pipelines_development: true # set development to true by default for pipelines trigger_pause_status: PAUSED # set pause_status to PAUSED by default for all triggers and schedules jobs_max_concurrent_runs: 10 # set max_concurrent runs to 10 by default for all jobs tags: dev: true ``` ## Tests * Existing process_target_mode tests that were adapted to use this new code * Unit tests specific for the new mutator * Unit tests for config loading and merging * Manual e2e testing --- bundle/config/mutator/apply_presets.go | 209 ++++++++++++++++++ bundle/config/mutator/apply_presets_test.go | 196 ++++++++++++++++ bundle/config/mutator/process_target_mode.go | 130 +++++------ .../mutator/process_target_mode_test.go | 150 ++++++++++++- bundle/config/presets.go | 32 +++ bundle/config/root.go | 5 + bundle/config/target.go | 4 + bundle/phases/initialize.go | 1 + bundle/tests/presets/databricks.yml | 22 ++ bundle/tests/presets_test.go | 28 +++ 10 files changed, 687 insertions(+), 90 deletions(-) create mode 100644 bundle/config/mutator/apply_presets.go create mode 100644 bundle/config/mutator/apply_presets_test.go create mode 100644 bundle/config/presets.go create mode 100644 bundle/tests/presets/databricks.yml create mode 100644 bundle/tests/presets_test.go diff --git a/bundle/config/mutator/apply_presets.go b/bundle/config/mutator/apply_presets.go new file mode 100644 index 00000000..42e6ab95 --- /dev/null +++ b/bundle/config/mutator/apply_presets.go @@ -0,0 +1,209 @@ +package mutator + +import ( + "context" + "path" + "slices" + "sort" + "strings" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/textutil" + "github.com/databricks/databricks-sdk-go/service/catalog" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/databricks/databricks-sdk-go/service/ml" +) + +type applyPresets struct{} + +// Apply all presets, e.g. the prefix presets that +// adds a prefix to all names of all resources. +func ApplyPresets() *applyPresets { + return &applyPresets{} +} + +type Tag struct { + Key string + Value string +} + +func (m *applyPresets) Name() string { + return "ApplyPresets" +} + +func (m *applyPresets) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics { + if d := validatePauseStatus(b); d != nil { + return d + } + + r := b.Config.Resources + t := b.Config.Presets + prefix := t.NamePrefix + tags := toTagArray(t.Tags) + + // Jobs presets: Prefix, Tags, JobsMaxConcurrentRuns, TriggerPauseStatus + for _, j := range r.Jobs { + j.Name = prefix + j.Name + if j.Tags == nil { + j.Tags = make(map[string]string) + } + for _, tag := range tags { + if j.Tags[tag.Key] == "" { + j.Tags[tag.Key] = tag.Value + } + } + if j.MaxConcurrentRuns == 0 { + j.MaxConcurrentRuns = t.JobsMaxConcurrentRuns + } + if t.TriggerPauseStatus != "" { + paused := jobs.PauseStatusPaused + if t.TriggerPauseStatus == config.Unpaused { + paused = jobs.PauseStatusUnpaused + } + + if j.Schedule != nil && j.Schedule.PauseStatus == "" { + j.Schedule.PauseStatus = paused + } + if j.Continuous != nil && j.Continuous.PauseStatus == "" { + j.Continuous.PauseStatus = paused + } + if j.Trigger != nil && j.Trigger.PauseStatus == "" { + j.Trigger.PauseStatus = paused + } + } + } + + // Pipelines presets: Prefix, PipelinesDevelopment + for i := range r.Pipelines { + r.Pipelines[i].Name = prefix + r.Pipelines[i].Name + if config.IsExplicitlyEnabled(t.PipelinesDevelopment) { + r.Pipelines[i].Development = true + } + if t.TriggerPauseStatus == config.Paused { + r.Pipelines[i].Continuous = false + } + + // As of 2024-06, pipelines don't yet support tags + } + + // Models presets: Prefix, Tags + for _, m := range r.Models { + m.Name = prefix + m.Name + for _, t := range tags { + exists := slices.ContainsFunc(m.Tags, func(modelTag ml.ModelTag) bool { + return modelTag.Key == t.Key + }) + if !exists { + // Only add this tag if the resource didn't include any tag that overrides its value. + m.Tags = append(m.Tags, ml.ModelTag{Key: t.Key, Value: t.Value}) + } + } + } + + // Experiments presets: Prefix, Tags + for _, e := range r.Experiments { + filepath := e.Name + dir := path.Dir(filepath) + base := path.Base(filepath) + if dir == "." { + e.Name = prefix + base + } else { + e.Name = dir + "/" + prefix + base + } + for _, t := range tags { + exists := false + for _, experimentTag := range e.Tags { + if experimentTag.Key == t.Key { + exists = true + break + } + } + if !exists { + e.Tags = append(e.Tags, ml.ExperimentTag{Key: t.Key, Value: t.Value}) + } + } + } + + // Model serving endpoint presets: Prefix + for i := range r.ModelServingEndpoints { + r.ModelServingEndpoints[i].Name = normalizePrefix(prefix) + r.ModelServingEndpoints[i].Name + + // As of 2024-06, model serving endpoints don't yet support tags + } + + // Registered models presets: Prefix + for i := range r.RegisteredModels { + r.RegisteredModels[i].Name = normalizePrefix(prefix) + r.RegisteredModels[i].Name + + // As of 2024-06, registered models don't yet support tags + } + + // Quality monitors presets: Prefix + if t.TriggerPauseStatus == config.Paused { + for i := range r.QualityMonitors { + // Remove all schedules from monitors, since they don't support pausing/unpausing. + // Quality monitors might support the "pause" property in the future, so at the + // CLI level we do respect that property if it is set to "unpaused." + if r.QualityMonitors[i].Schedule != nil && r.QualityMonitors[i].Schedule.PauseStatus != catalog.MonitorCronSchedulePauseStatusUnpaused { + r.QualityMonitors[i].Schedule = nil + } + } + } + + // Schemas: Prefix + for i := range r.Schemas { + prefix = "dev_" + b.Config.Workspace.CurrentUser.ShortName + "_" + r.Schemas[i].Name = prefix + r.Schemas[i].Name + // HTTP API for schemas doesn't yet support tags. It's only supported in + // the Databricks UI and via the SQL API. + } + + return nil +} + +func validatePauseStatus(b *bundle.Bundle) diag.Diagnostics { + p := b.Config.Presets.TriggerPauseStatus + if p == "" || p == config.Paused || p == config.Unpaused { + return nil + } + return diag.Diagnostics{{ + Summary: "Invalid value for trigger_pause_status, should be PAUSED or UNPAUSED", + Severity: diag.Error, + Locations: []dyn.Location{b.Config.GetLocation("presets.trigger_pause_status")}, + }} +} + +// toTagArray converts a map of tags to an array of tags. +// We sort tags so ensure stable ordering. +func toTagArray(tags map[string]string) []Tag { + var tagArray []Tag + if tags == nil { + return tagArray + } + for key, value := range tags { + tagArray = append(tagArray, Tag{Key: key, Value: value}) + } + sort.Slice(tagArray, func(i, j int) bool { + return tagArray[i].Key < tagArray[j].Key + }) + return tagArray +} + +// normalizePrefix prefixes strings like '[dev lennart] ' to 'dev_lennart_'. +// We leave unicode letters and numbers but remove all "special characters." +func normalizePrefix(prefix string) string { + prefix = strings.ReplaceAll(prefix, "[", "") + prefix = strings.Trim(prefix, " ") + + // If the prefix ends with a ']', we add an underscore to the end. + // This makes sure that we get names like "dev_user_endpoint" instead of "dev_userendpoint" + suffix := "" + if strings.HasSuffix(prefix, "]") { + suffix = "_" + } + + return textutil.NormalizeString(prefix) + suffix +} diff --git a/bundle/config/mutator/apply_presets_test.go b/bundle/config/mutator/apply_presets_test.go new file mode 100644 index 00000000..35dac1f7 --- /dev/null +++ b/bundle/config/mutator/apply_presets_test.go @@ -0,0 +1,196 @@ +package mutator_test + +import ( + "context" + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/mutator" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/require" +) + +func TestApplyPresetsPrefix(t *testing.T) { + tests := []struct { + name string + prefix string + job *resources.Job + want string + }{ + { + name: "add prefix to job", + prefix: "prefix-", + job: &resources.Job{ + JobSettings: &jobs.JobSettings{ + Name: "job1", + }, + }, + want: "prefix-job1", + }, + { + name: "add empty prefix to job", + prefix: "", + job: &resources.Job{ + JobSettings: &jobs.JobSettings{ + Name: "job1", + }, + }, + want: "job1", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := &bundle.Bundle{ + Config: config.Root{ + Resources: config.Resources{ + Jobs: map[string]*resources.Job{ + "job1": tt.job, + }, + }, + Presets: config.Presets{ + NamePrefix: tt.prefix, + }, + }, + } + + ctx := context.Background() + diag := bundle.Apply(ctx, b, mutator.ApplyPresets()) + + if diag.HasError() { + t.Fatalf("unexpected error: %v", diag) + } + + require.Equal(t, tt.want, b.Config.Resources.Jobs["job1"].Name) + }) + } +} + +func TestApplyPresetsTags(t *testing.T) { + tests := []struct { + name string + tags map[string]string + job *resources.Job + want map[string]string + }{ + { + name: "add tags to job", + tags: map[string]string{"env": "dev"}, + job: &resources.Job{ + JobSettings: &jobs.JobSettings{ + Name: "job1", + Tags: nil, + }, + }, + want: map[string]string{"env": "dev"}, + }, + { + name: "merge tags with existing job tags", + tags: map[string]string{"env": "dev"}, + job: &resources.Job{ + JobSettings: &jobs.JobSettings{ + Name: "job1", + Tags: map[string]string{"team": "data"}, + }, + }, + want: map[string]string{"env": "dev", "team": "data"}, + }, + { + name: "don't override existing job tags", + tags: map[string]string{"env": "dev"}, + job: &resources.Job{ + JobSettings: &jobs.JobSettings{ + Name: "job1", + Tags: map[string]string{"env": "prod"}, + }, + }, + want: map[string]string{"env": "prod"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := &bundle.Bundle{ + Config: config.Root{ + Resources: config.Resources{ + Jobs: map[string]*resources.Job{ + "job1": tt.job, + }, + }, + Presets: config.Presets{ + Tags: tt.tags, + }, + }, + } + + ctx := context.Background() + diag := bundle.Apply(ctx, b, mutator.ApplyPresets()) + + if diag.HasError() { + t.Fatalf("unexpected error: %v", diag) + } + + tags := b.Config.Resources.Jobs["job1"].Tags + require.Equal(t, tt.want, tags) + }) + } +} + +func TestApplyPresetsJobsMaxConcurrentRuns(t *testing.T) { + tests := []struct { + name string + job *resources.Job + setting int + want int + }{ + { + name: "set max concurrent runs", + job: &resources.Job{ + JobSettings: &jobs.JobSettings{ + Name: "job1", + MaxConcurrentRuns: 0, + }, + }, + setting: 5, + want: 5, + }, + { + name: "do not override existing max concurrent runs", + job: &resources.Job{ + JobSettings: &jobs.JobSettings{ + Name: "job1", + MaxConcurrentRuns: 3, + }, + }, + setting: 5, + want: 3, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := &bundle.Bundle{ + Config: config.Root{ + Resources: config.Resources{ + Jobs: map[string]*resources.Job{ + "job1": tt.job, + }, + }, + Presets: config.Presets{ + JobsMaxConcurrentRuns: tt.setting, + }, + }, + } + ctx := context.Background() + diag := bundle.Apply(ctx, b, mutator.ApplyPresets()) + + if diag.HasError() { + t.Fatalf("unexpected error: %v", diag) + } + + require.Equal(t, tt.want, b.Config.Resources.Jobs["job1"].MaxConcurrentRuns) + }) + } +} diff --git a/bundle/config/mutator/process_target_mode.go b/bundle/config/mutator/process_target_mode.go index fb71f751..92ed2868 100644 --- a/bundle/config/mutator/process_target_mode.go +++ b/bundle/config/mutator/process_target_mode.go @@ -2,17 +2,14 @@ package mutator import ( "context" - "path" "strings" "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config" "github.com/databricks/cli/libs/auth" "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dyn" "github.com/databricks/cli/libs/log" - "github.com/databricks/databricks-sdk-go/service/catalog" - "github.com/databricks/databricks-sdk-go/service/jobs" - "github.com/databricks/databricks-sdk-go/service/ml" ) type processTargetMode struct{} @@ -30,103 +27,75 @@ func (m *processTargetMode) Name() string { // Mark all resources as being for 'development' purposes, i.e. // changing their their name, adding tags, and (in the future) // marking them as 'hidden' in the UI. -func transformDevelopmentMode(ctx context.Context, b *bundle.Bundle) diag.Diagnostics { +func transformDevelopmentMode(ctx context.Context, b *bundle.Bundle) { if !b.Config.Bundle.Deployment.Lock.IsExplicitlyEnabled() { log.Infof(ctx, "Development mode: disabling deployment lock since bundle.deployment.lock.enabled is not set to true") disabled := false b.Config.Bundle.Deployment.Lock.Enabled = &disabled } - r := b.Config.Resources + t := &b.Config.Presets shortName := b.Config.Workspace.CurrentUser.ShortName - prefix := "[dev " + shortName + "] " - // Generate a normalized version of the short name that can be used as a tag value. - tagValue := b.Tagging.NormalizeValue(shortName) - - for i := range r.Jobs { - r.Jobs[i].Name = prefix + r.Jobs[i].Name - if r.Jobs[i].Tags == nil { - r.Jobs[i].Tags = make(map[string]string) - } - r.Jobs[i].Tags["dev"] = tagValue - if r.Jobs[i].MaxConcurrentRuns == 0 { - r.Jobs[i].MaxConcurrentRuns = developmentConcurrentRuns - } - - // Pause each job. As an exception, we don't pause jobs that are explicitly - // marked as "unpaused". This allows users to override the default behavior - // of the development mode. - if r.Jobs[i].Schedule != nil && r.Jobs[i].Schedule.PauseStatus != jobs.PauseStatusUnpaused { - r.Jobs[i].Schedule.PauseStatus = jobs.PauseStatusPaused - } - if r.Jobs[i].Continuous != nil && r.Jobs[i].Continuous.PauseStatus != jobs.PauseStatusUnpaused { - r.Jobs[i].Continuous.PauseStatus = jobs.PauseStatusPaused - } - if r.Jobs[i].Trigger != nil && r.Jobs[i].Trigger.PauseStatus != jobs.PauseStatusUnpaused { - r.Jobs[i].Trigger.PauseStatus = jobs.PauseStatusPaused - } + if t.NamePrefix == "" { + t.NamePrefix = "[dev " + shortName + "] " } - for i := range r.Pipelines { - r.Pipelines[i].Name = prefix + r.Pipelines[i].Name - r.Pipelines[i].Development = true - r.Pipelines[i].Continuous = false - // (pipelines don't yet support tags) + if t.Tags == nil { + t.Tags = map[string]string{} + } + _, exists := t.Tags["dev"] + if !exists { + t.Tags["dev"] = b.Tagging.NormalizeValue(shortName) } - for i := range r.Models { - r.Models[i].Name = prefix + r.Models[i].Name - r.Models[i].Tags = append(r.Models[i].Tags, ml.ModelTag{Key: "dev", Value: tagValue}) + if t.JobsMaxConcurrentRuns == 0 { + t.JobsMaxConcurrentRuns = developmentConcurrentRuns } - for i := range r.Experiments { - filepath := r.Experiments[i].Name - dir := path.Dir(filepath) - base := path.Base(filepath) - if dir == "." { - r.Experiments[i].Name = prefix + base - } else { - r.Experiments[i].Name = dir + "/" + prefix + base - } - r.Experiments[i].Tags = append(r.Experiments[i].Tags, ml.ExperimentTag{Key: "dev", Value: tagValue}) + if t.TriggerPauseStatus == "" { + t.TriggerPauseStatus = config.Paused } - for i := range r.ModelServingEndpoints { - prefix = "dev_" + b.Config.Workspace.CurrentUser.ShortName + "_" - r.ModelServingEndpoints[i].Name = prefix + r.ModelServingEndpoints[i].Name - // (model serving doesn't yet support tags) + if !config.IsExplicitlyDisabled(t.PipelinesDevelopment) { + enabled := true + t.PipelinesDevelopment = &enabled } - - for i := range r.RegisteredModels { - prefix = "dev_" + b.Config.Workspace.CurrentUser.ShortName + "_" - r.RegisteredModels[i].Name = prefix + r.RegisteredModels[i].Name - // (registered models in Unity Catalog don't yet support tags) - } - - for i := range r.QualityMonitors { - // Remove all schedules from monitors, since they don't support pausing/unpausing. - // Quality monitors might support the "pause" property in the future, so at the - // CLI level we do respect that property if it is set to "unpaused". - if r.QualityMonitors[i].Schedule != nil && r.QualityMonitors[i].Schedule.PauseStatus != catalog.MonitorCronSchedulePauseStatusUnpaused { - r.QualityMonitors[i].Schedule = nil - } - } - - for i := range r.Schemas { - prefix = "dev_" + b.Config.Workspace.CurrentUser.ShortName + "_" - r.Schemas[i].Name = prefix + r.Schemas[i].Name - // HTTP API for schemas doesn't yet support tags. It's only supported in - // the Databricks UI and via the SQL API. - } - - return nil } func validateDevelopmentMode(b *bundle.Bundle) diag.Diagnostics { + p := b.Config.Presets + u := b.Config.Workspace.CurrentUser + + // Make sure presets don't set the trigger status to UNPAUSED; + // this could be surprising since most users (and tools) expect triggers + // to be paused in development. + // (Note that there still is an exceptional case where users set the trigger + // status to UNPAUSED at the level of an individual object, whic hwas + // historically allowed.) + if p.TriggerPauseStatus == config.Unpaused { + return diag.Diagnostics{{ + Severity: diag.Error, + Summary: "target with 'mode: development' cannot set trigger pause status to UNPAUSED by default", + Locations: []dyn.Location{b.Config.GetLocation("presets.trigger_pause_status")}, + }} + } + + // Make sure this development copy has unique names and paths to avoid conflicts if path := findNonUserPath(b); path != "" { return diag.Errorf("%s must start with '~/' or contain the current username when using 'mode: development'", path) } + if p.NamePrefix != "" && !strings.Contains(p.NamePrefix, u.ShortName) && !strings.Contains(p.NamePrefix, u.UserName) { + // Resources such as pipelines require a unique name, e.g. '[dev steve] my_pipeline'. + // For this reason we require the name prefix to contain the current username; + // it's a pitfall for users if they don't include it and later find out that + // only a single user can do development deployments. + return diag.Diagnostics{{ + Severity: diag.Error, + Summary: "prefix should contain the current username or ${workspace.current_user.short_name} to ensure uniqueness when using 'mode: development'", + Locations: []dyn.Location{b.Config.GetLocation("presets.name_prefix")}, + }} + } return nil } @@ -183,10 +152,11 @@ func (m *processTargetMode) Apply(ctx context.Context, b *bundle.Bundle) diag.Di switch b.Config.Bundle.Mode { case config.Development: diags := validateDevelopmentMode(b) - if diags != nil { + if diags.HasError() { return diags } - return transformDevelopmentMode(ctx, b) + transformDevelopmentMode(ctx, b) + return diags case config.Production: isPrincipal := auth.IsServicePrincipal(b.Config.Workspace.CurrentUser.UserName) return validateProductionMode(ctx, b, isPrincipal) diff --git a/bundle/config/mutator/process_target_mode_test.go b/bundle/config/mutator/process_target_mode_test.go index 1a2e96fa..1c8671b4 100644 --- a/bundle/config/mutator/process_target_mode_test.go +++ b/bundle/config/mutator/process_target_mode_test.go @@ -9,6 +9,7 @@ import ( "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config" "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/libs/diag" "github.com/databricks/cli/libs/tags" sdkconfig "github.com/databricks/databricks-sdk-go/config" "github.com/databricks/databricks-sdk-go/service/catalog" @@ -51,6 +52,7 @@ func mockBundle(mode config.Mode) *bundle.Bundle { Schedule: &jobs.CronSchedule{ QuartzCronExpression: "* * * * *", }, + Tags: map[string]string{"existing": "tag"}, }, }, "job2": { @@ -129,12 +131,13 @@ func mockBundle(mode config.Mode) *bundle.Bundle { func TestProcessTargetModeDevelopment(t *testing.T) { b := mockBundle(config.Development) - m := ProcessTargetMode() + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) diags := bundle.Apply(context.Background(), b, m) require.NoError(t, diags.Error()) // Job 1 assert.Equal(t, "[dev lennart] job1", b.Config.Resources.Jobs["job1"].Name) + assert.Equal(t, b.Config.Resources.Jobs["job1"].Tags["existing"], "tag") assert.Equal(t, b.Config.Resources.Jobs["job1"].Tags["dev"], "lennart") assert.Equal(t, b.Config.Resources.Jobs["job1"].Schedule.PauseStatus, jobs.PauseStatusPaused) @@ -183,7 +186,8 @@ func TestProcessTargetModeDevelopmentTagNormalizationForAws(t *testing.T) { }) b.Config.Workspace.CurrentUser.ShortName = "Héllö wörld?!" - diags := bundle.Apply(context.Background(), b, ProcessTargetMode()) + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) require.NoError(t, diags.Error()) // Assert that tag normalization took place. @@ -197,7 +201,8 @@ func TestProcessTargetModeDevelopmentTagNormalizationForAzure(t *testing.T) { }) b.Config.Workspace.CurrentUser.ShortName = "Héllö wörld?!" - diags := bundle.Apply(context.Background(), b, ProcessTargetMode()) + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) require.NoError(t, diags.Error()) // Assert that tag normalization took place (Azure allows more characters than AWS). @@ -211,17 +216,53 @@ func TestProcessTargetModeDevelopmentTagNormalizationForGcp(t *testing.T) { }) b.Config.Workspace.CurrentUser.ShortName = "Héllö wörld?!" - diags := bundle.Apply(context.Background(), b, ProcessTargetMode()) + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) require.NoError(t, diags.Error()) // Assert that tag normalization took place. assert.Equal(t, "Hello_world", b.Config.Resources.Jobs["job1"].Tags["dev"]) } +func TestValidateDevelopmentMode(t *testing.T) { + // Test with a valid development mode bundle + b := mockBundle(config.Development) + diags := validateDevelopmentMode(b) + require.NoError(t, diags.Error()) + + // Test with a bundle that has a non-user path + b.Config.Workspace.RootPath = "/Shared/.bundle/x/y/state" + diags = validateDevelopmentMode(b) + require.ErrorContains(t, diags.Error(), "root_path") + + // Test with a bundle that has an unpaused trigger pause status + b = mockBundle(config.Development) + b.Config.Presets.TriggerPauseStatus = config.Unpaused + diags = validateDevelopmentMode(b) + require.ErrorContains(t, diags.Error(), "UNPAUSED") + + // Test with a bundle that has a prefix not containing the username or short name + b = mockBundle(config.Development) + b.Config.Presets.NamePrefix = "[prod]" + diags = validateDevelopmentMode(b) + require.Len(t, diags, 1) + assert.Equal(t, diag.Error, diags[0].Severity) + assert.Contains(t, diags[0].Summary, "") + + // Test with a bundle that has valid user paths + b = mockBundle(config.Development) + b.Config.Workspace.RootPath = "/Users/lennart@company.com/.bundle/x/y/state" + b.Config.Workspace.StatePath = "/Users/lennart@company.com/.bundle/x/y/state" + b.Config.Workspace.FilePath = "/Users/lennart@company.com/.bundle/x/y/files" + b.Config.Workspace.ArtifactPath = "/Users/lennart@company.com/.bundle/x/y/artifacts" + diags = validateDevelopmentMode(b) + require.NoError(t, diags.Error()) +} + func TestProcessTargetModeDefault(t *testing.T) { b := mockBundle("") - m := ProcessTargetMode() + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) diags := bundle.Apply(context.Background(), b, m) require.NoError(t, diags.Error()) assert.Equal(t, "job1", b.Config.Resources.Jobs["job1"].Name) @@ -307,7 +348,7 @@ func TestAllResourcesMocked(t *testing.T) { func TestAllResourcesRenamed(t *testing.T) { b := mockBundle(config.Development) - m := ProcessTargetMode() + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) diags := bundle.Apply(context.Background(), b, m) require.NoError(t, diags.Error()) @@ -337,8 +378,7 @@ func TestDisableLocking(t *testing.T) { ctx := context.Background() b := mockBundle(config.Development) - err := bundle.Apply(ctx, b, ProcessTargetMode()) - require.Nil(t, err) + transformDevelopmentMode(ctx, b) assert.False(t, b.Config.Bundle.Deployment.Lock.IsEnabled()) } @@ -348,7 +388,97 @@ func TestDisableLockingDisabled(t *testing.T) { explicitlyEnabled := true b.Config.Bundle.Deployment.Lock.Enabled = &explicitlyEnabled - err := bundle.Apply(ctx, b, ProcessTargetMode()) - require.Nil(t, err) + transformDevelopmentMode(ctx, b) assert.True(t, b.Config.Bundle.Deployment.Lock.IsEnabled(), "Deployment lock should remain enabled in development mode when explicitly enabled") } + +func TestPrefixAlreadySet(t *testing.T) { + b := mockBundle(config.Development) + b.Config.Presets.NamePrefix = "custom_lennart_deploy_" + + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) + require.NoError(t, diags.Error()) + + assert.Equal(t, "custom_lennart_deploy_job1", b.Config.Resources.Jobs["job1"].Name) +} + +func TestTagsAlreadySet(t *testing.T) { + b := mockBundle(config.Development) + b.Config.Presets.Tags = map[string]string{ + "custom": "tag", + "dev": "foo", + } + + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) + require.NoError(t, diags.Error()) + + assert.Equal(t, "tag", b.Config.Resources.Jobs["job1"].Tags["custom"]) + assert.Equal(t, "foo", b.Config.Resources.Jobs["job1"].Tags["dev"]) +} + +func TestTagsNil(t *testing.T) { + b := mockBundle(config.Development) + b.Config.Presets.Tags = nil + + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) + require.NoError(t, diags.Error()) + + assert.Equal(t, "lennart", b.Config.Resources.Jobs["job2"].Tags["dev"]) +} + +func TestTagsEmptySet(t *testing.T) { + b := mockBundle(config.Development) + b.Config.Presets.Tags = map[string]string{} + + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) + require.NoError(t, diags.Error()) + + assert.Equal(t, "lennart", b.Config.Resources.Jobs["job2"].Tags["dev"]) +} + +func TestJobsMaxConcurrentRunsAlreadySet(t *testing.T) { + b := mockBundle(config.Development) + b.Config.Presets.JobsMaxConcurrentRuns = 10 + + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) + require.NoError(t, diags.Error()) + + assert.Equal(t, 10, b.Config.Resources.Jobs["job1"].MaxConcurrentRuns) +} + +func TestJobsMaxConcurrentRunsDisabled(t *testing.T) { + b := mockBundle(config.Development) + b.Config.Presets.JobsMaxConcurrentRuns = 1 + + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) + require.NoError(t, diags.Error()) + + assert.Equal(t, 1, b.Config.Resources.Jobs["job1"].MaxConcurrentRuns) +} + +func TestTriggerPauseStatusWhenUnpaused(t *testing.T) { + b := mockBundle(config.Development) + b.Config.Presets.TriggerPauseStatus = config.Unpaused + + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) + require.ErrorContains(t, diags.Error(), "target with 'mode: development' cannot set trigger pause status to UNPAUSED by default") +} + +func TestPipelinesDevelopmentDisabled(t *testing.T) { + b := mockBundle(config.Development) + notEnabled := false + b.Config.Presets.PipelinesDevelopment = ¬Enabled + + m := bundle.Seq(ProcessTargetMode(), ApplyPresets()) + diags := bundle.Apply(context.Background(), b, m) + require.NoError(t, diags.Error()) + + assert.False(t, b.Config.Resources.Pipelines["pipeline1"].PipelineSpec.Development) +} diff --git a/bundle/config/presets.go b/bundle/config/presets.go new file mode 100644 index 00000000..61009a25 --- /dev/null +++ b/bundle/config/presets.go @@ -0,0 +1,32 @@ +package config + +const Paused = "PAUSED" +const Unpaused = "UNPAUSED" + +type Presets struct { + // NamePrefix to prepend to all resource names. + NamePrefix string `json:"name_prefix,omitempty"` + + // PipelinesDevelopment is the default value for the development field of pipelines. + PipelinesDevelopment *bool `json:"pipelines_development,omitempty"` + + // TriggerPauseStatus is the default value for the pause status of all triggers and schedules. + // Either config.Paused, config.Unpaused, or empty. + TriggerPauseStatus string `json:"trigger_pause_status,omitempty"` + + // JobsMaxConcurrentRuns is the default value for the max concurrent runs of jobs. + JobsMaxConcurrentRuns int `json:"jobs_max_concurrent_runs,omitempty"` + + // Tags to add to all resources. + Tags map[string]string `json:"tags,omitempty"` +} + +// IsExplicitlyEnabled tests whether this feature is explicitly enabled. +func IsExplicitlyEnabled(feature *bool) bool { + return feature != nil && *feature +} + +// IsExplicitlyDisabled tests whether this feature is explicitly disabled. +func IsExplicitlyDisabled(feature *bool) bool { + return feature != nil && !*feature +} diff --git a/bundle/config/root.go b/bundle/config/root.go index 2c6fe1a4..86dc3392 100644 --- a/bundle/config/root.go +++ b/bundle/config/root.go @@ -60,6 +60,10 @@ type Root struct { // RunAs section allows to define an execution identity for jobs and pipelines runs RunAs *jobs.JobRunAs `json:"run_as,omitempty"` + // Presets applies preset transformations throughout the bundle, e.g. + // adding a name prefix to deployed resources. + Presets Presets `json:"presets,omitempty"` + Experimental *Experimental `json:"experimental,omitempty"` // Permissions section allows to define permissions which will be @@ -307,6 +311,7 @@ func (r *Root) MergeTargetOverrides(name string) error { "resources", "sync", "permissions", + "presets", } { if root, err = mergeField(root, target, f); err != nil { return err diff --git a/bundle/config/target.go b/bundle/config/target.go index acc49357..a2ef4d73 100644 --- a/bundle/config/target.go +++ b/bundle/config/target.go @@ -20,6 +20,10 @@ type Target struct { // development purposes. Mode Mode `json:"mode,omitempty"` + // Mutator configurations that e.g. change the + // name prefix of deployed resources. + Presets Presets `json:"presets,omitempty"` + // Overrides the compute used for jobs and other supported assets. ComputeID string `json:"compute_id,omitempty"` diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go index fac3066d..7a1081de 100644 --- a/bundle/phases/initialize.go +++ b/bundle/phases/initialize.go @@ -47,6 +47,7 @@ func Initialize() bundle.Mutator { mutator.SetRunAs(), mutator.OverrideCompute(), mutator.ProcessTargetMode(), + mutator.ApplyPresets(), mutator.DefaultQueueing(), mutator.ExpandPipelineGlobPaths(), diff --git a/bundle/tests/presets/databricks.yml b/bundle/tests/presets/databricks.yml new file mode 100644 index 00000000..d83d3180 --- /dev/null +++ b/bundle/tests/presets/databricks.yml @@ -0,0 +1,22 @@ +bundle: + name: presets + +presets: + tags: + prod: true + team: finance + pipelines_development: true + +targets: + dev: + presets: + name_prefix: "myprefix" + pipelines_development: true + trigger_pause_status: PAUSED + jobs_max_concurrent_runs: 10 + tags: + dev: true + prod: false + prod: + presets: + pipelines_development: false diff --git a/bundle/tests/presets_test.go b/bundle/tests/presets_test.go new file mode 100644 index 00000000..5fcb5d95 --- /dev/null +++ b/bundle/tests/presets_test.go @@ -0,0 +1,28 @@ +package config_tests + +import ( + "testing" + + "github.com/databricks/cli/bundle/config" + "github.com/stretchr/testify/assert" +) + +func TestPresetsDev(t *testing.T) { + b := loadTarget(t, "./presets", "dev") + + assert.Equal(t, "myprefix", b.Config.Presets.NamePrefix) + assert.Equal(t, config.Paused, b.Config.Presets.TriggerPauseStatus) + assert.Equal(t, 10, b.Config.Presets.JobsMaxConcurrentRuns) + assert.Equal(t, true, *b.Config.Presets.PipelinesDevelopment) + assert.Equal(t, "true", b.Config.Presets.Tags["dev"]) + assert.Equal(t, "finance", b.Config.Presets.Tags["team"]) + assert.Equal(t, "false", b.Config.Presets.Tags["prod"]) +} + +func TestPresetsProd(t *testing.T) { + b := loadTarget(t, "./presets", "prod") + + assert.Equal(t, false, *b.Config.Presets.PipelinesDevelopment) + assert.Equal(t, "finance", b.Config.Presets.Tags["team"]) + assert.Equal(t, "true", b.Config.Presets.Tags["prod"]) +}