Repository: kubernetes-sigs/kueue

Issue:
Securing APIService for Visibility Feature
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:
https://github.com/kubernetes-sigs/kueue/blob/main/config/components/visibility/apiservice_v1beta1.yaml

APIService skips TLS checks so we will not enable this feature for our cluster.

Kueue should provide support for securing this service using certificates.

**Why is this needed**:

For my organization, any endpoint not guarded by TLS or using insecure options is flagged as a potential security vulnerability. Since this is an optional feature we are not deploying this in our Kueue instalation. It would be a good feature to have.

**Completion requirements**:

This enhancement requires the following artifacts:

- [ ] Design doc
- [ ] API change
- [ ] Docs update

The artifacts should be linked in subsequent comments.


Discussion:
@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@tenzen-y: /remove-lifecycle stale@kannon92: /help
@k8s-ci-robot: @kannon92: 
	This request has been marked as needing help from a contributor.

### Guidelines
Please ensure that the issue body includes answers to the following questions:
- Why are we solving this issue?
- To address this issue, are there any code changes? If there are code changes, what needs to be done in the code and what places can the assignee treat as reference points?
- How can the assignee reach out to you for help?


For more details on the requirements of such an issue, please see [here](https://www.kubernetes.dev/docs/guide/help-wanted/) and ensure that they are met.

If this request no longer meets these requirements, the label can be removed
by commenting with the `/remove-help` command.


<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/4433):

>/help
>


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@MaysaMacedo: /assign@MaysaMacedo: I would like to help with this issue if it's fine by everyone.@tenzen-y: > I would like to help with this issue if it's fine by everyone.

+1

Thank you for your help!@MaysaMacedo: @kannon92 hello,
would you say that the following would be the requirements for this issue?

- Generate the certificate like it's done for other services https://github.com/openshift/kubernetes-sigs-kueue/blob/main/config/components/certmanager/certificate-metrics.yaml
-  Adapt `controller-manager` manifest to include the new certificate that exist in a secret as a volume
- Adapt `APIService` to include [caBundle](https://issues.redhat.com/browse/OCPNODE-3624#apiservice-apiregistration-k8s-io-v1) from the certificate generated in the secret https://cert-manager.io/docs/concepts/ca-injector/
- Adapt helm related charts
@kannon92: That sounds like a good plan.

@tenzen-y @mimowo should we include e2e test relying on Cert Manager for this also?@kannon92: I would also add a bullet point for Helm related charts also.@tenzen-y: This was reached by https://github.com/kubernetes-sigs/kueue/pull/6798
@MaysaMacedo Thanks!
/close
@k8s-ci-robot: @tenzen-y: Closing this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/4433#issuecomment-3323968233):

>This was reached by https://github.com/kubernetes-sigs/kueue/pull/6798
>@MaysaMacedo Thanks!
>/close
>


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>

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/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/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/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 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/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/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)