fix: bad connection handling for in cluster dialer

pkg/k8s/dialer.go

@@ -1,13 +1,19 @@
 package k8s
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"io"
 	"net"
+	"os"
+	"regexp"
+	"strconv"
 	"sync"
+	"sync/atomic"
+	"syscall"
 	"time"
 
 	coreV1 "k8s.io/api/core/v1"
@@ -66,21 +72,108 @@ func (c *contextDialer) DialContext(ctx context.Context, network string, addr st
 		return nil, fmt.Errorf("unsupported network: %q", network)
 	}
 
-	execDone := make(chan struct{})
-	pr, pw, conn := newConn(execDone)
-
+	ctrStdin, ctrStdout, conn := newConn()
+	connectSuccess := make(chan struct{})
+	connectFailure := make(chan error, 1)
 	go func() {
-		defer close(execDone)
-		errOut := bytes.NewBuffer(nil)
-		err := c.exec(addr, pr, pw, errOut)
+		stderrBuff := bytes.NewBuffer(nil)
+		ctrStderr := io.MultiWriter(stderrBuff, detectConnSuccess(connectSuccess))
+
+		err := c.exec(addr, ctrStdin, ctrStdout, ctrStderr)
 		if err != nil {
-			err = fmt.Errorf("failed to exec in pod: %w (stderr: %q)", err, errOut.String())
-			_ = pr.CloseWithError(err)
-			_ = pw.CloseWithError(err)
+			stderrStr := stderrBuff.String()
+			socatErr := tryParseSocatError(network, addr, stderrStr)
+			if socatErr != nil {
+				err = fmt.Errorf("socat error: %w", socatErr)
+			} else {
+				err = fmt.Errorf("failed to exec in pod: %w (stderr: %q)", err, stderrStr)
+			}
 		}
+		_ = conn.closeWithError(err)
+		connectFailure <- err
 	}()
 
-	return conn, nil
+	select {
+	case <-connectSuccess:
+		return conn, nil
+	case err := <-connectFailure:
+		return nil, err
+	case <-ctx.Done():
+		_ = conn.closeWithError(ctx.Err())
+		return nil, ctx.Err()
+	}
+}
+
+var connSuccessfulRE = regexp.MustCompile("successfully connected")
+
+// Creates io.Writer which closes connectSuccess channel when string "successfully connected" is written to it.
+func detectConnSuccess(connectSuccess chan struct{}) io.Writer {
+	pr, pw := io.Pipe()
+	go func() {
+		ok := connSuccessfulRE.MatchReader(bufio.NewReader(pr))
+		if ok {
+			close(connectSuccess)
+		}
+		_, _ = io.Copy(io.Discard, pr)
+	}()
+	return pw
+}
+
+var (
+	connectionRefusedErrorRE = regexp.MustCompile(`E connect\(\d+, AF=\d+ (?P<hostport>[\[\]0-9.:a-z]+), \d+\): Connection refused`)
+	nameResolutionErrorRE    = regexp.MustCompile(`E getaddrinfo\("(?P<hostname>[a-zA-z-.0-9]+)",.*\): Name does not resolve`)
+)
+
+// tries to detect common errors from `socat` stderr
+func tryParseSocatError(network, address, stderr string) error {
+	groups := nameResolutionErrorRE.FindStringSubmatch(stderr)
+	if groups != nil {
+		var name string
+		if len(groups) > 1 {
+			name = groups[1]
+		}
+		return &net.OpError{
+			Op:     "dial",
+			Net:    network,
+			Source: nil,
+			Addr:   nil,
+			Err: &net.DNSError{
+				Err:        "no such host",
+				Name:       name,
+				IsNotFound: true,
+			},
+		}
+	}
+	groups = connectionRefusedErrorRE.FindStringSubmatch(stderr)
+	if groups != nil {
+		var (
+			addr net.IP
+			port int
+			zone string
+		)
+		if len(groups) > 1 {
+			h, p, err := net.SplitHostPort(groups[1])
+			if err == nil {
+				addr = net.ParseIP(h)
+				p, _ := strconv.ParseInt(p, 10, 16)
+				port = int(p)
+			}
+		}
+		return &net.OpError{
+			Op:  "dial",
+			Net: network,
+			Addr: &net.TCPAddr{
+				IP:   addr,
+				Port: port,
+				Zone: zone,
+			},
+			Err: &os.SyscallError{
+				Syscall: "connect",
+				Err:     syscall.ECONNREFUSED,
+			},
+		}
+	}
+	return nil
 }
 
 func (c *contextDialer) Close() error {
@@ -197,7 +290,7 @@ func (c *contextDialer) exec(hostPort string, in io.Reader, out, errOut io.Write
 		Namespace(c.namespace).
 		SubResource("exec")
 	req.VersionedParams(&coreV1.PodExecOptions{
-		Command:   []string{"socat", "-", fmt.Sprintf("TCP:%s", hostPort)},
+		Command:   []string{"socat", "-dd", "-", fmt.Sprintf("TCP:%s", hostPort)},
 		Container: c.podName,
 		Stdin:     true,
 		Stdout:    true,
@@ -324,50 +417,73 @@ func (a addr) String() string {
 }
 
 type conn struct {
-	pr       *io.PipeReader
-	pw       *io.PipeWriter
-	execDone <-chan struct{}
+	pr  *io.PipeReader
+	pw  *io.PipeWriter
+	err atomic.Pointer[error]
 }
 
-func (c conn) Read(b []byte) (n int, err error) {
-	return c.pr.Read(b)
+func (c *conn) Read(b []byte) (n int, err error) {
+	n, err = c.pr.Read(b)
+	if errors.Is(err, io.ErrClosedPipe) {
+		if p := c.err.Load(); p != nil {
+			err = *p
+		}
+	}
+	return
 }
 
-func (c conn) Write(b []byte) (n int, err error) {
-	return c.pw.Write(b)
+func (c *conn) Write(b []byte) (n int, err error) {
+	n, err = c.pw.Write(b)
+	if errors.Is(err, io.ErrClosedPipe) {
+		if p := c.err.Load(); p != nil {
+			err = *p
+		}
+	}
+	return
 }
 
-func (c conn) Close() error {
-	err := c.pw.Close()
+func (c *conn) closeWithError(err error) error {
+	if err == nil {
+		err = net.ErrClosed
+	}
+
+	{
+		e := err
+		c.err.CompareAndSwap(nil, &e)
+	}
+	err = c.pw.CloseWithError(io.EOF)
 	if err != nil {
 		return fmt.Errorf("failed to close writer: %w", err)
 	}
-	<-c.execDone
-	err = c.pr.Close()
+	err = c.pr.CloseWithError(net.ErrClosed)
 	if err != nil {
 		return fmt.Errorf("failed to close reader: %w", err)
 	}
 	return nil
 }
 
-func (c conn) LocalAddr() net.Addr {
+func (c *conn) Close() error {
+	return c.closeWithError(nil)
+}
+
+func (c *conn) LocalAddr() net.Addr {
 	return addr{}
 }
 
-func (c conn) RemoteAddr() net.Addr {
+func (c *conn) RemoteAddr() net.Addr {
 	return addr{}
 }
 
-func (c conn) SetDeadline(t time.Time) error { return nil }
+func (c *conn) SetDeadline(t time.Time) error { return nil }
 
-func (c conn) SetReadDeadline(t time.Time) error { return nil }
+func (c *conn) SetReadDeadline(t time.Time) error { return nil }
 
-func (c conn) SetWriteDeadline(t time.Time) error { return nil }
+func (c *conn) SetWriteDeadline(t time.Time) error { return nil }
 
-func newConn(execDone <-chan struct{}) (*io.PipeReader, *io.PipeWriter, conn) {
+func newConn() (*io.PipeReader, *io.PipeWriter, *conn) {
 	pr0, pw0 := io.Pipe()
 	pr1, pw1 := io.Pipe()
-	rwc := conn{pr: pr0, pw: pw1, execDone: execDone}
+	rwc := &conn{pr: pr0, pw: pw1}
 	return pr1, pw0, rwc
 }
 

pkg/k8s/dialer_test.go

@@ -168,20 +168,21 @@ func TestDialUnreachable(t *testing.T) {
 		dialer.Close()
 	})
 
-	transport := &http.Transport{
-		DialContext: dialer.DialContext,
+	_, err = dialer.DialContext(ctx, "tcp", "does-not.exists.svc:80")
+	if err == nil {
+		t.Error("error was expected but got nil")
+		return
 	}
-
-	var client = http.Client{
-		Transport: transport,
+	if !strings.Contains(err.Error(), "no such host") {
+		t.Errorf("error %q doesn't containe expected substring: ", err.Error())
 	}
 
-	_, err = client.Get("http://does-not.exists.svc")
+	_, err = dialer.DialContext(ctx, "tcp", "localhost:80")
 	if err == nil {
 		t.Error("error was expected but got nil")
 		return
 	}
-	if !strings.Contains(err.Error(), "not resolve") {
-		t.Errorf("error %q doesn't containe expected sub-string: ", err.Error())
+	if !strings.Contains(err.Error(), "connection refused") {
+		t.Errorf("error %q doesn't containe expected substring: ", err.Error())
 	}
 }