Merge pull request #663 from jetstack/pass_context

Use new client-go functions for contextual logging

Author: inteon

pkg/agent/dummy_data_gatherer.go

@@ -29,12 +29,12 @@ type dummyDataGatherer struct {
 FailedAttempts int
 }
 
-func (g *dummyDataGatherer) Run(stopCh <-chan struct{}) error {
+func (g *dummyDataGatherer) Run(ctx context.Context) error {
 // no async functionality, see Fetch
 return nil
 }
 
-func (g *dummyDataGatherer) WaitForCacheSync(stopCh <-chan struct{}) error {
+func (g *dummyDataGatherer) WaitForCacheSync(ctx context.Context) error {
 // no async functionality, see Fetch
 return nil
 }

pkg/agent/run.go

@@ -191,7 +191,7 @@ func Run(cmd *cobra.Command, args []string) (returnErr error) {
 // blocks until the supplied channel is closed.
 // For this reason, we must allow these errgroup Go routines to exit
 // without cancelling the other Go routines in the group.
-if err := newDg.Run(gctx.Done()); err != nil {
+if err := newDg.Run(gctx); err != nil {
 return fmt.Errorf("failed to start %q data gatherer %q: %v", kind, dgConfig.Name, err)
 }
 return nil
@@ -220,7 +220,7 @@ func Run(cmd *cobra.Command, args []string) (returnErr error) {
 // wait for the informer to complete an initial sync, we do this to
 // attempt to have an initial set of data for the first upload of
 // the run.
-if err := dg.WaitForCacheSync(bootCtx.Done()); err != nil {
+if err := dg.WaitForCacheSync(bootCtx); err != nil {
 // log sync failure, this might recover in future
 if errors.Is(err, k8s.ErrCacheSyncTimeout) {
 timedoutDGs = append(timedoutDGs, dgConfig.Name)

pkg/datagatherer/datagatherer.go

@@ -17,9 +17,9 @@ type DataGatherer interface {
 Fetch() (data interface{}, count int, err error)
 // Run starts the data gatherer's informers for resource collection.
 // Returns error if the data gatherer informer wasn't initialized
-Run(stopCh <-chan struct{}) error
+Run(ctx context.Context) error
 // WaitForCacheSync waits for the data gatherer's informers cache to sync.
-WaitForCacheSync(stopCh <-chan struct{}) error
+WaitForCacheSync(ctx context.Context) error
 // Delete, clear the cache of the DataGatherer if one is being used
 Delete() error
 }

pkg/datagatherer/k8s/discovery.go

@@ -47,12 +47,12 @@ type DataGathererDiscovery struct {
 cl *discovery.DiscoveryClient
 }
 
-func (g *DataGathererDiscovery) Run(stopCh <-chan struct{}) error {
+func (g *DataGathererDiscovery) Run(ctx context.Context) error {
 // no async functionality, see Fetch
 return nil
 }
 
-func (g *DataGathererDiscovery) WaitForCacheSync(stopCh <-chan struct{}) error {
+func (g *DataGathererDiscovery) WaitForCacheSync(ctx context.Context) error {
 // no async functionality, see Fetch
 return nil
 }

pkg/datagatherer/k8s/dynamic.go

@@ -181,7 +181,6 @@ func (c *ConfigDynamic) newDataGathererWithClient(ctx context.Context, cl dynami
 dgCache := cache.New(5*time.Minute, 30*time.Second)
 
 newDataGatherer := &DataGathererDynamic{
-ctx: ctx,
 groupVersionResource: c.GroupVersionResource,
 fieldSelector: fieldSelector.String(),
 namespaces: c.IncludeNamespaces,
@@ -217,7 +216,7 @@ func (c *ConfigDynamic) newDataGathererWithClient(ctx context.Context, cl dynami
 newDataGatherer.informer = factory.ForResource(c.GroupVersionResource).Informer()
 }
 
-registration, err := newDataGatherer.informer.AddEventHandler(k8scache.ResourceEventHandlerFuncs{
+registration, err := newDataGatherer.informer.AddEventHandlerWithOptions(k8scache.ResourceEventHandlerFuncs{
 AddFunc: func(obj interface{}) {
 onAdd(log, obj, dgCache)
 },
@@ -227,6 +226,8 @@ func (c *ConfigDynamic) newDataGathererWithClient(ctx context.Context, cl dynami
 DeleteFunc: func(obj interface{}) {
 onDelete(log, obj, dgCache)
 },
+}, k8scache.HandlerOptions{
+Logger: &log,
 })
 if err != nil {
 return nil, err
@@ -243,7 +244,6 @@ func (c *ConfigDynamic) newDataGathererWithClient(ctx context.Context, cl dynami
 // This is to allow us to support arbitrary CRDs and resources that Preflight
 // does not have registered as part of its `runtime.Scheme`.
 type DataGathererDynamic struct {
-ctx context.Context
 // groupVersionResource is the name of the API group, version and resource
 // that should be fetched by this data gatherer.
 groupVersionResource schema.GroupVersionResource
@@ -269,8 +269,8 @@ type DataGathererDynamic struct {
 // Run starts the dynamic data gatherer's informers for resource collection.
 // Returns error if the data gatherer informer wasn't initialized, Run blocks
 // until the stopCh is closed.
-func (g *DataGathererDynamic) Run(stopCh <-chan struct{}) error {
-log := klog.FromContext(g.ctx)
+func (g *DataGathererDynamic) Run(ctx context.Context) error {
+log := klog.FromContext(ctx)
 if g.informer == nil {
 return fmt.Errorf("informer was not initialized, impossible to start")
 }
@@ -288,7 +288,7 @@ func (g *DataGathererDynamic) Run(stopCh <-chan struct{}) error {
 }
 
 // start shared informer
-g.informer.Run(stopCh)
+g.informer.RunWithContext(ctx)
 
 return nil
 }
@@ -298,8 +298,9 @@ var ErrCacheSyncTimeout = fmt.Errorf("timed out waiting for Kubernetes cache to
 // WaitForCacheSync waits for the data gatherer's informers cache to sync before
 // collecting the resources. Use errors.Is(err, ErrCacheSyncTimeout) to check if
 // the cache sync failed.
-func (g *DataGathererDynamic) WaitForCacheSync(stopCh <-chan struct{}) error {
-if !k8scache.WaitForCacheSync(stopCh, g.registration.HasSynced) {
+func (g *DataGathererDynamic) WaitForCacheSync(ctx context.Context) error {
+// Don't use WaitForNamedCacheSync, since we don't want to log extra messages.
+if !k8scache.WaitForCacheSync(ctx.Done(), g.registration.HasSynced) {
 return ErrCacheSyncTimeout
 }
 

pkg/datagatherer/k8s/dynamic_test.go

@@ -134,7 +134,6 @@ func TestNewDataGathererWithClientAndDynamicInformer(t *testing.T) {
 }
 
 expected := &DataGathererDynamic{
-ctx: ctx,
 groupVersionResource: config.GroupVersionResource,
 // it's important that the namespaces are set as the IncludeNamespaces
 // during initialization
@@ -144,9 +143,6 @@ func TestNewDataGathererWithClientAndDynamicInformer(t *testing.T) {
 
 gatherer := dg.(*DataGathererDynamic)
 // test gatherer's fields
-if !reflect.DeepEqual(gatherer.ctx, expected.ctx) {
-t.Errorf("expected %v, got %v", expected, dg)
-}
 if !reflect.DeepEqual(gatherer.groupVersionResource, expected.groupVersionResource) {
 t.Errorf("expected %v, got %v", expected, dg)
 }
@@ -180,7 +176,6 @@ func TestNewDataGathererWithClientAndSharedIndexInformer(t *testing.T) {
 }
 
 expected := &DataGathererDynamic{
-ctx: ctx,
 groupVersionResource: config.GroupVersionResource,
 // it's important that the namespaces are set as the IncludeNamespaces
 // during initialization
@@ -189,9 +184,6 @@ func TestNewDataGathererWithClientAndSharedIndexInformer(t *testing.T) {
 
 gatherer := dg.(*DataGathererDynamic)
 // test gatherer's fields
-if !reflect.DeepEqual(gatherer.ctx, expected.ctx) {
-t.Errorf("expected %v, got %v", expected, dg)
-}
 if !reflect.DeepEqual(gatherer.groupVersionResource, expected.groupVersionResource) {
 t.Errorf("expected %v, got %v", expected, dg)
 }
@@ -693,11 +685,11 @@ func TestDynamicGatherer_Fetch(t *testing.T) {
 // start data gatherer informer
 dynamiDg := dg
 go func() {
-if err = dynamiDg.Run(ctx.Done()); err != nil {
+if err = dynamiDg.Run(ctx); err != nil {
 t.Errorf("unexpected client error: %+v", err)
 }
 }()
-err = dynamiDg.WaitForCacheSync(ctx.Done())
+err = dynamiDg.WaitForCacheSync(ctx)
 if err != nil {
 t.Fatalf("unexpected client error: %+v", err)
 }
@@ -1010,11 +1002,11 @@ func TestDynamicGathererNativeResources_Fetch(t *testing.T) {
 // start data gatherer informer
 dynamiDg := dg
 go func() {
-if err = dynamiDg.Run(ctx.Done()); err != nil {
+if err = dynamiDg.Run(ctx); err != nil {
 t.Errorf("unexpected client error: %+v", err)
 }
 }()
-err = dynamiDg.WaitForCacheSync(ctx.Done())
+err = dynamiDg.WaitForCacheSync(ctx)
 if err != nil {
 t.Fatalf("unexpected client error: %+v", err)
 }

pkg/datagatherer/local/local.go

@@ -38,7 +38,7 @@ func (c *Config) NewDataGatherer(ctx context.Context) (datagatherer.DataGatherer
 }, nil
 }
 
-func (g *DataGatherer) Run(stopCh <-chan struct{}) error {
+func (g *DataGatherer) Run(ctx context.Context) error {
 // no async functionality, see Fetch
 return nil
 }
@@ -48,7 +48,7 @@ func (g *DataGatherer) Delete() error {
 return nil
 }
 
-func (g *DataGatherer) WaitForCacheSync(stopCh <-chan struct{}) error {
+func (g *DataGatherer) WaitForCacheSync(ctx context.Context) error {
 // no async functionality, see Fetch
 return nil
 }