refactor(sophos): Implement consistent time-based polling with ticker

Author: JocLRojas

sophos/main.go

@@ -20,23 +20,32 @@ func main() {
 	panelServ := configuration.GetPanelServiceName()
 	client := utmconf.NewUTMClient(intKey, "http://"+panelServ)
 
-	for {
+	ticker := time.NewTicker(time.Second * delayCheck)
+	defer ticker.Stop()
+
+	startTime := time.Now().Add(-600 * time.Second).UTC()
+
+	for range ticker.C {
 		if err := utils.ConnectionChecker(configuration.CHECKCON); err != nil {
 			utils.Logger.ErrorF("External connection failure detected: %v", err)
 		}
 
+		endTime := startTime.Add(299 * time.Second).UTC()
+		startTimeStr := startTime.Format(time.RFC3339)
+		endTimeStr := endTime.Format(time.RFC3339)
+
 		moduleConfig, err := client.GetUTMConfig(enum.SOPHOS)
 		if err != nil {
 			if strings.Contains(err.Error(), "invalid character '<'") {
-				time.Sleep(time.Second * delayCheck)
+				startTime = endTime.Add(time.Second)
 				continue
 			}
 			if (err.Error() != "") && (err.Error() != " ") {
 				utils.Logger.ErrorF("error getting configuration of the SOPHOS module: %v", err)
 			}
 
 			utils.Logger.Info("sync complete waiting %v seconds", delayCheck)
-			time.Sleep(time.Second * delayCheck)
+			startTime = endTime.Add(time.Second)
 			continue
 		}
 
@@ -56,7 +65,7 @@ func main() {
 				}
 
 				if !skip {
-					processor.PullLogs(group)
+					processor.PullLogs(group, startTimeStr, endTimeStr)
 				}
 
 				wg.Done()
@@ -65,6 +74,7 @@ func main() {
 
 		wg.Wait()
 		utils.Logger.Info("sync complete waiting %d seconds", delayCheck)
-		time.Sleep(time.Second * delayCheck)
+
+		startTime = endTime.Add(time.Second)
 	}
 }

sophos/processor/pull.go

@@ -8,37 +8,30 @@ import (
 	"github.com/utmstack/config-client-go/types"
 )
 
-const delayCheck = 300
-
-var timeGroups = make(map[int]int)
 var nextKeys = make(map[int]string)
 
-func PullLogs(group types.ModuleGroup) *logger.Error {
-	utils.Logger.Info("starting log sync for : %s", group.GroupName)
-
-	epoch := int(time.Now().Unix())
+func PullLogs(group types.ModuleGroup, startTimeStr, endTimeStr string) *logger.Error {
+	utils.Logger.Info("starting log sync for : %s from %s to %s", group.GroupName, startTimeStr, endTimeStr)
 
-	_, ok := timeGroups[group.ModuleID]
-	if !ok {
-		timeGroups[group.ModuleID] = epoch - delayCheck
+	startTime, err := time.Parse(time.RFC3339, startTimeStr)
+	if err != nil {
+		return utils.Logger.ErrorF("error parsing start time: %v", err)
 	}
 
-	defer func() {
-		timeGroups[group.ModuleID] = epoch + 1
-	}()
+	startEpoch := int(startTime.Unix())
 
 	agent := getSophosCentralProcessor(group)
 
-	logs, newNextKey, err := agent.getLogs(timeGroups[group.ModuleID], nextKeys[group.ModuleID], group)
-	if err != nil {
-		return err
+	logs, newNextKey, logErr := agent.getLogs(startEpoch, nextKeys[group.ModuleID], group)
+	if logErr != nil {
+		return logErr
 	}
 
 	nextKeys[group.ModuleID] = newNextKey
 
-	err = SendToLogstash(logs)
-	if err != nil {
-		return err
+	sendErr := SendToLogstash(logs)
+	if sendErr != nil {
+		return sendErr
 	}
 
 	return nil