Add buckets for all containers and all pods

Now, we don't need to use the global ID registry to iterate - we
can iterate only through containers or only through pods, without
having to iterate through both.

Signed-off-by: Matthew Heon <matthew.heon@gmail.com>

Closes: #184
Approved by: baude

2 files changed, 78 insertions, 13 deletions

diff --git a/libpod/boltdb_state.go b/libpod/boltdb_state.go
index b27588b74..8f6bd579f 100644
--- a/libpod/boltdb_state.go
+++ b/libpod/boltdb_state.go
@@ -50,9 +50,15 @@ func NewBoltState(path, lockDir string, runtime *Runtime) (State, error) {
 		if _, err := tx.CreateBucketIfNotExists(ctrBkt); err != nil {
 			return errors.Wrapf(err, "error creating containers bucket")
 		}
+		if _, err := tx.CreateBucketIfNotExists(allCtrsBkt); err != nil {
+			return errors.Wrapf(err, "error creating all containers bucket")
+		}
 		if _, err := tx.CreateBucketIfNotExists(podBkt); err != nil {
 			return errors.Wrapf(err, "error creating pods bucket")
 		}
+		if _, err := tx.CreateBucketIfNotExists(allPodsBkt); err != nil {
+			return errors.Wrapf(err, "error creating all pods bucket")
+		}
 		if _, err := tx.CreateBucketIfNotExists(runtimeConfigBkt); err != nil {
 			return errors.Wrapf(err, "error creating runtime-config bucket")
 		}
@@ -563,7 +569,7 @@ func (s *BoltState) AllContainers() ([]*Container, error) {
 	defer db.Close()
 
 	err = db.View(func(tx *bolt.Tx) error {
-		idBucket, err := getIDBucket(tx)
+		allCtrsBucket, err := getAllCtrsBucket(tx)
 		if err != nil {
 			return err
 		}
@@ -573,14 +579,13 @@ func (s *BoltState) AllContainers() ([]*Container, error) {
 			return err
 		}
 
-		// Iterate through all IDs
-		// If they're containers, make a container from them and append
-		// it into the containers listing
-		err = idBucket.ForEach(func(id, name []byte) error {
+		err = allCtrsBucket.ForEach(func(id, name []byte) error {
+			// If performance becomes an issue, this check can be
+			// removed. But the error messages that come back will
+			// be much less helpful.
 			ctrExists := ctrBucket.Bucket(id)
 			if ctrExists == nil {
-				// It's a pod not a container
-				return nil
+				return errors.Wrapf(ErrInternal, "state is inconsistent - container ID %s in all containers, but container not found", string(id))
 			}
 
 			ctr := new(Container)
@@ -965,6 +970,11 @@ func (s *BoltState) AddPod(pod *Pod) error {
 			return err
 		}
 
+		allPodsBkt, err := getAllPodsBucket(tx)
+		if err != nil {
+			return err
+		}
+
 		idsBkt, err := getIDBucket(tx)
 		if err != nil {
 			return err
@@ -1008,6 +1018,9 @@ func (s *BoltState) AddPod(pod *Pod) error {
 		if err := namesBkt.Put(podName, podID); err != nil {
 			return errors.Wrapf(err, "error storing pod %s name in DB", pod.Name())
 		}
+		if err := allPodsBkt.Put(podID, podName); err != nil {
+			return errors.Wrapf(err, "error storing pod %s in all pods bucket in DB", pod.ID())
+		}
 
 		return nil
 	})
@@ -1044,6 +1057,11 @@ func (s *BoltState) RemovePod(pod *Pod) error {
 			return err
 		}
 
+		allPodsBkt, err := getAllPodsBucket(tx)
+		if err != nil {
+			return err
+		}
+
 		idsBkt, err := getIDBucket(tx)
 		if err != nil {
 			return err
@@ -1082,6 +1100,9 @@ func (s *BoltState) RemovePod(pod *Pod) error {
 		if err := namesBkt.Delete(podName); err != nil {
 			return errors.Wrapf(err, "error removing pod %s name (%s) from DB", pod.ID(), pod.Name())
 		}
+		if err := allPodsBkt.Delete(podID); err != nil {
+			return errors.Wrapf(err, "error removing pod %s ID from all pods bucket in DB", pod.ID())
+		}
 		if err := podBkt.DeleteBucket(podID); err != nil {
 			return errors.Wrapf(err, "error removing pod %s from DB", pod.ID())
 		}
@@ -1124,6 +1145,11 @@ func (s *BoltState) RemovePodContainers(pod *Pod) error {
 			return err
 		}
 
+		allCtrsBkt, err := getAllCtrsBucket(tx)
+		if err != nil {
+			return err
+		}
+
 		idsBkt, err := getIDBucket(tx)
 		if err != nil {
 			return err
@@ -1186,6 +1212,10 @@ func (s *BoltState) RemovePodContainers(pod *Pod) error {
 				return errors.Wrapf(err, "error deleting container %s name in DB", string(id))
 			}
 
+			if err := allCtrsBkt.Delete(id); err != nil {
+				return errors.Wrapf(err, "error deleting container %s ID from all containers bucket in DB", string(id))
+			}
+
 			return nil
 		})
 		if err != nil {
@@ -1277,7 +1307,7 @@ func (s *BoltState) AllPods() ([]*Pod, error) {
 	defer db.Close()
 
 	err = db.View(func(tx *bolt.Tx) error {
-		idBucket, err := getIDBucket(tx)
+		allPodsBucket, err := getAllPodsBucket(tx)
 		if err != nil {
 			return err
 		}
@@ -1287,13 +1317,12 @@ func (s *BoltState) AllPods() ([]*Pod, error) {
 			return err
 		}
 
-		// Iterate through all IDs
-		// If they're pod, make a container from them and append it
-		err = idBucket.ForEach(func(id, name []byte) error {
+		err = allPodsBucket.ForEach(func(id, name []byte) error {
 			podExists := podBucket.Bucket(id)
+			// This check can be removed if performance becomes an
+			// issue, but much less helpful errors will be produced
 			if podExists == nil {
-				// Container not pod
-				return nil
+				return errors.Wrapf(ErrInternal, "inconsistency in state - pod %s is in all pods bucket but pod not found", string(id))
 			}
 
 			pod := new(Pod)
diff --git a/libpod/boltdb_state_internal.go b/libpod/boltdb_state_internal.go
index f26a60905..033dbe8bd 100644
--- a/libpod/boltdb_state_internal.go
+++ b/libpod/boltdb_state_internal.go
@@ -15,7 +15,9 @@ const (
 	idRegistryName    = "id-registry"
 	nameRegistryName  = "name-registry"
 	ctrName           = "ctr"
+	allCtrsName       = "all-ctrs"
 	podName           = "pod"
+	allPodsName       = "allPods"
 	runtimeConfigName = "runtime-config"
 
 	configName       = "config"
@@ -30,7 +32,9 @@ var (
 	idRegistryBkt    = []byte(idRegistryName)
 	nameRegistryBkt  = []byte(nameRegistryName)
 	ctrBkt           = []byte(ctrName)
+	allCtrsBkt       = []byte(allCtrsName)
 	podBkt           = []byte(podName)
+	allPodsBkt       = []byte(allPodsName)
 	runtimeConfigBkt = []byte(runtimeConfigName)
 
 	configKey       = []byte(configName)
@@ -138,6 +142,14 @@ func getCtrBucket(tx *bolt.Tx) (*bolt.Bucket, error) {
 	return bkt, nil
 }
 
+func getAllCtrsBucket(tx *bolt.Tx) (*bolt.Bucket, error) {
+	bkt := tx.Bucket(allCtrsBkt)
+	if bkt == nil {
+		return nil, errors.Wrapf(ErrDBBadConfig, "all containers bucket not found in DB")
+	}
+	return bkt, nil
+}
+
 func getPodBucket(tx *bolt.Tx) (*bolt.Bucket, error) {
 	bkt := tx.Bucket(podBkt)
 	if bkt == nil {
@@ -146,6 +158,14 @@ func getPodBucket(tx *bolt.Tx) (*bolt.Bucket, error) {
 	return bkt, nil
 }
 
+func getAllPodsBucket(tx *bolt.Tx) (*bolt.Bucket, error) {
+	bkt := tx.Bucket(allPodsBkt)
+	if bkt == nil {
+		return nil, errors.Wrapf(ErrDBBadConfig, "all pods bucket not found in DB")
+	}
+	return bkt, nil
+}
+
 func getRuntimeConfigBucket(tx *bolt.Tx) (*bolt.Bucket, error) {
 	bkt := tx.Bucket(runtimeConfigBkt)
 	if bkt == nil {
@@ -279,6 +299,11 @@ func (s *BoltState) addContainer(ctr *Container, pod *Pod) error {
 			return err
 		}
 
+		allCtrsBucket, err := getAllCtrsBucket(tx)
+		if err != nil {
+			return err
+		}
+
 		// If a pod was given, check if it exists
 		var podDB *bolt.Bucket
 		var podCtrs *bolt.Bucket
@@ -319,6 +344,9 @@ func (s *BoltState) addContainer(ctr *Container, pod *Pod) error {
 		if err := namesBucket.Put(ctrName, ctrID); err != nil {
 			return errors.Wrapf(err, "error adding container %s name (%s) to DB", ctr.ID(), ctr.Name())
 		}
+		if err := allCtrsBucket.Put(ctrID, ctrName); err != nil {
+			return errors.Wrapf(err, "error adding container %s to all containers bucket in DB", ctr.ID())
+		}
 
 		newCtrBkt, err := ctrBucket.CreateBucket(ctrID)
 		if err != nil {
@@ -405,6 +433,11 @@ func removeContainer(ctr *Container, pod *Pod, tx *bolt.Tx) error {
 		return err
 	}
 
+	allCtrsBucket, err := getAllCtrsBucket(tx)
+	if err != nil {
+		return err
+	}
+
 	// Does the pod exist?
 	var podDB *bolt.Bucket
 	if pod != nil {
@@ -475,6 +508,9 @@ func removeContainer(ctr *Container, pod *Pod, tx *bolt.Tx) error {
 	if err := namesBucket.Delete(ctrName); err != nil {
 		return errors.Wrapf(err, "error deleting container %s name in DB", ctr.ID())
 	}
+	if err := allCtrsBucket.Delete(ctrID); err != nil {
+		return errors.Wrapf(err, "error deleting container %s from all containers bucket in DB", ctr.ID())
+	}
 
 	depCtrs := ctr.Dependencies()