From c0f01c58d694fba75fdffe857083d1c4c2061539 Mon Sep 17 00:00:00 2001 From: Jose Diaz-Gonzalez Date: Wed, 28 Feb 2024 17:05:08 -0500 Subject: [PATCH] feat: add initial support for interacting with ClusterTriggerAuthentication objects These will be provisioned to the keda namespace and referenced by Triggers if they exist (as a fallback to app-specific values). --- plugins/scheduler-k3s/functions.go | 33 ++++++- plugins/scheduler-k3s/k8s.go | 87 +++++++++++++++++++ plugins/scheduler-k3s/template.go | 9 ++ .../templates/chart/keda-cluster-secret.yaml | 25 ++++++ .../keda-cluster-trigger-authentication.yaml | 26 ++++++ plugins/scheduler-k3s/triggers.go | 20 ++--- 6 files changed, 186 insertions(+), 14 deletions(-) create mode 100644 plugins/scheduler-k3s/templates/chart/keda-cluster-secret.yaml create mode 100644 plugins/scheduler-k3s/templates/chart/keda-cluster-trigger-authentication.yaml diff --git a/plugins/scheduler-k3s/functions.go b/plugins/scheduler-k3s/functions.go index 0ee22af7b..6ae1cf38b 100644 --- a/plugins/scheduler-k3s/functions.go +++ b/plugins/scheduler-k3s/functions.go @@ -512,9 +512,14 @@ func getAutoscaling(input GetAutoscalingInput) (ProcessAutoscaling, error) { Metadata: metadata, } - if _, ok := input.KedaValues.Authentications[trigger.Type]; ok { + if auth, ok := input.KedaValues.Authentications[trigger.Type]; ok { trigger.AuthenticationRef = &ProcessAutoscalingTriggerAuthenticationRef{ - Name: fmt.Sprintf("%s-%s", input.AppName, trigger.Type), + Name: auth.Name, + } + } else if auth, ok := input.KedaValues.GlobalAuthentications[trigger.Type]; ok { + trigger.AuthenticationRef = &ProcessAutoscalingTriggerAuthenticationRef{ + Name: auth.Name, + Kind: "ClusterTriggerAuthentication", } } @@ -535,7 +540,7 @@ func getAutoscaling(input GetAutoscalingInput) (ProcessAutoscaling, error) { } // getKedaValues retrieves keda values for a given app and process type -func getKedaValues(appName string) (GlobalKedaValues, error) { +func getKedaValues(ctx context.Context, clientset KubernetesClient, appName string) (GlobalKedaValues, error) { properties, err := common.PropertyGetAllByPrefix("scheduler-k3s", appName, TriggerAuthPropertyPrefix) if err != nil { return GlobalKedaValues{}, fmt.Errorf("Error getting trigger-auth properties: %w", err) @@ -556,6 +561,7 @@ func getKedaValues(appName string) (GlobalKedaValues, error) { if _, ok := auths[authType]; !ok { auths[authType] = KedaAuthentication{ + Name: fmt.Sprintf("%s-%s", appName, authType), Type: authType, Secrets: make(map[string]string), } @@ -564,8 +570,27 @@ func getKedaValues(appName string) (GlobalKedaValues, error) { auths[authType].Secrets[secretKey] = base64.StdEncoding.EncodeToString([]byte(value)) } + items, err := clientset.ListClusterTriggerAuthentications(ctx, ListClusterTriggerAuthenticationsInput{}) + if err != nil { + return GlobalKedaValues{}, fmt.Errorf("Error listing cluster trigger authentications: %w", err) + } + + globalAuths := map[string]KedaAuthentication{} + for _, item := range items { + if !strings.HasPrefix(item.Name, "global-auth-") { + continue + } + + authType := strings.TrimPrefix(item.Name, "global-auth-") + globalAuths[authType] = KedaAuthentication{ + Name: item.Name, + Type: authType, + } + } + return GlobalKedaValues{ - Authentications: auths, + Authentications: auths, + GlobalAuthentications: globalAuths, }, nil } diff --git a/plugins/scheduler-k3s/k8s.go b/plugins/scheduler-k3s/k8s.go index 556d8d4ab..f7f13245e 100644 --- a/plugins/scheduler-k3s/k8s.go +++ b/plugins/scheduler-k3s/k8s.go @@ -7,6 +7,7 @@ import ( "github.com/dokku/dokku/plugins/common" "github.com/go-openapi/jsonpointer" + kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1" appsv1 "k8s.io/api/apps/v1" autoscalingv1 "k8s.io/api/autoscaling/v1" batchv1 "k8s.io/api/batch/v1" @@ -15,6 +16,7 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/clientcmd" @@ -35,6 +37,9 @@ type KubernetesClient struct { // Client is the Kubernetes client Client kubernetes.Clientset + // DynamicClient is the Kubernetes dynamic client + DynamicClient dynamic.Interface + // KubeConfigPath is the path to the Kubernetes config KubeConfigPath string @@ -72,8 +77,14 @@ func NewKubernetesClient() (KubernetesClient, error) { return KubernetesClient{}, err } + dynamicClient, err := dynamic.NewForConfig(restConf) + if err != nil { + return KubernetesClient{}, err + } + return KubernetesClient{ Client: *client, + DynamicClient: dynamicClient, KubeConfigPath: kubeconfigPath, RestConfig: *restConf, RestClient: restClient, @@ -337,6 +348,44 @@ func (k KubernetesClient) LabelNode(ctx context.Context, input LabelNodeInput) e return nil } +// ListClusterTriggerAuthenticationsInput contains all the information needed to list Kubernetes trigger authentications +type ListClusterTriggerAuthenticationsInput struct { + // Namespace is the Kubernetes namespace + Namespace string + + // LabelSelector is the Kubernetes label selector + LabelSelector string +} + +// ListClusterTriggerAuthentications lists Kubernetes trigger authentications +func (k KubernetesClient) ListClusterTriggerAuthentications(ctx context.Context, input ListClusterTriggerAuthenticationsInput) ([]kedav1alpha1.ClusterTriggerAuthentication, error) { + listOptions := metav1.ListOptions{LabelSelector: input.LabelSelector} + + gvr := schema.GroupVersionResource{ + Group: "keda.sh", + Version: "v1alpha1", + Resource: "clustertriggerauthentications", + } + + response, err := k.DynamicClient.Resource(gvr).Namespace(input.Namespace).List(ctx, listOptions) + if err != nil { + return []kedav1alpha1.ClusterTriggerAuthentication{}, err + } + + triggerAuthentications := []kedav1alpha1.ClusterTriggerAuthentication{} + for _, triggerAuthentication := range response.Items { + var ta kedav1alpha1.ClusterTriggerAuthentication + err := runtime.DefaultUnstructuredConverter.FromUnstructured(triggerAuthentication.Object, &ta) + if err != nil { + return []kedav1alpha1.ClusterTriggerAuthentication{}, err + } + + triggerAuthentications = append(triggerAuthentications, ta) + } + + return triggerAuthentications, nil +} + // ListCronJobsInput contains all the information needed to list Kubernetes cron jobs type ListCronJobsInput struct { // LabelSelector is the Kubernetes label selector @@ -451,6 +500,44 @@ func (k KubernetesClient) ListPods(ctx context.Context, input ListPodsInput) ([] return podList.Items, err } +// ListTriggerAuthenticationsInput contains all the information needed to list Kubernetes trigger authentications +type ListTriggerAuthenticationsInput struct { + // Namespace is the Kubernetes namespace + Namespace string + + // LabelSelector is the Kubernetes label selector + LabelSelector string +} + +// ListTriggerAuthentications lists Kubernetes trigger authentications +func (k KubernetesClient) ListTriggerAuthentications(ctx context.Context, input ListTriggerAuthenticationsInput) ([]kedav1alpha1.TriggerAuthentication, error) { + listOptions := metav1.ListOptions{LabelSelector: input.LabelSelector} + + gvr := schema.GroupVersionResource{ + Group: "keda.sh", + Version: "v1alpha1", + Resource: "triggerauthentications", + } + + response, err := k.DynamicClient.Resource(gvr).Namespace(input.Namespace).List(ctx, listOptions) + if err != nil { + return []kedav1alpha1.TriggerAuthentication{}, err + } + + triggerAuthentications := []kedav1alpha1.TriggerAuthentication{} + for _, triggerAuthentication := range response.Items { + var ta kedav1alpha1.TriggerAuthentication + err := runtime.DefaultUnstructuredConverter.FromUnstructured(triggerAuthentication.Object, &ta) + if err != nil { + return []kedav1alpha1.TriggerAuthentication{}, err + } + + triggerAuthentications = append(triggerAuthentications, ta) + } + + return triggerAuthentications, nil +} + // ScaleDeploymentInput contains all the information needed to scale a Kubernetes deployment type ScaleDeploymentInput struct { // Name is the Kubernetes deployment name diff --git a/plugins/scheduler-k3s/template.go b/plugins/scheduler-k3s/template.go index 435d414ce..3bf76fdfd 100644 --- a/plugins/scheduler-k3s/template.go +++ b/plugins/scheduler-k3s/template.go @@ -61,10 +61,16 @@ type GlobalNetwork struct { type GlobalKedaValues struct { // Authentications is a map of authentication objects to use for keda Authentications map[string]KedaAuthentication `yaml:"authentications"` + + // GlobalAuthentications is a map of global authentication objects to use for keda + GlobalAuthentications map[string]KedaAuthentication `yaml:"global_authentications"` } // KedaAuthentication contains the authentication configuration for keda type KedaAuthentication struct { + // Name is the name of the authentication object + Name string `yaml:"name"` + // Type is the type of authentication to use Type string `yaml:"type"` @@ -145,6 +151,9 @@ type ProcessAutoscalingTrigger struct { type ProcessAutoscalingTriggerAuthenticationRef struct { // Name is the name of the authentication object Name string `yaml:"name"` + + // Kind is the kind of authentication object + Kind string `yaml:"kind,omitempty"` } type ProcessHealthchecks struct { diff --git a/plugins/scheduler-k3s/templates/chart/keda-cluster-secret.yaml b/plugins/scheduler-k3s/templates/chart/keda-cluster-secret.yaml new file mode 100644 index 000000000..cd9d997eb --- /dev/null +++ b/plugins/scheduler-k3s/templates/chart/keda-cluster-secret.yaml @@ -0,0 +1,25 @@ +{{- if .Capabilities.APIVersions.Has "keda.sh/v1alpha1" -}} +--- +apiVersion: v1 +kind: Secret +metadata: + annotations: + app.kubernetes.io/version: {{ $.Values.global.deploment_id | quote }} + dokku.com/managed: "true" + {{- if and $.Values.global.annotations $.Values.global.annotations.keda_secret }} + {{- range $k, $v := $.Values.global.annotations.keda_secret }} + {{ $k }}: {{ $v | quote }} + {{- end }} + {{- end }} + labels: + app.kubernetes.io/instance: {{ $.Values.type }} + app.kubernetes.io/name: {{ $.Values.type }} + app.kubernetes.io/part-of: cluster + name: ckta-{{ $.Values.type }} + namespace: keda +data: +{{- with $.Values.secrets }} +data: + {{- toYaml . | nindent 2 }} +{{- end }} +{{- end }} diff --git a/plugins/scheduler-k3s/templates/chart/keda-cluster-trigger-authentication.yaml b/plugins/scheduler-k3s/templates/chart/keda-cluster-trigger-authentication.yaml new file mode 100644 index 000000000..b74b9d30b --- /dev/null +++ b/plugins/scheduler-k3s/templates/chart/keda-cluster-trigger-authentication.yaml @@ -0,0 +1,26 @@ +{{- if .Capabilities.APIVersions.Has "keda.sh/v1alpha1" -}} +--- +apiVersion: keda.sh/v1alpha1 +kind: ClusterTriggerAuthentication +metadata: + annotations: + app.kubernetes.io/version: {{ $.Values.global.deploment_id | quote }} + dokku.com/managed: "true" + {{- if and $.Values.global.annotations $.Values.global.annotations.keda_trigger_authentication }} + {{- range $k, $v := $.Values.global.annotations.keda_trigger_authentication }} + {{ $k }}: {{ $v | quote }} + {{- end }} + {{- end }} + labels: + app.kubernetes.io/instance: {{ $.Values.type }} + app.kubernetes.io/name: {{ $.Values.type }} + app.kubernetes.io/part-of: cluster + name: {{ $.Values.type }} +spec: + secretTargetRef: + {{- range $k, $v := $.Values.secrets }} + - key: {{ $k }} + name: ckta-{{ $.Values.type }} + parameter: {{ $k }} + {{- end }} +{{- end }} diff --git a/plugins/scheduler-k3s/triggers.go b/plugins/scheduler-k3s/triggers.go index 952a79a57..a3195c3ce 100644 --- a/plugins/scheduler-k3s/triggers.go +++ b/plugins/scheduler-k3s/triggers.go @@ -266,7 +266,16 @@ func TriggerSchedulerDeploy(scheduler string, appName string, imageTag string) e return fmt.Errorf("Error getting global labels: %w", err) } - kedaValues, err := getKedaValues(appName) + clientset, err := NewKubernetesClient() + if err != nil { + return fmt.Errorf("Error creating kubernetes client: %w", err) + } + + if err := clientset.Ping(); err != nil { + return fmt.Errorf("kubernetes api not available: %w", err) + } + + kedaValues, err := getKedaValues(ctx, clientset, appName) if err != nil { return fmt.Errorf("Error getting keda values: %w", err) } @@ -454,15 +463,6 @@ func TriggerSchedulerDeploy(scheduler string, appName string, imageTag string) e } } - clientset, err := NewKubernetesClient() - if err != nil { - return fmt.Errorf("Error creating kubernetes client: %w", err) - } - - if err := clientset.Ping(); err != nil { - return fmt.Errorf("kubernetes api not available: %w", err) - } - cronJobs, err := clientset.ListCronJobs(ctx, ListCronJobsInput{ LabelSelector: fmt.Sprintf("app.kubernetes.io/part-of=%s", appName), Namespace: namespace,