feat(job): remove useless runtime client interface (#1151)

* feat(job): remove useless runtime client interface

* update single job

* add  ut

* fix ut

Author: D0m021ng

pkg/common/k8s/discovery.go

@@ -25,6 +25,7 @@ var DiscoveryHandlerFunc = http.HandlerFunc(func(w http.ResponseWriter, req *htt
 			APIResources: []metav1.APIResource{
 				{Name: "queues", Namespaced: false, Kind: "Queue"},
 				{Name: "elasticresourcequotas", Namespaced: false, Kind: "ElasticResourceQuota"},
+				{Name: "podgroups", Namespaced: true, Kind: "PodGroup"},
 			},
 		}
 	case "/apis/sparkoperator.k8s.io/v1beta2":

pkg/common/schema/job.go

@@ -28,18 +28,12 @@ type Framework string
 type MemberRole string
 
 const (
-	EnvJobType        = "PF_JOB_TYPE"
-	EnvJobQueueName   = "PF_JOB_QUEUE_NAME"
-	EnvJobQueueID     = "PF_JOB_QUEUE_ID"
-	EnvJobClusterName = "PF_JOB_CLUSTER_NAME"
-	EnvJobClusterID   = "PF_JOB_CLUSTER_ID"
-	EnvJobNamespace   = "PF_JOB_NAMESPACE"
-	EnvJobUserName    = "PF_USER_NAME"
-	EnvJobFsID        = "PF_FS_ID"
-	EnvJobPVCName     = "PF_JOB_PVC_NAME"
-	EnvJobPriority    = "PF_JOB_PRIORITY"
-	EnvJobMode        = "PF_JOB_MODE"
-	EnvJobFramework   = "PF_JOB_FRAMEWORK"
+	EnvJobType      = "PF_JOB_TYPE"
+	EnvJobQueueName = "PF_JOB_QUEUE_NAME"
+	EnvJobNamespace = "PF_JOB_NAMESPACE"
+	EnvJobUserName  = "PF_USER_NAME"
+	EnvJobMode      = "PF_JOB_MODE"
+	EnvJobFramework = "PF_JOB_FRAMEWORK"
 	// EnvJobYamlPath Additional configuration for a specific job
 	EnvJobYamlPath  = "PF_JOB_YAML_PATH"
 	EnvIsCustomYaml = "PF_IS_CUSTOM_YAML"
@@ -52,32 +46,17 @@ const (
 	EnvEnableJobQueueSync = "PF_JOB_QUEUE_SYNC"
 
 	// EnvJobModePS env
-	EnvJobModePS          = "PS"
-	EnvJobPSPort          = "PF_JOB_PS_PORT"
-	EnvJobPServerReplicas = "PF_JOB_PSERVER_REPLICAS"
-	EnvJobPServerFlavour  = "PF_JOB_PSERVER_FLAVOUR"
-	EnvJobPServerCommand  = "PF_JOB_PSERVER_COMMAND"
-	EnvJobWorkerReplicas  = "PF_JOB_WORKER_REPLICAS"
-	EnvJobWorkerFlavour   = "PF_JOB_WORKER_FLAVOUR"
-	EnvJobWorkerCommand   = "PF_JOB_WORKER_COMMAND"
-
+	EnvJobModePS = "PS"
 	// EnvJobModeCollective env
 	EnvJobModeCollective   = "Collective"
-	EnvJobReplicas         = "PF_JOB_REPLICAS"
 	EnvJobFlavour          = "PF_JOB_FLAVOUR"
 	EnvJobLimitFlavour     = "PF_JOB_LIMIT_FLAVOUR"
 	EnvJobLimitFlavourNone = "NONE"
 
-	// EnvJobModePod env reuse EnvJobReplicas and EnvJobFlavour
-	EnvJobModePod = "Pod"
-
 	// spark job env
-	EnvJobSparkMainFile    = "PF_JOB_SPARK_MAIN_FILE"
-	EnvJobSparkMainClass   = "PF_JOB_SPARK_MAIN_CLASS"
-	EnvJobSparkArguments   = "PF_JOB_SPARK_ARGUMENTS"
-	EnvJobDriverFlavour    = "PF_JOB_DRIVER_FLAVOUR"
-	EnvJobExecutorReplicas = "PF_JOB_EXECUTOR_REPLICAS"
-	EnvJobExecutorFlavour  = "PF_JOB_EXECUTOR_FLAVOUR"
+	EnvJobSparkMainFile  = "PF_JOB_SPARK_MAIN_FILE"
+	EnvJobSparkMainClass = "PF_JOB_SPARK_MAIN_CLASS"
+	EnvJobSparkArguments = "PF_JOB_SPARK_ARGUMENTS"
 
 	// TODO move to framework
 	TypeVcJob      JobType = "vcjob"
@@ -155,9 +134,10 @@ const (
 	// JobKindGroupVersionAnnotation KindGroupVersion for job, format: {kind}.{group}/{version}
 	JobKindGroupVersionAnnotation = "paddleflow/job-kind-group-version"
 
-	VolcanoJobNameLabel  = "volcano.sh/job-name"
-	QueueLabelKey        = "volcano.sh/queue-name"
-	SparkAPPJobNameLabel = "sparkoperator.k8s.io/app-name"
+	VolcanoJobNameLabel     = "volcano.sh/job-name"
+	QueueLabelKey           = "volcano.sh/queue-name"
+	SchedulingQueueLabelKey = "scheduling.volcano.sh/queue-name"
+	SparkAPPJobNameLabel    = "sparkoperator.k8s.io/app-name"
 
 	QueueNamespaceAnnotation = "paddleflow/queue-namespace"
 

pkg/job/api/job_test.go

@@ -0,0 +1,60 @@
+/*
+Copyright (c) 2023 PaddlePaddle Authors. All Rights Reserve.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package api
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPFJob(t *testing.T) {
+	tfjob := PFJob{
+		ID:        "test-id",
+		Name:      "test-name",
+		Namespace: "default",
+	}
+	t.Run("test GetID func", func(t *testing.T) {
+		assert.Equal(t, tfjob.ID, tfjob.GetID())
+	})
+
+	t.Run("test NamespacedName func", func(t *testing.T) {
+		namespacedName := fmt.Sprintf("%s/%s", tfjob.Namespace, tfjob.ID)
+		assert.Equal(t, namespacedName, tfjob.NamespacedName())
+	})
+
+	t.Run("test UpdateLabels", func(t *testing.T) {
+		tfjob.UpdateLabels(map[string]string{
+			"a": "b",
+		})
+		assert.Equal(t, "b", tfjob.Labels["a"])
+	})
+
+	t.Run("test UpdateAnnotations", func(t *testing.T) {
+		tfjob.UpdateAnnotations(map[string]string{
+			"paddleflow/xx": "b",
+		})
+		assert.Equal(t, "b", tfjob.Annotations["paddleflow/xx"])
+	})
+
+	t.Run("test UpdateJobPriority", func(t *testing.T) {
+		newPC := "high"
+		tfjob.UpdateJobPriority(newPC)
+		assert.Equal(t, newPC, tfjob.PriorityClassName)
+	})
+}

pkg/job/api/queue_test.go

@@ -0,0 +1,94 @@
+/*
+Copyright (c) 2023 PaddlePaddle Authors. All Rights Reserve.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package api
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestQueueInfo_JobOrderFn(t *testing.T) {
+	currentTime := time.Now()
+	testCases := []struct {
+		name    string
+		left    *PFJob
+		right   *PFJob
+		wantAns bool
+	}{
+		{
+			name: "left < right",
+			left: &PFJob{
+				CreateTime: time.Now(),
+			},
+			right: &PFJob{
+				CreateTime: time.Now().Add(20 * time.Minute),
+			},
+			wantAns: true,
+		},
+		{
+			name: "left = right",
+			left: &PFJob{
+				ID:         "aa",
+				CreateTime: currentTime,
+			},
+			right: &PFJob{
+				ID:         "ab",
+				CreateTime: currentTime,
+			},
+			wantAns: true,
+		},
+	}
+
+	queueInfo := &QueueInfo{
+		SortPolicies: []SortPolicy{},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ans := queueInfo.JobOrderFn(tc.left, tc.right)
+			assert.Equal(t, tc.wantAns, ans)
+		})
+	}
+}
+
+func TestNewRegistry(t *testing.T) {
+	r := make(Registry)
+
+	pc := func(configuration Arguments) (SortPolicy, error) { return nil, nil }
+	t.Run("test registry", func(t *testing.T) {
+		err := r.Register("p1", pc)
+		assert.Equal(t, nil, err)
+
+		err = r.Register("p1", pc)
+		assert.Equal(t, fmt.Errorf("a sort policy named p1 already exists"), err)
+
+		err = r.Unregister("p1")
+		assert.Equal(t, nil, err)
+
+		err = r.Unregister("p2")
+		assert.Equal(t, fmt.Errorf("no sort policy named p2 exists"), err)
+	})
+
+	t.Run("test new registry", func(t *testing.T) {
+		QueueSortPolicies["priority"] = pc
+
+		ans := NewRegistry([]string{"priority"})
+		assert.Equal(t, 1, len(ans))
+	})
+}

pkg/job/runtime_v2/client/k3s_runtime_client.go

@@ -316,11 +316,6 @@ func (k3s *K3SRuntimeClient) startDynamicListener(listenerType string, stopCh <-
 	return err
 }
 
-// ListNodeQuota resource api for cluster nodes
-func (k3s *K3SRuntimeClient) ListNodeQuota(ctx context.Context) (pfschema.QuotaSummary, []pfschema.NodeQuotaInfo, error) {
-	return pfschema.QuotaSummary{}, nil, nil
-}
-
 func (k3s *K3SRuntimeClient) GetTaskLogV2(namespace, name string, logPage utils.LogPage) ([]pfschema.TaskLogInfo, error) {
 	return getTaskLogV2(k3s.Client, namespace, name, logPage)
 }

pkg/job/runtime_v2/client/k3s_runtime_client_test.go

@@ -124,16 +124,6 @@ func TestK3SRuntimeClient_Cluster(t *testing.T) {
 	assert.Contains(t, runtimeClient.ClusterName(), "mockNodeName")
 }
 
-func TestK3SRuntimeClient_ListNodeQuota(t *testing.T) {
-	// todo://add list node quota
-	var server = httptest.NewServer(k8s.DiscoveryHandlerFunc)
-	defer server.Close()
-	runtimeClient := NewFakeK3SRuntimeClient(server)
-	_, ret2, err := runtimeClient.ListNodeQuota(context.TODO())
-	assert.Nil(t, ret2)
-	assert.Nil(t, err)
-}
-
 func TestK3SExecutor(t *testing.T) {
 	var server = httptest.NewServer(k8s.DiscoveryHandlerFunc)
 	defer server.Close()

pkg/job/runtime_v2/client/kube_runtime_client.go

@@ -628,11 +628,6 @@ func (krc *KubeRuntimeClient) ClusterID() string {
 	return clusterID
 }
 
-func (krc *KubeRuntimeClient) ListNodeQuota(ctx context.Context) (pfschema.QuotaSummary, []pfschema.NodeQuotaInfo, error) {
-	// TODO: add ListNodeQuota logic
-	return pfschema.QuotaSummary{}, []pfschema.NodeQuotaInfo{}, nil
-}
-
 func (krc *KubeRuntimeClient) GetGVR(gvk schema.GroupVersionKind) (meta.RESTMapping, error) {
 	gvr, ok := krc.GVKToGVR.Load(gvk.String())
 	if ok {

pkg/job/runtime_v2/framework/interface.go

@@ -85,7 +85,4 @@ type RuntimeClientInterface interface {
 	RegisterListener(listenerType string, workQueue workqueue.RateLimitingInterface) error
 
 	StartListener(listenerType string, stopCh <-chan struct{}) error
-
-	// ListNodeQuota resource api for cluster nodes
-	ListNodeQuota(ctx context.Context) (pfschema.QuotaSummary, []pfschema.NodeQuotaInfo, error)
 }