利用 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 服务和函数计算服务模拟了消息在本地处理的过程，可以看出该框架非常适合用于边缘处理消息流。