Merge pull request #8 from ibuildthecloud/master

Ensure that polling doesn't skip initial rows

Author: ibuildthecloud

pkg/client/client.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -14,6 +15,10 @@ type Value struct {
 	Modified int64
 }
 
+var (
+	ErrNotFound = errors.New("etcdwrapper: key not found")
+)
+
 type Client interface {
 	Get(ctx context.Context, key string) (Value, error)
 	Put(ctx context.Context, key string, value []byte) error
@@ -59,7 +64,7 @@ func (c *client) Get(ctx context.Context, key string) (Value, error) {
 		}, nil
 	}
 
-	return Value{}, nil
+	return Value{}, ErrNotFound
 }
 
 func (c *client) Put(ctx context.Context, key string, value []byte) error {

pkg/drivers/generic/generic.go

@@ -199,9 +199,9 @@ func (d *Generic) execute(ctx context.Context, sql string, args ...interface{})
 	wait := strategy.Backoff(backoff.Linear(100 + time.Millisecond))
 	for i := uint(0); i < 20; i++ {
 		if i > 2 {
-			logrus.Infof("EXEC (%d) %v : %s", i, args, Stripped(sql))
+			logrus.Debugf("EXEC (try: %d) %v : %s", i, args, Stripped(sql))
 		} else {
-			logrus.Tracef("EXEC (%d) %v : %s", i, args, Stripped(sql))
+			logrus.Tracef("EXEC (try: %d) %v : %s", i, args, Stripped(sql))
 		}
 		result, err = d.DB.ExecContext(ctx, sql, args...)
 		if err != nil && d.Retry != nil && d.Retry(err) {

pkg/logstructured/sqllog/sql.go

@@ -41,10 +41,9 @@ type Dialect interface {
 	IsFill(key string) bool
 }
 
-func (s *SQLLog) Start(ctx context.Context) error {
+func (s *SQLLog) Start(ctx context.Context) (err error) {
 	s.ctx = ctx
-	go s.compact()
-	return nil
+	return
 }
 
 func (s *SQLLog) compact() {
@@ -266,14 +265,22 @@ func filter(events interface{}, checkPrefix bool, prefix string) ([]*server.Even
 }
 
 func (s *SQLLog) startWatch() (chan interface{}, error) {
+	pollStart, err := s.d.GetCompactRevision(s.ctx)
+	if err != nil {
+		return nil, err
+	}
+
 	c := make(chan interface{})
-	go s.poll(c)
+	// start compaction and polling at the same time to watch starts
+	// at the oldest revision, but compaction doesn't create gaps
+	go s.compact()
+	go s.poll(c, pollStart)
 	return c, nil
 }
 
-func (s *SQLLog) poll(result chan interface{}) {
+func (s *SQLLog) poll(result chan interface{}, pollStart int64) {
 	var (
-		last     int64
+		last     = pollStart
 		skip     int64
 		skipTime time.Time
 	)
@@ -293,15 +300,6 @@ func (s *SQLLog) poll(result chan interface{}) {
 		case <-wait.C:
 		}
 
-		if last == 0 {
-			if currentRev, err := s.CurrentRevision(s.ctx); err != nil {
-				logrus.Errorf("failed to find current revision: %v", err)
-				continue
-			} else {
-				last = currentRev
-			}
-		}
-
 		rows, err := s.d.After(s.ctx, "%", last)
 		if err != nil {
 			logrus.Errorf("fail to list latest changes: %v", err)