feat: add option to convert empty replicasets to delete event (#225)

Author: mikenorgate

cmd/dump/run.go

@@ -66,7 +66,7 @@ func run(ctx context.Context) error {
 
 	log = log.WithField("k8s_version", v.Full())
 
-	delta, err := controller.CollectSingleSnapshot(ctx, log, clusterID, clientset, dynamicClient, metricsClient, cfg.Controller, v)
+	delta, err := controller.CollectSingleSnapshot(ctx, log, clusterID, clientset, dynamicClient, metricsClient, cfg.Controller, v, cfg.SelfPod.Namespace)
 	if err != nil {
 		return err
 	}

internal/config/config.go

@@ -135,6 +135,7 @@ type Controller struct {
 	AnnotationsMaxLength           string        `mapstructure:"annotations_max_length"`
 	ForcePagination                bool          `mapstructure:"force_pagination"`
 	PageSize                       int64         `mapstructure:"page_size"`
+	FilterEmptyReplicaSets         bool          `mapstructure:"filter_empty_replica_sets"`
 
 	// DisabledInformers contains a list of informers to disable,
 	// for example:

internal/services/controller/controller.go

@@ -102,6 +102,7 @@ type conditionalInformer struct {
 	permissionVerbs   []string
 	isApplied         bool
 	isResourceInError bool
+	transformers      transformers.Transformers
 }
 
 func (i *conditionalInformer) Name() string {
@@ -120,6 +121,7 @@ func CollectSingleSnapshot(ctx context.Context,
 	metricsClient versioned.Interface,
 	cfg *config.Controller,
 	v version.Interface,
+	castwareNamespace string,
 ) (*castai.Delta, error) {
 	tweakListOptions := func(options *metav1.ListOptions) {
 		if cfg.ForcePagination && options.ResourceVersion == "0" {
@@ -131,7 +133,7 @@ func CollectSingleSnapshot(ctx context.Context,
 	f := informers.NewSharedInformerFactoryWithOptions(clientset, 0, informers.WithTweakListOptions(tweakListOptions))
 	df := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, 0, metav1.NamespaceAll, tweakListOptions)
 
-	defaultInformers := getDefaultInformers(f)
+	defaultInformers := getDefaultInformers(f, castwareNamespace, cfg.FilterEmptyReplicaSets)
 	conditionalInformers := getConditionalInformers(clientset, cfg, f, df, metricsClient, log)
 	additionalTransformers := createAdditionalTransformers(cfg)
 
@@ -146,7 +148,8 @@ func CollectSingleSnapshot(ctx context.Context,
 
 	handledInformers := map[string]*custominformers.HandledInformer{}
 	for typ, i := range defaultInformers {
-		handledInformers[typ.String()] = custominformers.NewHandledInformer(log, queue, i.informer, typ, i.filters, additionalTransformers...)
+		transformers := append(i.transformers, additionalTransformers...)
+		handledInformers[typ.String()] = custominformers.NewHandledInformer(log, queue, i.informer, typ, i.filters, transformers...)
 	}
 	for typ, i := range handledConditionalInformers {
 		handledInformers[typ] = i
@@ -208,6 +211,7 @@ func New(
 	agentVersion *config.AgentVersion,
 	healthzProvider *HealthzProvider,
 	selfSubjectAccessReview authorizationtypev1.SelfSubjectAccessReviewInterface,
+	castwareNamespace string,
 ) *Controller {
 	healthzProvider.Initializing()
 
@@ -225,7 +229,7 @@ func New(
 	df := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, defaultResync, metav1.NamespaceAll, tweakListOptions)
 	discovery := clientset.Discovery()
 
-	defaultInformers := getDefaultInformers(f)
+	defaultInformers := getDefaultInformers(f, castwareNamespace, cfg.FilterEmptyReplicaSets)
 	conditionalInformers := getConditionalInformers(clientset, cfg, f, df, metricsClient, log)
 	additionalTransformers := createAdditionalTransformers(cfg)
 
@@ -236,7 +240,8 @@ func New(
 		if !informerEnabled(cfg, name) {
 			continue
 		}
-		handledInformers[name] = custominformers.NewHandledInformer(log, queue, i.informer, typ, i.filters, additionalTransformers...)
+		transformers := append(i.transformers, additionalTransformers...)
+		handledInformers[name] = custominformers.NewHandledInformer(log, queue, i.informer, typ, i.filters, transformers...)
 	}
 
 	eventType := reflect.TypeOf(&corev1.Event{})
@@ -448,8 +453,9 @@ func startConditionalInformers(ctx context.Context,
 		log.Infof("Starting conditional informer for %v", informer.Name())
 		conditionalInformers[i].isApplied = true
 
+		transformers := append(informer.transformers, additionalTransformers...)
 		name := fmt.Sprintf("%s::%s", informer.groupVersion.String(), informer.kind)
-		handledInformer := custominformers.NewHandledInformer(log, queue, informer.informerFactory(), informer.apiType, nil, additionalTransformers...)
+		handledInformer := custominformers.NewHandledInformer(log, queue, informer.informerFactory(), informer.apiType, nil, transformers...)
 		handledInformers[name] = handledInformer
 
 		go handledInformer.Run(ctx.Done())
@@ -985,11 +991,12 @@ func getConditionalInformers(clientset kubernetes.Interface, cfg *config.Control
 }
 
 type defaultInformer struct {
-	informer cache.SharedInformer
-	filters  filters.Filters
+	informer     cache.SharedInformer
+	filters      filters.Filters
+	transformers transformers.Transformers
 }
 
-func getDefaultInformers(f informers.SharedInformerFactory) map[reflect.Type]defaultInformer {
+func getDefaultInformers(f informers.SharedInformerFactory, castwareNamespace string, filterEmptyReplicaSets bool) map[reflect.Type]defaultInformer {
 	return map[reflect.Type]defaultInformer{
 		reflect.TypeOf(&corev1.Node{}):                  {informer: f.Core().V1().Nodes().Informer()},
 		reflect.TypeOf(&corev1.Pod{}):                   {informer: f.Core().V1().Pods().Informer()},
@@ -998,11 +1005,28 @@ func getDefaultInformers(f informers.SharedInformerFactory) map[reflect.Type]def
 		reflect.TypeOf(&corev1.ReplicationController{}): {informer: f.Core().V1().ReplicationControllers().Informer()},
 		reflect.TypeOf(&corev1.Namespace{}):             {informer: f.Core().V1().Namespaces().Informer()},
 		reflect.TypeOf(&appsv1.Deployment{}):            {informer: f.Apps().V1().Deployments().Informer()},
-		reflect.TypeOf(&appsv1.ReplicaSet{}):            {informer: f.Apps().V1().ReplicaSets().Informer()},
-		reflect.TypeOf(&appsv1.DaemonSet{}):             {informer: f.Apps().V1().DaemonSets().Informer()},
-		reflect.TypeOf(&appsv1.StatefulSet{}):           {informer: f.Apps().V1().StatefulSets().Informer()},
-		reflect.TypeOf(&storagev1.StorageClass{}):       {informer: f.Storage().V1().StorageClasses().Informer()},
-		reflect.TypeOf(&batchv1.Job{}):                  {informer: f.Batch().V1().Jobs().Informer()},
+		reflect.TypeOf(&appsv1.ReplicaSet{}): {
+			informer: f.Apps().V1().ReplicaSets().Informer(),
+			transformers: transformers.Transformers{
+				func(e castai.EventType, obj interface{}) (castai.EventType, interface{}) {
+					replicaSet, ok := obj.(*appsv1.ReplicaSet)
+					if !ok || !filterEmptyReplicaSets {
+						return e, obj
+					}
+
+					if e == castai.EventDelete || replicaSet.Namespace == castwareNamespace ||
+						(replicaSet.Spec.Replicas != nil && *replicaSet.Spec.Replicas > 0 && replicaSet.Status.Replicas > 0) || replicaSet.OwnerReferences == nil {
+						return e, obj
+					}
+
+					return castai.EventDelete, obj
+				},
+			},
+		},
+		reflect.TypeOf(&appsv1.DaemonSet{}):       {informer: f.Apps().V1().DaemonSets().Informer()},
+		reflect.TypeOf(&appsv1.StatefulSet{}):     {informer: f.Apps().V1().StatefulSets().Informer()},
+		reflect.TypeOf(&storagev1.StorageClass{}): {informer: f.Storage().V1().StorageClasses().Informer()},
+		reflect.TypeOf(&batchv1.Job{}):            {informer: f.Batch().V1().Jobs().Informer()},
 		reflect.TypeOf(&corev1.Service{}): {
 			informer: f.Core().V1().Services().Informer(),
 			filters: filters.Filters{

internal/services/controller/controller_test.go

@@ -230,6 +230,7 @@ func TestController_ShouldReceiveDeltasBasedOnAvailableResources(t *testing.T) {
 				agentVersion,
 				NewHealthzProvider(defaultHealthzCfg, log),
 				fakeAuthorization.SelfSubjectAccessReviews(),
+				"",
 			)
 
 			if mockDiscovery != nil {
@@ -391,6 +392,7 @@ func TestController_ShouldSendByInterval(t *testing.T) {
 				agentVersion,
 				NewHealthzProvider(defaultHealthzCfg, log),
 				clientset.AuthorizationV1().SelfSubjectAccessReviews(),
+				"",
 			)
 
 			ctrl.Start(ctx.Done())
@@ -517,6 +519,7 @@ func TestController_ShouldCancelAndRestartAfterFailToSend(t *testing.T) {
 		agentVersion,
 		NewHealthzProvider(defaultHealthzCfg, log),
 		clientset.AuthorizationV1().SelfSubjectAccessReviews(),
+		"",
 	)
 
 	ctrl.Start(ctx.Done())
@@ -1266,6 +1269,7 @@ func TestCollectSingleSnapshot(t *testing.T) {
 			PrepTimeout: 10 * time.Second,
 		},
 		version,
+		"",
 	)
 	r.NoError(err)
 	r.NotNil(snapshot)

internal/services/controller/worker.go

@@ -60,6 +60,7 @@ func Loop(
 			agentVersion,
 			healthzProvider,
 			clientset.AuthorizationV1().SelfSubjectAccessReviews(),
+			cfg.SelfPod.Namespace,
 		)
 
 		ctrl.Start(ctrlCtx.Done())