Custom Go-based dataplane for xDS control plane APIs

Makefile

@@ -58,13 +58,15 @@ build:
 	$(GO) build -o ./bin/dataplane ./cmd/dataplane/main.go
 	$(GO) build -o ./bin/cl-controlplane ./cmd/cl-controlplane
 	$(GO) build -o ./bin/cl-dataplane ./cmd/cl-dataplane
+	$(GO) build -o ./bin/cl-go-dataplane ./cmd/cl-go-dataplane
 	$(GO) build -o ./bin/cl-adm ./cmd/cl-adm
 
 
 docker-build: build
 	docker build --progress=plain --rm --tag mbg .
 	docker build --progress=plain --rm --tag cl-controlplane -f ./cmd/cl-controlplane/Dockerfile .
 	docker build --progress=plain --rm --tag cl-dataplane -f ./cmd/cl-dataplane/Dockerfile .
+	docker build --progress=plain --rm --tag cl-go-dataplane -f ./cmd/cl-go-dataplane/Dockerfile .
 	docker build --progress=plain --rm --tag gwctl -f ./cmd/gwctl/Dockerfile .
 
 build-image:

cmd/cl-adm/cmd/create/create_peer.go

@@ -23,12 +23,15 @@ type PeerOptions struct {
 	Name string
 	// Dataplanes is the number of dataplanes to create.
 	Dataplanes uint16
+	// DataplaneType is the type of dataplane to create (envoy or go-based)
+	DataplaneType string
 }
 
 // AddFlags adds flags to fs and binds them to options.
 func (o *PeerOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&o.Name, "name", "", "Peer name.")
 	fs.Uint16Var(&o.Dataplanes, "dataplanes", 1, "Number of dataplanes.")
+	fs.StringVar(&o.DataplaneType, "dataplaneType", "envoy", "Type of dataplane")
 }
 
 // RequiredFlags are the names of flags that must be explicitly specified.
@@ -136,8 +139,9 @@ func (o *PeerOptions) Run() error {
 
 	// deployment configuration
 	args, err := templates.Config{
-		Peer:       o.Name,
-		Dataplanes: o.Dataplanes,
+		Peer:          o.Name,
+		Dataplanes:    o.Dataplanes,
+		DataplaneType: o.DataplaneType,
 	}.TemplateArgs()
 	if err != nil {
 		return err

cmd/cl-adm/templates/config.go

@@ -19,6 +19,9 @@ type Config struct {
 
 	// Dataplanes is the number of dataplane servers to run.
 	Dataplanes uint16
+
+	// DataplaneType is the type of dataplane to create (envoy or go-based)
+	DataplaneType string
 }
 
 // TemplateArgs returns arguments for instantiating a text/template
@@ -64,8 +67,9 @@ func (c Config) TemplateArgs() (map[string]interface{}, error) {
 	}
 
 	return map[string]interface{}{
-		"peer":       c.Peer,
-		"dataplanes": c.Dataplanes,
+		"peer":          c.Peer,
+		"dataplanes":    c.Dataplanes,
+		"dataplaneType": c.DataplaneType,
 
 		"fabricCA":         base64.StdEncoding.EncodeToString(fabricCA),
 		"peerCA":           base64.StdEncoding.EncodeToString(peerCA),

cmd/cl-adm/templates/docker.go

@@ -27,6 +27,7 @@ cl-controlplane \
 --log-level info \
 --log-file {{.persistencyDirectoryMountPath}}/log.log \
 
+{{ if (eq .dataplaneType "envoy") }}
 docker run -itd \
 --name {{.peer}}-dataplane \
 -v $FABRIC_DIR/{{.fabricCAPath}}:{{.dataplaneCAMountPath}} \
@@ -38,6 +39,19 @@ cl-dataplane \
 --controlplane-host {{.peer}}-controlplane \
 --log-level info \
 --log-file {{.persistencyDirectoryMountPath}}/log.log
+{{ else }}
+docker run -itd \
+--name {{.peer}}-dataplane \
+-v $FABRIC_DIR/{{.fabricCAPath}}:{{.dataplaneCAMountPath}} \
+-v $FABRIC_DIR/{{.dataplaneCertPath}}:{{.dataplaneCertMountPath}} \
+-v $FABRIC_DIR/{{.dataplaneKeyPath}}:{{.dataplaneKeyMountPath}} \
+-v $FABRIC_DIR/{{.dataplanePersistencyDirectory}}:{{.persistencyDirectoryMountPath}} \
+cl-go-dataplane \
+cl-go-dataplane \
+--controlplane-host {{.peer}}-controlplane \
+--log-level info \
+--log-file {{.persistencyDirectoryMountPath}}/log.log
+{{ end }}
 
 docker run -itd \
 --name {{.peer}}-gwctl \

cmd/cl-adm/templates/k8s.go

@@ -135,7 +135,11 @@ spec:
             secretName: cl-dataplane
       containers:
         - name: dataplane
+          {{ if (eq .dataplaneType "envoy") }}
           image: cl-dataplane
+          {{ else }}
+          image: cl-go-dataplane
+          {{ end }}
           imagePullPolicy: IfNotPresent
           args: ["--controlplane-host", "cl-controlplane", "--log-level", "info"]
           ports:

cmd/cl-go-dataplane/Dockerfile

@@ -0,0 +1,13 @@
+FROM alpine:3.14
+
+# Copy binary
+RUN mkdir -p /usr/local/bin
+COPY ./bin/cl-go-dataplane /usr/local/bin/cl-go-dataplane
+
+# Create directory for private keys
+RUN mkdir -p /etc/ssl/private
+
+# Create directory for certificates
+RUN mkdir -p /etc/ssl/certs
+
+ENTRYPOINT ["/usr/local/bin/cl-go-dataplane"]

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

@@ -0,0 +1,156 @@
+package app
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"strconv"
+
+	"github.com/google/uuid"
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+
+	cpapi "github.com/clusterlink-net/clusterlink/pkg/controlplane/api"
+	"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"
+	"github.com/clusterlink-net/clusterlink/pkg/util"
+)
+
+const (
+	// logLevel is the default log level.
+	logLevel = "warn"
+
+	// CAFile is the path to the certificate authority file.
+	CAFile = "/etc/ssl/certs/clink_ca.pem"
+	// CertificateFile is the path to the certificate file.
+	CertificateFile = "/etc/ssl/certs/clink-dataplane.pem"
+	// KeyFile is the path to the private-key file.
+	KeyFile = "/etc/ssl/private/clink-dataplane.pem"
+
+	// dataplaneServerAddress is the address of the dataplane HTTP server for accepting ingress dataplane connections.
+	dataplaneServerAddress = "127.0.0.1:8443"
+)
+
+// Options contains everything necessary to create and run a dataplane.
+type Options struct {
+	// ControlplaneHost is the IP/hostname of the controlplane.
+	ControlplaneHost string
+	// LogFile is the path to file where logs will be written.
+	LogFile string
+	// LogLevel is the log level.
+	LogLevel string
+}
+
+// AddFlags adds flags to fs and binds them to options.
+func (o *Options) AddFlags(fs *pflag.FlagSet) {
+	fs.StringVar(&o.ControlplaneHost, "controlplane-host", "",
+		"The controlplane IP/hostname.")
+	fs.StringVar(&o.LogFile, "log-file", "",
+		"Path to a file where logs will be written. If not specified, logs will be printed to stderr.")
+	fs.StringVar(&o.LogLevel, "log-level", logLevel,
+		"The log level. One of fatal, error, warn, info, debug.")
+}
+
+// RequiredFlags are the names of flags that must be explicitly specified.
+func (o *Options) RequiredFlags() []string {
+	return []string{"controlplane-host"}
+}
+
+// Run the go dataplane.
+func (o *Options) runGoDataplane(peerName, dataplaneID string, parsedCertData *util.ParsedCertData) error {
+	controlplaneTarget := net.JoinHostPort(o.ControlplaneHost, strconv.Itoa(cpapi.ListenPort))
+
+	log.Infof("Starting go dataplane, Name: %s, ID: %s", peerName, dataplaneID)
+
+	dataplane := dpserver.NewDataplane(dataplaneID, controlplaneTarget, peerName, parsedCertData)
+	go func() {
+		err := dataplane.StartDataplaneServer(dataplaneServerAddress)
+		log.Errorf("Failed to start dataplane server: %v.", err)
+	}()
+
+	go func() {
+		err := dataplane.StartSNIServer(dataplaneServerAddress)
+		log.Error("Failed to start dataplane server", err)
+	}()
+
+	// Start xDS client, if it fails to start we keep retrying to connect to the controlplane host
+	tlsConfig := parsedCertData.ClientConfig(cpapi.GRPCServerName(peerName))
+	xdsClient := dpclient.NewXDSClient(dataplane, controlplaneTarget, tlsConfig)
+	err := xdsClient.Run()
+	return fmt.Errorf("xDS Client stopped: %v", err)
+}
+
+// Run the dataplane.
+func (o *Options) Run() error {
+	// set log file
+	if o.LogFile != "" {
+		f, err := os.OpenFile(o.LogFile, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0666)
+		if err != nil {
+			return fmt.Errorf("unable to open log file: %v", err)
+		}
+
+		defer func() {
+			if err := f.Close(); err != nil {
+				log.Errorf("Cannot close log file: %v", err)
+			}
+		}()
+
+		log.SetOutput(f)
+	}
+
+	// set log level
+	logLevel, err := log.ParseLevel(o.LogLevel)
+	if err != nil {
+		return fmt.Errorf("unable to set log level: %v", err)
+	}
+	log.SetLevel(logLevel)
+
+	// parse TLS files
+	parsedCertData, err := util.ParseTLSFiles(CAFile, CertificateFile, KeyFile)
+	if err != nil {
+		return err
+	}
+
+	dnsNames := parsedCertData.DNSNames()
+	if len(dnsNames) != 1 {
+		return fmt.Errorf("expected peer certificate to contain a single DNS name, but got %d", len(dnsNames))
+	}
+
+	peerName, err := api.StripServerPrefix(dnsNames[0])
+	if err != nil {
+		return err
+	}
+
+	// generate random dataplane ID
+	dataplaneID := uuid.New().String()
+	log.Infof("Dataplane ID: %s.", dataplaneID)
+
+	return o.runGoDataplane(peerName, dataplaneID, parsedCertData)
+}
+
+// NewCLGoDataplaneCommand creates a *cobra.Command object with default parameters.
+func NewCLGoDataplaneCommand() *cobra.Command {
+	opts := &Options{}
+
+	cmd := &cobra.Command{
+		Use:          "cl-go-dataplane",
+		Long:         `cl-go-dataplane: dataplane agent for allowing network connectivity of remote clients and services`,
+		SilenceUsage: true,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return opts.Run()
+		},
+	}
+
+	opts.AddFlags(cmd.Flags())
+
+	for _, flag := range opts.RequiredFlags() {
+		if err := cmd.MarkFlagRequired(flag); err != nil {
+			fmt.Printf("Error marking required flag '%s': %v\n", flag, err)
+			os.Exit(1)
+		}
+	}
+
+	return cmd
+}

cmd/cl-go-dataplane/cl-go-dataplane.go

@@ -0,0 +1,15 @@
+// The cl-dataplane binary runs an instance of a clink dataplane.
+package main
+
+import (
+	"os"
+
+	"github.com/clusterlink-net/clusterlink/cmd/cl-go-dataplane/app"
+)
+
+func main() {
+	command := app.NewCLGoDataplaneCommand()
+	if err := command.Execute(); err != nil {
+		os.Exit(1)
+	}
+}

cmd/dataplane/main.go

@@ -50,7 +50,6 @@ func main() {
 
 	// Set Dataplane
 	dp := dp.NewDataplane(&store.Store{ID: id, CertAuthority: ca, Cert: cert, Key: key, Dataplane: dataplane}, controlplane)
-
 	dp.StartServer(port)
 	log.Infof("Dataplane main process is finished")
 }

pkg/dataplane/client/fetcher.go

@@ -0,0 +1,94 @@
+package client
+
+import (
+	"context"
+	"fmt"
+
+	cluster "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	listener "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	client "github.com/envoyproxy/go-control-plane/pkg/client/sotw/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/clusterlink-net/clusterlink/pkg/dataplane/server"
+)
+
+type fetcher struct {
+	client       client.ADSClient
+	resourceType string
+	dataplane    *server.Dataplane
+	logger       *logrus.Entry
+}
+
+func (f *fetcher) handleClusters(resources []*anypb.Any) error {
+	for _, r := range resources {
+		c := &cluster.Cluster{}
+		err := anypb.UnmarshalTo(r, c, proto.UnmarshalOptions{})
+		if err != nil {
+			return err
+		}
+
+		f.logger.Debugf("Cluster: %s.", c.Name)
+		f.dataplane.AddCluster(c)
+	}
+	return nil
+}
+
+func (f *fetcher) handleListeners(resources []*anypb.Any) error {
+	for _, r := range resources {
+		l := &listener.Listener{}
+		err := anypb.UnmarshalTo(r, l, proto.UnmarshalOptions{})
+		if err != nil {
+			return err
+		}
+		f.logger.Debugf("Listener: %s.", l.Name)
+		f.dataplane.AddListener(l)
+	}
+	return nil
+}
+
+func (f *fetcher) Run() error {
+	for {
+		resp, err := f.client.Fetch()
+		if err != nil {
+			f.logger.Errorf("Failed to fetch %s: %v.", f.resourceType, err)
+			return err
+		}
+		switch f.resourceType {
+		case resource.ClusterType:
+			err := f.handleClusters(resp.Resources)
+			if err != nil {
+				f.logger.Errorf("Failed to handle clusters: %v.", err)
+			}
+		case resource.ListenerType:
+			err := f.handleListeners(resp.Resources)
+			if err != nil {
+				f.logger.Errorf("Failed to handle listeners: %v.", err)
+			}
+		default:
+			return fmt.Errorf("unknown resource type")
+		}
+
+		err = f.client.Ack()
+		if err != nil {
+			f.logger.Errorf("Failed to ack: %v.", err)
+		}
+	}
+}
+
+func newFetcher(ctx context.Context, conn *grpc.ClientConn, resourceType string, dataplane *server.Dataplane) (*fetcher, error) {
+	client := client.NewADSClient(ctx, &core.Node{Id: dataplane.ID}, resourceType)
+	err := client.InitConnect(conn)
+	if err != nil {
+		return nil, err
+	}
+	return &fetcher{client: client,
+		resourceType: resourceType,
+		dataplane:    dataplane,
+		logger:       logrus.WithField("component", "fetcher.xds.client"),
+	}, nil
+}