influx: ExtractTimestamp and ExtractNanos

It turns out that ExtractTimestamp only works for - and is optimised
for - nanosecond timestamps. It's now been renamed to ExtractNanos and
a new ExtractTimestamp has been added which handles timestamps in any
precision. It's slightly slower but safe for all timestamp precisions.

filter/worker.go

@@ -160,7 +160,7 @@ func extractTimestamp(line []byte, defaultTs int64) int64 {
 		return defaultTs
 	}
 
-	out, _ := influx.ExtractTimestamp(line)
+	out, _ := influx.ExtractNanos(line)
 	if out == -1 {
 		return defaultTs
 	}

influx/timestamps.go

@@ -50,8 +50,8 @@ const (
 )
 
 var (
-	// maxTsLen is maximum number of characters a valid timestamp can be.
-	maxTsLen = len(fmt.Sprint(maxNanoTime))
+	// MaxTsLen is maximum number of characters a valid timestamp can be.
+	MaxTsLen = len(fmt.Sprint(maxNanoTime))
 
 	// ErrTimeOutOfRange gets returned when time is out of the representable range using int64 nanoseconds since the epoch.
 	ErrTimeOutOfRange = fmt.Errorf("time outside range %d - %d", minNanoTime, maxNanoTime)
@@ -62,6 +62,40 @@ var (
 // offset of -1 is returned.
 func ExtractTimestamp(line []byte) (int64, int) {
 	length := len(line)
+	if length < 6 {
+		return -1, -1
+	}
+
+	// Remove trailing newline, if present.
+	if line[length-1] == '\n' {
+		length--
+		line = line[:length]
+	}
+
+	to := length - MaxTsLen - 1
+	if to < 0 {
+		to = 0
+	}
+	for i := length - 1; i >= to; i-- {
+		if line[i] == ' ' {
+			out, err := convert.ToInt(line[i+1:])
+			if err != nil {
+				return -1, -1
+			}
+			return out, i + 1
+		}
+	}
+
+	return -1, -1
+}
+
+// ExtractNanos returns the value and offset of the timestamp in a
+// InfluxDB line protocol line. It is optimised for - and only works
+// for - timestamps in nanosecond precision. Use ExtractTimestamp() if
+// timestamps in other precisions may be present. If no valid
+// timestamp is present, an offset of -1 is returned.
+func ExtractNanos(line []byte) (int64, int) {
+	length := len(line)
 
 	if length < 6 {
 		return -1, -1
@@ -74,7 +108,7 @@ func ExtractTimestamp(line []byte) (int64, int) {
 	}
 
 	// Expect a space just before the timestamp.
-	from := length - maxTsLen - 1
+	from := length - MaxTsLen - 1
 	if from < 0 {
 		from = 0
 	}

influx/timestamps_small_test.go

@@ -27,7 +27,7 @@ import (
 )
 
 func TestExtractTimestamp(t *testing.T) {
-	ts := time.Date(1997, 6, 5, 4, 3, 2, 1, time.UTC).UnixNano()
+	ts := int64(12345)
 	tsStr := strconv.FormatInt(ts, 10)
 
 	check := func(input string, expectedTs int64, expectedOffset int) {
@@ -53,6 +53,42 @@ func TestExtractTimestamp(t *testing.T) {
 	check("weather,city=paris temp=60,humidity=100 "+tsStr, ts, 40)
 	check("weather,city=paris temp=60,humidity=100 "+tsStr+"\n", ts, 40)
 
+	// Various invalid timestamps
+	noTimestamp("weather temp=99 " + tsStr + " ")       // trailing whitespace
+	noTimestamp("weather temp=99 xxxxx")                // not digits
+	noTimestamp("weather temp=99 15x07")                // embedded non-digit
+	noTimestamp("weather temp=99 00000000000000000001") // too long
+	noTimestamp("weather temp=99 -" + tsStr)            // negative
+	noTimestamp(tsStr)                                  // timestamp only
+}
+
+func TestExtractNanos(t *testing.T) {
+	ts := time.Date(1997, 6, 5, 4, 3, 2, 1, time.UTC).UnixNano()
+	tsStr := strconv.FormatInt(ts, 10)
+
+	check := func(input string, expectedTs int64, expectedOffset int) {
+		ts, offset := influx.ExtractNanos([]byte(input))
+		assert.Equal(t, expectedTs, ts, "ExtractNanos(%q)", input)
+		assert.Equal(t, expectedOffset, offset, "ExtractNanos(%q)", input)
+	}
+
+	noTimestamp := func(input string) {
+		ts, offset := influx.ExtractNanos([]byte(input))
+		assert.Equal(t, -1, offset, "ExtractNanos(%q)", input)
+		assert.Equal(t, int64(-1), ts, "ExtractNanos(%q)", input)
+	}
+
+	noTimestamp("")
+	noTimestamp(" ")
+	noTimestamp("weather temp=99")
+	noTimestamp("weather,city=paris temp=60")
+	noTimestamp("weather,city=paris temp=99,humidity=100")
+	check("weather temp=99 "+tsStr, ts, 16)
+	check("weather temp=99 "+tsStr+"\n", ts, 16)
+	check("weather,city=paris temp=60 "+tsStr, ts, 27)
+	check("weather,city=paris temp=60,humidity=100 "+tsStr, ts, 40)
+	check("weather,city=paris temp=60,humidity=100 "+tsStr+"\n", ts, 40)
+
 	// Various invalid timestamps
 	noTimestamp("weather temp=99 " + tsStr + " ")       // trailing whitespace
 	noTimestamp("weather temp=99 xxxxxxxxxxxxxxxxxxx")  // not digits