Merge branch 'master' of github.com:ovh/data-processing-spark-submit

Author: David Morin

data_processing.go

@@ -61,7 +61,7 @@ type (
 		EngineVersion    string                `json:"engineVersion"`
 		EngineParameters []*JobEngineParameter `json:"engineParameters"`
 		Status           string                `json:"status"`
-		Ttl              string                `json:"ttl"`
+		TTL              string                `json:"ttl"`
 	}
 
 	// JobEngineParameter representation of JobEngineParameter in OVH API
@@ -90,7 +90,7 @@ type (
 		Engine           string                `json:"engine"`
 		Name             string                `json:"name"`
 		Region           string                `json:"region"`
-		Ttl              string                `json:"ttl,omitempty"`
+		TTL              string                `json:"ttl,omitempty"`
 		EngineVersion    string                `json:"engineVersion"`
 		EngineParameters []*JobEngineParameter `json:"engineParameters"`
 	}
@@ -141,3 +141,21 @@ func (c *Client) Kill(projectID string, jobID string) error {
 	path := fmt.Sprintf(DataProcessingStatus, url.QueryEscape(projectID), url.QueryEscape(jobID))
 	return c.OVH.Delete(path, nil)
 }
+
+// GetErrorDetails return the error details as a formatted string
+func GetErrorDetails(err *ovh.APIError) string {
+	if len(err.Details) == 0 {
+		return ""
+	}
+	details := "{"
+	first := true
+	for key, value := range err.Details {
+		if first {
+			first = false
+		} else {
+			details += ", "
+		}
+		details = fmt.Sprintf("%s \"%s\": \"%s\"", details, key, value)
+	}
+	return details + " }"
+}

go.mod

@@ -8,8 +8,7 @@ require (
 	github.com/dustin/go-humanize v1.0.0
 	github.com/gabriel-vasile/mimetype v1.1.0
 	github.com/ncw/swift v1.0.52
-	github.com/ovh/go-ovh v0.0.0-20181109152953-ba5adb4cf014
+	github.com/ovh/go-ovh v1.1.1-0.20211209132054-5bcee91ddcd5
 	github.com/peterhellberg/duration v0.0.0-20191119133758-ec6baeebcd10
-	github.com/smartystreets/goconvey v1.6.4 // indirect
-	gopkg.in/ini.v1 v1.55.0
+	gopkg.in/ini.v1 v1.57.0
 )

go.sum

@@ -16,8 +16,8 @@ github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/ncw/swift v1.0.52 h1:ACF3JufDGgeKp/9mrDgQlEgS8kRYC4XKcuzj/8EJjQU=
 github.com/ncw/swift v1.0.52/go.mod h1:23YIA4yWVnGwv2dQlN4bB7egfYX6YLn0Yo/S6zZO/ZM=
-github.com/ovh/go-ovh v0.0.0-20181109152953-ba5adb4cf014 h1:37VE5TYj2m/FLA9SNr4z0+A0JefvTmR60Zwf8XSEV7c=
-github.com/ovh/go-ovh v0.0.0-20181109152953-ba5adb4cf014/go.mod h1:joRatxRJaZBsY3JAOEMcoOp05CnZzsx4scTxi95DHyQ=
+github.com/ovh/go-ovh v1.1.1-0.20211209132054-5bcee91ddcd5 h1:FKZON4nzuqn7tAyvIxMJKRke1ZXNOU8CRiMc6OX2caQ=
+github.com/ovh/go-ovh v1.1.1-0.20211209132054-5bcee91ddcd5/go.mod h1:AxitLZ5HBRPyUd+Zl60Ajaag+rNTdVXWIkzfrVuTXWA=
 github.com/peterhellberg/duration v0.0.0-20191119133758-ec6baeebcd10 h1:Jf08dx6hxr6aNpHzUmYitsKGm6BmCFbwDGPb27/Boyc=
 github.com/peterhellberg/duration v0.0.0-20191119133758-ec6baeebcd10/go.mod h1:x5xjkH61fUOJVgCCDgqNzlJvdLXiYpmMzSuum2FBOaw=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -34,5 +34,5 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
-gopkg.in/ini.v1 v1.55.0 h1:E8yzL5unfpW3M6fz/eB7Cb5MQAYSZ7GKo4Qth+N2sgQ=
-gopkg.in/ini.v1 v1.55.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
+gopkg.in/ini.v1 v1.57.0 h1:9unxIsFcTt4I55uWluz+UmL95q4kdJ0buvQ1ZIqVQww=
+gopkg.in/ini.v1 v1.57.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=

submit.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"errors"
 	"fmt"
 	"log"
 	"os"
@@ -16,13 +17,15 @@ import (
 
 	"github.com/Pallinder/go-randomdata"
 	"github.com/alexflint/go-arg"
-	"github.com/peterhellberg/duration"
 	"github.com/ovh/go-ovh/ovh"
+	"github.com/peterhellberg/duration"
 	"gopkg.in/ini.v1"
 )
 
 const (
 	LoopWaitSecond = 2
+	OVHConfig      = "ovh"
+	SwiftConfig    = "swift"
 )
 
 var (
@@ -43,14 +46,15 @@ var (
 		Packages               string   `arg:"--packages" help:"Comma-delimited list of Maven coordinates"`
 		Repositories           string   `arg:"--repositories" help:"Comma-delimited list of additional repositories (or resolvers in SBT)"`
 		PropertiesFile         string   `arg:"--properties-file" help:"Read properties from the given file"`
-		Ttl                    string   `arg:"--ttl" help:"Maximum \"Time To Live\" (in RFC3339 (duration) eg. "P1DT30H4S") of this job, after which it will be automatically terminated"`
+		TTL                    string   `arg:"--ttl" help:"Maximum \"Time To Live\" (in RFC3339 (duration) eg. \"P1DT30H4S\") of this job, after which it will be automatically terminated"`
 		File                   string   `arg:"positional,required"`
 		Parameters             []string `arg:"positional"`
 	}
 )
 
 var (
-	configPath = "configuration.ini"
+	configPath         = "configuration.ini"
+	SupportedProtocols = []string{SwiftConfig}
 )
 
 type (
@@ -73,6 +77,11 @@ func main() {
 		log.Fatalf("Unable to load conf: %s", err)
 	}
 
+	protocols, err := validConfig(conf)
+	if err != nil {
+		log.Fatalf("Invalid conf: %s", err)
+	}
+
 	ovhConf := new(OVHConf)
 	if err := conf["ovh"].MapTo(ovhConf); err != nil {
 		log.Fatalf("Unable to parse \"ovh\" conf: %s", err)
@@ -93,20 +102,24 @@ func main() {
 		splitFile := strings.Split(args.File, "/")
 		protocol := strings.TrimSuffix(splitFile[0], ":")
 		containerName := splitFile[1]
-		storage, err := upload.New(conf[protocol], protocol)
-		if err != nil {
-			log.Fatalf("Error while Initialise Upload Storage: %s", err)
-		}
-
-		if storage == nil {
-			log.Fatalf("No configuration found for protocol %s", protocol)
-		}
-		filesList := strings.Split(args.Upload, ",")
-		for _, file := range filesList {
-			err = storage.Upload(file, containerName)
+		if inTheList(protocol, protocols) {
+			storage, err := upload.New(conf[protocol], protocol)
 			if err != nil {
-				log.Fatalf("Error while uploading file(s): %s", err)
+				log.Fatalf("Error while initializing upload storage configurations: %s", err)
+			}
+
+			if storage == nil {
+				log.Fatalf("No configuration found for protocol %s", protocol)
+			}
+			filesList := strings.Split(args.Upload, ",")
+			for _, file := range filesList {
+				err = storage.Upload(file, containerName)
+				if err != nil {
+					log.Fatalf("Error while uploading file(s): %s", err)
+				}
 			}
+		} else {
+			log.Fatalf("Error while initializing upload storage configurations: protocol %s isn't configured in %s or isn't supported", protocol, configPath)
 		}
 	}
 
@@ -117,6 +130,16 @@ func main() {
 
 	job, err := client.Submit(args.ProjectID, jobSubmitValue)
 	if err != nil {
+		if ovherr, ok := err.(*ovh.APIError); ok {
+			if err.Error() == "Error 422: \"Unprocessable Entity\"" {
+				log.Fatalf("Unable to submit job: %s :: %s :: %v. "+
+					"Please check that your requested job complies with the OVHcloud Data Processing capabilities "+
+					"(https://docs.ovh.com/gb/en/data-processing/capabilities/#the-apache-spark-job-in-data-processing-is-limited-to)", err, ovherr.Class, GetErrorDetails(ovherr))
+			} else {
+				log.Fatalf("Unable to submit job: %s :: %s :: %v.", err, ovherr.Class, GetErrorDetails(ovherr))
+			}
+		}
+
 		log.Fatalf("Unable to submit job: %s", err)
 	}
 
@@ -159,12 +182,13 @@ func ParsArgs() *JobSubmit {
 		Region:           args.Region,
 		EngineVersion:    args.SparkVersion,
 		EngineParameters: []*JobEngineParameter{},
-		Ttl:              args.Ttl,
+		TTL:              args.TTL,
 	}
 
 	if args.JobName == "" {
 		jobSubmit.Name = randomdata.SillyName()
 	}
+
 	if strings.EqualFold(filepath.Ext(args.File), ".jar") {
 		if args.Class == "" {
 			p.Fail("You must provide --class when using jar file")
@@ -258,12 +282,12 @@ func ParsArgs() *JobSubmit {
 		})
 	}
 
-	if args.Ttl != "" {
-        _, err := duration.Parse(args.Ttl);
-        if err != nil {
-            p.Fail("Invalid value for --ttl. It must be in RFC3339 (duration) format (i.e. PT30H for 30 hours)")
-        }
-    }
+	if args.TTL != "" {
+		_, err := duration.Parse(args.TTL)
+		if err != nil {
+			p.Fail("Invalid value for --ttl. It must be in RFC3339 (duration) format (i.e. PT30H for 30 hours)")
+		}
+	}
 
 	jobSubmit.EngineParameters = append(jobSubmit.EngineParameters, &JobEngineParameter{
 		Name:  ParameterExecutorNumber,
@@ -380,3 +404,31 @@ func PrintLog(jobLog []*Log) (lastPrintLog uint64) {
 	}
 	return lastPrintLog
 }
+
+// test if a value is in the given list
+func inTheList(value string, values []string) bool {
+	for _, element := range values {
+		if value == element {
+			return true
+		}
+	}
+	return false
+}
+
+// test if the given configurations are valid and list the protocols configured
+func validConfig(configSections map[string]*ini.Section) ([]string, error) {
+	var confList []string
+	var protocolsList []string
+	for name, _ := range configSections {
+		confList = append(confList, name)
+		if inTheList(name, SupportedProtocols) {
+			protocolsList = append(protocolsList, name)
+		}
+	}
+
+	if !inTheList(OVHConfig, confList) {
+		return protocolsList, errors.New("missing [ovh] configurations in " + configPath)
+	}
+
+	return protocolsList, nil
+}

submit_test.go

@@ -28,7 +28,7 @@ func TestParsArgsJava(t *testing.T) {
 		t.Fail()
 	}
 
-	if job.Ttl != "" {
+	if job.TTL != "" {
 		t.Fail()
 	}
 
@@ -90,7 +90,7 @@ func TestParsArgsPython(t *testing.T) {
 		t.Fail()
 	}
 
-	if job.Ttl != "" {
+	if job.TTL != "" {
 		t.Fail()
 	}
 
@@ -152,7 +152,7 @@ func TestParsArgs3(t *testing.T) {
 		t.Fail()
 	}
 
-	if job.Ttl != "P1DT30H4S" {
+	if job.TTL != "P1DT30H4S" {
 		t.Fail()
 	}