Support swarm service labels when using filter.labels

build.sh

@@ -6,7 +6,7 @@ cp -r /src /go/src/github.com/gliderlabs/logspout
 cd /go/src/github.com/gliderlabs/logspout
 export GOPATH=/go
 go get github.com/Masterminds/glide && $GOPATH/bin/glide install
-go build -ldflags "-X main.Version=$1" -o /bin/logspout
+go build -ldflags "-s -w -X main.Version=$1" -o /bin/logspout
 apk del go git mercurial build-base
 rm -rf /go /var/cache/apk/* /root/.glide
 

glide.yaml

@@ -2,6 +2,7 @@ package: github.com/gliderlabs/logspout
 excludeDirs:
 - custom
 import:
+- package: github.com/docker/docker/api/types/swarm
 - package: github.com/fsouza/go-dockerclient
 - package: github.com/gorilla/mux
 - package: golang.org/x/net

router/pump.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/docker/docker/api/types/swarm"
 	"github.com/fsouza/go-dockerclient"
 )
 
@@ -79,7 +80,7 @@ func logDriverSupported(container *docker.Container) bool {
 	}
 }
 
-func ignoreContainer(container *docker.Container) bool {
+func ignoreContainer(container *docker.Container, serviceLabels map[string]string) bool {
 	for _, kv := range container.Config.Env {
 		kvp := strings.SplitN(kv, "=", 2)
 		if len(kvp) == 2 && kvp[0] == "LOGSPOUT" && strings.ToLower(kvp[1]) == "ignore" {
@@ -96,7 +97,11 @@ func ignoreContainer(container *docker.Container) bool {
 		excludeLabel = excludeLabelArr[0]
 	}
 
-	if value, ok := container.Config.Labels[excludeLabel]; ok {
+	for k, v := range container.Config.Labels {
+		serviceLabels[k] = v
+	}
+
+	if value, ok := serviceLabels[excludeLabel]; ok {
 		return len(excludeLabel) > 0 && strings.ToLower(value) == strings.ToLower(excludeValue)
 	}
 	return false
@@ -190,12 +195,27 @@ func (p *LogsPump) Run() error {
 func (p *LogsPump) pumpLogs(event *docker.APIEvents, backlog bool, inactivityTimeout time.Duration) {
 	id := normalID(event.ID)
 	container, err := p.client.InspectContainer(id)
+	var service *swarm.Service
+	serviceLabels := make(map[string]string)
+	if err == nil {
+		serviceID := container.Config.Labels["com.docker.swarm.service.id"]
+		debug("Service ID: " + serviceID)
+		if serviceID != "" {
+			svc, err := p.client.InspectService(serviceID)
+			if err != nil {
+				debug("Error getting service", err)
+			} else {
+				service = svc
+				serviceLabels = service.Spec.Labels
+			}
+		}
+	}
 	assert(err, "pump")
 	if ignoreContainerTTY(container) {
 		debug("pump.pumpLogs():", id, "ignored: tty enabled")
 		return
 	}
-	if ignoreContainer(container) {
+	if ignoreContainer(container, serviceLabels) {
 		debug("pump.pumpLogs():", id, "ignored: environ ignore")
 		return
 	}
@@ -227,7 +247,7 @@ func (p *LogsPump) pumpLogs(event *docker.APIEvents, backlog bool, inactivityTim
 	}
 	outrd, outwr := io.Pipe()
 	errrd, errwr := io.Pipe()
-	p.pumps[id] = newContainerPump(container, outrd, errrd)
+	p.pumps[id] = newContainerPump(container, service, outrd, errrd)
 	p.mu.Unlock()
 	p.update(event)
 	go func() {
@@ -305,10 +325,15 @@ func (p *LogsPump) RoutingFrom(id string) bool {
 func (p *LogsPump) Route(route *Route, logstream chan *Message) {
 	p.mu.Lock()
 	for _, pump := range p.pumps {
+		var serviceLabels map[string]string
+		if pump.service != nil {
+			serviceLabels = pump.service.Spec.Labels
+		}
 		if route.MatchContainer(
 			normalID(pump.container.ID),
 			normalName(pump.container.Name),
-			pump.container.Config.Labels) {
+			pump.container.Config.Labels,
+			serviceLabels) {
 
 			pump.add(logstream, route)
 			defer pump.remove(logstream)
@@ -328,10 +353,15 @@ func (p *LogsPump) Route(route *Route, logstream chan *Message) {
 		case event := <-updates:
 			switch event.Status {
 			case "start", "restart":
+				var serviceLabels map[string]string
+				if event.pump.service != nil {
+					serviceLabels = event.pump.service.Spec.Labels
+				}
 				if route.MatchContainer(
 					normalID(event.pump.container.ID),
 					normalName(event.pump.container.Name),
-					event.pump.container.Config.Labels) {
+					event.pump.container.Config.Labels,
+					serviceLabels) {
 
 					event.pump.add(logstream, route)
 					defer event.pump.remove(logstream)
@@ -352,12 +382,14 @@ func (p *LogsPump) Route(route *Route, logstream chan *Message) {
 type containerPump struct {
 	sync.Mutex
 	container  *docker.Container
+	service    *swarm.Service
 	logstreams map[chan *Message]*Route
 }
 
-func newContainerPump(container *docker.Container, stdout, stderr io.Reader) *containerPump {
+func newContainerPump(container *docker.Container, service *swarm.Service, stdout, stderr io.Reader) *containerPump {
 	cp := &containerPump{
 		container:  container,
+		service:    service,
 		logstreams: make(map[chan *Message]*Route),
 	}
 	pump := func(source string, input io.Reader) {

router/pump_test.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"testing"
 
+	swarm "github.com/docker/docker/api/types/swarm"
 	docker "github.com/fsouza/go-dockerclient"
 )
 
@@ -67,11 +68,25 @@ func TestPumpIgnoreContainer(t *testing.T) {
 		{&docker.Config{Labels: map[string]string{"exclude": "false"}}, false},
 	}
 
+	serviceLabels := []struct {
+		in  map[string]string
+		out bool
+	}{
+		{map[string]string{"exclude": "true"}, true},
+		{map[string]string{"exclude": "false"}, false},
+	}
+
 	for _, conf := range containers {
-		if actual := ignoreContainer(&docker.Container{Config: conf.in}); actual != conf.out {
+		if actual := ignoreContainer(&docker.Container{Config: conf.in}, make(map[string]string)); actual != conf.out {
 			t.Errorf("expected %v got %v", conf.out, actual)
 		}
 	}
+
+	for _, serviceLabel := range serviceLabels {
+		if actual := ignoreContainer(&docker.Container{Config: &docker.Config{Labels: make(map[string]string)}}, serviceLabel.in); actual != serviceLabel.out {
+			t.Errorf("expected %v got %v", serviceLabel.out, actual)
+		}
+	}
 }
 
 func TestPumpIgnoreContainerCustomLabels(t *testing.T) {
@@ -85,11 +100,25 @@ func TestPumpIgnoreContainerCustomLabels(t *testing.T) {
 		{&docker.Config{Labels: map[string]string{"app": "demo-app"}}, false},
 	}
 
+	serviceLabels := []struct {
+		in  map[string]string
+		out bool
+	}{
+		{map[string]string{"k8s-app": "canal"}, true},
+		{map[string]string{"app": "demo-app"}, false},
+	}
+
 	for _, conf := range containers {
-		if actual := ignoreContainer(&docker.Container{Config: conf.in}); actual != conf.out {
+		if actual := ignoreContainer(&docker.Container{Config: conf.in}, make(map[string]string)); actual != conf.out {
 			t.Errorf("expected %v got %v", conf.out, actual)
 		}
 	}
+
+	for _, serviceLabel := range serviceLabels {
+		if actual := ignoreContainer(&docker.Container{Config: &docker.Config{Labels: make(map[string]string)}}, serviceLabel.in); actual != serviceLabel.out {
+			t.Errorf("expected %v got %v", serviceLabel.out, actual)
+		}
+	}
 }
 
 func TestPumpIgnoreContainerAllowTTYDefault(t *testing.T) {
@@ -139,6 +168,9 @@ func TestPumpContainerRename(t *testing.T) {
 		ID:   "8dfafdbc3a40",
 		Name: "bar",
 	}
+	service := &swarm.Service{
+		ID: "83nf93ndin2j",
+	}
 	client := newTestClient(&FakeRoundTripper{message: container, status: http.StatusOK})
 	p := &LogsPump{
 		client: &client,
@@ -153,7 +185,7 @@ func TestPumpContainerRename(t *testing.T) {
 		Name:   "foo",
 		Config: config,
 	}
-	p.pumps["8dfafdbc3a40"] = newContainerPump(container, os.Stdout, os.Stderr)
+	p.pumps["8dfafdbc3a40"] = newContainerPump(container, service, os.Stdout, os.Stderr)
 	if name := p.pumps["8dfafdbc3a40"].container.Name; name != "foo" {
 		t.Errorf("containerPump should have name: 'foo' got name: '%s'", name)
 	}
@@ -171,7 +203,10 @@ func TestPumpNewContainerPump(t *testing.T) {
 		ID:     "8dfafdbc3a40",
 		Config: config,
 	}
-	pump := newContainerPump(container, os.Stdout, os.Stderr)
+	service := &swarm.Service{
+		ID: "83nf93ndin2j",
+	}
+	pump := newContainerPump(container, service, os.Stdout, os.Stderr)
 	if pump == nil {
 		t.Error("pump nil")
 		return
@@ -186,7 +221,10 @@ func TestPumpContainerPump(t *testing.T) {
 		ID:     "8dfafdbc3a40",
 		Config: config,
 	}
-	pump := newContainerPump(container, os.Stdout, os.Stderr)
+	service := &swarm.Service{
+		ID: "83nf93ndin2j",
+	}
+	pump := newContainerPump(container, service, os.Stdout, os.Stderr)
 	logstream, route := make(chan *Message), &Route{}
 	go func() {
 		for msg := range logstream {