conntrack

Documentation

Index

• func IsClearConntrackNeeded(proto v1.Protocol) bool
• type Conntrack
  • func New() Conntrack
  • func (ct Conntrack) Callback(ch <-chan *client.ServiceEndpoints)
• type Flow
  • func (f Flow) Key() string
• type IPPort
  • func (i IPPort) Key() string
• type IPPortLeaf
• type Leaf
• type Sink
  • func NewSink() *Sink
  • func (ps *Sink) DeleteEndpoint(namespace, serviceName, key string)
  • func (ps *Sink) DeleteService(namespace, name string)
  • func (ps *Sink) Reset()
  • func (ps *Sink) SetEndpoint(namespace, serviceName, key string, endpoint *localv1.Endpoint)
  • func (ps *Sink) SetService(svc *localv1.Service)
  • func (ps *Sink) Setup()
  • func (s *Sink) Sync()

Examples

• Conntrack

Functions

func IsClearConntrackNeeded

func IsClearConntrackNeeded(proto v1.Protocol) bool

IsClearConntrackNeeded returns true if protocol requires conntrack cleanup for the stale connections

Types

type Conntrack

type Conntrack struct {
	// contains filtered or unexported fields
}

Example

package main

import (
	"context"
	"flag"
	"fmt"

	"k8s.io/klog/v2"
	"k8s.io/utils/exec"
	exectesting "k8s.io/utils/exec/testing"

	api "sigs.k8s.io/kpng/api/localv1"
	"sigs.k8s.io/kpng/client/localsink/fullstate"
)

func main() {
	// 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))

}

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
}

Output:

-- initial state --
/bin/conntrack [-D -p tcp --dport 80 --orig-dst 10.1.1.1]
/bin/conntrack [-D -p udp --dport 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 New

func New() Conntrack

func (Conntrack) Callback

func (ct Conntrack) Callback(ch <-chan *client.ServiceEndpoints)

type Flow

type Flow struct {
	IPPort
	EndpointIP string
	TargetPort int32
}

func (Flow) Key

func (f Flow) Key() string

type IPPort

type IPPort struct {
	Protocol api.Protocol
	DnatIP   string
	Port     int32
}

func (IPPort) Key

func (i IPPort) Key() string

type IPPortLeaf

type IPPortLeaf = diffstore.AnyLeaf[IPPort]

type Leaf

type Leaf = diffstore.AnyLeaf[Flow]

type Sink

type Sink struct {
	localsink.Config
	// contains filtered or unexported fields
}

func NewSink

func NewSink() *Sink

func (*Sink) DeleteEndpoint

func (ps *Sink) DeleteEndpoint(namespace, serviceName, key string)

func (*Sink) DeleteService

func (ps *Sink) DeleteService(namespace, name string)

func (*Sink) Reset

func (ps *Sink) Reset()

func (*Sink) SetEndpoint

func (ps *Sink) SetEndpoint(namespace, serviceName, key string, endpoint *localv1.Endpoint)

func (*Sink) SetService

func (ps *Sink) SetService(svc *localv1.Service)

func (*Sink) Setup

func (ps *Sink) Setup()

func (*Sink) Sync

func (s *Sink) Sync()