k0s/inttest/basic/basic_test.go

// SPDX-FileCopyrightText: 2020 k0s authors
// SPDX-License-Identifier: Apache-2.0

package basic

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"testing"

	"github.com/k0sproject/k0s/inttest/common"
	"github.com/k0sproject/k0s/pkg/apis/k0s/v1beta1"
	"github.com/k0sproject/k0s/pkg/constant"
	"github.com/k0sproject/k0s/pkg/kubernetes/watch"

	certificatesv1 "k8s.io/api/certificates/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/kubernetes"

	"github.com/BurntSushi/toml"
	"github.com/stretchr/testify/suite"
)

type BasicSuite struct {
	common.BootlooseSuite
}

type (
	CSR     = certificatesv1.CertificateSigningRequest
	CSRList = certificatesv1.CertificateSigningRequestList
)

func (s *BasicSuite) TestK0sGetsUp() {
	ctx := s.Context()
	customDataDir := "/var/lib/k0s/custom-data-dir"

	// Create an empty file to prove that k0s manage to rewrite a partially written file
	ssh, err := s.SSH(ctx, s.ControllerNode(0))
	s.Require().NoError(err)
	defer ssh.Disconnect()
	_, err = ssh.ExecWithOutput(ctx, fmt.Sprintf("mkdir -p %s/bin && touch -t 202201010000 %s/bin/kube-apiserver", customDataDir, customDataDir))
	s.Require().NoError(err)
	_, err = ssh.ExecWithOutput(ctx, "touch -t 202201010000 "+s.K0sFullPath)
	s.Require().NoError(err)
	_, err = ssh.ExecWithOutput(ctx, "mkdir -p /run/k0s/konnectivity-server/ && touch -t 202201010000 /run/k0s/konnectivity-server/konnectivity-server.sock")
	s.Require().NoError(err)

	dataDirOpt := "--data-dir=" + customDataDir
	s.Require().NoError(s.InitController(0, dataDirOpt))

	token, err := s.GetJoinToken("worker", dataDirOpt)
	s.Require().NoError(err)
	s.NoError(s.RunWorkersWithToken(token, `--labels="k0sproject.io/foo=bar"`, `--kubelet-extra-args=" --address=0.0.0.0  --event-burst=10"`))

	kc, err := s.KubeClient(s.ControllerNode(0), dataDirOpt)
	if err != nil {
		s.FailNow("failed to obtain Kubernetes client", err)
	}
	restConfig, err := s.GetKubeConfig(s.ControllerNode(0), "")
	s.Require().NoError(err)

	err = s.WaitForNodeReady(s.WorkerNode(0), kc)
	s.NoError(err)

	if labels, err := s.GetNodeLabels(s.WorkerNode(0), kc); s.NoError(err) {
		s.Equal("bar", labels["k0sproject.io/foo"])
	}

	err = s.WaitForNodeReady(s.WorkerNode(1), kc)
	s.NoError(err)

	s.AssertSomeKubeSystemPods(kc)

	s.T().Log("waiting to see kube-router pods ready")
	s.NoError(common.WaitForKubeRouterReady(ctx, kc), "kube-router did not start")

	s.Require().NoError(s.checkCertPerms(ctx, s.ControllerNode(0)))

	s.T().Log("Waiting for all worker CSRs to be approved")
	s.Require().NoError(s.checkCSRs(ctx, kc))

	s.Require().NoError(s.verifyKubeletAddressFlag(ctx, s.WorkerNode(0)))
	s.Require().NoError(s.verifyKubeletAddressFlag(ctx, s.WorkerNode(1)))
	for _, lease := range []string{"kube-scheduler", "kube-controller-manager"} {
		s.T().Logf("Waiting for %s lease", lease)
		_, err := common.WaitForLease(ctx, kc, lease, metav1.NamespaceSystem)
		s.Require().NoError(err, lease)
	}

	s.Require().NoError(common.VerifyKonnectivityMesh(ctx, restConfig, kc, s.T(), uint(s.ControllerCount), uint(s.WorkerCount)), "While verifying konnectivity mesh")
	var wg sync.WaitGroup
	for i := range s.WorkerCount {
		t, node := s.T(), s.WorkerNode(i)
		wg.Go(func() {
			if verifyCAdvisorMetrics(ctx, t, kc, node) {
				t.Log("Verified cAdvisor metrics on", node)
			}
		})
	}
	defer wg.Wait()

	s.T().Log("checking kube-router gobgp functionality")
	kubeRouterPods, err := kc.CoreV1().Pods(metav1.NamespaceSystem).List(ctx, metav1.ListOptions{LabelSelector: "k8s-app=kube-router"})
	s.Require().NoError(err)
	// Just take the first running pod for execing the gobgp command
	for _, pod := range kubeRouterPods.Items {
		if pod.Status.Phase == corev1.PodRunning {
			out, err := common.PodExecCmdOutput(kc, restConfig, pod.Name, metav1.NamespaceSystem, "gobgp global")
			s.Require().NoError(err)
			// Check that the output contains the default AS number, that's a sign that gobgp is working
			s.Regexp(`AS:\s+64512`, out)
			break
		}
	}

	s.verifyContainerdDefaultConfig(ctx)

	s.Require().NoError(s.probeCoreDNSAntiAffinity(ctx, kc))
}

func (s *BasicSuite) checkCertPerms(ctx context.Context, node string) error {
	ssh, err := s.SSH(ctx, node)
	if err != nil {
		return err
	}
	defer ssh.Disconnect()

	// Check that all .key files have 640 permissions
	keyOutput, err := ssh.ExecWithOutput(ctx, `find /var/lib/k0s/custom-data-dir/pki/ -name '*.key' -a \! -perm 0640`)
	if err != nil {
		return err
	}

	if keyOutput != "" {
		return fmt.Errorf("some private key files having non 640 permissions: %s", keyOutput)
	}

	// Check that .conf files have either 640 or 600 permissions (admin.conf, scheduler.conf, and ccm.conf use 600, others use 640)
	confOutput, err := ssh.ExecWithOutput(ctx, `find /var/lib/k0s/custom-data-dir/pki/ -name '*.conf' -a \! -perm 0640 -a \! -perm 0600`)
	if err != nil {
		return err
	}

	if confOutput != "" {
		return fmt.Errorf("some private conf files having non 640/600 permissions: %s", confOutput)
	}

	return nil
}

// Verifies that kubelet process has the address flag set
func (s *BasicSuite) verifyKubeletAddressFlag(ctx context.Context, node string) error {
	ssh, err := s.SSH(ctx, node)
	if err != nil {
		return err
	}
	defer ssh.Disconnect()

	output, err := ssh.ExecWithOutput(ctx, `grep -e '--address=0.0.0.0' /proc/$(pidof kubelet)/cmdline`)
	if err != nil {
		return err
	}
	if output != "--address=0.0.0.0" {
		return errors.New("kubelet does not have the address flag set")
	}

	return nil
}

func (s *BasicSuite) checkCSRs(ctx context.Context, kc *kubernetes.Clientset) error {
	// Wait until CSRs for all worker nodes got signed
	approvedNodes := map[string]struct{}{}

	return watch.FromClient[*CSRList, CSR](kc.CertificatesV1().CertificateSigningRequests()).
		WithFieldSelector(fields.OneTermEqualSelector("spec.signerName", "kubernetes.io/kubelet-serving")).
		WithErrorCallback(common.RetryWatchErrors(s.T().Logf)).
		Until(ctx, func(csr *CSR) (bool, error) {
			if !strings.HasPrefix(csr.Spec.Username, "system:node:worker") {
				return false, nil
			}
			if _, alreadyApproved := approvedNodes[csr.Spec.Username]; alreadyApproved {
				return false, nil
			}

			if reason, ok := getCSRApprovalReason(csr); !ok {
				s.T().Logf("CSR for %s is not yet approved", csr.Spec.Username)
				return false, nil
			} else if reason != "Autoapproved by K0s CSRApprover" {
				return false, fmt.Errorf("CSR for %s has an unexpected approval reason: %q", csr.Spec.Username, reason)
			}

			s.T().Logf("CSR for %s is approved", csr.Spec.Username)

			approvedNodes[csr.Spec.Username] = struct{}{}
			if len(approvedNodes) == s.WorkerCount {
				return true, nil
			}

			return false, nil
		})
}

func getCSRApprovalReason(csr *CSR) (string, bool) {
	for _, condition := range csr.Status.Conditions {
		if condition.Type != certificatesv1.CertificateApproved {
			continue
		}
		return condition.Reason, true
	}

	return "", false
}

func (s *BasicSuite) verifyContainerdDefaultConfig(ctx context.Context) {
	var defaultConfig bytes.Buffer
	ssh, err := s.SSH(ctx, s.WorkerNode(0))
	if !s.NoError(err) {
		return
	}
	defer ssh.Disconnect()

	if !s.NoError(ssh.Exec(ctx, "/var/lib/k0s/bin/containerd --config=/etc/k0s/containerd.toml config dump", common.SSHStreams{Out: &defaultConfig})) {
		return
	}

	var parsedConfig struct {
		Plugins struct {
			CRIIMages struct {
				PinnedImages struct {
					Sandbox string `toml:"sandbox"`
				} `toml:"pinned_images"`
			} `toml:"io.containerd.cri.v1.images"`
		} `toml:"plugins"`
	}

	_, err = toml.Decode(defaultConfig.String(), &parsedConfig)
	if !s.NoError(err) {
		return
	}

	s.Equal((&v1beta1.ImageSpec{
		Image:   constant.KubePauseContainerImage,
		Version: constant.KubePauseContainerImageVersion,
	}).URI(), parsedConfig.Plugins.CRIIMages.PinnedImages.Sandbox)
}

func (s *BasicSuite) probeCoreDNSAntiAffinity(ctx context.Context, kc *kubernetes.Clientset) error {
	// Wait until both CoreDNS Pods got assigned to a node
	pods := map[string]string{}

	return watch.Pods(kc.CoreV1().Pods(metav1.NamespaceSystem)).
		WithLabels(labels.Set{"k8s-app": "kube-dns"}).
		WithErrorCallback(common.RetryWatchErrors(s.T().Logf)).
		Until(ctx, func(pod *corev1.Pod) (bool, error) {
			// Keep waiting until there's anti-affinity and node assignment
			if a := pod.Spec.Affinity; a == nil || a.PodAntiAffinity == nil {
				s.T().Logf("Pod %s doesn't have any pod anti-affinity", pod.Name)
				return false, nil
			}
			nodeName := pod.Spec.NodeName
			if nodeName == "" {
				s.T().Logf("Pod %s not scheduled yet: %+v", pod.Name, pod.Status)
				return false, nil
			}

			if prevName, ok := pods[nodeName]; ok && pod.Name != prevName {
				return false, fmt.Errorf("multiple CoreDNS pods scheduled on node %s: %s and %s", nodeName, prevName, pod.Name)
			}

			s.T().Logf("Pod %s scheduled on %s", pod.Name, pod.Spec.NodeName)

			pods[nodeName] = pod.Name
			return len(pods) > 1, nil
		})
}

func TestBasicSuite(t *testing.T) {
	s := BasicSuite{
		common.BootlooseSuite{
			ControllerCount: 1,
			WorkerCount:     2,
		},
	}
	suite.Run(t, &s)
}