Merge pull request #523 from superseb/portcheck_timeout

Portcheck timeout

cluster/network.go

@@ -1,14 +1,14 @@
 package cluster
 
 import (
-	"bufio"
+	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"net"
 	"strings"
 
 	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/pkg/stdcopy"
 	"github.com/docker/go-connections/nat"
 	"github.com/rancher/rke/docker"
 	"github.com/rancher/rke/hosts"
@@ -402,13 +402,16 @@ func (c *Cluster) runServicePortChecks(ctx context.Context) error {
 }
 
 func checkPlaneTCPPortsFromHost(ctx context.Context, host *hosts.Host, portList []string, planeHosts []*hosts.Host, image string, prsMap map[string]v3.PrivateRegistry) error {
-	hosts := []string{}
+	var hosts []string
+	var portCheckLogs []string
+	var containerStdout bytes.Buffer
+	var containerStderr bytes.Buffer
+
 	for _, host := range planeHosts {
 		hosts = append(hosts, host.InternalAddress)
 	}
 	imageCfg := &container.Config{
 		Image: image,
-		Tty:   true,
 		Env: []string{
 			fmt.Sprintf("HOSTS=%s", strings.Join(hosts, " ")),
 			fmt.Sprintf("PORTS=%s", strings.Join(portList, " ")),
@@ -416,11 +419,14 @@ func checkPlaneTCPPortsFromHost(ctx context.Context, host *hosts.Host, portList
 		Cmd: []string{
 			"sh",
 			"-c",
-			"for host in $HOSTS; do for port in $PORTS ; do nc -z $host $port > /dev/null || echo $host $port ; done; done",
+			"for host in $HOSTS; do for port in $PORTS ; do echo \"Checking host ${host} on port ${port}\" >&1 & nc -w 5 -z $host $port > /dev/null || echo \"${host}:${port}\" >&2 & done; wait; done",
 		},
 	}
 	hostCfg := &container.HostConfig{
 		NetworkMode: "host",
+		LogConfig: container.LogConfig{
+			Type: "json-file",
+		},
 	}
 	if err := docker.DoRemoveContainer(ctx, host.DClient, PortCheckContainer, host.Address); err != nil {
 		return err
@@ -428,42 +434,28 @@ func checkPlaneTCPPortsFromHost(ctx context.Context, host *hosts.Host, portList
 	if err := docker.DoRunContainer(ctx, host.DClient, imageCfg, hostCfg, PortCheckContainer, host.Address, "network", prsMap); err != nil {
 		return err
 	}
-	if err := docker.WaitForContainer(ctx, host.DClient, host.Address, PortCheckContainer); err != nil {
-		return err
-	}
-	logs, err := docker.ReadContainerLogs(ctx, host.DClient, PortCheckContainer)
+
+	clogs, err := docker.ReadContainerLogs(ctx, host.DClient, PortCheckContainer)
 	if err != nil {
 		return err
 	}
-	defer logs.Close()
+	defer clogs.Close()
+
+	stdcopy.StdCopy(&containerStdout, &containerStderr, clogs)
+	containerLog := containerStderr.String()
+	logrus.Debugf("[network] containerLog [%s] on host: %s", containerLog, host.Address)
+
 	if err := docker.RemoveContainer(ctx, host.DClient, host.Address, PortCheckContainer); err != nil {
 		return err
 	}
-	portCheckLogs, err := getPortCheckLogs(logs)
-	if err != nil {
-		return err
-	}
-	if len(portCheckLogs) > 0 {
-
-		return fmt.Errorf("[network] Port check for ports: [%s] failed on host: [%s]", strings.Join(portCheckLogs, ", "), host.Address)
-
+	logrus.Debugf("[network] Length of portCheckLogs is [%d] on host: %s", len(portCheckLogs), host.Address)
+	if len(containerLog) > 0 {
+		portCheckLogs := strings.Join(strings.Split(strings.TrimSpace(containerLog), "\n"), ", ")
+		return fmt.Errorf("[network] Port check for ports: [%s] failed on host: [%s]", portCheckLogs, host.Address)
 	}
 	return nil
 }
 
-func getPortCheckLogs(reader io.ReadCloser) ([]string, error) {
-	logLines := bufio.NewScanner(reader)
-	hostPortLines := []string{}
-	for logLines.Scan() {
-		logLine := strings.Split(logLines.Text(), " ")
-		hostPortLines = append(hostPortLines, fmt.Sprintf("%s:%s", logLine[0], logLine[1]))
-	}
-	if err := logLines.Err(); err != nil {
-		return nil, err
-	}
-	return hostPortLines, nil
-}
-
 func getPortBindings(hostAddress string, portList []string) []nat.PortBinding {
 	portBindingList := []nat.PortBinding{}
 	for _, portNumber := range portList {

docker/docker.go

@@ -333,8 +333,7 @@ func ReadFileFromContainer(ctx context.Context, dClient *client.Client, hostname
 }
 
 func ReadContainerLogs(ctx context.Context, dClient *client.Client, containerName string) (io.ReadCloser, error) {
-	return dClient.ContainerLogs(ctx, containerName, types.ContainerLogsOptions{ShowStdout: true})
-
+	return dClient.ContainerLogs(ctx, containerName, types.ContainerLogsOptions{Follow: true, ShowStdout: true, ShowStderr: true, Timestamps: false})
 }
 
 func tryRegistryAuth(pr v3.PrivateRegistry) types.RequestPrivilegeFunc {

vendor/github.com/docker/docker/pkg/README.md

@@ -0,0 +1,11 @@
+pkg/ is a collection of utility packages used by the Moby project without being specific to its internals.
+
+Utility packages are kept separate from the moby core codebase to keep it as small and concise as possible.
+If some utilities grow larger and their APIs stabilize, they may be moved to their own repository under the
+Moby organization, to facilitate re-use by other projects. However that is not the priority.
+
+The directory `pkg` is named after the same directory in the camlistore project. Since Brad is a core
+Go maintainer, we thought it made sense to copy his methods for organizing Go code :) Thanks Brad!
+
+Because utility packages are small and neatly separated from the rest of the codebase, they are a good
+place to start for aspiring maintainers and contributors. Get in touch if you want to help maintain them!

vendor/github.com/docker/docker/pkg/stdcopy/stdcopy.go

@@ -0,0 +1,190 @@
+package stdcopy
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+)
+
+// StdType is the type of standard stream
+// a writer can multiplex to.
+type StdType byte
+
+const (
+	// Stdin represents standard input stream type.
+	Stdin StdType = iota
+	// Stdout represents standard output stream type.
+	Stdout
+	// Stderr represents standard error steam type.
+	Stderr
+	// Systemerr represents errors originating from the system that make it
+	// into the the multiplexed stream.
+	Systemerr
+
+	stdWriterPrefixLen = 8
+	stdWriterFdIndex   = 0
+	stdWriterSizeIndex = 4
+
+	startingBufLen = 32*1024 + stdWriterPrefixLen + 1
+)
+
+var bufPool = &sync.Pool{New: func() interface{} { return bytes.NewBuffer(nil) }}
+
+// stdWriter is wrapper of io.Writer with extra customized info.
+type stdWriter struct {
+	io.Writer
+	prefix byte
+}
+
+// Write sends the buffer to the underneath writer.
+// It inserts the prefix header before the buffer,
+// so stdcopy.StdCopy knows where to multiplex the output.
+// It makes stdWriter to implement io.Writer.
+func (w *stdWriter) Write(p []byte) (n int, err error) {
+	if w == nil || w.Writer == nil {
+		return 0, errors.New("Writer not instantiated")
+	}
+	if p == nil {
+		return 0, nil
+	}
+
+	header := [stdWriterPrefixLen]byte{stdWriterFdIndex: w.prefix}
+	binary.BigEndian.PutUint32(header[stdWriterSizeIndex:], uint32(len(p)))
+	buf := bufPool.Get().(*bytes.Buffer)
+	buf.Write(header[:])
+	buf.Write(p)
+
+	n, err = w.Writer.Write(buf.Bytes())
+	n -= stdWriterPrefixLen
+	if n < 0 {
+		n = 0
+	}
+
+	buf.Reset()
+	bufPool.Put(buf)
+	return
+}
+
+// NewStdWriter instantiates a new Writer.
+// Everything written to it will be encapsulated using a custom format,
+// and written to the underlying `w` stream.
+// This allows multiple write streams (e.g. stdout and stderr) to be muxed into a single connection.
+// `t` indicates the id of the stream to encapsulate.
+// It can be stdcopy.Stdin, stdcopy.Stdout, stdcopy.Stderr.
+func NewStdWriter(w io.Writer, t StdType) io.Writer {
+	return &stdWriter{
+		Writer: w,
+		prefix: byte(t),
+	}
+}
+
+// StdCopy is a modified version of io.Copy.
+//
+// StdCopy will demultiplex `src`, assuming that it contains two streams,
+// previously multiplexed together using a StdWriter instance.
+// As it reads from `src`, StdCopy will write to `dstout` and `dsterr`.
+//
+// StdCopy will read until it hits EOF on `src`. It will then return a nil error.
+// In other words: if `err` is non nil, it indicates a real underlying error.
+//
+// `written` will hold the total number of bytes written to `dstout` and `dsterr`.
+func StdCopy(dstout, dsterr io.Writer, src io.Reader) (written int64, err error) {
+	var (
+		buf       = make([]byte, startingBufLen)
+		bufLen    = len(buf)
+		nr, nw    int
+		er, ew    error
+		out       io.Writer
+		frameSize int
+	)
+
+	for {
+		// Make sure we have at least a full header
+		for nr < stdWriterPrefixLen {
+			var nr2 int
+			nr2, er = src.Read(buf[nr:])
+			nr += nr2
+			if er == io.EOF {
+				if nr < stdWriterPrefixLen {
+					return written, nil
+				}
+				break
+			}
+			if er != nil {
+				return 0, er
+			}
+		}
+
+		stream := StdType(buf[stdWriterFdIndex])
+		// Check the first byte to know where to write
+		switch stream {
+		case Stdin:
+			fallthrough
+		case Stdout:
+			// Write on stdout
+			out = dstout
+		case Stderr:
+			// Write on stderr
+			out = dsterr
+		case Systemerr:
+			// If we're on Systemerr, we won't write anywhere.
+			// NB: if this code changes later, make sure you don't try to write
+			// to outstream if Systemerr is the stream
+			out = nil
+		default:
+			return 0, fmt.Errorf("Unrecognized input header: %d", buf[stdWriterFdIndex])
+		}
+
+		// Retrieve the size of the frame
+		frameSize = int(binary.BigEndian.Uint32(buf[stdWriterSizeIndex : stdWriterSizeIndex+4]))
+
+		// Check if the buffer is big enough to read the frame.
+		// Extend it if necessary.
+		if frameSize+stdWriterPrefixLen > bufLen {
+			buf = append(buf, make([]byte, frameSize+stdWriterPrefixLen-bufLen+1)...)
+			bufLen = len(buf)
+		}
+
+		// While the amount of bytes read is less than the size of the frame + header, we keep reading
+		for nr < frameSize+stdWriterPrefixLen {
+			var nr2 int
+			nr2, er = src.Read(buf[nr:])
+			nr += nr2
+			if er == io.EOF {
+				if nr < frameSize+stdWriterPrefixLen {
+					return written, nil
+				}
+				break
+			}
+			if er != nil {
+				return 0, er
+			}
+		}
+
+		// we might have an error from the source mixed up in our multiplexed
+		// stream. if we do, return it.
+		if stream == Systemerr {
+			return written, fmt.Errorf("error from daemon in stream: %s", string(buf[stdWriterPrefixLen:frameSize+stdWriterPrefixLen]))
+		}
+
+		// Write the retrieved frame (without header)
+		nw, ew = out.Write(buf[stdWriterPrefixLen : frameSize+stdWriterPrefixLen])
+		if ew != nil {
+			return 0, ew
+		}
+
+		// If the frame has not been fully written: error
+		if nw != frameSize {
+			return 0, io.ErrShortWrite
+		}
+		written += int64(nw)
+
+		// Move the rest of the buffer to the beginning
+		copy(buf, buf[frameSize+stdWriterPrefixLen:])
+		// Move the index
+		nr -= frameSize + stdWriterPrefixLen
+	}
+}