helper_test.go

//go:build e2e

package e2e

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"math/rand"
	"net/http"
	"os"
	"strconv"
	"strings"
	"testing"
	"time"

	hrobot "github.com/syself/hrobot-go"
	corev1 "k8s.io/api/core/v1"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"

	"github.com/hetznercloud/hcloud-cloud-controller-manager/internal/annotation"
	"github.com/hetznercloud/hcloud-cloud-controller-manager/internal/hcops"
	"github.com/hetznercloud/hcloud-cloud-controller-manager/internal/testsupport"
	"github.com/hetznercloud/hcloud-cloud-controller-manager/internal/utils"
	"github.com/hetznercloud/hcloud-go/v2/hcloud"
)

// Load Balancer creation timeouts. Services that involve certificate
// provisioning take longer to become ready.
const (
	lbCreateTimeoutDefault = 8 * time.Minute
	lbCreateTimeoutCert    = 16 * time.Minute
)

// lbCreateTimeoutFor returns the timeout to use when waiting for the given
// Load Balancer service to become ready.
func lbCreateTimeoutFor(svc *corev1.Service) time.Duration {
	certAnnotations := []annotation.Name{
		annotation.LBSvcHTTPCertificates,
		annotation.LBSvcHTTPCertificateType,
		annotation.LBSvcHTTPManagedCertificateName,
		annotation.LBSvcHTTPManagedCertificateDomains,
	}
	for _, a := range certAnnotations {
		if _, ok := svc.Annotations[string(a)]; ok {
			return lbCreateTimeoutCert
		}
	}
	return lbCreateTimeoutDefault
}

var rng *rand.Rand

func init() {
	rng = rand.New(rand.NewSource(time.Now().UnixNano()))
}

// pollBackoff is the standard exponential backoff used while polling for
// k8s/hcloud state in this suite: 1s base, doubling, capped at 30s.
var pollBackoff = hcloud.ExponentialBackoffWithOpts(hcloud.ExponentialBackoffOpts{
	Base:       time.Second,
	Multiplier: 2,
	Cap:        30 * time.Second,
})

type TestCluster struct {
	hcloud    *hcloud.Client
	hrobot    hrobot.RobotClient
	k8sClient *kubernetes.Clientset

	// envName is used as the name prefix in all resources
	envName    string
	certDomain string

	certificates  *utils.SyncSet[int64]
	loadBalancers *utils.SyncSet[string]
}

func (tc *TestCluster) Start() error {
	// Hetzner Cloud Client
	token := os.Getenv("HCLOUD_TOKEN")
	if token == "" {
		return fmt.Errorf("no valid HCLOUD_TOKEN found")
	}

	opts := []hcloud.ClientOption{
		hcloud.WithToken(token),
		hcloud.WithApplication("hcloud-ccm-testsuite", "1.0"),
		hcloud.WithPollOpts(hcloud.PollOpts{
			BackoffFunc: hcloud.ExponentialBackoffWithOpts(hcloud.ExponentialBackoffOpts{
				Base:       time.Second,
				Multiplier: 2,
				Cap:        10 * time.Second,
			}),
		}),
	}
	tc.hcloud = hcloud.NewClient(opts...)

	// Hetzner Robot Client
	if enabled := os.Getenv("ROBOT_ENABLED"); enabled == "true" {
		robotUser := os.Getenv("ROBOT_USER")
		robotPassword := os.Getenv("ROBOT_PASSWORD")
		tc.hrobot = hrobot.NewBasicAuthClient(robotUser, robotPassword)
	}

	tc.envName = os.Getenv("ENV_NAME")

	loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
	configOverrides := &clientcmd.ConfigOverrides{}

	kubeConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides)
	clientConfig, err := kubeConfig.ClientConfig()
	if err != nil {
		return fmt.Errorf("kubeConfig.ClientConfig: %s", err)
	}

	tc.k8sClient, err = kubernetes.NewForConfig(clientConfig)
	if err != nil {
		return fmt.Errorf("kubernetes.NewForConfig: %s", err)
	}

	// Tests using this value should skip if empty
	// The domain specified here must be available in Hetzner DNS of the account running the tests.
	tc.certDomain = os.Getenv("CERT_DOMAIN")

	tc.certificates = utils.NewSyncSet[int64]()
	tc.loadBalancers = utils.NewSyncSet[string]()

	return nil
}

func (tc *TestCluster) Stop() error {
	errs := make([]error, 0, tc.loadBalancers.Size()+tc.certificates.Size())
	ctx := context.Background()

	uids := tc.loadBalancers.All()
	selector := fmt.Sprintf("%s in (%s)", hcops.LabelServiceUID, strings.Join(uids, ","))
	lbs, err := tc.hcloud.LoadBalancer.AllWithOpts(ctx, hcloud.LoadBalancerListOpts{
		ListOpts: hcloud.ListOpts{
			LabelSelector: selector,
		},
	})
	if err != nil {
		return fmt.Errorf("fetching load balancers via selector %s: %w", selector, err)
	}
	for _, lb := range lbs {
		// Stop is called after `m.Run()`, so we can't use t.Log anymore.
		fmt.Printf("force-deleting leaked load balancer %d (%s)\n", lb.ID, lb.Name)
		if _, err := tc.hcloud.LoadBalancer.Delete(ctx, lb); err != nil {
			errs = append(errs, fmt.Errorf("delete leaked load balancer %d failed: %w", lb.ID, err))
		}
	}

	for _, item := range tc.certificates.All() {
		// Stop is called after `m.Run()`, so we can't use t.Log anymore.
		fmt.Printf("deleting certificate %d\n", item)
		if _, err := tc.hcloud.Certificate.Delete(ctx, &hcloud.Certificate{ID: item}); err != nil {
			errs = append(errs, fmt.Errorf("delete certificate %d failed: %w", item, err))
		}
	}

	return errors.Join(errs...)
}

// CreateTLSCertificate creates a TLS certificate used for testing and posts it
// to the Hetzner Cloud backend.
//
// The baseName of the certificate gets a random number suffix attached.
// baseName and suffix are separated by a single "-" character.
func (tc *TestCluster) CreateTLSCertificate(t *testing.T, baseName string) (*hcloud.Certificate, error) {
	rndInt := rng.Int()
	name := fmt.Sprintf("%s-%d", baseName, rndInt)

	p := testsupport.NewTLSPair(t, fmt.Sprintf("www.example%d.com", rndInt))
	opts := hcloud.CertificateCreateOpts{
		Name:        name,
		Certificate: p.Cert,
		PrivateKey:  p.Key,
	}
	cert, _, err := tc.hcloud.Certificate.Create(t.Context(), opts)
	if err != nil {
		return nil, fmt.Errorf("%s: %w", name, err)
	}
	if cert == nil {
		return nil, errors.New("no certificate created")
	}

	tc.certificates.Add(cert.ID)

	return cert, nil
}

// NetworkName returns the network name.
func (tc *TestCluster) NetworkName() string {
	return tc.envName
}

// ControlNodeName returns the control node name.
func (tc *TestCluster) ControlNodeName() string {
	return fmt.Sprintf("%s-control", tc.envName)
}

// WorkerNodeName returns the worker node name, zero indexed.
func (tc *TestCluster) WorkerNodeName(index int) string {
	return fmt.Sprintf("%s-worker-%d", tc.envName, index)
}

type lbTestHelper struct {
	t *testing.T

	namespace string
	podName   string
	port      int
}

// DeployTestPod deploys a basic nginx pod within the k8s cluster
// and waits until it is "ready".
func (l *lbTestHelper) DeployTestPod() (*corev1.Pod, error) {
	l.t.Helper()

	ctx := l.t.Context()

	if l.namespace == "" {
		l.namespace = "hccm-test-" + strconv.Itoa(rand.Int())
	}
	_, err := testCluster.k8sClient.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{
			Name: l.namespace,
		},
	}, metav1.CreateOptions{})
	if err != nil && !k8serrors.IsAlreadyExists(err) {
		return nil, fmt.Errorf("error deploying test pod: %w", err)
	}

	podName := fmt.Sprintf("pod-%s", l.podName)
	testPod := corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name: podName,
			Labels: map[string]string{
				"app": podName,
			},
		},
		Spec: corev1.PodSpec{
			Containers: []corev1.Container{
				{
					Name:  "nginx-hello-world",
					Image: "nginxdemos/hello:plain-text",
					Ports: []corev1.ContainerPort{
						{
							ContainerPort: 80,
							Name:          "http",
						},
					},
				},
			},
		},
	}

	pod, err := testCluster.k8sClient.CoreV1().Pods(l.namespace).Create(ctx, &testPod, metav1.CreateOptions{})
	if err != nil {
		return nil, fmt.Errorf("could not create test pod: %w", err)
	}

	err = wait.PollUntilContextTimeout(ctx, 1*time.Second, 2*time.Minute, false, func(ctx context.Context) (done bool, err error) {
		p, err := testCluster.k8sClient.CoreV1().Pods(l.namespace).Get(ctx, podName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}
		for _, condition := range p.Status.Conditions {
			if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
				return true, nil
			}
		}
		pod = p
		return false, nil
	})
	if err != nil {
		return nil, fmt.Errorf("pod %s did not come up after 2 minutes: %w", podName, err)
	}

	return pod, nil
}

// ServiceDefinition returns a service definition for a Hetzner Cloud Load Balancer (k8s service).
func (l *lbTestHelper) ServiceDefinition(pod *corev1.Pod, annotations map[string]string) *corev1.Service {
	l.t.Helper()

	port := l.port
	if port == 0 {
		port = 80
	}

	return &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:        fmt.Sprintf("svc-%s", l.podName),
			Annotations: annotations,
		},
		Spec: corev1.ServiceSpec{
			Selector: map[string]string{
				"app": pod.Name,
			},
			Type: corev1.ServiceTypeLoadBalancer,
			Ports: []corev1.ServicePort{
				{
					Port:       int32(port),
					TargetPort: intstr.FromInt(80),
					Name:       "http",
				},
			},
		},
	}
}

// CreateService creates a k8s service based on the given service definition
// and waits until it is "ready". The wait timeout is derived from the service
// annotations: services that provision certificates get a longer timeout.
func (l *lbTestHelper) CreateService(lbSvc *corev1.Service) (*corev1.Service, error) {
	l.t.Helper()

	timeout := lbCreateTimeoutFor(lbSvc)

	lbSvc, err := testCluster.k8sClient.CoreV1().Services(l.namespace).Create(l.t.Context(), lbSvc, metav1.CreateOptions{})
	if err != nil {
		return nil, fmt.Errorf("could not create service: %w", err)
	}

	testCluster.loadBalancers.Add(string(lbSvc.UID))

	ctx, cancel := context.WithTimeout(l.t.Context(), timeout)
	defer cancel()

	retries := 0
	for {
		svc, err := testCluster.k8sClient.CoreV1().Services(l.namespace).Get(ctx, lbSvc.Name, metav1.GetOptions{})
		if err != nil {
			return nil, fmt.Errorf("error fetching load balancer service: %w", err)
		}

		if len(svc.Status.LoadBalancer.Ingress) > 0 {
			return svc, nil
		}

		select {
		case <-ctx.Done():
			return nil, fmt.Errorf("timed out waiting for load balancer service to receive ingress IPs")
		case <-time.After(pollBackoff(retries)):
			retries++
		}
	}
}

// TearDown deletes the created pod and service.
func (l *lbTestHelper) TearDown() {
	l.t.Helper()

	// No namespace was created yet (e.g. DeployTestPod never ran because a
	// prior step failed); nothing to clean up.
	if l.namespace == "" {
		return
	}

	// Use context.Background() rather than t.Context(): cleanup must run to
	// completion even when the test has already been cancelled or failed.
	ctx := context.Background()

	err := wait.PollUntilContextTimeout(ctx, 1*time.Second, 3*time.Minute, true, func(ctx context.Context) (bool, error) {
		err := testCluster.k8sClient.CoreV1().Namespaces().Delete(ctx, l.namespace, metav1.DeleteOptions{})
		if err != nil && !k8serrors.IsNotFound(err) {
			return false, err
		}
		return k8serrors.IsNotFound(err), nil
	})
	if err != nil {
		// The cluster is deleted afterward, so we can info log this error
		l.t.Logf("error tearing down test namespace: %v", err)
	}
}

// WaitForHTTPAvailable tries to connect to the given IP via HTTP or HTTPS
// (controlled by useHTTPS). It uses exponential backoff starting at 1s and
// capping at 30s, waiting up to 8 minutes for a successful HTTP 200 response.
// Each individual request has a 5s timeout.
func (l *lbTestHelper) WaitForHTTPAvailable(ingressIP string, useHTTPS bool) error {
	l.t.Helper()

	client := &http.Client{
		Timeout: 5 * time.Second,
		Transport: &http.Transport{
			TLSClientConfig: &tls.Config{
				InsecureSkipVerify: true, // nolint
			},
		},
	}
	proto := "http"
	if useHTTPS {
		proto = "https"
	}

	ctx, cancel := context.WithTimeout(l.t.Context(), 8*time.Minute)
	defer cancel()

	retries := 0
	for {
		resp, err := client.Get(fmt.Sprintf("%s://%s", proto, ingressIP))
		if err != nil {
			l.t.Logf("request to %s failed, keep waiting: %v", ingressIP, err)
		} else {
			resp.Body.Close()
			switch resp.StatusCode {
			case http.StatusOK:
				return nil
			case http.StatusServiceUnavailable:
				l.t.Log("service still unavailable, keep waiting")
			default:
				return fmt.Errorf("got unexpected HTTP status %d", resp.StatusCode)
			}
		}

		select {
		case <-ctx.Done():
			return fmt.Errorf("timed out after 8m waiting for %s to be available", ingressIP)
		case <-time.After(pollBackoff(retries)):
			retries++
		}
	}
}