content: change Writer/ReaderAt to take OCI

This change allows implementations to resolve the location of the actual data
using OCI descriptor fields such as MediaType.

No OCI descriptor field is written to the store.

No change on gRPC API.

Signed-off-by: Akihiro Suda <suda.akihiro@lab.ntt.co.jp>

vendor/github.com/containerd/cri/pkg/containerd/importer/importer.go

@@ -254,23 +254,24 @@ func writeDockerSchema2Manifest(ctx context.Context, cs content.Ingester, manife
 	for i, ch := range manifest.Layers {
 		labels[fmt.Sprintf("containerd.io/gc.ref.content.%d", i+1)] = ch.Digest.String()
 	}
-	if err := content.WriteBlob(ctx, cs, "manifest-"+manifestDigest.String(), manifestBytesR,
-		int64(len(manifestBytes)), manifestDigest, content.WithLabels(labels)); err != nil {
-		return nil, err
-	}
 
-	desc := &ocispec.Descriptor{
+	desc := ocispec.Descriptor{
 		MediaType: images.MediaTypeDockerSchema2Manifest,
 		Digest:    manifestDigest,
 		Size:      int64(len(manifestBytes)),
 	}
+	if err := content.WriteBlob(ctx, cs, "manifest-"+manifestDigest.String(), manifestBytesR,
+		desc, content.WithLabels(labels)); err != nil {
+		return nil, err
+	}
+
 	if arch != "" || os != "" {
 		desc.Platform = &ocispec.Platform{
 			Architecture: arch,
 			OS:           os,
 		}
 	}
-	return desc, nil
+	return &desc, nil
 }
 
 func onUntarManifestJSON(r io.Reader) ([]manifestDotJSON, error) {
@@ -290,7 +291,7 @@ func onUntarLayerTar(ctx context.Context, r io.Reader, cs content.Ingester, name
 	// name is like "foobar/layer.tar" ( guaranteed by isLayerTar() )
 	split := strings.Split(name, "/")
 	// note: split[0] is not expected digest here
-	cw, err := cs.Writer(ctx, "layer-"+split[0], size, "")
+	cw, err := cs.Writer(ctx, content.WithRef("layer-"+split[0]), content.WithDescriptor(ocispec.Descriptor{Size: size}))
 	if err != nil {
 		return nil, err
 	}
@@ -311,7 +312,7 @@ func onUntarDotJSON(ctx context.Context, r io.Reader, cs content.Ingester, name
 	config.desc.Size = size
 	// name is like "foobar.json" ( guaranteed by is DotJSON() )
 	split := strings.Split(name, ".")
-	cw, err := cs.Writer(ctx, "config-"+split[0], size, "")
+	cw, err := cs.Writer(ctx, content.WithRef("config-"+split[0]), content.WithDescriptor(ocispec.Descriptor{Size: size}))
 	if err != nil {
 		return nil, err
 	}

vendor/github.com/containerd/cri/pkg/server/events.go

@@ -116,15 +116,16 @@ func convertEvent(e *gogotypes.Any) (string, interface{}, error) {
 }
 
 // start starts the event monitor which monitors and handles all container events. It returns
-// a channel for the caller to wait for the event monitor to stop. start must be called after
-// subscribe.
-func (em *eventMonitor) start() (<-chan struct{}, error) {
+// an error channel for the caller to wait for stop errors from the event monitor.
+// start must be called after subscribe.
+func (em *eventMonitor) start() <-chan error {
+	errCh := make(chan error)
 	if em.ch == nil || em.errCh == nil {
-		return nil, errors.New("event channel is nil")
+		panic("event channel is nil")
 	}
-	closeCh := make(chan struct{})
 	backOffCheckCh := em.backOff.start()
 	go func() {
+		defer close(errCh)
 		for {
 			select {
 			case e := <-em.ch:
@@ -144,8 +145,11 @@ func (em *eventMonitor) start() (<-chan struct{}, error) {
 					em.backOff.enBackOff(cID, evt)
 				}
 			case err := <-em.errCh:
-				logrus.WithError(err).Error("Failed to handle event stream")
-				close(closeCh)
+				// Close errCh in defer directly if there is no error.
+				if err != nil {
+					logrus.WithError(err).Errorf("Failed to handle event stream")
+					errCh <- err
+				}
 				return
 			case <-backOffCheckCh:
 				cIDs := em.backOff.getExpiredContainers()
@@ -162,7 +166,7 @@ func (em *eventMonitor) start() (<-chan struct{}, error) {
 			}
 		}
 	}()
-	return closeCh, nil
+	return errCh
 }
 
 // stop stops the event monitor. It will close the event channel.

vendor/github.com/containerd/cri/pkg/server/helpers.go

@@ -345,7 +345,7 @@ func getImageInfo(ctx context.Context, image containerd.Image) (*imageInfo, erro
 	}
 	id := desc.Digest.String()
 
-	rb, err := content.ReadBlob(ctx, image.ContentStore(), desc.Digest)
+	rb, err := content.ReadBlob(ctx, image.ContentStore(), desc)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to read image config from content store")
 	}

vendor/github.com/containerd/cri/pkg/server/service.go

@@ -19,6 +19,7 @@ package server
 import (
 	"fmt"
 	"io"
+	"net/http"
 	"path/filepath"
 	"time"
 
@@ -179,10 +180,7 @@ func (c *criService) Run() error {
 
 	// Start event handler.
 	logrus.Info("Start event monitor")
-	eventMonitorCloseCh, err := c.eventMonitor.start()
-	if err != nil {
-		return errors.Wrap(err, "failed to start event monitor")
-	}
+	eventMonitorErrCh := c.eventMonitor.start()
 
 	// Start snapshot stats syncer, it doesn't need to be stopped.
 	logrus.Info("Start snapshots syncer")
@@ -195,27 +193,32 @@ func (c *criService) Run() error {
 
 	// Start streaming server.
 	logrus.Info("Start streaming server")
-	streamServerCloseCh := make(chan struct{})
+	streamServerErrCh := make(chan error)
 	go func() {
-		if err := c.streamServer.Start(true); err != nil {
+		defer close(streamServerErrCh)
+		if err := c.streamServer.Start(true); err != nil && err != http.ErrServerClosed {
 			logrus.WithError(err).Error("Failed to start streaming server")
+			streamServerErrCh <- err
 		}
-		close(streamServerCloseCh)
 	}()
 
 	// Set the server as initialized. GRPC services could start serving traffic.
 	c.initialized.Set()
 
+	var eventMonitorErr, streamServerErr error
 	// Stop the whole CRI service if any of the critical service exits.
 	select {
-	case <-eventMonitorCloseCh:
-	case <-streamServerCloseCh:
+	case eventMonitorErr = <-eventMonitorErrCh:
+	case streamServerErr = <-streamServerErrCh:
 	}
 	if err := c.Close(); err != nil {
 		return errors.Wrap(err, "failed to stop cri service")
 	}
-
-	<-eventMonitorCloseCh
+	// If the error is set above, err from channel must be nil here, because
+	// the channel is supposed to be closed. Or else, we wait and set it.
+	if err := <-eventMonitorErrCh; err != nil {
+		eventMonitorErr = err
+	}
 	logrus.Info("Event monitor stopped")
 	// There is a race condition with http.Server.Serve.
 	// When `Close` is called at the same time with `Serve`, `Close`
@@ -227,18 +230,27 @@ func (c *criService) Run() error {
 	// is fixed.
 	const streamServerStopTimeout = 2 * time.Second
 	select {
-	case <-streamServerCloseCh:
+	case err := <-streamServerErrCh:
+		if err != nil {
+			streamServerErr = err
+		}
 		logrus.Info("Stream server stopped")
 	case <-time.After(streamServerStopTimeout):
 		logrus.Errorf("Stream server is not stopped in %q", streamServerStopTimeout)
 	}
+	if eventMonitorErr != nil {
+		return errors.Wrap(eventMonitorErr, "event monitor error")
+	}
+	if streamServerErr != nil {
+		return errors.Wrap(streamServerErr, "stream server error")
+	}
 	return nil
 }
 
-// Stop stops the CRI service.
+// Close stops the CRI service.
+// TODO(random-liu): Make close synchronous.
 func (c *criService) Close() error {
 	logrus.Info("Stop CRI service")
-	// TODO(random-liu): Make event monitor stop synchronous.
 	c.eventMonitor.stop()
 	if err := c.streamServer.Stop(); err != nil {
 		return errors.Wrap(err, "failed to stop stream server")

vendor/github.com/containerd/cri/vendor.conf

@@ -4,7 +4,7 @@ github.com/boltdb/bolt e9cf4fae01b5a8ff89d0ec6b32f0d9c9f79aefdd
 github.com/BurntSushi/toml a368813c5e648fee92e5f6c30e3944ff9d5e8895
 github.com/containerd/cgroups fe281dd265766145e943a034aa41086474ea6130
 github.com/containerd/console cb7008ab3d8359b78c5f464cb7cf160107ad5925
-github.com/containerd/containerd 1e8b09cfc6825f7e6349884b5f76e86c1f04a5d4
+github.com/containerd/containerd 1f8c612a6c7ef2fc8328c953fb660adce2bf0a80 https://github.com/AkihiroSuda/containerd.git
 github.com/containerd/continuity 2d3749b4da569ac97ca63dccba5eee4f5ee2beab
 github.com/containerd/fifo 3d5202aec260678c48179c56f40e6f38a095738c
 github.com/containerd/go-cni f2d7272f12d045b16ed924f50e91f9f9cecc55a7