From 0ef0118150bb91d6d961c26dffd7c8a75e511a41 Mon Sep 17 00:00:00 2001
From: Tolga Ceylan <tolga.ceylan@gmail.com>
Date: Thu, 8 Mar 2018 15:46:32 -0800
Subject: [PATCH] fn: wait for async attach with success channel (#810)

* fn: wait for async attach with success channel

* fn: debug logs in test.sh

* fn: circleci test output as artifact

* fn: docker attach non-blocking adjustments

* fn: remove retry from risky NB attach
---
 .circleci/config.yml                      |  1 +
 api/agent/drivers/docker/docker.go        | 25 ++++++++++++++++++++---
 api/agent/drivers/docker/docker_client.go | 16 ++-------------
 3 files changed, 25 insertions(+), 17 deletions(-)

diff --git a/.circleci/config.yml b/.circleci/config.yml
index 1d4844894..623f08994 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -9,6 +9,7 @@ jobs:
       - GOVERSION=1.9.1
       - OS=linux
       - ARCH=amd64
+      - FN_LOG_LEVEL=debug
     steps:
       - checkout
       # install Go
diff --git a/api/agent/drivers/docker/docker.go b/api/agent/drivers/docker/docker.go
index d757f42e0..4a79a5a0f 100644
--- a/api/agent/drivers/docker/docker.go
+++ b/api/agent/drivers/docker/docker.go
@@ -345,17 +345,36 @@ func dockerMsg(derr *docker.Error) string {
 // Run executes the docker container. If task runs, drivers.RunResult will be returned. If something fails outside the task (ie: Docker), it will return error.
 // The docker driver will attempt to cast the task to a Auther. If that succeeds, private image support is available. See the Auther interface for how to implement this.
 func (drv *DockerDriver) run(ctx context.Context, container string, task drivers.ContainerTask) (drivers.WaitResult, error) {
+
+	attachSuccess := make(chan struct{})
 	mwOut, mwErr := task.Logger()
 
 	waiter, err := drv.docker.AttachToContainerNonBlocking(ctx, docker.AttachToContainerOptions{
-		Container: container, OutputStream: mwOut, ErrorStream: mwErr,
-		Stream: true, Stdout: true, Stderr: true,
-		Stdin: true, InputStream: task.Input()})
+		Success:      attachSuccess,
+		Container:    container,
+		OutputStream: mwOut,
+		ErrorStream:  mwErr,
+		Stream:       true,
+		Stdout:       true,
+		Stderr:       true,
+		Stdin:        true,
+		InputStream:  task.Input()})
+
 	if err != nil && ctx.Err() == nil {
 		// ignore if ctx has errored, rewrite status lay below
 		return nil, err
 	}
 
+	// Sync up with NB Attacher above before starting the task
+	if err == nil {
+		// WARNING: the I/O below requires docker hijack function to honor
+		// the contract below, specifically if an error is not returned
+		// from AttachToContainerNonBlocking, then max blocking time
+		// here should be what drv.docker dialer/client config was set to.
+		<-attachSuccess
+		attachSuccess <- struct{}{}
+	}
+
 	// we want to stop trying to collect stats when the container exits
 	// collectStats will stop when stopSignal is closed or ctx is cancelled
 	stopSignal := make(chan struct{})
diff --git a/api/agent/drivers/docker/docker_client.go b/api/agent/drivers/docker/docker_client.go
index 51c96ca30..6ba04b3c1 100644
--- a/api/agent/drivers/docker/docker_client.go
+++ b/api/agent/drivers/docker/docker_client.go
@@ -290,22 +290,10 @@ func (d *dockerWrap) Info(ctx context.Context) (info *docker.DockerInfo, err err
 	return d.docker.Info()
 }
 
-func (d *dockerWrap) AttachToContainerNonBlocking(ctx context.Context, opts docker.AttachToContainerOptions) (w docker.CloseWaiter, err error) {
+func (d *dockerWrap) AttachToContainerNonBlocking(ctx context.Context, opts docker.AttachToContainerOptions) (docker.CloseWaiter, error) {
 	ctx, span := trace.StartSpan(ctx, "docker_attach_container")
 	defer span.End()
-
-	logger := common.Logger(ctx).WithField("docker_cmd", "AttachContainer")
-	ctx, cancel := context.WithTimeout(ctx, retryTimeout)
-	defer cancel()
-	err = d.retry(ctx, logger, func() error {
-		w, err = d.docker.AttachToContainerNonBlocking(opts)
-		if err != nil {
-			// always retry if attach errors, task is running, we want logs!
-			err = temp(err)
-		}
-		return err
-	})
-	return w, err
+	return d.docker.AttachToContainerNonBlocking(opts)
 }
 
 func (d *dockerWrap) WaitContainerWithContext(id string, ctx context.Context) (code int, err error) {