GT PBX 개발기 5 — AMI 이벤트로 실시간 현황 만들기

GT PBX 개발기 시리즈 5/8편. 전체 목록은 gtpbx 태그에서 볼 수 있다.

목표: 1초 안에, 사후 집계와 같은 숫자로

M3 계약 초안은 10월 29일 11시 5분에 커밋했고 main 병합은 12시 33분이었다. 완료 기준은 계약 첫머리에 한 줄로 적었다. SIPp로 상담원 3명과 장비 10대를 돌린 뒤 월보드 숫자(대기, 상담원 상태, 당일 인입·응답·포기)가 queue_log 사후 집계와 같아야 하고, AMI 이벤트는 1초 안에 화면에 보여야 한다.

범위는 실시간 상태만으로 잡았다. 15분 집계와 통계 화면은 M4(6편)로, 알림 발송은 M5로 넘겼다. 경고는 색만 M3에서 칠한다.

AMI 연결을 하나 더 연다

M2의 웹 서버는 이미 AMI 연결 하나로 Action을 보내고 있었다. 이 연결은 Events: off다. 수집기는 같은 gtpbx-web 프로세스 안에 두되 Events: agent,system으로 연결을 하나 더 열었다. 같은 AMI 사용자를 쓰고, manager.conf의 read 권한에 이미 필요한 항목이 있어서 설정은 바꾸지 않았다.

연결을 둘로 나눈 영향은 OrbStack VM 통합 시험에서 따로 쟀다. AMI 연결 2개 상태에서 /api/health는 8번 모두 ami:true였다. 수집기 소켓만 끊으면 0.26초 뒤 ami:false가 되고 1.29초 뒤 다시 붙어 스냅샷이 queue show field와 같아졌다. 그동안 Action 연결로 하는 장비 생성과 PJSIPShowEndpoint는 정상이었다.

스냅샷 위에 이벤트를 쌓는다

연결할 때마다(재연결 포함) QueueStatus와 PJSIPShowContacts로 스냅샷을 만든다. 스냅샷을 받는 사이에 온 이벤트는 버퍼에 모았다가 스냅샷 뒤에 다시 적용한다. 상담원이 지금 상태에 들어간 시각(since)은 스냅샷 직후 queue_log에서 상담원별 마지막 관련 행을 읽어 채운다. 행이 없으면 스냅샷 시각을 쓰고 sinceExact=false로 표시한다.

수집기는 AMI 필드를 내부 이벤트로 바꾼 뒤 순수 함수 applyEvent(state, event)로 상태를 갱신한다. 단위 시험을 쉽게 하려는 목적과 함께, 나중에 사내 보조 PBX 어댑터가 같은 내부 이벤트를 쓰게 하려는 목적도 있었다. 상담원 상태는 아래 순서로 정한다.

필드 이름은 계약에 추정으로 적어 두고 구현 때 Asterisk 22.11 소스로 확인했다. 계약에는 등록 이벤트 값으로 Created를 적었는데, 22.11의 ContactStatus에서는 NonQualified로 나왔다. 그래서 NonQualified·Created·Updated·Reachable을 등록, Removed·Unreachable을 미등록으로 보고 Unknown은 무시한다. timestampevents = no라 이벤트 시각은 받은 시각을 쓴다.

상담원 기능 코드와 웹 버튼을 같은 모양으로

상담원은 전화기로 기능 코드를 눌러 상태를 바꾼다. 다이얼플랜의 기능 코드는 이렇다.

관리자와 슈퍼바이저는 월보드에서 같은 동작을 대신할 수 있게 했다. 이때 AMI QueueAdd·QueueRemove·QueuePause를 기능 코드와 같은 인자로 보낸다. 그래야 queue_log의 ADDMEMBER·PAUSE 행이 전화기에서 한 것과 같은 모양으로 남고, 6편의 통계가 둘을 구분하지 않아도 된다. 실제로 돌려 보니 callid 하나만 달랐다. 기능 코드의 로그인·로그아웃 행은 채널 uniqueid, 웹 대행은 MANAGER가 들어간다. 이건 Asterisk 동작이라 계약에 차이로 적었다.

SSE로 전체 스냅샷을 보낸다

브라우저에는 SSE(GET /api/realtime/stream)로 보낸다. 바뀔 때마다 차분이 아니라 전체 스냅샷을 보낸다. 상담원 수십 명, 대기 수십 건이면 수 KB라서 재접속과 누락 처리를 단순하게 두는 쪽을 골랐다. 같은 250ms 안의 변경은 한 번으로 합치고, 15초마다 하트비트 주석을 보낸다.

통합 시험(OrbStack)에서 queue_log·CEL 시각부터 SSE 스냅샷 at까지의 지연을 쟀다. 250ms 합치기가 포함된 값이다.

당일 지표(인입·응답·포기·응답률·서비스레벨)는 이벤트로 더하지 않는다. 0006 트리거가 queue_log로 만든 통화 요약 gtpbx_calls를 KST 오늘 0시부터 다시 센다. 통화가 끝나는 이벤트 뒤 1초 디바운스, 그리고 30초마다 다시 계산한다. 실시간 증분을 하면 사후 집계와 어긋날 여지가 생겨서 이렇게 했다.

화면은 /realtime 월보드, 메뉴 없이 큰 숫자만 보이는 TV 모드(/realtime?tv=1), 상담원용 /me 세 가지다. 조회자 계정의 첫 화면은 월보드로 바꿨다. /me는 보조 PBX 페이지 옆 360px 창에서 쓰도록 좁게 만들었다. 유지 시간은 클라이언트가 1초마다 now − since로 그린다.

등록 상태는 contact 단위로 모은다

장비·내선 목록에 "미등록 3시간 12분째 (마지막 등록 10-29 07:10)" 같은 문구를 보이려고 gtpbx_endpoint_presence 테이블에 마지막 등록 시각과 미등록 시작 시각을 남긴다. 처음에는 엔드포인트별로 마지막 ContactStatus 하나만 봤다.

독립 리뷰에서 이게 틀릴 수 있다는 지적이 나왔다. 전화기가 새 contact로 다시 등록한 뒤 옛 contact의 Removed가 늦게 오면 엔드포인트가 미등록으로 굳는다. 그래서 엔드포인트마다 contact URI 집합을 두고, 하나라도 등록이면 등록으로 본다. presence 테이블은 집계 결과가 바뀌거나 새 contact가 등록될 때만 쓴다.

새 장비 이름이 30초 비던 문제

통합 시험 ②에서 방금 만든 장비가 바로 전화를 걸면 월보드 대기 목록의 장비 이름이 비어 보였다. 수집기는 장비 이름·그룹 같은 메타를 30초마다 다시 읽는데, 그 사이에 만든 장비는 메타에 없었다. 최대 30초 동안 이름 없이 보이는 셈이다.

11시 58분 커밋(f254604)에서 대기 호출 메타를 갱신할 때 메타에 없는 장비가 있으면 장비 메타도 함께 읽도록 바꿨다. DB 통합 시험을 먼저 추가해 수정 전에 실패하는 것을 확인했다.

private async refreshCallMeta(): Promise<boolean> {
  const ids = [...this.state.callers.keys()];
  // 30초 갱신 사이에 만든 장비의 첫 호출이면 장비 메타(이름·그룹)도 다시 읽는다
  const unknownDevice = [...this.state.callers.values()].some((c) => c.endpoint !== null && !this.meta.devices.has(c.endpoint));
  const [calls, endpointMeta] = await Promise.all([loadCallMeta(this.db, ids), unknownDevice ? loadEndpointMeta(this.db) : null]);
  this.meta = { ...this.meta, ...endpointMeta, calls };
  return ids.every((id) => calls.has(id));
}

리뷰에서 고친 SSE 백프레셔

서버 코드만 따로 독립 리뷰를 맡겼고, 9건 중 8건을 12시 20분 커밋(6d46c69)에 반영했다. presence 집계 말고 SSE 쓰기도 지적됐다. 느린 클라이언트가 있으면 res.write()가 false를 돌려주는데, 처음 구현은 반환값을 보지 않고 계속 썼다. 그러면 보내지 못한 스냅샷이 서버 메모리에 쌓인다. 권한 검사가 세션을 읽는 사이에 이미 끊긴 요청을 스트림으로 등록해, 닫히지 않는 스트림이 남는 경로도 같이 지적됐다.

매번 전체 스냅샷을 보내니 중간 값은 버려도 된다. 밀리면 drain까지 기다렸다가 그동안 온 것 중 최신 값 하나만 보낸다. 그래도 버퍼가 1MB를 넘으면 스트림을 닫는다. TCP keepalive는 30초로 두었다.

export function sseSender(sink: SseSink, onOverflow: () => void, maxBufferedBytes = SSE_MAX_BUFFERED_BYTES): SseSender {
  let blocked = false;
  let pending: string | null = null;
  const write = (chunk: string): void => {
    if (sink.destroyed) return;
    if (sink.write(chunk)) return;
    if (sink.writableLength > maxBufferedBytes) {
      onOverflow();
      return;
    }
    blocked = true;
    sink.once("drain", () => {
      blocked = false;
      const next = pending;
      pending = null;
      if (next !== null) write(next);
    });
  };
  return {
    send(chunk) {
      if (blocked) pending = chunk;
      else write(chunk);
    },
    ping(chunk) {
      if (!blocked) write(chunk);
    },
  };
}

같은 커밋에서 당일 지표와 대기 호출 메타 조회에 요청 순번을 붙여 가장 최근 요청 결과만 반영하게 했다. 반영하지 않은 1건은 "탭 4개 이상이면 가장 오래된 스트림을 닫아 서로 밀어낸다"는 지적이다. 세션당 3개까지라는 계약대로 두고, 필요하면 event: evicted로 재접속을 막는 방법을 남은 일에 적었다.

AMI가 끊기면

수집기 연결이 끊기면 상태 모델에 ami: false를 두고 대기 목록과 상담원 상태를 비운다. 화면에는 "PBX 연결 끊김"이 뜬다. 처음 계약은 모든 값을 비우는 것이었는데, 12시 25분에 당일 지표는 계속 보이도록 정정했다(15a3549). 당일 지표는 AMI가 아니라 DB에서 세기 때문이다. 다시 붙으면 스냅샷부터 새로 만든다.

끊긴 동안 관리자가 상담원 상태를 대신 바꾸면, 수집기 반영을 2초 기다리지 않고 Action 연결의 QueueStatus로 바로 답한다. 이것도 리뷰 반영이다.

M3 통합 시험 tests/m3/run.sh --with-e2e는 OrbStack에서 119/119였다. 시험 뒤 Asterisk 로그는 ERROR 0, WARNING 0이었고, 웹 서버 경고는 AMI를 일부러 끊은 시나리오 ⑥의 ECONNRESET 1줄뿐이었다. EC2 스테이징 배포와 통합 회귀 기록은 12시 44분에 남겼다.

남은 것

• 실제 상담원 전화기와 현장 장비로 본 적이 없다. 모두 SIPp가 흉내 낸 전화기다.
• 장시간 SSE 연결과 메모리 사용량. 부하 시험 S5와 함께 보기로 했다.
• 조회자 세션이 8시간이라 TV 모드 화면이 하루에 몇 번 로그인 화면으로 떨어진다. 세션 길이는 아직 정하지 않았다.

다음 글

GT PBX 개발기 6 — 15분 사전 집계로 통계 만들기