send pop out pipeline id to etcd, and watch pop out in master

Author: chengjoey

modules/pipeline/endpoints/pipeline.go

@@ -262,7 +262,7 @@ func (e *Endpoints) pipelineCancel(ctx context.Context, r *http.Request, vars ma
 	}
 
 	if e.reconciler.QueueManager != nil {
-		e.reconciler.QueueManager.PopOutPipelineFromQueue(pipelineID)
+		e.reconciler.QueueManager.SendPopOutPipelineIDToEtcd(pipelineID)
 	}
 
 	return httpserver.OkResp(nil)

modules/pipeline/pipengine/reconciler/before_listen.go

@@ -38,5 +38,8 @@ func (r *Reconciler) beforeListen(ctx context.Context) error {
 	go func() {
 		r.QueueManager.ListenUpdatePriorityPipelineIDsFromEtcd(ctx)
 	}()
+	go func() {
+		r.QueueManager.ListenPopOutPipelineIDFromEtcd(ctx)
+	}()
 	return nil
 }

modules/pipeline/pipengine/reconciler/queuemanage/manager/queue.go

@@ -31,8 +31,9 @@ import (
 )
 
 const (
-	etcdQueueWatchPrefix         = "/devops/pipeline/queue_manager/actions/update/"
-	etcdQueuePipelineWatchPrefix = "/devops/pipeline/queue_manager/actions/batch-update/"
+	etcdQueueWatchPrefix               = "/devops/pipeline/queue_manager/actions/update/"
+	etcdQueuePipelineWatchPrefix       = "/devops/pipeline/queue_manager/actions/batch-update/"
+	etcdQueuePopOutPipelineWatchPrefix = "/devops/pipeline/queue_manager/actions/pop-out-pipeline/"
 )
 
 var (
@@ -75,6 +76,18 @@ func (mgr *defaultManager) SendQueueToEtcd(queueID uint64) {
 	})
 }
 
+func (mgr *defaultManager) SendPopOutPipelineIDToEtcd(pipelineID uint64) {
+	_ = loop.New(loop.WithDeclineRatio(2), loop.WithDeclineLimit(time.Second*60)).Do(func() (abort bool, err error) {
+		err = mgr.js.Put(context.Background(), fmt.Sprintf("%s%d", etcdQueuePopOutPipelineWatchPrefix, pipelineID), nil)
+		if err != nil {
+			logrus.Errorf("%s: send pop out pipeline to etcd failed, pipelineID: %d, err: %v", defaultQueueManagerLogPrefix, pipelineID, err)
+			return false, err
+		}
+		logrus.Infof("%s: send pop out pipeline to etcd successfully, pipelineID: %d", defaultQueueManagerLogPrefix, pipelineID)
+		return true, nil
+	})
+}
+
 func (mgr *defaultManager) SendUpdatePriorityPipelineIDsToEtcd(queueID uint64, pipelineIDS []uint64) {
 	_ = loop.New(loop.WithDeclineRatio(2), loop.WithDeclineLimit(time.Second*60)).Do(func() (abort bool, err error) {
 		err = mgr.js.Put(context.Background(), fmt.Sprintf("%s%d", etcdQueuePipelineWatchPrefix, queueID), pipelineIDS)
@@ -99,7 +112,7 @@ func (mgr *defaultManager) ListenInputQueueFromEtcd(ctx context.Context) {
 				func(key string, _ interface{}, t storetypes.ChangeType) (_ error) {
 					go func() {
 						logrus.Infof("%s: watched a key change: %s, changeType", key, t.String())
-						queueID, err := parseQueueIDFromWatchedKey(key, etcdQueueWatchPrefix)
+						queueID, err := parseIDFromWatchedKey(key, etcdQueueWatchPrefix)
 						if err != nil {
 							logrus.Errorf("%s: failed to parse queueID from watched key, key: %s, err: %v", defaultQueueManagerLogPrefix, key, err)
 							return
@@ -133,7 +146,7 @@ func (mgr *defaultManager) ListenUpdatePriorityPipelineIDsFromEtcd(ctx context.C
 			_ = mgr.js.IncludeWatch().Watch(ctx, etcdQueuePipelineWatchPrefix, true, true, false, []uint64{},
 				func(key string, value interface{}, t storetypes.ChangeType) (_ error) {
 					logrus.Infof("%s: watched a key change: %s, value: %v, changeType", key, value, t.String())
-					queueID, err := parseQueueIDFromWatchedKey(key, etcdQueuePipelineWatchPrefix)
+					queueID, err := parseIDFromWatchedKey(key, etcdQueuePipelineWatchPrefix)
 					if err != nil {
 						logrus.Errorf("%s: failed to parse queueID from watched key, key: %s, err: %v", defaultQueueManagerLogPrefix, key, err)
 						return
@@ -164,7 +177,39 @@ func (mgr *defaultManager) ListenUpdatePriorityPipelineIDsFromEtcd(ctx context.C
 	}
 }
 
-func parseQueueIDFromWatchedKey(key string, prefixKey string) (uint64, error) {
-	pipelineIDStr := strutil.TrimPrefixes(key, prefixKey)
-	return strconv.ParseUint(pipelineIDStr, 10, 64)
+func (mgr *defaultManager) ListenPopOutPipelineIDFromEtcd(ctx context.Context) {
+	logrus.Infof("%s: start listen pop out pipeline id", defaultQueueManagerLogPrefix)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			_ = mgr.js.IncludeWatch().Watch(ctx, etcdQueuePopOutPipelineWatchPrefix, true, true, true, nil,
+				func(key string, _ interface{}, t storetypes.ChangeType) (_ error) {
+					go func() {
+						defer func() {
+							if err := mgr.js.Remove(ctx, key, nil); err != nil {
+								logrus.Errorf("%s: failed to delete pop out pipeline key, key: %s, err: %v", defaultQueueManagerLogPrefix, key, err)
+							}
+						}()
+						logrus.Infof("%s: watched a key change: %s, changeType", key, t.String())
+						pipelineID, err := parseIDFromWatchedKey(key, etcdQueuePopOutPipelineWatchPrefix)
+						if err != nil {
+							logrus.Errorf("%s: failed to parse pipelineID from watched key, key: %s, err: %v", defaultQueueManagerLogPrefix, key, err)
+							return
+						}
+
+						mgr.PopOutPipelineFromQueue(pipelineID)
+						logrus.Infof("%s: pop out pipeline from queue successfully, pipeline id: %d", defaultQueueManagerLogPrefix, pipelineID)
+					}()
+
+					return nil
+				})
+		}
+	}
+}
+
+func parseIDFromWatchedKey(key string, prefixKey string) (uint64, error) {
+	IDStr := strutil.TrimPrefixes(key, prefixKey)
+	return strconv.ParseUint(IDStr, 10, 64)
 }

modules/pipeline/pipengine/reconciler/queuemanage/manager/queue_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"bou.ke/monkey"
+	"github.com/stretchr/testify/assert"
 
 	"github.com/erda-project/erda/pkg/jsonstore"
 )
@@ -49,3 +50,26 @@ func TestSendPipelineIDS(t *testing.T) {
 		mgr.SendUpdatePriorityPipelineIDsToEtcd(1, []uint64{1, 2, 3})
 	})
 }
+
+func TestParseIDFromWatchedKey(t *testing.T) {
+	queueID, err := parseIDFromWatchedKey("/devops/pipeline/queue_manager/actions/update/123", etcdQueueWatchPrefix)
+	assert.NoError(t, err)
+	assert.Equal(t, uint64(123), queueID)
+
+	pipelineID, err := parseIDFromWatchedKey("/devops/pipeline/queue_manager/actions/pop-out-pipeline/1001", etcdQueuePopOutPipelineWatchPrefix)
+	assert.NoError(t, err)
+	assert.Equal(t, uint64(1001), pipelineID)
+}
+
+func TestSendPopOutPipelineIDToEtcd(t *testing.T) {
+	js := &jsonstore.JsonStoreImpl{}
+	pm := monkey.PatchInstanceMethod(reflect.TypeOf(js), "Put", func(j *jsonstore.JsonStoreImpl, ctx context.Context, key string, object interface{}) error {
+		return nil
+	})
+	defer pm.Unpatch()
+
+	mgr := New(context.Background(), WithJsClient(js))
+	t.Run("SendPopOutPipelineIDToEtcd", func(t *testing.T) {
+		mgr.SendPopOutPipelineIDToEtcd(123)
+	})
+}

modules/pipeline/pipengine/reconciler/queuemanage/types/manager.go

@@ -34,5 +34,7 @@ type QueueManager interface {
 	ListenInputQueueFromEtcd(ctx context.Context)
 	SendUpdatePriorityPipelineIDsToEtcd(queueID uint64, pipelineIDS []uint64)
 	ListenUpdatePriorityPipelineIDsFromEtcd(ctx context.Context)
+	SendPopOutPipelineIDToEtcd(pipelineID uint64)
+	ListenPopOutPipelineIDFromEtcd(ctx context.Context)
 	snapshot.Snapshot
 }