feat(dns): embedded DNS server instead of coreDNS (#13124)

## Motivation

Add an embedded DNS server that resolves mesh local hostnames.
This feature is disabled by default and needs to be enabled with
`kuma_dns_proxy_port ` on DPs in universal to a non 0 value or simply
with `kuma_runtime_kubernetes_injector_builtin_dns_experimental_proxy
=true` on CP in Kubernetes (this leverages sidecar injection).

## Implementation information

We leveraged the same system as MeshMetric. The DNS map is exposed in
the `_kuma:dynamicconfig` listener, we poll this from the DP at a set
frequency and reload the data.

We improved the whole configfetcher to make it reusable across different
components. We also setup the endpoint so that it works with [etag
caching](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Headers/ETag)
while avoids unnecessary reloading and processing, we can therefore
refresh the configs more quickly as well.

For the moment I added a new matrix for e2e test. If this is stable I'll
switch the default.

---------

Signed-off-by: Charly Molter <[email protected]>

Author: lahabana

.github/workflows/_e2e.yaml

@@ -15,6 +15,7 @@ env:
   E2E_PARAM_CNI_NETWORK_PLUGIN: ${{ fromJSON(inputs.matrix).cniNetworkPlugin }}
   E2E_PARAM_ARCH: ${{ fromJSON(inputs.matrix).arch }}
   E2E_PARAM_SIDECAR_CONTAINERS: ${{ fromJSON(inputs.matrix).sidecarContainers }}
+  E2E_PARAM_DP_DNS: ${{ fromJSON(inputs.matrix).dpDNS }}
   E2E_PARAM_TARGET: ${{ fromJSON(inputs.matrix).target }}
   E2E_PARAM_PARALLELISM: ${{ fromJSON(inputs.matrix).parallelism }}
 jobs:
@@ -101,6 +102,9 @@ jobs:
           if [[ "${{ env.E2E_PARAM_SIDECAR_CONTAINERS }}" != "" ]]; then
             export KUMA_EXPERIMENTAL_SIDECAR_CONTAINERS=true
           fi
+          if [[ "${{ env.E2E_PARAM_DP_DNS }}" != "" ]]; then
+            export KUMA_EXPERIMENTAL_DP_DNS=true
+          fi
 
           if [[ "${{ env.E2E_PARAM_TARGET }}" == "" ]]; then
             export GINKGO_E2E_LABEL_FILTERS="job-${{ matrix.parallelRunnerId }}"

.github/workflows/_test.yaml

@@ -69,6 +69,8 @@ jobs:
                   {"target":"universal", "k8sVersion":"${{ env.K8S_MAX_VERSION }}"}
                 ],
                 "include":[
+                  {"dpDNS": "true", "k8sVersion": "${{ env.K8S_MAX_VERSION }}", "target": "kubernetes", "arch": "amd64"},
+                  {"dpDNS": "true", "k8sVersion": "kind", "target": "universal", "arch": "amd64"},
                   {"sidecarContainers": "sidecarContainers", "k8sVersion": "${{ env.K8S_MAX_VERSION }}", "target": "kubernetes", "arch": "amd64"},
                   {"k8sVersion": "${{ env.K8S_MIN_VERSION }}", "target": "multizone", "arch": "amd64"},
                   {"k8sVersion": "${{ env.K8S_MIN_VERSION }}", "target": "kubernetes", "arch": "amd64"},

app/kuma-dp/cmd/context.go

@@ -10,7 +10,6 @@ import (
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/envoy"
 	kumadp "github.com/kumahq/kuma/pkg/config/app/kuma-dp"
 	"github.com/kumahq/kuma/pkg/core/runtime/component"
-	core_xds "github.com/kumahq/kuma/pkg/core/xds"
 	"github.com/kumahq/kuma/pkg/log"
 	leader_memory "github.com/kumahq/kuma/pkg/plugins/leader/memory"
 )
@@ -25,8 +24,6 @@ type RootContext struct {
 	LogLevel                 log.LogLevel
 }
 
-var features = []string{core_xds.FeatureTCPAccessLogViaNamedPipe}
-
 // defaultDataplaneTokenGenerator uses only given tokens or paths from the
 // config.
 func defaultDataplaneTokenGenerator(cfg *kumadp.Config) (component.Component, error) {
@@ -54,7 +51,7 @@ func DefaultRootContext() *RootContext {
 	config := kumadp.DefaultConfig()
 	return &RootContext{
 		ComponentManager:         component.NewManager(leader_memory.NewNeverLeaderElector()),
-		BootstrapGenerator:       envoy.NewRemoteBootstrapGenerator(runtime.GOOS, features),
+		BootstrapGenerator:       envoy.NewRemoteBootstrapGenerator(runtime.GOOS),
 		Config:                   &config,
 		BootstrapDynamicMetadata: map[string]string{},
 		DataplaneTokenGenerator:  defaultDataplaneTokenGenerator,

app/kuma-dp/cmd/run.go

@@ -2,8 +2,10 @@ package cmd
 
 import (
 	"context"
+	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 	"time"
 
 	envoy_bootstrap_v3 "github.com/envoyproxy/go-control-plane/envoy/config/bootstrap/v3"
@@ -13,6 +15,8 @@ import (
 	mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1"
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/accesslogs"
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/certificate"
+	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/configfetcher"
+	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/dnsproxy"
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/dnsserver"
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/envoy"
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/meshmetrics"
@@ -27,7 +31,9 @@ import (
 	"github.com/kumahq/kuma/pkg/core/resources/model/rest"
 	"github.com/kumahq/kuma/pkg/core/runtime/component"
 	core_xds "github.com/kumahq/kuma/pkg/core/xds"
-	"github.com/kumahq/kuma/pkg/util/net"
+	dns_dpapi "github.com/kumahq/kuma/pkg/dns/dpapi"
+	meshmetric_dpapi "github.com/kumahq/kuma/pkg/plugins/policies/meshmetric/dpapi"
+	kuma_net "github.com/kumahq/kuma/pkg/util/net"
 	"github.com/kumahq/kuma/pkg/util/proto"
 	kuma_version "github.com/kumahq/kuma/pkg/version"
 	"github.com/kumahq/kuma/pkg/xds/bootstrap/types"
@@ -84,7 +90,7 @@ func newRunCmd(opts kuma_cmd.RunCmdOpts, rootCtx *RootContext) *cobra.Command {
 
 			proxyResource, err = readResource(cmd, &cfg.DataplaneRuntime)
 			if err != nil {
-				runLog.Error(err, "failed to read policy", "proxyType", cfg.Dataplane.ProxyType)
+				runLog.Error(err, "failed to read dataplane", "proxyType", cfg.Dataplane.ProxyType)
 
 				return err
 			}
@@ -205,6 +211,12 @@ func newRunCmd(opts kuma_cmd.RunCmdOpts, rootCtx *RootContext) *cobra.Command {
 			}
 			opts.AdminPort = bootstrap.GetAdmin().GetAddress().GetSocketAddress().GetPortValue()
 
+			confFetcher := configfetcher.NewConfigFetcher(
+				core_xds.MeshMetricsDynamicConfigurationSocketName(cfg.DataplaneRuntime.SocketDir),
+				time.NewTicker(cfg.DataplaneRuntime.DynamicConfiguration.RefreshInterval.Duration),
+				cfg.DataplaneRuntime.DynamicConfiguration.RefreshInterval.Duration,
+			)
+
 			if cfg.DNS.Enabled && !cfg.Dataplane.IsZoneProxy() {
 				dnsOpts := &dnsserver.Opts{
 					Config:   *cfg,
@@ -216,29 +228,49 @@ func newRunCmd(opts kuma_cmd.RunCmdOpts, rootCtx *RootContext) *cobra.Command {
 				if len(kumaSidecarConfiguration.Networking.CorefileTemplate) > 0 {
 					dnsOpts.ProvidedCorefileTemplate = kumaSidecarConfiguration.Networking.CorefileTemplate
 				}
+				if dnsOpts.Config.DNS.ProxyPort != 0 {
+					runLog.Info("Running with embedded DNS proxy port", "port", dnsOpts.Config.DNS.ProxyPort)
+					// Using embedded DNS
+					dnsproxyServer, err := dnsproxy.NewServer(net.JoinHostPort("localhost", strconv.Itoa(int(dnsOpts.Config.DNS.ProxyPort))))
+					if err != nil {
+						return err
+					}
+					if err := confFetcher.AddHandler(dns_dpapi.PATH, dnsproxyServer.ReloadMap); err != nil {
+						return err
+					}
+					components = append(components, dnsproxyServer)
+				} else {
+					dnsServer, err := dnsserver.New(dnsOpts)
+					if err != nil {
+						return err
+					}
 
-				dnsServer, err := dnsserver.New(dnsOpts)
-				if err != nil {
-					return err
-				}
+					version, err := dnsServer.GetVersion()
+					if err != nil {
+						return err
+					}
 
-				version, err := dnsServer.GetVersion()
-				if err != nil {
-					return err
+					rootCtx.BootstrapDynamicMetadata[core_xds.FieldPrefixDependenciesVersion+".coredns"] = version
+					components = append(components, dnsServer)
 				}
-
-				rootCtx.BootstrapDynamicMetadata[core_xds.FieldPrefixDependenciesVersion+".coredns"] = version
-
-				components = append(components, dnsServer)
 			}
 
 			envoyComponent, err := envoy.New(opts)
 			if err != nil {
 				return err
 			}
 			components = append(components, envoyComponent)
-
-			observabilityComponents := setupObservability(kumaSidecarConfiguration, bootstrap, cfg)
+			components = append(components, component.NewResilientComponent(
+				runLog.WithName("configfetcher"),
+				confFetcher,
+				cfg.Dataplane.ResilientComponentBaseBackoff.Duration,
+				cfg.Dataplane.ResilientComponentMaxBackoff.Duration,
+			))
+
+			observabilityComponents, err := setupObservability(gracefulCtx, kumaSidecarConfiguration, bootstrap, cfg, confFetcher)
+			if err != nil {
+				return err
+			}
 			components = append(components, observabilityComponents...)
 
 			var readinessReporter *readiness.Reporter
@@ -355,7 +387,7 @@ func getApplicationsToScrape(kumaSidecarConfiguration *types.KumaSidecarConfigur
 				Name:              item.Name,
 				Path:              item.Path,
 				Port:              item.Port,
-				IsIPv6:            net.IsAddressIPv6(item.Address),
+				IsIPv6:            kuma_net.IsAddressIPv6(item.Address),
 				QueryModifier:     metrics.RemoveQueryParameters,
 				MeshMetricMutator: metrics.AggregatedOtelMutator(),
 			})
@@ -382,18 +414,16 @@ func writeFile(filename string, data []byte, perm os.FileMode) error {
 	return os.WriteFile(filename, data, perm)
 }
 
-func setupObservability(kumaSidecarConfiguration *types.KumaSidecarConfiguration, bootstrap *envoy_bootstrap_v3.Bootstrap, cfg *kumadp.Config) []component.Component {
-	resilientComponentBaseBackoff := 5 * time.Second
-	resilientComponentMaxBackoff := 1 * time.Minute
+func setupObservability(ctx context.Context, kumaSidecarConfiguration *types.KumaSidecarConfiguration, bootstrap *envoy_bootstrap_v3.Bootstrap, cfg *kumadp.Config, fetcher *configfetcher.ConfigFetcher) ([]component.Component, error) {
 	baseApplicationsToScrape := getApplicationsToScrape(kumaSidecarConfiguration, bootstrap.GetAdmin().GetAddress().GetSocketAddress().GetPortValue())
 
 	accessLogStreamer := component.NewResilientComponent(
 		runLog.WithName("access-log-streamer"),
 		accesslogs.NewAccessLogStreamer(
 			core_xds.AccessLogSocketName(cfg.DataplaneRuntime.SocketDir, cfg.Dataplane.Name, cfg.Dataplane.Mesh),
 		),
-		resilientComponentBaseBackoff,
-		resilientComponentMaxBackoff,
+		cfg.Dataplane.ResilientComponentBaseBackoff.Duration,
+		cfg.Dataplane.ResilientComponentMaxBackoff.Duration,
 	)
 
 	openTelemetryProducer := metrics.NewAggregatedMetricsProducer(
@@ -410,21 +440,18 @@ func setupObservability(kumaSidecarConfiguration *types.KumaSidecarConfiguration
 		openTelemetryProducer,
 	)
 
-	meshMetricsConfigFetcher := component.NewResilientComponent(
-		runLog.WithName("mesh-metric-config-fetcher"),
-		meshmetrics.NewMeshMetricConfigFetcher(
-			core_xds.MeshMetricsDynamicConfigurationSocketName(cfg.DataplaneRuntime.SocketDir),
-			time.NewTicker(cfg.DataplaneRuntime.DynamicConfiguration.RefreshInterval.Duration),
-			metricsServer,
-			openTelemetryProducer,
-			kumaSidecarConfiguration.Networking.Address,
-			bootstrap.GetAdmin().GetAddress().GetSocketAddress().GetPortValue(),
-			bootstrap.GetAdmin().GetAddress().GetSocketAddress().GetAddress(),
-			cfg.Dataplane.DrainTime.Duration,
-		),
-		resilientComponentBaseBackoff,
-		resilientComponentMaxBackoff,
+	mm := meshmetrics.NewManager(
+		ctx,
+		metricsServer,
+		openTelemetryProducer,
+		kumaSidecarConfiguration.Networking.Address,
+		bootstrap.GetAdmin().GetAddress().GetSocketAddress().GetPortValue(),
+		bootstrap.GetAdmin().GetAddress().GetSocketAddress().GetAddress(),
+		cfg.Dataplane.DrainTime.Duration,
 	)
-
-	return []component.Component{accessLogStreamer, meshMetricsConfigFetcher, metricsServer}
+	err := fetcher.AddHandler(meshmetric_dpapi.PATH, mm.OnChange)
+	if err != nil {
+		return nil, err
+	}
+	return []component.Component{accessLogStreamer, metricsServer, mm}, nil
 }

app/kuma-dp/cmd/run_test.go

@@ -16,6 +16,7 @@ import (
 	envoy_bootstrap_v3 "github.com/envoyproxy/go-control-plane/envoy/config/bootstrap/v3"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/envoy"
 	kuma_cmd "github.com/kumahq/kuma/pkg/cmd"
@@ -36,6 +37,7 @@ var _ = Describe("run", func() {
 	var tmpDir string
 
 	BeforeEach(func() {
+		prometheus.DefaultRegisterer = prometheus.NewRegistry()
 		ctx, cancel = context.WithCancel(context.Background())
 		var err error
 		tmpDir, err = os.MkdirTemp("", "")