simplify the requests, posts, topics, routes and systems counters

This commit is contained in:
Azareal 2019-07-28 13:46:19 +10:00
parent 6d805a13cf
commit 01ccdb2087
5 changed files with 117 additions and 100 deletions

View File

@ -6,6 +6,7 @@ import (
c "github.com/Azareal/Gosora/common" c "github.com/Azareal/Gosora/common"
"github.com/Azareal/Gosora/query_gen" "github.com/Azareal/Gosora/query_gen"
"github.com/pkg/errors"
) )
var PostCounter *DefaultPostCounter var PostCounter *DefaultPostCounter
@ -19,40 +20,44 @@ type DefaultPostCounter struct {
func NewPostCounter() (*DefaultPostCounter, error) { func NewPostCounter() (*DefaultPostCounter, error) {
acc := qgen.NewAcc() acc := qgen.NewAcc()
counter := &DefaultPostCounter{ co := &DefaultPostCounter{
currentBucket: 0, currentBucket: 0,
insert: acc.Insert("postchunks").Columns("count, createdAt").Fields("?,UTC_TIMESTAMP()").Prepare(), insert: acc.Insert("postchunks").Columns("count, createdAt").Fields("?,UTC_TIMESTAMP()").Prepare(),
} }
c.AddScheduledFifteenMinuteTask(counter.Tick) c.AddScheduledFifteenMinuteTask(co.Tick)
//c.AddScheduledSecondTask(counter.Tick) //c.AddScheduledSecondTask(co.Tick)
c.AddShutdownTask(counter.Tick) c.AddShutdownTask(co.Tick)
return counter, acc.FirstError() return co, acc.FirstError()
} }
func (counter *DefaultPostCounter) Tick() (err error) { func (co *DefaultPostCounter) Tick() (err error) {
var oldBucket = counter.currentBucket oldBucket := co.currentBucket
var nextBucket int64 // 0 var nextBucket int64 // 0
if counter.currentBucket == 0 { if co.currentBucket == 0 {
nextBucket = 1 nextBucket = 1
} }
atomic.AddInt64(&counter.buckets[oldBucket], counter.buckets[nextBucket]) atomic.AddInt64(&co.buckets[oldBucket], co.buckets[nextBucket])
atomic.StoreInt64(&counter.buckets[nextBucket], 0) atomic.StoreInt64(&co.buckets[nextBucket], 0)
atomic.StoreInt64(&counter.currentBucket, nextBucket) atomic.StoreInt64(&co.currentBucket, nextBucket)
var previousViewChunk = counter.buckets[oldBucket] previousViewChunk := co.buckets[oldBucket]
atomic.AddInt64(&counter.buckets[oldBucket], -previousViewChunk) atomic.AddInt64(&co.buckets[oldBucket], -previousViewChunk)
return counter.insertChunk(previousViewChunk) err = co.insertChunk(previousViewChunk)
if err != nil {
return errors.Wrap(errors.WithStack(err),"post counter")
}
return nil
} }
func (counter *DefaultPostCounter) Bump() { func (co *DefaultPostCounter) Bump() {
atomic.AddInt64(&counter.buckets[counter.currentBucket], 1) atomic.AddInt64(&co.buckets[co.currentBucket], 1)
} }
func (counter *DefaultPostCounter) insertChunk(count int64) error { func (co *DefaultPostCounter) insertChunk(count int64) error {
if count == 0 { if count == 0 {
return nil return nil
} }
c.DebugLogf("Inserting a postchunk with a count of %d", count) c.DebugLogf("Inserting a postchunk with a count of %d", count)
_, err := counter.insert.Exec(count) _, err := co.insert.Exec(count)
return err return err
} }

View File

@ -4,8 +4,9 @@ import (
"database/sql" "database/sql"
"sync/atomic" "sync/atomic"
"github.com/Azareal/Gosora/common" c "github.com/Azareal/Gosora/common"
qgen "github.com/Azareal/Gosora/query_gen" qgen "github.com/Azareal/Gosora/query_gen"
"github.com/pkg/errors"
) )
// TODO: Rename this? // TODO: Rename this?
@ -20,40 +21,44 @@ type DefaultViewCounter struct {
} }
func NewGlobalViewCounter(acc *qgen.Accumulator) (*DefaultViewCounter, error) { func NewGlobalViewCounter(acc *qgen.Accumulator) (*DefaultViewCounter, error) {
counter := &DefaultViewCounter{ co := &DefaultViewCounter{
currentBucket: 0, currentBucket: 0,
insert: acc.Insert("viewchunks").Columns("count, createdAt, route").Fields("?,UTC_TIMESTAMP(),''").Prepare(), insert: acc.Insert("viewchunks").Columns("count, createdAt, route").Fields("?,UTC_TIMESTAMP(),''").Prepare(),
} }
common.AddScheduledFifteenMinuteTask(counter.Tick) // This is run once every fifteen minutes to match the frequency of the RouteViewCounter c.AddScheduledFifteenMinuteTask(co.Tick) // This is run once every fifteen minutes to match the frequency of the RouteViewCounter
//common.AddScheduledSecondTask(counter.Tick) //c.AddScheduledSecondTask(co.Tick)
common.AddShutdownTask(counter.Tick) c.AddShutdownTask(co.Tick)
return counter, acc.FirstError() return co, acc.FirstError()
} }
func (counter *DefaultViewCounter) Tick() (err error) { func (co *DefaultViewCounter) Tick() (err error) {
var oldBucket = counter.currentBucket oldBucket := co.currentBucket
var nextBucket int64 // 0 var nextBucket int64 // 0
if counter.currentBucket == 0 { if co.currentBucket == 0 {
nextBucket = 1 nextBucket = 1
} }
atomic.AddInt64(&counter.buckets[oldBucket], counter.buckets[nextBucket]) atomic.AddInt64(&co.buckets[oldBucket], co.buckets[nextBucket])
atomic.StoreInt64(&counter.buckets[nextBucket], 0) atomic.StoreInt64(&co.buckets[nextBucket], 0)
atomic.StoreInt64(&counter.currentBucket, nextBucket) atomic.StoreInt64(&co.currentBucket, nextBucket)
var previousViewChunk = counter.buckets[oldBucket] previousViewChunk := co.buckets[oldBucket]
atomic.AddInt64(&counter.buckets[oldBucket], -previousViewChunk) atomic.AddInt64(&co.buckets[oldBucket], -previousViewChunk)
return counter.insertChunk(previousViewChunk) err = co.insertChunk(previousViewChunk)
if err != nil {
return errors.Wrap(errors.WithStack(err), "req counter")
}
return nil
} }
func (counter *DefaultViewCounter) Bump() { func (co *DefaultViewCounter) Bump() {
atomic.AddInt64(&counter.buckets[counter.currentBucket], 1) atomic.AddInt64(&co.buckets[co.currentBucket], 1)
} }
func (counter *DefaultViewCounter) insertChunk(count int64) error { func (co *DefaultViewCounter) insertChunk(count int64) error {
if count == 0 { if count == 0 {
return nil return nil
} }
common.DebugLogf("Inserting a viewchunk with a count of %d", count) c.DebugLogf("Inserting a vchunk with a count of %d", count)
_, err := counter.insert.Exec(count) _, err := co.insert.Exec(count)
return err return err
} }

View File

@ -3,8 +3,9 @@ package counters
import ( import (
"database/sql" "database/sql"
"github.com/Azareal/Gosora/common" c "github.com/Azareal/Gosora/common"
qgen "github.com/Azareal/Gosora/query_gen" qgen "github.com/Azareal/Gosora/query_gen"
"github.com/pkg/errors"
) )
var RouteViewCounter *DefaultRouteViewCounter var RouteViewCounter *DefaultRouteViewCounter
@ -16,53 +17,53 @@ type DefaultRouteViewCounter struct {
} }
func NewDefaultRouteViewCounter(acc *qgen.Accumulator) (*DefaultRouteViewCounter, error) { func NewDefaultRouteViewCounter(acc *qgen.Accumulator) (*DefaultRouteViewCounter, error) {
var routeBuckets = make([]*RWMutexCounterBucket, len(routeMapEnum)) routeBuckets := make([]*RWMutexCounterBucket, len(routeMapEnum))
for bucketID, _ := range routeBuckets { for bucketID, _ := range routeBuckets {
routeBuckets[bucketID] = &RWMutexCounterBucket{counter: 0} routeBuckets[bucketID] = &RWMutexCounterBucket{counter: 0}
} }
counter := &DefaultRouteViewCounter{ co := &DefaultRouteViewCounter{
buckets: routeBuckets, buckets: routeBuckets,
insert: acc.Insert("viewchunks").Columns("count, createdAt, route").Fields("?,UTC_TIMESTAMP(),?").Prepare(), insert: acc.Insert("viewchunks").Columns("count, createdAt, route").Fields("?,UTC_TIMESTAMP(),?").Prepare(),
} }
common.AddScheduledFifteenMinuteTask(counter.Tick) // There could be a lot of routes, so we don't want to be running this every second c.AddScheduledFifteenMinuteTask(co.Tick) // There could be a lot of routes, so we don't want to be running this every second
//common.AddScheduledSecondTask(counter.Tick) //c.AddScheduledSecondTask(co.Tick)
common.AddShutdownTask(counter.Tick) c.AddShutdownTask(co.Tick)
return counter, acc.FirstError() return co, acc.FirstError()
} }
func (counter *DefaultRouteViewCounter) Tick() error { func (co *DefaultRouteViewCounter) Tick() error {
for routeID, routeBucket := range counter.buckets { for routeID, routeBucket := range co.buckets {
var count int var count int
routeBucket.RLock() routeBucket.RLock()
count = routeBucket.counter count = routeBucket.counter
routeBucket.counter = 0 routeBucket.counter = 0
routeBucket.RUnlock() routeBucket.RUnlock()
err := counter.insertChunk(count, routeID) // TODO: Bulk insert for speed? err := co.insertChunk(count, routeID) // TODO: Bulk insert for speed?
if err != nil { if err != nil {
return err return errors.Wrap(errors.WithStack(err), "route counter")
} }
} }
return nil return nil
} }
func (counter *DefaultRouteViewCounter) insertChunk(count int, route int) error { func (co *DefaultRouteViewCounter) insertChunk(count int, route int) error {
if count == 0 { if count == 0 {
return nil return nil
} }
var routeName = reverseRouteMapEnum[route] routeName := reverseRouteMapEnum[route]
common.DebugLogf("Inserting a viewchunk with a count of %d for route %s (%d)", count, routeName, route) c.DebugLogf("Inserting a vchunk with a count of %d for route %s (%d)", count, routeName, route)
_, err := counter.insert.Exec(count, routeName) _, err := co.insert.Exec(count, routeName)
return err return err
} }
func (counter *DefaultRouteViewCounter) Bump(route int) { func (co *DefaultRouteViewCounter) Bump(route int) {
// TODO: Test this check // TODO: Test this check
common.DebugDetail("counter.buckets[", route, "]: ", counter.buckets[route]) c.DebugDetail("co.buckets[", route, "]: ", co.buckets[route])
if len(counter.buckets) <= route || route < 0 { if len(co.buckets) <= route || route < 0 {
return return
} }
counter.buckets[route].Lock() co.buckets[route].Lock()
counter.buckets[route].counter++ co.buckets[route].counter++
counter.buckets[route].Unlock() co.buckets[route].Unlock()
} }

View File

@ -3,8 +3,9 @@ package counters
import ( import (
"database/sql" "database/sql"
"github.com/Azareal/Gosora/common" c "github.com/Azareal/Gosora/common"
"github.com/Azareal/Gosora/query_gen" qgen "github.com/Azareal/Gosora/query_gen"
"github.com/pkg/errors"
) )
var OSViewCounter *DefaultOSViewCounter var OSViewCounter *DefaultOSViewCounter
@ -19,49 +20,49 @@ func NewDefaultOSViewCounter(acc *qgen.Accumulator) (*DefaultOSViewCounter, erro
for bucketID, _ := range osBuckets { for bucketID, _ := range osBuckets {
osBuckets[bucketID] = &RWMutexCounterBucket{counter: 0} osBuckets[bucketID] = &RWMutexCounterBucket{counter: 0}
} }
counter := &DefaultOSViewCounter{ co := &DefaultOSViewCounter{
buckets: osBuckets, buckets: osBuckets,
insert: acc.Insert("viewchunks_systems").Columns("count, createdAt, system").Fields("?,UTC_TIMESTAMP(),?").Prepare(), insert: acc.Insert("viewchunks_systems").Columns("count, createdAt, system").Fields("?,UTC_TIMESTAMP(),?").Prepare(),
} }
common.AddScheduledFifteenMinuteTask(counter.Tick) c.AddScheduledFifteenMinuteTask(co.Tick)
//common.AddScheduledSecondTask(counter.Tick) //c.AddScheduledSecondTask(co.Tick)
common.AddShutdownTask(counter.Tick) c.AddShutdownTask(co.Tick)
return counter, acc.FirstError() return co, acc.FirstError()
} }
func (counter *DefaultOSViewCounter) Tick() error { func (co *DefaultOSViewCounter) Tick() error {
for id, bucket := range counter.buckets { for id, bucket := range co.buckets {
var count int var count int
bucket.RLock() bucket.RLock()
count = bucket.counter count = bucket.counter
bucket.counter = 0 // TODO: Add a SetZero method to reduce the amount of duplicate code between the OS and agent counters? bucket.counter = 0 // TODO: Add a SetZero method to reduce the amount of duplicate code between the OS and agent counters?
bucket.RUnlock() bucket.RUnlock()
err := counter.insertChunk(count, id) // TODO: Bulk insert for speed? err := co.insertChunk(count, id) // TODO: Bulk insert for speed?
if err != nil { if err != nil {
return err return errors.Wrap(errors.WithStack(err), "system counter")
} }
} }
return nil return nil
} }
func (counter *DefaultOSViewCounter) insertChunk(count int, os int) error { func (co *DefaultOSViewCounter) insertChunk(count int, os int) error {
if count == 0 { if count == 0 {
return nil return nil
} }
var osName = reverseOSMapEnum[os] osName := reverseOSMapEnum[os]
common.DebugLogf("Inserting a viewchunk with a count of %d for OS %s (%d)", count, osName, os) c.DebugLogf("Inserting a vchunk with a count of %d for OS %s (%d)", count, osName, os)
_, err := counter.insert.Exec(count, osName) _, err := co.insert.Exec(count, osName)
return err return err
} }
func (counter *DefaultOSViewCounter) Bump(id int) { func (co *DefaultOSViewCounter) Bump(id int) {
// TODO: Test this check // TODO: Test this check
common.DebugDetail("counter.buckets[", id, "]: ", counter.buckets[id]) c.DebugDetail("co.buckets[", id, "]: ", co.buckets[id])
if len(counter.buckets) <= id || id < 0 { if len(co.buckets) <= id || id < 0 {
return return
} }
counter.buckets[id].Lock() co.buckets[id].Lock()
counter.buckets[id].counter++ co.buckets[id].counter++
counter.buckets[id].Unlock() co.buckets[id].Unlock()
} }

View File

@ -4,8 +4,9 @@ import (
"database/sql" "database/sql"
"sync/atomic" "sync/atomic"
"github.com/Azareal/Gosora/common" c "github.com/Azareal/Gosora/common"
"github.com/Azareal/Gosora/query_gen" "github.com/Azareal/Gosora/query_gen"
"github.com/pkg/errors"
) )
var TopicCounter *DefaultTopicCounter var TopicCounter *DefaultTopicCounter
@ -19,40 +20,44 @@ type DefaultTopicCounter struct {
func NewTopicCounter() (*DefaultTopicCounter, error) { func NewTopicCounter() (*DefaultTopicCounter, error) {
acc := qgen.NewAcc() acc := qgen.NewAcc()
counter := &DefaultTopicCounter{ co := &DefaultTopicCounter{
currentBucket: 0, currentBucket: 0,
insert: acc.Insert("topicchunks").Columns("count, createdAt").Fields("?,UTC_TIMESTAMP()").Prepare(), insert: acc.Insert("topicchunks").Columns("count, createdAt").Fields("?,UTC_TIMESTAMP()").Prepare(),
} }
common.AddScheduledFifteenMinuteTask(counter.Tick) c.AddScheduledFifteenMinuteTask(co.Tick)
//common.AddScheduledSecondTask(counter.Tick) //c.AddScheduledSecondTask(co.Tick)
common.AddShutdownTask(counter.Tick) c.AddShutdownTask(co.Tick)
return counter, acc.FirstError() return co, acc.FirstError()
} }
func (counter *DefaultTopicCounter) Tick() (err error) { func (co *DefaultTopicCounter) Tick() (err error) {
var oldBucket = counter.currentBucket oldBucket := co.currentBucket
var nextBucket int64 // 0 var nextBucket int64 // 0
if counter.currentBucket == 0 { if co.currentBucket == 0 {
nextBucket = 1 nextBucket = 1
} }
atomic.AddInt64(&counter.buckets[oldBucket], counter.buckets[nextBucket]) atomic.AddInt64(&co.buckets[oldBucket], co.buckets[nextBucket])
atomic.StoreInt64(&counter.buckets[nextBucket], 0) atomic.StoreInt64(&co.buckets[nextBucket], 0)
atomic.StoreInt64(&counter.currentBucket, nextBucket) atomic.StoreInt64(&co.currentBucket, nextBucket)
var previousViewChunk = counter.buckets[oldBucket] previousViewChunk := co.buckets[oldBucket]
atomic.AddInt64(&counter.buckets[oldBucket], -previousViewChunk) atomic.AddInt64(&co.buckets[oldBucket], -previousViewChunk)
return counter.insertChunk(previousViewChunk) err = co.insertChunk(previousViewChunk)
if err != nil {
return errors.Wrap(errors.WithStack(err),"topics counter")
}
return nil
} }
func (counter *DefaultTopicCounter) Bump() { func (co *DefaultTopicCounter) Bump() {
atomic.AddInt64(&counter.buckets[counter.currentBucket], 1) atomic.AddInt64(&co.buckets[co.currentBucket], 1)
} }
func (counter *DefaultTopicCounter) insertChunk(count int64) error { func (co *DefaultTopicCounter) insertChunk(count int64) error {
if count == 0 { if count == 0 {
return nil return nil
} }
common.DebugLogf("Inserting a topicchunk with a count of %d", count) c.DebugLogf("Inserting a topicchunk with a count of %d", count)
_, err := counter.insert.Exec(count) _, err := co.insert.Exec(count)
return err return err
} }