Working task submission end to end

Author: Miguel Varela Ramos

pkg/operator/endpoints/submit_task.go

@@ -27,6 +27,7 @@ import (
 	"github.com/cortexlabs/cortex/pkg/operator/resources"
 	"github.com/cortexlabs/cortex/pkg/operator/resources/job/taskapi"
 	"github.com/cortexlabs/cortex/pkg/operator/schema"
+	"github.com/cortexlabs/cortex/pkg/types/spec"
 	"github.com/cortexlabs/cortex/pkg/types/userconfig"
 	"github.com/gorilla/mux"
 )
@@ -54,11 +55,16 @@ func SubmitTaskJob(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	submission := schema.TaskJobSubmission{}
+	submission := schema.TaskJobSubmission{
+		RuntimeJobConfig: spec.RuntimeJobConfig{Workers: 1},
+	}
 
 	err = json.Unmarshal(bodyBytes, &submission)
 	if err != nil {
-		respondError(w, r, errors.Append(err, fmt.Sprintf("\n\ntask job submission schema can be found at https://docs.cortex.dev/v/%s/", consts.CortexVersionMinor)))
+		respondError(w, r, errors.Append(err,
+			fmt.Sprintf("\n\ntask job submission schema can be found at https://docs.cortex.dev/v/%s/",
+				consts.CortexVersionMinor)),
+		)
 		return
 	}
 

pkg/operator/operator/k8s.go

@@ -36,7 +36,6 @@ import (
 	istioclientnetworking "istio.io/client-go/pkg/apis/networking/v1beta1"
 	kcore "k8s.io/api/core/v1"
 	kresource "k8s.io/apimachinery/pkg/api/resource"
-	kmeta "k8s.io/apimachinery/pkg/apis/meta/v1"
 	intstr "k8s.io/apimachinery/pkg/util/intstr"
 )
 
@@ -187,9 +186,9 @@ func TaskContainers(api *spec.API) ([]kcore.Container, []kcore.Volume) {
 		Env:             getTaskEnvVars(api, APIContainerName),
 		EnvFrom:         baseEnvVars(),
 		VolumeMounts:    apiPodVolumeMounts,
-		ReadinessProbe:  FileExistsProbe(_apiReadinessFile),
-		LivenessProbe:   _apiLivenessProbe,
-		Lifecycle:       nginxGracefulStopper(api.Kind),
+		//ReadinessProbe:  FileExistsProbe(_apiReadinessFile), // TODO: check if it makes sense to have probes
+		//LivenessProbe:   _apiLivenessProbe,
+		Lifecycle: nginxGracefulStopper(api.Kind),
 		Resources: kcore.ResourceRequirements{
 			Requests: apiPodResourceList,
 			Limits:   apiPodResourceLimitsList,
@@ -652,6 +651,10 @@ func getTaskEnvVars(api *spec.API, container string) []kcore.EnvVar {
 				Name:  "CORTEX_PROJECT_DIR",
 				Value: path.Join(_emptyDirMountPath, "project"),
 			},
+			kcore.EnvVar{
+				Name:  "CORTEX_CACHE_DIR",
+				Value: _specCacheDir,
+			},
 			kcore.EnvVar{
 				Name:  "CORTEX_API_SPEC",
 				Value: config.BucketPath(api.Key),
@@ -1059,13 +1062,3 @@ func GetEndpointFromVirtualService(virtualService *istioclientnetworking.Virtual
 
 	return endpoints.GetOne(), nil
 }
-
-func extractCortexAnnotations(obj kmeta.Object) map[string]string {
-	cortexAnnotations := make(map[string]string)
-	for key, value := range obj.GetAnnotations() {
-		if strings.Contains(key, "cortex.dev/") {
-			cortexAnnotations[key] = value
-		}
-	}
-	return cortexAnnotations
-}

pkg/workloads/cortex/serve/start/task.py

@@ -12,27 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import sys
-import os
-import argparse
-import inspect
-import time
 import json
-import threading
-import math
-import pathlib
-
-import boto3
-import botocore
+import os
 
-from cortex import consts
-from cortex.lib import util
-from cortex.lib.api import API, get_spec, get_api
+from cortex.lib.api import get_spec
 from cortex.lib.log import cx_logger as logger
-from cortex.lib.concurrency import LockedFile
-from cortex.lib.storage import S3, LocalStorage
-from cortex.lib.exceptions import UserRuntimeException
-
+from cortex.lib.storage import S3
 
 local_cache = {
     "api_spec": None,