Embedded cluster working group

.github/actions/cmx-versions/dist/index.js

@@ -7710,4 +7710,4 @@ getClusterVersions();
 
 module.exports = __webpack_exports__;
 /******/ })()
-;
+;

.github/workflows/regression.yaml

@@ -191,9 +191,9 @@ jobs:
             is_upgrade: "1"
           },
           {
-            name: "type=helmvm cluster, env=online, phase=new install, rbac=cluster admin",
-            backend_config: "helmvm-online-install-backend-config.tfvars",
-            terraform_script: "helmvm-online-install.sh"
+            name: "type=embeddedcluster cluster, env=online, phase=new install, rbac=cluster admin",
+            backend_config: "embeddedcluster-online-install-backend-config.tfvars",
+            terraform_script: "embeddedcluster-online-install.sh"
           }
         ]
     steps:

migrations/tables/embeded_cluster_tokens.yaml

@@ -0,0 +1,21 @@
+apiVersion: schemas.schemahero.io/v1alpha4
+kind: Table
+metadata:
+  name: embedded-cluster-tokens
+spec:
+  name: embedded_cluster_tokens
+  requires: []
+  schema:
+    rqlite:
+      strict: true
+      primaryKey:
+      - token
+      columns:
+      - name: token
+        type: text
+        constraints:
+          notNull: true
+      - name: roles
+        type: text
+        constraints:
+          notNull: true

pkg/helmvm/delete_node.go → pkg/embeddedcluster/delete_node.go

@@ -1,4 +1,4 @@
-package helmvm
+package embeddedcluster
 
 import (
 	"context"

pkg/helmvm/drain_node.go → pkg/embeddedcluster/drain_node.go

@@ -1,4 +1,4 @@
-package helmvm
+package embeddedcluster
 
 import (
 	"context"

pkg/helmvm/exec.go → pkg/embeddedcluster/exec.go

@@ -1,4 +1,4 @@
-package helmvm
+package embeddedcluster
 
 import (
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"

pkg/embeddedcluster/helmvm_node.go

@@ -0,0 +1,169 @@
+package embeddedcluster
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"strconv"
+	"strings"
+
+	"github.com/replicatedhq/kots/pkg/embeddedcluster/types"
+	"github.com/replicatedhq/kots/pkg/k8sutil"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	metricsv "k8s.io/metrics/pkg/client/clientset/versioned"
+)
+
+// GetNode will get a node with stats and podlist
+func GetNode(ctx context.Context, client kubernetes.Interface, nodeName string) (*types.Node, error) {
+	node, err := client.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("get node %s: %w", nodeName, err)
+	}
+
+	clientConfig, err := k8sutil.GetClusterConfig()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get cluster config: %w", err)
+	}
+
+	metricsClient, err := metricsv.NewForConfig(clientConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create metrics client: %w", err)
+	}
+
+	return nodeMetrics(ctx, client, metricsClient, *node)
+}
+
+func podsOnNode(ctx context.Context, client kubernetes.Interface, nodeName string) ([]corev1.Pod, error) {
+	namespaces, err := client.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("list namespaces: %w", err)
+	}
+
+	toReturn := []corev1.Pod{}
+
+	for _, ns := range namespaces.Items {
+		nsPods, err := client.CoreV1().Pods(ns.Name).List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("spec.nodeName=%s", nodeName)})
+		if err != nil {
+			return nil, fmt.Errorf("list pods on %s in namespace %s: %w", nodeName, ns.Name, err)
+		}
+
+		toReturn = append(toReturn, nsPods.Items...)
+	}
+	return toReturn, nil
+}
+
+// nodeMetrics takes a corev1.Node and gets metrics + status for that node
+func nodeMetrics(ctx context.Context, client kubernetes.Interface, metricsClient *metricsv.Clientset, node corev1.Node) (*types.Node, error) {
+	nodePods, err := podsOnNode(ctx, client, node.Name)
+	if err != nil {
+		return nil, fmt.Errorf("pods per node: %w", err)
+	}
+
+	cpuCapacity := types.CapacityUsed{}
+	memoryCapacity := types.CapacityUsed{}
+	podCapacity := types.CapacityUsed{}
+
+	memoryCapacity.Capacity = float64(node.Status.Capacity.Memory().Value()) / math.Pow(2, 30) // capacity in GB
+
+	cpuCapacity.Capacity, err = strconv.ParseFloat(node.Status.Capacity.Cpu().String(), 64)
+	if err != nil {
+		return nil, fmt.Errorf("parse CPU capacity %q for node %s: %w", node.Status.Capacity.Cpu().String(), node.Name, err)
+	}
+
+	podCapacity.Capacity = float64(node.Status.Capacity.Pods().Value())
+
+	nodeUsageMetrics, err := metricsClient.MetricsV1beta1().NodeMetricses().Get(ctx, node.Name, metav1.GetOptions{})
+	if err == nil {
+		if nodeUsageMetrics.Usage.Memory() != nil {
+			memoryCapacity.Used = float64(nodeUsageMetrics.Usage.Memory().Value()) / math.Pow(2, 30)
+		}
+
+		if nodeUsageMetrics.Usage.Cpu() != nil {
+			cpuCapacity.Used = nodeUsageMetrics.Usage.Cpu().AsApproximateFloat64()
+		}
+	} else {
+		// if we can't get metrics, we'll do nothing for now
+		// in the future we may decide to retry or log a warning
+	}
+
+	podCapacity.Used = float64(len(nodePods))
+
+	podInfo := []types.PodInfo{}
+
+	for _, pod := range nodePods {
+		newInfo := types.PodInfo{
+			Name:      pod.Name,
+			Namespace: pod.Namespace,
+			Status:    string(pod.Status.Phase),
+		}
+
+		podMetrics, err := metricsClient.MetricsV1beta1().PodMetricses(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{})
+		if err == nil {
+			podTotalMemory := 0.0
+			podTotalCPU := 0.0
+			for _, container := range podMetrics.Containers {
+				if container.Usage.Memory() != nil {
+					podTotalMemory += float64(container.Usage.Memory().Value()) / math.Pow(2, 20)
+				}
+				if container.Usage.Cpu() != nil {
+					podTotalCPU += container.Usage.Cpu().AsApproximateFloat64() * 1000
+				}
+			}
+			newInfo.Memory = fmt.Sprintf("%.1f MB", podTotalMemory)
+			newInfo.CPU = fmt.Sprintf("%.1f m", podTotalCPU)
+		}
+
+		podInfo = append(podInfo, newInfo)
+	}
+
+	return &types.Node{
+		Name:             node.Name,
+		IsConnected:      isConnected(node),
+		IsReady:          isReady(node),
+		IsPrimaryNode:    isPrimary(node),
+		CanDelete:        node.Spec.Unschedulable && !isConnected(node),
+		KubeletVersion:   node.Status.NodeInfo.KubeletVersion,
+		KubeProxyVersion: node.Status.NodeInfo.KubeProxyVersion,
+		OperatingSystem:  node.Status.NodeInfo.OperatingSystem,
+		KernelVersion:    node.Status.NodeInfo.KernelVersion,
+		CPU:              cpuCapacity,
+		Memory:           memoryCapacity,
+		Pods:             podCapacity,
+		Labels:           nodeRolesFromLabels(node.Labels),
+		Conditions:       findNodeConditions(node.Status.Conditions),
+		PodList:          podInfo,
+	}, nil
+}
+
+// nodeRolesFromLabels parses a map of k8s node labels, and returns the roles of the node
+func nodeRolesFromLabels(labels map[string]string) []string {
+	toReturn := []string{}
+
+	numRolesStr, ok := labels[types.EMBEDDED_CLUSTER_ROLE_LABEL]
+	if !ok {
+		// the first node will not initially have a role label, but is a 'controller'
+		if val, ok := labels["node-role.kubernetes.io/control-plane"]; ok && val == "true" {
+			return []string{"controller"}
+		}
+		return nil
+	}
+	numRoles, err := strconv.Atoi(strings.TrimPrefix(numRolesStr, "total-"))
+	if err != nil {
+		fmt.Printf("failed to parse role label %q: %s", numRolesStr, err.Error())
+
+		return nil
+	}
+
+	for i := 0; i < numRoles; i++ {
+		roleLabel, ok := labels[fmt.Sprintf("%s-%d", types.EMBEDDED_CLUSTER_ROLE_LABEL, i)]
+		if !ok {
+			fmt.Printf("failed to find role label %d", i)
+			continue
+		}
+		toReturn = append(toReturn, roleLabel)
+	}
+
+	return toReturn
+}

pkg/embeddedcluster/helmvm_nodes.go

@@ -0,0 +1,108 @@
+package embeddedcluster
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"github.com/replicatedhq/kots/pkg/embeddedcluster/types"
+	"github.com/replicatedhq/kots/pkg/k8sutil"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	metricsv "k8s.io/metrics/pkg/client/clientset/versioned"
+)
+
+// GetNodes will get a list of nodes with stats
+func GetNodes(ctx context.Context, client kubernetes.Interface) (*types.EmbeddedClusterNodes, error) {
+	nodes, err := client.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, errors.Wrap(err, "list nodes")
+	}
+
+	clientConfig, err := k8sutil.GetClusterConfig()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get cluster config")
+	}
+
+	metricsClient, err := metricsv.NewForConfig(clientConfig)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create metrics client")
+	}
+
+	toReturn := types.EmbeddedClusterNodes{}
+
+	for _, node := range nodes.Items {
+		nodeMet, err := nodeMetrics(ctx, client, metricsClient, node)
+		if err != nil {
+			return nil, errors.Wrap(err, "node metrics")
+		}
+
+		toReturn.Nodes = append(toReturn.Nodes, *nodeMet)
+	}
+
+	isEmbeddedCluster, err := IsEmbeddedCluster(client)
+	if err != nil {
+		return nil, errors.Wrap(err, "is embeddedcluster")
+	}
+	toReturn.IsEmbeddedClusterEnabled = isEmbeddedCluster
+
+	isHA, err := IsHA(client)
+	if err != nil {
+		return nil, errors.Wrap(err, "is ha")
+	}
+	toReturn.HA = isHA
+
+	return &toReturn, nil
+}
+
+func findNodeConditions(conditions []corev1.NodeCondition) types.NodeConditions {
+	discoveredConditions := types.NodeConditions{}
+	for _, condition := range conditions {
+		if condition.Type == "MemoryPressure" {
+			discoveredConditions.MemoryPressure = condition.Status == corev1.ConditionTrue
+		}
+		if condition.Type == "DiskPressure" {
+			discoveredConditions.DiskPressure = condition.Status == corev1.ConditionTrue
+		}
+		if condition.Type == "PIDPressure" {
+			discoveredConditions.PidPressure = condition.Status == corev1.ConditionTrue
+		}
+		if condition.Type == "Ready" {
+			discoveredConditions.Ready = condition.Status == corev1.ConditionTrue
+		}
+	}
+	return discoveredConditions
+}
+
+func isConnected(node corev1.Node) bool {
+	for _, taint := range node.Spec.Taints {
+		if taint.Key == "node.kubernetes.io/unreachable" {
+			return false
+		}
+	}
+
+	return true
+}
+
+func isReady(node corev1.Node) bool {
+	for _, condition := range node.Status.Conditions {
+		if condition.Type == "Ready" {
+			return condition.Status == corev1.ConditionTrue
+		}
+	}
+
+	return false
+}
+
+func isPrimary(node corev1.Node) bool {
+	for label := range node.ObjectMeta.Labels {
+		if label == "node-role.kubernetes.io/master" {
+			return true
+		}
+		if label == "node-role.kubernetes.io/control-plane" {
+			return true
+		}
+	}
+
+	return false
+}