Code review markups

README.md

@@ -11,6 +11,7 @@ Marathon-lb is a tool for managing HAProxy, by consuming [Marathon's](https://gi
  * DCOS integration
  * Automated Docker image builds ([mesosphere/marathon-lb](https://hub.docker.com/r/mesosphere/marathon-lb))
  * Global HAProxy templates which can be supplied at launch
+ * Supports IP-per-task integration, such as [Project Calico](https://github.com/projectcalico/calico-containers)
 
 ### Getting started
 
@@ -44,17 +45,6 @@ the [templates section](Longhelp.md#templates)
 You can access the HAProxy statistics via `:9090/haproxy?stats`, and you can
 retrieve the current HAProxy config from the `:9090/_haproxy_getconfig` endpoint.
 
-Marathon LB supports load balancing for applications that use the Mesos IP-per-task
-feature, whereby each task is assigned unique, accessible, IP addresses.  For these
-tasks services are directly accessible via the configured discovery ports and there
-is no host port mapping.  Note, that due to limitations with Marathon (see 
-[mesosphere/marathon#3636](https://github.com/mesosphere/marathon/issues/3636)) 
-configured service ports are not exposed to Marathon LB for IP-per-task apps.  
-For these apps, where the service ports are missing from the Marathon app data,
-Marathon LB will automatically assign port values from a configurable range.  The range
-is configured using the `--min-serv-port-ip-per-task` and `--max-serv-port-ip-per-task`
-options.
-
 ## Deployment
 The package is currently available [from the universe](https://github.com/mesosphere/universe).
 To deploy marathon-lb on the public slaves in your DCOS cluster,
@@ -248,6 +238,19 @@ An example minimal configuration for a [test instance of nginx is included here]
 
 Zero downtime deployments are accomplished through the use of a Lua module, which reports the number of HAProxy processes which are currently running by hitting the stats endpoint at the `/_haproxy_getpids`. After a restart, there will be multiple HAProxy PIDs until all remaining connections have gracefully terminated. By waiting for all connections to complete, you may safely and deterministically drain tasks. A caveat of this, however, is that if you have any long-lived connections on the same LB, HAProxy will continue to run and serve those connections until they complete, thereby breaking this technique.
 
+## Mesos with IP-per-task support
+
+Marathon-lb supports load balancing for applications that use the Mesos IP-per-task
+feature, whereby each task is assigned unique, accessible, IP addresses.  For these
+tasks services are directly accessible via the configured discovery ports and there
+is no host port mapping.  Note, that due to limitations with Marathon (see 
+[mesosphere/marathon#3636](https://github.com/mesosphere/marathon/issues/3636)) 
+configured service ports are not exposed to marathon-lb for IP-per-task apps.  
+For these apps, if the service ports are missing from the Marathon app data,
+marathon-lb will automatically assign port values from a configurable range.  The range
+is configured using the `--min-serv-port-ip-per-task` and `--max-serv-port-ip-per-task`
+options.
+
 ## Contributing
 
 PRs are welcome, but here are a few general guidelines:

marathon_lb.py

@@ -1041,6 +1041,10 @@ def get_apps(marathon):
 
         service_ports = SERVICE_PORT_ASSIGNER.get_service_ports(app)
         for i, servicePort in enumerate(service_ports):
+            if servicePort is None:
+                logger.warning("Skipping undefined service port")
+                continue
+
             service = MarathonService(
                         appId, servicePort, get_health_check(app, i))
 
@@ -1353,14 +1357,18 @@ def process_sse_events(marathon, config_file, groups,
                 'either specify both --min-serv-port-ip-per-task '
                 'and --max-serv-port-ip-per-task or neither (set both to zero '
                 'to disable auto assignment)')
+        if args.min_serv_port_ip_per_task > args.max_serv_port_ip_per_task:
+            arg_parser.error(
+                'cannot set --min-serv-port-ip-per-task to a higher value '
+                'than --max-serv-port-ip-per-task')
         if len(args.group) == 0:
             arg_parser.error('argument --group is required: please' +
                              'specify at least one group name')
 
     # Configure the service port assigner if min/max ports have been specified.
     if args.min_serv_port_ip_per_task and args.max_serv_port_ip_per_task:
-        SERVICE_PORT_ASSIGNER.set_ports(int(args.min_serv_port_ip_per_task),
-                                        int(args.max_serv_port_ip_per_task))
+        SERVICE_PORT_ASSIGNER.set_ports(args.min_serv_port_ip_per_task,
+                                        args.max_serv_port_ip_per_task)
 
     # Set request retries
     s = requests.Session()

tests/test_utils.py

@@ -76,6 +76,16 @@ def setUp(self):
         self.assigner = ServicePortAssigner()
         self.assigner.set_ports(10000, 10020)
 
+    def test_no_assignment_ports_not_set(self):
+        """
+        Test that no assignments are made if the port values are not set.
+        """
+        assigner = ServicePortAssigner()
+        app = _get_app(idx=1, num_ports=3, num_tasks=1)
+
+        # No ports set
+        self.assertEquals(assigner.get_service_ports(app), [])
+
     def test_not_ip_per_task(self):
         """
         Test a non-IP-per-task app returns the service ports defined in the
@@ -157,16 +167,31 @@ def test_ip_per_task_max_clash(self):
             ports = self.assigner.get_service_ports(app)
         self.assertEquals(ports, list(range(10000, 10010)))
 
+    def test_ip_per_task_exhausted(self):
+        """
+        Check that ports are returned as None when the ports list is
+        exhausted.
+        """
+        # Create an app with 2 more discovery ports than we are able to
+        # allocate.  Check the last two ports are unassigned, and check all
+        # ports are allocated from the correct range.
+        app = _get_app(idx=1, num_ports=24, num_tasks=1)
+        ports = self.assigner.get_service_ports(app)
+        self.assertEquals(ports[-3:], [None] * 3)
+        self.assertEquals(sorted(ports[:-3]), list(range(10000, 10021)))
+
 
 def _get_app(idx=1, num_ports=3, num_tasks=1, ip_per_task=True,
              inc_service_ports=False):
     app = {
-        "id": "app-%d" % idx
+        "id": "app-%d" % idx,
+        "tasks": [_get_task(idx*10 + idx2) for idx2 in range(num_tasks)],
+        "ports": [],
+        "ipAddress": None,
     }
+
     if inc_service_ports:
         app["ports"] = list(range(100, 100 + num_ports))
-    else:
-        app["ports"] = []
 
     if ip_per_task:
         app["ipAddress"] = {
@@ -177,12 +202,10 @@ def _get_app(idx=1, num_ports=3, num_tasks=1, ip_per_task=True,
             }
         }
 
-    app["tasks"] = [_get_task(idx*10, num_ports) for idx in range(num_tasks)]
-
     return app
 
 
-def _get_task(idx, num_ports):
+def _get_task(idx):
     return {
         "id": "task-%d" % idx,
         "ipAddresses": [{"ipAddress": "1.2.3.4"}]

utils.py

@@ -47,8 +47,9 @@ def __init__(self):
     def _assign_new_service_port(self, app, task_port):
         assert self.can_assign
 
-        remaining_ports = self.max_ports - len(self.ports_by_app)
-        assert remaining_ports, "Service ports are exhausted"
+        if self.max_ports <= len(self.ports_by_app):
+            logger.warning("Service ports are exhausted")
+            return None
 
         # We don't want to be searching forever, so limit the number of times
         # we clash to the number of remaining ports.
@@ -69,7 +70,7 @@ def _assign_new_service_port(self, app, task_port):
 
         # We must have assigned a unique port by now since we know there were
         # some available.
-        assert port and port not in ports
+        assert port and port not in ports, port
 
         logger.debug("Assigned new port: %d", port)
         return port
@@ -88,21 +89,26 @@ def set_ports(self, min_port, max_port):
         :param min_port: The minimum port value
         :param max_port: The maximum port value
         """
+        assert not self.ports_by_app
+        assert max_port >= min_port
         self.min_port = min_port
         self.max_port = max_port
         self.max_ports = max_port - min_port + 1
         self.can_assign = self.min_port and self.max_port
-        assert not self.ports_by_app
-        assert self.max_ports > 1
 
     def reset(self):
         """
         Reset the assigner so that ports are newly assigned.
-        :return:
         """
         self.ports_by_app = {}
 
     def get_service_ports(self, app):
+        """
+        Return a list of service ports for this app.
+        :param app: The application.
+        :return: The list of ports.    Note that if auto-assigning and ports
+        become exhausted, a port may be returned as None.
+        """
         ports = app['ports']
         if not ports and is_ip_per_task(app) and self.can_assign:
             logger.warning("Auto assigning service port for "