volumes: only send "create" event when actually creating volume

The VolumesService did not have information wether or not a volume
was _created_ or if a volume already existed in the driver, and
the existing volume was used.

As a result, multiple "create" events could be generated for the
same volume. For example:

1. Run `docker events` in a shell to start listening for events
2. Create a volume:

    docker volume create myvolume

3. Start a container that uses that volume:

    docker run -dit -v myvolume:/foo busybox

4. Check the events that were generated:

    2021-02-15T18:49:55.874621004+01:00 volume create myvolume (driver=local)
    2021-02-15T18:50:11.442759052+01:00 volume create myvolume (driver=local)
    2021-02-15T18:50:11.487104176+01:00 container create 45112157c8b1382626bf5e01ef18445a4c680f3846c5e32d01775dddee8ca6d1 (image=busybox, name=gracious_hypatia)
    2021-02-15T18:50:11.519288102+01:00 network connect a19f6bb8d44ff84d478670fa4e34c5bf5305f42786294d3d90e790ac74b6d3e0 (container=45112157c8b1382626bf5e01ef18445a4c680f3846c5e32d01775dddee8ca6d1, name=bridge, type=bridge)
    2021-02-15T18:50:11.526407799+01:00 volume mount myvolume (container=45112157c8b1382626bf5e01ef18445a4c680f3846c5e32d01775dddee8ca6d1, destination=/foo, driver=local, propagation=, read/write=true)
    2021-02-15T18:50:11.864134043+01:00 container start 45112157c8b1382626bf5e01ef18445a4c680f3846c5e32d01775dddee8ca6d1 (image=busybox, name=gracious_hypatia)

5. Notice that a "volume create" event is created twice;

    - once when `docker volume create` was ran
    - once when `docker run ...` was ran

This patch moves the generation of (most) events to the volume _store_, and only
generates an event if the volume did not yet exist.

Signed-off-by: Sebastiaan van Stijn <github@gone.nl>

integration-cli/docker_cli_events_unix_test.go

@@ -182,15 +182,14 @@ func (s *DockerSuite) TestVolumeEvents(c *testing.T) {
 	until := daemonUnixTime(c)
 	out, _ := dockerCmd(c, "events", "--since", since, "--until", until)
 	events := strings.Split(strings.TrimSpace(out), "\n")
-	assert.Assert(c, len(events) > 4)
+	assert.Assert(c, len(events) > 3)
 
 	volumeEvents := eventActionsByIDAndType(c, events, "test-event-volume-local", "volume")
-	assert.Equal(c, len(volumeEvents), 5)
+	assert.Equal(c, len(volumeEvents), 4)
 	assert.Equal(c, volumeEvents[0], "create")
-	assert.Equal(c, volumeEvents[1], "create")
-	assert.Equal(c, volumeEvents[2], "mount")
-	assert.Equal(c, volumeEvents[3], "unmount")
-	assert.Equal(c, volumeEvents[4], "destroy")
+	assert.Equal(c, volumeEvents[1], "mount")
+	assert.Equal(c, volumeEvents[2], "unmount")
+	assert.Equal(c, volumeEvents[3], "destroy")
 }
 
 func (s *DockerSuite) TestNetworkEvents(c *testing.T) {

integration/system/event_test.go

@@ -3,6 +3,7 @@ package system // import "github.com/docker/docker/integration/system"
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"io"
 	"net/http"
 	"net/url"
@@ -11,9 +12,12 @@ import (
 	"time"
 
 	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/events"
 	"github.com/docker/docker/api/types/filters"
+	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/api/types/strslice"
 	"github.com/docker/docker/api/types/versions"
+	"github.com/docker/docker/api/types/volume"
 	"github.com/docker/docker/integration/internal/container"
 	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/docker/docker/testutil/request"
@@ -122,3 +126,69 @@ func TestEventsBackwardsCompatible(t *testing.T) {
 	assert.Check(t, is.Equal(cID, containerCreateEvent.ID))
 	assert.Check(t, is.Equal("busybox", containerCreateEvent.From))
 }
+
+// TestEventsVolumeCreate verifies that volume create events are only fired
+// once: when creating the volume, and not when attaching to a container.
+func TestEventsVolumeCreate(t *testing.T) {
+	skip.If(t, testEnv.OSType == "windows", "FIXME: Windows doesn't trigger the events? Could be a race")
+
+	defer setupTest(t)()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	client := testEnv.APIClient()
+
+	since := request.DaemonUnixTime(ctx, t, client, testEnv)
+	volName := t.Name()
+	getEvents := func(messages <-chan events.Message, errs <-chan error) ([]events.Message, error) {
+		var evts []events.Message
+
+		for {
+			select {
+			case m := <-messages:
+				evts = append(evts, m)
+			case err := <-errs:
+				if err == io.EOF {
+					return evts, nil
+				}
+				return nil, err
+			case <-time.After(time.Second * 3):
+				return nil, errors.New("timeout hit")
+			}
+		}
+	}
+
+	_, err := client.VolumeCreate(ctx, volume.VolumeCreateBody{Name: volName})
+	assert.NilError(t, err)
+
+	filter := filters.NewArgs(
+		filters.Arg("type", "volume"),
+		filters.Arg("event", "create"),
+		filters.Arg("volume", volName),
+	)
+	messages, errs := client.Events(ctx, types.EventsOptions{
+		Since:   since,
+		Until:   request.DaemonUnixTime(ctx, t, client, testEnv),
+		Filters: filter,
+	})
+
+	volEvents, err := getEvents(messages, errs)
+	assert.NilError(t, err)
+	assert.Equal(t, len(volEvents), 1, "expected volume create event when creating a volume")
+
+	container.Create(ctx, t, client, container.WithMount(mount.Mount{
+		Type:   mount.TypeVolume,
+		Source: volName,
+		Target: "/tmp/foo",
+	}))
+
+	messages, errs = client.Events(ctx, types.EventsOptions{
+		Since:   since,
+		Until:   request.DaemonUnixTime(ctx, t, client, testEnv),
+		Filters: filter,
+	})
+
+	volEvents, err = getEvents(messages, errs)
+	assert.NilError(t, err)
+	assert.Equal(t, len(volEvents), 1, "expected volume create event to be fired only once")
+}

volume/service/service.go

@@ -23,7 +23,9 @@ type ds interface {
 	GetDriverList() []string
 }
 
-type volumeEventLogger interface {
+// VolumeEventLogger interface provides methods to log volume-related events
+type VolumeEventLogger interface {
+	// LogVolumeEvent generates an event related to a volume.
 	LogVolumeEvent(volumeID, action string, attributes map[string]string)
 }
 
@@ -33,17 +35,17 @@ type VolumesService struct {
 	vs           *VolumeStore
 	ds           ds
 	pruneRunning int32
-	eventLogger  volumeEventLogger
+	eventLogger  VolumeEventLogger
 }
 
 // NewVolumeService creates a new volume service
-func NewVolumeService(root string, pg plugingetter.PluginGetter, rootIDs idtools.Identity, logger volumeEventLogger) (*VolumesService, error) {
+func NewVolumeService(root string, pg plugingetter.PluginGetter, rootIDs idtools.Identity, logger VolumeEventLogger) (*VolumesService, error) {
 	ds := drivers.NewStore(pg)
 	if err := setupDefaultDriver(ds, root, rootIDs); err != nil {
 		return nil, err
 	}
 
-	vs, err := NewStore(root, ds)
+	vs, err := NewStore(root, ds, WithEventLogger(logger))
 	if err != nil {
 		return nil, err
 	}
@@ -71,7 +73,6 @@ func (s *VolumesService) Create(ctx context.Context, name, driverName string, op
 		return nil, err
 	}
 
-	s.eventLogger.LogVolumeEvent(v.Name(), "create", map[string]string{"driver": v.DriverName()})
 	apiV := volumeToAPIType(v)
 	return &apiV, nil
 }
@@ -161,10 +162,6 @@ func (s *VolumesService) Remove(ctx context.Context, name string, rmOpts ...opts
 	} else if IsNotExist(err) && cfg.PurgeOnError {
 		err = nil
 	}
-
-	if err == nil {
-		s.eventLogger.LogVolumeEvent(v.Name(), "destroy", map[string]string{"driver": v.DriverName()})
-	}
 	return err
 }
 

volume/service/store.go

@@ -68,8 +68,11 @@ func (v volumeWrapper) CachedPath() string {
 	return v.Volume.Path()
 }
 
+// StoreOpt sets options for a VolumeStore
+type StoreOpt func(store *VolumeStore) error
+
 // NewStore creates a new volume store at the given path
-func NewStore(rootPath string, drivers *drivers.Store) (*VolumeStore, error) {
+func NewStore(rootPath string, drivers *drivers.Store, opts ...StoreOpt) (*VolumeStore, error) {
 	vs := &VolumeStore{
 		locks:   &locker.Locker{},
 		names:   make(map[string]volume.Volume),
@@ -79,6 +82,12 @@ func NewStore(rootPath string, drivers *drivers.Store) (*VolumeStore, error) {
 		drivers: drivers,
 	}
 
+	for _, o := range opts {
+		if err := o(vs); err != nil {
+			return nil, err
+		}
+	}
+
 	if rootPath != "" {
 		// initialize metadata store
 		volPath := filepath.Join(rootPath, volumeDataDir)
@@ -108,6 +117,14 @@ func NewStore(rootPath string, drivers *drivers.Store) (*VolumeStore, error) {
 	return vs, nil
 }
 
+// WithEventLogger configures the VolumeStore with the given VolumeEventLogger
+func WithEventLogger(logger VolumeEventLogger) StoreOpt {
+	return func(store *VolumeStore) error {
+		store.eventLogger = logger
+		return nil
+	}
+}
+
 func (s *VolumeStore) getNamed(name string) (volume.Volume, bool) {
 	s.globalLock.RLock()
 	v, exists := s.names[name]
@@ -198,7 +215,9 @@ type VolumeStore struct {
 	labels map[string]map[string]string
 	// options stores volume options for each volume
 	options map[string]map[string]string
-	db      *bolt.DB
+
+	db          *bolt.DB
+	eventLogger VolumeEventLogger
 }
 
 func filterByDriver(names []string) filterFunc {
@@ -464,7 +483,7 @@ func (s *VolumeStore) Create(ctx context.Context, name, driverName string, creat
 	default:
 	}
 
-	v, err := s.create(ctx, name, driverName, cfg.Options, cfg.Labels)
+	v, created, err := s.create(ctx, name, driverName, cfg.Options, cfg.Labels)
 	if err != nil {
 		if _, ok := err.(*OpErr); ok {
 			return nil, err
@@ -472,6 +491,9 @@ func (s *VolumeStore) Create(ctx context.Context, name, driverName string, creat
 		return nil, &OpErr{Err: err, Name: name, Op: "create"}
 	}
 
+	if created && s.eventLogger != nil {
+		s.eventLogger.LogVolumeEvent(v.Name(), "create", map[string]string{"driver": v.DriverName()})
+	}
 	s.setNamed(v, cfg.Reference)
 	return v, nil
 }
@@ -552,7 +574,7 @@ func volumeExists(ctx context.Context, store *drivers.Store, v volume.Volume) (b
 //  for the given volume name, an error is returned after checking if the reference is stale.
 // If the reference is stale, it will be purged and this create can continue.
 // It is expected that callers of this function hold any necessary locks.
-func (s *VolumeStore) create(ctx context.Context, name, driverName string, opts, labels map[string]string) (volume.Volume, error) {
+func (s *VolumeStore) create(ctx context.Context, name, driverName string, opts, labels map[string]string) (volume.Volume, bool, error) {
 	// Validate the name in a platform-specific manner
 
 	// volume name validation is specific to the host os and not on container image
@@ -560,19 +582,19 @@ func (s *VolumeStore) create(ctx context.Context, name, driverName string, opts,
 	parser := volumemounts.NewParser(runtime.GOOS)
 	err := parser.ValidateVolumeName(name)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
 	v, err := s.checkConflict(ctx, name, driverName)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
 	if v != nil {
 		// there is an existing volume, if we already have this stored locally, return it.
 		// TODO: there could be some inconsistent details such as labels here
 		if vv, _ := s.getNamed(v.Name()); vv != nil {
-			return vv, nil
+			return vv, false, nil
 		}
 	}
 
@@ -580,7 +602,7 @@ func (s *VolumeStore) create(ctx context.Context, name, driverName string, opts,
 	if driverName == "" {
 		v, _ = s.getVolume(ctx, name, "")
 		if v != nil {
-			return v, nil
+			return v, false, nil
 		}
 	}
 
@@ -589,7 +611,7 @@ func (s *VolumeStore) create(ctx context.Context, name, driverName string, opts,
 	}
 	vd, err := s.drivers.CreateDriver(driverName)
 	if err != nil {
-		return nil, &OpErr{Op: "create", Name: name, Err: err}
+		return nil, false, &OpErr{Op: "create", Name: name, Err: err}
 	}
 
 	logrus.Debugf("Registering new volume reference: driver %q, name %q", vd.Name(), name)
@@ -599,7 +621,7 @@ func (s *VolumeStore) create(ctx context.Context, name, driverName string, opts,
 			if _, err := s.drivers.ReleaseDriver(driverName); err != nil {
 				logrus.WithError(err).WithField("driver", driverName).Error("Error releasing reference to volume driver")
 			}
-			return nil, err
+			return nil, false, err
 		}
 	}
 
@@ -617,9 +639,9 @@ func (s *VolumeStore) create(ctx context.Context, name, driverName string, opts,
 	}
 
 	if err := s.setMeta(name, metadata); err != nil {
-		return nil, err
+		return nil, true, err
 	}
-	return volumeWrapper{v, labels, vd.Scope(), opts}, nil
+	return volumeWrapper{v, labels, vd.Scope(), opts}, true, nil
 }
 
 // Get looks if a volume with the given name exists and returns it if so
@@ -802,6 +824,9 @@ func (s *VolumeStore) Remove(ctx context.Context, v volume.Volume, rmOpts ...opt
 			err = e
 		}
 	}
+	if err == nil && s.eventLogger != nil {
+		s.eventLogger.LogVolumeEvent(v.Name(), "destroy", map[string]string{"driver": v.DriverName()})
+	}
 	return err
 }