EKS/EC2 日志采集到 S3 与查询分析实战
EKS/EC2 日志采集到 S3 与查询分析实战
一套把容器和主机日志统一收到 S3、再按需查询的链路：fluent-bit 负责采集(EKS 用 DaemonSet,EC2 用 systemd 服务),日志落到 S3 后,要全文检索就用 Logstash 灌进 OpenSearch,临时排查就直接用 Athena 跑 SQL。下面的桶名、路径、endpoint 都换成了占位符,配置结构保持原样。
fluent-bit 采集 EKS Pod 日志到 S3
EKS 里用 DaemonSet 部署 fluent-bit,每个节点跑一个实例收集该节点上的容器日志。需要的 YAML 大致是 ServiceAccount + ClusterRole + ClusterRoleBinding + ConfigMap + DaemonSet 这一套。
ServiceAccount 通过 IRSA 绑定 IAM 角色(eks.amazonaws.com/role-arn 注解),这样 fluent-bit 写 S3 不必塞静态 AK/SK:
apiVersion: v1
kind: ServiceAccount
metadata:
name: fluent-bit
namespace: default
annotations:
eks.amazonaws.com/role-arn: aws_iam_role.eks_role.arn
ClusterRole 给它读 namespace/pod/log/node/event 的权限,再用 ClusterRoleBinding 绑到上面的 ServiceAccount:
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: fluent-bit-role
rules:
- apiGroups: [""]
resources: ["namespaces", "pods", "pods/log", "nodes", "events"]
verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: fluent-bit-role-binding
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: ClusterRole
name: fluent-bit-role
subjects:
- kind: ServiceAccount
name: fluent-bit
namespace: default
核心是 ConfigMap 里的 fluent-bit.conf。INPUT 用 tail 收 /var/log/containers/ 下指定服务的日志,OUTPUT 用 s3 插件按时间和 TAG 切分上传:
[SERVICE]
Flush 1
Log_Level info
Parsers_File parsers.conf
[INPUT]
Name tail
Parser docker
Path /var/log/containers/serviceA*.log,/var/log/containers/serviceB*.log # 只收这几个服务的日志
Tag <pod_name>
Tag_Regex (?<pod_name>[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)_(?<namespace_name>[^_]+)_(?<container_name>.+)-(?<container_id>[a-z0-9]{64})\.log$
[FILTER]
Name kubernetes
Match kube.*
Kube_Tag_Prefix kube.var.log.containers.
Regex_Parser custom-tag # 解析 TAG,使 INPUT 里能用 pod_name 命名 TAG
[OUTPUT]
Name s3
Match *
bucket my-pod-logs-bucket
region ap-southeast-1
total_file_size 5M # 单文件最大 5MB
upload_timeout 20m # 每 20 分钟触发一次上传
store_dir /buffers
store_dir_limit_size 2G # 缓存达到 2G 后不再接收,按实际调整
s3_key_format /logs/%Y-%m-%d/$TAG/%Y%m%d-%H-%M-%S-$UUID.log # S3 上的路径与文件名
json_date_key false
parsers.conf 配两个解析器,一个解 docker 的 JSON 时间戳,一个用正则从 TAG 里抠出 namespace/pod/container:
[PARSER]
Name docker
Format json
Time_Key time
Time_Format %Y-%m-%dT%H:%M:%S.%L
[PARSER]
Name custom-tag
Format regex
Regex ^(?<namespace_name>[^_]+)\.(?<pod_name>[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)\.(?<container_name>.+)\.(?<container_id>[a-z0-9]{64})
DaemonSet 把 /var/log 和上面的 ConfigMap 挂进容器。镜像用 fluent/fluent-bit:3.2.6-amd64,凭据从 Secret 注入(若已用 IRSA 可省掉这段 env):
apiVersion: apps/v1
kind: DaemonSet
metadata:
name: fluent-bit
namespace: default
spec:
selector:
matchLabels:
name: fluent-bit
template:
metadata:
labels:
name: fluent-bit
spec:
serviceAccountName: fluent-bit
containers:
- name: fluent-bit
image: fluent/fluent-bit:3.2.6-amd64
imagePullPolicy: IfNotPresent
env:
- name: AWS_ACCESS_KEY_ID
valueFrom:
secretKeyRef:
name: aws-ec2-access-secret
key: access_key
- name: AWS_SECRET_ACCESS_KEY
valueFrom:
secretKeyRef:
name: aws-ec2-access-secret
key: secret_key
volumeMounts:
- name: varlog
mountPath: /var/log
- name: config
mountPath: /fluent-bit/etc
volumes:
- name: varlog
hostPath:
path: /var/log
- name: config
configMap:
name: fluent-bit-config
dnsPolicy: ClusterFirst
restartPolicy: Always
整套用 kustomize 组织,kustomization.yaml 把这几个 resource 串起来一把 apply:
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
- cluster-role.yaml
- cluster-role-binding.yaml
- configmap.yaml
- serviceaccount.yaml
- daemonset.yaml
fluent-bit 采集 EC2(Amazon Linux 2)日志到 S3
非容器的 EC2 主机用安装脚本装 fluent-bit,然后配置文件 + systemd 服务跑起来。
安装(总是装最新发布版):
curl https://raw.githubusercontent.com/fluent/fluent-bit/master/install.sh | sh
/etc/fluent-bit/fluent-bit.conf 的关键是 OUTPUT 段,和 EKS 那套思路一样,只是 INPUT 换成主机侧来源(这里示例用 http,实际可换 tail/cpu 等):
[SERVICE]
flush 1
daemon Off
log_level info
parsers_file parsers.conf
plugins_file plugins.conf
storage.metrics on
[INPUT]
name http
listen 0.0.0.0
port 8080
[OUTPUT]
Name s3
Match *
bucket my-ec2-logs-bucket
region ap-southeast-1
total_file_size 5M
upload_timeout 5m
retry_limit 100
store_dir /buffers
store_dir_limit_size 3G
s3_key_format_tag_delimiters _
s3_key_format /logs/host-logs/$TAG[0]/%Y-%m-%d/$TAG-$UUID.log
json_date_key false
systemd 服务文件 /usr/lib/systemd/system/fluent-bit.service,注意把 AWS 凭据/配置路径用 Environment 指出来:
[Unit]
Description=Fluent Bit
Documentation=https://docs.fluentbit.io/manual/
Requires=network.target
After=network.target
[Service]
Type=simple
Environment="HOME=/root"
Environment="AWS_CONFIG_FILE=/root/.aws/config"
Environment="AWS_SHARED_CREDENTIALS_FILE=/root/.aws/credentials"
EnvironmentFile=-/etc/sysconfig/fluent-bit
EnvironmentFile=-/etc/default/fluent-bit
ExecStart=/opt/fluent-bit/bin/fluent-bit -c /etc/fluent-bit/fluent-bit.conf
Restart=always
[Install]
WantedBy=multi-user.target
之后 systemctl enable --now fluent-bit 即可。
Logstash 从 S3 拉日志写入 OpenSearch
要对落到 S3 的日志做全文检索,用 Logstash 做中转。要发 OpenSearch 2.x,Logstash 得装 8 以上版本。
装仓库和包:
sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
sudo tee /etc/yum.repos.d/logstash.repo <<EOF
[logstash-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF
sudo yum install logstash
sudo systemctl enable --now logstash
logstash.conf 三段:input 从 S3 指定前缀拉 JSON 日志;filter 用 grok 从 S3 路径里抠出 pod 名、再解析日志正文字段;output 写 OpenSearch:
input {
s3 {
access_key_id => "access_key_id"
secret_access_key => "secret_access_key"
region => "ap-southeast-1"
bucket => "my-pod-logs-bucket"
prefix => "logs/2025-03-04" # 只收这个目录下的日志
interval => 60 # 拉取间隔(秒)
type => "s3-logs"
codec => "json"
include_object_properties => true # 带上 S3 元信息,后面解析路径要用
}
}
filter {
mutate {
add_field => { "path" => "%{[@metadata][s3][key]}" }
}
grok {
match => {
"path" => "logs/(?<date>\d{4}-\d{2}-\d{2})/%{NOTSPACE:pod_name}/(?<raw>.+)" # 从 S3 路径里抠 pod 名
}
tag_on_failure => ["_s3_path_parse_failed"]
}
grok {
match => {
"log" => "%{TIMESTAMP_ISO8601:system_time} %{WORD:stream} %{WORD:flag} %{LOGLEVEL:log_level} %{TIMESTAMP_ISO8601:app_time}\s(?<raw>.+)"
}
}
}
output {
stdout { codec => rubydebug }
opensearch {
hosts => ["https://search-my-domain-xxxxxxxxxxxxxxxxxx.ap-southeast-1.es.amazonaws.com:443"]
index => "logstash-s3-logs"
user => "admin"
password => "<your-opensearch-password>"
ssl => true
ssl_certificate_verification => false
}
}
调 grok 表达式时 grokdebugger 很省事,内置 pattern 参照 logstash-patterns-core。OpenSearch 输出插件见 logstash-output-opensearch。
Athena 直接查 S3 上的日志
只是临时排查、不想搭 OpenSearch 时,Athena 直接对 S3 跑 SQL 更轻。
先给执行身份配好 IAM 权限,至少包含:AmazonAthenaFullAccess、AmazonS3FullAccess、AWSGlueConsoleFullAccess(用到 KMS 的话再加对应 KMS 策略)。在 Athena 控制台设好查询结果的输出位置。
建库建表,表的 LOCATION 指向日志所在的 S3 前缀,用 JSON SerDe 把每行当作一个 log 字段:
CREATE DATABASE IF NOT EXISTS logs_db;
CREATE EXTERNAL TABLE IF NOT EXISTS logs_db.logs (
log STRING
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
WITH SERDEPROPERTIES ('paths' = 'log')
LOCATION 's3://my-pod-logs-bucket/logs/2025-02-20/app-0';
然后就能像查表一样过滤日志,比如捞所有 ERROR:
SELECT *
FROM logs_db.logs
WHERE regexp_like(log, 'ERROR');