active supernode send dfget request to standby supernode

Signed-off-by: yunfeiyangbuaa <[email protected]>

supernode/daemon/mgr/ha/etcd_tool.go

@@ -164,6 +164,7 @@ func (etcd *EtcdMgr) Close() error {
 
 // SendStandbySupernodesInfo sends standby supernode's info to etcd.and if the standby supernode if off,the etcd will will notice that.
 func (etcd *EtcdMgr) SendStandbySupernodesInfo(key string, value string, timeout int64) error {
+	var respchan <-chan *clientv3.LeaseKeepAliveResponse
 	kv := clientv3.NewKV(etcd.client)
 	lease := clientv3.NewLease(etcd.client)
 	leaseResp, e := lease.Grant(context.TODO(), timeout)
@@ -172,10 +173,15 @@ func (etcd *EtcdMgr) SendStandbySupernodesInfo(key string, value string, timeout
 		return e
 	}
 	etcd.standbyLeaseResp = leaseResp
-	if _, e := lease.KeepAlive(context.TODO(), leaseResp.ID); e != nil {
+	if respchan, e = lease.KeepAlive(context.TODO(), leaseResp.ID); e != nil {
 		logrus.Errorf("failed to send heart beat to etcd to renew the lease %v", e)
 		return e
 	}
+	go func() {
+		for {
+			<-respchan
+		}
+	}()
 	return nil
 }
 

supernode/daemon/mgr/ha/manager.go

@@ -138,7 +138,6 @@ func (ha *Manager) SendGetCopy(params string) error {
 		e := ha.copyAPI.Get(urlCopy, "")
 		if e != nil {
 			logrus.Errorf("failed to send get copy,err: %v", e)
-			return e
 		}
 	}
 	return nil
@@ -151,7 +150,6 @@ func (ha *Manager) SendPostCopy(req interface{}, path string) error {
 		_, _, e := ha.copyAPI.Post(url, req, 5*time.Second)
 		if e != nil {
 			logrus.Errorf("failed to send post copy,err: %v", e)
-			return e
 		}
 	}
 	return nil

supernode/server/0.3_bridge.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 
 	"github.com/dragonflyoss/Dragonfly/apis/types"
@@ -52,7 +53,11 @@ var resultMap = map[string]string{
 
 func (s *Server) registry(ctx context.Context, rw http.ResponseWriter, req *http.Request) (err error) {
 	reader := req.Body
+	fmt.Println("url:", req.URL)
 	request := &types.TaskRegisterRequest{}
+	if s.HaMgr.GetSupernodeStatus() == constants.SupernodeUseHaActive {
+		s.HaMgr.SendPostCopy(request, "/peer/registry")
+	}
 	if err := json.NewDecoder(reader).Decode(request); err != nil {
 		return errors.Wrap(errTypes.ErrInvalidValue, err.Error())
 	}
@@ -106,10 +111,13 @@ func (s *Server) registry(ctx context.Context, rw http.ResponseWriter, req *http
 }
 
 func (s *Server) pullPieceTask(ctx context.Context, rw http.ResponseWriter, req *http.Request) (err error) {
+	if s.HaMgr.GetSupernodeStatus() == constants.SupernodeUseHaActive {
+		s.HaMgr.SendGetCopy(req.URL.String())
+	}
 	params := req.URL.Query()
 	taskID := params.Get("taskId")
 	srcCID := params.Get("srcCid")
-
+	fmt.Println("url:", req.URL)
 	request := &types.PiecePullRequest{
 		DfgetTaskStatus: statusMap[params.Get("status")],
 		PieceRange:      params.Get("range"),
@@ -178,7 +186,11 @@ func (s *Server) pullPieceTask(ctx context.Context, rw http.ResponseWriter, req
 }
 
 func (s *Server) reportPiece(ctx context.Context, rw http.ResponseWriter, req *http.Request) (err error) {
+	if s.HaMgr.GetSupernodeStatus() == constants.SupernodeUseHaActive {
+		s.HaMgr.SendGetCopy(req.URL.String())
+	}
 	params := req.URL.Query()
+	fmt.Println("url:", req.URL)
 	taskID := params.Get("taskId")
 	srcCID := params.Get("cid")
 	dstCID := params.Get("dstCid")
@@ -206,6 +218,10 @@ func (s *Server) reportPiece(ctx context.Context, rw http.ResponseWriter, req *h
 }
 
 func (s *Server) reportServiceDown(ctx context.Context, rw http.ResponseWriter, req *http.Request) (err error) {
+	if s.HaMgr.GetSupernodeStatus() == constants.SupernodeUseHaActive {
+		s.HaMgr.SendGetCopy(req.URL.String())
+	}
+	fmt.Println("url:", req.URL)
 	params := req.URL.Query()
 	taskID := params.Get("taskId")
 	cID := params.Get("cid")