Add container readiness probes

cmd/cl-controlplane/app/server.go

@@ -15,9 +15,12 @@ package app
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"net/http"
 	"os"
 	"path"
+	"syscall"
 
 	"github.com/bombsimon/logrusr/v4"
 	"github.com/sirupsen/logrus"
@@ -39,6 +42,7 @@ import (
 	"github.com/clusterlink-net/clusterlink/pkg/controlplane/xds"
 	"github.com/clusterlink-net/clusterlink/pkg/util/controller"
 	"github.com/clusterlink-net/clusterlink/pkg/util/grpc"
+	utilhttp "github.com/clusterlink-net/clusterlink/pkg/util/http"
 	"github.com/clusterlink-net/clusterlink/pkg/util/log"
 	"github.com/clusterlink-net/clusterlink/pkg/util/runnable"
 	"github.com/clusterlink-net/clusterlink/pkg/util/tls"
@@ -219,11 +223,27 @@ func (o *Options) Run() error {
 		return err
 	}
 
+	readinessListenAddress := fmt.Sprintf("0.0.0.0:%d", api.ReadinessListenPort)
+	httpServer := utilhttp.NewServer("controlplane-http", nil)
+	httpServer.Router().Get("/", func(w http.ResponseWriter, r *http.Request) {
+		resp, err := http.Get(fmt.Sprintf("https://127.0.0.1:%d", api.ListenPort))
+		if err == nil && resp.Body.Close() != nil {
+			logrus.Infof("Cannot close readiness response body: %v", err)
+		}
+		if errors.Is(err, syscall.ECONNREFUSED) ||
+			errors.Is(err, syscall.ECONNRESET) ||
+			!authzManager.IsReady() ||
+			!mgr.GetCache().WaitForCacheSync(r.Context()) {
+			w.WriteHeader(http.StatusServiceUnavailable)
+		}
+	})
+
 	runnableManager := runnable.NewManager()
 	runnableManager.Add(peerCertsWatcher)
 	runnableManager.Add(controller.NewManager(mgr))
 	runnableManager.Add(controlManager)
 	runnableManager.AddServer(controlplaneServerListenAddress, grpcServer)
+	runnableManager.AddServer(readinessListenAddress, httpServer)
 
 	return runnableManager.Run()
 }

cmd/cl-dataplane/app/envoy.go

@@ -26,12 +26,16 @@ import (
 
 const (
 	envoyPath = "/usr/local/bin/envoy"
+
+	adminPort = 1500
 )
 
 func (o *Options) runEnvoy(dataplaneID string) error {
 	envoyConfArgs := map[string]interface{}{
 		"dataplaneID": dataplaneID,
 
+		"adminPort": adminPort,
+
 		"controlplaneHost": o.ControlplaneHost,
 		"controlplanePort": cpapi.ListenPort,
 

cmd/cl-dataplane/app/envoyconf.go

@@ -22,7 +22,7 @@ admin:
   address:
     socket_address:
       address: 127.0.0.1
-      port_value: 1500
+      port_value: {{.adminPort}}
 bootstrap_extensions:
 - name: envoy.bootstrap.internal_listener
   typed_config:

cmd/cl-dataplane/app/server.go

@@ -15,13 +15,16 @@ package app
 
 import (
 	"fmt"
+	"net/http"
 	"os"
 
 	"github.com/google/uuid"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 
+	"github.com/clusterlink-net/clusterlink/pkg/dataplane/api"
+	utilhttp "github.com/clusterlink-net/clusterlink/pkg/util/http"
 	"github.com/clusterlink-net/clusterlink/pkg/util/log"
 )
 
@@ -36,8 +39,6 @@ const (
 	// KeyFile is the path to the private-key file.
 	KeyFile = "/etc/ssl/key/clink-dataplane.pem"
 
-	// Name is the app label of dataplane pods.
-	Name = "cl-dataplane"
 	// IngressSvcName is the ingress service name for the dataplane pods.
 	IngressSvcName = "clusterlink"
 )
@@ -85,6 +86,25 @@ func (o *Options) Run() error {
 	dataplaneID := uuid.New().String()
 	logrus.Infof("Dataplane ID: %s.", dataplaneID)
 
+	readinessListenAddress := fmt.Sprintf("0.0.0.0:%d", api.ReadinessListenPort)
+	httpServer := utilhttp.NewServer("dataplane-readiness-http", nil)
+	if err := httpServer.Listen(readinessListenAddress); err != nil {
+		return fmt.Errorf("cannot listen for readiness: %w", err)
+	}
+	httpServer.Router().Get("/", func(w http.ResponseWriter, r *http.Request) {
+		resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d/ready", adminPort))
+		if err == nil && resp.Body.Close() != nil {
+			logrus.Infof("Cannot close readiness response body: %v", err)
+		}
+		if err != nil || resp.StatusCode != http.StatusOK {
+			w.WriteHeader(http.StatusServiceUnavailable)
+		}
+	})
+	go func() {
+		err := httpServer.Start()
+		logrus.Errorf("Failed to start readiness server: %v.", err)
+	}()
+
 	return o.runEnvoy(dataplaneID)
 }
 

cmd/cl-go-dataplane/app/server.go

@@ -16,6 +16,7 @@ package app
 import (
 	"fmt"
 	"net"
+	"net/http"
 	"os"
 	"strconv"
 	"time"
@@ -32,6 +33,7 @@ import (
 	"github.com/clusterlink-net/clusterlink/pkg/dataplane/api"
 	dpclient "github.com/clusterlink-net/clusterlink/pkg/dataplane/client"
 	dpserver "github.com/clusterlink-net/clusterlink/pkg/dataplane/server"
+	utilhttp "github.com/clusterlink-net/clusterlink/pkg/util/http"
 	"github.com/clusterlink-net/clusterlink/pkg/util/log"
 	"github.com/clusterlink-net/clusterlink/pkg/util/tls"
 )
@@ -81,7 +83,7 @@ func (o *Options) runGoDataplane(dataplaneID string, parsedCertData *tls.ParsedC
 
 	controlplaneClient, err := grpc.NewClient(
 		controlplaneTarget,
-		grpc.WithTransportCredentials(credentials.NewTLS(parsedCertData.ClientConfig("cl-controlplane"))),
+		grpc.WithTransportCredentials(credentials.NewTLS(parsedCertData.ClientConfig(cpapi.Name))),
 		grpc.WithConnectParams(grpc.ConnectParams{
 			Backoff: backoff.Config{
 				BaseDelay:  100 * time.Millisecond,
@@ -101,8 +103,24 @@ func (o *Options) runGoDataplane(dataplaneID string, parsedCertData *tls.ParsedC
 		logrus.Errorf("Failed to start dataplane server: %v.", err)
 	}()
 
-	// Start xDS client, if it fails to start we keep retrying to connect to the controlplane host
 	xdsClient := dpclient.NewXDSClient(dataplane, controlplaneClient)
+
+	readinessListenAddress := fmt.Sprintf("0.0.0.0:%d", api.ReadinessListenPort)
+	httpServer := utilhttp.NewServer("go-dataplane-readiness-http", nil)
+	if err := httpServer.Listen(readinessListenAddress); err != nil {
+		return fmt.Errorf("cannot listen for readiness: %w", err)
+	}
+	httpServer.Router().Get("/", func(w http.ResponseWriter, r *http.Request) {
+		if !xdsClient.IsReady() || !dataplane.IsReady() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+		}
+	})
+	go func() {
+		err := httpServer.Start()
+		logrus.Errorf("Failed to start readiness server: %v.", err)
+	}()
+
+	// Start xDS client, if it fails to start we keep retrying to connect to the controlplane host
 	err = xdsClient.Run()
 	return fmt.Errorf("xDS Client stopped: %w", err)
 }

pkg/bootstrap/cert.go

@@ -18,6 +18,8 @@ import (
 	"path/filepath"
 
 	"github.com/clusterlink-net/clusterlink/cmd/clusterlink/config"
+	cpapi "github.com/clusterlink-net/clusterlink/pkg/controlplane/api"
+	dpapi "github.com/clusterlink-net/clusterlink/pkg/dataplane/api"
 )
 
 // Certificate represents a clusterlink certificate.
@@ -69,9 +71,9 @@ func CreateCACertificate() (*Certificate, error) {
 func CreateControlplaneCertificate(caCert *Certificate) (*Certificate, error) {
 	cert, err := createCertificate(&certificateConfig{
 		Parent:   caCert.cert,
-		Name:     "cl-controlplane",
+		Name:     cpapi.Name,
 		IsServer: true,
-		DNSNames: []string{"cl-controlplane"},
+		DNSNames: []string{cpapi.Name},
 	})
 	if err != nil {
 		return nil, err
@@ -84,9 +86,9 @@ func CreateControlplaneCertificate(caCert *Certificate) (*Certificate, error) {
 func CreateDataplaneCertificate(caCert *Certificate) (*Certificate, error) {
 	cert, err := createCertificate(&certificateConfig{
 		Parent:   caCert.cert,
-		Name:     "cl-dataplane",
+		Name:     dpapi.Name,
 		IsClient: true,
-		DNSNames: []string{"cl-dataplane"},
+		DNSNames: []string{dpapi.Name},
 	})
 	if err != nil {
 		return nil, err