Replace kubectl with Go server-side apply

Signed-off-by: Stefan Prodan <stefan.prodan@gmail.com>

.github/workflows/bootstrap.yaml

@@ -4,7 +4,7 @@ on:
   push:
     branches: [ main ]
   pull_request:
-    branches: [ main ]
+    branches: [ main, ssa ]
 
 jobs:
   github:

.github/workflows/e2e.yaml

@@ -4,7 +4,7 @@ on:
   push:
     branches: [ main ]
   pull_request:
-    branches: [ main ]
+    branches: [ main, ssa ]
 
 jobs:
   kind:

cmd/flux/check.go

@@ -18,9 +18,7 @@ package main
 
 import (
 	"context"
-	"encoding/json"
 	"os"
-	"os/exec"
 	"time"
 
 	"github.com/Masterminds/semver/v3"
@@ -73,18 +71,11 @@ func init() {
 }
 
 func runCheckCmd(cmd *cobra.Command, args []string) error {
-	ctx, cancel := context.WithTimeout(context.Background(), rootArgs.timeout)
-	defer cancel()
-
 	logger.Actionf("checking prerequisites")
 	checkFailed := false
 
 	fluxCheck()
 
-	if !kubectlCheck(ctx, ">=1.18.0-0") {
-		checkFailed = true
-	}
-
 	if !kubernetesCheck(">=1.16.0-0") {
 		checkFailed = true
 	}
@@ -130,42 +121,6 @@ func fluxCheck() {
 	}
 }
 
-func kubectlCheck(ctx context.Context, constraint string) bool {
-	_, err := exec.LookPath("kubectl")
-	if err != nil {
-		logger.Failuref("kubectl not found")
-		return false
-	}
-
-	kubectlArgs := []string{"version", "--client", "--output", "json"}
-	output, err := utils.ExecKubectlCommand(ctx, utils.ModeCapture, rootArgs.kubeconfig, rootArgs.kubecontext, kubectlArgs...)
-	if err != nil {
-		logger.Failuref("kubectl version can't be determined")
-		return false
-	}
-
-	kv := &kubectlVersion{}
-	if err = json.Unmarshal([]byte(output), kv); err != nil {
-		logger.Failuref("kubectl version output can't be unmarshalled")
-		return false
-	}
-
-	v, err := version.ParseVersion(kv.ClientVersion.GitVersion)
-	if err != nil {
-		logger.Failuref("kubectl version can't be parsed")
-		return false
-	}
-
-	c, _ := semver.NewConstraint(constraint)
-	if !c.Check(v) {
-		logger.Failuref("kubectl version %s < %s", v.Original(), constraint)
-		return false
-	}
-
-	logger.Successf("kubectl %s %s", v.String(), constraint)
-	return true
-}
-
 func kubernetesCheck(constraint string) bool {
 	cfg, err := utils.KubeConfig(rootArgs.kubeconfig, rootArgs.kubecontext)
 	if err != nil {

cmd/flux/check_test.go

@@ -23,13 +23,11 @@ func TestCheckPre(t *testing.T) {
 		t.Fatalf("Error unmarshalling: %v", err.Error())
 	}
 
-	clientVersion := strings.TrimPrefix(versions["clientVersion"].GitVersion, "v")
 	serverVersion := strings.TrimPrefix(versions["serverVersion"].GitVersion, "v")
 
 	cmd := cmdTestCase{
 		args: "check --pre",
 		assert: assertGoldenTemplateFile("testdata/check/check_pre.golden", map[string]string{
-			"clientVersion": clientVersion,
 			"serverVersion": serverVersion,
 		}),
 	}

cmd/flux/install.go

@@ -41,13 +41,13 @@ If a previous version is installed, then an in-place upgrade will be performed.`
   flux install --version=latest --namespace=flux-system
 
   # Install a specific version and a series of components
-  flux install --dry-run --version=v0.0.7 --components="source-controller,kustomize-controller"
+  flux install --version=v0.0.7 --components="source-controller,kustomize-controller"
 
   # Install Flux onto tainted Kubernetes nodes
   flux install --toleration-keys=node.kubernetes.io/dedicated-to-flux
 
-  # Dry-run install with manifests preview
-  flux install --dry-run --verbose
+  # Dry-run install
+  flux install --export | kubectl apply --dry-run=client -f- 
 
   # Write install manifests to file
   flux install --export > flux-system.yaml`,
@@ -102,6 +102,7 @@ func init() {
 		"list of toleration keys used to schedule the components pods onto nodes with matching taints")
 	installCmd.Flags().MarkHidden("manifests")
 	installCmd.Flags().MarkDeprecated("arch", "multi-arch container image is now available for AMD64, ARMv7 and ARM64")
+	installCmd.Flags().MarkDeprecated("dry-run", "use 'flux install --export | kubectl apply --dry-run=client -f-'")
 	rootCmd.AddCommand(installCmd)
 }
 
@@ -188,24 +189,18 @@ func installCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Successf("manifests build completed")
 	logger.Actionf("installing components in %s namespace", rootArgs.namespace)
-	applyOutput := utils.ModeStderrOS
-	if rootArgs.verbose {
-		applyOutput = utils.ModeOS
-	}
 
-	kubectlArgs := []string{"apply", "-f", filepath.Join(tmpDir, manifest.Path)}
 	if installArgs.dryRun {
-		kubectlArgs = append(kubectlArgs, "--dry-run=client")
-		applyOutput = utils.ModeOS
+		logger.Successf("install dry-run finished")
+		return nil
 	}
-	if _, err := utils.ExecKubectlCommand(ctx, applyOutput, rootArgs.kubeconfig, rootArgs.kubecontext, kubectlArgs...); err != nil {
+
+	applyOutput, err := utils.Apply(ctx, rootArgs.kubeconfig, rootArgs.kubecontext, filepath.Join(tmpDir, manifest.Path))
+	if err != nil {
 		return fmt.Errorf("install failed: %w", err)
 	}
 
-	if installArgs.dryRun {
-		logger.Successf("install dry-run finished")
-		return nil
-	}
+	fmt.Fprintln(os.Stderr, applyOutput)
 
 	kubeConfig, err := utils.KubeConfig(rootArgs.kubeconfig, rootArgs.kubecontext)
 	if err != nil {

cmd/flux/testdata/check/check_pre.golden

@@ -1,4 +1,3 @@
 ► checking prerequisites
-✔ kubectl {{ .clientVersion }} >=1.18.0-0
 ✔ Kubernetes {{ .serverVersion }} >=1.16.0-0
 ✔ prerequisites checks passed

go.mod

@@ -14,12 +14,13 @@ require (
 	github.com/fluxcd/notification-controller/api v0.17.0
 	github.com/fluxcd/pkg/apis/meta v0.10.0
 	github.com/fluxcd/pkg/runtime v0.12.0
+	github.com/fluxcd/pkg/ssa v0.0.1
 	github.com/fluxcd/pkg/ssh v0.0.5
 	github.com/fluxcd/pkg/untar v0.0.5
 	github.com/fluxcd/pkg/version v0.0.1
 	github.com/fluxcd/source-controller/api v0.16.0
 	github.com/go-git/go-git/v5 v5.4.2
-	github.com/google/go-cmp v0.5.5
+	github.com/google/go-cmp v0.5.6
 	github.com/google/go-containerregistry v0.2.0
 	github.com/manifoldco/promptui v0.7.0
 	github.com/mattn/go-shellwords v1.0.12
@@ -35,7 +36,7 @@ require (
 	sigs.k8s.io/cli-utils v0.25.1-0.20210608181808-f3974341173a
 	sigs.k8s.io/controller-runtime v0.10.1
 	sigs.k8s.io/kustomize/api v0.8.10
-	sigs.k8s.io/yaml v1.2.0
+	sigs.k8s.io/yaml v1.3.0
 )
 
 // drop LGPL dependency manifoldco/promptui -> juju/ansiterm

internal/bootstrap/bootstrap_plain_git.go

@@ -175,41 +175,14 @@ func (b *PlainGitBootstrapper) ReconcileComponents(ctx context.Context, manifest
 		// Apply components using any existing customisations
 		kfile := filepath.Join(filepath.Dir(componentsYAML), konfig.DefaultKustomizationFileName())
 		if _, err := os.Stat(kfile); err == nil {
-			tmpDir, err := os.MkdirTemp("", "gotk-crds")
-			defer os.RemoveAll(tmpDir)
-
-			// Extract the CRDs from the components manifest
-			crdsYAML := filepath.Join(tmpDir, "gotk-crds.yaml")
-			if err := utils.ExtractCRDs(componentsYAML, crdsYAML); err != nil {
-				return err
-			}
-
-			// Apply the CRDs
-			b.logger.Actionf("installing toolkit.fluxcd.io CRDs")
-			kubectlArgs := []string{"apply", "-f", crdsYAML}
-			if _, err = utils.ExecKubectlCommand(ctx, utils.ModeStderrOS, b.kubeconfig, b.kubecontext, kubectlArgs...); err != nil {
-				return err
-			}
-
-			// Wait for CRDs to be established
-			b.logger.Waitingf("waiting for CRDs to be reconciled")
-			kubectlArgs = []string{"wait", "--for", "condition=established", "-f", crdsYAML}
-			if _, err = utils.ExecKubectlCommand(ctx, utils.ModeStderrOS, b.kubeconfig, b.kubecontext, kubectlArgs...); err != nil {
-				return err
-			}
-			b.logger.Successf("CRDs reconciled successfully")
-
 			// Apply the components and their patches
 			b.logger.Actionf("installing components in %q namespace", options.Namespace)
-			kubectlArgs = []string{"apply", "-k", filepath.Dir(componentsYAML)}
-			if _, err = utils.ExecKubectlCommand(ctx, utils.ModeStderrOS, b.kubeconfig, b.kubecontext, kubectlArgs...); err != nil {
+			if _, err := utils.Apply(ctx, b.kubeconfig, b.kubecontext, kfile); err != nil {
 				return err
 			}
 		} else {
 			// Apply the CRDs and controllers
-			b.logger.Actionf("installing components in %q namespace", options.Namespace)
-			kubectlArgs := []string{"apply", "-f", componentsYAML}
-			if _, err = utils.ExecKubectlCommand(ctx, utils.ModeStderrOS, b.kubeconfig, b.kubecontext, kubectlArgs...); err != nil {
+			if _, err := utils.Apply(ctx, b.kubeconfig, b.kubecontext, componentsYAML); err != nil {
 				return err
 			}
 		}
@@ -336,10 +309,10 @@ func (b *PlainGitBootstrapper) ReconcileSyncConfig(ctx context.Context, options
 
 	// Apply to cluster
 	b.logger.Actionf("applying sync manifests")
-	kubectlArgs := []string{"apply", "-k", filepath.Join(b.git.Path(), filepath.Dir(kusManifests.Path))}
-	if _, err = utils.ExecKubectlCommand(ctx, utils.ModeStderrOS, b.kubeconfig, b.kubecontext, kubectlArgs...); err != nil {
+	if _, err := utils.Apply(ctx, b.kubeconfig, b.kubecontext, filepath.Join(b.git.Path(), kusManifests.Path)); err != nil {
 		return err
 	}
+
 	b.logger.Successf("reconciled sync configuration")
 
 	return nil

internal/utils/apply.go

@@ -0,0 +1,81 @@
+package utils
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/fluxcd/pkg/ssa"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+	"sigs.k8s.io/kustomize/api/konfig"
+
+	"github.com/fluxcd/flux2/pkg/manifestgen/kustomization"
+)
+
+// Apply is the equivalent of 'kubectl apply --server-side -f'.
+// If the given manifest is a kustomization.yaml, then apply performs the equivalent of 'kubectl apply --server-side -k'.
+func Apply(ctx context.Context, kubeConfigPath string, kubeContext string, manifestPath string) (string, error) {
+	cfg, err := KubeConfig(kubeConfigPath, kubeContext)
+	if err != nil {
+		return "", err
+	}
+	restMapper, err := apiutil.NewDynamicRESTMapper(cfg)
+	if err != nil {
+		return "", err
+	}
+	kubeClient, err := client.New(cfg, client.Options{Mapper: restMapper})
+	if err != nil {
+		return "", err
+	}
+	kubePoller := polling.NewStatusPoller(kubeClient, restMapper)
+
+	resourceManager := ssa.NewResourceManager(kubeClient, kubePoller, ssa.Owner{
+		Field: "flux",
+		Group: "fluxcd.io",
+	})
+
+	objs, err := readObjects(manifestPath)
+	if err != nil {
+		return "", err
+	}
+
+	if len(objs) < 1 {
+		return "", fmt.Errorf("no Kubernetes objects found at: %s", manifestPath)
+	}
+
+	changeSet, err := resourceManager.ApplyAllStaged(ctx, objs, false, time.Minute)
+	if err != nil {
+		return "", err
+	}
+
+	return changeSet.String(), nil
+}
+
+func readObjects(manifestPath string) ([]*unstructured.Unstructured, error) {
+	if _, err := os.Stat(manifestPath); err != nil {
+		return nil, err
+	}
+
+	if filepath.Base(manifestPath) == konfig.DefaultKustomizationFileName() {
+		resources, err := kustomization.Build(filepath.Dir(manifestPath))
+		if err != nil {
+			return nil, err
+		}
+		return ssa.ReadObjects(bytes.NewReader(resources))
+	}
+
+	ms, err := os.Open(manifestPath)
+	if err != nil {
+		return nil, err
+	}
+	defer ms.Close()
+
+	return ssa.ReadObjects(bufio.NewReader(ms))
+}

pkg/manifestgen/install/manifests.go

@@ -25,13 +25,11 @@ import (
 	"path"
 	"path/filepath"
 	"strings"
-	"sync"
 
+	"github.com/fluxcd/pkg/untar"
 	"sigs.k8s.io/kustomize/api/filesys"
-	"sigs.k8s.io/kustomize/api/krusty"
-	kustypes "sigs.k8s.io/kustomize/api/types"
 
-	"github.com/fluxcd/pkg/untar"
+	"github.com/fluxcd/flux2/pkg/manifestgen/kustomization"
 )
 
 func fetch(ctx context.Context, url, version, dir string) error {
@@ -114,56 +112,13 @@ func generate(base string, options Options) error {
 	return nil
 }
 
-var kustomizeBuildMutex sync.Mutex
-
 func build(base, output string) error {
-	// TODO(stefan): temporary workaround for concurrent map read and map write bug
-	// https://github.com/kubernetes-sigs/kustomize/issues/3659
-	kustomizeBuildMutex.Lock()
-	defer kustomizeBuildMutex.Unlock()
-
-	kfile := filepath.Join(base, "kustomization.yaml")
-
-	fs := filesys.MakeFsOnDisk()
-	if !fs.Exists(kfile) {
-		return fmt.Errorf("%s not found", kfile)
-	}
-
-	// TODO(hidde): work around for a bug in kustomize causing it to
-	//  not properly handle absolute paths on Windows.
-	//  Convert the path to a relative path to the working directory
-	//  as a temporary fix:
-	//  https://github.com/kubernetes-sigs/kustomize/issues/2789
-	if filepath.IsAbs(base) {
-		wd, err := os.Getwd()
-		if err != nil {
-			return err
-		}
-		base, err = filepath.Rel(wd, base)
-		if err != nil {
-			return err
-		}
-	}
-
-	buildOptions := &krusty.Options{
-		DoLegacyResourceSort: true,
-		LoadRestrictions:     kustypes.LoadRestrictionsNone,
-		AddManagedbyLabel:    false,
-		DoPrune:              false,
-		PluginConfig:         kustypes.DisabledPluginConfig(),
-	}
-
-	k := krusty.MakeKustomizer(buildOptions)
-	m, err := k.Run(fs, base)
-	if err != nil {
-		return err
-	}
-
-	resources, err := m.AsYaml()
+	resources, err := kustomization.Build(base)
 	if err != nil {
 		return err
 	}
 
+	fs := filesys.MakeFsOnDisk()
 	if err := fs.WriteFile(output, resources); err != nil {
 		return err
 	}

pkg/manifestgen/kustomization/kustomization.go

@@ -17,10 +17,14 @@ limitations under the License.
 package kustomization
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 
+	"sigs.k8s.io/kustomize/api/filesys"
 	"sigs.k8s.io/kustomize/api/konfig"
+	"sigs.k8s.io/kustomize/api/krusty"
 	"sigs.k8s.io/kustomize/api/provider"
 	kustypes "sigs.k8s.io/kustomize/api/types"
 	"sigs.k8s.io/yaml"
@@ -28,6 +32,8 @@ import (
 	"github.com/fluxcd/flux2/pkg/manifestgen"
 )
 
+// Generate scans the given directory for Kubernetes manifests and creates a kustomization.yaml
+// including all discovered manifests as resources.
 func Generate(options Options) (*manifestgen.Manifest, error) {
 	kfile := filepath.Join(options.TargetPath, konfig.DefaultKustomizationFileName())
 	abskfile := filepath.Join(options.BaseDir, kfile)
@@ -121,3 +127,57 @@ func Generate(options Options) (*manifestgen.Manifest, error) {
 		Content: string(kd),
 	}, nil
 }
+
+var kustomizeBuildMutex sync.Mutex
+
+// Build takes a Kustomize overlays and returns the resulting manifests as multi-doc YAML.
+func Build(base string) ([]byte, error) {
+	// TODO(stefan): temporary workaround for concurrent map read and map write bug
+	// https://github.com/kubernetes-sigs/kustomize/issues/3659
+	kustomizeBuildMutex.Lock()
+	defer kustomizeBuildMutex.Unlock()
+
+	kfile := filepath.Join(base, konfig.DefaultKustomizationFileName())
+
+	fs := filesys.MakeFsOnDisk()
+	if !fs.Exists(kfile) {
+		return nil, fmt.Errorf("%s not found", kfile)
+	}
+
+	// TODO(hidde): work around for a bug in kustomize causing it to
+	//  not properly handle absolute paths on Windows.
+	//  Convert the path to a relative path to the working directory
+	//  as a temporary fix:
+	//  https://github.com/kubernetes-sigs/kustomize/issues/2789
+	if filepath.IsAbs(base) {
+		wd, err := os.Getwd()
+		if err != nil {
+			return nil, err
+		}
+		base, err = filepath.Rel(wd, base)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	buildOptions := &krusty.Options{
+		DoLegacyResourceSort: true,
+		LoadRestrictions:     kustypes.LoadRestrictionsNone,
+		AddManagedbyLabel:    false,
+		DoPrune:              false,
+		PluginConfig:         kustypes.DisabledPluginConfig(),
+	}
+
+	k := krusty.MakeKustomizer(buildOptions)
+	m, err := k.Run(fs, base)
+	if err != nil {
+		return nil, err
+	}
+
+	resources, err := m.AsYaml()
+	if err != nil {
+		return nil, err
+	}
+
+	return resources, nil
+}