企业官网建设流程全解析

• 指标监控
• 监控大盘
• CLI
• 告警
• MCP 服务

导读

Kafka 是 Glances 众多导出（export）目标中的一员，用于把本地或远程主机的实时监控统计以 JSON 形式持续写入 Kafka topic，供下游流处理、日志采集或实时告警系统消费。本文以 docs/gw/kafka.rst 为核心，完整讲解[kafka]配置段、--export kafka启动方式、消息的 key/value 结构与消费端写法，并结合仓库中的导出模块源码与配置文件，剖析其底层实现原理，帮助读者在实际环境中快速落地「Glances → Kafka」数据管道。

Kafka 导出配置：配置文件中的[kafka]段

Kafka 导出的连接信息需要在 Glances 配置文件中定义。在默认配置文件 conf/glances.conf 中，[kafka]段如下（与官方文档 docs/gw/kafka.rst 保持一致）：

[kafka] # Configuration for the --export kafka option # http://kafka.apache.org/ host=localhost port=9092 topic=glances #compression=gzip # Tags will be added for all events #tags=foo:bar,spam:eggs # You can also use dynamic values #tags=hostname:`hostname -f`

各配置项的含义与说明如下：

从源码实现看，必填与可选配置项的划分与上述表格完全对应。glances/exports/glances_kafka/init.py 中通过基类的load_conf加载配置：

self.export_enable = self.load_conf( 'kafka', mandatories=['host', 'port', 'topic'], options=['compression', 'tags'] ) if not self.export_enable: exit('Missing KAFKA config')

也就是说，host、port、topic三个键缺失时，load_conf返回False，Glances 会直接以Missing KAFKA config退出；compression与tags缺失则不影响启用，仅以None处理。

tags 的静态与动态写法

tags支持两种写法：

• 静态标签：直接写出key:value对，多个标签用逗号分隔，例如tags=foo:bar,spam:eggs，最终每条记录都会带上"foo": "bar"、"spam": "eggs"这两个字段。
• 动态标签：值部分可以引用 shell 命令，用反引号包裹，例如tags=hostname:hostname -f``，这样会动态执行hostname -f并把结果作为hostname标签的值，便于在多主机场景下区分数据来源。

底层解析由基类 glances/exports/export.py 的parse_tags完成：先按逗号切分为多个key:value对，再对每个对按第一个冒号切分，拼装成字典；若某个对缺少冒号（ValueError），会记一条 info 日志并整体返回空字典：

d_tags = dict(x.split(":", 1) for x in tags.split(","))

关于 compression 的注意点

原文档明确指出：可以启用压缩（例如gzip），但压缩会消耗主机 CPU。因此对于 CPU 敏感或监控数据量不大的场景，建议保持注释状态，仅在网络带宽成为瓶颈时再开启，并需要权衡压缩带来的 CPU 开销。

启动 Kafka 导出：glances --export kafka

完成配置后，以如下命令启动 Glances 即可把统计写入 Kafka：

$ glances --export kafka

命令行层面的处理流程如下：

1. 参数解析：glances/main.py 定义--export参数，支持逗号分隔的导出模块列表（例如--export kafka,influxdb可同时导出到多个目标）。
2. 参数映射：glances/main.py 会把每个导出项映射为对应的布尔属性：

if args.export is not None: for p in args.export.split(','): setattr(args, 'export_' + p, True)

3. 动态加载：glances/stats.py 启动时会扫描glances/exports/目录下的所有glances_*子目录，把glances_kafka规约为导出器名kafka，当export_kafka为真时通过import_module导入该模块并实例化Export类，注册到活动导出器字典中。

需要说明的依赖前提：Kafka 导出模块依赖kafka-python库（见 pyproject.toml 的"kafka-python"依赖项）。使用前需确保该库已安装，例如通过pip install 'glances[kafka]'或手动安装kafka-python，否则模块导入会失败。

消息格式：key 为插件名，value 为 JSON 字典

统计数据以原生 JSON 格式发送到指定 topic，每条消息的结构约定为：

• key：插件名称（plugin name），例如mem、cpu、network等；
• value：JSON 字典，包含该插件的各项统计字段。

原文档给出的内存（memory）插件消息示例（Pythonkafka客户端打印的ConsumerRecord）如下：

ConsumerRecord(topic=u'glances', partition=0, offset=1305, timestamp=1490460592248, timestamp_type=0, key='mem', value=u'{"available": 2094710784, "used": 5777428480, "cached": 2513543168, "mem_careful": 50.0, "percent": 73.4, "free": 2094710784, "mem_critical": 90.0, "inactive": 2361626624, "shared": 475504640, "history_size": 28800.0, "mem_warning": 70.0, "total": 7872139264, "active": 4834361344, "buffers": 160112640}', checksum=214895201, serialized_key_size=3, serialized_value_size=303)

可以看到，value 中不仅包含total、used、free、available、cached、buffers、active、inactive、shared、percent等基础内存指标，还带有mem_warning、mem_careful、mem_critical这些阈值字段以及history_size历史窗口字段。这是 Glances 导出流程的通用行为：基类 glances/exports/export.py 的update方法会同时取插件的统计与阈值（limits），把两者合并后再逐插件调用各导出器。

发送逻辑的源码实现

glances/exports/glances_kafka/init.py 中的export方法完成实际发送：

def export(self, name, columns, points): """Write the points to the kafka server.""" logger.debug(f"Export {name} stats to Kafka") # Create DB input data = dict(zip(columns, points)) if self.tags is not None: data.update(self.parse_tags(self.tags)) # Send stats to the kafka topic # key=<plugin name> # value=JSON dict try: self.client.send( self.topic, # Kafka key name needs to be bytes #1593 key=name.encode('utf-8'), value=data, ) except Exception as e: logger.error(f"Cannot export {name} stats to Kafka ({e})")

几个关键细节：

• columns与points由基类 glances/exports/export.py 的build_export递归展开得到：字典的键统一转为小写字符串，布尔值转成 JSON 字符串，列表值以空格拼接，嵌套字典递归拍平，并支持通过[export]段的exclude_fields正则过滤不需要的字段。
• data由「字段名-字段值」交叉配对得到，若配置了tags，还会合并进标签字段。
• 消息key使用name.encode('utf-8')转为字节串——这是 Kafka key 的格式要求，对应源码注释中提到的 issue #1593。
• value直接以 dict 形式交给KafkaProducer，由构造时注册的value_serializer序列化为 JSON（见下文init实现）。
• 单条发送失败不会导致程序退出，只会记录 error 日志，符合监控类导出的容错设计。

生产端连接的初始化

glances/exports/glances_kafka/init.py 的init方法负责建立 KafkaProducer 连接：

server_uri = f'{self.host}:{self.port}' try: s = KafkaProducer( bootstrap_servers=server_uri, value_serializer=lambda v: json_dumps(v), compression_type=self.compression, ) except Exception as e: logger.critical(f"Cannot connect to Kafka server {server_uri} ({e})") sys.exit(2) else: logger.info(f"Connected to the Kafka server {server_uri}")

• broker 地址由host:port拼接而成；
• value_serializer使用 Glances 的json_dumps把 Python 字典序列化为 JSON 字节串；
• compression_type直接透传配置中的压缩类型（如gzip），为None时不启用压缩；
• 连接失败会记录 critical 日志并以退出码 2 终止进程；成功则记录一条 info 日志。

从源码结构看，当前init方法仅传递了bootstrap_servers、value_serializer与compression_type，未在代码中显式配置 SASL/SSL 等安全认证参数；若 Kafka broker 开启了认证，需要读者在 fork 或扩展该模块时自行补充，或优先使用支持安全协议的转发组件。

消费端示例：用 Python 读取 Glances 数据

原文档提供了基于kafka-python的消费端示例，用于验证 topic 中收到的 Glances 记录：

from kafka import KafkaConsumer import json consumer = KafkaConsumer('glances', value_deserializer=json.loads) for s in consumer: print(s)

• KafkaConsumer('glances', ...)订阅glancestopic；
• value_deserializer=json.loads把接收到的 JSON 字节串自动反序列化为 Python dict，与生产端value_serializer的 JSON 序列化正好配对；
• 循环中逐条打印ConsumerRecord，即可看到类似上文 memory 插件的完整记录（含 key、value、partition、offset、timestamp 等信息）。

消费端打印结果与文档中的ConsumerRecord示例一致，说明「生产端 JSON 序列化 → topic → 消费端 JSON 反序列化」整条链路是闭合的。实际工程中，也可基于该模式对接 Logstash、Kafka Connect、Spark Streaming 或自研消费程序做进一步的数据清洗、入库与告警。

收尾与资源释放：exit 方法

Glances 退出时，glances/exports/glances_kafka/init.py 的exit方法会先flush()确保未发送的消息全部送达，再close()关闭生产者连接，最后调用基类exit()完成统一收尾。这在长时间运行的监控导出场景中很重要，可避免进程退出时消息丢失或连接泄漏。

适用边界与注意事项

• 仅支持导出到 topic，不涉及消费：Glances 的 Kafka 模块只实现生产端（KafkaProducer），消费端需自行搭建（如上文的KafkaConsumer示例）。
• 压缩的取舍：compression=gzip能减少网络传输量，但会占用主机 CPU（原文档明确提示），建议按实际负载权衡。
• 配置缺失即退出：[kafka]段缺少host/port/topic任一必填项时，Glances 会打印Missing KAFKA config并退出，配置前请核对键名。
• 连接失败即退出：broker 不可达时进程以退出码 2 终止，因此建议先确认 broker 监听与网络连通性，再启动--export kafka。
• 动态标签依赖 shell：tags=hostname:hostname -f`` 这类写法会执行反引号中的命令，需确保目标命令在运行 Glances 的主机上可用。
• 依赖安装：需要kafka-python库（见 pyproject.toml），可通过pip install 'glances[kafka]'一并安装。
• 更多导出目标：Kafka 只是 Glances 导出体系的一员，--export支持逗号分隔的多个目标组合；完整的导出器清单见 glances/exports/ 目录及 docs/gw/index.rst 中的网关（gateway）文档索引。

小结

通过[kafka]配置段、glances --export kafka命令以及 topic 中「key=插件名、value=JSON dict」的消息约定，Glances 可以稳定地把 CPU、内存、网络等插件统计持续推送给 Kafka，为实时监控数据的中转、归档与分析提供了标准化的入口。本文结合 glances/exports/glances_kafka/init.py 的Export类实现、glances/exports/export.py 的基类导出流程以及 conf/glances.conf 的默认配置，从配置、命令、消息结构与源码四个层面完整还原了该功能，读者可按上文步骤直接搭建一套可运行的「Glances → Kafka → 消费者」监控数据管道。

• 指标监控
• 监控大盘
• CLI
• 告警
• MCP 服务