ENH: Handles updates for global services (#15)

* ENH: Handles of image updates for global services

* ENH: Considers nodeInfo when polling

service/eventservicelistener.go

@@ -66,11 +66,5 @@ func (s SwarmServiceListener) ListenForServiceEvents(eventChan chan<- Event) {
 
 // validEventNode returns true when event is valid (should be passed through)
 func (s SwarmServiceListener) validEventNode(msg events.Message) bool {
-	if msg.Action != "update" {
-		return true
-	}
-	if name, ok := msg.Actor.Attributes["updatestate.new"]; ok && len(name) > 0 {
-		return false
-	}
 	return true
 }

service/servicepoller.go

@@ -16,6 +16,7 @@ type SwarmServicePoller struct {
 	SSClient        SwarmServiceInspector
 	SSCache         SwarmServiceCacher
 	PollingInterval int
+	IncludeNodeInfo bool
 	MinifyFunc      func(SwarmService) SwarmServiceMini
 	Log             *log.Logger
 }
@@ -25,13 +26,15 @@ func NewSwarmServicePoller(
 	ssClient SwarmServiceInspector,
 	ssCache SwarmServiceCacher,
 	pollingInterval int,
+	includeNodeInfo bool,
 	minifyFunc func(SwarmService) SwarmServiceMini,
 	log *log.Logger,
 ) *SwarmServicePoller {
 	return &SwarmServicePoller{
 		SSClient:        ssClient,
 		SSCache:         ssCache,
 		PollingInterval: pollingInterval,
+		IncludeNodeInfo: includeNodeInfo,
 		MinifyFunc:      minifyFunc,
 		Log:             log,
 	}
@@ -45,17 +48,27 @@ func (s SwarmServicePoller) Run(
 		return
 	}
 
+	ctx := context.Background()
+
 	s.Log.Printf("Polling for Service Changes")
 	time.Sleep(time.Duration(s.PollingInterval) * time.Second)
+
 	for {
-		services, err := s.SSClient.SwarmServiceList(context.Background())
+		services, err := s.SSClient.SwarmServiceList(ctx)
 		if err != nil {
 			s.Log.Printf("ERROR (SwarmServicePolling): %v", err)
 		} else {
 			nowTimeNano := time.Now().UTC().UnixNano()
 			keys := s.SSCache.Keys()
 			for _, ss := range services {
 				delete(keys, ss.ID)
+
+				if s.IncludeNodeInfo {
+					if nodeInfo, err := s.SSClient.GetNodeInfo(ctx, ss); err == nil {
+						ss.NodeInfo = nodeInfo
+					}
+				}
+
 				ssMini := s.MinifyFunc(ss)
 				if s.SSCache.IsNewOrUpdated(ssMini) {
 					eventChan <- Event{

service/servicepoller_test.go

@@ -40,6 +40,7 @@ func (s *ServicePollerTestSuite) SetupTest() {
 		s.SSClientMock,
 		s.SSCacheMock,
 		1,
+		false,
 		s.MinifyFunc,
 		s.Logger,
 	)

service/swarmlistener.go

@@ -186,7 +186,7 @@ func NewSwarmListenerFromEnv(
 	}
 
 	ssPoller := NewSwarmServicePoller(
-		ssClient, ssCache, servicePollingInterval,
+		ssClient, ssCache, servicePollingInterval, includeNodeInfo,
 		func(ss SwarmService) SwarmServiceMini {
 			return MinifySwarmService(ss, ignoreKey, "com.docker.stack.namespace")
 		}, logger)

service/task.go

@@ -107,7 +107,10 @@ func GetTaskList(ctx context.Context, client *client.Client, serviceID string) (
 		rollback    bool
 	)
 
-	taskList := []swarm.Task{}
+	taskList, err := getUpToDateTasks()
+	if err != nil {
+		return taskList, err
+	}
 
 	for {
 		service, _, err := client.ServiceInspectWithRaw(ctx, serviceID, types.ServiceInspectOptions{})