forwarding dns query to envoy

pkg/consuldp/config.go

@@ -320,6 +320,13 @@ type EnvoyConfig struct {
 	ExtraArgs []string
 }
 
+const (
+	// VirtualDNSInlinePort is the port for Envoy's inline DNS table listener.
+	VirtualDNSInlinePort = 8653
+	// VirtualDNSEgressPort is the port for Envoy's external DNS forwarder listener.
+	VirtualDNSEgressPort = 8654
+)
+
 // XDSServer contains the configuration of the xDS server.
 type XDSServer struct {
 	// BindAddress is the address on which the Envoy xDS server will be available.

pkg/consuldp/consul_dataplane.go

@@ -47,6 +47,11 @@ type ConsulDataplane struct {
 	aclToken        string
 	metricsConfig   *metricsConfig
 	lifecycleConfig *lifecycleConfig
+	// upstreamIndex maps upstream service identities decoded from CDS SNIs on
+	// the proxied xDS stream. It is consulted during virtual-FQDN expansion so
+	// the dataplane fills missing tokens from the real upstream identity rather
+	// than the source proxy defaults.
+	upstreamIndex *dns.UpstreamIndex
 }
 
 // NewConsulDP creates a new instance of ConsulDataplane
@@ -66,8 +71,9 @@ func NewConsulDP(cfg *Config) (*ConsulDataplane, error) {
 	})
 
 	return &ConsulDataplane{
-		logger: logger,
-		cfg:    cfg,
+		logger:        logger,
+		cfg:           cfg,
+		upstreamIndex: dns.NewUpstreamIndex(),
 	}, nil
 }
 
@@ -186,7 +192,7 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 	if cdp.cfg.Mode == ModeTypeDNSProxy {
 		// start up DNS server with the configuration from the consul-dataplane flags / environment variables since
 		// envoy bootstrapping is bypassed.
-		if err = cdp.startDNSProxy(ctx, cdp.cfg.DNSServer, cdp.cfg.Proxy.Namespace, cdp.cfg.Proxy.Partition); err != nil {
+		if err = cdp.startDNSProxy(ctx, cdp.cfg.DNSServer, cdp.cfg.Proxy.Namespace, cdp.cfg.Proxy.Partition, ""); err != nil {
 			cdp.logger.Error("failed to start the dns proxy", "error", err)
 			return err
 		}
@@ -214,7 +220,7 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 	cdp.logger.Debug("generated envoy bootstrap params", "params", bootstrapParams)
 
 	// start up DNS server with envoy bootstrap params.
-	if err = cdp.startDNSProxy(ctx, cdp.cfg.DNSServer, bootstrapParams.Namespace, bootstrapParams.Partition); err != nil {
+	if err = cdp.startDNSProxy(ctx, cdp.cfg.DNSServer, bootstrapParams.Namespace, bootstrapParams.Partition, bootstrapParams.Datacenter); err != nil {
 		cdp.logger.Error("failed to start the dns proxy", "error", err)
 		return err
 	}
@@ -287,17 +293,28 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 }
 
 func (cdp *ConsulDataplane) startDNSProxy(ctx context.Context,
-	dnsConfig *DNSServerConfig, namespace, partition string) error {
+	dnsConfig *DNSServerConfig, namespace, partition, datacenter string) error {
 	dnsClientInterface := pbdns.NewDNSServiceClient(cdp.serverConn)
 
+	var (
+		virtualDNSInlineAddr string
+		virtualDNSEgressAddr string
+	)
+	virtualDNSInlineAddr = fmt.Sprintf("127.0.0.1:%d", VirtualDNSInlinePort)
+	virtualDNSEgressAddr = fmt.Sprintf("127.0.0.1:%d", VirtualDNSEgressPort)
+
 	dnsServer, err := dns.NewDNSServer(dns.DNSServerParams{
-		BindAddr:  dnsConfig.BindAddr,
-		Port:      dnsConfig.Port,
-		Client:    dnsClientInterface,
-		Logger:    cdp.logger,
-		Partition: partition,
-		Namespace: namespace,
-		Token:     cdp.aclToken,
+		BindAddr:             dnsConfig.BindAddr,
+		Port:                 dnsConfig.Port,
+		Client:               dnsClientInterface,
+		Logger:               cdp.logger,
+		Partition:            partition,
+		Namespace:            namespace,
+		Token:                cdp.aclToken,
+		Datacenter:           datacenter,
+		VirtualDNSInlineAddr: virtualDNSInlineAddr,
+		VirtualDNSEgressAddr: virtualDNSEgressAddr,
+		UpstreamIndex:        cdp.upstreamIndex,
 	})
 	if err == dns.ErrServerDisabled {
 		cdp.logger.Info("dns server disabled: configure the Consul DNS port to enable")

pkg/consuldp/xds.go

@@ -15,6 +15,8 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
+
+	"github.com/hashicorp/consul-dataplane/pkg/dns"
 )
 
 const (
@@ -118,15 +120,23 @@ func (cdp *ConsulDataplane) xdsServerExited() chan struct{} { return cdp.xdsServ
 
 func (cdp *ConsulDataplane) streamInterceptor() grpc.StreamServerInterceptor {
 	return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
-		return handler(srv, &metricServerStream{ss})
+		return handler(srv, &metricServerStream{ServerStream: ss, upstreamIndex: cdp.upstreamIndex})
 	}
 }
 
 type metricServerStream struct {
 	grpc.ServerStream
+	// upstreamIndex, when set, receives CDS cluster SNIs decoded from the
+	// server->Envoy frames that pass through this stream.
+	upstreamIndex *dns.UpstreamIndex
 }
 
 func (s *metricServerStream) SendMsg(m interface{}) error {
+	// Tap CDS frames on their way to Envoy to keep the upstream identity index
+	// current. This reads a copy of the frame's bytes and never mutates m, so
+	// opaque passthrough is unaffected.
+	tapClusterFrame(s.upstreamIndex, m)
+
 	err := s.ServerStream.SendMsg(m)
 	if err == nil {
 		metrics.SetGauge([]string{"envoy_connected"}, 1)

pkg/consuldp/xds_cluster_tap.go

@@ -0,0 +1,132 @@
+// Copyright IBM Corp. 2022, 2026
+// SPDX-License-Identifier: MPL-2.0
+
+package consuldp
+
+import (
+	"google.golang.org/protobuf/encoding/protowire"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	"github.com/hashicorp/consul-dataplane/pkg/dns"
+)
+
+// clusterTypeURL is the xDS type URL carried by CDS responses. Only responses
+// with this type URL contribute upstream identities to the index.
+const clusterTypeURL = "type.googleapis.com/envoy.config.cluster.v3.Cluster"
+
+// Field numbers within envoy.service.discovery.v3.DeltaDiscoveryResponse.
+// These mirror the generated proto (see go-control-plane discovery.pb.go) and
+// are the raw wire identifiers we scan for so the dataplane does not need to
+// pull in the full envoy proto dependency just to read cluster names.
+const (
+	deltaFieldResources        = 2 // repeated Resource
+	deltaFieldTypeURL          = 4 // string
+	deltaFieldRemovedResources = 6 // repeated string
+	resourceFieldName          = 3 // string (the cluster SNI)
+)
+
+// tapClusterFrame inspects a single server->Envoy xDS frame flowing through the
+// proxy. The proxy carries each frame as an opaque *emptypb.Empty whose unknown
+// fields hold the wire-format DiscoveryResponse, so we read those raw bytes and
+// hand-parse just the cluster names. When the frame is a CDS response, the
+// added and removed cluster SNIs are applied to the upstream index. Any other
+// frame (or an unparseable one) is ignored; the frame itself is never modified,
+// so passthrough is unaffected.
+func tapClusterFrame(index *dns.UpstreamIndex, m interface{}) {
+	if index == nil {
+		return
+	}
+	empty, ok := m.(*emptypb.Empty)
+	if !ok {
+		return
+	}
+	raw := empty.ProtoReflect().GetUnknown()
+	if len(raw) == 0 {
+		return
+	}
+	added, removed, isCluster := extractClusterSNIs(raw)
+	if !isCluster {
+		return
+	}
+	index.Update(added, removed)
+}
+
+// extractClusterSNIs scans a wire-format DeltaDiscoveryResponse for the cluster
+// SNIs it added/updated and removed. It returns isCluster=false (and no names)
+// when the frame is not a CDS response or cannot be parsed.
+func extractClusterSNIs(raw []byte) (added, removed []string, isCluster bool) {
+	var (
+		typeURL string
+		names   []string
+		gone    []string
+	)
+	b := raw
+	for len(b) > 0 {
+		num, typ, n := protowire.ConsumeTag(b)
+		if n < 0 {
+			return nil, nil, false
+		}
+		b = b[n:]
+		switch {
+		case num == deltaFieldTypeURL && typ == protowire.BytesType:
+			v, vn := protowire.ConsumeBytes(b)
+			if vn < 0 {
+				return nil, nil, false
+			}
+			typeURL = string(v)
+			b = b[vn:]
+		case num == deltaFieldResources && typ == protowire.BytesType:
+			v, vn := protowire.ConsumeBytes(b)
+			if vn < 0 {
+				return nil, nil, false
+			}
+			if name, ok := extractResourceName(v); ok {
+				names = append(names, name)
+			}
+			b = b[vn:]
+		case num == deltaFieldRemovedResources && typ == protowire.BytesType:
+			v, vn := protowire.ConsumeBytes(b)
+			if vn < 0 {
+				return nil, nil, false
+			}
+			gone = append(gone, string(v))
+			b = b[vn:]
+		default:
+			vn := protowire.ConsumeFieldValue(num, typ, b)
+			if vn < 0 {
+				return nil, nil, false
+			}
+			b = b[vn:]
+		}
+	}
+	if typeURL != clusterTypeURL {
+		return nil, nil, false
+	}
+	return names, gone, true
+}
+
+// extractResourceName scans a wire-format DeltaDiscoveryResponse.Resource for
+// its name field, which for a CDS cluster is the upstream SNI.
+func extractResourceName(raw []byte) (string, bool) {
+	b := raw
+	for len(b) > 0 {
+		num, typ, n := protowire.ConsumeTag(b)
+		if n < 0 {
+			return "", false
+		}
+		b = b[n:]
+		if num == resourceFieldName && typ == protowire.BytesType {
+			v, vn := protowire.ConsumeBytes(b)
+			if vn < 0 {
+				return "", false
+			}
+			return string(v), true
+		}
+		vn := protowire.ConsumeFieldValue(num, typ, b)
+		if vn < 0 {
+			return "", false
+		}
+		b = b[vn:]
+	}
+	return "", false
+}

pkg/consuldp/xds_cluster_tap_test.go

@@ -0,0 +1,102 @@
+// Copyright IBM Corp. 2022, 2026
+// SPDX-License-Identifier: MPL-2.0
+
+package consuldp
+
+import (
+	"testing"
+
+	"google.golang.org/protobuf/encoding/protowire"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	"github.com/hashicorp/consul-dataplane/pkg/dns"
+)
+
+func encodeResource(name string) []byte {
+	var b []byte
+	b = protowire.AppendTag(b, resourceFieldName, protowire.BytesType)
+	b = protowire.AppendBytes(b, []byte(name))
+	return b
+}
+
+func encodeDeltaResponse(typeURL string, names, removed []string) []byte {
+	var b []byte
+	b = protowire.AppendTag(b, deltaFieldTypeURL, protowire.BytesType)
+	b = protowire.AppendBytes(b, []byte(typeURL))
+	for _, n := range names {
+		b = protowire.AppendTag(b, deltaFieldResources, protowire.BytesType)
+		b = protowire.AppendBytes(b, encodeResource(n))
+	}
+	for _, r := range removed {
+		b = protowire.AppendTag(b, deltaFieldRemovedResources, protowire.BytesType)
+		b = protowire.AppendBytes(b, []byte(r))
+	}
+	return b
+}
+
+func emptyWithBytes(b []byte) *emptypb.Empty {
+	e := &emptypb.Empty{}
+	e.ProtoReflect().SetUnknown(protoreflect.RawFields(b))
+	return e
+}
+
+func TestExtractClusterSNIs(t *testing.T) {
+	const td = "e5b1a4d3.consul"
+
+	t.Run("cluster response", func(t *testing.T) {
+		raw := encodeDeltaResponse(clusterTypeURL,
+			[]string{"api.default.dc1.internal." + td},
+			[]string{"web.default.partition-vms.dc1.internal-v1." + td},
+		)
+		added, removed, ok := extractClusterSNIs(raw)
+		if !ok {
+			t.Fatal("expected cluster response")
+		}
+		if len(added) != 1 || added[0] != "api.default.dc1.internal."+td {
+			t.Fatalf("unexpected added: %v", added)
+		}
+		if len(removed) != 1 || removed[0] != "web.default.partition-vms.dc1.internal-v1."+td {
+			t.Fatalf("unexpected removed: %v", removed)
+		}
+	})
+
+	t.Run("non-cluster response ignored", func(t *testing.T) {
+		raw := encodeDeltaResponse(
+			"type.googleapis.com/envoy.config.listener.v3.Listener",
+			[]string{"ignored"}, nil)
+		if _, _, ok := extractClusterSNIs(raw); ok {
+			t.Fatal("expected non-cluster response to be ignored")
+		}
+	})
+
+	t.Run("empty bytes", func(t *testing.T) {
+		if _, _, ok := extractClusterSNIs(nil); ok {
+			t.Fatal("expected empty input to be ignored")
+		}
+	})
+}
+
+func TestTapClusterFrame(t *testing.T) {
+	const td = "e5b1a4d3.consul"
+	idx := dns.NewUpstreamIndex()
+
+	raw := encodeDeltaResponse(clusterTypeURL,
+		[]string{"api.default.partition-vms.dc1.internal-v1." + td}, nil)
+	tapClusterFrame(idx, emptyWithBytes(raw))
+
+	if _, ok := idx.Lookup("api", "", "partition-vms", "dc1"); !ok {
+		t.Fatal("expected cluster SNI to be indexed")
+	}
+
+	// Non-Empty message must be ignored without panicking.
+	tapClusterFrame(idx, "not-an-empty")
+
+	// A removal frame drops the entry.
+	rm := encodeDeltaResponse(clusterTypeURL, nil,
+		[]string{"api.default.partition-vms.dc1.internal-v1." + td})
+	tapClusterFrame(idx, emptyWithBytes(rm))
+	if _, ok := idx.Lookup("api", "", "partition-vms", "dc1"); ok {
+		t.Fatal("expected entry to be removed")
+	}
+}