Merge pull request #3660 from kalbasit/issue_3657

fix(logger): avoid hitting etcd for each log line

Author: Matthew Fisher

logger/drain/drain.go

@@ -6,26 +6,8 @@ import (
 	"net"
 	"net/url"
 	"os"
-
-	"github.com/coreos/go-etcd/etcd"
 )
 
-func GetDrain() string {
-	host := getopt("HOST", "127.0.0.1")
-
-	etcdPort := getopt("ETCD_PORT", "4001")
-	etcdPath := getopt("ETCD_PATH", "/deis/logs")
-
-	client := etcd.NewClient([]string{"http://" + host + ":" + etcdPort})
-
-	s, err := client.Get(etcdPath+"/drain", true, false)
-	if err != nil {
-		return ""
-	}
-
-	return s.Node.Value
-}
-
 func SendToDrain(m string, drain string) error {
 	u, err := url.Parse(drain)
 	if err != nil {

logger/main.go

@@ -18,6 +18,7 @@ import (
 var (
 	logAddr         string
 	logPort         int
+	drainURI        string
 	enablePublish   bool
 	publishHost     string
 	publishPath     string
@@ -29,6 +30,7 @@ var (
 func init() {
 	flag.StringVar(&logAddr, "log-addr", "0.0.0.0", "bind address for the logger")
 	flag.IntVar(&logPort, "log-port", 514, "bind port for the logger")
+	flag.StringVar(&drainURI, "drain-uri", "", "default drainURI, once set in etcd, this has no effect.")
 	flag.StringVar(&syslogd.LogRoot, "log-root", "/data/logs", "log path to store logs")
 	flag.BoolVar(&enablePublish, "enable-publish", false, "enable publishing to service discovery")
 	flag.StringVar(&publishHost, "publish-host", getopt("HOST", "127.0.0.1"), "service discovery hostname")
@@ -43,26 +45,49 @@ func main() {
 
 	client := etcd.NewClient([]string{"http://" + publishHost + ":" + publishPort})
 
-	// Wait for terminating signal
-	exitChan := make(chan os.Signal, 2)
+	signalChan := make(chan os.Signal, 1)
+	drainRespChan := make(chan *etcd.Response)
+	drainChan := make(chan string)
+	stopChan := make(chan bool)
+	exitChan := make(chan bool)
 	cleanupChan := make(chan bool)
-	signal.Notify(exitChan, syscall.SIGTERM, syscall.SIGINT)
+	signal.Notify(signalChan, syscall.SIGTERM, syscall.SIGINT)
 
-	go syslogd.Listen(exitChan, cleanupChan, fmt.Sprintf("%s:%d", logAddr, logPort))
+	// ensure the drain key exists in etcd.
+	if _, err := client.Get(publishPath+"/drain", false, false); err != nil {
+		setEtcd(client, publishPath+"/drain", drainURI, 0)
+	}
 
+	go client.Watch(publishPath+"/drain", 0, false, drainRespChan, stopChan)
+	go syslogd.Listen(exitChan, cleanupChan, drainChan, fmt.Sprintf("%s:%d", logAddr, logPort))
 	if enablePublish {
-		go publishService(client, publishHost, publishPath, strconv.Itoa(logPort), uint64(time.Duration(publishTTL).Seconds()))
+		go publishService(exitChan, client, publishHost, publishPath, strconv.Itoa(logPort), uint64(time.Duration(publishTTL).Seconds()))
 	}
 
-	// Wait for the proper shutdown of the syslog server before exit
-	<-cleanupChan
+	for {
+		select {
+		case er := <-drainRespChan:
+			drainChan <- er.Node.Value
+		case <-signalChan:
+			close(exitChan)
+			stopChan <- true
+		case <-cleanupChan:
+			return
+		}
+	}
 }
 
-func publishService(client *etcd.Client, host string, etcdPath string, port string, ttl uint64) {
+func publishService(exitChan chan bool, client *etcd.Client, host string, etcdPath string, port string, ttl uint64) {
+	t := time.NewTicker(time.Duration(publishInterval))
+
 	for {
-		setEtcd(client, etcdPath+"/host", host, ttl)
-		setEtcd(client, etcdPath+"/port", port, ttl)
-		time.Sleep(time.Duration(publishInterval))
+		select {
+		case <-t.C:
+			setEtcd(client, etcdPath+"/host", host, ttl)
+			setEtcd(client, etcdPath+"/port", port, ttl)
+		case <-exitChan:
+			return
+		}
 	}
 }
 

logger/syslogd/syslogd.go

@@ -13,12 +13,14 @@ import (
 	"github.com/deis/deis/logger/drain"
 )
 
+// LogRoot is the log path to store logs.
 var LogRoot string
 
 type handler struct {
 	// To simplify implementation of our handler we embed helper
 	// syslog.BaseHandler struct.
 	*syslog.BaseHandler
+	drainURI string
 }
 
 // Simple fiter for named/bind messages which can be used with BaseHandler
@@ -27,8 +29,11 @@ func filter(m syslog.SyslogMessage) bool {
 }
 
 func newHandler() *handler {
-	h := handler{syslog.NewBaseHandler(5, filter, false)}
-	go h.mainLoop() // BaseHandler needs some gorutine that reads from its queue
+	h := handler{
+		BaseHandler: syslog.NewBaseHandler(5, filter, false),
+	}
+
+	go h.mainLoop() // BaseHandler needs some goroutine that reads from its queue
 	return &h
 }
 
@@ -84,9 +89,8 @@ func (h *handler) mainLoop() {
 		if m == nil {
 			break
 		}
-		d := drain.GetDrain()
-		if d != "" {
-			drain.SendToDrain(m.String(), d)
+		if h.drainURI != "" {
+			drain.SendToDrain(m.String(), h.drainURI)
 		}
 		err := writeToDisk(m)
 		if err != nil {
@@ -97,7 +101,7 @@ func (h *handler) mainLoop() {
 }
 
 // Listen starts a new syslog server which runs until it receives a signal.
-func Listen(signalChan chan os.Signal, cleanupDone chan bool, bindAddr string) {
+func Listen(exitChan, cleanupDone chan bool, drainChan chan string, bindAddr string) {
 	fmt.Println("Starting syslog...")
 	// If LogRoot doesn't exist, create it
 	// equivalent to Python's `if not os.path.exists(filename)`
@@ -106,18 +110,24 @@ func Listen(signalChan chan os.Signal, cleanupDone chan bool, bindAddr string) {
 			log.Fatalf("unable to create LogRoot at %s: %v", LogRoot, err)
 		}
 	}
-	// Create a server with one handler and run one listen gorutine
+	// Create a server with one handler and run one listen goroutine
 	s := syslog.NewServer()
-	s.AddHandler(newHandler())
+	h := newHandler()
+	s.AddHandler(h)
 	s.Listen(bindAddr)
 	fmt.Println("Syslog server started...")
 	fmt.Println("deis-logger running")
 
 	// Wait for terminating signal
-	for _ = range signalChan {
-		// Shutdown the server
-		fmt.Println("Shutting down...")
-		s.Shutdown()
-		cleanupDone <- true
+	for {
+		select {
+		case <-exitChan:
+			// Shutdown the server
+			fmt.Println("Shutting down...")
+			s.Shutdown()
+			cleanupDone <- true
+		case d := <-drainChan:
+			h.drainURI = d
+		}
 	}
 }