Samples for Finagle Client Configuration

ResilientClientSamples.scala

/**
  * This is not meant to cover ALL possible ways of creating service clients; instead it focuses on the simplest way to
  * do so, while maintaining the capability to customize the clients based on service level agreements/expectations
  *
  * @note Most of the filters can be applied to the HTTP client as well but have been omitted from the sample code to improve
  * readability
  */

trait ClientSamples extends LazyLogging {
  private[this] lazy val config = ConfigFactory.load()
  private val useHttp = config.getBoolean("sample-service.use-http")

  /**
    * This is the simplest client that can be created supporting both HTTP and Thrift transports
    *
    * It is strongly recommended NOT to construct clients this way; without specifying basic/sane parameters for
    * timeouts, retry behavior, etc...
    */
  object SimpleClient {
    def apply(dest: String): SampleService[Future] = {
      useHttp match {
        case true =>
          val http = Http.client.newService(dest)
          val service = HttpClientFilter.withoutUpgrade(dest) andThen http
          new SampleService.FinagledClient(service)

        case _ =>
          Thrift.client.newIface[SampleService[Future]](dest)
      }
    }
  }

  /**
    * This client relies on the default methods that are provided by the `Thrift.client` object to set basic/sane
    * client stack parameters
    */
  object SampleClientWithBasicConfig {
    def apply(dest: String): SampleService[Future] = {
      useHttp match {
        case true =>
          val http = Http.client.newService(dest)
          val service = HttpClientFilter.withoutUpgrade(dest) andThen http
          new SampleService.FinagledClient(service)

        case _ =>
          Thrift.client

            /**
              * Define timeout requirement for this client.
              * @note The timeout is applied after a connection is acquired
              */
            .withRequestTimeout(200.milliseconds)

            /**
              * Define a requeue backoff policy for this client. By default, a `RequeueFilter` is configured in each client
              * stack and its default behavior is to have no delays, which is not recommended
              *
              * Failures that are known to be safe to retry (for example, exceptions that occurred before the bytes
              * were written to the wire) will be automatically retried by Finagle
              *
              * @see the extractor in [[com.twitter.finagle.service.RetryPolicy.RetryableWriteException]] for more details
              *      on which failures are considered retryable by the `RequeueFilter`
              * @see [[https://www.awsarchitectureblog.com/2015/03/backoff.html]] for a great explanation of how jittered
              *      backoffs work
              */
            .withRetryBackoff(Backoff.decorrelatedJittered(2.seconds, 32.seconds))

            /**
              * Define a retry budget for this client
              *
              * @see [[https://finagle.github.io/blog/2016/02/08/retry-budgets]] for an explanation of how retry budgets work
              */
            .withRetryBudget(RetryBudget(ttl = 5.seconds, minRetriesPerSec = 5, percentCanRetry = 0.1))

            /**
              * Configure Expiration module by defining a max life time and idle time for the client. By default, a client
              * session never expires. Configuring these parameters allows setting a maximum lifetime for the session and
              * a maximum idle time, where no requests are being sent from that client. Also, a connection acquisition 
              * timeout can be set (unbounded by default). Once any of these thresholds are met, the client is expired
              */
            .withSession.maxLifeTime(20.seconds)
            .withSession.maxIdleTime(10.seconds)
            .withSession.acquisitionTimeout(10.seconds)

            /**
              * The Fail Fast module is enabled by default for all clients except `Memcached`. Its function is to reduce
              * the number of requests dispatched to endpoints that are likely to fail. When a connection fails, the host
              * is marked down and a background process is launched to attempt reestablishing the connection (using a backoff
              * policy). Until the connection is reestablished, the load balancer will avoid using this host.
              *
              * @note In this example, the Fail Fast module is disabled. This is important in the case that only one host
              *       exists
              */
            .withSessionQualifier.noFailFast

            /**
              * The Failure Accrual module is enabled by default for all clients with a policy based on 5 consecutive failures
              * and an equal jittered backoff policy. This can be overriden as shown below.
              *
              * @note Failure Accrual can also be configured based on success rate as shown in this example:
              *
              * @example {{{
              *             .configured(FailureAccrualFactory.Param(() => FailureAccrualPolicy.successRate(
              * requiredSuccessRate = 0.99,
              * window = 100,
              * markDeadFor = Backoff.decorrelatedJittered(5.seconds, 20.seconds))))
              *          }}}
              */
            .configured(FailureAccrualFactory.Param(() =>
              FailureAccrualPolicy.consecutiveFailures(
                numFailures = 10,
                markDeadFor = Backoff.decorrelatedJittered(5.seconds, 20.seconds))))

            /**
              * Configure connection pooling. By default, the client stack is configured to have an overlay of caching and
              * watermark pools that have a low watermark of 0, an unbounded high watermark, and an unbounded TTL. Some of
              * these defaults can be overriden as shown below.
              *
              * @see [[https://twitter.github.io/finagle/guide/Clients.html#pooling]] for more information
              * @see `SampleClientWithAdvancedConfig` for a more advanced configuration
              */
            .withSessionPool.minSize(5)
            .withSessionPool.maxSize(10)
            .withSessionPool.maxWaiters(20) //Max number of connection requests to queue when exceeding high watermark

            /**
              * Configure response classifier. Using a response classifier can be very useful as it allows Finagle to gain
              * some understanding of your domain in order to properly classify failures, which leads to better failure
              * accrual handling and better metrics collection
              *
              * By default, Finagle will classify any `Return` type as a success and any `Failure` as a failure. The code
              * below changes this default behavior; it instructs Finagle to classify any deserialized Thrift exception as
              * a failure.
              *
              */
            .withResponseClassifier(ThriftResponseClassifier.ThriftExceptionsAsFailures)

            /**
              * Configure load balancer. By default, Finagle clients are configured to use a load balancer with a P2C algorithm
              * to distribute the load, picking the least loaded one. This can be substituted with several other setup including:
              * 1. Heap + Least Loaded
              * 2. Power of Two Choices (P2C) + Least Loaded (Default configuration)
              * 3. Power of Two Choices (P2C) + Peak EWMA
              * 4. Aperture + Least Loaded
              *
              * In this example, the Power of Two Choices (P2C) + Peak EWMA setup is used
              *
              * @see [[https://twitter.github.io/finagle/guide/Clients.html#load-balancing]] for more details
              * @see [[http://vkostyukov.net/posts/finagle-101]] for a good explanation of how the different setups work
              * @see [[https://blog.buoyant.io/2016/03/16/beyond-round-robin-load-balancing-for-latency/]] for a comparison
              *     of the different setups
              */
            .withLoadBalancer(Balancers.p2cPeakEwma(maxEffort = 100, decayTime = 100.seconds))

            .withStatsReceiver(TalonStatsReceiver())

            .newIface[SampleService[Future]](dest)
      }
    }
  }

  /**
    * This client shows a more advanced configuration than the previous two. For example, it shows how filters can be added
    * to the client stack during its creation.
    */
  object SampleClientWithAdvancedConfig {
    def apply(dest: String): SampleService[Future] = {
      useHttp match {
        case true =>
          val http = Http.client.newService(dest)
          val service = HttpClientFilter.withoutUpgrade(dest) andThen http
          new SampleService.FinagledClient(service)

        case _ =>
          val retryBudget = RetryBudget(ttl = 5.seconds, minRetriesPerSec = 5, percentCanRetry = 0.1)
          Thrift.client

          /**
            * Configure Expiration module by defining a max life time and idle time for the client. By default, a client
            * session never expires. Configuring these parameters allows setting a maximum lifetime for the session and
            * a maximum idle time, where no requests are being sent from that client. Also, a connection acquisition 
            * timeout can be set (unbounded by default). Once any of these thresholds are met, the client is expired
            */
          .withSession.maxLifeTime(20.seconds)
          .withSession.maxIdleTime(10.seconds)
          .withSession.acquisitionTimeout(10.seconds)

            /**
              * Add a `TimeoutFilter` to the stack with parameters that cannot be applied when using `withRequestTimeout`
              */
            .filtered(new TimeoutFilter(
              200.milliseconds,
              new IndividualRequestTimeoutException(200.milliseconds),
              HighResTimer.Default,
              TalonStatsReceiver()))

            /**
              * Add a `RetryFilter` to the stack. A `RetryPolicy` is used to configure the filter, taking in the number of
              * attempts to retry and a `PartialFunction` to determine if the request should be retried.
              * Alternatively, a `RetryPolicy` can be created using a backoff policy
              *
              * @note Generally, it is a good practice to leave the `RequeueFilter` to handle Finagle-safe failures and have
              *       the `RetryFilter` handle domain-specific and custom failures
              *
              * @note It is recommended to have a shared `RetryBudget` between the `RequeueFilter` and the `RetryFilter`
              *       to prevent retry storms.
              *
              * @see the extractor in [[com.twitter.finagle.service.RetryPolicy.RetryableWriteException]] for more details
              *      on which failures are considered retryable by the `RequeueFilter`
              * @see [[https://groups.google.com/forum/#!topic/finaglers/-MvcHJTxgjQ]] for an explanation of the differences
              *     between a `RetryFilter` and a `RequeueFilter`
              * @see [[https://twitter.github.io/finagle/guide/Clients.html#retries]] for more details on how retries work
              */
            .filtered(new RetryFilter(
              retryPolicy,
              HighResTimer.Default,
              TalonStatsReceiver(),
              retryBudget))

            .withRetryBudget(retryBudget)

            /**
              * The Fail Fast module is enabled by default for all clients except `Memcached`. Its function is to reduce
              * the number of requests dispatched to endpoints that are likely to fail. When a connection fails, the host
              * is marked down and a background process is launched to attempt reestablishing the connection (using a backoff
              * policy). Until the connection is reestablished, the load balancer will avoid using this host.
              *
              * @note In this example, the Fail Fast module is disabled. This is important in the case that only one host
              *       exists
              */
            .withSessionQualifier.noFailFast

            /**
              * The Failure Accrual module is enabled by default for all clients with a policy based on 5 consecutive failures
              * and an equal jittered backoff policy. This can be overriden as shown below.
              *
              * @note Failure Accrual can also be configured based on success rate as shown in this example:
              *
              * @example {{{
              *             .configured(FailureAccrualFactory.Param(() => FailureAccrualPolicy.successRate(
              * requiredSuccessRate = 0.99,
              * window = 100,
              * markDeadFor = Backoff.decorrelatedJittered(5.seconds, 20.seconds))))
              *          }}}
              */
            .configured(FailureAccrualFactory.Param(() =>
              FailureAccrualPolicy.consecutiveFailures(
                numFailures = 10,
                markDeadFor = Backoff.decorrelatedJittered(5.seconds, 20.seconds))))


            /**
              * A more advanced configuration for client pooling. In addition to overriding the `low`, `high`, and `maxWaiters`
              * parameters (which is possible with the `.withSessionPool` function), configuring pooling using the method
              * below allows overriding the `bufferSize` and `idleTime` for the connections as well.
              *
              */
            .configured(DefaultPool.Param(
              low = 5,
              high = 20,
              bufferSize = 0,
              idleTime = 60.milliseconds,
              maxWaiters = 20))

            /**
              * A more fine-tuned method can be used to configure response classifiers than simply classifying all Thrift
              * exceptions as failures as seen in [SampleClientWithBasicConfig]
              *
              * This can be accomplished by creating a custom `ResponseClassifier`. In the example below, the
              * `domainExceptionClassifier` classifies `DomainException`s as failures.
              *
              * @note At the time of writing, Finagle does not distinguish between `RetryableFailure`s and `NonRetryableFailure`s.
              *       They are added as groundwork for future enhancements
              *
              * @see [[https://twitter.github.io/finagle/guide/Clients.html#response-classification]] for more details on
              *     response classification
              */
            .withResponseClassifier(domainExceptionClassifier)

            /**
              * Configure load balancer. By default, Finagle clients are configured to use a load balancer with a P2C algorithm
              * to distribute the load, picking the least loaded one. This can be substituted with several other setup including:
              * 1. Heap + Least Loaded
              * 2. Power of Two Choices (P2C) + Least Loaded (Default configuration)
              * 3. Power of Two Choices (P2C) + Peak EWMA
              * 4. Aperture + Least Loaded
              *
              * In this example, Aperture + Least Loaded setup is used
              *
              * @see [[https://twitter.github.io/finagle/guide/Clients.html#load-balancing]] for more details
              * @see [[http://vkostyukov.net/posts/finagle-101]] for a good explanation of how the different setups work
              * @see [[https://blog.buoyant.io/2016/03/16/beyond-round-robin-load-balancing-for-latency/]] for a comparison
              *     of the different setups
              */
            .withLoadBalancer(Balancers.aperture(
              lowLoad = 1.0, highLoad = 2.0, // the load band adjusting an aperture
              minAperture = 10 // min aperture size
            ))

            .withStatsReceiver(TalonStatsReceiver())

            .newIface[SampleService[Future]](dest)
      }
    }

    private def retryPolicy[Req, Rep]: RetryPolicy[(Req, Try[Rep])] = RetryPolicy.tries(3, {
      case (_, Throw(Failure(Some(_: TimeoutException)))) | (_, Throw(_: TimeoutException)) =>
        logger.debug("Request timed out. Retrying")
        true
      case _ => false
    })

    private val domainExceptionClassifier: ResponseClassifier = {
      case ReqRep(_, Throw(_: DomainException)) =>
        ResponseClass.NonRetryableFailure
    }
  }
}