✨ sharded-test-server: support for running the caching layer

cmd/sharded-test-server/cache.go

@@ -0,0 +1,173 @@
+/*
+Copyright 2022 The KCP Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/abiosoft/lineprefix"
+	"github.com/fatih/color"
+
+	"k8s.io/client-go/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
+	"k8s.io/klog/v2"
+
+	"github.com/kcp-dev/kcp/cmd/test-server/helpers"
+	kcpclient "github.com/kcp-dev/kcp/pkg/client/clientset/versioned"
+	"github.com/kcp-dev/kcp/test/e2e/framework"
+)
+
+func startCacheServer(ctx context.Context, logDirPath, workingDir string) (<-chan error, string, error) {
+	red := color.New(color.BgHiRed, color.FgHiWhite).SprintFunc()
+	inverse := color.New(color.BgHiWhite, color.FgHiRed).SprintFunc()
+	out := lineprefix.New(
+		lineprefix.Prefix(red(strings.ToUpper("cache"))),
+		lineprefix.Color(color.New(color.FgHiRed)),
+	)
+	loggerOut := lineprefix.New(
+		lineprefix.Prefix(inverse(strings.ToUpper("cache"))),
+		lineprefix.Color(color.New(color.FgHiWhite)),
+	)
+	cacheWorkingDir := filepath.Join(workingDir, ".kcp-cache")
+	cachePort := 8012
+	commandLine := framework.DirectOrGoRunCommand("cache-server")
+	commandLine = append(
+		commandLine,
+		fmt.Sprintf("--root-directory=%s", cacheWorkingDir),
+		"--embedded-etcd-client-port=8010",
+		"--embedded-etcd-peer-port=8011",
+		fmt.Sprintf("--secure-port=%d", cachePort),
+	)
+	fmt.Fprintf(out, "running: %v\n", strings.Join(commandLine, " "))
+	cmd := exec.CommandContext(ctx, commandLine[0], commandLine[1:]...)
+
+	logFilePath := filepath.Join(logDirPath, ".kcp-cache", "out.log")
+	if err := os.MkdirAll(filepath.Dir(logFilePath), 0755); err != nil {
+		return nil, "", err
+	}
+	logFile, err := os.OpenFile(logFilePath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
+	if err != nil {
+		return nil, "", err
+	}
+
+	writer := helpers.NewHeadWriter(logFile, out)
+	cmd.Stdout = writer
+	cmd.Stdin = os.Stdin
+	cmd.Stderr = writer
+
+	if err := cmd.Start(); err != nil {
+		return nil, "", err
+	}
+
+	terminatedCh := make(chan error, 1)
+	go func() {
+		terminatedCh <- cmd.Wait()
+	}()
+
+	// wait for readiness
+	logger := klog.FromContext(ctx)
+	logger.Info("waiting for the cache server to be up")
+	cacheKubeconfigPath := filepath.Join(cacheWorkingDir, "cache.kubeconfig")
+	for {
+		time.Sleep(time.Second)
+
+		select {
+		case <-ctx.Done():
+			return nil, "", fmt.Errorf("context canceled")
+		case err := <-terminatedCh:
+			var exitErr *exec.ExitError
+			if err == nil {
+				return nil, "", fmt.Errorf("the cahce server terminated unexpectedly with exit code 0")
+			} else if errors.As(err, &exitErr) {
+				return nil, "", fmt.Errorf("the cache server terminated with exit code %d", exitErr.ExitCode())
+			}
+			return nil, "", fmt.Errorf("the cache server terminated with unknown error: %w", err)
+		default:
+		}
+
+		if _, err := os.Stat(filepath.Join(cacheWorkingDir, "apiserver.crt")); os.IsNotExist(err) {
+			logger.V(3).Info("failed to read the cache server certificate file", "err", err, "path", filepath.Join(cacheWorkingDir, "apiserver.crt"))
+			continue
+		}
+
+		if _, err := os.Stat(cacheKubeconfigPath); os.IsNotExist(err) {
+			cacheServerCert, err := ioutil.ReadFile(filepath.Join(cacheWorkingDir, "apiserver.crt"))
+			if err != nil {
+				return nil, "", err
+			}
+			cacheServerKubeConfig := clientcmdapi.Config{
+				Clusters: map[string]*clientcmdapi.Cluster{
+					"cache": {
+						Server:                   fmt.Sprintf("https://localhost:%d", cachePort),
+						CertificateAuthorityData: cacheServerCert,
+					},
+				},
+				Contexts: map[string]*clientcmdapi.Context{
+					"cache": {
+						Cluster: "cache",
+					},
+				},
+				CurrentContext: "cache",
+			}
+			if err = clientcmd.WriteToFile(cacheServerKubeConfig, cacheKubeconfigPath); err != nil {
+				return nil, "", err
+			}
+		}
+
+		cacheServerKubeConfig, err := clientcmd.LoadFromFile(cacheKubeconfigPath)
+		if err != nil {
+			return nil, "", err
+		}
+		cacheClientConfig := clientcmd.NewNonInteractiveClientConfig(*cacheServerKubeConfig, "cache", nil, nil)
+		cacheClientRestConfig, err := cacheClientConfig.ClientConfig()
+		if err != nil {
+			return nil, "", err
+		}
+		cacheClient, err := kcpclient.NewClusterForConfig(cacheClientRestConfig)
+		if err != nil {
+			return nil, "", err
+		}
+
+		res := cacheClient.RESTClient().Get().AbsPath("/readyz").Do(ctx)
+		if err := res.Error(); err != nil {
+			logger.V(3).Info("the cache server is not ready", "err", err)
+		} else {
+			var rc int
+			res.StatusCode(&rc)
+			if rc == http.StatusOK {
+				logger.V(3).Info("the cache server is ready")
+				break
+			}
+			if bs, err := res.Raw(); err != nil {
+				logger.V(3).Info("the cache server is not ready", "err", err)
+			} else {
+				logger.V(3).WithValues("rc", rc, "raw", string(bs)).Info("the cache server is not ready")
+			}
+		}
+	}
+	fmt.Fprintf(loggerOut, "the cache server is ready\n")
+	return terminatedCh, cacheKubeconfigPath, nil
+}

cmd/sharded-test-server/main.go

@@ -148,16 +148,30 @@ func start(proxyFlags, shardFlags []string, logDirPath, workDirPath string, numb
 		shardFlags = append(shardFlags, fmt.Sprintf("--shard-virtual-workspace-url=https://%s:7444", hostIP))
 	}
 
+	cacheServerErrCh := make(chan indexErrTuple)
+	cacheServerConfigPath := ""
+	if sets.NewString(shardFlags...).Has("--run-cache-server=true") {
+		cacheServerCh, configPath, err := startCacheServer(ctx, logDirPath, workDirPath)
+		if err != nil {
+			return fmt.Errorf("error starting the cache server: %w", err)
+		}
+		cacheServerConfigPath = configPath
+		go func() {
+			err := <-cacheServerCh
+			cacheServerErrCh <- indexErrTuple{0, err}
+		}()
+	}
+
 	// start shards
-	shardsErrCh := make(chan shardErrTuple)
+	shardsErrCh := make(chan indexErrTuple)
 	for i := 0; i < numberOfShards; i++ {
-		shardErrCh, err := startShard(ctx, i, shardFlags, servingCA, hostIP.String(), logDirPath, workDirPath)
+		shardErrCh, err := startShard(ctx, i, shardFlags, servingCA, hostIP.String(), logDirPath, workDirPath, cacheServerConfigPath)
 		if err != nil {
 			return err
 		}
 		go func(shardIndex int, shardErrCh <-chan error) {
 			err := <-shardErrCh
-			shardsErrCh <- shardErrTuple{shardIndex, err}
+			shardsErrCh <- indexErrTuple{shardIndex, err}
 
 		}(i, shardErrCh)
 	}
@@ -168,7 +182,7 @@ func start(proxyFlags, shardFlags []string, logDirPath, workDirPath string, numb
 	}
 
 	vwPort := "6444"
-	virtualWorkspacesErrCh := make(chan shardErrTuple)
+	virtualWorkspacesErrCh := make(chan indexErrTuple)
 	if standaloneVW {
 		// TODO: support multiple virtual workspace servers (i.e. multiple ports)
 		vwPort = "7444"
@@ -180,7 +194,7 @@ func start(proxyFlags, shardFlags []string, logDirPath, workDirPath string, numb
 			}
 			go func(vwIndex int, vwErrCh <-chan error) {
 				err := <-virtualWorkspaceErrCh
-				virtualWorkspacesErrCh <- shardErrTuple{vwIndex, err}
+				virtualWorkspacesErrCh <- indexErrTuple{vwIndex, err}
 			}(i, virtualWorkspaceErrCh)
 		}
 	}
@@ -195,12 +209,14 @@ func start(proxyFlags, shardFlags []string, logDirPath, workDirPath string, numb
 		return fmt.Errorf("shard %d exited: %w", shardIndexErr.index, shardIndexErr.error)
 	case vwIndexErr := <-virtualWorkspacesErrCh:
 		return fmt.Errorf("virtual workspaces %d exited: %w", vwIndexErr.index, vwIndexErr.error)
+	case cacheErr := <-cacheServerErrCh:
+		return fmt.Errorf("cache server exited: %w", cacheErr.error)
 	case <-ctx.Done():
 	}
 	return nil
 }
 
-type shardErrTuple struct {
+type indexErrTuple struct {
 	index int
 	error error
 }

cmd/sharded-test-server/shard.go

@@ -28,7 +28,7 @@ import (
 	shard "github.com/kcp-dev/kcp/cmd/test-server/kcp"
 )
 
-func startShard(ctx context.Context, n int, args []string, servingCA *crypto.CA, hostIP string, logDirPath, workDirPath string) (<-chan error, error) {
+func startShard(ctx context.Context, n int, args []string, servingCA *crypto.CA, hostIP string, logDirPath, workDirPath, cacheServerConfigPath string) (<-chan error, error) {
 	// create serving cert
 	hostnames := sets.NewString("localhost", hostIP)
 	klog.Infof("Creating shard server %d serving cert with hostnames %v", n, hostnames)
@@ -82,6 +82,9 @@ func startShard(ctx context.Context, n int, args []string, servingCA *crypto.CA,
 		fmt.Sprintf("--secure-port=%d", 6444+n),
 		"--virtual-workspaces-workspaces.authorization-cache.resync-period=1s",
 	)
+	if len(cacheServerConfigPath) > 0 {
+		args = append(args, fmt.Sprintf("--cache-server-kubeconfig-file=%s", cacheServerConfigPath))
+	}
 
 	return shard.Start(ctx,
 		fmt.Sprintf("kcp-%d", n),                              // name