feat/fix(endpoints) : Ensure endpoints for a serivce honors traffic p…

…olicy

This PR ensures that the endpoints built for a service in EDS honor
SMI traffic target policies. Only those destinations endpoints are
programmed on the envoy, if its destination pod has a service account
specified as a destination in any of of the applicable traffic targets.

Resolves issue openservicemesh#1658

Signed-off-by: Sneha Chhabria <snchh@microsoft.com>

pkg/catalog/endpoint.go

@@ -36,3 +36,17 @@ func (mc *MeshCatalog) GetResolvableServiceEndpoints(svc service.MeshService) ([
 	}
 	return endpoints, nil
 }
+
+// ListEndpointsForPodFromServiceAccount retrieves the list of IP addresses for the given service account
+func (mc *MeshCatalog) ListEndpointsForPodFromServiceAccount(sa service.K8sServiceAccount) []endpoint.Endpoint {
+	var endpoints []endpoint.Endpoint
+	for _, provider := range mc.endpointsProviders {
+		ep := provider.ListEndpointsForPodFromServiceAccount(sa)
+		if len(ep) == 0 {
+			log.Trace().Msgf("[%s] No endpoints found for service account=%s", provider.GetID(), sa)
+			continue
+		}
+		endpoints = append(endpoints, ep...)
+	}
+	return endpoints
+}

pkg/catalog/fake.go

@@ -104,7 +104,7 @@ func NewFakeMeshCatalog(kubeClient kubernetes.Interface) *MeshCatalog {
 	mockKubeController.EXPECT().IsMonitoredNamespace(tests.BookbuyerService.Namespace).Return(true).AnyTimes()
 	mockKubeController.EXPECT().IsMonitoredNamespace(tests.BookwarehouseService.Namespace).Return(true).AnyTimes()
 	mockKubeController.EXPECT().ListServiceAccountsForService(tests.BookstoreV1Service).Return([]service.K8sServiceAccount{tests.BookstoreServiceAccount}, nil).AnyTimes()
-	mockKubeController.EXPECT().ListServiceAccountsForService(tests.BookstoreV2Service).Return([]service.K8sServiceAccount{tests.BookstoreServiceAccount}, nil).AnyTimes()
+	mockKubeController.EXPECT().ListServiceAccountsForService(tests.BookstoreV2Service).Return([]service.K8sServiceAccount{tests.BookstoreV2ServiceAccount}, nil).AnyTimes()
 	mockKubeController.EXPECT().ListServiceAccountsForService(tests.BookbuyerService).Return([]service.K8sServiceAccount{tests.BookbuyerServiceAccount}, nil).AnyTimes()
 
 	return NewMeshCatalog(mockKubeController, kubeClient, meshSpec, certManager,

pkg/catalog/helpers_test.go

@@ -51,7 +51,7 @@ func newFakeMeshCatalogForRoutes(t *testing.T, testParams testParams) *MeshCatal
 	}
 
 	// Create a bookstoreV2 pod
-	bookstoreV2Pod := tests.NewPodFixture(tests.BookstoreV2Service.Namespace, tests.BookstoreV2Service.Name, tests.BookstoreServiceAccountName, tests.PodLabels)
+	bookstoreV2Pod := tests.NewPodFixture(tests.BookstoreV2Service.Namespace, tests.BookstoreV2Service.Name, tests.BookstoreV2ServiceAccountName, tests.PodLabels)
 	if _, err := kubeClient.CoreV1().Pods(tests.BookstoreV2Service.Namespace).Create(context.TODO(), &bookstoreV2Pod, metav1.CreateOptions{}); err != nil {
 		t.Fatalf("Error creating new pod: %s", err.Error())
 	}

pkg/catalog/ingress_test.go

@@ -70,8 +70,14 @@ func newFakeMeshCatalog() *MeshCatalog {
 
 	certManager := tresor.NewFakeCertManager(cfg)
 
-	// Create a pod
-	pod := tests.NewPodFixture(tests.Namespace, "pod-name", tests.BookstoreServiceAccountName, tests.PodLabels)
+	// Create a Bookstore-v1 pod
+	pod := tests.NewPodFixture(tests.Namespace, tests.BookstoreV1Service.Name, tests.BookstoreServiceAccountName, tests.PodLabels)
+	if _, err := kubeClient.CoreV1().Pods(tests.Namespace).Create(context.TODO(), &pod, metav1.CreateOptions{}); err != nil {
+		GinkgoT().Fatalf("Error creating new fake Mesh Catalog: %s", err.Error())
+	}
+
+	// Create a Bookstore-v2 pod
+	pod = tests.NewPodFixture(tests.Namespace, tests.BookstoreV2Service.Name, tests.BookstoreV2ServiceAccountName, tests.PodLabels)
 	if _, err := kubeClient.CoreV1().Pods(tests.Namespace).Create(context.TODO(), &pod, metav1.CreateOptions{}); err != nil {
 		GinkgoT().Fatalf("Error creating new fake Mesh Catalog: %s", err.Error())
 	}

pkg/catalog/routes_test.go

@@ -558,6 +558,10 @@ func TestListTrafficPolicies(t *testing.T) {
 			input:  tests.BookstoreV1Service,
 			output: []trafficpolicy.TrafficTarget{tests.BookstoreV1TrafficPolicy},
 		},
+		{
+			input:  tests.BookstoreV2Service,
+			output: []trafficpolicy.TrafficTarget{tests.BookstoreV2TrafficPolicy},
+		},
 		{
 			input:  tests.BookbuyerService,
 			output: []trafficpolicy.TrafficTarget{tests.BookstoreV1TrafficPolicy, tests.BookstoreV2TrafficPolicy, tests.BookstoreApexTrafficPolicy},
@@ -592,7 +596,7 @@ func TestGetTrafficPoliciesForService(t *testing.T) {
 					HTTPRouteMatches: tests.BookstoreV1TrafficPolicy.HTTPRouteMatches,
 				},
 				{
-					Name:             utils.GetTrafficTargetName(tests.TrafficTargetName, tests.BookbuyerService, tests.BookstoreV2Service),
+					Name:             utils.GetTrafficTargetName(tests.BookstoreV2TrafficTargetName, tests.BookbuyerService, tests.BookstoreV2Service),
 					Destination:      tests.BookstoreV2Service,
 					Source:           tests.BookbuyerService,
 					HTTPRouteMatches: tests.BookstoreV2TrafficPolicy.HTTPRouteMatches,

pkg/catalog/types.go

@@ -89,6 +89,9 @@ type MeshCataloger interface {
 	// ListEndpointsForService returns the list of individual instance endpoint backing a service
 	ListEndpointsForService(service.MeshService) ([]endpoint.Endpoint, error)
 
+	// ListEndpointsForPodFromServiceAccount retrieves the list of IP addresses for the given service account
+	ListEndpointsForPodFromServiceAccount(sa service.K8sServiceAccount) []endpoint.Endpoint
+
 	// GetResolvableServiceEndpoints returns the resolvable set of endpoint over which a service is accessible using its FQDN.
 	// These are the endpoint destinations we'd expect client applications sends the traffic towards to, when attempting to
 	// reach a specific service.

pkg/debugger/policy_test.go

@@ -53,6 +53,6 @@ func TestGetSMIPolicies(t *testing.T) {
 	responseRecorder := httptest.NewRecorder()
 	smiPoliciesHandler.ServeHTTP(responseRecorder, nil)
 	actualResponseBody := responseRecorder.Body.String()
-	expectedResponseBody := `{"traffic_splits":[{"metadata":{"name":"bar","namespace":"foo","creationTimestamp":null},"spec":{}}],"weighted_services":[{"service_name:omitempty":{"Namespace":"default","Name":"bookstore-v1"},"weight:omitempty":90,"root_service:omitempty":"bookstore-apex"},{"service_name:omitempty":{"Namespace":"default","Name":"bookstore-v2"},"weight:omitempty":10,"root_service:omitempty":"bookstore-apex"}],"service_accounts":[{"Namespace":"default","Name":"bookbuyer"}],"route_groups":[{"kind":"HTTPRouteGroup","apiVersion":"specs.smi-spec.io/v1alpha4","metadata":{"name":"bookstore-service-routes","namespace":"default","creationTimestamp":null},"spec":{"matches":[{"name":"buy-books","methods":["GET"],"pathRegex":"/buy","headers":[{"user-agent":"test-UA"}]},{"name":"sell-books","methods":["GET"],"pathRegex":"/sell","headers":[{"user-agent":"test-UA"}]},{"name":"allow-everything-on-header","headers":[{"user-agent":"test-UA"}]}]}}],"traffic_targets":[{"kind":"TrafficTarget","apiVersion":"access.smi-spec.io/v1alpha3","metadata":{"name":"bookbuyer-access-bookstore","namespace":"default","creationTimestamp":null},"spec":{"destination":{"kind":"Name","name":"bookstore","namespace":"default"},"sources":[{"kind":"Name","name":"bookbuyer","namespace":"default"}],"rules":[{"kind":"HTTPRouteGroup","name":"bookstore-service-routes","matches":["buy-books","sell-books"]}]}}]}`
+	expectedResponseBody := `{"traffic_splits":[{"metadata":{"name":"bar","namespace":"foo","creationTimestamp":null},"spec":{}}],"weighted_services":[{"service_name:omitempty":{"Namespace":"default","Name":"bookstore-v1"},"weight:omitempty":90,"root_service:omitempty":"bookstore-apex"},{"service_name:omitempty":{"Namespace":"default","Name":"bookstore-v2"},"weight:omitempty":10,"root_service:omitempty":"bookstore-apex"}],"service_accounts":[{"Namespace":"default","Name":"bookbuyer"}],"route_groups":[{"kind":"HTTPRouteGroup","apiVersion":"specs.smi-spec.io/v1alpha4","metadata":{"name":"bookstore-service-routes","namespace":"default","creationTimestamp":null},"spec":{"matches":[{"name":"buy-books","methods":["GET"],"pathRegex":"/buy","headers":[{"user-agent":"test-UA"}]},{"name":"sell-books","methods":["GET"],"pathRegex":"/sell","headers":[{"user-agent":"test-UA"}]},{"name":"allow-everything-on-header","headers":[{"user-agent":"test-UA"}]}]}}],"traffic_targets":[{"kind":"TrafficTarget","apiVersion":"access.smi-spec.io/v1alpha3","metadata":{"name":"bookbuyer-access-bookstore","namespace":"default","creationTimestamp":null},"spec":{"destination":{"kind":"ServiceAccount","name":"bookstore","namespace":"default"},"sources":[{"kind":"ServiceAccount","name":"bookbuyer","namespace":"default"}],"rules":[{"kind":"HTTPRouteGroup","name":"bookstore-service-routes","matches":["buy-books","sell-books"]}]}}]}`
 	assert.Equal(actualResponseBody, expectedResponseBody, "Actual value did not match expectations:\n%s", actualResponseBody)
 }

pkg/endpoint/providers/kube/client.go

@@ -67,6 +67,33 @@ func (c Client) ListEndpointsForService(svc service.MeshService) []endpoint.Endp
 	return endpoints
 }
 
+// ListEndpointsForPodFromServiceAccount retrieves the list of IP addresses for the given service account
+func (c Client) ListEndpointsForPodFromServiceAccount(sa service.K8sServiceAccount) []endpoint.Endpoint {
+	log.Trace().Msgf("[%s] Getting Endpoints for service account %s on Kubernetes", c.providerIdent, sa)
+	var endpoints []endpoint.Endpoint
+
+	podList := c.kubeController.ListPods()
+	for _, pod := range podList {
+		if pod.Namespace != sa.Namespace {
+			continue
+		}
+		if pod.Spec.ServiceAccountName != sa.Name {
+			continue
+		}
+
+		for _, podIP := range pod.Status.PodIPs {
+			ip := net.ParseIP(podIP.IP)
+			if ip == nil {
+				log.Error().Msgf("[%s] Error parsing IP address %s", c.providerIdent, podIP.IP)
+				break
+			}
+			ept := endpoint.Endpoint{IP: ip}
+			endpoints = append(endpoints, ept)
+		}
+	}
+	return endpoints
+}
+
 // GetServicesForServiceAccount retrieves a list of services for the given service account.
 func (c Client) GetServicesForServiceAccount(svcAccount service.K8sServiceAccount) ([]service.MeshService, error) {
 	services := mapset.NewSet()

pkg/endpoint/providers/kube/client_test.go

@@ -4,16 +4,20 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"testing"
 	"time"
 
+	"github.com/google/uuid"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
 	"github.com/golang/mock/gomock"
+	tassert "github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
+	testclient "k8s.io/client-go/kubernetes/fake"
 
 	"github.com/openservicemesh/osm/pkg/announcements"
 	"github.com/openservicemesh/osm/pkg/configurator"
@@ -552,3 +556,84 @@ var _ = Describe("Test Kube Client Provider (/w kubecontroller)", func() {
 		<-podsAndServiceChannel
 	})
 })
+
+func TestListEndpointsForPodFromServiceAccount(t *testing.T) {
+	assert := tassert.New(t)
+
+	testCases := []struct {
+		name                            string
+		serviceAccount                  service.K8sServiceAccount
+		outboundServiceAccountEndpoints map[service.K8sServiceAccount][]endpoint.Endpoint
+		expectedEndpoints               []endpoint.Endpoint
+	}{
+		{
+			name:           "get endpoints for pod with only one ip",
+			serviceAccount: tests.BookstoreServiceAccount,
+			outboundServiceAccountEndpoints: map[service.K8sServiceAccount][]endpoint.Endpoint{
+				tests.BookstoreServiceAccount: {{
+					IP: net.ParseIP(tests.ServiceIP),
+				}},
+			},
+			expectedEndpoints: []endpoint.Endpoint{{
+				IP: net.ParseIP(tests.ServiceIP),
+			}},
+		},
+		{
+			name:           "get endpoints for pod with multiple ips",
+			serviceAccount: tests.BookstoreServiceAccount,
+			outboundServiceAccountEndpoints: map[service.K8sServiceAccount][]endpoint.Endpoint{
+				tests.BookstoreServiceAccount: {
+					endpoint.Endpoint{
+						IP: net.ParseIP(tests.ServiceIP),
+					},
+					endpoint.Endpoint{
+						IP: net.ParseIP("9.9.9.9"),
+					},
+				},
+			},
+			expectedEndpoints: []endpoint.Endpoint{{
+				IP: net.ParseIP(tests.ServiceIP),
+			},
+				{
+					IP: net.ParseIP("9.9.9.9"),
+				}},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockCtrl := gomock.NewController(t)
+			kubeClient := testclient.NewSimpleClientset()
+			defer mockCtrl.Finish()
+
+			mockKubeController := k8s.NewMockController(mockCtrl)
+			mockConfigurator := configurator.NewMockConfigurator(mockCtrl)
+			providerID := "provider"
+
+			provider, err := NewProvider(kubeClient, mockKubeController, providerID, mockConfigurator)
+			assert.Nil(err)
+
+			var pods []*v1.Pod
+			for sa, endpoints := range tc.outboundServiceAccountEndpoints {
+				podlabels := map[string]string{
+					tests.SelectorKey:                tests.SelectorValue,
+					constants.EnvoyUniqueIDLabelName: uuid.New().String(),
+				}
+				pod := tests.NewPodFixture(sa.Namespace, sa.Name, sa.Name, podlabels)
+				var podIps []v1.PodIP
+				for _, ep := range endpoints {
+					podIps = append(podIps, v1.PodIP{IP: ep.IP.String()})
+				}
+				pod.Status.PodIPs = podIps
+				_, err := kubeClient.CoreV1().Pods(sa.Namespace).Create(context.TODO(), &pod, metav1.CreateOptions{})
+				assert.Nil(err)
+				pods = append(pods, &pod)
+			}
+			mockKubeController.EXPECT().ListPods().Return(pods).AnyTimes()
+
+			actual := provider.ListEndpointsForPodFromServiceAccount(tc.serviceAccount)
+			assert.NotNil(actual)
+			assert.ElementsMatch(actual, tc.expectedEndpoints)
+		})
+	}
+}

pkg/endpoint/providers/kube/fake.go

@@ -24,12 +24,18 @@ func NewFakeProvider() endpoint.Provider {
 			tests.BookstoreV2ServiceAccount: {tests.BookstoreV2Service},
 			tests.BookbuyerServiceAccount:   {tests.BookbuyerService},
 		},
+		svcAccountEndpoints: map[service.K8sServiceAccount][]endpoint.Endpoint{
+			tests.BookstoreServiceAccount:   {tests.Endpoint, tests.Endpoint},
+			tests.BookstoreV2ServiceAccount: {tests.Endpoint},
+			tests.BookbuyerServiceAccount:   {tests.Endpoint},
+		},
 	}
 }
 
 type fakeClient struct {
-	endpoints map[string][]endpoint.Endpoint
-	services  map[service.K8sServiceAccount][]service.MeshService
+	endpoints           map[string][]endpoint.Endpoint
+	services            map[service.K8sServiceAccount][]service.MeshService
+	svcAccountEndpoints map[service.K8sServiceAccount][]endpoint.Endpoint
 }
 
 // Retrieve the IP addresses comprising the given service.
@@ -40,6 +46,14 @@ func (f fakeClient) ListEndpointsForService(svc service.MeshService) []endpoint.
 	panic(fmt.Sprintf("You are asking for MeshService=%s but the fake Kubernetes client has not been initialized with this. What we have is: %+v", svc.String(), f.endpoints))
 }
 
+// Retrieve the IP addresses comprising the given service account.
+func (f fakeClient) ListEndpointsForPodFromServiceAccount(sa service.K8sServiceAccount) []endpoint.Endpoint {
+	if ep, ok := f.svcAccountEndpoints[sa]; ok {
+		return ep
+	}
+	panic(fmt.Sprintf("You are asking for K8sServiceAccount=%s but the fake Kubernetes client has not been initialized with this. What we have is: %+v", sa.String(), f.svcAccountEndpoints))
+}
+
 func (f fakeClient) GetServicesForServiceAccount(svcAccount service.K8sServiceAccount) ([]service.MeshService, error) {
 	services, ok := f.services[svcAccount]
 	if !ok {

pkg/endpoint/types.go

@@ -12,6 +12,9 @@ type Provider interface {
 	// Retrieve the IP addresses comprising the given service.
 	ListEndpointsForService(service.MeshService) []Endpoint
 
+	// ListEndpointsForPodFromServiceAccount retrieves the list of IP addresses for the given service account
+	ListEndpointsForPodFromServiceAccount(sa service.K8sServiceAccount) []Endpoint
+
 	// Retrieve the namespaced services for a given service account
 	GetServicesForServiceAccount(service.K8sServiceAccount) ([]service.MeshService, error)