[WIP] Optimize K8s Operator LLAP scaling

packaging/src/kubernetes/helm/hive-operator/crds/hiveclusters.hive.apache.org-v1.yml

@@ -129,6 +129,12 @@ spec:
                           replicas before scale-down occurs. Also acts as the cooldown
                           between consecutive scale-downs.
                         type: integer
+                      scaleDownThreshold:
+                        default: 20
+                        description: "Threshold that triggers scale-down (LLAP only):\
+                          \ average daemon utilization percentage (0-100) below which\
+                          \ scale-down triggers."
+                        type: integer
                       scaleUpStabilizationSeconds:
                         default: 60
                         description: Stabilization window in seconds for scale-up
@@ -138,8 +144,8 @@ spec:
                       scaleUpThreshold:
                         default: 80
                         description: "Threshold that triggers scale-up (component-specific:\
-                          \ sessions per pod for HS2, request rate for HMS, busy slots\
-                          \ per daemon for LLAP). Not used by TezAM (demand-based:\
+                          \ sessions per pod for HS2, request rate for HMS, load percentage\
+                          \ pending in TezAM for LLAP). Not used by TezAM (demand-based:\
                           \ 1 TezAM per session)."
                         type: integer
                     type: object
@@ -312,6 +318,12 @@ spec:
                             fewer replicas before scale-down occurs. Also acts as
                             the cooldown between consecutive scale-downs.
                           type: integer
+                        scaleDownThreshold:
+                          default: 20
+                          description: "Threshold that triggers scale-down (LLAP only):\
+                            \ average daemon utilization percentage (0-100) below\
+                            \ which scale-down triggers."
+                          type: integer
                         scaleUpStabilizationSeconds:
                           default: 60
                           description: Stabilization window in seconds for scale-up
@@ -321,9 +333,9 @@ spec:
                         scaleUpThreshold:
                           default: 80
                           description: "Threshold that triggers scale-up (component-specific:\
-                            \ sessions per pod for HS2, request rate for HMS, busy\
-                            \ slots per daemon for LLAP). Not used by TezAM (demand-based:\
-                            \ 1 TezAM per session)."
+                            \ sessions per pod for HS2, request rate for HMS, load\
+                            \ percentage pending in TezAM for LLAP). Not used by TezAM\
+                            \ (demand-based: 1 TezAM per session)."
                           type: integer
                       type: object
                     configOverrides:
@@ -467,6 +479,12 @@ spec:
                                 fewer replicas before scale-down occurs. Also acts
                                 as the cooldown between consecutive scale-downs.
                               type: integer
+                            scaleDownThreshold:
+                              default: 20
+                              description: "Threshold that triggers scale-down (LLAP\
+                                \ only): average daemon utilization percentage (0-100)\
+                                \ below which scale-down triggers."
+                              type: integer
                             scaleUpStabilizationSeconds:
                               default: 60
                               description: Stabilization window in seconds for scale-up
@@ -477,8 +495,8 @@ spec:
                               default: 80
                               description: "Threshold that triggers scale-up (component-specific:\
                                 \ sessions per pod for HS2, request rate for HMS,\
-                                \ busy slots per daemon for LLAP). Not used by TezAM\
-                                \ (demand-based: 1 TezAM per session)."
+                                \ load percentage pending in TezAM for LLAP). Not\
+                                \ used by TezAM (demand-based: 1 TezAM per session)."
                               type: integer
                           type: object
                         replicas:
@@ -545,6 +563,12 @@ spec:
                           replicas before scale-down occurs. Also acts as the cooldown
                           between consecutive scale-downs.
                         type: integer
+                      scaleDownThreshold:
+                        default: 20
+                        description: "Threshold that triggers scale-down (LLAP only):\
+                          \ average daemon utilization percentage (0-100) below which\
+                          \ scale-down triggers."
+                        type: integer
                       scaleUpStabilizationSeconds:
                         default: 60
                         description: Stabilization window in seconds for scale-up
@@ -554,8 +578,8 @@ spec:
                       scaleUpThreshold:
                         default: 80
                         description: "Threshold that triggers scale-up (component-specific:\
-                          \ sessions per pod for HS2, request rate for HMS, busy slots\
-                          \ per daemon for LLAP). Not used by TezAM (demand-based:\
+                          \ sessions per pod for HS2, request rate for HMS, load percentage\
+                          \ pending in TezAM for LLAP). Not used by TezAM (demand-based:\
                           \ 1 TezAM per session)."
                         type: integer
                     type: object
@@ -753,6 +777,12 @@ spec:
                           replicas before scale-down occurs. Also acts as the cooldown
                           between consecutive scale-downs.
                         type: integer
+                      scaleDownThreshold:
+                        default: 20
+                        description: "Threshold that triggers scale-down (LLAP only):\
+                          \ average daemon utilization percentage (0-100) below which\
+                          \ scale-down triggers."
+                        type: integer
                       scaleUpStabilizationSeconds:
                         default: 60
                         description: Stabilization window in seconds for scale-up
@@ -762,8 +792,8 @@ spec:
                       scaleUpThreshold:
                         default: 80
                         description: "Threshold that triggers scale-up (component-specific:\
-                          \ sessions per pod for HS2, request rate for HMS, busy slots\
-                          \ per daemon for LLAP). Not used by TezAM (demand-based:\
+                          \ sessions per pod for HS2, request rate for HMS, load percentage\
+                          \ pending in TezAM for LLAP). Not used by TezAM (demand-based:\
                           \ 1 TezAM per session)."
                         type: integer
                     type: object

packaging/src/kubernetes/helm/hive-operator/templates/hivecluster.yaml

@@ -157,6 +157,7 @@ spec:
       enabled: true
       minReplicas: {{ .autoscaling.minReplicas }}
       scaleUpThreshold: {{ .autoscaling.scaleUpThreshold }}
+      scaleDownThreshold: {{ .autoscaling.scaleDownThreshold }}
       scaleUpStabilizationSeconds: {{ .autoscaling.scaleUpStabilizationSeconds }}
       scaleDownStabilizationSeconds: {{ .autoscaling.scaleDownStabilizationSeconds }}
       gracePeriodSeconds: {{ .autoscaling.gracePeriodSeconds }}

packaging/src/kubernetes/helm/hive-operator/values.yaml

@@ -200,7 +200,8 @@ cluster:
     autoscaling:
       enabled: false
       minReplicas: 0
-      scaleUpThreshold: 10
+      scaleUpThreshold: 70
+      scaleDownThreshold: 20
       scaleUpStabilizationSeconds: 60
       scaleDownStabilizationSeconds: 900
       gracePeriodSeconds: 600

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/autoscaling/ComponentAutoscaler.java

@@ -94,8 +94,13 @@ public EvaluationResult evaluate(List<PodMetrics> metrics, AutoscalingSpec spec,
 
     int target;
     if (clamped > currentReplicas) {
-      // Scale up: use stabilized max (highest recommendation in window — don't under-scale)
-      target = scaleUpWindow.stabilizedMax();
+      if (component.startsWith(ConfigUtils.COMPONENT_LLAP + "-")) {
+        // HS2 sessions activation gate scales up the LLAP pods to atleast 1
+        // in presence of sessions. Avoid stabilizedMin in this start-up case.
+        target = currentReplicas == 0 ? clamped : scaleUpWindow.stabilizedMin();
+      } else {
+        target = scaleUpWindow.stabilizedMax();
+      }
     } else if (clamped < currentReplicas) {
       // Scale down: use stabilized max (highest/most conservative recommendation in window —
       // prevents premature scale-down, matches HPA selectPolicy: Max behavior).

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/autoscaling/HiveClusterAutoscaler.java

@@ -321,6 +321,21 @@ public List<PodMetrics> getHs2MetricsFromCache(HiveCluster cluster) {
     return metricsCache.getOrEmpty(key, maxStale);
   }
 
+  /**
+   * Returns cached TezAM metrics for the given LLAP cluster (used by LlapScalingStrategy).
+   */
+  public List<PodMetrics> getTezAmMetricsFromCache(HiveCluster cluster, String llapName) {
+    String namespace = cluster.getMetadata().getNamespace();
+    String clusterName = cluster.getMetadata().getName();
+    int maxStale = cluster.getSpec().llapClusters().stream()
+        .filter(l -> llapName.equals(l.name()))
+        .findFirst()
+        .map(l -> l.tezAm().autoscaling().metricsScrapeIntervalSeconds() * 3)
+        .orElse(30);
+    return metricsCache.getOrEmpty(
+        cacheKey(namespace, clusterName, ConfigUtils.tezAmComponentKey(llapName)), maxStale);
+  }
+
   private void evaluateComponent(HiveCluster cluster, KubernetesClient client,
       String namespace, String clusterName, String component,
       AutoscalingSpec autoscaling, int maxReplicas,

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/autoscaling/LlapScalingStrategy.java

@@ -28,11 +28,12 @@
 
 /**
  * Scaling strategy for LLAP daemons.
- * Formula: avg(QueuedRequests + Configured - Available) across all pods.
- * This represents average "busy slots + queued" per daemon.
- * desired = ceil(avg_busy / scaleUpThreshold)
+ * For Scale-Up: Pending Load across all TezAM pods should be above the threshold.
+ * For Scale-Down: Running Load across all LLAP pods should be below the threshold.
+ * desired = ceil(totalClusterLoad / capacityPerDaemon)
  * <p>
- * Activation gate: only scale if HS2 has open sessions (prevents zombie scaling).
+ * Activation gate: only scale if HS2 has open sessions, TezAM metrics are present &
+ * cluster has load (prevents zombie scaling).
  */
 public class LlapScalingStrategy implements ScalingStrategy {
 
@@ -41,7 +42,9 @@ public class LlapScalingStrategy implements ScalingStrategy {
   static final String METRIC_QUEUED = "hadoop_llapdaemon_executornumqueuedrequests";
   static final String METRIC_CONFIGURED = "hadoop_llapdaemon_executornumexecutorsconfigured";
   static final String METRIC_AVAILABLE = "hadoop_llapdaemon_executornumexecutorsavailable";
+  static final String METRIC_MAX_FREE_SLOTS_CONFIGURED = "hadoop_llapdaemon_executormaxfreeslotsconfigured";
   static final String METRIC_LLAP_TARGET_PREFIX = "hs2_llap_target_sessions_";
+  static final String METRIC_TEZ_PENDING_TASKS = "tez_am_pending_tasks";
 
   private final HiveClusterAutoscaler orchestrator;
   private final HiveCluster cluster;
@@ -83,36 +86,59 @@ public int computeDesiredReplicas(List<PodMetrics> podMetrics,
       return minReplica;
     }
 
-    // Compute average busy slots across all LLAP pods
-    double totalBusy = 0;
-    int podCount = 0;
+    List<PodMetrics> tezAmMetrics = orchestrator.getTezAmMetricsFromCache(cluster, llapName);
+    double totalPending = 0;
+    for (PodMetrics pm : tezAmMetrics) {
+      totalPending += pm.metrics().getOrDefault(METRIC_TEZ_PENDING_TASKS, 0.0);
+    }
+
+    double totalLLAPCapacity = 0;
+    double totalLLAPLoad = 0;
     for (PodMetrics pm : podMetrics) {
       double queued = pm.metrics().getOrDefault(METRIC_QUEUED, 0.0);
       double configured = pm.metrics().getOrDefault(METRIC_CONFIGURED, 0.0);
       double available = pm.metrics().getOrDefault(METRIC_AVAILABLE, 0.0);
-      double busy = queued + configured - available;
-      totalBusy += busy;
-      podCount++;
+      double capacity = pm.metrics().getOrDefault(METRIC_MAX_FREE_SLOTS_CONFIGURED, 0.0);
+      if (capacity <= 0) {
+        capacity = configured;
+      }
+      totalLLAPCapacity += capacity;
+      totalLLAPLoad += queued + configured - available;
     }
 
-    double avgBusy = totalBusy / podCount;
-    lastMetric = (int) Math.round(avgBusy);
-
-    if (avgBusy <= 0) {
-      // HS2 has sessions (passed activation gate above) but executors are idle between queries.
-      // Keep at least 1 daemon to avoid flapping: scaling to 0 here would cause immediate
-      // scale-back-up on the next evaluation when the empty-pod path triggers.
+    // HS2 has sessions (passed activation gate above) but either
+    // 1. there is no tezAM running, so LLAP running any work is zombie if any.
+    // 2. there are tezAMs running, but no tasks running or pending.
+    if(tezAmMetrics.isEmpty() || (totalPending + totalLLAPLoad) == 0) {
       return Math.max(1, autoscaling.minReplicas());
     }
 
-    if (LOG.isDebugEnabled()) {
-      LOG.debug("[llap] avgBusy={}, threshold={}", String.format("%.2f", avgBusy),
-          autoscaling.scaleUpThreshold());
+    double capacityPerDaemon = Math.max(1.0, totalLLAPCapacity / podMetrics.size());
+    double avgLLAPLoadPercent = totalLLAPCapacity > 0 ? (totalLLAPLoad / totalLLAPCapacity) * 100.00 : 0.0;
+
+    double totalClusterLoad = totalPending + totalLLAPLoad;
+    double pendingLoadPercent = (totalPending / totalClusterLoad) * 100.0;
+
+    int scaleUpThreshold = autoscaling.scaleUpThreshold();
+    int scaleDownThreshold = autoscaling.scaleDownThreshold();
+
+    lastMetric = (int) totalClusterLoad;
+
+    // Scale-up: pending load share of total load exceeds threshold
+    // Scale-down: no pending work AND daemon load below threshold
+    if (pendingLoadPercent >= scaleUpThreshold || (totalPending == 0 && avgLLAPLoadPercent <= scaleDownThreshold)) {
+      int desired = (int) Math.ceil(totalClusterLoad / capacityPerDaemon);
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("[llap-{}] totalClusterLoad={}, capacityPerDaemon={}, pendingLoadPercent={}, avgLLAPLoadPercent={}",
+            llapName, totalClusterLoad, capacityPerDaemon, String.format("%.2f", pendingLoadPercent),
+            String.format("%.2f", avgLLAPLoadPercent));
+      }
+      return desired;
     }
 
-    int threshold = Math.max(1, autoscaling.scaleUpThreshold());
-    int desired = (int) Math.ceil(avgBusy / threshold);
-    return Math.max(desired, autoscaling.minReplicas());
+    // Work is in flight (pending share below threshold) or daemons are moderately loaded.
+    // Return current pod count so the stabilization window keeps the replica count stable.
+    return podMetrics.size();
   }
 
   @Override

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/dependent/HiveDependentResource.java

@@ -854,14 +854,21 @@ private static String buildJmxExporterConfig(String component) {
       sb.append("- pattern: '").append(llapBean).append("ExecutorNumExecutors'\n");
       sb.append("  name: hadoop_llapdaemon_executornumexecutors\n");
       sb.append("  type: GAUGE\n");
+      sb.append("- pattern: '").append(llapBean).append("ExecutorMaxFreeSlotsConfigured'\n");
+      sb.append("  name: hadoop_llapdaemon_executormaxfreeslotsconfigured\n");
+      sb.append("  type: GAUGE\n");
       break;
     case ConfigUtils.COMPONENT_TEZAM:
       // LlapMetricsSystem registers beans under LlapTaskScheduler service
       // SchedulerDagStatus tracks if the AM is running a dag or is idle so exported as GAUGE.
+      // SchedulerPendingTaskCount tracks currently waiting tasks in the AM to be scheduled.
       String schedulerBean = "Hadoop<service=LlapTaskScheduler, name=LlapTaskSchedulerMetrics.+><>";
       sb.append("- pattern: '").append(schedulerBean).append("SchedulerDagStatus'\n");
       sb.append("  name: tez_am_dag_running\n");
       sb.append("  type: GAUGE\n");
+      sb.append("- pattern: '").append(schedulerBean).append("SchedulerPendingTaskCount'\n");
+      sb.append("  name: tez_am_pending_tasks\n");
+      sb.append("  type: GAUGE\n");
       break;
     default:
       sb.append("- pattern: '.*'\n");

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/model/spec/AutoscalingSpec.java

@@ -33,10 +33,14 @@ public record AutoscalingSpec(
     @Default("0")
     Integer minReplicas,
     @JsonPropertyDescription("Threshold that triggers scale-up (component-specific: "
-        + "sessions per pod for HS2, request rate for HMS, busy slots per daemon for LLAP). "
+        + "sessions per pod for HS2, request rate for HMS, load percentage pending in TezAM for LLAP). "
         + "Not used by TezAM (demand-based: 1 TezAM per session).")
     @Default("80")
     Integer scaleUpThreshold,
+    @JsonPropertyDescription("Threshold that triggers scale-down (LLAP only): "
+        + "average daemon utilization percentage (0-100) below which scale-down triggers.")
+    @Default("20")
+    Integer scaleDownThreshold,
     @JsonPropertyDescription("Stabilization window in seconds for scale-up decisions. "
         + "Picks the highest recommendation within this window to prevent flapping.")
     @Default("60")
@@ -73,6 +77,7 @@ public record AutoscalingSpec(
     enabled = enabled != null ? enabled : false;
     minReplicas = minReplicas != null ? minReplicas : 0;
     scaleUpThreshold = scaleUpThreshold != null ? scaleUpThreshold : 80;
+    scaleDownThreshold = scaleDownThreshold != null ? scaleDownThreshold : 20;
     scaleUpStabilizationSeconds = scaleUpStabilizationSeconds != null ? scaleUpStabilizationSeconds : 60;
     scaleDownStabilizationSeconds = scaleDownStabilizationSeconds != null ? scaleDownStabilizationSeconds : 600;
     gracePeriodSeconds = gracePeriodSeconds != null ? gracePeriodSeconds : 3600;

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/model/spec/HiveServer2Spec.java

@@ -63,6 +63,6 @@ public record HiveServer2Spec(
     extraVolumeMounts = extraVolumeMounts != null ? extraVolumeMounts : List.of();
     externalJars = externalJars != null ? externalJars : List.of();
     autoscaling = autoscaling != null ? autoscaling : new AutoscalingSpec(
-        false, 1, 80, 60, 600, 300, 10, 90, 30, null);
+        false, 1, 80, 0, 60, 600, 300, 10, 90, 30, null);
   }
 }

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/model/spec/LlapSpec.java

@@ -81,7 +81,7 @@ public record LlapTezAmSpec(
     public LlapTezAmSpec {
       replicas = replicas != null ? replicas : 1;
       autoscaling = autoscaling != null ? autoscaling : new AutoscalingSpec(
-          false, 0, 0, 60, 600, 120, 10, 0, 0, null);
+          false, 0, 0, 0, 60, 600, 120, 10, 0, 0, null);
     }
   }
 
@@ -102,7 +102,7 @@ public record LlapTezAmSpec(
     extraVolumes = extraVolumes != null ? extraVolumes : List.of();
     extraVolumeMounts = extraVolumeMounts != null ? extraVolumeMounts : List.of();
     autoscaling = autoscaling != null ? autoscaling : new AutoscalingSpec(
-        false, 0, 1, 60, 900, 600, 10, 0, 0, null);
+        false, 0, 1, 20, 60, 900, 600, 10, 0, 0, null);
     tezAm = tezAm != null ? tezAm : new LlapTezAmSpec(null, null);
   }
 

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/model/spec/MetastoreSpec.java

@@ -70,7 +70,7 @@ public record MetastoreSpec(
     extraVolumes = extraVolumes != null ? extraVolumes : List.of();
     extraVolumeMounts = extraVolumeMounts != null ? extraVolumeMounts : List.of();
     autoscaling = autoscaling != null ? autoscaling : new AutoscalingSpec(
-        false, 1, 75, 60, 300, 60, 10, 90, 30, null);
+        false, 1, 75, 0, 60, 300, 60, 10, 90, 30, null);
   }
 
   public boolean isEnabled() {

packaging/src/kubernetes/src/java/org/apache/hive/kubernetes/operator/model/spec/TezAmSpec.java

@@ -64,7 +64,7 @@ public record TezAmSpec(
     extraVolumes = extraVolumes != null ? extraVolumes : List.of();
     extraVolumeMounts = extraVolumeMounts != null ? extraVolumeMounts : List.of();
     autoscaling = autoscaling != null ? autoscaling : new AutoscalingSpec(
-        false, 0, 0, 60, 600, 120, 10, 0, 0, null);
+        false, 0, 0, 0, 60, 600, 120, 10, 0, 0, null);
   }
 
   public boolean isEnabled() {