Merge pull request #1676 from galal-hussein/update_kine_117

Dockerfile.dapper

@@ -22,7 +22,7 @@ ENV SELINUX $SELINUX
 
 ARG DQLITE=true
 ENV DQLITE $DQLITE
-COPY --from=rancher/dqlite-build:v1.3.1-r1 /dist/artifacts /usr/src/
+COPY --from=rancher/dqlite-build:v1.4.1-r1 /dist/artifacts /usr/src/
 RUN if [ "$DQLITE" = true ]; then \
         tar xzf /usr/src/dqlite.tgz -C / && \
         apk add --allow-untrusted /usr/local/packages/*.apk \

go.mod

@@ -64,7 +64,7 @@ require (
 	github.com/bhendo/go-powershell v0.0.0-20190719160123-219e7fb4e41e // indirect
 	github.com/bronze1man/goStrongswanVici v0.0.0-20190828090544-27d02f80ba40 // indirect
 	github.com/buger/jsonparser v0.0.0-20181115193947-bf1c66bbce23 // indirect
-	github.com/canonical/go-dqlite v1.3.0
+	github.com/canonical/go-dqlite v1.5.1
 	github.com/containerd/cgroups v0.0.0-00010101000000-000000000000 // indirect
 	github.com/containerd/containerd v1.3.0-beta.2.0.20190828155532-0293cbd26c69
 	github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6 // indirect
@@ -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
-	github.com/rancher/kine v0.3.5
+	github.com/rancher/kine v0.4.0
 	github.com/rancher/remotedialer v0.2.0
 	github.com/rancher/wrangler v0.4.0
 	github.com/rancher/wrangler-api v0.4.0

go.sum

@@ -116,6 +116,8 @@ github.com/caddyserver/caddy v1.0.3/go.mod h1:G+ouvOY32gENkJC+jhgl62TyhvqEsFaDiZ
 github.com/canonical/go-dqlite v1.2.0/go.mod h1:wp00vfMvPYgNCyxcPdHB5XExmDoCGoPUGymloAQT17Y=
 github.com/canonical/go-dqlite v1.3.0 h1:c+7eGZfh0K7yCmGrBkNRGZdY8R8+2jSSkz6Zr3YCjJE=
 github.com/canonical/go-dqlite v1.3.0/go.mod h1:wp00vfMvPYgNCyxcPdHB5XExmDoCGoPUGymloAQT17Y=
+github.com/canonical/go-dqlite v1.5.1 h1:1YjtIrFsC1A3XlgsX38ARAiKhvkZS63PqsEd8z3T4yU=
+github.com/canonical/go-dqlite v1.5.1/go.mod h1:wp00vfMvPYgNCyxcPdHB5XExmDoCGoPUGymloAQT17Y=
 github.com/cenkalti/backoff v2.1.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/census-instrumentation/opencensus-proto v0.2.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -722,6 +724,10 @@ github.com/rancher/helm-controller v0.4.1 h1:AbI3zXy2Qtw4e4A6IkxKozgnm6T3KKnUo6E
 github.com/rancher/helm-controller v0.4.1/go.mod h1:194LHuZRrxcD82bG1rJtOWsw98U4JbPhDWqvL7l3PAw=
 github.com/rancher/kine v0.3.5 h1:Tm4eOtejpnzs1WFBrXj76lCLvX9czLlTkgqUk9luCQk=
 github.com/rancher/kine v0.3.5/go.mod h1:xEMl0tLCva9/9me7mXJ3m9Vo6yqHgC4OU3NiK4CPrGQ=
+github.com/rancher/kine v0.3.6-0.20200422224205-0a0f5b924129 h1:4HYvCG8+pfkBlpYLv/5ZOdxagg3jTszMOqMjamMQ0hA=
+github.com/rancher/kine v0.3.6-0.20200422224205-0a0f5b924129/go.mod h1:xEMl0tLCva9/9me7mXJ3m9Vo6yqHgC4OU3NiK4CPrGQ=
+github.com/rancher/kine v0.4.0 h1:1IhWy3TzjExG8xnj46eyUEWdzqNAD1WrgL4eEBKm6Uc=
+github.com/rancher/kine v0.4.0/go.mod h1:IImtCJ68AIkE+VY/kUI0NkyJL5q5WzO8QvMsSXqbrpA=
 github.com/rancher/kubernetes v1.17.5-k3s1 h1:rE+lSQ7SszfLZ4KB94FPQtvqQPYoU6Jwul8HHBPqQcY=
 github.com/rancher/kubernetes v1.17.5-k3s1/go.mod h1:4A7Bm7SHrj2SCb9vxMBil1wu2uJy3k1zSjoD43SS8Zg=
 github.com/rancher/kubernetes/staging/src/k8s.io/api v1.17.5-k3s1 h1:/6jG49ioPbxPlk+wY3XQTj3JjVoGuE/5Ozx3tZGk32g=

vendor/github.com/canonical/go-dqlite/.gitignore

 .sqlite
+cmd/dqlite/dqlite
+cmd/dqlite-demo/dqlite-demo
 demo
 profile.coverprofile
 overalls.coverprofile

vendor/github.com/canonical/go-dqlite/README.md

@@ -10,9 +10,32 @@ Usage
 
 The best way to understand how to use the ```go-dqlite``` package is probably by
 looking at the source code of the [demo
-program](https://github.com/canonical/go-dqlite/tree/master/cmd/dqlite-demo) and
+program](https://github.com/canonical/go-dqlite/tree/master/cmd/dqlite-demo/main.go) and
 use it as example.
 
+In general your application will use code such as:
+
+
+```go
+dir := "/path/to/data/directory"
+address := "1.2.3.4:666" // Unique node address
+cluster := []string{...} // Optional list of existing nodes, when starting a new node
+app, err := app.New(dir, app.WithAddress(address), app.WithCluster(cluster))
+if err != nil {
+        // ...
+}
+
+db, err := app.Open(context.Background(), "my-database")
+if err != nil {
+        // ...
+}
+
+// db is a *sql.DB object
+if _, err := db.Exec("CREATE TABLE my_table (n INT)"); err != nil
+        // ...
+}
+```
+
 Build
 -----
 
@@ -50,42 +73,61 @@ go install -tags libsqlite3 ./cmd/dqlite-demo
 
 from the top-level directory of this repository.
 
-Once the ```dqlite-demo``` binary is installed, start three nodes of the demo
-application, respectively with IDs ```1```, ```2,``` and ```3```:
+This builds a demo dqlite application, which exposes a simple key/value store
+over an HTTP API.
+
+Once the `dqlite-demo` binary is installed (normally under `~/go/bin`),
+start three nodes of the demo application:
 
 ```bash
-dqlite-demo start 1 &
-dqlite-demo start 2 &
-dqlite-demo start 3 &
+dqlite-demo --api 127.0.0.1:8001 --db 127.0.0.1:9001 &
+dqlite-demo --api 127.0.0.1:8002 --db 127.0.0.1:9002 --join 127.0.0.1:9001 &
+dqlite-demo --api 127.0.0.1:8003 --db 127.0.0.1:9003 --join 127.0.0.1:9001 &
 ```
 
-The node with ID ```1``` automatically becomes the leader of a single node
-cluster, while the nodes with IDs ```2``` and ```3``` are waiting to be notified
-what cluster they belong to. Let's make nodes ```2``` and ```3``` join the
-cluster:
+The `--api` flag tells the demo program where to expose its HTTP API.
+
+The `--db` flag tells the demo program to use the given address for internal
+database replication.
+
+The `--join` flag is optional and should be used only for additional nodes after
+the first one. It informs them about the existing cluster, so they can
+automatically join it.
+
+Now we can start using the cluster. Let's insert a key pair:
 
 ```bash
-dqlite-demo add 2
-dqlite-demo add 3
+curl -X PUT -d my-key http://127.0.0.1:8001/my-value
 ```
 
-Now we can start using the cluster. The demo application is just a simple
-key/value store that stores data in a SQLite table. Let's insert a key pair:
+and then retrive it from the database:
 
 ```bash
-dqlite-demo update my-key my-value
+curl http://127.0.0.1:8001/my-value
 ```
 
-and then retrive it from the database:
+Currently the first node is the leader. If we stop it and then try to query the
+key again curl will fail, but we can simply change the endpoint to another node
+and things will work since an automatic failover has taken place:
 
 ```bash
-dqlite-demo query my-key
+kill -TERM %1; curl http://127.0.0.1:8002/my-value
 ```
 
-Currently node ```1``` is the leader. If we stop it and then try to query the
-key again we'll notice that the ```query``` command hangs for a bit waiting for
-the failover to occur and for another node to step up as leader:
+Shell
+------
+
+A basic SQLite-like dqlite shell can be built with:
 
 ```
-kill -TERM %1; sleep 0.1; dqlite-demo query my-key; dqlite-demo cluster
+go install -tags libsqlite3 ./cmd/dqlite
 ```
+
+You can test it with the `dqlite-demo` with:
+
+```
+dqlite -s 127.0.0.1:9001
+```
+
+It supports normal SQL queries plus the special `.cluster` and `.leader`
+commands to inspect the cluster members and the current leader.

vendor/github.com/canonical/go-dqlite/client/client.go

@@ -2,10 +2,6 @@ package client
 
 import (
 	"context"
-	"encoding/binary"
-	"io"
-	"net"
-	"strings"
 
 	"github.com/canonical/go-dqlite/internal/protocol"
 	"github.com/pkg/errors"
@@ -57,22 +53,13 @@ func New(ctx context.Context, address string, options ...Option) (*Client, error
 		return nil, errors.Wrap(err, "failed to establish network connection")
 	}
 
-	// Latest protocol version.
-	proto := make([]byte, 8)
-	binary.LittleEndian.PutUint64(proto, protocol.VersionOne)
-
-	// Perform the protocol handshake.
-	n, err := conn.Write(proto)
+	protocol, err := protocol.Handshake(ctx, conn, protocol.VersionOne)
 	if err != nil {
 		conn.Close()
-		return nil, errors.Wrap(err, "failed to send handshake")
-	}
-	if n != 8 {
-		conn.Close()
-		return nil, errors.Wrap(io.ErrShortWrite, "failed to send handshake")
+		return nil, err
 	}
 
-	client := &Client{protocol: protocol.NewProtocol(protocol.VersionOne, conn)}
+	client := &Client{protocol: protocol}
 
 	return client, nil
 }
@@ -274,10 +261,3 @@ func defaultOptions() *options {
 		LogFunc:  DefaultLogFunc,
 	}
 }
-
-func DefaultDialFunc(ctx context.Context, address string) (net.Conn, error) {
-	if strings.HasPrefix(address, "@") {
-		return protocol.UnixDial(ctx, address)
-	}
-	return protocol.TCPDial(ctx, address)
-}

vendor/github.com/canonical/go-dqlite/client/dial.go

+package client
+
+import (
+	"context"
+	"crypto/tls"
+	"net"
+	"strings"
+
+	"github.com/canonical/go-dqlite/internal/protocol"
+)
+
+// DefaultDialFunc is the default dial function, which can handle plain TCP and
+// Unix socket endpoints. You can customize it with WithDialFunc()
+func DefaultDialFunc(ctx context.Context, address string) (net.Conn, error) {
+	if strings.HasPrefix(address, "@") {
+		return protocol.UnixDial(ctx, address)
+	}
+	return protocol.TCPDial(ctx, address)
+}
+
+// DialFuncWithTLS returns a dial function that uses TLS encryption.
+//
+// The given dial function will be used to establish the network connection,
+// and the given TLS config will be used for encryption.
+func DialFuncWithTLS(dial DialFunc, config *tls.Config) DialFunc {
+	return func(ctx context.Context, addr string) (net.Conn, error) {
+		conn, err := dial(ctx, addr)
+		if err != nil {
+			return nil, err
+		}
+		return tls.Client(conn, config), nil
+	}
+}

vendor/github.com/canonical/go-dqlite/client/leader.go

@@ -2,10 +2,7 @@ package client
 
 import (
 	"context"
-	"time"
 
-	"github.com/Rican7/retry/backoff"
-	"github.com/Rican7/retry/strategy"
 	"github.com/canonical/go-dqlite/internal/protocol"
 )
 
@@ -18,10 +15,7 @@ func FindLeader(ctx context.Context, store NodeStore, options ...Option) (*Clien
 	}
 
 	config := protocol.Config{
-		Dial:           o.DialFunc,
-		AttemptTimeout: time.Second,
-		RetryStrategies: []strategy.Strategy{
-			strategy.Backoff(backoff.BinaryExponential(time.Millisecond))},
+		Dial: o.DialFunc,
 	}
 	connector := protocol.NewConnector(0, store, config, o.LogFunc)
 	protocol, err := connector.Connect(ctx)

vendor/github.com/canonical/go-dqlite/client/log.go

 package client
 
 import (
-	"fmt"
-	"log"
-	"os"
-
 	"github.com/canonical/go-dqlite/internal/logging"
 )
 
@@ -22,11 +18,5 @@ const (
 	LogError = logging.Error
 )
 
-var (
-	logger = log.New(os.Stdout, "", log.LstdFlags|log.Lmicroseconds)
-)
-
-// DefaultLogFunc emits messages using the stdlib's logger.
-func DefaultLogFunc(l LogLevel, format string, a ...interface{}) {
-	logger.Output(2, fmt.Sprintf("[%s]: %s", l.String(), fmt.Sprintf(format, a...)))
-}
+// DefaultLogFunc doesn't emit any message.
+func DefaultLogFunc(l LogLevel, format string, a ...interface{}) {}

vendor/github.com/canonical/go-dqlite/client/store.go

@@ -4,7 +4,11 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"io/ioutil"
+	"os"
+	"sync"
 
+	"github.com/ghodss/yaml"
 	"github.com/pkg/errors"
 
 	"github.com/canonical/go-dqlite/internal/protocol"
@@ -163,3 +167,65 @@ func (d *DatabaseNodeStore) Set(ctx context.Context, servers []NodeInfo) error {
 
 	return nil
 }
+
+// Persists a list addresses of dqlite nodes in a YAML file.
+type YamlNodeStore struct {
+	path    string
+	servers []NodeInfo
+	mu      sync.RWMutex
+}
+
+// NewYamlNodeStore creates a new YamlNodeStore backed by the given YAML file.
+func NewYamlNodeStore(path string) (*YamlNodeStore, error) {
+	servers := []NodeInfo{}
+
+	_, err := os.Stat(path)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			return nil, err
+		}
+	} else {
+		data, err := ioutil.ReadFile(path)
+		if err != nil {
+			return nil, err
+		}
+
+		if err := yaml.Unmarshal(data, &servers); err != nil {
+			return nil, err
+		}
+	}
+
+	store := &YamlNodeStore{
+		path:    path,
+		servers: servers,
+	}
+
+	return store, nil
+}
+
+// Get the current servers.
+func (s *YamlNodeStore) Get(ctx context.Context) ([]NodeInfo, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	return s.servers, nil
+}
+
+// Set the servers addresses.
+func (s *YamlNodeStore) Set(ctx context.Context, servers []NodeInfo) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	data, err := yaml.Marshal(servers)
+	if err != nil {
+		return err
+	}
+
+	if err := ioutil.WriteFile(s.path, data, 0600); err != nil {
+		return err
+	}
+
+	s.servers = servers
+
+	return nil
+}

vendor/github.com/canonical/go-dqlite/config.go

@@ -5,6 +5,7 @@ import (
 	"os"
 
 	"github.com/canonical/go-dqlite/internal/bindings"
+	"github.com/canonical/go-dqlite/internal/protocol"
 	"github.com/pkg/errors"
 )
 
@@ -26,7 +27,7 @@ import (
 // from setting Single-thread mode at all.
 func ConfigMultiThread() error {
 	if err := bindings.ConfigMultiThread(); err != nil {
-		if err, ok := err.(bindings.Error); ok && err.Code == 21 /* SQLITE_MISUSE */ {
+		if err, ok := err.(protocol.Error); ok && err.Code == 21 /* SQLITE_MISUSE */ {
 			return fmt.Errorf("SQLite is already initialized")
 		}
 		return errors.Wrap(err, "unknown error")

vendor/github.com/canonical/go-dqlite/internal/bindings/errors.go

-package bindings
-
-/*
-#include <sqlite3.h>
-*/
-import "C"
-
-// Error holds information about a SQLite error.
-type Error struct {
-	Code    int
-	Message string
-}
-
-func (e Error) Error() string {
-	if e.Message != "" {
-		return e.Message
-	}
-	return C.GoString(C.sqlite3_errstr(C.int(e.Code)))
-}

vendor/github.com/canonical/go-dqlite/internal/bindings/server.go

@@ -90,14 +90,14 @@ func init() {
 
 func ConfigSingleThread() error {
 	if rc := C.sqlite3ConfigSingleThread(); rc != 0 {
-		return Error{Code: int(rc)}
+		return protocol.Error{Message: C.GoString(C.sqlite3_errstr(rc)), Code: int(rc)}
 	}
 	return nil
 }
 
 func ConfigMultiThread() error {
 	if rc := C.sqlite3ConfigMultiThread(); rc != 0 {
-		return Error{Code: int(rc)}
+		return protocol.Error{Message: C.GoString(C.sqlite3_errstr(rc)), Code: int(rc)}
 	}
 	return nil
 }
@@ -197,6 +197,14 @@ func (s *Node) Recover(cluster []protocol.NodeInfo) error {
 	return nil
 }
 
+// GenerateID generates a unique ID for a server.
+func GenerateID(address string) uint64 {
+	caddress := C.CString(address)
+	defer C.free(unsafe.Pointer(caddress))
+	id := C.dqlite_generate_node_id(caddress)
+	return uint64(id)
+}
+
 // Extract the underlying socket from a connection.
 func connToSocket(conn net.Conn) (C.int, error) {
 	file, err := conn.(fileConn).File()

vendor/github.com/canonical/go-dqlite/internal/protocol/config.go

@@ -2,13 +2,14 @@ package protocol
 
 import (
 	"time"
-
-	"github.com/Rican7/retry/strategy"
 )
 
 // Config holds various configuration parameters for a dqlite client.
 type Config struct {
-	Dial            DialFunc            // Network dialer.
-	AttemptTimeout  time.Duration       // Timeout for each individual Dial attempt.
-	RetryStrategies []strategy.Strategy // Strategies used for retrying to connect to a leader.
+	Dial           DialFunc      // Network dialer.
+	DialTimeout    time.Duration // Timeout for establishing a network connection .
+	AttemptTimeout time.Duration // Timeout for each individual attempt to probe a server's leadership.
+	BackoffFactor  time.Duration // Exponential backoff factor for retries.
+	BackoffCap     time.Duration // Maximum connection retry backoff value,
+	RetryLimit     uint          // Maximum number of retries, or 0 for unlimited.
 }

vendor/github.com/canonical/go-dqlite/internal/protocol/connector.go

@@ -6,8 +6,11 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"time"
 
 	"github.com/Rican7/retry"
+	"github.com/Rican7/retry/backoff"
+	"github.com/Rican7/retry/strategy"
 	"github.com/canonical/go-dqlite/internal/logging"
 	"github.com/pkg/errors"
 )
@@ -27,6 +30,22 @@ type Connector struct {
 // NewConnector returns a new connector that can be used by a dqlite driver to
 // create new clients connected to a leader dqlite server.
 func NewConnector(id uint64, store NodeStore, config Config, log logging.Func) *Connector {
+	if config.DialTimeout == 0 {
+		config.DialTimeout = 10 * time.Second
+	}
+
+	if config.AttemptTimeout == 0 {
+		config.AttemptTimeout = 60 * time.Second
+	}
+
+	if config.BackoffFactor == 0 {
+		config.BackoffFactor = 100 * time.Millisecond
+	}
+
+	if config.BackoffCap == 0 {
+		config.BackoffCap = time.Second
+	}
+
 	connector := &Connector{
 		id:     id,
 		store:  store,
@@ -43,6 +62,8 @@ func NewConnector(id uint64, store NodeStore, config Config, log logging.Func) *
 func (c *Connector) Connect(ctx context.Context) (*Protocol, error) {
 	var protocol *Protocol
 
+	strategies := makeRetryStrategies(c.config.BackoffFactor, c.config.BackoffCap, c.config.RetryLimit)
+
 	// The retry strategy should be configured to retry indefinitely, until
 	// the given context is done.
 	err := retry.Retry(func(attempt uint) error {
@@ -61,17 +82,16 @@ func (c *Connector) Connect(ctx context.Context) (*Protocol, error) {
 		var err error
 		protocol, err = c.connectAttemptAll(ctx, log)
 		if err != nil {
-			log(logging.Debug, "connection failed err=%v", err)
 			return err
 		}
 
 		return nil
-	}, c.config.RetryStrategies...)
+	}, strategies...)
 
 	if err != nil {
-		// The retry strategy should never give up until success or
-		// context expiration.
-		panic("connect retry aborted unexpectedly")
+		// We exhausted the number of retries allowed by the configured
+		// strategy.
+		return nil, ErrNoAvailableLeader
 	}
 
 	if ctx.Err() != nil {
@@ -92,7 +112,7 @@ func (c *Connector) connectAttemptAll(ctx context.Context, log logging.Func) (*P
 	// Make an attempt for each address until we find the leader.
 	for _, server := range servers {
 		log := func(l logging.Level, format string, a ...interface{}) {
-			format += fmt.Sprintf(" address=%s id=%d", server.Address, server.ID)
+			format += fmt.Sprintf(" address=%s", server.Address)
 			log(l, format, a...)
 		}
 
@@ -107,68 +127,71 @@ func (c *Connector) connectAttemptAll(ctx context.Context, log logging.Func) (*P
 		}
 		if err != nil {
 			// This server is unavailable, try with the next target.
-			log(logging.Debug, "server connection failed err=%v", err)
+			log(logging.Warn, "server unavailable err=%v", err)
 			continue
 		}
 		if protocol != nil {
 			// We found the leader
-			log(logging.Info, "connected")
+			log(logging.Debug, "connected")
 			return protocol, nil
 		}
 		if leader == "" {
 			// This server does not know who the current leader is,
 			// try with the next target.
+			log(logging.Warn, "no known leader")
 			continue
 		}
 
 		// If we get here, it means this server reported that another
 		// server is the leader, let's close the connection to this
 		// server and try with the suggested one.
-		//logger = logger.With(zap.String("leader", leader))
+		log(logging.Debug, "connect to reported leader %s", leader)
+
+		ctx, cancel = context.WithTimeout(ctx, c.config.AttemptTimeout)
+		defer cancel()
+
 		protocol, leader, err = c.connectAttemptOne(ctx, leader, version)
 		if err != nil {
 			// The leader reported by the previous server is
 			// unavailable, try with the next target.
-			//logger.Info("leader server connection failed", zap.String("err", err.Error()))
+			log(logging.Warn, "reported leader unavailable err=%v", err)
 			continue
 		}
 		if protocol == nil {
 			// The leader reported by the target server does not consider itself
 			// the leader, try with the next target.
-			//logger.Info("reported leader server is not the leader")
+			log(logging.Warn, "reported leader server is not the leader")
 			continue
 		}
-		log(logging.Info, "connected")
+		log(logging.Debug, "connected")
 		return protocol, nil
 	}
 
 	return nil, ErrNoAvailableLeader
 }
 
-// Connect establishes a connection with a dqlite node.
-func Connect(ctx context.Context, dial DialFunc, address string, version uint64) (*Protocol, error) {
-	// Establish the connection.
-	conn, err := dial(ctx, address)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to establish network connection")
-	}
-
+// Perform the initial handshake using the given protocol version.
+func Handshake(ctx context.Context, conn net.Conn, version uint64) (*Protocol, error) {
 	// Latest protocol version.
 	protocol := make([]byte, 8)
 	binary.LittleEndian.PutUint64(protocol, version)
 
+	// Honor the ctx deadline, if present.
+	if deadline, ok := ctx.Deadline(); ok {
+		conn.SetDeadline(deadline)
+		defer conn.SetDeadline(time.Time{})
+	}
+
 	// Perform the protocol handshake.
 	n, err := conn.Write(protocol)
 	if err != nil {
-		conn.Close()
 		return nil, errors.Wrap(err, "failed to send handshake")
 	}
 	if n != 8 {
-		conn.Close()
 		return nil, errors.Wrap(io.ErrShortWrite, "failed to send handshake")
 	}
 
-	return NewProtocol(version, conn), nil
+	return newProtocol(version, conn), nil
 }
 
 // Connect to the given dqlite server and check if it's the leader.
@@ -181,8 +204,18 @@ func Connect(ctx context.Context, dial DialFunc, address string, version uint64)
 // - Target is the leader:                   -> server, "", nil
 //
 func (c *Connector) connectAttemptOne(ctx context.Context, address string, version uint64) (*Protocol, string, error) {
-	protocol, err := Connect(ctx, c.config.Dial, address, version)
+	dialCtx, cancel := context.WithTimeout(ctx, c.config.DialTimeout)
+	defer cancel()
+
+	// Establish the connection.
+	conn, err := c.config.Dial(dialCtx, address)
+	if err != nil {
+		return nil, "", errors.Wrap(err, "failed to establish network connection")
+	}
+
+	protocol, err := Handshake(ctx, conn, version)
 	if err != nil {
+		conn.Close()
 		return nil, "", err
 	}
 
@@ -219,8 +252,8 @@ func (c *Connector) connectAttemptOne(ctx context.Context, address string, versi
 		return nil, "", nil
 	case address:
 		// This server is the leader, register ourselves and return.
-		request.Reset()
-		response.Reset()
+		request.reset()
+		response.reset()
 
 		EncodeClient(&request, c.id)
 
@@ -247,4 +280,33 @@ func (c *Connector) connectAttemptOne(ctx context.Context, address string, versi
 	}
 }
 
+// Return a retry strategy with exponential backoff, capped at the given amount
+// of time and possibly with a maximum number of retries.
+func makeRetryStrategies(factor, cap time.Duration, limit uint) []strategy.Strategy {
+	backoff := backoff.BinaryExponential(factor)
+
+	strategies := []strategy.Strategy{}
+
+	if limit > 0 {
+		strategies = append(strategies, strategy.Limit(limit))
+	}
+
+	strategies = append(strategies,
+		func(attempt uint) bool {
+			if attempt > 0 {
+				duration := backoff(attempt)
+				// Duration might be negative in case of integer overflow.
+				if duration > cap || duration <= 0 {
+					duration = cap
+				}
+				time.Sleep(duration)
+			}
+
+			return true
+		},
+	)
+
+	return strategies
+}
+
 var errBadProtocol = fmt.Errorf("bad protocol")

vendor/github.com/canonical/go-dqlite/internal/protocol/dial.go

@@ -2,6 +2,7 @@ package protocol
 
 import (
 	"context"
+	"crypto/tls"
 	"net"
 )
 
@@ -18,3 +19,15 @@ func UnixDial(ctx context.Context, address string) (net.Conn, error) {
 	dialer := net.Dialer{}
 	return dialer.DialContext(ctx, "unix", address)
 }
+
+// TLSCipherSuites are the cipher suites by the go-dqlite TLS helpers.
+var TLSCipherSuites = []uint16{
+	tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
+	tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
+	tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
+	tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
+	tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,
+	tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,
+	tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
+	tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,
+}

vendor/github.com/canonical/go-dqlite/internal/protocol/errors.go

@@ -27,3 +27,13 @@ func (e ErrRequest) Error() string {
 // ErrRowsPart is returned when the first batch of a multi-response result
 // batch is done.
 var ErrRowsPart = fmt.Errorf("not all rows were returned in this response")
+
+// Error holds information about a SQLite error.
+type Error struct {
+	Code    int
+	Message string
+}
+
+func (e Error) Error() string {
+	return e.Message
+}

vendor/github.com/canonical/go-dqlite/internal/protocol/message.go

@@ -26,24 +26,23 @@ type Message struct {
 	flags  uint8
 	extra  uint16
 	header []byte // Statically allocated header buffer
-	body1  buffer // Statically allocated body data, using bytes
-	body2  buffer // Dynamically allocated body data
+	body   buffer // Message body data.
 }
 
-// Init initializes the message using the given size of the statically
-// allocated buffer (i.e. a buffer which is re-used across requests or
-// responses encoded or decoded using this message object).
-func (m *Message) Init(staticSize int) {
-	if (staticSize % messageWordSize) != 0 {
-		panic("static size is not aligned to word boundary")
+// Init initializes the message using the given initial size for the data
+// buffer, which is re-used across requests or responses encoded or decoded
+// using this message object.
+func (m *Message) Init(initialBufferSize int) {
+	if (initialBufferSize % messageWordSize) != 0 {
+		panic("initial buffer size is not aligned to word boundary")
 	}
 	m.header = make([]byte, messageHeaderSize)
-	m.body1.Bytes = make([]byte, staticSize)
-	m.Reset()
+	m.body.Bytes = make([]byte, initialBufferSize)
+	m.reset()
 }
 
 // Reset the state of the message so it can be used to encode or decode again.
-func (m *Message) Reset() {
+func (m *Message) reset() {
 	m.words = 0
 	m.mtype = 0
 	m.flags = 0
@@ -51,9 +50,7 @@ func (m *Message) Reset() {
 	for i := 0; i < messageHeaderSize; i++ {
 		m.header[i] = 0
 	}
-	m.body1.Offset = 0
-	m.body2.Bytes = nil
-	m.body2.Offset = 0
+	m.body.Offset = 0
 }
 
 // Append a byte slice to the message.
@@ -232,11 +229,11 @@ func (m *Message) putNamedValues(values NamedValues) {
 // Finalize the message by setting the message type and the number
 // of words in the body (calculated from the body size).
 func (m *Message) putHeader(mtype uint8) {
-	if m.body1.Offset <= 0 {
+	if m.body.Offset <= 0 {
 		panic("static offset is not positive")
 	}
 
-	if (m.body1.Offset % messageWordSize) != 0 {
+	if (m.body.Offset % messageWordSize) != 0 {
 		panic("static body is not aligned")
 	}
 
@@ -244,22 +241,7 @@ func (m *Message) putHeader(mtype uint8) {
 	m.flags = 0
 	m.extra = 0
 
-	m.words = uint32(m.body1.Offset) / messageWordSize
-
-	if m.body2.Bytes == nil {
-		m.finalize()
-		return
-	}
-
-	if m.body2.Offset <= 0 {
-		panic("dynamic offset is not positive")
-	}
-
-	if (m.body2.Offset % messageWordSize) != 0 {
-		panic("dynamic body is not aligned")
-	}
-
-	m.words += uint32(m.body2.Offset) / messageWordSize
+	m.words = uint32(m.body.Offset) / messageWordSize
 
 	m.finalize()
 }
@@ -276,27 +258,14 @@ func (m *Message) finalize() {
 }
 
 func (m *Message) bufferForPut(size int) *buffer {
-	if m.body2.Bytes != nil {
-		if (m.body2.Offset + size) > len(m.body2.Bytes) {
-			// Grow body2.
-			//
-			// TODO: find a good grow strategy.
-			bytes := make([]byte, m.body2.Offset+size)
-			copy(bytes, m.body2.Bytes)
-			m.body2.Bytes = bytes
-		}
-
-		return &m.body2
+	for (m.body.Offset + size) > len(m.body.Bytes) {
+		// Grow message buffer.
+		bytes := make([]byte, len(m.body.Bytes)*2)
+		copy(bytes, m.body.Bytes)
+		m.body.Bytes = bytes
 	}
 
-	if (m.body1.Offset + size) > len(m.body1.Bytes) {
-		m.body2.Bytes = make([]byte, size)
-		m.body2.Offset = 0
-
-		return &m.body2
-	}
-
-	return &m.body1
+	return &m.body
 }
 
 // Return the message type and its flags.
@@ -310,31 +279,6 @@ func (m *Message) getString() string {
 
 	index := bytes.IndexByte(b.Bytes[b.Offset:], 0)
 	if index == -1 {
-		// Check if the string overflows in the dynamic buffer.
-		if b == &m.body1 && m.body2.Bytes != nil {
-			// Assert that this is the first read of the dynamic buffer.
-			if m.body2.Offset != 0 {
-				panic("static buffer read after dynamic buffer one")
-			}
-			index = bytes.IndexByte(m.body2.Bytes[0:], 0)
-			if index != -1 {
-				// We found the trailing part of the string.
-				data := b.Bytes[b.Offset:]
-				data = append(data, m.body2.Bytes[0:index]...)
-
-				index++
-
-				if trailing := index % messageWordSize; trailing != 0 {
-					// Account for padding, moving index to the next word boundary.
-					index += messageWordSize - trailing
-				}
-
-				m.body1.Offset = len(m.body1.Bytes)
-				m.body2.Advance(index)
-
-				return string(data)
-			}
-		}
 		panic("no string found")
 	}
 	s := string(b.Bytes[b.Offset : b.Offset+index])
@@ -465,31 +409,23 @@ func (m *Message) getFiles() Files {
 
 func (m *Message) hasBeenConsumed() bool {
 	size := int(m.words * messageWordSize)
-	return (m.body1.Offset == size || m.body1.Offset == len(m.body1.Bytes)) &&
-		m.body1.Offset+m.body2.Offset == size
+	return m.body.Offset == size
 }
 
 func (m *Message) lastByte() byte {
 	size := int(m.words * messageWordSize)
-	if size > len(m.body1.Bytes) {
-		size = size - m.body1.Offset
-		return m.body2.Bytes[size-1]
-	}
-	return m.body1.Bytes[size-1]
+	return m.body.Bytes[size-1]
 }
 
 func (m *Message) bufferForGet() *buffer {
 	size := int(m.words * messageWordSize)
-	if m.body1.Offset == size || m.body1.Offset == len(m.body1.Bytes) {
-		// The static body has been exahusted, use the dynamic one.
-		if m.body1.Offset+m.body2.Offset == size {
-			err := fmt.Errorf("short message: type=%d words=%d off=%d", m.mtype, m.words, m.body1.Offset)
-			panic(err)
-		}
-		return &m.body2
+	// The static body has been exahusted, use the dynamic one.
+	if m.body.Offset == size {
+		err := fmt.Errorf("short message: type=%d words=%d off=%d", m.mtype, m.words, m.body.Offset)
+		panic(err)
 	}
 
-	return &m.body1
+	return &m.body
 }
 
 // Result holds the result of a statement.
@@ -639,7 +575,7 @@ func (r *Rows) Close() error {
 			err = fmt.Errorf("unexpected end of message")
 		}
 	}
-	r.message.Reset()
+	r.message.reset()
 	return err
 }
 
@@ -664,7 +600,7 @@ func (f *Files) Next() (string, []byte) {
 }
 
 func (f *Files) Close() {
-	f.message.Reset()
+	f.message.reset()
 }
 
 const (

vendor/github.com/canonical/go-dqlite/internal/protocol/protocol.go

@@ -13,31 +13,23 @@ import (
 
 // Protocol sends and receive the dqlite message on the wire.
 type Protocol struct {
-	version        uint64        // Protocol version
-	conn           net.Conn      // Underlying network connection.
-	contextTimeout time.Duration // Default context timeout.
-	closeCh        chan struct{} // Stops the heartbeat when the connection gets closed
-	mu             sync.Mutex    // Serialize requests
-	netErr         error         // A network error occurred
+	version uint64        // Protocol version
+	conn    net.Conn      // Underlying network connection.
+	closeCh chan struct{} // Stops the heartbeat when the connection gets closed
+	mu      sync.Mutex    // Serialize requests
+	netErr  error         // A network error occurred
 }
 
-func NewProtocol(version uint64, conn net.Conn) *Protocol {
+func newProtocol(version uint64, conn net.Conn) *Protocol {
 	protocol := &Protocol{
-		version:        version,
-		conn:           conn,
-		closeCh:        make(chan struct{}),
-		contextTimeout: 5 * time.Second,
+		version: version,
+		conn:    conn,
+		closeCh: make(chan struct{}),
 	}
 
 	return protocol
 }
 
-// SetContextTimeout sets the default context timeout when no deadline is
-// provided.
-func (p *Protocol) SetContextTimeout(timeout time.Duration) {
-	p.contextTimeout = timeout
-}
-
 // Call invokes a dqlite RPC, sending a request message and receiving a
 // response message.
 func (p *Protocol) Call(ctx context.Context, request, response *Message) (err error) {
@@ -50,21 +42,22 @@ func (p *Protocol) Call(ctx context.Context, request, response *Message) (err er
 		return p.netErr
 	}
 
-	// Honor the ctx deadline, if present, or use a default.
-	deadline, ok := ctx.Deadline()
-	if !ok {
-		deadline = time.Now().Add(p.contextTimeout)
-	}
+	var budget time.Duration
 
-	p.conn.SetDeadline(deadline)
+	// Honor the ctx deadline, if present.
+	if deadline, ok := ctx.Deadline(); ok {
+		p.conn.SetDeadline(deadline)
+		budget = time.Until(deadline)
+		defer p.conn.SetDeadline(time.Time{})
+	}
 
 	if err = p.send(request); err != nil {
-		err = errors.Wrap(err, "failed to send request")
+		err = errors.Wrapf(err, "send request (budget=%s)", budget)
 		goto err
 	}
 
 	if err = p.recv(response); err != nil {
-		err = errors.Wrap(err, "failed to receive response")
+		err = errors.Wrapf(err, "receive response (budget=%s)", budget)
 		goto err
 	}
 
@@ -91,14 +84,11 @@ func (p *Protocol) Interrupt(ctx context.Context, request *Message, response *Me
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	// Honor the ctx deadline, if present, or use a default.
-	deadline, ok := ctx.Deadline()
-	if !ok {
-		deadline = time.Now().Add(2 * time.Second)
+	// Honor the ctx deadline, if present.
+	if deadline, ok := ctx.Deadline(); ok {
+		p.conn.SetDeadline(deadline)
+		defer p.conn.SetDeadline(time.Time{})
 	}
-	p.conn.SetDeadline(deadline)
-
-	defer request.Reset()
 
 	EncodeInterrupt(request, 0)
 
@@ -108,12 +98,10 @@ func (p *Protocol) Interrupt(ctx context.Context, request *Message, response *Me
 
 	for {
 		if err := p.recv(response); err != nil {
-			response.Reset()
 			return errors.Wrap(err, "failed to receive response")
 		}
 
 		mtype, _ := response.getHeader()
-		response.Reset()
 
 		if mtype == ResponseEmpty {
 			break
@@ -155,7 +143,7 @@ func (p *Protocol) sendHeader(req *Message) error {
 }
 
 func (p *Protocol) sendBody(req *Message) error {
-	buf := req.body1.Bytes[:req.body1.Offset]
+	buf := req.body.Bytes[:req.body.Offset]
 	n, err := p.conn.Write(buf)
 	if err != nil {
 		return errors.Wrap(err, "failed to send static body")
@@ -165,24 +153,12 @@ func (p *Protocol) sendBody(req *Message) error {
 		return errors.Wrap(io.ErrShortWrite, "failed to write body")
 	}
 
-	if req.body2.Bytes == nil {
-		return nil
-	}
-
-	buf = req.body2.Bytes[:req.body2.Offset]
-	n, err = p.conn.Write(buf)
-	if err != nil {
-		return errors.Wrap(err, "failed to send dynamic body")
-	}
-
-	if n != len(buf) {
-		return errors.Wrap(io.ErrShortWrite, "failed to write body")
-	}
-
 	return nil
 }
 
 func (p *Protocol) recv(res *Message) error {
+	res.reset()
+
 	if err := p.recvHeader(res); err != nil {
 		return errors.Wrap(err, "failed to receive header")
 	}
@@ -209,30 +185,19 @@ func (p *Protocol) recvHeader(res *Message) error {
 
 func (p *Protocol) recvBody(res *Message) error {
 	n := int(res.words) * messageWordSize
-	n1 := n
-	n2 := 0
 
-	if n1 > len(res.body1.Bytes) {
-		// We need to allocate the dynamic buffer.
-		n1 = len(res.body1.Bytes)
-		n2 = n - n1
+	for n > len(res.body.Bytes) {
+		// Grow message buffer.
+		bytes := make([]byte, len(res.body.Bytes)*2)
+		res.body.Bytes = bytes
 	}
 
-	buf := res.body1.Bytes[:n1]
+	buf := res.body.Bytes[:n]
 
 	if err := p.recvPeek(buf); err != nil {
 		return errors.Wrap(err, "failed to read body")
 	}
 
-	if n2 > 0 {
-		res.body2.Bytes = make([]byte, n2)
-		res.body2.Offset = 0
-		buf = res.body2.Bytes
-		if err := p.recvPeek(buf); err != nil {
-			return errors.Wrap(err, "failed to read body")
-		}
-	}
-
 	return nil
 }
 

vendor/github.com/canonical/go-dqlite/internal/protocol/request.go

@@ -7,6 +7,7 @@ package protocol
 
 // EncodeLeader encodes a Leader request.
 func EncodeLeader(request *Message) {
+	request.reset()
 	request.putUint64(0)
 
 	request.putHeader(RequestLeader)
@@ -14,6 +15,7 @@ func EncodeLeader(request *Message) {
 
 // EncodeClient encodes a Client request.
 func EncodeClient(request *Message, id uint64) {
+	request.reset()
 	request.putUint64(id)
 
 	request.putHeader(RequestClient)
@@ -21,6 +23,7 @@ func EncodeClient(request *Message, id uint64) {
 
 // EncodeHeartbeat encodes a Heartbeat request.
 func EncodeHeartbeat(request *Message, timestamp uint64) {
+	request.reset()
 	request.putUint64(timestamp)
 
 	request.putHeader(RequestHeartbeat)
@@ -28,6 +31,7 @@ func EncodeHeartbeat(request *Message, timestamp uint64) {
 
 // EncodeOpen encodes a Open request.
 func EncodeOpen(request *Message, name string, flags uint64, vfs string) {
+	request.reset()
 	request.putString(name)
 	request.putUint64(flags)
 	request.putString(vfs)
@@ -37,6 +41,7 @@ func EncodeOpen(request *Message, name string, flags uint64, vfs string) {
 
 // EncodePrepare encodes a Prepare request.
 func EncodePrepare(request *Message, db uint64, sql string) {
+	request.reset()
 	request.putUint64(db)
 	request.putString(sql)
 
@@ -45,6 +50,7 @@ func EncodePrepare(request *Message, db uint64, sql string) {
 
 // EncodeExec encodes a Exec request.
 func EncodeExec(request *Message, db uint32, stmt uint32, values NamedValues) {
+	request.reset()
 	request.putUint32(db)
 	request.putUint32(stmt)
 	request.putNamedValues(values)
@@ -54,6 +60,7 @@ func EncodeExec(request *Message, db uint32, stmt uint32, values NamedValues) {
 
 // EncodeQuery encodes a Query request.
 func EncodeQuery(request *Message, db uint32, stmt uint32, values NamedValues) {
+	request.reset()
 	request.putUint32(db)
 	request.putUint32(stmt)
 	request.putNamedValues(values)
@@ -63,6 +70,7 @@ func EncodeQuery(request *Message, db uint32, stmt uint32, values NamedValues) {
 
 // EncodeFinalize encodes a Finalize request.
 func EncodeFinalize(request *Message, db uint32, stmt uint32) {
+	request.reset()
 	request.putUint32(db)
 	request.putUint32(stmt)
 
@@ -71,6 +79,7 @@ func EncodeFinalize(request *Message, db uint32, stmt uint32) {
 
 // EncodeExecSQL encodes a ExecSQL request.
 func EncodeExecSQL(request *Message, db uint64, sql string, values NamedValues) {
+	request.reset()
 	request.putUint64(db)
 	request.putString(sql)
 	request.putNamedValues(values)
@@ -80,6 +89,7 @@ func EncodeExecSQL(request *Message, db uint64, sql string, values NamedValues)
 
 // EncodeQuerySQL encodes a QuerySQL request.
 func EncodeQuerySQL(request *Message, db uint64, sql string, values NamedValues) {
+	request.reset()
 	request.putUint64(db)
 	request.putString(sql)
 	request.putNamedValues(values)
@@ -89,6 +99,7 @@ func EncodeQuerySQL(request *Message, db uint64, sql string, values NamedValues)
 
 // EncodeInterrupt encodes a Interrupt request.
 func EncodeInterrupt(request *Message, db uint64) {
+	request.reset()
 	request.putUint64(db)
 
 	request.putHeader(RequestInterrupt)
@@ -96,6 +107,7 @@ func EncodeInterrupt(request *Message, db uint64) {
 
 // EncodeAdd encodes a Add request.
 func EncodeAdd(request *Message, id uint64, address string) {
+	request.reset()
 	request.putUint64(id)
 	request.putString(address)
 
@@ -104,6 +116,7 @@ func EncodeAdd(request *Message, id uint64, address string) {
 
 // EncodeAssign encodes a Assign request.
 func EncodeAssign(request *Message, id uint64, role uint64) {
+	request.reset()
 	request.putUint64(id)
 	request.putUint64(role)
 
@@ -112,6 +125,7 @@ func EncodeAssign(request *Message, id uint64, role uint64) {
 
 // EncodeRemove encodes a Remove request.
 func EncodeRemove(request *Message, id uint64) {
+	request.reset()
 	request.putUint64(id)
 
 	request.putHeader(RequestRemove)
@@ -119,6 +133,7 @@ func EncodeRemove(request *Message, id uint64) {
 
 // EncodeDump encodes a Dump request.
 func EncodeDump(request *Message, name string) {
+	request.reset()
 	request.putString(name)
 
 	request.putHeader(RequestDump)
@@ -126,6 +141,7 @@ func EncodeDump(request *Message, name string) {
 
 // EncodeCluster encodes a Cluster request.
 func EncodeCluster(request *Message, format uint64) {
+	request.reset()
 	request.putUint64(format)
 
 	request.putHeader(RequestCluster)
@@ -133,6 +149,7 @@ func EncodeCluster(request *Message, format uint64) {
 
 // EncodeTransfer encodes a Transfer request.
 func EncodeTransfer(request *Message, id uint64) {
+	request.reset()
 	request.putUint64(id)
 
 	request.putHeader(RequestTransfer)

vendor/github.com/canonical/go-dqlite/internal/protocol/schema.sh

@@ -54,6 +54,7 @@ if [ "$entity" = "--request" ]; then
 
 // Encode${cmd} encodes a $cmd request.
 func Encode${cmd}(request *Message${args}) {
+	request.reset()
 EOF
 
     for i in "${@}"

vendor/github.com/canonical/go-dqlite/node.go

@@ -118,6 +118,16 @@ func (s *Node) Close() error {
 	return nil
 }
 
+// BootstrapID is a magic ID that should be used for the fist node in a
+// cluster. Alternatively ID 1 can be used as well.
+const BootstrapID = 0x2dc171858c3155be
+
+// GenerateID generates a unique ID for a new node, based on a hash of its
+// address and the current time.
+func GenerateID(address string) uint64 {
+	return bindings.GenerateID(address)
+}
+
 // Create a options object with sane defaults.
 func defaultOptions() *options {
 	return &options{

vendor/github.com/rancher/kine/pkg/drivers/mysql/mysql.go

@@ -35,6 +35,7 @@ var (
 			);`,
 	}
 	nameIdx     = "create index kine_name_index on kine (name)"
+	nameIDIdx   = "create index kine_name_id_index on kine (name,id)"
 	revisionIdx = "create unique index kine_name_prev_revision_uindex on kine (name, prev_revision)"
 	createDB    = "create database if not exists "
 )
@@ -87,6 +88,7 @@ func setup(db *sql.DB) error {
 	// check if duplicate indexes
 	indexes := []string{
 		nameIdx,
+		nameIDIdx,
 		revisionIdx}
 
 	for _, idx := range indexes {

vendor/github.com/rancher/kine/pkg/drivers/pgsql/pgsql.go

@@ -35,6 +35,7 @@ var (
  				old_value bytea
  			);`,
 		`CREATE INDEX IF NOT EXISTS kine_name_index ON kine (name)`,
+		`CREATE INDEX IF NOT EXISTS kine_name_id_index ON kine (name,id)`,
 		`CREATE UNIQUE INDEX IF NOT EXISTS kine_name_prev_revision_uindex ON kine (name, prev_revision)`,
 	}
 	createDB = "create database "

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

@@ -295,6 +295,7 @@ func (l *LogStructured) ttlEvents(ctx context.Context) chan *server.Event {
 
 func (l *LogStructured) ttl(ctx context.Context) {
 	// vary naive TTL support
+	mutex := &sync.Mutex{}
 	for event := range l.ttlEvents(ctx) {
 		go func(event *server.Event) {
 			select {
@@ -302,7 +303,9 @@ func (l *LogStructured) ttl(ctx context.Context) {
 				return
 			case <-time.After(time.Duration(event.KV.Lease) * time.Second):
 			}
+			mutex.Lock()
 			l.Delete(ctx, event.KV.Key, event.KV.ModRevision)
+			mutex.Unlock()
 		}(event)
 	}
 }

vendor/modules.txt

@@ -131,7 +131,7 @@ github.com/blang/semver
 github.com/bronze1man/goStrongswanVici
 # github.com/buger/jsonparser v0.0.0-20181115193947-bf1c66bbce23
 github.com/buger/jsonparser
-# github.com/canonical/go-dqlite v1.3.0
+# github.com/canonical/go-dqlite v1.5.1
 github.com/canonical/go-dqlite
 github.com/canonical/go-dqlite/client
 github.com/canonical/go-dqlite/driver
@@ -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.5
+# github.com/rancher/kine v0.4.0
 github.com/rancher/kine/pkg/broadcaster
 github.com/rancher/kine/pkg/client
 github.com/rancher/kine/pkg/drivers/dqlite