Repository: openservicemesh/osm (go)

Issue:
Avoid setting default meshroot certificate if one is already present in the cluster
**Please describe the Improvement and/or Feature Request**
<!-- A clear and concise description of the proposal. -->

If the default meshrootcertificate is rotated out and the bootstrapper restarts then it will then attempt to recreate the default meshrootcertificate. Once validation (#5220) is in place this will fail causing the bootstrapper to crashloop. 

One way to solve this would be to check if there are any `Active` Meshroot certificates and don't try to create a default. Another possible option is to right an annotation to the preset config map that it was already applied, if set then don't create an MRC.

**Scope (please mark with X where applicable)**
- New Functionality          [ ]
- Install                    [ ]
- SMI Traffic Access Policy  [ ]
- SMI Traffic Specs Policy   [ ]
- SMI Traffic Split Policy   [ ]
- Permissive Traffic Policy  [ ]
- Ingress                    [ ]
- Egress                     [ ]
- Envoy Control Plane        [ ]
- CLI Tool                   [ ]
- Metrics                    [ ]
- Certificate Management     [ ]
- Sidecar Injection          [ ]
- Logging                    [ ]
- Debugging                  [ ]
- Tests                      [ ]
- CI System                  [ ]
- Demo                       [ ]
- Project Release            [ ]

**Possible use cases**
<!-- A clear and concise description of possible use cases for this proposal. -->


Pull request description:
fix(certs): check for active MRCs on bootstrap startup
<!--

Please describe the motivation for this PR and provide enough
information so that others can review it.

-->
**Description**:

Checks if an active MRC exists before attempting to create the default MRC. Previously, if an active MRC existed that did not have the default MRC name, osm-mesh-root-certificate, the osm-bootstrap would error.

Resolves #5237

<!--

Please describe how this change was tested. You could include supporting information
such as logs, snippets, and screenshots.

-->
**Testing done**:
- unit tests

<!--

Please mark with X for applicable areas.

-->
**Affected area**:
| Functional Area            |     |
| -------------------------- | --- |
| New Functionality          | [x] |
| Certificate Management     | [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/cmd/osm-bootstrap/osm-bootstrap.go b/cmd/osm-bootstrap/osm-bootstrap.go
index d98316bc..5409abb4 100644
--- a/cmd/osm-bootstrap/osm-bootstrap.go
+++ b/cmd/osm-bootstrap/osm-bootstrap.go
@@ -402,16 +402,18 @@ func (b *bootstrap) ensureMeshRootCertificate() error {
 		return err
 	}
 
-	if len(meshRootCertificateList.Items) != 0 {
-		return nil
+	for _, mrc := range meshRootCertificateList.Items {
+		// if an MRC with Active intent exists in the mesh, do not create the default MRC
+		if mrc.Spec.Intent == configv1alpha2.ActiveIntent {
+			log.Debug().Msgf("Found MeshRootCertificate %s with %s intent. No default MeshRootCertificate will be created.", mrc.Name, mrc.Spec.Intent)
+			return nil
+		}
 	}
 
-	// create a MeshRootCertificate since none were found
 	return b.createMeshRootCertificate()
 }
 
 func (b *bootstrap) createMeshRootCertificate() error {
-	// TODO(5046): create the mrc per https://gist.github.com/keithmattix/9ac73abbcb5721b0ce58102ac3ebff29
 	// find preset config map to build the MeshRootCertificate from
 	presetMeshRootCertificate, err := b.kubeClient.CoreV1().ConfigMaps(b.namespace).Get(context.TODO(), presetMeshRootCertificateName, metav1.GetOptions{})
 	if err != nil {
@@ -425,7 +427,18 @@ func (b *bootstrap) createMeshRootCertificate() error {
 	}
 	_, err = b.configClient.ConfigV1alpha2().MeshRootCertificates(b.namespace).Create(context.TODO(), defaultMeshRootCertificate, metav1.CreateOptions{})
 	if apierrors.IsAlreadyExists(err) {
-		log.Info().Msgf("MeshRootCertificate already exists in %s. Skip creating.", b.namespace)
+		// check the intent of the existing MeshRootCertificate
+		existingDefaultMeshRootCertificate, err := b.configClient.ConfigV1alpha2().MeshRootCertificates(b.namespace).Get(context.TODO(), constants.DefaultMeshRootCertificateName, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+		if existingDefaultMeshRootCertificate.Spec.Intent != configv1alpha2.ActiveIntent {
+			log.Warn().Msgf("Existing default MeshRootCertificate %s in %s does not have an active intent. No active MeshRootCertificate will be created.",
+				constants.DefaultMeshRootCertificateName, b.namespace)
+			return nil
+		}
+
+		log.Info().Msgf("MeshRootCertificate %s already exists in %s. Skip creating.", constants.DefaultMeshRootCertificateName, b.namespace)
 		return nil
 	}
 	if err != nil {
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/cmd/osm-bootstrap/osm-bootstrap_test.go b/cmd/osm-bootstrap/osm-bootstrap_test.go
index e2f14d84..fd25f3da 100644
--- a/cmd/osm-bootstrap/osm-bootstrap_test.go
+++ b/cmd/osm-bootstrap/osm-bootstrap_test.go
@@ -88,12 +88,44 @@ var testPresetMeshConfigMap = &corev1.ConfigMap{
 	},
 }
 
-var testMeshRootCertificate = &configv1alpha2.MeshRootCertificate{
+var testActiveNonDefaultMeshRootCertificate = &configv1alpha2.MeshRootCertificate{
+	ObjectMeta: metav1.ObjectMeta{
+		Namespace: testNamespace,
+		Name:      "nondefaultmrc",
+	},
+	Spec: configv1alpha2.MeshRootCertificateSpec{
+		Intent: configv1alpha2.ActiveIntent,
+	},
+}
+
+var testPassiveNonDefaultMeshRootCertificate = &configv1alpha2.MeshRootCertificate{
+	ObjectMeta: metav1.ObjectMeta{
+		Namespace: testNamespace,
+		Name:      "nondefaultmrc",
+	},
+	Spec: configv1alpha2.MeshRootCertificateSpec{
+		Intent: configv1alpha2.PassiveIntent,
+	},
+}
+
+var testActiveDefaultMeshRootCertificate = &configv1alpha2.MeshRootCertificate{
+	ObjectMeta: metav1.ObjectMeta{
+		Namespace: testNamespace,
+		Name:      constants.DefaultMeshRootCertificateName,
+	},
+	Spec: configv1alpha2.MeshRootCertificateSpec{
+		Intent: configv1alpha2.ActiveIntent,
+	},
+}
+
+var testPassiveDefaultMeshRootCertificate = &configv1alpha2.MeshRootCertificate{
 	ObjectMeta: metav1.ObjectMeta{
 		Namespace: testNamespace,
 		Name:      constants.DefaultMeshRootCertificateName,
 	},
-	Spec: configv1alpha2.MeshRootCertificateSpec{},
+	Spec: configv1alpha2.MeshRootCertificateSpec{
+		Intent: configv1alpha2.PassiveIntent,
+	},
 }
 
 var testPresetMeshRootCertificate = &corev1.ConfigMap{
@@ -338,7 +370,7 @@ func TestCreateMeshRootCertificate(t *testing.T) {
 			name:                             "MeshRootCertificate already exists",
 			namespace:                        testNamespace,
 			kubeClient:                       fakeKube.NewSimpleClientset([]runtime.Object{testPresetMeshRootCertificate}...),
-			configClient:                     fakeConfig.NewSimpleClientset([]runtime.Object{testMeshRootCertificate}...),
+			configClient:                     fakeConfig.NewSimpleClientset([]runtime.Object{testActiveDefaultMeshRootCertificate}...),
 			expectDefaultMeshRootCertificate: true,
 			expectErr:                        false,
 		},
@@ -372,32 +404,60 @@ func TestCreateMeshRootCertificate(t *testing.T) {
 
 func TestEnsureMeshRootCertificate(t *testing.T) {
 	tests := []struct {
-		name         string
-		namespace    string
-		kubeClient   kubernetes.Interface
-		configClient configClientset.Interface
-		expectErr    bool
+		name                             string
+		namespace                        string
+		kubeClient                       kubernetes.Interface
+		configClient                     configClientset.Interface
+		expectDefaultMeshRootCertificate bool
+		expectErr                        bool
 	}{
 		{
-			name:         "MeshRootCertificate found",
-			namespace:    testNamespace,
-			kubeClient:   fakeKube.NewSimpleClientset(),
-			configClient: fakeConfig.NewSimpleClientset([]runtime.Object{testMeshRootCertificate}...),
-			expectErr:    false,
+			name:                             "active default MeshRootCertificate found",
+			namespace:                        testNamespace,
+			kubeClient:                       fakeKube.NewSimpleClientset(),
+			configClient:                     fakeConfig.NewSimpleClientset([]runtime.Object{testActiveDefaultMeshRootCertificate}...),
+			expectDefaultMeshRootCertificate: true,
+			expectErr:                        false,
 		},
 		{
-			name:         "MeshRootCertificate not found but successfully created",
-			namespace:    testNamespace,
-			kubeClient:   fakeKube.NewSimpleClientset([]runtime.Object{testPresetMeshRootCertificate}...),
-			configClient: fakeConfig.NewSimpleClientset(),
-			expectErr:    false,
+			name:                             "active non default MeshRootCertificate found",
+			namespace:                        testNamespace,
+			kubeClient:                       fakeKube.NewSimpleClientset(),
+			configClient:                     fakeConfig.NewSimpleClientset([]runtime.Object{testActiveNonDefaultMeshRootCertificate}...),
+			expectDefaultMeshRootCertificate: false,
+			expectErr:                        false,
 		},
 		{
-			name:         "MeshRootCertificate not found and error creating it",
-			namespace:    testNamespace,
-			kubeClient:   fakeKube.NewSimpleClientset(),
-			configClient: fakeConfig.NewSimpleClientset(),
-			expectErr:    true,
+			name:                             "MeshRootCertificate not found but successfully created",
+			namespace:                        testNamespace,
+			kubeClient:                       fakeKube.NewSimpleClientset([]runtime.Object{testPresetMeshRootCertificate}...),
+			configClient:                     fakeConfig.NewSimpleClientset(),
+			expectDefaultMeshRootCertificate: true,
+			expectErr:                        false,
+		},
+		{
+			name:                             "active MeshRootCertificate not found (passive non default MRC exists) but successfully created",
+			namespace:                        testNamespace,
+			kubeClient:                       fakeKube.NewSimpleClientset([]runtime.Object{testPresetMeshRootCertificate}...),
+			configClient:                     fakeConfig.NewSimpleClientset([]runtime.Object{testPassiveNonDefaultMeshRootCertificate}...),
+			expectDefaultMeshRootCertificate: true,
+			expectErr:                        false,
+		},
+		{
+			name:                             "active MeshRootCertificate not found (passive default MRC exists) and no active MeshRootCertificate created",
+			namespace:                        testNamespace,
+			kubeClient:                       fakeKube.NewSimpleClientset([]runtime.Object{testPresetMeshRootCertificate}...),
+			configClient:                     fakeConfig.NewSimpleClientset([]runtime.Object{testPassiveDefaultMeshRootCertificate}...),
+			expectDefaultMeshRootCertificate: true,
+			expectErr:                        false,
+		},
+		{
+			name:                             "MeshRootCertificate not found and error creating it",
+			namespace:                        testNamespace,
+			kubeClient:                       fakeKube.NewSimpleClientset(),
+			configClient:                     fakeConfig.NewSimpleClientset(),
+			expectDefaultMeshRootCertificate: false,
+			expectErr:                        true,
 		},
 	}
 
@@ -411,10 +471,18 @@ func TestEnsureMeshRootCertificate(t *testing.T) {
 			}
 
 			err := b.ensureMeshRootCertificate()
-			assert.Equal(tc.expectErr, err != nil)
+			if tc.expectErr {
+				assert.Error(err)
+			} else {
+				assert.NoError(err)
+			}
 
 			_, err = b.configClient.ConfigV1alpha2().MeshRootCertificates(b.namespace).Get(context.TODO(), constants.DefaultMeshRootCertificateName, metav1.GetOptions{})
-			assert.Equal(tc.expectErr, err != nil)
+			if tc.expectDefaultMeshRootCertificate {
+				assert.NoError(err)
+			} else {
+				assert.Error(err)
+			}
 		})
 	}
 }
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() {