@@ -36,12 +36,16 @@ import (
3636
3737 corev1 "k8s.io/api/core/v1"
3838 apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
39+ "k8s.io/apimachinery/pkg/api/meta"
3940 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
4041 "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
4142 "k8s.io/apimachinery/pkg/labels"
43+ "k8s.io/apimachinery/pkg/runtime/schema"
4244 "k8s.io/apimachinery/pkg/types"
45+ "k8s.io/apimachinery/pkg/util/sets"
4346 "k8s.io/utils/ptr"
4447 ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
48+ "sigs.k8s.io/controller-runtime/pkg/cluster"
4549 "sigs.k8s.io/controller-runtime/pkg/handler"
4650 "sigs.k8s.io/controller-runtime/pkg/manager"
4751 "sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -161,11 +165,191 @@ func Create(
161165 return nil , fmt .Errorf ("failed to setup local-side watch: %w" , err )
162166 }
163167
168+ if err := setupRelatedResourceWatches (c , localManager , remoteManager , pubRes , localDummy , remoteDummy , log ); err != nil {
169+ return nil , err
170+ }
171+
164172 log .Info ("Done setting up unmanaged controller." )
165173
166174 return c , nil
167175}
168176
177+ // setupRelatedResourceWatches sets up watches for all related resources that have a Watch
178+ // config, on their respective origin side, so that changes trigger primary reconciliation.
179+ func setupRelatedResourceWatches (
180+ c mccontroller.Controller ,
181+ localManager manager.Manager ,
182+ remoteManager mcmanager.Manager ,
183+ pubRes * syncagentv1alpha1.PublishedResource ,
184+ localDummy , remoteDummy * unstructured.Unstructured ,
185+ log * zap.SugaredLogger ,
186+ ) error {
187+ // Deduplication is per-origin to allow the same GVK on both sides.
188+ watchedKcpGVKs := sets .New [schema.GroupVersionKind ]()
189+ watchedServiceGVKs := sets .New [schema.GroupVersionKind ]()
190+
191+ for _ , relRes := range pubRes .Spec .Related {
192+ if relRes .Watch == nil {
193+ continue
194+ }
195+
196+ gvr := schema.GroupVersionResource {
197+ Group : relRes .Group ,
198+ Version : relRes .Version ,
199+ Resource : relRes .Resource ,
200+ }
201+
202+ // Use the REST mapper of the origin side: related resources may have projected GVKs
203+ // that differ between kcp and the service cluster, so we must resolve using the
204+ // mapper that actually knows about the GVR on that side.
205+ var originRESTMapper meta.RESTMapper
206+ if relRes .Origin == syncagentv1alpha1 .RelatedResourceOriginKcp {
207+ originRESTMapper = remoteManager .GetLocalManager ().GetRESTMapper ()
208+ } else {
209+ originRESTMapper = localManager .GetRESTMapper ()
210+ }
211+
212+ gvk , err := originRESTMapper .KindFor (gvr )
213+ if err != nil {
214+ return fmt .Errorf ("failed to determine Kind for related resource %v (origin: %s): %w" , gvr , relRes .Origin , err )
215+ }
216+
217+ relatedDummy := & unstructured.Unstructured {}
218+ relatedDummy .SetGroupVersionKind (gvk )
219+
220+ if relRes .Origin == syncagentv1alpha1 .RelatedResourceOriginKcp {
221+ if watchedKcpGVKs .Has (gvk ) {
222+ continue
223+ }
224+ watchedKcpGVKs .Insert (gvk )
225+
226+ enqueueForRelated , err := buildKcpRelatedHandler (relRes .Watch , gvk , remoteDummy , log )
227+ if err != nil {
228+ return err
229+ }
230+
231+ if err := c .MultiClusterWatch (mcsource .TypedKind (relatedDummy , enqueueForRelated )); err != nil {
232+ return fmt .Errorf ("failed to setup watch for kcp-origin related resource %v: %w" , gvk , err )
233+ }
234+ } else {
235+ if watchedServiceGVKs .Has (gvk ) {
236+ continue
237+ }
238+ watchedServiceGVKs .Insert (gvk )
239+
240+ enqueueForRelated , err := buildServiceRelatedHandler (relRes .Watch , gvk , localDummy , localManager , log )
241+ if err != nil {
242+ return err
243+ }
244+
245+ if err := c .Watch (source .TypedKind (localManager .GetCache (), relatedDummy , enqueueForRelated )); err != nil {
246+ return fmt .Errorf ("failed to setup watch for service-origin related resource %v: %w" , gvk , err )
247+ }
248+ }
249+
250+ log .Infow ("Set up watch for related resource" , "gvk" , gvk , "origin" , relRes .Origin )
251+ }
252+
253+ return nil
254+ }
255+
256+ // buildKcpRelatedHandler constructs the per-cluster event handler for a kcp-origin related resource.
257+ func buildKcpRelatedHandler (
258+ watch * syncagentv1alpha1.RelatedResourceWatch ,
259+ gvk schema.GroupVersionKind ,
260+ remoteDummy * unstructured.Unstructured ,
261+ log * zap.SugaredLogger ,
262+ ) (mchandler.TypedEventHandlerFunc [* unstructured.Unstructured , mcreconcile.Request ], error ) {
263+ switch {
264+ case watch .ByOwner != nil :
265+ ownerGVK := remoteDummy .GroupVersionKind ()
266+ return func (clusterName string , _ cluster.Cluster ) handler.TypedEventHandler [* unstructured.Unstructured , mcreconcile.Request ] {
267+ return & byOwnerEventHandler {
268+ clusterName : clusterName ,
269+ ownerGVK : ownerGVK ,
270+ }
271+ }, nil
272+
273+ case watch .BySelector != nil :
274+ labelSelector := watch .BySelector
275+ primaryDummy := remoteDummy .DeepCopy ()
276+ return func (clusterName string , cl cluster.Cluster ) handler.TypedEventHandler [* unstructured.Unstructured , mcreconcile.Request ] {
277+ return & bySelectorEventHandler {
278+ clusterName : clusterName ,
279+ client : cl .GetClient (),
280+ primaryDummy : primaryDummy ,
281+ labelSelector : labelSelector ,
282+ log : log ,
283+ }
284+ }, nil
285+
286+ default :
287+ return nil , fmt .Errorf ("related resource %v (origin: kcp) has Watch set but neither byOwner nor bySelector configured" , gvk )
288+ }
289+ }
290+
291+ // buildServiceRelatedHandler constructs the event handler for a service-cluster-origin related resource.
292+ // It maps the changed related resource back to the remote (kcp) primary via sync metadata on the local primary.
293+ func buildServiceRelatedHandler (
294+ watch * syncagentv1alpha1.RelatedResourceWatch ,
295+ gvk schema.GroupVersionKind ,
296+ localDummy * unstructured.Unstructured ,
297+ localManager manager.Manager ,
298+ log * zap.SugaredLogger ,
299+ ) (handler.TypedEventHandler [* unstructured.Unstructured , mcreconcile.Request ], error ) {
300+ localClient := localManager .GetClient ()
301+
302+ switch {
303+ case watch .ByOwner != nil :
304+ ownerGVK := localDummy .GroupVersionKind ()
305+ primaryDummy := localDummy .DeepCopy ()
306+ return handler .TypedEnqueueRequestsFromMapFunc (func (ctx context.Context , obj * unstructured.Unstructured ) []mcreconcile.Request {
307+ for _ , ref := range obj .GetOwnerReferences () {
308+ refGV , err := schema .ParseGroupVersion (ref .APIVersion )
309+ if err != nil || refGV .Group != ownerGVK .Group || refGV .Version != ownerGVK .Version || ref .Kind != ownerGVK .Kind {
310+ continue
311+ }
312+ localPrimary := primaryDummy .DeepCopy ()
313+ if err := localClient .Get (ctx , types.NamespacedName {Namespace : obj .GetNamespace (), Name : ref .Name }, localPrimary ); err != nil {
314+ log .Warnw ("Failed to fetch local primary for byOwner watch" , "owner" , ref .Name , "error" , err )
315+ return nil
316+ }
317+ if req := sync .RemoteNameForLocalObject (localPrimary ); req != nil {
318+ return []mcreconcile.Request {* req }
319+ }
320+ return nil
321+ }
322+ return nil
323+ }), nil
324+
325+ case watch .BySelector != nil :
326+ selector , err := metav1 .LabelSelectorAsSelector (watch .BySelector )
327+ if err != nil {
328+ return nil , fmt .Errorf ("failed to convert bySelector for service-origin related resource %v: %w" , gvk , err )
329+ }
330+ primaryDummy := localDummy .DeepCopy ()
331+ return handler .TypedEnqueueRequestsFromMapFunc (func (ctx context.Context , _ * unstructured.Unstructured ) []mcreconcile.Request {
332+ primaryList := & unstructured.UnstructuredList {}
333+ primaryList .SetAPIVersion (primaryDummy .GetAPIVersion ())
334+ primaryList .SetKind (primaryDummy .GetKind () + "List" )
335+ if err := localClient .List (ctx , primaryList , & ctrlruntimeclient.ListOptions {LabelSelector : selector }); err != nil {
336+ log .Warnw ("Failed to list local primary objects for bySelector watch" , "selector" , selector .String (), "error" , err )
337+ return nil
338+ }
339+ var reqs []mcreconcile.Request
340+ for i := range primaryList .Items {
341+ if req := sync .RemoteNameForLocalObject (& primaryList .Items [i ]); req != nil {
342+ reqs = append (reqs , * req )
343+ }
344+ }
345+ return reqs
346+ }), nil
347+
348+ default :
349+ return nil , fmt .Errorf ("related resource %v (origin: service) has Watch set but neither byOwner nor bySelector configured" , gvk )
350+ }
351+ }
352+
169353func (r * Reconciler ) Reconcile (ctx context.Context , request mcreconcile.Request ) (reconcile.Result , error ) {
170354 log := r .log .With ("cluster" , request .ClusterName , "request" , request .NamespacedName )
171355 log .Debug ("Processing" )
0 commit comments