Feat: Allow using in-cluster creds in control plane cluster in a multi-cluster deployment

docs/deployment/deployment/multicluster.rst

@@ -386,8 +386,22 @@ label has to be 1.
 
 .. note:: 
    This step will disable ``flytepropeller`` in the control plane cluster, leaving no possibility of running workflows there. If you require
-   the control plane to run workflows, edit the ``values-controlplane.yaml`` file and set ``flytepropeller.enabled`` to ``true``. Then, perform the ``helm upgrade`` operation and complete the steps in :ref:`this section <dataplane-deployment>` to configure it 
-   as a dataplane cluster.
+   the control plane to run workflows, edit the ``values-controlplane.yaml`` file and set ``flytepropeller.enabled`` to ``true`` and add one
+   additional cluster config for the control plane cluster itself:
+
+   .. code-block:: yaml
+      :caption: values-override.yaml
+
+      configmap:
+         clusters:
+            clusterConfigs:
+            - name: "dataplane_1"
+              ...
+            - name: "controlplane"
+              enabled: true
+              inCluster: true  # Use in-cluster credentials
+
+   Then, perform the ``helm upgrade`` operation.
 
 .. tab-set::
 

flyteadmin/pkg/flytek8s/client.go

@@ -54,7 +54,7 @@ func GetRestClientConfig(kubeConfigPathString, master string,
 			return nil, errors.NewFlyteAdminErrorf(codes.InvalidArgument, "Error building kubeconfig: %v", err)
 		}
 		logger.Debugf(context.Background(), "successfully loaded kube config from %s", kubeConfigPathString)
-	} else if k8sCluster != nil {
+	} else if k8sCluster != nil && !k8sCluster.InCluster {
 		kubeConfiguration, err = RemoteClusterConfig(k8sCluster.Endpoint, k8sCluster.Auth)
 		if err != nil {
 			return nil, err

flyteadmin/pkg/runtime/config_provider_test.go

@@ -32,22 +32,31 @@ func TestClusterConfig(t *testing.T) {
 	configProvider := NewConfigurationProvider()
 	clusterConfig := configProvider.ClusterConfiguration()
 	clusters := clusterConfig.GetClusterConfigs()
-	assert.Equal(t, 2, len(clusters))
+	assert.Equal(t, 3, len(clusters))
 
 	assert.Equal(t, "testcluster", clusters[0].Name)
 	assert.Equal(t, "testcluster_endpoint", clusters[0].Endpoint)
 	assert.Equal(t, "/path/to/testcluster/cert", clusters[0].Auth.CertPath)
 	assert.Equal(t, "/path/to/testcluster/token", clusters[0].Auth.TokenPath)
 	assert.Equal(t, "file_path", clusters[0].Auth.Type)
 	assert.False(t, clusters[0].Enabled)
+	assert.Equal(t, false, clusters[0].InCluster)
 
 	assert.Equal(t, "testcluster2", clusters[1].Name)
 	assert.Equal(t, "testcluster2_endpoint", clusters[1].Endpoint)
 	assert.Equal(t, "/path/to/testcluster2/cert", clusters[1].Auth.CertPath)
 	assert.Equal(t, "/path/to/testcluster2/token", clusters[1].Auth.TokenPath)
 	assert.True(t, clusters[1].Enabled)
-
 	assert.Equal(t, "file_path", clusters[1].Auth.Type)
+	assert.Equal(t, false, clusters[1].InCluster)
+
+	assert.Equal(t, "testcluster3", clusters[2].Name)
+	assert.Equal(t, "", clusters[2].Endpoint)
+	assert.Equal(t, "", clusters[2].Auth.CertPath)
+	assert.Equal(t, "", clusters[2].Auth.TokenPath)
+	assert.True(t, clusters[2].Enabled)
+	assert.Equal(t, "", clusters[2].Auth.Type)
+	assert.Equal(t, true, clusters[2].InCluster)
 }
 
 func TestGetCloudEventsConfig(t *testing.T) {

flyteadmin/pkg/runtime/interfaces/cluster_configuration.go

@@ -15,6 +15,7 @@ type ClusterConfig struct {
 	Auth             Auth                     `json:"auth"`
 	Enabled          bool                     `json:"enabled"`
 	KubeClientConfig *config.KubeClientConfig `json:"kubeClientConfig,omitempty"`
+	InCluster        bool                     `json:"inCluster"`
 }
 
 type Auth struct {

flyteadmin/pkg/runtime/testdata/clusters_config.yaml

@@ -13,3 +13,6 @@ clusters:
       type: "file_path"
       tokenPath: "/path/to/testcluster2/token"
       certPath: "/path/to/testcluster2/cert"
+  - name: "testcluster3"
+    enabled: true
+    inCluster: true