<s> from tapiriik . database import db , close_connections <EOL> from tapiriik . settings import RABBITMQ_USER_QUEUE_STATS_URL <EOL> from datetime import datetime , timedelta <EOL> import requests <EOL> distanceSyncedAggr = list ( db . sync_stats . aggregate ( [ { "<STR_LIT>" : { "<STR_LIT>" : None , "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } } ] ) ) <EOL> if distanceSyncedAggr : <EOL> distanceSynced = distanceSyncedAggr [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> else : <EOL> distanceSynced = <NUM_LIT:0> <EOL> lastDayDistanceSyncedAggr = list ( db . sync_stats . aggregate ( [ { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : datetime . utcnow ( ) - timedelta ( hours = <NUM_LIT> ) } } } , { "<STR_LIT>" : { "<STR_LIT>" : None , "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } } ] ) ) <EOL> if lastDayDistanceSyncedAggr : <EOL> lastDayDistanceSynced = lastDayDistanceSyncedAggr [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> else : <EOL> lastDayDistanceSynced = <NUM_LIT:0> <EOL> lastHourDistanceSyncedAggr = list ( db . sync_stats . aggregate ( [ { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : datetime . utcnow ( ) - timedelta ( hours = <NUM_LIT:1> ) } } } , { "<STR_LIT>" : { "<STR_LIT>" : None , "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } } ] ) ) <EOL> if lastHourDistanceSyncedAggr : <EOL> lastHourDistanceSynced = lastHourDistanceSyncedAggr [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> else : <EOL> lastHourDistanceSynced = <NUM_LIT:0> <EOL> enqueueHead = list ( db . users . find ( { "<STR_LIT>" : { "<STR_LIT>" : datetime . utcnow ( ) } , "<STR_LIT>" : None , "<STR_LIT>" : { "<STR_LIT>" : False } } , { "<STR_LIT>" : <NUM_LIT:1> } ) . sort ( "<STR_LIT>" ) . limit ( <NUM_LIT:10> ) ) <EOL> enqueueTime = timedelta ( <NUM_LIT:0> ) <EOL> if len ( enqueueHead ) : <EOL> for pendingEnqueueUser in enqueueHead : <EOL> enqueueTime += datetime . utcnow ( ) - pendingEnqueueUser [ "<STR_LIT>" ] <EOL> enqueueTime /= len ( enqueueHead ) <EOL> rmq_user_queue_stats = requests . get ( RABBITMQ_USER_QUEUE_STATS_URL ) . json ( ) <EOL> rmq_user_queue_length = rmq_user_queue_stats [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> rmq_user_queue_rate = rmq_user_queue_stats [ "<STR_LIT>" ] [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> rmq_user_queue_wait_time = rmq_user_queue_length / rmq_user_queue_rate <EOL> db . sync_worker_stats . remove ( { "<STR_LIT>" : { "<STR_LIT>" : datetime . utcnow ( ) - timedelta ( hours = <NUM_LIT:1> ) } } ) <EOL> timeUsedAgg = list ( db . sync_worker_stats . aggregate ( [ { "<STR_LIT>" : { "<STR_LIT>" : None , "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } } ] ) ) <EOL> totalSyncOps = db . sync_worker_stats . count ( ) <EOL> if timeUsedAgg : <EOL> timeUsed = timeUsedAgg [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> avgSyncTime = timeUsed / totalSyncOps <EOL> else : <EOL> timeUsed = <NUM_LIT:0> <EOL> avgSyncTime = <NUM_LIT:0> <EOL> lockedSyncRecords = list ( db . users . aggregate ( [ <EOL> { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : None } } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : None , "<STR_LIT:count>" : { "<STR_LIT>" : <NUM_LIT:1> } } } <EOL> ] ) ) <EOL> if len ( lockedSyncRecords ) > <NUM_LIT:0> : <EOL> lockedSyncRecords = lockedSyncRecords [ <NUM_LIT:0> ] [ "<STR_LIT:count>" ] <EOL> else : <EOL> lockedSyncRecords = <NUM_LIT:0> <EOL> pendingSynchronizations = list ( db . users . aggregate ( [ <EOL> { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : datetime . utcnow ( ) } } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : None , "<STR_LIT:count>" : { "<STR_LIT>" : <NUM_LIT:1> } } } <EOL> ] ) ) <EOL> if len ( pendingSynchronizations ) > <NUM_LIT:0> : <EOL> pendingSynchronizations = pendingSynchronizations [ <NUM_LIT:0> ] [ "<STR_LIT:count>" ] <EOL> else : <EOL> pendingSynchronizations = <NUM_LIT:0> <EOL> usersWithErrors = list ( db . users . aggregate ( [ <EOL> { "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT:0> } } } , <EOL> { "<STR_LIT>" : { "<STR_LIT>" : None , "<STR_LIT:count>" : { "<STR_LIT>" : <NUM_LIT:1> } } } <EOL> ] ) ) <EOL> if len ( usersWithErrors ) > <NUM_LIT:0> : <EOL> usersWithErrors = usersWithErrors [ <NUM_LIT:0> ] [ "<STR_LIT:count>" ] <EOL> else : <EOL> usersWithErrors = <NUM_LIT:0> <EOL> totalErrors = list ( db . users . aggregate ( [ <EOL> { "<STR_LIT>" : { "<STR_LIT>" : None , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } } <EOL> ] ) ) <EOL> if len ( totalErrors ) > <NUM_LIT:0> : <EOL> totalErrors = totalErrors [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> else : <EOL> totalErrors = <NUM_LIT:0> <EOL> db . sync_status_stats . insert ( { <EOL> "<STR_LIT>" : datetime . utcnow ( ) , <EOL> "<STR_LIT>" : lockedSyncRecords , <EOL> "<STR_LIT>" : pendingSynchronizations , <EOL> "<STR_LIT>" : usersWithErrors , <EOL> "<STR_LIT>" : totalErrors , <EOL> "<STR_LIT>" : timeUsed , <EOL> "<STR_LIT>" : enqueueTime . total_seconds ( ) , <EOL> "<STR_LIT>" : rmq_user_queue_wait_time <EOL> } ) <EOL> db . stats . update ( { } , { "<STR_LIT>" : { <EOL> "<STR_LIT>" : distanceSynced , <EOL> "<STR_LIT>" : lastDayDistanceSynced , <EOL> "<STR_LIT>" : lastHourDistanceSynced , <EOL> "<STR_LIT>" : timeUsed , <EOL> "<STR_LIT>" : avgSyncTime , <EOL> "<STR_LIT>" : totalSyncOps , <EOL> "<STR_LIT>" : enqueueTime . total_seconds ( ) , <EOL> "<STR_LIT>" : rmq_user_queue_wait_time , <EOL> "<STR_LIT>" : datetime . utcnow ( ) } } , upsert = True ) <EOL> def aggregateCommonErrors ( ) : <EOL> from bson . code import Code <EOL> map_operation = Code ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:}>" <EOL> ) <EOL> reduce_operation = Code ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:}>" ) <EOL> finalize_operation = Code ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:}>" <EOL> ) <EOL> db . connections . map_reduce ( map_operation , reduce_operation , "<STR_LIT>" , finalize = finalize_operation ) <EOL> aggregateCommonErrors ( ) <EOL> close_connections ( ) </s>
