[CORE-10975] Add QoS packet rate limit (#9938)

cni-plugin/Makefile

@@ -145,7 +145,7 @@ $(DEPLOY_CONTAINER_FIPS_MARKER): Dockerfile build build-cni-bins
 	touch $@
 
 # These are the files that we need to copy from the containernetworking-plugins project to our image.
-CN_FILES := host-local portmap loopback tuning bandwidth
+CN_FILES := host-local portmap loopback tuning
 
 CONTAINERNETWORKING_PLUGINS_CLONED=.containernetworking-plugins-$(CNI_VERSION).cloned
 
@@ -157,7 +157,7 @@ $(CONTAINERNETWORKING_PLUGINS_CLONED):
 
 CN_FLAGS=-ldflags "-X github.com/containernetworking/plugins/pkg/utils/buildversion.BuildVersion=$(GIT_VERSION)"
 
-$(BIN)/host-local $(BIN)/loopback $(BIN)/portmap $(BIN)/tuning $(BIN)/bandwidth &: $(CONTAINERNETWORKING_PLUGINS_CLONED)
+$(BIN)/host-local $(BIN)/loopback $(BIN)/portmap $(BIN)/tuning  &: $(CONTAINERNETWORKING_PLUGINS_CLONED)
 	docker run \
 		$(EXTRA_DOCKER_ARGS) \
 		-v $(CURDIR)/containernetworking-plugins:/go/src/github.com/containernetworking/plugins:z \
@@ -195,7 +195,7 @@ $(WINDOWS_BIN)/flannel.exe: $(FLANNEL_CNI_PLUGIN_CLONED)
 	cp flannel-cni-plugin/dist/flannel-$(ARCH).exe $(WINDOWS_BIN)/flannel.exe
 
 .PHONY: build-cni-bins build-win-cni-bins
-build-cni-bins: $(BIN)/flannel $(BIN)/loopback $(BIN)/host-local $(BIN)/portmap $(BIN)/tuning $(BIN)/bandwidth
+build-cni-bins: $(BIN)/flannel $(BIN)/loopback $(BIN)/host-local $(BIN)/portmap $(BIN)/tuning
 build-win-cni-bins: $(WINDOWS_BIN)/flannel.exe
 
 ###############################################################################

felix/bpf/tc/attach.go

@@ -31,6 +31,7 @@ import (
 	"github.com/projectcalico/calico/felix/bpf/hook"
 	"github.com/projectcalico/calico/felix/bpf/libbpf"
 	tcdefs "github.com/projectcalico/calico/felix/bpf/tc/defs"
+	"github.com/projectcalico/calico/felix/dataplane/linux/qos"
 )
 
 type AttachPoint struct {
@@ -318,7 +319,25 @@ func EnsureQdisc(ifaceName string) (bool, error) {
 		log.WithField("iface", ifaceName).Debug("Already have a clsact qdisc on this interface")
 		return true, nil
 	}
-	return false, libbpf.CreateQDisc(ifaceName)
+
+	// Clean up QoS config as it is currently not suppored by the BPF dataplane
+	// and should be removed when transitioning from iptables or nftables to BPF.
+	var errs []error
+	err = qos.RemoveIngressQdisc(ifaceName)
+	if err != nil {
+		errs = append(errs, fmt.Errorf("error removing QoS ingress qdisc from interface %s: %w", ifaceName, err))
+	}
+	err = qos.RemoveEgressQdisc(ifaceName)
+	if err != nil {
+		errs = append(errs, fmt.Errorf("error removing QoS egress qdisc from interface %s: %w", ifaceName, err))
+	}
+
+	err = libbpf.CreateQDisc(ifaceName)
+	if err != nil {
+		errs = append(errs, fmt.Errorf("error creating qdisc on interface %s: %w", ifaceName, err))
+	}
+
+	return false, errors.Join(errs...)
 }
 
 func HasQdisc(ifaceName string) (bool, error) {

felix/dataplane/driver.go

@@ -111,7 +111,7 @@ func StartDataplaneDriver(
 		// avoid allocating the others to minimize the number of bits in use.
 
 		// The accept bit is a long-lived bit used to communicate between chains.
-		var markAccept, markPass, markScratch0, markScratch1, markWireguard, markEndpointNonCaliEndpoint uint32
+		var markAccept, markPass, markScratch0, markScratch1, markWireguard, markLimitPacketRate, markEndpointNonCaliEndpoint uint32
 		markAccept, _ = markBitsManager.NextSingleBitMark()
 
 		// The pass bit is used to communicate from a policy chain up to the endpoint chain.
@@ -123,6 +123,8 @@ func StartDataplaneDriver(
 		markScratch0, _ = markBitsManager.NextSingleBitMark()
 		markScratch1, _ = markBitsManager.NextSingleBitMark()
 
+		markLimitPacketRate, _ = markBitsManager.NextSingleBitMark()
+
 		if configParams.WireguardEnabled || configParams.WireguardEnabledV6 {
 			log.Info("Wireguard enabled, allocating a mark bit")
 			markWireguard, _ = markBitsManager.NextSingleBitMark()
@@ -134,7 +136,7 @@ func StartDataplaneDriver(
 			}
 		}
 
-		if markAccept == 0 || markScratch0 == 0 || markPass == 0 || markScratch1 == 0 {
+		if markAccept == 0 || markScratch0 == 0 || markPass == 0 || markScratch1 == 0 || markLimitPacketRate == 0 {
 			log.WithFields(log.Fields{
 				"Name":     "felix-iptables",
 				"MarkMask": allowedMarkBits,
@@ -159,6 +161,7 @@ func StartDataplaneDriver(
 			"dropMark":            markDrop,
 			"scratch0Mark":        markScratch0,
 			"scratch1Mark":        markScratch1,
+			"limitPacketRateMark": markLimitPacketRate,
 			"endpointMark":        markEndpointMark,
 			"endpointMarkNonCali": markEndpointNonCaliEndpoint,
 		}).Info("Calculated iptables mark bits")
@@ -247,6 +250,7 @@ func StartDataplaneDriver(
 				MarkDrop:            markDrop,
 				MarkScratch0:        markScratch0,
 				MarkScratch1:        markScratch1,
+				MarkLimitPacketRate: markLimitPacketRate,
 				MarkEndpoint:        markEndpointMark,
 				MarkNonCaliEndpoint: markEndpointNonCaliEndpoint,
 

felix/dataplane/linux/endpoint_mgr.go

@@ -669,6 +669,16 @@ func (m *endpointManager) resolveWorkloadEndpoints() {
 	}
 
 	removeActiveWorkload := func(logCxt *log.Entry, oldWorkload *proto.WorkloadEndpoint, id types.WorkloadEndpointID) {
+		if !m.bpfEnabled {
+			// QoS state should be removed before the workload itself is removed
+			if oldWorkload != nil {
+				logCxt.Info("Deleting QoS bandwidth state if present")
+				err := m.maybeUpdateQoSBandwidth(oldWorkload, nil)
+				if err != nil {
+					logCxt.WithError(err).WithField("workload", oldWorkload).Debug("Error deleting QoS bandwidth state, workload may have been already removed.")
+				}
+			}
+		}
 		m.callbacks.InvokeRemoveWorkload(oldWorkload)
 		m.filterTable.RemoveChains(m.activeWlIDToChains[id])
 		delete(m.activeWlIDToChains, id)
@@ -807,6 +817,14 @@ func (m *endpointManager) resolveWorkloadEndpoints() {
 				m.activeWlIfaceNameToID[workload.Name] = id
 				delete(m.pendingWlEpUpdates, id)
 
+				if !m.bpfEnabled {
+					logCxt.Info("Updating QoS bandwidth state if changed")
+					err := m.maybeUpdateQoSBandwidth(oldWorkload, workload)
+					if err != nil {
+						logCxt.WithError(err).WithFields(log.Fields{"oldWorkload": oldWorkload, "newWorkload": workload}).Debug("Error updating QoS bandwidth state")
+					}
+				}
+
 				m.callbacks.InvokeUpdateWorkload(oldWorkload, workload)
 			} else {
 				logCxt.Info("Workload removed, deleting its chains.")
@@ -885,6 +903,7 @@ func (m *endpointManager) updateWorkloadEndpointChains(
 		adminUp,
 		tierGroups,
 		workload.ProfileIds,
+		workload.QosControls,
 	)
 	m.filterTable.UpdateChains(chains)
 	m.activeWlIDToChains[id] = chains

felix/dataplane/linux/endpoint_mgr_test.go

@@ -86,7 +86,7 @@ var wlDispatchEmpty = []*generictables.Chain{
 			},
 			{
 				Match:   iptables.Match(),
-				Action:  iptables.SetMaskedMarkAction{Mark: 0x0100, Mask: 0xff00},
+				Action:  iptables.SetMaskedMarkAction{Mark: 0x0200, Mask: 0xfe00},
 				Comment: []string{"Non-Cali endpoint mark"},
 			},
 		},
@@ -628,7 +628,7 @@ func chainsForIfaces(ipVersion uint8,
 			},
 			generictables.Rule{
 				Match:   iptables.Match(),
-				Action:  iptables.SetMaskedMarkAction{Mark: 0x0100, Mask: 0xff00},
+				Action:  iptables.SetMaskedMarkAction{Mark: 0x0200, Mask: 0xfe00},
 				Comment: []string{"Non-Cali endpoint mark"},
 			},
 		)
@@ -794,8 +794,9 @@ func endpointManagerTests(ipVersion uint8, flowlogs bool) func() {
 				MarkScratch0:           0x20,
 				MarkScratch1:           0x40,
 				MarkDrop:               0x80,
-				MarkEndpoint:           0xff00,
-				MarkNonCaliEndpoint:    0x0100,
+				MarkLimitPacketRate:    0x100,
+				MarkEndpoint:           0xfe00,
+				MarkNonCaliEndpoint:    0x0200,
 				KubeIPVSSupportEnabled: true,
 				WorkloadIfacePrefixes:  []string{"cali", "tap"},
 				VXLANPort:              4789,