metropolis/test/e2e/main_test.go

// Copyright 2020 The Monogon Project Authors.
//
// SPDX-License-Identifier: Apache-2.0
//
// 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 e2e

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	_ "net/http"
	_ "net/http/pprof"
	"net/url"
	"os"
	"strings"
	"testing"
	"time"

	"google.golang.org/grpc"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	podv1 "k8s.io/kubernetes/pkg/api/v1/pod"

	common "source.monogon.dev/metropolis/node"
	"source.monogon.dev/metropolis/node/core/identity"
	"source.monogon.dev/metropolis/node/core/rpc"
	apb "source.monogon.dev/metropolis/proto/api"
	"source.monogon.dev/metropolis/test/launch"
	"source.monogon.dev/metropolis/test/launch/cluster"
	"source.monogon.dev/metropolis/test/util"
)

const (
	// Timeout for the global test context.
	//
	// Bazel would eventually time out the test after 900s ("large") if, for
	// some reason, the context cancellation fails to abort it.
	globalTestTimeout = 600 * time.Second

	// Timeouts for individual end-to-end tests of different sizes.
	smallTestTimeout = 60 * time.Second
	largeTestTimeout = 120 * time.Second
)

// TestE2E is the main E2E test entrypoint for single-node freshly-bootstrapped
// E2E tests. It starts a full Metropolis node in bootstrap mode and then runs
// tests against it. The actual tests it performs are located in the RunGroup
// subtest.
func TestE2E(t *testing.T) {
	// Run pprof server for debugging
	addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
	if err != nil {
		panic(err)
	}

	pprofListen, err := net.ListenTCP("tcp", addr)
	if err != nil {
		launch.Fatal("Failed to listen on pprof port: %s", pprofListen.Addr())
	}

	launch.Log("E2E: pprof server listening on %s", pprofListen.Addr())
	go func() {
		launch.Log("E2E: pprof server returned an error: %v", http.Serve(pprofListen, nil))
		pprofListen.Close()
	}()

	// Set a global timeout to make sure this terminates
	ctx, cancel := context.WithTimeout(context.Background(), globalTestTimeout)
	defer cancel()

	// Launch cluster.
	clusterOptions := cluster.ClusterOptions{
		NumNodes: 2,
	}
	cluster, err := cluster.LaunchCluster(ctx, clusterOptions)
	if err != nil {
		t.Fatalf("LaunchCluster failed: %v", err)
	}
	defer func() {
		err := cluster.Close()
		if err != nil {
			t.Fatalf("cluster Close failed: %v", err)
		}
	}()

	launch.Log("E2E: Cluster running, starting tests...")

	// Dial first node's curator.
	creds := rpc.NewAuthenticatedCredentials(cluster.Owner, rpc.WantInsecure())
	remote := net.JoinHostPort(cluster.NodeIDs[0], common.CuratorServicePort.PortString())
	cl, err := grpc.Dial(remote, grpc.WithContextDialer(cluster.DialNode), grpc.WithTransportCredentials(creds))
	if err != nil {
		t.Fatalf("failed to dial first node's curator: %v", err)
	}
	defer cl.Close()
	mgmt := apb.NewManagementClient(cl)

	// This exists to keep the parent around while all the children race.
	// It currently tests both a set of OS-level conditions and Kubernetes
	// Deployments and StatefulSets
	t.Run("RunGroup", func(t *testing.T) {
		t.Run("Cluster", func(t *testing.T) {
			util.TestEventual(t, "Retrieving cluster directory sucessful", ctx, 60*time.Second, func(ctx context.Context) error {
				res, err := mgmt.GetClusterInfo(ctx, &apb.GetClusterInfoRequest{})
				if err != nil {
					return fmt.Errorf("GetClusterInfo: %w", err)
				}

				// Ensure that the expected node count is present.
				nodes := res.ClusterDirectory.Nodes
				if want, got := clusterOptions.NumNodes, len(nodes); want != got {
					return fmt.Errorf("wanted %d nodes in cluster directory, got %d", want, got)
				}

				// Ensure the nodes have the expected addresses.
				addresses := make(map[string]bool)
				for _, n := range nodes {
					if len(n.Addresses) != 1 {
						return fmt.Errorf("node %s has no addresss", identity.NodeID(n.PublicKey))
					}
					address := n.Addresses[0].Host
					addresses[address] = true
				}

				for _, address := range []string{"10.1.0.2", "10.1.0.3"} {
					if !addresses[address] {
						return fmt.Errorf("address %q not found in directory", address)
					}
				}
				return nil
			})
			util.TestEventual(t, "Heartbeat test successful", ctx, 20*time.Second, cluster.AllNodesHealthy)
			util.TestEventual(t, "Node rejoin successful", ctx, 60*time.Second, func(ctx context.Context) error {
				// Ensure nodes rejoin the cluster after a reboot by reboting the 1st node.
				if err := cluster.RebootNode(ctx, 1); err != nil {
					return fmt.Errorf("while rebooting a node: %w", err)
				}
				return nil
			})
			util.TestEventual(t, "Heartbeat test successful", ctx, 20*time.Second, cluster.AllNodesHealthy)
		})
		t.Run("Kubernetes", func(t *testing.T) {
			t.Parallel()
			// TODO(q3k): use SOCKS proxy.
			clientSet, err := GetKubeClientSet(cluster, cluster.Ports[uint16(common.KubernetesAPIWrappedPort)])
			if err != nil {
				t.Fatal(err)
			}
			util.TestEventual(t, "Add KubernetesWorker roles", ctx, smallTestTimeout, func(ctx context.Context) error {
				// Find all nodes that are non-controllers.
				var ids []string
				srvN, err := mgmt.GetNodes(ctx, &apb.GetNodesRequest{})
				if err != nil {
					return fmt.Errorf("GetNodes: %w", err)
				}
				defer srvN.CloseSend()
				for {
					node, err := srvN.Recv()
					if err == io.EOF {
						break
					}
					if err != nil {
						return fmt.Errorf("GetNodes.Recv: %w", err)
					}
					if node.Roles.KubernetesController != nil {
						continue
					}
					if node.Roles.ConsensusMember != nil {
						continue
					}
					ids = append(ids, identity.NodeID(node.Pubkey))
				}

				if len(ids) < 1 {
					return fmt.Errorf("no appropriate nodes found")
				}

				// Make all these nodes as KubernetesWorker.
				for _, id := range ids {
					tr := true
					_, err := mgmt.UpdateNodeRoles(ctx, &apb.UpdateNodeRolesRequest{
						Node: &apb.UpdateNodeRolesRequest_Id{
							Id: id,
						},
						KubernetesWorker: &tr,
					})
					if err != nil {
						return fmt.Errorf("could not make node %q into kubernetes worker: %w", id, err)
					}
				}
				return nil
			})
			util.TestEventual(t, "Node is registered and ready", ctx, largeTestTimeout, func(ctx context.Context) error {
				nodes, err := clientSet.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
				if err != nil {
					return err
				}
				if len(nodes.Items) < 1 {
					return errors.New("node not yet registered")
				}
				node := nodes.Items[0]
				for _, cond := range node.Status.Conditions {
					if cond.Type != corev1.NodeReady {
						continue
					}
					if cond.Status != corev1.ConditionTrue {
						return fmt.Errorf("node not ready: %v", cond.Message)
					}
				}
				return nil
			})
			util.TestEventual(t, "Simple deployment", ctx, largeTestTimeout, func(ctx context.Context) error {
				_, err := clientSet.AppsV1().Deployments("default").Create(ctx, makeTestDeploymentSpec("test-deploy-1"), metav1.CreateOptions{})
				return err
			})
			util.TestEventual(t, "Simple deployment is running", ctx, largeTestTimeout, func(ctx context.Context) error {
				res, err := clientSet.CoreV1().Pods("default").List(ctx, metav1.ListOptions{LabelSelector: "name=test-deploy-1"})
				if err != nil {
					return err
				}
				if len(res.Items) == 0 {
					return errors.New("pod didn't get created")
				}
				pod := res.Items[0]
				if podv1.IsPodAvailable(&pod, 1, metav1.NewTime(time.Now())) {
					return nil
				}
				events, err := clientSet.CoreV1().Events("default").List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.namespace=default", pod.Name)})
				if err != nil || len(events.Items) == 0 {
					return fmt.Errorf("pod is not ready: %v", pod.Status.Phase)
				} else {
					return fmt.Errorf("pod is not ready: %v", events.Items[0].Message)
				}
			})
			util.TestEventual(t, "Simple deployment with runc", ctx, largeTestTimeout, func(ctx context.Context) error {
				deployment := makeTestDeploymentSpec("test-deploy-2")
				var runcStr = "runc"
				deployment.Spec.Template.Spec.RuntimeClassName = &runcStr
				_, err := clientSet.AppsV1().Deployments("default").Create(ctx, deployment, metav1.CreateOptions{})
				return err
			})
			util.TestEventual(t, "Simple deployment is running on runc", ctx, largeTestTimeout, func(ctx context.Context) error {
				res, err := clientSet.CoreV1().Pods("default").List(ctx, metav1.ListOptions{LabelSelector: "name=test-deploy-2"})
				if err != nil {
					return err
				}
				if len(res.Items) == 0 {
					return errors.New("pod didn't get created")
				}
				pod := res.Items[0]
				if podv1.IsPodAvailable(&pod, 1, metav1.NewTime(time.Now())) {
					return nil
				}
				events, err := clientSet.CoreV1().Events("default").List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.namespace=default", pod.Name)})
				if err != nil || len(events.Items) == 0 {
					return fmt.Errorf("pod is not ready: %v", pod.Status.Phase)
				} else {
					var errorMsg strings.Builder
					for _, msg := range events.Items {
						errorMsg.WriteString(" | ")
						errorMsg.WriteString(msg.Message)
					}
					return fmt.Errorf("pod is not ready: %v", errorMsg.String())
				}
			})
			util.TestEventual(t, "Simple StatefulSet with PVC", ctx, largeTestTimeout, func(ctx context.Context) error {
				_, err := clientSet.AppsV1().StatefulSets("default").Create(ctx, makeTestStatefulSet("test-statefulset-1", corev1.PersistentVolumeFilesystem), metav1.CreateOptions{})
				return err
			})
			util.TestEventual(t, "Simple StatefulSet with PVC is running", ctx, largeTestTimeout, func(ctx context.Context) error {
				res, err := clientSet.CoreV1().Pods("default").List(ctx, metav1.ListOptions{LabelSelector: "name=test-statefulset-1"})
				if err != nil {
					return err
				}
				if len(res.Items) == 0 {
					return errors.New("pod didn't get created")
				}
				pod := res.Items[0]
				if podv1.IsPodAvailable(&pod, 1, metav1.NewTime(time.Now())) {
					return nil
				}
				events, err := clientSet.CoreV1().Events("default").List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.namespace=default", pod.Name)})
				if err != nil || len(events.Items) == 0 {
					return fmt.Errorf("pod is not ready: %v", pod.Status.Phase)
				} else {
					return fmt.Errorf("pod is not ready: %v", events.Items[0].Message)
				}
			})
			util.TestEventual(t, "Simple StatefulSet with Block PVC", ctx, largeTestTimeout, func(ctx context.Context) error {
				_, err := clientSet.AppsV1().StatefulSets("default").Create(ctx, makeTestStatefulSet("test-statefulset-2", corev1.PersistentVolumeBlock), metav1.CreateOptions{})
				return err
			})
			util.TestEventual(t, "Simple StatefulSet with Block PVC is running", ctx, largeTestTimeout, func(ctx context.Context) error {
				res, err := clientSet.CoreV1().Pods("default").List(ctx, metav1.ListOptions{LabelSelector: "name=test-statefulset-2"})
				if err != nil {
					return err
				}
				if len(res.Items) == 0 {
					return errors.New("pod didn't get created")
				}
				pod := res.Items[0]
				if podv1.IsPodAvailable(&pod, 1, metav1.NewTime(time.Now())) {
					return nil
				}
				events, err := clientSet.CoreV1().Events("default").List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.namespace=default", pod.Name)})
				if err != nil || len(events.Items) == 0 {
					return fmt.Errorf("pod is not ready: %v", pod.Status.Phase)
				} else {
					return fmt.Errorf("pod is not ready: %v", events.Items[0].Message)
				}
			})
			util.TestEventual(t, "In-cluster self-test job", ctx, smallTestTimeout, func(ctx context.Context) error {
				_, err := clientSet.BatchV1().Jobs("default").Create(ctx, makeSelftestSpec("selftest"), metav1.CreateOptions{})
				return err
			})
			util.TestEventual(t, "In-cluster self-test job passed", ctx, smallTestTimeout, func(ctx context.Context) error {
				res, err := clientSet.BatchV1().Jobs("default").Get(ctx, "selftest", metav1.GetOptions{})
				if err != nil {
					return err
				}
				if res.Status.Failed > 0 {
					pods, err := clientSet.CoreV1().Pods("default").List(ctx, metav1.ListOptions{
						LabelSelector: "job-name=selftest",
					})
					if err != nil {
						return util.Permanent(fmt.Errorf("job failed but failed to find pod: %w", err))
					}
					if len(pods.Items) < 1 {
						return fmt.Errorf("job failed but pod does not exist")
					}
					lines, err := getPodLogLines(ctx, clientSet, pods.Items[0].Name, 1)
					if err != nil {
						return fmt.Errorf("job failed but could not get logs: %w", err)
					}
					if len(lines) > 0 {
						return util.Permanent(fmt.Errorf("job failed, last log line: %s", lines[0]))
					}
					return util.Permanent(fmt.Errorf("job failed, empty log"))
				}
				if res.Status.Succeeded > 0 {
					return nil
				}
				return fmt.Errorf("job still running")
			})
			util.TestEventual(t, "Prometheus node metrics retrieved", ctx, smallTestTimeout, func(ctx context.Context) error {
				pool := x509.NewCertPool()
				pool.AddCert(cluster.CACertificate)
				cl := http.Client{
					Transport: &http.Transport{
						TLSClientConfig: &tls.Config{
							Certificates: []tls.Certificate{cluster.Owner},
							RootCAs:      pool,
						},
						DialContext: func(ctx context.Context, _, addr string) (net.Conn, error) {
							return cluster.DialNode(ctx, addr)
						},
					},
				}
				u := url.URL{
					Scheme: "https",
					Host:   net.JoinHostPort(cluster.NodeIDs[0], common.MetricsPort.PortString()),
					Path:   "/metrics/node",
				}
				res, err := cl.Get(u.String())
				if err != nil {
					return err
				}
				defer res.Body.Close()
				if res.StatusCode != 200 {
					return fmt.Errorf("status code %d", res.StatusCode)
				}

				body, err := io.ReadAll(res.Body)
				if err != nil {
					return err
				}
				needle := "node_uname_info"
				if !strings.Contains(string(body), needle) {
					return util.Permanent(fmt.Errorf("could not find %q in returned response", needle))
				}
				return nil
			})
			if os.Getenv("HAVE_NESTED_KVM") != "" {
				util.TestEventual(t, "Pod for KVM/QEMU smoke test", ctx, smallTestTimeout, func(ctx context.Context) error {
					runcRuntimeClass := "runc"
					_, err := clientSet.CoreV1().Pods("default").Create(ctx, &corev1.Pod{
						ObjectMeta: metav1.ObjectMeta{
							Name: "vm-smoketest",
						},
						Spec: corev1.PodSpec{
							Containers: []corev1.Container{{
								Name:            "vm-smoketest",
								ImagePullPolicy: corev1.PullNever,
								Image:           "bazel/metropolis/vm/smoketest:smoketest_container",
								Resources: corev1.ResourceRequirements{
									Limits: corev1.ResourceList{
										"devices.monogon.dev/kvm": *resource.NewQuantity(1, ""),
									},
								},
							}},
							RuntimeClassName: &runcRuntimeClass,
							RestartPolicy:    corev1.RestartPolicyNever,
						},
					}, metav1.CreateOptions{})
					return err
				})
				util.TestEventual(t, "KVM/QEMU smoke test completion", ctx, smallTestTimeout, func(ctx context.Context) error {
					pod, err := clientSet.CoreV1().Pods("default").Get(ctx, "vm-smoketest", metav1.GetOptions{})
					if err != nil {
						return fmt.Errorf("failed to get pod: %w", err)
					}
					if pod.Status.Phase == corev1.PodSucceeded {
						return nil
					}
					events, err := clientSet.CoreV1().Events("default").List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.namespace=default", pod.Name)})
					if err != nil || len(events.Items) == 0 {
						return fmt.Errorf("pod is not ready: %v", pod.Status.Phase)
					} else {
						return fmt.Errorf("pod is not ready: %v", events.Items[len(events.Items)-1].Message)
					}
				})
			}
		})
	})
}