feat: add profile/install/uninstall/upgrade command (#538)

pkg/cmd/hgctl/kubernetes/client.go

@@ -19,17 +19,21 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	kubescheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/remotecommand"
+	"k8s.io/client-go/util/retry"
+	ctrClient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type CLIClient interface {
@@ -44,6 +48,15 @@ type CLIClient interface {
 
 	// PodExec takes a command and the pod data to run the command in the specified pod.
 	PodExec(namespacedName types.NamespacedName, container string, command string) (stdout string, stderr string, err error)
+
+	// ApplyObject creates or updates unstructured object
+	ApplyObject(obj *unstructured.Unstructured) error
+
+	// DeleteObject delete unstructured object
+	DeleteObject(obj *unstructured.Unstructured) error
+
+	// CreateNamespace create namespace
+	CreateNamespace(namespace string) error
 }
 
 var _ CLIClient = &client{}
@@ -52,6 +65,7 @@ type client struct {
 	config     *rest.Config
 	restClient *rest.RESTClient
 	kube       kubernetes.Interface
+	ctrClient  ctrClient.Client
 }
 
 func NewCLIClient(clientConfig clientcmd.ClientConfig) (CLIClient, error) {
@@ -80,33 +94,13 @@ func newClientInternal(clientConfig clientcmd.ClientConfig) (*client, error) {
 		return nil, err
 	}
 
+	c.ctrClient, err = ctrClient.New(c.config, ctrClient.Options{})
+	if err != nil {
+		return nil, err
+	}
 	return &c, err
 }
 
-func setRestDefaults(config *rest.Config) *rest.Config {
-	if config.GroupVersion == nil || config.GroupVersion.Empty() {
-		config.GroupVersion = &corev1.SchemeGroupVersion
-	}
-	if len(config.APIPath) == 0 {
-		if len(config.GroupVersion.Group) == 0 {
-			config.APIPath = "/api"
-		} else {
-			config.APIPath = "/apis"
-		}
-	}
-	if len(config.ContentType) == 0 {
-		config.ContentType = runtime.ContentTypeJSON
-	}
-	if config.NegotiatedSerializer == nil {
-		// This codec factory ensures the resources are not converted. Therefore, resources
-		// will not be round-tripped through internal versions. Defaulting does not happen
-		// on the client.
-		config.NegotiatedSerializer = serializer.NewCodecFactory(kubescheme.Scheme).WithoutConversion()
-	}
-
-	return config
-}
-
 func (c *client) RESTConfig() *rest.Config {
 	if c.config == nil {
 		return nil
@@ -170,3 +164,85 @@ func (c *client) PodExec(namespacedName types.NamespacedName, container string,
 	stderr = stderrBuf.String()
 	return
 }
+
+// DeleteObject delete unstructured object
+func (c *client) DeleteObject(obj *unstructured.Unstructured) error {
+	err := c.ctrClient.Delete(context.TODO(), obj, ctrClient.PropagationPolicy(metav1.DeletePropagationBackground))
+	if err != nil {
+		if !errors.IsNotFound(err) {
+			return err
+		}
+	}
+	return nil
+}
+
+// ApplyObject creates or updates unstructured object
+func (c *client) ApplyObject(obj *unstructured.Unstructured) error {
+	if obj.GetKind() == "List" {
+		objList, err := obj.ToList()
+		if err != nil {
+			return err
+		}
+		for _, item := range objList.Items {
+			if err := c.ApplyObject(&item); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	key := ctrClient.ObjectKeyFromObject(obj)
+	receiver := &unstructured.Unstructured{}
+	receiver.SetGroupVersionKind(obj.GroupVersionKind())
+
+	if err := retry.RetryOnConflict(wait.Backoff{
+		Duration: time.Millisecond * 10,
+		Factor:   2,
+		Steps:    3,
+	}, func() error {
+		if err := c.ctrClient.Get(context.Background(), key, receiver); err != nil {
+			if errors.IsNotFound(err) {
+				if err := c.ctrClient.Create(context.Background(), obj); err != nil {
+					return err
+				}
+			}
+			return nil
+		}
+		if err := applyOverlay(receiver, obj); err != nil {
+			return err
+		}
+		if err := c.ctrClient.Update(context.Background(), receiver); err != nil {
+			return err
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// CreateNamespace create namespace
+func (c *client) CreateNamespace(namespace string) error {
+	key := ctrClient.ObjectKey{
+		Namespace: metav1.NamespaceSystem,
+		Name:      namespace,
+	}
+	if err := c.ctrClient.Get(context.Background(), key, &corev1.Namespace{}); err != nil {
+		if errors.IsNotFound(err) {
+			nsObj := &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: metav1.NamespaceSystem,
+					Name:      namespace,
+				},
+			}
+			if err := c.ctrClient.Create(context.Background(), nsObj); err != nil {
+				return err
+			}
+			return nil
+		}
+		return fmt.Errorf("failed to check if namespace %v exists: %v", namespace, err)
+	}
+
+	return nil
+}

pkg/cmd/hgctl/kubernetes/common.go

@@ -0,0 +1,157 @@
+// Copyright (c) 2022 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kubernetes
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	jsonpatch "github.com/evanphx/json-patch/v5"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	kubescheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"k8s.io/kubectl/pkg/scheme"
+)
+
+// applyOverlay applies an overlay using JSON patch strategy over the current Object in place.
+func applyOverlay(current, overlay *unstructured.Unstructured) error {
+	cj, err := runtime.Encode(unstructured.UnstructuredJSONScheme, current)
+	if err != nil {
+		return err
+	}
+
+	overlayUpdated := overlay.DeepCopy()
+	if strings.EqualFold(current.GetKind(), "service") {
+		if err := saveClusterIP(current, overlayUpdated); err != nil {
+			return err
+		}
+
+		saveNodePorts(current, overlayUpdated)
+	}
+
+	if current.GetKind() == "PersistentVolumeClaim" {
+		if err := savePersistentVolumeClaim(current, overlayUpdated); err != nil {
+			return err
+		}
+	}
+
+	uj, err := runtime.Encode(unstructured.UnstructuredJSONScheme, overlayUpdated)
+	if err != nil {
+		return err
+	}
+	merged, err := jsonpatch.MergePatch(cj, uj)
+	if err != nil {
+		return err
+	}
+	return runtime.DecodeInto(unstructured.UnstructuredJSONScheme, merged, current)
+}
+
+// createPortMap returns a map, mapping the value of the port and value of the nodePort
+func createPortMap(current *unstructured.Unstructured) map[string]uint32 {
+	portMap := make(map[string]uint32)
+	svc := &corev1.Service{}
+	if err := scheme.Scheme.Convert(current, svc, nil); err != nil {
+		return portMap
+	}
+	for _, p := range svc.Spec.Ports {
+		portMap[strconv.Itoa(int(p.Port))] = uint32(p.NodePort)
+	}
+	return portMap
+}
+
+// savePersistentVolumeClaim copies the storageClassName from the current cluster into the overlay
+func savePersistentVolumeClaim(current, overlay *unstructured.Unstructured) error {
+	// Save the value of spec.storageClassName set by the cluster
+	if storageClassName, found, err := unstructured.NestedString(current.Object, "spec",
+		"storageClassName"); err != nil {
+		return err
+	} else if found {
+		if _, _, err2 := unstructured.NestedString(overlay.Object, "spec",
+			"storageClassName"); err2 != nil {
+			// override when overlay storageClassName property is not existed
+			if err3 := unstructured.SetNestedField(overlay.Object, storageClassName, "spec",
+				"storageClassName"); err3 != nil {
+				return err3
+			}
+		}
+	}
+	return nil
+}
+
+// saveNodePorts transfers the port values from the current cluster into the overlay
+func saveNodePorts(current, overlay *unstructured.Unstructured) {
+	portMap := createPortMap(current)
+	ports, _, _ := unstructured.NestedFieldNoCopy(overlay.Object, "spec", "ports")
+	portList, ok := ports.([]any)
+	if !ok {
+		return
+	}
+	for _, port := range portList {
+		m, ok := port.(map[string]any)
+		if !ok {
+			continue
+		}
+		if nodePortNum, ok := m["nodePort"]; ok && fmt.Sprintf("%v", nodePortNum) == "0" {
+			if portNum, ok := m["port"]; ok {
+				if v, ok := portMap[fmt.Sprintf("%v", portNum)]; ok {
+					m["nodePort"] = v
+				}
+			}
+		}
+	}
+}
+
+// saveClusterIP copies the cluster IP from the current cluster into the overlay
+func saveClusterIP(current, overlay *unstructured.Unstructured) error {
+	// Save the value of spec.clusterIP set by the cluster
+	if clusterIP, found, err := unstructured.NestedString(current.Object, "spec",
+		"clusterIP"); err != nil {
+		return err
+	} else if found {
+		if err := unstructured.SetNestedField(overlay.Object, clusterIP, "spec",
+			"clusterIP"); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func setRestDefaults(config *rest.Config) *rest.Config {
+	if config.GroupVersion == nil || config.GroupVersion.Empty() {
+		config.GroupVersion = &corev1.SchemeGroupVersion
+	}
+	if len(config.APIPath) == 0 {
+		if len(config.GroupVersion.Group) == 0 {
+			config.APIPath = "/api"
+		} else {
+			config.APIPath = "/apis"
+		}
+	}
+	if len(config.ContentType) == 0 {
+		config.ContentType = runtime.ContentTypeJSON
+	}
+	if config.NegotiatedSerializer == nil {
+		// This codec factory ensures the resources are not converted. Therefore, resources
+		// will not be round-tripped through internal versions. Defaulting does not happen
+		// on the client.
+		config.NegotiatedSerializer = serializer.NewCodecFactory(kubescheme.Scheme).WithoutConversion()
+	}
+
+	return config
+}