add event handling and graceful shutdown for artifacts

- Add `Artifact` event type in `config.go` and `events.go`.
- Map `artStore.EventType` to standard libpod event statuses (Pull, Push, Remove, Create).
- Spawn a `libartifactEvents` goroutine in `makeRuntime` to poll the event channel and write to the central eventer.
- Implement graceful shutdown in `Runtime.Shutdown()` to process any pending artifact events safely before exiting.

Signed-off-by: Byounguk Lee <nimdrak@gmail.com>

Author: nimdrak

libpod/events/config.go

@@ -122,6 +122,8 @@ const (
 	Container Type = "container"
 	// Image - event is related to images
 	Image Type = "image"
+	// Artifact - event is related to artifacts
+	Artifact Type = "artifact"
 	// Network - event is related to networks
 	Network Type = "network"
 	// Pod - event is related to pods

libpod/events/events.go

@@ -111,6 +111,8 @@ func (s Status) String() string {
 // StringToType converts string to an EventType
 func StringToType(name string) (Type, error) {
 	switch name {
+	case Artifact.String():
+		return Artifact, nil
 	case Container.String():
 		return Container, nil
 	case Image.String():

libpod/runtime.go

@@ -68,18 +68,19 @@ type Runtime struct {
 	storageConfig storage.StoreOptions
 	storageSet    storageSet
 
-	state                  State
-	store                  storage.Store
-	storageService         *storageService
-	imageContext           types.SystemContext
-	defaultOCIRuntime      OCIRuntime
-	ociRuntimes            map[string]OCIRuntime
-	runtimeFlags           []string
-	network                nettypes.ContainerNetwork
-	conmonPath             string
-	libimageRuntime        *libimage.Runtime
-	libimageEventsShutdown chan bool
-	lockManager            lock.Manager
+	state                     State
+	store                     storage.Store
+	storageService            *storageService
+	imageContext              types.SystemContext
+	defaultOCIRuntime         OCIRuntime
+	ociRuntimes               map[string]OCIRuntime
+	runtimeFlags              []string
+	network                   nettypes.ContainerNetwork
+	conmonPath                string
+	libimageRuntime           *libimage.Runtime
+	libimageEventsShutdown    chan bool
+	libartifactEventsShutdown chan bool
+	lockManager               lock.Manager
 
 	// ArtifactStore returns the artifact store created from the runtime.
 	ArtifactStore func() (*artStore.ArtifactStore, error)
@@ -503,6 +504,8 @@ func makeRuntime(ctx context.Context, runtime *Runtime) (retErr error) {
 		runtime.ArtifactStore = sync.OnceValues(func() (*artStore.ArtifactStore, error) {
 			return artStore.NewArtifactStore(filepath.Join(runtime.storageConfig.GraphRoot, "artifacts"), runtime.SystemContext())
 		})
+		// Run the libartifact events routine.
+		runtime.libartifactEvents()
 	}
 
 	// We now need to see if the system has restarted
@@ -722,6 +725,71 @@ func (r *Runtime) libimageEvents() {
 	}()
 }
 
+// libartifactEventsMap translates a libartifact event type to a libpod event status.
+var libartifactEventsMap = map[artStore.EventType]events.Status{
+	artStore.EventTypeArtifactPull:   events.Pull,
+	artStore.EventTypeArtifactPush:   events.Push,
+	artStore.EventTypeArtifactRemove: events.Remove,
+	artStore.EventTypeArtifactAdd:    events.Create,
+}
+
+// libartifactEvents spawns a goroutine which will listen for events on
+// the artStore.ArtifactStore.  The goroutine will be cleaned up implicitly
+// when the main() exists.
+func (r *Runtime) libartifactEvents() {
+	r.libartifactEventsShutdown = make(chan bool)
+
+	toLibpodEventStatus := func(e *artStore.Event) events.Status {
+		status, found := libartifactEventsMap[e.Type]
+		if !found {
+			return "Unknown"
+		}
+		return status
+	}
+
+	store, err := r.ArtifactStore()
+	if err != nil {
+		logrus.Errorf("Unable to initialize artifact store for events: %q", err)
+		return
+	}
+	eventChannel := store.EventChannel()
+
+	go func() {
+		sawShutdown := false
+		for {
+			// Make sure to read and write all events before
+			// shutting down.
+			for len(eventChannel) > 0 {
+				libartifactEvent := <-eventChannel
+				e := events.Event{
+					ID:     libartifactEvent.ID,
+					Name:   libartifactEvent.Name,
+					Status: toLibpodEventStatus(libartifactEvent),
+					Time:   libartifactEvent.Time,
+					Type:   events.Artifact,
+				}
+				if libartifactEvent.Error != nil {
+					e.Error = libartifactEvent.Error.Error()
+				}
+				if err := r.eventer.Write(e); err != nil {
+					logrus.Errorf("Unable to write artifact event: %q", err)
+				}
+			}
+
+			if sawShutdown {
+				close(r.libartifactEventsShutdown)
+				return
+			}
+
+			select {
+			case <-r.libartifactEventsShutdown:
+				sawShutdown = true
+			case <-time.After(100 * time.Millisecond):
+			}
+		}
+	}()
+}
+
 // DeferredShutdown shuts down the runtime without exposing any
 // errors. This is only meant to be used when the runtime is being
 // shutdown within a defer statement; else use Shutdown
@@ -771,6 +839,13 @@ func (r *Runtime) Shutdown(force bool) error {
 			<-r.libimageEventsShutdown
 		}
 
+		if r.libartifactEventsShutdown != nil {
+			// Tell loop to shutdown
+			r.libartifactEventsShutdown <- true
+			// Wait for close to signal shutdown
+			<-r.libartifactEventsShutdown
+		}
+
 		// Note that the libimage runtime shuts down the store.
 		if err := r.libimageRuntime.Shutdown(force); err != nil {
 			lastError = fmt.Errorf("shutting down container storage: %w", err)