如何在Grafana中可视化S3存储的ELB日志？

我想在Grafana中可视化负载均衡器（ELB）收集并存储在S3中的日志，日志路径为：

test-bkp/AWSLogs/851725603941/elasticloadbalancing/us-east-2/year/month/date/*.gz

请问最佳实现方法是什么？可以使用哪些工具？能否通过Loki或InfluxDB实现？请给出建议并提供配置指导。

我曾尝试用Loki，但效果不好。我的环境如下：

• server1：已安装Grafana、Loki和Prometheus
• server2：运行由ELB托管的应用，日志已同步到S3，路径同上

我已经尝试的步骤：
在server1上安装了Promtail，配置文件promtail-local-config.yaml如下：

server:
  http_listen_port: 9080
  grpc_listen_port: 0

positions:
  filename: /data/loki/positions.yaml

clients:
  - url: http://x.x.x.x:3100/loki/api/v1/push

scrape_configs:
  - job_name: "local-s3-logs"
    static_configs:
      - targets: ['localhost']
        labels:
          job: "s3-logs"
          __path__: /var/log/s3-logs/**/*.log.gz

同步脚本sync-s3-logs.sh：

#!/bin/bash
# Sync S3 logs to local directory

BUCKET="s3://test-bkp/AWSLogs/851725603941/elasticloadbalancing/us-east-2/"
LOCAL_DIR="/var/log/s3-logs/"

# Create the directory if it doesn't exist
mkdir -p $LOCAL_DIR

# Sync the logs from S3 to the local directory
aws s3 sync $BUCKET $LOCAL_DIR

通过systemd运行同步脚本，配置文件：
/etc/systemd/system/s3-sync.service：

[Unit]
Description=S3 Sync Service for ELB Logs
Wants=network-online.target
After=network-online.target

[Service]
Type=simple
ExecStart=/opt/s3-sync/sync-s3-logs.sh
Restart=on-failure

[Install]
WantedBy=multi-user.target

/etc/systemd/system/s3-sync.timer：

[Unit]
Description=Run S3 Sync Service every 1 minute

[Timer]
OnBootSec=1min
OnUnitActiveSec=1min

[Install]
WantedBy=timers.target

一、可行工具选项 #

1. Loki（推荐优化现有方案） #

你的现有思路没问题，但需要调整配置解决效果不佳的问题。Loki适合日志场景，轻量化且和Grafana集成度高。

2. InfluxDB #

适合将日志解析为结构化指标后存储，比如ELB日志中的请求数、响应时间等，更适合做监控指标可视化，而非原始日志查询。

3. 其他工具 #

• AWS Athena + Grafana：直接查询S3中的日志，无需同步到本地，适合大规模日志场景，但依赖AWS服务。
• Fluentd/Fluent Bit：替代Promtail，直接从S3拉取日志并发送到Loki/InfluxDB，省去本地同步步骤。

二、Loki方案优化（针对你的现有配置） #

1. 修复Promtail配置问题 #

你的Promtail配置无法直接读取.gz压缩日志，需要添加解压处理器，同时优化路径匹配和标签：

server:
  http_listen_port: 9080
  grpc_listen_port: 0

positions:
  filename: /data/loki/positions.yaml

clients:
  - url: http://x.x.x.x:3100/loki/api/v1/push

scrape_configs:
  - job_name: "local-s3-logs"
    static_configs:
      - targets: ['localhost']
        labels:
          job: "s3-elb-logs"
          __path__: /var/log/s3-logs/**/*.gz
    # 添加解压处理器，处理gzip压缩文件
    pipeline_stages:
      - decompress:
          format: gzip
      # 解析ELB日志为结构化数据（可选，提升查询体验）
      - regex:
          expression: '^(?P<type>[^ ]*) (?P<timestamp>[^ ]*) (?P<elb>[^ ]*) (?P<client_ip>[^ ]*):(?P<client_port>[^ ]*) (?P<target_ip>[^ ]*):(?P<target_port>[^ ]*) (?P<request_processing_time>[^ ]*) (?P<target_processing_time>[^ ]*) (?P<response_processing_time>[^ ]*) (?P<elb_status_code>[^ ]*) (?P<target_status_code>[^ ]*) (?P<received_bytes>[^ ]*) (?P<sent_bytes>[^ ]*) "(?P<request_method>[^ ]*) (?P<request_url>[^ ]*) (?P<request_protocol>[^ ]*)" "(?P<user_agent>[^"]*)" (?P<ssl_cipher>[^ ]*) (?P<ssl_protocol>[^ ]*) (?P<target_group_arn>[^ ]*) "(?P<trace_id>[^"]*)" "(?P<domain_name>[^"]*)" "(?P<chosen_cert_arn>[^"]*)" (?P<matched_rule_priority>[^ ]*) (?P<request_creation_time>[^ ]*) "(?P<actions_executed>[^"]*)" "(?P<redirect_url>[^"]*)" "(?P<error_reason>[^"]*)" (?P<target_port_list>[^ ]*) (?P<target_status_code_list>[^ ]*) (?P<classification>[^ ]*) (?P<classification_reason>[^ ]*)'
      - labels:
          elb:
          client_ip:
          elb_status_code:
      - timestamp:
          source: timestamp
          format: "2006-01-02T15:04:05.000Z"

2. 优化S3同步脚本 #

避免重复处理已同步的日志，添加--exclude参数跳过已解压文件，或者直接在同步后解压：

#!/bin/bash
BUCKET="s3://test-bkp/AWSLogs/851725603941/elasticloadbalancing/us-east-2/"
LOCAL_DIR="/var/log/s3-logs/"

mkdir -p $LOCAL_DIR

# 只同步新的.gz文件
aws s3 sync $BUCKET $LOCAL_DIR --exclude "*" --include "*.gz"

# 解压所有未解压的.gz文件
find $LOCAL_DIR -name "*.gz" -exec gunzip {} \;

注意：如果选择解压后再收集，需要修改Promtail的__path__为/var/log/s3-logs/**/*.log，并去掉decompress阶段。

3. 调整systemd定时器 #

ELB日志通常按小时生成，没必要每分钟同步，改为每小时同步一次更合理：

[Timer]
OnBootSec=10min
OnUnitActiveSec=1h

三、InfluxDB方案实现 #

1. 日志解析与导入 #

使用telegraf将ELB日志解析为指标导入InfluxDB，配置telegraf.conf：

[[inputs.tail]]
  files = ["/var/log/s3-logs/**/*.log"]
  from_beginning = false
  pipe = false
  tag_keys = ["elb"]
  data_format = "regex"
  regex = '^(?P<type>[^ ]*) (?P<timestamp>[^ ]*) (?P<elb>[^ ]*) (?P<client_ip>[^ ]*):(?P<client_port>[^ ]*) (?P<target_ip>[^ ]*):(?P<target_port>[^ ]*) (?P<request_processing_time>[^ ]*) (?P<target_processing_time>[^ ]*) (?P<response_processing_time>[^ ]*) (?P<elb_status_code>[^ ]*) (?P<target_status_code>[^ ]*) (?P<received_bytes>[^ ]*) (?P<sent_bytes>[^ ]*) "(?P<request_method>[^ ]*) (?P<request_url>[^ ]*) (?P<request_protocol>[^ ]*)" "(?P<user_agent>[^"]*)" (?P<ssl_cipher>[^ ]*) (?P<ssl_protocol>[^ ]*) (?P<target_group_arn>[^ ]*) "(?P<trace_id>[^"]*)" "(?P<domain_name>[^"]*)" "(?P<chosen_cert_arn>[^"]*)" (?P<matched_rule_priority>[^ ]*) (?P<request_creation_time>[^ ]*) "(?P<actions_executed>[^"]*)" "(?P<redirect_url>[^"]*)" "(?P<error_reason>[^"]*)" (?P<target_port_list>[^ ]*) (?P<target_status_code_list>[^ ]*) (?P<classification>[^ ]*) (?P<classification_reason>[^ ]*)'
  timestamp_format = "2006-01-02T15:04:05.000Z"
  timestamp_field = "timestamp"

[[outputs.influxdb_v2]]
  urls = ["http://<influxdb-ip>:8086"]
  token = "<your-token>"
  organization = "<your-org>"
  bucket = "elb-logs"

2. Grafana可视化 #

在Grafana中添加InfluxDB数据源，然后创建面板查询指标，比如统计不同状态码的请求数、平均响应时间等。

四、AWS Athena + Grafana方案（无需本地同步） #

1. 在AWS中创建Athena表，指向S3中的ELB日志路径。
2. 在Grafana中添加Athena数据源。
3. 直接通过SQL查询S3中的日志，生成可视化面板。
   优点：无需本地存储和同步，适合大规模日志；缺点：依赖AWS服务，查询延迟较高。

内容的提问来源于stack exchange，提问作者sarjan kumar