Set DSCP for host endpoints

felix/calc/event_sequencer.go

@@ -468,6 +468,12 @@ func ModelWorkloadEndpointToProto(ep *model.WorkloadEndpoint, peerData *Endpoint
 }
 
 func ModelHostEndpointToProto(ep *model.HostEndpoint, tiers, untrackedTiers, preDNATTiers []*proto.TierInfo, forwardTiers []*proto.TierInfo) *proto.HostEndpoint {
+	var qosPolicies []*proto.QoSPolicy
+	if ep.QoSControls != nil && ep.QoSControls.DSCP != nil {
+		qosPolicies = append(qosPolicies, &proto.QoSPolicy{
+			Dscp: int32(ep.QoSControls.DSCP.ToUint8()),
+		})
+	}
 	return &proto.HostEndpoint{
 		Name:              ep.Name,
 		ExpectedIpv4Addrs: ipsToStrings(ep.ExpectedIPv4Addrs),
@@ -477,6 +483,7 @@ func ModelHostEndpointToProto(ep *model.HostEndpoint, tiers, untrackedTiers, pre
 		UntrackedTiers:    untrackedTiers,
 		PreDnatTiers:      preDNATTiers,
 		ForwardTiers:      forwardTiers,
+		QosPolicies:       qosPolicies,
 	}
 }
 

felix/calc/event_sequencer_test.go

@@ -278,6 +278,37 @@ var _ = DescribeTable("ModelHostEndpointToProto",
 			ProfileIds:        []string{"prof1"},
 		},
 	),
+	Entry("host endpoint with QoSControls",
+		model.HostEndpoint{
+			Name:              "eth0",
+			ExpectedIPv4Addrs: []net.IP{mustParseIP("10.28.0.13"), mustParseIP("10.28.0.14")},
+			ExpectedIPv6Addrs: []net.IP{mustParseIP("dead::beef"), mustParseIP("dead::bee5")},
+			Labels: uniquelabels.Make(map[string]string{
+				"a": "b",
+			}),
+			ProfileIDs: []string{"prof1"},
+			QoSControls: &model.QoSControls{
+				DSCP: &dscp,
+			},
+		},
+		[]*proto.TierInfo{{Name: "a", IngressPolicies: []string{"b"}}},
+		[]*proto.TierInfo{{Name: "a", EgressPolicies: []string{"c"}}},
+		[]*proto.TierInfo{{Name: "a", EgressPolicies: []string{"d"}}},
+		&proto.HostEndpoint{
+			Name:              "eth0",
+			ExpectedIpv4Addrs: []string{"10.28.0.13", "10.28.0.14"},
+			ExpectedIpv6Addrs: []string{"dead::beef", "dead::bee5"},
+			Tiers:             []*proto.TierInfo{{Name: "a", IngressPolicies: []string{"b"}}},
+			UntrackedTiers:    []*proto.TierInfo{{Name: "a", EgressPolicies: []string{"c"}}},
+			ForwardTiers:      []*proto.TierInfo{{Name: "a", EgressPolicies: []string{"d"}}},
+			ProfileIds:        []string{"prof1"},
+			QosPolicies: []*proto.QoSPolicy{
+				&proto.QoSPolicy{
+					Dscp: 38,
+				},
+			},
+		},
+	),
 )
 
 var _ = Describe("ServiceAccount update/remove", func() {

felix/dataplane/linux/qos_policy_mgr.go

@@ -28,11 +28,12 @@ import (
 type qosPolicyManager struct {
 	ipVersion    uint8
 	ruleRenderer rules.RuleRenderer
+	mangleTable  Table
 
-	// QoS policy
-	mangleTable Table
+	// QoS policies.
+	wepPolicies map[types.WorkloadEndpointID]rules.QoSPolicy
+	hepPolicies map[types.HostEndpointID]rules.QoSPolicy
 	dirty       bool
-	policies    map[types.WorkloadEndpointID]rules.QoSPolicy
 
 	logCxt *logrus.Entry
 }
@@ -43,30 +44,66 @@ func newQoSPolicyManager(
 	ipVersion uint8,
 ) *qosPolicyManager {
 	return &qosPolicyManager{
-		ipVersion:    ipVersion,
-		policies:     map[types.WorkloadEndpointID]rules.QoSPolicy{},
-		dirty:        true,
 		mangleTable:  mangleTable,
 		ruleRenderer: ruleRenderer,
+		ipVersion:    ipVersion,
+		wepPolicies:  map[types.WorkloadEndpointID]rules.QoSPolicy{},
+		hepPolicies:  map[types.HostEndpointID]rules.QoSPolicy{},
+		dirty:        true,
 		logCxt:       logrus.WithField("ipVersion", ipVersion),
 	}
 }
 
 func (m *qosPolicyManager) OnUpdate(msg interface{}) {
 	switch msg := msg.(type) {
+	case *proto.HostEndpointUpdate:
+		m.handleHEPUpdates(msg.GetId(), msg)
+	case *proto.HostEndpointRemove:
+		m.handleHEPUpdates(msg.GetId(), nil)
 	case *proto.WorkloadEndpointUpdate:
-		m.handleWlEndpointUpdates(msg.GetId(), msg)
+		m.handleWEPUpdates(msg.GetId(), msg)
 	case *proto.WorkloadEndpointRemove:
-		m.handleWlEndpointUpdates(msg.GetId(), nil)
+		m.handleWEPUpdates(msg.GetId(), nil)
+	}
+}
+
+func (m *qosPolicyManager) handleHEPUpdates(hepID *proto.HostEndpointID, msg *proto.HostEndpointUpdate) {
+	id := types.ProtoToHostEndpointID(hepID)
+	if msg == nil || len(msg.Endpoint.QosPolicies) == 0 {
+		_, exists := m.hepPolicies[id]
+		if exists {
+			delete(m.hepPolicies, id)
+			m.dirty = true
+		}
+		return
+	}
+
+	// We only support one policy per endpoint at this point.
+	dscp := msg.Endpoint.QosPolicies[0].Dscp
+
+	// This situation must be handled earlier.
+	if dscp > 63 || dscp < 0 {
+		logrus.WithField("id", id).Panicf("Invalid DSCP value %v", dscp)
+	}
+	ips := msg.Endpoint.ExpectedIpv4Addrs
+	if m.ipVersion == 6 {
+		ips = msg.Endpoint.ExpectedIpv6Addrs
+	}
+	if len(ips) != 0 {
+		m.hepPolicies[id] = rules.QoSPolicy{
+			SrcAddrs: normaliseSourceAddr(ips),
+			DSCP:     uint8(dscp),
+		}
+		m.dirty = true
 	}
 }
 
-func (m *qosPolicyManager) handleWlEndpointUpdates(wlID *proto.WorkloadEndpointID, msg *proto.WorkloadEndpointUpdate) {
-	id := types.ProtoToWorkloadEndpointID(wlID)
+func (m *qosPolicyManager) handleWEPUpdates(wepID *proto.WorkloadEndpointID, msg *proto.WorkloadEndpointUpdate) {
+	id := types.ProtoToWorkloadEndpointID(wepID)
 	if msg == nil || len(msg.Endpoint.QosPolicies) == 0 {
-		_, exists := m.policies[id]
+		_, exists := m.wepPolicies[id]
 		if exists {
-			delete(m.policies, id)
+			delete(m.wepPolicies, id)
 			m.dirty = true
 		}
 		return
@@ -84,7 +121,7 @@ func (m *qosPolicyManager) handleWlEndpointUpdates(wlID *proto.WorkloadEndpointI
 		ips = msg.Endpoint.Ipv6Nets
 	}
 	if len(ips) != 0 {
-		m.policies[id] = rules.QoSPolicy{
+		m.wepPolicies[id] = rules.QoSPolicy{
 			SrcAddrs: normaliseSourceAddr(ips),
 			DSCP:     uint8(dscp),
 		}
@@ -102,9 +139,12 @@ func normaliseSourceAddr(addrs []string) string {
 }
 
 func (m *qosPolicyManager) CompleteDeferredWork() error {
+	var policies []rules.QoSPolicy
 	if m.dirty {
-		var policies []rules.QoSPolicy
-		for _, p := range m.policies {
+		for _, p := range m.wepPolicies {
+			policies = append(policies, p)
+		}
+		for _, p := range m.hepPolicies {
 			policies = append(policies, p)
 		}
 		sort.Slice(policies, func(i, j int) bool {

felix/dataplane/linux/qos_policy_mgr_test.go

@@ -57,8 +57,8 @@ func qosPolicyManagerTests(ipVersion uint8) func() {
 			}}})
 		})
 
-		It("should handle workload updates correctly", func() {
-			By("sending workload endpoint updates with DSCP annotion")
+		It("should handle endpoint updates correctly", func() {
+			By("sending first workload endpoint update with DSCP annotation")
 			endpoint1 := &proto.WorkloadEndpoint{
 				State:       "active",
 				Name:        "cali12345-ab",
@@ -84,7 +84,7 @@ func qosPolicyManagerTests(ipVersion uint8) func() {
 				},
 			}}})
 
-			By("sending another workload endpoint updates with DSCP annotion")
+			By("sending another workload endpoint update with DSCP annotation")
 			endpoint2 := &proto.WorkloadEndpoint{
 				State:       "active",
 				Name:        "cali2",
@@ -115,7 +115,7 @@ func qosPolicyManagerTests(ipVersion uint8) func() {
 				},
 			}}})
 
-			By("verifying update to DSCP value takes effect")
+			By("verifying update to first workload DSCP value")
 			endpoint1.QosPolicies = []*proto.QoSPolicy{{Dscp: 13}}
 			manager.OnUpdate(&proto.WorkloadEndpointUpdate{
 				Id:       &wlEPID1,
@@ -140,7 +140,73 @@ func qosPolicyManagerTests(ipVersion uint8) func() {
 				},
 			}}})
 
-			By("verifying QoS policy rules removed when annotation is removed")
+			By("sending a host endpoint update with DSCP annotation")
+			hep1ID := &proto.HostEndpointID{
+				EndpointId: "id1",
+			}
+			hep1 := &proto.HostEndpoint{
+				Name:              "eth0",
+				ExpectedIpv4Addrs: []string{"192.168.1.2", "192.168.2.2"},
+				ExpectedIpv6Addrs: []string{"2001:db9:10::2", "dead:beff::20:2"},
+				QosPolicies:       []*proto.QoSPolicy{{Dscp: 44}},
+			}
+			manager.OnUpdate(&proto.HostEndpointUpdate{
+				Id:       hep1ID,
+				Endpoint: hep1,
+			})
+
+			err = manager.CompleteDeferredWork()
+			Expect(err).NotTo(HaveOccurred())
+
+			mangleTable.checkChains([][]*generictables.Chain{{{
+				Name: rules.ChainQoSPolicy,
+				Rules: []generictables.Rule{
+					// Rendered policies are sorted.
+					{
+						Action: iptables.DSCPAction{Value: 20},
+						Match:  iptables.Match().SourceNet(addrFromWlUpdate(endpoint2, ipVersion)),
+					},
+					{
+						Action: iptables.DSCPAction{Value: 13},
+						Match:  iptables.Match().SourceNet(addrFromWlUpdate(endpoint1, ipVersion)),
+					},
+					{
+						Action: iptables.DSCPAction{Value: 44},
+						Match:  iptables.Match().SourceNet(addrFromHepUpdate(hep1, ipVersion)),
+					},
+				},
+			}}})
+
+			By("verifying update to host endpoint DSCP value")
+			hep1.QosPolicies = []*proto.QoSPolicy{{Dscp: 30}}
+			manager.OnUpdate(&proto.HostEndpointUpdate{
+				Id:       hep1ID,
+				Endpoint: hep1,
+			})
+
+			err = manager.CompleteDeferredWork()
+			Expect(err).NotTo(HaveOccurred())
+
+			mangleTable.checkChains([][]*generictables.Chain{{{
+				Name: rules.ChainQoSPolicy,
+				Rules: []generictables.Rule{
+					// Rendered policies are sorted.
+					{
+						Action: iptables.DSCPAction{Value: 20},
+						Match:  iptables.Match().SourceNet(addrFromWlUpdate(endpoint2, ipVersion)),
+					},
+					{
+						Action: iptables.DSCPAction{Value: 13},
+						Match:  iptables.Match().SourceNet(addrFromWlUpdate(endpoint1, ipVersion)),
+					},
+					{
+						Action: iptables.DSCPAction{Value: 30},
+						Match:  iptables.Match().SourceNet(addrFromHepUpdate(hep1, ipVersion)),
+					},
+				},
+			}}})
+
+			By("verifying QoS policy rules removed when first workload annotation is removed")
 			endpoint1.QosPolicies = nil
 			manager.OnUpdate(&proto.WorkloadEndpointUpdate{
 				Id:       &wlEPID1,
@@ -157,17 +223,37 @@ func qosPolicyManagerTests(ipVersion uint8) func() {
 						Action: iptables.DSCPAction{Value: 20},
 						Match:  iptables.Match().SourceNet(addrFromWlUpdate(endpoint2, ipVersion)),
 					},
+					{
+						Action: iptables.DSCPAction{Value: 30},
+						Match:  iptables.Match().SourceNet(addrFromHepUpdate(hep1, ipVersion)),
+					},
 				},
 			}}})
 
-			By("verifying QoS policy rules removed when workload is removed")
+			By("verifying QoS policy rules removed when second workload is removed")
 			manager.OnUpdate(&proto.WorkloadEndpointRemove{
 				Id: &wlEPID2,
 			})
 
 			err = manager.CompleteDeferredWork()
 			Expect(err).NotTo(HaveOccurred())
+			mangleTable.checkChains([][]*generictables.Chain{{{
+				Name: rules.ChainQoSPolicy,
+				Rules: []generictables.Rule{
+					{
+						Action: iptables.DSCPAction{Value: 30},
+						Match:  iptables.Match().SourceNet(addrFromHepUpdate(hep1, ipVersion)),
+					},
+				},
+			}}})
+
+			By("verifying QoS policy rules removed when host endpoint is removed")
+			manager.OnUpdate(&proto.HostEndpointRemove{
+				Id: hep1ID,
+			})
 
+			err = manager.CompleteDeferredWork()
+			Expect(err).NotTo(HaveOccurred())
 			mangleTable.checkChains([][]*generictables.Chain{{{
 				Name:  rules.ChainQoSPolicy,
 				Rules: nil,
@@ -183,3 +269,11 @@ func addrFromWlUpdate(endpoint *proto.WorkloadEndpoint, ipVersion uint8) string
 	}
 	return normaliseSourceAddr(addr)
 }
+
+func addrFromHepUpdate(endpoint *proto.HostEndpoint, ipVersion uint8) string {
+	addr := endpoint.ExpectedIpv4Addrs
+	if ipVersion == 6 {
+		addr = endpoint.ExpectedIpv6Addrs
+	}
+	return normaliseSourceAddr(addr)
+}