Repository: kubernetes-sigs/kueue

Issue:
[FS] Workloads submitted to 0 weight ClusterQueue may end up in infinite preemption cycle
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

If the matter is security related, please disclose it privately via https://kubernetes.io/security/
-->

**What happened**:
When there are two CQs with weight 0 that point to the same Cohort, Kueue may end up in an infinite cycle of preemption. This is because the DRS share is always `maxInt`, no matter how many Workloads are submitted/preempted to a given CQ

I've also reproduced this with a different configuration by setting CQs' weights to a large number (10mils) so CQ's `.weightedShare` doesnt change after preemption. This seems to be the core of the issue

**What you expected to happen**:
No infinite preemption cycle

**How to reproduce it (as minimally and precisely as possible)**:
Manifest for queue's hierarchy:

```
apiVersion: kueue.x-k8s.io/v1beta1
kind: ResourceFlavor
metadata:
  name: default-flavor
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: Cohort
metadata:
  name: "top-cohort"
spec:
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default-flavor"
      resources:
      - name: "cpu"
        nominalQuota: 8
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "best-effort-cq-a"
spec:
  namespaceSelector: {}
  cohort: "top-cohort"
  fairSharing:
      weight: "0"  #1
  preemption:
    reclaimWithinCohort: Any
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default-flavor"
      resources:
      - name: "cpu"
        nominalQuota: 0
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: ClusterQueue
metadata:
  name: "best-effort-cq-b"
spec:
  namespaceSelector: {}
  cohort: "top-cohort"
  fairSharing:
    weight: "0"  #1
  preemption:
    reclaimWithinCohort: Any
  resourceGroups:
  - coveredResources: ["cpu"]
    flavors:
    - name: "default-flavor"
      resources:
      - name: "cpu"
        nominalQuota: 0
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  namespace: "default"
  name: "best-effort-lq-a"
spec:
  clusterQueue: "best-effort-cq-a"
---
apiVersion: kueue.x-k8s.io/v1beta1
kind: LocalQueue
metadata:
  namespace: "default"
  name: "best-effort-lq-b"
spec:
  clusterQueue: "best-effort-cq-b"
```

Job manifest: 
```
apiVersion: batch/v1
kind: Job
metadata:
  generateName: best-effort-b-
  namespace: default
  labels:
    kueue.x-k8s.io/queue-name: best-effort-lq-b
spec:
  parallelism: 1
  suspend: true
  template:
    spec:
      containers:
      - name: dummy-job
        image: alpine
        command: ["/bin/sleep", "infinity"]
        resources:
          requests:
            cpu: 4
      restartPolicy: Never
```

1. Submit 2 jobs to the CQ-A
2. Submit a job to the CQ-B
3. Observe that only one job from CQ-A is running and there's infinite preemption cycle between the second job from CQ-A and the job from CQ-B 

**Anything else we need to know?**:

**Environment**:
- Kubernetes version (use `kubectl version`):
- Kueue version (use `git describe --tags --dirty --always`):
- Cloud provider or hardware configuration:
- OS (e.g: `cat /etc/os-release`):
- Kernel (e.g. `uname -a`):
- Install tools:
- Others:

Discussion:
@PBundyra: One of the way to solve this issues is to slightly change FairSharing API and prevent setting `0` as `.fairSharing.weight`. Instead user would use `1m` to indicate a CQ/Cohort has the lowest weight possible. I've already discussed it with @gabesaba and agreed on this approach 

Since this would be change that breaks backward compatibility we would need to create a conversion webhook

cc @mimowo @tenzen-y @mimowo: Are there alternatives which were explored? 

Im not sure this is a role of convertion webhook. I would rather say that if 0 is illegal then we deprecated it as a possible value in 0.14 and drop in v1beta2. Then we could validate against 0 with a proper message. Also, it will allow us to properly support 0 one day if we need to.@mimowo: Or maybe i misunderstood the proposal, Is it to have a webhook between v1beta1 and v1beta2, converting 0 to 1, and validate against 0 in v1beta2. In v1beta1 we would just deprecated? 

I guess it makes sense, but can we say there are no use cases requiring 0, and we can cover them all with 1? 

Im not clear about it, maybe it is a good topic to ask on wg-batch.@PBundyra: > Or maybe i misunderstood the proposal, Is it to have a webhook between v1beta1 and v1beta2, converting 0 to 1, and validate against 0 in v1beta2. In v1beta1 we would just deprecated?

Yes, that's what I meant

> I guess it makes sense, but can we say there are no use cases requiring 0, and we can cover them all with 1?


0 weight was to indicate the lowest priority of a CQ/Cohort, and now users would need to use `1m`. Surely there are some edge cases (e.g. if users previously set `1m` for some CQ/Cohort to distinguish it from `0` weight CQ/Cohort than it doesn't work anymore) but I think those can be resolved by some adjustments by admins
 
> Im not clear about it, maybe it is a good topic to ask on wg-batch.

Sure, I'm happy to discuss this
@mimowo: Ok, so deprecating and validating against is an option.

The issue reminds me some other similar infinite preemption loops from the past which we solved by using deterministic tie-breaker (timestamps or UIDs). iiuc here, similarly, we need tie-breaking because both CQs get the same DRF=maxInt. 

Could you explain why this approach wouldn't work in this case? I think it would be useful to learn this nuance for better understanding of FS.
@amy: Linking this potentially relevant issue. Its not preemption cycling per se. But it is about candidate ordering as well as issues with rounding fractional values up to 1 for large enough clusters: https://github.com/kubernetes-sigs/kueue/issues/6621

So I know that rounding up is what we do today after: https://github.com/kubernetes-sigs/kueue/pull/6617
Just haven't gotten around to quantifying whether this solution is sufficient on our end given the immediate need to fix guarantee reclamation that this PR fixed.@gabesaba: Another option is to not allow CQs/Cohorts with 0 weight to issue preemptions. However, I prefer the option of making 0 weight an invalid value, as I think this will result in simpler code and easier to understand API.@mimowo: My question is if this is solvable easily with a better / deterministic tie-break strategy. If this would work I think the code could remain simple as well. 

Strengthening validation is often hard and requires time for the deprecation process to sink. @PBundyra: I've also tested another scenario and hit the same issue: I've set the weights to a large number (e.g. 10mils) and reproduced the same cycle. It seems like the infinite cycle appears whenever the preemption of a Workload doesn't change its CQ's `weightedShare`. Hence I'll rather focus on fixing the core of the issue rather than making `0` weight invalid@PBundyra: I've found the root cause of the problem. What happens is:
1. Two workloads from CQ-A are running
2. Workload from CQ-B comes and preempts one from CQ-A
3. In the next scheduling cycle there are two Workloads, one from CQ-A and one from CQ-B
4. Scheduler runs FS tournament and determines that CQ-A is a winner - this is a problem
5. Then workload from CQ-A is admitted and the cycle repeats

Now, why it happens? This is because of how DRS are calculated in this function: https://github.com/kubernetes-sigs/kueue/blob/main/pkg/scheduler/fair_sharing_iterator.go#L190-L218

With 0 (or very large) weight, admission of a single Workload doesn't impact DRS value, so
Before admitting second Workload, CQ-A's DRS is `maxInt` (or 1) and it's the same after simulating admission of the second CQ-A's Workload. For CQ-B the result of simulation is the same, DRS is `maxInt` (or 1). Then the ordering fallbacks to creation timestamp where workload from CQ-A wins @PBundyra: So this is related to the discussion @amy started: https://github.com/kubernetes-sigs/kueue/issues/6621
We either need better tie-breaker mechanism or prevent scheduler from leading to tie in the first place (maybe by some artificial change in DRS value if the admission simulation hasn't change it before)@PBundyra: Alternatively, since the impact of the preemption is negligible (the DRS value remains the same) maybe Kueue shouldn't preempt the workload from CQ-A in the first place. This happens because condition of `LessThanOrEqualToFinalShare` strategy are met (the final share is equal) I'm hesitant if we should apply this strategy when preempting. IIUC, the main motivation behind this strategy was to fairly reclaim borrowed quota: if there are two best effort CQs, then high-prio CQ should reclaim quota from those best effort CQs equally. And when it comes to preemption the intended main strategy was `LessThanInitialShare`. 

So my alternative proposal is to create a separation of responsibilities for those two strategies:
- `LessThanOrEqualToFinalShare` should only be used for fair quota reclamation
- `LessThanInitialShare` should only. be used from fair preemption when borrowing quota@gabesaba: > We either need better tie-breaker mechanism or prevent scheduler from leading to tie in the first place (maybe by some artificial change in DRS value if the admission simulation hasn't change it before

We may consider revisiting fractional FS score (https://github.com/kubernetes-sigs/kueue/pull/6617#discussion_r2285066030). This will resolve at least the problem of high weights/rounding. I don't think that the original conclusion of the discussion, that preemption storms will be caused by fractional scores, is correct. In fact, I think it may help prevent preemption storms/unstable preemptions in the case of high weights.

> So my alternative proposal is to create a separation of responsibilities for those two strategies

Maybe, but how to differentiate the cases? I think that this becomes quite complex, especially in conjuction with hierarchical cohorts. And a conceptually complex API for the user. I prefer a solution where these strategies result in a stable state after some number of preemptions. I think that this can be accomplished by implementing several of the options above:

1) prevent 0-weight preemption storm
  a) make 0-weight illegal OR 
  b) prevent 0-weight nodes from issuing preemptions (a simple, hacky way is to return false when preemptorNewShare is MaxInt [here](https://github.com/kubernetes-sigs/kueue/blob/bf2ad4a40a21e50bdd11452ec0ecc1bf7d60080e/pkg/scheduler/preemption/fairsharing/strategy.go#L35-L43))
2) fix tiebreak issue, preferably by getting rid of rounding (see the very first part of this comment)@PBundyra: I agree that a lot of issues come down to lack of precision with the current way of calculating DRS. However, I synced with @mimowo offline and we came up with a way of keeping the same API as now and improving precision. 

Currently when comparing two Workloads we calculate DRS which is:
$DRS_1=\frac{\frac{borrowed_1}{lendable_1}}{weight_1}  \gtreqless  \frac{\frac{borrowed_2}{lendable_2}}{weight_2}  = DRS_2$ 
This comes with two divisions which introduce loss of precision. We could also rewrite those in an equivalent form that eliminates division:
$DRS_1=\frac{\frac{borrowed_1}{lendable_1}}{weight_1} = \frac{borrowed_1}{lendable_1 \cdot weight_1} \gtreqless \frac{borrowed_2}{lendable_2 \cdot weight_2} = DRS_2 <=> borrowed_1 \cdot lendable_2 \cdot weight_2  \gtreqless borrowed_2 \cdot lendable_1 \cdot weight_1$

The results are equivalent but we instead of dividing we now multiply.

Now, there's no a single static value for a candidate, but it dynamically changes while still keeping total order.
(Of course there are edge cases with 0 weight, but those can be easily solved and I would rather show in a PR.)

We still expose the same DRS as previously but the internal calculation differ. This is desirable as we would prefer not to change existing API but the value on its own doesn't determine what happens in case of tie-break. To fully determine what should be the order of candidates a user would need to compare weights, lendable and borrowed quota. This is however not a regression as currently DRS on it's own is non-decisive (by definition) in case of tie-breaks 

/cc @mimowo @gabesaba @amy
@amy: I like this: https://github.com/kubernetes-sigs/kueue/issues/6774#issuecomment-3291706679

Regarding super high precision. I remember chatting with @pajakd at the time, and why the value wasn't a float64 in the first place. He mentioned the potential for preemption cycling for super high precision. Want to make sure this wouldn't introduce another kind of cycling. 

@PBundyra after you make the PR, can you assign me as reviewer for a pass? 🙏@mimowo: Nice, we also talked with @PBundyra that with that approach we could potentially eliminate this hack: https://github.com/kubernetes-sigs/kueue/pull/6617 (but requires a careful check).@amy: What I mean is, whether its float64 or its borrowed * lendable * weight, both seem like the same category where the end result is a "high precision number".

Will there be preemption cycling for high precision numbers? Where A & B are marginally different, but end up preempting each other? It seemed like that's why you originally had the int "bucketing" solution. @mimowo: > What I mean is, whether its float64 or its borrowed * lendable * weight, both seem like the same category where the end result is a "high precision number".

It is similar conceptually indeed, but keeping it in the multication form allows us to resolve the cases when weight=0.

> Will there be preemption cycling for high precision numbers? Where A & B are marginally different, but end up preempting each other? It seemed like that's why you originally had the int "bucketing" solution.

I don't think this was ever a "bucketing" solution by design. Probably more of an attempt to avoid surfacing floats in the to the API layer. https://github.com/kubernetes-sigs/kueue/tree/main/keps/1714-fair-sharing. I think it makes sense to assume that user does not need to see the super high resolution, but for Kueue internals, we sometimes use tie-breaks which aren't really surfaced to the user, like for example UIDs.@amy: Mainly want to make sure that `borrowed * lendable * weight` doesn't also result in preemption cycling. If it doesn't, LGTM!

Could there be a situation where there's 2 workloads with marginally different DRS values that end up continuously preempting each other? @mimowo: That is a valid concern to have about the two interleaving workloads. 

This was actually the problem with buckets, that the tie break continues to resolve the Equal in opposite directions each time.

This comes down to the FinallShare LessOrEqual semantics. because this is FinalShare, then the preemption should not be triggered as long as the Less case (strict inequality) happens. @PBundyra: > Could there be a situation where there's 2 workloads with marginally different DRS values that end up continuously preempting each other?

@gabesaba Created a formal proof that this won't happen - we'll put in the documentation soon. However the proof had assumption that for CQ's DRS changes on preemption/admission. While theoretically this is correct, our implementation violated this assumption because of the roundings (and 0 weight but that's a bit different story).

Now let's compare high vs low precision calculation and bucketing ints.
With bucketing ints there are a lot of preemption when using `LessThanOrEqualToFinalShare` strategy because of the `Equal` factor, just as @mimowo mentioned. Bucketing also breaks the formal proof and leads to preemption cycles. We could eliminate that by adding higher precision one way or another (this is an implementation detail). Whenever `LessThanOrEqualToFinalShare` is used this is a desirable solution.

There's also point that @tenzen-y made here: https://github.com/kubernetes-sigs/kueue/pull/6617#discussion_r2286363618 Bucketing ints prevents from `LessThanInitialShare` preemptions that rely on marginal (order of magnitude of rounding error) differences. This is the case only when `LessThanInitialShare` is the only present strategy - if `LessThanOrEqualToFinalShare` is also present the preemption would be issued anyway. Having said that, even with the greater number of `LessThanInitialShare` preemption, there would be no preemption cycles and the algorithm would converge to a stable state.

To sum up, introducing greater precision fixes current bugs that lead to infinite preemption cycles, and it lowers number of preemption whenever `LessThanOrEqualToFinalShare` is present. If `LessThanOrEqualToFinalShare` is not part of a config, higher precision can lead to greater number of preemption, but those preemption converge to a stable state, there is no infinite preemption cycles. Given that by default Kueue uses `LessThanOrEqualToFinalShare` I think we should introduce greater precision.@amy: @PBundyra thanks for the writeup! Interested to check out the proofing writeup. Please tag me to review doc/code changes when relevant. Thank you! 🙏@PBundyra: I've synced with @gabesaba that will take over the implementation
/assign @gabesaba @mimowo: I think as a follow up of increasing the resolution we can also drop this one: https://github.com/kubernetes-sigs/kueue/pull/6617

Solution patch:
diff --git a/apis/config/v1beta1/configuration_types.go b/apis/config/v1beta1/configuration_types.go
index f0f60a2a8..01c5cfb27 100644
--- a/apis/config/v1beta1/configuration_types.go
+++ b/apis/config/v1beta1/configuration_types.go
@@ -356,15 +356,16 @@ const (
 )
 
 type InternalCertManagement struct {
-	// Enable controls the use of internal cert management for the webhook
-	// and metrics endpoints.
+	// Enable controls the use of internal cert management for the webhook,
+	// metrics and visibility endpoints.
 	// When enabled Kueue is using libraries to generate and
 	// self-sign the certificates.
 	// When disabled, you need to provide the certificates for
-	// the webhooks and metrics through a third party certificate
+	// the webhooks, metrics and visibility through a third party certificate
 	// This secret is mounted to the kueue controller manager pod. The mount
-	// path for webhooks is /tmp/k8s-webhook-server/serving-certs, whereas for
-	// metrics endpoint the expected path is `/etc/kueue/metrics/certs`.
+	// path for webhooks is /tmp/k8s-webhook-server/serving-certs, for
+	// metrics endpoint the expected path is `/etc/kueue/metrics/certs` and for
+	// visibility endpoint the expected path is `/visibility`.
 	// The keys and certs are named tls.key and tls.crt.
 	Enable *bool `json:"enable,omitempty"`
 
diff --git a/apis/kueue/v1beta1/clusterqueue_types.go b/apis/kueue/v1beta1/clusterqueue_types.go
index eef40861a..4bac827c9 100644
--- a/apis/kueue/v1beta1/clusterqueue_types.go
+++ b/apis/kueue/v1beta1/clusterqueue_types.go
@@ -59,7 +59,7 @@ type ClusterQueueSpec struct {
 	// Each resource group defines the list of resources and a list of flavors
 	// that provide quotas for these resources.
 	// Each resource and each flavor can only form part of one resource group.
-	// resourceGroups can be up to 16.
+	// resourceGroups can be up to 16, with a max of 256 total flavors across all groups.
 	// +listType=atomic
 	// +kubebuilder:validation:MaxItems=16
 	ResourceGroups []ResourceGroup `json:"resourceGroups,omitempty"`
@@ -179,9 +179,10 @@ type ResourceGroup struct {
 	// coveredResources is the list of resources covered by the flavors in this
 	// group.
 	// Examples: cpu, memory, vendor.com/gpu.
-	// The list cannot be empty and it can contain up to 16 resources.
+	// The list cannot be empty and it can contain up to 64 resources. With a total
+	// of up to 256 covered resources across all resource groups in the ClusterQueue.
 	// +kubebuilder:validation:MinItems=1
-	// +kubebuilder:validation:MaxItems=16
+	// +kubebuilder:validation:MaxItems=64
 	CoveredResources []corev1.ResourceName `json:"coveredResources"`
 
 	// flavors is the list of flavors that provide the resources of this group.
@@ -190,11 +191,12 @@ type ResourceGroup struct {
 	// cpus).
 	// Each flavor MUST list all the resources listed for this group in the same
 	// order as the .resources field.
-	// The list cannot be empty and it can contain up to 16 flavors.
+	// The list cannot be empty and it can contain up to 64 flavors, with a max of
+	// 256 total flavors across all resource groups in the ClusterQueue.
 	// +listType=map
 	// +listMapKey=name
 	// +kubebuilder:validation:MinItems=1
-	// +kubebuilder:validation:MaxItems=16
+	// +kubebuilder:validation:MaxItems=64
 	Flavors []FlavorQuotas `json:"flavors"`
 }
 
diff --git a/charts/kueue/templates/certmanager/certificate-visibility-server.yaml b/charts/kueue/templates/certmanager/certificate-visibility-server.yaml
new file mode 100644
index 000000000..50f6f699a
--- /dev/null
+++ b/charts/kueue/templates/certmanager/certificate-visibility-server.yaml
@@ -0,0 +1,20 @@
+{{/* Code generated by yaml-processor. DO NOT EDIT. */}}
+
+{{- if .Values.enableCertManager }}
+apiVersion: cert-manager.io/v1
+kind: Certificate
+metadata:
+  labels:
+  {{- include "kueue.labels" . | nindent 4 }}
+  name: '{{ include "kueue.fullname" . }}-visibility-server-cert'
+  namespace: '{{ .Release.Namespace }}'
+spec:
+  dnsNames:
+  - '{{ include "kueue.fullname" . }}-visibility-server.{{ .Release.Namespace }}.svc'
+  - '{{ include "kueue.fullname" . }}-visibility-server.{{ .Release.Namespace }}.svc.{{ .Values.kubernetesClusterDomain }}'
+  commonName: '{{ include "kueue.fullname" . }}-visibility-server'
+  issuerRef:
+    kind: Issuer
+    name: '{{ include "kueue.fullname" . }}-selfsigned-issuer'
+  secretName: '{{ include "kueue.fullname" . }}-visibility-server-cert'
+{{- end }}
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
index 945231a56..011f0f343 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_clusterqueues.yaml
@@ -375,7 +375,7 @@ spec:
                     Each resource group defines the list of resources and a list of flavors
                     that provide quotas for these resources.
                     Each resource and each flavor can only form part of one resource group.
-                    resourceGroups can be up to 16.
+                    resourceGroups can be up to 16, with a max of 256 total flavors across all groups.
                   items:
                     properties:
                       coveredResources:
@@ -383,11 +383,12 @@ spec:
                           coveredResources is the list of resources covered by the flavors in this
                           group.
                           Examples: cpu, memory, vendor.com/gpu.
-                          The list cannot be empty and it can contain up to 16 resources.
+                          The list cannot be empty and it can contain up to 64 resources. With a total
+                          of up to 256 covered resources across all resource groups in the ClusterQueue.
                         items:
                           description: ResourceName is the name identifying various resources in a ResourceList.
                           type: string
-                        maxItems: 16
+                        maxItems: 64
                         minItems: 1
                         type: array
                       flavors:
@@ -398,7 +399,8 @@ spec:
                           cpus).
                           Each flavor MUST list all the resources listed for this group in the same
                           order as the .resources field.
-                          The list cannot be empty and it can contain up to 16 flavors.
+                          The list cannot be empty and it can contain up to 64 flavors, with a max of
+                          256 total flavors across all resource groups in the ClusterQueue.
                         items:
                           properties:
                             name:
@@ -483,7 +485,7 @@ spec:
                             - name
                             - resources
                           type: object
-                        maxItems: 16
+                        maxItems: 64
                         minItems: 1
                         type: array
                         x-kubernetes-list-map-keys:
diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_cohorts.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_cohorts.yaml
index e198cfb1b..95b71607c 100644
--- a/charts/kueue/templates/crd/kueue.x-k8s.io_cohorts.yaml
+++ b/charts/kueue/templates/crd/kueue.x-k8s.io_cohorts.yaml
@@ -132,11 +132,12 @@ spec:
                           coveredResources is the list of resources covered by the flavors in this
                           group.
                           Examples: cpu, memory, vendor.com/gpu.
-                          The list cannot be empty and it can contain up to 16 resources.
+                          The list cannot be empty and it can contain up to 64 resources. With a total
+                          of up to 256 covered resources across all resource groups in the ClusterQueue.
                         items:
                           description: ResourceName is the name identifying various resources in a ResourceList.
                           type: string
-                        maxItems: 16
+                        maxItems: 64
                         minItems: 1
                         type: array
                       flavors:
@@ -147,7 +148,8 @@ spec:
                           cpus).
                           Each flavor MUST list all the resources listed for this group in the same
                           order as the .resources field.
-                          The list cannot be empty and it can contain up to 16 flavors.
+                          The list cannot be empty and it can contain up to 64 flavors, with a max of
+                          256 total flavors across all resource groups in the ClusterQueue.
                         items:
                           properties:
                             name:
@@ -232,7 +234,7 @@ spec:
                             - name
                             - resources
                           type: object
-                        maxItems: 16
+                        maxItems: 64
                         minItems: 1
                         type: array
                         x-kubernetes-list-map-keys:
diff --git a/charts/kueue/templates/manager/manager.yaml b/charts/kueue/templates/manager/manager.yaml
index 57e8261a2..3b0f7de04 100644
--- a/charts/kueue/templates/manager/manager.yaml
+++ b/charts/kueue/templates/manager/manager.yaml
@@ -108,8 +108,6 @@ spec:
       - name: manager-config
         configMap:
           name: {{ include "kueue.fullname" . }}-manager-config
-      - name: visibility
-        emptyDir: {}
       {{- if .Values.enableCertManager }}
       - name: metrics-certs
         secret:
@@ -122,4 +120,18 @@ spec:
             path: tls.crt
           - key: tls.key
             path: tls.key
+      - name: visibility
+        secret:
+          secretName: {{ include "kueue.fullname" . }}-visibility-server-cert
+          optional: false
+          items:
+          - key: ca.crt
+            path: ca.crt
+          - key: tls.crt
+            path: tls.crt
+          - key: tls.key
+            path: tls.key
+      {{- else }}
+      - name: visibility
+        emptyDir: {}
       {{- end }}
diff --git a/charts/kueue/templates/visibility/apiservice_v1beta1.yaml b/charts/kueue/templates/visibility/apiservice_v1beta1.yaml
index d761fe681..5062954e0 100644
--- a/charts/kueue/templates/visibility/apiservice_v1beta1.yaml
+++ b/charts/kueue/templates/visibility/apiservice_v1beta1.yaml
@@ -3,13 +3,19 @@
 apiVersion: apiregistration.k8s.io/v1
 kind: APIService
 metadata:
+  {{- if .Values.enableCertManager }}
+  annotations:
+    cert-manager.io/inject-ca-from: '{{ .Release.Namespace }}/{{ include "kueue.fullname" . }}-visibility-server-cert'
+  {{- end }}
   labels:
   {{- include "kueue.labels" . | nindent 4 }}
   name: v1beta1.visibility.kueue.x-k8s.io
 spec:
+  {{- if not .Values.enableCertManager }}
+  insecureSkipTLSVerify: true
+  {{- end }}
   group: visibility.kueue.x-k8s.io
   groupPriorityMinimum: 100
-  insecureSkipTLSVerify: true
   service:
     name: '{{ include "kueue.fullname" . }}-visibility-server'
     namespace: '{{ .Release.Namespace }}'
diff --git a/cmd/kueue/main.go b/cmd/kueue/main.go
index 5d43061af..baba5b1f6 100644
--- a/cmd/kueue/main.go
+++ b/cmd/kueue/main.go
@@ -273,7 +273,7 @@ func main() {
 
 	if features.Enabled(features.VisibilityOnDemand) {
 		go func() {
-			if err := visibility.CreateAndStartVisibilityServer(ctx, queues); err != nil {
+			if err := visibility.CreateAndStartVisibilityServer(ctx, queues, *cfg.InternalCertManagement.Enable); err != nil {
 				setupLog.Error(err, "Unable to create and start visibility server")
 				os.Exit(1)
 			}
diff --git a/config/components/certmanager/certificate-visibility-server.yaml b/config/components/certmanager/certificate-visibility-server.yaml
new file mode 100644
index 000000000..9a27c114e
--- /dev/null
+++ b/config/components/certmanager/certificate-visibility-server.yaml
@@ -0,0 +1,18 @@
+# The following manifests contain a visibility server certificate CR.
+# More document can be found at https://docs.cert-manager.io
+apiVersion: cert-manager.io/v1
+kind: Certificate
+metadata:
+  name: visibility-server-cert  # this name should match the one appeared in kustomizeconfig.yaml
+  namespace: kueue-system
+spec:
+  commonName: visibility-server
+  dnsNames:
+  # VISIBILITY_SERVICE_NAME and VISIBILITY_SERVICE_NAMESPACE will be substituted by kustomize
+  # replacements in the config/default/kustomization.yaml file.
+  - $(VISIBILITY_SERVICE_NAME).$(VISIBILITY_SERVICE_NAMESPACE).svc
+  - $(VISIBILITY_SERVICE_NAME).$(VISIBILITY_SERVICE_NAMESPACE).svc.cluster.local
+  issuerRef:
+    kind: Issuer
+    name: selfsigned-issuer
+  secretName: visibility-server-cert
diff --git a/config/components/certmanager/kustomization.yaml b/config/components/certmanager/kustomization.yaml
index 7a4276d21..b82a7b1ee 100644
--- a/config/components/certmanager/kustomization.yaml
+++ b/config/components/certmanager/kustomization.yaml
@@ -1,6 +1,7 @@
 resources:
 - certificate.yaml
 - certificate-metrics.yaml
+- certificate-visibility-server.yaml
 
 configurations:
 - kustomizeconfig.yaml
diff --git a/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml b/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
index 34cc00c78..207dade41 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_clusterqueues.yaml
@@ -365,7 +365,7 @@ spec:
                   Each resource group defines the list of resources and a list of flavors
                   that provide quotas for these resources.
                   Each resource and each flavor can only form part of one resource group.
-                  resourceGroups can be up to 16.
+                  resourceGroups can be up to 16, with a max of 256 total flavors across all groups.
                 items:
                   properties:
                     coveredResources:
@@ -373,12 +373,13 @@ spec:
                         coveredResources is the list of resources covered by the flavors in this
                         group.
                         Examples: cpu, memory, vendor.com/gpu.
-                        The list cannot be empty and it can contain up to 16 resources.
+                        The list cannot be empty and it can contain up to 64 resources. With a total
+                        of up to 256 covered resources across all resource groups in the ClusterQueue.
                       items:
                         description: ResourceName is the name identifying various
                           resources in a ResourceList.
                         type: string
-                      maxItems: 16
+                      maxItems: 64
                       minItems: 1
                       type: array
                     flavors:
@@ -389,7 +390,8 @@ spec:
                         cpus).
                         Each flavor MUST list all the resources listed for this group in the same
                         order as the .resources field.
-                        The list cannot be empty and it can contain up to 16 flavors.
+                        The list cannot be empty and it can contain up to 64 flavors, with a max of
+                        256 total flavors across all resource groups in the ClusterQueue.
                       items:
                         properties:
                           name:
@@ -474,7 +476,7 @@ spec:
                         - name
                         - resources
                         type: object
-                      maxItems: 16
+                      maxItems: 64
                       minItems: 1
                       type: array
                       x-kubernetes-list-map-keys:
diff --git a/config/components/crd/bases/kueue.x-k8s.io_cohorts.yaml b/config/components/crd/bases/kueue.x-k8s.io_cohorts.yaml
index 878cdd653..f3b65f54b 100644
--- a/config/components/crd/bases/kueue.x-k8s.io_cohorts.yaml
+++ b/config/components/crd/bases/kueue.x-k8s.io_cohorts.yaml
@@ -116,12 +116,13 @@ spec:
                         coveredResources is the list of resources covered by the flavors in this
                         group.
                         Examples: cpu, memory, vendor.com/gpu.
-                        The list cannot be empty and it can contain up to 16 resources.
+                        The list cannot be empty and it can contain up to 64 resources. With a total
+                        of up to 256 covered resources across all resource groups in the ClusterQueue.
                       items:
                         description: ResourceName is the name identifying various
                           resources in a ResourceList.
                         type: string
-                      maxItems: 16
+                      maxItems: 64
                       minItems: 1
                       type: array
                     flavors:
@@ -132,7 +133,8 @@ spec:
                         cpus).
                         Each flavor MUST list all the resources listed for this group in the same
                         order as the .resources field.
-                        The list cannot be empty and it can contain up to 16 flavors.
+                        The list cannot be empty and it can contain up to 64 flavors, with a max of
+                        256 total flavors across all resource groups in the ClusterQueue.
                       items:
                         properties:
                           name:
@@ -217,7 +219,7 @@ spec:
                         - name
                         - resources
                         type: object
-                      maxItems: 16
+                      maxItems: 64
                       minItems: 1
                       type: array
                       x-kubernetes-list-map-keys:
diff --git a/config/default/apiservice_cainjection_patch.yaml b/config/default/apiservice_cainjection_patch.yaml
new file mode 100644
index 000000000..287524c54
--- /dev/null
+++ b/config/default/apiservice_cainjection_patch.yaml
@@ -0,0 +1,9 @@
+# This patch add annotation to APIService to inject CA from the visibility-server cert.
+# The variables $(VISIBILITY_CERTIFICATE_NAMESPACE) and $(VISIBILITY_SERVICE_NAME) will be substituted by kustomize.
+
+apiVersion: apiregistration.k8s.io/v1
+kind: APIService
+metadata:
+  name: v1beta1.visibility.kueue.x-k8s.io
+  annotations:
+    cert-manager.io/inject-ca-from: $(VISIBILITY_CERTIFICATE_NAMESPACE)/$(VISIBILITY_CERTIFICATE_NAME)
diff --git a/config/default/apiservice_insecure_removal.yaml b/config/default/apiservice_insecure_removal.yaml
new file mode 100644
index 000000000..ad97adbda
--- /dev/null
+++ b/config/default/apiservice_insecure_removal.yaml
@@ -0,0 +1,2 @@
+- op: remove
+  path: /spec/insecureSkipTLSVerify
diff --git a/config/default/cert_visibility_manager_patch.yaml b/config/default/cert_visibility_manager_patch.yaml
new file mode 100644
index 000000000..aa212c551
--- /dev/null
+++ b/config/default/cert_visibility_manager_patch.yaml
@@ -0,0 +1,18 @@
+# Add the visibility-server certs volume configuration
+- op: remove
+  path: /spec/template/spec/volumes/0
+
+- op: add
+  path: /spec/template/spec/volumes/-
+  value:
+    name: visibility
+    secret:
+      secretName: kueue-visibility-server-cert
+      optional: false
+      items:
+        - key: ca.crt
+          path: ca.crt
+        - key: tls.crt
+          path: tls.crt
+        - key: tls.key
+          path: tls.key
diff --git a/config/default/kustomization.yaml b/config/default/kustomization.yaml
index f27795131..a6c683456 100644
--- a/config/default/kustomization.yaml
+++ b/config/default/kustomization.yaml
@@ -58,14 +58,27 @@ patches:
 # - path: mutating_webhookcainjection_patch.yaml
 # - path: validating_webhookcainjection_patch.yaml
 
+
 # Uncomment the patches line if you enable Metrics and CertManager
 # [CERTMANAGER] To enable metrics protected with certManager, uncomment the following line.
 # This patch will protect the metrics with certManager self-signed certs.
 # - path: cert_metrics_manager_patch.yaml
-#  target:
+#   target:
+#    kind: Deployment
+
+# [CERTMANAGER] To enable visibility server protected with certManager, uncomment the following line.
+# - path: apiservice_cainjection_patch.yaml
+# - path: apiservice_insecure_removal.yaml # to remove insecureSkipTLSVerify from the APIService
+#   target:
+#     kind: APIService
+#     name: v1beta1.visibility.kueue.x-k8s.io
+
+# - path: cert_visibility_manager_patch.yaml
+#   target:
 #    kind: Deployment
 
 
+
 # the following config is for teaching kustomize how to do var substitution
 # vars:
 # [CERTMANAGER] To enable cert-manager, uncomment all sections with 'CERTMANAGER' prefix.
@@ -107,3 +120,29 @@ patches:
 #     name: controller-manager-metrics-service
 #   fieldref:
 #     fieldpath: metadata.namespace
+# - name: VISIBILITY_SERVICE_NAME
+#   objref:
+#     kind: Service
+#     version: v1
+#     name: visibility-server
+# - name: VISIBILITY_SERVICE_NAMESPACE
+#   objref:
+#     kind: Service
+#     name: visibility-server
+#     version: v1
+#   fieldref:
+#     fieldPath: metadata.namespace
+# - name: VISIBILITY_CERTIFICATE_NAMESPACE
+#   objref:
+#     kind: Certificate
+#     group: cert-manager.io
+#     version: v1
+#     name: visibility-server-cert # this name should match the one in certificate-visibility-server.yaml
+#   fieldref:
+#     fieldpath: metadata.namespace
+# - name: VISIBILITY_CERTIFICATE_NAME
+#   objref:
+#     kind: Certificate
+#     group: cert-manager.io
+#     version: v1
+#     name: visibility-server-cert # this name should match the one in certificate-visibility-server.yaml
diff --git a/hack/e2e-common.sh b/hack/e2e-common.sh
index 314e06a38..3cc8706e9 100644
--- a/hack/e2e-common.sh
+++ b/hack/e2e-common.sh
@@ -227,6 +227,9 @@ function deploy_with_certmanager() {
         $KUSTOMIZE edit add patch --path "mutating_webhookcainjection_patch.yaml"
         $KUSTOMIZE edit add patch --path "validating_webhookcainjection_patch.yaml"
         $KUSTOMIZE edit add patch --path "cert_metrics_manager_patch.yaml" --kind Deployment
+        $KUSTOMIZE edit add patch --path "cert_visibility_manager_patch.yaml" --kind Deployment
+        $KUSTOMIZE edit add patch --path "apiservice_cainjection_patch.yaml" --kind APIService
+        $KUSTOMIZE edit add patch --path "apiservice_insecure_removal.yaml" --kind APIService
     )
 
     build_and_apply_kueue_manifests "$1" "${ROOT_DIR}/test/e2e/config/certmanager"
diff --git a/hack/processing-plan.yaml b/hack/processing-plan.yaml
index cd07a9ed8..5c7e3593a 100644
--- a/hack/processing-plan.yaml
+++ b/hack/processing-plan.yaml
@@ -96,6 +96,9 @@ files:
       - type: APPEND
         key: .subjects.[].name
         value: '"{{ include \"kueue.fullname\" . }}-"'
+      - type: DELETE
+        key: .spec.insecureSkipTLSVerify
+        onFileCondition: '.kind == "APIService"'
     postOperations:
       - type: INSERT_TEXT
         key: .metadata
@@ -110,6 +113,23 @@ files:
           {{- include "kueue.selectorLabels" . | nindent 4 }}
         indentation: 2
         onFileCondition: '.kind == "Service"'
+      - type: INSERT_TEXT
+        key: .metadata
+        value: |
+          {{- if .Values.enableCertManager }}
+          annotations:
+            cert-manager.io/inject-ca-from: '{{ .Release.Namespace }}/{{ include "kueue.fullname" . }}-visibility-server-cert'
+          {{- end }}
+        indentation: 2
+        onFileCondition: '.kind == "APIService"'
+      - type: INSERT_TEXT
+        key: .spec
+        value: |
+          {{- if not .Values.enableCertManager }}
+          insecureSkipTLSVerify: true
+          {{- end }}
+        indentation: 2
+        onFileCondition: '.kind == "APIService"'
   - path: ./config/components/visibility-apf/*.yaml
     outputDir: ./charts/kueue/templates/visibility-apf
     removeComments: true
@@ -901,6 +921,14 @@ files:
           - '{{ include "kueue.fullname" . }}-controller-manager-metrics-service.{{ .Release.Namespace }}.svc.{{ .Values.kubernetesClusterDomain }}'
         indentation: 2
         onFileCondition: '.kind == "Certificate" and .metadata.name | contains("metrics-cert")'
+      - type: INSERT_TEXT
+        key: .spec
+        value: |
+          dnsNames:
+          - '{{ include "kueue.fullname" . }}-visibility-server.{{ .Release.Namespace }}.svc'
+          - '{{ include "kueue.fullname" . }}-visibility-server.{{ .Release.Namespace }}.svc.{{ .Values.kubernetesClusterDomain }}'
+        indentation: 2
+        onFileCondition: '.kind == "Certificate" and .metadata.name | contains("visibility-server-cert")'
       - type: INSERT_TEXT
         position: START
         value: |
diff --git a/hack/releasing/sync-notes.sh b/hack/releasing/sync-notes.sh
index 22c360803..93b3e6a74 100755
--- a/hack/releasing/sync-notes.sh
+++ b/hack/releasing/sync-notes.sh
@@ -57,9 +57,17 @@ git fetch "${UPSTREAM_REMOTE}" --tags
 find_previous_version() {
   local release_version="$1"
   IFS='.' read -r major minor patch <<< "${release_version#v}"
+  # If patch is 0, treat it as a minor or major version bump
   if [ "$patch" -eq 0 ]; then
-    echo "${release_version}-devel"
-    return 0
+    if [ "$minor" -gt 0 ]; then
+      # Decrease minor version and set patch to 0
+      echo "v$major.$((minor-1)).0"
+      return 0
+    elif [ "$major" -gt 0 ]; then
+      # Decrease major version, set minor and patch to 0
+      echo "v$((major-1)).0.0"
+      return 0
+    fi
   fi
   for tag in $(git tag -l | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' | sort -V -r); do
     IFS='.' read -r t_major t_minor t_patch <<< "${tag#v}"
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 80e10dd5f..3be9a9529 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -345,8 +345,8 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 			quotaReservedCondition := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadQuotaReserved)
 			quotaReservedWaitTime := r.clock.Since(quotaReservedCondition.LastTransitionTime.Time)
 			r.recorder.Eventf(&wl, corev1.EventTypeNormal, "Admitted", "Admitted by ClusterQueue %v, wait time since reservation was %.0fs", wl.Status.Admission.ClusterQueue, quotaReservedWaitTime.Seconds())
-			metrics.AdmittedWorkload(cqName, workload.GetWorkloadPriorityClass(&wl), queuedWaitTime)
-			metrics.AdmissionChecksWaitTime(cqName, quotaReservedWaitTime)
+			metrics.AdmittedWorkload(cqName, wl.Spec.PriorityClassName, queuedWaitTime)
+			metrics.ReportAdmissionChecksWaitTime(cqName, wl.Spec.PriorityClassName, quotaReservedWaitTime)
 			if features.Enabled(features.LocalQueueMetrics) {
 				metrics.LocalQueueAdmittedWorkload(metrics.LQRefFromWorkload(&wl), queuedWaitTime)
 				metrics.LocalQueueAdmissionChecksWaitTime(metrics.LQRefFromWorkload(&wl), quotaReservedWaitTime)
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index 9d4c6c931..183c4fa5c 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -134,7 +134,7 @@ The label 'result' can have the following values:
 			Subsystem: constants.KueueName,
 			Name:      "quota_reserved_workloads_total",
 			Help:      "The total number of quota reserved workloads per 'cluster_queue'",
-		}, []string{"cluster_queue", "workload_priority_class"},
+		}, []string{"cluster_queue", "priority_class"},
 	)
 
 	LocalQueueQuotaReservedWorkloadsTotal = prometheus.NewCounterVec(
@@ -151,7 +151,7 @@ The label 'result' can have the following values:
 			Name:      "quota_reserved_wait_time_seconds",
 			Help:      "The time between a workload was created or requeued until it got quota reservation, per 'cluster_queue'",
 			Buckets:   generateExponentialBuckets(14),
-		}, []string{"cluster_queue", "workload_priority_class"},
+		}, []string{"cluster_queue", "priority_class"},
 	)
 
 	PodsReadyToEvictedTimeSeconds = prometheus.NewHistogramVec(
@@ -190,7 +190,7 @@ The label 'underlying_cause' can have the following values:
 			Subsystem: constants.KueueName,
 			Name:      "admitted_workloads_total",
 			Help:      "The total number of admitted workloads per 'cluster_queue'",
-		}, []string{"cluster_queue", "workload_priority_class"},
+		}, []string{"cluster_queue", "priority_class"},
 	)
 
 	LocalQueueAdmittedWorkloadsTotal = prometheus.NewCounterVec(
@@ -207,7 +207,7 @@ The label 'underlying_cause' can have the following values:
 			Name:      "admission_wait_time_seconds",
 			Help:      "The time between a workload was created or requeued until admission, per 'cluster_queue'",
 			Buckets:   generateExponentialBuckets(14),
-		}, []string{"cluster_queue", "workload_priority_class"},
+		}, []string{"cluster_queue", "priority_class"},
 	)
 
 	queuedUntilReadyWaitTime = prometheus.NewHistogramVec(
@@ -237,13 +237,13 @@ The label 'underlying_cause' can have the following values:
 		}, []string{"name", "namespace"},
 	)
 
-	admissionChecksWaitTime = prometheus.NewHistogramVec(
+	AdmissionChecksWaitTime = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Subsystem: constants.KueueName,
 			Name:      "admission_checks_wait_time_seconds",
 			Help:      "The time from when a workload got the quota reservation until admission, per 'cluster_queue'",
 			Buckets:   generateExponentialBuckets(14),
-		}, []string{"cluster_queue"},
+		}, []string{"cluster_queue", "priority_class"},
 	)
 
 	localQueueAdmissionChecksWaitTime = prometheus.NewHistogramVec(
@@ -291,7 +291,7 @@ The label 'underlying_cause' can have the following values:
 - "AdmissionCheck" means that the workload was evicted by Kueue due to a rejected admission check.
 - "MaximumExecutionTimeExceeded" means that the workload was evicted by Kueue due to maximum execution time exceeded.
 - "RequeuingLimitExceeded" means that the workload was evicted by Kueue due to requeuing limit exceeded.`,
-		}, []string{"cluster_queue", "reason", "underlying_cause", "workload_priority_class"},
+		}, []string{"cluster_queue", "reason", "underlying_cause", "priority_class"},
 	)
 
 	ReplacedWorkloadSlicesTotal = prometheus.NewCounterVec(
@@ -343,7 +343,7 @@ The label 'detailed_reason' can have the following values:
 - "AdmissionCheck" means that the workload was evicted by Kueue due to a rejected admission check.
 - "MaximumExecutionTimeExceeded" means that the workload was evicted by Kueue due to maximum execution time exceeded.
 - "RequeuingLimitExceeded" means that the workload was evicted by Kueue due to requeuing limit exceeded.`,
-		}, []string{"cluster_queue", "reason", "detailed_reason", "workload_priority_class"},
+		}, []string{"cluster_queue", "reason", "detailed_reason", "priority_class"},
 	)
 
 	PreemptedWorkloadsTotal = prometheus.NewCounterVec(
@@ -510,9 +510,9 @@ func AdmissionAttempt(result AdmissionResult, duration time.Duration) {
 	admissionAttemptDuration.WithLabelValues(string(result)).Observe(duration.Seconds())
 }
 
-func QuotaReservedWorkload(cqName kueue.ClusterQueueReference, workloadPriorityClass string, waitTime time.Duration) {
-	QuotaReservedWorkloadsTotal.WithLabelValues(string(cqName), workloadPriorityClass).Inc()
-	QuotaReservedWaitTime.WithLabelValues(string(cqName), workloadPriorityClass).Observe(waitTime.Seconds())
+func QuotaReservedWorkload(cqName kueue.ClusterQueueReference, priorityClass string, waitTime time.Duration) {
+	QuotaReservedWorkloadsTotal.WithLabelValues(string(cqName), priorityClass).Inc()
+	QuotaReservedWaitTime.WithLabelValues(string(cqName), priorityClass).Observe(waitTime.Seconds())
 }
 
 func LocalQueueQuotaReservedWorkload(lq LocalQueueReference, waitTime time.Duration) {
@@ -520,9 +520,9 @@ func LocalQueueQuotaReservedWorkload(lq LocalQueueReference, waitTime time.Durat
 	localQueueQuotaReservedWaitTime.WithLabelValues(string(lq.Name), lq.Namespace).Observe(waitTime.Seconds())
 }
 
-func AdmittedWorkload(cqName kueue.ClusterQueueReference, workloadPriorityClass string, waitTime time.Duration) {
-	AdmittedWorkloadsTotal.WithLabelValues(string(cqName), workloadPriorityClass).Inc()
-	AdmissionWaitTime.WithLabelValues(string(cqName), workloadPriorityClass).Observe(waitTime.Seconds())
+func AdmittedWorkload(cqName kueue.ClusterQueueReference, priorityClass string, waitTime time.Duration) {
+	AdmittedWorkloadsTotal.WithLabelValues(string(cqName), priorityClass).Inc()
+	AdmissionWaitTime.WithLabelValues(string(cqName), priorityClass).Observe(waitTime.Seconds())
 }
 
 func LocalQueueAdmittedWorkload(lq LocalQueueReference, waitTime time.Duration) {
@@ -530,8 +530,8 @@ func LocalQueueAdmittedWorkload(lq LocalQueueReference, waitTime time.Duration)
 	localQueueAdmissionWaitTime.WithLabelValues(string(lq.Name), lq.Namespace).Observe(waitTime.Seconds())
 }
 
-func AdmissionChecksWaitTime(cqName kueue.ClusterQueueReference, waitTime time.Duration) {
-	admissionChecksWaitTime.WithLabelValues(string(cqName)).Observe(waitTime.Seconds())
+func ReportAdmissionChecksWaitTime(cqName kueue.ClusterQueueReference, workloadPriorityClass string, waitTime time.Duration) {
+	AdmissionChecksWaitTime.WithLabelValues(string(cqName), workloadPriorityClass).Observe(waitTime.Seconds())
 }
 
 func LocalQueueAdmissionChecksWaitTime(lq LocalQueueReference, waitTime time.Duration) {
@@ -564,8 +564,8 @@ func ReportLocalQueuePendingWorkloads(lq LocalQueueReference, active, inadmissib
 	LocalQueuePendingWorkloads.WithLabelValues(string(lq.Name), lq.Namespace, PendingStatusInadmissible).Set(float64(inadmissible))
 }
 
-func ReportEvictedWorkloads(cqName kueue.ClusterQueueReference, evictionReason, underlyingCause, workloadPriorityClass string) {
-	EvictedWorkloadsTotal.WithLabelValues(string(cqName), evictionReason, underlyingCause, workloadPriorityClass).Inc()
+func ReportEvictedWorkloads(cqName kueue.ClusterQueueReference, evictionReason, underlyingCause, priorityClass string) {
+	EvictedWorkloadsTotal.WithLabelValues(string(cqName), evictionReason, underlyingCause, priorityClass).Inc()
 }
 
 func ReportReplacedWorkloadSlices(cqName kueue.ClusterQueueReference) {
@@ -576,8 +576,8 @@ func ReportLocalQueueEvictedWorkloads(lq LocalQueueReference, reason, underlying
 	LocalQueueEvictedWorkloadsTotal.WithLabelValues(string(lq.Name), lq.Namespace, reason, underlyingCause).Inc()
 }
 
-func ReportEvictedWorkloadsOnce(cqName kueue.ClusterQueueReference, reason, underlyingCause, workloadPriorityClass string) {
-	EvictedWorkloadsOnceTotal.WithLabelValues(string(cqName), reason, underlyingCause, workloadPriorityClass).Inc()
+func ReportEvictedWorkloadsOnce(cqName kueue.ClusterQueueReference, reason, underlyingCause, priorityClass string) {
+	EvictedWorkloadsOnceTotal.WithLabelValues(string(cqName), reason, underlyingCause, priorityClass).Inc()
 }
 
 func ReportPreemption(preemptingCqName kueue.ClusterQueueReference, preemptingReason string, targetCqName kueue.ClusterQueueReference) {
@@ -600,7 +600,7 @@ func ClearClusterQueueMetrics(cqName string) {
 	PodsReadyToEvictedTimeSeconds.DeleteLabelValues(cqName)
 	AdmittedWorkloadsTotal.DeletePartialMatch(prometheus.Labels{"cluster_queue": cqName})
 	AdmissionWaitTime.DeletePartialMatch(prometheus.Labels{"cluster_queue": cqName})
-	admissionChecksWaitTime.DeleteLabelValues(cqName)
+	AdmissionChecksWaitTime.DeletePartialMatch(prometheus.Labels{"cluster_queue": cqName})
 	queuedUntilReadyWaitTime.DeleteLabelValues(cqName)
 	admittedUntilReadyWaitTime.DeleteLabelValues(cqName)
 	EvictedWorkloadsTotal.DeletePartialMatch(prometheus.Labels{"cluster_queue": cqName})
@@ -775,7 +775,7 @@ func Register() {
 		EvictedWorkloadsOnceTotal,
 		PreemptedWorkloadsTotal,
 		AdmissionWaitTime,
-		admissionChecksWaitTime,
+		AdmissionChecksWaitTime,
 		queuedUntilReadyWaitTime,
 		admittedUntilReadyWaitTime,
 		ClusterQueueResourceUsage,
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 2b3ce0efe..1e3745430 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -796,7 +796,7 @@ func (s *Scheduler) recordQuotaReservationMetrics(newWorkload, originalWorkload
 
 	s.recorder.Eventf(newWorkload, corev1.EventTypeNormal, "QuotaReserved", "Quota reserved in ClusterQueue %v, wait time since queued was %.0fs", admission.ClusterQueue, waitTime.Seconds())
 
-	metrics.QuotaReservedWorkload(admission.ClusterQueue, workload.GetWorkloadPriorityClass(newWorkload), waitTime)
+	metrics.QuotaReservedWorkload(admission.ClusterQueue, newWorkload.Spec.PriorityClassName, waitTime)
 	if features.Enabled(features.LocalQueueMetrics) {
 		metrics.LocalQueueQuotaReservedWorkload(metrics.LQRefFromWorkload(newWorkload), waitTime)
 	}
@@ -809,14 +809,14 @@ func (s *Scheduler) recordWorkloadAdmissionEvents(newWorkload, originalWorkload
 	}
 
 	s.recorder.Eventf(newWorkload, corev1.EventTypeNormal, "Admitted", "Admitted by ClusterQueue %v, wait time since reservation was 0s", admission.ClusterQueue)
-	metrics.AdmittedWorkload(admission.ClusterQueue, workload.GetWorkloadPriorityClass(newWorkload), waitTime)
+	metrics.AdmittedWorkload(admission.ClusterQueue, newWorkload.Spec.PriorityClassName, waitTime)
 
 	if features.Enabled(features.LocalQueueMetrics) {
 		metrics.LocalQueueAdmittedWorkload(metrics.LQRefFromWorkload(newWorkload), waitTime)
 	}
 
 	if len(newWorkload.Status.AdmissionChecks) > 0 {
-		metrics.AdmissionChecksWaitTime(admission.ClusterQueue, 0)
+		metrics.ReportAdmissionChecksWaitTime(admission.ClusterQueue, newWorkload.Spec.PriorityClassName, 0)
 		if features.Enabled(features.LocalQueueMetrics) {
 			metrics.LocalQueueAdmissionChecksWaitTime(metrics.LQRefFromWorkload(newWorkload), 0)
 		}
diff --git a/pkg/visibility/server.go b/pkg/visibility/server.go
index 2a467ce8c..567ec26b1 100644
--- a/pkg/visibility/server.go
+++ b/pkg/visibility/server.go
@@ -57,9 +57,9 @@ var (
 // +kubebuilder:rbac:groups=flowcontrol.apiserver.k8s.io,resources=flowschemas/status,verbs=patch
 
 // CreateAndStartVisibilityServer creates visibility server injecting KueueManager and starts it
-func CreateAndStartVisibilityServer(ctx context.Context, kueueMgr *qcache.Manager) error {
+func CreateAndStartVisibilityServer(ctx context.Context, kueueMgr *qcache.Manager, enableInternalCertManagement bool) error {
 	config := newVisibilityServerConfig()
-	if err := applyVisibilityServerOptions(config); err != nil {
+	if err := applyVisibilityServerOptions(config, enableInternalCertManagement); err != nil {
 		return fmt.Errorf("unable to apply VisibilityServerOptions: %w", err)
 	}
 
@@ -79,12 +79,17 @@ func CreateAndStartVisibilityServer(ctx context.Context, kueueMgr *qcache.Manage
 	return nil
 }
 
-func applyVisibilityServerOptions(config *genericapiserver.RecommendedConfig) error {
+func applyVisibilityServerOptions(config *genericapiserver.RecommendedConfig, enableInternalCertManagement bool) error {
 	o := genericoptions.NewRecommendedOptions("", api.Codecs.LegacyCodec(visibilityv1beta1.SchemeGroupVersion))
 	o.Etcd = nil
 	o.SecureServing.BindPort = 8082
-	// The directory where TLS certs will be created
-	o.SecureServing.ServerCert.CertDirectory = certDir
+	if enableInternalCertManagement {
+		// The directory where TLS certs will be created
+		o.SecureServing.ServerCert.CertDirectory = certDir
+	} else {
+		o.SecureServing.ServerCert.CertKey.CertFile = certDir + "/tls.crt"
+		o.SecureServing.ServerCert.CertKey.KeyFile = certDir + "/tls.key"
+	}
 	o.Admission.DisablePlugins = disabledPlugins
 	if err := o.SecureServing.MaybeDefaultWithSelfSignedCerts("localhost", nil, []net.IP{net.ParseIP("127.0.0.1")}); err != nil {
 		return fmt.Errorf("error creating self-signed certificates: %v", err)
diff --git a/pkg/webhooks/clusterqueue_webhook.go b/pkg/webhooks/clusterqueue_webhook.go
index 6205b7548..d412bd9b3 100644
--- a/pkg/webhooks/clusterqueue_webhook.go
+++ b/pkg/webhooks/clusterqueue_webhook.go
@@ -110,6 +110,9 @@ func ValidateClusterQueue(cq *kueue.ClusterQueue) field.ErrorList {
 		allErrs = append(allErrs, validatePreemption(cq.Spec.Preemption, path.Child("preemption"))...)
 	}
 	allErrs = append(allErrs, validateFairSharing(cq.Spec.FairSharing, path.Child("fairSharing"))...)
+	allErrs = append(allErrs, validateTotalFlavors(cq.Spec.ResourceGroups, path.Child("resourceGroups"))...)
+	allErrs = append(allErrs, validateTotalCoveredResources(cq.Spec.ResourceGroups, path.Child("resourceGroups"))...)
+	allErrs = append(allErrs, validateFlavorResourceCombinations(cq.Spec.ResourceGroups, path.Child("resourceGroups"))...)
 	return allErrs
 }
 
@@ -117,6 +120,50 @@ func ValidateClusterQueueUpdate(newObj *kueue.ClusterQueue) field.ErrorList {
 	return ValidateClusterQueue(newObj)
 }
 
+func validateTotalFlavors(resourceGroups []kueue.ResourceGroup, path *field.Path) field.ErrorList {
+	var allErrs field.ErrorList
+	total := 0
+	for _, rg := range resourceGroups {
+		total += len(rg.Flavors)
+	}
+	if total > 256 {
+		allErrs = append(allErrs, field.Invalid(path, total,
+			fmt.Sprintf("total number of flavors across all resourceGroups must be ≤ 256, got %d", total)))
+	}
+	return allErrs
+}
+
+func validateTotalCoveredResources(resourceGroups []kueue.ResourceGroup, path *field.Path) field.ErrorList {
+	var allErrs field.ErrorList
+	total := 0
+	for _, rg := range resourceGroups {
+		total += len(rg.CoveredResources)
+	}
+	if total > 256 {
+		allErrs = append(allErrs, field.Invalid(path, total,
+			fmt.Sprintf("total number of covered resources across all resourceGroups must be ≤ 256, got %d", total)))
+	}
+	return allErrs
+}
+
+func validateFlavorResourceCombinations(resourceGroups []kueue.ResourceGroup, path *field.Path) field.ErrorList {
+	var allErrs field.ErrorList
+	for i, rg := range resourceGroups {
+		total := 0
+		for _, fqs := range rg.Flavors {
+			total += len(fqs.Resources)
+		}
+		if total > 512 {
+			allErrs = append(allErrs, field.Invalid(
+				path.Index(i),
+				total,
+				fmt.Sprintf("number of flavor-resource combinations in a resourceGroup must be ≤ 512, got %d", total),
+			))
+		}
+	}
+	return allErrs
+}
+
 func validatePreemption(preemption *kueue.ClusterQueuePreemption, path *field.Path) field.ErrorList {
 	var allErrs field.ErrorList
 	if preemption.ReclaimWithinCohort == kueue.PreemptionPolicyNever &&
diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
index ff3e1d199..f3261896e 100644
--- a/pkg/workload/workload.go
+++ b/pkg/workload/workload.go
@@ -1123,7 +1123,7 @@ func Evict(ctx context.Context, c client.Client, recorder record.EventRecorder,
 	}
 	reportEvictedWorkload(recorder, wl, wl.Status.Admission.ClusterQueue, reason, underlyingCause, msg)
 	if reportWorkloadEvictedOnce {
-		metrics.ReportEvictedWorkloadsOnce(wl.Status.Admission.ClusterQueue, reason, underlyingCause, GetWorkloadPriorityClass(wl))
+		metrics.ReportEvictedWorkloadsOnce(wl.Status.Admission.ClusterQueue, reason, underlyingCause, wl.Spec.PriorityClassName)
 	}
 	return nil
 }
@@ -1145,7 +1145,7 @@ func resetUnhealthyNodes(w *kueue.Workload) {
 }
 
 func reportEvictedWorkload(recorder record.EventRecorder, wl *kueue.Workload, cqName kueue.ClusterQueueReference, reason, underlyingCause, message string) {
-	metrics.ReportEvictedWorkloads(cqName, reason, underlyingCause, GetWorkloadPriorityClass(wl))
+	metrics.ReportEvictedWorkloads(cqName, reason, underlyingCause, wl.Spec.PriorityClassName)
 	if podsReadyToEvictionTime := workloadsWithPodsReadyToEvictedTime(wl); podsReadyToEvictionTime != nil {
 		metrics.PodsReadyToEvictedTimeSeconds.WithLabelValues(string(cqName), reason, underlyingCause).Observe(podsReadyToEvictionTime.Seconds())
 	}
@@ -1214,12 +1214,3 @@ func setSchedulingStatsEviction(wl *kueue.Workload, newEvictionState kueue.Workl
 	}
 	return false
 }
-
-// GetWorkloadPriorityClass returns the WorkloadPriorityClass name if it exists.
-// Otherwise, it returns an empty string.
-func GetWorkloadPriorityClass(wl *kueue.Workload) string {
-	if wl.Spec.PriorityClassSource == constants.WorkloadPriorityClassSource {
-		return wl.Spec.PriorityClassName
-	}
-	return ""
-}
diff --git a/site/content/en/docs/concepts/fair_sharing.md b/site/content/en/docs/concepts/fair_sharing.md
index a6aa27468..271692c06 100644
--- a/site/content/en/docs/concepts/fair_sharing.md
+++ b/site/content/en/docs/concepts/fair_sharing.md
@@ -7,8 +7,80 @@ description: >
 ---
 
 
-### [Admission Fair Sharing](/docs/concepts/admission_fair_sharing)
+## [Admission Fair Sharing](/docs/concepts/admission_fair_sharing)
 
-A mechanism for ordering workloads based on the historical resource usage of their source LocalQueues, giving preference to those that have consumed fewer resources over time.
+A mechanism for ordering workloads based on the historical resource usage of
+their source LocalQueues, giving preference to those that have consumed fewer
+resources over time.
 
-### [Preemption based Fair Sharing](/docs/concepts/preemption/#fair-sharing)
\ No newline at end of file
+## [Preemption based Fair Sharing](/docs/concepts/preemption/#fair-sharing)
+
+### Proof that two workloads won't preempt each other
+
+Here we present a proof that FairSharing preemption
+[strategies](https://github.com/kubernetes-sigs/kueue/blob/efc4a1bc10eacdd8a86129ccae34eadf3dd75424/pkg/scheduler/preemption/fairsharing/strategy.go#L37-L45)
+won't result in a loop where two workloads in different ClusterQueues keep
+preempting each other.
+
+Suppose Workload A is trying to preempt Workload B, and that these workloads are
+in different ClusterQueues.  We define the following variables as Dominant
+Resource Shares (DRS, see [fair_sharing.go](https://github.com/kubernetes-sigs/kueue/blob/efc4a1bc10eacdd8a86129ccae34eadf3dd75424/pkg/cache/scheduler/fair_sharing.go))
+of ClusterQueues before and after admission of their respective workloads:
+  - DRS_A_pending
+  - DRS_A_admitted
+  - DRS_B_pending
+  - DRS_B_admitted
+
+We assume that, if the ClusterQueue is borrowing after admission, admission
+increases DRS and preemption decreases DRS: DRS_A_pending < DRS_A_admitted and
+DRS_B_pending < DRS_B_admitted. This assumption is necessary, as
+DRS_A_pending < DRS_A_admitted could fail to hold if the workloads in question
+are not using the dominant resource.
+
+We assume that both workloads are borrowing after admission. If either workload
+is not borrowing after admission, a preemption loop will not occur, as we
+exclude ClusterQueues/Cohorts which are not borrowing from preemption
+candidates. See
+[nextTarget](https://github.com/kubernetes-sigs/kueue/blob/efc4a1bc10eacdd8a86129ccae34eadf3dd75424/pkg/scheduler/preemption/fairsharing/ordering.go#L132-L138),
+where we prune Cohorts and ClusterQueues with DRS=0.
+
+#### Workload A Preempts Workload B
+
+For preemption to occur, either [DRS_A_admitted <= DRS_B_pending](https://github.com/kubernetes-sigs/kueue/blob/efc4a1bc10eacdd8a86129ccae34eadf3dd75424/pkg/scheduler/preemption/fairsharing/strategy.go#L37-L40),
+or [DRS_A_admitted < DRS_B_admitted](https://github.com/kubernetes-sigs/kueue/blob/efc4a1bc10eacdd8a86129ccae34eadf3dd75424/pkg/scheduler/preemption/fairsharing/strategy.go#L42-L45)
+must hold.
+
+Since DRS_B_pending < DRS_B_admitted, we conclude that DRS_A_admitted <
+DRS_B_admitted always holds when preemption is possible
+
+**Lemma 1** Workload A can preempt Workload B => DRS_A_admitted < DRS_B_admitted
+
+#### Workload B Preempts Workload A
+We use the same logic to create a contradiction.
+
+For preemption to occur, either DRS_B_admitted <= DRS_A_pending, or
+DRS_B_admitted < DRS_A_admitted must hold.
+
+Since DRS_A_pending < DRS_A_admitted, we conclude that DRS_B_admitted <
+DRS_A_admitted always holds when preemption is possible.
+
+**Lemma 2** Workload B can preempt Workload A => DRS_B_admitted < DRS_A_admitted
+
+#### Conclusion
+Assume that both preemption events are possible.
+
+From Lemma 1, we have DRS_A_admitted < DRS_B_admitted.  From Lemma 2, we have
+DRS_B_admitted < DRS_A_admitted.
+
+This is a contradiction. We conclude that, if workload A preempts workload
+B, then workload B will not be able to preempt Workload A afterwards.
+
+#### Corallaries
+* If workload A preempts a set of workloads {B_1, ..., B_k} and admits,
+  none of {B_1, ..., B_k} will have the right to preempt workload A afterwards.
+* This results generalizes to a chain of preemptions: If A->B->C, then
+  DRS_A_admitted < DRS_B_admitted < DRS_C_admitted. Therefore C cannot preempt A.
+
+#### Limitations
+A proof that a chain of preemptions won't cause
+a cycle is needed for the Hierarchical case.
diff --git a/site/content/en/docs/reference/kueue-config.v1beta1.md b/site/content/en/docs/reference/kueue-config.v1beta1.md
index a8a5ff264..46d2ce3bf 100644
--- a/site/content/en/docs/reference/kueue-config.v1beta1.md
+++ b/site/content/en/docs/reference/kueue-config.v1beta1.md
@@ -684,15 +684,16 @@ underlying job are changed.</p>
 <code>bool</code>
 </td>
 <td>
-   <p>Enable controls the use of internal cert management for the webhook
-and metrics endpoints.
+   <p>Enable controls the use of internal cert management for the webhook,
+metrics and visibility endpoints.
 When enabled Kueue is using libraries to generate and
 self-sign the certificates.
 When disabled, you need to provide the certificates for
-the webhooks and metrics through a third party certificate
+the webhooks, metrics and visibility through a third party certificate
 This secret is mounted to the kueue controller manager pod. The mount
-path for webhooks is /tmp/k8s-webhook-server/serving-certs, whereas for
-metrics endpoint the expected path is <code>/etc/kueue/metrics/certs</code>.
+path for webhooks is /tmp/k8s-webhook-server/serving-certs, for
+metrics endpoint the expected path is <code>/etc/kueue/metrics/certs</code> and for
+visibility endpoint the expected path is <code>/visibility</code>.
 The keys and certs are named tls.key and tls.crt.</p>
 </td>
 </tr>
diff --git a/site/content/en/docs/reference/kueue.v1beta1.md b/site/content/en/docs/reference/kueue.v1beta1.md
index 95914ba8a..84ce81ddf 100644
--- a/site/content/en/docs/reference/kueue.v1beta1.md
+++ b/site/content/en/docs/reference/kueue.v1beta1.md
@@ -975,7 +975,7 @@ It must be a DNS (RFC 1123) and has the maximum length of 253 characters.</p>
 Each resource group defines the list of resources and a list of flavors
 that provide quotas for these resources.
 Each resource and each flavor can only form part of one resource group.
-resourceGroups can be up to 16.</p>
+resourceGroups can be up to 16, with a max of 256 total flavors across all groups.</p>
 </td>
 </tr>
 <tr><td><code>cohort</code> <B>[Required]</B><br/>
@@ -2700,7 +2700,8 @@ nodes matching to the Resource Flavor node labels.</p>
    <p>coveredResources is the list of resources covered by the flavors in this
 group.
 Examples: cpu, memory, vendor.com/gpu.
-The list cannot be empty and it can contain up to 16 resources.</p>
+The list cannot be empty and it can contain up to 64 resources. With a total
+of up to 256 covered resources across all resource groups in the ClusterQueue.</p>
 </td>
 </tr>
 <tr><td><code>flavors</code> <B>[Required]</B><br/>
@@ -2713,7 +2714,8 @@ Typically, different flavors represent different hardware models
 cpus).
 Each flavor MUST list all the resources listed for this group in the same
 order as the .resources field.
-The list cannot be empty and it can contain up to 16 flavors.</p>
+The list cannot be empty and it can contain up to 64 flavors, with a max of
+256 total flavors across all resource groups in the ClusterQueue.</p>
 </td>
 </tr>
 </tbody>
diff --git a/site/content/en/docs/reference/metrics.md b/site/content/en/docs/reference/metrics.md
index 5bdb7a5a9..d70cf7c68 100644
--- a/site/content/en/docs/reference/metrics.md
+++ b/site/content/en/docs/reference/metrics.md
@@ -24,19 +24,19 @@ Use the following metrics to monitor the health of the kueue controllers:
 Use the following metrics to monitor the status of your ClusterQueues:
 
 
-| Metric name                                | Type      | Description                                                                         | Labels                                                                                                                                                                                                 |
-| -------------------------------------------- | ----------- | ------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `kueue_pending_workloads`                  | Gauge     | The number of pending workloads.                                                    | `cluster_queue`: the name of the ClusterQueue<br> `status`: possible values are `active` or `inadmissible`                                                                                             |
-| `kueue_quota_reserved_workloads_total`     | Counter   | The total number of quota reserved workloads.                                       | `cluster_queue`: the name of the ClusterQueue<br> workload_priority_class: the workload priority class name                                                                                            |
-| `kueue_quota_reserved_wait_time_seconds`   | Histogram | The time between a workload was created or requeued until it got quota reservation. | `cluster_queue`: the name of the ClusterQueue<br> workload_priority_class: the workload priority class name                                                                                            |
-| `kueue_admitted_workloads_total`           | Counter   | The total number of admitted workloads.                                             | `cluster_queue`: the name of the ClusterQueue<br> workload_priority_class: the workload priority class name                                                                                               |
-| `kueue_evicted_workloads_total`            | Counter   | The total number of evicted workloads.                                              | `cluster_queue`: the name of the ClusterQueue<br> `reason`: Possible values are `Preempted`, `PodsReadyTimeout`, `AdmissionCheck`, `ClusterQueueStopped` or `Deactivated`<br> workload_priority_class: the workload priority class name |
-| `kueue_evicted_workloads_once_total`       | Counter   | The number of unique workload evictions per 'cluster_queue'                         | `cluster_queue`: the name of the ClusterQueue<br> `reason`: Possible values are `Preempted`, `PodsReadyTimeout`, `AdmissionCheck`, `ClusterQueueStopped` or `Deactivated`<br> `detailedReason`: specifies a finer-grained explanation that complements the eviction cause<br> workload_priority_class: the workload priority class name |
-| `kueue_admission_wait_time_seconds`        | Histogram | The time between a workload was created or requeued until admission.                | `cluster_queue`: the name of the ClusterQueue<br> workload_priority_class: the workload priority class name                                                                                            |
-| `kueue_admission_checks_wait_time_seconds` | Histogram | The time from when a workload got the quota reservation until admission.            | `cluster_queue`: the name of the ClusterQueue                                                                                                                                                          |
-| `kueue_admitted_active_workloads`          | Gauge     | The number of admitted Workloads that are active (unsuspended and not finished)     | `cluster_queue`: the name of the ClusterQueue                                                                                                                                                          |
+| Metric name                                | Type      | Description                                                                         | Labels                                                                                                                                                                                            |
+| -------------------------------------------- | ----------- | ------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `kueue_pending_workloads`                  | Gauge     | The number of pending workloads.                                                    | `cluster_queue`: the name of the ClusterQueue<br> `status`: possible values are `active` or `inadmissible`                                                                                        |
+| `kueue_quota_reserved_workloads_total`     | Counter   | The total number of quota reserved workloads.                                       | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                         |
+| `kueue_quota_reserved_wait_time_seconds`   | Histogram | The time between a workload was created or requeued until it got quota reservation. | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                         |
+| `kueue_admitted_workloads_total`           | Counter   | The total number of admitted workloads.                                             | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                           |
+| `kueue_evicted_workloads_total`            | Counter   | The total number of evicted workloads.                                              | `cluster_queue`: the name of the ClusterQueue<br> `reason`: Possible values are `Preempted`, `PodsReadyTimeout`, `AdmissionCheck`, `ClusterQueueStopped` or `Deactivated`<br> priority_class: the priority class name |
+| `kueue_evicted_workloads_once_total`       | Counter   | The number of unique workload evictions per 'cluster_queue'                         | `cluster_queue`: the name of the ClusterQueue<br> `reason`: Possible values are `Preempted`, `PodsReadyTimeout`, `AdmissionCheck`, `ClusterQueueStopped` or `Deactivated`<br> `detailedReason`: specifies a finer-grained explanation that complements the eviction cause<br> priority_class: the priority class name |
+| `kueue_admission_wait_time_seconds`        | Histogram | The time between a workload was created or requeued until admission.                | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                         |
+| `kueue_admission_checks_wait_time_seconds` | Histogram | The time from when a workload got the quota reservation until admission.            | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                         |
+| `kueue_admitted_active_workloads`          | Gauge     | The number of admitted Workloads that are active (unsuspended and not finished)     | `cluster_queue`: the name of the ClusterQueue                                                                                                                                                     |
 | `kueue_cluster_queue_status`               | Gauge     | Reports the status of the ClusterQueue                                              | `cluster_queue`: The name of the ClusterQueue<br> `status`: Possible values are `pending`, `active` or `terminated`. For a ClusterQueue, the metric only reports a value of 1 for one of the statuses. |
-| `kueue_reserving_active_workloads`         | Gauge     | The number of Workloads that are reserving quota, per `cluster_queue`.              | `cluster_queue`: the name of the ClusterQueue                                                                                                                                                          |
+| `kueue_reserving_active_workloads`         | Gauge     | The number of Workloads that are reserving quota, per `cluster_queue`.              | `cluster_queue`: the name of the ClusterQueue                                                                                                                                                     |
 | `kueue_admission_cycle_preemption_skips`   | Gauge     | The number of Workloads in the ClusterQueue that got preemption candidates but had to be skipped because other ClusterQueues needed the same resources in the same cycle | `cluster_queue`: the name of the ClusterQueue                                                                     |
 | `kueue_preempted_workloads_total`          | Counter   | The number of preempted workloads per `preempting_cluster_queue`                    | `preempting_cluster_queue`: the name of the ClusterQueue<br> `reason`: possible values are `InClusterQueue` means that the workload was preempted by a workload in the same ClusterQueue; `InCohortReclamation` means that the workload was preempted by a workload in the same cohort due to reclamation of nominal quota; `InCohortFairSharing` means that the workload was preempted by a workload in the same cohort due to Fair Sharing; `InCohortReclaimWhileBorrowing` means that the workload was preempted by a workload in the same cohort due to reclamation of nominal quota while borrowing |
 
diff --git a/site/content/zh-CN/docs/reference/metrics.md b/site/content/zh-CN/docs/reference/metrics.md
index 6ef2e62d9..95d00324e 100644
--- a/site/content/zh-CN/docs/reference/metrics.md
+++ b/site/content/zh-CN/docs/reference/metrics.md
@@ -27,13 +27,13 @@ Use the following metrics to monitor the status of your ClusterQueues:
 | Metric name                                | Type      | Description                                                                         | Labels                                                                                                                                                                                                 |
 | -------------------------------------------- | ----------- | ------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `kueue_pending_workloads`                  | Gauge     | The number of pending workloads.                                                    | `cluster_queue`: the name of the ClusterQueue<br> `status`: possible values are `active` or `inadmissible`                                                                                             |
-| `kueue_quota_reserved_workloads_total`     | Counter   | The total number of quota reserved workloads.                                       | `cluster_queue`: the name of the ClusterQueue<br> workload_priority_class: the workload priority class name                                                                                            |
-| `kueue_quota_reserved_wait_time_seconds`   | Histogram | The time between a workload was created or requeued until it got quota reservation. | `cluster_queue`: the name of the ClusterQueue<br> workload_priority_class: the workload priority class name                                                                                            |
-| `kueue_admitted_workloads_total`           | Counter   | The total number of admitted workloads.                                             | `cluster_queue`: the name of the ClusterQueue<br> workload_priority_class: the workload priority class name                                                                                            |
-| `kueue_evicted_workloads_total`            | Counter   | The total number of evicted workloads.                                              | `cluster_queue`: the name of the ClusterQueue<br> `reason`: Possible values are `Preempted`, `PodsReadyTimeout`, `AdmissionCheck`, `ClusterQueueStopped` or `Deactivated`<br> workload_priority_class: the workload priority class name |
-| `kueue_evicted_workloads_once_total`       | Counter   | The number of unique workload evictions per 'cluster_queue'                         | `cluster_queue`: the name of the ClusterQueue<br> `reason`: Possible values are `Preempted`, `PodsReadyTimeout`, `AdmissionCheck`, `ClusterQueueStopped` or `Deactivated`<br> `detailedReason`: specifies a finer-grained explanation that complements the eviction cause<br> workload_priority_class: the workload priority class name |
-| `kueue_admission_wait_time_seconds`        | Histogram | The time between a workload was created or requeued until admission.                | `cluster_queue`: the name of the ClusterQueue<br> workload_priority_class: the workload priority class name                                                                                            |
-| `kueue_admission_checks_wait_time_seconds` | Histogram | The time from when a workload got the quota reservation until admission.            | `cluster_queue`: the name of the ClusterQueue                                                                                                                                                          |
+| `kueue_quota_reserved_workloads_total`     | Counter   | The total number of quota reserved workloads.                                       | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                            |
+| `kueue_quota_reserved_wait_time_seconds`   | Histogram | The time between a workload was created or requeued until it got quota reservation. | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                            |
+| `kueue_admitted_workloads_total`           | Counter   | The total number of admitted workloads.                                             | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                            |
+| `kueue_evicted_workloads_total`            | Counter   | The total number of evicted workloads.                                              | `cluster_queue`: the name of the ClusterQueue<br> `reason`: Possible values are `Preempted`, `PodsReadyTimeout`, `AdmissionCheck`, `ClusterQueueStopped` or `Deactivated`<br> priority_class: the priority class name |
+| `kueue_evicted_workloads_once_total`       | Counter   | The number of unique workload evictions per 'cluster_queue'                         | `cluster_queue`: the name of the ClusterQueue<br> `reason`: Possible values are `Preempted`, `PodsReadyTimeout`, `AdmissionCheck`, `ClusterQueueStopped` or `Deactivated`<br> `detailedReason`: specifies a finer-grained explanation that complements the eviction cause<br> priority_class: the priority class name |
+| `kueue_admission_wait_time_seconds`        | Histogram | The time between a workload was created or requeued until admission.                | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                            |
+| `kueue_admission_checks_wait_time_seconds` | Histogram | The time from when a workload got the quota reservation until admission.            | `cluster_queue`: the name of the ClusterQueue<br> priority_class: the priority class name                                                                                            |
 | `kueue_admitted_active_workloads`          | Gauge     | The number of admitted Workloads that are active (unsuspended and not finished)     | `cluster_queue`: the name of the ClusterQueue                                                                                                                                                          |
 | `kueue_cluster_queue_status`               | Gauge     | Reports the status of the ClusterQueue                                              | `cluster_queue`: The name of the ClusterQueue<br> `status`: Possible values are `pending`, `active` or `terminated`. For a ClusterQueue, the metric only reports a value of 1 for one of the statuses. |
 | `kueue_reserving_active_workloads`         | Gauge     | The number of Workloads that are reserving quota, per `cluster_queue`.              | `cluster_queue`: the name of the ClusterQueue                                                                                                                                                          |



Test patch:
diff --git a/pkg/webhooks/clusterqueue_webhook_test.go b/pkg/webhooks/clusterqueue_webhook_test.go
index 4a84114c7..f0a1fef94 100644
--- a/pkg/webhooks/clusterqueue_webhook_test.go
+++ b/pkg/webhooks/clusterqueue_webhook_test.go
@@ -17,11 +17,13 @@ limitations under the License.
 package webhooks
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/utils/ptr"
@@ -369,3 +371,142 @@ func TestValidateClusterQueueUpdate(t *testing.T) {
 		})
 	}
 }
+
+func makeFlavors(n int) []kueue.FlavorQuotas {
+	flavs := make([]kueue.FlavorQuotas, 0, n)
+	for i := range n {
+		flavs = append(flavs, kueue.FlavorQuotas{
+			Name: kueue.ResourceFlavorReference(fmt.Sprintf("f%03d", i)),
+			Resources: []kueue.ResourceQuota{{
+				Name:         corev1.ResourceCPU,
+				NominalQuota: resource.MustParse("1"),
+			}},
+		})
+	}
+	return flavs
+}
+
+func TestValidateTotalFlavors(t *testing.T) {
+	testcases := []struct {
+		name       string
+		numFlavors int
+		wantErr    bool
+	}{
+		{"within limit (10 flavors)", 10, false},
+		{"At limit (256 flavors)", 256, false},
+		{"over limit (257 flavors)", 257, true},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			cq := testingutil.MakeClusterQueue("cluster-queue").
+				ResourceGroup(makeFlavors(tc.numFlavors)...).Obj()
+
+			gotErr := ValidateClusterQueue(cq)
+
+			if diff := cmp.Diff(tc.wantErr, len(gotErr) > 0); diff != "" {
+				t.Errorf("Unexpected error (-want,+got):\n%s", diff)
+			}
+		})
+	}
+}
+
+func makeCoveredResources(n int) []kueue.ResourceGroup {
+	resources := make([]corev1.ResourceName, n)
+	quotas := make([]kueue.ResourceQuota, n)
+	for i := range n {
+		name := corev1.ResourceName(fmt.Sprintf("res%03d", i))
+		resources[i] = name
+		quotas[i] = kueue.ResourceQuota{
+			Name:         name,
+			NominalQuota: resource.MustParse("1"),
+		}
+	}
+
+	return []kueue.ResourceGroup{{
+		CoveredResources: resources,
+		Flavors: []kueue.FlavorQuotas{{
+			Name:      kueue.ResourceFlavorReference("default"),
+			Resources: quotas,
+		}},
+	}}
+}
+
+func TestValidateTotalCoveredResources(t *testing.T) {
+	testcases := []struct {
+		name          string
+		numCoveredRes int
+		wantErr       bool
+	}{
+		{"within limit (10 covered resources)", 10, false},
+		{"At limit (256 covered resources)", 256, false},
+		{"over limit (257 covered resources)", 257, true},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			cq := testingutil.MakeClusterQueue("cluster-queue").
+				Obj()
+			cq.Spec.ResourceGroups = makeCoveredResources(tc.numCoveredRes)
+
+			gotErr := ValidateClusterQueue(cq)
+
+			if diff := cmp.Diff(tc.wantErr, len(gotErr) > 0); diff != "" {
+				t.Errorf("Unexpected error (-want,+got):\n%s", diff)
+			}
+		})
+	}
+}
+
+func makeFlavorResourceCombinations(numFlavors, numResources int) []kueue.ResourceGroup {
+	flavors := make([]kueue.FlavorQuotas, 0, numFlavors)
+	resources := make([]corev1.ResourceName, numResources)
+	for j := range numResources {
+		resources[j] = corev1.ResourceName(fmt.Sprintf("res%03d", j))
+	}
+
+	for i := range numFlavors {
+		quotas := make([]kueue.ResourceQuota, numResources)
+		for j := range numResources {
+			quotas[j] = kueue.ResourceQuota{
+				Name:         resources[j],
+				NominalQuota: resource.MustParse("1"),
+			}
+		}
+		flavors = append(flavors, kueue.FlavorQuotas{
+			Name:      kueue.ResourceFlavorReference(fmt.Sprintf("f%03d", i)),
+			Resources: quotas,
+		})
+	}
+
+	return []kueue.ResourceGroup{{
+		CoveredResources: resources,
+		Flavors:          flavors,
+	}}
+}
+
+func TestValidateFlavorResourceCombinations(t *testing.T) {
+	testcases := []struct {
+		name         string
+		numFlavors   int
+		numResources int
+		wantErr      bool
+	}{
+		{"within limit (1 flavor × 10 resources = 10)", 1, 10, false},
+		{"at limit (8 flavors × 64 resources = 512)", 8, 64, false},
+		{"over limit (9 flavors × 64 resources = 576)", 9, 64, true},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			cq := testingutil.MakeClusterQueue("cluster-queue").Obj()
+			cq.Spec.ResourceGroups = makeFlavorResourceCombinations(tc.numFlavors, tc.numResources)
+
+			gotErr := ValidateClusterQueue(cq)
+
+			if diff := cmp.Diff(tc.wantErr, len(gotErr) > 0); diff != "" {
+				t.Errorf("Unexpected error (-want,+got):\n%s", diff)
+			}
+		})
+	}
+}
diff --git a/test/e2e/certmanager/suite_test.go b/test/e2e/certmanager/suite_test.go
index bda52e38f..17cf6f365 100644
--- a/test/e2e/certmanager/suite_test.go
+++ b/test/e2e/certmanager/suite_test.go
@@ -28,15 +28,17 @@ import (
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	visibilityv1beta1 "sigs.k8s.io/kueue/client-go/clientset/versioned/typed/visibility/v1beta1"
 	"sigs.k8s.io/kueue/test/util"
 )
 
 var (
-	k8sClient  client.WithWatch
-	ctx        context.Context
-	cfg        *rest.Config
-	restClient *rest.RESTClient
-	kueueNS    = util.GetKueueNamespace()
+	k8sClient        client.WithWatch
+	ctx              context.Context
+	cfg              *rest.Config
+	restClient       *rest.RESTClient
+	kueueNS          = util.GetKueueNamespace()
+	visibilityClient visibilityv1beta1.VisibilityV1beta1Interface
 )
 
 func TestAPIs(t *testing.T) {
@@ -57,6 +59,8 @@ var _ = ginkgo.BeforeSuite(func() {
 	k8sClient, cfg, err = util.CreateClientUsingCluster("")
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 	restClient = util.CreateRestClient(cfg)
+	visibilityClient, err = util.CreateVisibilityClient("")
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 	ctx = ginkgo.GinkgoT().Context()
 
 	waitForAvailableStart := time.Now()
diff --git a/test/e2e/certmanager/visibility_test.go b/test/e2e/certmanager/visibility_test.go
new file mode 100644
index 000000000..e3c865fec
--- /dev/null
+++ b/test/e2e/certmanager/visibility_test.go
@@ -0,0 +1,135 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed 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.
+*/
+
+package certmanager
+
+import (
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
+	"sigs.k8s.io/kueue/pkg/util/testing"
+	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
+	"sigs.k8s.io/kueue/test/util"
+)
+
+var _ = ginkgo.Describe("Kueue secure visibility server", func() {
+	const defaultFlavor = "default-flavor"
+
+	var (
+		defaultRF    *kueue.ResourceFlavor
+		localQueue   *kueue.LocalQueue
+		clusterQueue *kueue.ClusterQueue
+		ns           *corev1.Namespace
+		firstJob     *batchv1.Job
+		secondJob    *batchv1.Job
+	)
+
+	ginkgo.BeforeEach(func() {
+		ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "e2e-")
+		defaultRF = testing.MakeResourceFlavor(defaultFlavor).Obj()
+		util.MustCreate(ctx, k8sClient, defaultRF)
+
+		clusterQueue = testing.MakeClusterQueue("cluster-queue").
+			ResourceGroup(
+				*testing.MakeFlavorQuotas(defaultFlavor).
+					Resource(corev1.ResourceCPU, "1").
+					Obj(),
+			).
+			Obj()
+		util.MustCreate(ctx, k8sClient, clusterQueue)
+
+		localQueue = testing.MakeLocalQueue("local-queue", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+		util.MustCreate(ctx, k8sClient, localQueue)
+	})
+	ginkgo.AfterEach(func() {
+		gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+		util.ExpectAllPodsInNamespaceDeleted(ctx, k8sClient, ns)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, localQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+		util.ExpectObjectToBeDeleted(ctx, k8sClient, defaultRF, true)
+	})
+
+	ginkgo.When("Capacity is maxed by the admitted job", func() {
+		ginkgo.It("Should allow fetching information about pending workloads in ClusterQueue", func() {
+			ginkgo.By("Schedule a job that maxes out the cluster queue", func() {
+				firstJob = testingjob.MakeJob("job-1", ns.Name).
+					Queue(kueue.LocalQueueName(localQueue.Name)).
+					Image(util.GetAgnHostImage(), util.BehaviorWaitForDeletion).
+					RequestAndLimit(corev1.ResourceCPU, "1").
+					TerminationGracePeriod(1).
+					BackoffLimit(0).
+					Obj()
+				util.MustCreate(ctx, k8sClient, firstJob)
+			})
+			ginkgo.By("Wait for the job to be unsuspended", func() {
+				job := &batchv1.Job{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(firstJob), job)).To(gomega.Succeed())
+					g.Expect(job.Spec.Suspend).Should(gomega.Equal(ptr.To(false)))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Verify there are no pending workloads", func() {
+				info, err := visibilityClient.ClusterQueues().GetPendingWorkloadsSummary(ctx, clusterQueue.Name, metav1.GetOptions{})
+				gomega.Expect(err).NotTo(gomega.HaveOccurred())
+				gomega.Expect(info.Items).Should(gomega.BeEmpty())
+			})
+
+			ginkgo.By("Schedule a job which is pending due to low quota", func() {
+				secondJob = testingjob.MakeJob("job-2", ns.Name).
+					Queue(kueue.LocalQueueName(localQueue.Name)).
+					Image(util.GetAgnHostImage(), util.BehaviorExitFast).
+					RequestAndLimit(corev1.ResourceCPU, "1").
+					Obj()
+				util.MustCreate(ctx, k8sClient, secondJob)
+			})
+
+			ginkgo.By("Verify there is one pending workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					info, err := visibilityClient.ClusterQueues().GetPendingWorkloadsSummary(ctx, clusterQueue.Name, metav1.GetOptions{})
+					g.Expect(err).NotTo(gomega.HaveOccurred())
+					g.Expect(info.Items).Should(gomega.HaveLen(1))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Delete the first job to release the quota", func() {
+				gomega.Expect(util.DeleteAllPodsInNamespace(ctx, k8sClient, ns)).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("verify second job is not suspended", func() {
+				job := &batchv1.Job{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(secondJob), job)).To(gomega.Succeed())
+					g.Expect(job.Spec.Suspend).Should(gomega.Equal(ptr.To(false)))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Verify there are no pending workloads, once the last workload is admitted", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					info, err := visibilityClient.ClusterQueues().GetPendingWorkloadsSummary(ctx, clusterQueue.Name, metav1.GetOptions{})
+					g.Expect(err).NotTo(gomega.HaveOccurred())
+					g.Expect(info.Items).Should(gomega.BeEmpty())
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+})
diff --git a/test/e2e/config/certmanager/kustomization.yaml b/test/e2e/config/certmanager/kustomization.yaml
index 66a3060ac..608035735 100644
--- a/test/e2e/config/certmanager/kustomization.yaml
+++ b/test/e2e/config/certmanager/kustomization.yaml
@@ -71,3 +71,29 @@ vars:
     kind: Certificate
     name: serving-cert
     version: v1
+- name: VISIBILITY_SERVICE_NAME
+  objref:
+    kind: Service
+    version: v1
+    name: visibility-server
+- name: VISIBILITY_SERVICE_NAMESPACE
+  objref:
+    kind: Service
+    name: visibility-server
+    version: v1
+  fieldref:
+    fieldPath: metadata.namespace
+- name: VISIBILITY_CERTIFICATE_NAMESPACE
+  objref:
+    kind: Certificate
+    group: cert-manager.io
+    version: v1
+    name: visibility-server-cert # this name should match the one in certificate-visibility-server.yaml
+  fieldref:
+    fieldpath: metadata.namespace
+- name: VISIBILITY_CERTIFICATE_NAME
+  objref:
+    kind: Certificate
+    group: cert-manager.io
+    version: v1
+    name: visibility-server-cert # this name should match the one in certificate-visibility-server.yaml
diff --git a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
index 7d65d4559..dba6b6d54 100644
--- a/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
+++ b/test/integration/singlecluster/controller/jobs/job/job_controller_test.go
@@ -2140,6 +2140,146 @@ var _ = ginkgo.Describe("Interacting with scheduler", ginkgo.Ordered, ginkgo.Con
 			util.ExpectAdmittedWorkloadsTotalMetric(prodClusterQ, "", 1)
 		})
 	})
+
+	ginkgo.When("the cluster queue has admission check and workload priority class", func() {
+		var (
+			highPriorityClass *kueue.WorkloadPriorityClass
+			lowPriorityClass  *kueue.WorkloadPriorityClass
+			admissionCheck    *kueue.AdmissionCheck
+			resourceFlavor    *kueue.ResourceFlavor
+			clusterQueue      *kueue.ClusterQueue
+			localQueue        *kueue.LocalQueue
+		)
+
+		ginkgo.BeforeEach(func() {
+			highPriorityClass = testing.MakeWorkloadPriorityClass("high").PriorityValue(highPriorityValue).Obj()
+			util.MustCreate(ctx, k8sClient, highPriorityClass)
+
+			lowPriorityClass = testing.MakeWorkloadPriorityClass("low").PriorityValue(priorityValue).Obj()
+			util.MustCreate(ctx, k8sClient, lowPriorityClass)
+
+			admissionCheck = testing.MakeAdmissionCheck("ac").ControllerName("ac-controller").Obj()
+			util.MustCreate(ctx, k8sClient, admissionCheck)
+
+			util.SetAdmissionCheckActive(ctx, k8sClient, admissionCheck, metav1.ConditionTrue)
+
+			resourceFlavor = testing.MakeResourceFlavor("rf").NodeLabel(instanceKey, "on-demand").Obj()
+			util.MustCreate(ctx, k8sClient, resourceFlavor)
+
+			clusterQueue = testing.MakeClusterQueue("cq").
+				ResourceGroup(
+					*testing.MakeFlavorQuotas(resourceFlavor.Name).Resource(corev1.ResourceCPU, "5").Obj(),
+				).
+				AdmissionChecks("ac").
+				Preemption(kueue.ClusterQueuePreemption{
+					WithinClusterQueue: kueue.PreemptionPolicyLowerPriority,
+				}).
+				Obj()
+			util.MustCreate(ctx, k8sClient, clusterQueue)
+
+			localQueue = testing.MakeLocalQueue("lq", ns.Name).ClusterQueue(clusterQueue.Name).Obj()
+			util.MustCreate(ctx, k8sClient, localQueue)
+		})
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, localQueue, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, admissionCheck, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, lowPriorityClass, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, highPriorityClass, true)
+		})
+
+		ginkgo.It("Should readmit preempted Job with priorityClass in alternative flavor with admission check", func() {
+			lowJob := testingjob.MakeJob("low", ns.Name).
+				Queue(kueue.LocalQueueName(localQueue.Name)).
+				WorkloadPriorityClass(lowPriorityClass.Name).
+				Parallelism(5).
+				Request(corev1.ResourceCPU, "1").
+				Suspend(false).
+				Obj()
+			ginkgo.By("Creating a low priority job", func() {
+				util.MustCreate(ctx, k8sClient, lowJob)
+			})
+
+			ginkgo.By("Verifying the low priority job is suspended", func() {
+				createdJob := &batchv1.Job{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(lowJob), createdJob)).Should(gomega.Succeed())
+					g.Expect(createdJob.Spec.Suspend).Should(gomega.Equal(ptr.To(true)))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			createdLowWorkload := &kueue.Workload{}
+			lowWlLookupKey := types.NamespacedName{
+				Name:      workloadjob.GetWorkloadNameForJob(lowJob.Name, lowJob.UID),
+				Namespace: ns.Name,
+			}
+			ginkgo.By("Fetching the created low priority workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, lowWlLookupKey, createdLowWorkload)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Setting ready for admission checks in low priority workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, lowWlLookupKey, createdLowWorkload)).To(gomega.Succeed())
+					workload.SetAdmissionCheckState(&createdLowWorkload.Status.AdmissionChecks, kueue.AdmissionCheckState{
+						Name:  kueue.AdmissionCheckReference(admissionCheck.Name),
+						State: kueue.CheckStateReady,
+					}, clock.RealClock{})
+					g.Expect(k8sClient.Status().Update(ctx, createdLowWorkload)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			util.ExpectAdmissionChecksWaitTimeMetric(clusterQueue, lowPriorityClass.Name, 1)
+			util.ExpectAdmissionChecksWaitTimeMetric(clusterQueue, highPriorityClass.Name, 0)
+
+			highJob := testingjob.MakeJob("high", ns.Name).
+				Queue(kueue.LocalQueueName(localQueue.Name)).
+				WorkloadPriorityClass(highPriorityClass.Name).
+				Parallelism(5).
+				Request(corev1.ResourceCPU, "1").
+				Suspend(false).
+				Obj()
+			ginkgo.By("Creating a high priority job", func() {
+				util.MustCreate(ctx, k8sClient, highJob)
+			})
+
+			ginkgo.By("Verifying the high priority job is suspended", func() {
+				createdJob := &batchv1.Job{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(highJob), createdJob)).Should(gomega.Succeed())
+					g.Expect(createdJob.Spec.Suspend).Should(gomega.Equal(ptr.To(true)))
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			createdHighWorkload := &kueue.Workload{}
+			highWlLookupKey := types.NamespacedName{
+				Name:      workloadjob.GetWorkloadNameForJob(highJob.Name, highJob.UID),
+				Namespace: ns.Name,
+			}
+			ginkgo.By("Fetching the created high priority workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, highWlLookupKey, createdHighWorkload)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Setting ready for admission checks in high priority workload", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, highWlLookupKey, createdHighWorkload)).To(gomega.Succeed())
+					workload.SetAdmissionCheckState(&createdHighWorkload.Status.AdmissionChecks, kueue.AdmissionCheckState{
+						Name:  kueue.AdmissionCheckReference(admissionCheck.Name),
+						State: kueue.CheckStateReady,
+					}, clock.RealClock{})
+					g.Expect(k8sClient.Status().Update(ctx, createdHighWorkload)).Should(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			util.ExpectAdmissionChecksWaitTimeMetric(clusterQueue, lowPriorityClass.Name, 1)
+			util.ExpectAdmissionChecksWaitTimeMetric(clusterQueue, highPriorityClass.Name, 1)
+		})
+	})
 })
 
 var _ = ginkgo.Describe("Job controller interacting with Workload controller when waitForPodsReady is enabled", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
diff --git a/test/integration/singlecluster/webhook/core/clusterqueue_test.go b/test/integration/singlecluster/webhook/core/clusterqueue_test.go
index 0edcfb631..b84b62e0a 100644
--- a/test/integration/singlecluster/webhook/core/clusterqueue_test.go
+++ b/test/integration/singlecluster/webhook/core/clusterqueue_test.go
@@ -39,7 +39,7 @@ import (
 
 const (
 	resourcesMaxItems = 16
-	flavorsMaxItems   = 16
+	flavorsMaxItems   = 64
 )
 
 var _ = ginkgo.Describe("ClusterQueue Webhook", ginkgo.Ordered, func() {
diff --git a/test/integration/singlecluster/webhook/core/cohort_test.go b/test/integration/singlecluster/webhook/core/cohort_test.go
index cce83c3a6..841957e6a 100644
--- a/test/integration/singlecluster/webhook/core/cohort_test.go
+++ b/test/integration/singlecluster/webhook/core/cohort_test.go
@@ -18,6 +18,7 @@ package core
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
@@ -84,45 +85,30 @@ var _ = ginkgo.Describe("Cohort Webhook", ginkgo.Ordered, func() {
 					Obj(),
 				testing.BeForbiddenError()),
 			ginkgo.Entry("Should reject too many flavors in resource group",
-				testing.MakeCohort("cohort").ResourceGroup(
-					*testing.MakeFlavorQuotas("f0").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f1").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f2").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f3").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f4").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f5").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f6").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f7").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f8").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f9").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f10").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f11").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f12").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f13").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f14").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f15").Resource(corev1.ResourceCPU).Obj(),
-					*testing.MakeFlavorQuotas("f16").Resource(corev1.ResourceCPU).Obj()).Obj(),
+				func() *kueue.Cohort {
+					var flavors []kueue.FlavorQuotas
+					for i := range 65 {
+						flavors = append(flavors,
+							*testing.MakeFlavorQuotas(fmt.Sprintf("f%d", i)).
+								Resource(corev1.ResourceCPU).
+								Obj(),
+						)
+					}
+					return testing.MakeCohort("cohort").
+						ResourceGroup(flavors...).
+						Obj()
+				}(),
 				testing.BeInvalidError()),
 			ginkgo.Entry("Should reject too many resources in resource group",
-				testing.MakeCohort("cohort").ResourceGroup(
-					*testing.MakeFlavorQuotas("flavor").
-						Resource("cpu0").
-						Resource("cpu1").
-						Resource("cpu2").
-						Resource("cpu3").
-						Resource("cpu4").
-						Resource("cpu5").
-						Resource("cpu6").
-						Resource("cpu7").
-						Resource("cpu8").
-						Resource("cpu9").
-						Resource("cpu10").
-						Resource("cpu11").
-						Resource("cpu12").
-						Resource("cpu13").
-						Resource("cpu14").
-						Resource("cpu15").
-						Resource("cpu16").Obj()).Obj(),
+				func() *kueue.Cohort {
+					fq := testing.MakeFlavorQuotas("flavor")
+					for i := range 17 {
+						fq = fq.Resource(corev1.ResourceName(fmt.Sprintf("cpu%d", i)))
+					}
+					return testing.MakeCohort("cohort").
+						ResourceGroup(*fq.Obj()).
+						Obj()
+				}(),
 				testing.BeInvalidError()),
 			ginkgo.Entry("Should allow resource with valid name",
 				testing.MakeCohort("cohort").
diff --git a/test/util/util.go b/test/util/util.go
index 059303e4d..9031dccec 100644
--- a/test/util/util.go
+++ b/test/util/util.go
@@ -582,6 +582,15 @@ func ExpectAdmissionWaitTimeMetric(cq *kueue.ClusterQueue, workloadPriorityClass
 	}, Timeout, Interval).Should(gomega.Succeed())
 }
 
+func ExpectAdmissionChecksWaitTimeMetric(cq *kueue.ClusterQueue, workloadPriorityClass string, count int) {
+	gomega.EventuallyWithOffset(1, func(g gomega.Gomega) {
+		metric := metrics.AdmissionChecksWaitTime.WithLabelValues(cq.Name, workloadPriorityClass)
+		v, err := testutil.GetHistogramMetricCount(metric)
+		g.Expect(err).ToNot(gomega.HaveOccurred())
+		g.Expect(int(v)).Should(gomega.Equal(count))
+	}, Timeout, Interval).Should(gomega.Succeed())
+}
+
 func ExpectEvictedWorkloadsTotalMetric(cqName, reason, underlyingCause, workloadPriorityClass string, v int) {
 	metric := metrics.EvictedWorkloadsTotal.WithLabelValues(cqName, reason, underlyingCause, workloadPriorityClass)
 	expectCounterMetric(metric, v)