Python netlink 소켓으로 커널 이벤트 실시간 수신하기

인터페이스가 언제 올라오고 IP가 언제 바뀌는지 감시해야 할 때, 흔히 ip monitor를 subprocess로 띄우고 출력을 파싱한다. 텍스트 포맷이 배포판마다 조금씩 달라 깨지기 쉽고, 1초마다 ip addr를 폴링하는 방식은 그 사이에 일어난 변화를 놓친다. 커널은 이런 이벤트를 netlink 소켓으로 이미 푸시하고 있고, Python 표준 라이브러리의 socket과 struct만으로 직접 받을 수 있다. 이 글에서는 rtnetlink로 인터페이스 목록을 조회하고 링크·주소 변화를 실시간으로 받는 코드, uevent 수신, 그리고 실제로 이벤트를 유실시키는 ENOBUFS 상황까지 다룬다.

Contents:

netlink 메시지 구조

netlink는 AF_NETLINK 주소 패밀리를 쓰는 소켓이다. 프로토콜 번호로 어떤 커널 서브시스템과 대화할지 정하고, 모든 메시지는 nlmsghdr 헤더로 시작한다. 헤더 뒤에는 서브시스템별 고정 구조체가 오고, 그 뒤로 가변 길이 속성(rtattr)이 TLV로 붙는다.

구조체	struct 포맷	필드
`nlmsghdr`	`=IHHII`	len, type, flags, seq, pid
`ifinfomsg` (링크)	`=BBHiII`	family, pad, type, index, flags, change
`ifaddrmsg` (주소)	`=BBBBi`	family, prefixlen, flags, scope, index
`rtattr` (속성)	`=HH`	rta_len, rta_type + payload

모든 필드가 호스트 바이트 오더이므로 struct 포맷은 =(네이티브 오더, 패딩 없음)로 고정한다. 각 메시지와 각 속성은 4바이트 경계로 정렬되어 있어, 다음 항목으로 넘어갈 때 길이를 올림해야 한다.

def align4(n):
    return (n + 3) & ~3

RTM_GETLINK로 인터페이스 목록 조회하기

요청 메시지에 NLM_F_DUMP를 붙이면 커널이 전체 목록을 여러 메시지로 나눠 보내고 마지막에 NLMSG_DONE으로 끝을 알린다.

import os
import socket
import struct

NETLINK_ROUTE = 0
RTM_GETLINK, RTM_NEWLINK = 18, 16
NLM_F_REQUEST = 0x01
NLM_F_DUMP = 0x300          # NLM_F_ROOT | NLM_F_MATCH
NLMSG_ERROR, NLMSG_DONE = 2, 3
IFLA_ADDRESS, IFLA_IFNAME, IFLA_MTU = 1, 3, 4
IFF_UP, IFF_RUNNING = 0x1, 0x40

nlmsghdr = struct.Struct("=IHHII")      # len, type, flags, seq, pid
ifinfomsg = struct.Struct("=BBHiII")    # family, pad, type, index, flags, change
rtattr = struct.Struct("=HH")           # len, type


def align4(n):
    return (n + 3) & ~3


def parse_attrs(buf):
    """rtattr TLV 스트림을 {type: payload} 딕셔너리로."""
    attrs, off = {}, 0
    while off + rtattr.size <= len(buf):
        rta_len, rta_type = rtattr.unpack_from(buf, off)
        if rta_len < rtattr.size:
            break
        attrs[rta_type] = buf[off + rtattr.size:off + rta_len]
        off += align4(rta_len)
    return attrs


def dump_links():
    sock = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_ROUTE)
    sock.bind((0, 0))                    # pid=0 -> 커널이 알아서 할당

    req = nlmsghdr.pack(nlmsghdr.size + ifinfomsg.size, RTM_GETLINK,
                        NLM_F_REQUEST | NLM_F_DUMP, 1, 0)
    req += ifinfomsg.pack(socket.AF_UNSPEC, 0, 0, 0, 0, 0)
    sock.send(req)

    done = False
    while not done:
        data = sock.recv(65535)
        off = 0
        while off < len(data):
            msg_len, msg_type, flags, seq, pid = nlmsghdr.unpack_from(data, off)
            if msg_type == NLMSG_DONE:
                done = True
                break
            if msg_type == NLMSG_ERROR:
                errno = struct.unpack_from("=i", data, off + nlmsghdr.size)[0]
                raise OSError(-errno, os.strerror(-errno))
            if msg_type == RTM_NEWLINK:
                body = off + nlmsghdr.size
                _, _, _, index, if_flags, _ = ifinfomsg.unpack_from(data, body)
                attrs = parse_attrs(data[body + ifinfomsg.size:off + msg_len])
                name = attrs.get(IFLA_IFNAME, b"").rstrip(b"\x00").decode()
                mtu = struct.unpack("=I", attrs[IFLA_MTU])[0] if IFLA_MTU in attrs else 0
                mac = ":".join(f"{b:02x}" for b in attrs.get(IFLA_ADDRESS, b""))
                state = "UP" if if_flags & IFF_UP else "DOWN"
                running = "RUNNING" if if_flags & IFF_RUNNING else "-"
                print(f"{index:>3} {name:<12} {state:<5} {running:<8} mtu={mtu:<6} {mac}")
            off += align4(msg_len)
    sock.close()


if __name__ == "__main__":
    print(f"{'idx':>3} {'name':<12} {'state':<5} {'carrier':<8} {'mtu':<10} mac")
    dump_links()

$ python3 nl_dump_links.py
idx name         state carrier  mtu        mac
  1 lo           UP    RUNNING  mtu=65536  00:00:00:00:00:00
  2 enp0s3       UP    RUNNING  mtu=1500   08:00:27:f8:2b:2e
  3 docker0      UP    -        mtu=1500   a2:50:fd:29:f6:46

docker0의 carrier가 비어 있는 것에 주목한다. 관리 상태(IFF_UP)와 실제 링크 상태(IFF_RUNNING)는 별개라, 브리지에 연결된 컨테이너가 없으면 UP이면서 RUNNING이 아닌 상태가 된다.

멀티캐스트 그룹 구독으로 변화 감시하기

bind()의 두 번째 값에 그룹 비트마스크를 넣으면 요청 없이도 커널이 이벤트를 밀어 넣어준다. 링크와 IPv4/IPv6 주소 그룹을 한 소켓에서 함께 구독한다.

import socket
import struct
import sys

NETLINK_ROUTE = 0
RTMGRP_LINK = 0x1
RTMGRP_IPV4_IFADDR = 0x10
RTMGRP_IPV6_IFADDR = 0x100

RTM_NEWLINK, RTM_DELLINK = 16, 17
RTM_NEWADDR, RTM_DELADDR = 20, 21
NAMES = {RTM_NEWLINK: "NEWLINK", RTM_DELLINK: "DELLINK",
         RTM_NEWADDR: "NEWADDR", RTM_DELADDR: "DELADDR"}

IFLA_IFNAME = 3
IFA_ADDRESS, IFA_LABEL = 1, 3
IFF_UP, IFF_RUNNING = 0x1, 0x40

nlmsghdr = struct.Struct("=IHHII")
ifinfomsg = struct.Struct("=BBHiII")
ifaddrmsg = struct.Struct("=BBBBi")     # family, prefixlen, flags, scope, index
rtattr = struct.Struct("=HH")


def align4(n):
    return (n + 3) & ~3


def parse_attrs(buf):
    attrs, off = {}, 0
    while off + rtattr.size <= len(buf):
        rta_len, rta_type = rtattr.unpack_from(buf, off)
        if rta_len < rtattr.size:
            break
        attrs[rta_type] = buf[off + rtattr.size:off + rta_len]
        off += align4(rta_len)
    return attrs


def main():
    sock = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_ROUTE)
    # 수신 버퍼를 넉넉히 잡지 않으면 이벤트 폭주 시 ENOBUFS로 유실된다
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20)
    sock.bind((0, RTMGRP_LINK | RTMGRP_IPV4_IFADDR | RTMGRP_IPV6_IFADDR))
    print("listening... (Ctrl-C to stop)", flush=True)

    while True:
        data = sock.recv(65535)
        off = 0
        while off < len(data):
            msg_len, msg_type, *_ = nlmsghdr.unpack_from(data, off)
            body = off + nlmsghdr.size
            tag = NAMES.get(msg_type, str(msg_type))

            if msg_type in (RTM_NEWLINK, RTM_DELLINK):
                _, _, _, index, flags, _ = ifinfomsg.unpack_from(data, body)
                attrs = parse_attrs(data[body + ifinfomsg.size:off + msg_len])
                name = attrs.get(IFLA_IFNAME, b"").rstrip(b"\x00").decode()
                state = "UP" if flags & IFF_UP else "DOWN"
                carrier = "RUNNING" if flags & IFF_RUNNING else "NO-CARRIER"
                print(f"[{tag}] idx={index} {name} {state} {carrier}", flush=True)

            elif msg_type in (RTM_NEWADDR, RTM_DELADDR):
                family, prefixlen, _, _, index = ifaddrmsg.unpack_from(data, body)
                attrs = parse_attrs(data[body + ifaddrmsg.size:off + msg_len])
                raw = attrs.get(IFA_ADDRESS, b"")
                af = socket.AF_INET if family == socket.AF_INET else socket.AF_INET6
                addr = socket.inet_ntop(af, raw) if raw else "?"
                label = attrs.get(IFA_LABEL, b"").rstrip(b"\x00").decode()
                print(f"[{tag}] idx={index} {label or index} {addr}/{prefixlen}", flush=True)

            off += align4(msg_len)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(0)

모니터를 띄워둔 상태에서 dummy 인터페이스를 만들고 주소를 붙였다 떼어본다.

python3 nl_monitor.py &
sudo ip link add dummy0 type dummy
sudo ip link set dummy0 up
sudo ip addr add 10.99.0.1/24 dev dummy0
sudo ip addr del 10.99.0.1/24 dev dummy0
sudo ip link set dummy0 down
sudo ip link del dummy0

listening... (Ctrl-C to stop)
[NEWLINK] idx=4 dummy0 DOWN NO-CARRIER
[NEWLINK] idx=4 dummy0 DOWN NO-CARRIER
[NEWLINK] idx=4 dummy0 UP RUNNING
[NEWADDR] idx=4 4 fe80::f8ba:dff:fe7d:a435/64
[NEWADDR] idx=4 dummy0 10.99.0.1/24
[DELADDR] idx=4 dummy0 10.99.0.1/24
[NEWLINK] idx=4 dummy0 DOWN NO-CARRIER
[DELADDR] idx=4 4 fe80::f8ba:dff:fe7d:a435/64
[DELLINK] idx=4 dummy0 DOWN NO-CARRIER

인터페이스 생성 한 번에 NEWLINK가 두 번 오고(등록 직후와 속성 설정 후), 링크를 UP 시키면 요청하지 않은 IPv6 링크로컬 주소가 자동으로 붙는다. 이 NEWADDR에는 IFA_LABEL이 없어 이름 자리에 인덱스가 찍혔으므로, IPv6 주소 이벤트에서는 인덱스로 이름을 따로 조회해야 한다.

같은 메시지 타입이 반복해서 오므로, 상태 머신을 만들 때는 NEWLINK를 “새로 생겼다”가 아니라 “현재 상태는 이렇다”는 갱신 통지로 다뤄야 한다.

NETLINK_KOBJECT_UEVENT로 장치 이벤트 받기

장치 핫플러그는 rtnetlink가 아니라 uevent 소켓으로 온다. 메시지가 구조체가 아니라 NUL로 구분된 KEY=VALUE 문자열이라 파싱이 더 간단하다.

import socket

NETLINK_KOBJECT_UEVENT = 15

sock = socket.socket(socket.AF_NETLINK, socket.SOCK_DGRAM, NETLINK_KOBJECT_UEVENT)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20)
sock.bind((0, 1))                # group 1 = 커널이 직접 보내는 uevent

print("listening uevents...", flush=True)
while True:
    data = sock.recv(65535)
    if data.startswith(b"libudev\0"):
        continue                 # udev가 재전송하는 메시지는 건너뛴다
    fields = data.split(b"\x00")
    header = fields[0].decode(errors="replace")
    env = dict(f.decode().split("=", 1) for f in fields[1:] if b"=" in f)
    print(header, flush=True)
    for key in ("ACTION", "DEVPATH", "SUBSYSTEM", "INTERFACE", "SEQNUM"):
        if key in env:
            print(f"    {key}={env[key]}", flush=True)

$ id -u
1000
$ python3 nl_uevent.py
listening uevents...
add@/devices/virtual/net/dummy9
    ACTION=add
    DEVPATH=/devices/virtual/net/dummy9
    SUBSYSTEM=net
    INTERFACE=dummy9
    SEQNUM=3146
add@/devices/virtual/net/dummy9/queues/rx-0
    ACTION=add
    DEVPATH=/devices/virtual/net/dummy9/queues/rx-0
    SUBSYSTEM=queues
    SEQNUM=3147
remove@/devices/virtual/net/dummy9
    ACTION=remove
    DEVPATH=/devices/virtual/net/dummy9
    SUBSYSTEM=net
    INTERFACE=dummy9
    SEQNUM=3151

uid 1000으로 실행했는데도 커널 uevent가 그대로 들어온다 — 수신은 비특권 사용자에게도 허용되어 있다(NL_CFG_F_NONROOT_RECV). 장치 하나를 만들 때 하위 kobject(queues/rx-0 등)까지 각각 이벤트를 내므로, SUBSYSTEM으로 걸러내지 않으면 잡음이 많다.

이벤트 유실: ENOBUFS 재현하기

netlink 멀티캐스트는 TCP가 아니다. 수신 버퍼가 차면 커널은 메시지를 버리고 다음 recv()에 ENOBUFS를 올린다. 버퍼를 일부러 최소로 줄이고 이벤트를 쏟아부어 확인한다.

import socket
import time

sock = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, 0)   # NETLINK_ROUTE
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)    # 일부러 작게
sock.bind((0, 0x1 | 0x10))                                    # RTMGRP_LINK | RTMGRP_IPV4_IFADDR
print("buf =", sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF), flush=True)

time.sleep(6)          # 그동안 쌓이는 이벤트를 읽지 않는다

received = lost = 0
sock.settimeout(1.0)
while True:
    try:
        sock.recv(65535)
        received += 1
    except socket.timeout:
        break
    except OSError as e:
        lost += 1
        print(f"! ENOBUFS: {e}  (여기까지의 이벤트는 유실됨)", flush=True)
print(f"received={received} enobufs={lost}", flush=True)

python3 nl_enobufs.py &
sleep 2
for i in $(seq 1 120); do sudo ip link add nltest$i type dummy; done
for i in $(seq 1 120); do sudo ip link del nltest$i; done

buf = 2304
! ENOBUFS: [Errno 105] No buffer space available  (여기까지의 이벤트는 유실됨)
received=110 enobufs=1

240번의 링크 조작으로 그보다 많은 이벤트가 발생했는데 110개만 받았다. ENOBUFS가 한 번 뜨면 몇 개를 놓쳤는지 알 방법이 없으므로, 이 예외를 만나면 RTM_GETLINK 덤프로 전체 상태를 다시 읽어 동기화하는 것이 정석이다.

주의사항

bind()에 넘기는 pid는 프로세스 ID가 아니라 소켓 식별자다. 한 프로세스가 netlink 소켓을 여러 개 열면 값이 겹쳐 EADDRINUSE가 난다. 항상 0을 넘겨 커널이 할당하게 한다.
bind()의 그룹 마스크는 32비트라 그룹 번호 32 이상은 지정할 수 없다. 그 이상은 NETLINK_ADD_MEMBERSHIP 소켓 옵션을 써야 하고, generic netlink 계열은 대부분 여기 해당한다.
이벤트 수신은 비특권으로 되지만 송신은 다르다. uevent를 멀티캐스트 그룹으로 쏘거나 rtnetlink로 링크를 조작하려면 CAP_NET_ADMIN이 필요하다.
recv() 한 번에 메시지 여러 개가 붙어 올 수 있다. 반드시 nlmsg_len을 따라가며 루프를 돌아야 하고, 버퍼가 작으면 메시지가 잘리므로 MSG_TRUNC로 실제 크기를 확인한다.
여기서 쓴 상수(RTM_*, IFLA_*)는 Python 표준 라이브러리에 없어 직접 정의해야 한다. 값은 커널 헤더(/usr/include/linux/rtnetlink.h, if_link.h)가 정본이며, 커널 버전에 따라 속성이 추가될 수 있으니 모르는 rta_type은 무시하도록 짠다.
속성 파싱에서 4바이트 정렬 올림을 빼먹으면 다음 속성부터 값이 어긋나는데, 예외 없이 조용히 이상한 값이 나오므로 디버깅이 오래 걸린다.
이벤트 기반 감시라도 시작 시점의 상태는 알 수 없다. 구독을 먼저 걸고 그다음 전체 덤프를 읽어야 그 사이에 발생한 이벤트를 놓치지 않는다.
본격적인 라우팅 조작이나 generic netlink(nl80211, taskstats 등)까지 다뤄야 한다면 pyroute2가 훨씬 현실적이다. 직접 파싱은 의존성 없이 이벤트만 받으면 되는 에이전트에 어울린다.

마무리

표준 라이브러리만으로 커널 이벤트를 받는 코드가 100줄 남짓이다. ip monitor 출력 파싱과 비교하면 포맷이 바이너리로 고정되어 있어 배포판을 타지 않고, 폴링과 달리 변화를 놓치지 않는다. 대신 유실은 별개 문제라, ENOBUFS를 만났을 때 전체 덤프로 재동기화하는 경로를 반드시 함께 만들어야 실제로 믿고 쓸 수 있다.