feat(controller): introduce deis/logspout

Author: Matthew Fisher

Makefile

@@ -4,8 +4,8 @@
 
 include includes.mk
 
-COMPONENTS=builder cache controller database logger publisher registry router
-START_ORDER=publisher logger database cache registry controller builder router
+COMPONENTS=builder cache controller database logger logspout publisher registry router
+START_ORDER=publisher logger logspout database cache registry controller builder router
 CLIENTS=client deisctl
 
 all: build run

controller/scheduler/coreos.py

@@ -1,7 +1,6 @@
 import cStringIO
 import base64
 import copy
-import functools
 import json
 import httplib
 import paramiko
@@ -105,7 +104,6 @@ def create(self, name, image, command='', template=None, **kwargs):
         """Create a container"""
         self._create_container(name, image, command,
                                template or copy.deepcopy(CONTAINER_TEMPLATE), **kwargs)
-        self._create_log(name, image, command, copy.deepcopy(LOG_TEMPLATE))
 
     def _create_container(self, name, image, command, unit, **kwargs):
         l = locals().copy()
@@ -138,15 +136,6 @@ def _create_container(self, name, image, command, unit, **kwargs):
         # post unit to fleet
         self._put_unit(name, {"desiredState": "launched", "options": unit})
 
-    def _create_log(self, name, image, command, unit):
-        l = locals().copy()
-        l.update(re.match(MATCH, name).groupdict())
-        # construct unit from template
-        for f in unit:
-            f['value'] = f['value'].format(**l)
-        # post unit to fleet
-        self._put_unit(name+'-log', {"desiredState": "launched", "options": unit})
-
     def start(self, name):
         """Start a container"""
         self._wait_for_container(name)
@@ -181,23 +170,16 @@ def stop(self, name):
 
     def destroy(self, name):
         """Destroy a container"""
-        funcs = []
-        funcs.append(functools.partial(self._destroy_container, name))
-        funcs.append(functools.partial(self._destroy_log, name))
         # call all destroy functions, ignoring any errors
-        for f in funcs:
-            try:
-                f()
-            except:
-                pass
+        try:
+            self._destroy_container(name)
+        except:
+            pass
         self._wait_for_destroy(name)
 
     def _destroy_container(self, name):
         return self._delete_unit(name)
 
-    def _destroy_log(self, name):
-        return self._delete_unit(name+'-log')
-
     def run(self, name, image, entrypoint, command):  # noqa
         """Run a one-off command"""
         self._create_container(name, image, command, copy.deepcopy(RUN_TEMPLATE),
@@ -307,16 +289,6 @@ def attach(self, name):
 ]
 
 
-LOG_TEMPLATE = [
-    {"section": "Unit", "name": "Description", "value": "{name} log"},
-    {"section": "Unit", "name": "BindsTo", "value": "{name}.service"},
-    {"section": "Service", "name": "ExecStartPre", "value": '''/bin/sh -c "until docker inspect {name} >/dev/null 2>&1; do sleep 1; done"'''},  # noqa
-    {"section": "Service", "name": "ExecStart", "value": '''/bin/sh -c "docker logs -f {name} 2>&1 | logger -p local0.info -t {app}[{c_type}.{c_num}] --udp --server $(etcdctl get /deis/logs/host) --port $(etcdctl get /deis/logs/port)"'''},  # noqa
-    {"section": "Service", "name": "TimeoutStartSec", "value": "20m"},
-    {"section": "X-Fleet", "name": "MachineOf", "value": "{name}.service"},
-]
-
-
 RUN_TEMPLATE = [
     {"section": "Unit", "name": "Description", "value": "{name} admin command"},
     {"section": "Service", "name": "ExecStartPre", "value": '''/bin/sh -c "IMAGE=$(etcdctl get /deis/registry/host 2>&1):$(etcdctl get /deis/registry/port 2>&1)/{image}; docker pull $IMAGE"'''},  # noqa

deisctl/cmd/cmd.go

@@ -92,7 +92,7 @@ func startDefaultServices(b backend.Backend) error {
 	if err := Start(b, []string{"logger@1"}); err != nil {
 		return err
 	}
-	if err := Start(b, []string{"publisher", "cache@1", "router@1", "database@1", "controller@1", "registry@1", "builder@1"}); err != nil {
+	if err := Start(b, []string{"publisher", "logspout", "cache@1", "router@1", "database@1", "controller@1", "registry@1", "builder@1"}); err != nil {
 		return err
 	}
 	fmt.Println("Service containers launched.")
@@ -118,7 +118,7 @@ func StopPlatform(b backend.Backend) error {
 
 func stopDefaultServices(b backend.Backend) error {
 	fmt.Println("Stopping service containers...")
-	if err := Stop(b, []string{"publisher", "builder@1", "registry@1", "controller@1", "database@1", "cache@1", "router@1", "logger@1"}); err != nil {
+	if err := Stop(b, []string{"publisher", "logspout", "builder@1", "registry@1", "controller@1", "database@1", "cache@1", "router@1", "logger@1"}); err != nil {
 		return err
 	}
 	fmt.Println("Service containers stopped.")
@@ -196,7 +196,7 @@ func installDefaultServices(b backend.Backend) error {
 	if err := Scale(b, targets); err != nil {
 		return err
 	}
-	if err := b.Create([]string{"publisher"}); err != nil {
+	if err := b.Create([]string{"publisher", "logspout"}); err != nil {
 		return err
 	}
 	fmt.Println("Service containers scheduled.")
@@ -226,7 +226,7 @@ func uninstallAllServices(b backend.Backend) error {
 	if err := Scale(b, targets); err != nil {
 		return err
 	}
-	if err := b.Destroy([]string{"publisher"}); err != nil {
+	if err := b.Destroy([]string{"publisher", "logspout"}); err != nil {
 		return err
 	}
 	fmt.Println("Service containers destroyed.")
@@ -309,6 +309,7 @@ Options:
 		"deis-database-data.service",
 		"deis-logger.service",
 		"deis-logger-data.service",
+		"deis-logspout.service",
 		"deis-publisher.service",
 		"deis-registry.service",
 		"deis-registry-data.service",

deisctl/units/deis-logspout.service

@@ -0,0 +1,17 @@
+[Unit]
+Description=deis-logspout
+Requires=docker.socket etcd.service
+After=docker.socket etcd.service
+
+[Service]
+EnvironmentFile=/etc/environment
+ExecStartPre=/bin/sh -c "IMAGE=`/run/deis/bin/get_image /deis/logspout`; docker history $IMAGE >/dev/null || docker pull $IMAGE"
+ExecStartPre=/bin/sh -c "docker inspect deis-logspout >/dev/null && docker rm -f deis-logspout || true"
+ExecStart=/bin/sh -c "IMAGE=`/run/deis/bin/get_image /deis/logspout` && docker run --name deis-logspout --rm -v /var/run/docker.sock:/tmp/docker.sock -e ETCD_HOST=$COREOS_PRIVATE_IPV4 -e HOST=$COREOS_PRIVATE_IPV4 $IMAGE"
+ExecStopPost=/usr/bin/docker stop deis-logspout
+
+[Install]
+WantedBy=multi-user.target
+
+[X-Fleet]
+Global=true

logspout/Dockerfile

@@ -1,13 +1,6 @@
-FROM flynn/busybox
-MAINTAINER Jeff Lindsay <progrium@gmail.com>
+FROM deis/go:latest
+MAINTAINER OpDemand <info@opdemand.com>
 
-ADD ./stage/logspout /bin/logspout
-
-ENV DOCKER_HOST unix:///tmp/docker.sock
-ENV ROUTESPATH /mnt/routes
-VOLUME /mnt/routes
-
-EXPOSE 8000
-
-ENTRYPOINT ["/bin/logspout"]
-CMD []
+WORKDIR /go/src/github.com/deis/deis/logspout
+ADD . /go/src/github.com/deis/deis/logspout
+RUN go get .

logspout/Makefile

@@ -1,18 +1,67 @@
-build/container: stage/logspout Dockerfile
-	docker build --no-cache -t logspout .
-	touch build/container
+include ../includes.mk
 
-build/logspout: *.go
-	GOOS=linux GOARCH=amd64 go build -o build/logspout
+COMPONENT := logspout
+DOCKER_IMAGE := deis/$(COMPONENT)
+BUILD_IMAGE := $(DOCKER_IMAGE)-build
+RELEASE_IMAGE := $(DOCKER_IMAGE):$(BUILD_TAG)
+DEV_DOCKER_IMAGE := $(DEV_REGISTRY)/$(RELEASE_IMAGE)
 
-stage/logspout: build/logspout
-	mkdir -p stage
-	cp build/logspout stage/logspout
+build: check-docker
+	docker build -t $(BUILD_IMAGE) .
+	docker cp `docker run -d $(BUILD_IMAGE)`:/go/bin/logspout image/
+	docker build -t $(RELEASE_IMAGE) image
+	rm -rf image/logspout
+
+clean: check-docker check-registry
+	docker rmi $(RELEASE_IMAGE) $(BUILD_IMAGE)
+
+full-clean: check-docker check-registry
+	docker images -q $(DOCKER_IMAGE) | xargs docker rmi -f
+	docker images -q $(BUILD_IMAGE) | xargs docker rmi -f
+	docker images -q $(DEV_DOCKER_IMAGE) | xargs docker rmi -f
+
+install: check-deisctl
+	deisctl install $(COMPONENT)
+
+uninstall: check-deisctl
+	deisctl uninstall $(COMPONENT)
+
+start: check-deisctl
+	deisctl start $(COMPONENT)
+
+stop: check-deisctl
+	deisctl stop $(COMPONENT)
+
+restart: stop start
+
+run: install start
+
+dev-release: check-registry check-deisctl
+	docker tag $(RELEASE_IMAGE) $(DEV_DOCKER_IMAGE)
+	docker push $(DEV_DOCKER_IMAGE)
+	deisctl config $(COMPONENT) set image=$(DEV_DOCKER_IMAGE)
 
 release:
-	docker tag logspout progrium/logspout
-	docker push progrium/logspout
+	docker push $(RELEASE_IMAGE)
+
+deploy: build dev-release restart
+
+test: test-unit test-functional
+
+setup-root-gotools:
+	sudo GOPATH=/tmp/tmpGOPATH go get -u -v code.google.com/p/go.tools/cmd/cover
+	sudo GOPATH=/tmp/tmpGOPATH go get -u -v code.google.com/p/go.tools/cmd/vet
+	sudo rm -rf /tmp/tmpGOPATH
+
+setup-gotools:
+	go get -v github.com/golang/lint/golint
+
+test-style:
+	go vet -x ./...
+	-golint .
+
+test-unit: test-style
+	@echo no tests
 
-.PHONY: clean
-clean:
-	rm -rf build
+test-functional:
+	@echo no functional tests

logspout/attacher.go

@@ -80,7 +80,7 @@ func (m *AttachManager) attach(id string) {
 		m.Unlock()
 		success <- struct{}{}
 		m.send(&AttachEvent{ID: id, Name: name, Type: "attach"})
-		debug("attach:", id, "success")
+		debug("attach:", id, name, "success")
 		return
 	}
 	debug("attach:", id, "failure:", <-failure)

logspout/image/Dockerfile

@@ -0,0 +1,8 @@
+FROM flynn/busybox
+MAINTAINER OpDemand <info@opdemand.com>
+
+ENV DOCKER_HOST unix:///tmp/docker.sock
+ENV ROUTESPATH /tmp
+CMD ["/bin/logspout"]
+
+ADD logspout /bin/logspout

logspout/logspout.go

@@ -12,6 +12,7 @@ import (
 	"strings"
 
 	"code.google.com/p/go.net/websocket"
+	"github.com/coreos/go-etcd/etcd"
 	"github.com/fsouza/go-dockerclient"
 	"github.com/go-martini/martini"
 )
@@ -136,6 +137,21 @@ func main() {
 	attacher := NewAttachManager(client)
 	router := NewRouteManager(attacher)
 
+	// HACK: if we are connecting to etcd, get the logger's connection
+	// details from there
+	if etcdHost := os.Getenv("ETCD_HOST"); etcdHost != "" {
+		connectionString := []string{"http://" + etcdHost + ":4001"}
+		debug("etcd:", connectionString[0])
+		etcd := etcd.NewClient(connectionString)
+		hostResp, err := etcd.Get("/deis/logs/host", false, false)
+		assert(err, "url")
+		portResp, err := etcd.Get("/deis/logs/port", false, false)
+		assert(err, "url")
+		host := fmt.Sprintf("%s:%s", hostResp.Node.Value, portResp.Node.Value)
+		log.Println("routing all to " + host)
+		router.Add(&Route{Target: Target{Type: "syslog", Addr: host}})
+	}
+
 	if len(os.Args) > 1 {
 		u, err := url.Parse(os.Args[1])
 		assert(err, "url")