Implementation of Weighted Clusters and domain based routing (#16)

Gopkg.toml

@@ -66,7 +66,7 @@
 [[override]]
   name = "github.com/lyft/flytestdlib"
   source = "https://github.com/lyft/flytestdlib"
-  version = "^v0.2.12"
+  version = "^v0.2.22"
 
 [[constraint]]
   name = "github.com/magiconair/properties"

cmd/entrypoints/clusterresource.go

@@ -3,9 +3,8 @@ package entrypoints
 import (
 	"context"
 
-	"github.com/lyft/flyteadmin/pkg/executioncluster"
-
 	"github.com/lyft/flyteadmin/pkg/clusterresource"
+	executioncluster "github.com/lyft/flyteadmin/pkg/executioncluster/impl"
 
 	"github.com/lyft/flyteadmin/pkg/runtime"
 
@@ -57,7 +56,7 @@ var controllerRunCmd = &cobra.Command{
 			scope.NewSubScope("cluster"),
 			cfg.KubeConfig,
 			cfg.Master,
-			configuration.ClusterConfiguration())
+			configuration)
 
 		clusterResourceController := clusterresource.NewClusterResourceController(db, executionCluster, scope)
 		clusterResourceController.Run()
@@ -89,7 +88,7 @@ var controllerSyncCmd = &cobra.Command{
 			scope.NewSubScope("cluster"),
 			cfg.KubeConfig,
 			cfg.Master,
-			configuration.ClusterConfiguration())
+			configuration)
 
 		clusterResourceController := clusterresource.NewClusterResourceController(db, executionCluster, scope)
 		err := clusterResourceController.Sync(ctx)

pkg/clusterresource/controller.go

@@ -11,9 +11,9 @@ import (
 	"strings"
 	"time"
 
-	"sigs.k8s.io/controller-runtime/pkg/client"
+	"github.com/lyft/flyteadmin/pkg/executioncluster/interfaces"
 
-	"github.com/lyft/flyteadmin/pkg/executioncluster"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/lyft/flyteadmin/pkg/common"
 	"github.com/lyft/flyteidl/gen/pb-go/flyteidl/admin"
@@ -68,7 +68,7 @@ type NamespaceCache = map[NamespaceName]LastModTimeCache
 type controller struct {
 	db                     repositories.RepositoryInterface
 	config                 runtimeInterfaces.Configuration
-	executionCluster       executioncluster.ClusterInterface
+	executionCluster       interfaces.ClusterInterface
 	poller                 chan struct{}
 	metrics                controllerMetrics
 	lastAppliedTemplateDir string
@@ -340,7 +340,7 @@ func newMetrics(scope promutils.Scope) controllerMetrics {
 	}
 }
 
-func NewClusterResourceController(db repositories.RepositoryInterface, executionCluster executioncluster.ClusterInterface, scope promutils.Scope) Controller {
+func NewClusterResourceController(db repositories.RepositoryInterface, executionCluster interfaces.ClusterInterface, scope promutils.Scope) Controller {
 	config := runtime.NewConfigurationProvider()
 	return &controller{
 		db:               db,

pkg/executioncluster/execution_target.go

@@ -1,17 +1,17 @@
 package executioncluster
 
 import (
-	"github.com/lyft/flyteadmin/pkg/flytek8s"
-	runtime "github.com/lyft/flyteadmin/pkg/runtime/interfaces"
+	"github.com/lyft/flyteidl/gen/pb-go/flyteidl/core"
 	flyteclient "github.com/lyft/flytepropeller/pkg/client/clientset/versioned"
-	"github.com/lyft/flytestdlib/promutils"
-	"k8s.io/client-go/rest"
+	"github.com/lyft/flytestdlib/random"
+
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 // Spec to determine the execution target
 type ExecutionTargetSpec struct {
-	TargetID string
+	TargetID    string
+	ExecutionID *core.WorkflowExecutionIdentifier
 }
 
 // Client object of the target execution cluster
@@ -22,35 +22,6 @@ type ExecutionTarget struct {
 	Enabled     bool
 }
 
-func getRestClientFromKubeConfig(scope promutils.Scope, kubeConfiguration *rest.Config) (*flyteclient.Clientset, error) {
-	fc, err := flyteclient.NewForConfig(kubeConfiguration)
-	if err != nil {
-		scope.MustNewCounter(
-			"flyteclient_initialization_error",
-			"count of errors encountered initializing a flyte client from kube config").Inc()
-		return nil, err
-	}
-	return fc, nil
-}
-
-// Creates a new Execution target for a cluster based on config passed in.
-func NewExecutionTarget(scope promutils.Scope, k8sCluster runtime.ClusterConfig) (*ExecutionTarget, error) {
-	kubeConf, err := flytek8s.GetRestClientConfigForCluster(k8sCluster)
-	if err != nil {
-		return nil, err
-	}
-	flyteClient, err := getRestClientFromKubeConfig(scope, kubeConf)
-	if err != nil {
-		return nil, err
-	}
-	client, err := client.New(kubeConf, client.Options{})
-	if err != nil {
-		return nil, err
-	}
-	return &ExecutionTarget{
-		FlyteClient: flyteClient,
-		Client:      client,
-		ID:          k8sCluster.Name,
-		Enabled:     k8sCluster.Enabled,
-	}, nil
+func (e ExecutionTarget) Compare(to random.Comparable) bool {
+	return e.ID < to.(ExecutionTarget).ID
 }

pkg/executioncluster/impl/cluster_execution_target_provider.go

@@ -0,0 +1,46 @@
+package impl
+
+import (
+	"github.com/lyft/flyteadmin/pkg/executioncluster"
+	"github.com/lyft/flyteadmin/pkg/flytek8s"
+	runtime "github.com/lyft/flyteadmin/pkg/runtime/interfaces"
+	flyteclient "github.com/lyft/flytepropeller/pkg/client/clientset/versioned"
+	"github.com/lyft/flytestdlib/promutils"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type clusterExecutionTargetProvider struct{}
+
+// Creates a new Execution target for a cluster based on config passed in.
+func (c *clusterExecutionTargetProvider) GetExecutionTarget(scope promutils.Scope, k8sCluster runtime.ClusterConfig) (*executioncluster.ExecutionTarget, error) {
+	kubeConf, err := flytek8s.GetRestClientConfigForCluster(k8sCluster)
+	if err != nil {
+		return nil, err
+	}
+	flyteClient, err := getRestClientFromKubeConfig(scope, kubeConf)
+	if err != nil {
+		return nil, err
+	}
+	client, err := client.New(kubeConf, client.Options{})
+	if err != nil {
+		return nil, err
+	}
+	return &executioncluster.ExecutionTarget{
+		FlyteClient: flyteClient,
+		Client:      client,
+		ID:          k8sCluster.Name,
+		Enabled:     k8sCluster.Enabled,
+	}, nil
+}
+
+func getRestClientFromKubeConfig(scope promutils.Scope, kubeConfiguration *rest.Config) (*flyteclient.Clientset, error) {
+	fc, err := flyteclient.NewForConfig(kubeConfiguration)
+	if err != nil {
+		scope.MustNewCounter(
+			"flyteclient_initialization_error",
+			"count of errors encountered initializing a flyte client from kube config").Inc()
+		return nil, err
+	}
+	return fc, nil
+}

pkg/executioncluster/impl/factory.go

@@ -0,0 +1,24 @@
+package impl
+
+import (
+	executioncluster_interface "github.com/lyft/flyteadmin/pkg/executioncluster/interfaces"
+	"github.com/lyft/flyteadmin/pkg/runtime/interfaces"
+	"github.com/lyft/flytestdlib/promutils"
+)
+
+func GetExecutionCluster(scope promutils.Scope, kubeConfig, master string, config interfaces.Configuration) executioncluster_interface.ClusterInterface {
+	switch len(config.ClusterConfiguration().GetClusterConfigs()) {
+	case 0:
+		cluster, err := NewInCluster(scope, kubeConfig, master)
+		if err != nil {
+			panic(err)
+		}
+		return cluster
+	default:
+		cluster, err := NewRandomClusterSelector(scope, config.ClusterConfiguration(), &clusterExecutionTargetProvider{}, config.ApplicationConfiguration().GetDomainsConfig())
+		if err != nil {
+			panic(err)
+		}
+		return cluster
+	}
+}

pkg/executioncluster/in_cluster.go → pkg/executioncluster/impl/in_cluster.go

@@ -1,32 +1,34 @@
-package executioncluster
+package impl
 
 import (
 	"fmt"
 
+	"github.com/lyft/flyteadmin/pkg/executioncluster"
+	"github.com/lyft/flyteadmin/pkg/executioncluster/interfaces"
 	"github.com/lyft/flyteadmin/pkg/flytek8s"
 	"github.com/lyft/flytestdlib/promutils"
 	"github.com/pkg/errors"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type InCluster struct {
-	target ExecutionTarget
+	target executioncluster.ExecutionTarget
 }
 
-func (i InCluster) GetTarget(spec *ExecutionTargetSpec) (*ExecutionTarget, error) {
+func (i InCluster) GetTarget(spec *executioncluster.ExecutionTargetSpec) (*executioncluster.ExecutionTarget, error) {
 	if spec != nil && spec.TargetID != "" {
 		return nil, errors.New(fmt.Sprintf("remote target %s is not supported", spec.TargetID))
 	}
 	return &i.target, nil
 }
 
-func (i InCluster) GetAllValidTargets() []ExecutionTarget {
-	return []ExecutionTarget{
+func (i InCluster) GetAllValidTargets() []executioncluster.ExecutionTarget {
+	return []executioncluster.ExecutionTarget{
 		i.target,
 	}
 }
 
-func NewInCluster(scope promutils.Scope, kubeConfig, master string) (ClusterInterface, error) {
+func NewInCluster(scope promutils.Scope, kubeConfig, master string) (interfaces.ClusterInterface, error) {
 	clientConfig, err := flytek8s.GetRestClientConfig(kubeConfig, master, nil)
 	if err != nil {
 		return nil, err
@@ -40,7 +42,7 @@ func NewInCluster(scope promutils.Scope, kubeConfig, master string) (ClusterInte
 		return nil, err
 	}
 	return &InCluster{
-		target: ExecutionTarget{
+		target: executioncluster.ExecutionTarget{
 			Client:      client,
 			FlyteClient: flyteClient,
 		},

pkg/executioncluster/in_cluster_test.go → pkg/executioncluster/impl/in_cluster_test.go

@@ -1,14 +1,16 @@
-package executioncluster
+package impl
 
 import (
 	"testing"
 
+	"github.com/lyft/flyteadmin/pkg/executioncluster"
+
 	"github.com/stretchr/testify/assert"
 )
 
 func TestInClusterGetTarget(t *testing.T) {
 	cluster := InCluster{
-		target: ExecutionTarget{
+		target: executioncluster.ExecutionTarget{
 			ID: "t1",
 		},
 	}
@@ -19,15 +21,15 @@ func TestInClusterGetTarget(t *testing.T) {
 
 func TestInClusterGetRemoteTarget(t *testing.T) {
 	cluster := InCluster{
-		target: ExecutionTarget{},
+		target: executioncluster.ExecutionTarget{},
 	}
-	_, err := cluster.GetTarget(&ExecutionTargetSpec{TargetID: "t1"})
+	_, err := cluster.GetTarget(&executioncluster.ExecutionTargetSpec{TargetID: "t1"})
 	assert.EqualError(t, err, "remote target t1 is not supported")
 }
 
 func TestInClusterGetAllValidTargets(t *testing.T) {
 	cluster := InCluster{
-		target: ExecutionTarget{
+		target: executioncluster.ExecutionTarget{
 			ID: "t1",
 		},
 	}