如何为 pgwatch 定制 gRPC Sink对接自有时序存储的完整开发者指南【免费下载链接】pgwatchpgwatch: PostgreSQL metrics monitor/dashboard项目地址: https://gitcode.com/gh_mirrors/pg/pgwatchpgwatch 是一款功能强大的PostgreSQL 指标监控与仪表盘开源项目内置了 PostgreSQL、Prometheus、JSON 文件等存储后端Sink。当你需要把监控指标推送到 InfluxDB、ClickHouse、IoTDB 等自有时序数据库时pgwatch 的gRPC Sink就是官方预留的万能出口——只需实现一个约 5 个 RPC 接口的小服务就能让 PostgreSQL 监控数据自由落袋。本文带你从零定制属于自己的 gRPC Sink。 gRPC Sink 解决什么问题pgwatch 的数据链路是Reaper收割器定期从被监控的 PostgreSQL 实例采集指标 → 通过 Sink 写入存储 → Web UI / 仪表盘展示。官方文档在 docs/concept/components.md 中明确说明了各 Sink 的定位其中 gRPC Sink 的说明是当内置存储选项都不满足需求时你可以基于 pgwatch 的 protobuf 定义实现自己的 gRPC 服务器将指标推送到任意自定义存储后端。这意味着✅不受限于官方支持的数据库——任何语言、任何时序库都能接✅协议是 protobuf——高性能、强类型、跨语言Go、Java、Python、Rust 均可实现✅附带元数据能力——不仅能推数据还能同步指标定义与监控源的增删 先看懂 protobuf 契约只有 3 个 RPC整个 gRPC 契约就定义在 api/pb/pgwatch.proto 中服务名为Receiver包含 3 个方法RPC 方法请求类型作用你应如何处理UpdateMeasurementsMeasurementEnvelope推送一批指标数据含 DB 名、指标名、自定义标签、结构化数据行核心将数据写入你的时序库SyncMetricSyncReq同步指标/监控源的生命周期操作AddOp/DeleteOp/DefineOp更新本地元数据新增/删除某实例的某指标DefineMetricsgoogle.protobuf.Struct推送完整的指标定义JSON 化后的结构保存指标 schema便于建表/建模板三个方法的返回值统一是Reply其中只有一个logmsg字段——pgwatch 会把这个字段的内容打进日志这是官方给你的调试回音联调时非常好用。数据载荷的结构MeasurementEnvelope大致是DBName被监控的 PostgreSQL 实例名MetricName指标名如pg_stat_statements、pg_settingsCustomTags在 pgwatch 中为该实例打的自定义标签Datarepeated google.protobuf.Struct即每行一条mapstring, value的测量记录 提示internal/sinks/rpc.go中可以看到客户端侧的完整实现逻辑——指标定义是先转 JSON、再转 Struct发送的你按mapstring, any接收即可无需关心 Go 类型细节。⚙️ 一行配置把 pgwatch 指向你的服务器服务写好之后只需在 pgwatch 启动时加一个--sink参数可以重复使用与内置 Sink 并行写入。URI 格式见 docs/reference/sinks_options.md# 最简用法 --sinkgrpc://localhost:5000/ # 带认证 TLS --sinkgrpc://user:pwdlocalhost:5000/?sslrootca/home/user/ca.crt关键特性认证user:pwd会以 gRPC metadata 形式username/password两个字段转发给你的服务器服务器端自行校验不做任何强制TLS指定sslrootca参数加载 CA 证书后自动启用加密否则走明文连通性探测pgwatch 启动时会用一次SyncMetric空请求做 Ping只有服务端可用才认为 Sink 就绪Unavailable状态会立即报错退出代码侧的入口在internal/sinks/multiwriter.goSink 类型解析支持grpc和rpc两种写法都会路由到NewRPCWriter。️ 实现你的 gRPC 服务器4 步走第 1 步生成客户端代码拿 api/pb/pgwatch.proto 生成你所在语言的桩代码。以 Go 为例protoc --go_out. --go-grpc_out. api/pb/pgwatch.proto第 2 步实现Receiver服务以 Go 为例核心骨架示意func (s *MyServer) UpdateMeasurements(ctx context.Context, env *pb.MeasurementEnvelope) (*pb.Reply, error) { // 1. 从 ctx 的 metadata 中取出 username/password 做认证 // 2. 把 env.Data 中每行 Struct 转成 map写入你的时序库 // 3. 返回 pb.Reply{Logmsg: 写入 env.DBName} return pb.Reply{}, nil }第 3 步处理元数据同步SyncMetric里根据Operation枚举维护哪些实例在监控哪些指标DefineMetrics收到的是指标定义的 JSON 结构建议落盘或入库后续建表/建保留策略都能派上用场。第 4 步监听端口并返回 logmsg返回有意义的logmsg比如已写入 128 行到 tsdb它会被 pgwatch 原样打印到日志是验证数据链路是否打通的最快方式。 官方另有一个社区维护的 gRPC 服务器集合pgwatch-contrib/rpc见 docs/howto/implement_grpc_server.md针对常见存储方案提供了示例实现与搭建教程非常适合作为生产级服务的起点参考。 验证数据链路看 Web UI 是否健康配置生效后打开 pgwatch Web UI 确认 Reaper 与 Sink 均工作正常数据落库后你就可以在监控大盘中看到类似下面的指标网格与性能分析视图✅ 避坑清单与最佳实践场景建议批量写入UpdateMeasurements一次携带一整批Data务必批量插入你的时序库避免逐行写入拖慢 Reaper错误处理gRPC 错误会原样回传并记录日志写入失败请返回明确的codes状态码方便 pgwatch 侧定位认证若走内网可暂不认证跨网段/公网部署务必启用 TLSsslrootca并在服务端校验 metadata 中的凭据标签利用CustomTags是 pgwatch 实例维度的标签正好可映射为你时序库的 tag 维度做实例级筛选保留策略SyncMetric的DeleteOp提示某实例的某指标被移除是清理历史数据的天然钩子联调排障全程善用Reply.logmsg——pgwatch 日志里搜grpc关键字即可看到每批数据的行数与耗时小结pgwatch 的 gRPC Sink 把对接自有时序存储这件事压缩成了三件事读懂 api/pb/pgwatch.proto 中的 3 个 RPC、实现你的Receiver服务器、用一行--sinkgrpc://...完成接入。契约简单、协议标准无论你的存储是 InfluxDB、ClickHouse 还是自研方案一个下午就能让 PostgreSQL 监控指标流进自己的数据仓库。【免费下载链接】pgwatchpgwatch: PostgreSQL metrics monitor/dashboard项目地址: https://gitcode.com/gh_mirrors/pg/pgwatch创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考