Add support for exporting APIs (#1368)

Author: vishalbollu

cli/cluster/lib_http_client.go

@@ -197,6 +197,9 @@ func (client *OperatorClient) MakeRequest(operatorConfig OperatorConfig, request
 
 	response, err := client.Do(request)
 	if err != nil {
+		if operatorConfig.EnvName == "" {
+			return nil, errors.Wrap(err, "failed to connect to operator", operatorConfig.OperatorEndpoint)
+		}
 		return nil, ErrorFailedToConnectOperator(err, operatorConfig.EnvName, operatorConfig.OperatorEndpoint)
 	}
 	defer response.Body.Close()

cli/cmd/cluster.go

@@ -19,6 +19,7 @@ package cmd
 import (
 	"fmt"
 	"os"
+	"path"
 	"path/filepath"
 	"strings"
 	"time"
@@ -27,6 +28,7 @@ import (
 	"github.com/cortexlabs/cortex/cli/cluster"
 	"github.com/cortexlabs/cortex/cli/types/cliconfig"
 	"github.com/cortexlabs/cortex/pkg/consts"
+	"github.com/cortexlabs/cortex/pkg/lib/archive"
 	"github.com/cortexlabs/cortex/pkg/lib/aws"
 	cr "github.com/cortexlabs/cortex/pkg/lib/configreader"
 	"github.com/cortexlabs/cortex/pkg/lib/console"
@@ -43,6 +45,8 @@ import (
 	"github.com/cortexlabs/cortex/pkg/types"
 	"github.com/cortexlabs/cortex/pkg/types/clusterconfig"
 	"github.com/cortexlabs/cortex/pkg/types/clusterstate"
+	"github.com/cortexlabs/cortex/pkg/types/spec"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
 	"github.com/spf13/cobra"
 )
 
@@ -89,6 +93,11 @@ func clusterInit() {
 	addAWSCredentials(_downCmd)
 	_downCmd.Flags().BoolVarP(&_flagClusterDisallowPrompt, "yes", "y", false, "skip prompts")
 	_clusterCmd.AddCommand(_downCmd)
+
+	_exportCmd.Flags().SortFlags = false
+	addClusterConfigFlag(_exportCmd)
+	addAWSCredentials(_exportCmd)
+	_clusterCmd.AddCommand(_exportCmd)
 }
 
 func addClusterConfigFlag(cmd *cobra.Command) {
@@ -269,7 +278,7 @@ var _upCmd = &cobra.Command{
 			exit.Error(errors.Append(err, fmt.Sprintf("\n\nunable to locate operator load balancer; you can attempt to resolve this issue and configure your CLI environment by running `cortex cluster info --env %s`", _flagClusterEnv)))
 		}
 		if loadBalancer == nil {
-			exit.Error(ErrorNoOperatorLoadBalancer(_flagClusterEnv))
+			exit.Error(errors.Append(ErrorNoOperatorLoadBalancer(), fmt.Sprintf("; you can attempt to resolve this issue and configure your CLI environment by running `cortex cluster info --env %s`", _flagClusterEnv)))
 		}
 
 		newEnvironment := cliconfig.Environment{
@@ -434,6 +443,7 @@ var _downCmd = &cobra.Command{
 		if err != nil {
 			exit.Error(err)
 		}
+
 		warnIfNotAdmin(awsClient)
 
 		clusterState, err := clusterstate.GetClusterState(awsClient, accessConfig)
@@ -533,6 +543,138 @@ var _downCmd = &cobra.Command{
 	},
 }
 
+var _exportCmd = &cobra.Command{
+	Use:   "export",
+	Short: "download the code and configuration for all APIs deployed in a cluster",
+	Args:  cobra.NoArgs,
+	Run: func(cmd *cobra.Command, args []string) {
+		telemetry.Event("cli.cluster.export")
+
+		if _flagClusterConfig != "" {
+			// Deprecation: specifying aws creds in cluster configuration is no longer supported
+			if err := detectAWSCredsInConfigFile(cmd.Use, _flagClusterConfig); err != nil {
+				exit.Error(err)
+			}
+		}
+
+		accessConfig, err := getClusterAccessConfig(_flagClusterDisallowPrompt)
+		if err != nil {
+			exit.Error(err)
+		}
+
+		awsCreds, err := awsCredentialsForManagingCluster(*accessConfig, _flagClusterDisallowPrompt)
+		if err != nil {
+			exit.Error(err)
+		}
+
+		// Check AWS access
+		awsClient, err := newAWSClient(*accessConfig.Region, awsCreds)
+		if err != nil {
+			exit.Error(err)
+		}
+		warnIfNotAdmin(awsClient)
+
+		clusterState, err := clusterstate.GetClusterState(awsClient, accessConfig)
+		if err != nil {
+			exit.Error(err)
+		}
+
+		err = clusterstate.AssertClusterStatus(*accessConfig.ClusterName, *accessConfig.Region, clusterState.Status, clusterstate.StatusCreateComplete)
+		if err != nil {
+			exit.Error(err)
+		}
+
+		loadBalancer, err := awsClient.FindLoadBalancer(map[string]string{
+			clusterconfig.ClusterNameTag: *accessConfig.ClusterName,
+			"cortex.dev/load-balancer":   "operator",
+		})
+		if err != nil {
+			exit.Error(err)
+		}
+		if loadBalancer == nil {
+			exit.Error(ErrorNoOperatorLoadBalancer())
+		}
+
+		operatorConfig := cluster.OperatorConfig{
+			Telemetry:          isTelemetryEnabled(),
+			ClientID:           clientID(),
+			AWSAccessKeyID:     awsCreds.AWSAccessKeyID,
+			AWSSecretAccessKey: awsCreds.AWSSecretAccessKey,
+			OperatorEndpoint:   "https://" + *loadBalancer.DNSName,
+		}
+
+		info, err := cluster.Info(operatorConfig)
+		if err != nil {
+			exit.Error(err)
+		}
+
+		apisResponse, err := cluster.GetAPIs(operatorConfig)
+		if err != nil {
+			exit.Error(err)
+		}
+
+		var apiSpecs []spec.API
+
+		for _, batchAPI := range apisResponse.BatchAPIs {
+			apiSpecs = append(apiSpecs, batchAPI.Spec)
+		}
+
+		for _, realtimeAPI := range apisResponse.RealtimeAPIs {
+			apiSpecs = append(apiSpecs, realtimeAPI.Spec)
+		}
+
+		for _, trafficSplitter := range apisResponse.TrafficSplitters {
+			apiSpecs = append(apiSpecs, trafficSplitter.Spec)
+		}
+
+		if len(apiSpecs) == 0 {
+			fmt.Println(fmt.Sprintf("no apis found in cluster named %s in %s", *accessConfig.ClusterName, *accessConfig.Region))
+			exit.Ok()
+		}
+
+		exportPath := fmt.Sprintf("export-%s-%s", *accessConfig.Region, *accessConfig.ClusterName)
+
+		err = files.CreateDir(exportPath)
+		if err != nil {
+			exit.Error(err)
+		}
+
+		for _, apiSpec := range apiSpecs {
+			baseDir := filepath.Join(exportPath, apiSpec.Name)
+
+			fmt.Println(fmt.Sprintf("exporting %s to %s", apiSpec.Name, baseDir))
+
+			err = files.CreateDir(baseDir)
+			if err != nil {
+				exit.Error(err)
+			}
+
+			err = awsClient.DownloadFileFromS3(info.ClusterConfig.Bucket, apiSpec.RawAPIKey(), path.Join(baseDir, apiSpec.FileName))
+			if err != nil {
+				exit.Error(err)
+			}
+
+			if apiSpec.Kind != userconfig.TrafficSplitterKind {
+				zipFileLocation := path.Join(baseDir, path.Base(apiSpec.ProjectKey))
+				err = awsClient.DownloadFileFromS3(info.ClusterConfig.Bucket, apiSpec.ProjectKey, zipFileLocation)
+				if err != nil {
+					exit.Error(err)
+				}
+
+				_, err = archive.UnzipFileToDir(zipFileLocation, baseDir)
+				if err != nil {
+					exit.Error(err)
+				}
+
+				err := os.Remove(zipFileLocation)
+				if err != nil {
+					exit.Error(err)
+				}
+			}
+		}
+	},
+}
+
 var _emailPrompValidation = &cr.PromptValidation{
 	PromptItemValidations: []*cr.PromptItemValidation{
 		{

cli/cmd/errors.go

@@ -133,11 +133,10 @@ func ErrorInvalidOperatorEndpoint(endpoint string) error {
 	})
 }
 
-// err can be passed in as nil
-func ErrorNoOperatorLoadBalancer(envName string) error {
+func ErrorNoOperatorLoadBalancer() error {
 	return errors.WithStack(&errors.Error{
 		Kind:    ErrNoOperatorLoadBalancer,
-		Message: fmt.Sprintf("unable to locate operator load balancer; you can attempt to resolve this issue and configure your CLI environment by running `cortex cluster info --env %s`", envName),
+		Message: "unable to locate operator load balancer",
 	})
 }
 

dev/generate_cli_md.sh

@@ -37,6 +37,7 @@ commands=(
   "cluster up"
   "cluster info"
   "cluster configure"
+  "cluster export"
   "cluster down"
   "env configure"
   "env list"

docs/miscellaneous/cli.md

@@ -156,6 +156,21 @@ Flags:
   -h, --help                help for configure
 ```
 
+### cluster export
+
+```text
+download the code and configuration for all APIs deployed in a cluster
+
+Usage:
+  cortex cluster export [flags]
+
+Flags:
+  -c, --config string       path to a cluster configuration file
+      --aws-key string      aws access key id
+      --aws-secret string   aws secret access key
+  -h, --help                help for export
+```
+
 ### cluster down
 
 ```text

pkg/operator/resources/batchapi/api.go

@@ -49,6 +49,10 @@ func UpdateAPI(apiConfig *userconfig.API, projectID string) (*spec.API, string,
 			return nil, "", errors.Wrap(err, "upload api spec")
 		}
 
+		if err := config.AWS.UploadBytesToS3(api.RawYAMLBytes, config.Cluster.Bucket, api.RawAPIKey()); err != nil {
+			return nil, "", errors.Wrap(err, "upload raw api spec")
+		}
+
 		err = applyK8sResources(api, prevVirtualService)
 		if err != nil {
 			go deleteK8sResources(api.Name)
@@ -75,6 +79,10 @@ func UpdateAPI(apiConfig *userconfig.API, projectID string) (*spec.API, string,
 			return nil, "", errors.Wrap(err, "upload api spec")
 		}
 
+		if err := config.AWS.UploadBytesToS3(api.RawYAMLBytes, config.Cluster.Bucket, api.RawAPIKey()); err != nil {
+			return nil, "", errors.Wrap(err, "upload raw api spec")
+		}
+
 		err = applyK8sResources(api, prevVirtualService)
 		if err != nil {
 			return nil, "", err

pkg/operator/resources/realtimeapi/api.go

@@ -59,6 +59,10 @@ func UpdateAPI(apiConfig *userconfig.API, projectID string, force bool) (*spec.A
 			return nil, "", errors.Wrap(err, "upload api spec")
 		}
 
+		if err := config.AWS.UploadBytesToS3(api.RawYAMLBytes, config.Cluster.Bucket, api.RawAPIKey()); err != nil {
+			return nil, "", errors.Wrap(err, "upload raw api spec")
+		}
+
 		// Use api spec indexed by PredictorID for replicas to prevent rolling updates when SpecID changes without PredictorID changing
 		if err := config.AWS.UploadJSONToS3(api, config.Cluster.Bucket, api.PredictorKey); err != nil {
 			return nil, "", errors.Wrap(err, "upload predictor spec")
@@ -92,6 +96,10 @@ func UpdateAPI(apiConfig *userconfig.API, projectID string, force bool) (*spec.A
 			return nil, "", errors.Wrap(err, "upload api spec")
 		}
 
+		if err := config.AWS.UploadBytesToS3(api.RawYAMLBytes, config.Cluster.Bucket, api.RawAPIKey()); err != nil {
+			return nil, "", errors.Wrap(err, "upload raw api spec")
+		}
+
 		// Use api spec indexed by PredictorID for replicas to prevent rolling updates when SpecID changes without PredictorID changing
 		if err := config.AWS.UploadJSONToS3(api, config.Cluster.Bucket, api.PredictorKey); err != nil {
 			return nil, "", errors.Wrap(err, "upload predictor spec")

pkg/operator/resources/trafficsplitter/api.go

@@ -42,10 +42,16 @@ func UpdateAPI(apiConfig *userconfig.API, force bool) (*spec.API, string, error)
 		if err := config.AWS.UploadJSONToS3(api, config.Cluster.Bucket, api.Key); err != nil {
 			return nil, "", errors.Wrap(err, "upload api spec")
 		}
+
+		if err := config.AWS.UploadBytesToS3(api.RawYAMLBytes, config.Cluster.Bucket, api.RawAPIKey()); err != nil {
+			return nil, "", errors.Wrap(err, "upload raw api spec")
+		}
+
 		if err := applyK8sVirtualService(api, prevVirtualService); err != nil {
 			go deleteK8sResources(api.Name)
 			return nil, "", err
 		}
+
 		err = operator.AddAPIToAPIGateway(*api.Networking.Endpoint, api.Networking.APIGateway, false)
 		if err != nil {
 			go deleteK8sResources(api.Name)
@@ -58,9 +64,15 @@ func UpdateAPI(apiConfig *userconfig.API, force bool) (*spec.API, string, error)
 		if err := config.AWS.UploadJSONToS3(api, config.Cluster.Bucket, api.Key); err != nil {
 			return nil, "", errors.Wrap(err, "upload api spec")
 		}
+
+		if err := config.AWS.UploadBytesToS3(api.RawYAMLBytes, config.Cluster.Bucket, api.RawAPIKey()); err != nil {
+			return nil, "", errors.Wrap(err, "upload raw api spec")
+		}
+
 		if err := applyK8sVirtualService(api, prevVirtualService); err != nil {
 			return nil, "", err
 		}
+
 		if err := operator.UpdateAPIGatewayK8s(prevVirtualService, api, false); err != nil {
 			return nil, "", err
 		}

pkg/types/spec/api.go

@@ -154,6 +154,16 @@ func Key(apiName string, apiID string) string {
 	)
 }
 
+func (api API) RawAPIKey() string {
+	return filepath.Join(
+		"apis",
+		api.Name,
+		"raw_api",
+		api.ID,
+		consts.CortexVersion+"-cortex.yaml",
+	)
+}
+
 func MetadataRoot(apiName string) string {
 	return filepath.Join(
 		"apis",

pkg/types/spec/validations.go

@@ -43,6 +43,7 @@ import (
 	"github.com/cortexlabs/cortex/pkg/types"
 	"github.com/cortexlabs/cortex/pkg/types/clusterconfig"
 	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"github.com/cortexlabs/yaml"
 	kresource "k8s.io/apimachinery/pkg/api/resource"
 )
 
@@ -643,10 +644,15 @@ func ExtractAPIConfigs(
 				return nil, errors.Append(err, fmt.Sprintf("\n\napi configuration schema for Traffic Splitter can be found at https://docs.cortex.dev/v/%s/deployments/realtime-api/traffic-splitter", consts.CortexVersionMinor))
 			}
 		}
-
 		api.Index = i
 		api.FileName = configFileName
 
+		rawYAMLBytes, err := yaml.Marshal([]map[string]interface{}{data})
+		if err != nil {
+			return nil, errors.Wrap(err, api.Identify())
+		}
+		api.RawYAMLBytes = rawYAMLBytes
+
 		if resourceStruct.Kind == userconfig.RealtimeAPIKind || resourceStruct.Kind == userconfig.BatchAPIKind {
 			api.ApplyDefaultDockerPaths()
 		}