Merge pull request #1992 from bacongobbler/logspout

feat(logspout): 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),
@@ -309,16 +291,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
+TimeoutStartSec=20m
+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 -e DEBUG=1 $IMAGE"
+
+[Install]
+WantedBy=multi-user.target
+
+[X-Fleet]
+Global=true

deisctl/units/deis-publisher.service

@@ -9,7 +9,6 @@ TimeoutStartSec=20m
 ExecStartPre=/bin/sh -c "IMAGE=`/run/deis/bin/get_image /deis/publisher`; docker history $IMAGE >/dev/null || docker pull $IMAGE"
 ExecStartPre=/bin/sh -c "docker inspect deis-publisher >/dev/null && docker rm -f deis-publisher || true"
 ExecStart=/bin/sh -c "IMAGE=`/run/deis/bin/get_image /deis/publisher` && docker run --name deis-publisher --rm -e HOST=$COREOS_PRIVATE_IPV4 -e ETCD_HOST=$COREOS_PRIVATE_IPV4 -v /var/run/docker.sock:/var/run/docker.sock $IMAGE"
-ExecStopPost=/usr/bin/docker stop deis-publisher
 
 [Install]
 WantedBy=multi-user.target

logger/syslog/filehandler.go

@@ -19,7 +19,7 @@ type FileHandler struct {
 
 // NewFileHandler accepts all arguments expected by NewBaseHandler plus
 // filename which is the path to the log file.
-func NewFileHandler(filename string, qlen int, filter func(*Message) bool,
+func NewFileHandler(filename string, qlen int, filter func(SyslogMessage) bool,
 	ft bool) *FileHandler {
 
 	h := &FileHandler{
@@ -61,7 +61,7 @@ func (h *FileHandler) mainLoop() {
 	}
 }
 
-func (h *FileHandler) saveMessage(m *Message) {
+func (h *FileHandler) saveMessage(m SyslogMessage) {
 	var err error
 	if h.f == nil {
 		h.f, err = os.OpenFile(
@@ -90,6 +90,6 @@ func (h *FileHandler) checkErr(err error) bool {
 }
 
 // Handle queues and dispatches a message. See BaseHandler.Handle
-func (h *FileHandler) Handle(m *Message) *Message {
+func (h *FileHandler) Handle(m SyslogMessage) SyslogMessage {
 	return h.bh.Handle(m)
 }

logger/syslog/filehandler_test.go

@@ -2,7 +2,6 @@ package syslog
 
 import (
 	"testing"
-	"time"
 )
 
 type TestLogger struct{}
@@ -12,24 +11,24 @@ func (tl TestLogger) Printf(format string, v ...interface{}) {}
 func (tl TestLogger) Println(...interface{})                 {}
 
 func TestNewFileHandler(t *testing.T) {
-	fh := NewFileHandler("", 1, func(m *Message) bool { return true }, true)
+	fh := NewFileHandler("", 1, func(m SyslogMessage) bool { return true }, true)
 	if fh == nil {
 		t.Errorf("expected filehandler, got nil")
 	}
 }
 
 func TestSetLogger(t *testing.T) {
 	tl := TestLogger{}
-	fh := NewFileHandler("", 1, func(m *Message) bool { return true }, true)
+	fh := NewFileHandler("", 1, func(m SyslogMessage) bool { return true }, true)
 	fh.SetLogger(tl)
 	if fh.l != tl {
 		t.Errorf("expected the logger to be set")
 	}
 }
 
 func TestHandle(t *testing.T) {
-	fh := NewFileHandler("/tmp/test", 1, func(m *Message) bool { return true }, true)
-	handle := fh.Handle(&Message{time.Now(), nil, 0, 0, time.Now(), "localhost", "test", "message", "", ""})
+	fh := NewFileHandler("/tmp/test", 1, func(m SyslogMessage) bool { return true }, true)
+	handle := fh.Handle(&Message{"localhost test message"})
 	if handle == nil {
 		t.Errorf("expected a handle, got nil")
 	}

logger/syslog/handler.go

@@ -5,26 +5,26 @@ type Handler interface {
 	// Handle should return Message (maybe modified) for future processing by
 	// other handlers or return nil. If Handle is called with nil message it
 	// should complete all remaining work and properly shutdown before return.
-	Handle(*Message) *Message
+	Handle(SyslogMessage) SyslogMessage
 }
 
 // BaseHandler is designed to simplify the creation of real handlers. It
 // implements Handler interface using nonblocking queuing of messages and
 // simple message filtering.
 type BaseHandler struct {
-	queue  chan *Message
+	queue  chan SyslogMessage
 	end    chan struct{}
-	filter func(*Message) bool
+	filter func(SyslogMessage) bool
 	ft     bool
 }
 
 // NewBaseHandler creates BaseHandler using a specified filter. If filter is nil
 // or if it returns true messages are passed to BaseHandler internal queue
 // (of qlen length). If filter returns false or ft is true messages are returned
 // to server for future processing by other handlers.
-func NewBaseHandler(qlen int, filter func(*Message) bool, ft bool) *BaseHandler {
+func NewBaseHandler(qlen int, filter func(SyslogMessage) bool, ft bool) *BaseHandler {
 	return &BaseHandler{
-		queue:  make(chan *Message, qlen),
+		queue:  make(chan SyslogMessage, qlen),
 		end:    make(chan struct{}),
 		filter: filter,
 		ft:     ft,
@@ -34,7 +34,7 @@ func NewBaseHandler(qlen int, filter func(*Message) bool, ft bool) *BaseHandler
 // Handle inserts m in an internal queue. It immediately returns even if
 // queue is full. If m == nil it closes queue and waits for End method call
 // before return.
-func (h *BaseHandler) Handle(m *Message) *Message {
+func (h *BaseHandler) Handle(m SyslogMessage) SyslogMessage {
 	if m == nil {
 		close(h.queue) // signal that there is no more messages for processing
 		<-h.end        // wait for handler shutdown
@@ -58,7 +58,7 @@ func (h *BaseHandler) Handle(m *Message) *Message {
 // Get returns first message from internal queue. It waits for message if queue
 // is empty. It returns nil if there is no more messages to process and handler
 // should shutdown.
-func (h *BaseHandler) Get() *Message {
+func (h *BaseHandler) Get() SyslogMessage {
 	m, ok := <-h.queue
 	if ok {
 		return m
@@ -70,7 +70,7 @@ func (h *BaseHandler) Get() *Message {
 // it directly, especially if your handler needs to select from multiple channels
 // or have to work without blocking. You need to check if this channel is closed by
 // sender and properly shutdown in this case.
-func (h *BaseHandler) Queue() <-chan *Message {
+func (h *BaseHandler) Queue() <-chan SyslogMessage {
 	return h.queue
 }
 

logger/syslog/message.go

@@ -2,47 +2,18 @@ package syslog
 
 import (
 	"fmt"
-	"net"
-	"time"
+	"strings"
 )
 
-// Message defines an RFC 3164 syslog message.
-type Message struct {
-	Time      time.Time // time the message was logged
-	Source    net.Addr  // source address of the log message
-	Facility            // facility tag (see type Facility)
-	Severity            // severity tag (see type Severity)
-	Timestamp time.Time // optional
-	Hostname  string    // optional
-	Tag       string    // message tag as defined in RFC 3164
-	Content   string    // message content as defined in RFC 3164
-	Tag1      string    // alternate message tag (white rune as separator)
-	Content1  string    // alternate message content (white rune as separator)
+type SyslogMessage interface {
+	fmt.Stringer
 }
 
-// NetSrc only network part of Source as string (IP for UDP or Name for UDS)
-func (m *Message) NetSrc() string {
-	switch a := m.Source.(type) {
-	case *net.UDPAddr:
-		return a.IP.String()
-	case *net.UnixAddr:
-		return a.Name
-	case *net.TCPAddr:
-		return a.IP.String()
-	}
-	// Unknown type
-	return m.Source.String()
+// Message defines a syslog message.
+type Message struct {
+	Msg string
 }
 
-// String returns the Message in a string format. This satisfies the fmt.Stringer
-// interface.
 func (m *Message) String() string {
-	timeLayout := "2006-01-02 15:04:05"
-	return fmt.Sprintf(
-		"%s %s %s%s",
-		m.Time.Format(timeLayout),
-		m.Hostname,
-		m.Tag,
-		m.Content,
-	)
+	return strings.TrimSuffix(m.Msg, "\n")
 }