fix: ensure conversion webhook CA bundles are injected during upgrades

During upgrades with pre-existing LQ/CQ, the API server fails conversion
with 'x509: certificate signed by unknown authority' because cert-controller
doesn't inject CA bundles into CRD conversion webhook configurations.

This fix manually injects CA bundles into kueue.x-k8s.io CRDs, registers
webhooks before manager start so conversion endpoints are ready when caches
sync, and sets publishNotReadyAddresses: true to allow API server access
during startup.

Fixes #7344

Author: sohankunkerkar

charts/kueue/README.md

@@ -162,4 +162,5 @@ The following table lists the configurable parameters of the kueue chart and the
 | webhookService.ipDualStack.ipFamilies | list | `["IPv6","IPv4"]` | webhookService's ipDualStack ipFamilies |
 | webhookService.ipDualStack.ipFamilyPolicy | string | `"PreferDualStack"` | webhookService's ipDualStack ipFamilyPolicy |
 | webhookService.ports | list | `[{"port":443,"protocol":"TCP","targetPort":9443}]` | webhookService's ports |
+| webhookService.publishNotReadyAddresses | bool | `true` | webhookService's publishNotReadyAddresses ensures conversion traffic reaches pods before they are Ready |
 | webhookService.type | string | `"ClusterIP"` | webhookService's type |

charts/kueue/templates/rbac/role.yaml

@@ -38,6 +38,7 @@ rules:
       - limitranges
       - namespaces
       - nodes
+      - secrets
     verbs:
       - get
       - list

charts/kueue/templates/webhook/service.yaml

@@ -24,6 +24,7 @@ metadata:
   name: '{{ include "kueue.fullname" . }}-webhook-service'
   namespace: '{{ .Release.Namespace }}'
 spec:
+  publishNotReadyAddresses: true
   type: '{{ .Values.webhookService.type }}'
   {{- if .Values.webhookService.ipDualStack.enabled }}
   ipFamilies: {{ toYaml .Values.webhookService.ipDualStack.ipFamilies | nindent 4 }}

charts/kueue/values.yaml

@@ -200,6 +200,8 @@ webhookService:
     ipFamilies: ["IPv6", "IPv4"]
     # -- webhookService's ipDualStack ipFamilyPolicy
     ipFamilyPolicy: "PreferDualStack"
+  # -- webhookService's publishNotReadyAddresses ensures conversion traffic reaches pods before they are Ready
+  publishNotReadyAddresses: true
   # -- webhookService's ports
   ports:
     - port: 443

cmd/kueue/main.go

@@ -265,6 +265,13 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Register webhooks before starting the manager so conversion webhooks
+	// are available when caches start syncing.
+	if failedWebhook, err := webhooks.Setup(mgr); err != nil {
+		setupLog.Error(err, "unable to create webhook", "webhook", failedWebhook)
+		os.Exit(1)
+	}
+
 	// Cert won't be ready until manager starts, so start a goroutine here which
 	// will block until the cert is ready before setting up the controllers.
 	// Controllers who register after manager starts will start directly.
@@ -395,10 +402,6 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, cCache *schdcache.C
 		}
 	}
 
-	if failedWebhook, err := webhooks.Setup(mgr); err != nil {
-		return fmt.Errorf("unable to create webhook %s: %w", failedWebhook, err)
-	}
-
 	opts := []jobframework.Option{
 		jobframework.WithManageJobsWithoutQueueName(cfg.ManageJobsWithoutQueueName),
 		jobframework.WithWaitForPodsReady(cfg.WaitForPodsReady),

config/components/rbac/role.yaml

@@ -19,6 +19,7 @@ rules:
   - limitranges
   - namespaces
   - nodes
+  - secrets
   verbs:
   - get
   - list

config/components/webhook/service.yaml

@@ -4,6 +4,7 @@ metadata:
   name: webhook-service
   namespace: system
 spec:
+  publishNotReadyAddresses: true
   ports:
     - port: 443
       protocol: TCP

go.mod

@@ -25,6 +25,7 @@ require (
 	golang.org/x/exp v0.0.0-20250718183923-645b1fa84792
 	golang.org/x/sync v0.18.0
 	k8s.io/api v0.34.1
+	k8s.io/apiextensions-apiserver v0.34.1
 	k8s.io/apimachinery v0.34.1
 	k8s.io/apiserver v0.34.1
 	k8s.io/autoscaler/cluster-autoscaler/apis v0.0.0-20240830133931-adfbe7e45277
@@ -144,7 +145,6 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/apiextensions-apiserver v0.34.1 // indirect
 	k8s.io/code-generator v0.34.1 // indirect
 	k8s.io/gengo/v2 v2.0.0-20250820003526-c297c0c1eb9d // indirect
 	k8s.io/kms v0.34.1 // indirect

pkg/util/cert/cert.go

@@ -17,23 +17,37 @@ limitations under the License.
 package cert
 
 import (
+	"bytes"
+	"context"
+	"errors"
 	"fmt"
+	"sort"
 	"strings"
+	"time"
 
 	"github.com/go-logr/logr"
 	cert "github.com/open-policy-agent/cert-controller/pkg/rotator"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/util/retry"
 	ctrl "sigs.k8s.io/controller-runtime"
 
 	config "sigs.k8s.io/kueue/apis/config/v1beta2"
 )
 
 const (
-	caName               = "kueue-ca"
-	caOrganization       = "kueue"
-	webhookServiceSuffix = "-webhook-service"
+	caName                 = "kueue-ca"
+	caOrganization         = "kueue"
+	webhookServiceSuffix   = "-webhook-service"
+	conversionGroup        = "kueue.x-k8s.io"
+	conversionQueryTimeout = 10 * time.Second
 )
 
+// +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch
 // +kubebuilder:rbac:groups="admissionregistration.k8s.io",resources=mutatingwebhookconfigurations,verbs=get;list;watch;update
 // +kubebuilder:rbac:groups="admissionregistration.k8s.io",resources=validatingwebhookconfigurations,verbs=get;list;watch;update
 // +kubebuilder:rbac:groups="apiextensions.k8s.io",resources=customresourcedefinitions,verbs=get;list;watch;update
@@ -48,6 +62,28 @@ func ManageCerts(mgr ctrl.Manager, cfg config.Configuration, setupFinished chan
 	mutatingWebhookName := buildWebhookConfigurationName(webhookBaseName, "mutating")
 	validatingWebhookName := buildWebhookConfigurationName(webhookBaseName, "validating")
 
+	webhooks := []cert.WebhookInfo{{
+		Type: cert.Validating,
+		Name: validatingWebhookName,
+	}, {
+		Type: cert.Mutating,
+		Name: mutatingWebhookName,
+	}}
+
+	ctx, cancel := context.WithTimeout(context.Background(), conversionQueryTimeout)
+	defer cancel()
+	conversionCRDs := discoverConversionCRDs(ctx, mgr)
+	conversionReady := ensureConversionCRDCABundles(mgr, cfg, conversionCRDs)
+
+	internalReady := make(chan struct{})
+	go func() {
+		<-internalReady
+		if conversionReady != nil {
+			<-conversionReady
+		}
+		close(setupFinished)
+	}()
+
 	return cert.AddRotator(mgr, &cert.CertRotator{
 		SecretKey: types.NamespacedName{
 			Namespace: *cfg.Namespace,
@@ -57,23 +93,8 @@ func ManageCerts(mgr ctrl.Manager, cfg config.Configuration, setupFinished chan
 		CAName:         caName,
 		CAOrganization: caOrganization,
 		DNSName:        dnsName,
-		IsReady:        setupFinished,
-		Webhooks: []cert.WebhookInfo{{
-			Type: cert.Validating,
-			Name: validatingWebhookName,
-		}, {
-			Type: cert.Mutating,
-			Name: mutatingWebhookName,
-		}, {
-			Type: cert.CRDConversion,
-			Name: "localqueues.kueue.x-k8s.io",
-		}, {
-			Type: cert.CRDConversion,
-			Name: "clusterqueues.kueue.x-k8s.io",
-		}, {
-			Type: cert.CRDConversion,
-			Name: "workloads.kueue.x-k8s.io",
-		}},
+		IsReady:        internalReady,
+		Webhooks:       webhooks,
 		// When kueue is running in the leader election mode,
 		// we expect webhook server will run in primary and secondary instance
 		RequireLeaderElection: false,
@@ -96,3 +117,119 @@ func deriveWebhookBaseName(webhookServiceName string) string {
 func buildWebhookConfigurationName(baseName, webhookType string) string {
 	return fmt.Sprintf("%s-%s-webhook-configuration", baseName, webhookType)
 }
+
+func discoverConversionCRDs(ctx context.Context, mgr ctrl.Manager) []string {
+	log := ctrl.Log.WithName("cert")
+	clientset, err := apiextensionsclient.NewForConfig(mgr.GetConfig())
+	if err != nil {
+		log.Error(err, "failed to create clientset")
+		return nil
+	}
+
+	crds, err := clientset.ApiextensionsV1().CustomResourceDefinitions().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		log.Error(err, "failed to list CRDs")
+		return nil
+	}
+
+	var names []string
+	for _, crd := range crds.Items {
+		if crd.Spec.Group == conversionGroup &&
+			crd.Spec.Conversion != nil &&
+			crd.Spec.Conversion.Strategy == apiextensionsv1.WebhookConverter &&
+			crd.Spec.Conversion.Webhook != nil &&
+			crd.Spec.Conversion.Webhook.ClientConfig != nil {
+			names = append(names, crd.Name)
+		}
+	}
+	sort.Strings(names)
+	log.Info("Discovered conversion CRDs", "count", len(names), "names", names)
+	return names
+}
+
+func ensureConversionCRDCABundles(mgr ctrl.Manager, cfg config.Configuration, crdNames []string) <-chan struct{} {
+	if len(crdNames) == 0 {
+		return nil
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		log := ctrl.Log.WithName("cert")
+
+		caData, err := waitForCAData(mgr, cfg, log)
+		if err != nil {
+			return
+		}
+
+		if err := injectCABundlesIntoCRDs(mgr, crdNames, caData, log); err != nil {
+			log.Error(err, "failed to inject CA bundles")
+		}
+	}()
+	return done
+}
+
+func waitForCAData(mgr ctrl.Manager, cfg config.Configuration, log logr.Logger) ([]byte, error) {
+	if cfg.InternalCertManagement == nil || cfg.InternalCertManagement.WebhookSecretName == nil || cfg.Namespace == nil {
+		return nil, errors.New("cert management disabled")
+	}
+
+	coreClient, err := kubernetes.NewForConfig(mgr.GetConfig())
+	if err != nil {
+		log.Error(err, "failed to create clientset")
+		return nil, err
+	}
+
+	secretKey := types.NamespacedName{
+		Namespace: *cfg.Namespace,
+		Name:      *cfg.InternalCertManagement.WebhookSecretName,
+	}
+
+	for range [30]int{} {
+		secret, err := coreClient.CoreV1().Secrets(secretKey.Namespace).Get(context.Background(), secretKey.Name, metav1.GetOptions{})
+		if err == nil {
+			if data, ok := secret.Data["ca.crt"]; ok && len(data) > 0 {
+				return append([]byte(nil), data...), nil
+			}
+		} else if !apierrors.IsNotFound(err) {
+			log.Error(err, "failed to fetch secret")
+			return nil, err
+		}
+		time.Sleep(2 * time.Second)
+	}
+	err = errors.New("timeout waiting for CA certificate")
+	log.Error(err, "CA bundle injection failed")
+	return nil, err
+}
+
+func injectCABundlesIntoCRDs(mgr ctrl.Manager, crdNames []string, caData []byte, log logr.Logger) error {
+	clientset, err := apiextensionsclient.NewForConfig(mgr.GetConfig())
+	if err != nil {
+		log.Error(err, "failed to create clientset")
+		return err
+	}
+
+	for _, name := range crdNames {
+		err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+			crd, err := clientset.ApiextensionsV1().CustomResourceDefinitions().Get(context.Background(), name, metav1.GetOptions{})
+			if err != nil {
+				return err
+			}
+			if crd.Spec.Conversion == nil || crd.Spec.Conversion.Webhook == nil || crd.Spec.Conversion.Webhook.ClientConfig == nil {
+				return nil
+			}
+			if bytes.Equal(crd.Spec.Conversion.Webhook.ClientConfig.CABundle, caData) {
+				return nil
+			}
+			crd.Spec.Conversion.Webhook.ClientConfig.CABundle = caData
+			_, err = clientset.ApiextensionsV1().CustomResourceDefinitions().Update(context.Background(), crd, metav1.UpdateOptions{})
+			return err
+		})
+		if err != nil {
+			log.Error(err, "failed to inject CA bundle", "crd", name)
+		} else {
+			log.Info("Injected conversion CRD CA bundle", "crd", name)
+		}
+	}
+	return nil
+}