eventstream: clean up mutex locks

First, the last event's timestamp will now be fetched before entering
the checkMissedStateChanges resp. checkMissedAcknowledgements method
four times, which shouldn't be a problem.. unless it becomes one.

Also to clearly signal what should be locked, the initial declarations
in the Process method were protected, even when not necessary.

Author: oxzi

internal/eventstream/client.go

@@ -193,8 +193,11 @@ func (client *Client) handleEvent(ev *event.Event, source string) {
 // loop takes care of reconnections, all those events will be logged while generated Events will be dispatched to the
 // callback function.
 func (client *Client) Process() {
+	client.eventsHandlerMutex.Lock()
 	client.eventsRingBuffer = make([]uint64, 1024)
 	client.eventsRingBufferPos = 0
+	client.eventsLastTs = time.Time{}
+	client.eventsHandlerMutex.Unlock()
 
 	defer client.Logger.Info("Event Stream Client has stopped")
 
@@ -220,9 +223,13 @@ func (client *Client) Process() {
 			client.Logger.Errorf("Cannot reestablish an API connection: %v", err)
 		}
 
-		go client.checkMissedStateChanges("host")
-		go client.checkMissedStateChanges("service")
-		go client.checkMissedAcknowledgements("host")
-		go client.checkMissedAcknowledgements("service")
+		client.eventsHandlerMutex.RLock()
+		lastEventTime := client.eventsLastTs
+		client.eventsHandlerMutex.RUnlock()
+
+		go client.checkMissedStateChanges("host", lastEventTime)
+		go client.checkMissedStateChanges("service", lastEventTime)
+		go client.checkMissedAcknowledgements("host", lastEventTime)
+		go client.checkMissedAcknowledgements("service", lastEventTime)
 	}
 }

internal/eventstream/client_api.go

@@ -193,11 +193,8 @@ func (client *Client) checkMissedChanges(objType, filterExpr string, attrsCallba
 }
 
 // checkMissedStateChanges fetches missed Host or Service state changes and feeds them into the handler.
-func (client *Client) checkMissedStateChanges(objType string) {
-	client.eventsHandlerMutex.RLock()
-	filterExpr := fmt.Sprintf("%s.last_state_change>%f",
-		objType, float64(client.eventsLastTs.UnixMicro())/1_000_000.0)
-	client.eventsHandlerMutex.RUnlock()
+func (client *Client) checkMissedStateChanges(objType string, since time.Time) {
+	filterExpr := fmt.Sprintf("%s.last_state_change>%f", objType, float64(since.UnixMicro())/1_000_000.0)
 
 	client.checkMissedChanges(objType, filterExpr, func(attrs *HostServiceRuntimeAttributes, host, service string) {
 		ev, err := client.buildHostServiceEvent(attrs.LastCheckResult, attrs.State, host, service)
@@ -211,11 +208,9 @@ func (client *Client) checkMissedStateChanges(objType string) {
 }
 
 // checkMissedAcknowledgements fetches missed set Host or Service Acknowledgements and feeds them into the handler.
-func (client *Client) checkMissedAcknowledgements(objType string) {
-	client.eventsHandlerMutex.RLock()
+func (client *Client) checkMissedAcknowledgements(objType string, since time.Time) {
 	filterExpr := fmt.Sprintf("%s.acknowledgement && %s.acknowledgement_last_change>%f",
-		objType, objType, float64(client.eventsLastTs.UnixMicro())/1_000_000.0)
-	client.eventsHandlerMutex.RUnlock()
+		objType, objType, float64(since.UnixMicro())/1_000_000.0)
 
 	client.checkMissedChanges(objType, filterExpr, func(attrs *HostServiceRuntimeAttributes, host, service string) {
 		ackComment, err := client.fetchAcknowledgementComment(host, service, attrs.AcknowledgementLastChange.Time)