TaskAPI (#1717)

Co-authored-by: Robert Lucian Chiriac <robert.lucian.chiriac@gmail.com>

Author: Miguel Varela Ramos

build/build-image.sh

@@ -33,7 +33,7 @@ if [[ "$image" == *"-slim" ]]; then
   build_args="--build-arg SLIM=true"
 fi
 
-if [ "${image}" == "python-predictor-gpu-slim" ]; then
+if [ "${image}" == *"gpu-slim" ]; then
   cuda=("10.0" "10.1" "10.1" "10.2" "10.2" "11.0" "11.1")
   cudnn=("7" "7" "8" "7" "8" "8" "8")
   for i in ${!cudnn[@]}; do

cli/cluster/delete.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cortexlabs/cortex/pkg/lib/prompt"
 	s "github.com/cortexlabs/cortex/pkg/lib/strings"
 	"github.com/cortexlabs/cortex/pkg/operator/schema"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
 )
 
 func Delete(operatorConfig OperatorConfig, apiName string, keepCache bool, force bool) (schema.DeleteResponse, error) {
@@ -73,13 +74,20 @@ func getReadyRealtimeAPIReplicasOrNil(operatorConfig OperatorConfig, apiName str
 	return &totalReady
 }
 
-func StopJob(operatorConfig OperatorConfig, apiName string, jobID string) (schema.DeleteResponse, error) {
+func StopJob(operatorConfig OperatorConfig, kind userconfig.Kind, apiName string, jobID string) (schema.DeleteResponse, error) {
 	params := map[string]string{
 		"apiName": apiName,
 		"jobID":   jobID,
 	}
 
-	httpRes, err := HTTPDelete(operatorConfig, path.Join("/batch", apiName), params)
+	var endpointComponent string
+	if kind == userconfig.BatchAPIKind {
+		endpointComponent = "batch"
+	} else {
+		endpointComponent = "tasks"
+	}
+
+	httpRes, err := HTTPDelete(operatorConfig, path.Join("/"+endpointComponent, apiName), params)
 	if err != nil {
 		return schema.DeleteResponse{}, err
 	}

cli/cluster/get.go

@@ -65,16 +65,31 @@ func GetAPIByID(operatorConfig OperatorConfig, apiName string, apiID string) ([]
 	return apiRes, nil
 }
 
-func GetJob(operatorConfig OperatorConfig, apiName string, jobID string) (schema.JobResponse, error) {
+func GetBatchJob(operatorConfig OperatorConfig, apiName string, jobID string) (schema.BatchJobResponse, error) {
 	endpoint := path.Join("/batch", apiName)
 	httpRes, err := HTTPGet(operatorConfig, endpoint, map[string]string{"jobID": jobID})
 	if err != nil {
-		return schema.JobResponse{}, err
+		return schema.BatchJobResponse{}, err
 	}
 
-	var jobRes schema.JobResponse
+	var jobRes schema.BatchJobResponse
 	if err = json.Unmarshal(httpRes, &jobRes); err != nil {
-		return schema.JobResponse{}, errors.Wrap(err, endpoint, string(httpRes))
+		return schema.BatchJobResponse{}, errors.Wrap(err, endpoint, string(httpRes))
+	}
+
+	return jobRes, nil
+}
+
+func GetTaskJob(operatorConfig OperatorConfig, apiName string, jobID string) (schema.TaskJobResponse, error) {
+	endpoint := path.Join("/tasks", apiName)
+	httpRes, err := HTTPGet(operatorConfig, endpoint, map[string]string{"jobID": jobID})
+	if err != nil {
+		return schema.TaskJobResponse{}, err
+	}
+
+	var jobRes schema.TaskJobResponse
+	if err = json.Unmarshal(httpRes, &jobRes); err != nil {
+		return schema.TaskJobResponse{}, errors.Wrap(err, endpoint, string(httpRes))
 	}
 
 	return jobRes, nil

cli/cluster/logs.go

@@ -42,20 +42,6 @@ func StreamJobLogs(operatorConfig OperatorConfig, apiName string, jobID string)
 	return streamLogs(operatorConfig, "/logs/"+apiName, map[string]string{"jobID": jobID})
 }
 
-func GetGCPLogsURL(operatorConfig OperatorConfig, apiName string) (schema.GCPLogsResponse, error) {
-	httpRes, err := HTTPGet(operatorConfig, "/logs/"+apiName)
-	if err != nil {
-		return schema.GCPLogsResponse{}, err
-	}
-
-	var gcpLogsResponse schema.GCPLogsResponse
-	if err = json.Unmarshal(httpRes, &gcpLogsResponse); err != nil {
-		return schema.GCPLogsResponse{}, errors.Wrap(err, "/logs/"+apiName, string(httpRes))
-	}
-
-	return gcpLogsResponse, nil
-}
-
 func streamLogs(operatorConfig OperatorConfig, path string, qParams ...map[string]string) error {
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt)

cli/cmd/delete.go

@@ -70,7 +70,12 @@ var _deleteCmd = &cobra.Command{
 
 		var deleteResponse schema.DeleteResponse
 		if len(args) == 2 {
-			deleteResponse, err = cluster.StopJob(MustGetOperatorConfig(env.Name), args[0], args[1])
+			apisRes, err := cluster.GetAPI(MustGetOperatorConfig(env.Name), args[0])
+			if err != nil {
+				exit.Error(err)
+			}
+
+			deleteResponse, err = cluster.StopJob(MustGetOperatorConfig(env.Name), apisRes[0].Spec.Kind, args[0], args[1])
 			if err != nil {
 				exit.Error(err)
 			}

cli/cmd/get.go

@@ -127,7 +127,17 @@ var _getCmd = &cobra.Command{
 					return "", err
 				}
 
-				jobTable, err := getJob(env, args[0], args[1])
+				apisRes, err := cluster.GetAPI(MustGetOperatorConfig(envName), args[0])
+				if err != nil {
+					return "", err
+				}
+
+				var jobTable string
+				if apisRes[0].Spec.Kind == userconfig.BatchAPIKind {
+					jobTable, err = getBatchJob(env, args[0], args[1])
+				} else {
+					jobTable, err = getTaskJob(env, args[0], args[1])
+				}
 				if err != nil {
 					return "", err
 				}
@@ -189,6 +199,8 @@ func getAPIsInAllEnvironments() (string, error) {
 	var allRealtimeAPIEnvs []string
 	var allBatchAPIs []schema.APIResponse
 	var allBatchAPIEnvs []string
+	var allTaskAPIs []schema.APIResponse
+	var allTaskAPIEnvs []string
 	var allTrafficSplitters []schema.APIResponse
 	var allTrafficSplitterEnvs []string
 
@@ -219,6 +231,9 @@ func getAPIsInAllEnvironments() (string, error) {
 				case userconfig.RealtimeAPIKind:
 					allRealtimeAPIEnvs = append(allRealtimeAPIEnvs, env.Name)
 					allRealtimeAPIs = append(allRealtimeAPIs, api)
+				case userconfig.TaskAPIKind:
+					allTaskAPIEnvs = append(allTaskAPIEnvs, env.Name)
+					allTaskAPIs = append(allTaskAPIs, api)
 				case userconfig.TrafficSplitterKind:
 					allTrafficSplitterEnvs = append(allTrafficSplitterEnvs, env.Name)
 					allTrafficSplitters = append(allTrafficSplitters, api)
@@ -243,7 +258,7 @@ func getAPIsInAllEnvironments() (string, error) {
 
 	out := ""
 
-	if len(allRealtimeAPIs) == 0 && len(allBatchAPIs) == 0 && len(allTrafficSplitters) == 0 {
+	if len(allRealtimeAPIs) == 0 && len(allBatchAPIs) == 0 && len(allTrafficSplitters) == 0 && len(allTaskAPIs) == 0 {
 		// check if any environments errorred
 		if len(errorsMap) != len(cliConfig.Environments) {
 			if len(errorsMap) == 0 {
@@ -271,20 +286,26 @@ func getAPIsInAllEnvironments() (string, error) {
 			out += t.MustFormat()
 		}
 
-		if len(allRealtimeAPIs) > 0 {
-			t := realtimeAPIsTable(allRealtimeAPIs, allRealtimeAPIEnvs)
-
+		if len(allTaskAPIs) > 0 {
+			t := taskAPIsTable(allTaskAPIs, allTaskAPIEnvs)
 			if len(allBatchAPIs) > 0 {
 				out += "\n"
 			}
+			out += t.MustFormat()
+		}
 
+		if len(allRealtimeAPIs) > 0 {
+			t := realtimeAPIsTable(allRealtimeAPIs, allRealtimeAPIEnvs)
+			if len(allBatchAPIs) > 0 || len(allTaskAPIs) > 0 {
+				out += "\n"
+			}
 			out += t.MustFormat()
 		}
 
 		if len(allTrafficSplitters) > 0 {
 			t := trafficSplitterListTable(allTrafficSplitters, allTrafficSplitterEnvs)
 
-			if len(allRealtimeAPIs) > 0 || len(allBatchAPIs) > 0 {
+			if len(allRealtimeAPIs) > 0 || len(allBatchAPIs) > 0 || len(allTaskAPIs) > 0 {
 				out += "\n"
 			}
 
@@ -319,20 +340,23 @@ func getAPIsByEnv(env cliconfig.Environment, printEnv bool) (string, error) {
 
 	var allRealtimeAPIs []schema.APIResponse
 	var allBatchAPIs []schema.APIResponse
+	var allTaskAPIs []schema.APIResponse
 	var allTrafficSplitters []schema.APIResponse
 
 	for _, api := range apisRes {
 		switch api.Spec.Kind {
 		case userconfig.BatchAPIKind:
 			allBatchAPIs = append(allBatchAPIs, api)
+		case userconfig.TaskAPIKind:
+			allTaskAPIs = append(allTaskAPIs, api)
 		case userconfig.RealtimeAPIKind:
 			allRealtimeAPIs = append(allRealtimeAPIs, api)
 		case userconfig.TrafficSplitterKind:
 			allTrafficSplitters = append(allTrafficSplitters, api)
 		}
 	}
 
-	if len(allRealtimeAPIs) == 0 && len(allBatchAPIs) == 0 && len(allTrafficSplitters) == 0 {
+	if len(allRealtimeAPIs) == 0 && len(allBatchAPIs) == 0 && len(allTaskAPIs) == 0 && len(allTrafficSplitters) == 0 {
 		return console.Bold("no apis are deployed"), nil
 	}
 
@@ -350,6 +374,22 @@ func getAPIsByEnv(env cliconfig.Environment, printEnv bool) (string, error) {
 		out += t.MustFormat()
 	}
 
+	if len(allTaskAPIs) > 0 {
+		envNames := []string{}
+		for range allTaskAPIs {
+			envNames = append(envNames, env.Name)
+		}
+
+		t := taskAPIsTable(allTaskAPIs, envNames)
+		t.FindHeaderByTitle(_titleEnvironment).Hidden = true
+
+		if len(allBatchAPIs) > 0 {
+			out += "\n"
+		}
+
+		out += t.MustFormat()
+	}
+
 	if len(allRealtimeAPIs) > 0 {
 		envNames := []string{}
 		for range allRealtimeAPIs {
@@ -359,7 +399,7 @@ func getAPIsByEnv(env cliconfig.Environment, printEnv bool) (string, error) {
 		t := realtimeAPIsTable(allRealtimeAPIs, envNames)
 		t.FindHeaderByTitle(_titleEnvironment).Hidden = true
 
-		if len(allBatchAPIs) > 0 {
+		if len(allBatchAPIs) > 0 || len(allTaskAPIs) > 0 {
 			out += "\n"
 		}
 
@@ -375,7 +415,7 @@ func getAPIsByEnv(env cliconfig.Environment, printEnv bool) (string, error) {
 		t := trafficSplitterListTable(allTrafficSplitters, envNames)
 		t.FindHeaderByTitle(_titleEnvironment).Hidden = true
 
-		if len(allBatchAPIs) > 0 || len(allRealtimeAPIs) > 0 {
+		if len(allBatchAPIs) > 0 || len(allTaskAPIs) > 0 || len(allRealtimeAPIs) > 0 {
 			out += "\n"
 		}
 
@@ -412,6 +452,8 @@ func getAPI(env cliconfig.Environment, apiName string) (string, error) {
 		return trafficSplitterTable(apiRes, env)
 	case userconfig.BatchAPIKind:
 		return batchAPITable(apiRes), nil
+	case userconfig.TaskAPIKind:
+		return taskAPITable(apiRes), nil
 	default:
 		return "", errors.ErrorUnexpected(fmt.Sprintf("encountered unexpected kind %s for api %s", apiRes.Spec.Kind, apiRes.Spec.Name))
 	}

cli/cmd/lib_batch_apis.go

@@ -49,7 +49,7 @@ func batchAPIsTable(batchAPIs []schema.APIResponse, envNames []string) table.Tab
 		latestJobID := "-"
 		runningJobs := 0
 
-		for _, job := range batchAPI.JobStatuses {
+		for _, job := range batchAPI.BatchJobStatuses {
 			if job.StartTime.After(latestStartTime) {
 				latestStartTime = job.StartTime
 				latestJobID = job.ID + fmt.Sprintf(" (submitted %s ago)", libtime.SinceStr(&latestStartTime))
@@ -82,14 +82,14 @@ func batchAPIsTable(batchAPIs []schema.APIResponse, envNames []string) table.Tab
 }
 
 func batchAPITable(batchAPI schema.APIResponse) string {
-	jobRows := make([][]interface{}, 0, len(batchAPI.JobStatuses))
+	jobRows := make([][]interface{}, 0, len(batchAPI.BatchJobStatuses))
 
 	out := ""
-	if len(batchAPI.JobStatuses) == 0 {
-		out = console.Bold("no submitted jobs\n")
+	if len(batchAPI.BatchJobStatuses) == 0 {
+		out = console.Bold("no submitted batch jobs\n")
 	} else {
 		totalFailed := 0
-		for _, job := range batchAPI.JobStatuses {
+		for _, job := range batchAPI.BatchJobStatuses {
 			succeeded := 0
 			failed := 0
 
@@ -144,8 +144,8 @@ func batchAPITable(batchAPI schema.APIResponse) string {
 	return out
 }
 
-func getJob(env cliconfig.Environment, apiName string, jobID string) (string, error) {
-	resp, err := cluster.GetJob(MustGetOperatorConfig(env.Name), apiName, jobID)
+func getBatchJob(env cliconfig.Environment, apiName string, jobID string) (string, error) {
+	resp, err := cluster.GetBatchJob(MustGetOperatorConfig(env.Name), apiName, jobID)
 	if err != nil {
 		return "", err
 	}
@@ -216,12 +216,11 @@ func getJob(env cliconfig.Environment, apiName string, jobID string) (string, er
 	out += titleStr("batch stats") + t.MustFormat(&table.Opts{BoldHeader: pointer.Bool(false)})
 
 	if job.Status == status.JobEnqueuing {
-		out += "\nstill enqueuing, workers have not been allocated for this job yet\n"
+		out += "\n" + "still enqueuing, workers have not been allocated for this job yet\n"
 	} else if job.Status.IsCompleted() {
-		out += "\nworker stats are not available because this job is not currently running\n"
+		out += "\n" + "worker stats are not available because this job is not currently running\n"
 	} else {
 		out += titleStr("worker stats")
-
 		if job.WorkerCounts != nil {
 			t := table.Table{
 				Headers: []table.Header{
@@ -253,7 +252,7 @@ func getJob(env cliconfig.Environment, apiName string, jobID string) (string, er
 
 	out += "\n" + console.Bold("job endpoint: ") + resp.Endpoint + "\n"
 
-	jobSpecStr, err := libjson.Pretty(job.Job)
+	jobSpecStr, err := libjson.Pretty(job.BatchJob)
 	if err != nil {
 		return "", err
 	}