Yaroslav/scheduler queue

app/config/config.go

@@ -17,7 +17,6 @@ type AppConfig struct {
 	CacheTTL time.Duration `json:"cacheTtl" pflag:",TTL for app status cache"`
 }
 
-
 const appConfigSectionKey = "apps"
 
 var defaultAppConfig = &AppConfig{

app/internal/k8s/app_client_test.go

@@ -581,7 +581,6 @@ func TestSubscribe_ReceivesEvent(t *testing.T) {
 	}
 }
 
-
 func TestSubscribe_AppSpecificDoesNotReceiveOtherApps(t *testing.T) {
 	c := testClient(t)
 	ch := c.Subscribe("app1")

app/internal/service/internal_app_service_test.go

@@ -15,8 +15,8 @@ import (
 	appconfig "github.com/flyteorg/flyte/v2/app/internal/config"
 	flyteapp "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/app"
 	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/app/appconnect"
-	flytecoreapp "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/core"
 	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/common"
+	flytecoreapp "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/core"
 )
 
 // mockAppK8sClient is a testify mock for AppK8sClientInterface.

app/internal/setup.go

@@ -11,7 +11,6 @@ import (
 	appconfig "github.com/flyteorg/flyte/v2/app/internal/config"
 	appk8s "github.com/flyteorg/flyte/v2/app/internal/k8s"
 	"github.com/flyteorg/flyte/v2/app/internal/service"
-	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/app/appconnect"
 	knativeapp "github.com/flyteorg/flyte/v2/flytestdlib/app"
 )
 

executor/pkg/controller/taskaction_controller_test.go

@@ -49,8 +49,10 @@ type fakePlugin struct {
 	abortCalls int
 }
 
-func (f *fakePlugin) GetID() string                              { return f.id }
-func (f *fakePlugin) GetProperties() pluginsCore.PluginProperties { return pluginsCore.PluginProperties{} }
+func (f *fakePlugin) GetID() string { return f.id }
+func (f *fakePlugin) GetProperties() pluginsCore.PluginProperties {
+	return pluginsCore.PluginProperties{}
+}
 func (f *fakePlugin) Handle(_ context.Context, _ pluginsCore.TaskExecutionContext) (pluginsCore.Transition, error) {
 	return pluginsCore.UnknownTransition, nil
 }
@@ -540,8 +542,8 @@ var _ = Describe("TaskAction Controller", func() {
 		BeforeEach(func() {
 			resource := &flyteorgv1.TaskAction{
 				ObjectMeta: metav1.ObjectMeta{
-					Name:      abortResourceName,
-					Namespace: "default",
+					Name:       abortResourceName,
+					Namespace:  "default",
 					Finalizers: []string{taskActionFinalizer},
 				},
 				Spec: flyteorgv1.TaskActionSpec{

executor/setup.go

@@ -18,7 +18,6 @@
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
-	"github.com/flyteorg/flyte/v2/flytestdlib/app"
 	flyteorgv1 "github.com/flyteorg/flyte/v2/executor/api/v1"
 	"github.com/flyteorg/flyte/v2/executor/pkg/config"
 	"github.com/flyteorg/flyte/v2/executor/pkg/controller"
@@ -28,6 +27,8 @@
 	"github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/catalog"
 	cachecatalog "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/catalog/cache_service"
 	webhookConfig "github.com/flyteorg/flyte/v2/flyteplugins/go/tasks/pluginmachinery/secret/config"
+	"github.com/flyteorg/flyte/v2/flytestdlib/app"
+	"git
 	"github.com/flyteorg/flyte/v2/flytestdlib/promutils"
 	"github.com/flyteorg/flyte/v2/flytestdlib/storage"
 	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow/workflowconnect"

flyteidl2/task/run.proto

@@ -84,9 +84,9 @@ message RunSpec {
   // Deprecated, please use CacheConfig.OverwriteCache instead
   bool overwrite_cache = 5 [deprecated = true];
 
-  // the specific cluster that this action should be executed on. this value will be used as the
-  // default for all actions in the run unless overridden.
-  string cluster = 6;
+  // the queue name that this run should be executed on.
+  // Queue name is unique within an org so it can be used as an id to reference the queue.
+  string queue = 6;
 
   // Encapsulates user settings pertaining to offloaded data (i.e. Blobs, Schema, query data, etc.).
   RawDataStorage raw_data_storage = 7;

flyteidl2/workflow/run_definition.proto

@@ -42,9 +42,9 @@ message TaskAction {
   // Enables caching when set and specifies the cache version to use.
   google.protobuf.StringValue cache_key = 3;
 
-  // the specific cluster that this action should be executed on. if not set, the cluster from the
-  // `RunSpec` will be used.
-  string cluster = 4;
+  // the specific queue that this action should be executed on.
+  // this value will be used to resolve the full queue id and store it in 'assigned_queue' field.
+  string queue = 4;
 }
 
 // TraceAction is used to define a trace action that can be used to track the execution of an action that's managed