Merge pull request #1603 from mlaventure/fix-windows-flaky-tests

windows: ensure a init process exist after "Create()"

container_test.go

@@ -203,9 +203,9 @@ func TestContainerOutput(t *testing.T) {
 	}
 
 	status := <-statusC
-	code, _, _ := status.Result()
+	code, _, err := status.Result()
 	if code != 0 {
-		t.Errorf("expected status 0 but received %d", code)
+		t.Errorf("expected status 0 but received %d: %v", code, err)
 	}
 	if _, err := task.Delete(ctx); err != nil {
 		t.Error(err)

windows/process.go

@@ -5,6 +5,7 @@ package windows
 import (
 	"context"
 	"io"
+	"sync"
 	"syscall"
 	"time"
 
@@ -19,13 +20,14 @@ import (
 
 // process implements containerd.Process and containerd.State
 type process struct {
+	sync.Mutex
+
 	hcs hcsshim.Process
 
-	id     string
-	pid    uint32
-	io     *pipeSet
-	status runtime.Status
-	task   *task
+	id   string
+	pid  uint32
+	io   *pipeSet
+	task *task
 
 	exitCh   chan struct{}
 	exitCode uint32
@@ -50,6 +52,9 @@ func (p *process) State(ctx context.Context) (runtime.State, error) {
 }
 
 func (p *process) Status() runtime.Status {
+	p.Lock()
+	defer p.Unlock()
+
 	if p.task.getStatus() == runtime.PausedStatus {
 		return runtime.PausedStatus
 	}
@@ -69,15 +74,24 @@ func (p *process) Status() runtime.Status {
 
 func (p *process) Kill(ctx context.Context, sig uint32, all bool) error {
 	// On windows all signals kill the process
+	if p.Status() == runtime.CreatedStatus {
+		return errors.Wrap(errdefs.ErrFailedPrecondition, "process was not started")
+	}
 	return errors.Wrap(p.hcs.Kill(), "failed to kill process")
 }
 
 func (p *process) ResizePty(ctx context.Context, size runtime.ConsoleSize) error {
+	if p.Status() == runtime.CreatedStatus {
+		return errors.Wrap(errdefs.ErrFailedPrecondition, "process was not started")
+	}
 	err := p.hcs.ResizeConsole(uint16(size.Width), uint16(size.Height))
 	return errors.Wrap(err, "failed to resize process console")
 }
 
 func (p *process) CloseIO(ctx context.Context) error {
+	if p.Status() == runtime.CreatedStatus {
+		return errors.Wrap(errdefs.ErrFailedPrecondition, "process was not started")
+	}
 	return errors.Wrap(p.hcs.CloseStdin(), "failed to close stdin")
 }
 
@@ -93,6 +107,13 @@ func (p *process) ExitCode() (uint32, time.Time, error) {
 }
 
 func (p *process) Start(ctx context.Context) (err error) {
+	p.Lock()
+	defer p.Unlock()
+
+	if p.hcs != nil {
+		return errors.Wrap(errdefs.ErrFailedPrecondition, "process already started")
+	}
+
 	// If we fail, close the io right now
 	defer func() {
 		if err != nil {
@@ -132,6 +153,7 @@ func (p *process) Start(ctx context.Context) (err error) {
 	if p.io.stderr != nil {
 		go ioCopy("stderr", p.io.stderr, stderr)
 	}
+	p.hcs = hp
 
 	// Wait for the process to exit to get the exit status
 	go func() {
@@ -174,8 +196,6 @@ func (p *process) Start(ctx context.Context) (err error) {
 		// Cleanup HCS resources
 		hp.Close()
 	}()
-	p.status = runtime.RunningStatus
-	p.hcs = hp
 	return nil
 }
 

windows/runtime.go

@@ -304,6 +304,11 @@ func (r *windowsRuntime) newTask(ctx context.Context, namespace, id string, spec
 		hcsContainer:      ctr,
 		terminateDuration: createOpts.TerminateDuration,
 	}
+	// Create the new process but don't start it
+	pconf := newWindowsProcessConfig(t.spec.Process, t.io)
+	if _, err = t.newProcess(ctx, t.id, pconf, t.io); err != nil {
+		return nil, err
+	}
 	r.tasks.Add(ctx, t)
 
 	var rootfs []*containerdtypes.Mount

windows/task.go

@@ -111,13 +111,16 @@ func (t *task) Info() runtime.TaskInfo {
 }
 
 func (t *task) Start(ctx context.Context) error {
-	conf := newWindowsProcessConfig(t.spec.Process, t.io)
-	p, err := t.newProcess(ctx, t.id, conf, t.io)
-	if err != nil {
-		return err
+	p := t.getProcess(t.id)
+	if p == nil {
+		panic("init process is missing")
 	}
+
+	if p.Status() != runtime.CreatedStatus {
+		return errors.Wrap(errdefs.ErrFailedPrecondition, "process was already started")
+	}
+
 	if err := p.Start(ctx); err != nil {
-		t.removeProcess(t.id)
 		return err
 	}
 	t.publisher.Publish(ctx,
@@ -136,13 +139,13 @@ func (t *task) Pause(ctx context.Context) error {
 			t.Lock()
 			t.status = runtime.PausedStatus
 			t.Unlock()
-		}
-		if err == nil {
+
 			t.publisher.Publish(ctx,
 				runtime.TaskPausedEventTopic,
 				&eventsapi.TaskPaused{
 					ContainerID: t.id,
 				})
+			return nil
 		}
 		return errors.Wrap(err, "hcsshim failed to pause task")
 	}
@@ -157,13 +160,13 @@ func (t *task) Resume(ctx context.Context) error {
 			t.Lock()
 			t.status = runtime.RunningStatus
 			t.Unlock()
-		}
-		if err == nil {
+
 			t.publisher.Publish(ctx,
 				runtime.TaskResumedEventTopic,
 				&eventsapi.TaskResumed{
 					ContainerID: t.id,
 				})
+			return nil
 		}
 		return errors.Wrap(err, "hcsshim failed to resume task")
 	}
@@ -313,7 +316,6 @@ func (t *task) newProcess(ctx context.Context, id string, conf *hcsshim.ProcessC
 		id:     id,
 		pid:    pid,
 		io:     pset,
-		status: runtime.CreatedStatus,
 		task:   t,
 		exitCh: make(chan struct{}),
 		conf:   conf,