conntrack: basic Example test

client/plugins/conntrack/conntrack.go

@@ -3,6 +3,7 @@ package conntrack
 import (
 	"sync"
 
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/kpng/client"
 	"sigs.k8s.io/kpng/client/diffstore2"
 	"sigs.k8s.io/kpng/client/localsink/fullstate"
@@ -61,6 +62,8 @@ func (ct Conntrack) Callback(ch <-chan *client.ServiceEndpoints) {
 					continue
 				}
 
+				klog.V(1).Info("svc IP ", svcIP)
+
 				ipp := IPPort{
 					Protocol: svcPort.Protocol,
 					DnatIP:   svcIP,
@@ -96,13 +99,19 @@ func (ct Conntrack) Callback(ch <-chan *client.ServiceEndpoints) {
 
 	ct.done()
 
+	klog.V(2).Info("received the new conntrack state")
+
 	for _, item := range ct.ipPorts.Changed() {
 		if item.Created() {
-			cleanupIPPortEntries(item.Value().Get())
+			ipp := item.Value().Get()
+			klog.V(1).Infof("cleaning conntrack entries for new service IP:port %v", ipp)
+			cleanupIPPortEntries(ipp)
 		}
 	}
 
 	for _, item := range ct.flows.Deleted() {
-		cleanupFlowEntries(item.Value().Get())
+		flow := item.Value().Get()
+		klog.V(1).Infof("cleaning conntrack entries for delete flow %v", flow)
+		cleanupFlowEntries(flow)
 	}
 }

client/plugins/conntrack/conntrack_test.go

@@ -0,0 +1,116 @@
+package conntrack
+
+import (
+	"context"
+	"flag"
+	"fmt"
+
+	"k8s.io/klog/v2"
+	"k8s.io/utils/exec"
+	exectesting "k8s.io/utils/exec/testing"
+
+	api "sigs.k8s.io/kpng/api/localnetv1"
+	"sigs.k8s.io/kpng/client/localsink/fullstate"
+)
+
+func ExampleConntrack() {
+	// setup
+	klog.InitFlags(nil)
+	flag.Set("v", "4")
+	execer = printCmdsExecer{}
+
+	ct := New()
+
+	// initial state
+	state := []*fullstate.ServiceEndpoints{
+		{
+			Service: &api.Service{
+				Namespace: "test-ns",
+				Name:      "test-svc",
+				Type:      "ClusterIP",
+				IPs: &api.ServiceIPs{
+					ClusterIPs: api.NewIPSet("10.1.1.1"),
+				},
+				Ports: []*api.PortMapping{
+					{
+						Name:       "p1",
+						Protocol:   api.Protocol_TCP,
+						Port:       80,
+						TargetPort: 8080,
+					},
+					{
+						Name:       "p2",
+						Protocol:   api.Protocol_UDP,
+						Port:       53,
+						TargetPort: 5353,
+					},
+				},
+			},
+			Endpoints: []*api.Endpoint{
+				{
+					IPs: api.NewIPSet("10.1.2.1"),
+				},
+			},
+		},
+	}
+
+	fmt.Println("-- initial state --")
+	ct.Callback(arrayCh(state))
+
+	fmt.Println("-- add one endpoint --")
+	state[0].Endpoints = append(state[0].Endpoints, &api.Endpoint{IPs: api.NewIPSet("10.1.3.1")})
+	ct.Callback(arrayCh(state))
+
+	fmt.Println("-- remove one endpoint --")
+	state[0].Endpoints = state[0].Endpoints[:1]
+	ct.Callback(arrayCh(state))
+
+	fmt.Println("-- remove one service --")
+	state = state[:0]
+	ct.Callback(arrayCh(state))
+
+	// Output:
+	// -- initial state --
+	// /bin/conntrack [-D -p tcp --sport 80 --orig-dst 10.1.1.1]
+	// /bin/conntrack [-D -p udp --sport 53 --orig-dst 10.1.1.1]
+	// -- add one endpoint --
+	// -- remove one endpoint --
+	// /bin/conntrack [-D -p udp --dport 53 --dst-nat 10.1.3.1 --orig-dst 10.1.1.1]
+	// -- remove one service --
+	// /bin/conntrack [-D -p udp --dport 53 --dst-nat 10.1.2.1 --orig-dst 10.1.1.1]
+
+}
+
+func arrayCh[T any](ts []T) <-chan T {
+	ch := make(chan T, 1)
+	go func() {
+		for _, t := range ts {
+			ch <- t
+		}
+		close(ch)
+	}()
+	return ch
+}
+
+type printCmdsExecer struct{}
+
+var _ exec.Interface = printCmdsExecer{}
+
+func (e printCmdsExecer) Command(cmd string, args ...string) exec.Cmd {
+	fmt.Println(cmd, args)
+	return exectesting.InitFakeCmd(&exectesting.FakeCmd{
+		CombinedOutputScript: []exectesting.FakeAction{
+			func() ([]byte, []byte, error) {
+				return []byte{}, []byte{}, nil
+			},
+		},
+	}, cmd, args...)
+}
+
+func (e printCmdsExecer) CommandContext(ctx context.Context, cmd string, args ...string) exec.Cmd {
+	return e.Command(cmd, args...)
+}
+
+func (e printCmdsExecer) LookPath(file string) (string, error) {
+	return "/bin/" + file, nil
+}