Round #4

Author: vishalbollu

cli/cmd/errors.go

@@ -69,6 +69,7 @@ const (
 	ErrClusterAlreadyDeleted         = "cli.cluster_already_deleted"
 	ErrFailedClusterStatus           = "cli.failed_cluster_status"
 	ErrClusterDoesNotExist           = "cli.cluster_does_not_exist"
+	ErrTensorFlowDirTooManyFiles     = "cli.tensorflow_dir_too_many_files"
 )
 
 func ErrorCLINotConfigured(env string) error {
@@ -299,3 +300,10 @@ func ErrorFailedClusterStatus(status clusterstate.Status, clusterName string, re
 		Message: fmt.Sprintf("cluster %s in %s encountered an unexpected status %s, please try to delete the cluster with `cortex cluster down` or delete the cloudformation stacks manually in your AWS console %s", clusterName, region, string(status), getCloudFormationURL(clusterName, region)),
 	})
 }
+
+func ErrorTensorFlowDirTooManyFiles(count int32) error {
+	return errors.WithStack(&errors.Error{
+		Kind:    ErrTensorFlowDirTooManyFiles,
+		Message: fmt.Sprintf("more than %d many files found in tensorflow directory", count),
+	})
+}

cli/cmd/local.go

@@ -5,8 +5,11 @@ import (
 	"fmt"
 	"path"
 	"path/filepath"
+	"strings"
+	"time"
 
 	"github.com/cortexlabs/cortex/cli/local"
+	"github.com/cortexlabs/cortex/pkg/lib/aws"
 	"github.com/cortexlabs/cortex/pkg/lib/debug"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/exit"
@@ -18,6 +21,7 @@ import (
 	"github.com/cortexlabs/cortex/pkg/lib/zip"
 	"github.com/cortexlabs/cortex/pkg/operator/schema"
 	"github.com/cortexlabs/cortex/pkg/types/spec"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
 	"github.com/docker/docker/api/types"
 	dockertypes "github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/filters"
@@ -140,6 +144,78 @@ var localCmd = &cobra.Command{
 	},
 }
 
+func cacheModel(api *userconfig.API) {
+	if strings.HasPrefix(*api.Predictor.Model, "s3://") {
+
+	} else {
+
+	}
+}
+
+func cacheModelFromS3(api *userconfig.API) (string, error) {
+	awsClient, err := aws.NewFromEnvS3Path(*api.Predictor.Model)
+	if err != nil {
+		return "", err
+	}
+
+	s3Objects, err := awsClient.ListPathPrefix(*api.Predictor.Model, 1001)
+	if err != nil {
+		return "", err
+	}
+
+	if len(s3Objects) == 1001 {
+		return "", ErrorTensorFlowDirTooManyFiles(1000)
+	}
+	var mostRecentUpdateDate *time.Time
+	for _, obj := range s3Objects {
+		mostRecentUpdateDate = obj.LastModified
+	}
+
+	modelPathHash := hash.String(*api.Predictor.Model)
+	modelDir := filepath.Join(*api.Predictor.Model, modelPathHash)
+	modelVersionDir := filepath.Join(modelDir, mostRecentUpdateDate.Format("2006-01-02T15:04:05"))
+
+	if files.IsFile(filepath.Join(modelVersionDir, "_SUCCESS")) {
+		return modelVersionDir, nil
+	}
+
+	err = files.DeleteDir(modelDir)
+	if err != nil {
+		return "", err
+	}
+
+	_, err = files.CreateDirIfMissing(modelVersionDir)
+	if err != nil {
+		return "", nil
+	}
+
+	bucket, fullPathKey, err := aws.SplitS3Path(*api.Predictor.Model)
+	if err != nil {
+		return "", err
+	}
+	for _, obj := range s3Objects {
+		if *obj.Size == 0 { // TODO test creation of empty files
+			continue
+		}
+
+		if strings.HasSuffix(*obj.Key, "/") {
+			continue
+		}
+
+		localKey := (*obj.Key)[len(fullPathKey):]
+		fileBytes, err := awsClient.ReadBytesFromS3(bucket, *obj.Key)
+		if err != nil {
+			return "", err
+		}
+
+		err = files.WriteFile(fileBytes, filepath.Join(modelVersionDir, localKey))
+		if err != nil {
+			return "", err
+		}
+	}
+	return modelVersionDir, nil
+}
+
 var localGet = &cobra.Command{
 	Use:   "local-get",
 	Short: "local an application",

cli/local/config.go

@@ -18,7 +18,6 @@ package local
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
@@ -35,6 +34,7 @@ var _cachedDockerClient *dockerclient.Client
 var CWD string
 var LocalDir string
 var LocalWorkspace string
+var ModelCacheDir string
 
 func init() {
 	cwd, err := os.Getwd()
@@ -58,7 +58,13 @@ func init() {
 	}
 
 	LocalWorkspace = filepath.Join(LocalDir, "local_workspace")
-	fmt.Println(LocalWorkspace)
+	err = os.MkdirAll(LocalWorkspace, os.ModePerm)
+	if err != nil {
+		err := errors.Wrap(err, "unable to write to home directory", LocalDir)
+		exit.Error(err)
+	}
+
+	ModelCacheDir = filepath.Join(LocalDir, "model_cache")
 	err = os.MkdirAll(LocalWorkspace, os.ModePerm)
 	if err != nil {
 		err := errors.Wrap(err, "unable to write to home directory", LocalDir)

pkg/operator/operator/validations.go

@@ -42,7 +42,7 @@ func ValidateClusterAPIs(apis []userconfig.API, projectFileMap map[string][]byte
 	}
 
 	for i := range apis {
-		if err := spec.ValidateAPI(&apis[i], projectFileMap); err != nil {
+		if err := spec.ValidateAPI(&apis[i], projectFileMap, "aws"); err != nil {
 			return err
 		}
 		if err := validateK8s(&apis[i], config.Cluster, virtualServices, maxMem); err != nil {

pkg/types/spec/errors.go

@@ -49,7 +49,7 @@ const (
 	ErrFieldNotSupportedByPredictorType   = "spec.field_not_supported_by_predictor_type"
 	ErrNoAvailableNodeComputeLimit        = "spec.no_available_node_compute_limit"
 	ErrCortexPrefixedEnvVarNotAllowed     = "spec.cortex_prefixed_env_var_not_allowed"
-	ErrLocalPathNotSupportedByProvider    = "spec.local_path_not_supported_by_provider"
+	ErrLocalPathNotSupportedByAWSProvider = "spec.local_path_not_supported_by_aws_provider"
 )
 
 func ErrorMalformedConfig() error {
@@ -236,9 +236,9 @@ func ErrorCortexPrefixedEnvVarNotAllowed() error {
 	})
 }
 
-func ErrorLocalPathNotSupportedByProvider(provider string) error {
+func ErrorLocalModelPathNotSupportedByAWSProvider() error {
 	return errors.WithStack(&errors.Error{
-		Kind:    ErrLocalPathNotSupportedByProvider,
-		Message: fmt.Sprintf("environment variables starting with CORTEX_ are reserved"),
+		Kind:    ErrLocalPathNotSupportedByAWSProvider,
+		Message: fmt.Sprintf("local model paths are not supported for aws provider, please specify an S3 path"),
 	})
 }

pkg/types/spec/validations.go

@@ -469,7 +469,23 @@ func validateTensorFlowPredictor(predictor *userconfig.Predictor, providerType s
 			predictor.Model = pointer.String(path)
 		}
 	} else {
+		if providerType == "aws" {
+			return errors.Wrap(ErrorLocalModelPathNotSupportedByAWSProvider(), model, userconfig.ModelKey)
+		}
 
+		if strings.HasPrefix(model, ".zip") {
+			if err := files.CheckFile(model); err != nil {
+				return errors.Wrap(err, userconfig.ModelKey)
+			}
+		} else {
+			path, err := getTFServingExportFromLocalPath(model)
+			if err != nil {
+				return errors.Wrap(err, userconfig.ModelKey)
+			} else if path == "" {
+				return errors.Wrap(ErrorInvalidTensorFlowDir(model), userconfig.ModelKey)
+			}
+			predictor.Model = pointer.String(path)
+		}
 	}
 
 	return nil
@@ -492,7 +508,13 @@ func validateONNXPredictor(predictor *userconfig.Predictor, providerType string)
 			return errors.Wrap(ErrorS3FileNotFound(model), userconfig.ModelKey)
 		}
 	} else {
+		if providerType == "aws" {
+			return errors.Wrap(ErrorLocalModelPathNotSupportedByAWSProvider(), model, userconfig.ModelKey)
+		}
 
+		if err := files.CheckFile(model); err != nil {
+			return errors.Wrap(err, userconfig.ModelKey)
+		}
 	}
 
 	if predictor.SignatureKey != nil {
@@ -567,6 +589,9 @@ func isValidTensorFlowS3Directory(path string, awsClient *aws.Client) bool {
 }
 
 func getTFServingExportFromLocalPath(path string) (string, error) {
+	if !files.IsDir(path) {
+		return "", ErrorDirNotFoundOrEmpty(path)
+	}
 	paths, err := files.ListDirRecursive(path, true, files.IgnoreHiddenFiles, files.IgnoreHiddenFolders)
 	if err != nil {
 		return "", err