feat(kuma-dp): rework on the virtual probes to support probing tcp and grpc ports

UPGRADE.md

@@ -174,6 +174,32 @@ Migration step:
 3. Deploy the gateway and verify if traffic works correctly.
 4. Remove the old resources.
 
+### Introduction to Application Probe Proxy and deprecation of Virtual Probes
+
+To support more types of application probes on Kubernetes, in version 2.9, we introduced a new feature named "Application Probe Proxy" which supports HTTP Get, TCP Socket and gRPC application probes. Starting from `2.9.x`, Virtual Probes is deprecated, and Application Probe Proxy is enabled by default.
+
+Application workloads using Virtual Probes will be migrated to Application Probe Proxy automatically on next restart/redeploy on Kubernetes, without other operations. 
+
+Application Probe Proxy will by default listen on port `9000`, the same port that Virtual Probes Listener uses. If you'd customized the Virtual Probes port, you might also want to customize the port of Application Probe Proxy. You may do so using one of these methods:
+
+1. Configuring on the control plane to apply on all dataplanes: set the port onto configuration key `runtime.kubernetes.injector.sidecarContainer.applicationProbeProxyPort` 
+1. Configuring on the control plane to apply on all dataplanes: set the port using environment variable `KUMA_RUNTIME_KUBERNETES_APPLICATION_PROBE_PROXY_PORT` 
+1. Configuring for certain dataplanes: set the port using pod annotation `kuma.io/application-probe-proxy-port`
+
+By setting the port to `0`, Application Probe Proxy feature will be disabled.
+
+When the Application Probe Proxy is disabled, Virtual Probes still works as usual before Virtual Probes is removed.
+
+Because of deprecation of Virtual Probes, the following items are considered deprecated:
+
+- Pod annotation `kuma.io/virtual-probes`
+- Pod annotation `kuma.io/virtual-probes-port`
+- Control plane configuration key `runtime.kubernetes.injector.sidecarContainer.virtualProbesEnabled`
+- Control plane configuration key `runtime.kubernetes.injector.sidecarContainer.virtualProbesPort`
+- Control plane environment variable `KUMA_RUNTIME_KUBERNETES_VIRTUAL_PROBES_ENABLED`
+- Control plane environment variable `KUMA_RUNTIME_KUBERNETES_VIRTUAL_PROBES_PORT`
+- Data field `probes` on `Dataplane` objects
+
 ### kumactl
 
 #### Default prometheus scrape config removes `service`

api/mesh/v1alpha1/dataplane.proto

@@ -338,5 +338,7 @@ message Dataplane {
   // See
   // https://kuma.io/docs/latest/policies/service-health-probes/#virtual-probes
   // for more information.
+  // Deprecated: this feature will be removed for Universal; on Kubernetes, it's
+  // not needed anymore.
   Probes probes = 3;
 }

app/cni/pkg/cni/annotations_linux.go

@@ -16,6 +16,7 @@ const (
 	defaultIPFamilyMode        = "dualstack"
 	defaultBuiltinDNSPort      = "15053"
 	defaultNoRedirectUID       = "5678"
+	defaultAppProbeProxyPort   = "9000"
 	defaultRedirectExcludePort = defaultProxyStatusPort
 )
 
@@ -36,6 +37,7 @@ var annotationRegistry = map[string]*annotationParam{
 	"iptablesLogs":                {"traffic.kuma.io/iptables-logs", "false", alwaysValidFunc},
 	"excludeInboundIPs":           {"traffic.kuma.io/exclude-inbound-ips", "", validateIPs},
 	"excludeOutboundIPs":          {"traffic.kuma.io/exclude-outbound-ips", "", validateIPs},
+	"applicationProbeProxyPort":   {"kuma.io/application-probe-proxy-port", defaultAppProbeProxyPort, validateSinglePort},
 }
 
 type IntermediateConfig struct {
@@ -148,6 +150,13 @@ func validatePortList(ports string) error {
 	return nil
 }
 
+func validateSinglePort(portString string) error {
+	if _, err := parsePort(portString); err != nil {
+		return err
+	}
+	return nil
+}
+
 func validateIpFamilyMode(val string) error {
 	if val == "" {
 		return errors.New("value is empty")
@@ -181,6 +190,7 @@ func getAnnotationOrDefault(name string, annotations map[string]string) (string,
 // NewIntermediateConfig returns a new IntermediateConfig Object constructed from a list of ports and annotations
 func NewIntermediateConfig(annotations map[string]string) (*IntermediateConfig, error) {
 	intermediateConfig := &IntermediateConfig{}
+	valDefaultProbeProxyPort := defaultAppProbeProxyPort
 
 	allFields := map[string]*string{
 		"outboundPort":                &intermediateConfig.targetPort,
@@ -193,6 +203,7 @@ func NewIntermediateConfig(annotations map[string]string) (*IntermediateConfig,
 		"builtinDNSPort":              &intermediateConfig.builtinDNSPort,
 		"excludeOutboundPortsForUIDs": &intermediateConfig.excludeOutboundPortsForUIDs,
 		"noRedirectUID":               &intermediateConfig.noRedirectUID,
+		"applicationProbeProxyPort":   &valDefaultProbeProxyPort,
 		"dropInvalidPackets":          &intermediateConfig.dropInvalidPackets,
 		"iptablesLogs":                &intermediateConfig.iptablesLogs,
 		"excludeInboundIPs":           &intermediateConfig.excludeInboundIPs,
@@ -205,6 +216,7 @@ func NewIntermediateConfig(annotations map[string]string) (*IntermediateConfig,
 		}
 	}
 
+	excludeAppProbeProxyPort(allFields)
 	return intermediateConfig, nil
 }
 
@@ -216,3 +228,18 @@ func mapAnnotation(annotations map[string]string, field *string, fieldName strin
 	*field = val
 	return nil
 }
+
+func excludeAppProbeProxyPort(allFields map[string]*string) {
+	inboundPortsToExclude := allFields["excludeInboundPorts"]
+	applicationProbeProxyPort := *allFields["applicationProbeProxyPort"]
+	if applicationProbeProxyPort == "0" {
+		return
+	}
+
+	existingExcludes := *inboundPortsToExclude
+	if existingExcludes == "" {
+		*inboundPortsToExclude = applicationProbeProxyPort
+	} else {
+		*inboundPortsToExclude = fmt.Sprintf("%s,%s", existingExcludes, applicationProbeProxyPort)
+	}
+}

app/cni/pkg/cni/annotations_linux_test.go

@@ -37,4 +37,14 @@ var _ = Describe("NewIntermediateConfig", func() {
 		Expect(err).ToNot(HaveOccurred())
 		Expect(cfg.inboundPort).To(Equal("1234"))
 	})
+
+	It("should exclude application probe proxy ports", func() {
+		a := map[string]string{
+			"kuma.io/application-probe-proxy-port":  "19988",
+			"traffic.kuma.io/exclude-inbound-ports": "3355",
+		}
+		cfg, err := NewIntermediateConfig(a)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(cfg.excludeInboundPorts).To(Equal("3355,19988"))
+	})
 })

app/cni/pkg/cni/main_linux.go

@@ -124,7 +124,7 @@ func cmdAdd(args *skel.CmdArgs) error {
 	ctx := context.Background()
 	conf, err := parseConfig(args.StdinData)
 	if err != nil {
-		return errors.Wrap(err, "error parsing kuma-cni cmdAdd config")
+		return errorLogged(log, err, "error parsing kuma-cni cmdAdd config")
 	}
 
 	mainProcessStderr, err := hijackMainProcessStderr(conf.LogLevel)
@@ -139,7 +139,7 @@ func cmdAdd(args *skel.CmdArgs) error {
 	// Determine if running under k8s by checking the CNI args
 	k8sArgs := K8sArgs{}
 	if err := types.LoadArgs(args.Args, &k8sArgs); err != nil {
-		return errors.Wrap(err, "error loading kuma-cni cmdAdd args")
+		return errorLogged(log, err, "error loading kuma-cni cmdAdd args")
 	}
 	logger := log.WithValues(
 		"pod", string(k8sArgs.K8S_POD_NAME),
@@ -162,7 +162,7 @@ func cmdAdd(args *skel.CmdArgs) error {
 
 	containerCount, initContainersMap, annotations, err := getPodInfoWithRetries(ctx, conf, k8sArgs)
 	if err != nil {
-		return errors.Wrap(err, "pod excluded - error getting pod info")
+		return errorLogged(logger, err, "pod excluded - error getting pod info")
 	}
 
 	if isInitContainerPresent(initContainersMap) {
@@ -184,10 +184,10 @@ func cmdAdd(args *skel.CmdArgs) error {
 	}
 
 	if intermediateConfig, configErr := NewIntermediateConfig(annotations); configErr != nil {
-		return errors.Wrap(configErr, "pod excluded - pod intermediateConfig failed due to bad params")
+		return errorLogged(logger, configErr, "pod excluded - pod intermediateConfig failed due to bad params")
 	} else {
 		if err := Inject(ctx, args.Netns, intermediateConfig, logger); err != nil {
-			return errors.Wrap(err, "pod excluded - could not inject rules into namespace")
+			return errorLogged(logger, err, "pod excluded - could not inject rules into namespace")
 		}
 	}
 	logger.Info("successfully injected iptables rules")
@@ -208,6 +208,11 @@ func prepareResult(conf *PluginConf, logger logr.Logger) error {
 	return types.PrintResult(result, conf.CNIVersion)
 }
 
+func errorLogged(logger logr.Logger, err error, message string) error {
+	logger.Info(fmt.Sprintf("[WARNING] %s", message), "err", err)
+	return errors.Wrap(err, message)
+}
+
 func excludeByMissingSidecarInjectedAnnotation(annotations map[string]string) bool {
 	excludePod := false
 	val, ok := annotations[metadata.KumaSidecarInjectedAnnotation]

app/kuma-dp/cmd/run.go

@@ -17,6 +17,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/probes"
 	kuma_cmd "github.com/kumahq/kuma/pkg/cmd"
 	"github.com/kumahq/kuma/pkg/config"
 	kumadp "github.com/kumahq/kuma/pkg/config/app/kuma-dp"
@@ -240,6 +241,13 @@ func newRunCmd(opts kuma_cmd.RunCmdOpts, rootCtx *RootContext) *cobra.Command {
 				return err
 			}
 
+			if opts.Config.ApplicationProbeProxyServer.Port > 0 {
+				prober := probes.NewProber(kumaSidecarConfiguration.Networking.Address, opts.Config.ApplicationProbeProxyServer.Port)
+				if err := rootCtx.ComponentManager.Add(prober); err != nil {
+					return err
+				}
+			}
+
 			stopComponents := make(chan struct{})
 			go func() {
 				var draining bool