如何通过Elasticsearch将CloudWatch日志流式传输到Kibana？求免费方案指引

一、本地/自建ELK环境部署（核心免费方案） #

• 用Docker Compose快速部署开源免费版ELK（OSS版本，无付费限制）：
  编写docker-compose.yml配置文件：

  version: '3.8'
  services:
    elasticsearch:
      image: docker.elastic.co/elasticsearch/elasticsearch-oss:7.17.16
      environment:
        - discovery.type=single-node
        - ES_JAVA_OPTS=-Xms512m -Xmx512m
      ports:
        - "9200:9200"
        - "9300:9300"
      volumes:
        - esdata:/usr/share/elasticsearch/data
    logstash:
      image: docker.elastic.co/logstash/logstash-oss:7.17.16
      volumes:
        - ./logstash/config:/usr/share/logstash/config
        - ./logstash/pipeline:/usr/share/logstash/pipeline
      ports:
        - "5044:5044"
      depends_on:
        - elasticsearch
    kibana:
      image: docker.elastic.co/kibana/kibana-oss:7.17.16
      ports:
        - "5601:5601"
      depends_on:
        - elasticsearch
  volumes:
    esdata:
      driver: local
  

  执行启动命令：docker-compose up -d，验证服务状态：访问http://localhost:5601确认Kibana加载，http://localhost:9200查看Elasticsearch健康状态。

二、AWS侧日志转发配置（两种免费方式） #

方式1：CloudWatch + Lambda（利用AWS免费额度） #

• 创建Lambda函数（Python运行时），负责将CloudWatch日志转发到Logstash：

  import json
  import base64
  import socket
  
  def lambda_handler(event, context):
      logstash_host = "你的ELK服务器IP"
      logstash_port = 5044
      sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
      sock.connect((logstash_host, logstash_port))
      
      # 解码CloudWatch日志数据
      payload = json.loads(base64.b64decode(event['awslogs']['data']))
      for log_event in payload['logEvents']:
          log_entry = json.dumps(log_event) + '\n'
          sock.send(log_entry.encode('utf-8'))
      sock.close()
      return {'statusCode': 200}
  

• 给Lambda配置触发源：选择目标CloudWatch日志组，添加订阅过滤器。
• 配置IAM角色权限：给角色添加logs:DescribeLogGroups、logs:GetLogEvents权限，以及VPC访问权限（若ELK部署在AWS VPC内）。

方式2：Filebeat（轻量免费替代方案） #

• 在AWS免费额度内的EC2实例上部署Filebeat OSS版本，配置拉取CloudWatch日志：
  编辑filebeat.yml：

  filebeat.inputs:
  - type: aws_cloudwatch
    log_group_arns:
      - "arn:aws:logs:你的区域:你的账号ID:log-group:目标日志组名称:*"
    aws_access_key_id: "你的AWS Access Key"
    aws_secret_access_key: "你的AWS Secret Key"
    region: "你的AWS区域"
  
  output.logstash:
    hosts: ["你的Logstash地址:5044"]
  

  启动Filebeat：sudo systemctl start filebeat，它会自动拉取日志并发送到Logstash。

三、Logstash日志处理配置 #

• 在Logstash的pipeline目录创建cloudwatch.conf：

  input {
    tcp {
      port => 5044
      codec => json_lines
    }
  }
  
  filter {
    # 解析日志字段，按需调整
    json {
      source => "message"
    }
    # 同步时间戳到ELK标准字段
    date {
      match => ["timestamp", "ISO8601"]
      target => "@timestamp"
    }
  }
  
  output {
    elasticsearch {
      hosts => ["elasticsearch:9200"]
      index => "cloudwatch-logs-%{+YYYY.MM.dd}"
    }
  }
  

• 重启Logstash生效：docker-compose restart logstash

四、Kibana可视化配置 #

• 创建索引模式：进入Stack Management > Index Patterns，创建匹配cloudwatch-logs-*的模式，选择@timestamp作为时间字段。
• 构建仪表板：
  1. 在Discover中验证日志是否正确拉取、字段是否解析正常。
  2. 进入Dashboard，添加可视化组件（比如日志量趋势柱状图、错误日志统计表格、日志来源分布饼图等），按需自定义布局。

关键注意事项 #

• 网络连通性：确保AWS资源（Lambda/EC2）能访问ELK服务器，本地部署需开放5044端口，AWS VPC内需配置安全组和路由。
• 免费额度控制：AWS Lambda、EC2、CloudWatch均有免费额度，ELK OSS版本无费用，控制资源使用量可零成本运行。
• 日志过滤：在Filebeat或Logstash中添加过滤规则，只拉取必要日志，减少存储和带宽消耗。

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