feat(runtime): remove runtime v1 from job manager (#890)

* feat(runtime): remove runtime v1 from job manager

* update ut for job manager

Author: D0m021ng

pkg/job/job_manager.go

@@ -18,7 +18,6 @@ package job
 
 import (
 	"fmt"
-	"os"
 	"sync"
 	"time"
 
@@ -28,7 +27,6 @@ import (
 	"github.com/PaddlePaddle/PaddleFlow/pkg/common/config"
 	"github.com/PaddlePaddle/PaddleFlow/pkg/common/schema"
 	"github.com/PaddlePaddle/PaddleFlow/pkg/job/api"
-	"github.com/PaddlePaddle/PaddleFlow/pkg/job/runtime"
 	"github.com/PaddlePaddle/PaddleFlow/pkg/job/runtime_v2"
 	"github.com/PaddlePaddle/PaddleFlow/pkg/metrics"
 	"github.com/PaddlePaddle/PaddleFlow/pkg/model"
@@ -40,8 +38,6 @@ const (
 	defaultCacheSize  = 500
 	defaultExpireTime = 30
 	defaultJobLoop    = 1
-	// EnvRuntimeVersion contains the version of PaddleFlow runtime
-	EnvRuntimeVersion = "PF_RUNTIME_VERSION"
 )
 
 type ActiveClustersFunc func() []model.ClusterInfo
@@ -65,8 +61,6 @@ type JobManagerImpl struct {
 	// clusterRuntimes contains cluster status and runtime services
 	clusterRuntimes   ClusterRuntimes
 	clusterSyncPeriod time.Duration
-
-	isRuntimeV2 bool
 }
 
 func NewJobManagerImpl() (*JobManagerImpl, error) {
@@ -109,50 +103,7 @@ func (m *JobManagerImpl) Start(activeClusters ActiveClustersFunc, activeQueueJob
 	/// init config for job manager
 	m.init()
 	// start job manager
-	rVersion := os.Getenv(EnvRuntimeVersion)
-	if rVersion == "v1" {
-		m.startRuntime()
-	} else {
-		m.isRuntimeV2 = true
-		m.startRuntimeV2()
-	}
-}
-
-func (m *JobManagerImpl) startRuntime() {
-	log.Infof("Start job manager on runtime!")
-	// submit job to cluster
-	go m.pJobProcessLoop()
-
-	for {
-		// get active clusters
-		clusters := m.activeClusters()
-
-		for _, cluster := range clusters {
-			clusterID := api.ClusterID(cluster.ID)
-			// skip when cluster status is offline
-			if cluster.Status == model.ClusterStatusOffLine {
-				log.Warnf("cluster[%s] status is %s, skip it", cluster.ID, model.ClusterStatusOffLine)
-				m.stopClusterRuntime(clusterID)
-				continue
-			}
-
-			_, find := m.clusterRuntimes.Get(clusterID)
-			if !find {
-				runtimeSvc, err := runtime.GetOrCreateRuntime(cluster)
-				if err != nil {
-					log.Errorf("new runtime for cluster[%s] failed, err: %v. skip it", cluster.ID, err)
-					continue
-				}
-				log.Infof("Create new runtime with cluster <%s>", cluster.ID)
-
-				cr := NewClusterRuntimeInfo(cluster.Name, runtimeSvc)
-				m.clusterRuntimes.Store(clusterID, cr)
-				// start runtime for new cluster
-				go m.Run(runtimeSvc, cr.StopCh, clusterID)
-			}
-		}
-		time.Sleep(m.clusterSyncPeriod)
-	}
+	m.startRuntime()
 }
 
 func (m *JobManagerImpl) stopClusterRuntime(clusterID api.ClusterID) {
@@ -163,21 +114,10 @@ func (m *JobManagerImpl) stopClusterRuntime(clusterID api.ClusterID) {
 		close(cr.StopCh)
 	}
 	m.clusterRuntimes.Delete(clusterID)
-	runtime.PFRuntimeMap.Delete(clusterID)
 	runtime_v2.PFRuntimeMap.Delete(clusterID)
 	m.stopClusterQueueSubmit(clusterID)
 }
 
-func (m *JobManagerImpl) Run(runtimeService runtime.RuntimeService, stopCh <-chan struct{}, clusterID api.ClusterID) {
-	log.Infof("Start %s!", runtimeService.Name())
-	// start queue sync
-	go runtimeService.SyncQueue(stopCh)
-	// start job sync
-	go runtimeService.SyncJob(stopCh)
-	// start job gc
-	go runtimeService.GCJob(stopCh)
-}
-
 func (m *JobManagerImpl) pJobProcessLoop() {
 	log.Infof("start job process loop ...")
 	for {
@@ -245,27 +185,19 @@ func (m *JobManagerImpl) pSubmitQueueJob(jobQueue *api.JobQueue, clusterRuntime
 				log.Infof("Leaving submit %s job in queue %s, total elapsed time: %s", job.ID, name, time.Since(startTime))
 			} else {
 				// TODO: add to config
-				// time.Sleep(m.jobLoopPeriod)
 				time.Sleep(200 * time.Millisecond)
 			}
 		}
 	}
 }
 
-func (m *JobManagerImpl) submitJob(clusterRuntime *ClusterRuntimeInfo, job *api.PFJob) {
-	if clusterRuntime == nil || job == nil {
+// submitJob submit a job to cluster
+func (m *JobManagerImpl) submitJob(clusterRuntime *ClusterRuntimeInfo, jobInfo *api.PFJob) {
+	if clusterRuntime == nil || jobInfo == nil {
 		log.Errorf("submit job to cluster failed, err: clusterRuntime or job is nil")
 		return
 	}
-	if m.isRuntimeV2 {
-		m.submitJobV1(clusterRuntime.RuntimeV2Svc.SubmitJob, job)
-	} else {
-		m.submitJobV1(clusterRuntime.RuntimeSvc.SubmitJob, job)
-	}
-}
 
-// submitJob submit a job to cluster
-func (m *JobManagerImpl) submitJobV1(jobSubmit func(*api.PFJob) error, jobInfo *api.PFJob) {
 	log.Infof("begin to submit job %s to cluster", jobInfo.ID)
 	startTime := time.Now()
 	job, err := storage.Job.GetJobByID(jobInfo.ID)
@@ -277,7 +209,7 @@ func (m *JobManagerImpl) submitJobV1(jobSubmit func(*api.PFJob) error, jobInfo *
 	if job.Status == schema.StatusJobInit {
 		var jobStatus schema.JobStatus
 		var msg string
-		err = jobSubmit(jobInfo)
+		err = clusterRuntime.RuntimeSvc.SubmitJob(jobInfo)
 		if err != nil {
 			// new job failed, update db and skip this job
 			msg = fmt.Sprintf("submit job to cluster failed, err: %s", err)
@@ -320,8 +252,8 @@ func (m *JobManagerImpl) stopQueueSubmit(queueID api.QueueID) {
 }
 
 // PaddleFlow runtime v2
-// startRuntimeV2 start job manager on runtime v2
-func (m *JobManagerImpl) startRuntimeV2() {
+// startRuntime start job manager on runtime v2
+func (m *JobManagerImpl) startRuntime() {
 	log.Infof("Start job manager on runtime v2!")
 	// submit job to cluster
 	go m.pJobProcessLoop()
@@ -348,7 +280,7 @@ func (m *JobManagerImpl) startRuntimeV2() {
 				}
 				log.Infof("Create new runtime with cluster <%s>", cluster.ID)
 
-				cr := NewClusterRuntimeV2Info(cluster.Name, runtimeSvc)
+				cr := NewClusterRuntimeInfo(cluster.Name, runtimeSvc)
 				m.clusterRuntimes.Store(clusterID, cr)
 				// start runtime for new cluster
 				go runtimeSvc.SyncController(cr.StopCh)
@@ -407,28 +339,19 @@ func (m *JobManagerImpl) GetQueue(queueID api.QueueID) (*clusterQueue, bool) {
 
 // ClusterRuntimeInfo defines cluster runtime
 type ClusterRuntimeInfo struct {
-	Name         string
-	StopCh       chan struct{}
-	RuntimeSvc   runtime.RuntimeService
-	RuntimeV2Svc runtime_v2.RuntimeService
+	Name       string
+	StopCh     chan struct{}
+	RuntimeSvc runtime_v2.RuntimeService
 }
 
-func NewClusterRuntimeInfo(name string, r runtime.RuntimeService) *ClusterRuntimeInfo {
+func NewClusterRuntimeInfo(name string, r runtime_v2.RuntimeService) *ClusterRuntimeInfo {
 	return &ClusterRuntimeInfo{
 		Name:       name,
 		StopCh:     make(chan struct{}),
 		RuntimeSvc: r,
 	}
 }
 
-func NewClusterRuntimeV2Info(name string, r runtime_v2.RuntimeService) *ClusterRuntimeInfo {
-	return &ClusterRuntimeInfo{
-		Name:         name,
-		StopCh:       make(chan struct{}),
-		RuntimeV2Svc: r,
-	}
-}
-
 // ClusterRuntimes contains cluster runtimes
 type ClusterRuntimes struct {
 	sync.RWMutex

pkg/job/job_manager_test.go

@@ -17,44 +17,90 @@ limitations under the License.
 package job
 
 import (
+	"reflect"
 	"testing"
 	"time"
 
+	"github.com/agiledragon/gomonkey/v2"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/PaddlePaddle/PaddleFlow/pkg/common/config"
+	"github.com/PaddlePaddle/PaddleFlow/pkg/common/schema"
+	"github.com/PaddlePaddle/PaddleFlow/pkg/job/api"
+	runtime "github.com/PaddlePaddle/PaddleFlow/pkg/job/runtime_v2"
 	"github.com/PaddlePaddle/PaddleFlow/pkg/model"
 	"github.com/PaddlePaddle/PaddleFlow/pkg/storage"
 	"github.com/PaddlePaddle/PaddleFlow/pkg/storage/driver"
 )
 
+const (
+	mockQueueID   = "queue-test"
+	mockClusterID = "cluster-test"
+)
+
 func TestJobManager(t *testing.T) {
 	config.GlobalServerConfig = &config.ServerConfig{}
 
-	clusterInfo := &model.ClusterInfo{
-		Name:   "test-cluster",
-		Status: model.ClusterStatusOnLine,
+	mockCluster := &model.ClusterInfo{
+		Model: model.Model{
+			ID: mockClusterID,
+		},
+		Name:        "test-cluster",
+		ClusterType: schema.KubernetesType,
+		Status:      model.ClusterStatusOnLine,
+	}
+	mockQueue := &model.Queue{
+		Model: model.Model{
+			ID: mockQueueID,
+		},
+		Status:    schema.StatusQueueOpen,
+		ClusterId: mockClusterID,
 	}
 	jobM, err := NewJobManagerImpl()
 	assert.Equal(t, nil, err)
 	testCases := []struct {
 		name       string
 		jobManager *JobManagerImpl
+		job        *model.Job
 		err        error
 	}{
 		{
 			name:       "start job manager v2",
 			jobManager: jobM,
-			err:        nil,
+			job: &model.Job{
+				Name:    "test-job",
+				Status:  schema.StatusJobInit,
+				QueueID: mockQueueID,
+				Config:  &schema.Conf{},
+			},
+			err: nil,
 		},
 	}
 
 	driver.InitMockDB()
-	err = storage.Cluster.CreateCluster(clusterInfo)
+	err = storage.Cluster.CreateCluster(mockCluster)
+	assert.Equal(t, nil, err)
+	err = storage.Queue.CreateQueue(mockQueue)
 	assert.Equal(t, nil, err)
+	// mock cluster
+	rts := &runtime.KubeRuntime{}
+	var p1 = gomonkey.ApplyPrivateMethod(reflect.TypeOf(rts), "Init", func() error {
+		return nil
+	})
+	defer p1.Reset()
+	var p2 = gomonkey.ApplyPrivateMethod(reflect.TypeOf(rts), "SyncController", func(<-chan struct{}) error {
+		return nil
+	})
+	defer p2.Reset()
+	var p3 = gomonkey.ApplyPrivateMethod(reflect.TypeOf(rts), "SubmitJob", func(*api.PFJob) error {
+		return nil
+	})
+	defer p3.Reset()
 
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
+			err = storage.Job.CreateJob(testCase.job)
+			assert.Equal(t, nil, err)
 			go testCase.jobManager.Start(storage.Cluster.ActiveClusters, storage.Job.ListQueueJob)
 			time.Sleep(2 * time.Second)
 		})