Add support for KEDA

[nicolaferraro]

Fix #1107

This adds support for KEDA autoscalers. It supports both manual configuration and fully automated configuration based on Kamelet metadata. In the fully automated scenario, Kamelets are annotated with special KEDA markers to tell the KEDA trait how to configure properties. I've added documentation in the Kamelet dev and user guide.

Everything is developed under addons. The only bits that are added to the main codebase are the roles/bindings for which a major refactoring was needed.

I've not directly pointed to the KEDA repo to import the API since they use a single module. We can ask to provide APIs in a separate module in the future. Until then, we need to sync the APIs from time to time. We only use 2 objects and I've included only fields relevant for us.

The KEDA trait allows configuring multiple triggers for an integration. In order to do it, I had to extend the CLI to support slices of complex objects. KEDA also requires that the resources being scaled have a spec -> replicas field, that we don't put by default. So, when the KEDA trait is enabled there's a specific hack in place that sets that field explicitly to make KEDA work.
Among manual configs, the user can indicate a secret per trigger that contains authentication options, other than plain metadata.

Auto-configuration works with Kamelet. By annotating Kamelets/properties with special markers, the trait is able to map Kamelet configuration to KEDA configuration, so that a KameletBinding can automatically create all KEDA resources without additional input from the user.

I've also created some KEDA-enabled Kamelets that I'll publish when this PR gets merged.
The trick is to add something like this in the Kamelet definition:

    properties:
      topic:
        title: Topic Names
        description: Comma separated list of Kafka topic names
        type: string
        x-descriptors:
        - urn:keda:metadata:topic
        - urn:keda:required

That tells to the trait that the topic Kamelet property corresponds to the topic KEDA parameter. There are more sophisticated configuration explained in the dev guide.

So when you bind e.g. a Kafka Source Kamelet to a destination, you need to add a trait.camel.apache.org/keda.enabled=true annotation and nothing else and you'll have autoscaling and scaling to zero from Kafka.

I've also added metadata to the AWS-SQS Kamelet and plan to do a demo with both.
The new Kamelet looks like this:

# ---------------------------------------------------------------------------
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.
# ---------------------------------------------------------------------------

apiVersion: camel.apache.org/v1alpha1
kind: Kamelet
metadata:
  name: aws-sqs-source
  annotations:
    camel.apache.org/kamelet.support.level: "Preview"
    camel.apache.org/catalog.version: "main-SNAPSHOT"
    camel.apache.org/kamelet.icon: "data:image/svg+xml;base64,PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiIHN0YW5kYWxvbmU9Im5vIj8+CjxzdmcKICAgeG1sbnM6ZGM9Imh0dHA6Ly9wdXJsLm9yZy9kYy9lbGVtZW50cy8xLjEvIgogICB4bWxuczpjYz0iaHR0cDovL2NyZWF0aXZlY29tbW9ucy5vcmcvbnMjIgogICB4bWxuczpyZGY9Imh0dHA6Ly93d3cudzMub3JnLzE5OTkvMDIvMjItcmRmLXN5bnRheC1ucyMiCiAgIHhtbG5zOnN2Zz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciCiAgIHhtbG5zPSJodHRwOi8vd3d3LnczLm9yZy8yMDAwL3N2ZyIKICAgeG1sbnM6c29kaXBvZGk9Imh0dHA6Ly9zb2RpcG9kaS5zb3VyY2Vmb3JnZS5uZXQvRFREL3NvZGlwb2RpLTAuZHRkIgogICB4bWxuczppbmtzY2FwZT0iaHR0cDovL3d3dy5pbmtzY2FwZS5vcmcvbmFtZXNwYWNlcy9pbmtzY2FwZSIKICAgZW5hYmxlLWJhY2tncm91bmQ9Im5ldyAwIDAgMTAwIDEwMCIKICAgaGVpZ2h0PSI3Mi4xOTk5OTciCiAgIHZlcnNpb249IjEuMSIKICAgdmlld0JveD0iMCAwIDU5Ljg0OTk5OCA3Mi4xOTk5OTciCiAgIHdpZHRoPSI1OS44NDk5OTgiCiAgIHhtbDpzcGFjZT0icHJlc2VydmUiCiAgIGlkPSJzdmcxNDUiCiAgIHNvZGlwb2RpOmRvY25hbWU9ImRvd25sb2FkLnN2ZyIKICAgaW5rc2NhcGU6dmVyc2lvbj0iMS4wLjIgKGU4NmM4NzA4NzksIDIwMjEtMDEtMTUpIj48bWV0YWRhdGEKICAgICBpZD0ibWV0YWRhdGExNTEiPjxyZGY6UkRGPjxjYzpXb3JrCiAgICAgICAgIHJkZjphYm91dD0iIj48ZGM6Zm9ybWF0PmltYWdlL3N2Zyt4bWw8L2RjOmZvcm1hdD48ZGM6dHlwZQogICAgICAgICAgIHJkZjpyZXNvdXJjZT0iaHR0cDovL3B1cmwub3JnL2RjL2RjbWl0eXBlL1N0aWxsSW1hZ2UiIC8+PGRjOnRpdGxlPjwvZGM6dGl0bGU+PC9jYzpXb3JrPjwvcmRmOlJERj48L21ldGFkYXRhPjxkZWZzCiAgICAgaWQ9ImRlZnMxNDkiIC8+PHNvZGlwb2RpOm5hbWVkdmlldwogICAgIHBhZ2Vjb2xvcj0iI2ZmZmZmZiIKICAgICBib3JkZXJjb2xvcj0iIzY2NjY2NiIKICAgICBib3JkZXJvcGFjaXR5PSIxIgogICAgIG9iamVjdHRvbGVyYW5jZT0iMTAiCiAgICAgZ3JpZHRvbGVyYW5jZT0iMTAiCiAgICAgZ3VpZGV0b2xlcmFuY2U9IjEwIgogICAgIGlua3NjYXBlOnBhZ2VvcGFjaXR5PSIwIgogICAgIGlua3NjYXBlOnBhZ2VzaGFkb3c9IjIiCiAgICAgaW5rc2NhcGU6d2luZG93LXdpZHRoPSIxOTIwIgogICAgIGlua3NjYXBlOndpbmRvdy1oZWlnaHQ9IjEwMTYiCiAgICAgaWQ9Im5hbWVkdmlldzE0NyIKICAgICBzaG93Z3JpZD0iZmFsc2UiCiAgICAgZml0LW1hcmdpbi10b3A9IjAuMSIKICAgICBmaXQtbWFyZ2luLWxlZnQ9IjAuMSIKICAgICBmaXQtbWFyZ2luLXJpZ2h0PSIwLjEiCiAgICAgZml0LW1hcmdpbi1ib3R0b209IjAuMSIKICAgICBpbmtzY2FwZTp6b29tPSI4LjE5IgogICAgIGlua3NjYXBlOmN4PSIyOS45MjUiCiAgICAgaW5rc2NhcGU6Y3k9IjM2LjEiCiAgICAgaW5rc2NhcGU6d2luZG93LXg9IjAiCiAgICAgaW5rc2NhcGU6d2luZG93LXk9IjI3IgogICAgIGlua3NjYXBlOndpbmRvdy1tYXhpbWl6ZWQ9IjEiCiAgICAgaW5rc2NhcGU6Y3VycmVudC1sYXllcj0ic3ZnMTQ1IiAvPjxnCiAgICAgaWQ9IkFtYXpvbl9DbG91ZFNlYXJjaCIKICAgICB0cmFuc2Zvcm09InRyYW5zbGF0ZSgtMjAuMDc1LC0xMy45KSI+PGcKICAgICAgIGlkPSJnMTQyIj48cG9seWdvbgogICAgICAgICBmaWxsPSIjZDlhNzQxIgogICAgICAgICBwb2ludHM9IjIzLjk4NywzNi4yMDEgNTQuNDYyLDQwLjQ5NCA1NC40Niw1OS41MDYgMjMuOTg1LDYzLjc5MyAiCiAgICAgICAgIGlkPSJwb2x5Z29uMTIwIiAvPjxwb2x5Z29uCiAgICAgICAgIGZpbGw9IiM4NzY5MjkiCiAgICAgICAgIHBvaW50cz0iNTAuMDAzLDE0IDIwLjE3OSwyOC45MDggMjAuMTc5LDM3LjM0NCA1MC4wMDMsMjguMzk5ICIKICAgICAgICAgaWQ9InBvbHlnb24xMjIiIC8+PHBvbHlnb24KICAgICAgICAgZmlsbD0iIzg3NjkyOSIKICAgICAgICAgcG9pbnRzPSI0OS45OTcsODYgMjAuMTksNzEuMDk0IDIwLjE5LDYyLjY1NCA0OS45OTksNzEuNiAiCiAgICAgICAgIGlkPSJwb2x5Z29uMTI0IiAvPjxwb2x5Z29uCiAgICAgICAgIGZpbGw9IiNkOWE3NDEiCiAgICAgICAgIHBvaW50cz0iNTAuMDAzLDE0IDc5LjgyNSwyOC45MTQgNzkuODIzLDM3LjM1IDUwLjAwMywyOC4zOTkgIgogICAgICAgICBpZD0icG9seWdvbjEyNiIgLz48cG9seWdvbgogICAgICAgICBmaWxsPSIjZDlhNzQxIgogICAgICAgICBwb2ludHM9IjQ5Ljk5Nyw4NiA3OS44MDYsNzEuMDk5IDc5LjgwNiw2Mi42NiA0OS45OTksNzEuNiAiCiAgICAgICAgIGlkPSJwb2x5Z29uMTI4IiAvPjxwb2x5Z29uCiAgICAgICAgIGZpbGw9IiM4NzY5MjkiCiAgICAgICAgIHBvaW50cz0iMjAuMTc5LDI4LjkwOCAyMy45ODksMjcuMDA0IDIzLjk4NSw3Mi45OSAyMC4xNzUsNzEuMDg2ICIKICAgICAgICAgaWQ9InBvbHlnb24xMzAiIC8+PHBvbHlnb24KICAgICAgICAgZmlsbD0iIzg3NjkyOSIKICAgICAgICAgcG9pbnRzPSI1MC4wMDEsNDAuODMyIDM5LjAxOSw0Mi4yMjkgMzkuMDE3LDU3Ljc2MiA0OS45OTksNTkuMTYgIgogICAgICAgICBpZD0icG9seWdvbjEzMiIgLz48cG9seWdvbgogICAgICAgICBmaWxsPSIjODc2OTI5IgogICAgICAgICBwb2ludHM9IjM1LjA4Myw0Mi43MjkgMjcuOTU0LDQzLjYzNyAyNy45NTQsNTYuMzU0IDM1LjA4MSw1Ny4yNiAiCiAgICAgICAgIGlkPSJwb2x5Z29uMTM0IiAvPjxwb2x5Z29uCiAgICAgICAgIGZpbGw9IiM2MjRhMWUiCiAgICAgICAgIHBvaW50cz0iNzkuODIzLDM3LjM1IDU0LjQ4MSw0MC40OTYgMjMuOTg3LDM2LjIwMSA1MC4wMDMsMjguMzk5ICIKICAgICAgICAgaWQ9InBvbHlnb24xMzYiIC8+PHBvbHlnb24KICAgICAgICAgZmlsbD0iI2ZhZDc5MSIKICAgICAgICAgcG9pbnRzPSI3OS44MDYsNjIuNjYgNTQuNDYsNTkuNTA2IDIzLjk4NSw2My43OTMgNDkuOTk5LDcxLjYgIgogICAgICAgICBpZD0icG9seWdvbjEzOCIgLz48cG9seWdvbgogICAgICAgICBmaWxsPSIjZDlhNzQxIgogICAgICAgICBwb2ludHM9IjUwLjAwMSw0MC44MzIgNzkuODA4LDQ0LjYyOSA3OS44MDgsNTUuMzMgNDkuOTk5LDU5LjA5MiAiCiAgICAgICAgIGlkPSJwb2x5Z29uMTQwIiAvPjwvZz48L2c+PC9zdmc+Cg=="
    camel.apache.org/provider: "Apache Software Foundation"
    camel.apache.org/kamelet.group: "AWS SQS"
    camel.apache.org/keda.type: "aws-sqs-queue"
  labels:
    camel.apache.org/kamelet.type: "source"
spec:
  definition:
    title: "AWS SQS Source"
    description: |-
      Receive data from AWS SQS.
    required:
      - queueNameOrArn
      - accessKey
      - secretKey
      - region
    type: object
    properties:
      queueNameOrArn:
        title: Queue Name
        description: The SQS Queue Name or ARN
        type: string
        x-descriptors:
        - urn:keda:metadata:queueURL
        - urn:keda:required
      deleteAfterRead:
        title: Auto-delete Messages
        description: Delete messages after consuming them
        type: boolean
        x-descriptors:
        - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
        default: true
      accessKey:
        title: Access Key
        description: The access key obtained from AWS
        type: string
        format: password
        x-descriptors:
        - urn:alm:descriptor:com.tectonic.ui:password
        - urn:camel:group:credentials
        - urn:keda:authentication:awsAccessKeyID
        - urn:keda:required
      secretKey:
        title: Secret Key
        description: The secret key obtained from AWS
        type: string
        format: password
        x-descriptors:
        - urn:alm:descriptor:com.tectonic.ui:password
        - urn:camel:group:credentials
        - urn:keda:authentication:awsSecretAccessKey
        - urn:keda:required
      region:
        title: AWS Region
        description: The AWS region to connect to
        type: string
        example: eu-west-1
        x-descriptors:
        - urn:keda:metadata:awsRegion
        - urn:keda:required
      autoCreateQueue:
        title: Autocreate Queue
        description: Setting the autocreation of the SQS queue. 
        type: boolean
        x-descriptors:
        - 'urn:alm:descriptor:com.tectonic.ui:checkbox'
        default: false
      amazonAWSHost:
        title: AWS Host
        description: The hostname of the Amazon AWS cloud. 
        type: string
      protocol:
        title: Protocol
        description: The underlying protocol used to communicate with SQS
        type: string
        example: http or https
        default: https
  dependencies:
    - "camel:aws2-sqs"
    - "camel:kamelet"
  flow:
    from:
      uri: "aws2-sqs:{{queueNameOrArn}}"
      parameters:
        autoCreateQueue: "{{autoCreateQueue}}"
        secretKey: "{{secretKey}}"
        accessKey: "{{accessKey}}"
        region: "{{region}}"
        deleteAfterRead: "{{deleteAfterRead}}"
        amazonAWSHost: "{{?amazonAWSHost}}"
        protocol: "{{?protocol}}"
      steps:
      - to: "kamelet:sink"

While developing I did a bit of refactoring in the serverside apply part, but later I didn't use it (but kept the refactoring).

Release Note

Added support for KEDA autoscalers with automatic configuration when using Kamelets

[astefanutti]

That's awesome!

Maybe a section could also be added to the existing scaling documentation.

[astefanutti]

I'm sure you've come to the conclusion it's necessary, but that smells for an outside reader. My understand from the Keda documentation is that the only requirement for custom resources to be Keda-scalable is to declare a scale sub-resource. Also, the path to the replicas spec is variable, and abstracted by the scale sub-resource. Does that mean Keda access the replicas spec directly, rather than via the scale endpoint?

[nicolaferraro]

I need to investigate.. yeah I called it "hack" because it smells :)

[nicolaferraro]

Turns out that the error is thrown by Kube: error 500 when KEDA tries to get the scale subresource: https://github.com/kedacore/keda/blob/9ac7a0a28e446335376ef8f5e82494cf05456ecb/controllers/keda/scaledobject_controller.go#L293.

I've already seen this in the past...

[astefanutti]

Would it be possible that the Integration hasn't been reconciled yet and the .status.replicas field hasn't been set when the Keda controller reconciles the ScaledObject? If that is the case, maybe we could have it set by default earlier, either when the Integration is created, or even possibly defaulted in the CRD?

[nicolaferraro]

No, the .status.replicas field is not considered. Here's a test I ran with a fully deployed klb (with Keda trait disabled).

$ kubectl get klb sqs-to-telegram -o "jsonpath={.status.replicas}"
1

$ curl --silent http://localhost:8088/apis/camel.apache.org/v1alpha1/namespaces/test/kameletbindings/sqs-to-telegram/scale | jq
{
  "kind": "Status",
  "apiVersion": "v1",
  "metadata": {},
  "status": "Failure",
  "message": "Internal error occurred: the spec replicas field \".spec.replicas\" does not exist",
  "reason": "InternalError",
  "details": {
    "causes": [
      {
        "message": "the spec replicas field \".spec.replicas\" does not exist"
      }
    ]
  },
  "code": 500
}

So the error comes from the API server.
Then, if I set the .spec.replicas everything goes well:

$ kubectl get klb sqs-to-telegram -o "jsonpath={.spec.replicas}"


$ kubectl scale klb sqs-to-telegram --replicas 1
kameletbinding.camel.apache.org/sqs-to-telegram scaled


$ kubectl get klb sqs-to-telegram -o "jsonpath={.spec.replicas}"
1

$ curl --silent http://localhost:8088/apis/camel.apache.org/v1alpha1/namespaces/test/kameletbindings/sqs-to-telegram/scale | jq
{
  "kind": "Scale",
  "apiVersion": "autoscaling/v1",
  "metadata": {
    "name": "sqs-to-telegram",
    "namespace": "test",
    "uid": "0b93916c-72f1-47ae-b2d1-74f9e3f7e5b6",
    "resourceVersion": "512151",
    "creationTimestamp": "2022-01-03T22:56:18Z"
  },
  "spec": {
    "replicas": 1
  },
  "status": {
    "replicas": 1,
    "selector": "camel.apache.org/integration=sqs-to-telegram"
  }
}

May it be my version of Kube... it's not latest, but it's a recent one:

$ kubectl version
Client Version: version.Info{Major:"1", Minor:"22", GitVersion:"v1.22.2", GitCommit:"8b5a19147530eaac9476b0ab82980b4088bbc1b2", GitTreeState:"clean", BuildDate:"2021-09-15T21:38:50Z", GoVersion:"go1.16.8", Compiler:"gc", Platform:"linux/amd64"}
Server Version: version.Info{Major:"1", Minor:"22", GitVersion:"v1.22.1", GitCommit:"632ed300f2c34f6d6d15ca4cef3d3c7073412212", GitTreeState:"clean", BuildDate:"2021-08-19T15:39:34Z", GoVersion:"go1.16.7", Compiler:"gc", Platform:"linux/amd64"}

[zroubalik]

Yeah this might be seen as an obvious approach, but unfortunately it doesn't work this way. There are certain states where we need to compare the actual number of replicas to multiple values in ScaledObject and act based on the result. It is not just assignig one value as a replica count.

I mean, everything could be done, right? :) So maybe if we refactor to code a lot it could be done, but I am not 100% sure about this. It would definitely make the code less logical.

[zroubalik]

Okay, I thought about it more and it cannot be done, there are certain KEDA features that won't work. For example there's an optional setting to reset replica count back to original number, once ScaledObject is deleted. If we don't know the number of replicas in the moment when we create the ScaledObject, we don't know the value that should be used to restore the replicas count, once ScaledObject is deleted.
And there definitely other things, that would be broken.

[astefanutti]

Right, sorry to drag you on this. I'm totally OK with KEDA mandating the .spec.replicas field to be set. I'm still trying to understand what's the best decision on this in the general case.

[astefanutti]

If we don't know the number of replicas in the moment when we create the ScaledObject, we don't know the value that should be used to restore the replicas count, once ScaledObject is deleted.

For that case, it was not set, so it could be unset. It may be a dubious parallel, but it's like a pointer in Golang, when it's needed to differentiate between the default value for the type, and not initialised.

[zroubalik]

Right, sorry to drag you on this. I'm totally OK with KEDA mandating the .spec.replicas field to be set. I'm still trying to understand what's the best decision on this in the general case.

Yeah, no worries :) It is always great to discuss stuff, it gives you another perspective.

For that case, it was not set, so it could be unset. It may be a dubious parallel, but it's like a pointer in Golang, when it's needed to differentiate between the default value for the type, and not initialised.

But behind those numbers (pointers), there are actuall replicas of a workload that needs to be set. I am not sure what unset means in this case :)

[astefanutti]

I forgot to mention that it seems you've also fixed #1616 in the process, maybe partially though?

[nicolaferraro]

I forgot to mention that it seems you've also fixed #1616 in the process, maybe partially though?

Yes, iirc those cases should be covered. I'll add some other tests to verify.

[nicolaferraro]

I think there's something wrong in the new applier. Maybe we can fix it in another PR, wdyt @astefanutti ?

[astefanutti]

@nicolaferraro sure, we can figure it out later. I was about to suggest you exactly that also for the point about the scale sub-resource detection.