databricks-cli/bundle/python/warning.go

102 lines
2.6 KiB
Go

package python
import (
"context"
"strings"
"github.com/databricks/cli/bundle"
"github.com/databricks/cli/bundle/libraries"
"github.com/databricks/cli/libs/diag"
"github.com/databricks/cli/libs/log"
"github.com/databricks/databricks-sdk-go"
"golang.org/x/mod/semver"
)
type wrapperWarning struct {
}
func WrapperWarning() bundle.Mutator {
return &wrapperWarning{}
}
func (m *wrapperWarning) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
if isPythonWheelWrapperOn(b) {
return nil
}
if hasIncompatibleWheelTasks(ctx, b) {
return diag.Errorf("python wheel tasks with local libraries require compute with DBR 13.1+. Please change your cluster configuration or set experimental 'python_wheel_wrapper' setting to 'true'")
}
return nil
}
func isPythonWheelWrapperOn(b *bundle.Bundle) bool {
return b.Config.Experimental != nil && b.Config.Experimental.PythonWheelWrapper
}
func hasIncompatibleWheelTasks(ctx context.Context, b *bundle.Bundle) bool {
tasks := libraries.FindAllWheelTasksWithLocalLibraries(b)
for _, task := range tasks {
if task.NewCluster != nil {
if lowerThanExpectedVersion(ctx, task.NewCluster.SparkVersion) {
return true
}
}
if task.JobClusterKey != "" {
for _, job := range b.Config.Resources.Jobs {
for _, cluster := range job.JobClusters {
if task.JobClusterKey == cluster.JobClusterKey && cluster.NewCluster != nil {
if lowerThanExpectedVersion(ctx, cluster.NewCluster.SparkVersion) {
return true
}
}
}
}
}
if task.ExistingClusterId != "" {
version, err := getSparkVersionForCluster(ctx, b.WorkspaceClient(), task.ExistingClusterId)
// If there's error getting spark version for cluster, do not mark it as incompatible
if err != nil {
log.Warnf(ctx, "unable to get spark version for cluster %s, err: %s", task.ExistingClusterId, err.Error())
return false
}
if lowerThanExpectedVersion(ctx, version) {
return true
}
}
}
return false
}
func lowerThanExpectedVersion(ctx context.Context, sparkVersion string) bool {
parts := strings.Split(sparkVersion, ".")
if len(parts) < 2 {
return false
}
if parts[1][0] == 'x' { // treat versions like 13.x as the very latest minor (13.99)
parts[1] = "99"
}
v := "v" + parts[0] + "." + parts[1]
return semver.Compare(v, "v13.1") < 0
}
// Name implements bundle.Mutator.
func (m *wrapperWarning) Name() string {
return "PythonWrapperWarning"
}
func getSparkVersionForCluster(ctx context.Context, w *databricks.WorkspaceClient, clusterId string) (string, error) {
details, err := w.Clusters.GetByClusterId(ctx, clusterId)
if err != nil {
return "", err
}
return details.SparkVersion, nil
}