Clean mqueue task now show before/after

openldap-log-parser/cmd/root.go

@@ -125,7 +125,7 @@ var (
 	File   os.File
 	Writer *bufio.Writer
 
-	Version = "0.6.6"
+	Version = "0.6.10"
 
 	BuildInfo = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Name: "openldaplogparser_build_info",
@@ -217,7 +217,7 @@ func OlcToFlat(olc *OpenLdapConnection) []OpenLdapConnectionFlat {
 
 	for i := range olc.Operations {
 		olcf[i] = OpenLdapConnectionFlat{
-			Time:           olc.Time,
+			Time:           olc.Operations[i].Time,
 			Hostname:       olc.Hostname,
 			Process:        olc.Process,
 			ClientIp:       olc.ClientIp,
@@ -321,35 +321,39 @@ func writeOut(msg string, filename string) error {
 	return nil
 }
 
-// Every 24H, remove sent, milter-rejected and deferred that entered queue more than 5 days ago
+func cleanMQueue(mqueue map[string]*OpenLdapConnection, mqMtx *sync.Mutex, age time.Duration) {
-/*
+    var ok bool
-func periodicallyCleanMQueue(mqueue map[int]*PostfixLogParser, mqMtx *sync.Mutex) {
-	var ok int
 
-	for range time.Tick(time.Hour * 24) {
+	log.Printf("Start cleaning queue task: %d items in queue", len(mqueue))
-		// Do we need read lock?
+
-		for _, inmail := range mqueue {
+	// Do we need read lock?
-			ok = 0
+	for uid, ldcon := range mqueue {
-			// Check all mails were sent (multiple destinations mails)
+		ok = false
-			//  or rejected
+		// Check if a close operation exist
-			for _, outmail := range inmail.Messages {
+		for _, op := range ldcon.Operations {
-				if outmail.Status == "sent" || outmail.Status == "milter-reject" {
+			if op.OpType == "close" {
-					ok += 1
+				if op.Time.Add(age).Before(time.Now()) {
-				} else if outmail.Status == "deferred" {
+					ok = true
-					if inmail.Time.Add(time.Hour * 5 * 24).Before(time.Now()) {
-						ok += 1
-					}
 				}
 			}
-			if ok == len(inmail.Messages) {
+		}
-				mqMtx.Lock()
+		if ok == true {
-				delete(mqueue, inmail.MessageId)
+			mqMtx.Lock()
-				mqMtx.Unlock()
+			delete(mqueue, uid)
-			}
+			mqMtx.Unlock()
 		}
 	}
+	log.Printf("Finished cleaning queue task: %d items in queue", len(mqueue))
+}
+
+
+
+// Every 24H, remove connections closed more than 24 hours ago
+func periodicallyCleanMQueue(mqueue map[string]*OpenLdapConnection, mqMtx *sync.Mutex) {
+	for range time.Tick(time.Hour * 24) {
+		cleanMQueue(mqueue, mqMtx, 24 * time.Hour)
+	}
 }
-*/
 
 func initConfig() {}
 
@@ -975,7 +979,18 @@ func processLogs(cmd *cobra.Command, args []string) {
 	}
 
 	// Cleaner thread
-	//go periodicallyCleanMQueue(mQueue, &mqMtx)
+	go periodicallyCleanMQueue(mQueue, &mqMtx)
+
+    // On demand Mqueue cleaning... For debug, dont try this at home, kids!
+/*	sig2 := make(chan os.Signal)
+	signal.Notify(sig2, syscall.SIGUSR2)
+	go func() {
+		for {
+			<-sig2
+			cleanMQueue(mQueue, &mqMtx, 5 * time.Minute)
+		}
+	}()
+*/
 
 	// Initialize Stdin input...
 	if true == strings.EqualFold(gSyslogListenAddress, "do-not-listen") {