feat: replace in-memory scheduling with Kueue

helm/java-generator-chart/templates/deployment.yaml

@@ -57,17 +57,11 @@ spec:
             - name: QUARKUS_OTEL_RESOURCE_ATTRIBUTES
               value: "service.name=sbomer-java-generator"
 
-            # Generator Logic
-            - name: SBOMER_GENERATOR_MAX_CONCURRENT
-              value: {{ .Values.config.maxConcurrent | quote }}
-            - name: SBOMER_GENERATOR_OOM_RETRIES
-              value: {{ .Values.config.oomRetries | quote }}
-            - name: SBOMER_GENERATOR_MEMORY_MULTIPLIER
-              value: {{ .Values.config.memoryMultiplier | quote }}
-            - name: SBOMER_GENERATOR_MAVEN_DEFAULT_MEMORY
-              value: {{ .Values.config.defaultMemory.maven | quote }}
-            - name: SBOMER_GENERATOR_DOMINO_DEFAULT_MEMORY
-              value: {{ .Values.config.defaultMemory.domino | quote }}
+            # Kueue Configuration
+            {{- if .Values.kueue.includeKueue }}
+            - name: SBOMER_GENERATOR_KUEUE_QUEUE_NAME
+              value: {{ .Values.kueue.localQueue.name | quote }}
+            {{- end }}
 
 
             # Tekton Task Identity

helm/java-generator-chart/templates/kueue/clusterqueue.yaml

@@ -0,0 +1,25 @@
+{{- if .Values.kueue.includeKueue -}}
+apiVersion: kueue.x-k8s.io/v1beta1
+kind: ClusterQueue
+metadata:
+  name: {{ .Values.kueue.clusterQueue.name }}
+  labels:
+    {{- include "java-generator-chart.labels" . | nindent 4 }}
+spec:
+  namespaceSelector: {}  # Allow all namespaces
+  resourceGroups:
+  - coveredResources: ["cpu", "memory"]
+    flavors:
+    - name: {{ .Values.kueue.resourceFlavor.name }}
+      resources:
+      # CPU quota for all concurrent tasks
+      # Note: Kueue uses resource requests (not limits) for admission control
+      - name: "cpu"
+        nominalQuota: {{ .Values.kueue.clusterQueue.quotas.cpu | quote }}
+      # Memory quota for all concurrent tasks
+      # Note: Kueue uses resource requests (not limits) for admission control
+      - name: "memory"
+        nominalQuota: {{ .Values.kueue.clusterQueue.quotas.memory | quote }}
+  # Admission control configuration
+  queueingStrategy: {{ .Values.kueue.clusterQueue.queueingStrategy }}
+{{- end }}

helm/java-generator-chart/templates/kueue/localqueue.yaml

@@ -0,0 +1,11 @@
+{{- if .Values.kueue.includeKueue -}}
+apiVersion: kueue.x-k8s.io/v1beta1
+kind: LocalQueue
+metadata:
+  name: {{ .Values.kueue.localQueue.name }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    {{- include "java-generator-chart.labels" . | nindent 4 }}
+spec:
+  clusterQueue: {{ .Values.kueue.clusterQueue.name }}
+{{- end }}

helm/java-generator-chart/templates/kueue/resourceflavor.yaml

@@ -0,0 +1,9 @@
+{{- if .Values.kueue.includeKueue -}}
+apiVersion: kueue.x-k8s.io/v1beta1
+kind: ResourceFlavor
+metadata:
+  name: {{ .Values.kueue.resourceFlavor.name }}
+  labels:
+    {{- include "java-generator-chart.labels" . | nindent 4 }}
+spec: {}
+{{- end }}

helm/java-generator-chart/templates/tekton/cdx-maven-plugin-generation-task.yaml

@@ -40,11 +40,11 @@ spec:
       imagePullPolicy: {{ .Values.task.agent.pullPolicy }}
       resources:
         requests:
-          cpu: 500m
-          memory: 1000Mi
+          cpu: {{ .Values.task.resources.cdxMavenPlugin.requests.cpu }}
+          memory: {{ .Values.task.resources.cdxMavenPlugin.requests.memory }}
         limits:
-          cpu: 800m
-          memory: 2000Mi
+          cpu: {{ .Values.task.resources.cdxMavenPlugin.limits.cpu }}
+          memory: {{ .Values.task.resources.cdxMavenPlugin.limits.memory }}
       env:
         - name: TRACEPARENT
           value: "$(params.trace-parent)"

helm/java-generator-chart/templates/tekton/domino-generation-task.yaml

@@ -34,11 +34,11 @@ spec:
       imagePullPolicy: {{ .Values.task.agent.pullPolicy }}
       resources:
         requests:
-          cpu: 500m
-          memory: 1000Mi
+          cpu: {{ .Values.task.resources.domino.requests.cpu }}
+          memory: {{ .Values.task.resources.domino.requests.memory }}
         limits:
-          cpu: 800m
-          memory: 2000Mi
+          cpu: {{ .Values.task.resources.domino.limits.cpu }}
+          memory: {{ .Values.task.resources.domino.limits.memory }}
       env:
         - name: TRACEPARENT
           value: "$(params.trace-parent)"

helm/java-generator-chart/values.yaml

@@ -43,24 +43,20 @@ rbac:
     - apiGroups: ["tekton.dev"]
       resources: ["taskruns", "pipelineruns", "tasks"]
       verbs: ["create", "get", "list", "watch", "update", "patch", "delete", "deletecollection"]
+    # Include Kueue for queue management
+    - apiGroups: ["kueue.x-k8s.io"]
+      resources: ["localqueues", "clusterqueues", "resourceflavors", "workloads"]
+      verbs: ["get", "list", "watch"]
 
 # --- APPLICATION CONFIG ---
 config:
   logLevel: INFO
   otel:
     protocol: grpc
     endpoint: "" # placeholder
-  maxConcurrent: 20
-  oomRetries: 3
   # Default internal cluster URL for storage
   storageUrl: "http://manifest-storage-service:8085"
 
-  # OOM Memory scaling configurations
-  memoryMultiplier: "1.5"
-  defaultMemory:
-    maven: "1Gi"
-    domino: "2Gi"
-
   kafka:
     bootstrapServers: "kafka:9092"
     schemaRegistryUrl: "http://schema-registry:8080/apis/registry/v2"
@@ -69,10 +65,38 @@ config:
     # Leave empty to let Quarkus auto-discover the K8s API (Standard K8s behavior)
     # The Super-Chart can override this if needed.
     kubernetesClientMasterUrl: ""
-    # Default to false (Secure). 
+    # Default to false (Secure).
     # Can override this to "true" for local dev/minikube.
     trustCerts: "false"
 
+# --- KUEUE CONFIG ---
+# Kueue is mandatory for queue management
+kueue:
+  includeKueue: true
+  # LocalQueue configuration (namespace-scoped)
+  localQueue:
+    name: "java-generator-queue"
+
+  # ClusterQueue configuration (cluster-scoped)
+  clusterQueue:
+    name: "java-generator-cluster-queue"
+    queueingStrategy: BestEffortFIFO
+    quotas:
+      # Total CPU quota across all task types
+      # Calculated as: maxConcurrent * highest CPU request among task types
+      # Example: 20 pods * 500m (cdxMavenPlugin/domino request) = 10000m
+      cpu: "10000m"
+      # Total memory quota across all task types
+      # Calculated as: maxConcurrent * highest memory request among task types
+      # Example: 20 pods * 1000Mi = 20Gi
+      memory: "20Gi"
+      # Maximum number of concurrent TaskRuns (across all types)
+      pods: 20
+
+  # ResourceFlavor configuration
+  resourceFlavor:
+    name: "default-flavor"
+
 # --- TEKTON TASK CONFIG ---
 task:
   name:
@@ -86,6 +110,26 @@ task:
     # If tag is empty, the template will default to .Chart.AppVersion (The Git SHA)
     tag: ""
     pullPolicy: IfNotPresent
+  # Resource limits for TaskRun steps
+  # These values control CPU and memory allocation for generation tasks
+  # Adjust based on your workload requirements and cluster capacity
+  resources:
+    # CycloneDX Maven Plugin task resources
+    cdxMavenPlugin:
+      requests:
+        cpu: 500m
+        memory: 1000Mi
+      limits:
+        cpu: 800m
+        memory: 2000Mi
+    # Domino task resources (typically needs more memory)
+    domino:
+      requests:
+        cpu: 500m
+        memory: 1000Mi
+      limits:
+        cpu: 800m
+        memory: 2000Mi
 
   maven:
     configMapName: java-generator-maven-settings

src/main/java/org/jboss/sbomer/java/generator/adapter/in/TaskReconciler.java

@@ -8,7 +8,6 @@
 import org.jboss.sbomer.java.generator.core.port.api.GenerationOrchestrator;
 import org.jboss.sbomer.java.generator.core.port.spi.FailureNotifier;
 import org.jboss.sbomer.java.generator.core.utility.FailureUtility;
-import org.jboss.sbomer.java.generator.core.utility.TraceUtility;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -20,8 +19,7 @@
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.StatusCode;
-import io.opentelemetry.api.trace.Tracer;
-import io.opentelemetry.context.Scope;
+import io.opentelemetry.instrumentation.annotations.WithSpan;
 import jakarta.inject.Inject;
 import lombok.extern.slf4j.Slf4j;
 
@@ -38,39 +36,28 @@ public class TaskReconciler implements Reconciler<TaskRun> {
     @Inject
     ObjectMapper objectMapper;
 
-    @Inject
-    Tracer tracer;
-
     private static final String REASON_OOM_KILLED = "OOMKilled";
 
     private static final String GENERATION_ID_LABEL = "sbomer.jboss.org/generation-id";
     private static final String RESULT_NAME_SBOM_URL = "sbom-url";
-    private static final String TRACEPARENT_ANNOTATION = "sbomer.jboss.org/traceparent";
 
+    @WithSpan
     @Override
     public UpdateControl<TaskRun> reconcile(TaskRun taskRun, Context<TaskRun> context) {
         String taskName = taskRun.getMetadata().getName();
         String generationId = taskRun.getMetadata().getLabels().get(GENERATION_ID_LABEL);
 
-        // Read trace context from TaskRun annotations
-        Map<String, String> annotations = taskRun.getMetadata().getAnnotations();
-        String traceParent = annotations != null ? annotations.get(TRACEPARENT_ANNOTATION) : null;
-
-        // Extract status for span attributes and logging
-        String taskRunStatus = getConditionStatus(taskRun);
+        // Extract status for logging and tracing
         String taskRunReason = getConditionReason(taskRun);
 
-        // Create a child span under the original trace from the Kafka consumer
-        Span span = TraceUtility.childSpanBuilder(tracer, "TaskReconciler.reconcile", traceParent, generationId)
-                .setAttribute("taskrun.name", taskName != null ? taskName : "unknown")
-                .setAttribute("taskrun.status", taskRunStatus)
-                .setAttribute("taskrun.reason", taskRunReason)
-                .startSpan();
-        try (Scope ignored = span.makeCurrent()) {
-            return doReconcile(taskRun, taskName, generationId, taskRunReason);
-        } finally {
-            span.end();
+        // Add span attributes for observability
+        Span.current().setAttribute("taskrun.name", taskName != null ? taskName : "unknown");
+        Span.current().setAttribute("taskrun.reason", taskRunReason);
+        if (generationId != null) {
+            Span.current().setAttribute("generation.id", generationId);
         }
+
+        return doReconcile(taskRun, taskName, generationId, taskRunReason);
     }
 
     private UpdateControl<TaskRun> doReconcile(TaskRun taskRun, String taskName, String generationId, String statusReason) {

src/main/java/org/jboss/sbomer/java/generator/adapter/out/TektonGenerationExecutor.java

@@ -10,7 +10,6 @@
 
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.tekton.v1beta1.TaskRun;
-import io.opentelemetry.instrumentation.annotations.SpanAttribute;
 import io.opentelemetry.instrumentation.annotations.WithSpan;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -66,54 +65,8 @@ public void scheduleGeneration(GenerationTask generationTask) {
         }
 
         // Execute against the cluster
+        // Kueue manages the TaskRun lifecycle, including cleanup after completion
         kubernetesClient.resources(TaskRun.class).inNamespace(namespace).resource(taskRun).create();
     }
 
-    @WithSpan
-    @Override
-    public void abortGeneration(@SpanAttribute("generation.id") String generationId) {
-        log.info("Aborting generation: {}", generationId);
-        kubernetesClient.resources(TaskRun.class)
-                .inNamespace(namespace)
-                .withLabel(LABEL_GENERATION_ID, generationId)
-                .delete();
-    }
-
-    // In this specific implementation, basically same logic as abortGeneration
-    @WithSpan
-    @Override
-    public void cleanupGeneration(@SpanAttribute("generation.id") String generationId) {
-         log.info("Cleaning up generation: {}", generationId);
-         kubernetesClient.resources(TaskRun.class)
-                 .inNamespace(namespace)
-                 .withLabel(LABEL_GENERATION_ID, generationId)
-                 .delete();
-    }
-
-    @Override
-    public int countActiveExecutions() {
-        // Count TaskRuns for THIS generator that are NOT finished.
-        // This is the input for the Throttling logic.
-        return (int) kubernetesClient.resources(TaskRun.class).inNamespace(namespace)
-                .withLabel(LABEL_GENERATOR_TYPE, LABEL_GENERATOR_VALUE)
-                .list()
-                .getItems()
-                .stream()
-                .filter(tr -> !isFinished(tr))
-                .count();
-    }
-
-    /**
-     * Helper to check Tekton Status Conditions
-     */
-    private boolean isFinished(TaskRun taskRun) {
-        if (taskRun.getStatus() == null || taskRun.getStatus().getConditions() == null) {
-            return false; // No status means it's initializing/running
-        }
-
-        // Check for "Succeeded" condition with Status "True" or "False" (False means failed, but it is still 'finished')
-        return taskRun.getStatus().getConditions().stream()
-                .anyMatch(c -> "Succeeded".equals(c.getType()) &&
-                        ("True".equals(c.getStatus()) || "False".equals(c.getStatus())));
-    }
 }

src/main/java/org/jboss/sbomer/java/generator/core/domain/model/GenerationTask.java

@@ -5,19 +5,14 @@
 import org.jboss.sbomer.events.common.GenerationRequestSpec;
 
 /**
- * Internal domain model representing a unit of work waiting in the queue.
+ * Internal domain model representing a generation task.
  * It decouples the internal scheduling logic from the external Kafka event structure.
  */
 public record GenerationTask(
     String generationId,
     GenerationRequestSpec spec,
-    int retryCount, // NOT max retries, the number of it retries it's currently on
-    String memoryOverride, // i.e. 2Gi
     Map<String, String> generatorOptions,
     Map<String, String> handlerProvidedOptions,
     String traceParent // W3C traceparent header (00-<traceId>-<spanId>-<traceFlags>)
 ) {
-    public GenerationTask(String generationId, GenerationRequestSpec spec, Map<String, String> generatorOptions, Map<String, String> handlerProvidedOptions, String traceParent) {
-        this(generationId, spec, 0, null, generatorOptions, handlerProvidedOptions, traceParent);
-    }
 }

src/main/java/org/jboss/sbomer/java/generator/core/port/spi/GenerationExecutor.java

@@ -15,39 +15,10 @@ public interface GenerationExecutor {
      * Schedules the generation payload for execution.
      * <p>
      * In a Kubernetes/Tekton implementation, this creates the TaskRun resource.
+     * Kueue manages the TaskRun lifecycle, including cleanup after completion.
      * </p>
      *
      * @param generationTask The object carrying information about a generation task
      */
     void scheduleGeneration(GenerationTask generationTask);
-
-    /**
-     * Aborts resources associated with a specific generation.
-     * <p>
-     * Used for manual cancellation.
-     * </p>
-     *
-     * @param generationId The unique ID to identify the resources.
-     */
-    void abortGeneration(String generationId);
-
-    /**
-     * Cleans up resources associated with a specific generation.
-     * <p>
-     * Used for cleaning up the environment after the generation has ended.
-     * </p>
-     *
-     * @param generationId The unique ID to identify the resources.
-     */
-    void cleanupGeneration(String generationId);
-
-    /**
-     * Returns the number of currently active/running executions managed by this generator.
-     * <p>
-     * This is critical for the Core Domain's "Throttling" logic.
-     * </p>
-     *
-     * @return count of active jobs.
-     */
-    int countActiveExecutions();
 }