Repository: kubernetes-sigs/kueue

Issue:
Flaky Integration Test: Scheduler when Preemption is enabled in fairsharing and there are large values of quota and weights Queue can reclaim its nominal quota
<!-- 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**:

`Scheduler Fair Sharing Suite: [It] Scheduler when Preemption is enabled in fairsharing and there are large values of quota and weights Queue can reclaim its nominal quota [slow]` failed in release-0.12 periodic Job.

```shell
{Timed out after 10.000s.
The function passed to Eventually failed at /home/prow/go/src/kubernetes-sigs/kueue/test/util/util_scheduling.go:67 with:
Not enough workloads evicted
Expected
    <int>: 0
to equal
    <int>: 3 failed [FAILED] Timed out after 10.000s.
The function passed to Eventually failed at /home/prow/go/src/kubernetes-sigs/kueue/test/util/util_scheduling.go:67 with:
Not enough workloads evicted
Expected
    <int>: 0
to equal
    <int>: 3
In [It] at: /home/prow/go/src/kubernetes-sigs/kueue/test/integration/singlecluster/scheduler/fairsharing/fair_sharing_test.go:499 @ 09/13/25 07:51:18.383
}
```

**What you expected to happen**:

No errors.

**How to reproduce it (as minimally and precisely as possible)**:

<img width="3168" height="482" alt="Image" src="https://github.com/user-attachments/assets/7259e200-1e88-4336-b761-da6a3d415555" />

https://prow.k8s.io/view/gs/kubernetes-ci-logs/logs/periodic-kueue-test-integration-release-0-12/1966767439517585408

**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:
@tenzen-y: /kind flake

IIUC, we added this case in https://github.com/kubernetes-sigs/kueue/pull/6617

cc @pajakd @PBundyra @mimowo: https://prow.k8s.io/view/gs/kubernetes-ci-logs/logs/periodic-kueue-test-integration-release-0-12/1969848044652335104@pajakd: /assign@mbobrovskyi: It also happens in main: https://prow.k8s.io/view/gs/kubernetes-ci-logs/pr-logs/pull/kubernetes-sigs_kueue/7122/pull-kueue-test-integration-extended-main/1973390165825032192.@mbobrovskyi: /retitle Flaky Integration Test: Scheduler when Preemption is enabled in fairsharing and there are large values of quota and weights Queue can reclaim its nominal quota@pajakd: I looked at the logs and I think there is the following race condition:
https://github.com/kubernetes-sigs/kueue/blob/0d6a4aee95cb221f172cb82b88d6ebb68f1c57e0/test/integration/singlecluster/scheduler/fairsharing/fair_sharing_test.go#L837-L846

- The last of the 10 workloads in queue "a" is created and reserves the quota.
- Before it gets admitted, the workload in queue "b" gets created. 
- I can see in the logs that the workload in "b" evicts 2 workloads (for some reason the error in the test failure says 0 evictions).

IMO to deflake the test we should wait for the actual admission of the 10 workloads.@mimowo: Is it becuase the metric `ReservingActiveWorkloadsMetric` is bumped before the workloads are actually reserving the quota? 

If this is the case, then it looks like an issue on its own, as we try to bump metrics only once the transition happens. 

Still, the metrics fix may not be trivial and the issue is likely low priority. Finally, testing the metric is not the intention of the test.

So, +1 on the pragmatic flakiness fix.

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/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/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/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 5de77bd60..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.ReportAdmissionChecksWaitTime(cqName, workload.GetWorkloadPriorityClass(&wl), 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 a3d2b8eca..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(
@@ -243,7 +243,7 @@ The label 'underlying_cause' can have the following values:
 			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", "workload_priority_class"},
+		}, []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) {
@@ -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) {
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index 362be632a..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.ReportAdmissionChecksWaitTime(admission.ClusterQueue, workload.GetWorkloadPriorityClass(newWorkload), 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/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/metrics.md b/site/content/en/docs/reference/metrics.md
index 161d663af..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<br> workload_priority_class: the workload 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                                                                                                                                                          |
+| 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/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/tas/tas_test.go b/test/integration/singlecluster/tas/tas_test.go
index b87cd71d1..443648fe2 100644
--- a/test/integration/singlecluster/tas/tas_test.go
+++ b/test/integration/singlecluster/tas/tas_test.go
@@ -37,6 +37,7 @@ import (
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/controller/admissionchecks/provisioning"
 	"sigs.k8s.io/kueue/pkg/controller/tas"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/admissioncheck"
 	utiltas "sigs.k8s.io/kueue/pkg/util/tas"
 	"sigs.k8s.io/kueue/pkg/util/testing"
@@ -1139,7 +1140,8 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 				})
 			})
 
-			ginkgo.It("should update workload TopologyAssignment after a node recovers", func() {
+			ginkgo.It("should update workload TopologyAssignment after a node becomes available", func() {
+				features.SetFeatureGateDuringTest(ginkgo.GinkgoTB(), features.TASFailedNodeReplacementFailFast, false)
 				var wl1, wl2 *kueue.Workload
 				nodeName := nodes[0].Name
 
@@ -1225,8 +1227,8 @@ var _ = ginkgo.Describe("Topology Aware Scheduling", ginkgo.Ordered, func() {
 							&kueue.TopologyAssignment{
 								Levels: []string{corev1.LabelHostname},
 								Domains: []kueue.TopologyDomainAssignment{
-									{Count: 1, Values: []string{"x1"}},
 									{Count: 1, Values: []string{"x2"}},
+									{Count: 1, Values: []string{"x3"}},
 								},
 							},
 						))