diff options
26 files changed, 183 insertions, 213 deletions
diff --git a/cloud/pkg/cloudhub/cloudhub.go b/cloud/pkg/cloudhub/cloudhub.go index 324d04da7..3f954a212 100644 --- a/cloud/pkg/cloudhub/cloudhub.go +++ b/cloud/pkg/cloudhub/cloudhub.go @@ -9,6 +9,7 @@ import ( "github.com/kubeedge/beehive/pkg/common/config" "github.com/kubeedge/beehive/pkg/core" + beehiveContext "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/kubeedge/cloud/pkg/cloudhub/channelq" "github.com/kubeedge/kubeedge/cloud/pkg/cloudhub/common/util" chconfig "github.com/kubeedge/kubeedge/cloud/pkg/cloudhub/config" @@ -62,6 +63,7 @@ func (a *cloudHub) Start() { func (a *cloudHub) Cleanup() { a.cancel() + beehiveContext.Cleanup(a.Name()) } func initHubConfig() { diff --git a/cloud/pkg/devicecontroller/devicecontroller.go b/cloud/pkg/devicecontroller/devicecontroller.go index aa9ae25b5..48140cabb 100644 --- a/cloud/pkg/devicecontroller/devicecontroller.go +++ b/cloud/pkg/devicecontroller/devicecontroller.go @@ -8,6 +8,7 @@ import ( "k8s.io/klog" "github.com/kubeedge/beehive/pkg/core" + beehiveContext "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/kubeedge/cloud/pkg/devicecontroller/config" "github.com/kubeedge/kubeedge/cloud/pkg/devicecontroller/constants" "github.com/kubeedge/kubeedge/cloud/pkg/devicecontroller/controller" @@ -62,6 +63,7 @@ func (dctl *DeviceController) Start() { // Cleanup controller func (dctl *DeviceController) Cleanup() { dctl.cancel() + beehiveContext.Cleanup(dctl.Name()) } func initConfig() { diff --git a/cloud/pkg/edgecontroller/edgecontroller.go b/cloud/pkg/edgecontroller/edgecontroller.go index b5a268785..07e8a351d 100644 --- a/cloud/pkg/edgecontroller/edgecontroller.go +++ b/cloud/pkg/edgecontroller/edgecontroller.go @@ -7,6 +7,7 @@ import ( "k8s.io/klog" "github.com/kubeedge/beehive/pkg/core" + beehiveContext "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/kubeedge/cloud/pkg/edgecontroller/config" "github.com/kubeedge/kubeedge/cloud/pkg/edgecontroller/constants" "github.com/kubeedge/kubeedge/cloud/pkg/edgecontroller/controller" @@ -59,6 +60,7 @@ func (ctl *EdgeController) Start() { // Cleanup controller func (ctl *EdgeController) Cleanup() { ctl.cancel() + beehiveContext.Cleanup(ctl.Name()) } func initConfig() { diff --git a/edge/mocks/beehive/fake_module.go b/edge/mocks/beehive/fake_module.go index 5fa00359e..ecc3f9267 100644 --- a/edge/mocks/beehive/fake_module.go +++ b/edge/mocks/beehive/fake_module.go @@ -1,15 +1,13 @@ // Code generated by MockGen. DO NOT EDIT. -// Source: github.com/kubeedge/kubeedge/beehive/pkg/core (interfaces: Module) +// Source: staging/src/github.com/kubeedge/beehive/pkg/core/module.go -// Package beehive is a generated GoMock package. +// Package mock_core is a generated GoMock package. package beehive import ( reflect "reflect" gomock "github.com/golang/mock/gomock" - - context "github.com/kubeedge/beehive/pkg/core/context" ) // MockModule is a mock of Module interface @@ -35,18 +33,23 @@ func (m *MockModule) EXPECT() *MockModuleMockRecorder { return m.recorder } -// Cleanup mocks base method -func (m *MockModule) Cleanup() { - m.ctrl.Call(m, "Cleanup") +// Name mocks base method +func (m *MockModule) Name() string { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Name") + ret0, _ := ret[0].(string) + return ret0 } -// Cleanup indicates an expected call of Cleanup -func (mr *MockModuleMockRecorder) Cleanup() *gomock.Call { - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Cleanup", reflect.TypeOf((*MockModule)(nil).Cleanup)) +// Name indicates an expected call of Name +func (mr *MockModuleMockRecorder) Name() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Name", reflect.TypeOf((*MockModule)(nil).Name)) } // Group mocks base method func (m *MockModule) Group() string { + m.ctrl.T.Helper() ret := m.ctrl.Call(m, "Group") ret0, _ := ret[0].(string) return ret0 @@ -54,27 +57,30 @@ func (m *MockModule) Group() string { // Group indicates an expected call of Group func (mr *MockModuleMockRecorder) Group() *gomock.Call { + mr.mock.ctrl.T.Helper() return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Group", reflect.TypeOf((*MockModule)(nil).Group)) } -// Name mocks base method -func (m *MockModule) Name() string { - ret := m.ctrl.Call(m, "Name") - ret0, _ := ret[0].(string) - return ret0 +// Start mocks base method +func (m *MockModule) Start() { + m.ctrl.T.Helper() + m.ctrl.Call(m, "Start") } -// Name indicates an expected call of Name -func (mr *MockModuleMockRecorder) Name() *gomock.Call { - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Name", reflect.TypeOf((*MockModule)(nil).Name)) +// Start indicates an expected call of Start +func (mr *MockModuleMockRecorder) Start() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockModule)(nil).Start)) } -// Start mocks base method -func (m *MockModule) Start(arg0 *context.Context) { - m.ctrl.Call(m, "Start", arg0) +// Cleanup mocks base method +func (m *MockModule) Cleanup() { + m.ctrl.T.Helper() + m.ctrl.Call(m, "Cleanup") } -// Start indicates an expected call of Start -func (mr *MockModuleMockRecorder) Start(arg0 interface{}) *gomock.Call { - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockModule)(nil).Start), arg0) +// Cleanup indicates an expected call of Cleanup +func (mr *MockModuleMockRecorder) Cleanup() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Cleanup", reflect.TypeOf((*MockModule)(nil).Cleanup)) } diff --git a/edge/pkg/devicetwin/devicetwin.go b/edge/pkg/devicetwin/devicetwin.go index a71284b0f..5f87b0b59 100644 --- a/edge/pkg/devicetwin/devicetwin.go +++ b/edge/pkg/devicetwin/devicetwin.go @@ -6,6 +6,7 @@ import ( "k8s.io/klog" "github.com/kubeedge/beehive/pkg/core" + beehiveContext "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/kubeedge/edge/pkg/common/modules" "github.com/kubeedge/kubeedge/edge/pkg/devicetwin/dtclient" "github.com/kubeedge/kubeedge/edge/pkg/devicetwin/dtcontext" @@ -56,4 +57,5 @@ func (dt *DeviceTwin) Start() { //Cleanup clean resource after quit func (dt *DeviceTwin) Cleanup() { dt.cancel() + beehiveContext.Cleanup(dt.Name()) } diff --git a/edge/pkg/devicetwin/devicetwin_test.go b/edge/pkg/devicetwin/devicetwin_test.go index d38eaa4e4..d10b022cd 100644 --- a/edge/pkg/devicetwin/devicetwin_test.go +++ b/edge/pkg/devicetwin/devicetwin_test.go @@ -75,8 +75,6 @@ func TestStart(t *testing.T) { var querySeterMock *beego.MockQuerySeter // fakeModule is mocked implementation of TestModule. var fakeModule *beehive.MockModule - // mainContext is beehive context used for communication between modules. - var mainContext *beehiveContext.Context mockCtrl := gomock.NewController(t) defer mockCtrl.Finish() @@ -87,21 +85,19 @@ func TestStart(t *testing.T) { fakeModule.EXPECT().Name().Return(TestModule).Times(3) core.Register(fakeModule) - mainContext = beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - mainContext.AddModule(TestModule) + beehiveContext.AddModule(TestModule) core.Register(&DeviceTwin{}) dt := DeviceTwin{} - mainContext.AddModule(dt.Name()) - mainContext.AddModuleGroup(dt.Name(), dt.Group()) - dt.Context = mainContext + beehiveContext.AddModule(dt.Name()) + beehiveContext.AddModuleGroup(dt.Name(), dt.Group()) ormerMock.EXPECT().QueryTable(gomock.Any()).Return(querySeterMock).Times(1) querySeterMock.EXPECT().All(gomock.Any()).Return(int64(1), nil).Times(1) - go dt.Start(mainContext) + go dt.Start() time.Sleep(1 * time.Millisecond) // Sending a message from devicetwin module to the created fake module(TestModule) to check context is initialized properly. - dt.Context.Send(TestModule, test) - _, err := mainContext.Receive(TestModule) + beehiveContext.Send(TestModule, test) + _, err := beehiveContext.Receive(TestModule) t.Run("MessagePingTest", func(t *testing.T) { if err != nil { t.Errorf("Error while receiving message: %v", err) @@ -154,22 +150,20 @@ func TestStart(t *testing.T) { func TestCleanup(t *testing.T) { //test is for sending test messages from devicetwin module. var test model.Message - mainContext := beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - mainContext.AddModule(TestModule) + beehiveContext.AddModule(TestModule) core.Register(&DeviceTwin{}) dt := DeviceTwin{} - mainContext.AddModule(dt.Name()) - mainContext.AddModuleGroup(dt.Name(), dt.Group()) + beehiveContext.AddModule(dt.Name()) + beehiveContext.AddModuleGroup(dt.Name(), dt.Group()) _, cancel := context.WithCancel(context.Background()) deviceTwin := DeviceTwin{ - Context: mainContext, - cancel: cancel, + cancel: cancel, } deviceTwin.Cleanup() //Send message to avoid deadlock if channel deletion has failed after cleanup - go mainContext.Send(DeviceTwinModuleName, test) - _, err := mainContext.Receive(DeviceTwinModuleName) + go beehiveContext.Send(DeviceTwinModuleName, test) + _, err := beehiveContext.Receive(DeviceTwinModuleName) t.Run("CheckCleanUp", func(t *testing.T) { if err == nil { t.Errorf("DeviceTwin Module still has channel after cleanup") diff --git a/edge/pkg/devicetwin/dtmanager/communicate_test.go b/edge/pkg/devicetwin/dtmanager/communicate_test.go index 8ecd629af..ff721cd1c 100644 --- a/edge/pkg/devicetwin/dtmanager/communicate_test.go +++ b/edge/pkg/devicetwin/dtmanager/communicate_test.go @@ -22,7 +22,6 @@ import ( "testing" "time" - "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/beehive/pkg/core/model" cloudconn "github.com/kubeedge/kubeedge/edge/pkg/common/cloudconnection" "github.com/kubeedge/kubeedge/edge/pkg/devicetwin/dtcommon" @@ -32,9 +31,8 @@ import ( // TestStartAction is function to test Start() when value is passed in ReceiverChan. func TestStartAction(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContextStateConnected, _ := dtcontext.InitDTContext(mainContext) + dtContextStateConnected, _ := dtcontext.InitDTContext() dtContextStateConnected.State = dtcommon.Connected receiveChanActionPresent := make(chan interface{}, 1) @@ -95,8 +93,7 @@ func TestStartAction(t *testing.T) { // TestStartHeartBeat is function to test Start() when value is passed in HeartBeatChan. func TestStartHeartBeat(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContexts, _ := dtcontext.InitDTContext(mainContext) + dtContexts, _ := dtcontext.InitDTContext() heartChanStop := make(chan interface{}, 1) heartChanPing := make(chan interface{}, 1) heartChanStop <- "stop" @@ -135,7 +132,7 @@ func TestStartHeartBeat(t *testing.T) { if test.Worker.HeartBeatChan == heartChanPing { _, exist := test.Worker.DTContexts.ModulesHealth.Load("group") if !exist { - t.Errorf("Start Failed to add module in context") + t.Errorf("Start Failed to add module in beehiveContext") } } }) @@ -144,9 +141,8 @@ func TestStartHeartBeat(t *testing.T) { // TestDealSendToCloud is function to test dealSendToCloud(). func TestDealSendToCloud(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContextStateDisconnected, _ := dtcontext.InitDTContext(mainContext) - dtContextStateConnected, _ := dtcontext.InitDTContext(mainContext) + dtContextStateDisconnected, _ := dtcontext.InitDTContext() + dtContextStateConnected, _ := dtcontext.InitDTContext() dtContextStateConnected.State = dtcommon.Connected msg := &model.Message{ Header: model.MessageHeader{ @@ -192,7 +188,7 @@ func TestDealSendToCloud(t *testing.T) { t.Errorf("dealSendToCloud() error = %v, wantErr %v", err, test.wantErr) return } - // Testing whether the message is properly stored in ConfirmMap of context when correct message is passed + // Testing whether the message is properly stored in ConfirmMap of beehiveContext when correct message is passed if err == nil && test.context.State == dtcommon.Connected { gotMsg, exist := test.context.ConfirmMap.Load("message") if !exist { @@ -209,8 +205,7 @@ func TestDealSendToCloud(t *testing.T) { // TestDealLifeCycle is function to test dealLifeCycle(). func TestDealLifeCycle(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContext, _ := dtcontext.InitDTContext(mainContext) + dtContext, _ := dtcontext.InitDTContext() tests := []struct { name string context *dtcontext.DTContext @@ -249,8 +244,7 @@ func TestDealLifeCycle(t *testing.T) { // TestDealConfirm is function to test dealConfirm(). func TestDealConfirm(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContext, _ := dtcontext.InitDTContext(mainContext) + dtContext, _ := dtcontext.InitDTContext() tests := []struct { name string context *dtcontext.DTContext @@ -281,7 +275,7 @@ func TestDealConfirm(t *testing.T) { if err == nil { _, exist := test.context.ConfirmMap.Load("parentId") if exist { - t.Errorf("dealConfirm failed() ParentMessageId still present in context ConfirmMap") + t.Errorf("dealConfirm failed() ParentMessageId still present in beehiveContext ConfirmMap") } } }) @@ -290,8 +284,7 @@ func TestDealConfirm(t *testing.T) { // TestCheckConfirm is function to test checkConfirm(). func TestCheckConfirm(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContext, _ := dtcontext.InitDTContext(mainContext) + dtContext, _ := dtcontext.InitDTContext() dtContext.State = dtcommon.Connected dtContext.ConfirmMap.Store("emptyMessage", &dttype.DTMessage{}) dtContext.ConfirmMap.Store("actionMessage", &dttype.DTMessage{ diff --git a/edge/pkg/devicetwin/dtmanager/device_test.go b/edge/pkg/devicetwin/dtmanager/device_test.go index f8947bf7d..0950e236b 100644 --- a/edge/pkg/devicetwin/dtmanager/device_test.go +++ b/edge/pkg/devicetwin/dtmanager/device_test.go @@ -26,7 +26,6 @@ import ( "github.com/astaxie/beego/orm" "github.com/golang/mock/gomock" - "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/beehive/pkg/core/model" "github.com/kubeedge/kubeedge/edge/mocks/beego" "github.com/kubeedge/kubeedge/edge/pkg/common/dbm" @@ -46,8 +45,7 @@ func testAction(context *dtcontext.DTContext, resource string, msg interface{}) // TestDeviceStartAction is function to test Start() when value is passed in ReceiverChan. func TestDeviceStartAction(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContextStateConnected, _ := dtcontext.InitDTContext(mainContext) + dtContextStateConnected, _ := dtcontext.InitDTContext() dtContextStateConnected.State = dtcommon.Connected content := dttype.DeviceUpdate{} bytes, _ := json.Marshal(content) @@ -104,8 +102,7 @@ func TestDeviceStartAction(t *testing.T) { // TestDevicetHeartBeat is function to test Start() when value is passed in HeartBeatChan. func TestDeviceStartHeartBeat(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContexts, _ := dtcontext.InitDTContext(mainContext) + dtContexts, _ := dtcontext.InitDTContext() heartChanStop := make(chan interface{}, 1) heartChanPing := make(chan interface{}, 1) heartChanStop <- "stop" @@ -143,7 +140,7 @@ func TestDeviceStartHeartBeat(t *testing.T) { if test.Worker.HeartBeatChan == heartChanPing { _, exist := test.Worker.DTContexts.ModulesHealth.Load("group") if !exist { - t.Errorf("Start Failed to add module in context") + t.Errorf("Start Failed to add module in beehiveContext") } } }) @@ -162,8 +159,7 @@ func TestDealDeviceStateUpdate(t *testing.T) { querySeterMock = beego.NewMockQuerySeter(mockCtrl) dbm.DBAccess = ormerMock - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContexts, err := dtcontext.InitDTContext(mainContext) + dtContexts, err := dtcontext.InitDTContext() if err != nil { t.Errorf("InitDTContext error %v", err) return @@ -266,8 +262,7 @@ func TestDealDeviceStateUpdate(t *testing.T) { //TestDealDeviceUpdated is function to test dealDeviceUpdated(). func TestDealDeviceUpdated(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContexts, _ := dtcontext.InitDTContext(mainContext) + dtContexts, _ := dtcontext.InitDTContext() content := dttype.DeviceUpdate{} bytes, err := json.Marshal(content) if err != nil { @@ -344,8 +339,7 @@ func TestDeviceUpdated(t *testing.T) { Cols: make(map[string]interface{}), }) - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContexts, _ := dtcontext.InitDTContext(mainContext) + dtContexts, _ := dtcontext.InitDTContext() dtContexts.DeviceList.Store("EmptyDevice", "Device") devA := &dttype.Device{ID: "DeviceA"} @@ -471,13 +465,12 @@ func TestDeviceUpdated(t *testing.T) { // TestDealMsgAttr is function to test DealMsgAttr(). func TestDealMsgAttr(t *testing.T) { - mainContext := context.InitContext(context.MsgCtxTypeChannel) - dtContextsEmptyAttributes, err := dtcontext.InitDTContext(mainContext) + dtContextsEmptyAttributes, err := dtcontext.InitDTContext() if err != nil { t.Errorf("initDtcontext error %v", err) return } - dtContextsNonEmptyAttributes, err := dtcontext.InitDTContext(mainContext) + dtContextsNonEmptyAttributes, err := dtcontext.InitDTContext() if err != nil { t.Errorf("initDtcontext error %v", err) return diff --git a/edge/pkg/devicetwin/process_test.go b/edge/pkg/devicetwin/process_test.go index f5f2d583b..f11af45ef 100644 --- a/edge/pkg/devicetwin/process_test.go +++ b/edge/pkg/devicetwin/process_test.go @@ -27,7 +27,6 @@ import ( "github.com/astaxie/beego/orm" "github.com/golang/mock/gomock" - beehiveContext "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/beehive/pkg/core/model" "github.com/kubeedge/kubeedge/edge/mocks/beego" "github.com/kubeedge/kubeedge/edge/pkg/common/dbm" @@ -67,8 +66,7 @@ func createFakeDeviceTwin() *[]dtclient.DeviceTwin { //TestRegisterDTModule is function to test RegisterDTmodule(). func TestRegisterDTModule(t *testing.T) { - mainContext := beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - dtContexts, _ := dtcontext.InitDTContext(mainContext) + dtContexts, _ := dtcontext.InitDTContext() var moduleRegistered bool _, cancel := context.WithCancel(context.Background()) dtc := &DeviceTwin{ @@ -117,13 +115,11 @@ func TestRegisterDTModule(t *testing.T) { //TestDTController_distributeMsg is function to test distributeMsg(). func TestDTController_distributeMsg(t *testing.T) { - mainContext := beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - dtContexts, _ := dtcontext.InitDTContext(mainContext) + dtContexts, _ := dtcontext.InitDTContext() dtc := &DeviceTwin{ HeartBeatToModule: make(map[string]chan interface{}), DTModules: make(map[string]dtmodule.DTModule), DTContexts: dtContexts, - Context: mainContext, } payload := dttype.MembershipUpdate{ @@ -214,8 +210,7 @@ func TestSyncSqlite(t *testing.T) { querySeterMock = beego.NewMockQuerySeter(mockCtrl) dbm.DBAccess = ormerMock - mainContext := beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - dtContexts, _ := dtcontext.InitDTContext(mainContext) + dtContexts, _ := dtcontext.InitDTContext() // fakeDevice is used to set the argument of All function fakeDevice := createFakeDevice() // fakeDeviceAttr is used to set the argument of All function @@ -306,8 +301,7 @@ func TestSyncDeviceFromSqlite(t *testing.T) { querySeterMock = beego.NewMockQuerySeter(mockCtrl) dbm.DBAccess = ormerMock - mainContext := beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - dtContext, _ := dtcontext.InitDTContext(mainContext) + dtContext, _ := dtcontext.InitDTContext() // fakeDevice is used to set the argument of All function fakeDevice := createFakeDevice() // fakeDeviceAttr is used to set the argument of All function diff --git a/edge/pkg/edged/edged.go b/edge/pkg/edged/edged.go index d837820f7..c5be79003 100644 --- a/edge/pkg/edged/edged.go +++ b/edge/pkg/edged/edged.go @@ -350,6 +350,7 @@ func (e *edged) Start() { func (e *edged) Cleanup() { e.cancel() + beehiveContext.Cleanup(e.Name()) } // isInitPodReady is used to safely return initPodReady flag diff --git a/edge/pkg/edgehub/edgehub.go b/edge/pkg/edgehub/edgehub.go index 89fce4be1..235af135d 100644 --- a/edge/pkg/edgehub/edgehub.go +++ b/edge/pkg/edgehub/edgehub.go @@ -8,6 +8,7 @@ import ( "k8s.io/klog" "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/clients" @@ -106,4 +107,5 @@ func (eh *EdgeHub) Start() { //Cleanup sets up context cleanup through Edgehub name func (eh *EdgeHub) Cleanup() { eh.cancel() + beehiveContext.Cleanup(eh.Name()) } diff --git a/edge/pkg/edgehub/process_test.go b/edge/pkg/edgehub/process_test.go index c6ade049f..8b3b38a50 100644 --- a/edge/pkg/edgehub/process_test.go +++ b/edge/pkg/edgehub/process_test.go @@ -125,8 +125,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) - defer beehiveContext.DestroyContext() message := model.NewMessage("test_id") tests := []struct { name string @@ -182,8 +180,6 @@ func TestSendToKeepChannel(t *testing.T) { //TestDispatch() tests whether the messages are properly dispatched to their respective modules func TestDispatch(t *testing.T) { - beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - defer beehiveContext.DestroyContext() tests := []struct { name string @@ -245,8 +241,6 @@ func TestRouteToEdge(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() mockAdapter := edgehub.NewMockAdapter(mockCtrl) - beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - defer beehiveContext.DestroyContext() tests := []struct { name string @@ -291,8 +285,6 @@ func TestSendToCloud(t *testing.T) { mockCtrl := gomock.NewController(t) defer mockCtrl.Finish() mockAdapter := edgehub.NewMockAdapter(mockCtrl) - beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - defer beehiveContext.DestroyContext() msg := model.NewMessage("").BuildHeader("test_id", "", 1) msg.Header.Sync = true @@ -382,8 +374,6 @@ func TestRouteToCloud(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() mockAdapter := edgehub.NewMockAdapter(mockCtrl) - beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) - defer beehiveContext.DestroyContext() tests := []struct { name string diff --git a/edge/pkg/eventbus/eventbus.go b/edge/pkg/eventbus/eventbus.go index fff27119b..91a02005c 100644 --- a/edge/pkg/eventbus/eventbus.go +++ b/edge/pkg/eventbus/eventbus.go @@ -122,6 +122,7 @@ func (eb *eventbus) Start() { func (eb *eventbus) Cleanup() { eb.cancel() + beehiveContext.Cleanup(eb.Name()) } func pubMQTT(topic string, payload []byte) { diff --git a/edge/pkg/metamanager/metamanager.go b/edge/pkg/metamanager/metamanager.go index 986ac7fec..1ec91e8b9 100644 --- a/edge/pkg/metamanager/metamanager.go +++ b/edge/pkg/metamanager/metamanager.go @@ -27,8 +27,7 @@ func Register() { } type metaManager struct { - context *beehiveContext.Context - cancel context.CancelFunc + cancel context.CancelFunc } func (*metaManager) Name() string { @@ -39,9 +38,8 @@ func (*metaManager) Group() string { return modules.MetaGroup } -func (m *metaManager) Start(c *beehiveContext.Context) { +func (m *metaManager) Start() { var ctx context.Context - m.context = c ctx, m.cancel = context.WithCancel(context.Background()) InitMetaManagerConfig() @@ -56,7 +54,7 @@ func (m *metaManager) Start(c *beehiveContext.Context) { case <-timer.C: timer.Reset(period) msg := model.NewMessage("").BuildRouter(MetaManagerModuleName, GroupResource, model.ResourceTypePodStatus, OperationMetaSync) - m.context.Send(MetaManagerModuleName, *msg) + beehiveContext.Send(MetaManagerModuleName, *msg) } } }() @@ -66,7 +64,7 @@ func (m *metaManager) Start(c *beehiveContext.Context) { func (m *metaManager) Cleanup() { m.cancel() - m.context.Cleanup(m.Name()) + beehiveContext.Cleanup(m.Name()) } func getSyncInterval() time.Duration { diff --git a/edge/pkg/metamanager/process.go b/edge/pkg/metamanager/process.go index 447711bc4..b95a0f8bd 100644 --- a/edge/pkg/metamanager/process.go +++ b/edge/pkg/metamanager/process.go @@ -65,37 +65,37 @@ func InitMetaManagerConfig() { } } -func feedbackError(err error, info string, request model.Message, c *beehiveContext.Context) { +func feedbackError(err error, info string, request model.Message) { errInfo := "Something wrong" if err != nil { errInfo = fmt.Sprintf(info+": %v", err) } errResponse := model.NewErrorMessage(&request, errInfo).SetRoute(MetaManagerModuleName, request.GetGroup()) if request.GetSource() == modules.EdgedModuleName { - sendToEdged(errResponse, request.IsSync(), c) + sendToEdged(errResponse, request.IsSync()) } else { - sendToCloud(errResponse, c) + sendToCloud(errResponse) } } -func sendToEdged(message *model.Message, sync bool, c *beehiveContext.Context) { +func sendToEdged(message *model.Message, sync bool) { if sync { - c.SendResp(*message) + beehiveContext.SendResp(*message) } else { - c.Send(modules.EdgedModuleName, *message) + beehiveContext.Send(modules.EdgedModuleName, *message) } } -func sendToEdgeMesh(message *model.Message, sync bool, c *beehiveContext.Context) { +func sendToEdgeMesh(message *model.Message, sync bool) { if sync { - c.SendResp(*message) + beehiveContext.SendResp(*message) } else { - c.Send(modules.EdgeMeshModuleName, *message) + beehiveContext.Send(modules.EdgeMeshModuleName, *message) } } -func sendToCloud(message *model.Message, c *beehiveContext.Context) { - c.SendToGroup(sendModuleGroupName, *message) +func sendToCloud(message *model.Message) { + beehiveContext.SendToGroup(sendModuleGroupName, *message) } // Resource format: <namespace>/<restype>[/resid] @@ -156,7 +156,7 @@ func (m *metaManager) processInsert(message model.Message) { content, err = json.Marshal(message.GetContent()) if err != nil { klog.Errorf("marshal update message content failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to marshal message content", message, m.context) + feedbackError(err, "Error to marshal message content", message) return } } @@ -169,20 +169,20 @@ func (m *metaManager) processInsert(message model.Message) { err = dao.SaveMeta(meta) if err != nil { klog.Errorf("save meta failed, %s: %v", msgDebugInfo(&message), err) - feedbackError(err, "Error to save meta to DB", message, m.context) + feedbackError(err, "Error to save meta to DB", message) return } if resType == constants.ResourceTypeService || resType == constants.ResourceTypeEndpoints { // Notify edgemesh - sendToEdgeMesh(&message, false, m.context) + sendToEdgeMesh(&message, false) } else { // Notify edged - sendToEdged(&message, false, m.context) + sendToEdged(&message, false) } resp := message.NewRespByMessage(&message, OK) - sendToCloud(resp, m.context) + sendToCloud(resp) } func (m *metaManager) processUpdate(message model.Message) { @@ -195,7 +195,7 @@ func (m *metaManager) processUpdate(message model.Message) { content, err = json.Marshal(message.GetContent()) if err != nil { klog.Errorf("marshal update message content failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to marshal message content", message, m.context) + feedbackError(err, "Error to marshal message content", message) return } } @@ -208,7 +208,7 @@ func (m *metaManager) processUpdate(message model.Message) { err = json.Unmarshal(content, &epsList) if err != nil { klog.Errorf("Unmarshal update message content failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to unmarshal", message, m.context) + feedbackError(err, "Error to unmarshal", message) return } for _, eps := range epsList { @@ -228,16 +228,16 @@ func (m *metaManager) processUpdate(message model.Message) { continue } } - sendToEdgeMesh(&message, false, m.context) + sendToEdgeMesh(&message, false) resp := message.NewRespByMessage(&message, OK) - sendToCloud(resp, m.context) + sendToCloud(resp) return case constants.ResourceTypeServiceList: var svcList []v1.Service err = json.Unmarshal(content, &svcList) if err != nil { klog.Errorf("Unmarshal update message content failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to unmarshal", message, m.context) + feedbackError(err, "Error to unmarshal", message) return } for _, svc := range svcList { @@ -257,9 +257,9 @@ func (m *metaManager) processUpdate(message model.Message) { continue } } - sendToEdgeMesh(&message, false, m.context) + sendToEdgeMesh(&message, false) resp := message.NewRespByMessage(&message, OK) - sendToCloud(resp, m.context) + sendToCloud(resp) return case model.ResourceTypePodlist: meta := &dao.Meta{ @@ -269,12 +269,12 @@ func (m *metaManager) processUpdate(message model.Message) { err = dao.InsertOrUpdate(meta) if err != nil { klog.Errorf("Update meta failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to update meta to DB", message, m.context) + feedbackError(err, "Error to update meta to DB", message) return } - sendToEdgeMesh(&message, false, m.context) + sendToEdgeMesh(&message, false) resp := message.NewRespByMessage(&message, OK) - sendToCloud(resp, m.context) + sendToCloud(resp) return default: klog.Warningf("Resource type %s unknown", resType) @@ -284,7 +284,7 @@ func (m *metaManager) processUpdate(message model.Message) { if resourceUnchanged(resType, resKey, content) { resp := message.NewRespByMessage(&message, OK) - sendToEdged(resp, message.IsSync(), m.context) + sendToEdged(resp, message.IsSync()) klog.Infof("resource[%s] unchanged, no notice", resKey) return } @@ -296,28 +296,28 @@ func (m *metaManager) processUpdate(message model.Message) { err = dao.InsertOrUpdate(meta) if err != nil { klog.Errorf("update meta failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to update meta to DB", message, m.context) + feedbackError(err, "Error to update meta to DB", message) return } switch message.GetSource() { //case core.EdgedModuleName: case modules.EdgedModuleName: - sendToCloud(&message, m.context) + sendToCloud(&message) resp := message.NewRespByMessage(&message, OK) - sendToEdged(resp, message.IsSync(), m.context) + sendToEdged(resp, message.IsSync()) case CloudControlerModel: if resType == constants.ResourceTypeService || resType == constants.ResourceTypeEndpoints { - sendToEdgeMesh(&message, message.IsSync(), m.context) + sendToEdgeMesh(&message, message.IsSync()) } else { - sendToEdged(&message, message.IsSync(), m.context) + sendToEdged(&message, message.IsSync()) } resp := message.NewRespByMessage(&message, OK) - sendToCloud(resp, m.context) + sendToCloud(resp) case CloudFunctionModel: - m.context.Send(EdgeFunctionModel, message) + beehiveContext.Send(EdgeFunctionModel, message) case EdgeFunctionModel: - sendToCloud(&message, m.context) + sendToCloud(&message) } } @@ -332,7 +332,7 @@ func (m *metaManager) processResponse(message model.Message) { content, err = json.Marshal(message.GetContent()) if err != nil { klog.Errorf("marshal response message content failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to marshal message content", message, m.context) + feedbackError(err, "Error to marshal message content", message) return } } @@ -345,20 +345,20 @@ func (m *metaManager) processResponse(message model.Message) { err = dao.InsertOrUpdate(meta) if err != nil { klog.Errorf("update meta failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to update meta to DB", message, m.context) + feedbackError(err, "Error to update meta to DB", message) return } // Notify edged or edgemesh if the data if coming from cloud if message.GetSource() == CloudControlerModel { if resType == constants.ResourceTypeService || resType == constants.ResourceTypeEndpoints { - sendToEdgeMesh(&message, message.IsSync(), m.context) + sendToEdgeMesh(&message, message.IsSync()) } else { - sendToEdged(&message, message.IsSync(), m.context) + sendToEdged(&message, message.IsSync()) } } else { // Send to cloud if the update request is coming from edged - sendToCloud(&message, m.context) + sendToCloud(&message) } } @@ -366,22 +366,22 @@ func (m *metaManager) processDelete(message model.Message) { err := dao.DeleteMetaByKey(message.GetResource()) if err != nil { klog.Errorf("delete meta failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to delete meta to DB", message, m.context) + feedbackError(err, "Error to delete meta to DB", message) return } _, resType, _ := parseResource(message.GetResource()) if resType == constants.ResourceTypeService || resType == constants.ResourceTypeEndpoints { // Notify edgemesh - sendToEdgeMesh(&message, false, m.context) + sendToEdgeMesh(&message, false) resp := message.NewRespByMessage(&message, OK) - sendToCloud(resp, m.context) + sendToCloud(resp) return } // Notify edged - sendToEdged(&message, false, m.context) + sendToEdged(&message, false) resp := message.NewRespByMessage(&message, OK) - sendToCloud(resp, m.context) + sendToCloud(resp) } func (m *metaManager) processQuery(message model.Message) { @@ -395,7 +395,7 @@ func (m *metaManager) processQuery(message model.Message) { } else { resp := message.NewRespByMessage(&message, *metas) resp.SetRoute(MetaManagerModuleName, resp.GetGroup()) - sendToEdged(resp, message.IsSync(), m.context) + sendToEdged(resp, message.IsSync()) } return } @@ -408,14 +408,14 @@ func (m *metaManager) processQuery(message model.Message) { } if err != nil { klog.Errorf("query meta failed, %s", msgDebugInfo(&message)) - feedbackError(err, "Error to query meta in DB", message, m.context) + feedbackError(err, "Error to query meta in DB", message) } else { resp := message.NewRespByMessage(&message, *metas) resp.SetRoute(MetaManagerModuleName, resp.GetGroup()) if resType == constants.ResourceTypeService || resType == constants.ResourceTypeEndpoints || resType == model.ResourceTypePodlist { - sendToEdgeMesh(resp, message.IsSync(), m.context) + sendToEdgeMesh(resp, message.IsSync()) } else { - sendToEdged(resp, message.IsSync(), m.context) + sendToEdged(resp, message.IsSync()) } } } @@ -425,11 +425,11 @@ func (m *metaManager) processRemoteQuery(message model.Message) { // TODO: retry originalID := message.GetID() message.UpdateID() - resp, err := m.context.SendSync(sendModuleName, message, 60*time.Second) // TODO: configurable + resp, err := beehiveContext.SendSync(sendModuleName, message, 60*time.Second) // TODO: configurable klog.Infof("########## process get: req[%+v], resp[%+v], err[%+v]", message, resp, err) if err != nil { klog.Errorf("remote query failed: %v", err) - feedbackError(err, "Error to query meta in DB", message, m.context) + feedbackError(err, "Error to query meta in DB", message) return } @@ -441,7 +441,7 @@ func (m *metaManager) processRemoteQuery(message model.Message) { content, err = json.Marshal(resp.GetContent()) if err != nil { klog.Errorf("marshal remote query response content failed, %s", msgDebugInfo(&resp)) - feedbackError(err, "Error to marshal message content", message, m.context) + feedbackError(err, "Error to marshal message content", message) return } } @@ -457,9 +457,9 @@ func (m *metaManager) processRemoteQuery(message model.Message) { } resp.BuildHeader(resp.GetID(), originalID, resp.GetTimestamp()) if resType == constants.ResourceTypeService || resType == constants.ResourceTypeEndpoints { - sendToEdgeMesh(&resp, message.IsSync(), m.context) + sendToEdgeMesh(&resp, message.IsSync()) } else { - sendToEdged(&resp, message.IsSync(), m.context) + sendToEdged(&resp, message.IsSync()) } }() } @@ -520,7 +520,7 @@ func (m *metaManager) syncPodStatus() { } msg := model.NewMessage("").BuildRouter(MetaManagerModuleName, GroupResource, namespace+constants.ResourceSep+model.ResourceTypePodStatus, model.UpdateOperation).FillBody(content) - sendToCloud(msg, m.context) + sendToCloud(msg) klog.Infof("sync pod status successful, %s", msgDebugInfo(msg)) } @@ -535,7 +535,7 @@ func (m *metaManager) processFunctionAction(message model.Message) { content, err = json.Marshal(message.GetContent()) if err != nil { klog.Errorf("marshal save message content failed, %s: %v", msgDebugInfo(&message), err) - feedbackError(err, "Error to marshal message content", message, m.context) + feedbackError(err, "Error to marshal message content", message) return } } @@ -548,11 +548,11 @@ func (m *metaManager) processFunctionAction(message model.Message) { err = dao.SaveMeta(meta) if err != nil { klog.Errorf("save meta failed, %s: %v", msgDebugInfo(&message), err) - feedbackError(err, "Error to save meta to DB", message, m.context) + feedbackError(err, "Error to save meta to DB", message) return } - m.context.Send(EdgeFunctionModel, message) + beehiveContext.Send(EdgeFunctionModel, message) } func (m *metaManager) processFunctionActionResult(message model.Message) { @@ -565,7 +565,7 @@ func (m *metaManager) processFunctionActionResult(message model.Message) { content, err = json.Marshal(message.GetContent()) if err != nil { klog.Errorf("marshal save message content failed, %s: %v", msgDebugInfo(&message), err) - feedbackError(err, "Error to marshal message content", message, m.context) + feedbackError(err, "Error to marshal message content", message) return } } @@ -578,24 +578,24 @@ func (m *metaManager) processFunctionActionResult(message model.Message) { err = dao.SaveMeta(meta) if err != nil { klog.Errorf("save meta failed, %s: %v", msgDebugInfo(&message), err) - feedbackError(err, "Error to save meta to DB", message, m.context) + feedbackError(err, "Error to save meta to DB", message) return } - sendToCloud(&message, m.context) + sendToCloud(&message) } func (m *metaManager) processVolume(message model.Message) { klog.Info("process volume started") - back, err := m.context.SendSync(modules.EdgedModuleName, message, constants.CSISyncMsgRespTimeout) + back, err := beehiveContext.SendSync(modules.EdgedModuleName, message, constants.CSISyncMsgRespTimeout) klog.Infof("process volume get: req[%+v], back[%+v], err[%+v]", message, back, err) if err != nil { klog.Errorf("process volume send to edged failed: %v", err) } resp := message.NewRespByMessage(&message, back.GetContent()) - sendToCloud(resp, m.context) + sendToCloud(resp) klog.Infof("process volume send to cloud resp[%+v]", resp) } @@ -638,7 +638,7 @@ func (m *metaManager) runMetaManager(ctx context.Context) { default: } - if msg, err := m.context.Receive(m.Name()); err == nil { + if msg, err := beehiveContext.Receive(m.Name()); err == nil { klog.Infof("get a message %+v", msg) m.process(msg) } else { diff --git a/edge/pkg/servicebus/servicebus.go b/edge/pkg/servicebus/servicebus.go index 6c251ab9b..fb834ec36 100644 --- a/edge/pkg/servicebus/servicebus.go +++ b/edge/pkg/servicebus/servicebus.go @@ -25,8 +25,7 @@ const ( // servicebus struct type servicebus struct { - context *beehiveContext.Context - cancel context.CancelFunc + cancel context.CancelFunc } // Register register servicebus @@ -43,10 +42,9 @@ func (*servicebus) Group() string { return modules.BusGroup } -func (sb *servicebus) Start(c *beehiveContext.Context) { +func (sb *servicebus) Start() { // no need to call TopicInit now, we have fixed topic var ctx context.Context - sb.context = c ctx, sb.cancel = context.WithCancel(context.Background()) var htc = new(http.Client) htc.Timeout = time.Second * 10 @@ -63,7 +61,7 @@ func (sb *servicebus) Start(c *beehiveContext.Context) { default: } - msg, err := sb.context.Receive("servicebus") + msg, err := beehiveContext.Receive("servicebus") if err != nil { klog.Warningf("servicebus receive msg error %v", err) continue @@ -81,7 +79,7 @@ func (sb *servicebus) Start(c *beehiveContext.Context) { klog.Warningf(m) code := http.StatusBadRequest if response, err := buildErrorResponse(msg.GetID(), m, code); err == nil { - sb.context.SendToGroup(modules.HubGroup, response) + beehiveContext.SendToGroup(modules.HubGroup, response) } return } @@ -91,7 +89,7 @@ func (sb *servicebus) Start(c *beehiveContext.Context) { m := "error to marshal request msg content" code := http.StatusBadRequest if response, err := buildErrorResponse(msg.GetID(), m, code); err == nil { - sb.context.SendToGroup(modules.HubGroup, response) + beehiveContext.SendToGroup(modules.HubGroup, response) } return } @@ -101,7 +99,7 @@ func (sb *servicebus) Start(c *beehiveContext.Context) { code := http.StatusBadRequest klog.Errorf(m, err) if response, err := buildErrorResponse(msg.GetID(), m, code); err == nil { - sb.context.SendToGroup(modules.HubGroup, response) + beehiveContext.SendToGroup(modules.HubGroup, response) } return } @@ -113,7 +111,7 @@ func (sb *servicebus) Start(c *beehiveContext.Context) { code := http.StatusNotFound klog.Errorf(m, err) if response, err := buildErrorResponse(msg.GetID(), m, code); err == nil { - sb.context.SendToGroup(modules.HubGroup, response) + beehiveContext.SendToGroup(modules.HubGroup, response) } return } @@ -127,7 +125,7 @@ func (sb *servicebus) Start(c *beehiveContext.Context) { code := http.StatusInternalServerError klog.Errorf(m, err) if response, err := buildErrorResponse(msg.GetID(), m, code); err == nil { - sb.context.SendToGroup(modules.HubGroup, response) + beehiveContext.SendToGroup(modules.HubGroup, response) } return } @@ -136,14 +134,14 @@ func (sb *servicebus) Start(c *beehiveContext.Context) { responseMsg := model.NewMessage(msg.GetID()) responseMsg.Content = response responseMsg.SetRoute("servicebus", modules.UserGroup) - sb.context.SendToGroup(modules.HubGroup, *responseMsg) + beehiveContext.SendToGroup(modules.HubGroup, *responseMsg) }() } } func (sb *servicebus) Cleanup() { sb.cancel() - sb.context.Cleanup(sb.Name()) + beehiveContext.Cleanup(sb.Name()) } func buildErrorResponse(parentID string, content string, statusCode int) (model.Message, error) { diff --git a/edge/test/test.go b/edge/test/test.go index 1cd494eb4..613f8bba7 100644 --- a/edge/test/test.go +++ b/edge/test/test.go @@ -29,7 +29,6 @@ func Register() { } type testManager struct { - context *beehiveContext.Context moduleWait *sync.WaitGroup } @@ -121,7 +120,7 @@ func (tm *testManager) podHandler(w http.ResponseWriter, req *http.Request) { ns = p.Namespace } msgReq := message.BuildMsg("resource", string(p.UID), "edgecontroller", ns+"/pod/"+string(p.Name), operation, p) - tm.context.Send("metaManager", *msgReq) + beehiveContext.Send("metaManager", *msgReq) klog.Infof("send message to metaManager is %+v\n", msgReq) } } @@ -152,7 +151,7 @@ func (tm *testManager) deviceHandler(w http.ResponseWriter, req *http.Request) { operation = model.UpdateOperation } msgReq := message.BuildMsg("edgehub", "", "edgemgr", "membership", operation, Content) - tm.context.Send("twin", *msgReq) + beehiveContext.Send("twin", *msgReq) klog.Infof("send message to twingrp is %+v\n", msgReq) } } @@ -182,7 +181,7 @@ func (tm *testManager) secretHandler(w http.ResponseWriter, req *http.Request) { } msgReq := message.BuildMsg("edgehub", string(p.UID), "test", "fakeNamespace/secret/"+string(p.UID), operation, p) - tm.context.Send("metaManager", *msgReq) + beehiveContext.Send("metaManager", *msgReq) klog.Infof("send message to metaManager is %+v\n", msgReq) } } @@ -212,13 +211,12 @@ func (tm *testManager) configmapHandler(w http.ResponseWriter, req *http.Request } msgReq := message.BuildMsg("edgehub", string(p.UID), "test", "fakeNamespace/configmap/"+string(p.UID), operation, p) - tm.context.Send("metaManager", *msgReq) + beehiveContext.Send("metaManager", *msgReq) klog.Infof("send message to metaManager is %+v\n", msgReq) } } -func (tm *testManager) Start(c *beehiveContext.Context) { - tm.context = c +func (tm *testManager) Start() { defer tm.Cleanup() http.HandleFunc("/pods", tm.podHandler) @@ -232,5 +230,5 @@ func (tm *testManager) Start(c *beehiveContext.Context) { } func (tm *testManager) Cleanup() { - tm.context.Cleanup(tm.Name()) + beehiveContext.Cleanup(tm.Name()) } diff --git a/edgemesh/pkg/module.go b/edgemesh/pkg/module.go index d42f701ee..0cd622363 100644 --- a/edgemesh/pkg/module.go +++ b/edgemesh/pkg/module.go @@ -15,8 +15,7 @@ import ( //EdgeMesh defines EdgeMesh object structure type EdgeMesh struct { - context *beehiveContext.Context - cancel context.CancelFunc + cancel context.CancelFunc } // Register register edgemesh @@ -35,8 +34,7 @@ func (em *EdgeMesh) Group() string { } //Start sets context and starts the controller -func (em *EdgeMesh) Start(c *beehiveContext.Context) { - em.context = c +func (em *EdgeMesh) Start() { var ctx context.Context ctx, em.cancel = context.WithCancel(context.Background()) proxy.Init() @@ -49,7 +47,7 @@ func (em *EdgeMesh) Start(c *beehiveContext.Context) { return default: } - msg, err := em.context.Receive(constant.ModuleNameEdgeMesh) + msg, err := beehiveContext.Receive(constant.ModuleNameEdgeMesh) if err != nil { klog.Warningf("edgemesh receive msg error %v", err) continue @@ -62,5 +60,5 @@ func (em *EdgeMesh) Start(c *beehiveContext.Context) { //Cleanup sets up context cleanup through EdgeMesh name func (em *EdgeMesh) Cleanup() { em.cancel() - em.context.Cleanup(em.Name()) + beehiveContext.Cleanup(em.Name()) } diff --git a/edgemesh/pkg/proxy/proxy.go b/edgemesh/pkg/proxy/proxy.go index d0238c85e..78f259e03 100644 --- a/edgemesh/pkg/proxy/proxy.go +++ b/edgemesh/pkg/proxy/proxy.go @@ -11,7 +11,6 @@ import ( "syscall" "time" - "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/beehive/pkg/core/model" "github.com/kubeedge/kubeedge/common/constants" "github.com/kubeedge/kubeedge/edge/pkg/metamanager/client" @@ -67,8 +66,7 @@ func Init() { go func() { unused = make([]string, 0) addrByService = &addrTable{} - c := context.InitContext(context.MsgCtxTypeChannel) - metaClient = client.New(c) + metaClient = client.New() //create virtual network device for { err := vdev.CreateDevice() diff --git a/edgemesh/pkg/registry/registry.go b/edgemesh/pkg/registry/registry.go index e776d4e20..162305a28 100644 --- a/edgemesh/pkg/registry/registry.go +++ b/edgemesh/pkg/registry/registry.go @@ -9,7 +9,6 @@ import ( v1 "k8s.io/api/core/v1" "k8s.io/klog" - "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/kubeedge/edge/pkg/metamanager/client" "github.com/kubeedge/kubeedge/edgemesh/pkg/common" ) @@ -40,9 +39,8 @@ func toProtocolMap(address v1.EndpointAddress, ports []v1.EndpointPort) map[stri } func NewServiceDiscovery(options registry.Options) registry.ServiceDiscovery { - c := context.InitContext(context.MsgCtxTypeChannel) return &ServiceDiscovery{ - metaClient: client.New(c), + metaClient: client.New(), Name: EdgeRegistry, } } diff --git a/edgemesh/pkg/server/dns.go b/edgemesh/pkg/server/dns.go index bba1066de..8034efe9c 100644 --- a/edgemesh/pkg/server/dns.go +++ b/edgemesh/pkg/server/dns.go @@ -13,7 +13,6 @@ import ( "k8s.io/klog" - "github.com/kubeedge/beehive/pkg/core/context" "github.com/kubeedge/kubeedge/edge/pkg/metamanager/client" "github.com/kubeedge/kubeedge/edgemesh/pkg/common" "github.com/kubeedge/kubeedge/edgemesh/pkg/proxy" @@ -78,9 +77,7 @@ func DnsStart() { // startDnsServer start the DNS Server func startDnsServer() { - // init meta client - c := context.InitContext(context.MsgCtxTypeChannel) - metaClient = client.New(c) + metaClient = client.New() //get DNS server name lip, err := getIP() if err != nil { diff --git a/staging/src/github.com/kubeedge/beehive/go.mod b/staging/src/github.com/kubeedge/beehive/go.mod index a08138ee3..d5923c9c4 100644 --- a/staging/src/github.com/kubeedge/beehive/go.mod +++ b/staging/src/github.com/kubeedge/beehive/go.mod @@ -5,6 +5,7 @@ go 1.12 require ( github.com/go-chassis/go-archaius v0.20.0 github.com/go-chassis/go-chassis v1.7.1 // indirect + github.com/golang/mock v1.3.1 github.com/gopherjs/gopherjs v0.0.0-20181103185306-d547d1d9531e // indirect github.com/onsi/ginkgo v1.10.1 // indirect github.com/onsi/gomega v1.7.0 // indirect diff --git a/staging/src/github.com/kubeedge/beehive/go.sum b/staging/src/github.com/kubeedge/beehive/go.sum index 1a0e88a1d..3db9810e7 100644 --- a/staging/src/github.com/kubeedge/beehive/go.sum +++ b/staging/src/github.com/kubeedge/beehive/go.sum @@ -31,6 +31,8 @@ github.com/go-mesh/openlogging v1.0.1-0.20181205082104-3d418c478b2d h1:fs2hiWn8t github.com/go-mesh/openlogging v1.0.1-0.20181205082104-3d418c478b2d/go.mod h1:qaKi+amO+hsGin2q1GmW+/NcbZpMPnTufwrWzDmIuuU= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= +github.com/golang/mock v1.3.1 h1:qGJ6qTW+x6xX/my+8YUVl4WNpX9B7+/l2tRsHGZ7f2s= +github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y= github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= @@ -101,6 +103,7 @@ golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f h1:Bl/8QSvNqXvPGPGXa2z5xUTmV7VDcZyvRZ+QQXkXTZQ= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -110,6 +113,7 @@ golang.org/x/sys v0.0.0-20190318195719-6c81ef8f67ca/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= +golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/staging/src/github.com/kubeedge/beehive/pkg/core/context/context_factory.go b/staging/src/github.com/kubeedge/beehive/pkg/core/context/context_factory.go index 9fb647866..4001f50ab 100644 --- a/staging/src/github.com/kubeedge/beehive/pkg/core/context/context_factory.go +++ b/staging/src/github.com/kubeedge/beehive/pkg/core/context/context_factory.go @@ -20,8 +20,12 @@ var ( once sync.Once ) +func init() { + InitContext(MsgCtxTypeChannel) +} + // InitContext gets global context instance -func InitContext(contextType string) *Context { +func InitContext(contextType string) { once.Do(func() { context = &Context{} switch contextType { @@ -33,12 +37,6 @@ func InitContext(contextType string) *Context { klog.Fatalf("Do not support context type:%s", contextType) } }) - return context -} - -// DestroyContext is used for Unit test -func DestroyContext() { - context = nil } // AddModule adds module into module context diff --git a/staging/src/github.com/kubeedge/beehive/pkg/core/core.go b/staging/src/github.com/kubeedge/beehive/pkg/core/core.go index 056b9383d..e8cf939f3 100644 --- a/staging/src/github.com/kubeedge/beehive/pkg/core/core.go +++ b/staging/src/github.com/kubeedge/beehive/pkg/core/core.go @@ -37,7 +37,6 @@ func GracefulShutdown() { modules := GetModules() for name, module := range modules { klog.Infof("Cleanup module %v", name) - beehiveContext.Cleanup(name) module.Cleanup() } } diff --git a/staging/src/github.com/kubeedge/beehive/pkg/core/module.go b/staging/src/github.com/kubeedge/beehive/pkg/core/module.go index d2209a621..08e91859c 100644 --- a/staging/src/github.com/kubeedge/beehive/pkg/core/module.go +++ b/staging/src/github.com/kubeedge/beehive/pkg/core/module.go @@ -6,7 +6,7 @@ import ( "k8s.io/klog" "github.com/kubeedge/beehive/pkg/common/config" - "github.com/kubeedge/beehive/pkg/core/context" + beehiveContext "github.com/kubeedge/beehive/pkg/core/context" ) const ( @@ -64,6 +64,8 @@ func isModuleEnabled(m string) bool { type moduleChangeCallback struct{} func (cb moduleChangeCallback) Callback(k string, v interface{}) { + beehiveContext.InitContext(beehiveContext.MsgCtxTypeChannel) + retryReadKey := func() interface{} { for times := 0; times < tryReadKeyTimes; times++ { // try to read the key again @@ -109,12 +111,11 @@ func (cb moduleChangeCallback) Callback(k string, v interface{}) { break } Register(module) - coreContext := context.InitContext(context.MsgCtxTypeChannel) //Init the module - coreContext.AddModule(module.Name()) + beehiveContext.AddModule(module.Name()) //Assemble typeChannels for sendToGroup - coreContext.AddModuleGroup(module.Name(), module.Group()) - go module.Start(coreContext) + beehiveContext.AddModuleGroup(module.Name(), module.Group()) + go module.Start() delete(disabledModules, m) klog.Infof("Callback: Module %s is enabled", m) } |