• 利用 Function 函数计算服务进行消息处理
    • 操作流程
    • 消息处理测试
      • Baetyl 启动
      • MQTTBox 建立连接
      • 消息处理验证

    利用 Function 函数计算服务进行消息处理

    声明：

    • 本文测试所用设备系统为 Ubuntu 18.04
    • 本文测试使用的函数运行时是 Python3
    • 模拟 MQTT client 行为的客户端为 MQTTBox提示：Darwin 系统可以通过源码安装 Baetyl，可参考 源码安装 Baetyl。

    与基于 Hub 服务实现设备间消息转发不同的是，本文主要介绍利用本地函数计算服务进行消息处理。其中 Hub 服务用于建立 Baetyl 与 MQTT 客户端之间的连接，Python 运行时服务用于处理 MQTT 消息，而本地函数计算服务则通过 MQTT 消息上下文衔接本地 Hub 服务与 Python 运行时服务。

    本文将以 TCP 连接方式为例，展示本地函数计算服务的消息处理、计算功能。

    操作流程

    • 步骤一：安装 Baetyl，并导入示例配置包。参考 快速安装 Baetyl 进行操作；
    • 步骤二：依据测试需求修改导入的配置信息，执行 sudo systemctl start baetyl 以容器模式启动 Baetyl，然后执行 sudo systemctl status baetyl 来查看 Baetyl 是否正常运行。如果 Baetyl 已经启动，执行 sudo systemctl start baetyl 重启来加载新的配置。
    • 步骤三：通过 MQTTBox 以 TCP 方式与 Baetyl Hub 服务 建立连接；
      • 若成功与 Hub 服务建立连接，则依据配置的主题权限信息向有权限的主题发布消息，同时向拥有订阅权限的主题订阅消息，并观察 Baetyl 日志信息；
        • 若 Baetyl 日志显示已经启动 Python 运行时服务，则表明发布的消息受到了预期的函数处理；
        • 若 Baetyl 日志显示未成功启动 Python 运行时服务，则重复上述步骤，直至看到 Baetyl 主程序成功启动了 Python 运行时服务。
      • 若与 Baetyl Hub 建立连接失败，则重复步骤三操作，直至 MQTTBox 与 Baetyl Hub 服务成功建立连接为止。
    • 步骤四：通过 MQTTBox 查看对应主题消息的收发状态。../_images/python-flow.png基于本地函数计算服务实现设备消息处理流程

    消息处理测试

    依据 步骤一 导入示例配置包后，确认一下应用配置、 Hub 服务配置以及函数计算服务配置。

    将 Baetyl 应用配置改成如下配置：

    1. # /usr/local/var/db/baetyl/application.yml
    2. version: v0
    3. services:
    4.   - name: localhub
    5.     image: hub.baidubce.com/baetyl/baetyl-hub
    6.     replica: 1
    7.     ports:
    8.       - 1883:1883
    9.     mounts:
    10.       - name: localhub-conf
    11.         path: etc/baetyl
    12.         readonly: true
    13.       - name: localhub-data
    14.         path: var/db/baetyl/data
    15.       - name: localhub-log
    16.         path: var/log/baetyl
    17.   - name: function-manager
    18.     image: hub.baidubce.com/baetyl/baetyl-function-manager
    19.     replica: 1
    20.     mounts:
    21.       - name: function-manager-conf
    22.         path: etc/baetyl
    23.         readonly: true
    24.       - name: function-manager-log
    25.         path: var/log/baetyl
    26.   - name: function-python27-sayhi
    27.     image: hub.baidubce.com/baetyl/baetyl-function-python27
    28.     replica: 0
    29.     mounts:
    30.       - name: function-sayhi-conf
    31.         path: etc/baetyl
    32.         readonly: true
    33.       - name: function-sayhi-code
    34.         path: var/db/baetyl/function-sayhi
    35.         readonly: true
    36.   - name: function-python36-sayhi
    37.     image: hub.baidubce.com/baetyl/baetyl-function-python36
    38.     replica: 0
    39.     mounts:
    40.       - name: function-sayhi-conf
    41.         path: etc/baetyl
    42.         readonly: true
    43.       - name: function-sayhi-code
    44.         path: var/db/baetyl/function-sayhi
    45.         readonly: true
    46.   - name: function-node85-sayhi
    47.     image: hub.baidubce.com/baetyl/baetyl-function-node85
    48.     replica: 0
    49.     mounts:
    50.       - name: function-sayjs-conf
    51.         path: etc/baetyl
    52.         readonly: true
    53.       - name: function-sayjs-code
    54.         path: var/db/baetyl/function-sayhi
    55.         readonly: true
    56.   - name: function-sql-filter
    57.     image: hub.baidubce.com/baetyl/baetyl-function-sql
    58.     replica: 0
    59.     mounts:
    60.       - name: function-filter-conf
    61.         path: etc/baetyl
    62.         readonly: true
    63. volumes:
    64.   # hub
    65.   - name: localhub-conf
    66.     path: var/db/baetyl/localhub-conf
    67.   - name: localhub-data
    68.     path: var/db/baetyl/localhub-data
    69.   - name: localhub-cert
    70.     path: var/db/baetyl/localhub-cert-only-for-test
    71.   - name: localhub-log
    72.     path: var/db/baetyl/localhub-log
    73.   # function
    74.   - name: function-manager-conf
    75.     path: var/db/baetyl/function-manager-conf
    76.   - name: function-manager-log
    77.     path: var/db/baetyl/function-manager-log
    78.   - name: function-sayhi-conf
    79.     path: var/db/baetyl/function-sayhi-conf
    80.   - name: function-sayhi-code
    81.     path: var/db/baetyl/function-sayhi-code
    82.   - name: function-sayjs-conf
    83.     path: var/db/baetyl/function-sayjs-conf
    84.   - name: function-sayjs-code
    85.     path: var/db/baetyl/function-sayjs-code
    86.   - name: function-filter-conf
    87.     path: var/db/baetyl/function-filter-conf

    Baetyl Hub 服务配置改成如下配置：

    1. # /usr/local/var/db/baetyl/localhub-conf/service.yml
    2. listen:
    3.   - tcp://0.0.0.0:1883
    4. principals:
    5.   - username: test
    6.     password: hahaha
    7.     permissions:
    8.       - action: 'pub'
    9.         permit: ['#']
    10.       - action: 'sub'
    11.         permit: ['#']
    12. subscriptions:
    13.   - source:
    14.       topic: 't'
    15.     target:
    16.       topic: 't/topic'
    17. logger:
    18.   path: var/log/baetyl/service.log
    19.   level: "debug"

    Baetyl 本地函数计算服务相关配置无需修改，具体配置如下：

    1. # /usr/local/var/db/baetyl/function-manager-conf/service.yml
    2. hub:
    3.   address: tcp://localhub:1883
    4.   username: test
    5.   password: hahaha
    6. rules:
    7.   - clientid: func-python27-sayhi-1
    8.     subscribe:
    9.       topic: t
    10.     function:
    11.       name: python27-sayhi
    12.     publish:
    13.       topic: t/py2hi
    14.   - clientid: func-sql-filter-1
    15.     subscribe:
    16.       topic: t
    17.       qos: 1
    18.     function:
    19.       name: sql-filter
    20.     publish:
    21.       topic: t/sqlfilter
    22.       qos: 1
    23.   - clientid: func-python36-sayhi-1
    24.     subscribe:
    25.       topic: t
    26.     function:
    27.       name: python36-sayhi
    28.     publish:
    29.       topic: t/py3hi
    30.   - clientid: func-node85-sayhi-1
    31.     subscribe:
    32.       topic: t
    33.     function:
    34.       name: node85-sayhi
    35.     publish:
    36.       topic: t/node8hi
    37. functions:
    38.   - name: python27-sayhi
    39.     service: function-python27-sayhi
    40.     instance:
    41.       min: 0
    42.       max: 10
    43.   - name: sql-filter
    44.     service: function-sql-filter
    45.   - name: python36-sayhi
    46.     service: function-python36-sayhi
    47.   - name: node85-sayhi
    48.     service: function-node85-sayhi
    49. logger:
    50.   path: var/log/baetyl/service.log
    51.   level: "debug"
    52.  
    53. # /usr/local/var/db/baetyl/function-filter-conf/service.yml
    54. functions:
    55.   - name: sql-filter
    56.     handler: 'select qos() as qos, topic() as topic, * where id < 10'
    57.  
    58. # /usr/local/var/db/baetyl/function-sayhi-conf/service.yml
    59. functions:
    60.   - name: 'python27-sayhi'
    61.     handler: 'index.handler'
    62.     codedir: 'var/db/baetyl/function-sayhi'
    63.   - name: 'python36-sayhi'
    64.     handler: 'index.handler'
    65.     codedir: 'var/db/baetyl/function-sayhi'
    66.  
    67. # /usr/local/var/db/baetyl/function-sayjs-conf/service.yml
    68. functions:
    69.   - name: 'node85-sayhi'
    70.     handler: 'index.handler'
    71.     codedir: 'var/db/baetyl/function-sayhi'

    Python 函数代码无需修。/usr/local/var/db/baetyl/function-sayhi-code/index.py 实现如下：

    1. #!/usr/bin/env python
    2. # -*- coding:utf-8 -*-
    3. """
    4. function to say hi in python
    5. """
    6.  
    7. def handler(event, context):
    8.     """
    9.     function handler
    10.     """ 
    11.     res = {}
    12.     if isinstance(event, dict):
    13.         if "err" in event:
    14.             raise TypeError(event['err'])
    15.         res = event
    16.     elif isinstance(event, bytes):
    17.         res['bytes'] = event.decode("utf-8")
    18.  
    19.     if 'messageQOS' in context:
    20.         res['messageQOS'] = context['messageQOS']
    21.     if 'messageTopic' in context:
    22.         res['messageTopic'] = context['messageTopic']
    23.     if 'messageTimestamp' in context:
    24.         res['messageTimestamp'] = context['messageTimestamp']
    25.     if 'functionName' in context:
    26.         res['functionName'] = context['functionName']
    27.     if 'functionInvokeID' in context:
    28.         res['functionInvokeID'] = context['functionInvokeID']
    29.  
    30.     res['Say'] = 'Hello Baetyl'
    31.     return res

    Node 函数代码无需修。/usr/local/var/db/baetyl/function-sayjs-code/index.js 实现如下：

    1. #!/usr/bin/env node
    2.  
    3. const hasAttr = (obj, attr) => {
    4.     if (obj instanceof Object && !(obj instanceof Array)) {
    5.         if (obj[attr] != undefined) {
    6.             return true;
    7.         }
    8.     }
    9.     return false;
    10. };
    11.  
    12. const passParameters = (event, context) => {
    13.     if (hasAttr(context, 'messageQOS')) {
    14.         event['messageQOS'] = context['messageQOS'];
    15.     }
    16.     if (hasAttr(context, 'messageTopic')) {
    17.         event['messageTopic'] = context['messageTopic'];
    18.     }
    19.     if (hasAttr(context, 'messageTimestamp')) {
    20.         event['messageTimestamp'] = context['messageTimestamp'];
    21.     }
    22.     if (hasAttr(context, 'functionName')) {
    23.         event['functionName'] = context['functionName'];
    24.     }
    25.     if (hasAttr(context, 'functionInvokeID')) {
    26.         event['functionInvokeID'] = context['functionInvokeID'];
    27.     }
    28. };
    29.  
    30. exports.handler = (event, context, callback) => {
    31.     // support Buffer & json object
    32.     if (Buffer.isBuffer(event)) {
    33.         const message = event.toString();
    34.         event = {}
    35.         event["bytes"] = message;
    36.     }
    37.     else if("err" in event) {
    38.         return callback(new TypeError(event['err']))
    39.     }
    40.  
    41.     passParameters(event, context);
    42.     event['Say'] = 'Hello Baetyl'
    43.     callback(null, event);
    44. };

    如上配置，假若 MQTTBox 基于上述配置信息已与 Hub 服务建立连接，向 Hub 发送主题为 t 的消息，函数计算服务会降消息分别路由给 python27-sayhi、python36-sayhi、node85-sayhi 和 sql-filter 函数处理，并分别输出主题为 t/py2hi、t/py3hi、t/node8hi 和 t/sqlfilter 的消息。这时订阅主题 # 的 MQTT client 将会接收到这这些消息，以及原消息 t 和 Hub 服务直接转主题的消息 t/topic。

    提示：凡是在 rules 消息路由配置项中出现、用到的函数，必须在 functions 配置项中进行函数实例的配置，否则无法正常启动函数运行时实例。

    Baetyl 启动

    依据 步骤二，执行 sudo systemctl start baetyl 以容器模式启动 Baetyl，如果 Baetyl 已经启动，执行 sudo systemctl restart baetyl 来重启。

    查看 Baetyl 主程序的日志，执行 sudo tail -f -n 40 /usr/local/var/log/baetyl/baetyl.log 显示如下：

    ../_images/function-start-log.pngBaetyl 加载、启动日志

    同样，我们也可以通过执行命令 docker ps 查看系统当前正在运行的 docker 容器列表，具体如下图示。

    ../_images/docker-ps1.png通过 docker ps 命令查看系统当前运行 docker 容器列表

    经过对比，不难发现，本次 Baetyl 启动时已经成功加载了 Hub 服务和函数计算服务，函数运行时服务实例并没有启动，因为函数运行时服务实例在有消息触发时才会动态创建。

    MQTTBox 建立连接

    本次测试中，我们采用 TCP 连接方式对 MQTTBox 进行连接信息配置，然后点击 Add subscriber 按钮订阅主题 # ，该主题用于接收所有 Hub 服务收到的消息。

    消息处理验证

    通过查看 /usr/local/var/db/baetyl/function-sayhi-code/index.py 代码文件可以发现，在接收到某字典类格式的消息后，函数 handler 会对其进行一系列处理，然后将处理结果返回。返回的结果中包括各种追加的上下文信息，比如 messageTopic、functionName 等。

    这里，我们通过 MQTTBox 将消息 {"id":1} 发布给主题 t ，然后观察 MQTTBox 接收到的消息如下。

    ../_images/mqttbox-tcp-process-success.pngMQTTBox 接收消息

    发送消息后，我们快速执行命令 docker ps 查看系统当前正在运行的容器列表，所有函数运行时服务实例都被启动了，其结果如下图示。

    ../_images/docker-ps-after-trigger.png通过 docker ps 命令查看系统当前正在运行的容器列表

    综上，我们通过 Hub 服务和函数计算服务模拟了消息在本地处理的过程，可以看出该框架非常适合用于边缘处理消息流。