如何编写Shell脚本读取日志新行并推送到Elasticsearch去重

嘿，针对你的两个问题，我来分享几个在日常运维中实测好用的方案，帮你解决实时日志处理和ES重复推送的问题👇

最常用的工具就是tail -f（或者tail -F，支持日志轮转场景），它会持续输出文件的新增内容，结合管道或循环就能实现自定义的读写逻辑。

基础用法：实时输出+备份新行 #

如果只是简单地把新行同时输出到终端和写入备份文件，用tee就能快速搞定：

# 监控/var/log/app.log的新行，同时输出到终端和追加到app_new_lines.log
tail -F /var/log/app.log | tee -a /var/log/app_new_lines.log

自定义处理逻辑：解析并加工新行 #

如果需要更复杂的操作（比如过滤关键词、格式化内容、写入数据库等），可以用while read循环来逐行处理：

tail -F /var/log/app.log | while read -r line; do
  # 跳过空行
  [[ -z "$line" ]] && continue

  # 示例1：只处理包含"ERROR"的行，写入错误日志并加上时间戳
  if echo "$line" | grep -q "ERROR"; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $line" >> /var/log/app_errors.log
  fi

  # 示例2：如果是JSON格式日志，格式化后写入另一个文件
  if echo "$line" | jq . >/dev/null 2>&1; then
    echo "$line" | jq '.processed_at = now' >> /var/log/formatted_app.log
  fi
done

核心思路是只处理从未推送过的新增行，我们可以通过记录已处理行的唯一标识（比如哈希值），或者追踪文件的读取位置来实现去重。

方案1：tail跟踪+哈希去重（简单易实现） #

这个方案用tail -F跟踪日志，通过SHA256哈希值标记已处理的行，避免重复推送：

#!/bin/bash

# 配置参数
LOG_FILE="/path/to/your/target.log"
ES_URL="http://your-es-server:9200/your-log-index/_doc"
# 存储已处理行哈希的临时文件
PROCESSED_HASHES="/tmp/processed_log_hashes.txt"

# 初始化文件
touch "$PROCESSED_HASHES"

# 实时跟踪日志新行（-F支持日志轮转自动切换）
tail -F "$LOG_FILE" --retry | while read -r line; do
  # 跳过空行
  [[ -z "$line" ]] && continue

  # 生成当前行的唯一哈希值
  line_hash=$(echo "$line" | sha256sum | awk '{print $1}')

  # 检查哈希是否已存在，不存在则推送ES
  if ! grep -q "$line_hash" "$PROCESSED_HASHES"; then
    # 推送至Elasticsearch，可根据你的日志格式调整JSON内容
    curl -X POST "$ES_URL" \
      -H "Content-Type: application/json" \
      -d "{
        \"@timestamp\": \"$(date -u +"%Y-%m-%dT%H:%M:%SZ")\",
        \"message\": \"$line\",
        \"source\": \"$(basename "$LOG_FILE")\"
      }"

    # 将哈希写入文件，标记为已处理
    echo "$line_hash" >> "$PROCESSED_HASHES"

    # 可选：定期清理哈希文件，避免占用过多磁盘（比如保留最近7天的记录）
    find "$PROCESSED_HASHES" -mtime +7 -exec truncate -s 0 {} \;
  fi
done

方案2：inotifywait监控文件修改（更精准） #

如果对实时性要求更高，或者需要处理特殊的日志更新场景，可以用inotifywait监控文件的修改事件，只读取新增的内容：

#!/bin/bash

LOG_FILE="/path/to/your/target.log"
ES_URL="http://your-es-server:9200/your-log-index/_doc"
PROCESSED_HASHES="/tmp/processed_log_hashes.txt"
# 记录上次读取的文件字节位置
LAST_POSITION="/tmp/log_last_position.txt"

# 初始化文件
touch "$PROCESSED_HASHES" "$LAST_POSITION"

# 首次运行时，记录当前文件大小作为起始位置
if [[ ! -s "$LAST_POSITION" ]]; then
  stat -c %s "$LOG_FILE" > "$LAST_POSITION"
fi

# 持续监控文件修改事件
while inotifywait -e modify "$LOG_FILE"; do
  current_size=$(stat -c %s "$LOG_FILE")
  last_size=$(cat "$LAST_POSITION")

  # 读取新增的内容（从上次位置到当前位置的字节）
  if [[ "$current_size" -gt "$last_size" ]]; then
    tail -c +$((last_size + 1)) "$LOG_FILE" | while read -r line; do
      [[ -z "$line" ]] && continue

      line_hash=$(echo "$line" | sha256sum | awk '{print $1}')
      if ! grep -q "$line_hash" "$PROCESSED_HASHES"; then
        # 推送ES
        curl -X POST "$ES_URL" \
          -H "Content-Type: application/json" \
          -d "{
            \"@timestamp\": \"$(date -u +"%Y-%m-%dT%H:%M:%SZ")\",
            \"message\": \"$line\"
          }"

        echo "$line_hash" >> "$PROCESSED_HASHES"
      fi
    done

    # 更新最后读取的位置
    echo "$current_size" > "$LAST_POSITION"
  fi
done

注意事项 #

• 日志轮转处理：tail -F会自动识别日志轮转后的新文件（比如logrotate生成的新文件），而inotifywait需要额外处理文件替换的情况（可以监控move_self事件）。
• 去重逻辑调整：如果你的日志行本身包含唯一ID（比如request_id），可以直接用这个ID代替哈希值，效率更高。
• 推送重试：如果ES偶尔不可用，可以给curl加--retry 3参数实现自动重试。
• 性能优化：如果日志量极大，while read循环可能会有瓶颈，可以考虑用awk来批量处理行，提升效率。

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