kube-ovn/pkg/daemon/flow_rules_linux.go at 9550667899be04b3ba5498408d2e63d799b6e047 · kubeovn/kube-ovn · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
package daemon

import (
	"fmt"
	"strconv"
	"strings"

	"k8s.io/klog/v2"

	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
	"github.com/kubeovn/kube-ovn/pkg/ovs"
	"github.com/kubeovn/kube-ovn/pkg/util"
)

const flowKindUnderlayService = "usvc"

func (c *Controller) AddOrUpdateUnderlaySubnetSvcLocalFlowCache(serviceIP string, port uint16, protocol, dstMac, underlayNic, bridgeName, subnetName string) error {
	inPort, err := c.getPortID(underlayNic)
	if err != nil {
		return err
	}

	patchPortName := fmt.Sprintf("patch-localnet.%s-to-br-int", subnetName)
	outPort, err := c.getPortID(patchPortName)
	if err != nil {
		klog.V(5).Infof("patch-localnet port %s not found on bridge %s, skipping underlay service flow for %s:%d (subnet %s may not have pods on this node yet)", patchPortName, bridgeName, serviceIP, port, subnetName)
		return nil
	}

	isIPv6 := util.CheckProtocol(serviceIP) == kubeovnv1.ProtocolIPv6
	protoStr := ""
	switch strings.ToUpper(protocol) {
	case "TCP":
		protoStr = "tcp"
		if isIPv6 {
			protoStr = "tcp6"
		}
	case "UDP":
		protoStr = "udp"
		if isIPv6 {
			protoStr = "udp6"
		}
	default:
		return fmt.Errorf("unsupported protocol %s", protocol)
	}

	cookie := fmt.Sprintf("0x%x", util.UnderlaySvcLocalOpenFlowCookieV4)
	nwDst := "nw_dst"
	if isIPv6 {
		cookie = fmt.Sprintf("0x%x", util.UnderlaySvcLocalOpenFlowCookieV6)
		nwDst = "ipv6_dst"
	}

	flow := fmt.Sprintf("cookie=%s,priority=%d,in_port=%d,%s,%s=%s,tp_dst=%d "+
		"actions=mod_dl_dst:%s,output:%d",
		cookie, util.UnderlaySvcLocalOpenFlowPriority, inPort, protoStr, nwDst, serviceIP, port, dstMac, outPort)

	key := buildFlowKey(flowKindUnderlayService, serviceIP, port, protocol, "")
	c.setFlowCache(c.flowCache, bridgeName, key, []string{flow})

	klog.V(5).Infof("updated underlay flow cache for service %s", key)
	c.requestFlowSync()
	return nil
}

func (c *Controller) deleteUnderlaySubnetSvcLocalFlowCache(bridgeName, serviceIP string, port uint16, protocol string) {
	key := buildFlowKey(flowKindUnderlayService, serviceIP, port, protocol, "")

	c.deleteFlowCache(c.flowCache, bridgeName, key)

	klog.V(5).Infof("deleted underlay flow cache for service %s", key)
	c.requestFlowSync()
}

func buildFlowKey(kind, ip string, port uint16, protocol, extra string) string {
	if extra == "" {
		return fmt.Sprintf("%s-%s-%s-%d", kind, ip, protocol, port)
	}
	return fmt.Sprintf("%s-%s-%s-%d-%s", kind, ip, protocol, port, extra)
}

func (c *Controller) getPortID(portName string) (int, error) {
	ofportStr, err := ovs.Get("Interface", portName, "ofport", "", true)
	if err != nil {
		return 0, fmt.Errorf("failed to get ofport for interface %s: %w", portName, err)
	}

	portID, err := strconv.Atoi(strings.TrimSpace(ofportStr))
	if err != nil {
		return 0, fmt.Errorf("failed to parse ofport %q: %w", ofportStr, err)
	}

	return portID, nil
}