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

声明

  • 本文测试所用设备系统为 Ubuntu 18.04
  • python 版本为 3.6,2.7 版本配置流程相同,但需要在 python 脚本中注意语言差异
  • 模拟 MQTT client 行为的客户端为 MQTTBox
  • 本文所用镜像为依赖 Baetyl 源码自行编译所得,具体请查看 如何从源码构建镜像

提示: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 应用配置改成如下配置:

# /usr/local/var/db/baetyl/application.yml
version: v0
services:
  - name: localhub
    image: hub.baidubce.com/baetyl/baetyl-hub
    replica: 1
    ports:
      - 1883:1883
    mounts:
      - name: localhub-conf
        path: etc/baetyl
        readonly: true
      - name: localhub-data
        path: var/db/baetyl/data
      - name: localhub-log
        path: var/log/baetyl
  - name: function-manager
    image: hub.baidubce.com/baetyl/baetyl-function-manager
    replica: 1
    mounts:
      - name: function-manager-conf
        path: etc/baetyl
        readonly: true
      - name: function-manager-log
        path: var/log/baetyl
  - name: function-python27-sayhi
    image: hub.baidubce.com/baetyl/baetyl-function-python27
    replica: 0
    mounts:
      - name: function-sayhi-conf
        path: etc/baetyl
        readonly: true
      - name: function-sayhi-code
        path: var/db/baetyl/function-sayhi
        readonly: true
  - name: function-python36-sayhi
    image: hub.baidubce.com/baetyl/baetyl-function-python36
    replica: 0
    mounts:
      - name: function-sayhi-conf
        path: etc/baetyl
        readonly: true
      - name: function-sayhi-code
        path: var/db/baetyl/function-sayhi
        readonly: true
  - name: function-node85-sayhi
    image: hub.baidubce.com/baetyl/baetyl-function-node85
    replica: 0
    mounts:
      - name: function-sayjs-conf
        path: etc/baetyl
        readonly: true
      - name: function-sayjs-code
        path: var/db/baetyl/function-sayhi
        readonly: true
  - name: function-sql-filter
    image: hub.baidubce.com/baetyl/baetyl-function-sql
    replica: 0
    mounts:
      - name: function-filter-conf
        path: etc/baetyl
        readonly: true
volumes:
  # hub
  - name: localhub-conf
    path: var/db/baetyl/localhub-conf
  - name: localhub-data
    path: var/db/baetyl/localhub-data
  - name: localhub-cert
    path: var/db/baetyl/localhub-cert-only-for-test
  - name: localhub-log
    path: var/db/baetyl/localhub-log
  # function
  - name: function-manager-conf
    path: var/db/baetyl/function-manager-conf
  - name: function-manager-log
    path: var/db/baetyl/function-manager-log
  - name: function-sayhi-conf
    path: var/db/baetyl/function-sayhi-conf
  - name: function-sayhi-code
    path: var/db/baetyl/function-sayhi-code
  - name: function-sayjs-conf
    path: var/db/baetyl/function-sayjs-conf
  - name: function-sayjs-code
    path: var/db/baetyl/function-sayjs-code
  - name: function-filter-conf
    path: var/db/baetyl/function-filter-conf

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

# /usr/local/var/db/baetyl/localhub-conf/service.yml
listen:
  - tcp://0.0.0.0:1883
principals:
  - username: test
    password: hahaha
    permissions:
      - action: 'pub'
        permit: ['#']
      - action: 'sub'
        permit: ['#']
subscriptions:
  - source:
      topic: 't'
    target:
      topic: 't/topic'
logger:
  path: var/log/baetyl/service.log
  level: "debug"

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

# /usr/local/var/db/baetyl/function-manager-conf/service.yml
hub:
  address: tcp://localhub:1883
  username: test
  password: hahaha
rules:
  - clientid: func-python27-sayhi-1
    subscribe:
      topic: t
    function:
      name: python27-sayhi
    publish:
      topic: t/py2hi
  - clientid: func-sql-filter-1
    subscribe:
      topic: t
      qos: 1
    function:
      name: sql-filter
    publish:
      topic: t/sqlfilter
      qos: 1
  - clientid: func-python36-sayhi-1
    subscribe:
      topic: t
    function:
      name: python36-sayhi
    publish:
      topic: t/py3hi
  - clientid: func-node85-sayhi-1
    subscribe:
      topic: t
    function:
      name: node85-sayhi
    publish:
      topic: t/node8hi
functions:
  - name: python27-sayhi
    service: function-python27-sayhi
    instance:
      min: 0
      max: 10
  - name: sql-filter
    service: function-sql-filter
  - name: python36-sayhi
    service: function-python36-sayhi
  - name: node85-sayhi
    service: function-node85-sayhi
logger:
  path: var/log/baetyl/service.log
  level: "debug"

# /usr/local/var/db/baetyl/function-filter-conf/service.yml
functions:
  - name: sql-filter
    handler: 'select qos() as qos, topic() as topic, * where id < 10'

# /usr/local/var/db/baetyl/function-sayhi-conf/service.yml
functions:
  - name: 'python27-sayhi'
    handler: 'index.handler'
    codedir: 'var/db/baetyl/function-sayhi'
  - name: 'python36-sayhi'
    handler: 'index.handler'
    codedir: 'var/db/baetyl/function-sayhi'

# /usr/local/var/db/baetyl/function-sayjs-conf/service.yml
functions:
  - name: 'node85-sayhi'
    handler: 'index.handler'
    codedir: 'var/db/baetyl/function-sayhi'

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

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
function to say hi in python
"""

def handler(event, context):
    """
    function handler
    """ 
    res = {}
    if isinstance(event, dict):
        if "err" in event:
            raise TypeError(event['err'])
        res = event
    elif isinstance(event, bytes):
        res['bytes'] = event.decode("utf-8")

    if 'messageQOS' in context:
        res['messageQOS'] = context['messageQOS']
    if 'messageTopic' in context:
        res['messageTopic'] = context['messageTopic']
    if 'messageTimestamp' in context:
        res['messageTimestamp'] = context['messageTimestamp']
    if 'functionName' in context:
        res['functionName'] = context['functionName']
    if 'functionInvokeID' in context:
        res['functionInvokeID'] = context['functionInvokeID']

    res['Say'] = 'Hello Baetyl'
    return res

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

#!/usr/bin/env node

const hasAttr = (obj, attr) => {
    if (obj instanceof Object && !(obj instanceof Array)) {
        if (obj[attr] != undefined) {
            return true;
        }
    }
    return false;
};

const passParameters = (event, context) => {
    if (hasAttr(context, 'messageQOS')) {
        event['messageQOS'] = context['messageQOS'];
    }
    if (hasAttr(context, 'messageTopic')) {
        event['messageTopic'] = context['messageTopic'];
    }
    if (hasAttr(context, 'messageTimestamp')) {
        event['messageTimestamp'] = context['messageTimestamp'];
    }
    if (hasAttr(context, 'functionName')) {
        event['functionName'] = context['functionName'];
    }
    if (hasAttr(context, 'functionInvokeID')) {
        event['functionInvokeID'] = context['functionInvokeID'];
    }
};

exports.handler = (event, context, callback) => {
    // support Buffer & json object
    if (Buffer.isBuffer(event)) {
        const message = event.toString();
        event = {}
        event["bytes"] = message;
    }
    else if("err" in event) {
        return callback(new TypeError(event['err']))
    }

    passParameters(event, context);
    event['Say'] = 'Hello Baetyl'
    callback(null, event);
};

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

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

Baetyl 启动

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

提示:Darwin 系统通过源码安装 Baetyl,可执行 sudo baetyl start 以容器模式启动 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 会对其进行一系列处理,然后将处理结果返回。返回的结果中包括各种追加的上下文信息,比如 messageTopicfunctionName 等。

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

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

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

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

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