Store and use log offset

cmd/ohm/main.go

@@ -55,15 +55,17 @@ func main() {
 
 	// Start the producer
 	go func() {
+		var err error
 		var retryCount int
+		var lineID string
 
 		// TODO move to constants
 		clampMaxRetrySleep := 60 * time.Second
 		resetAfter := 10 * time.Minute
 
 		lastCrash := time.Now()
 		for {
-			err := nextdns.StreamLogs(logC)
+			lineID, err = nextdns.StreamLogs(logC, lineID)
 			logStreamerRestarts.Inc()
 			if time.Since(lastCrash) > resetAfter {
 				retryCount = 0

src/nextdns/api.go

@@ -19,9 +19,11 @@ const (
 )
 
 var (
-	StreamingLogLineRegex = regexp.MustCompile(`^data:\s+`)
-	APIKey                string
-	Profile               string
+	StreamingLogLineIDRegex = regexp.MustCompile(`^id:\s+`)
+	StreamingLogLineRegex   = regexp.MustCompile(`^data:\s+`)
+
+	APIKey  string
+	Profile string
 )
 
 func MakeUrl(pathParts ...string) string {
@@ -50,30 +52,42 @@ func GetLogs() (*http.Response, error) {
 	return http.DefaultClient.Do(req)
 }
 
-func StreamLogs(logC chan types.LogData) error {
+func StreamLogs(logC chan types.LogData, lastID string) (string, error) {
 	req, err := Get(MakeUrl("profiles", Profile, "logs", "stream"))
+	if lastID != "" {
+		log.Printf("streaming from id %s onwards", lastID)
+		q := req.URL.Query()
+		q.Add("id", lastID)
+		req.URL.RawQuery = q.Encode()
+	}
+
 	if err != nil {
-		return err
+		return lastID, err
 	}
 
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		return err
+		return lastID, err
 	}
 
 	reader := bufio.NewReader(resp.Body)
+	var data []byte
 	// TODO cancel via context
 	for {
 		line, err := reader.ReadBytes('\n')
 		if err != nil {
-			return err
+			return lastID, err
 		}
-		prefix := StreamingLogLineRegex.FindSubmatchIndex(line)
-		// could be blank line or metadata
-		if len(prefix) == 0 {
+
+		if idPrefix := StreamingLogLineIDRegex.FindSubmatchIndex(line); len(idPrefix) > 0 {
+			lastID = strings.TrimSpace(string(line[idPrefix[1]:]))
+			continue
+		} else if prefix := StreamingLogLineRegex.FindSubmatchIndex(line); len(prefix) > 0 {
+			data = line[prefix[1]:]
+		} else {
+			// could be blank line or be other metadata
 			continue
 		}
-		data := line[prefix[1]:]
 
 		logData := types.LogData{}
 		err = json.Unmarshal(data, &logData)

src/ohm/states.go

@@ -30,7 +30,7 @@ func Ready(key string, _, _, _ time.Duration, logC chan types.LogData) Handler {
 }
 
 func Monitoring(key string, allowance, cooldown, lockout time.Duration, logC chan types.LogData) Handler {
-	log.Printf("monitoring: %s (cooldown: %s, lockout: %s)", key, cooldown, lockout)
+	log.Printf("monitoring: %s (allowance: %s, cooldown: %s, lockout: %s)", key, allowance, cooldown, lockout)
 	end := time.Now().Add(allowance)
 
 	// if a cooldown timer is provided, use that

src/types/types.go

@@ -6,6 +6,8 @@ type Status string
 
 const StatusBlocked Status = "blocked"
 
+type LogEntryID string
+
 type LogData struct {
 	Timestamp time.Time `json:"timestamp"`
 	Domain    string    `json:"domain"`