diff --git a/test-integ/peering_commontopo/ac5_2_pq_failover_test.go b/test-integ/peering_commontopo/ac5_2_pq_failover_test.go
index ff4a8bea32..3bf8c9be9a 100644
--- a/test-integ/peering_commontopo/ac5_2_pq_failover_test.go
+++ b/test-integ/peering_commontopo/ac5_2_pq_failover_test.go
@@ -86,7 +86,10 @@ func (s *ac5_2PQFailoverSuite) setupDC(ct *commonTopo, clu, peerClu *topology.Cl
 		Service: NewFortioServiceWithDefaults(
 			clu.Datacenter,
 			serverSID,
-			nil,
+			func(s *topology.Service) {
+				s.EnvoyAdminPort = 0
+				s.DisableServiceMesh = true
+			},
 		),
 		Exports: []api.ServiceConsumer{{Peer: peer}},
 	}
@@ -149,7 +152,10 @@ func (s *ac5_2PQFailoverSuite) setupDC3(ct *commonTopo, clu, peer1, peer2 *topol
 		Service: NewFortioServiceWithDefaults(
 			clu.Datacenter,
 			serverSID,
-			nil,
+			func(s *topology.Service) {
+				s.EnvoyAdminPort = 0
+				s.DisableServiceMesh = true
+			},
 		),
 		Exports: func() []api.ServiceConsumer {
 			var consumers []api.ServiceConsumer
diff --git a/test-integ/peering_commontopo/ac7_1_rotate_gw_test.go b/test-integ/peering_commontopo/ac7_1_rotate_gw_test.go
index b8da4e34bf..4973cb6d04 100644
--- a/test-integ/peering_commontopo/ac7_1_rotate_gw_test.go
+++ b/test-integ/peering_commontopo/ac7_1_rotate_gw_test.go
@@ -141,8 +141,12 @@ func (s *suiteRotateGW) setup(t *testing.T, ct *commonTopo) {
 
 	// add a second mesh gateway "new"
 	s.newMGWNodeName = fmt.Sprintf("new-%s-default-mgw", clu.Name)
+	nodeKind := topology.NodeKindClient
+	if clu.Datacenter == agentlessDC {
+		nodeKind = topology.NodeKindDataplane
+	}
 	clu.Nodes = append(clu.Nodes, newTopologyMeshGatewaySet(
-		topology.NodeKindClient,
+		nodeKind,
 		"default",
 		s.newMGWNodeName,
 		1,
diff --git a/test-integ/peering_commontopo/asserter.go b/test-integ/peering_commontopo/asserter.go
index 1fa1b81ef1..b5aa71ebae 100644
--- a/test-integ/peering_commontopo/asserter.go
+++ b/test-integ/peering_commontopo/asserter.go
@@ -41,6 +41,7 @@ type asserter struct {
 type sprawlLite interface {
 	HTTPClientForCluster(clusterName string) (*http.Client, error)
 	APIClientForNode(clusterName string, nid topology.NodeID, token string) (*api.Client, error)
+	APIClientForCluster(clusterName string, token string) (*api.Client, error)
 	Topology() *topology.Topology
 }
 
@@ -58,18 +59,12 @@ func (a *asserter) mustGetHTTPClient(t *testing.T, cluster string) *http.Client
 }
 
 func (a *asserter) mustGetAPIClient(t *testing.T, cluster string) *api.Client {
-	cl, err := a.apiClientFor(cluster)
+	clu := a.sp.Topology().Clusters[cluster]
+	cl, err := a.sp.APIClientForCluster(clu.Name, "")
 	require.NoError(t, err)
 	return cl
 }
 
-func (a *asserter) apiClientFor(cluster string) (*api.Client, error) {
-	clu := a.sp.Topology().Clusters[cluster]
-	// TODO: this always goes to the first client, but we might want to balance this
-	cl, err := a.sp.APIClientForNode(cluster, clu.FirstClient().ID(), "")
-	return cl, err
-}
-
 // httpClientFor returns a pre-configured http.Client that proxies requests
 // through the embedded squid instance in each LAN.
 //
diff --git a/test-integ/peering_commontopo/commontopo.go b/test-integ/peering_commontopo/commontopo.go
index d0e2c8e55d..c53dbb0ccb 100644
--- a/test-integ/peering_commontopo/commontopo.go
+++ b/test-integ/peering_commontopo/commontopo.go
@@ -52,6 +52,8 @@ type commonTopo struct {
 	services map[string]map[topology.ServiceID]struct{}
 }
 
+const agentlessDC = "dc2"
+
 func NewCommonTopo(t *testing.T) *commonTopo {
 	t.Helper()
 
@@ -84,11 +86,9 @@ func NewCommonTopo(t *testing.T) *commonTopo {
 	peerings = append(peerings, addPeerings(dc1, dc3)...)
 	peerings = append(peerings, addPeerings(dc2, dc3)...)
 
-	addMeshGateways(dc1, topology.NodeKindClient)
-	addMeshGateways(dc2, topology.NodeKindClient)
-	addMeshGateways(dc3, topology.NodeKindClient)
-	// TODO: consul-topology doesn't support this yet
-	// addMeshGateways(dc2, topology.NodeKindDataplane)
+	addMeshGateways(dc1)
+	addMeshGateways(dc2)
+	addMeshGateways(dc3)
 
 	setupGlobals(dc1)
 	setupGlobals(dc2)
@@ -131,7 +131,7 @@ func (ct *commonTopo) postLaunchChecks(t *testing.T) {
 	)
 
 	// check that exports line up as expected
-	for _, clu := range ct.Sprawl.Config().Clusters {
+	for _, clu := range ct.Sprawl.Topology().Clusters {
 		// expected exports per peer
 		type key struct {
 			peer      string
@@ -191,9 +191,6 @@ func LocalPeerName(clu *topology.Cluster, partition string) string {
 type serviceExt struct {
 	*topology.Service
 
-	// default NodeKindClient
-	NodeKind topology.NodeKind
-
 	Exports    []api.ServiceConsumer
 	Config     *api.ServiceConfigEntry
 	Intentions *api.ServiceIntentionsConfigEntry
@@ -227,8 +224,15 @@ func (ct *commonTopo) AddServiceNode(clu *topology.Cluster, svc serviceExt) *top
 		return n
 	}
 
+	nodeKind := topology.NodeKindClient
+	// TODO: bug in deployer somewhere; it should guard against a KindDataplane node with
+	// DisableServiceMesh services on it; dataplane is only for service-mesh
+	if !svc.DisableServiceMesh && clu.Datacenter == agentlessDC {
+		nodeKind = topology.NodeKindDataplane
+	}
+
 	node := &topology.Node{
-		Kind:      topology.NodeKindClient,
+		Kind:      nodeKind,
 		Name:      serviceHostnameString(clu.Datacenter, svc.ID),
 		Partition: svc.ID.Partition,
 		Addresses: []*topology.Address{
@@ -239,9 +243,6 @@ func (ct *commonTopo) AddServiceNode(clu *topology.Cluster, svc serviceExt) *top
 		},
 		Cluster: clusterName,
 	}
-	if svc.NodeKind != "" {
-		node.Kind = svc.NodeKind
-	}
 	clu.Nodes = append(clu.Nodes, node)
 
 	// Export if necessary
@@ -265,7 +266,7 @@ func (ct *commonTopo) AddServiceNode(clu *topology.Cluster, svc serviceExt) *top
 }
 
 func (ct *commonTopo) APIClientForCluster(t *testing.T, clu *topology.Cluster) *api.Client {
-	cl, err := ct.Sprawl.APIClientForNode(clu.Name, clu.FirstClient().ID(), "")
+	cl, err := ct.Sprawl.APIClientForCluster(clu.Name, "")
 	require.NoError(t, err)
 	return cl
 }
@@ -372,10 +373,14 @@ func setupGlobals(clu *topology.Cluster) {
 
 // addMeshGateways adds a mesh gateway for every partition in the cluster.
 // Assumes that the LAN network name is equal to datacenter name.
-func addMeshGateways(c *topology.Cluster, kind topology.NodeKind) {
+func addMeshGateways(c *topology.Cluster) {
+	nodeKind := topology.NodeKindClient
+	if c.Datacenter == agentlessDC {
+		nodeKind = topology.NodeKindDataplane
+	}
 	for _, p := range c.Partitions {
 		c.Nodes = topology.MergeSlices(c.Nodes, newTopologyMeshGatewaySet(
-			kind,
+			nodeKind,
 			p.Name,
 			fmt.Sprintf("%s-%s-mgw", c.Name, p.Name),
 			1,
diff --git a/testing/deployer/sprawl/catalog.go b/testing/deployer/sprawl/catalog.go
index eb355739a6..6c924f019b 100644
--- a/testing/deployer/sprawl/catalog.go
+++ b/testing/deployer/sprawl/catalog.go
@@ -318,6 +318,18 @@ func serviceToCatalogRegistration(
 			Address: node.LocalAddress(),
 		},
 	}
+	if svc.IsMeshGateway {
+		reg.Service.Kind = api.ServiceKindMeshGateway
+		reg.Service.Proxy = &api.AgentServiceConnectProxyConfig{
+			Config: map[string]interface{}{
+				"envoy_gateway_no_default_bind":       true,
+				"envoy_gateway_bind_tagged_addresses": true,
+			},
+			MeshGateway: api.MeshGatewayConfig{
+				Mode: api.MeshGatewayModeLocal,
+			},
+		}
+	}
 	if node.HasPublicAddress() {
 		reg.TaggedAddresses = map[string]string{
 			"lan":      node.LocalAddress(),
@@ -325,6 +337,26 @@ func serviceToCatalogRegistration(
 			"wan":      node.PublicAddress(),
 			"wan_ipv4": node.PublicAddress(),
 		}
+		// TODO: not sure what the difference is between these, but with just the
+		// top-level set, it appeared to not get set in either :/
+		reg.Service.TaggedAddresses = map[string]api.ServiceAddress{
+			"lan": {
+				Address: node.LocalAddress(),
+				Port:    svc.Port,
+			},
+			"lan_ipv4": {
+				Address: node.LocalAddress(),
+				Port:    svc.Port,
+			},
+			"wan": {
+				Address: node.PublicAddress(),
+				Port:    svc.Port,
+			},
+			"wan_ipv4": {
+				Address: node.PublicAddress(),
+				Port:    svc.Port,
+			},
+		}
 	}
 	if cluster.Enterprise {
 		reg.Partition = svc.ID.Partition
diff --git a/testing/deployer/sprawl/internal/build/docker.go b/testing/deployer/sprawl/internal/build/docker.go
index 7007d44c84..56ec520572 100644
--- a/testing/deployer/sprawl/internal/build/docker.go
+++ b/testing/deployer/sprawl/internal/build/docker.go
@@ -23,9 +23,10 @@ COPY --from=0 /bin/consul /bin/consul
 
 // FROM hashicorp/consul-dataplane:latest
 // COPY --from=busybox:uclibc /bin/sh /bin/sh
+// TODO: busybox:latest doesn't work, see https://hashicorp.slack.com/archives/C03EUN3QF1C/p1691784078972959
 const dockerfileDataplane = `
 ARG DATAPLANE_IMAGE
-FROM busybox:latest
+FROM busybox:1.34
 FROM ${DATAPLANE_IMAGE}
 COPY --from=0 /bin/busybox /bin/busybox
 USER 0:0
diff --git a/testing/deployer/sprawl/internal/tfgen/agent.go b/testing/deployer/sprawl/internal/tfgen/agent.go
index d884d200d2..00e7276783 100644
--- a/testing/deployer/sprawl/internal/tfgen/agent.go
+++ b/testing/deployer/sprawl/internal/tfgen/agent.go
@@ -13,14 +13,14 @@ import (
 	"github.com/hashicorp/consul/testing/deployer/topology"
 )
 
-func (g *Generator) generateAgentHCL(node *topology.Node) (string, error) {
+func (g *Generator) generateAgentHCL(node *topology.Node) string {
 	if !node.IsAgent() {
-		return "", fmt.Errorf("not an agent")
+		panic("generateAgentHCL only applies to agents")
 	}
 
 	cluster, ok := g.topology.Clusters[node.Cluster]
 	if !ok {
-		return "", fmt.Errorf("no such cluster: %s", node.Cluster)
+		panic(fmt.Sprintf("no such cluster: %s", node.Cluster))
 	}
 
 	var b HCLBuilder
@@ -167,7 +167,7 @@ func (g *Generator) generateAgentHCL(node *topology.Node) (string, error) {
 		}
 	}
 
-	return b.String(), nil
+	return b.String()
 }
 
 type HCLBuilder struct {
diff --git a/testing/deployer/sprawl/internal/tfgen/dns.go b/testing/deployer/sprawl/internal/tfgen/dns.go
index 6c3e5ca624..20dca878eb 100644
--- a/testing/deployer/sprawl/internal/tfgen/dns.go
+++ b/testing/deployer/sprawl/internal/tfgen/dns.go
@@ -9,7 +9,6 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"text/template"
 
 	"github.com/hashicorp/consul/testing/deployer/topology"
 	"github.com/hashicorp/consul/testing/deployer/util"
@@ -179,5 +178,3 @@ server IN A %s ; Consul server
 
 	return buf.Bytes()
 }
-
-var tfCorednsT = template.Must(template.ParseFS(content, "templates/container-coredns.tf.tmpl"))
diff --git a/testing/deployer/sprawl/internal/tfgen/docker.go b/testing/deployer/sprawl/internal/tfgen/docker.go
index bc8bd65756..63f628c15a 100644
--- a/testing/deployer/sprawl/internal/tfgen/docker.go
+++ b/testing/deployer/sprawl/internal/tfgen/docker.go
@@ -11,6 +11,9 @@ import (
 var invalidResourceName = regexp.MustCompile(`[^a-z0-9-]+`)
 
 func DockerImageResourceName(image string) string {
+	if image == "" {
+		panic(`image must not be ""`)
+	}
 	return invalidResourceName.ReplaceAllLiteralString(image, "-")
 }
 
diff --git a/testing/deployer/sprawl/internal/tfgen/nodes.go b/testing/deployer/sprawl/internal/tfgen/nodes.go
index 7b7addfb20..33a820a3e3 100644
--- a/testing/deployer/sprawl/internal/tfgen/nodes.go
+++ b/testing/deployer/sprawl/internal/tfgen/nodes.go
@@ -5,9 +5,6 @@ package tfgen
 
 import (
 	"fmt"
-	"sort"
-	"strconv"
-	"text/template"
 
 	"github.com/hashicorp/consul/testing/deployer/topology"
 )
@@ -22,32 +19,6 @@ type terraformPod struct {
 	DockerNetworkName string
 }
 
-type terraformConsulAgent struct {
-	terraformPod
-	ImageResource     string
-	HCL               string
-	EnterpriseLicense string
-	Env               []string
-}
-
-type terraformMeshGatewayService struct {
-	terraformPod
-	EnvoyImageResource string
-	Service            *topology.Service
-	Command            []string
-}
-
-type terraformService struct {
-	terraformPod
-	AppImageResource       string
-	EnvoyImageResource     string // agentful
-	DataplaneImageResource string // agentless
-	Service                *topology.Service
-	Env                    []string
-	Command                []string
-	EnvoyCommand           []string // agentful
-}
-
 func (g *Generator) generateNodeContainers(
 	step Step,
 	cluster *topology.Cluster,
@@ -82,156 +53,99 @@ func (g *Generator) generateNodeContainers(
 	}
 	pod.DockerNetworkName = net.DockerName
 
-	var (
-		containers []Resource
-	)
+	containers := []Resource{}
 
 	if node.IsAgent() {
-		agentHCL, err := g.generateAgentHCL(node)
-		if err != nil {
-			return nil, err
-		}
-
-		agent := terraformConsulAgent{
-			terraformPod:      pod,
-			ImageResource:     DockerImageResourceName(node.Images.Consul),
-			HCL:               agentHCL,
-			EnterpriseLicense: g.license,
-			Env:               node.AgentEnv,
-		}
-
 		switch {
 		case node.IsServer() && step.StartServers(),
 			!node.IsServer() && step.StartAgents():
-			containers = append(containers, Eval(tfConsulT, &agent))
+			containers = append(containers, Eval(tfConsulT, struct {
+				terraformPod
+				ImageResource     string
+				HCL               string
+				EnterpriseLicense string
+			}{
+				terraformPod:      pod,
+				ImageResource:     DockerImageResourceName(node.Images.Consul),
+				HCL:               g.generateAgentHCL(node),
+				EnterpriseLicense: g.license,
+			}))
 		}
 	}
 
+	svcContainers := []Resource{}
 	for _, svc := range node.SortedServices() {
-		if svc.IsMeshGateway {
-			if node.Kind == topology.NodeKindDataplane {
-				panic("NOT READY YET")
-			}
-			gw := terraformMeshGatewayService{
-				terraformPod:       pod,
-				EnvoyImageResource: DockerImageResourceName(node.Images.EnvoyConsulImage()),
-				Service:            svc,
-				Command: []string{
-					"consul", "connect", "envoy",
-					"-register",
-					"-mesh-gateway",
-				},
-			}
-			if token := g.sec.ReadServiceToken(node.Cluster, svc.ID); token != "" {
-				gw.Command = append(gw.Command, "-token", token)
-			}
-			if cluster.Enterprise {
-				gw.Command = append(gw.Command,
-					"-partition",
-					svc.ID.Partition,
-				)
-			}
-			gw.Command = append(gw.Command,
-				"-address",
-				`{{ GetInterfaceIP \"eth0\" }}:`+strconv.Itoa(svc.Port),
-				"-wan-address",
-				`{{ GetInterfaceIP \"eth1\" }}:`+strconv.Itoa(svc.Port),
-			)
-			gw.Command = append(gw.Command,
-				"-grpc-addr", "http://127.0.0.1:8502",
-				"-admin-bind",
-				// for demo purposes
-				"0.0.0.0:"+strconv.Itoa(svc.EnvoyAdminPort),
-				"--",
-				"-l",
-				"trace",
-			)
-			if step.StartServices() {
-				containers = append(containers, Eval(tfMeshGatewayT, &gw))
-			}
-		} else {
-			tfsvc := terraformService{
-				terraformPod:     pod,
-				AppImageResource: DockerImageResourceName(svc.Image),
-				Service:          svc,
-				Command:          svc.Command,
-			}
-			tfsvc.Env = append(tfsvc.Env, svc.Env...)
-			if step.StartServices() {
-				containers = append(containers, Eval(tfAppT, &tfsvc))
+		token := g.sec.ReadServiceToken(node.Cluster, svc.ID)
+		switch {
+		case svc.IsMeshGateway && !node.IsDataplane():
+			svcContainers = append(svcContainers, Eval(tfMeshGatewayT, struct {
+				terraformPod
+				ImageResource string
+				Enterprise    bool
+				Service       *topology.Service
+				Token         string
+			}{
+				terraformPod:  pod,
+				ImageResource: DockerImageResourceName(node.Images.EnvoyConsulImage()),
+				Enterprise:    cluster.Enterprise,
+				Service:       svc,
+				Token:         token,
+			}))
+		case svc.IsMeshGateway && node.IsDataplane():
+			svcContainers = append(svcContainers, Eval(tfMeshGatewayDataplaneT, &struct {
+				terraformPod
+				ImageResource string
+				Enterprise    bool
+				Service       *topology.Service
+				Token         string
+			}{
+				terraformPod:  pod,
+				ImageResource: DockerImageResourceName(node.Images.LocalDataplaneImage()),
+				Enterprise:    cluster.Enterprise,
+				Service:       svc,
+				Token:         token,
+			}))
+
+		case !svc.IsMeshGateway:
+			svcContainers = append(svcContainers, Eval(tfAppT, struct {
+				terraformPod
+				ImageResource string
+				Service       *topology.Service
+			}{
+				terraformPod:  pod,
+				ImageResource: DockerImageResourceName(svc.Image),
+				Service:       svc,
+			}))
+
+			if svc.DisableServiceMesh {
+				break
 			}
 
-			setenv := func(k, v string) {
-				tfsvc.Env = append(tfsvc.Env, k+"="+v)
+			tmpl := tfAppSidecarT
+			var img string
+			if node.IsDataplane() {
+				tmpl = tfAppDataplaneT
+				img = DockerImageResourceName(node.Images.LocalDataplaneImage())
+			} else {
+				img = DockerImageResourceName(node.Images.EnvoyConsulImage())
 			}
+			svcContainers = append(svcContainers, Eval(tmpl, struct {
+				terraformPod
+				ImageResource string
+				Service       *topology.Service
+				Token         string
+				Enterprise    bool
+			}{
+				terraformPod:  pod,
+				ImageResource: img,
+				Service:       svc,
+				Token:         token,
+				Enterprise:    cluster.Enterprise,
+			}))
+		}
 
-			if !svc.DisableServiceMesh {
-				if node.IsDataplane() {
-					tfsvc.DataplaneImageResource = DockerImageResourceName(node.Images.LocalDataplaneImage())
-					tfsvc.EnvoyImageResource = ""
-					tfsvc.EnvoyCommand = nil
-					// --- REQUIRED ---
-					setenv("DP_CONSUL_ADDRESSES", "server."+node.Cluster+"-consulcluster.lan")
-					setenv("DP_SERVICE_NODE_NAME", node.PodName())
-					setenv("DP_PROXY_SERVICE_ID", svc.ID.Name+"-sidecar-proxy")
-				} else {
-					tfsvc.DataplaneImageResource = ""
-					tfsvc.EnvoyImageResource = DockerImageResourceName(node.Images.EnvoyConsulImage())
-					tfsvc.EnvoyCommand = []string{
-						"consul", "connect", "envoy",
-						"-sidecar-for", svc.ID.Name,
-					}
-				}
-				if cluster.Enterprise {
-					if node.IsDataplane() {
-						setenv("DP_SERVICE_NAMESPACE", svc.ID.Namespace)
-						setenv("DP_SERVICE_PARTITION", svc.ID.Partition)
-					} else {
-						tfsvc.EnvoyCommand = append(tfsvc.EnvoyCommand,
-							"-partition",
-							svc.ID.Partition,
-							"-namespace",
-							svc.ID.Namespace,
-						)
-					}
-				}
-				if token := g.sec.ReadServiceToken(node.Cluster, svc.ID); token != "" {
-					if node.IsDataplane() {
-						setenv("DP_CREDENTIAL_TYPE", "static")
-						setenv("DP_CREDENTIAL_STATIC_TOKEN", token)
-					} else {
-						tfsvc.EnvoyCommand = append(tfsvc.EnvoyCommand, "-token", token)
-					}
-				}
-				if node.IsDataplane() {
-					setenv("DP_ENVOY_ADMIN_BIND_ADDRESS", "0.0.0.0") // for demo purposes
-					setenv("DP_ENVOY_ADMIN_BIND_PORT", "19000")
-					setenv("DP_LOG_LEVEL", "trace")
-
-					setenv("DP_CA_CERTS", "/consul/config/certs/consul-agent-ca.pem")
-					setenv("DP_CONSUL_GRPC_PORT", "8503")
-					setenv("DP_TLS_SERVER_NAME", "server."+node.Datacenter+".consul")
-				} else {
-					tfsvc.EnvoyCommand = append(tfsvc.EnvoyCommand,
-						"-grpc-addr", "http://127.0.0.1:8502",
-						"-admin-bind",
-						// for demo purposes
-						"0.0.0.0:"+strconv.Itoa(svc.EnvoyAdminPort),
-						"--",
-						"-l",
-						"trace",
-					)
-				}
-				if step.StartServices() {
-					sort.Strings(tfsvc.Env)
-
-					if node.IsDataplane() {
-						containers = append(containers, Eval(tfAppDataplaneT, &tfsvc))
-					} else {
-						containers = append(containers, Eval(tfAppSidecarT, &tfsvc))
-					}
-				}
-			}
+		if step.StartServices() {
+			containers = append(containers, svcContainers...)
 		}
 	}
 
@@ -243,10 +157,3 @@ func (g *Generator) generateNodeContainers(
 
 	return containers, nil
 }
-
-var tfPauseT = template.Must(template.ParseFS(content, "templates/container-pause.tf.tmpl"))
-var tfConsulT = template.Must(template.ParseFS(content, "templates/container-consul.tf.tmpl"))
-var tfMeshGatewayT = template.Must(template.ParseFS(content, "templates/container-mgw.tf.tmpl"))
-var tfAppT = template.Must(template.ParseFS(content, "templates/container-app.tf.tmpl"))
-var tfAppSidecarT = template.Must(template.ParseFS(content, "templates/container-app-sidecar.tf.tmpl"))
-var tfAppDataplaneT = template.Must(template.ParseFS(content, "templates/container-app-dataplane.tf.tmpl"))
diff --git a/testing/deployer/sprawl/internal/tfgen/proxy.go b/testing/deployer/sprawl/internal/tfgen/proxy.go
index 0fafa9b925..0fcaf841ee 100644
--- a/testing/deployer/sprawl/internal/tfgen/proxy.go
+++ b/testing/deployer/sprawl/internal/tfgen/proxy.go
@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"text/template"
 
 	"github.com/hashicorp/consul/testing/deployer/topology"
 	"github.com/hashicorp/consul/testing/deployer/util"
@@ -86,5 +85,3 @@ func (g *Generator) getForwardProxyContainer(
 
 	return Eval(tfForwardProxyT, &proxy)
 }
-
-var tfForwardProxyT = template.Must(template.ParseFS(content, "templates/container-proxy.tf.tmpl"))
diff --git a/testing/deployer/sprawl/internal/tfgen/templates/container-app-dataplane.tf.tmpl b/testing/deployer/sprawl/internal/tfgen/templates/container-app-dataplane.tf.tmpl
index bfb0705e6d..040cabbf11 100644
--- a/testing/deployer/sprawl/internal/tfgen/templates/container-app-dataplane.tf.tmpl
+++ b/testing/deployer/sprawl/internal/tfgen/templates/container-app-dataplane.tf.tmpl
@@ -1,8 +1,8 @@
 resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}-sidecar" {
 	name = "{{.Node.DockerName}}-{{.Service.ID.TFString}}-sidecar"
     network_mode = "container:${docker_container.{{.PodName}}.id}"
-    image        = docker_image.{{.DataplaneImageResource}}.latest
-    restart  = "on-failure"
+    image        = docker_image.{{.ImageResource}}.latest
+    restart      = "on-failure"
 
 {{- range $k, $v := .Labels }}
   labels {
@@ -18,9 +18,24 @@ resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}-sidec
   }
 
   env = [
-{{- range .Env }}
-      "{{.}}",
-{{- end}}
+			"DP_CONSUL_ADDRESSES=server.{{.Node.Cluster}}-consulcluster.lan",
+			"DP_SERVICE_NODE_NAME={{.Node.PodName}}",
+			"DP_PROXY_SERVICE_ID={{.Service.ID.Name}}-sidecar-proxy",
+    {{ if .Enterprise }}
+      "DP_SERVICE_NAMESPACE={{.Service.ID.Namespace}}",
+      "DP_SERVICE_PARTITION={{.Service.ID.Partition}}",
+    {{ end }}
+    {{ if .Token }}
+      "DP_CREDENTIAL_TYPE=static",
+      "DP_CREDENTIAL_STATIC_TOKEN={{.Token}}",
+    {{ end }}
+      // for demo purposes
+			"DP_ENVOY_ADMIN_BIND_ADDRESS=0.0.0.0",
+			"DP_ENVOY_ADMIN_BIND_PORT=19000",
+			"DP_LOG_LEVEL=trace",
+			"DP_CA_CERTS=/consul/config/certs/consul-agent-ca.pem",
+			"DP_CONSUL_GRPC_PORT=8503",
+			"DP_TLS_SERVER_NAME=server.{{.Node.Datacenter}}.consul",
   ]
 
   command = [
diff --git a/testing/deployer/sprawl/internal/tfgen/templates/container-app-sidecar.tf.tmpl b/testing/deployer/sprawl/internal/tfgen/templates/container-app-sidecar.tf.tmpl
index 6abb397c25..15a5f69226 100644
--- a/testing/deployer/sprawl/internal/tfgen/templates/container-app-sidecar.tf.tmpl
+++ b/testing/deployer/sprawl/internal/tfgen/templates/container-app-sidecar.tf.tmpl
@@ -1,7 +1,7 @@
 resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}-sidecar" {
 	name = "{{.Node.DockerName}}-{{.Service.ID.TFString}}-sidecar"
     network_mode = "container:${docker_container.{{.PodName}}.id}"
-    image        = docker_image.{{.EnvoyImageResource}}.latest
+    image        = docker_image.{{.ImageResource}}.latest
     restart  = "on-failure"
 
 {{- range $k, $v := .Labels }}
@@ -17,15 +17,21 @@ resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}-sidec
     read_only      = true
   }
 
-  env = [
-{{- range .Env }}
-      "{{.}}",
-{{- end}}
-  ]
-
   command = [
-{{- range .EnvoyCommand }}
-    "{{.}}",
-{{- end }}
+    "consul", "connect", "envoy",
+    "-sidecar-for={{.Service.ID.Name}}",
+    "-grpc-addr=http://127.0.0.1:8502",
+    // for demo purposes (TODO: huh?)
+    "-admin-bind=0.0.0.0:{{.Service.EnvoyAdminPort}}",
+  {{if .Enterprise}} 
+    "-partition={{.Service.ID.Partition}}",
+    "-namespace={{.Service.ID.Namespace}}",
+  {{end}}
+  {{if .Token }}
+    "-token={{.Token}}",
+  {{end}}
+    "--",
+    "-l",
+    "trace",
   ]
 }
diff --git a/testing/deployer/sprawl/internal/tfgen/templates/container-app.tf.tmpl b/testing/deployer/sprawl/internal/tfgen/templates/container-app.tf.tmpl
index b1b390f0f1..d6033587b1 100644
--- a/testing/deployer/sprawl/internal/tfgen/templates/container-app.tf.tmpl
+++ b/testing/deployer/sprawl/internal/tfgen/templates/container-app.tf.tmpl
@@ -1,7 +1,7 @@
 resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}" {
 	name = "{{.Node.DockerName}}-{{.Service.ID.TFString}}"
     network_mode = "container:${docker_container.{{.PodName}}.id}"
-    image        = docker_image.{{.AppImageResource}}.latest
+    image        = docker_image.{{.ImageResource}}.latest
     restart  = "on-failure"
 
 {{- range $k, $v := .Labels }}
@@ -12,13 +12,13 @@ resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}" {
 {{- end }}
 
   env = [
-{{- range .Env }}
+{{- range .Service.Env }}
       "{{.}}",
 {{- end}}
   ]
 
   command = [
-{{- range .Command }}
+{{- range .Service.Command }}
     "{{.}}",
 {{- end }}
   ]
diff --git a/testing/deployer/sprawl/internal/tfgen/templates/container-consul.tf.tmpl b/testing/deployer/sprawl/internal/tfgen/templates/container-consul.tf.tmpl
index 01f7f3fb4d..fac148f29c 100644
--- a/testing/deployer/sprawl/internal/tfgen/templates/container-consul.tf.tmpl
+++ b/testing/deployer/sprawl/internal/tfgen/templates/container-consul.tf.tmpl
@@ -8,9 +8,6 @@ resource "docker_container" "{{.Node.DockerName}}" {
       "CONSUL_UID=0",
       "CONSUL_GID=0",
       "CONSUL_LICENSE={{.EnterpriseLicense}}",
-{{- range .Env }}
-      "{{.}}",
-{{- end}}
   ]
 
 {{- range $k, $v := .Labels }}
diff --git a/testing/deployer/sprawl/internal/tfgen/templates/container-mgw-dataplane.tf.tmpl b/testing/deployer/sprawl/internal/tfgen/templates/container-mgw-dataplane.tf.tmpl
new file mode 100644
index 0000000000..18152e68eb
--- /dev/null
+++ b/testing/deployer/sprawl/internal/tfgen/templates/container-mgw-dataplane.tf.tmpl
@@ -0,0 +1,45 @@
+resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}" {
+	name = "{{.Node.DockerName}}-{{.Service.ID.TFString}}"
+    network_mode = "container:${docker_container.{{.PodName}}.id}"
+    image        = docker_image.{{.ImageResource}}.latest
+    restart      = "on-failure"
+
+{{- range $k, $v := .Labels }}
+  labels {
+    label = "{{ $k }}"
+    value = "{{ $v }}"
+  }
+{{- end }}
+
+  volumes {
+    volume_name    = "{{.TLSVolumeName}}"
+    container_path = "/consul/config/certs"
+    read_only      = true
+  }
+
+  env = [
+			"DP_CONSUL_ADDRESSES=server.{{.Node.Cluster}}-consulcluster.lan",
+			"DP_SERVICE_NODE_NAME={{.Node.PodName}}",
+			"DP_PROXY_SERVICE_ID={{.Service.ID.Name}}",
+    {{ if .Enterprise }}
+      "DP_SERVICE_NAMESPACE={{.Service.ID.Namespace}}",
+      "DP_SERVICE_PARTITION={{.Service.ID.Partition}}",
+    {{ end }}
+    {{ if .Token }}
+      "DP_CREDENTIAL_TYPE=static",
+      "DP_CREDENTIAL_STATIC_TOKEN={{.Token}}",
+    {{ end }}
+      // for demo purposes
+			"DP_ENVOY_ADMIN_BIND_ADDRESS=0.0.0.0",
+			"DP_ENVOY_ADMIN_BIND_PORT=19000",
+			"DP_LOG_LEVEL=trace",
+			"DP_CA_CERTS=/consul/config/certs/consul-agent-ca.pem",
+			"DP_CONSUL_GRPC_PORT=8503",
+			"DP_TLS_SERVER_NAME=server.{{.Node.Datacenter}}.consul",
+  ]
+
+  command = [
+    "/usr/local/bin/consul-dataplane",
+  ]
+}
+
diff --git a/testing/deployer/sprawl/internal/tfgen/templates/container-mgw.tf.tmpl b/testing/deployer/sprawl/internal/tfgen/templates/container-mgw.tf.tmpl
index ec25665f3e..78c4abe8df 100644
--- a/testing/deployer/sprawl/internal/tfgen/templates/container-mgw.tf.tmpl
+++ b/testing/deployer/sprawl/internal/tfgen/templates/container-mgw.tf.tmpl
@@ -1,7 +1,7 @@
 resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}" {
     name = "{{.Node.DockerName}}-{{.Service.ID.TFString}}"
     network_mode = "container:${docker_container.{{.PodName}}.id}"
-    image        = docker_image.{{.EnvoyImageResource}}.latest
+    image        = docker_image.{{.ImageResource}}.latest
     restart  = "on-failure"
 
 {{- range $k, $v := .Labels }}
@@ -18,8 +18,22 @@ resource "docker_container" "{{.Node.DockerName}}-{{.Service.ID.TFString}}" {
   }
 
   command = [
-{{- range .Command }}
-    "{{.}}",
-{{- end }}
+    "consul", "connect", "envoy",
+    "-register",
+    "-mesh-gateway",
+    "-address={{`{{ GetInterfaceIP \"eth0\" }}`}}:{{.Service.Port}}",
+    "-wan-address={{`{{ GetInterfaceIP \"eth1\" }}`}}:{{.Service.Port}}",
+    "-grpc-addr=http://127.0.0.1:8502",
+    // for demo purposes (TODO: huh?)
+    "-admin-bind=0.0.0.0:{{.Service.EnvoyAdminPort}}",
+  {{ if .Enterprise }} 
+    "-partition={{.Service.ID.Partition}}",
+  {{end}}
+  {{ if .Token }}
+    "-token={{.Token}}",
+  {{end}}
+    "--",
+    "-l",
+    "trace",
   ]
 }
diff --git a/testing/deployer/sprawl/internal/tfgen/tfgen.go b/testing/deployer/sprawl/internal/tfgen/tfgen.go
index e752528ef1..6df36dcb2d 100644
--- a/testing/deployer/sprawl/internal/tfgen/tfgen.go
+++ b/testing/deployer/sprawl/internal/tfgen/tfgen.go
@@ -5,6 +5,7 @@ package tfgen
 
 import (
 	"embed"
+	"text/template"
 )
 
 //go:embed templates/container-app-dataplane.tf.tmpl
@@ -12,7 +13,20 @@ import (
 //go:embed templates/container-app.tf.tmpl
 //go:embed templates/container-consul.tf.tmpl
 //go:embed templates/container-mgw.tf.tmpl
+//go:embed templates/container-mgw-dataplane.tf.tmpl
 //go:embed templates/container-pause.tf.tmpl
 //go:embed templates/container-proxy.tf.tmpl
 //go:embed templates/container-coredns.tf.tmpl
 var content embed.FS
+
+var (
+	tfAppDataplaneT         = template.Must(template.ParseFS(content, "templates/container-app-dataplane.tf.tmpl"))
+	tfAppSidecarT           = template.Must(template.ParseFS(content, "templates/container-app-sidecar.tf.tmpl"))
+	tfAppT                  = template.Must(template.ParseFS(content, "templates/container-app.tf.tmpl"))
+	tfConsulT               = template.Must(template.ParseFS(content, "templates/container-consul.tf.tmpl"))
+	tfMeshGatewayT          = template.Must(template.ParseFS(content, "templates/container-mgw.tf.tmpl"))
+	tfMeshGatewayDataplaneT = template.Must(template.ParseFS(content, "templates/container-mgw-dataplane.tf.tmpl"))
+	tfPauseT                = template.Must(template.ParseFS(content, "templates/container-pause.tf.tmpl"))
+	tfForwardProxyT         = template.Must(template.ParseFS(content, "templates/container-proxy.tf.tmpl"))
+	tfCorednsT              = template.Must(template.ParseFS(content, "templates/container-coredns.tf.tmpl"))
+)
diff --git a/testing/deployer/sprawl/sprawl.go b/testing/deployer/sprawl/sprawl.go
index a4b2759762..40f1571a92 100644
--- a/testing/deployer/sprawl/sprawl.go
+++ b/testing/deployer/sprawl/sprawl.go
@@ -113,6 +113,21 @@ func (s *Sprawl) APIClientForNode(clusterName string, nid topology.NodeID, token
 	)
 }
 
+// APIClientForCluster is a convenience wrapper for APIClientForNode that returns
+// an API client for an agent node in the cluster, preferring clients, then servers
+func (s *Sprawl) APIClientForCluster(clusterName, token string) (*api.Client, error) {
+	clu := s.topology.Clusters[clusterName]
+	// TODO: this always goes to the first client, but we might want to balance this
+	firstAgent := clu.FirstClient()
+	if firstAgent == nil {
+		firstAgent = clu.FirstServer()
+	}
+	if firstAgent == nil {
+		return nil, fmt.Errorf("failed to find agent in cluster %s", clusterName)
+	}
+	return s.APIClientForNode(clusterName, firstAgent.ID(), token)
+}
+
 func copyConfig(cfg *topology.Config) (*topology.Config, error) {
 	dup, err := copystructure.Copy(cfg)
 	if err != nil {
diff --git a/testing/deployer/sprawl/tls.go b/testing/deployer/sprawl/tls.go
index 4ba26432f1..bc5489206c 100644
--- a/testing/deployer/sprawl/tls.go
+++ b/testing/deployer/sprawl/tls.go
@@ -88,7 +88,9 @@ fi
 			"-i",
 			"--net=none",
 			"-v", cluster.TLSVolumeName + ":/data",
-			"busybox:latest",
+			// TODO: latest busted?
+			// https://hashicorp.slack.com/archives/C03EUN3QF1C/p1691784078972959
+			"busybox:1.34",
 			"sh", "-c",
 			// Need this so the permissions stick; docker seems to treat unused volumes differently.
 			`touch /data/VOLUME_PLACEHOLDER && chown -R ` + consulUserArg + ` /data`,
diff --git a/testing/deployer/topology/images.go b/testing/deployer/topology/images.go
index d58d7a1213..836ddb091c 100644
--- a/testing/deployer/topology/images.go
+++ b/testing/deployer/topology/images.go
@@ -66,6 +66,7 @@ func (i Images) EnvoyConsulImage() string {
 	return "local/" + name1 + "-and-" + name2 + ":" + tag1 + "-with-" + tag2
 }
 
+// TODO: what is this for and why do we need to do this and why is it named this?
 func (i Images) ChooseNode(kind NodeKind) Images {
 	switch kind {
 	case NodeKindServer:
diff --git a/testing/deployer/topology/topology.go b/testing/deployer/topology/topology.go
index b485436612..a71dbaa451 100644
--- a/testing/deployer/topology/topology.go
+++ b/testing/deployer/topology/topology.go
@@ -290,6 +290,7 @@ func (c *Cluster) ServerByAddr(addr string) *Node {
 
 func (c *Cluster) FirstServer() *Node {
 	for _, node := range c.Nodes {
+		// TODO: not sure why we check that it has 8500 exposed?
 		if node.IsServer() && !node.Disabled && node.ExposedPort(8500) > 0 {
 			return node
 		}
@@ -432,9 +433,6 @@ type Node struct {
 	// the enclosing Cluster.
 	Images Images
 
-	// AgentEnv contains optional environment variables to attach to Consul agents.
-	AgentEnv []string
-
 	Disabled bool `json:",omitempty"`
 
 	Addresses []*Address