feat: Maintain Scaler CRD

deploy/helm/commons-operator/Chart.yaml

@@ -54,3 +54,8 @@ annotations:
       name: openlineageconnections.lineage.stackable.tech
       displayName: OpenLineageConnection
       description: A reusable definition of a connection to an OpenLineage backend.
+    - kind: Scaler
+      version: v1alpha1
+      name: scalers.autoscaling.stackable.tech
+      displayName: Scaler
+      description: Controls the replica count of a Stackable component, integrating with the Kubernetes scale subresource to enable horizontal autoscaling.

extra/crds.yaml

@@ -803,3 +803,154 @@ spec:
         type: object
     served: true
     storage: true
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: scalers.autoscaling.stackable.tech
+spec:
+  group: autoscaling.stackable.tech
+  names:
+    kind: Scaler
+    plural: scalers
+    singular: scaler
+  scope: Namespaced
+  versions:
+  - name: v1alpha1
+    schema:
+      openAPIV3Schema:
+        description: Controls the replica count of a Stackable component, integrating with the Kubernetes scale subresource to enable horizontal autoscaling.
+        properties:
+          spec:
+            properties:
+              replicas:
+                description: |-
+                  Desired replica count.
+
+                  Written by the horizontal pod autoscaling mechanism via the /scale subresource.
+
+                  NOTE: This and other replica fields)use a [`u16`] instead of a [`i32`] used by
+                  [`k8s_openapi`] types to force a non-negative replica count. All [`u16`]s can be
+                  converted losslessly to [`i32`]s where needed.
+
+                  Upstream issues:
+
+                  - <https://github.com/kubernetes/kubernetes/issues/105533>
+                  - <https://github.com/Arnavion/k8s-openapi/issues/136>
+                format: uint16
+                maximum: 65535.0
+                minimum: 0.0
+                type: integer
+            required:
+            - replicas
+            type: object
+          status:
+            description: Status of a StackableScaler.
+            nullable: true
+            properties:
+              lastTransitionTime:
+                description: Timestamp indicating when the scaler state last transitioned.
+                format: date-time
+                type: string
+              replicas:
+                description: |-
+                  The current total number of replicas targeted by the managed StatefulSet.
+
+                  Exposed via the `/scale` subresource for horizontal pod autoscaling consumption.
+                format: uint16
+                maximum: 65535.0
+                minimum: 0.0
+                type: integer
+              selector:
+                description: Label selector string for HPA pod counting. Written at `.status.selector`.
+                nullable: true
+                type: string
+              state:
+                description: The current state of the scaler state machine.
+                oneOf:
+                - required:
+                  - idle
+                - required:
+                  - preScaling
+                - required:
+                  - scaling
+                - required:
+                  - postScaling
+                - required:
+                  - failed
+                properties:
+                  failed:
+                    description: |-
+                      A hook returned an error.
+
+                      The scaler stays here until the user applies the [`Annotation::autoscaling_retry`] annotation
+                      to trigger a reset to [`ScalerState::Idle`].
+                    properties:
+                      failedIn:
+                        description: Which stage produced the error.
+                        enum:
+                        - PreScaling
+                        - Scaling
+                        - PostScaling
+                        type: string
+                      reason:
+                        description: Human-readable error message from the hook.
+                        type: string
+                    required:
+                    - failedIn
+                    - reason
+                    type: object
+                  idle:
+                    description: No scaling operation is in progress.
+                    type: object
+                  postScaling:
+                    description: |-
+                      Running the `post_scale` hook (e.g. cluster rebalance).
+
+                      This stage additionally tracks the previous replica count to be able derive the direction
+                      of the scaling operation.
+                    properties:
+                      previousReplicas:
+                        format: uint16
+                        maximum: 65535.0
+                        minimum: 0.0
+                        type: integer
+                    required:
+                    - previousReplicas
+                    type: object
+                  preScaling:
+                    description: Running the `pre_scale` hook (e.g. data offload).
+                    type: object
+                  scaling:
+                    description: |-
+                      Waiting for the StatefulSet to converge to the new replica count.
+
+                      This stage additionally tracks the previous replica count to be able derive the direction
+                      of the scaling operation.
+                    properties:
+                      previousReplicas:
+                        format: uint16
+                        maximum: 65535.0
+                        minimum: 0.0
+                        type: integer
+                    required:
+                    - previousReplicas
+                    type: object
+                type: object
+            required:
+            - lastTransitionTime
+            - replicas
+            - state
+            type: object
+        required:
+        - spec
+        title: Scaler
+        type: object
+    served: true
+    storage: true
+    subresources:
+      scale:
+        labelSelectorPath: .status.selector
+        specReplicasPath: .spec.replicas
+        statusReplicasPath: .status.replicas
+      status: {}

rust/operator-binary/src/main.rs

@@ -13,6 +13,7 @@ use stackable_operator::{
         authentication::core::{self, AuthenticationClass, AuthenticationClassVersion},
         openlineage::{self, OpenLineageConnection, OpenLineageConnectionVersion},
         s3::{self, S3Bucket, S3BucketVersion, S3Connection, S3ConnectionVersion},
+        scaler::{self, Scaler, ScalerVersion},
     },
     eos::EndOfSupportChecker,
     kube::CustomResourceExt,
@@ -67,6 +68,8 @@ async fn main() -> anyhow::Result<()> {
                 .print_yaml_schema(built_info::PKG_VERSION, &SerializeOptions::default())?;
             OpenLineageConnection::merged_crd(OpenLineageConnectionVersion::V1Alpha1)?
                 .print_yaml_schema(built_info::PKG_VERSION, &SerializeOptions::default())?;
+            Scaler::merged_crd(ScalerVersion::V1Alpha1)?
+                .print_yaml_schema(built_info::PKG_VERSION, &SerializeOptions::default())?;
         }
         Command::Run(CommonsOperatorRunArguments {
             common:
@@ -129,6 +132,11 @@ async fn main() -> anyhow::Result<()> {
                 "CRD {crd} established",
                 crd = openlineage::v1alpha1::OpenLineageConnection::crd_name()
             ));
+            let scaler_crd_check = readiness_checks.register(format!(
+                "CRD {crd} established",
+                crd = scaler::v1alpha1::Scaler::crd_name()
+            ));
+
             let webhook_server = create_webhook_server(
                 ctx.clone(),
                 &operator_environment,
@@ -156,20 +164,29 @@ async fn main() -> anyhow::Result<()> {
                 .run(sigterm_watcher.handle())
                 .map_err(|err| anyhow!(err).context("failed to run webhook"));
 
+            // NOTE (@Techassi): This is far from optimal, I don't really like it. I will come up
+            // with something better down the line.
             let crds_established = async {
                 signal::crd_established(&client, core::v1alpha1::AuthenticationClass::crd_name())
                     .await?;
                 authentication_class_crd_check.mark_passed();
+
                 signal::crd_established(&client, s3::v1alpha1::S3Connection::crd_name()).await?;
                 s3_connection_crd_check.mark_passed();
+
                 signal::crd_established(&client, s3::v1alpha1::S3Bucket::crd_name()).await?;
                 s3_bucket_crd_check.mark_passed();
+
                 signal::crd_established(
                     &client,
                     openlineage::v1alpha1::OpenLineageConnection::crd_name(),
                 )
                 .await?;
                 openlineage_connection_crd_check.mark_passed();
+
+                signal::crd_established(&client, scaler::v1alpha1::Scaler::crd_name()).await?;
+                scaler_crd_check.mark_passed();
+
                 anyhow::Ok(())
             };
 

rust/operator-binary/src/webhooks/conversion.rs

@@ -3,6 +3,7 @@ use stackable_operator::{
         authentication::core::{AuthenticationClass, AuthenticationClassVersion},
         openlineage::{OpenLineageConnection, OpenLineageConnectionVersion},
         s3::{S3Bucket, S3BucketVersion, S3Connection, S3ConnectionVersion},
+        scaler::{Scaler, ScalerVersion},
     },
     kube::Client,
     webhook::webhooks::{ConversionWebhook, ConversionWebhookOptions, Webhook},
@@ -13,21 +14,30 @@ use crate::FIELD_MANAGER;
 pub fn create_webhook(disable_crd_maintenance: bool, client: Client) -> Box<impl Webhook> {
     let crds_and_handlers = vec![
         (
-            AuthenticationClass::merged_crd(AuthenticationClassVersion::V1Alpha1).unwrap(),
+            AuthenticationClass::merged_crd(AuthenticationClassVersion::V1Alpha1)
+                .expect("the AuthenticationClass CRD versions must be mergeable"),
             AuthenticationClass::try_convert as fn(_) -> _,
         ),
         (
-            S3Connection::merged_crd(S3ConnectionVersion::V1Alpha1).unwrap(),
+            S3Connection::merged_crd(S3ConnectionVersion::V1Alpha1)
+                .expect("the S3Connection CRD versions must be mergeable"),
             S3Connection::try_convert as fn(_) -> _,
         ),
         (
-            S3Bucket::merged_crd(S3BucketVersion::V1Alpha1).unwrap(),
+            S3Bucket::merged_crd(S3BucketVersion::V1Alpha1)
+                .expect("the S3Bucket CRD versions must be mergeable"),
             S3Bucket::try_convert as fn(_) -> _,
         ),
         (
-            OpenLineageConnection::merged_crd(OpenLineageConnectionVersion::V1Alpha1).unwrap(),
+            OpenLineageConnection::merged_crd(OpenLineageConnectionVersion::V1Alpha1)
+                .expect("the OpenLineageConnection CRD versions must be mergeable"),
             OpenLineageConnection::try_convert as fn(_) -> _,
         ),
+        (
+            Scaler::merged_crd(ScalerVersion::V1Alpha1)
+                .expect("the Scaler CRD versions must be mergeable"),
+            Scaler::try_convert as fn(_) -> _,
+        ),
     ];
 
     let conversion_webhook_options = ConversionWebhookOptions {