Merge pull request #1489 from crosbymichael/list-state

Handle locking and errors for process state

linux/shim/init.go

@@ -222,8 +222,13 @@ func (p *initProcess) ExitedAt() time.Time {
 }
 
 func (p *initProcess) Status(ctx context.Context) (string, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	c, err := p.runtime.State(ctx, p.id)
 	if err != nil {
+		if os.IsNotExist(err) {
+			return "stopped", nil
+		}
 		return "", p.runtimeError(err, "OCI runtime state failed")
 	}
 	return c.Status, nil

linux/task.go

@@ -5,6 +5,8 @@ package linux
 import (
 	"context"
 
+	"google.golang.org/grpc"
+
 	"github.com/containerd/cgroups"
 	"github.com/containerd/containerd/api/types/task"
 	"github.com/containerd/containerd/errdefs"
@@ -63,7 +65,10 @@ func (t *Task) State(ctx context.Context) (runtime.State, error) {
 		ID: t.id,
 	})
 	if err != nil {
-		return runtime.State{}, errdefs.FromGRPC(err)
+		if err != grpc.ErrServerStopped {
+			return runtime.State{}, errdefs.FromGRPC(err)
+		}
+		return runtime.State{}, errdefs.ErrNotFound
 	}
 	var status runtime.Status
 	switch response.Status {

services/tasks/service.go

@@ -274,17 +274,22 @@ func (s *Service) List(ctx context.Context, r *api.ListTasksRequest) (*api.ListT
 		if err != nil {
 			return nil, errdefs.ToGRPC(err)
 		}
-		for _, t := range tasks {
+		addTasks(ctx, resp, tasks)
-			tt, err := processFromContainerd(ctx, t)
-			if err != nil {
-				return nil, err
-			}
-			resp.Tasks = append(resp.Tasks, tt)
-		}
 	}
 	return resp, nil
 }
 
+func addTasks(ctx context.Context, r *api.ListTasksResponse, tasks []runtime.Task) {
+	for _, t := range tasks {
+		tt, err := processFromContainerd(ctx, t)
+		if err != nil {
+			log.G(ctx).WithError(err).WithField("id", t.ID()).Error("converting task to protobuf")
+			continue
+		}
+		r.Tasks = append(r.Tasks, tt)
+	}
+}
+
 func (s *Service) Pause(ctx context.Context, r *api.PauseTaskRequest) (*google_protobuf.Empty, error) {
 	t, err := s.getTask(ctx, r.ContainerID)
 	if err != nil {