watch etcd to update queue instead of in memory

modules/pipeline/endpoints/queue_management.go

@@ -148,7 +148,7 @@ func (e *Endpoints) updatePipelineQueue(ctx context.Context, r *http.Request, va
 	}
 
 	// update queue in manager
-	e.reconciler.QueueManager.IdempotentAddQueue(queue)
+	e.reconciler.QueueManager.SendQueueToEtcd(queue.ID)
 
 	return httpserver.OkResp(queue)
 }
@@ -204,9 +204,7 @@ func (e *Endpoints) batchUpgradePipelinePriority(ctx context.Context, r *http.Re
 		return errorresp.ErrResp(err)
 	}
 
-	if err = e.reconciler.QueueManager.BatchUpdatePipelinePriorityInQueue(queue, req.PipelineIDsOrderByPriorityFromHighToLow); err != nil {
-		return apierrors.ErrUpgradePipelinePriority.InternalError(err).ToResp(), nil
-	}
+	e.reconciler.QueueManager.SendUpdatePriorityPipelineIDsToEtcd(queue.ID, req.PipelineIDsOrderByPriorityFromHighToLow)
 
 	return httpserver.OkResp(nil)
 }

modules/pipeline/initialize.go

@@ -160,6 +160,9 @@ func (p *provider) do() error {
 	if err != nil {
 		return fmt.Errorf("failed to init reconciler, err: %v", err)
 	}
+	if err := r.LoadQueueManger(context.Background()); err != nil {
+		return fmt.Errorf("failed to load reconciler queue manager, err: %v", err)
+	}
 	if err := engine.OnceDo(r); err != nil {
 		return err
 	}

modules/pipeline/pipengine/reconciler/before_listen.go

@@ -22,8 +22,8 @@ import (
 )
 
 func (r *Reconciler) beforeListen(ctx context.Context) error {
-	// init before listen
-	if err := retry.DoWithInterval(func() error { return r.loadQueueManger(ctx) }, 3, time.Second*10); err != nil {
+	// init before listen, use new queue manager cover the initialize context
+	if err := retry.DoWithInterval(func() error { return r.LoadQueueManger(ctx) }, 3, time.Second*10); err != nil {
 		return err
 	}
 	if err := retry.DoWithInterval(func() error { return r.loadThrottler(ctx) }, 3, time.Second*10); err != nil {
@@ -32,5 +32,11 @@ func (r *Reconciler) beforeListen(ctx context.Context) error {
 	go func() {
 		r.continueBackupQueueUsage(ctx)
 	}()
+	go func() {
+		r.QueueManager.ListenInputQueueFromEtcd(ctx)
+	}()
+	go func() {
+		r.QueueManager.ListenUpdatePriorityPipelineIDsFromEtcd(ctx)
+	}()
 	return nil
 }

modules/pipeline/pipengine/reconciler/define.go

@@ -53,7 +53,7 @@ type Reconciler struct {
 	dbClient *dbclient.Client
 
 	QueueManager  types.QueueManager
-	TaskThrottler throttler.Throttler // TODO remove the throttler.Throttler, after 1.0 iteration throttler is not necessary
+	TaskThrottler throttler.Throttler // TODO remove the throttler.Throttler, after release/1.3 iteration throttler is not necessary
 
 	// processingTasks store task id which is in processing
 	processingTasks sync.Map

modules/pipeline/pipengine/reconciler/queue.go

@@ -17,6 +17,8 @@ package reconciler
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"strings"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -27,9 +29,9 @@ import (
 const logPrefixContinueBackupQueueUsage = "[queue usage backup]"
 
 // loadQueueManger
-func (r *Reconciler) loadQueueManger(ctx context.Context) error {
+func (r *Reconciler) LoadQueueManger(ctx context.Context) error {
 	// init queue manager
-	r.QueueManager = manager.New(ctx, manager.WithDBClient(r.dbClient), manager.WithEtcdClient(r.etcd))
+	r.QueueManager = manager.New(ctx, manager.WithDBClient(r.dbClient), manager.WithEtcdClient(r.etcd), manager.WithJsClient(r.js))
 
 	return nil
 }
@@ -48,13 +50,20 @@ func (r *Reconciler) continueBackupQueueUsage(ctx context.Context) {
 			queueSnapshot := manager.SnapshotObj{}
 			if err := json.Unmarshal(backup, &queueSnapshot); err != nil {
 				errDone <- err
+				return
 			}
+			errs := []string{}
 			for qID, qMsg := range queueSnapshot.QueueUsageByID {
 				if err := r.etcd.Put(ctx, manager.MakeQueueUsageBackupKey(qID), string(qMsg)); err != nil {
-					errDone <- err
-					return
+					errs = append(errs, fmt.Sprintf("%v", err))
+					continue
 				}
 			}
+			if len(errs) > 0 {
+				usageErr := fmt.Errorf(strings.Join(errs, ","))
+				errDone <- usageErr
+				return
+			}
 			done <- struct{}{}
 		}()
 

modules/pipeline/pipengine/reconciler/queue_test.go

@@ -16,6 +16,7 @@ package reconciler
 
 import (
 	"context"
+	"fmt"
 	"reflect"
 	"testing"
 	"time"
@@ -29,7 +30,7 @@ import (
 func TestContinueBackupQueueUsage(t *testing.T) {
 	etcdClient := &etcd.Store{}
 	pm := monkey.PatchInstanceMethod(reflect.TypeOf(etcdClient), "Put", func(j *etcd.Store, ctx context.Context, key string, value string) error {
-		return nil
+		return fmt.Errorf("failed to put msg")
 	})
 	defer pm.Unpatch()
 	q := manager.New(context.Background(), manager.WithEtcdClient(etcdClient))
@@ -54,4 +55,5 @@ func TestContinueBackupQueueUsage(t *testing.T) {
 		time.Sleep(2 * time.Second)
 		cancel()
 	})
+
 }

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

@@ -23,6 +23,7 @@ import (
 
 	"github.com/erda-project/erda/modules/pipeline/dbclient"
 	"github.com/erda-project/erda/modules/pipeline/pipengine/reconciler/queuemanage/types"
+	"github.com/erda-project/erda/pkg/jsonstore"
 	"github.com/erda-project/erda/pkg/jsonstore/etcd"
 )
 
@@ -37,6 +38,7 @@ type defaultManager struct {
 
 	dbClient *dbclient.Client
 	etcd     *etcd.Store
+	js       jsonstore.JsonStore
 }
 
 // New return a new queue manager.
@@ -80,3 +82,9 @@ func WithEtcdClient(etcdClient *etcd.Store) Option {
 		mgr.etcd = etcdClient
 	}
 }
+
+func WithJsClient(js jsonstore.JsonStore) Option {
+	return func(mgr *defaultManager) {
+		mgr.js = js
+	}
+}

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

@@ -15,9 +15,28 @@
 package manager
 
 import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/sirupsen/logrus"
+
 	"github.com/erda-project/erda/apistructs"
 	"github.com/erda-project/erda/modules/pipeline/pipengine/reconciler/queuemanage/queue"
 	"github.com/erda-project/erda/modules/pipeline/pipengine/reconciler/queuemanage/types"
+	"github.com/erda-project/erda/pkg/jsonstore/storetypes"
+	"github.com/erda-project/erda/pkg/loop"
+	"github.com/erda-project/erda/pkg/strutil"
+)
+
+const (
+	etcdQueueWatchPrefix         = "/devops/pipeline/queue_manager/actions/update/"
+	etcdQueuePipelineWatchPrefix = "/devops/pipeline/queue_manager/actions/batch-update/"
+)
+
+var (
+	defaultQueueManagerLogPrefix = "[default queue manager]"
 )
 
 // IdempotentAddQueue add to to manager idempotent.
@@ -43,3 +62,109 @@ func (mgr *defaultManager) IdempotentAddQueue(pq *apistructs.PipelineQueue) type
 
 	return newQueue
 }
+
+func (mgr *defaultManager) SendQueueToEtcd(queueID 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", etcdQueueWatchPrefix, queueID), nil)
+		if err != nil {
+			logrus.Errorf("%s: send to queue failed, err: %v", defaultQueueManagerLogPrefix, err)
+			return false, err
+		}
+		logrus.Infof("%s: queue id: %d add to queue success", defaultQueueManagerLogPrefix, queueID)
+		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)
+		if err != nil {
+			logrus.Errorf("%s: send to queue pipelines failed, err: %v", defaultQueueManagerLogPrefix, err)
+			return false, err
+		}
+		logrus.Infof("%s: queue id: %d add to queue pipelines success", defaultQueueManagerLogPrefix, queueID)
+		return true, nil
+	})
+}
+
+// Listen leader node should listen etcd, watch queue information
+func (mgr *defaultManager) ListenInputQueueFromEtcd(ctx context.Context) {
+	logrus.Infof("%s: start listen", defaultQueueManagerLogPrefix)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			_ = mgr.js.IncludeWatch().Watch(ctx, etcdQueueWatchPrefix, true, true, true, nil,
+				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)
+						if err != nil {
+							logrus.Errorf("%s: failed to parse queueID from watched key, key: %s, err: %v", defaultQueueManagerLogPrefix, key, err)
+							return
+						}
+						pq, exist, err := mgr.dbClient.GetPipelineQueue(queueID)
+						if err != nil {
+							logrus.Errorf("%s: failed to get queue, id: %d, err: %v", defaultQueueManagerLogPrefix, queueID, err)
+							return
+						}
+						if !exist {
+							logrus.Errorf("%s: queue not existed, id: %d, err: %v", defaultQueueManagerLogPrefix, queueID, err)
+							return
+						}
+
+						_ = mgr.IdempotentAddQueue(pq)
+					}()
+
+					return nil
+				})
+		}
+	}
+}
+
+func (mgr *defaultManager) ListenUpdatePriorityPipelineIDsFromEtcd(ctx context.Context) {
+	logrus.Infof("%s: start listen pipeline ids", defaultQueueManagerLogPrefix)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			_ = 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)
+					if err != nil {
+						logrus.Errorf("%s: failed to parse queueID from watched key, key: %s, err: %v", defaultQueueManagerLogPrefix, key, err)
+						return
+					}
+					pq, exist, err := mgr.dbClient.GetPipelineQueue(queueID)
+					if err != nil {
+						logrus.Errorf("%s: failed to get queue, id: %d, err: %v", defaultQueueManagerLogPrefix, queueID, err)
+						return
+					}
+					if !exist {
+						logrus.Errorf("%s: queue not existed, id: %d, err: %v", defaultQueueManagerLogPrefix, queueID, err)
+						return
+					}
+
+					pipelineIDS, ok := value.(*[]uint64)
+					if !ok {
+						logrus.Errorf("%s: failed convert value: %v to pipeline ids", defaultQueueManagerLogPrefix, value)
+						return
+					}
+					if err := mgr.BatchUpdatePipelinePriorityInQueue(pq, *pipelineIDS); err != nil {
+						logrus.Errorf("%s: failed to batch update pipeline priority in queue, err: %v", defaultQueueManagerLogPrefix, err)
+						return
+					}
+
+					return nil
+				})
+		}
+	}
+}
+
+func parseQueueIDFromWatchedKey(key string, prefixKey string) (uint64, error) {
+	pipelineIDStr := strutil.TrimPrefixes(key, prefixKey)
+	return strconv.ParseUint(pipelineIDStr, 10, 64)
+}

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

@@ -0,0 +1,51 @@
+// Copyright (c) 2021 Terminus, Inc.
+//
+// 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 manager
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	"bou.ke/monkey"
+
+	"github.com/erda-project/erda/pkg/jsonstore"
+)
+
+func TestSend(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("Send", func(t *testing.T) {
+		mgr.SendQueueToEtcd(1)
+	})
+}
+
+func TestSendPipelineIDS(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("Send", func(t *testing.T) {
+		mgr.SendUpdatePriorityPipelineIDsToEtcd(1, []uint64{1, 2, 3})
+	})
+}

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

@@ -39,7 +39,7 @@ func (mgr *defaultManager) Export() json.RawMessage {
 	return b
 }
 
-// Import default queue manager execute in memory, don't need import
+// Import default queue manager execute in memory, don't need load from database
 func (mgr *defaultManager) Import(rawMsg json.RawMessage) error {
 	return nil
 }

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

@@ -27,7 +27,7 @@ import (
 )
 
 func MakeQueueUsageBackupKey(qID string) string {
-	return fmt.Sprintf("/devops/pipeline/queue/reconciler/usage/%s", qID)
+	return fmt.Sprintf("/devops/pipeline/queue_manager/actions/usage/%s", qID)
 }
 
 func (mgr *defaultManager) QueryQueueUsage(pq *apistructs.PipelineQueue) *pb.QueueUsage {

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

@@ -15,6 +15,8 @@
 package types
 
 import (
+	"context"
+
 	"github.com/erda-project/erda-proto-go/pipeline/pb"
 	"github.com/erda-project/erda/apistructs"
 	"github.com/erda-project/erda/modules/pipeline/pipengine/queue/snapshot"
@@ -28,5 +30,9 @@ type QueueManager interface {
 	PopOutPipelineFromQueue(pipelineID uint64)
 	BatchUpdatePipelinePriorityInQueue(pq *apistructs.PipelineQueue, pipelineIDs []uint64) error
 	Stop()
+	SendQueueToEtcd(queueID uint64)
+	ListenInputQueueFromEtcd(ctx context.Context)
+	SendUpdatePriorityPipelineIDsToEtcd(queueID uint64, pipelineIDS []uint64)
+	ListenUpdatePriorityPipelineIDsFromEtcd(ctx context.Context)
 	snapshot.Snapshot
 }

modules/pipeline/pipengine/reconciler/throttler.go

@@ -50,7 +50,7 @@ func (r *Reconciler) loadThrottler(ctx context.Context) error {
 	logrus.Warnf("reconciler: failed to load throttler, ignore, import err: %v", err)
 	// load from database
 
-	// continueBackupThrottler after 1.0 iteration, throttler is not necessary
+	// continueBackupThrottler after release/1.3 iteration, throttler is not necessary
 	//r.continueBackupThrottler(ctx)
 	return nil
 }