hectorzhao/lisglosips
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: initialize upgraded analytics collector cursor

Browse Source
This commit is contained in:
hectorzhao
2026-09-01 10:11:27 +08:00
parent 7f8825cf3a
commit 29dbeb6981
1 changed files with 1 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/worker-cdr/src/analytics-main.ts
+1 -1
View File
@@ -35,5 +35,5 @@ async function collect(){
async function consumeEntries(items:Array<[string,string[]]>){for(const[id,fields]of items){try{const i=fields.findIndex(v=>v==='event');if(i<0||!fields[i+1])throw new Error('Missing event field');const event=validateAnalyticsEvent(JSON.parse(fields[i+1]));await store.consume(event);ingestLagMs=Math.max(0,(event.receivedAt??Date.now())-event.at);if(!dataThrough||event.at>Date.parse(dataThrough))dataThrough=new Date(event.at).toISOString();}catch(error){const message=error instanceof Error?error.message.slice(0,500):'Invalid queued event';const deadId=analyticsHash([ANALYTICS_STREAM,id]);await db.$executeRaw`INSERT IGNORE INTO caller_analysis_dead_letters(id,stream_id,payload,error,status,created_at,updated_at) VALUES (${deadId},${id},${JSON.stringify(fields)},${message},'OPEN',NOW(3),NOW(3))`;gap=true;}await redis.xack(ANALYTICS_STREAM,ANALYTICS_GROUP,id);}}
async function producerCount():Promise<number>{const response=await fetch(process.env.ANALYTICS_MI_URL??'http://127.0.0.1:8888/mi',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({jsonrpc:'2.0',id:1,method:'get_statistics',params:{statistics:['cra_event_total']}}),signal:AbortSignal.timeout(2000)});const body=await response.json()as{result?:Record<string,number>};const value=body.result?.['script:cra_event_total'];if(!response.ok||typeof value!=='number'||!Number.isSafeInteger(value))throw new Error('Producer counter unavailable');return value;}
async function observeActive(){const response=await fetch(process.env.ANALYTICS_MI_URL??'http://127.0.0.1:8888/mi',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({jsonrpc:'2.0',id:1,method:'dlg_list',params:[]}),signal:AbortSignal.timeout(2000)});const body=await response.json()as{result?:{Dialogs?:Array<{callid:string;state:number}>}};if(!response.ok||!body.result?.Dialogs)throw new Error('Dialog observation unavailable');const live=new Map(body.result.Dialogs.map(d=>[d.callid,Number(d.state)]));let lastId='';for(;;){const calls=await db.$queryRaw<Array<{id:string;payload:AnalyticsCall}>>`SELECT c.id,c.payload FROM caller_analysis_calls c WHERE c.id>${lastId} AND EXISTS (SELECT 1 FROM caller_analysis_states s WHERE s.call_key=c.id AND s.view='original' AND CAST(JSON_UNQUOTE(JSON_EXTRACT(s.counts,'$.activeCalls')) AS SIGNED)>0) ORDER BY c.id LIMIT 500`;if(!calls.length)break;const now=Date.now();for(const{payload}of calls)for(const leg of Object.values(payload.legs)){if(leg.endedAt!==null||leg.talkStartedAt==null||leg.talkMs>0)continue;if((live.get(leg.event.callId)??0)===4&&now>leg.talkStartedAt)await store.consume({...leg.event,kind:'DURATION',at:now,talkMs:now-leg.talkStartedAt,source:'dialog-acked-observation',eventId:analyticsHash([leg.event.callUid,leg.event.attemptId,'positive-duration'])});}lastId=calls[calls.length-1].id;}}
async function main(){try{bootId=(await readFile('/proc/sys/kernel/random/boot_id','utf8')).trim();}catch{bootId=`${hostname()}-unknown`;}try{const saved=JSON.parse(await readFile(checkpoint,'utf8'))as CollectorCursor;cursor=saved;gap=saved.gap===true;}catch{/* first start */}await redis.connect();await db.$connect();try{await redis.xgroup('CREATE',ANALYTICS_STREAM,ANALYTICS_GROUP,'0','MKSTREAM');}catch(e){if(!(e instanceof Error)||!e.message.includes('BUSYGROUP'))throw e;}let pendingCursor='0-0';while(!stopping){let degraded=gap;let reason='';try{await collect();const produced=await producerCount();if(cursor.lastProducerCount!==undefined&&produced<cursor.lastProducerCount){gap=degraded=true;reason='PRODUCER_RESTART';await recordGap(reason,{previous:cursor.lastProducerCount,current:produced});cursor.eventsSinceProducerStart=0;}cursor.lastProducerCount=produced;await persistCursor();const claimed=await redis.xautoclaim(ANALYTICS_STREAM,ANALYTICS_GROUP,consumer,5000,pendingCursor,'COUNT',100)as[string,Array<[string,string[]]>];pendingCursor=claimed[0];await consumeEntries(claimed[1]);const fresh=await redis.xreadgroup('GROUP',ANALYTICS_GROUP,consumer,'COUNT',200,'BLOCK',100,'STREAMS',ANALYTICS_STREAM,'>')as Array<[string,Array<[string,string[]]>]>|null;if(fresh)for(const[,batch]of fresh)await consumeEntries(batch);await observeActive();const groups=await redis.xinfo('GROUPS',ANALYTICS_STREAM)as Array<Array<string|number>>;const group=groups.map(g=>Object.fromEntries(Array.from({length:g.length/2},(_,i)=>[g[i*2],g[i*2+1]]))).find(g=>g.name===ANALYTICS_GROUP);if(Number(group?.pending??0)||Number(group?.lag??0)){degraded=true;reason='QUEUE_BACKLOG';}if(produced!==(cursor.eventsSinceProducerStart??0)){degraded=true;reason='PRODUCER_COLLECTOR_COUNT_MISMATCH';}if(!Number(group?.pending??0)&&Date.now()-lastTrim>60000){await redis.xtrim(ANALYTICS_STREAM,'MAXLEN','~',Number(process.env.ANALYTICS_STREAM_MAXLEN??1000000));lastTrim=Date.now();}if(Date.now()-lastAlerts>15000&&!degraded){await evaluateCallerAlerts(db);lastAlerts=Date.now();}}catch(e){degraded=true;reason=e instanceof Error?e.message.slice(0,200):'Worker error';logger.error({reason},'Analytics cycle failed');}if(degraded&&Date.now()-lastAlertSuspend>15000){try{await suspendCallerAlerts(db,reason||'ANALYTICS_DEGRADED');lastAlertSuspend=Date.now();}catch{logger.error('Analytics alert suspension failed');}}try{const spoolSize=await stat(spool).then(s=>s.size).catch(()=>null);const coverageStatus=!degraded&&!gap&&cursor.lastProducerCount===cursor.eventsSinceProducerStart?'CONTINUOUS':'DEGRADED';const payload=JSON.stringify({degraded,reason,dataThrough,lagMs:ingestLagMs,coverageStatus,captureSince:process.env.ANALYTICS_CAPTURE_SINCE??null,spoolOffset:cursor.offset,spoolUnreadBytes:spoolSize===null?null:Math.max(0,spoolSize-cursor.offset),gap,producerEventCount:cursor.lastProducerCount??null,collectedEventCount:cursor.eventsSinceProducerStart??null,bootId,updatedAt:new Date().toISOString()});await db.$executeRaw(Prisma.sql`INSERT INTO caller_analysis_health(id,payload,updated_at) VALUES ('worker',${payload},NOW(3)) ON DUPLICATE KEY UPDATE payload=VALUES(payload),updated_at=VALUES(updated_at)`);}catch{logger.error('Analytics heartbeat write failed');}await new Promise(resolve=>setTimeout(resolve,700));}redis.disconnect();await db.$disconnect();}
async function main(){try{bootId=(await readFile('/proc/sys/kernel/random/boot_id','utf8')).trim();}catch{bootId=`${hostname()}-unknown`;}try{const saved=JSON.parse(await readFile(checkpoint,'utf8'))as CollectorCursor;cursor={...cursor,...saved,eventsSinceProducerStart:saved.eventsSinceProducerStart??0};gap=saved.gap===true;}catch{/* first start */}await redis.connect();await db.$connect();try{await redis.xgroup('CREATE',ANALYTICS_STREAM,ANALYTICS_GROUP,'0','MKSTREAM');}catch(e){if(!(e instanceof Error)||!e.message.includes('BUSYGROUP'))throw e;}let pendingCursor='0-0';while(!stopping){let degraded=gap;let reason='';try{await collect();const produced=await producerCount();if(cursor.lastProducerCount!==undefined&&produced<cursor.lastProducerCount){gap=degraded=true;reason='PRODUCER_RESTART';await recordGap(reason,{previous:cursor.lastProducerCount,current:produced});cursor.eventsSinceProducerStart=0;}cursor.lastProducerCount=produced;await persistCursor();const claimed=await redis.xautoclaim(ANALYTICS_STREAM,ANALYTICS_GROUP,consumer,5000,pendingCursor,'COUNT',100)as[string,Array<[string,string[]]>];pendingCursor=claimed[0];await consumeEntries(claimed[1]);const fresh=await redis.xreadgroup('GROUP',ANALYTICS_GROUP,consumer,'COUNT',200,'BLOCK',100,'STREAMS',ANALYTICS_STREAM,'>')as Array<[string,Array<[string,string[]]>]>|null;if(fresh)for(const[,batch]of fresh)await consumeEntries(batch);await observeActive();const groups=await redis.xinfo('GROUPS',ANALYTICS_STREAM)as Array<Array<string|number>>;const group=groups.map(g=>Object.fromEntries(Array.from({length:g.length/2},(_,i)=>[g[i*2],g[i*2+1]]))).find(g=>g.name===ANALYTICS_GROUP);if(Number(group?.pending??0)||Number(group?.lag??0)){degraded=true;reason='QUEUE_BACKLOG';}if(produced!==(cursor.eventsSinceProducerStart??0)){degraded=true;reason='PRODUCER_COLLECTOR_COUNT_MISMATCH';}if(!Number(group?.pending??0)&&Date.now()-lastTrim>60000){await redis.xtrim(ANALYTICS_STREAM,'MAXLEN','~',Number(process.env.ANALYTICS_STREAM_MAXLEN??1000000));lastTrim=Date.now();}if(Date.now()-lastAlerts>15000&&!degraded){await evaluateCallerAlerts(db);lastAlerts=Date.now();}}catch(e){degraded=true;reason=e instanceof Error?e.message.slice(0,200):'Worker error';logger.error({reason},'Analytics cycle failed');}if(degraded&&Date.now()-lastAlertSuspend>15000){try{await suspendCallerAlerts(db,reason||'ANALYTICS_DEGRADED');lastAlertSuspend=Date.now();}catch{logger.error('Analytics alert suspension failed');}}try{const spoolSize=await stat(spool).then(s=>s.size).catch(()=>null);const coverageStatus=!degraded&&!gap&&cursor.lastProducerCount===cursor.eventsSinceProducerStart?'CONTINUOUS':'DEGRADED';const payload=JSON.stringify({degraded,reason,dataThrough,lagMs:ingestLagMs,coverageStatus,captureSince:process.env.ANALYTICS_CAPTURE_SINCE??null,spoolOffset:cursor.offset,spoolUnreadBytes:spoolSize===null?null:Math.max(0,spoolSize-cursor.offset),gap,producerEventCount:cursor.lastProducerCount??null,collectedEventCount:cursor.eventsSinceProducerStart??null,bootId,updatedAt:new Date().toISOString()});await db.$executeRaw(Prisma.sql`INSERT INTO caller_analysis_health(id,payload,updated_at) VALUES ('worker',${payload},NOW(3)) ON DUPLICATE KEY UPDATE payload=VALUES(payload),updated_at=VALUES(updated_at)`);}catch{logger.error('Analytics heartbeat write failed');}await new Promise(resolve=>setTimeout(resolve,700));}redis.disconnect();await db.$disconnect();}
void main().catch(e=>{logger.error({message:e instanceof Error?e.message:'Startup failed'},'Analytics failed');process.exitCode=1;redis.disconnect();void db.$disconnect();});