如何为从Oracle gv$session轮询至Elasticsearch的SQL执行生成稳定文档ID

咱们先把你遇到的核心矛盾点拆明白，这样才能精准破局：

• Oracle会在SQL执行中途偷偷改sql_exec_id和sql_exec_start，之前靠这些字段做唯一ID的思路直接作废
• 轮询的快照经常乱序跑到Logstash，靠单调指标（比如elapsed_time）判断执行连续性的逻辑直接翻车
• 同一个会话里跑相同SQL的重叠执行，会把聚合逻辑搅得一团糟
• Logstash重启后，计数器类的ID直接重置，完全没了稳定性

经过验证的可行方案思路 #

方案1：会话+SQL的基础键+Redis状态跟踪（解决中途改ID和重叠执行） #

这个思路的核心是不一开始就依赖Oracle的不稳定执行标识，而是用会话+SQL的稳定组合做基础，再用Redis跟踪每个执行的全生命周期：

1. 基础键设计：用(sid, serial#, logon_time, sql_id)作为Redis中跟踪会话内SQL执行的基础前缀——这几个字段是绝对稳定的，不会中途变化。
2. 跟踪正在进行的执行：在Redis中给每个基础前缀维护一个哈希表，存储所有正在进行的SQL执行项，每个项包含：
   • 当前的sql_exec_id/sql_exec_start（即使中途变化也能实时更新）
   • 该执行的最大单调指标值（elapsed_time、buffer_gets、cpu_time）
   • 唯一执行标识（用Redis的全局持久化计数器生成，或者第一次捕获的微秒级时间戳+基础键哈希）
3. 乱序快照的处理逻辑：
   • 收到新快照时，先和Redis中该基础键下的所有执行项对比：如果某个执行项的所有单调指标都≤新快照的对应值（加小容差避免Oracle统计波动），就判定是同一执行的后续快照，直接更新指标和最新的Oracle执行标识。
   • 如果找不到匹配项，说明是新执行（包括Oracle中途改了sql_exec_id的情况），直接生成新的执行标识加入跟踪列表。
4. 执行完成的判定：如果某个执行项连续20秒（比5秒轮询间隔长3倍）没收到新快照，或者会话的SQL ID变了，就把这个执行项聚合为ES文档，文档ID用基础键+唯一执行标识，绝对稳定。

方案2：借助Oracle的ASH或SQL Monitor视图（如果环境允许） #

如果你的环境能访问gv$active_session_history（ASH）或者gv$sql_monitor，那问题会简单很多：

• gv$sql_monitor会为每个SQL执行生成不会中途变化的sql_exec_id，直接用这个字段做执行标识的一部分就行。
• ASH提供更细粒度的执行轨迹，能精准区分重叠执行的同一SQL，避免聚合混淆。

方案3：持久化全局计数器（解决Logstash重启丢ID的问题） #

如果必须用计数器生成唯一ID，一定要把计数器存在持久化的Redis里：

• Logstash启动时先从Redis读取当前计数器的最大值，后续每次生成新执行标识时原子自增并写回Redis。
• 重启后先扫描Redis中所有正在跟踪的执行项，把超时的直接聚合，未超时的继续跟踪，完全不影响ID稳定性。

优化后的Logstash Ruby代码片段 #

ruby {
  code => "
    require 'redis'
    require 'digest/md5'

    # 初始化Redis连接（务必开启Redis的RDB/AOF持久化！）
    redis = Redis.new(host: '127.0.0.1', port: 6379)
    # 稳定的基础键：不会中途变化
    base_key = [event.get('username'), event.get('sid'), event.get('serial#'), event.get('logon_time'), event.get('sql_id')].map(&:to_s).join(':')
    # 当前快照的指标和执行信息
    current_metrics = {
      elapsed_time: event.get('elapsed_time').to_i,
      buffer_gets: event.get('buffer_gets').to_i,
      cpu_time: event.get('cpu_time').to_i
    }
    current_exec = {
      sql_exec_id: event.get('sql_exec_id').to_s,
      sql_exec_start: event.get('sql_exec_start').to_s,
      last_seen: event.get('@timestamp').to_i,
      metrics: current_metrics
    }

    # 读取当前会话+SQL下的所有正在执行的任务
    ongoing_execs = redis.hgetall("#{base_key}:ongoing") || {}
    matched_exec_key = nil

    # 匹配同一执行：靠单调指标判断连续性，加小容差避免Oracle统计波动
    ongoing_execs.each do |exec_key, exec_data|
      parsed_data = JSON.parse(exec_data)
      is_continuous = current_metrics[:elapsed_time] >= (parsed_data['metrics']['elapsed_time'] - 1000) &&
                      current_metrics[:buffer_gets] >= (parsed_data['metrics']['buffer_gets'] - 100) &&
                      current_metrics[:cpu_time] >= (parsed_data['metrics']['cpu_time'] - 1000)
      if is_continuous
        matched_exec_key = exec_key
        break
      end
    end

    if matched_exec_key
      # 更新已匹配的执行项：取指标最大值，同步最新的Oracle执行标识
      updated_data = JSON.parse(ongoing_execs[matched_exec_key])
      updated_data['last_seen'] = current_exec[:last_seen]
      updated_data['sql_exec_id'] = current_exec[:sql_exec_id]
      updated_data['sql_exec_start'] = current_exec[:sql_exec_start]
      updated_data['metrics']['elapsed_time'] = [updated_data['metrics']['elapsed_time'], current_metrics[:elapsed_time]].max
      updated_data['metrics']['buffer_gets'] = [updated_data['metrics']['buffer_gets'], current_metrics[:buffer_gets]].max
      updated_data['metrics']['cpu_time'] = [updated_data['metrics']['cpu_time'], current_metrics[:cpu_time]].max
      redis.hset("#{base_key}:ongoing", matched_exec_key, JSON.generate(updated_data))
    else
      # 生成唯一执行标识：用全局持久化计数器，重启也不会乱
      global_counter = redis.incr('persistent_sql_exec_global_counter')
      exec_identifier = "exec_#{global_counter}"
      # 初始化新执行项，记录首次出现时间
      new_exec_data = current_exec.merge({
        first_seen: event.get('@timestamp').to_i
      })
      redis.hset("#{base_key}:ongoing", exec_identifier, JSON.generate(new_exec_data))
      # 设置哈希表超时，自动清理残留数据
      redis.expire("#{base_key}:ongoing", 30)
    end

    # 清理超时执行项，写入Elasticsearch
    ongoing_execs.each do |exec_key, exec_data|
      parsed_data = JSON.parse(exec_data)
      # 20秒没收到更新，判定为执行完成
      if (Time.now.to_i - parsed_data['last_seen']) > 20
        # 生成稳定的ES文档ID：基础键+执行标识的哈希
        es_doc_id = Digest::MD5.hexdigest("#{base_key}_#{exec_key}")
        # 构造最终的ES文档
        es_doc = {
          username: event.get('username'),
          sid: event.get('sid'),
          serial#: event.get('serial#'),
          logon_time: event.get('logon_time'),
          sql_id: event.get('sql_id'),
          sql_exec_id: parsed_data['sql_exec_id'],
          sql_exec_start: parsed_data['sql_exec_start'],
          first_seen: parsed_data['first_seen'],
          last_seen: parsed_data['last_seen'],
          metrics: parsed_data['metrics']
        }
        # 可以用ES输出插件发送，或者先丢到Redis队列批量处理
        redis.rpush('es_exec_docs', JSON.generate(es_doc))
        # 移除已完成的执行项
        redis.hdel("#{base_key}:ongoing", exec_key)
      end
    end

    event.cancel
  "
}

关键踩坑提醒 #

• Redis必须持久化：一定要开RDB或AOF，不然Logstash重启后，所有跟踪的执行状态和计数器都会丢，之前的努力全白费。
• 超时时间要合理：设置成轮询间隔的3-4倍（比如5秒轮询，超时20秒），避免因为网络延迟或Oracle统计慢误判执行完成。
• 指标加容差：Oracle的单调指标偶尔会有微小波动，给判断逻辑加个小容差（比如elapsed_time允许差1000以内），能避免把同一执行当成新执行。
• 重叠执行自动处理：Redis的哈希表会自动维护同一会话/同一SQL的多个执行项，每个项有独立标识，完全不会互相干扰。

内容来源于stack exchange