Improve socat conn success detection

It appears that Go's MatchReader() returns true only if some new lines
are written to the writer after the searched word. It mostly all right
because socat actually writes several more lines to stderr. However this
new implementation is better it should signal immediately when searched
patter is written to the writer.

Signed-off-by: Matej Vašek <[email protected]>

pkg/k8s/dialer.go

@@ -1,7 +1,6 @@
 package k8s
 
 import (
-	"bufio"
 	"bytes"
 	"context"
 	"errors"
@@ -107,19 +106,72 @@ func (c *contextDialer) DialContext(ctx context.Context, network string, addr st
 	}
 }
 
-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)
+	return newKMPWriter("successfully connected", connectSuccess)
+}
+
+// kmpWriter is a writer that search word w using KMP algorithm in the text written to the writer.
+// When searched word w appears in the input the channel ch is closed.
+// This can be used to detect if particular character sequence was written to the writer.
+type kmpWriter struct {
+	w     string
+	k     int
+	t     []int
+	ch    chan<- struct{}
+	found bool
+}
+
+func newKMPWriter(w string, ch chan<- struct{}) *kmpWriter {
+	// Building KMP table.
+	T := make([]int, len(w)+1)
+	T[0] = -1
+	pos := 1
+	cnd := 0
+	for pos < len(w) {
+		if w[pos] == w[cnd] {
+			T[pos] = T[cnd]
+		} else {
+			T[pos] = cnd
+			for cnd >= 0 && w[pos] != w[cnd] {
+				cnd = T[cnd]
+			}
 		}
-		_, _ = io.Copy(io.Discard, pr)
-	}()
-	return pw
+		pos++
+		cnd++
+	}
+	T[pos] = cnd
+
+	return &kmpWriter{
+		w:  w,
+		t:  T,
+		ch: ch,
+	}
+}
+
+func (d *kmpWriter) Write(s []byte) (n int, err error) {
+	if d.found {
+		return len(s), nil
+	}
+	j := 0
+	for j < len(s) {
+		if d.w[d.k] == s[j] {
+			j++
+			d.k++
+			if d.k == len(d.w) {
+				d.found = true
+				close(d.ch)
+				return len(s), nil
+			}
+		} else {
+			d.k = d.t[d.k]
+			if d.k < 0 {
+				j++
+				d.k++
+			}
+		}
+	}
+	return j, nil
 }
 
 var (

pkg/k8s/dialer_unit_test.go

@@ -0,0 +1,25 @@
+package k8s
+
+import (
+	"testing"
+	"time"
+)
+
+func TestDetectSocatSuccess(t *testing.T) {
+	ch := make(chan struct{}, 1)
+	w := detectConnSuccess(ch)
+	_, err := w.Write([]byte("some data successucces"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = w.Write([]byte("sfully connected"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-ch:
+		t.Log("OK")
+	case <-time.After(time.Millisecond * 100):
+		t.Error("NOK")
+	}
+}