Add support for SQL driver

staging/src/k8s.io/apiserver/pkg/storage/storagebackend/config.go

@@ -25,6 +25,7 @@ import (
 
 const (
 	StorageTypeUnset = ""
+	StorageTypeKVSQL = "kvsql"
 	StorageTypeETCD3 = "etcd3"
 
 	DefaultCompactInterval = 5 * time.Minute

staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/factory.go

@@ -19,6 +19,7 @@ package factory
 import (
 	"fmt"
 
+	"github.com/ibuildthecloud/kvsql"
 	"k8s.io/apiserver/pkg/storage"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
 )
@@ -29,7 +30,13 @@ type DestroyFunc func()
 // Create creates a storage backend based on given config.
 func Create(c storagebackend.Config) (storage.Interface, DestroyFunc, error) {
 	switch c.Type {
-	case storagebackend.StorageTypeUnset, storagebackend.StorageTypeETCD3:
+	case storagebackend.StorageTypeUnset, storagebackend.StorageTypeKVSQL:
+		return factory.NewKVSQLStorage(c)
+	case storagebackend.StorageTypeETCD3:
+		// TODO: We have the following features to implement:
+		// - Support secure connection by using key, cert, and CA files.
+		// - Honor "https" scheme to support secure connection in gRPC.
+		// - Support non-quorum read.
 		return newETCD3Storage(c)
 	default:
 		return nil, nil, fmt.Errorf("unknown storage type: %s", c.Type)
@@ -39,7 +46,9 @@ func Create(c storagebackend.Config) (storage.Interface, DestroyFunc, error) {
 // CreateHealthCheck creates a healthcheck function based on given config.
 func CreateHealthCheck(c storagebackend.Config) (func() error, error) {
 	switch c.Type {
-	case storagebackend.StorageTypeUnset, storagebackend.StorageTypeETCD3:
+	case storagebackend.StorageTypeUnset, storagebackend.StorageTypeKVSQL:
+		return factory.NewKVSQLHealthCheck(c)
+	case storagebackend.StorageTypeETCD3:
 		return newETCD3HealthCheck(c)
 	default:
 		return nil, fmt.Errorf("unknown storage type: %s", c.Type)

vendor/github.com/docker/docker/pkg/locker/README.md

+Locker
+=====
+
+locker provides a mechanism for creating finer-grained locking to help
+free up more global locks to handle other tasks.
+
+The implementation looks close to a sync.Mutex, however, the user must provide a
+reference to use to refer to the underlying lock when locking and unlocking,
+and unlock may generate an error.
+
+If a lock with a given name does not exist when `Lock` is called, one is
+created.
+Lock references are automatically cleaned up on `Unlock` if nothing else is
+waiting for the lock.
+
+
+## Usage
+
+```go
+package important
+
+import (
+	"sync"
+	"time"
+
+	"github.com/docker/docker/pkg/locker"
+)
+
+type important struct {
+	locks *locker.Locker
+	data  map[string]interface{}
+	mu    sync.Mutex
+}
+
+func (i *important) Get(name string) interface{} {
+	i.locks.Lock(name)
+	defer i.locks.Unlock(name)
+	return i.data[name]
+}
+
+func (i *important) Create(name string, data interface{}) {
+	i.locks.Lock(name)
+	defer i.locks.Unlock(name)
+
+	i.createImportant(data)
+
+	i.mu.Lock()
+	i.data[name] = data
+	i.mu.Unlock()
+}
+
+func (i *important) createImportant(data interface{}) {
+	time.Sleep(10 * time.Second)
+}
+```
+
+For functions dealing with a given name, always lock at the beginning of the
+function (or before doing anything with the underlying state), this ensures any
+other function that is dealing with the same name will block.
+
+When needing to modify the underlying data, use the global lock to ensure nothing
+else is modifying it at the same time.
+Since name lock is already in place, no reads will occur while the modification
+is being performed.
+

vendor/github.com/docker/docker/pkg/locker/locker.go

+/*
+Package locker provides a mechanism for creating finer-grained locking to help
+free up more global locks to handle other tasks.
+
+The implementation looks close to a sync.Mutex, however the user must provide a
+reference to use to refer to the underlying lock when locking and unlocking,
+and unlock may generate an error.
+
+If a lock with a given name does not exist when `Lock` is called, one is
+created.
+Lock references are automatically cleaned up on `Unlock` if nothing else is
+waiting for the lock.
+*/
+package locker // import "github.com/docker/docker/pkg/locker"
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+)
+
+// ErrNoSuchLock is returned when the requested lock does not exist
+var ErrNoSuchLock = errors.New("no such lock")
+
+// Locker provides a locking mechanism based on the passed in reference name
+type Locker struct {
+	mu    sync.Mutex
+	locks map[string]*lockCtr
+}
+
+// lockCtr is used by Locker to represent a lock with a given name.
+type lockCtr struct {
+	mu sync.Mutex
+	// waiters is the number of waiters waiting to acquire the lock
+	// this is int32 instead of uint32 so we can add `-1` in `dec()`
+	waiters int32
+}
+
+// inc increments the number of waiters waiting for the lock
+func (l *lockCtr) inc() {
+	atomic.AddInt32(&l.waiters, 1)
+}
+
+// dec decrements the number of waiters waiting on the lock
+func (l *lockCtr) dec() {
+	atomic.AddInt32(&l.waiters, -1)
+}
+
+// count gets the current number of waiters
+func (l *lockCtr) count() int32 {
+	return atomic.LoadInt32(&l.waiters)
+}
+
+// Lock locks the mutex
+func (l *lockCtr) Lock() {
+	l.mu.Lock()
+}
+
+// Unlock unlocks the mutex
+func (l *lockCtr) Unlock() {
+	l.mu.Unlock()
+}
+
+// New creates a new Locker
+func New() *Locker {
+	return &Locker{
+		locks: make(map[string]*lockCtr),
+	}
+}
+
+// Lock locks a mutex with the given name. If it doesn't exist, one is created
+func (l *Locker) Lock(name string) {
+	l.mu.Lock()
+	if l.locks == nil {
+		l.locks = make(map[string]*lockCtr)
+	}
+
+	nameLock, exists := l.locks[name]
+	if !exists {
+		nameLock = &lockCtr{}
+		l.locks[name] = nameLock
+	}
+
+	// increment the nameLock waiters while inside the main mutex
+	// this makes sure that the lock isn't deleted if `Lock` and `Unlock` are called concurrently
+	nameLock.inc()
+	l.mu.Unlock()
+
+	// Lock the nameLock outside the main mutex so we don't block other operations
+	// once locked then we can decrement the number of waiters for this lock
+	nameLock.Lock()
+	nameLock.dec()
+}
+
+// Unlock unlocks the mutex with the given name
+// If the given lock is not being waited on by any other callers, it is deleted
+func (l *Locker) Unlock(name string) error {
+	l.mu.Lock()
+	nameLock, exists := l.locks[name]
+	if !exists {
+		l.mu.Unlock()
+		return ErrNoSuchLock
+	}
+
+	if nameLock.count() == 0 {
+		delete(l.locks, name)
+	}
+	nameLock.Unlock()
+
+	l.mu.Unlock()
+	return nil
+}

vendor/github.com/docker/docker/pkg/locker/locker_test.go

+package locker // import "github.com/docker/docker/pkg/locker"
+
+import (
+	"math/rand"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestLockCounter(t *testing.T) {
+	l := &lockCtr{}
+	l.inc()
+
+	if l.waiters != 1 {
+		t.Fatal("counter inc failed")
+	}
+
+	l.dec()
+	if l.waiters != 0 {
+		t.Fatal("counter dec failed")
+	}
+}
+
+func TestLockerLock(t *testing.T) {
+	l := New()
+	l.Lock("test")
+	ctr := l.locks["test"]
+
+	if ctr.count() != 0 {
+		t.Fatalf("expected waiters to be 0, got :%d", ctr.waiters)
+	}
+
+	chDone := make(chan struct{})
+	go func() {
+		l.Lock("test")
+		close(chDone)
+	}()
+
+	chWaiting := make(chan struct{})
+	go func() {
+		for range time.Tick(1 * time.Millisecond) {
+			if ctr.count() == 1 {
+				close(chWaiting)
+				break
+			}
+		}
+	}()
+
+	select {
+	case <-chWaiting:
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for lock waiters to be incremented")
+	}
+
+	select {
+	case <-chDone:
+		t.Fatal("lock should not have returned while it was still held")
+	default:
+	}
+
+	if err := l.Unlock("test"); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case <-chDone:
+	case <-time.After(3 * time.Second):
+		t.Fatalf("lock should have completed")
+	}
+
+	if ctr.count() != 0 {
+		t.Fatalf("expected waiters to be 0, got: %d", ctr.count())
+	}
+}
+
+func TestLockerUnlock(t *testing.T) {
+	l := New()
+
+	l.Lock("test")
+	l.Unlock("test")
+
+	chDone := make(chan struct{})
+	go func() {
+		l.Lock("test")
+		close(chDone)
+	}()
+
+	select {
+	case <-chDone:
+	case <-time.After(3 * time.Second):
+		t.Fatalf("lock should not be blocked")
+	}
+}
+
+func TestLockerConcurrency(t *testing.T) {
+	l := New()
+
+	var wg sync.WaitGroup
+	for i := 0; i <= 10000; i++ {
+		wg.Add(1)
+		go func() {
+			l.Lock("test")
+			// if there is a concurrency issue, will very likely panic here
+			l.Unlock("test")
+			wg.Done()
+		}()
+	}
+
+	chDone := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(chDone)
+	}()
+
+	select {
+	case <-chDone:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timeout waiting for locks to complete")
+	}
+
+	// Since everything has unlocked this should not exist anymore
+	if ctr, exists := l.locks["test"]; exists {
+		t.Fatalf("lock should not exist: %v", ctr)
+	}
+}
+
+func BenchmarkLocker(b *testing.B) {
+	l := New()
+	for i := 0; i < b.N; i++ {
+		l.Lock("test")
+		l.Unlock("test")
+	}
+}
+
+func BenchmarkLockerParallel(b *testing.B) {
+	l := New()
+	b.SetParallelism(128)
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			l.Lock("test")
+			l.Unlock("test")
+		}
+	})
+}
+
+func BenchmarkLockerMoreKeys(b *testing.B) {
+	l := New()
+	var keys []string
+	for i := 0; i < 64; i++ {
+		keys = append(keys, strconv.Itoa(i))
+	}
+	b.SetParallelism(128)
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			k := keys[rand.Intn(len(keys))]
+			l.Lock(k)
+			l.Unlock(k)
+		}
+	})
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/client.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	"google.golang.org/grpc"
+)
+
+// Client provides and manages an etcd v3 client session.
+type Client struct {
+	Cluster
+	KV
+	Lease
+	Watcher
+	callOpts []grpc.CallOption
+}
+
+// New creates a new etcdv3 client from a given configuration.
+func New(cfg Config) (*Client, error) {
+	c := &Client{
+		Lease: &lessor{},
+	}
+	kv, err := newKV(cfg)
+	if err != nil {
+		return nil, err
+	}
+	c.KV = kv
+	c.Watcher = kv
+	return c, nil
+}
+
+// Close shuts down the client's etcd connections.
+func (c *Client) Close() error {
+	if c.Watcher != nil {
+		return c.Watcher.Close()
+	}
+	return nil
+}
+
+// Endpoints lists the registered endpoints for the client.
+func (c *Client) Endpoints() (eps []string) {
+	return
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/cluster.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	"golang.org/x/net/context"
+)
+
+type Cluster struct {
+}
+
+func (c *Cluster) MemberList(ctx context.Context) (interface{}, error) {
+	return nil, nil
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/compact_op.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+// CompactOp represents a compact operation.
+type CompactOp struct {
+	revision int64
+}
+
+// CompactOption configures compact operation.
+type CompactOption func(*CompactOp)
+
+func (op *CompactOp) applyCompactOpts(opts []CompactOption) {
+	for _, opt := range opts {
+		opt(op)
+	}
+}
+
+// OpCompact wraps slice CompactOption to create a CompactOp.
+func OpCompact(rev int64, opts ...CompactOption) CompactOp {
+	ret := CompactOp{revision: rev}
+	ret.applyCompactOpts(opts)
+	return ret
+}
+
+func (op CompactOp) toRequest() *pb.CompactionRequest {
+	return &pb.CompactionRequest{Revision: op.revision}
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/compare.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+type CompareTarget int
+type CompareResult int
+
+const (
+	CompareVersion CompareTarget = iota
+	CompareCreated
+	CompareModified
+	CompareValue
+)
+
+type Cmp pb.Compare
+
+func Compare(cmp Cmp, result string, v interface{}) Cmp {
+	var r pb.Compare_CompareResult
+
+	switch result {
+	case "=":
+		r = pb.Compare_EQUAL
+	case "!=":
+		r = pb.Compare_NOT_EQUAL
+	case ">":
+		r = pb.Compare_GREATER
+	case "<":
+		r = pb.Compare_LESS
+	default:
+		panic("Unknown result op")
+	}
+
+	cmp.Result = r
+	switch cmp.Target {
+	case pb.Compare_VALUE:
+		val, ok := v.(string)
+		if !ok {
+			panic("bad compare value")
+		}
+		cmp.TargetUnion = &pb.Compare_Value{Value: []byte(val)}
+	case pb.Compare_VERSION:
+		cmp.TargetUnion = &pb.Compare_Version{Version: mustInt64(v)}
+	case pb.Compare_CREATE:
+		cmp.TargetUnion = &pb.Compare_CreateRevision{CreateRevision: mustInt64(v)}
+	case pb.Compare_MOD:
+		cmp.TargetUnion = &pb.Compare_ModRevision{ModRevision: mustInt64(v)}
+	default:
+		panic("Unknown compare type")
+	}
+	return cmp
+}
+
+func Version(key string) Cmp {
+	return Cmp{Key: []byte(key), Target: pb.Compare_VERSION}
+}
+
+func ModRevision(key string) Cmp {
+	return Cmp{Key: []byte(key), Target: pb.Compare_MOD}
+}
+
+// mustInt64 panics if val isn't an int or int64. It returns an int64 otherwise.
+func mustInt64(val interface{}) int64 {
+	if v, ok := val.(int64); ok {
+		return v
+	}
+	if v, ok := val.(int); ok {
+		return int64(v)
+	}
+	panic("bad value")
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/config.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	"crypto/tls"
+	"time"
+
+	"google.golang.org/grpc"
+)
+
+type Config struct {
+	// Endpoints is a list of URLs.
+	Endpoints []string `json:"endpoints"`
+
+	// DialKeepAliveTime is the time in seconds after which client pings the server to see if
+	// transport is alive.
+	DialKeepAliveTime time.Duration `json:"dial-keep-alive-time"`
+
+	// DialKeepAliveTimeout is the time in seconds that the client waits for a response for the
+	// keep-alive probe.  If the response is not received in this time, the connection is closed.
+	DialKeepAliveTimeout time.Duration `json:"dial-keep-alive-timeout"`
+
+	// TLS holds the client secure credentials, if any.
+	TLS *tls.Config
+
+	DialTimeout time.Duration
+
+	DialOptions []grpc.DialOption
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/driver/client.go

+package driver
+
+import (
+	"errors"
+)
+
+var (
+	ErrExists        = errors.New("key exists")
+	ErrNotExists     = errors.New("key and or Revision does not exists")
+	ErrRevisionMatch = errors.New("revision does not match")
+)
+
+type KeyValue struct {
+	ID             int64
+	Key            string
+	Value          []byte
+	OldValue       []byte
+	OldRevision    int64
+	CreateRevision int64
+	Revision       int64
+	TTL            int64
+	Version        int64
+	Del            int64
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/driver/driver.go

+package driver
+
+import (
+	"context"
+)
+
+type Driver interface {
+	List(ctx context.Context, revision, limit int64, rangeKey, startKey string) (kvs []*KeyValue, listRevision int64, err error)
+
+	Delete(ctx context.Context, key string, revision int64) ([]*KeyValue, error)
+
+	// Update should return ErrNotExist when the key does not exist and ErrRevisionMatch when revision doesn't match
+	Update(ctx context.Context, key string, value []byte, revision, ttl int64) (oldKv *KeyValue, newKv *KeyValue, err error)
+
+	Watch(ctx context.Context, key string, revision int64) <-chan Event
+
+	Close() error
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/driver/generic.go

+package driver
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/ibuildthecloud/kvsql/pkg/broadcast"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	utiltrace "k8s.io/utils/trace"
+)
+
+type Generic struct {
+	db *sql.DB
+
+	CleanupSQL      string
+	GetSQL          string
+	ListSQL         string
+	ListRevisionSQL string
+	ListResumeSQL   string
+	ReplaySQL       string
+	InsertSQL       string
+	GetRevisionSQL  string
+	ToDeleteSQL     string
+	DeleteOldSQL    string
+	revision        int64
+
+	changes     chan *KeyValue
+	broadcaster broadcast.Broadcaster
+	cancel      func()
+}
+
+func (g *Generic) Start(ctx context.Context, db *sql.DB) error {
+	g.db = db
+	g.changes = make(chan *KeyValue, 1024)
+
+	row := db.QueryRowContext(ctx, g.GetRevisionSQL)
+	rev := sql.NullInt64{}
+	if err := row.Scan(&rev); err != nil {
+		return errors.Wrap(err, "Failed to initialize revision")
+	}
+	if rev.Int64 == 0 {
+		g.revision = 1
+	} else {
+		g.revision = rev.Int64
+	}
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(time.Minute):
+				_, err := g.ExecContext(ctx, g.CleanupSQL, time.Now().Unix())
+				if err != nil {
+					logrus.Errorf("Failed to purge expired TTL entries")
+				}
+
+				err = g.cleanup(ctx)
+				if err != nil {
+					logrus.Errorf("Failed to cleanup duplicate entries")
+				}
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (g *Generic) cleanup(ctx context.Context) error {
+	rows, err := g.QueryContext(ctx, g.ToDeleteSQL)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	toDelete := map[string]int64{}
+	for rows.Next() {
+		var (
+			count, revision int64
+			name            string
+		)
+		err := rows.Scan(&count, &name, &revision)
+		if err != nil {
+			return err
+		}
+		toDelete[name] = revision
+	}
+
+	rows.Close()
+
+	for name, rev := range toDelete {
+		_, err = g.ExecContext(ctx, g.DeleteOldSQL, name, rev, rev)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (g *Generic) Get(ctx context.Context, key string) (*KeyValue, error) {
+	kvs, _, err := g.List(ctx, 0, 1, key, "")
+	if err != nil {
+		return nil, err
+	}
+	if len(kvs) > 0 {
+		return kvs[0], nil
+	}
+	return nil, nil
+}
+
+func (g *Generic) replayEvents(ctx context.Context, key string, revision int64) ([]*KeyValue, error) {
+	rows, err := g.QueryContext(ctx, g.ReplaySQL, key, revision)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var resp []*KeyValue
+	for rows.Next() {
+		value := KeyValue{}
+		if err := scan(rows.Scan, &value); err != nil {
+			return nil, err
+		}
+		resp = append(resp, &value)
+	}
+
+	return resp, nil
+}
+
+func (g *Generic) List(ctx context.Context, revision, limit int64, rangeKey, startKey string) ([]*KeyValue, int64, error) {
+	var (
+		rows *sql.Rows
+		err  error
+	)
+
+	if limit == 0 {
+		limit = 1000000
+	} else {
+		limit = limit + 1
+	}
+
+	listRevision := atomic.LoadInt64(&g.revision)
+	if !strings.HasSuffix(rangeKey, "%") && revision <= 0 {
+		rows, err = g.QueryContext(ctx, g.GetSQL, rangeKey, limit)
+	} else if revision <= 0 {
+		rows, err = g.QueryContext(ctx, g.ListSQL, rangeKey, limit)
+	} else if len(startKey) > 0 {
+		listRevision = revision
+		rows, err = g.QueryContext(ctx, g.ListResumeSQL, revision, rangeKey, startKey, limit)
+	} else {
+		rows, err = g.QueryContext(ctx, g.ListRevisionSQL, revision, rangeKey, limit)
+	}
+
+	if err != nil {
+		return nil, 0, err
+	}
+	defer rows.Close()
+
+	var resp []*KeyValue
+	for rows.Next() {
+		value := KeyValue{}
+		if err := scan(rows.Scan, &value); err != nil {
+			return nil, 0, err
+		}
+		if value.Revision > listRevision {
+			listRevision = value.Revision
+		}
+		if value.Del == 0 {
+			resp = append(resp, &value)
+		}
+	}
+
+	return resp, listRevision, nil
+}
+
+func (g *Generic) Delete(ctx context.Context, key string, revision int64) ([]*KeyValue, error) {
+	if strings.HasSuffix(key, "%") {
+		panic("can not delete list revision")
+	}
+
+	_, err := g.mod(ctx, true, key, []byte{}, revision, 0)
+	return nil, err
+}
+
+func (g *Generic) Update(ctx context.Context, key string, value []byte, revision, ttl int64) (*KeyValue, *KeyValue, error) {
+	kv, err := g.mod(ctx, false, key, value, revision, ttl)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if kv.Version == 1 {
+		return nil, kv, nil
+	}
+
+	oldKv := *kv
+	oldKv.Revision = oldKv.OldRevision
+	oldKv.Value = oldKv.OldValue
+	return &oldKv, kv, nil
+}
+
+func (g *Generic) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
+	trace := utiltrace.New(fmt.Sprintf("SQL DB ExecContext query: %s keys: %v", query, args))
+	defer trace.LogIfLong(500 * time.Millisecond)
+
+	return g.db.ExecContext(ctx, query, args...)
+}
+
+func (g *Generic) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
+	trace := utiltrace.New(fmt.Sprintf("SQL DB QueryContext query: %s keys: %v", query, args))
+	defer trace.LogIfLong(500 * time.Millisecond)
+
+	return g.db.QueryContext(ctx, query, args...)
+}
+
+func (g *Generic) mod(ctx context.Context, delete bool, key string, value []byte, revision int64, ttl int64) (*KeyValue, error) {
+	oldKv, err := g.Get(ctx, key)
+	if err != nil {
+		return nil, err
+	}
+
+	if revision > 0 && oldKv == nil {
+		return nil, ErrNotExists
+	}
+
+	if revision > 0 && oldKv.Revision != revision {
+		return nil, ErrRevisionMatch
+	}
+
+	if ttl > 0 {
+		ttl = int64(time.Now().Unix()) + ttl
+	}
+
+	newRevision := atomic.AddInt64(&g.revision, 1)
+	result := &KeyValue{
+		Key:            key,
+		Value:          value,
+		Revision:       newRevision,
+		TTL:            int64(ttl),
+		CreateRevision: newRevision,
+		Version:        1,
+	}
+	if oldKv != nil {
+		result.OldRevision = oldKv.Revision
+		result.OldValue = oldKv.Value
+		result.TTL = oldKv.TTL
+		result.CreateRevision = oldKv.CreateRevision
+		result.Version = oldKv.Version + 1
+	}
+
+	if delete {
+		result.Del = 1
+	}
+
+	_, err = g.ExecContext(ctx, g.InsertSQL,
+		result.Key,
+		result.Value,
+		result.OldValue,
+		result.OldRevision,
+		result.CreateRevision,
+		result.Revision,
+		result.TTL,
+		result.Version,
+		result.Del,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	g.changes <- result
+	return result, nil
+}
+
+type scanner func(dest ...interface{}) error
+
+func scan(s scanner, out *KeyValue) error {
+	return s(
+		&out.ID,
+		&out.Key,
+		&out.Value,
+		&out.OldValue,
+		&out.OldRevision,
+		&out.CreateRevision,
+		&out.Revision,
+		&out.TTL,
+		&out.Version,
+		&out.Del)
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/driver/mysql/sqlite.go

+package sqlite
+
+import (
+	"database/sql"
+	"github.com/ibuildthecloud/kvsql/clientv3/driver"
+	"github.com/ibuildthecloud/kvsql/clientv3/driver/sqlite"
+)
+
+var (
+	schema = []string{
+		`create table if not exists key_value
+			(
+				name int not null,
+				value MEDIUMTEXT not null,
+				create_revision int not null,
+				revision int not null,
+				ttl int not null,
+				version int not null,
+				del int not null,
+				old_value MEDIUMTEXT not null,
+				id int auto_increment,
+				old_revision int not null,
+				constraint key_value_pk
+				primary key (id)
+			)`,
+	}
+
+	idx = []string{
+		"create index key_value__name_idx on key_value (name)",
+		"create index key_value__revision_idx on key_value (revision)",
+	}
+)
+
+func NewMYSQL() *driver.Generic {
+	return sqlite.NewSQLite()
+}
+
+func Open(dataSourceName string) (*sql.DB, error) {
+	db, err := sql.Open("mysql", dataSourceName)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, stmt := range schema {
+		_, err := db.Exec(stmt)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	for _, stmt := range idx {
+		db.Exec(stmt)
+	}
+
+	return db, nil
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/driver/sqlite/sqlite.go

+package sqlite
+
+import (
+	"database/sql"
+	"strings"
+
+	"github.com/ibuildthecloud/kvsql/clientv3/driver"
+)
+
+var (
+	fieldList = "name, value, old_value, old_revision, create_revision, revision, ttl, version, del"
+	baseList  = `
+SELECT kv.id, kv.name, kv.value, kv.old_value, kv.old_revision, kv.create_revision, kv.revision, kv.ttl, kv.version, kv.del
+FROM key_value kv
+  INNER JOIN
+    (
+      SELECT MAX(revision) revision, kvi.name
+      FROM key_value kvi
+		%REV%
+        GROUP BY kvi.name
+    ) AS r
+    ON r.name = kv.name AND r.revision = kv.revision
+WHERE kv.name like ? %RES% ORDER BY kv.name ASC limit ?
+`
+	insertSQL = `
+INSERT INTO key_value(` + fieldList + `)
+   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`
+
+	schema = []string{
+		`create table if not exists key_value
+			(
+				name INTEGER,
+				value BLOB,
+				create_revision INTEGER,
+				revision INTEGER,
+				ttl INTEGER,
+				version INTEGER,
+				del INTEGER,
+				old_value BLOB,
+				id INTEGER primary key autoincrement,
+				old_revision INTEGER
+			)`,
+		`create index if not exists name_idx on key_value (name)`,
+		`create index if not exists revision_idx on key_value (revision)`,
+	}
+)
+
+func NewSQLite() *driver.Generic {
+	return &driver.Generic{
+		CleanupSQL:      "DELETE FROM key_value WHERE ttl > 0 AND ttl < ?",
+		GetSQL:          "SELECT id, " + fieldList + " FROM key_value WHERE name = ? ORDER BY revision DESC limit ?",
+		ListSQL:         strings.Replace(strings.Replace(baseList, "%REV%", "", -1), "%RES%", "", -1),
+		ListRevisionSQL: strings.Replace(strings.Replace(baseList, "%REV%", "WHERE kvi.revision >= ?", -1), "%RES%", "", -1),
+		ListResumeSQL: strings.Replace(strings.Replace(baseList, "%REV%", "WHERE kvi.revision <= ?", -1),
+			"%RES%", "and kv.name > ? ", -1),
+		InsertSQL:      insertSQL,
+		ReplaySQL:      "SELECT id, " + fieldList + " FROM key_value WHERE name like ? and revision > ? ORDER BY revision ASC",
+		GetRevisionSQL: "SELECT MAX(revision) FROM key_value",
+		ToDeleteSQL:    "SELECT count(*) c, name, max(revision) FROM key_value GROUP BY name HAVING c > 1 or (c = 1 and del = 1)",
+		DeleteOldSQL:   "DELETE FROM key_value WHERE name = ? AND (revision < ? OR (revision = ? AND del = 1))",
+	}
+}
+
+func Open(dataSourceName string) (*sql.DB, error) {
+	if dataSourceName == "" {
+		dataSourceName = "./state.db?_journal=WAL&cache=shared"
+	}
+	db, err := sql.Open("sqlite3", dataSourceName)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, stmt := range schema {
+		_, err := db.Exec(stmt)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return db, nil
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/driver/watcher.go

+package driver
+
+import (
+	"context"
+	"io"
+	"strings"
+)
+
+type Event struct {
+	KV    *KeyValue
+	Err   error
+	Start bool
+}
+
+func matchesKey(prefix bool, key string, kv *KeyValue) bool {
+	if kv == nil {
+		return false
+	}
+	if prefix {
+		return strings.HasPrefix(kv.Key, key[:len(key)-1])
+	}
+	return kv.Key == key
+}
+
+func (g *Generic) globalWatcher() (chan map[string]interface{}, error) {
+	ctx, cancel := context.WithCancel(context.Background())
+	g.cancel = cancel
+	result := make(chan map[string]interface{}, 100)
+
+	go func() {
+		defer close(result)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case e := <-g.changes:
+				result <- map[string]interface{}{
+					"data": e,
+				}
+			}
+		}
+	}()
+
+	return result, nil
+}
+
+func (g *Generic) Watch(ctx context.Context, key string, revision int64) <-chan Event {
+	ctx, parentCancel := context.WithCancel(ctx)
+
+	prefix := strings.HasSuffix(key, "%")
+
+	events, err := g.broadcaster.Subscribe(ctx, g.globalWatcher)
+	if err != nil {
+		panic(err)
+	}
+
+	watchChan := make(chan Event)
+	go func() (returnErr error) {
+		defer func() {
+			sendErrorAndClose(watchChan, returnErr)
+			parentCancel()
+		}()
+
+		start(watchChan)
+
+		if revision > 0 {
+			keys, err := g.replayEvents(ctx, key, revision)
+			if err != nil {
+				return err
+			}
+
+			for _, k := range keys {
+				watchChan <- Event{KV: k}
+			}
+		}
+
+		for e := range events {
+			k, ok := e["data"].(*KeyValue)
+			if ok && matchesKey(prefix, key, k) {
+				watchChan <- Event{KV: k}
+			}
+		}
+
+		return nil
+	}()
+
+	return watchChan
+}
+
+func start(watchResponses chan Event) {
+	watchResponses <- Event{
+		Start: true,
+	}
+}
+
+func sendErrorAndClose(watchResponses chan Event, err error) {
+	if err == nil {
+		err = io.EOF
+	}
+	watchResponses <- Event{Err: err}
+	close(watchResponses)
+}
+
+// Close closes the watcher and cancels all watch requests.
+func (g *Generic) Close() error {
+	if g.cancel != nil {
+		g.cancel()
+		g.cancel = nil
+	}
+	return nil
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/kv.go

+// Copyright 2015 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	"bytes"
+	"database/sql"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/ibuildthecloud/kvsql/clientv3/driver"
+	"github.com/ibuildthecloud/kvsql/clientv3/driver/sqlite"
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+	"github.com/coreos/etcd/mvcc/mvccpb"
+	"github.com/docker/docker/pkg/locker"
+	"golang.org/x/net/context"
+)
+
+type (
+	CompactResponse pb.CompactionResponse
+	PutResponse     pb.PutResponse
+	GetResponse     pb.RangeResponse
+	DeleteResponse  pb.DeleteRangeResponse
+	TxnResponse     pb.TxnResponse
+)
+
+var (
+	connections map[string]*kv
+	connectionsCtx context.Context
+	CloseDB func()
+	connectionsLock sync.Mutex
+)
+
+type KV interface {
+	// Put puts a key-value pair into etcd.
+	// Note that key,value can be plain bytes array and string is
+	// an immutable representation of that bytes array.
+	// To get a string of bytes, do string([]byte{0x10, 0x20}).
+	Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error)
+
+	// Get retrieves keys.
+	// By default, Get will return the value for "key", if any.
+	// When passed WithRange(end), Get will return the keys in the range [key, end).
+	// When passed WithFromKey(), Get returns keys greater than or equal to key.
+	// When passed WithRev(rev) with rev > 0, Get retrieves keys at the given revision;
+	// if the required revision is compacted, the request will fail with ErrCompacted .
+	// When passed WithLimit(limit), the number of returned keys is bounded by limit.
+	// When passed WithSort(), the keys will be sorted.
+	Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error)
+
+	// Delete deletes a key, or optionally using WithRange(end), [key, end).
+	Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error)
+
+	// Compact compacts etcd KV history before the given rev.
+	Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error)
+
+	// Txn creates a transaction.
+	Txn(ctx context.Context) Txn
+}
+
+type kv struct {
+	l locker.Locker
+	d driver.Driver
+}
+
+func newKV(cfg Config) (*kv, error) {
+	connectionsLock.Lock()
+	defer connectionsLock.Unlock()
+
+	if len(cfg.Endpoints) != 1 {
+		return nil, fmt.Errorf("exactly one endpoint required for DB setting, got %v", cfg.Endpoints)
+	}
+
+	key := cfg.Endpoints[0]
+
+	if kv, ok := connections[key]; ok {
+		return kv, nil
+	}
+
+	if connections == nil {
+		connections = map[string]*kv{}
+		connectionsCtx, CloseDB = context.WithCancel(context.Background())
+	}
+
+	parts := strings.SplitN(key, "://", 2)
+	if len(parts) != 2 {
+		return nil, fmt.Errorf("invalid kvsql string")
+	}
+
+	var (
+		db *sql.DB
+		driver *driver.Generic
+		err error
+	)
+
+	switch parts[0] {
+	case "sqlite":
+		if db, err = sqlite.Open(parts[1]); err != nil {
+			return nil, err
+		}
+		driver = sqlite.NewSQLite()
+	}
+
+	if err := driver.Start(context.TODO(), db); err != nil {
+		db.Close()
+		return nil, err
+	}
+
+	kv := &kv{
+		d:driver,
+	}
+	connections[key] = kv
+
+	return  kv, nil
+}
+
+func (k *kv) Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error) {
+	//trace := utiltrace.New(fmt.Sprintf("SQL Put key: %s", key))
+	//defer trace.LogIfLong(500 * time.Millisecond)
+	k.l.Lock(key)
+	defer k.l.Unlock(key)
+
+	op := OpPut(key, val, opts...)
+	return k.opPut(ctx, op)
+}
+
+func (k *kv) opPut(ctx context.Context, op Op) (*PutResponse, error) {
+	oldR, r, err := k.d.Update(ctx, op.key, op.val, op.rev, int64(op.leaseID))
+	if err != nil {
+		return nil, err
+	}
+	return getPutResponse(oldR, r), nil
+}
+
+func (k *kv) Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error) {
+	//trace := utiltrace.New(fmt.Sprintf("SQL Get key: %s", key))
+	//defer trace.LogIfLong(500 * time.Millisecond)
+	op := OpGet(key, opts...)
+	return k.opGet(ctx, op)
+}
+
+func (k *kv) opGet(ctx context.Context, op Op) (*GetResponse, error) {
+	var (
+		rangeKey string
+		startKey string
+	)
+
+	if op.boundingKey == "" {
+		rangeKey = op.key
+		startKey = ""
+	} else {
+		rangeKey = op.boundingKey
+		startKey = string(bytes.SplitN([]byte(op.key), []byte{'\x00'}, -1)[0])
+	}
+
+	kvs, rev, err := k.d.List(ctx, op.rev, op.limit, rangeKey, startKey)
+	if err != nil {
+		return nil, err
+	}
+
+	return getResponse(kvs, rev, op.limit, op.countOnly), nil
+}
+
+func getPutResponse(oldValue *driver.KeyValue, value *driver.KeyValue) *PutResponse {
+	return &PutResponse{
+		Header: &pb.ResponseHeader{
+			Revision: value.Revision,
+		},
+		PrevKv: toKeyValue(oldValue),
+	}
+}
+
+func toKeyValue(v *driver.KeyValue) *mvccpb.KeyValue {
+	if v == nil {
+		return nil
+	}
+
+	return &mvccpb.KeyValue{
+		Key:            []byte(v.Key),
+		CreateRevision: v.CreateRevision,
+		ModRevision:    v.Revision,
+		Version:        v.Version,
+		Value:          v.Value,
+		Lease:          v.TTL,
+	}
+}
+
+func getDeleteResponse(values []*driver.KeyValue) *DeleteResponse {
+	gr := getResponse(values, 0, 0, false)
+	return &DeleteResponse{
+		Header: &pb.ResponseHeader{
+			Revision: gr.Header.Revision,
+		},
+		PrevKvs: gr.Kvs,
+	}
+}
+
+func getResponse(values []*driver.KeyValue, revision, limit int64, count bool) *GetResponse {
+	gr := &GetResponse{
+		Header: &pb.ResponseHeader{
+			Revision: revision,
+		},
+	}
+
+	for _, v := range values {
+		kv := toKeyValue(v)
+		if kv.ModRevision > gr.Header.Revision {
+			gr.Header.Revision = kv.ModRevision
+		}
+
+		gr.Kvs = append(gr.Kvs, kv)
+	}
+
+	gr.Count = int64(len(gr.Kvs))
+	if limit > 0 && gr.Count > limit {
+		gr.Kvs = gr.Kvs[:limit]
+		gr.More = true
+	}
+
+	if count {
+		gr.Kvs = nil
+	}
+
+	return gr
+}
+
+func (k *kv) Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error) {
+	//trace := utiltrace.New(fmt.Sprintf("SQL Delete key: %s", key))
+	//defer trace.LogIfLong(500 * time.Millisecond)
+	k.l.Lock(key)
+	defer k.l.Unlock(key)
+
+	op := OpDelete(key, opts...)
+	return k.opDelete(ctx, op)
+}
+
+func (k *kv) opDelete(ctx context.Context, op Op) (*DeleteResponse, error) {
+	r, err := k.d.Delete(ctx, op.key, op.rev)
+	if err != nil {
+		return nil, err
+	}
+	return getDeleteResponse(r), nil
+}
+
+func (k *kv) Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error) {
+	return &CompactResponse{
+		Header: &pb.ResponseHeader{},
+	}, nil
+}
+
+func (k *kv) Txn(ctx context.Context) Txn {
+	return &txn{
+		kv:  k,
+		ctx: ctx,
+	}
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/lease.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+	"golang.org/x/net/context"
+)
+
+type (
+	LeaseRevokeResponse pb.LeaseRevokeResponse
+	LeaseID             int64
+)
+
+// LeaseGrantResponse wraps the protobuf message LeaseGrantResponse.
+type LeaseGrantResponse struct {
+	*pb.ResponseHeader
+	ID    LeaseID
+	TTL   int64
+	Error string
+}
+
+type Lease interface {
+	// Grant creates a new lease.
+	Grant(ctx context.Context, ttl int64) (*LeaseGrantResponse, error)
+}
+
+type lessor struct {
+}
+
+func (l *lessor) Grant(ctx context.Context, ttl int64) (*LeaseGrantResponse, error) {
+	return &LeaseGrantResponse{
+		ResponseHeader: &pb.ResponseHeader{},
+		TTL:            ttl,
+		ID:             LeaseID(ttl),
+	}, nil
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/op.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+type opType int
+
+const (
+	// A default Op has opType 0, which is invalid.
+	tRange opType = iota + 1
+	tPut
+	tDeleteRange
+	tTxn
+)
+
+var (
+	noPrefixEnd = []byte{0}
+)
+
+// Op represents an Operation that kv can execute.
+type Op struct {
+	t           opType
+	key         string
+	boundingKey string
+
+	// for range
+	limit     int64
+	countOnly bool
+
+	// for range, watch
+	rev int64
+
+	// for watch, put, delete
+	prevKV bool
+
+	// for put
+	val     []byte
+	leaseID LeaseID
+
+	// txn
+	cmps    []Cmp
+	thenOps []Op
+	elseOps []Op
+}
+
+// accessors / mutators
+
+func (op Op) IsTxn() bool              { return op.t == tTxn }
+func (op Op) Txn() ([]Cmp, []Op, []Op) { return op.cmps, op.thenOps, op.elseOps }
+
+// Rev returns the requested revision, if any.
+func (op Op) Rev() int64 { return op.rev }
+
+// IsPut returns true iff the operation is a Put.
+func (op Op) IsPut() bool { return op.t == tPut }
+
+// IsGet returns true iff the operation is a Get.
+func (op Op) IsGet() bool { return op.t == tRange }
+
+// IsDelete returns true iff the operation is a Delete.
+func (op Op) IsDelete() bool { return op.t == tDeleteRange }
+
+// IsCountOnly returns whether countOnly is set.
+func (op Op) IsCountOnly() bool { return op.countOnly == true }
+
+// ValueBytes returns the byte slice holding the Op's value, if any.
+func (op Op) ValueBytes() []byte { return op.val }
+
+// WithValueBytes sets the byte slice for the Op's value.
+func (op *Op) WithValueBytes(v []byte) { op.val = v }
+
+func (op Op) toRangeRequest() *pb.RangeRequest {
+	if op.t != tRange {
+		panic("op.t != tRange")
+	}
+	r := &pb.RangeRequest{
+		Key:       []byte(op.key),
+		RangeEnd:  []byte(op.boundingKey),
+		Limit:     op.limit,
+		Revision:  op.rev,
+		CountOnly: op.countOnly,
+	}
+	return r
+}
+
+func OpGet(key string, opts ...OpOption) Op {
+	ret := Op{t: tRange, key: key}
+	ret.applyOpts(opts)
+	return ret
+}
+
+func OpDelete(key string, opts ...OpOption) Op {
+	ret := Op{t: tDeleteRange, key: key}
+	ret.applyOpts(opts)
+	switch {
+	case ret.leaseID != 0:
+		panic("unexpected lease in delete")
+	case ret.limit != 0:
+		panic("unexpected limit in delete")
+	case ret.rev != 0:
+		panic("unexpected revision in delete")
+	case ret.countOnly:
+		panic("unexpected countOnly in delete")
+	}
+	return ret
+}
+
+func OpPut(key, val string, opts ...OpOption) Op {
+	ret := Op{t: tPut, key: key, val: []byte(val)}
+	ret.applyOpts(opts)
+	switch {
+	case len(ret.key) > 0 && ret.key[len(ret.key)-1] == '%':
+		panic("unexpected range in put")
+	case ret.limit != 0:
+		panic("unexpected limit in put")
+	case ret.rev != 0:
+		panic("unexpected revision in put")
+	case ret.countOnly:
+		panic("unexpected countOnly in put")
+	}
+	return ret
+}
+
+func (op *Op) applyOpts(opts []OpOption) {
+	for _, opt := range opts {
+		opt(op)
+	}
+}
+
+// OpOption configures Operations like Get, Put, Delete.
+type OpOption func(*Op)
+
+// WithLease attaches a lease ID to a key in 'Put' request.
+func WithLease(leaseID LeaseID) OpOption {
+	return func(op *Op) { op.leaseID = leaseID }
+}
+
+// WithLimit limits the number of results to return from 'Get' request.
+// If WithLimit is given a 0 limit, it is treated as no limit.
+func WithLimit(n int64) OpOption { return func(op *Op) { op.limit = n } }
+
+// WithRev specifies the store revision for 'Get' request.
+// Or the start revision of 'Watch' request.
+func WithRev(rev int64) OpOption { return func(op *Op) { op.rev = rev } }
+
+// GetPrefixRangeEnd gets the range end of the prefix.
+// 'Get(foo, WithPrefix())' is equal to 'Get(foo, WithRange(GetPrefixRangeEnd(foo))'.
+func GetPrefixRangeEnd(prefix string) string {
+	return prefix + "%"
+}
+
+// WithPrefix enables 'Get', 'Delete', or 'Watch' requests to operate
+// on the keys with matching prefix. For example, 'Get(foo, WithPrefix())'
+// can return 'foo1', 'foo2', and so on.
+func WithPrefix() OpOption {
+	return func(op *Op) {
+		op.key += "%"
+	}
+}
+
+// WithRange specifies the range of 'Get', 'Delete', 'Watch' requests.
+// For example, 'Get' requests with 'WithRange(end)' returns
+// the keys in the range [key, end).
+// endKey must be lexicographically greater than start key.
+func WithRange(endKey string) OpOption {
+	return func(op *Op) { op.boundingKey = endKey }
+}
+
+// WithSerializable makes 'Get' request serializable. By default,
+// it's linearizable. Serializable requests are better for lower latency
+// requirement.
+func WithSerializable() OpOption {
+	return func(op *Op) {}
+}
+
+// WithCountOnly makes the 'Get' request return only the count of keys.
+func WithCountOnly() OpOption {
+	return func(op *Op) { op.countOnly = true }
+}
+
+// WithPrevKV gets the previous key-value pair before the event happens. If the previous KV is already compacted,
+// nothing will be returned.
+func WithPrevKV() OpOption {
+	return func(op *Op) {
+		op.prevKV = true
+	}
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/txn.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+	"golang.org/x/net/context"
+	utiltrace "k8s.io/utils/trace"
+)
+
+type Txn interface {
+	// If takes a list of comparison. If all comparisons passed in succeed,
+	// the operations passed into Then() will be executed. Or the operations
+	// passed into Else() will be executed.
+	If(cs ...Cmp) Txn
+
+	// Then takes a list of operations. The Ops list will be executed, if the
+	// comparisons passed in If() succeed.
+	Then(ops ...Op) Txn
+
+	// Else takes a list of operations. The Ops list will be executed, if the
+	// comparisons passed in If() fail.
+	Else(ops ...Op) Txn
+
+	// Commit tries to commit the transaction.
+	Commit() (*TxnResponse, error)
+}
+
+type txn struct {
+	kv  *kv
+	ctx context.Context
+
+	mu    sync.Mutex
+	cif   bool
+	cthen bool
+	celse bool
+
+	cmps []*pb.Compare
+
+	sus []Op
+	fas []Op
+}
+
+func (txn *txn) If(cs ...Cmp) Txn {
+	txn.mu.Lock()
+	defer txn.mu.Unlock()
+
+	if txn.cif {
+		panic("cannot call If twice!")
+	}
+
+	if txn.cthen {
+		panic("cannot call If after Then!")
+	}
+
+	if txn.celse {
+		panic("cannot call If after Else!")
+	}
+
+	txn.cif = true
+
+	for i := range cs {
+		txn.cmps = append(txn.cmps, (*pb.Compare)(&cs[i]))
+	}
+
+	return txn
+}
+
+func (txn *txn) Then(ops ...Op) Txn {
+	txn.mu.Lock()
+	defer txn.mu.Unlock()
+
+	if txn.cthen {
+		panic("cannot call Then twice!")
+	}
+	if txn.celse {
+		panic("cannot call Then after Else!")
+	}
+
+	txn.cthen = true
+
+	for _, op := range ops {
+		txn.sus = append(txn.sus, op)
+	}
+
+	return txn
+}
+
+func (txn *txn) Else(ops ...Op) Txn {
+	txn.mu.Lock()
+	defer txn.mu.Unlock()
+
+	if txn.celse {
+		panic("cannot call Else twice!")
+	}
+
+	txn.celse = true
+
+	for _, op := range ops {
+		txn.fas = append(txn.fas, op)
+	}
+
+	return txn
+}
+
+func (txn *txn) do(op Op) (*pb.ResponseOp, error) {
+	switch op.t {
+	case tRange:
+		r, err := txn.kv.opGet(txn.ctx, op)
+		if err != nil {
+			return nil, err
+		}
+		return &pb.ResponseOp{
+			Response: &pb.ResponseOp_ResponseRange{
+				ResponseRange: (*pb.RangeResponse)(r),
+			},
+		}, nil
+	case tPut:
+		r, err := txn.kv.opPut(txn.ctx, op)
+		if err != nil {
+			return nil, err
+		}
+		return &pb.ResponseOp{
+			Response: &pb.ResponseOp_ResponsePut{
+				ResponsePut: (*pb.PutResponse)(r),
+			},
+		}, nil
+	case tDeleteRange:
+		r, err := txn.kv.opDelete(txn.ctx, op)
+		if err != nil {
+			return nil, err
+		}
+		return &pb.ResponseOp{
+			Response: &pb.ResponseOp_ResponseDeleteRange{
+				ResponseDeleteRange: (*pb.DeleteRangeResponse)(r),
+			},
+		}, nil
+	default:
+		return nil, fmt.Errorf("unknown op in txn: %#v", op)
+	}
+}
+
+func (txn *txn) Commit() (*TxnResponse, error) {
+	trace := utiltrace.New("SQL Commit")
+	defer trace.LogIfLong(500 * time.Millisecond)
+
+	locks := map[string]bool{}
+	resp := &TxnResponse{
+		Header: &pb.ResponseHeader{},
+	}
+
+	good := true
+	for _, c := range txn.cmps {
+		k := string(c.Key)
+		if !locks[k] {
+			txn.kv.l.Lock(k)
+			trace.Step(fmt.Sprintf("lock acquired: %s", k))
+			locks[k] = true
+			defer txn.kv.l.Unlock(k)
+		}
+		gr, err := txn.kv.Get(txn.ctx, k)
+		if err != nil {
+			return nil, err
+		}
+
+		switch c.Target {
+		case pb.Compare_VERSION:
+			ver := int64(0)
+			if len(gr.Kvs) > 0 {
+				ver = gr.Kvs[0].Version
+			}
+			cv, _ := c.TargetUnion.(*pb.Compare_Version)
+			if ver != cv.Version {
+				good = false
+			}
+		case pb.Compare_MOD:
+			mod := int64(0)
+			if len(gr.Kvs) > 0 {
+				mod = gr.Kvs[0].ModRevision
+			}
+			cv, _ := c.TargetUnion.(*pb.Compare_ModRevision)
+			if mod != cv.ModRevision {
+				good = false
+			}
+		default:
+			return nil, fmt.Errorf("unknown txn target %v", c.Target)
+		}
+
+		trace.Step(fmt.Sprintf("condition key %s good %v", k, good))
+	}
+
+	resp.Succeeded = good
+	ops := txn.sus
+	if !good {
+		ops = txn.fas
+	}
+
+	for _, op := range ops {
+		r, err := txn.do(op)
+		if err != nil {
+			return nil, err
+		}
+		resp.Responses = append(resp.Responses, r)
+		trace.Step(fmt.Sprintf("op key %s op %v", op.key, op.t))
+	}
+
+	return resp, nil
+}

vendor/github.com/ibuildthecloud/kvsql/clientv3/watch.go

+// Copyright 2016 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientv3
+
+import (
+	v3rpc "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+	"github.com/coreos/etcd/mvcc/mvccpb"
+
+	"golang.org/x/net/context"
+)
+
+const (
+	EventTypeDelete = mvccpb.DELETE
+	EventTypePut    = mvccpb.PUT
+)
+
+type Event mvccpb.Event
+
+type WatchChan <-chan WatchResponse
+
+type Watcher interface {
+	// Watch watches on a key or prefix. The watched events will be returned
+	// through the returned channel. If revisions waiting to be sent over the
+	// watch are compacted, then the watch will be canceled by the server, the
+	// client will post a compacted error watch response, and the channel will close.
+	Watch(ctx context.Context, key string, opts ...OpOption) WatchChan
+
+	// Close closes the watcher and cancels all watch requests.
+	Close() error
+}
+
+func (k *kv) Watch(ctx context.Context, key string, opts ...OpOption) WatchChan {
+	op := OpGet(key, opts...)
+	c := k.d.Watch(ctx, op.key, op.rev)
+
+	result := make(chan WatchResponse)
+	go func() {
+		defer close(result)
+		for e := range c {
+			if e.Err != nil {
+				result <- NewWatchResponseErr(e.Err)
+				continue
+			} else if e.Start {
+				result <- WatchResponse{
+					Created: true,
+				}
+				continue
+			}
+
+			k := e.KV
+
+			event := &Event{}
+			if k.Del == 0 {
+				event.Type = mvccpb.PUT
+			} else {
+				event.Type = mvccpb.DELETE
+			}
+
+			event.Kv = toKeyValue(k)
+			if event.Kv.Version > 1 && k.OldRevision > 0 {
+				oldKV := *event.Kv
+				oldKV.ModRevision = k.OldRevision
+				oldKV.Value = k.OldValue
+				event.PrevKv = &oldKV
+			}
+
+			wr := WatchResponse{
+				Header: pb.ResponseHeader{
+					Revision: event.Kv.ModRevision,
+				},
+				Events: []*Event{
+					event,
+				},
+			}
+			result <- wr
+		}
+	}()
+
+	return result
+}
+
+func (k *kv) Close() error {
+	return k.d.Close()
+}
+
+type WatchResponse struct {
+	Header pb.ResponseHeader
+	Events []*Event
+
+	// CompactRevision is the minimum revision the watcher may receive.
+	CompactRevision int64
+
+	// Canceled is used to indicate watch failure.
+	// If the watch failed and the stream was about to close, before the channel is closed,
+	// the channel sends a final response that has Canceled set to true with a non-nil Err().
+	Canceled bool
+
+	// Created is used to indicate the creation of the watcher.
+	Created bool
+
+	closeErr error
+}
+
+func NewWatchResponseErr(err error) WatchResponse {
+	return WatchResponse{
+		Canceled: true,
+		closeErr: err,
+	}
+}
+
+// IsCreate returns true if the event tells that the key is newly created.
+func (e *Event) IsCreate() bool {
+	return e.Type == EventTypePut && e.Kv.CreateRevision == e.Kv.ModRevision
+}
+
+// IsModify returns true if the event tells that a new value is put on existing key.
+func (e *Event) IsModify() bool {
+	return e.Type == EventTypePut && e.Kv.CreateRevision != e.Kv.ModRevision
+}
+
+// Err is the error value if this WatchResponse holds an error.
+func (wr *WatchResponse) Err() error {
+	switch {
+	case wr.closeErr != nil:
+		return v3rpc.Error(wr.closeErr)
+	case wr.CompactRevision != 0:
+		return v3rpc.ErrCompacted
+	}
+	return nil
+}
+
+// IsProgressNotify returns true if the WatchResponse is progress notification.
+func (wr *WatchResponse) IsProgressNotify() bool {
+	return len(wr.Events) == 0 && !wr.Canceled && !wr.Created && wr.CompactRevision == 0 && wr.Header.Revision != 0
+}

vendor/github.com/ibuildthecloud/kvsql/factory.go

+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package factory
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"time"
+
+	"github.com/ibuildthecloud/kvsql/clientv3"
+	etcd3 "github.com/ibuildthecloud/kvsql/storage"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apiserver/pkg/storage"
+	"k8s.io/apiserver/pkg/storage/storagebackend"
+	"k8s.io/apiserver/pkg/storage/value"
+)
+
+func NewKVSQLHealthCheck(c storagebackend.Config) (func() error, error) {
+	// constructing the etcd v3 client blocks and times out if etcd is not available.
+	// retry in a loop in the background until we successfully create the client, storing the client or error encountered
+
+	clientValue := &atomic.Value{}
+
+	clientErrMsg := &atomic.Value{}
+	clientErrMsg.Store("etcd client connection not yet established")
+
+	go wait.PollUntil(time.Second, func() (bool, error) {
+		client, err := newETCD3Client(c)
+		if err != nil {
+			clientErrMsg.Store(err.Error())
+			return false, nil
+		}
+		clientValue.Store(client)
+		clientErrMsg.Store("")
+		return true, nil
+	}, wait.NeverStop)
+
+	return func() error {
+		if errMsg := clientErrMsg.Load().(string); len(errMsg) > 0 {
+			return fmt.Errorf(errMsg)
+		}
+		client := clientValue.Load().(*clientv3.Client)
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel()
+		if _, err := client.Cluster.MemberList(ctx); err != nil {
+			return fmt.Errorf("error listing etcd members: %v", err)
+		}
+		return nil
+	}, nil
+}
+
+func newETCD3Client(c storagebackend.Config) (*clientv3.Client, error) {
+	cfg := clientv3.Config{
+		Endpoints: c.Transport.ServerList,
+	}
+
+	if len(cfg.Endpoints) == 0 {
+		cfg.Endpoints = []string{"sqlite://"}
+	}
+
+	client, err := clientv3.New(cfg)
+	return client, err
+}
+
+func NewKVSQLStorage(c storagebackend.Config) (storage.Interface, func(), error) {
+	client, err := newETCD3Client(c)
+	if err != nil {
+		return nil, nil, err
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	etcd3.StartCompactor(ctx, client, c.CompactionInterval)
+	destroyFunc := func() {
+		cancel()
+		client.Close()
+	}
+	transformer := c.Transformer
+	if transformer == nil {
+		transformer = value.IdentityTransformer
+	}
+
+	return etcd3.New(client, c.Codec, c.Prefix, transformer, c.Paging), destroyFunc, nil
+}

vendor/github.com/ibuildthecloud/kvsql/pkg/broadcast/broadcaster.go

+package broadcast
+
+import (
+	"context"
+	"sync"
+)
+
+type ConnectFunc func() (chan map[string]interface{}, error)
+
+type Broadcaster struct {
+	sync.Mutex
+	running bool
+	subs    map[chan map[string]interface{}]struct{}
+}
+
+func (b *Broadcaster) Subscribe(ctx context.Context, connect ConnectFunc) (chan map[string]interface{}, error) {
+	b.Lock()
+	defer b.Unlock()
+
+	if !b.running {
+		if err := b.start(connect); err != nil {
+			return nil, err
+		}
+	}
+
+	sub := make(chan map[string]interface{}, 100)
+	if b.subs == nil {
+		b.subs = map[chan map[string]interface{}]struct{}{}
+	}
+	b.subs[sub] = struct{}{}
+	go func() {
+		<-ctx.Done()
+		b.unsub(sub, true)
+	}()
+
+	return sub, nil
+}
+
+func (b *Broadcaster) unsub(sub chan map[string]interface{}, lock bool) {
+	if lock {
+		b.Lock()
+	}
+	if _, ok := b.subs[sub]; ok {
+		close(sub)
+		delete(b.subs, sub)
+	}
+	if lock {
+		b.Unlock()
+	}
+}
+
+func (b *Broadcaster) start(connect ConnectFunc) error {
+	c, err := connect()
+	if err != nil {
+		return err
+	}
+
+	go b.stream(c)
+	b.running = true
+	return nil
+}
+
+func (b *Broadcaster) stream(input chan map[string]interface{}) {
+	for item := range input {
+		b.Lock()
+		for sub := range b.subs {
+			newItem := cloneMap(item)
+			select {
+			case sub <- newItem:
+			default:
+				// Slow consumer, drop
+				go b.unsub(sub, true)
+			}
+		}
+		b.Unlock()
+	}
+
+	b.Lock()
+	for sub := range b.subs {
+		b.unsub(sub, false)
+	}
+	b.running = false
+	b.Unlock()
+}
+
+func cloneMap(data map[string]interface{}) map[string]interface{} {
+	if data == nil {
+		return nil
+	}
+
+	result := map[string]interface{}{}
+	for k, v := range data {
+		result[k] = cloneValue(v)
+	}
+
+	return result
+}
+
+func cloneValue(v interface{}) interface{} {
+	switch t := v.(type) {
+	case []interface{}:
+		return cloneSlice(t)
+	case []map[string]interface{}:
+		return cloneMapSlice(t)
+	case map[string]interface{}:
+		return cloneMap(t)
+	default:
+		return v
+	}
+}
+
+func cloneMapSlice(data []map[string]interface{}) []interface{} {
+	result := make([]interface{}, len(data))
+	for i := range data {
+		result[i] = cloneValue(data[i])
+	}
+	return result
+}
+
+func cloneSlice(data []interface{}) []interface{} {
+	result := make([]interface{}, len(data))
+	for i := range data {
+		result[i] = cloneValue(data[i])
+	}
+	return result
+}

vendor/github.com/ibuildthecloud/kvsql/storage/compact.go

+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd3
+
+import (
+	"context"
+	"strconv"
+	"sync"
+	"time"
+
+	"k8s.io/klog"
+	"github.com/ibuildthecloud/kvsql/clientv3"
+)
+
+const (
+	compactRevKey = "compact_rev_key"
+)
+
+var (
+	endpointsMapMu sync.Mutex
+	endpointsMap   map[string]struct{}
+)
+
+func init() {
+	endpointsMap = make(map[string]struct{})
+}
+
+// StartCompactor starts a compactor in the background to compact old version of keys that's not needed.
+// By default, we save the most recent 10 minutes data and compact versions > 10minutes ago.
+// It should be enough for slow watchers and to tolerate burst.
+// TODO: We might keep a longer history (12h) in the future once storage API can take advantage of past version of keys.
+func StartCompactor(ctx context.Context, client *clientv3.Client, compactInterval time.Duration) {
+	endpointsMapMu.Lock()
+	defer endpointsMapMu.Unlock()
+
+	// In one process, we can have only one compactor for one cluster.
+	// Currently we rely on endpoints to differentiate clusters.
+	for _, ep := range client.Endpoints() {
+		if _, ok := endpointsMap[ep]; ok {
+			klog.V(4).Infof("compactor already exists for endpoints %v", client.Endpoints())
+			return
+		}
+	}
+	for _, ep := range client.Endpoints() {
+		endpointsMap[ep] = struct{}{}
+	}
+
+	if compactInterval != 0 {
+		go compactor(ctx, client, compactInterval)
+	}
+}
+
+// compactor periodically compacts historical versions of keys in etcd.
+// It will compact keys with versions older than given interval.
+// In other words, after compaction, it will only contain keys set during last interval.
+// Any API call for the older versions of keys will return error.
+// Interval is the time interval between each compaction. The first compaction happens after "interval".
+func compactor(ctx context.Context, client *clientv3.Client, interval time.Duration) {
+	// Technical definitions:
+	// We have a special key in etcd defined as *compactRevKey*.
+	// compactRevKey's value will be set to the string of last compacted revision.
+	// compactRevKey's version will be used as logical time for comparison. THe version is referred as compact time.
+	// Initially, because the key doesn't exist, the compact time (version) is 0.
+	//
+	// Algorithm:
+	// - Compare to see if (local compact_time) = (remote compact_time).
+	// - If yes, increment both local and remote compact_time, and do a compaction.
+	// - If not, set local to remote compact_time.
+	//
+	// Technical details/insights:
+	//
+	// The protocol here is lease based. If one compactor CAS successfully, the others would know it when they fail in
+	// CAS later and would try again in 10 minutes. If an APIServer crashed, another one would "take over" the lease.
+	//
+	// For example, in the following diagram, we have a compactor C1 doing compaction in t1, t2. Another compactor C2
+	// at t1' (t1 < t1' < t2) would CAS fail, set its known oldRev to rev at t1', and try again in t2' (t2' > t2).
+	// If C1 crashed and wouldn't compact at t2, C2 would CAS successfully at t2'.
+	//
+	//                 oldRev(t2)     curRev(t2)
+	//                                  +
+	//   oldRev        curRev           |
+	//     +             +              |
+	//     |             |              |
+	//     |             |    t1'       |     t2'
+	// +---v-------------v----^---------v------^---->
+	//     t0           t1             t2
+	//
+	// We have the guarantees:
+	// - in normal cases, the interval is 10 minutes.
+	// - in failover, the interval is >10m and <20m
+	//
+	// FAQ:
+	// - What if time is not accurate? We don't care as long as someone did the compaction. Atomicity is ensured using
+	//   etcd API.
+	// - What happened under heavy load scenarios? Initially, each apiserver will do only one compaction
+	//   every 10 minutes. This is very unlikely affecting or affected w.r.t. server load.
+
+	var compactTime int64
+	var rev int64
+	var err error
+	for {
+		select {
+		case <-time.After(interval):
+		case <-ctx.Done():
+			return
+		}
+
+		compactTime, rev, err = compact(ctx, client, compactTime, rev)
+		if err != nil {
+			klog.Errorf("etcd: endpoint (%v) compact failed: %v", client.Endpoints(), err)
+			continue
+		}
+	}
+}
+
+// compact compacts etcd store and returns current rev.
+// It will return the current compact time and global revision if no error occurred.
+// Note that CAS fail will not incur any error.
+func compact(ctx context.Context, client *clientv3.Client, t, rev int64) (int64, int64, error) {
+	resp, err := client.KV.Txn(ctx).If(
+		clientv3.Compare(clientv3.Version(compactRevKey), "=", t),
+	).Then(
+		clientv3.OpPut(compactRevKey, strconv.FormatInt(rev, 10)), // Expect side effect: increment Version
+	).Else(
+		clientv3.OpGet(compactRevKey),
+	).Commit()
+	if err != nil {
+		return t, rev, err
+	}
+
+	curRev := resp.Header.Revision
+
+	if !resp.Succeeded {
+		curTime := resp.Responses[0].GetResponseRange().Kvs[0].Version
+		return curTime, curRev, nil
+	}
+	curTime := t + 1
+
+	if rev == 0 {
+		// We don't compact on bootstrap.
+		return curTime, curRev, nil
+	}
+	if _, err = client.Compact(ctx, rev); err != nil {
+		return curTime, curRev, err
+	}
+	klog.V(4).Infof("etcd: compacted rev (%d), endpoints (%v)", rev, client.Endpoints())
+	return curTime, curRev, nil
+}

vendor/github.com/ibuildthecloud/kvsql/storage/errors.go

+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd3
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+
+	etcdrpc "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+)
+
+func interpretWatchError(err error) error {
+	switch {
+	case err == etcdrpc.ErrCompacted:
+		return errors.NewResourceExpired("The resourceVersion for the provided watch is too old.")
+	}
+	return err
+}
+
+const (
+	expired         string = "The resourceVersion for the provided list is too old."
+	continueExpired string = "The provided continue parameter is too old " +
+		"to display a consistent list result. You can start a new list without " +
+		"the continue parameter."
+	inconsistentContinue string = "The provided continue parameter is too old " +
+		"to display a consistent list result. You can start a new list without " +
+		"the continue parameter, or use the continue token in this response to " +
+		"retrieve the remainder of the results. Continuing with the provided " +
+		"token results in an inconsistent list - objects that were created, " +
+		"modified, or deleted between the time the first chunk was returned " +
+		"and now may show up in the list."
+)
+
+func interpretListError(err error, paging bool, continueKey, keyPrefix string) error {
+	switch {
+	case err == etcdrpc.ErrCompacted:
+		if paging {
+			return handleCompactedErrorForPaging(continueKey, keyPrefix)
+		}
+		return errors.NewResourceExpired(expired)
+	}
+	return err
+}
+
+func handleCompactedErrorForPaging(continueKey, keyPrefix string) error {
+	// continueToken.ResoureVersion=-1 means that the apiserver can
+	// continue the list at the latest resource version. We don't use rv=0
+	// for this purpose to distinguish from a bad token that has empty rv.
+	newToken, err := encodeContinue(continueKey, keyPrefix, -1)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return errors.NewResourceExpired(continueExpired)
+	}
+	statusError := errors.NewResourceExpired(inconsistentContinue)
+	statusError.ErrStatus.ListMeta.Continue = newToken
+	return statusError
+}

vendor/github.com/ibuildthecloud/kvsql/storage/event.go

+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd3
+
+import (
+	"github.com/ibuildthecloud/kvsql/clientv3"
+	"github.com/coreos/etcd/mvcc/mvccpb"
+)
+
+type event struct {
+	key       string
+	value     []byte
+	prevValue []byte
+	rev       int64
+	isDeleted bool
+	isCreated bool
+}
+
+// parseKV converts a KeyValue retrieved from an initial sync() listing to a synthetic isCreated event.
+func parseKV(kv *mvccpb.KeyValue) *event {
+	return &event{
+		key:       string(kv.Key),
+		value:     kv.Value,
+		prevValue: nil,
+		rev:       kv.ModRevision,
+		isDeleted: false,
+		isCreated: true,
+	}
+}
+
+func parseEvent(e *clientv3.Event) *event {
+	ret := &event{
+		key:       string(e.Kv.Key),
+		value:     e.Kv.Value,
+		rev:       e.Kv.ModRevision,
+		isDeleted: e.Type == clientv3.EventTypeDelete,
+		isCreated: e.IsCreate(),
+	}
+	if e.PrevKv != nil {
+		ret.prevValue = e.PrevKv.Value
+	}
+	return ret
+}

vendor/github.com/ibuildthecloud/kvsql/storage/lease_manager.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package etcd3
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/ibuildthecloud/kvsql/clientv3"
+)
+
+// leaseManager is used to manage leases requested from etcd. If a new write
+// needs a lease that has similar expiration time to the previous one, the old
+// lease will be reused to reduce the overhead of etcd, since lease operations
+// are expensive. In the implementation, we only store one previous lease,
+// since all the events have the same ttl.
+type leaseManager struct {
+	client                  *clientv3.Client // etcd client used to grant leases
+	leaseMu                 sync.Mutex
+	prevLeaseID             clientv3.LeaseID
+	prevLeaseExpirationTime time.Time
+	// The period of time in seconds and percent of TTL that each lease is
+	// reused. The minimum of them is used to avoid unreasonably large
+	// numbers. We use var instead of const for testing purposes.
+	leaseReuseDurationSeconds int64
+	leaseReuseDurationPercent float64
+}
+
+// newDefaultLeaseManager creates a new lease manager using default setting.
+func newDefaultLeaseManager(client *clientv3.Client) *leaseManager {
+	return newLeaseManager(client, 60, 0.05)
+}
+
+// newLeaseManager creates a new lease manager with the number of buffered
+// leases, lease reuse duration in seconds and percentage. The percentage
+// value x means x*100%.
+func newLeaseManager(client *clientv3.Client, leaseReuseDurationSeconds int64, leaseReuseDurationPercent float64) *leaseManager {
+	return &leaseManager{
+		client: client,
+		leaseReuseDurationSeconds: leaseReuseDurationSeconds,
+		leaseReuseDurationPercent: leaseReuseDurationPercent,
+	}
+}
+
+// setLeaseReuseDurationSeconds is used for testing purpose. It is used to
+// reduce the extra lease duration to avoid unnecessary timeout in testing.
+func (l *leaseManager) setLeaseReuseDurationSeconds(duration int64) {
+	l.leaseMu.Lock()
+	defer l.leaseMu.Unlock()
+	l.leaseReuseDurationSeconds = duration
+}
+
+// GetLease returns a lease based on requested ttl: if the cached previous
+// lease can be reused, reuse it; otherwise request a new one from etcd.
+func (l *leaseManager) GetLease(ctx context.Context, ttl int64) (clientv3.LeaseID, error) {
+	now := time.Now()
+	l.leaseMu.Lock()
+	defer l.leaseMu.Unlock()
+	// check if previous lease can be reused
+	reuseDurationSeconds := l.getReuseDurationSecondsLocked(ttl)
+	valid := now.Add(time.Duration(ttl) * time.Second).Before(l.prevLeaseExpirationTime)
+	sufficient := now.Add(time.Duration(ttl+reuseDurationSeconds) * time.Second).After(l.prevLeaseExpirationTime)
+	if valid && sufficient {
+		return l.prevLeaseID, nil
+	}
+	// request a lease with a little extra ttl from etcd
+	ttl += reuseDurationSeconds
+	lcr, err := l.client.Lease.Grant(ctx, ttl)
+	if err != nil {
+		return clientv3.LeaseID(0), err
+	}
+	// cache the new lease id
+	l.prevLeaseID = lcr.ID
+	l.prevLeaseExpirationTime = now.Add(time.Duration(ttl) * time.Second)
+	return lcr.ID, nil
+}
+
+// getReuseDurationSecondsLocked returns the reusable duration in seconds
+// based on the configuration. Lock has to be acquired before calling this
+// function.
+func (l *leaseManager) getReuseDurationSecondsLocked(ttl int64) int64 {
+	reuseDurationSeconds := int64(l.leaseReuseDurationPercent * float64(ttl))
+	if reuseDurationSeconds > l.leaseReuseDurationSeconds {
+		reuseDurationSeconds = l.leaseReuseDurationSeconds
+	}
+	return reuseDurationSeconds
+}