Merge pull request #1321 from erikwilson/upgrade-kine

go.mod

@@ -99,7 +99,7 @@ require (
 	github.com/rakelkar/gonetsh v0.0.0-20190719023240-501daadcadf8 // indirect
 	github.com/rancher/dynamiclistener v0.2.0
 	github.com/rancher/helm-controller v0.4.1-0.20191223195946-ae918063bc54
-	github.com/rancher/kine v0.3.2
+	github.com/rancher/kine v0.3.3
 	github.com/rancher/remotedialer v0.2.0
 	github.com/rancher/wrangler v0.4.0
 	github.com/rancher/wrangler-api v0.4.0

go.sum

@@ -717,8 +717,8 @@ github.com/rancher/flannel v0.11.0-k3s.1 h1:mIwnfWDafjzQgFkZeJ1AkFrrAT3EdBaA1giE
 github.com/rancher/flannel v0.11.0-k3s.1/go.mod h1:Hn4ZV+eq0LhLZP63xZnxdGwXEoRSxs5sxELxu27M3UA=
 github.com/rancher/helm-controller v0.4.1-0.20191223195946-ae918063bc54 h1:S/IAloMp6MYYtoHKfJUS5M3Wv0CU+5Ixe1nnSv2pQUU=
 github.com/rancher/helm-controller v0.4.1-0.20191223195946-ae918063bc54/go.mod h1:194LHuZRrxcD82bG1rJtOWsw98U4JbPhDWqvL7l3PAw=
-github.com/rancher/kine v0.3.2 h1:2kP48ojBWVoZ6vlzixc9jc9uKRk7Yn1a7kWoOsJi7Sg=
-github.com/rancher/kine v0.3.2/go.mod h1:xEMl0tLCva9/9me7mXJ3m9Vo6yqHgC4OU3NiK4CPrGQ=
+github.com/rancher/kine v0.3.3 h1:FTOQN1qHQMzMfA8/nd9/rWfdR3Xk4PbXcOqw5P3dJDw=
+github.com/rancher/kine v0.3.3/go.mod h1:xEMl0tLCva9/9me7mXJ3m9Vo6yqHgC4OU3NiK4CPrGQ=
 github.com/rancher/kubernetes v1.17.0-k3s.1 h1:g1xvTHOHMJxwWtseblor0gighLRHpL7Bf9MwX8HR3W0=
 github.com/rancher/kubernetes v1.17.0-k3s.1/go.mod h1:NbNV+69yL3eKiKDJ+ZEjqOplN3BFXKBeunzkoOy8WLo=
 github.com/rancher/kubernetes/staging/src/k8s.io/api v1.17.0-k3s.1 h1:L2mS7D+Kv/0ZUg9uJZcPfKuDCYcKOTprTQsK35i2hFg=

vendor/github.com/rancher/kine/pkg/drivers/sqlite/sqlite_nocgo.go

@@ -3,24 +3,24 @@
 package sqlite
 
 import (
-        "errors"
+	"context"
 	"database/sql"
+	"errors"
 
 	"github.com/rancher/kine/pkg/drivers/generic"
 	"github.com/rancher/kine/pkg/server"
-
 )
 
 var errNoCgo = errors.New("this binary is built without CGO, sqlite is disabled")
 
-func New(dataSourceName string) (server.Backend, error) {
-        return nil, errNoCgo
+func New(ctx context.Context, dataSourceName string) (server.Backend, error) {
+	return nil, errNoCgo
 }
 
 func NewVariant(driverName, dataSourceName string) (server.Backend, *generic.Generic, error) {
-        return nil, nil, errNoCgo
+	return nil, nil, errNoCgo
 }
 
 func setup(db *sql.DB) error {
-        return errNoCgo
+	return errNoCgo
 }

vendor/github.com/rancher/kine/pkg/logstructured/sqllog/sql.go

@@ -162,7 +162,11 @@ func (s *SQLLog) After(ctx context.Context, prefix string, revision, limit int64
 		return 0, nil, err
 	}
 
-	rev, _, result, err := RowsToEvents(rows)
+	rev, compact, result, err := RowsToEvents(rows)
+	if revision > 0 && revision < compact {
+		return rev, result, server.ErrCompacted
+	}
+
 	return rev, result, err
 }
 

vendor/github.com/rancher/kine/pkg/server/types.go

@@ -3,12 +3,12 @@ package server
 import (
 	"context"
 
-	"github.com/pkg/errors"
+	"go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
 )
 
 var (
-	ErrKeyExists = errors.New("key exists")
-	ErrCompacted = errors.New("revision has been compact")
+	ErrKeyExists = rpctypes.ErrGRPCDuplicateKey
+	ErrCompacted = rpctypes.ErrGRPCCompacted
 )
 
 type Backend interface {

vendor/github.com/rancher/kine/pkg/server/watch.go

@@ -29,9 +29,10 @@ func (s *KVServerBridge) Watch(ws etcdserverpb.Watch_WatchServer) error {
 		}
 
 		if msg.GetCreateRequest() != nil {
-			w.Start(msg.GetCreateRequest())
+			w.Start(ws.Context(), msg.GetCreateRequest())
 		} else if msg.GetCancelRequest() != nil {
-			w.Cancel(msg.GetCancelRequest().WatchId)
+			logrus.Debugf("WATCH CANCEL REQ id=%d", msg.GetCancelRequest().GetWatchId())
+			w.Cancel(msg.GetCancelRequest().WatchId, nil)
 		}
 	}
 }
@@ -45,11 +46,11 @@ type watcher struct {
 	watches map[int64]func()
 }
 
-func (w *watcher) Start(r *etcdserverpb.WatchCreateRequest) {
+func (w *watcher) Start(ctx context.Context, r *etcdserverpb.WatchCreateRequest) {
 	w.Lock()
 	defer w.Unlock()
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(ctx)
 
 	id := atomic.AddInt64(&watchID, 1)
 	w.watches[id] = cancel
@@ -57,7 +58,7 @@ func (w *watcher) Start(r *etcdserverpb.WatchCreateRequest) {
 
 	key := string(r.Key)
 
-	logrus.Debugf("WATCH START id=%d, key=%s, revision=%d", id, key, r.StartRevision)
+	logrus.Debugf("WATCH START id=%d, count=%d, key=%s, revision=%d", id, len(w.watches), key, r.StartRevision)
 
 	go func() {
 		defer w.wg.Done()
@@ -66,7 +67,7 @@ func (w *watcher) Start(r *etcdserverpb.WatchCreateRequest) {
 			Created: true,
 			WatchId: id,
 		}); err != nil {
-			w.Cancel(id)
+			w.Cancel(id, err)
 			return
 		}
 
@@ -86,10 +87,11 @@ func (w *watcher) Start(r *etcdserverpb.WatchCreateRequest) {
 				WatchId: id,
 				Events:  toEvents(events...),
 			}); err != nil {
-				w.Cancel(id)
+				w.Cancel(id, err)
 				continue
 			}
 		}
+		w.Cancel(id, nil)
 		logrus.Debugf("WATCH CLOSE id=%d, key=%s", id, key)
 	}()
 }
@@ -116,28 +118,35 @@ func toEvent(event *Event) *mvccpb.Event {
 	return e
 }
 
-func (w *watcher) Cancel(watchID int64) {
+func (w *watcher) Cancel(watchID int64, err error) {
 	w.Lock()
-	defer w.Unlock()
 	if cancel, ok := w.watches[watchID]; ok {
 		cancel()
 		delete(w.watches, watchID)
 	}
-	err := w.server.Send(&etcdserverpb.WatchResponse{
-		Header:   &etcdserverpb.ResponseHeader{},
-		Canceled: true,
-		WatchId:  watchID,
+	w.Unlock()
+
+	reason := ""
+	if err != nil {
+		reason = err.Error()
+	}
+	logrus.Debugf("WATCH CANCEL id=%d reason=%s", watchID, reason)
+	err = w.server.Send(&etcdserverpb.WatchResponse{
+		Header:       &etcdserverpb.ResponseHeader{},
+		Canceled:     true,
+		CancelReason: "watch closed",
+		WatchId:      watchID,
 	})
 	if err != nil {
-		logrus.Errorf("Failed to send cancel response for watchID %d: %v", watchID, err)
+		logrus.Errorf("WATCH Failed to send cancel response for watchID %d: %v", watchID, err)
 	}
 }
 
 func (w *watcher) Close() {
 	w.Lock()
-	defer w.Unlock()
 	for _, v := range w.watches {
 		v()
 	}
+	w.Unlock()
 	w.wg.Wait()
 }

vendor/modules.txt

@@ -726,7 +726,7 @@ github.com/rancher/helm-controller/pkg/generated/informers/externalversions/helm
 github.com/rancher/helm-controller/pkg/generated/informers/externalversions/internalinterfaces
 github.com/rancher/helm-controller/pkg/generated/listers/helm.cattle.io/v1
 github.com/rancher/helm-controller/pkg/helm
-# github.com/rancher/kine v0.3.2
+# github.com/rancher/kine v0.3.3
 github.com/rancher/kine/pkg/broadcaster
 github.com/rancher/kine/pkg/client
 github.com/rancher/kine/pkg/drivers/dqlite