[DCOS-38138] Update Spark CLI for shell-escape fix #388
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -6,20 +6,30 @@ import ( | |
| "encoding/json" | ||
| "errors" | ||
| "fmt" | ||
| "log" | ||
| "net/url" | ||
| "os" | ||
| "regexp" | ||
| "strconv" | ||
| "strings" | ||
|
|
||
| "github.com/mattn/go-shellwords" | ||
| "github.com/mesosphere/dcos-commons/cli/client" | ||
| "github.com/mesosphere/dcos-commons/cli/config" | ||
| "gopkg.in/alecthomas/kingpin.v3-unstable" | ||
| ) | ||
|
|
||
| var keyWhitespaceValPattern = regexp.MustCompile("(.+)\\s+(.+)") | ||
| var backslashNewlinePattern = regexp.MustCompile("\\s*\\\\s*\\n\\s+") | ||
| var collapseSpacesPattern = regexp.MustCompile(`[\s\p{Zs}]{2,}`) | ||
|
|
||
| // AcceptedSparkAppExtensions lists the accepted extensions for Spark application artifacts | ||
| var AcceptedSparkAppExtensions = []string{ | ||
| ".jar", | ||
| ".py", | ||
| ".R", | ||
| } | ||
|
|
||
| type sparkVal struct { | ||
| flagName string | ||
| propName string | ||
|
|
@@ -146,8 +156,10 @@ Args: | |
| StringVar(&args.mainClass) // note: spark-submit can autodetect, but only for file://local.jar | ||
| submit.Flag("properties-file", "Path to file containing whitespace-separated Spark property defaults."). | ||
| PlaceHolder("PATH").ExistingFileVar(&args.propertiesFile) | ||
| submit.Flag("conf", "Custom Spark configuration properties. "+ | ||
| "If submitting properties with multiple values, "+ | ||
| "wrap in single quotes e.g. --conf prop='val1 val2'"). | ||
| PlaceHolder("prop=value").StringMapVar(&args.properties) | ||
| submit.Flag("kerberos-principal", "Principal to be used to login to KDC."). | ||
| PlaceHolder("user@REALM").Default("").StringVar(&args.kerberosPrincipal) | ||
| submit.Flag("keytab-secret-path", "Path to Keytab in secret store to be used in the Spark drivers"). | ||
|
|
@@ -280,75 +292,87 @@ func parseApplicationFile(args *sparkArgs) error { | |
| return nil | ||
| } | ||
|
|
||
| // we use Kingpin to parse CLI commands and options | ||
| // spark-submit by convention uses '--arg val' while kingpin only supports --arg=val | ||
|
Contributor
There was a problem hiding this comment. Could we create a ticket to address this. Kingpin supports both
Contributor
Author
There was a problem hiding this comment. here you go: https://jira.mesosphere.com/browse/DCOS-41107 |
||
| // cleanUpSubmitArgs transforms the former to the latter | ||
| func transformSubmitArgs(argsStr string, boolVals []*sparkVal) ([]string, []string) { | ||
| // clean up any instances of shell-style escaped newlines: "arg1\\narg2" => "arg1 arg2" | ||
| argsStr = strings.TrimSpace(backslashNewlinePattern.ReplaceAllLiteralString(argsStr, " ")) | ||
| // collapse two or more spaces to one | ||
| argsStr = collapseSpacesPattern.ReplaceAllString(argsStr, " ") | ||
| // parse argsStr into []string args maintaining shell escaped sequences | ||
| args, err := shellwords.Parse(argsStr) | ||
| if err != nil { | ||
| log.Fatalf("Could not parse string args correctly. Error: %v", err) | ||
| } | ||
| sparkArgs, appArgs := make([]string, 0), make([]string, 0) | ||
| LOOP: | ||
| for i := 0; i < len(args); { | ||
| current := strings.TrimSpace(args[i]) | ||
| switch { | ||
| // The main assumption with --submit-args is that all spark-submit flags come before the spark jar URL | ||
| // if current is a spark jar/app, we've processed all flags | ||
| case isSparkApp(current): | ||
| sparkArgs = append(sparkArgs, args[i]) | ||
| appArgs = append(appArgs, args[i+1:]...) | ||
| break LOOP | ||
| case strings.HasPrefix(current, "--"): | ||
| isBool, skipNext := checkIfBooleanFlag(boolVals, args, i) | ||
| if isBool { | ||
| sparkArgs = append(sparkArgs, current) | ||
| i++ | ||
| if skipNext { | ||
| i++ | ||
| } | ||
| continue LOOP | ||
| } | ||
| if strings.Contains(current, "=") { | ||
| // already in the form arg=val, no merge required | ||
| sparkArgs = append(sparkArgs, current) | ||
| i++ | ||
| continue LOOP | ||
| } | ||
| // otherwise, merge current with next into form arg=val; eg --driver-memory=512m | ||
| next := args[i+1] | ||
| sparkArgs = append(sparkArgs, current+"="+next) | ||
| i += 2 | ||
| default: | ||
| // if not a flag or jar, current is a continuation of the last arg and should not have been split | ||
| // eg extraJavaOptions="-Dparam1 -Dparam2" was parsed as [extraJavaOptions, -Dparam1, -Dparam2] | ||
|
Contributor
There was a problem hiding this comment. Oh, this isn't handled by shellwords? (It sounded like it might based on the README.)
Contributor
Author
There was a problem hiding this comment. No unfortunately the library splits the string on non-escaped spaces so this |
||
| combined := sparkArgs[len(sparkArgs)-1] + " " + current | ||
| sparkArgs = append(sparkArgs[:len(sparkArgs)-1], combined) | ||
| i++ | ||
| } | ||
| } | ||
| if config.Verbose { | ||
| client.PrintVerbose("Translated spark-submit arguments: '%s'", strings.Join(sparkArgs, ", ")) | ||
| client.PrintVerbose("Translated application arguments: '%s'", strings.Join(appArgs, ", ")) | ||
| } | ||
| return sparkArgs, appArgs | ||
| } | ||
|
|
||
| func isSparkApp(str string) bool { | ||
| for _, ext := range AcceptedSparkAppExtensions { | ||
| if strings.HasSuffix(str, ext) { | ||
| return true | ||
| } | ||
| } | ||
| return false | ||
| } | ||
|
|
||
| func checkIfBooleanFlag(boolVals []*sparkVal, args []string, idx int) (isBool, skipNext bool) { | ||
| // check if current is a boolean flag (eg --supervise) but skip any accompanying boolean values (eg --supervise true) | ||
| current := args[idx] | ||
| for _, boolVal := range boolVals { | ||
|
Contributor
There was a problem hiding this comment. Looking at this again this morning. We should break out of this loop as soon as we have found the matching flag -- either by returning, or by calling |
||
| if boolVal.flagName == current[2:] { | ||
| isBool = true | ||
| next := args[idx+1] | ||
| if _, err := strconv.ParseBool(next); err == nil { | ||
| skipNext = true | ||
|
||
| } | ||
| } | ||
| } | ||
| return | ||
| } | ||
|
|
||
| func getValsFromPropertiesFile(path string) map[string]string { | ||
|
|
@@ -416,7 +440,7 @@ func buildSubmitJson(cmd *SparkCommand, marathonConfig map[string]interface{}) ( | |
| // then map flags | ||
| submit, args := sparkSubmitArgSetup() // setup | ||
| // convert and get application flags, add them to the args passed to the spark app | ||
| submitArgs, appFlags := transformSubmitArgs(cmd.submitArgs, args.boolVals) | ||
| args.appArgs = append(args.appArgs, appFlags...) | ||
| _, err := submit.Parse(submitArgs) | ||
|
|
||
|
|
@@ -509,7 +533,7 @@ func buildSubmitJson(cmd *SparkCommand, marathonConfig map[string]interface{}) ( | |
| } else { | ||
| client.PrintMessage("Using image '%s' for the driver and the executors (from %s).", | ||
| args.properties["spark.mesos.executor.docker.image"], imageSource) | ||
| client.PrintMessage("To disable this image on executors, set " + | ||
| "spark.mesos.executor.docker.forcePullImage=false") | ||
| args.properties["spark.mesos.executor.docker.forcePullImage"] = "true" | ||
| } | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does this have to be public? Also, we could make the list local to the only function where it is used.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
sorry had a Scala slip of the brain there, forgot caps make go code public. Was aiming to denote it as a constant 🙃
fixed in 7b2bb6c