Swarm ingress controller fixes + documentation/ example YAML (#679)

Signed-off-by: David Chung <[email protected]>

Author: David Chung

docs/controller/ingress/example.yml

@@ -0,0 +1,77 @@
+#
+# Example Ingress YML that will sync services in Docker Swarm to the backend.
+# The backend is specified by the L4Plugin property.
+#
+# To run this:
+#
+# INFRAKIT_MANAGER_BACKEND=swarm infrakit plugin start manager simulator ingress
+#
+# In another console:
+# infrakit ingress controller commit -y path/to/this/file
+#
+# Creating Docker services
+# docker network create --driver overlay --ingress ingress
+# docker service create --network ingress --name t2 --publish 7777:80 nginx
+#
+# Verify that the route has been added in the simulator/lb1:
+# infrakit simulator/lb1 routes ls
+#
+# FRONTEND PORT    PROTOCOL   BACKEND PORT  CERT
+# 7777             TCP        7777  <-- from the swarm service
+# 80               http       8080  <-- from the static route in the config
+#
+# In Docker:
+#  docker service ls
+#ID                  NAME                MODE                REPLICAS            IMAGE               PORTS
+#m4fghruao79i        t2                  replicated          1/1                 nginx:latest        *:7777->80/tcp
+#
+# Remove the service
+# docker service rm test1
+#
+# infrakit simulator/lb1 routes ls  # should be empty
+#
+# FRONTEND PORT    PROTOCOL   BACKEND PORT  CERT
+# 80               http       8080  <-- from the static route
+
+kind: ingress
+metadata:
+  name: test.com
+  tags:
+    project: testing
+    user: chungers
+
+# options block map to pkg/controller/ingress/types/Options
+options:
+  # SyncInterval is how often to sync changes between the services and the LB
+  SyncInterval: 1s  # syntax is a string form of Go time.Duration
+
+# properties block map to pkg/controller/ingress/types/Properties
+properties:
+  - Backends:
+    Groups:
+
+      # This is a group at socket(group), groupID(cattle).
+      - group/cattle
+
+    # This is the plugin name of the L4 plugin. When you run `infrakit plugin start ... simulator`
+    # the default socket file name is 'simulator' and there's a default lb2 in the RPC object.
+    L4Plugin: simulator/lb1
+
+    # Here we have a static route that is always present.
+    Routes:
+      - LoadBalancerPort: 80
+        LoadBalancerProtocol: https
+        Port: 8080
+        Protocol: http
+
+    # Plus all the services in Swarm that have --publish <frontend-port>:<container_port>
+    RouteSources:
+      swarm:
+        Host: unix:///var/run/docker.sock
+
+    # This field here is required and leave it at 'default'. (TODO - make this not required).
+    # default will match any services that don't have vhost information.
+    # There are ways to specify the vhost for a service in the label.  If that information
+    # isn't there, the service will have to be lumped to a 'default' vhost.  Otherwise,
+    # the value of this field will be used to group and match the service which loadbalancer.
+    Vhost: default

pkg/controller/ingress/controller.go

@@ -9,6 +9,7 @@ import (
 	"github.com/docker/infrakit/pkg/discovery"
 	"github.com/docker/infrakit/pkg/manager"
 	"github.com/docker/infrakit/pkg/types"
+	"golang.org/x/net/context"
 )
 
 // NewController returns a controller implementation
@@ -76,7 +77,7 @@ func (m *managed) Enforce(spec types.Spec) (*types.Object, error) {
 	if err != nil {
 		return nil, err
 	}
-	m.start()
+	m.Start()
 	return m.object(), nil
 }
 
@@ -88,15 +89,41 @@ func (m *managed) Inspect() (*types.Object, error) {
 // Free implements internal/Managed
 func (m *managed) Free() (*types.Object, error) {
 	if m.started() {
-		m.stop()
+		m.Stop()
 	}
 	return m.Inspect()
 }
 
 // Terminate implements internal/Managed
 func (m *managed) Terminate() (*types.Object, error) {
 	if m.started() {
-		m.stop()
+		m.Stop()
 	}
 	return m.Inspect()
 }
+
+// Start implements internal/ControlLoop
+func (m *managed) Start() {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	if m.process != nil && m.poller != nil {
+		go m.poller.Run(context.Background())
+	}
+}
+
+// Stop implements internal/ControlLoop
+func (m *managed) Stop() error {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	if m.process != nil && m.poller != nil {
+		m.poller.Stop()
+	}
+	return nil
+}
+
+// Running implements internal/ControlLoop
+func (m *managed) Running() bool {
+	return m.started()
+}

pkg/controller/ingress/fsm.go

@@ -170,7 +170,7 @@ func (c *managed) init(in types.Spec) (err error) {
 	}
 
 	if c.ticker == nil {
-		interval := c.options.SyncInterval
+		interval := c.options.SyncInterval.Duration()
 		if interval == 0 {
 			interval = ingress.DefaultSyncInterval
 		}

pkg/controller/ingress/fsm_test.go

@@ -86,22 +86,22 @@ func TestControllerInitSpec(t *testing.T) {
 
 	managedObject := &managed{
 		options: ingress.Options{
-			SyncInterval: expectedInterval,
+			SyncInterval: types.Duration(expectedInterval),
 		},
 	}
 
 	err := managedObject.init(types.Spec{})
 	require.NoError(t, err)
 
 	t.Log("verify that the default value remains despite no Options in the spec")
-	require.Equal(t, expectedInterval, managedObject.options.SyncInterval)
+	require.Equal(t, expectedInterval, managedObject.options.SyncInterval.Duration())
 
 	t.Log("verify that spec's option value makes into the ingress.Options")
 	managedObject = &managed{}
 
 	expectedOptions := ingress.Options{
 		HardSync:     true,
-		SyncInterval: expectedInterval,
+		SyncInterval: types.Duration(expectedInterval),
 	}
 
 	err = managedObject.init(types.Spec{

pkg/controller/ingress/l4.go

@@ -93,16 +93,14 @@ func configureL4(elb loadbalancer.L4, desired []loadbalancer.Route, options type
 		log.Info("CHANGED", "name", elb.Name(), "listener", l)
 	}
 	for _, l := range toRemove {
-		log.Info("REMOVE", "name", elb.Name(), "listener", l)
 
-		if options.RemoveListeners {
-			_, err := elb.Unpublish(l.LoadBalancerPort)
-			if err != nil {
-				log.Warn("err unpublishing", "route", l, "err", err)
-				return err
-			}
-			log.Info("REMOVED", "name", elb.Name(), "listener", l)
+		log.Info("REMOVE", "name", elb.Name(), "listener", l)
+		_, err := elb.Unpublish(l.LoadBalancerPort)
+		if err != nil {
+			log.Warn("err unpublishing", "route", l, "err", err)
+			return err
 		}
+		log.Info("REMOVED", "name", elb.Name(), "listener", l)
 	}
 	return nil
 }

pkg/controller/ingress/managed.go

@@ -18,7 +18,6 @@ import (
 	"github.com/docker/infrakit/pkg/spi/instance"
 	"github.com/docker/infrakit/pkg/spi/loadbalancer"
 	"github.com/docker/infrakit/pkg/types"
-	"golang.org/x/net/context"
 )
 
 var log = logutil.New("module", "controller/ingress")
@@ -101,37 +100,9 @@ func (c *managed) l4Client(spec ingress.Spec) (loadbalancer.L4, error) {
 	return loadbalancer_rpc.NewClient(spec.L4Plugin, endpoint.Address)
 }
 
-// Run starts the controller given the spec it needs to maintain
-func (c *managed) Run(spec types.Spec) error {
-	err := c.init(spec)
-	if err != nil {
-		return err
-	}
-	c.start()
-	return nil
-}
-
 func (c *managed) started() bool {
 	c.lock.RLock()
 	defer c.lock.RUnlock()
 
 	return c.process != nil && c.poller != nil
 }
-
-func (c *managed) start() {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
-	if c.process != nil && c.poller != nil {
-		go c.poller.Run(context.Background())
-	}
-}
-
-func (c *managed) stop() {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
-	if c.process != nil && c.poller != nil {
-		c.poller.Stop()
-	}
-}

pkg/controller/ingress/managed_test.go

@@ -22,8 +22,7 @@ func TestManagedStartStop(t *testing.T) {
 	ticker := make(chan time.Time, 1)
 	leader := make(chan bool, 1)
 
-	doneWork := make(chan interface{})
-
+	doneWork := make(chan int, 1)
 	managedObject := &managed{
 		Leadership:   fakeLeadership(leader),
 		ticker:       ticker,
@@ -33,9 +32,7 @@ func TestManagedStartStop(t *testing.T) {
 
 		routes: func() (map[ingress.Vhost][]loadbalancer.Route, error) {
 			// if this function is called then we know we've done work in the state transition
-			// from syncing to waiting
-			close(doneWork)
-
+			doneWork <- 1
 			return nil, nil
 		},
 	}
@@ -56,7 +53,7 @@ func TestManagedStartStop(t *testing.T) {
 	err := managedObject.init(spec)
 	require.NoError(t, err)
 
-	managedObject.start()
+	managedObject.Start()
 
 	t.Log("verify initial state machine is in the follower state")
 	require.Equal(t, follower, managedObject.stateMachine.State())
@@ -96,20 +93,14 @@ func TestManagedStartStop(t *testing.T) {
 	leader <- true
 
 	// here we change the routes function to test for another close
-	doneWork2 := make(chan interface{})
-	managedObject.routes = func() (map[ingress.Vhost][]loadbalancer.Route, error) {
-		// if this function is called then we know we've done work in the state transition
-		// from syncing to waiting
-		close(doneWork2)
-		return nil, nil
-	}
 
 	ticker <- time.Now()
 
-	t.Log("verify state machine moved to the waiting state")
+	<-doneWork // if not called, the test will hang here
 
-	<-doneWork2 // if not called, the test will hang here
+	t.Log("verify state machine moved to the waiting state")
 	require.Equal(t, waiting, managedObject.stateMachine.State())
 
-	managedObject.stop()
+	managedObject.Stop()
+
 }

pkg/controller/ingress/swarm/info.go

@@ -1,7 +1,6 @@
 package swarm
 
 import (
-	log "github.com/Sirupsen/logrus"
 	"github.com/docker/docker/client"
 	"golang.org/x/net/context"
 )
@@ -10,7 +9,7 @@ import (
 func AmISwarmLeader(ctx context.Context, client client.APIClient) (bool, error) {
 	info, err := client.Info(ctx)
 
-	log.Debugln("info=", info, "err=", err)
+	log.Debug("Swarm info", "info", info, "err", err)
 
 	if err != nil {
 		return false, err
@@ -19,19 +18,19 @@ func AmISwarmLeader(ctx context.Context, client client.APIClient) (bool, error)
 	// inspect itself to see if i am the leader
 	node, _, err := client.NodeInspectWithRaw(ctx, info.Swarm.NodeID)
 
-	log.Debugln("nodeId=", info.Swarm.NodeID, "node=", node, "err=", err)
+	log.Debug("Inspect node", "nodeID", info.Swarm.NodeID, "node", node, "err", err, "V", debugV)
 
 	if err != nil {
 		return false, err
 	}
 
-	log.Debugln("manager=", node.ManagerStatus)
+	log.Debug("manager status", "status", node.ManagerStatus, "V", debugV)
 
 	if node.ManagerStatus == nil {
 		return false, nil
 	}
 
-	log.Debugln("leader=", node.ManagerStatus.Leader)
+	log.Debug("leader status", "leader", node.ManagerStatus.Leader)
 
 	return node.ManagerStatus.Leader, nil
 }

pkg/controller/ingress/swarm/init.go

@@ -5,11 +5,19 @@ import (
 
 	"github.com/docker/go-connections/tlsconfig"
 	ingress "github.com/docker/infrakit/pkg/controller/ingress/types"
+	logutil "github.com/docker/infrakit/pkg/log"
 	"github.com/docker/infrakit/pkg/spi/loadbalancer"
 	"github.com/docker/infrakit/pkg/types"
 	"github.com/docker/infrakit/pkg/util/docker"
 )
 
+var log = logutil.New("module", "controller/ingress/swarm")
+
+const (
+	debugV  = logutil.V(300)
+	debugV2 = logutil.V(310)
+)
+
 func init() {
 
 	// Register the swarm based ingress route finder.  This will be included when the package is imported
@@ -20,11 +28,13 @@ func init() {
 	)
 }
 
+// Docker is alias for docker connection information
+type Docker docker.ConnectInfo
+
 // Spec is the struct that captures the configuration of the swarm-based ingress route finder
 type Spec struct {
-
 	// Docker holds the connection params to the Docker engine for join tokens, etc.
-	Docker docker.ConnectInfo
+	Docker `json:",inline" yaml:",inline"`
 }
 
 // RoutesFromSwarmServices determines the routes based on the services running in the Docker swarm
@@ -52,6 +62,7 @@ func RoutesFromSwarmServices(properties *types.Any,
 		return nil, err
 	}
 
+	log.Info("Connected to Docker", "client", dockerClient)
 	routes, err := NewServiceRoutes(dockerClient).SetOptions(options).Build()
 	if err != nil {
 		return nil, err

pkg/controller/ingress/swarm/init_test.go

@@ -19,9 +19,9 @@ func TestParseSpec(t *testing.T) {
 			RouteSources: map[string]*types.Any{
 				"swarm": types.AnyValueMust(
 					Spec{
-						Docker: docker.ConnectInfo{
+						Docker(docker.ConnectInfo{
 							Host: "/var/run/docker.sock",
-						},
+						}),
 					},
 				),
 			},