Repository: openservicemesh/osm (go)

Issue:
Allow specifying the removal of HTTP query string from access logs
**Please describe the Improvement and/or Feature Request**
<!-- A clear and concise description of the proposal. -->
Currently, OSM does not have a way to remove query string from the reuqest path that is a part of the access log. For example, consider the below access log:
`[2022-10-19T20:23:53.392Z] authority=httpbin.httpbin:14001 path=/get?secret=password bytes_received=0 bytes_sent=0`

Since it's possible for the query string to contain sensitive info, provide the capability to remove the query string from the path logged in the access log, otherwise it could result in a [vulnerability arising from information exposure through query strings](https://owasp.org/www-community/vulnerabilities/Information_exposure_through_query_strings_in_url).

Envoy implements a formatter to remove query strings from access logs: https://www.envoyproxy.io/docs/envoy/latest/api-v3/extensions/formatter/req_without_query/v3/req_without_query.proto using the command operator `%REQ_WITHOUT_QUERY(X?Y):Z%`. Leverage this capability in OSM's access log functionality where if the `%REQ_WITHOUT_QUERY` is present in the access log format, configure the appropriate Envoy access log formatter to strip query strings from the request path.



**Scope (please mark with X where applicable)**
- New Functionality          [X]

**Possible use cases**
<!-- A clear and concise description of possible use cases for this proposal. -->
Prevent [vulnerability arising from information exposure through query strings](https://owasp.org/www-community/vulnerabilities/Information_exposure_through_query_strings_in_url)


Pull request description:
envoy/access-logs: allow removing HTTP query string
<!--

Please describe the motivation for this PR and provide enough
information so that others can review it.

-->
**Description**:
Allows specifying the removal of HTTP query strings from access logs as they may contain sensitive
info. If the %REQ_WITHOUT_QUERY command operator
is specified instead of the regular %REQ operator, then a query string formatter is used such that
the HTTP path without the query string is rendered in the access logs.

Resolves #5241

<!--

Please describe how this change was tested. You could include supporting information
such as logs, snippets, and screenshots.

-->
**Testing done**:
Tested query string masking using the `%REQ(X-ENVOY-ORIGINAL-PATH?:PATH)%` 
for both JSON and text based access log formats.

E.g. for the query `http://httpbin.httpbin:14001/status/200?foo=bar`, the `foo=bar` query string is stripped off when using the `%REQ_WITHOUT_QUERY` instead of the `%REQ` operator. 

TEXT
```
apiVersion: policy.openservicemesh.io/v1alpha1
kind: Telemetry
metadata:
  name: httpbin-log
  namespace: httpbin
spec:
  accessLog:
    format: "[%START_TIME%] \"%REQ(:METHOD)% %REQ_WITHOUT_QUERY(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL%\" %RESPONSE_CODE% %RESPONSE_FLAGS%\n"
```
```
[2022-11-14T20:07:40.010Z] "HEAD /status/200 HTTP/1.1" 200 -
```

JSON 
```
apiVersion: policy.openservicemesh.io/v1alpha1
kind: Telemetry
metadata:
  name: httpbin-log
  namespace: httpbin
spec:
  accessLog:
    format: '{"path":"%REQ_WITHOUT_QUERY(X-ENVOY-ORIGINAL-PATH?:PATH)%","bytes_received":"%BYTES_RECEIVED%","bytes_sent":"%BYTES_SENT%"}'
```
```
{"path":"/status/200","bytes_received":0,"bytes_sent":0}
```

<!--

Please mark with X for applicable areas.

-->
**Affected area**:
| Functional Area            |     |
| -------------------------- | --- |
| New Functionality          | [X] |
| Observability              | [X] |

Please answer the following questions with yes/no.

1. Does this change contain code from or inspired by another project? `no`
    -   Did you notify the maintainers and provide attribution?

2. Is this a breaking change? `no`

3. Has documentation corresponding to this change been updated in the [osm-docs](https://github.com/openservicemesh/osm-docs) repo (if applicable)? `no`

Patch:
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index 37bb30f3..a7428cbb 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -241,7 +241,7 @@ jobs:
       matrix:
         k8s_version: [""]
         focus: [""]
-        bucket: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
+        bucket: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
         include:
           - k8s_version: v1.22.9
             focus: "Test traffic flowing from client to server with a Kubernetes Service for the Source: HTTP"
diff --git a/.github/workflows/nightly-fips.yml b/.github/workflows/nightly-fips.yml
index 2c040fde..09a59a92 100644
--- a/.github/workflows/nightly-fips.yml
+++ b/.github/workflows/nightly-fips.yml
@@ -22,7 +22,7 @@ jobs:
       matrix:
         k8s_version: [""]
         focus: [""]
-        bucket: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+        bucket: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
         include:
           - k8s_version: v1.22.9
             focus: "Test traffic flowing from client to server with a Kubernetes Service for the Source: HTTP"
diff --git a/pkg/cli/verifier/envoy_config.go b/pkg/cli/verifier/envoy_config.go
index 7e3896e2..d0c1dfef 100644
--- a/pkg/cli/verifier/envoy_config.go
+++ b/pkg/cli/verifier/envoy_config.go
@@ -19,6 +19,7 @@ import (
 	configv1alpha2 "github.com/openservicemesh/osm/pkg/apis/config/v1alpha2"
 	"github.com/openservicemesh/osm/pkg/compute/kube"
 	"github.com/openservicemesh/osm/pkg/envoy"
+	"github.com/openservicemesh/osm/pkg/k8s"
 	"github.com/openservicemesh/osm/pkg/trafficpolicy"
 
 	"github.com/openservicemesh/osm/pkg/constants"
@@ -479,11 +480,16 @@ func (v *EnvoyConfigVerifier) getDstMeshServicesForK8sSvc(svc corev1.Service) ([
 		// us to retrieve the TargetPort for the MeshService.
 		meshSvc.TargetPort = kube.GetTargetPortFromEndpoints(portSpec.Name, *endpoints)
 
-		if !kube.IsHeadlessService(svc) {
+		// Even if the service is headless, add it so it can be targeted
+
+		if !k8s.IsHeadlessService(svc) {
 			meshServices = append(meshServices, meshSvc)
 			continue
 		}
 
+		// If there's not at least 1 subdomain-ed MeshService added,
+		// add the entire headless service
+		var added bool
 		for _, subset := range endpoints.Subsets {
 			for _, address := range subset.Addresses {
 				if address.Hostname == "" {
@@ -498,8 +504,13 @@ func (v *EnvoyConfigVerifier) getDstMeshServicesForK8sSvc(svc corev1.Service) ([
 					Protocol:   meshSvc.Protocol,
 				}
 				meshServices = append(meshServices, mSvc)
+				added = true
 			}
 		}
+
+		if !added {
+			meshServices = append(meshServices, meshSvc)
+		}
 	}
 
 	return meshServices, nil
diff --git a/pkg/compute/kube/client.go b/pkg/compute/kube/client.go
index 135fa3c4..8ad0cdca 100644
--- a/pkg/compute/kube/client.go
+++ b/pkg/compute/kube/client.go
@@ -697,11 +697,14 @@ func (c *client) serviceToMeshServices(svc corev1.Service) []service.MeshService
 		} else {
 			log.Warn().Msgf("k8s service %s/%s does not have endpoints but is being represented as a MeshService", svc.Namespace, svc.Name)
 		}
-		if !IsHeadlessService(svc) || endpoints == nil {
+
+		if !k8s.IsHeadlessService(svc) || endpoints == nil {
 			meshServices = append(meshServices, meshSvc)
 			continue
 		}
-
+		// If there's not at least 1 subdomain-ed MeshService added,
+		// add the entire headless service
+		var added bool
 		for _, subset := range endpoints.Subsets {
 			for _, address := range subset.Addresses {
 				if address.Hostname == "" {
@@ -715,8 +718,13 @@ func (c *client) serviceToMeshServices(svc corev1.Service) []service.MeshService
 					TargetPort: meshSvc.TargetPort,
 					Protocol:   meshSvc.Protocol,
 				})
+				added = true
 			}
 		}
+
+		if !added {
+			meshServices = append(meshServices, meshSvc)
+		}
 	}
 	return meshServices
 }
diff --git a/pkg/compute/kube/util.go b/pkg/compute/kube/util.go
index 4669b091..a2990228 100644
--- a/pkg/compute/kube/util.go
+++ b/pkg/compute/kube/util.go
@@ -35,9 +35,3 @@ func GetTargetPortFromEndpoints(endpointName string, endpoints corev1.Endpoints)
 	}
 	return
 }
-
-// IsHeadlessService determines whether or not a corev1.Service is a headless service
-// TODO(4863): unexport this method, it should not be used outside of this package.
-func IsHeadlessService(svc corev1.Service) bool {
-	return len(svc.Spec.ClusterIP) == 0 || svc.Spec.ClusterIP == corev1.ClusterIPNone
-}
diff --git a/pkg/envoy/generator/lds/accesslog.go b/pkg/envoy/generator/lds/accesslog.go
index cd7d12f0..633c3e50 100644
--- a/pkg/envoy/generator/lds/accesslog.go
+++ b/pkg/envoy/generator/lds/accesslog.go
@@ -3,12 +3,14 @@ package lds
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 
 	xds_accesslog "github.com/envoyproxy/go-control-plane/envoy/config/accesslog/v3"
 	xds_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	xds_grpc_accesslog "github.com/envoyproxy/go-control-plane/envoy/extensions/access_loggers/grpc/v3"
 	xds_otel_accesslog "github.com/envoyproxy/go-control-plane/envoy/extensions/access_loggers/open_telemetry/v3"
 	xds_accesslog_stream "github.com/envoyproxy/go-control-plane/envoy/extensions/access_loggers/stream/v3"
+	xds_formatter "github.com/envoyproxy/go-control-plane/envoy/extensions/formatter/req_without_query/v3"
 	otlpCommon "go.opentelemetry.io/proto/otlp/common/v1"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/anypb"
@@ -18,6 +20,10 @@ import (
 	"github.com/openservicemesh/osm/pkg/protobuf"
 )
 
+const (
+	reqWithoutQuery = "%REQ_WITHOUT_QUERY"
+)
+
 var (
 	defaultAccessLogJSONFields = &structpb.Struct{
 		Fields: map[string]*structpb.Value{
@@ -44,6 +50,13 @@ var (
 	}
 
 	defaultAccessLogJSONStr = protobuf.MustToJSON(defaultAccessLogJSONFields)
+
+	accessLogFormatters = []*xds_core.TypedExtensionConfig{
+		{
+			Name:        "envoy.formatter.req_without_query",
+			TypedConfig: protobuf.MustMarshalAny(&xds_formatter.ReqWithoutQuery{}),
+		},
+	}
 )
 
 type accessLogBuilder struct {
@@ -137,7 +150,7 @@ func (ab *accessLogBuilder) Build() ([]*xds_accesslog.AccessLog, error) {
 }
 
 func buildStdoutJSONAccessLog(format *structpb.Struct) *xds_accesslog_stream.StdoutAccessLog {
-	return &xds_accesslog_stream.StdoutAccessLog{
+	config := &xds_accesslog_stream.StdoutAccessLog{
 		AccessLogFormat: &xds_accesslog_stream.StdoutAccessLog_LogFormat{
 			LogFormat: &xds_core.SubstitutionFormatString{
 				Format: &xds_core.SubstitutionFormatString_JsonFormat{
@@ -146,10 +159,24 @@ func buildStdoutJSONAccessLog(format *structpb.Struct) *xds_accesslog_stream.Std
 			},
 		},
 	}
+
+	stripReqQuery := false
+	for _, value := range format.Fields {
+		if strings.Contains(value.GetStringValue(), reqWithoutQuery) {
+			stripReqQuery = true
+			break
+		}
+	}
+
+	if stripReqQuery {
+		config.GetLogFormat().Formatters = accessLogFormatters
+	}
+
+	return config
 }
 
 func buildStdoutTextAccessLog(format string) *xds_accesslog_stream.StdoutAccessLog {
-	return &xds_accesslog_stream.StdoutAccessLog{
+	config := &xds_accesslog_stream.StdoutAccessLog{
 		AccessLogFormat: &xds_accesslog_stream.StdoutAccessLog_LogFormat{
 			LogFormat: &xds_core.SubstitutionFormatString{
 				Format: &xds_core.SubstitutionFormatString_TextFormatSource{
@@ -162,6 +189,13 @@ func buildStdoutTextAccessLog(format string) *xds_accesslog_stream.StdoutAccessL
 			},
 		},
 	}
+
+	stripReqQuery := strings.Contains(format, reqWithoutQuery)
+	if stripReqQuery {
+		config.GetLogFormat().Formatters = accessLogFormatters
+	}
+
+	return config
 }
 
 func buildOpenTelemetryAccessLogConfig(cluster string, body string, attributes map[string]string) *xds_otel_accesslog.OpenTelemetryAccessLogConfig {


Test patch:
diff --git a/pkg/compute/kube/client_test.go b/pkg/compute/kube/client_test.go
index d16ce162..4c1f490d 100644
--- a/pkg/compute/kube/client_test.go
+++ b/pkg/compute/kube/client_test.go
@@ -2372,6 +2372,61 @@ func TestServiceToMeshServices(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "k8s headless service with single port and endpoint (no hostname), no appProtocol set",
+			// Single port on the service maps to a single MeshService.
+			// Since no appProtocol is specified, MeshService.Protocol should default
+			// to http because Port.Protocol=TCP
+			svc: corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns1",
+					Name:      "s1",
+				},
+				Spec: corev1.ServiceSpec{
+					Ports: []corev1.ServicePort{
+						{
+							Name:     "p1",
+							Port:     80,
+							Protocol: corev1.ProtocolTCP,
+						},
+					},
+					ClusterIP: corev1.ClusterIPNone,
+				},
+			},
+			svcEndpoints: []runtime.Object{
+				&corev1.Endpoints{
+					ObjectMeta: metav1.ObjectMeta{
+						// Should match svc.Name and svc.Namespace
+						Namespace: "ns1",
+						Name:      "s1",
+					},
+					Subsets: []corev1.EndpointSubset{
+						{
+							Addresses: []corev1.EndpointAddress{
+								{
+									IP: "10.1.0.1",
+								},
+							},
+							Ports: []corev1.EndpointPort{
+								{
+									// Must match the port of 'svc.Spec.Ports[0]'
+									Port: 8080, // TargetPort
+								},
+							},
+						},
+					},
+				},
+			},
+			expected: []service.MeshService{
+				{
+					Namespace:  "ns1",
+					Name:       "s1",
+					Port:       80,
+					TargetPort: 8080,
+					Protocol:   "http",
+				},
+			},
+		},
 		{
 			name: "multiple ports on k8s service with appProtocol specified",
 			svc: corev1.Service{
@@ -2514,6 +2569,78 @@ func TestServiceToMeshServices(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "multiple ports on k8s headless service with appProtocol specified and no hostname in the endpoints",
+			svc: corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns1",
+					Name:      "s1",
+				},
+				Spec: corev1.ServiceSpec{
+					ClusterIP: corev1.ClusterIPNone,
+					Ports: []corev1.ServicePort{
+						{
+							Name:        "p1",
+							Port:        80,
+							AppProtocol: pointer.StringPtr("http"),
+						},
+						{
+							Name:        "p2",
+							Port:        90,
+							AppProtocol: pointer.StringPtr("tcp"),
+						},
+					},
+				},
+			},
+			svcEndpoints: []runtime.Object{
+				&corev1.Endpoints{
+					ObjectMeta: metav1.ObjectMeta{
+						// Should match svc.Name and svc.Namespace
+						Namespace: "ns1",
+						Name:      "s1",
+					},
+					Subsets: []corev1.EndpointSubset{
+						{
+							Addresses: []corev1.EndpointAddress{
+								{
+									IP: "10.1.0.1",
+								},
+							},
+							Ports: []corev1.EndpointPort{
+								{
+									// Must match the port of 'svc.Spec.Ports[0]'
+									Name:        "p1",
+									Port:        8080, // TargetPort
+									AppProtocol: pointer.StringPtr("http"),
+								},
+								{
+									// Must match the port of 'svc.Spec.Ports[1]'
+									Name:        "p2",
+									Port:        9090, // TargetPort
+									AppProtocol: pointer.StringPtr("tcp"),
+								},
+							},
+						},
+					},
+				},
+			},
+			expected: []service.MeshService{
+				{
+					Namespace:  "ns1",
+					Name:       "s1",
+					Port:       80,
+					TargetPort: 8080,
+					Protocol:   "http",
+				},
+				{
+					Namespace:  "ns1",
+					Name:       "s1",
+					Port:       90,
+					TargetPort: 9090,
+					Protocol:   "tcp",
+				},
+			},
+		},
 		{
 			name: "duplicate ports on k8s service with different protocols",
 			svc: corev1.Service{
@@ -2682,7 +2809,7 @@ func TestGetMeshService(t *testing.T) {
 			expectErr:          true,
 		},
 		{
-			name: "TargetPort not found as Endpoint does not exist",
+			name: "TargetPort not found as matching Endpoint does not exist",
 			svc: &corev1.Service{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "s1",
@@ -2693,6 +2820,7 @@ func TestGetMeshService(t *testing.T) {
 						Name: "p1",
 						Port: 80,
 					}},
+					ClusterIP: corev1.ClusterIPNone,
 				},
 			},
 			endpoints: &corev1.Endpoints{
@@ -2707,14 +2835,18 @@ func TestGetMeshService(t *testing.T) {
 								Name: "invalid", // does not match svc port
 								Port: 8080,
 							},
+							{
+								Name: "invalid2",
+								Port: 8081, // also does not match svc port; having 2 ports triggers the name filter logic
+							},
 						},
 					},
 				},
 			},
 			namespacedSvc:      types.NamespacedName{Namespace: "ns1", Name: "s1"}, // matches svc
 			port:               80,                                                 // matches svc
-			expectedTargetPort: 0,                                                  // matches endpoint's 'p1' port
-			expectErr:          true,
+			expectedTargetPort: 0,                                                  // does not match either of the endpoint's ports
+			expectErr:          false,                                              // port matches, so no error should be expected
 		},
 	}
 
diff --git a/pkg/envoy/generator/lds/accesslog_test.go b/pkg/envoy/generator/lds/accesslog_test.go
index b5097063..7b6c735e 100644
--- a/pkg/envoy/generator/lds/accesslog_test.go
+++ b/pkg/envoy/generator/lds/accesslog_test.go
@@ -26,11 +26,23 @@ func TestAccessLogBuild(t *testing.T) {
 			expectErr:             false,
 		},
 		{
-			name:                  "custom  stream access log JSON format",
+			name:                  "custom stream access log text format with query formatter",
+			ab:                    NewAccessLogBuilder().Name("foo").Format("[%START_TIME%] \"%REQ(:METHOD)% %REQ_WITHOUT_QUERY(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL%\" %RESPONSE_CODE% %RESPONSE_FLAGS%\n"),
+			numExpectedAccessLogs: 1, // STDOUT stream access log
+			expectErr:             false,
+		},
+		{
+			name:                  "custom stream access log JSON format",
 			ab:                    NewAccessLogBuilder().Name("foo").Format(`{"authority":"%REQ(:AUTHORITY)%","bytes_received":"%BYTES_RECEIVED%","bytes_sent":"%BYTES_SENT%"}`),
 			numExpectedAccessLogs: 1, // STDOUT stream access log
 			expectErr:             false,
 		},
+		{
+			name:                  "custom stream access log JSON format with query formatter",
+			ab:                    NewAccessLogBuilder().Name("foo").Format(`{"authority":"%REQ_WITHOUT_QUERY(:AUTHORITY)%","bytes_received":"%BYTES_RECEIVED%","bytes_sent":"%BYTES_SENT%"}`),
+			numExpectedAccessLogs: 1, // STDOUT stream access log
+			expectErr:             false,
+		},
 		{
 			name:                  "default stream access log and OpenTelemetry access log",
 			ab:                    NewAccessLogBuilder().Name("foo").OpenTelemetryCluster("otel-collector"),
diff --git a/tests/e2e/e2e_client_server_connectivity_test.go b/tests/e2e/e2e_client_server_connectivity_test.go
index f978ecac..9c299904 100644
--- a/tests/e2e/e2e_client_server_connectivity_test.go
+++ b/tests/e2e/e2e_client_server_connectivity_test.go
@@ -12,6 +12,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/openservicemesh/osm/pkg/apis/config/v1alpha2"
+	"github.com/openservicemesh/osm/pkg/tests"
 
 	"github.com/openservicemesh/osm/pkg/constants"
 	. "github.com/openservicemesh/osm/tests/framework"
@@ -26,7 +27,7 @@ var _ = OSMDescribe("Test HTTP traffic from 1 pod client -> 1 pod server",
 	func() {
 		Context("Test traffic flowing from client to server with a Kubernetes Service for the Source: HTTP", func() {
 			withSourceKubernetesService := true
-			testTraffic(withSourceKubernetesService, PodCommandDefault)
+			testTraffic(withSourceKubernetesService, PodCommandDefault, false)
 		})
 
 		Context("Test traffic flowing from client to server without a Kubernetes Service for the Source: HTTP", func() {
@@ -34,18 +35,26 @@ var _ = OSMDescribe("Test HTTP traffic from 1 pod client -> 1 pod server",
 			// for the Envoy proxy to be configured for outbound traffic to some remote server.
 			// This test ensures we test this scenario: client Pod is not associated w/ a service.
 			withSourceKubernetesService := false
-			testTraffic(withSourceKubernetesService, PodCommandDefault)
+			testTraffic(withSourceKubernetesService, PodCommandDefault, false)
 		})
 
 		Context("Test traffic flowing from client to a server with a podIP bind", func() {
 			// Prior iterations of OSM didn't allow mesh services to bind to the podIP
 			// This test ensures that that behavior is configurable via MeshConfig
 			withSourceKubernetesService := true
-			testTraffic(withSourceKubernetesService, []string{"gunicorn", "-b", "$(POD_IP):80", "httpbin:app", "-k", "gevent"}, WithLocalProxyMode(v1alpha2.LocalProxyModePodIP))
+			testTraffic(withSourceKubernetesService, []string{"gunicorn", "-b", "$(POD_IP):80", "httpbin:app", "-k", "gevent"}, false, WithLocalProxyMode(v1alpha2.LocalProxyModePodIP))
+		})
+
+		Context("Test traffic flowing from client to a headless service without a Kubernetes Service for the Source: HTTP", func() {
+			// Prior iterations of OSM required that a source pod belong to a Kubernetes service
+			// for the Envoy proxy to be configured for outbound traffic to some remote server.
+			// This test ensures we test this scenario: client Pod is not associated w/ a service.
+			withSourceKubernetesService := true
+			testTraffic(withSourceKubernetesService, PodCommandDefault, true)
 		})
 	})
 
-func testTraffic(withSourceKubernetesService bool, destPodCommand []string, installOpts ...InstallOsmOpt) {
+func testTraffic(withSourceKubernetesService bool, destPodCommand []string, destServiceHeadless bool, installOpts ...InstallOsmOpt) {
 	const sourceName = "client"
 	const destName = "server"
 	var ns = []string{sourceName, destName}
@@ -68,7 +77,7 @@ func testTraffic(withSourceKubernetesService bool, destPodCommand []string, inst
 		Expect(err).NotTo(HaveOccurred())
 		_, err = Td.CreatePod(destName, podDef)
 		Expect(err).NotTo(HaveOccurred())
-		dstSvc, err := Td.CreateService(destName, svcDef)
+		dstSvc, err := Td.CreateService(destName, *tests.HeadlessSvc(&svcDef))
 		Expect(err).NotTo(HaveOccurred())
 
 		// Expect it to be up and running in it's receiver namespace
diff --git a/tests/e2e/e2e_deployment_client_headless_server_test.go b/tests/e2e/e2e_deployment_client_headless_server_test.go
new file mode 100644
index 00000000..cee4a1d2
--- /dev/null
+++ b/tests/e2e/e2e_deployment_client_headless_server_test.go
@@ -0,0 +1,216 @@
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/openservicemesh/osm/pkg/tests"
+	. "github.com/openservicemesh/osm/tests/framework"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var _ = OSMDescribe("Test HTTP traffic from N deployment client -> 1 deployment server",
+	OSMDescribeInfo{
+		Tier:   1,
+		Bucket: 11,
+	},
+	func() {
+		const (
+			// to name the header we will use to identify the server that replies
+			HTTPHeaderName = "podname"
+		)
+
+		var maxTestDuration = 150 * time.Second
+
+		Context("DeploymentsClientServer", func() {
+			var (
+				destApp           = "server"
+				sourceAppBaseName = "client"
+				sourceNamespaces  = []string{}
+
+				// Total (numberOfClientApps x replicaSetPerApp) pods
+				numberOfClientServices = 5
+				replicaSetPerService   = 5
+			)
+
+			// Used across the test to wait for concurrent steps to finish
+			var wg sync.WaitGroup
+
+			for i := 0; i < numberOfClientServices; i++ {
+				// base names for each client service
+				sourceNamespaces = append(sourceNamespaces, fmt.Sprintf("%s%d", sourceAppBaseName, i))
+			}
+
+			It("Tests HTTP traffic from multiple client deployments to a headless service deployment", func() {
+				// Install OSM
+				Expect(Td.InstallOSM(Td.GetOSMInstallOpts())).To(Succeed())
+
+				// Server NS
+				Expect(Td.CreateNs(destApp, nil)).To(Succeed())
+				Expect(Td.AddNsToMesh(true, destApp)).To(Succeed())
+
+				// Client Applications
+				for _, srcClient := range sourceNamespaces {
+					Expect(Td.CreateNs(srcClient, nil)).To(Succeed())
+					Expect(Td.AddNsToMesh(true, srcClient)).To(Succeed())
+				}
+
+				// Use a deployment with multiple replicaset at serverside
+				svcAccDef, deploymentDef, svcDef, err := Td.SimpleDeploymentApp(
+					SimpleDeploymentAppDef{
+						DeploymentName:     destApp,
+						Namespace:          destApp,
+						ServiceName:        destApp,
+						ServiceAccountName: destApp,
+						ReplicaCount:       int32(replicaSetPerService),
+						Image:              "simonkowallik/httpbin",
+						Ports:              []int{DefaultUpstreamServicePort},
+						Command:            HttpbinCmd,
+						OS:                 Td.ClusterOS,
+					})
+				Expect(err).NotTo(HaveOccurred())
+
+				// Expose an env variable such as XHTTPBIN_X_POD_NAME:
+				// This httpbin fork will pick certain env variable formats and reply the values as headers.
+				// We will expose pod name as one of these env variables, and will use it
+				// to identify the pod that replies to the request, and validate the test
+				deploymentDef.Spec.Template.Spec.Containers[0].Env = []v1.EnvVar{
+					{
+						Name: fmt.Sprintf("XHTTPBIN_%s", HTTPHeaderName),
+						ValueFrom: &v1.EnvVarSource{
+							FieldRef: &v1.ObjectFieldSelector{
+								FieldPath: "metadata.name",
+							},
+						},
+					},
+				}
+
+				_, err = Td.CreateServiceAccount(destApp, &svcAccDef)
+				Expect(err).NotTo(HaveOccurred())
+				_, err = Td.CreateDeployment(destApp, deploymentDef)
+				Expect(err).NotTo(HaveOccurred())
+				_, err = Td.CreateService(destApp, *tests.HeadlessSvc(&svcDef))
+				Expect(err).NotTo(HaveOccurred())
+
+				wg.Add(1)
+				go func(wg *sync.WaitGroup, srcClient string) {
+					defer GinkgoRecover()
+					defer wg.Done()
+					Expect(Td.WaitForPodsRunningReady(destApp, 200*time.Second, replicaSetPerService, nil)).To(Succeed())
+				}(&wg, destApp)
+
+				// Create all client deployments, also with replicaset
+				for _, srcClient := range sourceNamespaces {
+					svcAccDef, deploymentDef, svcDef, err = Td.SimpleDeploymentApp(
+						SimpleDeploymentAppDef{
+							DeploymentName:     srcClient,
+							Namespace:          srcClient,
+							ServiceAccountName: srcClient,
+							ServiceName:        srcClient,
+							ContainerName:      srcClient,
+							ReplicaCount:       int32(replicaSetPerService),
+							Command:            []string{"/bin/bash", "-c", "--"},
+							Args:               []string{"while true; do sleep 30; done;"},
+							Image:              "songrgg/alpine-debug",
+							Ports:              []int{DefaultUpstreamServicePort}, // Can't deploy services with empty/no ports
+							OS:                 Td.ClusterOS,
+						})
+					Expect(err).NotTo(HaveOccurred())
+
+					_, err = Td.CreateServiceAccount(srcClient, &svcAccDef)
+					Expect(err).NotTo(HaveOccurred())
+					_, err = Td.CreateDeployment(srcClient, deploymentDef)
+					Expect(err).NotTo(HaveOccurred())
+					_, err = Td.CreateService(srcClient, svcDef)
+					Expect(err).NotTo(HaveOccurred())
+
+					wg.Add(1)
+					go func(wg *sync.WaitGroup, srcClient string) {
+						defer GinkgoRecover()
+						defer wg.Done()
+						Expect(Td.WaitForPodsRunningReady(srcClient, 200*time.Second, replicaSetPerService, nil)).To(Succeed())
+					}(&wg, srcClient)
+				}
+				wg.Wait()
+
+				// Create traffic rules
+				// Deploy allow rules (10x)clients -> server
+				for _, srcClient := range sourceNamespaces {
+					httpRG, trafficTarget := Td.CreateSimpleAllowPolicy(
+						SimpleAllowPolicy{
+							RouteGroupName:    srcClient,
+							TrafficTargetName: srcClient,
+
+							SourceNamespace:      srcClient,
+							SourceSVCAccountName: srcClient,
+
+							DestinationNamespace:      destApp,
+							DestinationSvcAccountName: destApp,
+						})
+
+					// Configs have to be put into a monitored NS, and osm-system can't be by cli
+					_, err = Td.CreateHTTPRouteGroup(destApp, httpRG)
+					Expect(err).NotTo(HaveOccurred())
+					_, err = Td.CreateTrafficTarget(destApp, trafficTarget)
+					Expect(err).NotTo(HaveOccurred())
+				}
+
+				// Create Multiple HTTP request structure and fill it with appropriate details
+				requests := HTTPMultipleRequest{
+					Sources: []HTTPRequestDef{},
+				}
+				for _, ns := range sourceNamespaces {
+					pods, err := Td.Client.CoreV1().Pods(ns).List(context.Background(), metav1.ListOptions{})
+					Expect(err).To(BeNil())
+
+					for _, pod := range pods.Items {
+						requests.Sources = append(requests.Sources, HTTPRequestDef{
+							SourceNs:        ns,
+							SourcePod:       pod.Name,
+							SourceContainer: ns, // container_name == NS for this test
+
+							Destination: fmt.Sprintf("%s.%s:%d", destApp, destApp, DefaultUpstreamServicePort),
+						})
+					}
+				}
+
+				var results HTTPMultipleResults
+				var serversSeen map[string]bool = map[string]bool{} // Just counts unique servers seen
+				success := Td.WaitForRepeatedSuccess(func() bool {
+					// Issue all calls, get results
+					results = Td.MultipleHTTPRequest(&requests)
+
+					// Care, depending on variables there could be a lot of results
+					Td.PrettyPrintHTTPResults(&results)
+
+					// Verify success conditions
+					for _, ns := range results {
+						for _, podResult := range ns {
+							if podResult.Err != nil || podResult.StatusCode != 200 {
+								return false
+							}
+							// We should see pod header populated
+							dstPod, ok := podResult.Headers[HTTPHeaderName]
+							if ok {
+								// Store and mark that we have seen a response for this server pod
+								serversSeen[dstPod] = true
+							}
+						}
+					}
+					Td.T.Logf("Unique servers replied %d/%d", len(serversSeen), replicaSetPerService)
+					Expect(len(serversSeen)).To(Equal(replicaSetPerService))
+
+					return true
+				}, 5, maxTestDuration)
+
+				Expect(success).To(BeTrue())
+			})
+		})
+	})
diff --git a/tests/e2e/e2e_statefulsets_test.go b/tests/e2e/e2e_statefulsets_test.go
index 864e7d6c..5e0b6c63 100644
--- a/tests/e2e/e2e_statefulsets_test.go
+++ b/tests/e2e/e2e_statefulsets_test.go
@@ -22,7 +22,7 @@ import (
 var _ = OSMDescribe("Test traffic among Statefulset members",
 	OSMDescribeInfo{
 		Tier:   1,
-		Bucket: 9,
+		Bucket: 12,
 		OS:     OSCrossPlatform,
 	},
 	func() {