Openstack_通用技術_RPC 遠程異步調用

目錄

• 目錄
• RPC
• 一個經過 HTTP Request 調用操做函數的 RPC 實現樣例
  • 環境
  • 接收 HTTP Request
  • RPC 調用具體的操做函數
• 測試

RPC

RPC: 同一個項目內的不一樣服務進程之間的交互方式。爲不一樣的進程服務提供了 call()(同步) 和 cast()(異步) 兩種調用方式。

問題 1: 在一個 Openstack 項目中擁有多個不一樣的進程服務，EG. API Service/Manage Service。 當咱們經過 Client 發送 API Request 去調用 Manage Service 執行一個操做任務時，咱們會但願這個調用的結果是可以快速響應到 Client 的(保證用戶體驗)。

問題 2: 並且進程服務之間的調用咱們還須要考慮如何有效的避免進程服務之間調用的阻塞問題。EG. API Service 調用 Manage Service 時，若是不能及時的將 API Service 釋放掉，那麼 API Request 就會由於被佔用，而沒法處理新的請求。

對於上面兩個問題，咱們能夠經過將具體的執行過程和響應過程分離來達到理想的效果。這也是 RPC 和 API 存在的緣由之一。

一個經過 HTTP Request 調用操做函數的 RPC 實現樣例

包含了下列兩個過程實現：

• 接收 HTTP Request
• RPC 調用操做函數

環境

Devstack 用的 L 版，樣例在自定義的 Openstack 項目中實現，詳見自動化生成 Openstack 新項目開發框架。

接收 HTTP Request

• 定義 HTTP Request URL 路由轉發的 Resource
  詳見Openstack Restful API 開發框架 Paste + PasteDeploy + Routes + WebOb 和
  Openstack Paste.ini 文件詳解

# project/api/v1/router.py

from project.api.v1 import new_resource

class APIRouter(octopunch.api.openstack.APIRouter):
    """Routes requests on the API to the appropriate controller and method."""
    ExtensionManager = extensions.ExtensionManager

    def _setup_routes(self, mapper, ext_mgr):
        self.resources['versions'] = versions.create_resource()
        mapper.connect("versions", "/",
                       controller=self.resources['versions'],
                       action='show')

        mapper.redirect("", "/")

        self.resources['new_resource'] = new_resource.create_resource(ext_mgr)
        mapper.resource('new_resource', 'new_resource',
                        controller=self.resources['new_resource'])

• 將 Resource 和 HTTP 方法綁定到 Controller 的 Action 函數中

# project/api/v1/new_resource.py

class NewResourceController(wsgi.Controller):

    def __init__(self, ext_mgr):
        self.ext_mgr = ext_mgr
        super(NewResourceController, self).__init__()

    # Create() 對應了 HTTP 的 POST 方法
    @wsgi.serializers()
    def create(self, req, body):
        """Create a NewResource."""
        context = req.environ['project.context']

        # Sync the project database.
        self.new_resource_api.db_sync(context)

def create_resource(ext_mgr):
    """project resource factory method."""
    return wsgi.Resource(NewResourceController(ext_mgr))

上述兩個文件實現了經過 HTTP Request 的 POST 方法來執行指定的 Create() 函數。

RPC 調用具體的操做函數

• 經過 API 調用 RPC-API

# project/new_resource/api.py

# import rpcapi module
from project.new_resource import rpcapi as new_resource_rpcapi

class API(base.Base):
    """API for interacting with the new_resource manager."""

    def __init__(self, db_driver=None, image_service=None):
        self.new_resource_rpcapi = new_resource_rpcapi.NewResourceAPI()
        super(API, self).__init__(db_driver)

    # 定義調用 rpcapi 的接口函數
    def db_sync(self, context):
        """Call rpc api to start db sync for new_resource database."""
        self.new_resource_rpcapi.db_sync(context)

API 的存在是爲了可以快速的響應請求，至於以後的執行過程交由 RPC-API 和 Manager 來處理

• rpc-api.py 調用 manager.py
  在 rpcapi.py 定義的 RPC 接口函數會自動的映射到 manager.py 中指定的處理函數。

# project/new_resource/rpcapi.py

class NewResourceAPI(object):

    def __init__(self):
        super(NewResourceAPI, self).__init__()
        target = messaging.Target(topic=CONF.manage_topic,
                                  version=self.RPC_API_VERSION)
        serializer = objects_base.ProjectObjectSerializer()
        self.client = rpc.get_client(target, version_cap='1.8',
                                     serializer=serializer)

    # 定義 rpcapi 函數，使用 cast 異步調用方式
    def db_sync(self, context):
        cctxt = self.client.prepare()
        # 指定 rpcapi 的調用方式，和指定映射到 manager.py 的處理函數
        cctxt.cast(context, 'db_sync')

RPC-API 的存在是爲了快速的響應進程服務之間的調用請求。

# project/new_resource/manager.py
class NewResourceManager(manager.Manager):

    RPC_API_VERSION = '1.8'

    target = messaging.Target(version=RPC_API_VERSION)

    def __init__(self, service_name=None, *args, **kwargs):
        super(NewResourceManager, self).__init__(*args, **kwargs)
        self._startup_delay = True

    def init_host_with_rpc(self):
        eventlet.sleep(CONF.periodic_interval)
        self._startup_delay = False

    def db_sync(self, context):
        print "這裏是具體的 RPC 操做函數"

小結：
Openstack 的 PRC 調用的過程爲： api.py ⇒ rpcapi.py ⇒ manager.py
詳見：Openstack Nova 源碼分析 — RPC 遠程調用過程

測試

• 啓動 API 服務

project-api --config-file /etc/project/proname.conf

• 啓動 Manager 服務

project-manager --config-file /etc/project/project.conf

• 發送 HTTP 請求

curl -i 'http://<Service_host_ip>:<service_port>/v1/<project_id>/<ResourceUrl>' -X POST -H "Content-Type: application/json" -H "X-Auth-Project-Id: admin" -H "X-Auth-Token: <token_id>" -d '<body_content_dict>'

注意： 樣例須要根據自身開發環境進行調整，