Fix test flakiness by only reacting to MeshConfigs and MRCs in target namespaces

Signed-off-by: Keith Mattix II <[email protected]>

Author: keithmattix

pkg/certificate/fake_manager.go

@@ -34,7 +34,7 @@ func (c *fakeMRCClient) List() ([]*v1alpha2.MeshRootCertificate, error) {
 	}}, nil
 }
 
-func (c *fakeMRCClient) Watch(ctx context.Context) (<-chan MRCEvent, error) {
+func (c *fakeMRCClient) Watch(ctx context.Context, namespace string) (<-chan MRCEvent, error) {
 	ch := make(chan MRCEvent)
 	go func() {
 		ch <- MRCEvent{
@@ -100,6 +100,7 @@ func FakeCertManager() (*Manager, error) {
 		getCertValidityDuration,
 		nil,
 		1*time.Hour,
+		"",
 	)
 	if err != nil {
 		return nil, fmt.Errorf("error creating fakeCertManager, err: %w", err)

pkg/certificate/manager.go

@@ -17,14 +17,14 @@ import (
 )
 
 // NewManager creates a new CertificateManager with the passed MRCClient and options
-func NewManager(ctx context.Context, mrcClient MRCClient, getServiceCertValidityPeriod func() time.Duration, getIngressCertValidityDuration func() time.Duration, msgBroker *messaging.Broker, checkInterval time.Duration) (*Manager, error) {
+func NewManager(ctx context.Context, mrcClient MRCClient, getServiceCertValidityPeriod func() time.Duration, getIngressCertValidityDuration func() time.Duration, msgBroker *messaging.Broker, checkInterval time.Duration, ns string) (*Manager, error) {
 	m := &Manager{
 		serviceCertValidityDuration: getServiceCertValidityPeriod,
 		ingressCertValidityDuration: getIngressCertValidityDuration,
 		msgBroker:                   msgBroker,
 	}
 
-	err := m.start(ctx, mrcClient)
+	err := m.start(ctx, mrcClient, ns)
 	if err != nil {
 		return nil, err
 	}
@@ -49,12 +49,12 @@ func (m *Manager) startRotationTicker(ctx context.Context, checkInterval time.Du
 	}()
 }
 
-func (m *Manager) start(ctx context.Context, mrcClient MRCClient) error {
+func (m *Manager) start(ctx context.Context, mrcClient MRCClient, ns string) error {
 	// start a watch and we wait until the manager is initialized so that
 	// the caller gets a manager that's ready to be used
 	var once sync.Once
 	var wg sync.WaitGroup
-	mrcEvents, err := mrcClient.Watch(ctx)
+	mrcEvents, err := mrcClient.Watch(ctx, ns)
 	if err != nil {
 		return err
 	}

pkg/certificate/manager_test.go

@@ -83,7 +83,7 @@ func TestRotor(t *testing.T) {
 	stop := make(chan struct{})
 	defer close(stop)
 	msgBroker := messaging.NewBroker(stop)
-	certManager, err := NewManager(context.Background(), &fakeMRCClient{}, getServiceCertValidityPeriod, getIngressGatewayCertValidityPeriod, msgBroker, 5*time.Second)
+	certManager, err := NewManager(context.Background(), &fakeMRCClient{}, getServiceCertValidityPeriod, getIngressGatewayCertValidityPeriod, msgBroker, 5*time.Second, "")
 	require.NoError(err)
 
 	certA, err := certManager.IssueCertificate(cnPrefix, Service)

pkg/certificate/providers/compat.go

@@ -15,7 +15,7 @@ func (c *MRCCompatClient) List() ([]*v1alpha2.MeshRootCertificate, error) {
 }
 
 // Watch is a basic Watch implementation for the MRC attached to the compat client
-func (c *MRCCompatClient) Watch(ctx context.Context) (<-chan certificate.MRCEvent, error) {
+func (c *MRCCompatClient) Watch(ctx context.Context, namespace string) (<-chan certificate.MRCEvent, error) {
 	ch := make(chan certificate.MRCEvent)
 	go func() {
 		ch <- certificate.MRCEvent{

pkg/certificate/providers/config.go

@@ -100,7 +100,7 @@ func NewCertificateManager(ctx context.Context, kubeClient kubernetes.Interface,
 		mrcClient = c
 	}
 
-	return certificate.NewManager(ctx, mrcClient, cfg.GetServiceCertValidityPeriod, cfg.GetIngressGatewayCertValidityPeriod, msgBroker, checkInterval)
+	return certificate.NewManager(ctx, mrcClient, cfg.GetServiceCertValidityPeriod, cfg.GetIngressGatewayCertValidityPeriod, msgBroker, checkInterval, providerNamespace)
 }
 
 // GetCertIssuerForMRC returns a certificate.Issuer generated from the provided MRC.

pkg/certificate/providers/mrc.go

@@ -8,6 +8,7 @@ import (
 	"github.com/openservicemesh/osm/pkg/apis/config/v1alpha2"
 	"github.com/openservicemesh/osm/pkg/certificate"
 	"github.com/openservicemesh/osm/pkg/k8s/informers"
+	"github.com/rs/zerolog/log"
 )
 
 // MRCComposer is a composer object that allows consumers
@@ -41,11 +42,16 @@ func (m *MRCComposer) List() ([]*v1alpha2.MeshRootCertificate, error) {
 // from the informerCollection's MRC store. Channels returned from multiple invocations of
 // Watch() are unique and have no coordination with each other. Events are guaranteed
 // to be ordered for any particular resources, but NOT across different resources.
-func (m *MRCComposer) Watch(ctx context.Context) (<-chan certificate.MRCEvent, error) {
+func (m *MRCComposer) Watch(ctx context.Context, namespace string) (<-chan certificate.MRCEvent, error) {
 	eventChan := make(chan certificate.MRCEvent)
 	m.informerCollection.AddEventHandler(informers.InformerKeyMeshRootCertificate, cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
+			log.Debug().Msg("received MRC add event")
 			mrc := obj.(*v1alpha2.MeshRootCertificate)
+			// If there's a specific namespace passed in don't send the MRCEvent unless the MRC's namespace matches
+			if namespace != "" && mrc.GetNamespace() != namespace {
+				return
+			}
 			eventChan <- certificate.MRCEvent{
 				Type: certificate.MRCEventAdded,
 				MRC:  mrc,
@@ -54,7 +60,11 @@ func (m *MRCComposer) Watch(ctx context.Context) (<-chan certificate.MRCEvent, e
 		// We don't really care about the previous version
 		// since the "state machine" of the MRC is well defined
 		UpdateFunc: func(_, newObj interface{}) {
+			log.Debug().Msg("received MRC update event")
 			mrc := newObj.(*v1alpha2.MeshRootCertificate)
+			if namespace != "" && mrc.GetNamespace() != namespace {
+				return
+			}
 			eventChan <- certificate.MRCEvent{
 				Type: certificate.MRCEventUpdated,
 				MRC:  mrc,

pkg/certificate/providers/tresor/fake/fake.go

@@ -39,7 +39,7 @@ func (c *fakeMRCClient) List() ([]*v1alpha2.MeshRootCertificate, error) {
 	return []*v1alpha2.MeshRootCertificate{{Spec: v1alpha2.MeshRootCertificateSpec{TrustDomain: "fake.example.com"}}}, nil
 }
 
-func (c *fakeMRCClient) Watch(ctx context.Context) (<-chan certificate.MRCEvent, error) {
+func (c *fakeMRCClient) Watch(ctx context.Context, namespace string) (<-chan certificate.MRCEvent, error) {
 	ch := make(chan certificate.MRCEvent)
 	go func() {
 		ch <- certificate.MRCEvent{
@@ -83,7 +83,7 @@ func NewFake(msgBroker *messaging.Broker, checkInterval time.Duration) *certific
 
 // NewFakeWithValidityDuration constructs a fake certificate manager with specified cert validity duration
 func NewFakeWithValidityDuration(getCertValidityDuration func() time.Duration, msgBroker *messaging.Broker, checkInterval time.Duration) *certificate.Manager {
-	tresorCertManager, err := certificate.NewManager(context.Background(), &fakeMRCClient{}, getCertValidityDuration, getCertValidityDuration, msgBroker, checkInterval)
+	tresorCertManager, err := certificate.NewManager(context.Background(), &fakeMRCClient{}, getCertValidityDuration, getCertValidityDuration, msgBroker, checkInterval, "")
 	if err != nil {
 		log.Error().Err(err).Msg("error encountered creating fake cert manager")
 		return nil

pkg/certificate/types.go

@@ -147,7 +147,7 @@ var (
 // MRCEventBroker describes any type that allows the caller to Watch() MRCEvents
 type MRCEventBroker interface {
 	// Watch allows the caller to subscribe to events surrounding
-	// MRCs that belong to this particular mesh. Watch returns a channel
+	// MRCs within a specific namespace. Watch returns a channel
 	// that emits events, and an error if the subscription goes awry.
-	Watch(context.Context) (<-chan MRCEvent, error)
+	Watch(context.Context, string) (<-chan MRCEvent, error)
 }

pkg/configurator/client.go

@@ -6,6 +6,7 @@ import (
 
 	"k8s.io/client-go/tools/cache"
 
+	"github.com/openservicemesh/osm/pkg/apis/config/v1alpha2"
 	configv1alpha2 "github.com/openservicemesh/osm/pkg/apis/config/v1alpha2"
 	"github.com/openservicemesh/osm/pkg/k8s/informers"
 
@@ -31,15 +32,21 @@ func NewConfigurator(informerCollection *informers.InformerCollection, osmNamesp
 		Delete: announcements.MeshConfigDeleted,
 	}
 
-	informerCollection.AddEventHandler(informers.InformerKeyMeshConfig, k8s.GetEventHandlerFuncs(nil, meshConfigEventTypes, msgBroker))
+	informerCollection.AddEventHandler(informers.InformerKeyMeshConfig, k8s.GetEventHandlerFuncs(func(obj interface{}) bool {
+		mc := obj.(*v1alpha2.MeshConfig)
+		return mc.GetName() == meshConfigName
+	}, meshConfigEventTypes, msgBroker))
 	informerCollection.AddEventHandler(informers.InformerKeyMeshConfig, c.metricsHandler())
 
 	meshRootCertificateEventTypes := k8s.EventTypes{
 		Add:    announcements.MeshRootCertificateAdded,
 		Update: announcements.MeshRootCertificateUpdated,
 		Delete: announcements.MeshRootCertificateDeleted,
 	}
-	informerCollection.AddEventHandler(informers.InformerKeyMeshRootCertificate, k8s.GetEventHandlerFuncs(nil, meshRootCertificateEventTypes, msgBroker))
+	informerCollection.AddEventHandler(informers.InformerKeyMeshRootCertificate, k8s.GetEventHandlerFuncs(func(obj interface{}) bool {
+		mrc := obj.(*v1alpha2.MeshRootCertificate)
+		return mrc.GetNamespace() == osmNamespace
+	}, meshRootCertificateEventTypes, msgBroker))
 
 	return c
 }