update pkg/storage to support rollback

reviewable/pr1155/r1
fibonacci1729 8 years ago
parent 36606cf152
commit 383a9c186a

@ -30,21 +30,16 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
kberrs "k8s.io/kubernetes/pkg/api/errors" kberrs "k8s.io/kubernetes/pkg/api/errors"
client "k8s.io/kubernetes/pkg/client/unversioned" client "k8s.io/kubernetes/pkg/client/unversioned"
kblabels "k8s.io/kubernetes/pkg/labels"
) )
var _ Driver = (*ConfigMaps)(nil)
// ConfigMapsDriverName is the string name of the driver. // ConfigMapsDriverName is the string name of the driver.
const ConfigMapsDriverName = "ConfigMap" const ConfigMapsDriverName = "ConfigMap"
var b64 = base64.StdEncoding var b64 = base64.StdEncoding
// labels is a map of key value pairs to be included as metadata in a configmap object.
type labels map[string]string
func (lbs *labels) init() { *lbs = labels(make(map[string]string)) }
func (lbs labels) get(key string) string { return lbs[key] }
func (lbs labels) set(key, val string) { lbs[key] = val }
func (lbs labels) toMap() map[string]string { return lbs }
// ConfigMaps is a wrapper around an implementation of a kubernetes // ConfigMaps is a wrapper around an implementation of a kubernetes
// ConfigMapsInterface. // ConfigMapsInterface.
type ConfigMaps struct { type ConfigMaps struct {
@ -112,9 +107,41 @@ func (cfgmaps *ConfigMaps) List(filter func(*rspb.Release) bool) ([]*rspb.Releas
return results, nil return results, nil
} }
// Query fetches all releases that match the provided map of labels.
// An error is returned if the configmap fails to retrieve the releases.
func (cfgmaps *ConfigMaps) Query(labels map[string]string) ([]*rspb.Release, error) {
ls := kblabels.Set{}
for k, v := range labels {
ls[k] = v
}
opts := api.ListOptions{LabelSelector: ls.AsSelector()}
list, err := cfgmaps.impl.List(opts)
if err != nil {
logerrf(err, "query: failed to query with labels")
return nil, err
}
if len(list.Items) == 0 {
return nil, ErrReleaseNotFound
}
var results []*rspb.Release
for _, item := range list.Items {
rls, err := decodeRelease(item.Data["release"])
if err != nil {
logerrf(err, "query: failed to decode release: %s", err)
continue
}
results = append(results, rls)
}
return results, nil
}
// Create creates a new ConfigMap holding the release. If the // Create creates a new ConfigMap holding the release. If the
// ConfigMap already exists, ErrReleaseExists is returned. // ConfigMap already exists, ErrReleaseExists is returned.
func (cfgmaps *ConfigMaps) Create(rls *rspb.Release) error { func (cfgmaps *ConfigMaps) Create(key string, rls *rspb.Release) error {
// set labels for configmaps object meta data // set labels for configmaps object meta data
var lbs labels var lbs labels
@ -122,7 +149,7 @@ func (cfgmaps *ConfigMaps) Create(rls *rspb.Release) error {
lbs.set("CREATED_AT", strconv.Itoa(int(time.Now().Unix()))) lbs.set("CREATED_AT", strconv.Itoa(int(time.Now().Unix())))
// create a new configmap to hold the release // create a new configmap to hold the release
obj, err := newConfigMapsObject(rls, lbs) obj, err := newConfigMapsObject(key, rls, lbs)
if err != nil { if err != nil {
logerrf(err, "create: failed to encode release %q", rls.Name) logerrf(err, "create: failed to encode release %q", rls.Name)
return err return err
@ -141,7 +168,7 @@ func (cfgmaps *ConfigMaps) Create(rls *rspb.Release) error {
// Update updates the ConfigMap holding the release. If not found // Update updates the ConfigMap holding the release. If not found
// the ConfigMap is created to hold the release. // the ConfigMap is created to hold the release.
func (cfgmaps *ConfigMaps) Update(rls *rspb.Release) error { func (cfgmaps *ConfigMaps) Update(key string, rls *rspb.Release) error {
// set labels for configmaps object meta data // set labels for configmaps object meta data
var lbs labels var lbs labels
@ -149,7 +176,7 @@ func (cfgmaps *ConfigMaps) Update(rls *rspb.Release) error {
lbs.set("MODIFIED_AT", strconv.Itoa(int(time.Now().Unix()))) lbs.set("MODIFIED_AT", strconv.Itoa(int(time.Now().Unix())))
// create a new configmap object to hold the release // create a new configmap object to hold the release
obj, err := newConfigMapsObject(rls, lbs) obj, err := newConfigMapsObject(key, rls, lbs)
if err != nil { if err != nil {
logerrf(err, "update: failed to encode release %q", rls.Name) logerrf(err, "update: failed to encode release %q", rls.Name)
return err return err
@ -194,7 +221,7 @@ func (cfgmaps *ConfigMaps) Delete(key string) (rls *rspb.Release, err error) {
// "OWNER" - owner of the configmap, currently "TILLER". // "OWNER" - owner of the configmap, currently "TILLER".
// "NAME" - name of the release. // "NAME" - name of the release.
// //
func newConfigMapsObject(rls *rspb.Release, lbs labels) (*api.ConfigMap, error) { func newConfigMapsObject(key string, rls *rspb.Release, lbs labels) (*api.ConfigMap, error) {
const owner = "TILLER" const owner = "TILLER"
// encode the release // encode the release
@ -216,7 +243,7 @@ func newConfigMapsObject(rls *rspb.Release, lbs labels) (*api.ConfigMap, error)
// create and return configmap object // create and return configmap object
return &api.ConfigMap{ return &api.ConfigMap{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
Name: rls.Name, Name: key,
Labels: lbs.toMap(), Labels: lbs.toMap(),
}, },
Data: map[string]string{"release": s}, Data: map[string]string{"release": s},

@ -1,12 +1,9 @@
/* /*
Copyright 2016 The Kubernetes Authors All rights reserved. Copyright 2016 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License"); Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License. you may not use this file except in compliance with the License.
You may obtain a copy of the License at You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0 http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS, distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@ -21,26 +18,22 @@ import (
"testing" "testing"
rspb "k8s.io/helm/pkg/proto/hapi/release" rspb "k8s.io/helm/pkg/proto/hapi/release"
"k8s.io/kubernetes/pkg/api"
kberrs "k8s.io/kubernetes/pkg/api/errors"
"k8s.io/kubernetes/pkg/client/unversioned"
) )
var _ Driver = &ConfigMaps{}
func TestConfigMapName(t *testing.T) { func TestConfigMapName(t *testing.T) {
c := newTestFixture(t) c := newTestFixtureCfgMaps(t)
if c.Name() != ConfigMapsDriverName { if c.Name() != ConfigMapsDriverName {
t.Errorf("Expected name to be %q, got %q", ConfigMapsDriverName, c.Name()) t.Errorf("Expected name to be %q, got %q", ConfigMapsDriverName, c.Name())
} }
} }
func TestConfigMapGet(t *testing.T) { func TestConfigMapGet(t *testing.T) {
key := "key-1" vers := int32(1)
rel := newTestRelease(key, 1, rspb.Status_DEPLOYED) name := "smug-pigeon"
key := testKey(name, vers)
rel := releaseStub(name, vers, rspb.Status_DEPLOYED)
cfgmaps := newTestFixture(t, []*rspb.Release{rel}...) cfgmaps := newTestFixtureCfgMaps(t, []*rspb.Release{rel}...)
// get release with key // get release with key
got, err := cfgmaps.Get(key) got, err := cfgmaps.Get(key)
@ -54,13 +47,13 @@ func TestConfigMapGet(t *testing.T) {
} }
func TestConfigMapList(t *testing.T) { func TestConfigMapList(t *testing.T) {
cfgmaps := newTestFixture(t, []*rspb.Release{ cfgmaps := newTestFixtureCfgMaps(t, []*rspb.Release{
newTestRelease("key-1", 1, rspb.Status_DELETED), releaseStub("key-1", 1, rspb.Status_DELETED),
newTestRelease("key-2", 1, rspb.Status_DELETED), releaseStub("key-2", 1, rspb.Status_DELETED),
newTestRelease("key-3", 1, rspb.Status_DEPLOYED), releaseStub("key-3", 1, rspb.Status_DEPLOYED),
newTestRelease("key-4", 1, rspb.Status_DEPLOYED), releaseStub("key-4", 1, rspb.Status_DEPLOYED),
newTestRelease("key-5", 1, rspb.Status_SUPERSEDED), releaseStub("key-5", 1, rspb.Status_SUPERSEDED),
newTestRelease("key-6", 1, rspb.Status_SUPERSEDED), releaseStub("key-6", 1, rspb.Status_SUPERSEDED),
}...) }...)
// list all deleted releases // list all deleted releases
@ -101,13 +94,15 @@ func TestConfigMapList(t *testing.T) {
} }
func TestConfigMapCreate(t *testing.T) { func TestConfigMapCreate(t *testing.T) {
cfgmaps := newTestFixture(t) cfgmaps := newTestFixtureCfgMaps(t)
key := "key-1" vers := int32(1)
rel := newTestRelease(key, 1, rspb.Status_DEPLOYED) name := "smug-pigeon"
key := testKey(name, vers)
rel := releaseStub(name, vers, rspb.Status_DEPLOYED)
// store the release in a configmap // store the release in a configmap
if err := cfgmaps.Create(rel); err != nil { if err := cfgmaps.Create(key, rel); err != nil {
t.Fatalf("Failed to create release with key %q: %s", key, err) t.Fatalf("Failed to create release with key %q: %s", key, err)
} }
@ -124,16 +119,18 @@ func TestConfigMapCreate(t *testing.T) {
} }
func TestConfigMapUpdate(t *testing.T) { func TestConfigMapUpdate(t *testing.T) {
key := "key-1" vers := int32(1)
rel := newTestRelease(key, 1, rspb.Status_DEPLOYED) name := "smug-pigeon"
key := testKey(name, vers)
rel := releaseStub(name, vers, rspb.Status_DEPLOYED)
cfgmaps := newTestFixture(t, []*rspb.Release{rel}...) cfgmaps := newTestFixtureCfgMaps(t, []*rspb.Release{rel}...)
// modify release status code & version // modify release status code
rel = newTestRelease(key, 2, rspb.Status_SUPERSEDED) rel.Info.Status.Code = rspb.Status_SUPERSEDED
// perform the update // perform the update
if err := cfgmaps.Update(rel); err != nil { if err := cfgmaps.Update(key, rel); err != nil {
t.Fatalf("Failed to update release: %s", err) t.Fatalf("Failed to update release: %s", err)
} }
@ -144,85 +141,7 @@ func TestConfigMapUpdate(t *testing.T) {
} }
// check release has actually been updated by comparing modified fields // check release has actually been updated by comparing modified fields
switch { if rel.Info.Status.Code != got.Info.Status.Code {
case rel.Info.Status.Code != got.Info.Status.Code:
t.Errorf("Expected status %s, got status %s", rel.Info.Status.Code, got.Info.Status.Code) t.Errorf("Expected status %s, got status %s", rel.Info.Status.Code, got.Info.Status.Code)
case rel.Version != got.Version:
t.Errorf("Expected version %d, got version %d", rel.Version, got.Version)
}
}
// newTestFixture initializes a MockConfigMapsInterface.
// ConfigMaps are created for each release provided.
func newTestFixture(t *testing.T, releases ...*rspb.Release) *ConfigMaps {
var mock MockConfigMapsInterface
mock.Init(t, releases...)
return NewConfigMaps(&mock)
}
// newTestRelease creates a release object for testing.
func newTestRelease(key string, version int32, status rspb.Status_Code) *rspb.Release {
return &rspb.Release{Name: key, Info: &rspb.Info{Status: &rspb.Status{Code: status}}, Version: version}
}
// MockConfigMapsInterface mocks a kubernetes ConfigMapsInterface
type MockConfigMapsInterface struct {
unversioned.ConfigMapsInterface
objects map[string]*api.ConfigMap
}
func (mock *MockConfigMapsInterface) Init(t *testing.T, releases ...*rspb.Release) {
mock.objects = map[string]*api.ConfigMap{}
for _, rls := range releases {
cfgmap, err := newConfigMapsObject(rls, nil)
if err != nil {
t.Fatalf("Failed to create configmap: %s", err)
}
mock.objects[rls.Name] = cfgmap
}
}
func (mock *MockConfigMapsInterface) Get(name string) (*api.ConfigMap, error) {
object, ok := mock.objects[name]
if !ok {
return nil, kberrs.NewNotFound(api.Resource("tests"), name)
}
return object, nil
}
func (mock *MockConfigMapsInterface) List(opts api.ListOptions) (*api.ConfigMapList, error) {
var list api.ConfigMapList
for _, cfgmap := range mock.objects {
list.Items = append(list.Items, *cfgmap)
}
return &list, nil
}
func (mock *MockConfigMapsInterface) Create(cfgmap *api.ConfigMap) (*api.ConfigMap, error) {
name := cfgmap.ObjectMeta.Name
if object, ok := mock.objects[name]; ok {
return object, kberrs.NewAlreadyExists(api.Resource("tests"), name)
}
mock.objects[name] = cfgmap
return cfgmap, nil
}
func (mock *MockConfigMapsInterface) Update(cfgmap *api.ConfigMap) (*api.ConfigMap, error) {
name := cfgmap.ObjectMeta.Name
if _, ok := mock.objects[name]; !ok {
return nil, kberrs.NewNotFound(api.Resource("tests"), name)
}
mock.objects[name] = cfgmap
return cfgmap, nil
}
func (mock *MockConfigMapsInterface) Delete(name string) error {
if _, ok := mock.objects[name]; !ok {
return kberrs.NewNotFound(api.Resource("tests"), name)
} }
delete(mock.objects, name)
return nil
} }

@ -27,6 +27,8 @@ var (
ErrReleaseNotFound = errors.New("release: not found") ErrReleaseNotFound = errors.New("release: not found")
// ErrReleaseExists indicates that a release already exists. // ErrReleaseExists indicates that a release already exists.
ErrReleaseExists = errors.New("release: already exists") ErrReleaseExists = errors.New("release: already exists")
// ErrInvalidKey indicates that a release key could not be parsed.
ErrInvalidKey = errors.New("release: invalid key")
) )
// Creator is the interface that wraps the Create method. // Creator is the interface that wraps the Create method.
@ -34,7 +36,7 @@ var (
// Create stores the release or returns ErrReleaseExists // Create stores the release or returns ErrReleaseExists
// if an identical release already exists. // if an identical release already exists.
type Creator interface { type Creator interface {
Create(rls *rspb.Release) error Create(key string, rls *rspb.Release) error
} }
// Updator is the interface that wraps the Update method. // Updator is the interface that wraps the Update method.
@ -42,7 +44,7 @@ type Creator interface {
// Update updates an existing release or returns // Update updates an existing release or returns
// ErrReleaseNotFound if the release does not exist. // ErrReleaseNotFound if the release does not exist.
type Updator interface { type Updator interface {
Update(rls *rspb.Release) error Update(key string, rls *rspb.Release) error
} }
// Deletor is the interface that wraps the Delete method. // Deletor is the interface that wraps the Delete method.
@ -59,9 +61,12 @@ type Deletor interface {
// if the release does not exist. // if the release does not exist.
// //
// List returns the set of all releases that satisfy the filter predicate. // List returns the set of all releases that satisfy the filter predicate.
//
// Query returns the set of all releases that match the provided label set.
type Queryor interface { type Queryor interface {
Get(key string) (*rspb.Release, error) Get(key string) (*rspb.Release, error)
List(filter func(*rspb.Release) bool) ([]*rspb.Release, error) List(filter func(*rspb.Release) bool) ([]*rspb.Release, error)
Query(labels map[string]string) ([]*rspb.Release, error)
} }
// Driver is the interface composed of Creator, Updator, Deletor, Queryor // Driver is the interface composed of Creator, Updator, Deletor, Queryor

@ -0,0 +1,50 @@
package driver
import (
"bytes"
"fmt"
"io"
)
// labels is a map of key value pairs to be included as metadata in a configmap object.
type labels map[string]string
func (lbs *labels) init() { *lbs = labels(make(map[string]string)) }
func (lbs labels) get(key string) string { return lbs[key] }
func (lbs labels) set(key, val string) { lbs[key] = val }
func (lbs labels) keys() (ls []string) {
for key := range lbs {
ls = append(ls, key)
}
return
}
func (lbs labels) match(set labels) bool {
for _, key := range set.keys() {
if lbs.get(key) != set.get(key) {
return false
}
}
return true
}
func (lbs labels) toMap() map[string]string { return lbs }
func (lbs *labels) fromMap(kvs map[string]string) {
for k, v := range kvs {
lbs.set(k, v)
}
}
func (lbs labels) dump(w io.Writer) error {
var b bytes.Buffer
fmt.Fprintln(&b, "labels:")
for k, v := range lbs {
fmt.Fprintf(&b, "\t- %q -> %q\n", k, v)
}
_, err := w.Write(b.Bytes())
return err
}

@ -0,0 +1,33 @@
package driver
import (
"testing"
)
func TestLabelsMatch(t *testing.T) {
var tests = []struct {
desc string
set1 labels
set2 labels
expect bool
}{
{
"equal labels sets",
labels(map[string]string{"KEY_A": "VAL_A", "KEY_B": "VAL_B"}),
labels(map[string]string{"KEY_A": "VAL_A", "KEY_B": "VAL_B"}),
true,
},
{
"disjoint label sets",
labels(map[string]string{"KEY_C": "VAL_C", "KEY_D": "VAL_D"}),
labels(map[string]string{"KEY_A": "VAL_A", "KEY_B": "VAL_B"}),
false,
},
}
for _, tt := range tests {
if !tt.set1.match(tt.set2) && tt.expect {
t.Fatalf("Expected match '%s'\n", tt.desc)
}
}
}

@ -14,26 +14,33 @@ See the License for the specific language governing permissions and
limitations under the License. limitations under the License.
*/ */
package driver // import "k8s.io/helm/pkg/storage/driver" package driver
import ( import (
"bytes"
"fmt"
"io"
"strconv"
"strings"
"sync" "sync"
rspb "k8s.io/helm/pkg/proto/hapi/release" rspb "k8s.io/helm/pkg/proto/hapi/release"
) )
var _ Driver = (*Memory)(nil)
// MemoryDriverName is the string name of this driver. // MemoryDriverName is the string name of this driver.
const MemoryDriverName = "Memory" const MemoryDriverName = "Memory"
// Memory is the in-memory storage driver implementation. // Memory is the in-memory storage driver implementation.
type Memory struct { type Memory struct {
sync.RWMutex sync.RWMutex
cache map[string]*rspb.Release cache map[string]records
} }
// NewMemory initializes a new memory driver. // NewMemory initializes a new memory driver.
func NewMemory() *Memory { func NewMemory() *Memory {
return &Memory{cache: map[string]*rspb.Release{}} return &Memory{cache: map[string]records{}}
} }
// Name returns the name of the driver. // Name returns the name of the driver.
@ -45,42 +52,81 @@ func (mem *Memory) Name() string {
func (mem *Memory) Get(key string) (*rspb.Release, error) { func (mem *Memory) Get(key string) (*rspb.Release, error) {
defer unlock(mem.rlock()) defer unlock(mem.rlock())
if rls, ok := mem.cache[key]; ok { switch elems := strings.Split(key, ".v"); len(elems) {
return rls, nil case 2:
name, ver := elems[0], elems[1]
if _, err := strconv.Atoi(ver); err != nil {
return nil, ErrInvalidKey
}
if recs, ok := mem.cache[name]; ok {
if r := recs.Get(key); r != nil {
return r.rls, nil
}
} }
return nil, ErrReleaseNotFound return nil, ErrReleaseNotFound
default:
return nil, ErrInvalidKey
}
} }
// List returns the list of all releases such that filter(release) == true // List returns the list of all releases such that filter(release) == true
func (mem *Memory) List(filter func(*rspb.Release) bool) ([]*rspb.Release, error) { func (mem *Memory) List(filter func(*rspb.Release) bool) ([]*rspb.Release, error) {
defer unlock(mem.rlock()) defer unlock(mem.rlock())
var releases []*rspb.Release var ls []*rspb.Release
for k := range mem.cache { for _, recs := range mem.cache {
if filter(mem.cache[k]) { recs.Iter(func(_ int, rec *record) bool {
releases = append(releases, mem.cache[k]) if filter(rec.rls) {
ls = append(ls, rec.rls)
}
return true
})
}
return ls, nil
}
// Query returns the set of releases that match the provided set of labels
func (mem *Memory) Query(keyvals map[string]string) ([]*rspb.Release, error) {
defer unlock(mem.rlock())
var lbs labels
lbs.init()
lbs.fromMap(keyvals)
var ls []*rspb.Release
for _, recs := range mem.cache {
recs.Iter(func(_ int, rec *record) bool {
if rec.lbs.match(lbs) {
ls = append(ls, rec.rls)
} }
return true
})
} }
return releases, nil return ls, nil
} }
// Create creates a new release or returns ErrReleaseExists. // Create creates a new release or returns ErrReleaseExists.
func (mem *Memory) Create(rls *rspb.Release) error { func (mem *Memory) Create(key string, rls *rspb.Release) error {
defer unlock(mem.wlock()) defer unlock(mem.wlock())
if _, ok := mem.cache[rls.Name]; ok { if recs, ok := mem.cache[rls.Name]; ok {
return ErrReleaseExists if err := recs.Add(newRecord(key, rls)); err != nil {
return err
}
mem.cache[rls.Name] = recs
return nil
} }
mem.cache[rls.Name] = rls mem.cache[rls.Name] = records{newRecord(key, rls)}
return nil return nil
} }
// Update updates a release or returns ErrReleaseNotFound. // Update updates a release or returns ErrReleaseNotFound.
func (mem *Memory) Update(rls *rspb.Release) error { func (mem *Memory) Update(key string, rls *rspb.Release) error {
defer unlock(mem.wlock()) defer unlock(mem.wlock())
if _, ok := mem.cache[rls.Name]; ok { if rs, ok := mem.cache[rls.Name]; ok && rs.Exists(key) {
mem.cache[rls.Name] = rls rs.Replace(key, newRecord(key, rls))
return nil return nil
} }
return ErrReleaseNotFound return ErrReleaseNotFound
@ -90,27 +136,56 @@ func (mem *Memory) Update(rls *rspb.Release) error {
func (mem *Memory) Delete(key string) (*rspb.Release, error) { func (mem *Memory) Delete(key string) (*rspb.Release, error) {
defer unlock(mem.wlock()) defer unlock(mem.wlock())
if old, ok := mem.cache[key]; ok { switch elems := strings.Split(key, ".v"); len(elems) {
delete(mem.cache, key) case 2:
return old, nil name, ver := elems[0], elems[1]
if _, err := strconv.Atoi(ver); err != nil {
return nil, ErrInvalidKey
}
if recs, ok := mem.cache[name]; ok {
if r := recs.Remove(key); r != nil {
return r.rls, nil
}
}
return nil, ErrReleaseNotFound
default:
return nil, ErrInvalidKey
} }
return nil, ErrReleaseNotFound return nil, ErrReleaseNotFound
} }
func (mem *Memory) dump(w io.Writer) error {
var b bytes.Buffer
fmt.Fprintln(&b, "memory:")
for key, recs := range mem.cache {
fmt.Fprintf(&b, "\t# %q\n", key)
recs.Iter(func(index int, r *record) bool {
fmt.Fprintf(&b, "\t\t- [%d] v%d (status = %s)\n",
index,
r.rls.Version,
r.rls.Info.Status.Code,
)
return true
})
}
_, err := w.Write(b.Bytes())
return err
}
// wlock locks mem for writing // wlock locks mem for writing
func (mem *Memory) wlock() func() { func (mem *Memory) wlock() func() {
mem.Lock() mem.Lock()
return func() { return func() { mem.Unlock() }
mem.Unlock()
}
} }
// rlock locks mem for reading // rlock locks mem for reading
func (mem *Memory) rlock() func() { func (mem *Memory) rlock() func() {
mem.RLock() mem.RLock()
return func() { return func() { mem.RUnlock() }
mem.RUnlock()
}
} }
// unlock calls fn which reverses a mem.rlock or mem.wlock. e.g: // unlock calls fn which reverses a mem.rlock or mem.wlock. e.g:

@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License. limitations under the License.
*/ */
package driver // import "k8s.io/helm/pkg/storage/driver" package driver
import ( import (
"reflect" "reflect"
@ -23,79 +23,146 @@ import (
rspb "k8s.io/helm/pkg/proto/hapi/release" rspb "k8s.io/helm/pkg/proto/hapi/release"
) )
var _ Driver = &Memory{}
func TestMemoryName(t *testing.T) { func TestMemoryName(t *testing.T) {
mem := NewMemory() if mem := NewMemory(); mem.Name() != MemoryDriverName {
if mem.Name() != MemoryDriverName {
t.Errorf("Expected name to be %q, got %q", MemoryDriverName, mem.Name()) t.Errorf("Expected name to be %q, got %q", MemoryDriverName, mem.Name())
} }
} }
func TestMemoryGet(t *testing.T) { func TestMemoryCreate(t *testing.T) {
key := "test-1" var tests = []struct {
rls := &rspb.Release{Name: key} desc string
rls *rspb.Release
err bool
}{
{
"create should success",
releaseStub("rls-c", 1, rspb.Status_DEPLOYED),
false,
},
{
"create should fail (release already exists)",
releaseStub("rls-a", 1, rspb.Status_DEPLOYED),
true,
},
}
ts := tsFixtureMemory(t)
for _, tt := range tests {
key := testKey(tt.rls.Name, tt.rls.Version)
rls := tt.rls
if err := ts.Create(key, rls); err != nil {
if !tt.err {
t.Fatalf("failed to create %q: %s", tt.desc, err)
}
}
}
}
mem := NewMemory() func TestMemoryGet(t *testing.T) {
if err := mem.Create(rls); err != nil { var tests = []struct {
t.Fatalf("Failed create: %s", err) desc string
key string
err bool
}{
{"release key should exist", "rls-a.v1", false},
{"release key should not exist", "rls-a.v5", true},
} }
res, err := mem.Get(key) ts := tsFixtureMemory(t)
if err != nil { for _, tt := range tests {
t.Errorf("Could not get %s: %s", key, err) if _, err := ts.Get(tt.key); err != nil {
if !tt.err {
t.Fatalf("Failed %q to get '%s': %q\n", tt.desc, tt.key, err)
}
} }
if res.Name != key {
t.Errorf("Expected %s, got %s", key, res.Name)
} }
} }
func TestMemoryCreate(t *testing.T) { func TestMemoryQuery(t *testing.T) {
key := "test-1" var tests = []struct {
rls := &rspb.Release{Name: key} desc string
xlen int
lbs map[string]string
}{
{
"should be 2 query results",
2,
map[string]string{"STATUS": "DEPLOYED"},
},
}
ts := tsFixtureMemory(t)
for _, tt := range tests {
l, err := ts.Query(tt.lbs)
if err != nil {
t.Fatalf("Failed to query: %s\n", err)
}
mem := NewMemory() if tt.xlen != len(l) {
if err := mem.Create(rls); err != nil { t.Fatalf("Expected %d results, actual %d\n", tt.xlen, len(l))
t.Fatalf("Failed created: %s", err)
} }
if mem.cache[key].Name != key {
t.Errorf("Unexpected release name: %s", mem.cache[key].Name)
} }
} }
func TestMemoryUpdate(t *testing.T) { func TestMemoryUpdate(t *testing.T) {
key := "test-1" var tests = []struct {
rls := &rspb.Release{Name: key} desc string
key string
mem := NewMemory() rls *rspb.Release
if err := mem.Create(rls); err != nil { err bool
t.Fatalf("Failed create: %s", err) }{
{
"update release status",
"rls-a.v4",
releaseStub("rls-a", 4, rspb.Status_SUPERSEDED),
false,
},
{
"update release does not exist",
"rls-z.v1",
releaseStub("rls-z", 1, rspb.Status_DELETED),
true,
},
}
ts := tsFixtureMemory(t)
for _, tt := range tests {
if err := ts.Update(tt.key, tt.rls); err != nil {
if !tt.err {
t.Fatalf("Failed %q: %s\n", tt.desc, err)
}
continue
}
r, err := ts.Get(tt.key)
if err != nil {
t.Fatalf("Failed to get: %s\n", err)
} }
if err := mem.Update(rls); err != nil {
t.Fatalf("Failed update: %s", err) if !reflect.DeepEqual(r, tt.rls) {
t.Fatalf("Expected %s, actual %s\n", tt.rls, r)
} }
if mem.cache[key].Name != key {
t.Errorf("Unexpected release name: %s", mem.cache[key].Name)
} }
} }
func TestMemoryDelete(t *testing.T) { func TestMemoryDelete(t *testing.T) {
key := "test-1" var tests = []struct {
rls := &rspb.Release{Name: key} desc string
key string
mem := NewMemory() err bool
if err := mem.Create(rls); err != nil { }{
t.Fatalf("Failed create: %s", err) {"release key should exist", "rls-a.v1", false},
{"release key should not exist", "rls-a.v5", true},
} }
res, err := mem.Delete(key) ts := tsFixtureMemory(t)
if err != nil { for _, tt := range tests {
t.Fatalf("Failed delete: %s", err) if _, err := ts.Delete(tt.key); err != nil {
if !tt.err {
t.Fatalf("Failed %q to get '%s': %q\n", tt.desc, tt.key, err)
} }
if mem.cache[key] != nil {
t.Errorf("Expected nil, got %s", mem.cache[key])
} }
if !reflect.DeepEqual(rls, res) {
t.Errorf("Expected %s, got %s", rls, res)
} }
} }

@ -0,0 +1,116 @@
package driver
import (
"sort"
"strconv"
rspb "k8s.io/helm/pkg/proto/hapi/release"
)
// records holds a list of in-memory release records
type records []*record
func (rs records) Len() int { return len(rs) }
func (rs records) Swap(i, j int) { rs[i], rs[j] = rs[j], rs[i] }
func (rs records) Less(i, j int) bool { return rs[i].rls.Version < rs[j].rls.Version }
func (rs *records) Add(r *record) error {
if r == nil {
return nil
}
if rs.Exists(r.key) {
return ErrReleaseExists
}
*rs = append(*rs, r)
sort.Sort(*rs)
return nil
}
func (rs records) Get(key string) *record {
if i, ok := rs.Index(key); ok {
return rs[i]
}
return nil
}
func (rs *records) Iter(fn func(int, *record) bool) {
cp := make([]*record, len(*rs))
copy(cp, *rs)
for i, r := range cp {
if !fn(i, r) {
return
}
}
}
func (rs *records) Index(key string) (int, bool) {
for i, r := range *rs {
if r.key == key {
return i, true
}
}
return -1, false
}
func (rs records) Exists(key string) bool {
_, ok := rs.Index(key)
return ok
}
func (rs *records) Remove(key string) (r *record) {
if i, ok := rs.Index(key); ok {
return rs.removeAt(i)
}
return nil
}
func (rs *records) Replace(key string, rec *record) *record {
if i, ok := rs.Index(key); ok {
old := (*rs)[i]
(*rs)[i] = rec
return old
}
return nil
}
func (rs records) FindByVersion(vers int32) (int, bool) {
i := sort.Search(len(rs), func(i int) bool {
return rs[i].rls.Version == vers
})
if i < len(rs) && rs[i].rls.Version == vers {
return i, true
}
return i, false
}
func (rs *records) removeAt(index int) *record {
r := (*rs)[index]
(*rs)[index] = nil
copy((*rs)[index:], (*rs)[index+1:])
*rs = (*rs)[:len(*rs)-1]
return r
}
// record is the data structure used to cache releases
// for the in-memory storage driver
type record struct {
key string
lbs labels
rls *rspb.Release
}
// newRecord creates a new in-memory release record
func newRecord(key string, rls *rspb.Release) *record {
var lbs labels
lbs.init()
lbs.set("NAME", rls.Name)
lbs.set("STATUS", rspb.Status_Code_name[int32(rls.Info.Status.Code)])
lbs.set("VERSION", strconv.Itoa(int(rls.Version)))
return &record{key: key, lbs: lbs, rls: rls}
}

@ -0,0 +1,71 @@
package driver
import (
"testing"
rspb "k8s.io/helm/pkg/proto/hapi/release"
)
func TestRecordsAdd(t *testing.T) {
rs := records([]*record{
newRecord("rls-a.v1", releaseStub("rls-a", 1, rspb.Status_SUPERSEDED)),
newRecord("rls-a.v2", releaseStub("rls-a", 2, rspb.Status_DEPLOYED)),
})
var tests = []struct {
desc string
key string
ok bool
rec *record
}{
{
"add valid key",
"rls-a.v3",
false,
newRecord("rls-a.v3", releaseStub("rls-a", 3, rspb.Status_SUPERSEDED)),
},
{
"add already existing key",
"rls-a.v1",
true,
newRecord("rls-a.v1", releaseStub("rls-a", 1, rspb.Status_DEPLOYED)),
},
}
for _, tt := range tests {
if err := rs.Add(tt.rec); err != nil {
if !tt.ok {
t.Fatalf("failed: %q: %s\n", tt.desc, err)
}
}
}
}
func TestRecordsRemove(t *testing.T) {
var tests = []struct {
desc string
key string
ok bool
}{
{"remove valid key", "rls-a.v1", false},
{"remove invalid key", "rls-a.v", true},
{"remove non-existant key", "rls-z.v1", true},
}
rs := records([]*record{
newRecord("rls-a.v1", releaseStub("rls-a", 1, rspb.Status_SUPERSEDED)),
newRecord("rls-a.v2", releaseStub("rls-a", 2, rspb.Status_DEPLOYED)),
})
for _, tt := range tests {
if r := rs.Remove(tt.key); r == nil {
if !tt.ok {
t.Fatalf("Failed to %q (key = %s). Expected nil, got %s",
tt.desc,
tt.key,
r,
)
}
}
}
}

@ -0,0 +1,119 @@
package driver
import (
"fmt"
"testing"
rspb "k8s.io/helm/pkg/proto/hapi/release"
"k8s.io/kubernetes/pkg/client/unversioned"
kberrs "k8s.io/kubernetes/pkg/api/errors"
"k8s.io/kubernetes/pkg/api"
)
func releaseStub(name string, vers int32, code rspb.Status_Code) *rspb.Release {
return &rspb.Release{
Name: name,
Version: vers,
Info: &rspb.Info{Status: &rspb.Status{Code: code}},
}
}
func testKey(name string, vers int32) string {
return fmt.Sprintf("%s.v%d", name, vers)
}
func tsFixtureMemory(t *testing.T) *Memory {
hs := []*rspb.Release{
// rls-a
releaseStub("rls-a", 4, rspb.Status_DEPLOYED),
releaseStub("rls-a", 1, rspb.Status_SUPERSEDED),
releaseStub("rls-a", 3, rspb.Status_SUPERSEDED),
releaseStub("rls-a", 2, rspb.Status_SUPERSEDED),
// rls-b
releaseStub("rls-b", 4, rspb.Status_DEPLOYED),
releaseStub("rls-b", 1, rspb.Status_SUPERSEDED),
releaseStub("rls-b", 3, rspb.Status_SUPERSEDED),
releaseStub("rls-b", 2, rspb.Status_SUPERSEDED),
}
mem := NewMemory()
for _, tt := range hs {
err := mem.Create(testKey(tt.Name, tt.Version), tt)
if err != nil {
t.Fatalf("Test setup failed to create: %s\n", err)
}
}
return mem
}
// newTestFixture initializes a MockConfigMapsInterface.
// ConfigMaps are created for each release provided.
func newTestFixtureCfgMaps(t *testing.T, releases ...*rspb.Release) *ConfigMaps {
var mock MockConfigMapsInterface
mock.Init(t, releases...)
return NewConfigMaps(&mock)
}
// MockConfigMapsInterface mocks a kubernetes ConfigMapsInterface
type MockConfigMapsInterface struct {
unversioned.ConfigMapsInterface
objects map[string]*api.ConfigMap
}
func (mock *MockConfigMapsInterface) Init(t *testing.T, releases ...*rspb.Release) {
mock.objects = map[string]*api.ConfigMap{}
for _, rls := range releases {
objkey := testKey(rls.Name, rls.Version)
cfgmap, err := newConfigMapsObject(objkey, rls, nil)
if err != nil {
t.Fatalf("Failed to create configmap: %s", err)
}
mock.objects[objkey] = cfgmap
}
}
func (mock *MockConfigMapsInterface) Get(name string) (*api.ConfigMap, error) {
object, ok := mock.objects[name]
if !ok {
return nil, kberrs.NewNotFound(api.Resource("tests"), name)
}
return object, nil
}
func (mock *MockConfigMapsInterface) List(opts api.ListOptions) (*api.ConfigMapList, error) {
var list api.ConfigMapList
for _, cfgmap := range mock.objects {
list.Items = append(list.Items, *cfgmap)
}
return &list, nil
}
func (mock *MockConfigMapsInterface) Create(cfgmap *api.ConfigMap) (*api.ConfigMap, error) {
name := cfgmap.ObjectMeta.Name
if object, ok := mock.objects[name]; ok {
return object, kberrs.NewAlreadyExists(api.Resource("tests"), name)
}
mock.objects[name] = cfgmap
return cfgmap, nil
}
func (mock *MockConfigMapsInterface) Update(cfgmap *api.ConfigMap) (*api.ConfigMap, error) {
name := cfgmap.ObjectMeta.Name
if _, ok := mock.objects[name]; !ok {
return nil, kberrs.NewNotFound(api.Resource("tests"), name)
}
mock.objects[name] = cfgmap
return cfgmap, nil
}
func (mock *MockConfigMapsInterface) Delete(name string) error {
if _, ok := mock.objects[name]; !ok {
return kberrs.NewNotFound(api.Resource("tests"), name)
}
delete(mock.objects, name)
return nil
}

@ -17,6 +17,7 @@ limitations under the License.
package storage // import "k8s.io/helm/pkg/storage" package storage // import "k8s.io/helm/pkg/storage"
import ( import (
"fmt"
"log" "log"
rspb "k8s.io/helm/pkg/proto/hapi/release" rspb "k8s.io/helm/pkg/proto/hapi/release"
@ -30,34 +31,34 @@ type Storage struct {
// Get retrieves the release from storage. An error is returned // Get retrieves the release from storage. An error is returned
// if the storage driver failed to fetch the release, or the // if the storage driver failed to fetch the release, or the
// release identified by key does not exist. // release identified by the key, version pair does not exist.
func (s *Storage) Get(key string) (*rspb.Release, error) { func (s *Storage) Get(name string, version int32) (*rspb.Release, error) {
log.Printf("Getting release %q from storage\n", key) log.Printf("Getting release %q (v%d) from storage\n", name, version)
return s.Driver.Get(key) return s.Driver.Get(makeKey(name, version))
} }
// Create creates a new storage entry holding the release. An // Create creates a new storage entry holding the release. An
// error is returned if the storage driver failed to store the // error is returned if the storage driver failed to store the
// release, or a release with identical an key already exists. // release, or a release with identical an key already exists.
func (s *Storage) Create(rls *rspb.Release) error { func (s *Storage) Create(rls *rspb.Release) error {
log.Printf("Create release %q in storage\n", rls.Name) log.Printf("Create release %q (v%d) in storage\n", rls.Name, rls.Version)
return s.Driver.Create(rls) return s.Driver.Create(makeKey(rls.Name, rls.Version), rls)
} }
// Update update the release in storage. An error is returned if the // Update update the release in storage. An error is returned if the
// storage backend fails to update the release or if the release // storage backend fails to update the release or if the release
// does not exist. // does not exist.
func (s *Storage) Update(rls *rspb.Release) error { func (s *Storage) Update(rls *rspb.Release) error {
log.Printf("Updating %q in storage\n", rls.Name) log.Printf("Updating %q (v%d) in storage\n", rls.Name, rls.Version)
return s.Driver.Update(rls) return s.Driver.Update(makeKey(rls.Name, rls.Version), rls)
} }
// Delete deletes the release from storage. An error is returned if // Delete deletes the release from storage. An error is returned if
// the storage backend fails to delete the release or if the release // the storage backend fails to delete the release or if the release
// does not exist. // does not exist.
func (s *Storage) Delete(key string) (*rspb.Release, error) { func (s *Storage) Delete(name string, version int32) (*rspb.Release, error) {
log.Printf("Deleting release %q from storage\n", key) log.Printf("Deleting release %q (v%d) from storage\n", name, version)
return s.Driver.Delete(key) return s.Driver.Delete(makeKey(name, version))
} }
// ListReleases returns all releases from storage. An error is returned if the // ListReleases returns all releases from storage. An error is returned if the
@ -105,6 +106,32 @@ func (s *Storage) ListFilterAny(filters ...FilterFunc) ([]*rspb.Release, error)
}) })
} }
// Deployed returns the deployed release with the provided release name, or
// returns ErrReleaseNotFound if not found.
func (s *Storage) Deployed(name string) (*rspb.Release, error) {
log.Printf("Getting deployed release from '%s' history\n", name)
ls, err := s.Driver.Query(map[string]string{
"NAME": name,
"STATUS": "DEPLOYED",
})
switch {
case err != nil:
return nil, err
case len(ls) == 0:
return nil, fmt.Errorf("'%s' has no deployed releases", name)
default:
return ls[0], nil
}
}
// makeKey concatenates a release name and version into
// a string with format ```<release_name>#v<version>```.
// This key is used to uniquely identify storage objects.
func makeKey(rlsname string, version int32) string {
return fmt.Sprintf("%s.v%d", rlsname, version)
}
// Init initializes a new storage backend with the driver d. // Init initializes a new storage backend with the driver d.
// If d is nil, the default in-memory driver is used. // If d is nil, the default in-memory driver is used.
func Init(d driver.Driver) *Storage { func Init(d driver.Driver) *Storage {

@ -30,11 +30,15 @@ func TestStorageCreate(t *testing.T) {
storage := Init(driver.NewMemory()) storage := Init(driver.NewMemory())
// create fake release // create fake release
rls := ReleaseTestData{Name: "angry-beaver"}.ToRelease() rls := ReleaseTestData{
Name: "angry-beaver",
Version: 1,
}.ToRelease()
assertErrNil(t.Fatal, storage.Create(rls), "StoreRelease") assertErrNil(t.Fatal, storage.Create(rls), "StoreRelease")
// fetch the release // fetch the release
res, err := storage.Get(rls.Name) res, err := storage.Get(rls.Name, rls.Version)
assertErrNil(t.Fatal, err, "QueryRelease") assertErrNil(t.Fatal, err, "QueryRelease")
// verify the fetched and created release are the same // verify the fetched and created release are the same
@ -48,16 +52,20 @@ func TestStorageUpdate(t *testing.T) {
storage := Init(driver.NewMemory()) storage := Init(driver.NewMemory())
// create fake release // create fake release
rls := ReleaseTestData{Name: "angry-beaver"}.ToRelease() rls := ReleaseTestData{
Name: "angry-beaver",
Version: 1,
Status: rspb.Status_DEPLOYED,
}.ToRelease()
assertErrNil(t.Fatal, storage.Create(rls), "StoreRelease") assertErrNil(t.Fatal, storage.Create(rls), "StoreRelease")
// modify the release // modify the release
rls.Version = 2 rls.Info.Status.Code = rspb.Status_DELETED
rls.Manifest = "new-manifest"
assertErrNil(t.Fatal, storage.Update(rls), "UpdateRelease") assertErrNil(t.Fatal, storage.Update(rls), "UpdateRelease")
// retrieve the updated release // retrieve the updated release
res, err := storage.Get(rls.Name) res, err := storage.Get(rls.Name, rls.Version)
assertErrNil(t.Fatal, err, "QueryRelease") assertErrNil(t.Fatal, err, "QueryRelease")
// verify updated and fetched releases are the same. // verify updated and fetched releases are the same.
@ -71,11 +79,15 @@ func TestStorageDelete(t *testing.T) {
storage := Init(driver.NewMemory()) storage := Init(driver.NewMemory())
// create fake release // create fake release
rls := ReleaseTestData{Name: "angry-beaver"}.ToRelease() rls := ReleaseTestData{
Name: "angry-beaver",
Version: 1,
}.ToRelease()
assertErrNil(t.Fatal, storage.Create(rls), "StoreRelease") assertErrNil(t.Fatal, storage.Create(rls), "StoreRelease")
// delete the release // delete the release
res, err := storage.Delete(rls.Name) res, err := storage.Delete(rls.Name, rls.Version)
assertErrNil(t.Fatal, err, "DeleteRelease") assertErrNil(t.Fatal, err, "DeleteRelease")
// verify updated and fetched releases are the same. // verify updated and fetched releases are the same.
@ -134,6 +146,46 @@ func TestStorageList(t *testing.T) {
} }
} }
func TestStorageDeployed(t *testing.T) {
storage := Init(driver.NewMemory())
const name = "angry-bird"
const vers = int32(4)
// setup storage with test releases
setup := func() {
// release records
rls0 := ReleaseTestData{Name: name, Version: 1, Status: rspb.Status_SUPERSEDED}.ToRelease()
rls1 := ReleaseTestData{Name: name, Version: 2, Status: rspb.Status_SUPERSEDED}.ToRelease()
rls2 := ReleaseTestData{Name: name, Version: 3, Status: rspb.Status_SUPERSEDED}.ToRelease()
rls3 := ReleaseTestData{Name: name, Version: 4, Status: rspb.Status_DEPLOYED}.ToRelease()
// create the release records in the storage
assertErrNil(t.Fatal, storage.Create(rls0), "Storing release 'angry-bird' (v1)")
assertErrNil(t.Fatal, storage.Create(rls1), "Storing release 'angry-bird' (v2)")
assertErrNil(t.Fatal, storage.Create(rls2), "Storing release 'angry-bird' (v3)")
assertErrNil(t.Fatal, storage.Create(rls3), "Storing release 'angry-bird' (v4)")
}
setup()
rls, err := storage.Deployed(name)
if err != nil {
t.Fatalf("Failed to query for deployed release: %s\n", err)
}
switch {
case rls == nil:
t.Fatalf("Release is nil")
case rls.Name != name:
t.Fatalf("Expected release name %q, actual %q\n", name, rls.Name)
case rls.Version != vers:
t.Fatalf("Expected release version %d, actual %d\n", vers, rls.Version)
case rls.Info.Status.Code != rspb.Status_DEPLOYED:
t.Fatalf("Expected release status 'DEPLOYED', actual %s\n", rls.Info.Status.Code)
}
}
type ReleaseTestData struct { type ReleaseTestData struct {
Name string Name string
Version int32 Version int32

Loading…
Cancel
Save