Use log timestamps from docker logs.

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file.
 - @gbolo added option to harden the TLS client
 - @chopmann added option to bind the http server to an address
 - @ibrokethecloud added ability to add custom key:value pairs as EXCLUDE_LABEL.
+- @localghost added option to read original logs timestamps from containers
 
 ### Removed
 

README.md

@@ -187,6 +187,7 @@ If you use multiline logging with raw, it's recommended to json encode the Data
 * `MULTILINE_PATTERN` - pattern for multiline logging, see: [MULTILINE_MATCH](#multiline_match) (default: `^\s`)
 * `MULTILINE_FLUSH_AFTER` - maximum time between the first and last lines of a multiline log entry in milliseconds (default: 500)
 * `MULTILINE_SEPARATOR` - separator between lines for output (default: `\n`)
+* `ORIGINAL_TIMESTAMPS` - read original log timestamps from containers (default: generate timestamps as the logs are processed by logspout)
 
 #### Raw Format
 

router/pump.go

@@ -33,6 +33,10 @@ func getopt(name, dfault string) string {
 	return value
 }
 
+func originalTimestamps() bool {
+	return getopt("ORIGINAL_TIMESTAMPS", "false") == "true"
+}
+
 func debug(v ...interface{}) {
 	if os.Getenv("DEBUG") != "" {
 		log.Println(v...)
@@ -244,6 +248,7 @@ func (p *LogsPump) pumpLogs(event *docker.APIEvents, backlog bool, inactivityTim
 				Since:             sinceTime.Unix(),
 				InactivityTimeout: inactivityTimeout,
 				RawTerminal:       rawTerminal,
+				Timestamps:        originalTimestamps(),
 			})
 			if err != nil {
 				debug("pump.pumpLogs():", id, "stopped with error:", err)
@@ -370,10 +375,11 @@ func newContainerPump(container *docker.Container, stdout, stderr io.Reader) *co
 				}
 				return
 			}
+			logMessage, logTime := parseLogLine(line, originalTimestamps())
 			cp.send(&Message{
-				Data:      strings.TrimSuffix(line, "\n"),
+				Data:      logMessage,
 				Container: container,
-				Time:      time.Now(),
+				Time:      logTime,
 				Source:    source,
 			})
 		}
@@ -405,3 +411,22 @@ func (cp *containerPump) remove(logstream chan *Message) {
 	defer cp.Unlock()
 	delete(cp.logstreams, logstream)
 }
+
+func parseLogLine(line string, originalTimestamps bool) (string, time.Time) {
+	line = strings.TrimSuffix(line, "\n")
+
+	if ! originalTimestamps {
+		return line, time.Now()
+	}
+
+	logEntry := strings.SplitN(line, " ", 2)
+	logTime, err := time.Parse(time.RFC3339Nano, logEntry[0])
+	if err != nil {
+		return line, time.Now()
+	}
+
+	if len(logEntry) == 2 {
+		return logEntry[1], logTime
+	}
+	return "", logTime
+}