DuplicateRequestFilter.scala

package com.tumblr.fibr.service.filter

import com.tumblr.fibr.config.FilterConfig
import com.tumblr.fibr.tsdb.{TsdbRequest, TsdbResponse}
import com.google.common.cache.{Cache, CacheBuilder}
import com.twitter.finagle.Service
import com.twitter.util.Future
import java.util.concurrent.{Callable, TimeUnit}

/**
 * If we see the same request within timeoutInSeconds, return the same response
 */
class DuplicateRequestFilter(override val filterConfig: FilterConfig,
                             timeoutInSeconds: Int = 300)
  extends ServiceFilter
{
  type TsdbService = Service[TsdbRequest, TsdbResponse]
  val NotFoundInCache = mkStatName("query", "cache", "miss")
  val FoundInCache = mkStatName("query", "cache", "hit")
  val RequestedFromCache = mkStatName("query", "cache", "request")

  private[this] val cache: Cache[TsdbRequest,Future[TsdbResponse]] = CacheBuilder.newBuilder()
      .maximumSize(500)
      .expireAfterAccess(timeoutInSeconds, TimeUnit.SECONDS)
      .build()

  override def apply(request: TsdbRequest, service: TsdbService): Future[TsdbResponse] = {
    val wasPresent = cache.getIfPresent(request)
    cache.get(request, new Callable[Future[TsdbResponse]] {
      override def call: Future[TsdbResponse] = {
        service(request)
      }
    }).ensure {
      wasPresent match {
        case null =>
          Stats.incr(NotFoundInCache)
        case o =>
          Stats.incr(FoundInCache)
          log.info("Found query in cache: %s".format(request.toString))
      }
      Stats.incr(RequestedFromCache)
    }
  }

  val EvictedFromCache = mkStatName("query", "cache", "evict")
  Stats.addGauge(EvictedFromCache) { cache.stats.evictionCount }
}