Merge pull request canonical#6 from canonical/MK-66/grpc-keepalive

Configure keepalive and move to etcd v3.5

Author: ktsakalozos

go.mod

@@ -1,17 +1,22 @@
 module github.com/rancher/kine
 
-go 1.12
+go 1.15
 
 require (
 	github.com/Rican7/retry v0.1.0
-	github.com/canonical/go-dqlite v1.5.1
-	github.com/go-sql-driver/mysql v1.4.1
-	github.com/lib/pq v1.1.1
-	github.com/mattn/go-sqlite3 v1.10.0
-	github.com/pkg/errors v0.8.1
-	github.com/rancher/wrangler v0.4.0
-	github.com/sirupsen/logrus v1.4.2
+	github.com/canonical/go-dqlite v1.8.0
+	github.com/go-sql-driver/mysql v1.6.0
+	github.com/lib/pq v1.10.2
+	github.com/mattn/go-sqlite3 v2.0.3+incompatible
+	github.com/peterh/liner v1.2.0 // indirect
+	github.com/pkg/errors v0.9.1
+	github.com/rancher/wrangler v0.8.3
+	github.com/sirupsen/logrus v1.7.0
 	github.com/urfave/cli v1.21.0
 	go.etcd.io/etcd v0.0.0-20191023171146-3cf2f69b5738
-	google.golang.org/grpc v1.23.1
+	go.etcd.io/etcd/api/v3 v3.5.0
+	go.etcd.io/etcd/client/pkg/v3 v3.5.0
+	go.etcd.io/etcd/client/v3 v3.5.0
+	go.etcd.io/etcd/server/v3 v3.5.0
+	google.golang.org/grpc v1.38.0
 )

go.sum

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/rancher/kine/pkg/endpoint"
-	"go.etcd.io/etcd/clientv3"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 type Value struct {
@@ -26,6 +26,7 @@ type Client interface {
 	Put(ctx context.Context, key string, value []byte) error
 	Create(ctx context.Context, key string, value []byte) error
 	Update(ctx context.Context, key string, revision int64, value []byte) error
+	Delete(ctx context.Context, key string, revision int64) error
 	Close() error
 }
 
@@ -128,6 +129,21 @@ func (c *client) Update(ctx context.Context, key string, revision int64, value [
 	return nil
 }
 
+func (c *client) Delete(ctx context.Context, key string, revision int64) error {
+	resp, err := c.c.Txn(ctx).
+		If(clientv3.Compare(clientv3.ModRevision(key), "=", revision)).
+		Then(clientv3.OpDelete(key)).
+		Else(clientv3.OpGet(key)).
+		Commit()
+	if err != nil {
+		return err
+	}
+	if !resp.Succeeded {
+		return fmt.Errorf("revision %d doesnt match", revision)
+	}
+	return nil
+}
+
 func (c *client) Close() error {
 	return c.c.Close()
 }

pkg/client/client.go

@@ -15,7 +15,9 @@ import (
 	"github.com/rancher/kine/pkg/server"
 	"github.com/rancher/kine/pkg/tls"
 	"github.com/sirupsen/logrus"
+	"go.etcd.io/etcd/server/v3/embed"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/keepalive"
 )
 
 const (
@@ -112,7 +114,18 @@ func grpcServer(config Config) *grpc.Server {
 	if config.GRPCServer != nil {
 		return config.GRPCServer
 	}
-	return grpc.NewServer()
+	gopts := []grpc.ServerOption{
+		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
+			MinTime:             embed.DefaultGRPCKeepAliveMinTime,
+			PermitWithoutStream: false,
+		}),
+		grpc.KeepaliveParams(keepalive.ServerParameters{
+			Time:    embed.DefaultGRPCKeepAliveInterval,
+			Timeout: embed.DefaultGRPCKeepAliveTimeout,
+		}),
+	}
+
+	return grpc.NewServer(gopts...)
 }
 
 func getKineStorageBackend(ctx context.Context, driver, dsn string, cfg Config) (bool, server.Backend, error) {

pkg/endpoint/endpoint.go

@@ -3,10 +3,12 @@ package server
 import (
 	"context"
 
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 )
 
 func isCompact(txn *etcdserverpb.TxnRequest) bool {
+	// See https://github.com/kubernetes/kubernetes/blob/442a69c3bdf6fe8e525b05887e57d89db1e2f3a5/staging/src/k8s.io/apiserver/pkg/storage/etcd3/compact.go#L72
 	return len(txn.Compare) == 1 &&
 		txn.Compare[0].Target == etcdserverpb.Compare_VERSION &&
 		txn.Compare[0].Result == etcdserverpb.Compare_EQUAL &&
@@ -18,14 +20,19 @@ func isCompact(txn *etcdserverpb.TxnRequest) bool {
 }
 
 func (l *LimitedServer) compact(ctx context.Context) (*etcdserverpb.TxnResponse, error) {
+	// return comparison failure so that the apiserver does not bother compacting
 	return &etcdserverpb.TxnResponse{
 		Header:    &etcdserverpb.ResponseHeader{},
-		Succeeded: true,
+		Succeeded: false,
 		Responses: []*etcdserverpb.ResponseOp{
 			{
-				Response: &etcdserverpb.ResponseOp_ResponsePut{
-					ResponsePut: &etcdserverpb.PutResponse{
+				Response: &etcdserverpb.ResponseOp_ResponseRange{
+					ResponseRange: &etcdserverpb.RangeResponse{
 						Header: &etcdserverpb.ResponseHeader{},
+						Kvs: []*mvccpb.KeyValue{
+							&mvccpb.KeyValue{},
+						},
+						Count: 1,
 					},
 				},
 			},

pkg/server/compact.go

@@ -3,7 +3,7 @@ package server
 import (
 	"context"
 
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
 
 func isCreate(txn *etcdserverpb.TxnRequest) *etcdserverpb.PutRequest {

pkg/server/create.go

@@ -3,7 +3,7 @@ package server
 import (
 	"context"
 
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
 
 func isDelete(txn *etcdserverpb.TxnRequest) (int64, string, bool) {

pkg/server/delete.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
 
 func (l *LimitedServer) get(ctx context.Context, r *etcdserverpb.RangeRequest) (*RangeResponse, error) {

pkg/server/get.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
 
 func (s *KVServerBridge) LeaseGrant(ctx context.Context, req *etcdserverpb.LeaseGrantRequest) (*etcdserverpb.LeaseGrantResponse, error) {

pkg/server/lease.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
 
 type LimitedServer struct {

pkg/server/limited.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"strings"
 
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
 
 func (l *LimitedServer) list(ctx context.Context, r *etcdserverpb.RangeRequest) (*RangeResponse, error) {

pkg/server/list.go

@@ -5,8 +5,8 @@ import (
 	"fmt"
 
 	"github.com/sirupsen/logrus"
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
-	"go.etcd.io/etcd/mvcc/mvccpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health"
 	healthpb "google.golang.org/grpc/health/grpc_health_v1"

pkg/server/server.go

@@ -3,7 +3,7 @@ package server
 import (
 	"context"
 
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
 
 func isUpdate(txn *etcdserverpb.TxnRequest) (int64, string, []byte, int64, bool) {

pkg/server/update.go

@@ -6,8 +6,8 @@ import (
 	"sync/atomic"
 
 	"github.com/sirupsen/logrus"
-	"go.etcd.io/etcd/etcdserver/etcdserverpb"
-	"go.etcd.io/etcd/mvcc/mvccpb"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 )
 
 var (

pkg/server/watch.go

@@ -3,7 +3,7 @@ package tls
 import (
 	"crypto/tls"
 
-	"go.etcd.io/etcd/pkg/transport"
+	"go.etcd.io/etcd/client/pkg/v3/transport"
 )
 
 type Config struct {