diff --git a/changelog/v1.18.0-beta35/cluster_names.yaml b/changelog/v1.18.0-beta35/cluster_names.yaml
new file mode 100644
index 00000000000..7be47c3358d
--- /dev/null
+++ b/changelog/v1.18.0-beta35/cluster_names.yaml
@@ -0,0 +1,7 @@
+changelog:
+  - type: NEW_FEATURE
+    issueLink: https://github.com/solo-io/solo-projects/issues/7105
+    resolvesIssue: false
+    description: >-
+      Kubernetes gateway cluster names for kubernetes services will have a new more parsable format.
+      This behavior for kubernetes gateway can be reverted for now via GG_K8S_GW_LEGACY_CLUSTER_NAMES.
\ No newline at end of file
diff --git a/projects/gateway2/krtcollections/endpoints.go b/projects/gateway2/krtcollections/endpoints.go
index 6bcc3c0011a..c5a3716fca9 100644
--- a/projects/gateway2/krtcollections/endpoints.go
+++ b/projects/gateway2/krtcollections/endpoints.go
@@ -226,12 +226,6 @@ func transformK8sEndpoints(ctx context.Context, inputs EndpointsInputs) func(kct
 			return nil
 		}
 
-		if len(endpointSlices) == 0 {
-			warnsToLog = append(warnsToLog, fmt.Sprintf("EndpointSlices not found for service %v/%v", svcNs, svcName))
-			logger.Debug("EndpointSlices not found for service")
-			return nil
-		}
-
 		// Initialize the returned EndpointsForUpstream
 		settings := krt.FetchOne(kctx, inputs.EndpointsSettings.AsCollection())
 		enableAutoMtls := settings.EnableAutoMtls
@@ -397,7 +391,7 @@ func findPortInEndpointSlice(endpointSlice *discoveryv1.EndpointSlice, singlePor
 
 // TODO: use exported version from translator?
 func GetEndpointClusterName(upstream *v1.Upstream) string {
-	clusterName := translator.UpstreamToClusterName(upstream.GetMetadata().Ref())
+	clusterName := translator.KubeGatewayUpstreamToClusterName(upstream)
 	endpointClusterName, err := translator.GetEndpointClusterName(clusterName, upstream)
 	if err != nil {
 		panic(err)
diff --git a/projects/gateway2/proxy_syncer/kube_gw_translator_syncer.go b/projects/gateway2/proxy_syncer/kube_gw_translator_syncer.go
index 0eaab0eefd8..3c28ffc99ef 100644
--- a/projects/gateway2/proxy_syncer/kube_gw_translator_syncer.go
+++ b/projects/gateway2/proxy_syncer/kube_gw_translator_syncer.go
@@ -14,6 +14,7 @@ import (
 	v1 "github.com/solo-io/gloo/projects/gloo/pkg/api/v1"
 	v1snap "github.com/solo-io/gloo/projects/gloo/pkg/api/v1/gloosnapshot"
 	"github.com/solo-io/gloo/projects/gloo/pkg/plugins"
+	"github.com/solo-io/gloo/projects/gloo/pkg/translator"
 	"github.com/solo-io/gloo/projects/gloo/pkg/xds"
 	"github.com/solo-io/solo-kit/pkg/api/v1/control-plane/cache"
 
@@ -63,7 +64,12 @@ func (s *ProxyTranslator) buildXdsSnapshot(
 		Messages: map[*core.ResourceRef][]string{},
 	}
 
-	xdsSnapshot, reports, proxyReport := s.translator.NewTranslator(ctx, settings).Translate(params, proxy)
+	tx := s.translator.NewTranslator(
+		ctx,
+		settings,
+		translator.ForKubeGatewayAPI(),
+	)
+	xdsSnapshot, reports, proxyReport := tx.Translate(params, proxy)
 
 	// Messages are aggregated during translation, and need to be added to reports
 	for _, messages := range params.Messages {
@@ -107,7 +113,6 @@ func (s *ProxyTranslator) syncXds(
 	// a default initial fetch timeout
 	snap.MakeConsistent()
 	s.xdsCache.SetSnapshot(proxyKey, snap)
-
 }
 
 func (s *ProxyTranslator) syncStatus(
diff --git a/projects/gateway2/validation/validator_test.go b/projects/gateway2/validation/validator_test.go
index 65013b2feb6..a0e1fabd2e3 100644
--- a/projects/gateway2/validation/validator_test.go
+++ b/projects/gateway2/validation/validator_test.go
@@ -81,6 +81,7 @@ var _ = Describe("Kube Gateway API Policy Validation Helper", func() {
 			settings,
 			pluginRegistry,
 			translator.EnvoyCacheResourcesListToFnvHash,
+			translator.ForKubeGatewayAPI(),
 		)
 		vc = gloovalidation.ValidatorConfig{
 			Ctx: context.Background(),
diff --git a/projects/gloo/constants/gloo_gateway.go b/projects/gloo/constants/gloo_gateway.go
index bbbc7bed0ba..2712e9561ac 100644
--- a/projects/gloo/constants/gloo_gateway.go
+++ b/projects/gloo/constants/gloo_gateway.go
@@ -2,4 +2,5 @@ package constants
 
 const (
 	GlooGatewayEnableK8sGwControllerEnv = "GG_K8S_GW_CONTROLLER"
+	GlooGatewayKubeStyleClusterNames    = "GG_K8S_GW_LEGACY_CLUSTER_NAMES"
 )
diff --git a/projects/gloo/pkg/plugins/kubernetes/uds_convert.go b/projects/gloo/pkg/plugins/kubernetes/uds_convert.go
index b52048b3ae2..3d657e974b5 100644
--- a/projects/gloo/pkg/plugins/kubernetes/uds_convert.go
+++ b/projects/gloo/pkg/plugins/kubernetes/uds_convert.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strconv"
 
 	"github.com/solo-io/gloo/projects/gloo/pkg/plugins/kubernetes/serviceconverter"
 	"github.com/solo-io/go-utils/contextutils"
@@ -20,6 +21,39 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
+// these labels are used to propagate internal data
+// on synthetic Gloo resources generated from other Kubernetes
+// resources (generally Service).
+// The `~` is an invalid character that prevents these labels from ending up
+// on actual Kubernetes resources.
+const (
+	// KubeSourceResourceLabel indicates the kind of resource that the synthetic
+	// resource is based on.
+	KubeSourceResourceLabel = "~internal.solo.io/kubernetes-source-resource"
+	// KubeSourceResourceLabel indicates the original name of the resource that
+	// the synthetic resource is based on.
+	KubeNameLabel = "~internal.solo.io/kubernetes-name"
+	// KubeSourceResourceLabel indicates the original namespace of the resource
+	// that the synthetic resource is based on.
+	KubeNamespaceLabel = "~internal.solo.io/kubernetes-namespace"
+	// KubeSourceResourceLabel indicates the service port when applicable.
+	KubeServicePortLabel = "~internal.solo.io/kubernetes-service-port"
+)
+
+// ClusterNameForKube builds the cluster name based on _internal_ labels.
+// All of the kind, name, namespace and port must be provided.
+func ClusterNameForKube(us *v1.Upstream) (string, bool) {
+	labels := us.GetMetadata().GetLabels()
+	kind, kok := labels[KubeSourceResourceLabel]
+	name, nok := labels[KubeNameLabel]
+	ns, nsok := labels[KubeNamespaceLabel]
+	port, pok := labels[KubeServicePortLabel]
+	if !(kok && nok && nsok && pok) {
+		return "", false
+	}
+	return fmt.Sprintf("%s_%s_%s_%s", kind, name, ns, port), true
+}
+
 type UpstreamConverter interface {
 	UpstreamsForService(ctx context.Context, svc *corev1.Service) v1.UpstreamList
 }
@@ -48,7 +82,15 @@ func (uc *KubeUpstreamConverter) CreateUpstream(ctx context.Context, svc *corev1
 	coremeta.ResourceVersion = ""
 	coremeta.Name = UpstreamName(meta.Namespace, meta.Name, port.Port)
 	labels := coremeta.GetLabels()
-	coremeta.Labels = make(map[string]string)
+	coremeta.Labels = map[string]string{
+		// preserve parts of the source service in a structured way
+		// so we don't rely on string parsing to recover these
+		// this is more extensible than relying on casting Spec to Upstream_Kube
+		KubeSourceResourceLabel: "kube-svc",
+		KubeNameLabel:           meta.Name,
+		KubeNamespaceLabel:      meta.Namespace,
+		KubeServicePortLabel:    strconv.Itoa(int(port.Port)),
+	}
 
 	us := &v1.Upstream{
 		Metadata: coremeta,
diff --git a/projects/gloo/pkg/syncer/setup/translator.go b/projects/gloo/pkg/syncer/setup/translator.go
index 03a29f82baf..bef76b24430 100644
--- a/projects/gloo/pkg/syncer/setup/translator.go
+++ b/projects/gloo/pkg/syncer/setup/translator.go
@@ -13,12 +13,13 @@ type TranslatorFactory struct {
 	PluginRegistry plugins.PluginRegistryFactory
 }
 
-func (tf TranslatorFactory) NewTranslator(ctx context.Context, settings *v1.Settings) translator.Translator {
+func (tf TranslatorFactory) NewTranslator(ctx context.Context, settings *v1.Settings, opts ...translator.Option) translator.Translator {
 	return translator.NewTranslatorWithHasher(
 		sslutils.NewSslConfigTranslator(),
 		settings,
 		tf.PluginRegistry(ctx),
 		translator.EnvoyCacheResourcesListToFnvHash,
+		opts...,
 	)
 }
 
diff --git a/projects/gloo/pkg/translator/clusters.go b/projects/gloo/pkg/translator/clusters.go
index 697ca6fccff..ed5ae3ab4ec 100644
--- a/projects/gloo/pkg/translator/clusters.go
+++ b/projects/gloo/pkg/translator/clusters.go
@@ -69,7 +69,7 @@ func (t *translatorInstance) computeClusters(
 }
 
 // This function is intented to be used when translating a single upstream outside of the context of a full snapshot.
-// This happens in GGv2 krt implementation.
+// This happens in the kube gateway krt implementation.
 func (t *translatorInstance) TranslateCluster(
 	params plugins.Params,
 	upstream *v1.Upstream,
@@ -138,9 +138,14 @@ func (t *translatorInstance) initializeCluster(
 		errorList = append(errorList, err)
 	}
 
+	clusterName := UpstreamToClusterName(upstream.GetMetadata().Ref())
+	if t.opts.kubeGatewayAPI {
+		clusterName = KubeGatewayUpstreamToClusterName(upstream)
+	}
+
 	circuitBreakers := t.settings.GetGloo().GetCircuitBreakers()
 	out := &envoy_config_cluster_v3.Cluster{
-		Name:             UpstreamToClusterName(upstream.GetMetadata().Ref()),
+		Name:             clusterName,
 		Metadata:         new(envoy_config_core_v3.Metadata),
 		CircuitBreakers:  getCircuitBreakers(upstream.GetCircuitBreakers(), circuitBreakers),
 		LbSubsetConfig:   createLbConfig(upstream),
diff --git a/projects/gloo/pkg/translator/translator.go b/projects/gloo/pkg/translator/translator.go
index 838c28d01bb..aeb1fd54200 100644
--- a/projects/gloo/pkg/translator/translator.go
+++ b/projects/gloo/pkg/translator/translator.go
@@ -54,12 +54,11 @@ type ClusterTranslator interface {
 	) (*envoy_config_cluster_v3.Cluster, []error)
 }
 
-var (
-	_ Translator = new(translatorInstance)
-)
+var _ Translator = new(translatorInstance)
 
 // translatorInstance is the implementation for a Translator used during Gloo translation
 type translatorInstance struct {
+	opts                        translatorOpts
 	lock                        sync.Mutex
 	pluginRegistry              plugins.PluginRegistry
 	settings                    *v1.Settings
@@ -68,8 +67,17 @@ type translatorInstance struct {
 	shouldEnforceNamespaceMatch bool
 }
 
-func NewDefaultTranslator(settings *v1.Settings, pluginRegistry plugins.PluginRegistry) *translatorInstance {
-	return NewTranslatorWithHasher(utils.NewSslConfigTranslator(), settings, pluginRegistry, EnvoyCacheResourcesListToFnvHash)
+type Option func(o *translatorOpts)
+
+func ForKubeGatewayAPI() Option {
+	return func(o *translatorOpts) {
+		o.kubeGatewayAPI = true
+	}
+}
+
+type translatorOpts struct {
+	// kubeGatewayAPI should only be set for translating kubeGatewayAPI proxies.
+	kubeGatewayAPI bool
 }
 
 func NewTranslatorWithHasher(
@@ -77,6 +85,7 @@ func NewTranslatorWithHasher(
 	settings *v1.Settings,
 	pluginRegistry plugins.PluginRegistry,
 	hasher func(resources []envoycache.Resource) (uint64, error),
+	opts ...Option,
 ) *translatorInstance {
 	shouldEnforceStr := os.Getenv(api_conversion.MatchingNamespaceEnv)
 	shouldEnforceNamespaceMatch := false
@@ -87,6 +96,12 @@ func NewTranslatorWithHasher(
 			// TODO: what to do here?
 		}
 	}
+
+	o := translatorOpts{}
+	for _, opt := range opts {
+		opt(&o)
+	}
+
 	return &translatorInstance{
 		lock:                        sync.Mutex{},
 		pluginRegistry:              pluginRegistry,
@@ -94,6 +109,7 @@ func NewTranslatorWithHasher(
 		hasher:                      hasher,
 		listenerTranslatorFactory:   NewListenerSubsystemTranslatorFactory(pluginRegistry, sslConfigTranslator, settings),
 		shouldEnforceNamespaceMatch: shouldEnforceNamespaceMatch,
+		opts:                        o,
 	}
 }
 
diff --git a/projects/gloo/pkg/translator/utils.go b/projects/gloo/pkg/translator/utils.go
index 27a5914bff7..e394bb83637 100644
--- a/projects/gloo/pkg/translator/utils.go
+++ b/projects/gloo/pkg/translator/utils.go
@@ -12,13 +12,27 @@ import (
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/any"
+	"github.com/solo-io/gloo/pkg/utils/envutils"
+	"github.com/solo-io/gloo/projects/gloo/constants"
+	v1 "github.com/solo-io/gloo/projects/gloo/pkg/api/v1"
+	"github.com/solo-io/gloo/projects/gloo/pkg/plugins/kubernetes"
 	"github.com/solo-io/gloo/projects/gloo/pkg/utils"
 	"github.com/solo-io/solo-kit/pkg/api/v1/resources/core"
 )
 
 // returns the name of the cluster created for a given upstream
-func UpstreamToClusterName(upstream *core.ResourceRef) string {
+// used only for kube gateway api (aka ggv2) proxies.
+func KubeGatewayUpstreamToClusterName(upstream *v1.Upstream) string {
+	legacyClusterNames := envutils.IsEnvTruthy(constants.GlooGatewayKubeStyleClusterNames)
+	clusterName, ok := kubernetes.ClusterNameForKube(upstream)
+	if !ok || legacyClusterNames {
+		return UpstreamToClusterName(upstream.GetMetadata().Ref())
+	}
+	return clusterName
+}
 
+// returns the name of the cluster created for a given upstream
+func UpstreamToClusterName(upstream *core.ResourceRef) string {
 	// For non-namespaced resources, return only name
 	if upstream.GetNamespace() == "" {
 		return upstream.GetName()
@@ -30,7 +44,6 @@ func UpstreamToClusterName(upstream *core.ResourceRef) string {
 
 // returns the ref of the upstream for a given cluster
 func ClusterToUpstreamRef(cluster string) (*core.ResourceRef, error) {
-
 	split := strings.Split(cluster, "_")
 	if len(split) > 2 || len(split) < 1 {
 		return nil, errors.Errorf("unable to convert cluster %s back to upstream ref", cluster)
@@ -47,7 +60,6 @@ func ClusterToUpstreamRef(cluster string) (*core.ResourceRef, error) {
 }
 
 func NewFilterWithTypedConfig(name string, config proto.Message) (*envoy_config_listener_v3.Filter, error) {
-
 	s := &envoy_config_listener_v3.Filter{
 		Name: name,
 	}
@@ -107,14 +119,12 @@ func IsIpv4Address(bindAddress string) (validIpv4, strictIPv4 bool, err error) {
 	if bindIP == nil {
 		// If bindAddress is not a valid textual representation of an IP address
 		return false, false, errors.Errorf("bindAddress %s is not a valid IP address", bindAddress)
-
 	} else if bindIP.To4() == nil {
 		// If bindIP is not an IPv4 address, To4 returns nil.
 		// so this is not an acceptable ipv4
 		return false, false, nil
 	}
 	return true, isPureIPv4Address(bindAddress), nil
-
 }
 
 // isPureIPv4Address checks the string to see if it is