feat(kuma-dp): add a separate component to handle kuma-sidecar readin…

…ess probes (backport of #11107) (#11241)

app/kuma-dp/cmd/run.go

@@ -16,6 +16,7 @@ import (
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/envoy"
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/meshmetrics"
 	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/metrics"
+	"github.com/kumahq/kuma/app/kuma-dp/pkg/dataplane/readiness"
 	kuma_cmd "github.com/kumahq/kuma/pkg/cmd"
 	"github.com/kumahq/kuma/pkg/config"
 	kumadp "github.com/kumahq/kuma/pkg/config/app/kuma-dp"
@@ -178,6 +179,7 @@ func newRunCmd(opts kuma_cmd.RunCmdOpts, rootCtx *RootContext) *cobra.Command {
 			bootstrap, kumaSidecarConfiguration, err := rootCtx.BootstrapGenerator(gracefulCtx, opts.Config.ControlPlane.URL, opts.Config, envoy.BootstrapParams{
 				Dataplane:           opts.Dataplane,
 				DNSPort:             cfg.DNS.EnvoyDNSPort,
+				ReadinessPort:       cfg.Dataplane.ReadinessPort,
 				EmptyDNSPort:        cfg.DNS.CoreDNSEmptyPort,
 				EnvoyVersion:        *envoyVersion,
 				Workdir:             cfg.DataplaneRuntime.SocketDir,
@@ -233,6 +235,15 @@ func newRunCmd(opts kuma_cmd.RunCmdOpts, rootCtx *RootContext) *cobra.Command {
 
 			observabilityComponents := setupObservability(kumaSidecarConfiguration, bootstrap, cfg)
 			components = append(components, observabilityComponents...)
+
+			var readinessReporter *readiness.Reporter
+			if cfg.Dataplane.ReadinessPort > 0 {
+				readinessReporter = readiness.NewReporter(
+					bootstrap.GetAdmin().GetAddress().GetSocketAddress().GetAddress(),
+					cfg.Dataplane.ReadinessPort)
+				components = append(components, readinessReporter)
+			}
+
 			if err := rootCtx.ComponentManager.Add(components...); err != nil {
 				return err
 			}
@@ -245,6 +256,9 @@ func newRunCmd(opts kuma_cmd.RunCmdOpts, rootCtx *RootContext) *cobra.Command {
 					if err := envoyComponent.DrainConnections(); err != nil {
 						runLog.Error(err, "could not drain connections")
 					} else {
+						if readinessReporter != nil {
+							readinessReporter.Terminating()
+						}
 						runLog.Info("waiting for connections to be drained", "waitTime", cfg.Dataplane.DrainTime)
 						select {
 						case <-time.After(cfg.Dataplane.DrainTime.Duration):

app/kuma-dp/pkg/dataplane/envoy/envoy.go

@@ -30,6 +30,7 @@ var runLog = core.Log.WithName("kuma-dp").WithName("run").WithName("envoy")
 type BootstrapParams struct {
 	Dataplane           rest.Resource
 	DNSPort             uint32
+	ReadinessPort       uint32
 	EmptyDNSPort        uint32
 	EnvoyVersion        EnvoyVersion
 	DynamicMetadata     map[string]string

app/kuma-dp/pkg/dataplane/envoy/remote_bootstrap.go

@@ -179,6 +179,7 @@ func (b *remoteBootstrap) requestForBootstrap(ctx context.Context, client *http.
 		},
 		DynamicMetadata:     params.DynamicMetadata,
 		DNSPort:             params.DNSPort,
+		ReadinessPort:       params.ReadinessPort,
 		EmptyDNSPort:        params.EmptyDNSPort,
 		OperatingSystem:     b.operatingSystem,
 		Features:            b.features,

app/kuma-dp/pkg/dataplane/readiness/component.go

@@ -0,0 +1,111 @@
+package readiness
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"sync/atomic"
+	"time"
+
+	"github.com/asaskevich/govalidator"
+	"github.com/bakito/go-log-logr-adapter/adapter"
+
+	"github.com/kumahq/kuma/pkg/core"
+	"github.com/kumahq/kuma/pkg/core/runtime/component"
+)
+
+const (
+	pathPrefixReady  = "/ready"
+	stateReady       = "READY"
+	stateTerminating = "TERMINATING"
+)
+
+// Reporter reports the health status of this Kuma Dataplane Proxy
+type Reporter struct {
+	localListenAddr string
+	localListenPort uint32
+	isTerminating   atomic.Bool
+}
+
+var logger = core.Log.WithName("readiness")
+
+func NewReporter(localIPAddr string, localListenPort uint32) *Reporter {
+	return &Reporter{
+		localListenPort: localListenPort,
+		localListenAddr: localIPAddr,
+	}
+}
+
+func (r *Reporter) Start(stop <-chan struct{}) error {
+	protocol := "tcp"
+	addr := r.localListenAddr
+	if govalidator.IsIPv6(addr) {
+		protocol = "tcp6"
+		addr = fmt.Sprintf("[%s]", addr)
+	}
+	lis, err := net.Listen(protocol, fmt.Sprintf("%s:%d", addr, r.localListenPort))
+	if err != nil {
+		return err
+	}
+
+	defer func() {
+		_ = lis.Close()
+	}()
+
+	logger.Info("starting readiness reporter", "addr", lis.Addr().String())
+
+	mux := http.NewServeMux()
+	mux.HandleFunc(pathPrefixReady, r.handleReadiness)
+	server := &http.Server{
+		ReadHeaderTimeout: time.Second,
+		Handler:           mux,
+		ErrorLog:          adapter.ToStd(logger),
+	}
+
+	errCh := make(chan error)
+	go func() {
+		if err := server.Serve(lis); err != nil {
+			errCh <- err
+		}
+	}()
+
+	select {
+	case err := <-errCh:
+		return err
+	case <-stop:
+		logger.Info("stopping readiness reporter")
+		return server.Shutdown(context.Background())
+	}
+}
+
+func (r *Reporter) Terminating() {
+	r.isTerminating.Store(true)
+}
+
+func (r *Reporter) handleReadiness(writer http.ResponseWriter, req *http.Request) {
+	state := stateReady
+	stateHTTPStatus := http.StatusOK
+	if r.isTerminating.Load() {
+		state = stateTerminating
+		stateHTTPStatus = http.StatusServiceUnavailable
+	}
+
+	stateBytes := []byte(state)
+	writer.Header().Set("content-type", "text/plain")
+	writer.Header().Set("content-length", fmt.Sprintf("%d", len(stateBytes)))
+	writer.Header().Set("cache-control", "no-cache, max-age=0")
+	writer.Header().Set("x-powered-by", "kuma-dp")
+	writer.WriteHeader(stateHTTPStatus)
+	_, err := writer.Write(stateBytes)
+	logger.V(1).Info("responding readiness state", "state", state, "client", req.RemoteAddr)
+	if err != nil {
+		logger.Info("[WARNING] could not write response", "err", err)
+	}
+}
+
+func (r *Reporter) NeedLeaderElection() bool {
+	return false
+}
+
+var _ component.Component = &Reporter{}

pkg/config/app/kuma-dp/config.go

@@ -25,10 +25,11 @@ var DefaultConfig = func() Config {
 			},
 		},
 		Dataplane: Dataplane{
-			Mesh:      "",
-			Name:      "", // Dataplane name must be set explicitly
-			DrainTime: config_types.Duration{Duration: 30 * time.Second},
-			ProxyType: "dataplane",
+			Mesh:          "",
+			Name:          "", // Dataplane name must be set explicitly
+			DrainTime:     config_types.Duration{Duration: 30 * time.Second},
+			ProxyType:     "dataplane",
+			ReadinessPort: 9902,
 		},
 		DataplaneRuntime: DataplaneRuntime{
 			BinaryPath: "envoy",
@@ -133,6 +134,8 @@ type Dataplane struct {
 	ProxyType string `json:"proxyType,omitempty" envconfig:"kuma_dataplane_proxy_type"`
 	// Drain time for listeners.
 	DrainTime config_types.Duration `json:"drainTime,omitempty" envconfig:"kuma_dataplane_drain_time"`
+	// Port that exposes kuma-dp readiness status on localhost, set this value to 0 to provide readiness by "/ready" endpoint from Envoy adminAPI
+	ReadinessPort uint32 `json:"readinessPort,omitempty" envconfig:"kuma_readiness_port"`
 }
 
 func (d *Dataplane) PostProcess() error {
@@ -304,6 +307,10 @@ func (d *Dataplane) Validate() error {
 		errs = multierr.Append(errs, errors.Errorf(".DrainTime must be positive"))
 	}
 
+	if d.ReadinessPort > 65353 {
+		return errors.New(".ReadinessPort has to be in [0, 65353] range")
+	}
+
 	return errs
 }
 

pkg/config/app/kuma-dp/testdata/default-config.golden.yaml

@@ -8,6 +8,7 @@ controlPlane:
 dataplane:
   drainTime: 30s
   proxyType: dataplane
+  readinessPort: 9902
 dataplaneRuntime:
   binaryPath: envoy
   dynamicConfiguration:

pkg/core/xds/metadata.go

@@ -21,6 +21,7 @@ const (
 	// Supported Envoy node metadata fields.
 	FieldDataplaneAdminPort         = "dataplane.admin.port"
 	FieldDataplaneAdminAddress      = "dataplane.admin.address"
+	FieldDataplaneReadinessPort     = "dataplane.readinessReporter.port"
 	FieldDataplaneDNSPort           = "dataplane.dns.port"
 	FieldDataplaneDNSEmptyPort      = "dataplane.dns.empty.port"
 	FieldDataplaneDataplaneResource = "dataplane.resource"
@@ -54,6 +55,7 @@ type DataplaneMetadata struct {
 	Resource            model.Resource
 	AdminPort           uint32
 	AdminAddress        string
+	ReadinessPort       uint32
 	DNSPort             uint32
 	EmptyDNSPort        uint32
 	DynamicMetadata     map[string]string
@@ -117,6 +119,13 @@ func (m *DataplaneMetadata) GetAdminPort() uint32 {
 	return m.AdminPort
 }
 
+func (m *DataplaneMetadata) GetReadinessPort() uint32 {
+	if m == nil {
+		return 0
+	}
+	return m.ReadinessPort
+}
+
 func (m *DataplaneMetadata) GetAdminAddress() string {
 	if m == nil {
 		return ""
@@ -165,6 +174,7 @@ func DataplaneMetadataFromXdsMetadata(xdsMetadata *structpb.Struct, tmpDir strin
 	}
 	metadata.AdminPort = uint32Metadata(xdsMetadata, FieldDataplaneAdminPort)
 	metadata.AdminAddress = xdsMetadata.Fields[FieldDataplaneAdminAddress].GetStringValue()
+	metadata.ReadinessPort = uint32Metadata(xdsMetadata, FieldDataplaneReadinessPort)
 	metadata.DNSPort = uint32Metadata(xdsMetadata, FieldDataplaneDNSPort)
 	metadata.EmptyDNSPort = uint32Metadata(xdsMetadata, FieldDataplaneDNSEmptyPort)
 	if value := xdsMetadata.Fields[FieldDataplaneDataplaneResource]; value != nil {

pkg/core/xds/metadata_test.go

@@ -59,6 +59,11 @@ var _ = Describe("DataplaneMetadataFromXdsMetadata", func() {
 							StringValue: "8001",
 						},
 					},
+					"dataplane.readinessReporter.port": {
+						Kind: &structpb.Value_StringValue{
+							StringValue: "9300",
+						},
+					},
 					"accessLogSocketPath": {
 						Kind: &structpb.Value_StringValue{
 							StringValue: "/tmp/logs",
@@ -77,6 +82,7 @@ var _ = Describe("DataplaneMetadataFromXdsMetadata", func() {
 				EmptyDNSPort:        8001,
 				AccessLogSocketPath: "/tmp/logs",
 				MetricsSocketPath:   "/tmp/metrics",
+				ReadinessPort:       9300,
 			},
 		}),
 		Entry("should ignore dependencies version provided through metadata if version is not set at all", testCase{

pkg/xds/bootstrap/generator.go

@@ -116,6 +116,7 @@ func (b *bootstrapGenerator) Generate(ctx context.Context, request types.Bootstr
 		},
 		DynamicMetadata:     request.DynamicMetadata,
 		DNSPort:             request.DNSPort,
+		ReadinessPort:       request.ReadinessPort,
 		EmptyDNSPort:        request.EmptyDNSPort,
 		ProxyType:           request.ProxyType,
 		Features:            request.Features,

pkg/xds/bootstrap/parameters.go

@@ -30,6 +30,7 @@ type configParameters struct {
 	Service             string
 	AdminAddress        string
 	AdminPort           uint32
+	ReadinessPort       uint32
 	AdminAccessLogPath  string
 	XdsHost             string
 	XdsPort             uint32

pkg/xds/bootstrap/template_v3.go

@@ -335,6 +335,9 @@ func genConfig(parameters configParameters, proxyConfig xds.Proxy, enableReloada
 	if parameters.EmptyDNSPort != 0 {
 		res.Node.Metadata.Fields[core_xds.FieldDataplaneDNSEmptyPort] = util_proto.MustNewValueForStruct(strconv.Itoa(int(parameters.EmptyDNSPort)))
 	}
+	if parameters.ReadinessPort != 0 {
+		res.Node.Metadata.Fields[core_xds.FieldDataplaneReadinessPort] = util_proto.MustNewValueForStruct(strconv.Itoa(int(parameters.ReadinessPort)))
+	}
 	if parameters.ProxyType != "" {
 		res.Node.Metadata.Fields[core_xds.FieldDataplaneProxyType] = util_proto.MustNewValueForStruct(parameters.ProxyType)
 	}

pkg/xds/bootstrap/types/bootstrap_request.go

@@ -13,6 +13,7 @@ type BootstrapRequest struct {
 	CaCert              string            `json:"caCert"`
 	DynamicMetadata     map[string]string `json:"dynamicMetadata"`
 	DNSPort             uint32            `json:"dnsPort,omitempty"`
+	ReadinessPort       uint32            `json:"readinessPort,omitempty"`
 	EmptyDNSPort        uint32            `json:"emptyDnsPort,omitempty"`
 	OperatingSystem     string            `json:"operatingSystem"`
 	Features            []string          `json:"features"`

pkg/xds/envoy/names/resource_names.go

@@ -65,6 +65,10 @@ func GetMetricsHijackerClusterName() string {
 	return Join("kuma", "metrics", "hijacker")
 }
 
+func GetDPPReadinessClusterName() string {
+	return Join("kuma", "readiness")
+}
+
 func GetPrometheusListenerName() string {
 	return Join("kuma", "metrics", "prometheus")
 }