Merge pull request #2135 from AkihiroSuda/oci-content-store

content: change Writer/ReaderAt to take OCI descriptor

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