feature: tcp & udp streams (#6)

Dockerfile

@@ -22,6 +22,8 @@ ADD src /opt/vergilius
 RUN cd /opt/vergilius/ && python setup.py install
 WORKDIR /opt/vergilius/
 
-EXPOSE 80 443
+EXPOSE 80 443 7000-8000
+
+ENV DHPARAM_LENGTH 4096
 
 RUN apt-get clean && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*

README.md

@@ -37,10 +37,16 @@ Consul service config example
 }
 ```
 
-Vergilius looks for registered services with tags `http` and `http2`, creates upstream with all containers of this service,
+Vergilius looks for registered services with tags `http` and `http2` creates upstream with all containers of this service,
 routes requests from `(www.)?service.example.com` and `*.(www.)?service.example.com` to containers using nginx
 `least_conn` balancing algorithm.
 
+You can also add `tcp` and `udp` tags to service, vergilus will stream this protocols too.
+ External ports for this services are stored in consul KV at `vergilius/ports/%service_name%`.
+ You can configure external ports range with `PROXY_PORTS` env, for ex.: `5000-6000`. 
+ It's strongly recommended to use vergilius in `net=host` mode or disable `userland-proxy`,
+  because docker will create as much userland proxies as `PROXY_PORTS` you have.
+
 #### how http2 works
 
 To use `http2` proxy, use `http2` tag instead of `http` or use both. Vergilius will try to acquire certificate from

circle.yml

@@ -20,9 +20,14 @@ test:
   pre:
     - docker run -d -p 8500:8500 e96tech/consul-server -advertise 127.0.0.1 -bootstrap -dc circle -domain local
     - curl --retry 10 --retry-delay 5 -v http://localhost:8500
+    - sed -i '1i load_module "modules/ngx_stream_module.so";' /home/ubuntu/vergilius/src/vergilius/templates/service_validate.html
 
 deployment:
-  dockerhub:
+  dockerhub_master:
     branch: master
     commands:
-      - 'curl -H "Content-Type: application/json" --data "{\"source_type\": \"Branch\", \"source_name\": \"master\"}" -X POST https://registry.hub.docker.com/u/devopsftw/vergilius/trigger/ea3f932c-49b9-47e8-af0c-ec1d8615cda4/'
+      - 'curl -H "Content-Type: application/json" --data "{\"source_type\": \"Branch\", \"source_name\": \"master\"}" -X POST https://registry.hub.docker.com/u/devopsftw/vergilius/trigger/ea3f932c-49b9-47e8-af0c-ec1d8615cda4/'
+  dockerhub_tag:
+    tag: /.*/
+    commands:
+      - 'curl -H "Content-Type: application/json" --data "{\"source_type\": \"Tag\", \"source_name\": \"$CIRCLE_TAG\"}" -X POST https://registry.hub.docker.com/u/devopsftw/vergilius/trigger/ea3f932c-49b9-47e8-af0c-ec1d8615cda4/'

nginx/nginx.conf

@@ -41,6 +41,16 @@ http {
 
     ssl_dhparam /etc/nginx/dhparam/dhparam.pem;
 
-    include /etc/nginx/conf.d/*.conf;
+    include /etc/nginx/conf.d/*.upstream.conf;
+    include /etc/nginx/conf.d/*.http.conf;
+    include /etc/nginx/conf.d/*.http2.conf;
+    include /etc/nginx/conf.d/default.conf;
     include /etc/nginx/sites-enabled/*.conf;
 }
+
+stream {
+    include /etc/nginx/conf.d/*.upstream.conf;
+    include /etc/nginx/conf.d/*.tcp.conf;
+    include /etc/nginx/conf.d/*.udp.conf;
+}
+

services/nginx.sh

@@ -7,8 +7,8 @@ function die {
 
 if [ ! -f /etc/nginx/dhparam/dhparam.pem ]; then
     mkdir -p /etc/nginx/dhparam/
-    echo "dhparam file /etc/nginx/dhparam/dhparam.pem does not exist. Generating one with 4086 bit. This will take a while..."
-    openssl dhparam -out /etc/nginx/dhparam/dhparam.pem 4096 || die "Could not generate dhparam file"
+    echo "dhparam file /etc/nginx/dhparam/dhparam.pem does not exist. Generating one with $DHPARAM_LENGTH bit. This will take a while..."
+    openssl dhparam -out /etc/nginx/dhparam/dhparam.pem $DHPARAM_LENGTH || die "Could not generate dhparam file"
     echo "Finished. Starting nginx now..."
 fi
 

src/vergilius/components/port_allocator.py

@@ -0,0 +1,20 @@
+from vergilius.config import PROXY_PORTS
+
+allocated = set()
+
+
+def allocate():
+    min_port = PROXY_PORTS[0]
+    max_port = PROXY_PORTS[1]
+
+    while min_port < max_port:
+        if min_port not in allocated:
+            allocated.add(min_port)
+            return min_port
+        min_port += 1
+
+    raise Exception('Failed to allocate port')
+
+
+def release(port):
+    allocated.discard(int(port))

src/vergilius/config.py

@@ -8,6 +8,7 @@
 NGINX_BINARY = os.environ.get('NGINX_BINARY', '/usr/sbin/nginx')
 NGINX_HTTP_PORT = os.environ.get('NGINX_HTTP_PORT', 80)
 NGINX_HTTP2_PORT = os.environ.get('NGINX_HTTP2_PORT', 443)
+PROXY_PORTS = [int(s) for s in os.environ.get('PROXY_PORTS', '7000-8000').split('-')]
 
 ACME_DIRECTORY_URL = os.environ.get('ACME_DIRECTORY_URL', 'https://acme-staging.api.letsencrypt.org/directory')
 

src/vergilius/loop/service_watcher.py

@@ -23,7 +23,8 @@ def watch_services(self):
 
     def check_services(self, data):
         # check if service has any of our tags
-        services_to_publish = dict((k, v) for k, v in data.items() if any(x in v for x in [u'http', u'http2']))
+        services_to_publish = dict(
+                (k, v) for k, v in data.items() if any(x in v for x in [u'http', u'http2', u'tcp', u'udp']))
         for service_name in services_to_publish:
             if service_name not in self.services:
                 vergilius.logger.info('[service watcher]: new service: %s' % service_name)

src/vergilius/models/service.py

@@ -4,8 +4,12 @@
 import tempfile
 import unicodedata
 
+import itertools
 from consul import tornado, base, ConsulException
+from shutil import rmtree
+
 from vergilius import config, consul_tornado, consul, logger, template_loader
+from vergilius.components import port_allocator
 from vergilius.loop.nginx_reloader import NginxReloader
 from vergilius.models.certificate import Certificate
 
@@ -20,9 +24,12 @@ def __init__(self, name):
         logger.info('[service][%s]: new and loading' % self.name)
         self.allow_crossdomain = False
         self.nodes = {}
-        self.domains = {
+        self.port = None
+        self.binds = {
             u'http': set(),
-            u'http2': set()
+            u'http2': set(),
+            u'tcp': set(),
+            u'udp': set()
         }
 
         self.active = True
@@ -35,15 +42,15 @@ def __init__(self, name):
         self.watch()
 
     def fetch(self):
-        index, data = consul.health.service(self.name, passing=True)
+        index, data = consul.health.service(self.id, passing=True)
         self.parse_data(data)
 
     @tornado.gen.coroutine
     def watch(self):
         index = None
         while True and self.active:
             try:
-                index, data = yield consul_tornado.health.service(self.name, index, wait=None, passing=True)
+                index, data = yield consul_tornado.health.service(self.id, index, wait=None, passing=True)
                 self.parse_data(data)
             except ConsulException as e:
                 logger.error('consul exception: %s' % e)
@@ -55,18 +62,18 @@ def parse_data(self, data):
 
         :type data: set[]
         """
-        for protocol in self.domains.iterkeys():
-            self.domains[protocol].clear()
+        for protocol in self.binds.iterkeys():
+            self.binds[protocol].clear()
 
         allow_crossdomain = False
         self.nodes = {}
         for node in data:
             if not node[u'Service'][u'Port']:
-                logger.warn('[service][%s]: Node %s is ignored due no ServicePort' % (self.id, node[u'Node']))
+                logger.warn('[service][%s]: Node %s is ignored due no Service Port' % (self.id, node[u'Node'][u'Node']))
                 continue
 
             if node[u'Service'][u'Tags'] is None:
-                logger.warn('[service][%s]: Node %s is ignored due no ServiceTags' % (self.id, node[u'Node']))
+                logger.warn('[service][%s]: Node %s is ignored due no Service Tags' % (self.id, node[u'Node'][u'Node']))
                 continue
 
             self.nodes[node['Node']['Node']] = {
@@ -80,76 +87,97 @@ def parse_data(self, data):
 
             for protocol in [u'http', u'http2']:
                 if protocol in node[u'Service'][u'Tags']:
-                    self.domains[protocol].update(
+                    self.binds[protocol].update(
                             tag.replace(protocol + ':', '') for tag in node[u'Service'][u'Tags'] if
                             tag.startswith(protocol + ':')
                     )
 
+            for protocol in ['tcp', 'udp']:
+                self.binds[protocol].update({node[u'Service'][u'Port']})
+
         self.allow_crossdomain = allow_crossdomain
 
         self.flush_nginx_config()
 
-    def get_nginx_config(self):
+    def get_nginx_config(self, config_type):
         """
         Generate nginx config from service attributes
+        :param config_type: string
         """
-        if self.domains[u'http2']:
+        if config_type == 'http2' and len(self.binds['http2']):
             self.check_certificate()
-        return template_loader.load('service.html').generate(service=self, config=config)
+
+        if config_type in ['tcp', 'udp']:
+            self.check_port()
+
+        return template_loader.load('service_%s.html' % config_type).generate(service=self, config=config)
 
     def flush_nginx_config(self):
         if not self.validate():
             logger.error('[service][%s]: failed to validate nginx config!' % self.id)
             return False
 
-        nginx_config = self.get_nginx_config()
-        deployed_nginx_config = None
+        has_changes = False
 
-        try:
-            deployed_nginx_config = self.read_nginx_config_file()
-        except IOError:
-            pass
+        for config_type in self.get_config_types():
+            nginx_config = self.get_nginx_config(config_type)
+            deployed_nginx_config = None
 
-        if deployed_nginx_config != nginx_config:
-            config_file = open(self.get_nginx_config_path(), 'w+')
-            config_file.write(nginx_config)
-            config_file.close()
-            logger.info('[service][%s]: got new nginx config %s' % (self.name, self.get_nginx_config_path()))
+            try:
+                deployed_nginx_config = self.read_nginx_config_file(config_type)
+            except IOError:
+                pass
+
+            if deployed_nginx_config != nginx_config:
+                config_file = open(self.get_nginx_config_path(config_type), 'w+')
+                config_file.write(nginx_config)
+                config_file.close()
+                has_changes = True
+
+        if has_changes:
             NginxReloader.queue_reload()
+            logger.info('[service][%s]: got new nginx config' % self.name)
 
-    def get_nginx_config_path(self):
-        return os.path.join(config.NGINX_CONFIG_PATH, self.id + '.conf')
+    def get_nginx_config_path(self, config_type):
+        return os.path.join(config.NGINX_CONFIG_PATH, '%s.%s.conf' % (self.id, config_type))
 
-    def read_nginx_config_file(self):
-        with open(self.get_nginx_config_path(), 'r') as config_file:
+    def read_nginx_config_file(self, config_type):
+        with open(self.get_nginx_config_path(config_type), 'r') as config_file:
             config_content = config_file.read()
             config_file.close()
             return config_content
 
+    def get_config_types(self):
+        return itertools.chain(self.binds.keys(), ['upstream'])
+
     def validate(self):
         """
         Deploy temporary service & nginx config and validate it with nginx
         :return: bool
         """
-        service_config_file = tempfile.NamedTemporaryFile(delete=False)
-        service_config_file.write(self.get_nginx_config())
-        service_config_file.close()
-
-        nginx_config_file = tempfile.NamedTemporaryFile(delete=False)
-        nginx_config_file.write(template_loader.load('service_validate.html')
-                                .generate(service_config=service_config_file.name,
-                                          pid_file='%s.pid' % service_config_file.name)
-                                )
+
+        temp_dir = tempfile.mkdtemp()
+
+        files = {}
+        for config_type in self.get_config_types():
+            path = os.path.join(temp_dir, config_type)
+            config_file = open(path, 'w+')
+            config_file.write(self.get_nginx_config(config_type))
+            config_file.close()
+            files['service_%s' % config_type] = path
+
+        files['pid_file'] = os.path.join(temp_dir, 'pid')
+
+        nginx_config_file = open(os.path.join(temp_dir, 'service'), 'w+')
+        nginx_config_file.write(template_loader.load('service_validate.html').generate(**files))
         nginx_config_file.close()
 
         try:
             return_code = subprocess.check_call([config.NGINX_BINARY, '-t', '-c', nginx_config_file.name])
         except subprocess.CalledProcessError:
             return_code = 1
         finally:
-            os.unlink(service_config_file.name)
-            os.unlink('%s.pid' % service_config_file.name)
-            os.unlink(nginx_config_file.name)
+            rmtree(temp_dir, ignore_errors=True)
 
         return return_code == 0
 
@@ -161,10 +189,14 @@ def delete(self):
         logger.info('[service][%s]: deleting' % self.name)
         self.active = False
 
-        try:
-            os.remove(self.get_nginx_config_path())
-        except OSError:
-            pass
+        if self.port:
+            self.release_port()
+
+        for config_type in self.get_config_types():
+            try:
+                os.remove(self.get_nginx_config_path(config_type))
+            except OSError:
+                pass
 
     def __del__(self):
         if self.active:
@@ -182,4 +214,14 @@ def slugify(cls, string):
 
     def check_certificate(self):
         if not self.certificate:
-            self.certificate = Certificate(service=self, domains=self.domains[u'http2'])
+            self.certificate = Certificate(service=self, domains=self.binds['http2'])
+
+    def check_port(self):
+        if not self.port:
+            self.port = port_allocator.allocate()
+            consul.kv.put('vergilius/ports/%s' % self.name, str(self.port))
+
+    def release_port(self):
+        if self.port:
+            port_allocator.release(self.port)
+            consul.kv.delete('vergilius/ports/%s' % self.name)