Port 9402 port api search identifier

pkg/handlers/controllers.go

@@ -2,20 +2,19 @@ package handlers
 
 import (
 	"context"
-	"time"
-
 	"github.com/port-labs/port-k8s-exporter/pkg/config"
-	"github.com/port-labs/port-k8s-exporter/pkg/port/integration"
-
 	"github.com/port-labs/port-k8s-exporter/pkg/crd"
 	"github.com/port-labs/port-k8s-exporter/pkg/goutils"
 	"github.com/port-labs/port-k8s-exporter/pkg/k8s"
 	"github.com/port-labs/port-k8s-exporter/pkg/port"
 	"github.com/port-labs/port-k8s-exporter/pkg/port/cli"
+	"github.com/port-labs/port-k8s-exporter/pkg/port/integration"
 	"github.com/port-labs/port-k8s-exporter/pkg/signal"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/klog/v2"
+	"sync"
+	"time"
 )
 
 type ControllersHandler struct {
@@ -71,56 +70,73 @@ func NewControllersHandler(exporterConfig *port.Config, portConfig *port.Integra
 func (c *ControllersHandler) Handle() {
 	klog.Info("Starting informers")
 	c.informersFactory.Start(c.stopCh)
-	klog.Info("Waiting for informers cache sync")
+
+	currentEntitiesSets := make([]map[string]interface{}, 0)
+	shouldDeleteStaleEntities := true
+	var syncWg sync.WaitGroup
+
 	for _, controller := range c.controllers {
+		controller := controller
+
+		go func() {
+			<-c.stopCh
+			klog.Info("Shutting down controllers")
+			controller.Shutdown()
+			klog.Info("Exporter exiting")
+		}()
+
+		klog.Infof("Waiting for informer cache to sync for resource '%s'", controller.Resource.Kind)
 		if err := controller.WaitForCacheSync(c.stopCh); err != nil {
 			klog.Fatalf("Error while waiting for informer cache sync: %s", err.Error())
 		}
-	}
 
-	currentEntitiesSet := make([]map[string]interface{}, 0)
-	for _, controller := range c.controllers {
-		controllerEntitiesSet, rawDataExamples, err := controller.GetEntitiesSet()
-		if err != nil {
-			klog.Errorf("error getting controller entities set: %s", err.Error())
-		}
-		currentEntitiesSet = append(currentEntitiesSet, controllerEntitiesSet)
-		if len(rawDataExamples) > 0 {
-			err = integration.PostIntegrationKindExample(c.portClient, c.stateKey, controller.Resource.Kind, rawDataExamples)
-			if err != nil {
-				klog.Warningf("failed to post integration kind example: %s", err.Error())
+		syncWg.Add(1)
+		go func() {
+			defer syncWg.Done()
+			klog.Infof("Starting full initial resync for resource '%s'", controller.Resource.Kind)
+			initialSyncResult := controller.RunInitialSync()
+			klog.Infof("Done full initial resync, starting live events sync for resource '%s'", controller.Resource.Kind)
+			controller.RunEventsSync(1, c.stopCh)
+			if initialSyncResult.EntitiesSet != nil {
+				currentEntitiesSets = append(currentEntitiesSets, initialSyncResult.EntitiesSet)
 			}
-		}
-	}
-
-	klog.Info("Deleting stale entities")
-	c.RunDeleteStaleEntities(currentEntitiesSet)
-	klog.Info("Starting controllers")
-	for _, controller := range c.controllers {
-		controller.Run(1, c.stopCh)
+			if len(initialSyncResult.RawDataExamples) > 0 {
+				err := integration.PostIntegrationKindExample(c.portClient, c.stateKey, controller.Resource.Kind, initialSyncResult.RawDataExamples)
+				if err != nil {
+					klog.Warningf("failed to post integration kind example: %s", err.Error())
+				}
+			}
+			shouldDeleteStaleEntities = shouldDeleteStaleEntities && initialSyncResult.ShouldDeleteStaleEntities
+		}()
 	}
+	syncWg.Wait()
 
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	defer cancelCtx()
 	go func() {
 		<-c.stopCh
-		klog.Info("Shutting down controllers")
-		for _, controller := range c.controllers {
-			controller.Shutdown()
-		}
-		klog.Info("Exporter exiting")
+		cancelCtx()
 	}()
+
+	if shouldDeleteStaleEntities {
+		klog.Info("Deleting stale entities")
+		c.runDeleteStaleEntities(ctx, currentEntitiesSets)
+		klog.Info("Done deleting stale entities")
+	} else {
+		klog.Warning("Skipping delete of stale entities due to a failure in getting all current entities from k8s")
+	}
 }
 
-func (c *ControllersHandler) RunDeleteStaleEntities(currentEntitiesSet []map[string]interface{}) {
-	_, err := c.portClient.Authenticate(context.Background(), c.portClient.ClientID, c.portClient.ClientSecret)
+func (c *ControllersHandler) runDeleteStaleEntities(ctx context.Context, currentEntitiesSet []map[string]interface{}) {
+	_, err := c.portClient.Authenticate(ctx, c.portClient.ClientID, c.portClient.ClientSecret)
 	if err != nil {
 		klog.Errorf("error authenticating with Port: %v", err)
 	}
 
-	err = c.portClient.DeleteStaleEntities(context.Background(), c.stateKey, goutils.MergeMaps(currentEntitiesSet...))
+	err = c.portClient.DeleteStaleEntities(ctx, c.stateKey, goutils.MergeMaps(currentEntitiesSet...))
 	if err != nil {
 		klog.Errorf("error deleting stale entities: %s", err.Error())
 	}
-	klog.Info("Done deleting stale entities")
 }
 
 func (c *ControllersHandler) Stop() {