Add Gorm v2-based backend

go.mod

@@ -5,13 +5,21 @@ go 1.12
 require (
 	github.com/Rican7/retry v0.1.0
 	github.com/canonical/go-dqlite v1.5.1
-	github.com/go-sql-driver/mysql v1.4.1
-	github.com/lib/pq v1.1.1
-	github.com/mattn/go-sqlite3 v1.10.0
+	github.com/denisenkom/go-mssqldb v0.0.0-20200428022330-06a60b6afbbc
+	github.com/go-sql-driver/mysql v1.5.0
+	github.com/lib/pq v1.3.0
+	github.com/mattn/go-sqlite3 v1.14.0
 	github.com/pkg/errors v0.8.1
+	github.com/pkg/math v0.0.0-20141027224758-f2ed9e40e245
 	github.com/rancher/wrangler v0.4.0
+	github.com/satori/go.uuid v1.2.0
 	github.com/sirupsen/logrus v1.4.2
 	github.com/urfave/cli v1.21.0
 	go.etcd.io/etcd v0.0.0-20191023171146-3cf2f69b5738
 	google.golang.org/grpc v1.23.1
+	gorm.io/driver/mysql v1.0.0
+	gorm.io/driver/postgres v1.0.0
+	gorm.io/driver/sqlite v1.1.0
+	gorm.io/driver/sqlserver v1.0.0
+	gorm.io/gorm v1.9.19
 )

main.go

@@ -4,21 +4,30 @@ import (
 	"context"
 	"os"
 
-	"github.com/rancher/kine/pkg/endpoint"
+	"github.com/pkg/math"
 	"github.com/rancher/wrangler/pkg/signals"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
+
+	"github.com/rancher/kine/pkg/endpoint"
 )
 
 var (
 	config endpoint.Config
 )
 
 func main() {
+	cli.VersionFlag = cli.BoolFlag{
+		Name:  "version, V",
+		Usage: "print the version",
+	}
+
 	app := cli.NewApp()
 	app.Name = "kine"
 	app.Description = "Minimal etcd v3 API to support custom Kubernetes storage engines"
 	app.Usage = "Mini"
+	app.Version = "0.4.0"
+
 	app.Flags = []cli.Flag{
 		cli.StringFlag{
 			Name:        "listen-address",
@@ -63,7 +72,21 @@ func main() {
 			Usage:       "Key file for DB connection",
 			Destination: &config.KeyFile,
 		},
-		cli.BoolFlag{Name: "debug"},
+		cli.BoolFlag{
+			Name:  "debug",
+			Usage: "Enable debug mode for diagnostics",
+		},
+		cli.IntFlag{
+			Name:        "v, verbose",
+			Usage:       "Set the verbosity level (0=Panic Only, 1=Fatal, ..., 6=Trace)",
+			Value:       -1,
+			Destination: &config.Features.VerboseLevel,
+		},
+		cli.BoolFlag{
+			Name:        "alpha-use-new-backend",
+			Usage:       "Use the new experimental gorm based database backend",
+			Destination: &config.Features.UseAlphaBackend,
+		},
 	}
 	app.Action = run
 
@@ -74,8 +97,20 @@ func main() {
 
 func run(c *cli.Context) error {
 	if c.Bool("debug") {
-		logrus.SetLevel(logrus.DebugLevel)
+		if config.IsValidVerboseLevel() {
+			logrus.Warn("you've requested verbose level and debug mode together yourself, whoever is more verbose will be superseding")
+		}
+		config.Features.VerboseLevel = math.Max(config.Features.VerboseLevel, int(logrus.DebugLevel))
+	}
+
+	if config.IsValidVerboseLevel() {
+		logrus.SetLevel(logrus.Level(config.Features.VerboseLevel))
 	}
+
+	if config.Features.UseAlphaBackend {
+		logrus.Warn("you are using an experimental backend powered by gorm. it is not guaranteed everything will work for now")
+	}
+
 	ctx := signals.SetupSignalHandler(context.Background())
 	_, err := endpoint.Listen(ctx, config)
 	if err != nil {

pkg/alpha/gorm/backend.go

@@ -0,0 +1,64 @@
+package gorm
+
+import (
+	"context"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"gorm.io/gorm"
+
+	"github.com/rancher/kine/pkg/tls"
+)
+
+type DatabaseErrorHandler interface {
+	HandleInsertionError(err error) error
+}
+
+type Driver interface {
+	PrepareDSN(dataSourceName string, tlsInfo tls.Config) (string, error)
+	GetOpenFunctor() func(string) gorm.Dialector
+	DatabaseErrorHandler
+}
+
+type DatabaseBackend struct {
+	DB *gorm.DB
+	DatabaseErrorHandler
+}
+
+func New(ctx context.Context, dialect gorm.Dialector, handlers DatabaseErrorHandler) (backend *DatabaseBackend, err error) {
+	db, err := gorm.Open(dialect, &gorm.Config{
+		Logger:      &Logger{},
+		PrepareStmt: true,
+	})
+
+	if err != nil {
+		return
+	}
+
+	rawDB, err := db.DB()
+	if err != nil {
+		return
+	}
+
+	// Actually, I don't think this is much needed
+	// GORM offers automatic pinging so no worries
+majorScope:
+	for i := 0; i < 300; i++ {
+		for i := 0; i < 3; i++ {
+			if err = rawDB.Ping(); err == nil {
+				break majorScope
+			}
+		}
+
+		err = rawDB.Close()
+		logrus.WithError(err).Error("failed to ping connection")
+		select {
+		case <-ctx.Done():
+			err = ctx.Err()
+			return
+		case <-time.After(time.Second):
+		}
+	}
+	backend = &DatabaseBackend{DB: db, DatabaseErrorHandler: handlers}
+	return
+}

pkg/alpha/gorm/compat/compat.go

@@ -0,0 +1,17 @@
+package compat
+
+import (
+	"github.com/rancher/kine/pkg/alpha/gorm"
+)
+
+type Backend struct {
+	gorm.DatabaseBackend
+}
+
+func New(backend *gorm.DatabaseBackend) (*Backend, error) {
+	if err := backend.DB.AutoMigrate(&KineEntry{}); err != nil {
+		return nil, err
+	}
+	compatBackend := &Backend{*backend}
+	return compatBackend, nil
+}

pkg/alpha/gorm/compat/dialect.go

@@ -0,0 +1,146 @@
+package compat
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+
+	"gorm.io/gorm"
+)
+
+func (g *Backend) ListCurrent(ctx context.Context, prefix string, limit int64, includeDeleted bool) (*sql.Rows, error) {
+	tx := g.ListQuery(ctx, prefix, limit, includeDeleted, nil)
+	return tx.Rows()
+}
+
+func (g *Backend) List(ctx context.Context, prefix, startKey string, limit, revision int64, includeDeleted bool) (*sql.Rows, error) {
+	subquery := g.DB.WithContext(ctx).
+		Where("id <= ?", revision)
+
+	if startKey != "" {
+		subsubquery := g.FindBestLatestKeyBoundByRevision(ctx, startKey, revision)
+		subquery = subquery.Where("id > (?)", subsubquery)
+	}
+
+	tx := g.ListQuery(ctx, prefix, limit, includeDeleted, subquery)
+	return tx.Rows()
+}
+
+func (g *Backend) Count(ctx context.Context, prefix string) (int64, int64, error) {
+	kv := KineEntry{}
+	tx := g.CurrentRevisionQuery(ctx).Find(&kv)
+	if tx.Error == nil {
+		var children int64
+		tx := g.ListCurrentWithPrefixQuery(ctx, prefix, false, nil, "id as theid").
+			Select("COUNT(theid)").
+			Count(&children)
+		if tx.Error == nil {
+			return int64(kv.ID), children, nil
+		}
+	}
+	return 0, 0, tx.Error
+}
+
+func (g *Backend) CurrentRevision(ctx context.Context) (revision int64, err error) {
+	kv := KineEntry{}
+	tx := g.CurrentRevisionQuery(ctx).Find(&kv)
+	if errors.Is(tx.Error, gorm.ErrRecordNotFound) {
+		return
+	}
+	return int64(kv.ID), tx.Error
+}
+
+func (g *Backend) After(ctx context.Context, prefix string, rev, limit int64) (*sql.Rows, error) {
+	tx := g.DB.WithContext(ctx).
+		Model(&KineEntry{}).
+		Limit(int(limit)).
+		Order("id ASC").
+		Where("name LIKE ?", prefix).
+		Where("id > ?", rev).
+		Select(
+			fmt.Sprintf("(?), (?), %s", columns),
+			g.CurrentRevisionQuery(ctx), g.GetCompactRevisionQuery(ctx),
+		)
+
+	return tx.Rows()
+}
+
+func (g *Backend) Insert(
+	ctx context.Context, key string, create, delete bool, createRevision, previousRevision int64, ttl int64, value,
+	prevValue []byte,
+) (revision int64, err error) {
+	defer func() {
+		if interceptErr := g.HandleInsertionError(err); interceptErr != nil {
+			err = interceptErr
+		}
+	}()
+
+	entity := KineEntry{
+		Name:           key,
+		Created:        create,
+		Deleted:        delete,
+		CreateRevision: uint64(createRevision),
+		PrevRevision:   uint64(previousRevision),
+		Lease:          uint64(ttl),
+		Value:          value,
+		OldValue:       prevValue,
+	}
+
+	tx := g.DB.WithContext(ctx).
+		Save(&entity)
+	return int64(entity.ID), tx.Error
+}
+
+func (g *Backend) GetRevision(ctx context.Context, revision int64) (*sql.Rows, error) {
+	return g.DB.WithContext(ctx).
+		Where(&KineEntry{ID: uint64(revision)}).
+		Select(
+			fmt.Sprintf("0, 0, %s", columns),
+		).
+		Rows()
+}
+
+func (g *Backend) DeleteRevision(ctx context.Context, revision int64) error {
+	tx := g.DB.WithContext(ctx).
+		Delete(&KineEntry{ID: uint64(revision)})
+	return tx.Error
+}
+
+func (g *Backend) GetCompactRevision(ctx context.Context) (revision int64, err error) {
+	var kv KineEntry
+	tx := g.GetCompactRevisionQuery(ctx).
+		Last(&kv)
+	if errors.Is(tx.Error, gorm.ErrRecordNotFound) {
+		return
+	}
+	return int64(kv.PrevRevision), tx.Error
+}
+
+func (g *Backend) SetCompactRevision(ctx context.Context, revision int64) error {
+	tx := g.GetCompactRevisionQuery(ctx).
+		Updates(&KineEntry{PrevRevision: uint64(revision)})
+	return tx.Error
+}
+
+func (g *Backend) Fill(ctx context.Context, revision int64) error {
+	tx := g.DB.WithContext(ctx).Create(
+		&KineEntry{
+			ID:             uint64(revision),
+			Name:           fmt.Sprintf("gap-%d", revision),
+			Created:        false,
+			Deleted:        true,
+			CreateRevision: 0,
+			PrevRevision:   0,
+			Lease:          0,
+			Value:          nil,
+			OldValue:       nil,
+		},
+	)
+	return tx.Error
+}
+
+func (g *Backend) IsFill(key string) bool {
+	return strings.HasPrefix(key, "gap-")
+}