简介
当我们使用 Shifu 采集数据时，需要主动调用 Shifu 对应的接口才能获取数据。但是如果需要将数据发送到 MQTT Broker 中，需要手动编写推送程序。 Shifu 在 v0.4.0 加入了 Telemetry Service(遥测服务)， 能够将遥测收集的数据发布到遥测服务器（如MQTT Broker）中。
Shifu Telemetry Service 简介
Shifu Telemetry Service 是 Shifu 的一个组件，可以将遥测收集的数据通过MQTT等协议推送或发布到指定服务器中。
MQTT 简介
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。
架构图
准备 🗂
- kubectl v1.25.2
- docker
- kind v0.16.0
- git
- python3
拉取 Shifu 代码
使用 git 命令下载 Shifu 代码。
git clone https://github.com/Edgenesis/shifu.git
cd shifu
创建 kind 集群
使用一下命令创建 Kind 集群。
$ kind delete cluster && kind create cluster
Creating cluster "kind" ...
✓ Ensuring node image (kindest/node:v1.25.2) 🖼
✓ Preparing nodes 📦
✓ Writing configuration 📜
✓ Starting control-plane 🕹️
✓ Installing CNI 🔌
✓ Installing StorageClass 💾
Set kubectl context to "kind-kind"
You can now use your cluster with:
kubectl cluster-info --context kind-kind
Have a question, bug, or feature request? Let us know! https://kind.sigs.k8s.io/#community 🙂
安装 Shifu
使用 Shifu 安装命令，一键安装 Shifu。
$ kubectl apply -f pkg/k8s/crd/install/shifu_install.yml
$ kubectl get pod -n shifu-crd-system
NAME READY STATUS RESTARTS AGE
shifu-crd-controller-manager-5bbdb4d786-s6h4m 2/2 Running 0 1s
创建一个 mosquitto MQTT broker
使用以下命令在docker中启动一个mosquitto MQTT broker运行在后台，并映射到本地的1883和9001端口。
$ docker run -itd -p 1883:1883 -p 9001:9001 --name mosquitto eclipse-mosquitto:2.0.15
当输出为以下值时表示 mosquitto 已创建成功。
$ docker ps | grep mosquitto
35dcebdf3979 eclipse-mosquitto:2.0.15 "/docker-entrypoint.…" 12 seconds ago Up 11 seconds 0.0.0.0:1883->1883/tcp, 0.0.0.0:9001->9001/tcp mosquitto
部署 Shifu Telemetry Service
使用以下命令在集群中安装 Telemetry Service
$ kubectl apply -f examples/telemetryservice
deployment.apps/telemetryservice created
service/telemetryservice created
通过以下命令查看TelemetryService的运行情况。如果STATUS为RUNNING则表示Telemetry Service运行成功。
$ kubectl get po -n shifu-service
NAME READY STATUS RESTARTS AGE
telemetryservice-846f6bcf6c-ckqf6 1/1 Running 0 16s
运行 虚拟设备
安装 Python 依赖（Flask包）。
$ python3 -m pip install -r examples/mqttTelemetryService/requirements.txt
使用Python脚本模拟一个用于测试遥测服务的虚拟设备。当向该虚拟设备发送HTTP请求时会返回 “status api” 字符串。
$ python3 examples/mqttTelemetryService/mockdevice.py
* Serving Flask app 'telemetryservices'
* Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
* Running on all addresses (0.0.0.0)
* Running on http://127.0.0.1:12345
* Running on http://192.168.14.163:12345
Press CTRL+C to quit
创建 deviceshifu
修改 Edgedevice 配置文件
将 examples/mqttTelemetryService/deployment/http-edgedevice.yaml 中的 address 字段修改为本机 IP 地址，以便 Shifu 能够连接上设备。
spec:
sku: "HTTP Device"
connection: Ethernet
address: 192.168.14.163:12345 # edit it
protocol: HTTP
修改 TelemetryService 配置文件
将 "examples/mqttTelemetryService/deployment/http-deviceshifu-telemetryservice.yaml" 中的 MQTTServerAddress 字段修改为本机 IP 地址，或您 MQTT Broker 的地址。
apiVersion: shifu.edgenesis.io/v1alpha1
kind: TelemetryService
metadata:
name: push-endpoint-1
namespace: devices
spec:
telemetrySeriveEndpoint: http://telemetryservice.shifu-service.svc.cluster.local
serviceSettings:
MQTTSetting:
MQTTServerAddress: 192.168.14.163:1883 # edit it
MQTTTopic: /test/test
安装 deviceShifu
使用以下命令安装 deviceShifu 以及 Telemetry Service 的推送配置。
$ kubectl apply -f examples/mqttTelemetryService/deployment
configmap/http-configmap-0.0.1 created
deployment.apps/deviceshifu-http-deployment created
service/deviceshifu-http created
telemetryservice.shifu.edgenesis.io/push-endpoint-1 created
edgedevice.shifu.edgenesis.io/edgedevice-http created
使用以下命令，可以查看 deviceShifu 已经部署到集群中。
$ kubectl get pods -n deviceshifu
NAME READY STATUS RESTARTS AGE
deviceshifu-http-deployment-7f5878fb79-ct7zz 1/1 Running 0 4m33s
测试结果
我们需要在命令行中执行以下命令来订阅 mosquitto 以便能够观察到数据变化。
$ docker exec -it mosquitto mosquitto_sub -v -t /test/test
status api
status api
status api
我们可以观察到每隔 6 秒 mosquitto MQTT broker 会收到一条消息来自 telemetryService。
总结
通过本文，我们介绍了如何通过 Shifu 以及 Telemetry Service 实现定时采集数据并发布到遥测服务器 MQTT Broker 中。我们可以通过订阅 MQTT 的 Topic 实现数据整合、设备发现、数据监控等操作来充分利用我们采集到的数据。
本文由边无际授权发布