mirrors/apollo-backend
2
0
Fork 0
mirror of https://github.com/christianselig/apollo-backend synced 2024-11-10 22:17:44 +00:00
Code Issues Projects Releases Packages Wiki Activity

measure if we are falling behind

Browse source
This commit is contained in:
Andre Medeiros 2021-07-07 18:09:29 -04:00
parent 3b9a1ecd3d
commit ec7bb411fd
1 changed files with 12 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
cmd/apollo-worker/main.go
View file

@ -28,8 +28,11 @@ type application struct {
client *reddit.Client client *reddit.Client
} }
var workers int = runtime.NumCPU() * 6 var (
var rate float64 = 0.1 workers int = runtime.NumCPU() * 6
rate float64 = 0.1
backoff int64 = 5
)
func accountWorker(id int, rc *reddit.Client, db *sql.DB, logger *log.Logger, statsd *statsd.Client, quit chan bool) { func accountWorker(id int, rc *reddit.Client, db *sql.DB, logger *log.Logger, statsd *statsd.Client, quit chan bool) {
authKey, err := token.AuthKeyFromBytes([]byte(os.Getenv("APPLE_KEY_PKEY"))) authKey, err := token.AuthKeyFromBytes([]byte(os.Getenv("APPLE_KEY_PKEY")))
@ -62,11 +65,11 @@ func accountWorker(id int, rc *reddit.Client, db *sql.DB, logger *log.Logger, st
query := ` query := `
SELECT id, username, access_token, refresh_token, expires_at, last_message_id, last_checked_at FROM accounts SELECT id, username, access_token, refresh_token, expires_at, last_message_id, last_checked_at FROM accounts
WHERE last_checked_at <= $1 - 5 WHERE last_checked_at <= $1 - $2
ORDER BY last_checked_at ORDER BY last_checked_at
LIMIT 1 LIMIT 1
FOR UPDATE SKIP LOCKED` FOR UPDATE SKIP LOCKED`
args := []interface{}{now} args := []interface{}{now, backoff}
account := &data.Account{} account := &data.Account{}
err = tx.QueryRow(query, args...).Scan(&account.ID, &account.Username, &account.AccessToken, &account.RefreshToken, &account.ExpiresAt, &account.LastMessageID, &account.LastCheckedAt) err = tx.QueryRow(query, args...).Scan(&account.ID, &account.Username, &account.AccessToken, &account.RefreshToken, &account.ExpiresAt, &account.LastMessageID, &account.LastCheckedAt)
@ -77,6 +80,11 @@ func accountWorker(id int, rc *reddit.Client, db *sql.DB, logger *log.Logger, st
continue continue
} }
if account.LastCheckedAt > 0 {
latency := now - account.LastCheckedAt - backoff
statsd.Histogram("apollo.queue.delay", float64(latency), []string{}, rate)
}
_, err = tx.Exec(`UPDATE accounts SET last_checked_at = $1 WHERE id = $2`, now, account.ID) _, err = tx.Exec(`UPDATE accounts SET last_checked_at = $1 WHERE id = $2`, now, account.ID)
rac := rc.NewAuthenticatedClient(account.RefreshToken, account.AccessToken) rac := rc.NewAuthenticatedClient(account.RefreshToken, account.AccessToken)