Apply Helm Stack in-memory

cmd/controller/controller.go

@@ -42,6 +42,7 @@ import (
 	"github.com/k0sproject/k0s/pkg/constant"
 	"github.com/k0sproject/k0s/pkg/k0scontext"
 	"github.com/k0sproject/k0s/pkg/kubernetes"
+	"github.com/k0sproject/k0s/pkg/leaderelection"
 	"github.com/k0sproject/k0s/pkg/performance"
 	"github.com/k0sproject/k0s/pkg/telemetry"
 	"github.com/k0sproject/k0s/pkg/token"
@@ -311,7 +312,7 @@ func (c *command) start(ctx context.Context, flags *config.ControllerOptions, de
 
 	// One leader elector per controller
 	if singleController {
-		leaderElector = &leaderelector.Dummy{Leader: true}
+		leaderElector = leaderelector.Off()
 	} else {
 		// The name used to be hardcoded in the component itself
 		// At some point we need to rename this.
@@ -324,7 +325,10 @@ func (c *command) start(ctx context.Context, flags *config.ControllerOptions, de
 			K0sVars:           c.K0sVars,
 			KubeClientFactory: adminClientFactory,
 			IgnoredStacks: []string{
+				controller.AutopilotStackName,
 				controller.ClusterConfigStackName,
+				controller.EtcdMemberStackName,
+				controller.HelmExtensionStackName,
 				controller.SystemRBACStackName,
 				controller.WindowsStackName,
 			},
@@ -373,8 +377,8 @@ func (c *command) start(ctx context.Context, flags *config.ControllerOptions, de
 		if err != nil {
 			return err
 		}
-		clusterComponents.Add(ctx, controller.NewCRD(c.K0sVars.ManifestsDir, "etcd", controller.WithStackName("etcd-member")))
-		nodeComponents.Add(ctx, etcdReconciler)
+		clusterComponents.Add(ctx, controller.NewCRDStack(adminClientFactory, "etcd", controller.WithStackName(controller.EtcdMemberStackName)))
+		clusterComponents.Add(ctx, etcdReconciler)
 	}
 
 	perfTimer.Checkpoint("starting-certificates-init")
@@ -420,7 +424,7 @@ func (c *command) start(ctx context.Context, flags *config.ControllerOptions, de
 	if flags.EnableDynamicConfig {
 		clusterComponents.Add(ctx, controller.NewClusterConfigInitializer(
 			adminClientFactory,
-			leaderElector,
+			func() leaderelection.Status { status, _ := leaderElector.CurrentStatus(); return status },
 			nodeConfig,
 		))
 
@@ -439,7 +443,6 @@ func (c *command) start(ctx context.Context, flags *config.ControllerOptions, de
 	))
 
 	if !slices.Contains(flags.DisableComponents, constant.HelmComponentName) {
-		clusterComponents.Add(ctx, controller.NewCRD(c.K0sVars.ManifestsDir, "helm"))
 		clusterComponents.Add(ctx, controller.NewExtensionsController(
 			c.K0sVars,
 			adminClientFactory,
@@ -448,13 +451,13 @@ func (c *command) start(ctx context.Context, flags *config.ControllerOptions, de
 	}
 
 	if !slices.Contains(flags.DisableComponents, constant.AutopilotComponentName) {
-		clusterComponents.Add(ctx, controller.NewCRD(c.K0sVars.ManifestsDir, "autopilot"))
+		clusterComponents.Add(ctx, controller.NewCRDStack(adminClientFactory, controller.AutopilotStackName))
 	}
 
 	if enableK0sEndpointReconciler {
 		clusterComponents.Add(ctx, controller.NewEndpointReconciler(
 			nodeConfig,
-			leaderElector,
+			func() leaderelection.Status { status, _ := leaderElector.CurrentStatus(); return status },
 			adminClientFactory,
 			net.DefaultResolver,
 			nodeConfig.PrimaryAddressFamily(),

inttest/addons/addons_test.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"crypto"
 	"crypto/x509"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
@@ -24,7 +25,6 @@ import (
 	"github.com/cloudflare/cfssl/initca"
 	"github.com/cloudflare/cfssl/signer"
 	"github.com/cloudflare/cfssl/signer/local"
-	"github.com/k0sproject/k0s/internal/pkg/templatewriter"
 	"github.com/k0sproject/k0s/inttest/common"
 	helmv1beta1 "github.com/k0sproject/k0s/pkg/apis/helm/v1beta1"
 	k0sv1beta1 "github.com/k0sproject/k0s/pkg/apis/k0s/v1beta1"
@@ -34,11 +34,14 @@ import (
 	"github.com/stretchr/testify/require"
 	"helm.sh/helm/v3/pkg/action"
 	"helm.sh/helm/v3/pkg/cli"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
-	k8s "k8s.io/client-go/kubernetes"
-	"k8s.io/utils/ptr"
+	"k8s.io/client-go/dynamic"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	crlog "sigs.k8s.io/controller-runtime/pkg/log"
@@ -173,13 +176,7 @@ func (as *AddonsSuite) TestHelmBasedAddons() {
 
 	as.Run("Rename chart in Helm extension", func() { as.renameChart(ctx) })
 
-	values := map[string]any{
-		"replicaCount": 2,
-		"image": map[string]any{
-			"pullPolicy": "Always",
-		},
-	}
-	as.doTestAddonUpdate(addonName, values)
+	as.doTestAddonUpdate(ctx, addonName)
 	chart := as.waitForTestRelease(addonName, "0.6.0", metav1.NamespaceDefault, 2)
 	as.Require().NoError(as.checkCustomValues(chart.Status.ReleaseName))
 	as.deleteRelease(chart)
@@ -243,19 +240,22 @@ func (as *AddonsSuite) renameChart(ctx context.Context) {
 func (as *AddonsSuite) deleteRelease(chart *helmv1beta1.Chart) {
 	ctx := as.Context()
 	as.T().Logf("Deleting chart %s/%s", chart.Namespace, chart.Name)
-	ssh, err := as.SSH(ctx, as.ControllerNode(0))
+	kubeconfig, err := as.GetKubeConfig(as.ControllerNode(0))
 	as.Require().NoError(err)
-	defer ssh.Disconnect()
-	_, err = ssh.ExecWithOutput(ctx, "rm /var/lib/k0s/manifests/helm/0_helm_extension_test-addon.yaml")
+	dyn, err := dynamic.NewForConfig(kubeconfig)
 	as.Require().NoError(err)
-	cfg, err := as.GetKubeConfig(as.ControllerNode(0))
-	as.Require().NoError(err)
-	k8sclient, err := k8s.NewForConfig(cfg)
+
+	_, err = patchResource(
+		ctx, dyn.Resource(k0sv1beta1.SchemeGroupVersion.WithResource("clusterconfigs")).Namespace(metav1.NamespaceSystem), "k0s",
+		patchOp{"test", "/spec/extensions/helm/charts/0/name", chart.Status.ReleaseName},
+		patchOp{"remove", "/spec/extensions/helm/charts/0", nil},
+	)
 	as.Require().NoError(err)
 	as.Require().NoError(wait.PollUntilContextCancel(ctx, 1*time.Second, true, func(pollCtx context.Context) (done bool, err error) {
 		as.T().Logf("Expecting have no secrets left for release %s/%s", chart.Status.Namespace, chart.Status.ReleaseName)
-		items, err := k8sclient.CoreV1().Secrets(chart.Status.Namespace).List(pollCtx, metav1.ListOptions{
-			LabelSelector: "name=" + chart.Status.ReleaseName,
+		secrets := dyn.Resource(corev1.SchemeGroupVersion.WithResource("secrets"))
+		items, err := secrets.Namespace(chart.Status.Namespace).List(pollCtx, metav1.ListOptions{
+			LabelSelector: fields.OneTermEqualSelector("name", chart.Status.ReleaseName).String(),
 		})
 		if err != nil {
 			if ctxErr := context.Cause(ctx); ctxErr != nil {
@@ -271,19 +271,16 @@ func (as *AddonsSuite) deleteRelease(chart *helmv1beta1.Chart) {
 		return true, nil
 	}))
 
-	chartClient, err := client.New(cfg, client.Options{Scheme: k0sscheme.Scheme})
-	as.Require().NoError(err)
-
 	as.T().Logf("Expecting chart %s/%s to be deleted", chart.Namespace, chart.Name)
 	var lastResourceVersion string
 	as.Require().NoError(wait.PollUntilContextCancel(ctx, 1*time.Second, true, func(ctx context.Context) (bool, error) {
-		var found helmv1beta1.Chart
-		err := chartClient.Get(ctx, client.ObjectKey{Namespace: chart.Namespace, Name: chart.Name}, &found)
+		charts := dyn.Resource(helmv1beta1.SchemeGroupVersion.WithResource("charts"))
+		found, err := charts.Namespace(chart.Namespace).Get(ctx, chart.Name, metav1.GetOptions{})
 		switch {
 		case err == nil:
-			if lastResourceVersion == "" || lastResourceVersion != found.ResourceVersion {
+			if lastResourceVersion == "" || lastResourceVersion != found.GetResourceVersion() {
 				as.T().Log("Chart not yet deleted")
-				lastResourceVersion = found.ResourceVersion
+				lastResourceVersion = found.GetResourceVersion()
 			}
 			return false, nil
 
@@ -461,33 +458,25 @@ func (as *AddonsSuite) checkCustomValues(releaseName string) error {
 	})
 }
 
-func (as *AddonsSuite) doTestAddonUpdate(addonName string, values map[string]any) {
-	path := fmt.Sprintf("/var/lib/k0s/manifests/helm/0_helm_extension_%s.yaml", addonName)
-	valuesBytes, err := yaml.Marshal(values)
+func (as *AddonsSuite) doTestAddonUpdate(ctx context.Context, addonName string) {
+	kubeconfig, err := as.GetKubeConfig(as.ControllerNode(0))
+	as.Require().NoError(err)
+	dyn, err := dynamic.NewForConfig(kubeconfig)
 	as.Require().NoError(err)
-	tw := templatewriter.TemplateWriter{
-		Name:     "testChartUpdate",
-		Template: chartCrdTemplate,
-		Data: struct {
-			Name         string
-			ChartName    string
-			Values       string
-			Version      string
-			TargetNS     string
-			ForceUpgrade *bool
-		}{
-			Name:         "test-addon",
-			ChartName:    "ealenn/echo-server",
-			Values:       string(valuesBytes),
-			Version:      "0.5.0",
-			TargetNS:     metav1.NamespaceDefault,
-			ForceUpgrade: ptr.To(false),
-		},
-	}
-	buf := bytes.NewBuffer([]byte{})
-	as.Require().NoError(tw.WriteToBuffer(buf))
 
-	as.PutFile(as.ControllerNode(0), path, buf.String())
+	_, err = patchResource(
+		ctx, dyn.Resource(k0sv1beta1.SchemeGroupVersion.WithResource("clusterconfigs")).Namespace(metav1.NamespaceSystem), "k0s",
+		patchOp{"test", "/spec/extensions/helm/charts/0/name", addonName},
+		patchOp{"replace", "/spec/extensions/helm/charts/0", map[string]any{
+			"name":         addonName,
+			"chartname":    "ealenn/echo-server",
+			"values":       `{"replicaCount": 2, "image": {"pullPolicy": "Always"}}`,
+			"version":      "0.5.0",
+			"namespace":    metav1.NamespaceDefault,
+			"forceUpgrade": false,
+		}},
+	)
+	as.Require().NoError(err)
 }
 
 func TestAddonsSuite(t *testing.T) {
@@ -507,6 +496,21 @@ func TestAddonsSuite(t *testing.T) {
 	suite.Run(t, &s)
 }
 
+type patchOp = struct {
+	Op    string `json:"op"`
+	Path  string `json:"path"`
+	Value any    `json:"value,omitempty"`
+}
+
+//nolint:unparam // returned object not yet used
+func patchResource(ctx context.Context, c dynamic.ResourceInterface, name string, ops ...patchOp) (*unstructured.Unstructured, error) {
+	if patch, err := json.Marshal(ops); err != nil {
+		return nil, err
+	} else {
+		return c.Patch(ctx, name, types.JSONPatchType, patch, metav1.PatchOptions{})
+	}
+}
+
 type k0sConfigParams struct {
 	BasicAddonName string
 
@@ -550,25 +554,3 @@ spec:
 `
 
 var k0sConfigWithAddonTemplate = template.Must(template.New("k0sConfigWithAddon").Parse(k0sConfigWithAddonRawTemplate))
-
-// TODO: this actually duplicates logic from the controller code
-// better to somehow handle it by programmatic api
-const chartCrdTemplate = `
-apiVersion: helm.k0sproject.io/v1beta1
-kind: Chart
-metadata:
-  name: k0s-addon-chart-{{ .Name }}
-  namespace: ` + metav1.NamespaceSystem + `
-  finalizers:
-    - helm.k0sproject.io/uninstall-helm-release 
-spec:
-  chartName: {{ .ChartName }}
-  releaseName: {{ .Name }}
-  values: |
-{{ .Values | nindent 4 }}
-  version: {{ .Version }}
-  namespace: {{ .TargetNS }}
-{{- if ne .ForceUpgrade nil }}
-  forceUpgrade: {{ .ForceUpgrade }}
-{{- end }}
-`

pkg/applier/manager_test.go

@@ -38,13 +38,13 @@ import (
 func TestManager_AppliesStacks(t *testing.T) {
 	k0sVars, err := config.NewCfgVars(nil, t.TempDir())
 	require.NoError(t, err)
-	leaderElector := leaderelector.Dummy{Leader: true}
+	leaderElector := leaderelector.Off()
 	clients := testutil.NewFakeClientFactory()
 
 	underTest := applier.Manager{
 		K0sVars:           k0sVars,
 		KubeClientFactory: clients,
-		LeaderElector:     &leaderElector,
+		LeaderElector:     leaderElector,
 	}
 
 	// A stack that exists on disk before the manager is started.
@@ -100,14 +100,14 @@ data: {}
 func TestManager_IgnoresStacks(t *testing.T) {
 	k0sVars, err := config.NewCfgVars(nil, t.TempDir())
 	require.NoError(t, err)
-	leaderElector := leaderelector.Dummy{Leader: true}
+	leaderElector := leaderelector.Off()
 	clients := testutil.NewFakeClientFactory()
 
 	underTest := applier.Manager{
 		K0sVars:           k0sVars,
 		IgnoredStacks:     []string{"ignored"},
 		KubeClientFactory: clients,
-		LeaderElector:     &leaderElector,
+		LeaderElector:     leaderElector,
 	}
 
 	ignored := filepath.Join(k0sVars.ManifestsDir, "ignored")

pkg/applier/stack.go

@@ -4,11 +4,13 @@
 package applier
 
 import (
+	"cmp"
 	"context"
 	"crypto/md5"
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"io"
 	"slices"
 	"sync"
 	"time"
@@ -23,15 +25,60 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/cli-runtime/pkg/resource"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/restmapper"
 	"k8s.io/utils/ptr"
 
+	"github.com/avast/retry-go"
 	jsonpatch "github.com/evanphx/json-patch"
 	"github.com/sirupsen/logrus"
 )
 
+func ApplyStack(ctx context.Context, clients kubernetes.ClientFactoryInterface, src io.Reader, srcName, stackName string) error {
+	infos, err := resource.NewLocalBuilder().
+		Unstructured().
+		Stream(src, srcName).
+		Flatten().
+		Do().
+		Infos()
+	if err != nil {
+		return err
+	}
+
+	resources := make([]*unstructured.Unstructured, len(infos))
+	for i := range infos {
+		resources[i] = infos[i].Object.(*unstructured.Unstructured)
+	}
+
+	var lastErr error
+	if err := retry.Do(
+		func() error {
+			stack := Stack{
+				Name:      stackName,
+				Resources: resources,
+				Clients:   clients,
+			}
+			lastErr = stack.Apply(ctx, true)
+			return lastErr
+		},
+		retry.Context(ctx),
+		retry.LastErrorOnly(true),
+		retry.OnRetry(func(attempt uint, err error) {
+			logrus.WithFields(logrus.Fields{
+				"component": "applier",
+				"stack":     stackName,
+				"attempt":   attempt + 1,
+			}).WithError(err).Debug("Failed to apply stack, retrying after backoff")
+		}),
+	); err != nil {
+		return cmp.Or(lastErr, err)
+	}
+
+	return nil
+}
+
 // Stack is a k8s resource bundle
 type Stack struct {
 	Name          string