fix(weblog): fix up CPU issues

The way we spawned and checked to ensure the weblog server was running was a bit backwards,
causing the logger to run with massive CPU usage after a few hours because of the state loop.
Changing the server logic to instead defer to net/http for serving requests and shutting down
the connection at the net.Listener reduces CPU usage significantly.

Additionally the Redis adapter had a similar issue; creating connections, writing every second
then closing the connection. Re-using the connection will reduce CPU cycle usage significantly.

Author: Matthew Fisher

main.go

@@ -2,6 +2,9 @@ package main
 
 import (
 	l "log"
+	"net/http"
+
+	_ "net/http/pprof"
 
 	"github.com/deis/logger/log"
 	"github.com/deis/logger/storage"
@@ -29,16 +32,19 @@ func main() {
 	if err != nil {
 		l.Fatal("Error starting log aggregator: ", err)
 	}
+	defer aggregator.Stop()
 	l.Println("Log aggregator running")
 
-	weblogServer, err := weblog.NewServer(storageAdapter)
-	if err != nil {
-		l.Fatal("Error creating weblog server: ", err)
-	}
-	serverErrCh := weblogServer.Listen()
-	l.Println("Weblog server running")
+	weblogServer := weblog.NewServer(storageAdapter)
+	weblogServer.Start()
+	defer weblogServer.Close()
+	l.Printf("Weblog server serving at %s\n", weblogServer.URL)
+
+	// start a Go Profiler
+	go func() {
+		l.Println(http.ListenAndServe("0.0.0.0:8099", nil))
+	}()
 
-	defer aggregator.Stop()
 	stoppedCh := aggregator.Stopped()
 	select {
 	case stopErr := <-stoppedCh:
@@ -47,7 +53,5 @@ func main() {
 		} else {
 			l.Fatal("Log aggregator has stopped with no error")
 		}
-	case serverErr := <-serverErrCh:
-		l.Fatal("Weblog server failed: ", serverErr)
 	}
 }

storage/redis_adapter.go

@@ -54,12 +54,9 @@ func (mp messagePipeliner) execPipeline() {
 			mp.errCh <- fmt.Errorf("Error adding ltrim of %s to the pipeline: %s", app, err)
 		}
 	}
-	go func() {
-		defer mp.pipeline.Close()
-		if _, err := mp.pipeline.Exec(); err != nil {
-			mp.errCh <- fmt.Errorf("Error executing pipeline: %s", err)
-		}
-	}()
+	if _, err := mp.pipeline.Exec(); err != nil {
+		mp.errCh <- fmt.Errorf("Error executing pipeline: %s", err)
+	}
 }
 
 type redisAdapter struct {
@@ -105,29 +102,20 @@ func (a *redisAdapter) Start() {
 		errCh := make(chan error)
 		mp := newMessagePipeliner(a.bufferSize, a.redisClient, a.config.PipelineTimeout, errCh)
 		go func() {
+			defer mp.pipeline.Close()
 			for {
 				select {
 				case err := <-errCh:
 					log.Println(err)
 				case <-a.stopCh:
 					return
-				}
-			}
-		}()
-		go func() {
-			for {
-				select {
 				case message := <-a.messageChannel:
 					mp.addMessage(message)
 					if mp.messageCount == a.config.PipelineLength {
-						mp.execPipeline()
-						mp = newMessagePipeliner(a.bufferSize, a.redisClient, a.config.PipelineTimeout, errCh)
+						go mp.execPipeline()
 					}
 				case <-mp.timeoutTicker.C:
-					mp.execPipeline()
-					mp = newMessagePipeliner(a.bufferSize, a.redisClient, a.config.PipelineTimeout, errCh)
-				case <-a.stopCh:
-					return
+					go mp.execPipeline()
 				}
 			}
 		}()

weblog/router.go

@@ -1,20 +1,10 @@
 package weblog
 
 import (
-	"log"
-	"net/http"
-
-	_ "net/http/pprof"
-
 	"github.com/gorilla/mux"
 )
 
 func newRouter(rh *requestHandler) *mux.Router {
-
-	go func() {
-		log.Println(http.ListenAndServe("0.0.0.0:8099", nil))
-	}()
-
 	r := mux.NewRouter()
 	r.HandleFunc("/healthz", rh.getHealthz).Methods("GET")
 	r.HandleFunc("/healthz/", rh.getHealthz).Methods("GET")

weblog/server.go

@@ -2,49 +2,65 @@ package weblog
 
 import (
 	"fmt"
-	"log"
+	"net"
 	"net/http"
 
-	"github.com/gorilla/mux"
-
 	"github.com/deis/logger/storage"
 )
 
 const (
-	bindHost = "0.0.0.0"
-	bindPort = 8088
+	bindAddr = "0.0.0.0:8088"
 )
 
-// Server implements a simple HTTP server that handles GET and DELETE requests for application
-// logs.  These actions are accomplished by delegating to a storage.Adapter.
+// Server implements an HTTP server.
 type Server struct {
-	listening bool
-	router    *mux.Router
-	errCh     chan error
+	// The Listener used for incoming HTTP connections
+	Listener net.Listener
+	// Server may be changed before calling Start()
+	Server *http.Server
+	// base URL of form http://ipaddr:port with no trailing slash
+	URL string
+	// started defines whether the server has started or not.
+	started bool
 }
 
-// NewServer returns a pointer to a new Server instance.
-func NewServer(storageAdapter storage.Adapter) (*Server, error) {
-	return &Server{
-		router: newRouter(newRequestHandler(storageAdapter)),
-	}, nil
+// New returns a new HTTP Server. The caller should call Start to start it and Close when finished
+// to shut it down.
+func NewServer(storageAdapter storage.Adapter) *Server {
+	s := &Server{
+		Listener: defaultListener(),
+		Server:   &http.Server{Handler: newRouter(newRequestHandler(storageAdapter))},
+	}
+	return s
 }
 
-// Listen starts the server's main loop.
-func (s *Server) Listen() <-chan error {
-	// Should only ever be called once
-	if !s.listening {
-		s.listening = true
-		go func() {
-			s.errCh <- s.listen()
-		}()
-		log.Printf("weblog server running on %s:%d", bindHost, bindPort)
+// Start starts an HTTP server.
+func (s *Server) Start() {
+	if s.started {
+		panic("weblog: server already started")
+	}
+	if s.URL == "" {
+		s.URL = "http://" + s.Listener.Addr().String()
 	}
-	return s.errCh
+	go func() {
+		s.Server.Serve(s.Listener)
+	}()
+	s.started = true
 }
 
-func (s *Server) listen() error {
-	mux := http.NewServeMux()
-	mux.Handle("/", s.router)
-	return http.ListenAndServe(fmt.Sprintf("%s:%d", bindHost, bindPort), mux)
+// Close closes the HTTP Server from listening for the inbound requests.
+func (s *Server) Close() {
+	s.Server.SetKeepAlivesEnabled(false)
+	s.Listener.Close()
+	s.started = false
+}
+
+// defaultListener provides a net.Listener on bindAddr, panicking if it cannot listen on that
+// address.
+func defaultListener() net.Listener {
+	l, err := net.Listen("tcp", bindAddr)
+	if err != nil {
+		panic(fmt.Sprintf("weblog: failed to listen on %v: %v", bindAddr, err))
+	}
+	return l
 }

weblog/server_test.go

@@ -0,0 +1,114 @@
+package weblog
+
+import (
+	"bufio"
+	"fmt"
+	"net"
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/deis/logger/storage"
+)
+
+// TODO(bacongobbler): stop relying that port 6666 is not in use
+var testBindAddr string = "127.0.0.1:6666"
+
+// testListener provides a net.Listener for testing, panicking if it cannot listen on that
+// address.
+func newTestListener(t *testing.T) net.Listener {
+	l, err := net.Listen("tcp", testBindAddr)
+	if err != nil {
+		t.Fatalf("failed to listen on %s: %v", testBindAddr, err)
+	}
+	return l
+}
+
+func newTestStorageAdapter(t *testing.T) storage.Adapter {
+	storageAdapter, err := storage.NewAdapter("memory", 1)
+	if err != nil {
+		t.Fatalf("Error creating storage adapter: %v", err)
+	}
+	return storageAdapter
+}
+
+func TestServerStart(t *testing.T) {
+	storageAdapter := newTestStorageAdapter(t)
+	storageAdapter.Start()
+	defer storageAdapter.Stop()
+
+	s := &Server{
+		Listener: newTestListener(t),
+		Server:   &http.Server{Handler: newRouter(newRequestHandler(storageAdapter))},
+	}
+
+	s.Start()
+
+	conn, err := net.Dial("tcp", testBindAddr)
+	if err != nil {
+		t.Fatalf("could not connect to test server: %v", err)
+	}
+	defer conn.Close()
+	fmt.Fprintf(conn, "GET /healthz HTTP/1.0\r\n\r\n")
+	status, err := bufio.NewReader(conn).ReadString('\n')
+	if err != nil {
+		t.Errorf("there was an error reading from the response: %v", err)
+	}
+	if !strings.Contains(status, "200 OK") {
+		t.Errorf("Did not receive 200 OK, got '%s'", status)
+	}
+
+	// explicitly close the connection so other tests can run
+	s.Close()
+}
+
+func TestServerClose(t *testing.T) {
+	storageAdapter := newTestStorageAdapter(t)
+	storageAdapter.Start()
+	defer storageAdapter.Stop()
+
+	s := &Server{
+		Listener: newTestListener(t),
+		Server:   &http.Server{Handler: newRouter(newRequestHandler(storageAdapter))},
+	}
+
+	s.Start()
+	s.Close()
+
+	// try reading from the server, expecting it to fail
+	_, err := net.Dial("tcp", testBindAddr)
+	if err == nil {
+		t.Error("server returned nil. Calling s.Close() did not close the server connection!")
+	}
+}
+
+func TestServerURL(t *testing.T) {
+	storageAdapter := newTestStorageAdapter(t)
+	storageAdapter.Start()
+	defer storageAdapter.Stop()
+
+	s := &Server{
+		Listener: newTestListener(t),
+		Server:   &http.Server{Handler: newRouter(newRequestHandler(storageAdapter))},
+		URL:      "foo",
+	}
+
+	s.Start()
+
+	if s.URL != "foo" {
+		t.Errorf("URL is not 'foo', got '%s'", s.URL)
+	}
+
+	s.Close()
+
+	s.URL = ""
+
+	s.Start()
+
+	if s.URL != "http://" + testBindAddr {
+		t.Errorf("URL is not 'http://%s', got '%s'", testBindAddr, s.URL)
+	}
+
+	// explicitly close the connection so other tests can run
+	s.Close()
+}