starting kafka config once

port-labs · Dec 13, 2023 · f1ef258 · f1ef258
1 parent 62a731d
commit f1ef258
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 15 deletions.
diff --git a/pkg/event_listener/event_listener_handler.go b/pkg/event_listener/event_listener_handler.go
@@ -6,7 +6,6 @@ import (
 	"github.com/port-labs/port-k8s-exporter/pkg/config"
 	"github.com/port-labs/port-k8s-exporter/pkg/event_listener/consumer"
 	"github.com/port-labs/port-k8s-exporter/pkg/event_listener/polling"
-	"github.com/port-labs/port-k8s-exporter/pkg/goutils"
 	"github.com/port-labs/port-k8s-exporter/pkg/handlers"
 	"github.com/port-labs/port-k8s-exporter/pkg/port"
 	"github.com/port-labs/port-k8s-exporter/pkg/port/cli"
@@ -24,19 +23,20 @@ type IncomingMessage struct {
 	} `json:"diff"`
 }
 
-var kafkaConfig = &consumer.KafkaConfiguration{
-	Brokers:                 config.NewString("event-listener-brokers", "localhost:9092", "Kafka brokers"),
-	SecurityProtocol:        config.NewString("event-listener-security-protocol", "plaintext", "Kafka security protocol"),
-	AuthenticationMechanism: config.NewString("event-listener-authentication-mechanism", "none", "Kafka authentication mechanism"),
-}
-
 type EventListener struct {
 	settings          port.EventListenerSettings
 	stateKey          string
 	controllerHandler *handlers.ControllersHandler
 	portClient        *cli.PortClient
 }
 
+var kafkaConfig = &consumer.KafkaConfiguration{
+	Brokers:                 config.NewString("event-listener-brokers", "localhost:9092", "Kafka brokers"),
+	SecurityProtocol:        config.NewString("event-listener-security-protocol", "plaintext", "Kafka security protocol"),
+	AuthenticationMechanism: config.NewString("event-listener-authentication-mechanism", "none", "Kafka authentication mechanism"),
+}
+var pollingListenerRate = config.NewUInt("event-listener-polling-rate", 60, "Polling rate for the polling event listener")
+
 func shouldResync(stateKey string, message *IncomingMessage) bool {
 	return message.Diff != nil &&
 		message.Diff.After != nil &&
@@ -102,9 +102,8 @@ func startKafkaEventListener(l *EventListener, resync func()) error {
 
 func startPollingEventListener(l *EventListener, resync func()) {
 	klog.Infof("Starting polling event listener")
-	pollingRate := goutils.GetUintEnvOrDefault("EVENT_LISTENER__POLLING_RATE", 60)
-	klog.Infof("Polling rate set to %d seconds", pollingRate)
-	pollingHandler := polling.NewPollingHandler(pollingRate, l.stateKey, l.portClient)
+	klog.Infof("Polling rate set to %d seconds", pollingListenerRate)
+	pollingHandler := polling.NewPollingHandler(pollingListenerRate, l.stateKey, l.portClient)
 	pollingHandler.Run(resync)
 }
 

diff --git a/pkg/event_listener/polling/polling.go b/pkg/event_listener/polling/polling.go
@@ -11,15 +11,15 @@ import (
 	"time"
 )
 
-type PollingHandler struct {
+type HandlerSettings struct {
 	ticker      *time.Ticker
 	stateKey    string
 	portClient  *cli.PortClient
-	pollingRate uint64
+	pollingRate uint
 }
 
-func NewPollingHandler(pollingRate uint64, stateKey string, portClient *cli.PortClient) *PollingHandler {
-	rv := &PollingHandler{
+func NewPollingHandler(pollingRate uint, stateKey string, portClient *cli.PortClient) *HandlerSettings {
+	rv := &HandlerSettings{
 		ticker:      time.NewTicker(time.Second * time.Duration(pollingRate)),
 		stateKey:    stateKey,
 		portClient:  portClient,
@@ -28,7 +28,7 @@ func NewPollingHandler(pollingRate uint64, stateKey string, portClient *cli.Port
 	return rv
 }
 
-func (h *PollingHandler) Run(resync func()) {
+func (h *HandlerSettings) Run(resync func()) {
 	klog.Infof("Starting polling handler")
 	currentState, err := integration.GetIntegrationConfig(h.portClient, h.stateKey)
 	if err != nil {