Merge pull request #2741 from LiilyZhang/zhangl/Issue2555

Issue 2555 - Race condition when agent un-registering leaves agent as…

events/events.go

@@ -121,6 +121,9 @@ const (
 
 	// Secret related
 	UPDATED_SECRETS EventId = "SECRET_UPDATES"
+
+	// ESS related
+	ESS_UNCONFIG EventId = "ESS_UNCONFIG"
 )
 
 type EndContractCause string
@@ -1759,6 +1762,36 @@ func NewNodeShutdownCompleteMessage(id EventId, errorMsg string) *NodeShutdownCo
 	}
 }
 
+type SyncServiceCleanedUpMessage struct {
+	event Event
+	err   string
+}
+
+func (n *SyncServiceCleanedUpMessage) Event() Event {
+	return n.event
+}
+
+func (n SyncServiceCleanedUpMessage) String() string {
+	return n.ShortString()
+}
+
+func (n SyncServiceCleanedUpMessage) ShortString() string {
+	return fmt.Sprintf("Event: %v, Error: %v", n.event, n.err)
+}
+
+func (n SyncServiceCleanedUpMessage) Err() string {
+	return n.err
+}
+
+func NewSyncServiceCleanedUpMessage(id EventId, errorMsg string) *SyncServiceCleanedUpMessage {
+	return &SyncServiceCleanedUpMessage{
+		event: Event{
+			Id: id,
+		},
+		err: errorMsg,
+	}
+}
+
 // This is a special message that the message dispatcher knows about.
 type WorkerStopMessage struct {
 	event Event

governance/governance.go

@@ -56,6 +56,7 @@ type GovernanceWorker struct {
 	limitedRetryEC    exchange.ExchangeContext
 	exchErrors        cache.Cache
 	noworkDispatch    int64 // The last time the NoWorkHandler was dispatched.
+	essCleanedUp      bool
 }
 
 func NewGovernanceWorker(name string, cfg *config.HorizonConfig, db *bolt.DB, pm *policy.PolicyManager) *GovernanceWorker {
@@ -83,6 +84,7 @@ func NewGovernanceWorker(name string, cfg *config.HorizonConfig, db *bolt.DB, pm
 		limitedRetryEC:  lrec,
 		exchErrors:      cache.NewSimpleMapCache(),
 		noworkDispatch:  time.Now().Unix(),
+		essCleanedUp:    false,
 	}
 
 	// Start the worker and set the no work interval to 10 seconds.
@@ -311,6 +313,10 @@ func (w *GovernanceWorker) NewEvent(incoming events.Message) {
 			w.Commands <- worker.NewTerminateCommand("shutdown")
 		}
 
+	case *events.SyncServiceCleanedUpMessage:
+		w.essCleanedUp = true
+		glog.V(5).Infof(logString(fmt.Sprintf("Receive SyncServiceCleanedUpMessage event, set ess.CleanUp to %t for governance worker.", w.essCleanedUp)))
+
 	case *events.NodeHeartbeatStateChangeMessage:
 		msg, _ := incoming.(*events.NodeHeartbeatStateChangeMessage)
 		switch msg.Event().Id {
@@ -1401,12 +1407,14 @@ func (w *GovernanceWorker) NoWorkHandler() {
 
 	// When all subworkers are down, start the shutdown process.
 	if w.IsWorkerShuttingDown() && w.ShuttingDownCmd != nil {
-		if w.AreAllSubworkersTerminated() {
+		if w.AreAllSubworkersTerminated() && w.essCleanedUp {
 			glog.V(5).Infof(logString(fmt.Sprintf("GovernanceWorker initiating async shutdown.")))
 			cmd := w.ShuttingDownCmd
 			// This is one of the few go routines that should NOT be abstracted as a subworker.
 			go w.nodeShutdown(cmd)
 			w.ShuttingDownCmd = nil
+		} else if !w.essCleanedUp {
+			glog.V(5).Infof(logString(fmt.Sprintf("GovernanceWorker waiting for ESS to finish cleanup.")))
 		} else {
 			glog.V(5).Infof(logString(fmt.Sprintf("GovernanceWorker waiting for subworkers to terminate.")))
 		}

resource/commands.go

@@ -26,7 +26,7 @@ func NewNodeConfigCommand(msg *events.EdgeRegisteredExchangeMessage) *NodeConfig
 
 // This worker command is used to tell the worker than the node is done shutting down and so it can terminate itself.
 type NodeUnconfigCommand struct {
-	msg *events.NodeShutdownCompleteMessage
+	msg *events.NodeShutdownMessage
 }
 
 func (n NodeUnconfigCommand) String() string {
@@ -37,7 +37,7 @@ func (n NodeUnconfigCommand) ShortString() string {
 	return fmt.Sprintf("NodeUnconfig Command, Msg: %v", n.msg)
 }
 
-func NewNodeUnconfigCommand(msg *events.NodeShutdownCompleteMessage) *NodeUnconfigCommand {
+func NewNodeUnconfigCommand(msg *events.NodeShutdownMessage) *NodeUnconfigCommand {
 	return &NodeUnconfigCommand{
 		msg: msg,
 	}

resource/resource_worker.go

@@ -77,10 +77,10 @@ func (w *ResourceWorker) NewEvent(incoming events.Message) {
 		w.EC = worker.NewExchangeContext(fmt.Sprintf("%v/%v", msg.Org(), msg.DeviceId()), msg.Token(), w.Config.Edge.ExchangeURL, w.Config.GetCSSURL(), w.Config.Collaborators.HTTPClientFactory)
 		w.Commands <- NewNodeConfigCommand(msg)
 
-	case *events.NodeShutdownCompleteMessage:
-		msg, _ := incoming.(*events.NodeShutdownCompleteMessage)
+	case *events.NodeShutdownMessage:
+		msg, _ := incoming.(*events.NodeShutdownMessage)
 		switch msg.Event().Id {
-		case events.UNCONFIGURE_COMPLETE:
+		case events.START_UNCONFIGURE:
 			w.Commands <- NewNodeUnconfigCommand(msg)
 		}
 
@@ -103,9 +103,13 @@ func (w *ResourceWorker) CommandHandler(command worker.Command) bool {
 
 	case *NodeUnconfigCommand:
 		cmd, _ := command.(*NodeUnconfigCommand)
-		if err := w.handleNodeUnconfigCommand(cmd); err != nil {
+		err := w.handleNodeUnconfigCommand(cmd)
+		errMsg := ""
+		if err != nil {
 			glog.Errorf(reslog(fmt.Sprintf("Error handling node unconfig command: %v", err)))
+			errMsg = err.Error()
 		}
+		w.Messages() <- events.NewSyncServiceCleanedUpMessage(events.ESS_UNCONFIG, errMsg)
 
 	default:
 		return false