Repository: m3db/m3db-operator
Pull request #222 (2020-06-09 17:46:30)

[controller] Only list objects created by operator
Continuously listing every pod and statefulset in large clusters can
slow down the operator control loops significantly. We know every pod
and statefulset created by an M3DBCluster will have predetermined
labels, and can only list objects from the Kubernetes API with those
labels.

codecov[bot]: # [Codecov](https://codecov.io/gh/m3db/m3db-operator/pull/222?src=pr&el=h1) Report
> Merging [#222](https://codecov.io/gh/m3db/m3db-operator/pull/222?src=pr&el=desc) into [master](https://codecov.io/gh/m3db/m3db-operator/commit/cac232c5d824452f90f79d424ffea336a7e378a2&el=desc) will **increase** coverage by `0.02%`.
> The diff coverage is `100.00%`.


[![Impacted file tree graph](https://codecov.io/gh/m3db/m3db-operator/pull/222/graphs/tree.svg?width=650&height=150&src=pr&token=jtHDyWhsto)](https://codecov.io/gh/m3db/m3db-operator/pull/222?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #222      +/-   ##
==========================================
+ Coverage   75.50%   75.53%   +0.02%     
==========================================
  Files          30       30              
  Lines        2111     2113       +2     
==========================================
+ Hits         1594     1596       +2     
  Misses        391      391              
  Partials      126      126              
```


------

[Continue to review full report at Codecov](https://codecov.io/gh/m3db/m3db-operator/pull/222?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/m3db/m3db-operator/pull/222?src=pr&el=footer). Last update [cac232c...371d8e7](https://codecov.io/gh/m3db/m3db-operator/pull/222?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/cmd/m3db-operator/main.go b/cmd/m3db-operator/main.go
index 1cb1c60..ecbf55c 100644
--- a/cmd/m3db-operator/main.go
+++ b/cmd/m3db-operator/main.go
@@ -33,12 +33,16 @@ import (
 	clientset "github.com/m3db/m3db-operator/pkg/client/clientset/versioned"
 	informers "github.com/m3db/m3db-operator/pkg/client/informers/externalversions"
 	"github.com/m3db/m3db-operator/pkg/controller"
+	"github.com/m3db/m3db-operator/pkg/k8sops/labels"
 	"github.com/m3db/m3db-operator/pkg/k8sops/m3db"
 	"github.com/m3db/m3db-operator/pkg/k8sops/podidentity"
 
 	"github.com/m3db/m3x/instrument"
 
 	apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	klabels "k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
 	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
@@ -53,7 +57,7 @@ import (
 
 const (
 	// Informers will resync on this interval
-	_informerSyncDuration = time.Minute
+	_informerSyncDuration = 5 * time.Minute
 )
 
 var (
@@ -168,23 +172,42 @@ func main() {
 		logger.Fatal("failed to create k8sclient", zap.Error(err))
 	}
 
-	stopCh := make(chan struct{})
-	var kubeInformerFactory kubeinformers.SharedInformerFactory
-	if _namespace == "all" {
-		kubeInformerFactory = kubeinformers.NewSharedInformerFactory(kubeClient, _informerSyncDuration)
-	} else {
-		kubeInformerFactory = kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, _informerSyncDuration, kubeinformers.WithNamespace(_namespace))
+	labelReq, err := klabels.NewRequirement(labels.App, selection.Exists, nil)
+	if err != nil {
+		logger.Fatal("failed to construct selector requirement", zap.Error(err))
 	}
+	tweakOpts := kubeinformers.WithTweakListOptions(func(opts *metav1.ListOptions) {
+		// PodSelector provides a selector for listing only M3DB pods to annotate.
+		podSelector := klabels.NewSelector().Add(*labelReq)
+		opts.LabelSelector = podSelector.String()
+	})
 
-	nodeLister := kubeInformerFactory.Core().V1().Nodes().Lister()
+	var (
+		m3dbClusterInformerFactory informers.SharedInformerFactory
+		// filteredInformerFactory is used to create the pod and statefulset
+		// informers. We know that every pod and StatefulSet created by the operator
+		// will have its hardcoded labels. The pod annotation and statefulset loop
+		// may slow down in clusters with many pods or statefulsets if it must
+		// process every objects in the cluster. The filter only lists objects that
+		// have labels we expect.
+		filteredInformerFactory kubeinformers.SharedInformerFactory
+		// The kube informer factory is still required to list nodes, as they don't
+		// have our label.
+		kubeInformerFactory kubeinformers.SharedInformerFactory
+	)
 
-	var m3dbClusterInformerFactory informers.SharedInformerFactory
 	if _namespace == "all" {
+		kubeInformerFactory = kubeinformers.NewSharedInformerFactory(kubeClient, _informerSyncDuration)
 		m3dbClusterInformerFactory = informers.NewSharedInformerFactory(crdClient, _informerSyncDuration)
+		filteredInformerFactory = kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, _informerSyncDuration, tweakOpts)
 	} else {
+		kubeInformerFactory = kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, _informerSyncDuration, kubeinformers.WithNamespace(_namespace))
 		m3dbClusterInformerFactory = informers.NewSharedInformerFactoryWithOptions(crdClient, _informerSyncDuration, informers.WithNamespace(_namespace))
+		filteredInformerFactory = kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, _informerSyncDuration, kubeinformers.WithNamespace(_namespace), tweakOpts)
 	}
 
+	nodeLister := kubeInformerFactory.Core().V1().Nodes().Lister()
+
 	clusterLogger := logger.With(zap.String("controller", "m3db-cluster-controller"))
 	idLogger := logger.With(zap.String("component", "pod-identity-provider"))
 	idProvider, err := podidentity.NewProvider(
@@ -203,6 +226,7 @@ func main() {
 	opts := []controller.Option{
 		controller.WithConfig(config),
 		controller.WithKubeInformerFactory(kubeInformerFactory),
+		controller.WithFilteredInformerFactory(filteredInformerFactory),
 		controller.WithM3DBClusterInformerFactory(m3dbClusterInformerFactory),
 		controller.WithPodIdentityProvider(idProvider),
 		controller.WithLogger(clusterLogger),
@@ -223,7 +247,9 @@ func main() {
 		logger.Fatal("failed to create controller", zap.Error(err))
 	}
 
+	stopCh := make(chan struct{})
 	go kubeInformerFactory.Start(stopCh)
+	go filteredInformerFactory.Start(stopCh)
 	go m3dbClusterInformerFactory.Start(stopCh)
 
 	// Trap the INT and TERM signals
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 471cfde..bd21bdd 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -56,10 +56,10 @@ import (
 	"k8s.io/client-go/util/workqueue"
 
 	jsonpatch "github.com/evanphx/json-patch"
-	"k8s.io/utils/pointer"
 	pkgerrors "github.com/pkg/errors"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
+	"k8s.io/utils/pointer"
 )
 
 const (
@@ -132,8 +132,6 @@ func NewM3DBController(opts ...Option) (*M3DBController, error) {
 		return nil, err
 	}
 
-	kubeInformerFactory := options.kubeInformerFactory
-	m3dbClusterInformerFactory := options.m3dbClusterInformerFactory
 	kclient := options.kclient
 	kubeClient := options.kubeClient
 	crdClient := options.crdClient
@@ -150,9 +148,9 @@ func NewM3DBController(opts ...Option) (*M3DBController, error) {
 		multiClient.clusterURLFn = clusterURLProxy
 	}
 
-	statefulSetInformer := kubeInformerFactory.Apps().V1().StatefulSets()
-	podInformer := kubeInformerFactory.Core().V1().Pods()
-	m3dbClusterInformer := m3dbClusterInformerFactory.Operator().V1alpha1().M3DBClusters()
+	statefulSetInformer := options.filteredInformerFactory.Apps().V1().StatefulSets()
+	podInformer := options.filteredInformerFactory.Core().V1().Pods()
+	m3dbClusterInformer := options.m3dbClusterInformerFactory.Operator().V1alpha1().M3DBClusters()
 
 	samplescheme.AddToScheme(scheme.Scheme)
 
@@ -695,7 +693,7 @@ func (c *M3DBController) handleStatefulSetUpdate(obj interface{}) {
 		c.logger.Info("recovered object from tombstone", zap.String("name", object.GetName()))
 	}
 
-	c.logger.Info("processing statefulset", zap.String("name", object.GetName()))
+	c.logger.Info("processing statefulset", zap.String("statefulset.namespace", object.GetNamespace()), zap.String("statefulset.name", object.GetName()))
 
 	owner := metav1.GetControllerOf(object)
 	// TODO(schallert): const
@@ -705,7 +703,7 @@ func (c *M3DBController) handleStatefulSetUpdate(obj interface{}) {
 
 	cluster, err := c.clusterLister.M3DBClusters(object.GetNamespace()).Get(owner.Name)
 	if err != nil {
-		c.logger.Info("ignoring orphaned object", zap.String("m3dbcluster", owner.Name), zap.String("statefulset", object.GetName()))
+		c.logger.Info("ignoring orphaned object", zap.String("m3dbcluster", owner.Name), zap.String("namespace", object.GetNamespace()), zap.String("statefulset", object.GetName()))
 		return
 	}
 
@@ -803,8 +801,8 @@ func (c *M3DBController) handlePodUpdate(pod *corev1.Pod) error {
 		return err
 	}
 
-	podLogger := c.logger.With(zap.String("pod", pod.Name))
-	podLogger.Debug("processing pod")
+	podLogger := c.logger.With(zap.String("pod.namespace", pod.Namespace), zap.String("pod.name", pod.Name))
+	podLogger.Info("processing pod")
 
 	cluster, err := c.getParentCluster(pod)
 	if err != nil {
diff --git a/pkg/controller/options.go b/pkg/controller/options.go
index a7ddb2f..354c50f 100644
--- a/pkg/controller/options.go
+++ b/pkg/controller/options.go
@@ -49,6 +49,7 @@ type options struct {
 	crdClient                  clientset.Interface
 	kubeClient                 kubernetes.Interface
 	kubeInformerFactory        kubeinformers.SharedInformerFactory
+	filteredInformerFactory    kubeinformers.SharedInformerFactory
 	m3dbClusterInformerFactory informers.SharedInformerFactory
 	kubectlProxy               bool
 }
@@ -101,6 +102,13 @@ func WithKubeInformerFactory(f kubeinformers.SharedInformerFactory) Option {
 	})
 }
 
+// WithFilteredInformerFactory sets the Filteredrnetes base type informer factory.
+func WithFilteredInformerFactory(f kubeinformers.SharedInformerFactory) Option {
+	return optionFn(func(o *options) {
+		o.filteredInformerFactory = f
+	})
+}
+
 // WithM3DBClusterInformerFactory sets the factory for our custom type.
 func WithM3DBClusterInformerFactory(f informers.SharedInformerFactory) Option {
 	return optionFn(func(o *options) {


Test patch:
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index a72ec09..3ba5ad7 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -79,6 +79,7 @@ func TestNew(t *testing.T) {
 		WithCRDClient(crdClient),
 		WithKubeClient(kubeClient),
 		WithKubeInformerFactory(kubeinformers.NewSharedInformerFactory(kubeClient, 0)),
+		WithFilteredInformerFactory(kubeinformers.NewSharedInformerFactory(kubeClient, 0)),
 		WithM3DBClusterInformerFactory(m3dbinformers.NewSharedInformerFactory(crdClient, 0)),
 	}