Checks that the expected services are running on bootstrap / join

Currently, we're doing only a few checks that would ensure a successful
installation of the k8s-snap (the containerd path is free for us to use,
the ports meant for Kubernetes services are free, binaries are sane).

However, due to various reasons, the bootstrap / join process can still
succeed, even if some of the services are not running (e.g.: invalid
extra service arguments given in the bootstrap config, leading to a
service to fail due to an unknown or invalid argument). At the moment,
we're only verifying that kube-apiserver is accessible.

We're adding a few checks that the Kubernetes services are not stopped /
failed after we've started them.

Adds integration test for this scenario.

k8s/lib.sh

@@ -178,10 +178,10 @@ k8s::common::execute_service() {
 
   k8s::common::setup_env
 
+  set -xe
   # Source arguments and substitute environment variables. Will fail if we cannot read the file.
   declare -a args="($(cat "${SNAP_COMMON}/args/${service_name}"))"
 
-  set -xe
   exec "${SNAP}/bin/${service_name}" "${args[@]}"
 }
 

src/k8s/pkg/k8sd/app/cluster_util.go

@@ -4,10 +4,12 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"time"
 
 	"github.com/canonical/k8s/pkg/k8sd/setup"
 	"github.com/canonical/k8s/pkg/snap"
 	snaputil "github.com/canonical/k8s/pkg/snap/util"
+	"github.com/canonical/k8s/pkg/utils/control"
 	mctypes "github.com/canonical/microcluster/v2/rest/types"
 )
 
@@ -61,6 +63,20 @@ func waitApiServerReady(ctx context.Context, snap snap.Snap) error {
 	return nil
 }
 
+func waitControlPlaneServices(ctx context.Context, snap snap.Snap) error {
+	// The services may be able to start, appearing to be "active", but they might eventually fail due to
+	// various reasons, and they may be restarted. We're checking their activity a few times.
+	if err := control.Consistently(ctx, 3, 5*time.Second, func() error {
+		if err := snaputil.CheckControlPlaneServicesStates(ctx, snap, "active"); err != nil {
+			return fmt.Errorf("failed to ensure all control plane services are active: %w", err)
+		}
+		return nil
+	}); err != nil {
+		return fmt.Errorf("failed after retry: %w", err)
+	}
+	return nil
+}
+
 func DetermineLocalhostAddress(clusterMembers []mctypes.ClusterMember) (string, error) {
 	// Check if any of the cluster members have an IPv6 address, if so return "::1"
 	// if one member has an IPv6 address, other members should also have IPv6 interfaces

src/k8s/pkg/k8sd/app/hooks_bootstrap.go

@@ -283,6 +283,20 @@ func (a *App) onBootstrapWorkerNode(ctx context.Context, s state.State, encodedT
 		return fmt.Errorf("failed after retry: %w", err)
 	}
 
+	log.Info("Checking worker node services")
+	// The services may be able to start, appearing to be "active", but they might eventually fail due to
+	// various reasons, and they may be restarted. We're checking their activity a few times.
+	if err := control.Consistently(ctx, 3, 5*time.Second, func() error {
+		if err := snaputil.CheckWorkerServicesStates(ctx, snap, "active"); err != nil {
+			return fmt.Errorf("failed to ensure all worker services are active: %w", err)
+		}
+		return nil
+	}); err != nil {
+		log.Error(err, "Not all worker node services entered an active state. Stopping worker node services.")
+		return err
+	}
+
+	log.Info("Worker node services are ready")
 	return nil
 }
 
@@ -508,8 +522,14 @@ func (a *App) onBootstrapControlPlane(ctx context.Context, s state.State, bootst
 	if err := waitApiServerReady(ctx, snap); err != nil {
 		return fmt.Errorf("kube-apiserver did not become ready in time: %w", err)
 	}
-	log.Info("API server is ready - notify controllers")
 
+	log.Info("API server is ready - waiting for control plane services")
+	if err := waitControlPlaneServices(ctx, snap); err != nil {
+		log.Error(err, "Not all control plane services entered an active state. Stopping control plane services.")
+		return err
+	}
+
+	log.Info("Control plane services are ready - notify controllers")
 	a.NotifyFeatureController(
 		cfg.Network.GetEnabled(),
 		cfg.Gateway.GetEnabled(),

src/k8s/pkg/k8sd/app/hooks_join.go

@@ -236,5 +236,12 @@ func (a *App) onPostJoin(ctx context.Context, s state.State, initConfig map[stri
 		return fmt.Errorf("failed to wait for kube-apiserver to become ready: %w", err)
 	}
 
+	log.Info("API server is ready - waiting for control plane services")
+	if err := waitControlPlaneServices(ctx, snap); err != nil {
+		log.Error(err, "Not all control plane services entered an active state. Stopping control plane services.")
+		return err
+	}
+
+	log.Info("Control plane node services are ready")
 	return nil
 }

src/k8s/pkg/snap/interface.go

@@ -19,9 +19,10 @@ type Snap interface {
 	GID() int         // GID is the group ID to set on config files.
 	Hostname() string // Hostname is the name of the node.
 
-	StartService(ctx context.Context, serviceName string) error   // snapctl start $service
-	StopService(ctx context.Context, serviceName string) error    // snapctl stop $service
-	RestartService(ctx context.Context, serviceName string) error // snapctl restart $service
+	StartService(ctx context.Context, serviceName string) error              // snapctl start $service
+	StopService(ctx context.Context, serviceName string) error               // snapctl stop $service
+	RestartService(ctx context.Context, serviceName string) error            // snapctl restart $service
+	GetServiceState(ctx context.Context, serviceName string) (string, error) // systemctl is-running $service
 
 	SnapctlGet(ctx context.Context, args ...string) ([]byte, error) // snapctl get $args...
 	SnapctlSet(ctx context.Context, args ...string) error           // snapctl set $args...

src/k8s/pkg/snap/mock/mock.go

@@ -99,6 +99,10 @@ func (s *Snap) RestartService(ctx context.Context, name string) error {
 	return s.RestartServiceErr
 }
 
+func (s *Snap) GetServiceState(ctx context.Context, name string) (string, error) {
+	return "active", nil
+}
+
 func (s *Snap) Refresh(ctx context.Context, opts types.RefreshOpts) (string, error) {
 	if len(s.RefreshCalledWith) == 0 {
 		s.RefreshCalledWith = []types.RefreshOpts{opts}

src/k8s/pkg/snap/pebble.go

@@ -1,17 +1,24 @@
 package snap
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/canonical/k8s/pkg/k8sd/types"
 	"github.com/canonical/k8s/pkg/log"
 	"github.com/canonical/k8s/pkg/utils"
 )
 
+const (
+	stateActive   = "active"
+	stateInactive = "inactive"
+)
+
 type PebbleOpts struct {
 	SnapDir           string
 	SnapCommonDir     string
@@ -66,6 +73,31 @@ func (s *pebble) RestartService(ctx context.Context, name string) error {
 	return s.runCommand(ctx, []string{filepath.Join(s.snapDir, "bin", "pebble"), "restart", name})
 }
 
+// GetServiceState returns a k8s service state. The name can be either prefixed or not.
+func (s *pebble) GetServiceState(ctx context.Context, name string) (string, error) {
+	var b bytes.Buffer
+	err := s.runCommand(ctx, []string{filepath.Join(s.snapDir, "bin", "pebble"), "services", name}, func(c *exec.Cmd) { c.Stdout = &b })
+	if err != nil {
+		return "", err
+	}
+
+	output := b.String()
+	// We're expecting output like this:
+	// Service  Startup  Current   Since
+	// kubelet  enabled  inactive  -
+	lines := strings.Split(output, "\n")
+	if len(lines) < 2 {
+		return "", fmt.Errorf("Unexpected output when checking service %s state", name)
+	}
+
+	fields := strings.Fields(lines[1])
+	if len(fields) < 3 || (!strings.EqualFold(stateActive, fields[2]) && !strings.EqualFold(stateInactive, fields[2])) {
+		return "", fmt.Errorf("Unexpected output when checking service %s state", name)
+	}
+
+	return fields[2], nil
+}
+
 func (s *pebble) Refresh(ctx context.Context, to types.RefreshOpts) (string, error) {
 	switch {
 	case to.Revision != "":

src/k8s/pkg/snap/service.go

@@ -13,3 +13,11 @@ func serviceName(serviceName string) string {
 	}
 	return fmt.Sprintf("k8s.%s", serviceName)
 }
+
+// systemdServiceName infers the name of the systemd service from the service name.
+func systemdServiceName(serviceName string) string {
+	if strings.HasPrefix(serviceName, "snap.k8s.") {
+		return serviceName
+	}
+	return fmt.Sprintf("snap.k8s.%s", serviceName)
+}

src/k8s/pkg/snap/snap.go

@@ -89,6 +89,24 @@ func (s *snap) RestartService(ctx context.Context, name string) error {
 	return s.runCommand(ctx, []string{"snapctl", "restart", serviceName(name)})
 }
 
+// GetServiceState returns a k8s service state. The name can be either prefixed or not.
+func (s *snap) GetServiceState(ctx context.Context, name string) (string, error) {
+	log.FromContext(ctx).V(2).WithCallDepth(1).Info("Getting service state", "service", name)
+
+	var b bytes.Buffer
+	err := s.runCommand(ctx, []string{"systemctl", "is-active", systemdServiceName(name)}, func(c *exec.Cmd) { c.Stdout = &b })
+	if err != nil {
+		// systemctl is-active has exit code 3 if the given service is not active.
+		unwrapped := errors.Unwrap(err)
+		if exitErr, ok := unwrapped.(*exec.ExitError); ok && exitErr.ExitCode() == 3 {
+			return "failed", nil
+		}
+		return "", err
+	}
+
+	return strings.TrimSpace(b.String()), nil
+}
+
 // Refresh refreshes the snap to a different track, revision or custom snap.
 func (s *snap) Refresh(ctx context.Context, to types.RefreshOpts) (string, error) {
 	if s.Strict() {

src/k8s/pkg/snap/util/services.go

@@ -3,6 +3,7 @@ package snaputil
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"github.com/canonical/k8s/pkg/snap"
 )
@@ -98,6 +99,34 @@ func StopK8sDqliteServices(ctx context.Context, snap snap.Snap) error {
 	return nil
 }
 
+// CheckWorkerServicesStates checks if the worker services are in the expected state.
+// CheckWorkerServicesStates will return on the first error or service state mismatch.
+func CheckWorkerServicesStates(ctx context.Context, snap snap.Snap, state string) error {
+	return checkServicesStates(ctx, snap, workerServices, state)
+}
+
+// CheckControlPlaneServicesStates checks if the control plane services are in the expected state.
+// CheckControlPlaneServicesStates will return on the first error or service state mismatch.
+func CheckControlPlaneServicesStates(ctx context.Context, snap snap.Snap, state string) error {
+	return checkServicesStates(ctx, snap, controlPlaneServices, state)
+}
+
+// CheckK8sDqliteServices checks if the k8s-dqlite datastore service is in the expected state.
+func CheckK8sDqliteServices(ctx context.Context, snap snap.Snap, state string) error {
+	return checkServicesStates(ctx, snap, []string{"k8s-dqlite"}, state)
+}
+
+func checkServicesStates(ctx context.Context, snap snap.Snap, services []string, state string) error {
+	for _, service := range services {
+		if actualState, err := snap.GetServiceState(ctx, service); err != nil {
+			return fmt.Errorf("failed to check service %s: %w", service, err)
+		} else if !strings.EqualFold(state, actualState) {
+			return fmt.Errorf("expected %s to be in state %s, but it is %s", service, state, actualState)
+		}
+	}
+	return nil
+}
+
 // ServiceArgsFromMap processes a map of string pointers and categorizes them into update and delete lists.
 // - If the value pointer is nil, it adds the argument name to the delete list.
 // - If the value pointer is not nil, it adds the argument and its value to the update map.

src/k8s/pkg/utils/control/retry.go

@@ -22,3 +22,20 @@ func RetryFor(ctx context.Context, retryCount int, delayBetweenRetry time.Durati
 	}
 	return err
 }
+
+// Consistently will retry a given function for the given amount of times, expecting all calls to succeed without
+// errors. It will wait for backoff between retries.
+func Consistently(ctx context.Context, retryCount int, delayBetweenRetry time.Duration, retryFunc func() error) error {
+	for i := 0; i < retryCount; i++ {
+		if err := retryFunc(); err != nil {
+			return err
+		}
+		select {
+		case <-ctx.Done():
+			return context.Canceled
+		case <-time.After(delayBetweenRetry):
+			continue
+		}
+	}
+	return nil
+}

tests/integration/templates/bootstrap-foo-lish-arg.yaml

@@ -0,0 +1,3 @@
+# Contains the bootstrap configuration for the test ensuring k8s services start as expected.
+extra-node-kubelet-args:
+  --foo: "lish"

tests/integration/tests/test_bootstrap.py

@@ -15,3 +15,23 @@ def test_microk8s_installed(instances: List[harness.Instance]):
     instance.exec("snap install microk8s --classic".split())
     result = instance.exec("k8s bootstrap".split(), capture_output=True, check=False)
     assert "Error: microk8s snap is installed" in result.stderr.decode()
+
+
+@pytest.mark.node_count(1)
+@pytest.mark.disable_k8s_bootstrapping()
+@pytest.mark.tags(tags.NIGHTLY)
+def test_k8s_service_invalid_extra_args(instances: List[harness.Instance]):
+    instance = instances[0]
+    foo_lish_bootstrap_config = (
+        config.MANIFESTS_DIR / "bootstrap-foo-lish-arg.yaml"
+    ).read_text()
+
+    result = instance.exec(
+        ["k8s", "bootstrap", "--file", "-"],
+        input=str.encode(foo_lish_bootstrap_config),
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert "expected kubelet to be in state active" in result.stderr