@@ -58,6 +58,7 @@ func NewController(
5858 consumerDynamicInformer informers.GenericInformer ,
5959 providerDynamicInformer multinsinformer.GetterInformer ,
6060 serviceNamespaceInformer dynamic.Informer [bindlisters.APIServiceNamespaceLister ],
61+ namespaceCreationNotifyChan <- chan string ,
6162) (* controller , error ) {
6263 queue := workqueue .NewTypedRateLimitingQueueWithConfig (workqueue .DefaultTypedControllerRateLimiter [string ](), workqueue.TypedRateLimitingQueueConfig [string ]{Name : controllerName })
6364
@@ -66,6 +67,9 @@ func NewController(
6667 providerConfig = rest .CopyConfig (providerConfig )
6768 providerConfig = rest .AddUserAgent (providerConfig , controllerName )
6869
70+ consumerConfig = rest .CopyConfig (consumerConfig )
71+ consumerConfig = rest .AddUserAgent (consumerConfig , controllerName )
72+
6973 providerClient , err := dynamicclient .NewForConfig (providerConfig )
7074 if err != nil {
7175 return nil , err
@@ -195,6 +199,9 @@ type controller struct {
195199
196200 providerNamespace string
197201
202+ // Channel to receive notifications when new namespaces are created
203+ namespaceCreationNotifyChan <- chan string
204+
198205 readReconciler
199206}
200207
@@ -240,34 +247,24 @@ func (c *controller) enqueueConsumer(logger klog.Logger, obj interface{}) {
240247 runtime .HandleError (err )
241248 return
242249 }
250+
251+ c .enqueueConsumerByKey (logger , key )
252+ }
253+
254+ // enqueueConsumerByKey will enqueue the given namespaced object. This is split from enqueueConsumer
255+ // so we can accept channel requests from namespace creation watcher.
256+ func (c * controller ) enqueueConsumerByKey (logger klog.Logger , key string ) {
243257 ns , name , err := cache .SplitMetaNamespaceKey (key )
244258 if err != nil {
245259 runtime .HandleError (err )
246260 return
247261 }
248-
249262 if ns != "" { // Namespaced object.
250263 sn , err := c .serviceNamespaceInformer .Lister ().APIServiceNamespaces (c .providerNamespace ).Get (ns )
251264 if err != nil {
252265 if errors .IsNotFound (err ) {
253- // No namespace - create one.
254- // TODO: This is not quite right place for this code. We should refactor this to be somewhere else.
255- _ , err := c .createServiceNamespace (context .TODO (), & kubebindv1alpha2.APIServiceNamespace {
256- ObjectMeta : metav1.ObjectMeta {
257- Name : ns ,
258- Namespace : c .providerNamespace ,
259- OwnerReferences : []metav1.OwnerReference {
260- * metav1 .NewControllerRef (c .apiServiceExport , kubebindv1alpha2 .SchemeGroupVersion .WithKind ("APIServiceExport" )),
261- },
262- },
263- })
264- if err != nil && ! errors .IsAlreadyExists (err ) {
265- runtime .HandleError (fmt .Errorf ("failed to create APIServiceNamespace %q: %w" , ns , err ))
266- return
267- }
268- // Requeue when the APIServiceNamespace is created.
269- logger .V (2 ).Info ("created APIServiceNamespace, requeueing" , "namespace" , ns )
270- // Once APIServiceNamespace is created, the informer will pick it up and requeue the objects.
266+ // No APIServiceNamespace - means claimedresourcesnamespaces controller should create it
267+ logger .V (2 ).Info ("no APIServiceNamespace found for claimed object. Will retry" , "namespace" , ns , "object" , name )
271268 return
272269 }
273270 runtime .HandleError (fmt .Errorf ("failed to get APIServiceNamespace %q: %w" , ns , err ))
@@ -285,7 +282,7 @@ func (c *controller) enqueueConsumer(logger klog.Logger, obj interface{}) {
285282 return
286283 }
287284
288- logger .V (2 ).Info ("queueing Unstructured" , "key" , key )
285+ logger .V (2 ).Info ("queueing Unstructured" , "key" , key , "reason" , "Consumer" )
289286 c .queue .Add (key )
290287}
291288
@@ -427,6 +424,14 @@ func (c *controller) Start(ctx context.Context, numThreads int) {
427424 },
428425 })
429426
427+ // Start a goroutine to listen for namespace creation notifications.
428+ go func () {
429+ for key := range c .namespaceCreationNotifyChan {
430+ logger .Info ("received namespace creation notification" , "namespace" , key )
431+ c .enqueueConsumerByKey (logger , key )
432+ }
433+ }()
434+
430435 for i := 0 ; i < numThreads ; i ++ {
431436 go wait .UntilWithContext (ctx , c .startWorker , time .Second )
432437 }
@@ -450,7 +455,7 @@ func (c *controller) processNextWorkItem(ctx context.Context) bool {
450455
451456 logger := klog .FromContext (ctx ).WithValues ("key" , key )
452457 ctx = klog .NewContext (ctx , logger )
453- logger .V ( 2 ). Info ("processing key" )
458+ logger .Info ("processing key" )
454459
455460 // No matter what, tell the queue we're done with this key, to unblock
456461 // other workers.
@@ -474,12 +479,3 @@ func (c *controller) process(ctx context.Context, key string) error {
474479
475480 return c .reconcile (ctx , namespace , name )
476481}
477-
478- func (c * controller ) createServiceNamespace (ctx context.Context , sns * kubebindv1alpha2.APIServiceNamespace ) (* kubebindv1alpha2.APIServiceNamespace , error ) {
479- return c .providerBindClient .KubeBindV1alpha2 ().APIServiceNamespaces (sns .Namespace ).Create (ctx , sns , metav1.CreateOptions {})
480- }
481-
482- // configMap test-namespace/test
483- // goes into index - indexers.ServiceNamespaceByNamespace with "local name" ->
484-
485- // providerNamespace/<api-service-namspaces.....>
0 commit comments