Fix for ad_server time bomb.

time_bomb_diffuser.rb

#!/usr/bin/env ruby
require File.dirname(__FILE__) + "/../ad_server_dependencies.rb"

class StatKeyFix

  # This fixes ad stats that were incorrectly keyed in redis
  # according to ContentCacheAdvertisement#id instead of the
  # expected ContentInstance#id. The issue only affected ads
  # with a ContentCacheAdvertisement.id greater than 100,000

  def initialize(params)
    @update = params[:update] || false
    @limit  = params[:limit]  || 1
    @ccas   = params[:ccas]   || nil

    @redis  = Redis.new(redis_config(ENV['RACK_ENV']))
    minute = Time.now.strftime("%Y%m%d%H%m")
    @log = Logger.new(
      "#{AD_SERVER_ROOT}/log/time_bomb_fix_#{minute}.log")
  end

  def run
    where_condition = @ccas ?  "and id in (#{@ccas.join(',')})" : ""
    ccas = ContentCacheAdvertisement.connection.select_all(
      "select * from content_cache_advertisements where id >= 100000 #{where_condition} order by id limit #{@limit}")
    log "Found #{ccas.length} ContentCacheAdvertisements. Looping."

    ccas.each do |cca|

      cca_id              = cca['id']
      site_id             = cca['site_id']
      content_instance_id = cca['content_instance_id']
      log "Processing CCA #{cca_id}"

      wild_keys = generate_wild_keys(cca_id, site_id)

      process_keys(wild_keys, cca_id, content_instance_id)

    end

  end

  def generate_wild_keys(cca_id, site_id)
    wild_keys = []
    ['starts', 'views', 'clicks'].each do |stat_type|
      wild_keys += [
        "2011:09:??:??:#{cca_id}:#{site_id}:#{stat_type}",
        "2011:09:??:#{cca_id}:#{site_id}:#{stat_type}",
        "2011:week??:#{cca_id}:#{site_id}:#{stat_type}", 
        "2011:09:#{cca_id}:#{site_id}:#{stat_type}",
        "2011:#{cca_id}:#{site_id}:#{stat_type}",
        "total:#{cca_id}:#{site_id}:#{stat_type}"]
    end
    wild_keys
  end

  def process_keys(wild_keys, cca_id, content_instance_id)
    wild_keys.each do |wild_key|
      matching_keys = @redis.keys(wild_key)
      log "Found #{matching_keys.length} keys matching #{wild_key}"
      matching_keys.each do |key|
        real_key = key.gsub(cca_id, content_instance_id)
        update_key(key, real_key)
      end
    end
  end

  def update_key(key, real_key)
    total = @redis.get(key)
    log "Starting update of #{real_key}"
    begin
      if @update
        @redis.incrby(real_key, total.to_i) 
        log "Incremented #{real_key} by #{total.to_i}"
        mark_key_as_processed(key)
      else
        log "Dry run so not updating, would increment #{real_key} by #{total.to_i}"
      end
    rescue Exception => e
      log e
      log e.backtrace.join("\n")
    end
  end

  def mark_key_as_processed(key)
    new_key = key.strip + ':processed'
    if @redis.renamenx(key, new_key) 
      log "Source key successfully marked as processed."
    else
      log "Source key could not be marked as proccessed."
    end
  end

  def log(message)
    @log.info message
    puts message
  end

end