Merge pull request #20 from burmanm/add_crd_upgrader

Add modified version of the CRD upgrade tool for Helm installations

Makefile

@@ -14,7 +14,7 @@ SHELL = /usr/bin/env bash -o pipefail
 .SHELLFLAGS = -ec
 
 # ENVTEST_K8S_VERSION refers to the version of kubebuilder assets to be downloaded by envtest binary.
-ENVTEST_K8S_VERSION = 1.27.x
+ENVTEST_K8S_VERSION = 1.28.x
 
 .PHONY: all
 all: build

cmd/kubectl-k8ssandra/helm/crds.go

@@ -0,0 +1,122 @@
+package helm
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/k8ssandra/k8ssandra-client/pkg/helmutil"
+	"github.com/k8ssandra/k8ssandra-client/pkg/kubernetes"
+	"github.com/spf13/cobra"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+)
+
+var (
+	upgraderExample = `
+	# update CRDs in the namespace to targetVersion
+	%[1]s crds --chartName <chartName> --targetVersion <targetVersion> [<args>]
+
+	# update CRDs in the namespace to targetVersion with non-default chartRepo (helm.k8ssandra.io)
+	%[1]s crds --chartName <chartName> --targetVersion <targetVersion> --chartRepo <repository> [<args>]
+	`
+	errNotEnoughParameters = fmt.Errorf("not enough parameters, requires chartName and targetVersion")
+)
+
+type options struct {
+	configFlags *genericclioptions.ConfigFlags
+	genericclioptions.IOStreams
+	namespace     string
+	chartName     string
+	targetVersion string
+	chartRepo     string
+	repoURL       string
+}
+
+func newOptions(streams genericclioptions.IOStreams) *options {
+	return &options{
+		configFlags: genericclioptions.NewConfigFlags(true),
+		IOStreams:   streams,
+	}
+}
+
+// NewCmd provides a cobra command wrapping cqlShOptions
+func NewUpgradeCmd(streams genericclioptions.IOStreams) *cobra.Command {
+	o := newOptions(streams)
+
+	cmd := &cobra.Command{
+		Use:          "upgrade <targetVersion> [flags]",
+		Short:        "upgrade k8ssandra CRDs to target release version",
+		Example:      fmt.Sprintf(upgraderExample, "kubectl k8ssandra helm crds"),
+		SilenceUsage: true,
+		RunE: func(c *cobra.Command, args []string) error {
+			if err := o.Complete(c, args); err != nil {
+				return err
+			}
+			if err := o.Validate(); err != nil {
+				return err
+			}
+			if err := o.Run(); err != nil {
+				return err
+			}
+
+			return nil
+		},
+	}
+
+	fl := cmd.Flags()
+	fl.StringVar(&o.chartName, "chartName", "", "chartName to upgrade")
+	fl.StringVar(&o.targetVersion, "targetVersion", "", "targetVersion to upgrade to")
+	fl.StringVar(&o.chartRepo, "chartRepo", "", "optional chart repository name to override the default (k8ssandra)")
+	fl.StringVar(&o.repoURL, "repoURL", "", "optional chart repository url to override the default (helm.k8ssandra.io)")
+	o.configFlags.AddFlags(fl)
+
+	return cmd
+}
+
+// Complete parses the arguments and necessary flags to options
+func (c *options) Complete(cmd *cobra.Command, args []string) error {
+	var err error
+	if len(args) < 2 {
+		return errNotEnoughParameters
+	}
+
+	if c.repoURL == "" {
+		c.repoURL = helmutil.StableK8ssandraRepoURL
+	}
+
+	if c.chartRepo == "" {
+		c.chartRepo = helmutil.K8ssandraRepoName
+	}
+
+	c.targetVersion = args[0]
+	c.namespace, _, err = c.configFlags.ToRawKubeConfigLoader().Namespace()
+	return err
+}
+
+// Validate ensures that all required arguments and flag values are provided
+func (c *options) Validate() error {
+	// TODO Validate that the targetVersion is valid
+	return nil
+}
+
+// Run removes the finalizers for a release X in the given namespace
+func (c *options) Run() error {
+	restConfig, err := c.configFlags.ToRESTConfig()
+	if err != nil {
+		return err
+	}
+
+	kubeClient, err := kubernetes.GetClientInNamespace(restConfig, c.namespace)
+	if err != nil {
+		return err
+	}
+
+	ctx := context.Background()
+
+	upgrader, err := helmutil.NewUpgrader(kubeClient, c.chartRepo, c.repoURL, c.chartName)
+	if err != nil {
+		return err
+	}
+
+	_, err = upgrader.Upgrade(ctx, c.targetVersion)
+	return err
+}

cmd/kubectl-k8ssandra/helm/helm.go

@@ -0,0 +1,36 @@
+package helm
+
+import (
+	"github.com/spf13/cobra"
+	"k8s.io/cli-runtime/pkg/genericclioptions"
+)
+
+type ClientOptions struct {
+	configFlags *genericclioptions.ConfigFlags
+	genericclioptions.IOStreams
+}
+
+// NewClientOptions provides an instance of NamespaceOptions with default values
+func NewHelmOptions(streams genericclioptions.IOStreams) *ClientOptions {
+	return &ClientOptions{
+		configFlags: genericclioptions.NewConfigFlags(true),
+		IOStreams:   streams,
+	}
+}
+
+// NewCmd provides a cobra command wrapping NamespaceOptions
+func NewHelmCmd(streams genericclioptions.IOStreams) *cobra.Command {
+	o := NewHelmOptions(streams)
+
+	cmd := &cobra.Command{
+		Use: "k8ssandra [subcommand] [flags]",
+	}
+
+	// Add subcommands
+	cmd.AddCommand(NewUpgradeCmd(streams))
+
+	// cmd.Flags().BoolVar(&o.listNamespaces, "list", o.listNamespaces, "if true, print the list of all namespaces in the current KUBECONFIG")
+	o.configFlags.AddFlags(cmd.Flags())
+
+	return cmd
+}

internal/envtest/envtest.go

@@ -12,6 +12,8 @@ import (
 	k8ssandrataskapi "github.com/k8ssandra/k8ssandra-operator/apis/control/v1alpha1"
 
 	"github.com/k8ssandra/k8ssandra-client/pkg/kubernetes"
+	apiextensions "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/client-go/rest"
 	"k8s.io/kubectl/pkg/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -76,6 +78,10 @@ func (e *Environment) start() {
 		panic(err)
 	}
 
+	if err := apiextensions.AddToScheme(scheme.Scheme); err != nil {
+		panic(err)
+	}
+
 	//+kubebuilder:scaffold:scheme
 
 	k8sClient, err := client.New(cfg, client.Options{Scheme: scheme.Scheme})
@@ -101,3 +107,7 @@ func (e *Environment) CreateNamespace(t *testing.T) string {
 
 	return namespace
 }
+
+func (e *Environment) RestConfig() *rest.Config {
+	return e.env.Config
+}

pkg/helmutil/crds.go

@@ -0,0 +1,154 @@
+package helmutil
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/pkg/errors"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	deser "k8s.io/apimachinery/pkg/runtime/serializer/yaml"
+	k8syaml "k8s.io/apimachinery/pkg/util/yaml"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// Upgrader is a utility to update the CRDs in a helm chart's pre-upgrade hook
+type Upgrader struct {
+	client    client.Client
+	repoName  string
+	repoURL   string
+	chartName string
+}
+
+// NewUpgrader returns a new Upgrader client
+func NewUpgrader(c client.Client, repoName, repoURL, chartName string) (*Upgrader, error) {
+	return &Upgrader{
+		client:    c,
+		repoName:  repoName,
+		repoURL:   repoURL,
+		chartName: chartName,
+	}, nil
+}
+
+// Upgrade installs the missing CRDs or updates them if they exists already
+func (u *Upgrader) Upgrade(ctx context.Context, targetVersion string) ([]unstructured.Unstructured, error) {
+	chartDir, err := GetChartTargetDir(u.chartName, targetVersion)
+	if err != nil {
+		return nil, err
+	}
+
+	if _, err := os.Stat(chartDir); os.IsNotExist(err) {
+		downloadDir, err := DownloadChartRelease(u.repoName, u.repoURL, u.chartName, targetVersion)
+		if err != nil {
+			return nil, err
+		}
+
+		extractDir, err := ExtractChartRelease(downloadDir, u.chartName, targetVersion)
+		if err != nil {
+			return nil, err
+		}
+		chartDir = extractDir
+	}
+
+	// defer os.RemoveAll(downloadDir)
+
+	crds := make([]unstructured.Unstructured, 0)
+
+	// For each dir under the charts subdir, check the "crds/"
+	paths, _ := findCRDDirs(chartDir)
+
+	for _, path := range paths {
+		err = parseChartCRDs(&crds, path)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	for _, obj := range crds {
+		existingCrd := obj.DeepCopy()
+		err = u.client.Get(ctx, client.ObjectKey{Name: obj.GetName()}, existingCrd)
+		if apierrors.IsNotFound(err) {
+			if err = u.client.Create(ctx, &obj); err != nil {
+				return nil, errors.Wrapf(err, "failed to create CRD %s", obj.GetName())
+			}
+		} else if err != nil {
+			return nil, errors.Wrapf(err, "failed to fetch state of %s", obj.GetName())
+		} else {
+			obj.SetResourceVersion(existingCrd.GetResourceVersion())
+			if err = u.client.Update(ctx, &obj); err != nil {
+				return nil, errors.Wrapf(err, "failed to update CRD %s", obj.GetName())
+			}
+		}
+	}
+
+	return crds, err
+}
+
+func findCRDDirs(chartDir string) ([]string, error) {
+	dirs := make([]string, 0)
+	err := filepath.Walk(chartDir, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if info.IsDir() {
+			if strings.HasSuffix(path, "crds") {
+				dirs = append(dirs, path)
+			}
+			return nil
+		}
+		return nil
+	})
+	return dirs, err
+}
+
+func parseChartCRDs(crds *[]unstructured.Unstructured, crdDir string) error {
+	errOuter := filepath.Walk(crdDir, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+
+		if info.IsDir() {
+			return nil
+		}
+
+		// Add to CRDs ..
+		b, err := os.ReadFile(path)
+		if err != nil {
+			return err
+		}
+
+		if len(b) == 0 {
+			return nil
+		}
+
+		reader := k8syaml.NewYAMLReader(bufio.NewReader(bytes.NewReader(b)))
+		doc, err := reader.Read()
+		if err != nil {
+			return err
+		}
+
+		crd := unstructured.Unstructured{}
+
+		dec := deser.NewDecodingSerializer(unstructured.UnstructuredJSONScheme)
+
+		_, gvk, err := dec.Decode(doc, nil, &crd)
+		if err != nil {
+			return nil
+		}
+
+		if gvk.Kind != "CustomResourceDefinition" {
+			return nil
+		}
+
+		*crds = append(*crds, crd)
+
+		return nil
+	})
+
+	return errOuter
+}