add bats tests

Author: aojea

.github/workflows/bats.yml

@@ -0,0 +1,73 @@
+name: bats
+
+on:
+  push:
+    branches:
+      - 'main'
+    tags:
+      - 'v*'
+  pull_request:
+    branches: [ main ]
+  workflow_dispatch:
+
+env:
+  GO_VERSION: "1.23"
+  K8S_VERSION: "v1.32.0"
+  KIND_VERSION: "v0.27.0"
+
+jobs:
+  bats_tests:
+    runs-on: ubuntu-22.04
+    name: Bats e2e tests
+    steps:
+      - name: Checkout
+        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+      - name: Setup Bats and bats libs
+        id: setup-bats
+        uses: bats-core/bats-action@3.0.0
+      - name: Bats tests
+        shell: bash
+        env:
+         BATS_LIB_PATH: ${{ steps.setup-bats.outputs.lib-path }}
+         TERM: xterm
+        run: bats -o _artifacts tests/
+
+      - name: Upload logs
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: kind-logs-${{ env.JOB_NAME }}-${{ github.run_id }}
+          path: ./_artifacts
+ 
+  bats_mac_tests:
+    runs-on: macos-13
+    name: Bats e2e tests on Mac
+    steps:
+      - name: Checkout
+        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+      - name: Set up environment (download dependencies)
+        run: |
+          brew install docker colima
+          brew install kind
+          brew install golang
+          brew install bats-core
+          brew install kubectl
+          LIMA_SSH_PORT_FORWARDER=true colima start
+      - name: Bats tests
+        env:
+          TERM: linux
+        run: |
+          bash --version
+          bash -c "time bats -o _artifacts --trace --formatter tap tests/"
+      - name: Debug info
+        if: always()
+        run: |
+          docker ps
+
+      - name: Upload logs
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: kind-logs-${{ env.JOB_NAME }}-${{ github.run_id }}
+          path: ./_artifacts
+         

.github/workflows/k8s.yml

@@ -32,7 +32,7 @@ jobs:
       id: go
 
     - name: Check out code
-      uses: actions/checkout@v2
+      uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
 
     - name: Enable ipv4 and ipv6 forwarding
       run: |

examples/loadbalancer_udp_tcp.yaml

@@ -6,7 +6,7 @@ spec:
   selector:
     matchLabels:
       app: multiprotocol
-  replicas: 2
+  replicas: 1
   strategy:
     rollingUpdate:
       maxSurge: 0

pkg/loadbalancer/tunnel.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"time"
 
 	"k8s.io/klog/v2"
 
@@ -99,6 +100,7 @@ func (t *tunnelManager) removeTunnels(containerName string) error {
 // tunnel listens on localIP:localPort and proxies the connection to remoteIP:remotePort
 type tunnel struct {
 	listener   net.Listener
+	udpConn    *net.UDPConn
 	localIP    string
 	localPort  string
 	protocol   string
@@ -118,35 +120,57 @@ func NewTunnel(localIP, localPort, protocol, remoteIP, remotePort string) *tunne
 
 func (t *tunnel) Start() error {
 	klog.Infof("Starting tunnel on %s", net.JoinHostPort(t.localIP, t.localPort))
-	ln, err := net.Listen(t.protocol, net.JoinHostPort(t.localIP, t.localPort))
-	if err != nil {
-		return err
-	}
+	if t.protocol == "udp" {
+		localAddrStr := net.JoinHostPort(t.localIP, t.localPort)
+		udpAddr, err := net.ResolveUDPAddr("udp", localAddrStr)
+		if err != nil {
+			return err
+		}
+		conn, err := net.ListenUDP("udp", udpAddr)
+		t.udpConn = conn
+		go func() {
+			for {
+				err = t.handleUDPConnection(conn)
+				if err != nil {
+					klog.Infof("unexpected error on connection: %v", err)
+				}
+			}
+		}()
 
-	t.listener = ln
+	} else {
+		ln, err := net.Listen(t.protocol, net.JoinHostPort(t.localIP, t.localPort))
+		if err != nil {
+			return err
+		}
 
-	go func() {
-		for {
-			conn, err := ln.Accept()
-			if err != nil {
-				klog.Infof("unexpected error listening: %v", err)
-				return
-			} else {
-				go func() {
-					err := t.handleConnection(conn)
-					if err != nil {
-						klog.Infof("unexpected error on connection: %v", err)
-					}
-				}()
+		t.listener = ln
+
+		go func() {
+			for {
+				conn, err := ln.Accept()
+				if err != nil {
+					klog.Infof("unexpected error listening: %v", err)
+					return
+				} else {
+					go func() {
+						err := t.handleConnection(conn)
+						if err != nil {
+							klog.Infof("unexpected error on connection: %v", err)
+						}
+					}()
+				}
 			}
-		}
-	}()
+		}()
+	}
 	return nil
 }
 
 func (t *tunnel) Stop() error {
 	if t.listener != nil {
-		return t.listener.Close()
+		_ = t.listener.Close()
+	}
+	if t.udpConn != nil {
+		_ = t.udpConn.Close()
 	}
 	return nil
 }
@@ -171,3 +195,42 @@ func (t *tunnel) handleConnection(local net.Conn) error {
 	wg.Wait()
 	return nil
 }
+
+func (t *tunnel) handleUDPConnection(conn *net.UDPConn) error {
+	buf := make([]byte, 1500)
+	nRead, srcAddr, err := conn.ReadFromUDP(buf)
+	if err != nil {
+		return err
+	}
+	klog.V(4).Infof("Read %d bytes from %s", nRead, srcAddr.String())
+
+	klog.V(4).Infof("Connecting to %s", net.JoinHostPort(t.remoteIP, t.remotePort))
+	remoteConn, err := net.Dial(t.protocol, net.JoinHostPort(t.remoteIP, t.remotePort))
+	if err != nil {
+		return fmt.Errorf("can't connect to server %q: %v", net.JoinHostPort(t.remoteIP, t.remotePort), err)
+	}
+	defer remoteConn.Close()
+
+	nWrite, err := remoteConn.Write(buf)
+	if err != nil {
+		return fmt.Errorf("fail to write to remote %s: %s", remoteConn.RemoteAddr(), err)
+	} else if nWrite < nRead {
+		klog.V(2).Infof("Buffer underflow %d < %d to remote %s", nWrite, nRead, remoteConn.RemoteAddr())
+	}
+	klog.V(4).Infof("Wrote %d bytes to to %s", nWrite, remoteConn.RemoteAddr().String())
+
+	buf = make([]byte, 1500)
+	err = remoteConn.SetReadDeadline(time.Now().Add(5 * time.Second)) // Add deadline to ensure it doesn't block forever
+	if err != nil {
+		return fmt.Errorf("can not set read deadline: %v", err)
+	}
+	nRead, err = remoteConn.Read(buf)
+	if err != nil {
+		return fmt.Errorf("fail to read from remote %s: %s", remoteConn.RemoteAddr(), err)
+	}
+	klog.V(4).Infof("Read %d bytes from %s", nRead, remoteConn.RemoteAddr().String())
+
+	_, err = conn.WriteToUDP(buf, srcAddr)
+
+	return err
+}

tests/README.md

@@ -0,0 +1,8 @@
+# Integration tests
+
+
+1. Install `bats` https://bats-core.readthedocs.io/en/stable/installation.html
+
+2. Install `kind` https://kind.sigs.k8s.io/
+
+3. Run `bats tests/`

tests/setup_suite.bash

@@ -0,0 +1,37 @@
+#!/bin/bash
+
+set -eu
+
+function setup_suite {
+  export BATS_TEST_TIMEOUT=120
+  # Define the name of the kind cluster
+  export CLUSTER_NAME="ccm-kind"
+  # Build the cloud-provider-kind
+ 
+  mkdir -p "$BATS_TEST_DIRNAME"/../_artifacts
+  rm -rf "$BATS_TEST_DIRNAME"/../_artifacts/*
+  # create cluster
+  cat <<EOF | kind create cluster \
+  --name $CLUSTER_NAME           \
+  -v7 --wait 1m --retain --config=-
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+nodes:
+- role: control-plane
+- role: worker
+- role: worker
+EOF
+
+  cd "$BATS_TEST_DIRNAME"/.. && make
+  nohup "$BATS_TEST_DIRNAME"/../bin/cloud-provider-kind -v 2 --enable-log-dumping --logs-dir "$BATS_TEST_DIRNAME"/../_artifacts > "$BATS_TEST_DIRNAME"/../_artifacts/ccm-kind.log 2>&1 &
+  export CCM_PID=$!
+
+  # test depend on external connectivity that can be very flaky
+  sleep 5
+}
+
+function teardown_suite {
+    kill "$CCM_PID"
+    kind export logs "$BATS_TEST_DIRNAME"/../_artifacts --name "$CLUSTER_NAME"
+    kind delete cluster --name "$CLUSTER_NAME"
+}