[storage] Add an SQL storage driver

This commits adds the possibility to back Tiller (or the future
Tiller-less Helm CLI) with any SQL database (only postgres has been
tested so far) to store release information.

The main motivation for this commit was to use a storage backend that
would allow releases larger that 1MB in size (ConfigMap or Secret
drivers don't, because of limits on value size in the underlying etcd
key-value store).
Signed-off-by: Étienne Lafarge <etienne.lafarge@gmail.com>

Co-authored-by: Elliot Maincourt <e.maincourt@gmail.com> (@emaincourt)
Co-authored-by: Paul Borensztein <hi@0x01.fr> (@commit-master)

cmd/tiller/tiller.go

@@ -66,6 +66,7 @@ const (
 	storageMemory    = "memory"
 	storageConfigMap = "configmap"
 	storageSecret    = "secret"
+	storageSQL       = "sql"
 
 	traceAddr = ":44136"
 
@@ -77,8 +78,13 @@ var (
 	grpcAddr      = flag.String("listen", fmt.Sprintf(":%v", environment.DefaultTillerPort), "address:port to listen on")
 	probeAddr     = flag.String("probe-listen", fmt.Sprintf(":%v", environment.DefaultTillerProbePort), "address:port to listen on for probes")
 	enableTracing = flag.Bool("trace", false, "enable rpc tracing")
-	store                = flag.String("storage", storageConfigMap, "storage driver to use. One of 'configmap', 'memory', or 'secret'")
+	store         = flag.String("storage", storageConfigMap, "storage driver to use. One of 'configmap', 'memory', 'sql' or 'secret'")
+
+	sqlDialect          = flag.String("sql-dialect", "postgres", "SQL dialect to use (only postgres is supported for now")
+	sqlConnectionString = flag.String("sql-connection-string", "", "SQL connection string to use")
+
 	remoteReleaseModules = flag.Bool("experimental-release", false, "enable experimental release modules")
+
 	tlsEnable    = flag.Bool("tls", tlsEnableEnvVarDefault(), "enable TLS")
 	tlsVerify    = flag.Bool("tls-verify", tlsVerifyEnvVarDefault(), "enable TLS and verify remote certificate")
 	keyFile      = flag.String("tls-key", tlsDefaultsFromEnv("tls-key"), "path to TLS private key file")
@@ -143,6 +149,18 @@ func start() {
 
 		env.Releases = storage.Init(secrets)
 		env.Releases.Log = newLogger("storage").Printf
+	case storageSQL:
+		sqlDriver, err := driver.NewSQL(
+			*sqlDialect,
+			*sqlConnectionString,
+			newLogger("storage/driver").Printf,
+		)
+		if err != nil {
+			logger.Fatalf("Cannot initialize SQL storage driver: %v", err)
+		}
+
+		env.Releases = storage.Init(sqlDriver)
+		env.Releases.Log = newLogger("storage").Printf
 	}
 
 	if *maxHistory > 0 {

docs/install.md

@@ -353,7 +353,10 @@ in JSON format.
 
 ### Storage backends
 By default, `tiller` stores release information in `ConfigMaps` in the namespace
-where it is running. As of Helm 2.7.0, there is now a beta storage backend that
+where it is running.
+
+#### Secret storage backend
+As of Helm 2.7.0, there is now a beta storage backend that
 uses `Secrets` for storing release information. This was added for additional
 security in protecting charts in conjunction with the release of `Secret`
 encryption in Kubernetes.
@@ -369,6 +372,31 @@ Currently, if you want to switch from the default backend to the secrets
 backend, you'll have to do the migration for this on your own. When this backend
 graduates from beta, there will be a more official path of migration
 
+#### SQL storage backend
+As of Helm 2.14.0 there is now a beta SQL storage backend that stores release
+information in an SQL database (only postgres has been tested so far).
+
+Using such a storage backend is particularly useful if your release information
+weighs more than 1MB (in which case, it can't be stored in ConfigMaps/Secrets
+because of internal limits in Kubernetes' underlying etcd key-value store).
+
+To enable the SQL backend, you'll need to [deploy an SQL
+database](./sql-storage.md) and init Tiller with the following options:
+
+```shell
+helm init \
+  --override \
+    'spec.template.spec.containers[0].args'='{--storage=sql,--sql-dialect=postgres,--sql-connection-string=postgresql://tiller-postgres:5432/helm?user=helm&password=changemeforgodssake&sslmode=disable}'
+```
+
+**PRODUCTION NOTES**: it's recommended to change the username and password of
+the SQL database in production deployments. Enabling SSL is also a good idea.
+Last, but not least, perform regular backups/snapshots of your SQL database.
+
+Currently, if you want to switch from the default backend to the SQL backend,
+you'll have to do the migration for this on your own. When this backend
+graduates from beta, there will be a more official migration path.
+
 ## Conclusion
 
 In most cases, installation is as simple as getting a pre-built `helm` binary

docs/sql-storage.md

+# Store release information in an SQL database
+
+You may be willing to store release information in an SQL database - in
+particular, if your releases weigh more than 1MB and therefore [can't be stored in ConfigMaps or Secrets](https://github.com/helm/helm/issues/1413).
+
+We recommend using [PostgreSQL](https://www.postgresql.org/).
+
+This document describes how to deploy `postgres` atop Kubernetes. This being
+said, using an out-of-cluster (managed or not) PostreSQL instance is totally
+possible as well.
+
+Here's a Kubernetes manifest you can apply to get a minimal PostreSQL pod
+running on your Kubernetes cluster. **Don't forget to change the credentials
+and, optionally, enable TLS in production deployments**.
+
+```yaml
+apiVersion: v1
+kind: Service
+metadata:
+  name: tiller-postgres
+  namespace: kube-system
+spec:
+  ports:
+    - port: 5432
+  selector:
+    app: helm
+    name: postgres
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: tiller-postgres
+  namespace: kube-system
+spec:
+  serviceName: tiller-postgres
+  selector:
+    matchLabels:
+      app: helm
+      name: postgres
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: helm
+        name: postgres
+    spec:
+      containers:
+        - name: postgres
+          image: postgres:11-alpine
+          imagePullPolicy: Always
+          ports:
+            - containerPort: 5432
+          env:
+          - name: POSTGRES_DB
+            value: helm
+          - name: POSTGRES_USER
+            value: helm
+          - name: POSTGRES_PASSWORD
+            value: changemeforgodssake
+          - name: PGDATA
+            value: /var/lib/postgresql/data/pgdata
+          resources:
+            limits:
+              memory: 128Mi
+            requests:
+              cpu: 50m
+              memory: 128Mi
+          volumeMounts:
+            - mountPath: /var/lib/postgresql/data
+              name: tiller-postgres-data
+  volumeClaimTemplates:
+  - metadata:
+      name: tiller-postgres-data
+    spec:
+      accessModes: ["ReadWriteOnce"]
+      storageClassName: default
+      resources:
+        requests:
+          storage: 5Gi
+```
+
+Once postgres is deployed, you'll need to install Tiller using `helm init`, with
+a few custom CLI flags:
+
+```shell
+helm init \
+  --override \
+    'spec.template.spec.containers[0].args'='{--storage=sql,--sql-dialect=postgres,--sql-connection-string=postgresql://tiller-postgres:5432/helm?user=helm&password=changemeforgodssake&sslmode=disable}'
+```

glide.lock

@@ -173,10 +173,18 @@ imports:
   version: 9316a62528ac99aaecb4e47eadd6dc8aa6533d58
 - name: github.com/inconshreveable/mousetrap
   version: 76626ae9c91c4f2a10f34cad8ce83ea42c93bb75
+- name: github.com/jmoiron/sqlx
+  version: d161d7a76b5661016ad0b085869f77fd410f3e6a
+  subpackages:
+  - reflectx
 - name: github.com/json-iterator/go
   version: ab8a2e0c74be9d3be70b3184d9acc634935ded82
 - name: github.com/liggitt/tabwriter
   version: 89fcab3d43de07060e4fd4c1547430ed57e87f24
+- name: github.com/lib/pq
+  version: 88edab0803230a3898347e77b474f8c1820a1f20
+  subpackages:
+  - oid
 - name: github.com/mailru/easyjson
   version: 2f5df55504ebc322e4d52d34df6a1f5b503bf26d
   subpackages:
@@ -235,6 +243,10 @@ imports:
   version: 8a290539e2e8629dbc4e6bad948158f790ec31f4
 - name: github.com/PuerkitoBio/urlesc
   version: 5bd2802263f21d8788851d5305584c82a5c75d7e
+- name: github.com/rubenv/sql-migrate
+  version: 1007f53448d75fe14190968f5de4d95ed63ebb83
+  subpackages:
+  - sqlparse
 - name: github.com/russross/blackfriday
   version: 300106c228d52c8941d4b3de6054a6062a86dda3
 - name: github.com/shurcooL/sanitized_anchor_name
@@ -366,6 +378,8 @@ imports:
   - stats
   - status
   - tap
+- name: gopkg.in/gorp.v1
+  version: 6a667da9c028871f98598d85413e3fc4c6daa52e
 - name: gopkg.in/inf.v0
   version: 3887ee99ecf07df5b447e9b00d9c0b2adaa9f3e4
 - name: gopkg.in/square/go-jose.v2
@@ -807,6 +821,8 @@ imports:
   subpackages:
   - sortorder
 testImports:
+- name: github.com/DATA-DOG/go-sqlmock
+  version: 472e287dbafe67e526a3797165b64cb14f34705a
 - name: github.com/pmezard/go-difflib
   version: 5d4384ee4fb2527b0a1256a821ebfc92f91efefc
   subpackages:

glide.yaml

@@ -6,7 +6,6 @@ import:
   - package: golang.org/x/sync
     subpackages:
     - semaphore
-  # This is temporary and can probably be removed the next time gRPC is updated
   - package: golang.org/x/sys
     version: b90733256f2e882e81d52f9126de08df5615afd9
     subpackages:
@@ -17,7 +16,6 @@ import:
   - package: github.com/spf13/pflag
     version: ~1.0.1
   - package: github.com/Masterminds/vcs
-  # Pin version of mergo that is compatible with both sprig and Kubernetes
   - package: github.com/imdario/mergo
     version: v0.3.5
   - package: github.com/Masterminds/sprig
@@ -66,9 +64,14 @@ import:
     version: kubernetes-1.14.1
   - package: github.com/cyphar/filepath-securejoin
     version: ^0.2.1
+  - package: github.com/jmoiron/sqlx
+    version: ^1.2.0
+  - package: github.com/rubenv/sql-migrate
 
 testImports:
   - package: github.com/stretchr/testify
     version: ^1.1.4
     subpackages:
       - assert
+  - package: github.com/DATA-DOG/go-sqlmock
+    version: ^1.3.2

pkg/storage/driver/mock_test.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"testing"
 
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/jmoiron/sqlx"
 	"k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -221,3 +223,17 @@ func (mock *MockSecretsInterface) Delete(name string, opts *metav1.DeleteOptions
 	delete(mock.objects, name)
 	return nil
 }
+
+// newTestFixtureSQL mocks the SQL database (for testing purposes)
+func newTestFixtureSQL(t *testing.T, releases ...*rspb.Release) (*SQL, sqlmock.Sqlmock) {
+	sqlDB, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("error when opening stub database connection: %v", err)
+	}
+
+	sqlxDB := sqlx.NewDb(sqlDB, "sqlmock")
+	return &SQL{
+		db:  sqlxDB,
+		Log: func(_ string, _ ...interface{}) {},
+	}, mock
+}

pkg/storage/driver/sql.go

+/*
+Copyright The Helm 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 driver
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+	migrate "github.com/rubenv/sql-migrate"
+
+	// Import pq for potgres dialect
+	_ "github.com/lib/pq"
+
+	rspb "k8s.io/helm/pkg/proto/hapi/release"
+	storageerrors "k8s.io/helm/pkg/storage/errors"
+)
+
+var _ Driver = (*SQL)(nil)
+
+var labelMap = map[string]string{
+	"MODIFIED_AT": "modified_at",
+	"CREATED_AT":  "created_at",
+	"VERSION":     "version",
+	"STATUS":      "status",
+	"OWNER":       "owner",
+	"NAME":        "name",
+}
+
+// SQLDriverName is the string name of this driver.
+const SQLDriverName = "SQL"
+
+// SQL is the sql storage driver implementation.
+type SQL struct {
+	db  *sqlx.DB
+	Log func(string, ...interface{})
+}
+
+// Name returns the name of the driver.
+func (s *SQL) Name() string {
+	return SQLDriverName
+}
+
+func (s *SQL) ensureDBSetup() error {
+	// Populate the database with the relations we need if they don't exist yet
+	migrations := &migrate.MemoryMigrationSource{
+		Migrations: []*migrate.Migration{
+			{
+				Id: "init",
+				Up: []string{
+					`
+            CREATE TABLE releases (
+		      		key VARCHAR(67) PRIMARY KEY,
+              body TEXT NOT NULL,
+
+              name VARCHAR(64) NOT NULL,
+              version INTEGER NOT NULL,
+		      		status TEXT NOT NULL,
+		      		owner TEXT NOT NULL,
+		      		created_at INTEGER NOT NULL,
+		      		modified_at INTEGER NOT NULL DEFAULT 0
+            );
+
+            CREATE INDEX ON releases (key);
+            CREATE INDEX ON releases (version);
+            CREATE INDEX ON releases (status);
+            CREATE INDEX ON releases (owner);
+            CREATE INDEX ON releases (created_at);
+            CREATE INDEX ON releases (modified_at);
+		      `,
+				},
+				Down: []string{
+					`
+            DROP TABLE releases;
+          `,
+				},
+			},
+		},
+	}
+
+	_, err := migrate.Exec(s.db.DB, "postgres", migrations, migrate.Up)
+	return err
+}
+
+// Release describes a Helm release
+type Release struct {
+	Key  string `db:"key"`
+	Body string `db:"body"`
+
+	Name       string `db:"name"`
+	Version    int    `db:"version"`
+	Status     string `db:"status"`
+	Owner      string `db:"owner"`
+	CreatedAt  int    `db:"created_at"`
+	ModifiedAt int    `db:"modified_at"`
+}
+
+// NewSQL initializes a new memory driver.
+func NewSQL(dialect, connectionString string, logger func(string, ...interface{})) (*SQL, error) {
+	db, err := sqlx.Connect(dialect, connectionString)
+	if err != nil {
+		return nil, err
+	}
+
+	driver := &SQL{
+		db:  db,
+		Log: logger,
+	}
+
+	if err := driver.ensureDBSetup(); err != nil {
+		return nil, err
+	}
+
+	return driver, nil
+}
+
+// Get returns the release named by key.
+func (s *SQL) Get(key string) (*rspb.Release, error) {
+	var record Release
+	// Get will return an error if the result is empty
+	err := s.db.Get(&record, "SELECT body FROM releases WHERE key = $1", key)
+	if err != nil {
+		s.Log("got SQL error when getting release %s: %v", key, err)
+		return nil, storageerrors.ErrReleaseNotFound(key)
+	}
+
+	release, err := decodeRelease(record.Body)
+	if err != nil {
+		s.Log("get: failed to decode data %q: %v", key, err)
+		return nil, err
+	}
+
+	return release, nil
+}
+
+// List returns the list of all releases such that filter(release) == true
+func (s *SQL) List(filter func(*rspb.Release) bool) ([]*rspb.Release, error) {
+	var records = []Release{}
+	if err := s.db.Select(&records, "SELECT body FROM releases WHERE owner = 'TILLER'"); err != nil {
+		s.Log("list: failed to list: %v", err)
+		return nil, err
+	}
+
+	var releases []*rspb.Release
+	for _, record := range records {
+		release, err := decodeRelease(record.Body)
+		if err != nil {
+			s.Log("list: failed to decode release: %v: %v", record, err)
+			continue
+		}
+		if filter(release) {
+			releases = append(releases, release)
+		}
+	}
+
+	return releases, nil
+}
+
+// Query returns the set of releases that match the provided set of labels.
+func (s *SQL) Query(labels map[string]string) ([]*rspb.Release, error) {
+	var sqlFilterKeys []string
+	sqlFilter := map[string]interface{}{}
+	for key, val := range labels {
+		// Build a slice of where filters e.g
+		// labels = map[string]string{ "foo": "foo", "bar": "bar" }
+		// []string{ "foo=?", "bar=?" }
+		if dbField, ok := labelMap[key]; ok {
+			sqlFilterKeys = append(sqlFilterKeys, strings.Join([]string{dbField, "=:", dbField}, ""))
+			sqlFilter[dbField] = val
+		} else {
+			s.Log("unknown label %s", key)
+			return nil, fmt.Errorf("unknow label %s", key)
+		}
+	}
+	sort.Strings(sqlFilterKeys)
+
+	// Build our query
+	query := strings.Join([]string{
+		"SELECT body FROM releases",
+		"WHERE",
+		strings.Join(sqlFilterKeys, " AND "),
+	}, " ")
+
+	rows, err := s.db.NamedQuery(query, sqlFilter)
+	if err != nil {
+		s.Log("failed to query with labels: %v", err)
+		return nil, err
+	}
+
+	var releases []*rspb.Release
+	for rows.Next() {
+		var record Release
+		if err = rows.StructScan(&record); err != nil {
+			s.Log("failed to scan record %q: %v", record, err)
+			return nil, err
+		}
+
+		release, err := decodeRelease(record.Body)
+		if err != nil {
+			s.Log("failed to decode release: %v", err)
+			continue
+		}
+		releases = append(releases, release)
+	}
+
+	if len(releases) == 0 {
+		return nil, storageerrors.ErrReleaseNotFound(labels["NAME"])
+	}
+
+	return releases, nil
+}
+
+// Create creates a new release.
+func (s *SQL) Create(key string, rls *rspb.Release) error {
+	body, err := encodeRelease(rls)
+	if err != nil {
+		s.Log("failed to encode release: %v", err)
+		return err
+	}
+
+	transaction, err := s.db.Beginx()
+	if err != nil {
+		s.Log("failed to start SQL transaction: %v", err)
+		return fmt.Errorf("error beginning transaction: %v", err)
+	}
+
+	if _, err := transaction.NamedExec("INSERT INTO releases (key, body, name, version, status, owner, created_at) VALUES (:key, :body, :name, :version, :status, :owner, :created_at)",
+		&Release{
+			Key:  key,
+			Body: body,
+
+			Name:      rls.Name,
+			Version:   int(rls.Version),
+			Status:    rspb.Status_Code_name[int32(rls.Info.Status.Code)],
+			Owner:     "TILLER",
+			CreatedAt: int(time.Now().Unix()),
+		},
+	); err != nil {
+		defer transaction.Rollback()
+		var record Release
+		if err := transaction.Get(&record, "SELECT key FROM releases WHERE key = ?", key); err == nil {
+			s.Log("release %s already exists", key)
+			return storageerrors.ErrReleaseExists(key)
+		}
+
+		s.Log("failed to store release %s in SQL database: %v", key, err)
+		return err
+	}
+	defer transaction.Commit()
+
+	return nil
+}
+
+// Update updates a release.
+func (s *SQL) Update(key string, rls *rspb.Release) error {
+	body, err := encodeRelease(rls)
+	if err != nil {
+		s.Log("failed to encode release: %v", err)
+		return err
+	}
+
+	if _, err := s.db.NamedExec("UPDATE releases SET body=:body, name=:name, version=:version, status=:status, owner=:owner, modified_at=:modified_at WHERE key=:key",
+		&Release{
+			Key:  key,
+			Body: body,
+
+			Name:       rls.Name,
+			Version:    int(rls.Version),
+			Status:     rspb.Status_Code_name[int32(rls.Info.Status.Code)],
+			Owner:      "TILLER",
+			ModifiedAt: int(time.Now().Unix()),
+		},
+	); err != nil {
+		s.Log("failed to update release %s in SQL database: %v", key, err)
+		return err
+	}
+
+	return nil
+}
+
+// Delete deletes a release or returns ErrReleaseNotFound.
+func (s *SQL) Delete(key string) (*rspb.Release, error) {
+	transaction, err := s.db.Beginx()
+	if err != nil {
+		s.Log("failed to start SQL transaction: %v", err)
+		return nil, fmt.Errorf("error beginning transaction: %v", err)
+	}
+
+	var record Release
+	err = transaction.Get(&record, "SELECT body FROM releases WHERE key = $1", key)
+	if err != nil {
+		s.Log("release %s not found: %v", key, err)
+		return nil, storageerrors.ErrReleaseNotFound(key)
+	}
+
+	release, err := decodeRelease(record.Body)
+	if err != nil {
+		s.Log("failed to decode release %s: %v", key, err)
+		transaction.Rollback()
+		return nil, err
+	}
+	defer transaction.Commit()
+
+	_, err = transaction.Exec("DELETE FROM releases WHERE key = $1", key)
+	return release, err
+}

pkg/storage/driver/sql_test.go

+/*
+Copyright The Helm 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 driver
+
+import (
+	"fmt"
+	"reflect"
+	"regexp"
+	"testing"
+	"time"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	rspb "k8s.io/helm/pkg/proto/hapi/release"
+)
+
+func TestSQLName(t *testing.T) {
+	sqlDriver, _ := newTestFixtureSQL(t)
+	if sqlDriver.Name() != SQLDriverName {
+		t.Errorf("Expected name to be %q, got %q", SQLDriverName, sqlDriver.Name())
+	}
+}
+
+func TestSQLGet(t *testing.T) {
+	vers := int32(1)
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+	rel := releaseStub(name, vers, namespace, rspb.Status_DEPLOYED)
+
+	body, _ := encodeRelease(rel)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+	mock.
+		ExpectQuery("SELECT body FROM releases WHERE key = ?").
+		WithArgs(key).
+		WillReturnRows(
+			mock.NewRows([]string{
+				"body",
+			}).AddRow(
+				body,
+			),
+		).RowsWillBeClosed()
+
+	got, err := sqlDriver.Get(key)
+	if err != nil {
+		t.Fatalf("Failed to get release: %v", err)
+	}
+
+	if !reflect.DeepEqual(rel, got) {
+		t.Errorf("Expected release {%q}, got {%q}", rel, got)
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}
+
+func TestSQLList(t *testing.T) {
+	body1, _ := encodeRelease(releaseStub("key-1", 1, "default", rspb.Status_DELETED))
+	body2, _ := encodeRelease(releaseStub("key-2", 1, "default", rspb.Status_DELETED))
+	body3, _ := encodeRelease(releaseStub("key-3", 1, "default", rspb.Status_DEPLOYED))
+	body4, _ := encodeRelease(releaseStub("key-4", 1, "default", rspb.Status_DEPLOYED))
+	body5, _ := encodeRelease(releaseStub("key-5", 1, "default", rspb.Status_SUPERSEDED))
+	body6, _ := encodeRelease(releaseStub("key-6", 1, "default", rspb.Status_SUPERSEDED))
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+
+	for i := 0; i < 3; i++ {
+		mock.
+			ExpectQuery("SELECT body FROM releases WHERE owner = 'TILLER'").
+			WillReturnRows(
+				mock.NewRows([]string{
+					"body",
+				}).
+					AddRow(body1).
+					AddRow(body2).
+					AddRow(body3).
+					AddRow(body4).
+					AddRow(body5).
+					AddRow(body6),
+			).RowsWillBeClosed()
+	}
+
+	// list all deleted releases
+	del, err := sqlDriver.List(func(rel *rspb.Release) bool {
+		return rel.Info.Status.Code == rspb.Status_DELETED
+	})
+	// check
+	if err != nil {
+		t.Errorf("Failed to list deleted: %v", err)
+	}
+	if len(del) != 2 {
+		t.Errorf("Expected 2 deleted, got %d:\n%v\n", len(del), del)
+	}
+
+	// list all deployed releases
+	dpl, err := sqlDriver.List(func(rel *rspb.Release) bool {
+		return rel.Info.Status.Code == rspb.Status_DEPLOYED
+	})
+	// check
+	if err != nil {
+		t.Errorf("Failed to list deployed: %v", err)
+	}
+	if len(dpl) != 2 {
+		t.Errorf("Expected 2 deployed, got %d:\n%v\n", len(dpl), dpl)
+	}
+
+	// list all superseded releases
+	ssd, err := sqlDriver.List(func(rel *rspb.Release) bool {
+		return rel.Info.Status.Code == rspb.Status_SUPERSEDED
+	})
+	// check
+	if err != nil {
+		t.Errorf("Failed to list superseded: %v", err)
+	}
+	if len(ssd) != 2 {
+		t.Errorf("Expected 2 superseded, got %d:\n%v\n", len(ssd), ssd)
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}
+
+func TestSqlCreate(t *testing.T) {
+	vers := int32(1)
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+	rel := releaseStub(name, vers, namespace, rspb.Status_DEPLOYED)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+	body, _ := encodeRelease(rel)
+
+	mock.ExpectBegin()
+	mock.
+		ExpectExec(regexp.QuoteMeta("INSERT INTO releases (key, body, name, version, status, owner, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)")).
+		WithArgs(key, body, rel.Name, int(rel.Version), rspb.Status_Code_name[int32(rel.Info.Status.Code)], "TILLER", int(time.Now().Unix())).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectCommit()
+
+	if err := sqlDriver.Create(key, rel); err != nil {
+		t.Fatalf("failed to create release with key %q: %v", key, err)
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}
+
+func TestSqlCreateAlreadyExists(t *testing.T) {
+	vers := int32(1)
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+	rel := releaseStub(name, vers, namespace, rspb.Status_DEPLOYED)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+	body, _ := encodeRelease(rel)
+
+	// Insert fails (primary key already exists)
+	mock.ExpectBegin()
+	mock.
+		ExpectExec(regexp.QuoteMeta("INSERT INTO releases (key, body, name, version, status, owner, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)")).
+		WithArgs(key, body, rel.Name, int(rel.Version), rspb.Status_Code_name[int32(rel.Info.Status.Code)], "TILLER", int(time.Now().Unix())).
+		WillReturnError(fmt.Errorf("dialect dependent SQL error"))
+
+	// Let's check that we do make sure the error is due to a release already existing
+	mock.
+		ExpectQuery(regexp.QuoteMeta("SELECT key FROM releases WHERE key = ?")).
+		WithArgs(key).
+		WillReturnRows(
+			mock.NewRows([]string{
+				"body",
+			}).AddRow(
+				body,
+			),
+		).RowsWillBeClosed()
+	mock.ExpectRollback()
+
+	if err := sqlDriver.Create(key, rel); err == nil {
+		t.Fatalf("failed to create release with key %q: %v", key, err)
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}
+
+func TestSqlUpdate(t *testing.T) {
+	vers := int32(1)
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+	rel := releaseStub(name, vers, namespace, rspb.Status_DEPLOYED)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+	body, _ := encodeRelease(rel)
+
+	mock.
+		ExpectExec(regexp.QuoteMeta("UPDATE releases SET body=?, name=?, version=?, status=?, owner=?, modified_at=? WHERE key=?")).
+		WithArgs(body, rel.Name, int(rel.Version), rspb.Status_Code_name[int32(rel.Info.Status.Code)], "TILLER", int(time.Now().Unix()), key).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	if err := sqlDriver.Update(key, rel); err != nil {
+		t.Fatalf("failed to update release with key %q: %v", key, err)
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}
+
+func TestSqlQuery(t *testing.T) {
+	// Reflect actual use cases in ../storage.go
+	labelSetDeployed := map[string]string{
+		"NAME":   "smug-pigeon",
+		"OWNER":  "TILLER",
+		"STATUS": "DEPLOYED",
+	}
+	labelSetAll := map[string]string{
+		"NAME":  "smug-pigeon",
+		"OWNER": "TILLER",
+	}
+
+	supersededRelease := releaseStub("smug-pigeon", 1, "default", rspb.Status_SUPERSEDED)
+	supersededReleaseBody, _ := encodeRelease(supersededRelease)
+	deployedRelease := releaseStub("smug-pigeon", 2, "default", rspb.Status_DEPLOYED)
+	deployedReleaseBody, _ := encodeRelease(deployedRelease)
+
+	// Let's actually start our test
+	sqlDriver, mock := newTestFixtureSQL(t)
+
+	mock.
+		ExpectQuery(regexp.QuoteMeta("SELECT body FROM releases WHERE name=? AND owner=? AND status=?")).
+		WithArgs("smug-pigeon", "TILLER", "DEPLOYED").
+		WillReturnRows(
+			mock.NewRows([]string{
+				"body",
+			}).AddRow(
+				deployedReleaseBody,
+			),
+		).RowsWillBeClosed()
+
+	mock.
+		ExpectQuery(regexp.QuoteMeta("SELECT body FROM releases WHERE name=? AND owner=?")).
+		WithArgs("smug-pigeon", "TILLER").
+		WillReturnRows(
+			mock.NewRows([]string{
+				"body",
+			}).AddRow(
+				supersededReleaseBody,
+			).AddRow(
+				deployedReleaseBody,
+			),
+		).RowsWillBeClosed()
+
+	results, err := sqlDriver.Query(labelSetDeployed)
+	if err != nil {
+		t.Fatalf("failed to query for deployed smug-pigeon release: %v", err)
+	}
+
+	for _, res := range results {
+		if !reflect.DeepEqual(res, deployedRelease) {
+			t.Errorf("Expected release {%q}, got {%q}", deployedRelease, res)
+		}
+	}
+
+	results, err = sqlDriver.Query(labelSetAll)
+	if err != nil {
+		t.Fatalf("failed to query release history for smug-pigeon: %v", err)
+	}
+
+	if len(results) != 2 {
+		t.Errorf("expected a resultset of size 2, got %d", len(results))
+	}
+
+	for _, res := range results {
+		if !reflect.DeepEqual(res, deployedRelease) && !reflect.DeepEqual(res, supersededRelease) {
+			t.Errorf("Expected release {%q} or {%q}, got {%q}", deployedRelease, supersededRelease, res)
+		}
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}
+
+func TestSqlDelete(t *testing.T) {
+	vers := int32(1)
+	name := "smug-pigeon"
+	namespace := "default"
+	key := testKey(name, vers)
+	rel := releaseStub(name, vers, namespace, rspb.Status_DEPLOYED)
+
+	body, _ := encodeRelease(rel)
+
+	sqlDriver, mock := newTestFixtureSQL(t)
+
+	mock.ExpectBegin()
+	mock.
+		ExpectQuery("SELECT body FROM releases WHERE key = ?").
+		WithArgs(key).
+		WillReturnRows(
+			mock.NewRows([]string{
+				"body",
+			}).AddRow(
+				body,
+			),
+		).RowsWillBeClosed()
+
+	mock.
+		ExpectExec(regexp.QuoteMeta("DELETE FROM releases WHERE key = $1")).
+		WithArgs(key).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectCommit()
+
+	deletedRelease, err := sqlDriver.Delete(key)
+	if err != nil {
+		t.Fatalf("failed to delete release with key %q: %v", key, err)
+	}
+
+	if !reflect.DeepEqual(rel, deletedRelease) {
+		t.Errorf("Expected release {%q}, got {%q}", rel, deletedRelease)
+	}
+
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Errorf("sql expectations weren't met: %v", err)
+	}
+}