EntitySource implementation for the v1 registry client #84

ankitathomas · 2023-01-17T18:17:22Z

An implementation of a RegistryEntitySource that watches CatalogSources on all namespaces. It maintains an internal cache

See #76

pkg/deppy/input/catalogsource/registry_cache_querier.go

pkg/deppy/input/catalogsource/registry_bundle_to_entity.go

pkg/deppy/input/catalogsource/registry_cache_querier.go

perdasilva · 2023-02-17T14:02:14Z

pkg/deppy/input/catalogsource/registry_bundle_to_entity.go

+	SkipRange string   `json:"skipRange,omitempty"`
+}
+
+func entityFromBundle(catsrcID string, bundle *catalogsourceapi.Bundle) (*input.Entity, error) {


it might make sense to break this function down into sub-functions to make it a bit smaller and more readable. You may also want to make a type declaration for the key of the property map, e.g. type NamespacedName string or some such.

perdasilva · 2023-02-17T14:11:22Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+func (r *CachedRegistryEntitySource) StopCache() {
+	r.Mutex.Lock()
+	defer r.Mutex.Unlock()
+	r.done <- struct{}{}


you can just close(r.done)

perdasilva · 2023-02-17T14:22:49Z

pkg/deppy/input/catalogsource/registry_grpc_client.go

+	"github.com/operator-framework/deppy/pkg/lib/grpc"
+)
+
+type RegistryClient interface {


can't we straight up use the registry client coming from operator-registry?

oh nvm - we do

perdasilva · 2023-02-23T10:53:31Z

pkg/deppy/input/catalogsource/catalogsource_unit_test.go

+func TestDeppy(t *testing.T) {
+	RegisterFailHandler(Fail)
+
+	RunSpecs(t, "Deppy Suite")


Should it be "RegistryQuerier Suite" or something like that?

perdasilva · 2023-02-23T10:54:30Z

pkg/deppy/input/catalogsource/registry_bundle_to_entity.go

+		}
+	}
+
+	for _, prvAPI := range bundle.ProvidedApis {


It might be good to break these steps into their own functions to make this function a bit smaller and more readable. wdyt?

perdasilva · 2023-02-23T10:55:36Z

pkg/deppy/input/catalogsource/registry_bundle_to_entity.go

+	}
+
+	// Since multiple instances of bundle may exist for different channels, entityID must include reference to channel
+	entityIDFromBundle := func(catsrcID string, bundle *catalogsourceapi.Bundle) deppy.Identifier {


should we just push this out into a proper function rather than an inline one?

perdasilva · 2023-02-23T10:56:38Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+	}
+}
+
+func NewCachedRegistryQuerier(client client.WithWatch, rClient RegistryClient, logger *logr.Logger, options ...Option) *CachedRegistryEntitySource {


could the registry client be internal to this struct?

might also be worth adding the logger as an option with zap.New() as the default value.
if you want to allow for custom registry client implementations, it might also be worth moving this to an option and using yours as the default.

perdasilva · 2023-02-23T11:01:27Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+	r.logger.Info("Completed cache update", "catalogSource", catalogSourceKey)
+}
+
+func (r *CachedRegistryEntitySource) Get(ctx context.Context, id deppy.Identifier) *input.Entity {


rather than iterating over the whole cache to find the item, could you break out the catalog source component of the key and go directly to the right map?

perdasilva · 2023-02-23T11:02:25Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+	return nil
+}
+
+func (r *CachedRegistryEntitySource) Filter(ctx context.Context, filter input.Predicate) (input.EntityList, error) {


you could re-use Iterate here to reduce /reuse some of the code

perdasilva · 2023-02-23T11:03:10Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+	return resultSet, nil
+}
+
+func (r *CachedRegistryEntitySource) GroupBy(ctx context.Context, fn input.GroupByFunction) (input.EntityListMap, error) {


same here - you could re-use Iterate

perdasilva · 2023-02-23T11:13:55Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+		r.logger.Error(err, "failed to start catalogsource watch")
+		return
+	}
+	if err := r.populate(ctx); err != nil {


I wonder if we should have a client-go style WaitForCacheSync method, here's an example:

// stop signal for the informer stopper := make(chan struct{}) defer close(stopper) informer := podInformer.Informer() // start informer -> go factory.Start(stopper) // start to sync and call list if !cache.WaitForCacheSync(stopper, informer.HasSynced) { runtime.HandleError(fmt.Errorf("Timed out waiting for caches to sync")) return }

perdasilva · 2023-02-23T11:14:45Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+
+			switch entry.Type {
+			case watch.Deleted:
+				func() {


I'd say move this to its own function, wdyt?

perdasilva · 2023-02-23T11:15:51Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+		case <-r.done:
+			return
+		case entry := <-catalogSourceWatch.ResultChan():
+			encodedObj, err := json.Marshal(entry.Object)


another option might to be just cast it to a *v1alpha1.CatalogSource, e.g.

catalogsource, ok := entry.Object.(*v1alpha1.CatalogSource)

perdasilva · 2023-02-23T11:17:29Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+	}
+	// watching catalogSource for changes works only with OLM managed catalogSources.
+	go r.ProcessQueue(ctx)
+	for {


might be worth wrapping this into its own function and using adding defer r.Stop() + any other shutdown bits

perdasilva · 2023-02-23T11:18:30Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+				func() {
+					r.RWMutex.Lock()
+					defer r.RWMutex.Unlock()
+					catalogSourceKey := types.NamespacedName{Namespace: catalogSource.Namespace, Name: catalogSource.Name}


maybe push this to its own function. I think I've seen it around a couple of times in the code

perdasilva · 2023-02-23T11:23:51Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+					r.logger.Info("Completed cache delete", "catalogSource", catalogSourceKey)
+				}()
+			case watch.Added, watch.Modified:
+				r.syncCatalogSource(ctx, catalogSource)


we should consider using the workqueue to drive sync in a different go routine. When something is added or modified add that resource to the workqueue. This could also help us with error handling and retries. E.g. if there's a connection failure or transient issue, we can just re-enqueue the resource after some time.

careful with modified because it can happen frequently - e.g. everytime the connection state changes - we should only trigger a resync for certain changes to the resource. Maybe when the connection state is READY?

perdasilva · 2023-02-23T12:19:26Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+				r.logger.Info("cannot find catalogSource, skipping cache update", "CatalogSource", item)
+				return
+			}
+			r.syncCatalogSource(ctx, catalogSource)


we should catch any errors here and re-enqueue the catalog if it they happen. We may want to understand how this work queue works so we can have an exponential backoff. It may also be worth emitting a kube event that says that the cache couldn't be sync'ed for a particular catalog and that it will retry in so many seconds.

you can also do the check for whether the catalog is managed or not and re-enqueue here

perdasilva · 2023-02-23T12:19:48Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+	}
+	r.cache[catalogSourceKey.String()] = sourceCache{
+		Items: entities,
+		//imageID: imageID,


is this comment still valid?

perdasilva · 2023-02-23T12:21:21Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+		}
+		return
+	}
+	r.cache[catalogSourceKey.String()] = sourceCache{


nice - change the whole entry at once - makes it atomic ^^

perdasilva · 2023-02-23T12:22:25Z

pkg/deppy/input/catalogsource/registry_cache_querier.go

+const defaultCatalogSourceSyncInterval = 5 * time.Minute
+
+type CachedRegistryEntitySource struct {
+	sync.RWMutex


how come this is exported rather than an internal variable? is this a Go pattern?

perdasilva · 2023-02-23T12:23:58Z

pkg/deppy/input/catalogsource/registry_grpc_client.go

+func (r *registryGRPCClient) ListEntities(ctx context.Context, catalogSource *v1alpha1.CatalogSource) ([]*input.Entity, error) {
+	// TODO: create GRPC connections separately
+	conn, err := grpc.ConnectWithTimeout(ctx, catalogSource.Address(), r.timeout)
+	if conn != nil {


nit: you could change this to

if err != nil { return ... } defer conn.Close()

perdasilva · 2023-02-23T12:25:54Z

pkg/deppy/input/catalogsource/registry_grpc_client.go

+			catalogPackages[packageKey] = pkg
+		}
+
+		entity, err := EntityFromBundle(catalogSourceID, pkg, bundle)


since this function is in the same package we can get away with not exporting it - or did you export it to make testing easier?

perdasilva · 2023-02-23T12:26:32Z

pkg/deppy/input/catalogsource/registry_grpc_client.go

+
+		entity, err := EntityFromBundle(catalogSourceID, pkg, bundle)
+		if err != nil {
+			return entities, fmt.Errorf("failed to parse entity %s: %v", entity.Identifier(), err)


do you mean to return partial results? it might be worth just returning nil

perdasilva · 2023-02-23T12:27:08Z

pkg/deppy/input/catalogsource/registry_grpc_client_test.go

+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	api2 "github.com/operator-framework/operator-registry/pkg/api"


maybe registryapi rather than api2?

or registry

perdasilva · 2023-02-23T12:27:57Z

pkg/deppy/input/catalogsource/registry_grpc_client_test.go

+				"olm.bundle.path":            "quay.io/openshift-community-operators/prometheus@sha256:6fbd3eaa123054c5023323d1f9ab7cbea178087fcb7cb4f3e83872c6a88d39a1",
+			},
+		}}))
+	})


might be worth adding error test cases as well

perdasilva · 2023-02-23T12:28:41Z

pkg/lib/grpc/grpc.go

@@ -0,0 +1,125 @@
+package grpc


doesn't need to be in this PR, but I think we should open a ticket and move all of this to the registry repo. It seems like it could be useful beyond this use-case

perdasilva · 2023-02-23T12:29:27Z

pkg/lib/grpc/grpc.go

+	return grpc.Dial(address, dialOptions...)
+}
+
+func grpcProxyURL(addr string) (*url.URL, error) {


I feel like I've seen this code before. Is it from the olm repo? if we move this to operator-registry we should also update olm.

perdasilva · 2023-02-23T12:31:36Z

pkg/deppy/input/catalogsource/registry_cache_querier_test.go

+}
+
+var _ = Describe("Registry EntitySource", func() {
+	var querier input.EntitySource


maybe just make it a straigh up RegistryCacheQuerier so you don't need to cast it back in the AfterEach?

perdasilva · 2023-02-23T12:33:41Z

pkg/deppy/input/catalogsource/registry_bundle_to_entity_test.go

+)
+
+var _ = Describe("RegistryBundleConverter", func() {
+	It("generates entity from bundle", func() {


you may want to also poke at some of the error cases

perdasilva

Looks really good and like you've put a lot of effort into this. Thank you!
I think we should move all of this code to the operator-controller and remove the dependencies on olm and registry from deppy.
I've left a few comments that I hope will help ^^ Please let me know if you need any help or have any questions =D

github-actions · 2023-03-26T00:03:43Z

This PR has become stale because it has been open for 30 days with no activity. Please update this PR or remove the lifecycle/stale label before it is automatically closed in 30 days. Adding the lifecycle/frozen label will cause this PR to ignore lifecycle events.

ankitathomas · 2023-03-29T21:05:42Z

Closing in favor of operator-framework/operator-controller#129

ankitathomas requested a review from a team as a code owner January 17, 2023 18:17

openshift-ci bot added the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Jan 17, 2023

ankitathomas force-pushed the watchclient branch 2 times, most recently from 3eb56a3 to 65fc0d5 Compare January 17, 2023 18:25

tmshort reviewed Jan 17, 2023

View reviewed changes

pkg/deppy/input/catalogsource/registry_cache_querier.go Outdated Show resolved Hide resolved

pkg/deppy/input/catalogsource/registry_cache_querier.go Outdated Show resolved Hide resolved

pkg/deppy/input/catalogsource/registry_cache_querier.go Outdated Show resolved Hide resolved

perdasilva reviewed Jan 18, 2023

View reviewed changes

pkg/deppy/input/catalogsource/registry_cache_querier.go Outdated Show resolved Hide resolved

perdasilva reviewed Jan 18, 2023

View reviewed changes

pkg/deppy/input/catalogsource/registry_bundle_to_entity.go Outdated Show resolved Hide resolved

perdasilva reviewed Jan 18, 2023

View reviewed changes

pkg/deppy/input/catalogsource/registry_cache_querier.go Outdated Show resolved Hide resolved

ankitathomas added 4 commits January 30, 2023 12:51

Add RegistryEntitySource

d44935b

improve logging, have separate entities for each channel per bundle

678b19a

match e2e demo format for entities

41a9d94

use local hardcoded address

2395840

varshaprasad96 mentioned this pull request Feb 9, 2023

OLM V1 Milestone 1 operator-framework/enhancements#120

Closed

ankitathomas force-pushed the watchclient branch from 43feec0 to 861ab51 Compare February 14, 2023 05:27

ankitathomas changed the title ~~[WIP] EntitySource implementation for the v1 registry client~~ EntitySource implementation for the v1 registry client Feb 14, 2023

openshift-ci bot removed the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Feb 14, 2023

ankitathomas force-pushed the watchclient branch 4 times, most recently from 229c10e to 636c6e6 Compare February 15, 2023 21:02

add catalogsource queue for non-olm catalogsources, unit and e2e tests

fe8cc76

ankitathomas force-pushed the watchclient branch from 636c6e6 to fe8cc76 Compare February 15, 2023 21:13

perdasilva reviewed Feb 17, 2023

View reviewed changes

perdasilva reviewed Feb 23, 2023

View reviewed changes

perdasilva suggested changes Feb 23, 2023

View reviewed changes

github-actions bot added the lifecycle/stale Denotes an issue or PR has remained open with no activity and has become stale. label Mar 26, 2023

ankitathomas closed this Mar 29, 2023

EntitySource implementation for the v1 registry client #84

EntitySource implementation for the v1 registry client #84

Uh oh!

Conversation

ankitathomas commented Jan 17, 2023

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

perdasilva left a comment

Choose a reason for hiding this comment