Merge pull request #8711 from duglin/Issue8160

Add import/pull events to the stream

graph/import.go

@@ -4,6 +4,7 @@ import (
 	"net/http"
 	"net/url"
 
+	log "github.com/Sirupsen/logrus"
 	"github.com/docker/docker/engine"
 	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/utils"
@@ -57,5 +58,12 @@ func (s *TagStore) CmdImport(job *engine.Job) engine.Status {
 		}
 	}
 	job.Stdout.Write(sf.FormatStatus("", img.ID))
+	logID := img.ID
+	if tag != "" {
+		logID += ":" + tag
+	}
+	if err = job.Eng.Job("log", "import", logID, "").Run(); err != nil {
+		log.Errorf("Error logging event 'import' for %s: %s", logID, err)
+	}
 	return engine.StatusOK
 }

graph/pull.go

@@ -139,6 +139,11 @@ func (s *TagStore) CmdPull(job *engine.Job) engine.Status {
 		mirrors = s.mirrors
 	}
 
+	logName := localName
+	if tag != "" {
+		logName += ":" + tag
+	}
+
 	if len(mirrors) == 0 && (isOfficial || endpoint.Version == registry.APIVersion2) {
 		j := job.Eng.Job("trust_update_base")
 		if err = j.Run(); err != nil {
@@ -146,6 +151,9 @@ func (s *TagStore) CmdPull(job *engine.Job) engine.Status {
 		}
 
 		if err := s.pullV2Repository(job.Eng, r, job.Stdout, localName, remoteName, tag, sf, job.GetenvBool("parallel")); err == nil {
+			if err = job.Eng.Job("log", "pull", logName, "").Run(); err != nil {
+				log.Errorf("Error logging event 'pull' for %s: %s", logName, err)
+			}
 			return engine.StatusOK
 		} else if err != registry.ErrDoesNotExist {
 			log.Errorf("Error from V2 registry: %s", err)
@@ -156,6 +164,10 @@ func (s *TagStore) CmdPull(job *engine.Job) engine.Status {
 		return job.Error(err)
 	}
 
+	if err = job.Eng.Job("log", "pull", logName, "").Run(); err != nil {
+		log.Errorf("Error logging event 'pull' for %s: %s", logName, err)
+	}
+
 	return engine.StatusOK
 }
 

integration-cli/docker_cli_events_test.go

@@ -215,3 +215,57 @@ func TestEventsRedirectStdout(t *testing.T) {
 
 	logDone("events - redirect stdout")
 }
+
+func TestEventsImagePull(t *testing.T) {
+	since := time.Now().Unix()
+	pullCmd := exec.Command(dockerBinary, "pull", "scratch")
+	if out, _, err := runCommandWithOutput(pullCmd); err != nil {
+		t.Fatal("pulling the scratch image from has failed: %s, %v", out, err)
+	}
+
+	eventsCmd := exec.Command(dockerBinary, "events",
+		fmt.Sprintf("--since=%d", since),
+		fmt.Sprintf("--until=%d", time.Now().Unix()))
+	out, _, _ := runCommandWithOutput(eventsCmd)
+
+	events := strings.Split(strings.TrimSpace(out), "\n")
+	event := strings.TrimSpace(events[len(events)-1])
+
+	if !strings.HasSuffix(event, "scratch:latest: pull") {
+		t.Fatalf("Missing pull event - got:%q", event)
+	}
+
+	logDone("events - image pull is logged")
+}
+
+func TestEventsImageImport(t *testing.T) {
+	since := time.Now().Unix()
+
+	server, err := fileServer(map[string]string{
+		"/cirros.tar.gz": "/cirros.tar.gz",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer server.Close()
+	fileURL := fmt.Sprintf("%s/cirros.tar.gz", server.URL)
+	importCmd := exec.Command(dockerBinary, "import", fileURL)
+	out, _, err := runCommandWithOutput(importCmd)
+	if err != nil {
+		t.Errorf("import failed with errors: %v, output: %q", err, out)
+	}
+
+	eventsCmd := exec.Command(dockerBinary, "events",
+		fmt.Sprintf("--since=%d", since),
+		fmt.Sprintf("--until=%d", time.Now().Unix()))
+	out, _, _ = runCommandWithOutput(eventsCmd)
+
+	events := strings.Split(strings.TrimSpace(out), "\n")
+	event := strings.TrimSpace(events[len(events)-1])
+
+	if !strings.HasSuffix(event, ": import") {
+		t.Fatalf("Missing pull event - got:%q", event)
+	}
+
+	logDone("events - image import is logged")
+}