세션 생성 과정 조사

[rjwharry]

Session creation Flow

1. Run backend.ai run ...

   • def run() (client/cli/run.py)
     • asyncio_run(_run_cases())
       • _run()
         • session.ComputeSession.get_or_create(…)

           • ComputeSession class는 client/func/session.py 에 정의되어 있으며, get_or_create 함수에서 Request 객체를 생성하여 manager에 세션 생성 요청을 보낸다.

           • Request URL : POST http://127.0.0.1:8091/session

             rqst = Request("POST", f"/{prefix}")
             ...
             async with rqst.fetch() as resp:
             	data = await resp.json()
               o = cls(name, owner_access_key)  # type: ignore
               if api_session.get().api_version[0] >= 5:
             	  o.id = UUID(data["sessionId"])
               o.created = data.get("created", True)  # True is for legacy
               o.status = data.get("status", "RUNNING")
               o.service_ports = data.get("servicePorts", [])
               o.domain = domain_name
               o.group = group_name
               return o
             

2. Manager

   • create_from_params 함수가 cli로부터 받은 요청을 처리 (manager/api/session.py)

     manager가 실행될 때, manager/server.py의 build_root_app()함수에서 각 서브 패키지 URL앞에 패키지 이름을 붙인다. 따라서 아래 코드에서 빈 URL이지만, 앞에 “session” prefix가 붙는다. 그래서 create_from_params가 http://127.0.0.1:8091/session 요청을 핸들링하게 된다.

     def create_app(
         default_cors_options: CORSOptions,
     ) -> Tuple[web.Application, Iterable[WebMiddleware]]:
         app = web.Application()
         app.on_startup.append(init)
         app.on_shutdown.append(shutdown)
         app["api_versions"] = (1, 2, 3, 4)
         app["session.context"] = PrivateContext()
         cors = aiohttp_cors.setup(app, defaults=default_cors_options)
         cors.add(app.router.add_route("POST", "", create_from_params))
     

     • 사용자 권한 체크

     • _create() (manager/api/session.py)

       • Check work directory and reserved name directory
       • Resolve the image reference
       • Check existing session instance
         • if exist, return existing session info
       • registry.enqueue_session() (manager/registry.py)
         • check_scaling_group
         • prepare_vfolder_mounts
         • prepare_dotfiles
         • kernel db에 새로 생성할 kernel 정보 추가
         • _enqueue() → 다른 디비(어떤 디비인진 모르겠음)에 kernel 데이터 추가
         • publish POST_ENQUEUE_SESSION Event to hook_plugin_ctx
         • publish SessionEnqueuedEvent

     • schedule() (manager/schedule/dispatcher.py)

       SchedulerDispatcher class 생성자에서 아래와 같이 schedule 함수를 SessionEnqueuedEvent 핸들러로 등록하고 있음. 정확한 연결점이 맞는진 잘 모르겠음

       evd.consume(
         SessionEnqueuedEvent, None, self.schedule, coalescing_opts, name="dispatcher.enq"
       )
       

       • scheduler**in_sgroup() (manager/schedule/dispatcher.py)
         • _load_scheduler() (manager/schedule/dispatcher.py)
           • load_scheduler() (manager/schedule/dispatcher.py)
             • return scheduler
         • pending session과 cancelled session 리스트 조회
         • cancelled session 먼저 취소 처리
           • kernel db에 취소 세션 정보 업데이트
           • publish SessionCancelledEvent
         • pending session 리스트가 비워질 때까지 반복문
           • pending session 하나 고른 후 리스트에서 삭제
           • predicate 검사
             • 하나라도 실패하면 kernel db에 “predicate-checks-failed” 상태 반영
         • single 혹은 multi node에 따라 스케쥴링 진행 (single 가정)
         • _schedule_single_node_session() (manager/schedule/dispatcher.py)
           • scheduler가 적합한 agent 선택
             • 만약 가용한 agent가 없으면 kernel db에 “no-available-instance” 상태 업데이트
             • 다른 에러가 발생하면 kernel db에 “scheduler-error”로 상태 업데이트
           • 문제가 없다면 선택된 agent 정보를 kernel db에 업데이트
           • publish SessionScheduledEvent
         • publish DoPrepareEvent

     • prepare() (manager/schedule/dispatcher.py)

       schedule 함수와 마찬가지로 SchedulerDispatcher class 생성자에서 아래와 같이 prepare 함수를 DoPrepareEvent 핸드러로 등록하고 있다.

       evd.consume(DoPrepareEvent, None, self.prepare)
       

       • “SCHEDULED” 상태인 세션들을 “PREPARING” 상태로 업데이트한 후 반복문을 돌며 세션 시작
       • start_session() (manager/schedule/dispatcher.py)

         • registry.start_session() (manager/registry.py)

           • kernelAgentBinding

           • get resource policy for the session

           • Aggreate image registry information

           • update single or multi node and create local network with RPCContext

           • make ClusterInfo

           • Update session’s environment variable

           • _create_kernels_in_one_agent() (manager/registry.py)

             • 이전에 미리 고른 agent 정보를 기반으로 rpc instance를 생성하고 rpc call을 통해 해당 agent의 create_kernel 함수를 실행시킨다

             async with RPCContext(
             	agent_alloc_ctx.agent_id,
               agent_alloc_ctx.agent_addr,
               invoke_timeout=None,
               order_key=str(scheduled_session.session_id),
               keepalive_timeout=self.rpc_keepalive_timeout,
             ) as rpc:
             	...
             	created_infos = await rpc.call.create_kernels(
               kernel_creation_id,
               str(scheduled_session.session_id),
               [str(binding.kernel.kernel_id) for binding in items],
             	...)
             

         • 예외가 발생하면 “failed-to-start” 상태로 kernel db 업데이트

3. Agent

   • create_kernels() (agent/server.py)
     • create_kernel() (agent/server.py)
       • init kernel context
       • kernel context로 compute session 생성 준비
         • image 준비
         • resource 할당
         • mount vfolder
         • apply network
         • preopen port (ssh, etc)
       • start_container() (agent/docker/agent.py or agent/kubernetes/agent.py)
         • async docker 라이브러리로 container 생성