Merge pull request #38 from messiaen/merge-apps-by-label

Merge apps by label

README.md

@@ -216,6 +216,25 @@ Alias for [time.Now](https://golang.org/pkg/time/#Now)
 # Generated by nixy {{datetime}}
 ```
 
+#### MergeAppsByLabel
+
+Sometimes it is useful to implement the same service with apps within marathon.
+In this case you can use `.MergeAppsByLabel("some-label")` instead of `.Apps` in your template
+to merge multiple apps into a single service.
+
+For example if the following apps running in marathon implement the same API:
+
+`my-service-a`
+`my-service-b`
+
+And they both have the label `servicename=my-service` you could use `MergeAppsByLabel("servicename")` 
+to access both implementations as `nixy.marathon.mesos:12345`.
+
+- When the apps are merged, `Labels` from each original app are added
+  to each Task from that original app. This is necessary if you want
+  to have implementation specific labels.  For example, if one
+  implementation is faster, we could route more traffic there.
+
 ### TCP/UDP Load Balancing / Proxy
 
 It is possible to use Nixy to configure nginx as a proxy for TCP or UDP traffic.

nginx-merge-app-by-id.tmpl

@@ -0,0 +1,40 @@
+# Generated by nixy {{datetime}}
+user www-data;
+worker_processes auto;
+
+pid /var/run/nginx.pid
+
+events {
+    use epoll;
+    worker_connections 2048;
+    multi_accept on;
+}
+
+stream {
+    {{- range $appid, $app := .MergeAppsByLabel "streamservicename"}}
+    {{- if eq (index $app.Labels "internal") "stream"}}
+    {{- range $id, $definition := $app.PortDefinitions}}
+    {{- if ne (index $app.Labels "streamservicename") ""}}
+    upstream {{ (index $app.Labels "streamservicename") }}-{{ $id }} {
+    {{- else}}
+    upstream {{index $app.Hosts 0}}-{{ $id }} {
+    {{- end}}
+        least_conn;
+        {{- range $task := $app.Tasks}}
+        server {{ $task.Host }}:{{ index $task.Ports $id}}{{- with index $task.Labels "weight"}} weight={{ .  }}{{- end}};
+        {{- end}}
+    }
+    server {
+        {{- if eq $definition.Protocol "tcp" }}
+        listen {{ $definition.Port }};
+        {{- else}}
+        listen {{ $definition.Port }} {{ $definition.Protocol }};
+        {{- end}}
+        {{- if ne (index $app.Labels "streamservicename") ""}}
+        proxy_pass {{ (index $app.Labels "streamservicename") }}-{{ $id }};
+        {{- else}}
+        proxy_pass {{index $app.Hosts 0}}-{{ $id }};
+        {{- end}}
+    }
+}
+

nixy.go

@@ -30,6 +30,7 @@ type Task struct {
 	StartedAt    string
 	State        string
 	Version      string
+	Labels       map[string]string
 }
 
 // PortDefinitions struct
@@ -141,6 +142,63 @@ var eventqueue = make(chan bool, 2)
 // Global http transport for connection reuse
 var tr = &http.Transport{MaxIdleConnsPerHost: 10}
 
+func (c *Config) MergeAppsByLabel(label string) map[string]App {
+	apps := make(map[string]App, 0)
+	labeledApps := make(map[string][]App, 0)
+	for appID, app := range c.Apps {
+		if labelValue, has := app.Labels[label]; has {
+			labeledApps[labelValue] = append(labeledApps[labelValue], app)
+		} else {
+			apps[appID] = app
+		}
+	}
+
+	for id, appGroup := range labeledApps {
+		apps[id] = mergeApps(appGroup)
+	}
+	return apps
+}
+
+func mergeApps(apps []App) App {
+	tasks := make([]Task, 0)
+	labels := make(map[string]string, 0)
+	env := make(map[string]string, 0)
+	hosts := make([]string, 0)
+	portDefs := make([]PortDefinitions, 0)
+	seenPorts := make(map[int64]bool, 0)
+
+	for _, app := range apps {
+		for k, v := range app.Labels {
+			labels[k] = v
+		}
+		for k, v := range app.Env {
+			env[k] = v
+		}
+		for _, h := range app.Hosts {
+			hosts = append(hosts, h)
+		}
+		for _, t := range app.Tasks {
+			t.Labels = app.Labels
+			tasks = append(tasks, t)
+		}
+		for _, def := range app.PortDefinitions {
+			if _, seen := seenPorts[def.Port]; !seen {
+				seenPorts[def.Port] = true
+				portDefs = append(portDefs, def)
+			}
+		}
+	}
+	return App{
+		Tasks:           tasks,
+		Labels:          labels,
+		Env:             env,
+		Hosts:           hosts,
+		PortDefinitions: portDefs,
+		HealthChecks:    apps[0].HealthChecks,
+		Container:       Container{},
+	}
+}
+
 func newHealth() Health {
 	var h Health
 	for _, ep := range config.Marathon {