upstream/icingadb
1
0
Fork 0
mirror of https://github.com/Icinga/icingadb.git synced 2026-05-28 04:35:54 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Make counters atomic

Browse source 
refs #50
This commit is contained in:
Justin Müller 2019-12-04 11:22:04 +01:00
parent 448c3b4b0d
commit d25443859c
2 changed files with 25 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
configobject/history/history.go
View file

@ -11,7 +11,7 @@ import (
"github.com/prometheus/client_golang/prometheus"
log "github.com/sirupsen/logrus"
"strconv"
"sync"
"sync/atomic"
"time"
)
@ -34,23 +34,20 @@ var mysqlObservers = struct {
}
var historyCounter = struct {
state int
notification int
usernotification int
downtime int
comment int
flapping int
acknowledgement int
state uint64
notification uint64
usernotification uint64
downtime uint64
comment uint64
flapping uint64
acknowledgement uint64
}{}
var historyCounterLock = sync.Mutex{}
func printAndResetHistoryCounter(counter *uint64, historyType string) {
amount := atomic.SwapUint64(counter, 0)
func printAndResetHistoryCounter(counter *int, historyType string) {
if *counter > 0 {
log.Infof("Added %d %s history entries in the last 20 seconds", *counter, historyType)
historyCounterLock.Lock()
*counter = 0
historyCounterLock.Unlock()
if amount > 0 {
log.Infof("Added %d %s history entries in the last 20 seconds", amount, historyType)
}
}
@ -505,7 +502,7 @@ func acknowledgementHistoryWorker(super *supervisor.Supervisor) {
historyWorker(super, "acknowledgement", statements, dataFunctions, mysqlObservers.acknowledgement, &historyCounter.acknowledgement)
}
func historyWorker(super *supervisor.Supervisor, historyType string, preparedStatements []string, dataFunctions []func(map[string]interface{}) []interface{}, observer prometheus.Observer, counter *int) {
func historyWorker(super *supervisor.Supervisor, historyType string, preparedStatements []string, dataFunctions []func(map[string]interface{}) []interface{}, observer prometheus.Observer, counter *uint64) {
if super.EnvId == nil {
log.Debug(historyType + "History: Waiting for EnvId to be set")
time.Sleep(time.Second)
@ -574,10 +571,7 @@ func historyWorker(super *supervisor.Supervisor, historyType string, preparedSta
super.Rdbw.XDel("icinga:history:stream:"+historyType, storedEntryIds...)
count := len(storedEntryIds) - brokenEntries
historyCounterLock.Lock()
*counter++
historyCounterLock.Unlock()
atomic.AddUint64(counter, 1)
log.Debugf("%d %s history entries synced", count, historyType)
log.Debugf("%d %s history entries broken", brokenEntries, historyType)

21
configobject/statesync/statesync.go
View file

@ -12,13 +12,14 @@ import (
log "github.com/sirupsen/logrus"
"strconv"
"sync"
"sync/atomic"
"time"
)
// syncCounter counts on how many host/service states have synced since the last logSyncCounters().
var syncCounter = struct {
host int
service int
host uint64
service uint64
}{}
var syncCounterLock = sync.Mutex{}
@ -57,18 +58,18 @@ func logSyncCounters() {
for {
<-every20s.C
if syncCounter.host > 0 || syncCounter.service > 0 {
log.Infof("Synced %d host and %d service states in the last 20 seconds", syncCounter.host, syncCounter.service)
syncCounterLock.Lock()
syncCounter.host = 0
syncCounter.service = 0
syncCounterLock.Unlock()
host := atomic.SwapUint64(&syncCounter.host, 0)
service := atomic.SwapUint64(&syncCounter.service, 0)
if host > 0 || service > 0 {
log.Infof("Synced %d host and %d service states in the last 20 seconds", host, service)
}
}
}
// syncStates tries to sync the states of given object type every second.
func syncStates(super *supervisor.Supervisor, objectType string, counter *int, observer prometheus.Observer) {
func syncStates(super *supervisor.Supervisor, objectType string, counter *uint64, observer prometheus.Observer) {
if super.EnvId == nil {
log.Debug("StateSync: Waiting for EnvId to be set")
time.Sleep(time.Second)
@ -182,7 +183,7 @@ func syncStates(super *supervisor.Supervisor, objectType string, counter *int, o
log.Debugf("%d %s state synced", len(storedStateIds)-brokenStates, objectType)
log.Debugf("%d %s state broken", brokenStates, objectType)
syncCounterLock.Lock()
*counter += len(storedStateIds)
atomic.AddUint64(counter, uint64(len(storedStateIds)))
syncCounterLock.Unlock()
StateSyncsTotal.WithLabelValues(objectType).Add(float64(len(storedStateIds)))
}