目录 背景介绍 EdgeX 简介 InfluxDB 简介 开发过程 架构设计 EdgeX application service 前提条件 Application Service Docker 容器 InfluxDB 环境搭建 Telegraf 配置 Docker 容器 实验结果 MQTT broker 内容查看 InfluxDB Graph 开源代码 关于亿琪软件 欢迎关注 EdgeX Foundry 中文社区公众号，及时获取最新资料和分享。 日常情况下，在边缘计算端侧，一般不需要数据持久化到本地磁盘或数据库，但是，某些特殊应用下，需要做本地数据保存和分析时，这项功能就非常有必要。本文将介绍 EdgeX 边缘计算框架在本地保存 telemetry 数据到 InfluxDB 的具体方法和结果。 背景介绍 EdgeX 简介 EdgeX Foundry 边缘计算框架的分层和服务为边缘设备/节点和云/企业应用之间提供了一个双向转换引擎。EdgeX 作为边缘计算平台是非常不错的选择。 本文需要采用自定义开发的 EdgeX Application service 来完成数据持久化的工作。 InfluxDB 简介 InfluxDB 是一个由 InfluxData 公司开发的开源时序型数据库。它由Go写成，着力于高性能地查询与存储时序型数据。InfluxDB 被广泛应用于存储系统的监控数据，IoT行业的实时数据等场景。 EdgeX 数据写入（存入）InfluxDB 是非常适合且有必要的，本文使用 MQTT 和 Telegraf 作为 InfluxDB plugin 来实现具体功能。 开发过程 架构设计 EdgeX 应用程序服务(app-influxdb-export)通过 MQTT 将读数导出到 InfluxDB。此服务示例是使用 EdgeX v2.3.0 版本创建的，如需其他版本，请与我们联系或自行开发。 这个管道服务，有两部分组成： 将 EdgeX 事件/读取对象转换为 Influx line 协议； 将结果导出到 MQTT Broker，这是 Telegraf / Influx 用来监听并从中获取数据的 MQTT 主题； InfluxDB 实现数据持久化入库工作； EdgeX application service 前提条件 需要启动 EdgeX 框架； 启动 MQTT Broker 容器； Application Service 目录结构很简单，文件不多； ├── Dockerfile // docker 文件 ├── LICENSE ├── Makefile // make build ├── README.md ├── app-service-influx.png ├── go.mod // gomod 配置 ├── go.sum ├── main.go // 主程序 ├── pkg │ └── transforms │ └── conversions.go // 转换器 └── res └── configuration.toml // 配置文件 主程序，不复杂，遵循 SDK 原则即可； 这里省略源文件，请到github下载。 尝试编译，能成功生成二进制文件； make build Docker 容器 Dockerfile 描述文件； 这里省略源文件，请到github下载。 创建 docker image; docker build . -t edgexfoundry/app-influxdb:2.3.0 docker compose 片段，加入到你的 docker-compose.yml ; app-service-influxdb-export: image: edgexfoundry/app-influxdb:2.3.0 ports: - 127.0.0.1:59780:59780/tcp container_name: edgex-app-influxdb-export hostname: edgex-app-influxdb-export depends_on: - consul - data environment: CLIENTS_CORE_COMMAND_HOST: edgex-core-command CLIENTS_CORE_DATA_HOST: edgex-core-data CLIENTS_CORE_METADATA_HOST: edgex-core-metadata CLIENTS_SUPPORT_NOTIFICATIONS_HOST: edgex-support-notifications CLIENTS_SUPPORT_SCHEDULER_HOST: edgex-support-scheduler DATABASES_PRIMARY_HOST: edgex-redis EDGEX_SECURITY_SECRET_STORE: "false" MESSAGEQUEUE_HOST: edgex-redis REGISTRY_HOST: edgex-core-consul SERVICE_HOST: edgex-app-influxdb-export TRIGGER_EDGEXMESSAGEBUS_PUBLISHHOST_HOST: edgex-redis TRIGGER_EDGEXMESSAGEBUS_SUBSCRIBEHOST_HOST: edgex-redis MQTTCONFIG_BROKERADDRESS: edgex-mqtt-broker:1883 read_only: true restart: always networks: - edgex-network security_opt: - no-new-privileges:true user: 2002:2001 InfluxDB 环境搭建 Telegraf 配置 配置文件，注意保留你需要的部分； [[inputs.mqtt_consumer]] ## Broker URLs for the MQTT server or cluster. To connect to multiple ## clusters or standalone servers, use a separate plugin instance. ## example: servers = ["tcp://localhost:1883"] ## servers = ["ssl://localhost:1883"] ## servers = ["ws://localhost:1883"] servers = ["tcp://192.168.123.12:1883"] ## Topics that will be subscribed to. topics = [ "telegraf/host01/cpu", "telegraf/+/mem", "sensors/#", "edgex/EdgeXEvents" ] ## The message topic will be stored in a tag specified by this value. If set ## to the empty string no topic tag will be created. # topic_tag = "topic" ## QoS policy for messages ## 0 = at most once ## 1 = at least once ## 2 = exactly once ## ## When using a QoS of 1 or 2, you should enable persistent_session to allow ## resuming unacknowledged messages. # qos = 0 ## Connection timeout for initial connection in seconds # connection_timeout = "30s" ## Max undelivered messages ## This plugin uses tracking metrics, which ensure messages are read to ## outputs before acknowledging them to the original broker to ensure data ## is not lost. This option sets the maximum messages to read from the ## broker that have not been written by an output. ## ## This value needs to be picked with awareness of the agent's ## metric_batch_size value as well. Setting max undelivered messages too high ## can result in a constant stream of data batches to the output. While ## setting it too low may never flush the broker's messages. # max_undelivered_messages = 1000 ## Persistent session disables clearing of the client session on connection. ## In order for this option to work you must also set client_id to identify ## the client. To receive messages that arrived while the client is offline, ## also set the qos option to 1 or 2 and don't forget to also set the QoS when ## publishing. # persistent_session = false ## If unset, a random client ID will be generated. # client_id = "" ## Username and password to connect MQTT server. # username = "telegraf" # password = "metricsmetricsmetricsmetrics" ## Optional TLS Config # tls_ca = "/etc/telegraf/ca.pem" # tls_cert = "/etc/telegraf/cert.pem" # tls_key = "/etc/telegraf/key.pem" ## Use TLS but skip chain & host verification # insecure_skip_verify = false ## Client trace messages ## When set to true, and debug mode enabled in the agent settings, the MQTT ## client's messages are included in telegraf logs. These messages are very ## noisey, but essential for debugging issues. # client_trace = false ## Data format to consume. ## Each data format has its own unique set of configuration options, read ## more about them here: ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md data_format = "influx" ## Enable extracting tag values from MQTT topics ## _ denotes an ignored entry in the topic path # [[inputs.mqtt_consumer.topic_parsing]] # topic = "" # measurement = "" # tags = "" # fields = "" ## Value supported is int, float, unit # [[inputs.mqtt_consumer.topic.types]] # key = type # Configuration for sending metrics to InfluxDB 2.0 [[outputs.influxdb_v2]] ## The URLs of the InfluxDB cluster nodes. ## ## Multiple URLs can be specified for a single cluster, only ONE of the ## urls will be written to each interval. ## ex: urls = ["https://us-west-2-1.aws.cloud2.influxdata.com"] urls = ["http://192.168.123.12:8086"] ## Token for authentication. token = "PlbXqGQqYrnmZh22q88C0eS7CUyVSa9n1t2CjDY1nkA9RbXwM3BHBm0FIK3f3hT6p3oJKOrtFWLqeLgY9WQmyQ==" ## Organization is the name of the organization you wish to write to. organization = "yiqisoft" ## Destination bucket to write into. bucket = "edgex" ## The value of this tag will be used to determine the bucket. If this ## tag is not set the 'bucket' option is used as the default. # bucket_tag = "" ## If true, the bucket tag will not be added to the metric. # exclude_bucket_tag = false ## Timeout for HTTP messages. # timeout = "5s" ## Additional HTTP headers # http_headers = {"X-Special-Header" = "Special-Value"} ## HTTP Proxy override, if unset values the standard proxy environment ## variables are consulted to determine which proxy, if any, should be used. # http_proxy = "http://corporate.proxy:3128" ## HTTP User-Agent # user_agent = "telegraf" ## Content-Encoding for write request body, can be set to "gzip" to ## compress body or "identity" to apply no encoding. # content_encoding = "gzip" ## Enable or disable uint support for writing uints influxdb 2.0. # influx_uint_support = false ## HTTP/2 Timeouts ## The following values control the HTTP/2 client's timeouts. These settings ## are generally not required unless a user is seeing issues with client ## disconnects. If a user does see issues, then it is suggested to set these ## values to "15s" for ping timeout and "30s" for read idle timeout and ## retry. ## ## Note that the timer for read_idle_timeout begins at the end of the last ## successful write and not at the beginning of the next write. # ping_timeout = "0s" # read_idle_timeout = "0s" ## Optional TLS Config for use on HTTP connections. # tls_ca = "/etc/telegraf/ca.pem" # tls_cert = "/etc/telegraf/cert.pem" # tls_key = "/etc/telegraf/key.pem" ## Use TLS but skip chain & host verification # insecure_skip_verify = false Docker 容器 docker-compose.yml services: influxdb: image: influxdb:latest container_name: influxdb2 volumes: - /opt/devel/influxdb/data:/var/lib/influxdb2:rw environment: - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=mytoken ports: - 8086:8086 restart: unless-stopped telegraf: image: telegraf:latest container_name: telegraf volumes: # Sync timezone with host - /etc/localtime:/etc/localtime:ro # Map Telegraf configuration file - /opt/devel/influxdb/telegraf.conf:/etc/telegraf/telegraf.conf:ro # Map /tmp to permanent storage (this includes /tmp/metrics.out) - /opt/devel/influxdb:/tmp:rw environment: - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=mytoken restart: unless-stopped depends_on: - influxdb 启动容器 docker compose up -d 实验结果 MQTT broker 内容查看 确认 mqtt 内容为 Influx line protocol； InfluxDB Graph 确认 InfluxDB 数据接收正常。 开源代码 完整的源代码在：https://github.com/yiqisoft/edgex-app-influxdb-export.git 关于亿琪软件 上海亿琪软件有限公司成立于2016年，专注于 5G 通信、AI 人工智能、边缘计算和大数据网络安全多项技术领域，致力于物联网领域前沿技术的创新，为用户提供全方位、智能化和安全的物联网解决方案。 2023年，公司发布“YiFUSION｜工业边缘智能融合一体机”产品，为工业客户提供一整套的边缘计算+AI能力：高性能数据采集、多类型数据融合、AI算法集成、云端业务对接。在边缘网关的基础上，集成了 IoT 平台的边缘协同能力、本地Web SCADA 和 HMI 功能、本地数据存储、边缘 AI 视频分析、行业应用集成等。 2022年，公司推出 “YiCLOUD｜亿琪云”一站式物联网应用解决方案。公司的业务涵盖了智慧城市、智慧农业、智能工厂和智慧园区等多个领域，公司软硬件产品和解决方案获得华为技术认证，得到中国移动OCP认证，公司还是边缘计算产业联盟ECC成员。 感知物联，畅快无限。 联系我们 网站：http://yiqisoft.cn 邮件：support@yiqisoft.cn 电话：021-68863086 手机：186-1666-9123