feat(kuma-cp) support new format of Dataplane

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 Changes:
 
+* feature: support new format of the Dataplane including scraping metrics from Gateway Dataplane
+  [#578](https://github.com/Kong/kuma/pull/579)
 * feature: new Dataplane format
   [#578](https://github.com/Kong/kuma/pull/576)
 * feature: validate value of `protocol` tag on a Dataplane resource

api/mesh/v1alpha1/dataplane_helpers.go

@@ -128,6 +128,33 @@ func ParseOutboundInterface(text string) (OutboundInterface, error) {
 	}, nil
 }
 
+func (n *Dataplane_Networking) GetOutboundInterfaces() ([]OutboundInterface, error) {
+	if n == nil {
+		return nil, nil
+	}
+	ofaces := make([]OutboundInterface, len(n.Outbound))
+	for i, outbound := range n.Outbound {
+		if outbound.Interface != "" { // legacy format
+			oface, err := ParseOutboundInterface(outbound.Interface)
+			if err != nil {
+				return nil, err
+			}
+			ofaces[i] = oface
+		} else {
+			oface := OutboundInterface{
+				DataplanePort: outbound.Port,
+			}
+			if outbound.Address != "" {
+				oface.DataplaneIP = outbound.Address
+			} else {
+				oface.DataplaneIP = "127.0.0.1"
+			}
+			ofaces[i] = oface
+		}
+	}
+	return ofaces, nil
+}
+
 func ParsePort(text string) (uint32, error) {
 	port, err := strconv.ParseUint(text, 10, 32)
 	if err != nil {
@@ -147,15 +174,12 @@ func ParseIP(text string) (string, error) {
 }
 
 func (n *Dataplane_Networking) GetInboundInterface(service string) (*InboundInterface, error) {
-	for _, inbound := range n.Inbound {
+	for i, inbound := range n.Inbound {
 		if inbound.Tags[ServiceTag] != service {
 			continue
 		}
-		iface, err := ParseInboundInterface(inbound.Interface)
-		if err != nil {
-			return nil, err
-		}
-		return &iface, nil
+		iface, err := n.GetInboundInterfaceByIdx(i)
+		return &iface, err
 	}
 	return nil, errors.Errorf("Dataplane has no Inbound Interface for service %q", service)
 }
@@ -165,8 +189,8 @@ func (n *Dataplane_Networking) GetInboundInterfaces() ([]InboundInterface, error
 		return nil, nil
 	}
 	ifaces := make([]InboundInterface, len(n.Inbound))
-	for i, inbound := range n.Inbound {
-		iface, err := ParseInboundInterface(inbound.Interface)
+	for i, _ := range n.Inbound {
+		iface, err := n.GetInboundInterfaceByIdx(i)
 		if err != nil {
 			return nil, err
 		}
@@ -175,6 +199,31 @@ func (n *Dataplane_Networking) GetInboundInterfaces() ([]InboundInterface, error
 	return ifaces, nil
 }
 
+func (n *Dataplane_Networking) GetInboundInterfaceByIdx(idx int) (InboundInterface, error) {
+	if idx >= len(n.Inbound) {
+		return InboundInterface{}, errors.Errorf("there is no inbound for index %d. Dataplane has %d inbounds", idx, len(n.Inbound))
+	}
+	inbound := n.Inbound[idx]
+	if inbound.Interface != "" {
+		return ParseInboundInterface(inbound.Interface)
+	} else {
+		iface := InboundInterface{
+			DataplanePort: inbound.Port,
+		}
+		if inbound.Address != "" {
+			iface.DataplaneIP = inbound.Address
+		} else {
+			iface.DataplaneIP = n.Address
+		}
+		if inbound.ServicePort != 0 {
+			iface.WorkloadPort = inbound.ServicePort
+		} else {
+			iface.WorkloadPort = inbound.Port
+		}
+		return iface, nil
+	}
+}
+
 // Matches is simply an alias for MatchTags to make source code more aesthetic.
 func (d *Dataplane) Matches(selector TagSelector) bool {
 	if d != nil {

api/mesh/v1alpha1/dataplane_helpers_test.go

@@ -319,6 +319,89 @@ var _ = Describe("ParseOutboundInterface(..)", func() {
 
 var _ = Describe("Dataplane_Networking", func() {
 
+	Describe("GetOutboundInterfaces()", func() {
+		Context("valid input values", func() {
+			type testCase struct {
+				input    *Dataplane_Networking
+				expected []OutboundInterface
+			}
+
+			DescribeTable("should parse valid input values",
+				func(given testCase) {
+					// when
+					ofaces, err := given.input.GetOutboundInterfaces()
+					// then
+					Expect(err).ToNot(HaveOccurred())
+					// and
+					Expect(ofaces).To(Equal(given.expected))
+				},
+				Entry("nil", testCase{
+					input:    nil,
+					expected: nil,
+				}),
+				Entry("empty", testCase{
+					input:    &Dataplane_Networking{},
+					expected: []OutboundInterface{},
+				}),
+				Entry("legacy - 2 outbound interfaces", testCase{
+					input: &Dataplane_Networking{
+						Outbound: []*Dataplane_Networking_Outbound{
+							{Interface: ":8080"},
+							{Interface: "192.168.0.1:443"},
+						},
+					},
+					expected: []OutboundInterface{
+						{DataplaneIP: "127.0.0.1", DataplanePort: 8080},
+						{DataplaneIP: "192.168.0.1", DataplanePort: 443},
+					},
+				}),
+				Entry("2 outbound interfaces", testCase{
+					input: &Dataplane_Networking{
+						Outbound: []*Dataplane_Networking_Outbound{
+							{
+								Port: 8080,
+							},
+							{
+								Address: "192.168.0.1",
+								Port:    443,
+							},
+						},
+					},
+					expected: []OutboundInterface{
+						{DataplaneIP: "127.0.0.1", DataplanePort: 8080},
+						{DataplaneIP: "192.168.0.1", DataplanePort: 443},
+					},
+				}),
+			)
+		})
+
+		Context("invalid input values", func() {
+			type testCase struct {
+				input       *Dataplane_Networking
+				expectedErr gomega_types.GomegaMatcher
+			}
+
+			DescribeTable("should fail on invalid input values",
+				func(given testCase) {
+					// when
+					ifaces, err := given.input.GetOutboundInterfaces()
+					// then
+					Expect(ifaces).To(BeNil())
+					// and
+					Expect(err.Error()).To(given.expectedErr)
+				},
+				Entry("dataplane IP address is missing", testCase{
+					input: &Dataplane_Networking{
+						Outbound: []*Dataplane_Networking_Outbound{
+							{Interface: ":443:8443"},
+						},
+					},
+					expectedErr: Equal(`invalid format: expected "[ IPv4 | '[' IPv6 ']' ] ':' DATAPLANE_PORT", got ":443:8443"`),
+				}),
+			)
+		})
+	})
+
 	Describe("GetInboundInterfaces()", func() {
 
 		Context("valid input values", func() {
@@ -344,7 +427,7 @@ var _ = Describe("Dataplane_Networking", func() {
 					input:    &Dataplane_Networking{},
 					expected: []InboundInterface{},
 				}),
-				Entry("2 inbound interfaces", testCase{
+				Entry("legacy - 2 inbound interfaces", testCase{
 					input: &Dataplane_Networking{
 						Inbound: []*Dataplane_Networking_Inbound{
 							{Interface: "192.168.0.1:80:8080"},
@@ -356,6 +439,25 @@ var _ = Describe("Dataplane_Networking", func() {
 						{DataplaneIP: "192.168.0.1", DataplanePort: 443, WorkloadPort: 8443},
 					},
 				}),
+				Entry("2 inbound interfaces", testCase{
+					input: &Dataplane_Networking{
+						Address: "192.168.0.1",
+						Inbound: []*Dataplane_Networking_Inbound{
+							{
+								Port: 80,
+							},
+							{
+								Address:     "192.168.0.2",
+								Port:        443,
+								ServicePort: 8443,
+							},
+						},
+					},
+					expected: []InboundInterface{
+						{DataplaneIP: "192.168.0.1", DataplanePort: 80, WorkloadPort: 80},
+						{DataplaneIP: "192.168.0.2", DataplanePort: 443, WorkloadPort: 8443},
+					},
+				}),
 			)
 		})
 

pkg/core/permissions/matcher.go

@@ -12,9 +12,9 @@ type TrafficPermissionsMatcher struct {
 	ResourceManager manager.ResourceManager
 }
 
-type MatchedPermissions map[string]*mesh_core.TrafficPermissionResourceList
+type MatchedPermissions map[mesh_proto.InboundInterface]*mesh_core.TrafficPermissionResourceList
 
-func (m MatchedPermissions) Get(inbound string) *mesh_core.TrafficPermissionResourceList {
+func (m MatchedPermissions) Get(inbound mesh_proto.InboundInterface) *mesh_core.TrafficPermissionResourceList {
 	matched, ok := m[inbound]
 	if ok {
 		return matched
@@ -28,24 +28,27 @@ func (m *TrafficPermissionsMatcher) Match(ctx context.Context, dataplane *mesh_c
 	if err := m.ResourceManager.List(ctx, permissions, store.ListByMesh(dataplane.GetMeta().GetMesh())); err != nil {
 		return nil, err
 	}
-	return MatchDataplaneTrafficPermissions(&dataplane.Spec, permissions), nil
+	return MatchDataplaneTrafficPermissions(&dataplane.Spec, permissions)
 }
 
-func MatchDataplaneTrafficPermissions(dataplane *mesh_proto.Dataplane, permissions *mesh_core.TrafficPermissionResourceList) MatchedPermissions {
+func MatchDataplaneTrafficPermissions(dataplane *mesh_proto.Dataplane, permissions *mesh_core.TrafficPermissionResourceList) (MatchedPermissions, error) {
 	matchedPermissions := make(MatchedPermissions)
-	for _, inbound := range dataplane.GetNetworking().GetInbound() {
-		matchedPermissions[inbound.Interface] = &mesh_core.TrafficPermissionResourceList{
+	ifaces, err := dataplane.GetNetworking().GetInboundInterfaces()
+	if err != nil {
+		return nil, err
+	}
+	for i, inbound := range dataplane.GetNetworking().GetInbound() {
+		matchedPermissions[ifaces[i]] = &mesh_core.TrafficPermissionResourceList{
 			Items: matchInbound(inbound, permissions),
 		}
 	}
-	return matchedPermissions
+	return matchedPermissions, nil
 }
 
 func matchInbound(inbound *mesh_proto.Dataplane_Networking_Inbound, trafficPermissions *mesh_core.TrafficPermissionResourceList) []*mesh_core.TrafficPermissionResource {
 	matchedPerms := []*mesh_core.TrafficPermissionResource{}
 	for _, perm := range trafficPermissions.Items {
 		if len(perm.Spec.Sources) == 0 {
-			// todo(jakubdyszkiewicz) there shouldn't be any rule with 0 sources. Move to validation logic in a manager
 			continue
 		}
 		for _, dest := range perm.Spec.Destinations {

pkg/core/permissions/matcher_test.go

@@ -145,10 +145,15 @@ var _ = Describe("Matcher", func() {
 		}
 
 		// when
-		matchedPerms := MatchDataplaneTrafficPermissions(&dataplane, &permissions)
+		matchedPerms, err := MatchDataplaneTrafficPermissions(&dataplane, &permissions)
 
 		// then
-		backendMatches := matchedPerms.Get("192.168.0.1:8080:80")
+		Expect(err).ToNot(HaveOccurred())
+		backendMatches := matchedPerms.Get(mesh_proto.InboundInterface{
+			DataplaneIP:   "192.168.0.1",
+			DataplanePort: 8080,
+			WorkloadPort:  80,
+		})
 		expectedBackendMatches := core_mesh.TrafficPermissionResourceList{
 			Items: []*core_mesh.TrafficPermissionResource{
 				{
@@ -204,7 +209,11 @@ var _ = Describe("Matcher", func() {
 				},
 			},
 		}
-		webMatches := matchedPerms.Get("192.168.0.1:8090:90")
+		webMatches := matchedPerms.Get(mesh_proto.InboundInterface{
+			DataplaneIP:   "192.168.0.1",
+			DataplanePort: 8090,
+			WorkloadPort:  90,
+		})
 		Expect(*webMatches).To(Equal(expectedWebMatches))
 	})
 })

pkg/core/resources/apis/mesh/dataplane_helpers.go

@@ -56,11 +56,11 @@ func (d *DataplaneResource) UsesInboundInterface(address net.IP, port uint32) bo
 	if d == nil {
 		return false
 	}
-	for _, inbound := range d.Spec.Networking.GetInbound() {
-		iface, err := mesh_proto.ParseInboundInterface(inbound.Interface)
-		if err != nil {
-			continue
-		}
+	ifaces, err := d.Spec.Networking.GetInboundInterfaces()
+	if err != nil {
+		return false
+	}
+	for _, iface := range ifaces {
 		// compare against port and IP address of the dataplane
 		if port == iface.DataplanePort && overlap(address, net.ParseIP(iface.DataplaneIP)) {
 			return true
@@ -77,11 +77,11 @@ func (d *DataplaneResource) UsesOutboundInterface(address net.IP, port uint32) b
 	if d == nil {
 		return false
 	}
-	for _, outbound := range d.Spec.Networking.GetOutbound() {
-		oface, err := mesh_proto.ParseOutboundInterface(outbound.Interface)
-		if err != nil {
-			continue
-		}
+	ofaces, err := d.Spec.Networking.GetOutboundInterfaces()
+	if err != nil {
+		return false
+	}
+	for _, oface := range ofaces {
 		// compare against port and IP address of the dataplane
 		if port == oface.DataplanePort && overlap(address, net.ParseIP(oface.DataplaneIP)) {
 			return true
@@ -113,12 +113,16 @@ func (d *DataplaneResource) GetIP() string {
 	if d == nil {
 		return ""
 	}
-	ifaces, err := d.Spec.Networking.GetInboundInterfaces()
-	if err != nil {
+	if d.Spec.Networking.Address != "" {
+		return d.Spec.Networking.Address
+	}
+	// fallback to legacy format
+	if len(d.Spec.Networking.Inbound) == 0 {
 		return ""
 	}
-	if len(ifaces) == 0 {
+	iface, err := d.Spec.Networking.GetInboundInterfaceByIdx(0)
+	if err != nil {
 		return ""
 	}
-	return ifaces[0].DataplaneIP
+	return iface.DataplaneIP
 }