import { EventEmitter } from "events" ; import { delay } from "awaiting" ; import { isEqual } from "lodash" ; import { how_long_ago_m } from "@cocalc/util/misc" ; import { ALERT_HIGH_PCT , ALERT_DISK_FREE , RAISE_ALERT_AFTER_MIN , } from "./const" ; import { ProjectStatus , Alert , AlertType , ComponentName } from "./types" ; import { ProjectInfoServer , get_ProjectInfoServer } from "../project-info" ; import { ProjectInfo } from "../project-info/types" ; import { version } from "@cocalc/util/smc-version" ; import { cgroup_stats } from "./utils" ; import { getLogger } from "@cocalc/project/logger" ; const winston = getLogger ( "ProjectStatusServer" ) ; interface Elevated { cpu : number | null ; memory : number | null ; disk : number | null ; } export class ProjectStatusServer extends EventEmitter { private readonly dbg : Function ; private running = false ; private readonly testing : boolean ; private readonly project_info : ProjectInfoServer ; private info ? : ProjectInfo ; private status ? : ProjectStatus ; private last ? : ProjectStatus ; private elevated : Elevated = { cpu : null , disk : null , memory : null , } ; private elevated_cpu_procs : { [ pid ] : number ; } = { } ; private components : { [ name in ComponentName ] ? : number | null ; } = { } ; constructor ( testing = false ) { super ( ) ; this . testing = testing ; this . dbg = ( ... msg ) => winston . debug ( ... msg ) ; this . project_info = get_ProjectInfoServer ( ) ; } private async init ( ) { this . project_info . start ( ) ; this . project_info . on ( "info" , ( info ) => { this . info = info ; this . update ( ) ; } ) ; } public setComponentAlert ( name ) { if ( this . components [ name ] == null ) { this . components [ name ] = Date . now ( ) ; } } public clearComponentAlert ( name ) { delete this . components [ name ] ; } private update_alerts ( ) { if ( this . info == null ) return ; const du = this . info . disk_usage . project ; const ts = this . info . timestamp ; const do_alert = ( type , is_bad ) => { if ( is_bad ) { if ( this . elevated [ type ] == null ) { this . elevated [ type ] = ts ; } } else { this . elevated [ type ] = null ; } } ; do_alert ( "disk" , du . free < ALERT_DISK_FREE ) ; const cg = this . info . cgroup ; const du_tmp = this . info . disk_usage . tmp ; if ( cg != null ) { const { mem_pct , cpu_pct } = cgroup_stats ( cg , du_tmp ) ; do_alert ( "memory" , mem_pct > ALERT_HIGH_PCT ) ; do_alert ( "cpu-cgroup" , cpu_pct > ALERT_HIGH_PCT ) ; } } private alert_cpu_processes ( ) { const pids = [ ] ; if ( this . info == null ) return [ ] ; const ts = this . info . timestamp ; const ecp = this . elevated_cpu_procs ; const leftovers = new Set ( Object . keys ( ecp ) ) ; for ( const [ pid , proc ] of Object . entries ( this . info . processes ?? { } ) ) { leftovers . delete ( pid ) ; if ( proc . cpu . pct > ALERT_HIGH_PCT ) { if ( ecp [ pid ] == null ) { ecp [ pid ] = ts ; } } else { delete ecp [ pid ] ; } } for ( const pid of leftovers ) { delete ecp [ pid ] ; } for ( const [ pid , ts ] of Object . entries ( ecp ) ) { if ( ts != null && how_long_ago_m ( ts ) > RAISE_ALERT_AFTER_MIN ) { pids . push ( pid ) ; } } pids . sort ( ) ; return pids ; } private alerts ( ) { this . update_alerts ( ) ; const alerts = [ ] ; const alert_keys = [ "cpu-cgroup" , "disk" , "memory" ] ; for ( const k of alert_keys ) { const ts = this . elevated [ k ] ; if ( ts != null && how_long_ago_m ( ts ) > RAISE_ALERT_AFTER_MIN ) { alerts . push ( { type : k } as Alert ) ; } } const pids = this . alert_cpu_processes ( ) ; if ( pids . length > 0 ) alerts . push ( { type : "cpu-process" , pids } ) ; const componentNames = [ ] ; for ( const [ k , ts ] of Object . entries ( this . components ) ) { if ( ts == null ) continue ; componentNames . push ( k as ComponentName ) ; } if ( componentNames . length > 0 ) { alerts . push ( { type : "component" , names : componentNames } ) ; } return alerts ; } private update ( ) { this . status = { alerts : this . alerts ( ) , version : version , } ; if ( ! isEqual ( this . status , this . last ) ) { this . emit ( "status" , this . status ) ; } this . last = this . status ; } private async get_status ( ) { this . update ( ) ; return this . status ; } public stop ( ) { this . running = false ; } public async start ( ) { if ( this . running ) { this . dbg ( "project-status/server: already running, cannot be started twice" ) ; } else { await this . _start ( ) ; } } private async _start ( ) { this . dbg ( "start" ) ; if ( this . running ) { throw Error ( "Cannot start ProjectStatusServer twice" ) ; } this . running = true ; await this . init ( ) ; const status = await this . get_status ( ) ; this . emit ( "status" , status ) ; while ( this . testing ) { await delay ( 5000 ) ; const status = await this . get_status ( ) ; this . emit ( "status" , status ) ; } } } let _status = undefined ; export function get_ProjectStatusServer ( ) { if ( _status != null ) return _status ; _status = new ProjectStatusServer ( ) ; return _status ; } if ( require . main === module ) { const pss = new ProjectStatusServer ( true ) ; pss . start ( ) ; let cnt = 0 ; pss . on ( "status" , ( status ) => { console . log ( JSON . stringify ( status , null , 2 ) ) ; cnt += 1 ; if ( cnt >= 2 ) process . exit ( ) ; } ) ; }