penguin/monok8s

clitools/pkg/render/agent_apply.go

raw · 5524 bytes

package render

import (
	"context"
	"fmt"
	"reflect"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/kubernetes"
)

func ApplyAgentDaemonSets(ctx context.Context, kubeClient kubernetes.Interface, conf AgentConf) error {
	objs, err := buildAgentDaemonSetObjects(conf)
	if err != nil {
		return err
	}

	if err := applyAgentNamespace(ctx, kubeClient, buildAgentNamespace(conf)); err != nil {
		return fmt.Errorf("apply namespace: %w", err)
	}

	for _, obj := range objs {
		if err := applyAgentObject(ctx, kubeClient, obj); err != nil {
			return err
		}
	}

	return nil
}

func applyAgentObject(ctx context.Context, kubeClient kubernetes.Interface, obj runtime.Object) error {
	switch want := obj.(type) {
	case *corev1.ServiceAccount:
		return applyAgentServiceAccount(ctx, kubeClient, want)
	case *rbacv1.ClusterRole:
		return applyAgentClusterRole(ctx, kubeClient, want)
	case *rbacv1.ClusterRoleBinding:
		return applyAgentClusterRoleBinding(ctx, kubeClient, want)
	case *appsv1.DaemonSet:
		return applyAgentDaemonSet(ctx, kubeClient, want)
	default:
		return fmt.Errorf("unsupported agent object type %T", obj)
	}
}

func applyAgentNamespace(ctx context.Context, kubeClient kubernetes.Interface, want *corev1.Namespace) error {
	existing, err := kubeClient.CoreV1().Namespaces().Get(ctx, want.Name, metav1.GetOptions{})
	if apierrors.IsNotFound(err) {
		_, err = kubeClient.CoreV1().Namespaces().Create(ctx, want, metav1.CreateOptions{})
		return err
	}
	if err != nil {
		return err
	}

	labels, changed := mergeStringMapsInto(existing.Labels, want.Labels)
	if !changed {
		return nil
	}

	existing.Labels = labels
	_, err = kubeClient.CoreV1().Namespaces().Update(ctx, existing, metav1.UpdateOptions{})
	return err
}

func applyAgentServiceAccount(ctx context.Context, kubeClient kubernetes.Interface, want *corev1.ServiceAccount) error {
	existing, err := kubeClient.CoreV1().ServiceAccounts(want.Namespace).Get(ctx, want.Name, metav1.GetOptions{})
	if apierrors.IsNotFound(err) {
		_, err = kubeClient.CoreV1().ServiceAccounts(want.Namespace).Create(ctx, want, metav1.CreateOptions{})
		return err
	}
	if err != nil {
		return err
	}

	changed := false
	if !reflect.DeepEqual(existing.Labels, want.Labels) {
		existing.Labels = want.Labels
		changed = true
	}

	if !changed {
		return nil
	}

	_, err = kubeClient.CoreV1().ServiceAccounts(want.Namespace).Update(ctx, existing, metav1.UpdateOptions{})
	return err
}

func applyAgentClusterRole(ctx context.Context, kubeClient kubernetes.Interface, want *rbacv1.ClusterRole) error {
	existing, err := kubeClient.RbacV1().ClusterRoles().Get(ctx, want.Name, metav1.GetOptions{})
	if apierrors.IsNotFound(err) {
		_, err = kubeClient.RbacV1().ClusterRoles().Create(ctx, want, metav1.CreateOptions{})
		return err
	}
	if err != nil {
		return err
	}

	changed := false
	if !reflect.DeepEqual(existing.Labels, want.Labels) {
		existing.Labels = want.Labels
		changed = true
	}
	if !reflect.DeepEqual(existing.Rules, want.Rules) {
		existing.Rules = want.Rules
		changed = true
	}

	if !changed {
		return nil
	}

	_, err = kubeClient.RbacV1().ClusterRoles().Update(ctx, existing, metav1.UpdateOptions{})
	return err
}

func applyAgentClusterRoleBinding(ctx context.Context, kubeClient kubernetes.Interface, want *rbacv1.ClusterRoleBinding) error {
	existing, err := kubeClient.RbacV1().ClusterRoleBindings().Get(ctx, want.Name, metav1.GetOptions{})
	if apierrors.IsNotFound(err) {
		_, err = kubeClient.RbacV1().ClusterRoleBindings().Create(ctx, want, metav1.CreateOptions{})
		return err
	}
	if err != nil {
		return err
	}

	// roleRef is immutable. If it differs, fail loudly instead of pretending we can patch it.
	if !reflect.DeepEqual(existing.RoleRef, want.RoleRef) {
		return fmt.Errorf("existing ClusterRoleBinding %q has different roleRef and must be recreated", want.Name)
	}

	changed := false
	if !reflect.DeepEqual(existing.Labels, want.Labels) {
		existing.Labels = want.Labels
		changed = true
	}
	if !reflect.DeepEqual(existing.Subjects, want.Subjects) {
		existing.Subjects = want.Subjects
		changed = true
	}

	if !changed {
		return nil
	}

	_, err = kubeClient.RbacV1().ClusterRoleBindings().Update(ctx, existing, metav1.UpdateOptions{})
	return err
}

func applyAgentDaemonSet(ctx context.Context, kubeClient kubernetes.Interface, want *appsv1.DaemonSet) error {
	existing, err := kubeClient.AppsV1().DaemonSets(want.Namespace).Get(ctx, want.Name, metav1.GetOptions{})
	if apierrors.IsNotFound(err) {
		_, err = kubeClient.AppsV1().DaemonSets(want.Namespace).Create(ctx, want, metav1.CreateOptions{})
		return err
	}
	if err != nil {
		return err
	}

	changed := false
	if !reflect.DeepEqual(existing.Labels, want.Labels) {
		existing.Labels = want.Labels
		changed = true
	}
	if !reflect.DeepEqual(existing.Spec, want.Spec) {
		existing.Spec = want.Spec
		changed = true
	}

	if !changed {
		return nil
	}

	_, err = kubeClient.AppsV1().DaemonSets(want.Namespace).Update(ctx, existing, metav1.UpdateOptions{})
	return err
}

func mergeStringMapsInto(dst map[string]string, src map[string]string) (map[string]string, bool) {
	if len(src) == 0 {
		return dst, false
	}

	changed := false
	if dst == nil {
		dst = map[string]string{}
		changed = true
	}

	for k, v := range src {
		if dst[k] != v {
			dst[k] = v
			changed = true
		}
	}

	return dst, changed
}