chore(storage-naming): update storage to allow querying by labels

pkg/storage/driver/cfgmaps.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	kberrs "k8s.io/kubernetes/pkg/api/errors"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
+	kblabels "k8s.io/kubernetes/pkg/labels"
 )
 
 // ConfigMapsDriverName is the string name of the driver.
@@ -95,6 +96,10 @@ func (cfgmaps *ConfigMaps) List(filter func(*rspb.Release) bool) ([]*rspb.Releas
 		return nil, err
 	}
 
+	if len(list.Items) == 0 {
+		return nil, ErrReleaseNotFound
+	}
+
 	var results []*rspb.Release
 
 	// iterate over the configmaps object list
@@ -112,6 +117,38 @@ func (cfgmaps *ConfigMaps) List(filter func(*rspb.Release) bool) ([]*rspb.Releas
 	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
 // ConfigMap already exists, ErrReleaseExists is returned.
 func (cfgmaps *ConfigMaps) Create(key string, rls *rspb.Release) error {

pkg/storage/driver/cfgmaps_test.go

@@ -102,6 +102,10 @@ func TestConfigMapList(t *testing.T) {
 	}
 }
 
+func TestConfigMapQuery(t *testing.T) {
+	t.Skip("TestConfigMapQuery")
+}
+
 func TestConfigMapCreate(t *testing.T) {
 	cfgmaps := newTestFixture(t)
 

pkg/storage/driver/driver.go

@@ -53,15 +53,18 @@ type Deletor interface {
 	Delete(key string) (*rspb.Release, error)
 }
 
-// Queryor is the interface that wraps the Get and List methods.
+// Queryor is the interface that wraps the Get, List, and Query methods.
 //
 // Get returns the release named by key or returns ErrReleaseNotFound
 // if the release does not exist.
 //
 // List returns the set of all releases that satisfy the filter predicate.
+//
+// Query returns the set of all releases that match the provided set of labels.
 type Queryor interface {
 	Get(key string) (*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

pkg/storage/driver/memory.go

@@ -17,6 +17,7 @@ limitations under the License.
 package driver // import "k8s.io/helm/pkg/storage/driver"
 
 import (
+	"fmt"
 	"sync"
 
 	rspb "k8s.io/helm/pkg/proto/hapi/release"
@@ -64,6 +65,11 @@ func (mem *Memory) List(filter func(*rspb.Release) bool) ([]*rspb.Release, error
 	return releases, nil
 }
 
+// Query does not apply to in-memory storage.
+func (mem *Memory) Query(_ map[string]string) ([]*rspb.Release, error) {
+	return nil, fmt.Errorf("memory: cannot apply query by labels to in-memory storage")
+}
+
 // Create creates a new release or returns ErrReleaseExists.
 func (mem *Memory) Create(key string, rls *rspb.Release) error {
 	defer unlock(mem.wlock())

pkg/storage/driver/memory_test.go

@@ -135,5 +135,5 @@ func TestMemoryDelete(t *testing.T) {
 }
 
 func testKey(name string, version int32) string {
-	return fmt.Sprintf("%s#v%d", name, version)
+	return fmt.Sprintf("%s.v%d", name, version)
 }

pkg/storage/storage.go

@@ -34,6 +34,23 @@ type Storage struct {
 // release identified by the key, version pair does not exist.
 func (s *Storage) Get(name string, version int32) (*rspb.Release, error) {
 	log.Printf("Getting release %q (v%d) from storage\n", name, version)
+	// an unspecified version implies latest so we need to select from the
+	// set of releases any such that NAME == "name" and STATUS == "DEPLOYED"
+	if version == 0 {
+		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("bad query")
+		default:
+			return ls[0], nil
+		}
+	}
+
 	return s.Driver.Get(makeKey(name, version))
 }
 
@@ -117,8 +134,8 @@ func Init(d driver.Driver) *Storage {
 }
 
 // makeKey concatenates a release name and version into
-// a string with format ```<release_name>#v<version>```.
+// 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)
+	return fmt.Sprintf("%s.v%d", rlsname, version)
 }