ENH: Adds service polling (#11)

Author: thomasjpfan

Dockerfile

@@ -16,7 +16,9 @@ ENV DF_DOCKER_HOST="unix:///var/run/docker.sock" \
     DF_RETRY="50" \
     DF_RETRY_INTERVAL="5" \
     DF_NOTIFY_LABEL="com.df.notify" \
-    DF_INCLUDE_NODE_IP_INFO="false"
+    DF_INCLUDE_NODE_IP_INFO="false" \
+    DF_SERVICE_POLLING_INTERVAL="-1" \
+    DF_USE_DOCKER_SERVICE_EVENTS="true"
 
 EXPOSE 8080
 

args.go

@@ -6,12 +6,14 @@ import (
 )
 
 type args struct {
-	Retry         int
-	RetryInterval int
+	ServicePollingInterval int
+	Retry                  int
+	RetryInterval          int
 }
 
 func getArgs() *args {
 	return &args{
+		ServicePollingInterval: getValue(-1, "DF_SERVICE_POLLING_INTERVAL"),
 		Retry:         getValue(1, "DF_RETRY"),
 		RetryInterval: getValue(0, "DF_RETRY_INTERVAL"),
 	}

docs/config.md

@@ -13,3 +13,5 @@ The following environment variables can be used when creating the `swarm-listene
 |DF_NOTIFY_REMOVE_NODE_URL |Comma separated list of URLs that will be used to send notification requests when a node is remove.<br>**Example**: `url1,url2`|
 |DF_RETRY           |Number of notification request retries<br>**Default**: `50`<br>**Example**: `100`|
 |DF_RETRY_INTERVAL  |Time between each notificationo request retry, in seconds.<br>**Default**: `5`<br>**Example**:`10`|
+|DF_SERVICE_POLLING_INTERVAL |Time between each service polling request, in seconds. When this value is set less than or equal to zero, service polling is disabled.<br>**Default**: `-1`<br>**Example**:`20`|
+|DF_USE_DOCKER_SERVICE_EVENTS|Use docker events api to get service updates.<br>**Default**:`true`|

main.go

@@ -12,7 +12,7 @@ func main() {
 
 	l.Printf("Starting Docker Flow: Swarm Listener")
 	args := getArgs()
-	swarmListener, err := service.NewSwarmListenerFromEnv(args.Retry, args.RetryInterval, l)
+	swarmListener, err := service.NewSwarmListenerFromEnv(args.Retry, args.RetryInterval, args.ServicePollingInterval, l)
 	if err != nil {
 		l.Printf("Failed to initialize Docker Flow: Swarm Listener")
 		l.Printf("ERROR: %v", err)

service/mocks.go

@@ -67,6 +67,11 @@ func (m *swarmServiceCacherMock) InsertAndCheck(ss SwarmServiceMini) bool {
 	return args.Bool(0)
 }
 
+func (m *swarmServiceCacherMock) IsNewOrUpdated(ss SwarmServiceMini) bool {
+	args := m.Called(ss)
+	return args.Bool(0)
+}
+
 func (m *swarmServiceCacherMock) Delete(ID string) {
 	m.Called(ID)
 }
@@ -81,6 +86,11 @@ func (m *swarmServiceCacherMock) Len() int {
 	return args.Int(0)
 }
 
+func (m *swarmServiceCacherMock) Keys() map[string]struct{} {
+	args := m.Called()
+	return args.Get(0).(map[string]struct{})
+}
+
 type nodeListeningMock struct {
 	mock.Mock
 }
@@ -136,3 +146,11 @@ func (m *notifyDistributorMock) HasServiceListeners() bool {
 func (m *notifyDistributorMock) HasNodeListeners() bool {
 	return m.Called().Bool(0)
 }
+
+type swarmServicePollingMock struct {
+	mock.Mock
+}
+
+func (m *swarmServicePollingMock) Run(eventChan chan<- Event) {
+	m.Called(eventChan)
+}

service/servicecache.go

@@ -5,9 +5,11 @@ import "sync"
 // SwarmServiceCacher caches sevices
 type SwarmServiceCacher interface {
 	InsertAndCheck(ss SwarmServiceMini) bool
+	IsNewOrUpdated(ss SwarmServiceMini) bool
 	Delete(ID string)
 	Get(ID string) (SwarmServiceMini, bool)
 	Len() int
+	Keys() map[string]struct{}
 }
 
 // SwarmServiceCache implements `SwarmServiceCacher`
@@ -36,6 +38,15 @@ func (c *SwarmServiceCache) InsertAndCheck(ss SwarmServiceMini) bool {
 
 }
 
+// IsNewOrUpdated returns true if service is new or updated
+func (c *SwarmServiceCache) IsNewOrUpdated(ss SwarmServiceMini) bool {
+	c.mux.RLock()
+	defer c.mux.RUnlock()
+
+	cachedService, ok := c.cache[ss.ID]
+	return !ok || !ss.Equal(cachedService)
+}
+
 // Delete delets service from cache
 func (c *SwarmServiceCache) Delete(ID string) {
 	c.mux.Lock()
@@ -57,3 +68,14 @@ func (c *SwarmServiceCache) Len() int {
 	defer c.mux.RUnlock()
 	return len(c.cache)
 }
+
+// Keys returns the keys of the cache
+func (c *SwarmServiceCache) Keys() map[string]struct{} {
+	c.mux.RLock()
+	defer c.mux.RUnlock()
+	output := map[string]struct{}{}
+	for key := range c.cache {
+		output[key] = struct{}{}
+	}
+	return output
+}

service/servicecache_test.go

@@ -132,9 +132,44 @@ func (s *SwarmServiceCacheTestSuite) Test_GetAndRemove_InCache_ReturnsSwarmServi
 	s.Cache.Delete(s.SSMini.ID)
 	s.AssertNotInCache(s.SSMini)
 	s.Equal(s.SSMini, removedSSMini)
+}
+
+func (s *SwarmServiceCacheTestSuite) Test_Keys() {
+	s.Cache.InsertAndCheck(s.SSMini)
+	s.AssertInCache(s.SSMini)
+
+	keys := s.Cache.Keys()
+
+	s.Require().Len(keys, 1)
+	s.Contains(keys, s.SSMini.ID)
+
+}
+
+func (s *SwarmServiceCacheTestSuite) Test_IsNewOrUpdated_ServiceInCache() {
+	s.Cache.InsertAndCheck(s.SSMini)
+	s.AssertInCache(s.SSMini)
 
+	newOrUpdated := s.Cache.IsNewOrUpdated(s.SSMini)
+	s.False(newOrUpdated)
 }
 
+func (s *SwarmServiceCacheTestSuite) Test_IsNewOrUpdated_ServiceNotInCache() {
+	newOrUpdated := s.Cache.IsNewOrUpdated(s.SSMini)
+	s.True(newOrUpdated)
+}
+
+func (s *SwarmServiceCacheTestSuite) Test_IsNewOrUpdated_ServiceIsDifferentCache() {
+
+	s.Cache.InsertAndCheck(s.SSMini)
+	s.AssertInCache(s.SSMini)
+
+	anotherSSMini := getNewSwarmServiceMini()
+	anotherSSMini.Name = "anotherName"
+
+	newOrUpdated := s.Cache.IsNewOrUpdated(anotherSSMini)
+	s.True(newOrUpdated)
+
+}
 func (s *SwarmServiceCacheTestSuite) Test_GetAndRemove_NotInCache_ReturnsFalse() {
 
 	_, ok := s.Cache.Get(s.SSMini.ID)

service/servicepoller.go

@@ -0,0 +1,82 @@
+package service
+
+import (
+	"context"
+	"log"
+	"time"
+)
+
+// SwarmServicePolling provides an interface for polling service changes
+type SwarmServicePolling interface {
+	Run(eventChan chan<- Event)
+}
+
+// SwarmServicePoller implements `SwarmServicePoller`
+type SwarmServicePoller struct {
+	SSClient        SwarmServiceInspector
+	SSCache         SwarmServiceCacher
+	PollingInterval int
+	MinifyFunc      func(SwarmService) SwarmServiceMini
+	Log             *log.Logger
+}
+
+// NewSwarmServicePoller creates a new `SwarmServicePoller`
+func NewSwarmServicePoller(
+	ssClient SwarmServiceInspector,
+	ssCache SwarmServiceCacher,
+	pollingInterval int,
+	minifyFunc func(SwarmService) SwarmServiceMini,
+	log *log.Logger,
+) *SwarmServicePoller {
+	return &SwarmServicePoller{
+		SSClient:        ssClient,
+		SSCache:         ssCache,
+		PollingInterval: pollingInterval,
+		MinifyFunc:      minifyFunc,
+		Log:             log,
+	}
+}
+
+// Run starts poller and places events onto `eventChan`
+func (s SwarmServicePoller) Run(
+	eventChan chan<- Event) {
+
+	if s.PollingInterval <= 0 {
+		return
+	}
+
+	s.Log.Printf("Polling for Service Changes")
+	time.Sleep(time.Duration(s.PollingInterval) * time.Second)
+	for {
+		services, err := s.SSClient.SwarmServiceList(context.Background())
+		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)
+				ssMini := s.MinifyFunc(ss)
+				if s.SSCache.IsNewOrUpdated(ssMini) {
+					eventChan <- Event{
+						Type:     EventTypeCreate,
+						ID:       ss.ID,
+						TimeNano: nowTimeNano,
+						UseCache: true,
+					}
+				}
+			}
+
+			// Remaining keys are removal events
+			for k := range keys {
+				eventChan <- Event{
+					Type:     EventTypeRemove,
+					ID:       k,
+					TimeNano: nowTimeNano,
+					UseCache: true,
+				}
+			}
+		}
+		time.Sleep(time.Duration(s.PollingInterval) * time.Second)
+	}
+}