Port 5363 UI for the kubernetes exporter moving the config into port

main.go

@@ -85,7 +85,7 @@ func main() {
 	}
 
 	klog.Info("Starting controllers handler")
-	err = event_handler.StartEventHandler(eventListener, func() (event_handler.IStoppableRsync, error) {
+	err = event_handler.Start(eventListener, func() (event_handler.IStoppableRsync, error) {
 		return initiateHandler(exporterConfig, k8sClient, portClient)
 	})
 

pkg/event_handler/consumer/consumer_test.go

@@ -18,6 +18,7 @@ type Fixture struct {
 
 type MockConsumer struct {
 	pollData kafka.Event
+	close    func()
 }
 
 func (m *MockConsumer) SubscribeTopics(topics []string, rebalanceCb kafka.RebalanceCb) (err error) {
@@ -36,6 +37,10 @@ func (m *MockConsumer) SubscribeTopics(topics []string, rebalanceCb kafka.Rebala
 }
 
 func (m *MockConsumer) Poll(timeoutMs int) (event kafka.Event) {
+	// The consumer will poll this in while true loop so we need to close it inorder not to spam the logs
+	defer func() {
+		m.close()
+	}()
 	return m.pollData
 }
 
@@ -50,6 +55,7 @@ func (m *MockConsumer) Close() (err error) {
 func NewFixture(t *testing.T) *Fixture {
 	mock := &MockConsumer{}
 	consumer, err := NewConsumer(&config.KafkaConfiguration{}, mock)
+	mock.close = consumer.Close
 	if err != nil {
 		t.Fatalf("Error creating consumer: %v", err)
 	}

pkg/event_handler/event_handler.go

@@ -14,16 +14,16 @@ type IStoppableRsync interface {
 	Stop()
 }
 
-func StartEventHandler(eventListener IListener, controllerHandlerFactory func() (IStoppableRsync, error)) error {
-	controllerHandler, err := controllerHandlerFactory()
+func Start(eventListener IListener, initControllerHandler func() (IStoppableRsync, error)) error {
+	controllerHandler, err := initControllerHandler()
 	if err != nil {
 		return err
 	}
 
 	return eventListener.Run(func() {
 		klog.Infof("Resync request received. Recreating controllers for the new port configuration")
 		controllerHandler.Stop()
-		newController, resyncErr := controllerHandlerFactory()
+		newController, resyncErr := initControllerHandler()
 		controllerHandler = newController
 
 		if resyncErr != nil {

pkg/event_handler/event_handler_test.go

@@ -27,6 +27,9 @@ func (e *EventListenerMock) Run(resync func()) error {
 }
 
 func TestStartKafkaEventListener(t *testing.T) {
+	// Test should get a new controller handler on each call to the passed function and stop the previous one
+	// The flow for this test will be: create controller handler -> resync and stop the controller handler & create a
+	// new controller handler X 2 which will result the last controller handler to not be stopped
 	eventListenerMock := &EventListenerMock{}
 	firstResponse := &ControllerHandlerMock{}
 	secondResponse := &ControllerHandlerMock{}
@@ -37,7 +40,7 @@ func TestStartKafkaEventListener(t *testing.T) {
 		thirdResponse,
 	}
 
-	err := StartEventHandler(eventListenerMock, func() (IStoppableRsync, error) {
+	err := Start(eventListenerMock, func() (IStoppableRsync, error) {
 		r := responses[0]
 		responses = responses[1:]
 

pkg/event_handler/polling/event_listener.go

@@ -9,20 +9,21 @@ import (
 type EventListener struct {
 	stateKey   string
 	portClient *cli.PortClient
+	handler    *Handler
 }
 
 func NewEventListener(stateKey string, portClient *cli.PortClient) *EventListener {
 	return &EventListener{
 		stateKey:   stateKey,
 		portClient: portClient,
+		handler:    NewPollingHandler(config.PollingListenerRate, stateKey, portClient, nil),
 	}
 }
 
 func (l *EventListener) Run(resync func()) error {
 	klog.Infof("Starting polling event listener")
 	klog.Infof("Polling rate set to %d seconds", config.PollingListenerRate)
-	pollingHandler := NewPollingHandler(config.PollingListenerRate, l.stateKey, l.portClient, nil)
-	pollingHandler.Run(resync)
+	l.handler.Run(resync)
 
 	return nil
 }

pkg/event_handler/polling/polling.go

@@ -33,19 +33,19 @@ func (t *Ticker) GetC() <-chan time.Time {
 	return t.ticker.C
 }
 
-type HandlerSettings struct {
+type Handler struct {
 	ticker      ITicker
 	stateKey    string
 	portClient  *cli.PortClient
 	pollingRate uint
 }
 
-func NewPollingHandler(pollingRate uint, stateKey string, portClient *cli.PortClient, tickerOverride ITicker) *HandlerSettings {
+func NewPollingHandler(pollingRate uint, stateKey string, portClient *cli.PortClient, tickerOverride ITicker) *Handler {
 	ticker := tickerOverride
 	if ticker == nil {
 		ticker = NewTicker(time.Second * time.Duration(pollingRate))
 	}
-	rv := &HandlerSettings{
+	rv := &Handler{
 		ticker:      ticker,
 		stateKey:    stateKey,
 		portClient:  portClient,
@@ -54,7 +54,7 @@ func NewPollingHandler(pollingRate uint, stateKey string, portClient *cli.PortCl
 	return rv
 }
 
-func (h *HandlerSettings) Run(resync func()) {
+func (h *Handler) Run(resync func()) {
 	klog.Infof("Starting polling handler")
 	currentState, err := integration.GetIntegration(h.portClient, h.stateKey)
 	if err != nil {

pkg/port/integration/integration.go

@@ -31,6 +31,7 @@ func NewIntegration(portClient *cli.PortClient, stateKey string, eventListenerTy
 	return nil
 }
 
+// ToDo: remove this function
 func GetIntegrationConfig(portClient *cli.PortClient, stateKey string) (*port.AppConfig, error) {
 	_, err := portClient.Authenticate(context.Background(), portClient.ClientID, portClient.ClientSecret)
 	if err != nil {