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).
This commit is contained in:
Jose Diaz-Gonzalez
2024-02-28 17:05:08 -05:00
parent 94125ce787
commit c0f01c58d6
6 changed files with 186 additions and 14 deletions

View File

@@ -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
}

View File

@@ -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

View File

@@ -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 {

View File

@@ -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 }}

View File

@@ -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 }}

View File

@@ -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,