You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
This commit adds the infrastructure for watching and querying
arbitrarily-typed pipeline targets, in remote clusters as well as the
local cluster.
The basic shape is this: for each target that needs to be examined,
the reconciler uses `watchTargetAndGetReader(..., target)`. This
procedure encapsulates the detail of making sure there's a cache for
the target's cluster and type, and supplies the client.Reader needed
for fetching the target object.
A `cache.Cache` is kept for each {cluster, type}. `cache.Cache` is the
smallest piece of machinery that can be torn down, because the next
layer down, `Informer` objects, can't be removed once created. This is
important for being able to stop watching targets when they are no
longer targets.
Target object updates will come from all the caches, which come and
(in principle) go; but, the handler must be statically installed in
SetupWithManager(). So, targets are looked up in an index to get the
corresponding pipeline (if there is one), and that pipeline is put
into a `source.Channel`. The channel source multiplexes the dynamic
event handlers into a static pipeline requeue handler.
NB:
* I've put the remote cluster test in its own Test* wrapper, because
it needs to start another testenv to be the remote cluster.
* Supporting arbitrary types means using `unstructured.Unstructured`
when querying for target objects, and this complicates checking
their status. Since the caches are per-type, in theory there could
be code for uerying known types (HelmRelease and Kustomize), with
`Unstructured` as a fallback. So long at the object passed to
`watchTargetAndGetReader(...) is the same one used with
client.Get(...), it should all work.
* A cache per {cluster, type} is not the only possible scheme. The
watching could be more precise -- meaning fewer spurious events,
and narrower permissions needed -- by having a cache per {cluster,
namespace, type}, with the trade-off being managing more
goroutines, and other overheads. I've chosen the chunkier scheme
based on an informed guess that it'll be more efficient for low
numbers of clusters and targets.
// TODO future: get the secret via the cluster object, connect to remote cluster
340
-
returnnil, fmt.Errorf("remote clusters not supported yet")
341
-
}
342
381
343
-
// targetObjectKey returns the object key (namespaced name) for a target. The Pipeline is passed in as well as the Target, since the definition can be spread between these specs.
// this must be done with the lock held, because if it fails, we can't use the cache and shouldn't add it to the map.
453
+
iferr:=r.manager.Add(c); err!=nil { // this will start it asynchronously
454
+
r.cachesMu.Unlock()
455
+
returnnil, false, err
456
+
}
457
+
458
+
typeCache=c
459
+
r.caches[cacheKey] =typeCache
460
+
}
461
+
r.cachesMu.Unlock()
462
+
}
463
+
464
+
// Now we have a cache; make sure the object type in question is being watched, so we can query it and get updates.
465
+
466
+
// The informer is retrieved whether we created the cache or not, because we want to know if it's synced and thus ready to be queried.
467
+
inf, err:=typeCache.GetInformer(ctx, target) // NB not InformerForKind(...), because that uses the typed value cache specifically (see the method comment).
468
+
iferr!=nil {
469
+
returnnil, false, err
470
+
}
471
+
472
+
if!cacheFound { // meaning: we created the cache, this time around
473
+
enqueuePipelinesForTarget:=func(objinterface{}) {
474
+
eventObj, ok:=obj.(client.Object)
475
+
if!ok {
476
+
logger.Info("value to look up in index was not a client.Object", "object", eventObj)
// We're just looking up the name in the index so it'll be the same for the old as for the new object.
499
+
// However, this might change elsewhere, so to be defensive, run both. The queue will deduplicate,
500
+
// though it means we do a bit more lookup work.
501
+
enqueuePipelinesForTarget(oldObj)
502
+
enqueuePipelinesForTarget(newObj)
503
+
},
504
+
})
505
+
iferr!=nil {
506
+
returnnil, false, err
507
+
}
508
+
}
509
+
510
+
returntypeCache, inf.HasSynced(), nil
349
511
}
350
512
351
-
// clusterPrefix returns a string naming the cluster containing an app, to prepend to the usual namespace/name format of the app object itself. So that it can be empty, the separator is include in the return value.
513
+
// clusterPrefix returns a string naming the cluster containing an app, to prepend to the usual namespace/name format of the app object itself.
514
+
// So that it can be empty, the separator is include in the return value.
0 commit comments