Add traffic splitter API kind (#1213)

* updated kind with apisplitter

* added apisplitter useconfig

* apisplit yaml example

* created apisplitter resource barebones

* added apisplitter validations

* migrated logic to overloading api config

* basic post spec validation done

* adding splitting logic to virtual service

* working splitter update

* api gateway support for trafficsplit

* delete useless stuff for trafficsplitter

* add get command for trafficsplitter

* add trafficsplit table and get all deployments

* cleanup and comments

* cli update to get status of trafficsplitter

* fix test error

* fix comparison on virtual services

* new traffic split get table

* add new get all table for trafficsplitter

* add endpoint duplication check

* add check if api used by apisplitter

* fix bug virtual service naming and remove print statements

* apisplitter not supported on local check

* imporve cli output format

* fixes and formatting

* allow 0 weight trafficsplitter

* update cli trafficsplit get table

* improve error messages

* Update apisplit.yaml

* rewrite virtual service destination definition

* code review improvments

* review improvements

* code review improvements

* remove apisplit example

* correct table for cortex get

* remove print statements

* code review improvements

* add API uniqueness check && review improvements

* Nits

* Only allow logs command for Sync APIs

Co-authored-by: vishal <vishalbollu@users.noreply.github.com>
Co-authored-by: Vishal Bollu <vishal.bollu@gmail.com>

Author: tthebst

cli/cmd/get.go

@@ -34,7 +34,6 @@ import (
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/exit"
 	"github.com/cortexlabs/cortex/pkg/lib/json"
-	"github.com/cortexlabs/cortex/pkg/lib/sets/strset"
 	s "github.com/cortexlabs/cortex/pkg/lib/strings"
 	"github.com/cortexlabs/cortex/pkg/lib/table"
 	"github.com/cortexlabs/cortex/pkg/lib/telemetry"
@@ -49,18 +48,21 @@ import (
 )
 
 const (
-	_titleEnvironment = "env"
-	_titleAPI         = "api"
-	_titleStatus      = "status"
-	_titleUpToDate    = "up-to-date"
-	_titleStale       = "stale"
-	_titleRequested   = "requested"
-	_titleFailed      = "failed"
-	_titleLastupdated = "last update"
-	_titleAvgRequest  = "avg request"
-	_title2XX         = "2XX"
-	_title4XX         = "4XX"
-	_title5XX         = "5XX"
+	_titleEnvironment   = "env"
+	_titleAPI           = "api"
+	_titleAPISplitter   = "api splitter"
+	_titleAPIs          = "apis"
+	_apiSplitterWeights = "weights"
+	_titleStatus        = "status"
+	_titleUpToDate      = "up-to-date"
+	_titleStale         = "stale"
+	_titleRequested     = "requested"
+	_titleFailed        = "failed"
+	_titleLastupdated   = "last update"
+	_titleAvgRequest    = "avg request"
+	_title2XX           = "2XX"
+	_title4XX           = "4XX"
+	_title5XX           = "5XX"
 )
 
 var (
@@ -146,8 +148,11 @@ func getAPIsInAllEnvironments() (string, error) {
 	}
 
 	var allSyncAPIs []schema.SyncAPI
-	var allEnvs []string
+	var allAPISplitters []schema.APISplitter
+	var allEnvsSyncAPI []string
+	var allEnvsAPISplitter []string
 	errorsMap := map[string]error{}
+	// get apis from both environments
 	for _, env := range cliConfig.Environments {
 		var apisRes schema.GetAPIsResponse
 		var err error
@@ -159,34 +164,35 @@ func getAPIsInAllEnvironments() (string, error) {
 
 		if err == nil {
 			for range apisRes.SyncAPIs {
-				allEnvs = append(allEnvs, env.Name)
+				allEnvsSyncAPI = append(allEnvsSyncAPI, env.Name)
+			}
+			for range apisRes.APISplitter {
+				allEnvsAPISplitter = append(allEnvsAPISplitter, env.Name)
 			}
-
 			allSyncAPIs = append(allSyncAPIs, apisRes.SyncAPIs...)
+			if env.Provider == types.AWSProviderType {
+				allAPISplitters = append(allAPISplitters, apisRes.APISplitter...)
+			}
 		} else {
 			errorsMap[env.Name] = err
 		}
 	}
 
 	out := ""
+	var apiSplitTable table.Table
+	var syncAPITable table.Table
 
+	// build different table depending on kinds that are deployed
 	if len(allSyncAPIs) == 0 {
-		if len(errorsMap) == 1 {
-			// Print the error if there is just one
-			exit.Error(errors.FirstErrorInMap(errorsMap))
-		}
-		// if all envs errored, skip it "no apis are deployed" since it's misleading
-		if len(errorsMap) != len(cliConfig.Environments) {
-			out += console.Bold("no apis are deployed") + "\n"
-		}
+		apiSplitTable = apiSplitterListTable(allAPISplitters, allEnvsAPISplitter)
+		out = apiSplitTable.MustFormat()
+	} else if len(allAPISplitters) == 0 {
+		syncAPITable = apiTable(allSyncAPIs, allEnvsSyncAPI)
+		out = syncAPITable.MustFormat()
 	} else {
-		t := apiTable(allSyncAPIs, allEnvs)
-
-		if strset.New(allEnvs...).IsEqual(strset.New(types.LocalProviderType.String())) {
-			hideReplicaCountColumns(&t)
-		}
-
-		out += t.MustFormat()
+		apiSplitTable = apiSplitterListTable(allAPISplitters, allEnvsAPISplitter)
+		syncAPITable = apiTable(allSyncAPIs, allEnvsSyncAPI)
+		out = syncAPITable.MustFormat() + "\n" + apiSplitTable.MustFormat()
 	}
 
 	if len(errorsMap) == 1 {
@@ -229,7 +235,7 @@ func getAPIs(env cliconfig.Environment, printEnv bool) (string, error) {
 		}
 	}
 
-	if len(apisRes.SyncAPIs) == 0 {
+	if len(apisRes.SyncAPIs) == 0 && len(apisRes.APISplitter) == 0 {
 		return console.Bold("no apis are deployed"), nil
 	}
 
@@ -238,14 +244,30 @@ func getAPIs(env cliconfig.Environment, printEnv bool) (string, error) {
 		envNames = append(envNames, env.Name)
 	}
 
-	t := apiTable(apisRes.SyncAPIs, envNames)
-
-	t.FindHeaderByTitle(_titleEnvironment).Hidden = true
+	var apiSplitTable table.Table
+	var syncAPITable table.Table
+	var out string
 
-	out := t.MustFormat()
+	// build different table depending on kinds that are deployed
+	if len(apisRes.SyncAPIs) == 0 {
+		apiSplitTable = apiSplitterListTable(apisRes.APISplitter, envNames)
+		apiSplitTable.FindHeaderByTitle(_titleEnvironment).Hidden = true
+		out = apiSplitTable.MustFormat()
+	} else if len(apisRes.APISplitter) == 0 {
+		syncAPITable = apiTable(apisRes.SyncAPIs, envNames)
+		syncAPITable.FindHeaderByTitle(_titleEnvironment).Hidden = true
+		out = syncAPITable.MustFormat()
+	} else {
+		apiSplitTable = apiSplitterListTable(apisRes.APISplitter, envNames)
+		syncAPITable = apiTable(apisRes.SyncAPIs, envNames)
+		apiSplitTable.FindHeaderByTitle(_titleEnvironment).Hidden = true
+		syncAPITable.FindHeaderByTitle(_titleEnvironment).Hidden = true
+		out = syncAPITable.MustFormat() + "\n" + apiSplitTable.MustFormat()
+	}
 
 	if env.Provider == types.LocalProviderType {
-		hideReplicaCountColumns(&t)
+		// apisplitter not supported in local
+		hideReplicaCountColumns(&syncAPITable)
 		mismatchedVersionAPIsErrorMessage, _ := getLocalVersionMismatchedAPIsMessage()
 		if len(mismatchedVersionAPIsErrorMessage) > 0 {
 			out += "\n" + mismatchedVersionAPIsErrorMessage
@@ -292,7 +314,111 @@ func getAPI(env cliconfig.Environment, apiName string) (string, error) {
 			return "", err
 		}
 	}
-	return syncAPITable(apiRes.SyncAPI, env)
+	if apiRes.SyncAPI != nil {
+		return syncAPITable(apiRes.SyncAPI, env)
+	}
+	if apiRes.APISplitter != nil {
+		return apiSplitterTable(apiRes.APISplitter, env)
+	}
+	return "", nil
+}
+
+func apiSplitterTable(apiSplitter *schema.APISplitter, env cliconfig.Environment) (string, error) {
+	var out string
+
+	lastUpdated := time.Unix(apiSplitter.Spec.LastUpdated, 0)
+	out += console.Bold("kind: ") + apiSplitter.Spec.Kind.String() + "\n\n"
+	out += console.Bold("last updated: ") + libtime.SinceStr(&lastUpdated) + "\n\n"
+
+	t, err := trafficSplitTable(*apiSplitter, env)
+	if err != nil {
+		return "", err
+	}
+	t.FindHeaderByTitle(_titleEnvironment).Hidden = true
+
+	out += t.MustFormat()
+
+	apiEndpoint := apiSplitter.BaseURL
+	if env.Provider == types.AWSProviderType {
+		apiEndpoint = urls.Join(apiSplitter.BaseURL, *apiSplitter.Spec.Networking.Endpoint)
+		if apiSplitter.Spec.Networking.APIGateway == userconfig.NoneAPIGatewayType {
+			apiEndpoint = strings.Replace(apiEndpoint, "https://", "http://", 1)
+		}
+	}
+
+	out += "\n" + console.Bold("endpoint: ") + apiEndpoint
+
+	out += fmt.Sprintf("\n%s curl %s -X POST -H \"Content-Type: application/json\" -d @sample.json\n", console.Bold("curl:"), apiEndpoint)
+
+	out += titleStr("configuration") + strings.TrimSpace(apiSplitter.Spec.UserStr(env.Provider))
+
+	return out, nil
+}
+
+func trafficSplitTable(apiSplitter schema.APISplitter, env cliconfig.Environment) (table.Table, error) {
+	rows := make([][]interface{}, 0, len(apiSplitter.Spec.APIs))
+
+	for _, api := range apiSplitter.Spec.APIs {
+		apiRes, err := cluster.GetAPI(MustGetOperatorConfig(env.Name), api.Name)
+		if err != nil {
+			return table.Table{}, err
+		}
+		lastUpdated := time.Unix(apiRes.SyncAPI.Spec.LastUpdated, 0)
+		rows = append(rows, []interface{}{
+			env.Name,
+			apiRes.SyncAPI.Spec.Name,
+			api.Weight,
+			apiRes.SyncAPI.Status.Message(),
+			apiRes.SyncAPI.Status.Requested,
+			libtime.SinceStr(&lastUpdated),
+			latencyStr(&apiRes.SyncAPI.Metrics),
+			code2XXStr(&apiRes.SyncAPI.Metrics),
+			code5XXStr(&apiRes.SyncAPI.Metrics),
+		})
+	}
+
+	return table.Table{
+		Headers: []table.Header{
+			{Title: _titleEnvironment},
+			{Title: _titleAPIs},
+			{Title: _apiSplitterWeights},
+			{Title: _titleStatus},
+			{Title: _titleRequested},
+			{Title: _titleLastupdated},
+			{Title: _titleAvgRequest},
+			{Title: _title2XX},
+			{Title: _title5XX},
+		},
+		Rows: rows,
+	}, nil
+}
+
+func apiSplitterListTable(apiSplitter []schema.APISplitter, envNames []string) table.Table {
+	rows := make([][]interface{}, 0, len(apiSplitter))
+	for i, splitAPI := range apiSplitter {
+		lastUpdated := time.Unix(splitAPI.Spec.LastUpdated, 0)
+		var apis []string
+		for _, api := range splitAPI.Spec.APIs {
+			apis = append(apis, api.Name+":"+s.Int(api.Weight))
+		}
+		apisStr := s.TruncateEllipses(strings.Join(apis, " "), 100)
+		rows = append(rows, []interface{}{
+			envNames[i],
+			splitAPI.Spec.Name,
+			apisStr,
+			libtime.SinceStr(&lastUpdated),
+		})
+	}
+
+	return table.Table{
+		Headers: []table.Header{
+			{Title: _titleEnvironment},
+			{Title: _titleAPISplitter},
+			{Title: _titleAPIs},
+			{Title: _titleLastupdated},
+		},
+		Rows: rows,
+	}
 }
 
 func syncAPITable(syncAPI *schema.SyncAPI, env cliconfig.Environment) (string, error) {
@@ -302,6 +428,8 @@ func syncAPITable(syncAPI *schema.SyncAPI, env cliconfig.Environment) (string, e
 	t.FindHeaderByTitle(_titleEnvironment).Hidden = true
 	t.FindHeaderByTitle(_titleAPI).Hidden = true
 
+	out += console.Bold("kind: ") + syncAPI.Spec.Kind.String() + "\n\n"
+
 	out += t.MustFormat()
 
 	if env.Provider != types.LocalProviderType && syncAPI.Spec.Monitoring != nil {
@@ -340,7 +468,6 @@ func syncAPITable(syncAPI *schema.SyncAPI, env cliconfig.Environment) (string, e
 
 func apiTable(syncAPIs []schema.SyncAPI, envNames []string) table.Table {
 	rows := make([][]interface{}, 0, len(syncAPIs))
-
 	var totalFailed int32
 	var totalStale int32
 	var total4XX int

pkg/lib/k8s/virtual_service.go

@@ -33,17 +33,35 @@ var _virtualServiceTypeMeta = kmeta.TypeMeta{
 }
 
 type VirtualServiceSpec struct {
-	Name        string
-	Gateways    []string
+	Name         string
+	Gateways     []string
+	Destinations []Destination
+	Path         string
+	Rewrite      *string
+	Labels       map[string]string
+	Annotations  map[string]string
+}
+
+type Destination struct {
 	ServiceName string
-	ServicePort int32
-	Path        string
-	Rewrite     *string
-	Labels      map[string]string
-	Annotations map[string]string
+	Weight      int32
+	Port        uint32
 }
 
 func VirtualService(spec *VirtualServiceSpec) *istioclientnetworking.VirtualService {
+	destinations := []*istionetworking.HTTPRouteDestination{}
+	for _, destination := range spec.Destinations {
+		destinations = append(destinations, &istionetworking.HTTPRouteDestination{
+			Destination: &istionetworking.Destination{
+				Host: destination.ServiceName,
+				Port: &istionetworking.PortSelector{
+					Number: destination.Port,
+				},
+			},
+			Weight: destination.Weight,
+		})
+	}
+
 	virtualService := &istioclientnetworking.VirtualService{
 		TypeMeta: _virtualServiceTypeMeta,
 		ObjectMeta: kmeta.ObjectMeta{
@@ -65,16 +83,7 @@ func VirtualService(spec *VirtualServiceSpec) *istioclientnetworking.VirtualServ
 							},
 						},
 					},
-					Route: []*istionetworking.HTTPRouteDestination{
-						{
-							Destination: &istionetworking.Destination{
-								Host: spec.ServiceName,
-								Port: &istionetworking.PortSelector{
-									Number: uint32(spec.ServicePort),
-								},
-							},
-						},
-					},
+					Route: destinations,
 				},
 			},
 		},

pkg/operator/endpoints/logs.go

@@ -20,6 +20,7 @@ import (
 	"net/http"
 
 	"github.com/cortexlabs/cortex/pkg/operator/resources"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
 	"github.com/gorilla/mux"
 	"github.com/gorilla/websocket"
 )
@@ -32,11 +33,14 @@ func ReadLogs(w http.ResponseWriter, r *http.Request) {
 		respondError(w, r, err)
 		return
 	}
-
 	if deployedResource == nil {
 		respondError(w, r, resources.ErrorAPINotDeployed(apiName))
 		return
 	}
+	if deployedResource.Kind != userconfig.SyncAPIKind {
+		respondError(w, r, resources.ErrorOperationNotSupportedForKind(deployedResource.Kind))
+		return
+	}
 
 	upgrader := websocket.Upgrader{}
 	socket, err := upgrader.Upgrade(w, r, nil)