fix attach bug

manager/workload/attach.go

@@ -3,6 +3,7 @@ package workload
 import (
 	"bufio"
 	"context"
+	"fmt"
 	"io"
 	"strings"
 	"sync"
@@ -67,12 +68,16 @@ func (m *Manager) attach(ctx context.Context, ID string) {
 	wg := &sync.WaitGroup{}
 	pump := func(typ string, source io.Reader) {
 		defer wg.Done()
+		log.Debugf("[attach] attach pump %s %s %s start", workloadName, ID, typ)
+		defer log.Debugf("[attach] attach pump %s %s %s finished", workloadName, ID, typ)
+
 		buf := bufio.NewReader(source)
 		for {
 			data, err := buf.ReadString('\n')
+			fmt.Println("attach:", data, err)
 			if err != nil {
 				if err != io.EOF {
-					log.Errorf("[attach] attach pump %s %s %s %s", workloadName, ID, typ, err)
+					log.Errorf("[attach] attach pump %s %s %s failed, err: %v", workloadName, ID, typ, err)
 				}
 				return
 			}

runtime/docker/docker.go

@@ -140,6 +140,9 @@ func (d *Docker) AttachWorkload(ctx context.Context, ID string) (io.Reader, io.R
 			resp.Close()
 			outw.Close()
 			errw.Close()
+			outr.Close()
+			errr.Close()
+			log.Debugf("[attach] %v buf pipes closed", ID)
 		}()
 
 		_, err = stdcopy.StdCopy(outw, errw, resp.Reader)

utils/bufpipe.go

@@ -3,6 +3,7 @@ package utils
 import (
 	"bytes"
 	"errors"
+	"fmt"
 	"io"
 	"sync"
 )
@@ -49,9 +50,12 @@ func NewBufPipe(bufCap int64) (*PipeReader, *PipeWriter) {
 func (r *PipeReader) Read(data []byte) (int, error) {
 	r.cond.L.Lock()
 	defer r.cond.L.Unlock()
+	fmt.Println("read start")
+	defer fmt.Println("read done")
 
 RETRY:
 	n, err := r.buf.Read(data)
+	fmt.Println("bufpipe:", n, err, r.rerr)
 	// If not closed and no read, wait for writing.
 	if err == io.EOF && r.rerr == nil && n == 0 {
 		r.cond.Wait()
@@ -121,5 +125,6 @@ func (w *PipeWriter) CloseWithError(err error) error {
 		err = io.EOF
 	}
 	w.rerr = err
+	w.cond.Signal()
 	return nil
 }

utils/bufpipe_test.go

@@ -0,0 +1,33 @@
+package utils
+
+import (
+	"bufio"
+	"fmt"
+	"github.com/docker/go-units"
+	"github.com/stretchr/testify/assert"
+	"io"
+	"testing"
+	"time"
+)
+
+func TestBufPipe(t *testing.T)  {
+	size, _ := units.RAMInBytes("10M")
+	r, w := NewBufPipe(size)
+	w.Write([]byte("test"))
+	go func() {
+		time.Sleep(time.Second)
+		w.Close()
+		r.Close()
+	}()
+
+	reader := bufio.NewReader(r)
+
+	for {
+		_, err := reader.ReadString('\n')
+		if err != nil {
+			fmt.Println(err)
+			assert.Equal(t, err, io.EOF)
+			return
+		}
+	}
+}