Use InactivityTimeout to work around a bug in docker (#204)

* Use InactivityTimeout to work around a bug in docker.

Log rotation causes 'docker logs --follow' to stop outputing logs
(moby/moby#23913).

We work around this by detecting inactivity in a log stream and
restarting its pump. We restart the pump with argument
Since=Now-InactivityTimeout to catch any logs we may have "missed" due
to log rotation (this is harmless for truly silent log streams).

* gofmt -w router/pump.go

* Add mention of INACTIVITY_TIMEOUT flag to README.

README.md

@@ -109,6 +109,10 @@ Routes are stored on disk, so by default routes are ephemeral. You can mount a v
 
 See [routesapi module](http://github.com/gliderlabs/logspout/blob/master/routesapi) for all options.
 
+#### Detecting timeouts in Docker log streams
+
+Logspout relies on the Docker API to retrieve container logs. A failure in the API may cause a log stream to hang. Logspout can detect and restart inactive Docker log streams. Use the environment variable `INACTIVITY_TIMEOUT` to enable this feature. E.g.: `INACTIVITY_TIMEOUT=1m` for a 1-minute threshold.
+
 ## Modules
 
 The standard distribution of logspout comes with all modules defined in this repository. You can remove or add new modules with custom builds of logspout. Just edit the `modules.go` file and do a `docker build`.

router/pump.go

@@ -76,6 +76,12 @@ func ignoreContainer(container *docker.Container) bool {
 	return false
 }
 
+func getInactivityTimeoutFromEnv() time.Duration {
+	inactivityTimeout, err := time.ParseDuration(getopt("INACTIVITY_TIMEOUT", "0"))
+	assert(err, "Couldn't parse env var INACTIVITY_TIMEOUT. See https://golang.org/pkg/time/#ParseDuration for valid format.")
+	return inactivityTimeout
+}
+
 type update struct {
 	*docker.APIEvents
 	pump *containerPump
@@ -112,6 +118,9 @@ func (p *LogsPump) rename(event *docker.APIEvents) {
 }
 
 func (p *LogsPump) Run() error {
+	inactivityTimeout := getInactivityTimeoutFromEnv()
+	debug("pump.Run(): using inactivity timeout: ", inactivityTimeout)
+
 	containers, err := p.client.ListContainers(docker.ListContainersOptions{})
 	if err != nil {
 		return err
@@ -120,7 +129,7 @@ func (p *LogsPump) Run() error {
 		p.pumpLogs(&docker.APIEvents{
 			ID:     normalID(listing.ID),
 			Status: "start",
-		}, false)
+		}, false, inactivityTimeout)
 	}
 	events := make(chan *docker.APIEvents)
 	err = p.client.AddEventListener(events)
@@ -131,7 +140,7 @@ func (p *LogsPump) Run() error {
 		debug("pump.Run() event:", normalID(event.ID), event.Status)
 		switch event.Status {
 		case "start", "restart":
-			go p.pumpLogs(event, true)
+			go p.pumpLogs(event, true, inactivityTimeout)
 		case "rename":
 			go p.rename(event)
 		case "die":
@@ -141,7 +150,7 @@ func (p *LogsPump) Run() error {
 	return errors.New("docker event stream closed")
 }
 
-func (p *LogsPump) pumpLogs(event *docker.APIEvents, backlog bool) {
+func (p *LogsPump) pumpLogs(event *docker.APIEvents, backlog bool, inactivityTimeout time.Duration) {
 	id := normalID(event.ID)
 	container, err := p.client.InspectContainer(id)
 	assert(err, "pump")
@@ -180,14 +189,15 @@ func (p *LogsPump) pumpLogs(event *docker.APIEvents, backlog bool) {
 		for {
 			debug("pump.pumpLogs():", id, "started")
 			err := p.client.Logs(docker.LogsOptions{
-				Container:    id,
-				OutputStream: outwr,
-				ErrorStream:  errwr,
-				Stdout:       true,
-				Stderr:       true,
-				Follow:       true,
-				Tail:         "all",
-				Since:        sinceTime.Unix(),
+				Container:         id,
+				OutputStream:      outwr,
+				ErrorStream:       errwr,
+				Stdout:            true,
+				Stderr:            true,
+				Follow:            true,
+				Tail:              "all",
+				Since:             sinceTime.Unix(),
+				InactivityTimeout: inactivityTimeout,
 			})
 			if err != nil {
 				debug("pump.pumpLogs():", id, "stopped with error:", err)
@@ -196,6 +206,9 @@ func (p *LogsPump) pumpLogs(event *docker.APIEvents, backlog bool) {
 			}
 
 			sinceTime = time.Now()
+			if err == docker.ErrInactivityTimeout {
+				sinceTime = sinceTime.Add(-inactivityTimeout)
+			}
 
 			container, err := p.client.InspectContainer(id)
 			if err != nil {