如何将自定义服务的CSV/JSON统计数据导入Grafana/Prometheus？

针对你的场景，最优最简的方案是改造现有Python抓取脚本，直接暴露Prometheus兼容的指标端点，无需额外引入重型组件，完全复用现有工作流，实现长期存储与时间维度查询。

方案1：改造Python脚本暴露Prometheus指标 #

你已经有Python脚本在定期抓取30台服务器的CSV数据，只需添加少量代码，即可将统计字段转换为Prometheus可识别的计数器指标，让Prometheus主动拉取并存储历史数据，最后通过Grafana查询可视化。

具体步骤： #

1. 安装依赖：引入prometheus-client库，这是官方的Python Prometheus指标工具

   pip install prometheus-client pandas
   

2. 改造脚本核心逻辑：

   • 定义带维度标签的Counter指标（对应你的service_name、user等维度）
   • 抓取CSV数据后，计算当前值与上次值的差值（因为原数据是累计计数器，Prometheus Counter只能递增）
   • 启动一个HTTP服务，暴露/metrics端点供Prometheus拉取

   示例代码片段：

   from prometheus_client import start_http_server, Counter
   import pandas as pd
   import time
   from typing import Dict, Tuple
   
   # 全局存储上次指标值，用于计算增量
   last_metrics: Dict[Tuple[str, str, str], int] = {}
   
   # 定义Prometheus Counter指标，包含维度标签
   service_stats = Counter(
       "service_stats",
       "自定义服务统计计数器",
       ["service_name", "user", "metric_name"]
   )
   
   def fetch_service_stats(server_url: str, service_name: str):
       """抓取单台服务器的CSV统计并更新Prometheus指标"""
       try:
           df = pd.read_csv(server_url)
           for _, row in df.iterrows():
               user = row["user"]
               # 定义需要监控的计数器字段（根据你的CSV表头调整）
               metric_map = {
                   "responses": row["responses"],
                   "auth_errors": row["auth_errors"],
                   "line_errors": row["line_errors"],
                   "encode_errors": row["encode_errors"],
                   "internal_error": row["internal_error"]
               }
   
               for metric_name, current_val in metric_map.items():
                   key = (service_name, user, metric_name)
                   last_val = last_metrics.get(key, 0)
                   # 仅当当前值大于上次值时更新（处理服务重启后计数器重置的情况）
                   if current_val > last_val:
                       service_stats.labels(
                           service_name=service_name,
                           user=user,
                           metric_name=metric_name
                       ).inc(current_val - last_val)
                   last_metrics[key] = current_val
       except Exception as e:
           print(f"抓取{service_name}数据失败: {str(e)}")
   
   if __name__ == "__main__":
       # 启动Prometheus指标服务，端口自定义（比如9091）
       start_http_server(9091)
       # 配置你的30台服务器信息（URL + 服务名）
       servers = [
           ("https://server1:port/my/stats/url/csv", "service-1"),
           ("https://server2:port/my/stats/url/csv", "service-2"),
           # ... 其他服务器配置
       ]
   
       while True:
           for server_url, service_name in servers:
               fetch_service_stats(server_url, service_name)
           time.sleep(1)  # 每秒抓取一次，与服务数据更新频率同步
   

3. 配置Prometheus：
   在Prometheus的prometheus.yml中添加抓取目标：

   scrape_configs:
     - job_name: "custom_service_stats"
       scrape_interval: 1s  # 与数据更新频率一致
       static_configs:
         - targets: ["your-script-host:9091"]  # 运行Python脚本的主机和端口
   

4. Grafana可视化：

   • 添加Prometheus数据源，指向你的Prometheus实例
   • 新建面板，使用PromQL查询指标，例如：
     • 查看某个服务的每秒响应数：rate(service_stats{service_name="service-1", metric_name="responses"}[1m])
     • 查看某个用户的认证错误趋势：rate(service_stats{user="myusername", metric_name="auth_errors"}[5m])

方案优势： #

• 完全复用现有Python抓取逻辑，开发量极小
• 无额外组件开销，仅需一个Python脚本
• 天然支持Prometheus的长期存储与时间维度查询，完美适配Grafana

备选方案：使用Telegraf采集（适合已有Telegraf环境） #

如果你的环境中已经部署了Telegraf，可以通过它的CSV输入插件和Prometheus输出插件实现数据采集：

• 配置Telegraf的inputs.csv插件，抓取所有服务器的CSV端点
• 配置outputs.prometheus_client插件，暴露指标端点供Prometheus拉取
• 后续步骤同方案1的Prometheus配置与Grafana可视化

此方案需要额外维护Telegraf配置，但适合已有监控体系的场景。

关键注意事项： #

• 必须计算计数器增量：原数据是服务重启后从0开始的累计值，直接写入Prometheus Counter会导致脚本重启后重复累计，因此必须记录上次值并计算差值
• 调整Prometheus抓取间隔：建议与服务数据更新频率（每秒）保持一致，避免数据遗漏
• 使用rate()函数计算速率：Grafana中查询计数器时，用rate()或irate()函数将累计值转换为每秒增长率，更适合趋势分析

内容的提问来源于stack exchange，提问作者Allen Austin