fix fatal both in client and in unstructured (#73)

* fix fatal both in client and in unstructured

* enabling injecting portClient

* fix: remove mutex

* drill down application config

* fix usage of fixtures in tests

* using a more performent deep copy

* using centric config

* chore: cr fixes

main.go

@@ -25,9 +25,6 @@ func initiateHandler(exporterConfig *port.Config, k8sClient *k8s.Client, portCli
 
 	}
 
-	cli.WithDeleteDependents(i.Config.DeleteDependents)(portClient)
-	cli.WithCreateMissingRelatedEntities(i.Config.CreateMissingRelatedEntities)(portClient)
-
 	newHandler := handlers.NewControllersHandler(exporterConfig, i.Config, k8sClient, portClient)
 	newHandler.Handle()
 
@@ -50,15 +47,7 @@ func main() {
 	if err != nil {
 		klog.Fatalf("Error building K8s client: %s", err.Error())
 	}
-
-	portClient, err := cli.New(config.ApplicationConfig.PortBaseURL,
-		cli.WithClientID(config.ApplicationConfig.PortClientId), cli.WithClientSecret(config.ApplicationConfig.PortClientSecret),
-		cli.WithHeader("User-Agent", fmt.Sprintf("port-k8s-exporter/^0.3.4 (statekey/%s)", applicationConfig.StateKey)),
-	)
-
-	if err != nil {
-		klog.Fatalf("Error building Port client: %s", err.Error())
-	}
+	portClient := cli.New(config.ApplicationConfig)
 
 	if err := defaults.InitIntegration(portClient, applicationConfig); err != nil {
 		klog.Fatalf("Error initializing Port integration: %s", err.Error())

pkg/crd/crd_test.go

@@ -26,7 +26,7 @@ func deleteDefaultResources(portClient *cli.PortClient) {
 	_ = blueprint.DeleteBlueprint(portClient, "testkind")
 }
 
-func newFixture(t *testing.T, portClientId string, portClientSecret string, userAgent string, namespaced bool, crdsDiscoveryPattern string) *Fixture {
+func newFixture(t *testing.T, userAgent string, namespaced bool, crdsDiscoveryPattern string) *Fixture {
 	apiExtensionsFakeClient := fakeapiextensionsv1.FakeApiextensionsV1{Fake: &clienttesting.Fake{}}
 
 	apiExtensionsFakeClient.AddReactor("list", "customresourcedefinitions", func(action clienttesting.Action) (handled bool, ret runtime.Object, err error) {
@@ -100,22 +100,12 @@ func newFixture(t *testing.T, portClientId string, portClientSecret string, user
 		return true, fakeCrd, nil
 	})
 
-	if portClientId == "" {
-		portClientId = config.ApplicationConfig.PortClientId
-	}
-	if portClientSecret == "" {
-		portClientSecret = config.ApplicationConfig.PortClientSecret
-	}
 	if userAgent == "" {
 		userAgent = "port-k8s-exporter/0.1"
 	}
 
-	portClient, err := cli.New(config.ApplicationConfig.PortBaseURL, cli.WithHeader("User-Agent", userAgent),
-		cli.WithClientID(portClientId), cli.WithClientSecret(portClientSecret))
+	portClient := cli.New(config.ApplicationConfig)
 	deleteDefaultResources(portClient)
-	if err != nil {
-		t.Errorf("Error building Port client: %s", err.Error())
-	}
 
 	return &Fixture{
 		t:                  t,
@@ -271,7 +261,7 @@ func checkBlueprintAndActionsProperties(t *testing.T, f *Fixture, namespaced boo
 }
 
 func TestCRD_crd_autoDiscoverCRDsToActionsClusterScoped(t *testing.T) {
-	f := newFixture(t, "", "", "", false, "true")
+	f := newFixture(t, "", false, "true")
 
 	AutodiscoverCRDsToActions(f.portConfig, f.apiextensionClient, f.portClient)
 
@@ -281,7 +271,7 @@ func TestCRD_crd_autoDiscoverCRDsToActionsClusterScoped(t *testing.T) {
 }
 
 func TestCRD_crd_autoDiscoverCRDsToActionsNamespaced(t *testing.T) {
-	f := newFixture(t, "", "", "", true, "true")
+	f := newFixture(t, "", true, "true")
 
 	AutodiscoverCRDsToActions(f.portConfig, f.apiextensionClient, f.portClient)
 
@@ -291,7 +281,7 @@ func TestCRD_crd_autoDiscoverCRDsToActionsNamespaced(t *testing.T) {
 }
 
 func TestCRD_crd_autoDiscoverCRDsToActionsNoCRDs(t *testing.T) {
-	f := newFixture(t, "", "", "", false, "false")
+	f := newFixture(t, "", false, "false")
 
 	AutodiscoverCRDsToActions(f.portConfig, f.apiextensionClient, f.portClient)
 

pkg/defaults/defaults_test.go

@@ -1,7 +1,6 @@
 package defaults
 
 import (
-	"fmt"
 	"testing"
 
 	guuid "github.com/google/uuid"
@@ -23,11 +22,7 @@ type Fixture struct {
 
 func NewFixture(t *testing.T) *Fixture {
 	stateKey := guuid.NewString()
-	portClient, err := cli.New(config.ApplicationConfig.PortBaseURL, cli.WithHeader("User-Agent", fmt.Sprintf("port-k8s-exporter/0.1 (statekey/%s)", stateKey)),
-		cli.WithClientID(config.ApplicationConfig.PortClientId), cli.WithClientSecret(config.ApplicationConfig.PortClientSecret))
-	if err != nil {
-		t.Errorf("Error building Port client: %s", err.Error())
-	}
+	portClient := cli.New(config.ApplicationConfig)
 
 	deleteDefaultResources(portClient, stateKey)
 	return &Fixture{

pkg/event_handler/event_listener_factory.go

@@ -2,6 +2,7 @@ package event_handler
 
 import (
 	"fmt"
+
 	"github.com/port-labs/port-k8s-exporter/pkg/event_handler/consumer"
 	"github.com/port-labs/port-k8s-exporter/pkg/event_handler/polling"
 	"github.com/port-labs/port-k8s-exporter/pkg/port/cli"

pkg/event_handler/polling/polling_test.go

@@ -1,17 +1,17 @@
 package polling
 
 import (
-	"fmt"
 	_ "github.com/port-labs/port-k8s-exporter/test_utils"
 
+	"testing"
+	"time"
+
 	guuid "github.com/google/uuid"
 	"github.com/port-labs/port-k8s-exporter/pkg/config"
 	"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/stretchr/testify/assert"
-	"testing"
-	"time"
 )
 
 type Fixture struct {
@@ -31,14 +31,10 @@ func (m *MockTicker) GetC() <-chan time.Time {
 
 func NewFixture(t *testing.T, c chan time.Time) *Fixture {
 	stateKey := guuid.NewString()
-	portClient, err := cli.New(config.ApplicationConfig.PortBaseURL, cli.WithHeader("User-Agent", fmt.Sprintf("port-k8s-exporter/0.1 (statekey/%s)", stateKey)),
-		cli.WithClientID(config.ApplicationConfig.PortClientId), cli.WithClientSecret(config.ApplicationConfig.PortClientSecret))
-	if err != nil {
-		t.Errorf("Error building Port client: %s", err.Error())
-	}
+	portClient := cli.New(config.ApplicationConfig)
 
 	_ = integration.DeleteIntegration(portClient, stateKey)
-	err = integration.CreateIntegration(portClient, stateKey, "", &port.IntegrationAppConfig{
+	err := integration.CreateIntegration(portClient, stateKey, "", &port.IntegrationAppConfig{
 		Resources: []port.Resource{},
 	})
 	if err != nil {

pkg/handlers/controllers.go

@@ -2,9 +2,11 @@ package handlers
 
 import (
 	"context"
-	"github.com/port-labs/port-k8s-exporter/pkg/port/integration"
 	"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"
@@ -51,7 +53,7 @@ func NewControllersHandler(exporterConfig *port.Config, portConfig *port.Integra
 		}
 
 		informer := informersFactory.ForResource(gvr)
-		controller := k8s.NewController(port.AggregatedResource{Kind: kind, KindConfigs: kindConfigs}, portClient, informer, portConfig)
+		controller := k8s.NewController(port.AggregatedResource{Kind: kind, KindConfigs: kindConfigs}, informer, portConfig, config.ApplicationConfig)
 		controllers = append(controllers, controller)
 	}
 

pkg/jq/parser.go

@@ -5,15 +5,12 @@ import (
 	"os"
 	"reflect"
 	"strings"
-	"sync"
 
 	"github.com/itchyny/gojq"
 	"github.com/port-labs/port-k8s-exporter/pkg/goutils"
 	"k8s.io/klog/v2"
 )
 
-var mutex = &sync.Mutex{}
-
 func runJQQuery(jqQuery string, obj interface{}) (interface{}, error) {
 	query, err := gojq.Parse(jqQuery)
 	if err != nil {
@@ -30,10 +27,7 @@ func runJQQuery(jqQuery string, obj interface{}) (interface{}, error) {
 		klog.Warningf("failed to compile jq query: %s", jqQuery)
 		return nil, err
 	}
-
-	mutex.Lock()
 	queryRes, ok := code.Run(obj).Next()
-	mutex.Unlock()
 
 	if !ok {
 		return nil, fmt.Errorf("query should return at least one value")

pkg/k8s/controller.go

@@ -5,19 +5,21 @@ import (
 	"fmt"
 	"time"
 
+	"hash/fnv"
+	"strconv"
+
 	"github.com/port-labs/port-k8s-exporter/pkg/config"
 	"github.com/port-labs/port-k8s-exporter/pkg/jq"
 	"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/mapping"
-	"hash/fnv"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/informers"
-	"strconv"
 
 	"encoding/json"
+
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
@@ -49,7 +51,12 @@ type Controller struct {
 	workqueue         workqueue.RateLimitingInterface
 }
 
-func NewController(resource port.AggregatedResource, portClient *cli.PortClient, informer informers.GenericInformer, integrationConfig *port.IntegrationAppConfig) *Controller {
+func NewController(resource port.AggregatedResource, informer informers.GenericInformer, integrationConfig *port.IntegrationAppConfig, applicationConfig *config.ApplicationConfiguration) *Controller {
+	// We create a new Port client for each controller because the Resty client is not thread-safe.
+	portClient := cli.New(applicationConfig)
+
+	cli.WithDeleteDependents(integrationConfig.DeleteDependents)(portClient)
+	cli.WithCreateMissingRelatedEntities(integrationConfig.CreateMissingRelatedEntities)(portClient)
 	controller := &Controller{
 		Resource:          resource,
 		portClient:        portClient,
@@ -247,7 +254,7 @@ func (c *Controller) getObjectEntities(obj interface{}, selector port.Selector,
 		return nil, nil, fmt.Errorf("error casting to unstructured")
 	}
 	var structuredObj interface{}
-	err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredObj.Object, &structuredObj)
+	err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredObj.DeepCopy().Object, &structuredObj)
 	if err != nil {
 		return nil, nil, fmt.Errorf("error converting from unstructured: %v", err)
 	}