Add support for 'image' volume driver

We added the concept of image volumes in 2.2.0, to support
inspecting an image from within a container. However, this is a
strictly read-only mount, with no modification allowed.

By contrast, the new `image` volume driver creates a c/storage
container as its underlying storage, so we have a read/write
layer. This, in and of itself, is not especially interesting, but
what it will enable in the future is. If we add a new command to
allow these image volumes to be committed, we can now distribute
volumes - and changes to them - via a standard OCI image registry
(which is rather new and quite exciting).

Future work in this area:
- Add support for `podman volume push` (commit volume changes and
  push resulting image to OCI registry).
- Add support for `podman volume pull` (currently, we require
  that the image a volume is created from be already pulled; it
  would be simpler if we had a dedicated command that did the
  pull and made a volume from it)
- Add support for scratch images (make an empty image on demand
  to use as the base of the volume)
- Add UOR support to `podman volume push` and
  `podman volume pull` to enable both with non-image volume
  drivers

Signed-off-by: Matthew Heon <matthew.heon@pm.me>

libpod/boltdb_state.go

@@ -109,6 +109,7 @@ func NewBoltState(path string, runtime *Runtime) (State, error) {
 		runtimeConfigBkt,
 		exitCodeBkt,
 		exitCodeTimeStampBkt,
+		volCtrsBkt,
 	}
 
 	// Does the DB need an update?
@@ -2551,6 +2552,11 @@ func (s *BoltState) AddVolume(volume *Volume) error {
 			return err
 		}
 
+		volCtrsBkt, err := getVolumeContainersBucket(tx)
+		if err != nil {
+			return err
+		}
+
 		// Check if we already have a volume with the given name
 		volExists := allVolsBkt.Get(volName)
 		if volExists != nil {
@@ -2580,6 +2586,12 @@ func (s *BoltState) AddVolume(volume *Volume) error {
 			}
 		}
 
+		if volume.config.StorageID != "" {
+			if err := volCtrsBkt.Put([]byte(volume.config.StorageID), volName); err != nil {
+				return fmt.Errorf("storing volume %s container ID in DB: %w", volume.Name(), err)
+			}
+		}
+
 		if err := allVolsBkt.Put(volName, volName); err != nil {
 			return fmt.Errorf("storing volume %s in all volumes bucket in DB: %w", volume.Name(), err)
 		}
@@ -2619,6 +2631,11 @@ func (s *BoltState) RemoveVolume(volume *Volume) error {
 			return err
 		}
 
+		volCtrIDBkt, err := getVolumeContainersBucket(tx)
+		if err != nil {
+			return err
+		}
+
 		// Check if the volume exists
 		volDB := volBkt.Bucket(volName)
 		if volDB == nil {
@@ -2665,6 +2682,11 @@ func (s *BoltState) RemoveVolume(volume *Volume) error {
 		if err := volBkt.DeleteBucket(volName); err != nil {
 			return fmt.Errorf("removing volume %s from DB: %w", volume.Name(), err)
 		}
+		if volume.config.StorageID != "" {
+			if err := volCtrIDBkt.Delete([]byte(volume.config.StorageID)); err != nil {
+				return fmt.Errorf("removing volume %s container ID from DB: %w", volume.Name(), err)
+			}
+		}
 
 		return nil
 	})
@@ -3618,3 +3640,34 @@ func (s *BoltState) AllPods() ([]*Pod, error) {
 
 	return pods, nil
 }
+
+// ContainerIDIsVolume checks if the given c/storage container ID is used as
+// backing storage for a volume.
+func (s *BoltState) ContainerIDIsVolume(id string) (bool, error) {
+	if !s.valid {
+		return false, define.ErrDBClosed
+	}
+
+	isVol := false
+
+	db, err := s.getDBCon()
+	if err != nil {
+		return false, err
+	}
+	defer s.deferredCloseDBCon(db)
+
+	err = db.View(func(tx *bolt.Tx) error {
+		volCtrsBkt, err := getVolumeContainersBucket(tx)
+		if err != nil {
+			return err
+		}
+
+		volName := volCtrsBkt.Get([]byte(id))
+		if volName != nil {
+			isVol = true
+		}
+
+		return nil
+	})
+	return isVol, err
+}