Merge pull request #2635 from fisherxu/automated-cherry-pick-of-#2614-upstream-release-1.6

Automated cherry pick of #2614: remove syncKeeper in edgehub

3 files changed, 19 insertions, 221 deletions

diff --git a/edge/pkg/edgehub/edgehub.go b/edge/pkg/edgehub/edgehub.go
index 5e98ba2a9..01acc19f7 100644
--- a/edge/pkg/edgehub/edgehub.go
+++ b/edge/pkg/edgehub/edgehub.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/kubeedge/beehive/pkg/core"
 	beehiveContext "github.com/kubeedge/beehive/pkg/core/context"
-	"github.com/kubeedge/beehive/pkg/core/model"
 	"github.com/kubeedge/kubeedge/edge/pkg/common/modules"
 	"github.com/kubeedge/kubeedge/edge/pkg/edgehub/certificate"
 	"github.com/kubeedge/kubeedge/edge/pkg/edgehub/clients"
@@ -28,7 +27,6 @@ type EdgeHub struct {
 	certManager   certificate.CertManager
 	chClient      clients.Adapter
 	reconnectChan chan struct{}
-	syncKeeper    map[string]chan model.Message
 	keeperLock    sync.RWMutex
 	enable        bool
 }
@@ -36,7 +34,6 @@ type EdgeHub struct {
 func newEdgeHub(enable bool) *EdgeHub {
 	return &EdgeHub{
 		reconnectChan: make(chan struct{}),
-		syncKeeper:    make(map[string]chan model.Message),
 		enable:        enable,
 	}
 }
diff --git a/edge/pkg/edgehub/process.go b/edge/pkg/edgehub/process.go
index 173914834..0173296b3 100644
--- a/edge/pkg/edgehub/process.go
+++ b/edge/pkg/edgehub/process.go
@@ -37,47 +37,8 @@ func (eh *EdgeHub) initial() (err error) {
 	return nil
 }
 
-func (eh *EdgeHub) addKeepChannel(msgID string) chan model.Message {
-	eh.keeperLock.Lock()
-	defer eh.keeperLock.Unlock()
-
-	tempChannel := make(chan model.Message, 1)
-	eh.syncKeeper[msgID] = tempChannel
-
-	return tempChannel
-}
-
-func (eh *EdgeHub) deleteKeepChannel(msgID string) {
-	eh.keeperLock.Lock()
-	defer eh.keeperLock.Unlock()
-
-	delete(eh.syncKeeper, msgID)
-}
-
 func (eh *EdgeHub) isSyncResponse(msgID string) bool {
-	eh.keeperLock.RLock()
-	defer eh.keeperLock.RUnlock()
-
-	_, exist := eh.syncKeeper[msgID]
-	return exist
-}
-
-func (eh *EdgeHub) sendToKeepChannel(message model.Message) error {
-	eh.keeperLock.RLock()
-	defer eh.keeperLock.RUnlock()
-	channel, exist := eh.syncKeeper[message.GetParentID()]
-	if !exist {
-		klog.Errorf("failed to get sync keeper channel, messageID:%+v", message)
-		return fmt.Errorf("failed to get sync keeper channel, messageID:%+v", message)
-	}
-	// send response into synckeep channel
-	select {
-	case channel <- message:
-	default:
-		klog.Errorf("failed to send message to sync keep channel")
-		return fmt.Errorf("failed to send message to sync keep channel")
-	}
-	return nil
+	return msgID != ""
 }
 
 func (eh *EdgeHub) dispatch(message model.Message) error {
@@ -98,11 +59,13 @@ func (eh *EdgeHub) dispatch(message model.Message) error {
 	}
 
 	isResponse := eh.isSyncResponse(message.GetParentID())
-	if !isResponse {
-		beehiveContext.SendToGroup(md, message)
+	if isResponse {
+		beehiveContext.SendResp(message)
 		return nil
 	}
-	return eh.sendToKeepChannel(message)
+
+	beehiveContext.SendToGroup(md, message)
+	return nil
 }
 
 func (eh *EdgeHub) routeToEdge() {
@@ -138,24 +101,6 @@ func (eh *EdgeHub) sendToCloud(message model.Message) error {
 		return fmt.Errorf("failed to send message, error: %v", err)
 	}
 
-	syncKeep := func(message model.Message) {
-		tempChannel := eh.addKeepChannel(message.GetID())
-		sendTimer := time.NewTimer(time.Duration(config.Config.Heartbeat) * time.Second)
-		select {
-		case response := <-tempChannel:
-			sendTimer.Stop()
-			beehiveContext.SendResp(response)
-			eh.deleteKeepChannel(response.GetParentID())
-		case <-sendTimer.C:
-			klog.Warningf("timeout to receive response for message: %+v", message)
-			eh.deleteKeepChannel(message.GetID())
-		}
-	}
-
-	if message.IsSync() {
-		go syncKeep(message)
-	}
-
 	return nil
 }
 
diff --git a/edge/pkg/edgehub/process_test.go b/edge/pkg/edgehub/process_test.go
index 324d593fb..a7784bffd 100644
--- a/edge/pkg/edgehub/process_test.go
+++ b/edge/pkg/edgehub/process_test.go
@@ -33,59 +33,6 @@ import (
 	"github.com/kubeedge/kubeedge/edge/pkg/edgehub/config"
 )
 
-//TestAddKeepChannel() tests the addition of channel to the syncKeeper
-func TestAddKeepChannel(t *testing.T) {
-	beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel)
-	tests := []struct {
-		name  string
-		hub   *EdgeHub
-		msgID string
-	}{
-		{
-			name: "Adding a valid keep channel",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
-			msgID: "test",
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got := tt.hub.addKeepChannel(tt.msgID)
-			if !reflect.DeepEqual(tt.hub.syncKeeper[tt.msgID], got) {
-				t.Errorf("TestController_addKeepChannel() = %v, want %v", got, tt.hub.syncKeeper[tt.msgID])
-			}
-		})
-	}
-}
-
-//TestDeleteKeepChannel() tests the deletion of channel in the syncKeeper
-func TestDeleteKeepChannel(t *testing.T) {
-	beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel)
-	tests := []struct {
-		name  string
-		hub   *EdgeHub
-		msgID string
-	}{
-		{
-			name: "Deleting a valid keep channel",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
-			msgID: "test",
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			tt.hub.addKeepChannel(tt.msgID)
-			tt.hub.deleteKeepChannel(tt.msgID)
-			if _, exist := tt.hub.syncKeeper[tt.msgID]; exist {
-				t.Errorf("TestController_deleteKeepChannel = %v, want %v", tt.hub.syncKeeper[tt.msgID], nil)
-			}
-		})
-	}
-}
-
 //TestIsSyncResponse() tests whether there exists a channel with the given message_id in the syncKeeper
 func TestIsSyncResponse(t *testing.T) {
 	beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel)
@@ -96,27 +43,20 @@ func TestIsSyncResponse(t *testing.T) {
 		want  bool
 	}{
 		{
-			name: "Sync message response case",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
+			name:  "Sync message response case",
+			hub:   &EdgeHub{},
 			msgID: "test",
 			want:  true,
 		},
 		{
-			name: "Non sync message response  case",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
+			name:  "Non sync message response  case",
+			hub:   &EdgeHub{},
 			msgID: "",
 			want:  false,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if tt.want {
-				tt.hub.addKeepChannel(tt.msgID)
-			}
 			if got := tt.hub.isSyncResponse(tt.msgID); got != tt.want {
 				t.Errorf("TestController_isSyncResponse() = %v, want %v", got, tt.want)
 			}
@@ -124,54 +64,6 @@ func TestIsSyncResponse(t *testing.T) {
 	}
 }
 
-//TestSendToKeepChannel() tests the reception of response in the syncKeep channel
-func TestSendToKeepChannel(t *testing.T) {
-	beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel)
-	message := model.NewMessage("test_id")
-	tests := []struct {
-		name                string
-		hub                 *EdgeHub
-		message             *model.Message
-		keepChannelParentID string
-		expectedError       error
-	}{
-		{
-			name: "SyncKeeper Error Case in send to keep channel",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
-			message:             message,
-			keepChannelParentID: "wrong_id",
-			expectedError:       fmt.Errorf("failed to get sync keeper channel, messageID:%+v", *message),
-		},
-		{
-			name: "Send to keep channel with valid input",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
-			message:             model.NewMessage("test_id"),
-			keepChannelParentID: "test_id",
-			expectedError:       nil,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			keep := tt.hub.addKeepChannel(tt.keepChannelParentID)
-			if tt.expectedError == nil {
-				receive := func() {
-					<-keep
-				}
-				go receive()
-			}
-			time.Sleep(1 * time.Second)
-			err := tt.hub.sendToKeepChannel(*tt.message)
-			if !reflect.DeepEqual(err, tt.expectedError) {
-				t.Errorf("TestController_sendToKeepChannel() error = %v, expectedError %v", err, tt.expectedError)
-			}
-		})
-	}
-}
-
 //TestDispatch() tests whether the messages are properly dispatched to their respective modules
 func TestDispatch(t *testing.T) {
 	beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel)
@@ -183,28 +75,22 @@ func TestDispatch(t *testing.T) {
 		isResponse    bool
 	}{
 		{
-			name: "dispatch with valid input",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
+			name:          "dispatch with valid input",
+			hub:           &EdgeHub{},
 			message:       model.NewMessage("").BuildRouter(ModuleNameEdgeHub, module.TwinGroup, "", ""),
 			expectedError: nil,
 			isResponse:    false,
 		},
 		{
-			name: "Error Case in dispatch",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
+			name:          "Error Case in dispatch",
+			hub:           &EdgeHub{},
 			message:       model.NewMessage("test").BuildRouter(ModuleNameEdgeHub, module.EdgedGroup, "", ""),
 			expectedError: fmt.Errorf("msg_group not found"),
 			isResponse:    true,
 		},
 		{
-			name: "Response Case in dispatch",
-			hub: &EdgeHub{
-				syncKeeper: make(map[string]chan model.Message),
-			},
+			name:          "Response Case in dispatch",
+			hub:           &EdgeHub{},
 			message:       model.NewMessage("test").BuildRouter(ModuleNameEdgeHub, module.TwinGroup, "", ""),
 			expectedError: nil,
 			isResponse:    true,
@@ -212,14 +98,6 @@ func TestDispatch(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if tt.expectedError == nil && !tt.isResponse {
-				receive := func() {
-					keepChannel := tt.hub.addKeepChannel(tt.message.GetParentID())
-					<-keepChannel
-				}
-				go receive()
-			}
-			time.Sleep(1 * time.Second)
 			err := tt.hub.dispatch(*tt.message)
 			if !reflect.DeepEqual(err, tt.expectedError) {
 				t.Errorf("TestController_dispatch() error = %v, wantErr %v", err, tt.expectedError)
@@ -281,44 +159,37 @@ func TestSendToCloud(t *testing.T) {
 		hub             *EdgeHub
 		message         model.Message
 		expectedError   error
-		waitError       bool
 		mockError       error
 		HeartbeatPeriod int32
 	}{
 		{
 			name: "send to cloud with proper input",
 			hub: &EdgeHub{
-				chClient:   mockAdapter,
-				syncKeeper: make(map[string]chan model.Message),
+				chClient: mockAdapter,
 			},
 			HeartbeatPeriod: 6,
 			message:         *msg,
 			expectedError:   nil,
-			waitError:       false,
 			mockError:       nil,
 		},
 		{
 			name: "Wait Error in send to cloud",
 			hub: &EdgeHub{
-				chClient:   mockAdapter,
-				syncKeeper: make(map[string]chan model.Message),
+				chClient: mockAdapter,
 			},
 			HeartbeatPeriod: 3,
 			message:         *msg,
 			expectedError:   nil,
-			waitError:       true,
 			mockError:       nil,
 		},
 		{
 			name: "Send Failure in send to cloud",
 			hub: &EdgeHub{
-				chClient:   mockAdapter,
-				syncKeeper: make(map[string]chan model.Message),
+				chClient: mockAdapter,
 			},
 			HeartbeatPeriod: 3,
 			message:         model.Message{},
 			expectedError:   fmt.Errorf("failed to send message, error: Connection Refused"),
-			waitError:       false,
 			mockError:       errors.New("Connection Refused"),
 		},
 	}
@@ -326,25 +197,10 @@ func TestSendToCloud(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			mockAdapter.EXPECT().Send(gomock.Any()).Return(tt.mockError).Times(1)
 			config.Config.Heartbeat = tt.HeartbeatPeriod
-			if !tt.waitError && tt.expectedError == nil {
-				go tt.hub.sendToCloud(tt.message)
-				time.Sleep(1 * time.Second)
-				tempChannel := tt.hub.syncKeeper["test_id"]
-				tempChannel <- *model.NewMessage("test_id")
-				time.Sleep(1 * time.Second)
-				if _, exist := tt.hub.syncKeeper["test_id"]; exist {
-					t.Errorf("SendToCloud() error in receiving message")
-				}
-				return
-			}
 			err := tt.hub.sendToCloud(tt.message)
 			if !reflect.DeepEqual(err, tt.expectedError) {
 				t.Errorf("SendToCloud() error = %v, wantErr %v", err, tt.expectedError)
 			}
-			time.Sleep(time.Duration(tt.HeartbeatPeriod+2) * time.Second)
-			if _, exist := tt.hub.syncKeeper["test_id"]; exist {
-				t.Errorf("SendToCloud() error in waiting for timeout")
-			}
 		})
 	}
 }