FlexRIC xApp에서 KPM Monitoring과 RAN Control을 동시에 구현하기

현재 OAI gNB에 Free5GC Core Network를 연결하고, gNB 위에 FlexRIC을 올려서 RAN의 자원을 할당하는 방식.

이후, RIC 위에 AI Agent Control Plane을 올려서 AI Agent 기반 5G Slice 자원 제어 시스템을 구현할 것.

 

이때 FlexRIC에서 다음 2가지 요구 사항을 반영해야 한다.

• KPM 모니터링: 100ms 주기로 슬라이스별 PRB 사용량, 처리량, 지연을 SQLite에 지속 저장
• RC 제어: Python AI Agent가 5분마다 새로운 PRB 할당 정책을 xApp에 전달

이 두 가지를 어떻게 하나의 xApp 프로세스 안에서 안정적으로 구현할지가 중요하다.

 

 

 

xApp을 2개 띄우는 방안

---

가장 직관적인 접근은 KPM 전용 xApp과 RC 제어 전용 xApp을 각각 실행하는 것이다.

터미널 1: xapp_kpm_moni  (KPM 구독, 메트릭 수집)
터미널 2: xapp_kpm_rc_prb (RC 제어)

FlexRIC nearRT-RIC는 여러 xApp의 동시 연결을 지원하고, 실제로 xApp ID = 7처럼 각 xApp에 고유 ID를 부여한다.

그런데 새로운 prb 정책이 내려올 때마다 xapp_kpm_rc_prb xApp을 새로 띄우게 된다면?

 

현재 AI Agent가 호출하는 policy_execution_tool이 5분마다 xApp 프로세스스를 새로 실행하는 방식이다.

Control Interval 1 (t=0):
  Python → subprocess: xapp_kpm_rc_prb (XAPP_DURATION=10, SLICE1_MAX_PRB=40)
           → E2 연결 → KPM 구독 → 백그라운드에서 계속 실행 → 구독 해제 → 종료

Control Interval 2 (t=300):
  Python → subprocess: xapp_kpm_rc_prb (XAPP_DURATION=10, SLICE1_MAX_PRB=35)
           → E2 연결 → KPM 구독 → RC 제어 1회 → 구독 해제 → 종료

 

이 방식에는 두 가지 문제가 있다.

 

1. E2 연결/해제 오버헤드: 매번 SCTP 연결을 새로 맺는다. E2 SETUP-REQUEST → SETUP-RESPONSE 왕복이 발생하고, 이 과정에서 추가적인 지연이 생긴다.
2. gNB E2 에이전트 부하: 잦은 KPM 구독/해제가 gNB의 E2 에이전트에 부담을 준다.
   실제로 OAI gNB 로그에서 RIC_SUBSCRIPTION_REQUEST / RIC_SUBSCRIPTION_DELETE_REQUEST가 5분마다 반복되는 것을 확인할 수 있다.

 

 

해결책: Named Pipe 기반 단일 xApp 아키텍처

---

하나의 xApp 프로세스가 KPM 모니터링과 RC 제어를 모두 담당하되, 
Python이 named pipe를 통해 제어 명령을 전달하는 방식이다.

xApp (XAPP_DURATION=86400, 24시간 상시 실행)
├── KPM 콜백 스레드 (100ms마다)
│   └── 메트릭 수집 → SQLite 저장  ← 끊김 없이 지속
│
├── Pipe 감시 스레드
│   └── /tmp/xapp_ctrl.pipe 대기
│       → JSON 수신 시 RC 제어 실행
│       → /tmp/xapp_ctrl_result.pipe에 결과 반환
│
└── 메인 스레드
    └── xapp_wait_end_api() 대기 (duration 만료 시 종료)

Python policy_execution_tool (5분마다 호출)
  → /tmp/xapp_ctrl.pipe에 JSON 쓰기
  → /tmp/xapp_ctrl_result.pipe에서 결과 읽기

이를 위해서는 xApp에 2개의 Thread를 추가해야 한다.

 

 

1. Pipe 감시 Thread : named pipe에서 JSON을 읽어 RC 제어를 실행한다.

---

static void* pipe_ctrl_thread(void* arg)
{
  mkfifo(CTRL_PIPE_PATH, 0666);
  mkfifo(RESULT_PIPE_PATH, 0666);

  while (g_ctrl_running) {
    // open()은 writer가 나타날 때까지 블로킹 — CPU 낭비 없음
    int fd = open(CTRL_PIPE_PATH, O_RDONLY);
    char buf[1024] = {0};
    read(fd, buf, sizeof(buf) - 1);
    close(fd);

    // JSON 파싱 후 메인 스레드에 신호
    ctrl_cmd_t cmd = parse_json_command(buf);
    pthread_mutex_lock(&g_ctrl_mutex);
    g_ctrl_cmd = cmd;
    pthread_cond_signal(&g_ctrl_cond);
    pthread_mutex_unlock(&g_ctrl_mutex);
  }
  return NULL;
}

 

 

2. 메인 Thread: xapp_wait_end_api()를 별도 스레드에서 실행하고, 메인은 pipe 명령을 처리한다

---

// xapp_wait_end_api()를 백그라운드 스레드에서 실행
pthread_create(&wait_tid, NULL, wait_thread_fn, NULL);

// 메인 루프: pipe 명령 처리
while (!g_wait_done) {
    pthread_mutex_lock(&g_ctrl_mutex);
    pthread_cond_timedwait(&g_ctrl_cond, &g_ctrl_mutex, &ts);  // 1초 대기

    if (g_ctrl_cmd.pending) {
        apply_rc_control(&g_ctrl_cmd);  // RC 제어 실행
        g_ctrl_cmd.pending = false;
    }
    pthread_mutex_unlock(&g_ctrl_mutex);
}

 

 

 

3. Python 구현 핵심

---

policy_execution_tool이 pipe 존재 여부를 확인해 자동으로 방식을 선택한다.

async def apply_ran_policy(allocations, ...):
    if os.path.exists("/tmp/xapp_ctrl.pipe"):
        # xApp이 실행 중 → pipe로 명령 전달 (빠름, 오버헤드 없음)
        return await _apply_ran_policy_via_pipe(allocations)
    else:
        # xApp이 없음 → subprocess 실행 (폴백)
        return await _apply_ran_policy_subprocess(allocations, ...)

 

pipe 방식의 통신 흐름:

# 1. 결과 pipe를 먼저 열어둠 (xApp이 결과를 쓸 수 있도록)
result_fd = os.open(RESULT_PIPE_PATH, os.O_RDONLY | os.O_NONBLOCK)

# 2. 제어 명령 전송
ctrl_fd = os.open(CTRL_PIPE_PATH, os.O_WRONLY)
os.write(ctrl_fd, json.dumps(cmd).encode())

# 3. 결과 수신 (최대 10초 대기)
result = poll_result(result_fd, timeout=10.0)
# → {"status": "ok", "timestamp": "2026-05-03T07:01:00"}

 

이를 통해 KPM 콜백은 100ms마다 계속 실행되면서 DB에 메트릭을 쌓고, RC 제어는 pipe 명령이 올 때만 실행된다.

두 동작이 mutex로 보호되어 충돌 없이 공존할 수 있게 된다.

 

 

 

동작 확인

---

1) Near-RT RIC 실행

---

먼저 RIC을 실행:

cd ~/openairinterface5g/openair2/E2AP
./flexric/build/examples/ric/nearRT-RIC

 

 

2) gNB / UE attach

---

다음으로 gNB와 UE를 실행한다.

cd ~/openairinterface5g/cmake_targets/ran_build/build
sudo ./nr-softmodem -O ../../../targets/PROJECTS/GENERIC-NR-5GC/CONF/gnb.sa.band78.fr1.106PRB.usrpb210.conf --gNBs.[0].min_rxtxtime 6 --rfsim

cd ~/openairinterface5g/tools/scripts
sudo ./multi-ue.sh -c1
sudo ./multi-ue.sh -o1

sudo ip netns exec ue1 /home/user/openairinterface5g/cmake_targets/ran_build/build/nr-uesoftmodem \
  -O /home/user/openairinterface5g/targets/PROJECTS/GENERIC-NR-5GC/CONF/ue1.conf \
  -r 106 \
  --numerology 1 \
  --band 78 \
  -C 3619200000 \
  --rfsim \
  --uicc0.imsi 208930000000001 \
  --rfsimulator.serveraddr 10.201.1.100 \
  --telnetsrv \
  --telnetsrv.listenport 9095

 

 

3) xApp 실행 및 초기화

---

이제 xApp을 동작시킨다.

cd ~/openairinterface5g/openair2/E2AP/flexric
XAPP_DURATION=60 ./build/examples/xApp/c/kpm_rc/xapp_kpm_rc_prb

 

이후 다른 터미널에서 제어 명령을 전송한다.

echo '{"slice1_min_prb":5,"slice1_max_prb":40,"slice1_dedicated_prb":35,
       "slice2_min_prb":3,"slice2_max_prb":30,"slice2_dedicated_prb":25,
       "slice3_min_prb":3,"slice3_max_prb":30,"slice3_dedicated_prb":25}' \
  > /tmp/xapp_ctrl.pipe

 

그럼 다음과 같은 로그가 생성된다.

 

 

<xApp 로그>

 

이 로그는 정상적인 1회 PIPE 기반 RC 제어 cycle이다.

흐름은 다음과 같다.

1. PIPE 명령 수신
2. slice별 PRB 값 파싱
3. SST1/slice-sst1에 대한 RC Control 메시지 생성
4. UE key=1 대상으로 CONTROL-REQUEST 송신
5. CONTROL ACK 수신
6. 적용 성공

 

가장 중요한 시작점은 아래 로그로 여기서 Python/MCP 쪽에서 pipe로 제어 명령이 들어온 것이다.

[PIPE-CTRL] Received command: {...}
[PIPE-CTRL] Applying RC control: slice1(min=5,max=40,ded=35) ...

 

그 다음 이 부분에서 지금 로그에 실제로 dump된 RC 메시지는 slice1, 즉 SST=1, SD=000001 대상임을 볼 수 있다.

target profile = slice-sst1
snssai=(1,000001)

 

제어 타입은 다음과 같다.

ric_style_type = 2
ctrl_act_id = 6

 

이것의 의미는 다음과 같다.

Style 2 = Radio Resource Allocation Control
Action 6 = Slice-level PRB quota

 

실제 들어간 PRB 값은 여기다.

ran_param_id = 11 → min PRB = 5
ran_param_id = 12 → max PRB = 40
ran_param_id = 13 → dedicated PRB = 35

 

대상 UE는 여기서 확인된다.

UE ID type = gNB, amf_ue_ngap_id = 4
ran_ue_id = 1
...
Sending RC control for UE key=1, profile=slice-sst1

 

즉 UE key 1번에 대해 slice-sst1 정책을 보낸 것이다.

 

마지막이 성공 여부다.

[xApp]: CONTROL-REQUEST tx
[xApp]: CONTROL ACK rx
[xApp]: Successfully received CONTROL-ACK
[PIPE-CTRL] RC control applied successfully

 

이건 RIC/gNB 쪽에서 CONTROL ACK까지 받은 상태라서, E2AP 절차상 제어 요청은 성공했음을 확인할 수 있다.

 

 

 

<RIC 로그>

 

첫째, 제어 시작점은 아래 로그로, 여기서부터 PIPE 명령이 xApp을 통해 RIC control로 변환된 것이다.

[RIC-CORE][CTRL-FWD] fwd_ric_control_request: start

 

 

둘째, 실제 E2 control 송신 로그 시점에 nearRT-RIC가 gNB 쪽으로 CONTROL REQUEST를 보냈다.

[RIC-CORE][CTRL-FWD] CONTROL SERVICE sent

 

 

셋째, 성공 응답. 여기까지 와야 1 cycle이 정상 완료다.

[NEAR-RIC]: CONTROL ACKNOWLEDGE rx
[iApp]: RIC_CONTROL_ACKNOWLEDGE tx

 

 

그리고 앞뒤로 계속 보이는 이 부분은 monitoring 쪽 KPM에서 indication이 계속 들어오는 구간을 의미한다.

sctp_msg_arrived_event: decoded msg.type = 6
e2ap_msg_handle_ric() produced ans.type = 43

 

 

 

 

<DB에서 PRB 할당 변경 확인>

sqlite3 -header -column /tmp/slice_metrics.db \
"SELECT timestamp, throughput_bps, prb_usage_percent, prb_ratio_allocated, latency_ms \
 FROM slice_metrics \
 ORDER BY id;"

 

 

 

 

핵심 개념: xApp의 서비스화

---

이번 구조 변경의 본질은 “실행하고 종료되는 xApp”을 “상시 대기하며 명령을 처리하는 제어 서비스”로 전환으로 볼 수 있다.

Before: Python → xApp 실행 → RC Control 1회 → 종료
After : Python → PIPE 명령 전달 → 실행 중인 xApp이 즉시 RC Control 수행

 

즉, xApp은 더 이상 일회성 실행 파일이 아니라, Near-RT RIC 내부에서 동작하는 control execution engine 역할을 한다.

 

이것의 목표는 Slice PRB quota를 실시간으로 반복 제어하기 위해서이다.

pipe 기반 구조에서는 xApp이 계속 살아 있으므로,
이미 수집한 UE 정보와 RIC 연결 상태를 유지한 채 제어 명령만 빠르게 주입할 수 있다.

 

이 구조는 역할을 명확히 분리한다.

Python = 정책 판단 계층 / Control Plane
xApp   = RIC 제어 실행 계층 / Execution Engine
PIPE   = 두 계층을 연결하는 lightweight control channel

 

이러한 Python 기반의 제어는 “어떤 slice에 얼마만큼 PRB를 줄 것인가”를 결정하는데,
이 과정에서 AI Plane에서 동작하는 AI Agent와 쉽게 통합될 수 있다.

 

또한 xApp은 그 결정을 E2SM-RC 메시지로 변환해 gNB에 실제 적용하는 역할을 하는데,

이는 AI Agent의 잘못된 정책을 Block하는 Guardrail로서의 기능을 수행한다.