aurora-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From zma...@apache.org
Subject [18/51] [partial] aurora git commit: Move packages from com.twitter.common to org.apache.aurora.common
Date Wed, 26 Aug 2015 21:00:08 GMT
http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/thrift/callers/RetryingCaller.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/thrift/callers/RetryingCaller.java b/commons/src/main/java/org/apache/aurora/common/thrift/callers/RetryingCaller.java
new file mode 100644
index 0000000..84a0f4e
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/thrift/callers/RetryingCaller.java
@@ -0,0 +1,224 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.thrift.callers;
+
+import java.lang.reflect.Method;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Logger;
+
+import javax.annotation.Nullable;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Predicate;
+import com.google.common.base.Throwables;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+
+import org.apache.aurora.common.thrift.TResourceExhaustedException;
+import org.apache.thrift.async.AsyncMethodCallback;
+
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+import org.apache.aurora.common.stats.StatsProvider;
+
+/**
+* A caller that will retry calls to the wrapped caller.
+*
+* @author William Farner
+*/
+public class RetryingCaller extends CallerDecorator {
+  private static final Logger LOG = Logger.getLogger(RetryingCaller.class.getName());
+
+  @VisibleForTesting
+  public static final Amount<Long, Time> NONBLOCKING_TIMEOUT = Amount.of(-1L, Time.MILLISECONDS);
+
+  private final StatsProvider statsProvider;
+  private final String serviceName;
+  private final int retries;
+  private final ImmutableSet<Class<? extends Exception>> retryableExceptions;
+  private final boolean debug;
+
+  /**
+   * Creates a new retrying caller. The retrying caller will attempt to call invoked methods on the
+   * underlying caller at most {@code retries} times.  A retry will be performed only when one of
+   * the {@code retryableExceptions} is caught.
+   *
+   * @param decoratedCall The caller to decorate with retries.
+   * @param async Whether the caller is asynchronous.
+   * @param statsProvider The stat provider to export retry statistics through.
+   * @param serviceName The service name that calls are being invoked on.
+   * @param retries The maximum number of retries to perform.
+   * @param retryableExceptions The exceptions that can be retried.
+   * @param debug Whether to include debugging information when retries are being performed.
+   */
+  public RetryingCaller(Caller decoratedCall, boolean async, StatsProvider statsProvider,
+      String serviceName, int retries, ImmutableSet<Class<? extends Exception>> retryableExceptions,
+      boolean debug) {
+    super(decoratedCall, async);
+    this.statsProvider = statsProvider;
+    this.serviceName = serviceName;
+    this.retries = retries;
+    this.retryableExceptions = retryableExceptions;
+    this.debug = debug;
+  }
+
+  private final LoadingCache<Method, AtomicLong> stats =
+      CacheBuilder.newBuilder().build(new CacheLoader<Method, AtomicLong>() {
+        @Override public AtomicLong load(Method method) {
+          // Thrift does not support overloads - so just the name disambiguates all calls.
+          return statsProvider.makeCounter(serviceName + "_" + method.getName() + "_retries");
+        }
+      });
+
+  @Override public Object call(final Method method, final Object[] args,
+      @Nullable final AsyncMethodCallback callback,
+      @Nullable final Amount<Long, Time> connectTimeoutOverride) throws Throwable {
+    final AtomicLong retryCounter = stats.get(method);
+    final AtomicInteger attempts = new AtomicInteger();
+    final List<Throwable> exceptions = Lists.newArrayList();
+
+    final ResultCapture capture = new ResultCapture() {
+      @Override public void success() {
+        // No-op.
+      }
+
+      @Override public boolean fail(Throwable t) {
+        if (!isRetryable(t)) {
+          if (debug) {
+            LOG.warning(String.format(
+                "Call failed with un-retryable exception of [%s]: %s, previous exceptions: %s",
+                t.getClass().getName(), t.getMessage(), combineStackTraces(exceptions)));
+          }
+
+          return true;
+        } else if (attempts.get() >= retries) {
+          exceptions.add(t);
+
+          if (debug) {
+            LOG.warning(String.format("Retried %d times, last error: %s, exceptions: %s",
+                attempts.get(), t, combineStackTraces(exceptions)));
+          }
+
+          return true;
+        } else {
+          exceptions.add(t);
+
+          if (isAsync() && attempts.incrementAndGet() <= retries) {
+            try {
+              retryCounter.incrementAndGet();
+              // override connect timeout in ThriftCaller to prevent blocking for a connection
+              // for async retries (since this is within the callback in the selector thread)
+              invoke(method, args, callback, this, NONBLOCKING_TIMEOUT);
+            } catch (Throwable throwable) {
+              return fail(throwable);
+            }
+          }
+
+          return false;
+        }
+      }
+    };
+
+    boolean continueLoop;
+    do {
+      try {
+        // If this is an async call, the looping will be handled within the capture.
+        return invoke(method, args, callback, capture, connectTimeoutOverride);
+      } catch (Throwable t) {
+        if (!isRetryable(t)) {
+          Throwable propagated = t;
+
+          if (!exceptions.isEmpty() && (t instanceof TResourceExhaustedException)) {
+            // If we've been trucking along through retries that have had remote call failures
+            // and we suddenly can't immediately get a connection on the next retry, throw the
+            // previous remote call failure - the idea here is that the remote call failure is
+            // more interesting than a transient inability to get an immediate connection.
+            propagated = exceptions.remove(exceptions.size() - 1);
+          }
+
+          if (isAsync()) {
+            callback.onError((Exception) propagated);
+          } else {
+            throw propagated;
+          }
+        }
+      }
+
+      continueLoop = !isAsync() && attempts.incrementAndGet() <= retries;
+      if (continueLoop) retryCounter.incrementAndGet();
+    } while (continueLoop);
+
+    Throwable lastRetriedException = Iterables.getLast(exceptions);
+    if (debug) {
+      if (!exceptions.isEmpty()) {
+        LOG.warning(
+            String.format("Retried %d times, last error: %s, previous exceptions: %s",
+                attempts.get(), lastRetriedException, combineStackTraces(exceptions)));
+      } else {
+        LOG.warning(
+            String.format("Retried 1 time, last error: %s", lastRetriedException));
+      }
+    }
+
+    if (!isAsync()) throw lastRetriedException;
+    return null;
+  }
+
+  private boolean isRetryable(Throwable throwable) {
+    return isRetryable.getUnchecked(throwable.getClass());
+  }
+
+  private final LoadingCache<Class<? extends Throwable>, Boolean> isRetryable =
+      CacheBuilder.newBuilder().build(new CacheLoader<Class<? extends Throwable>, Boolean>() {
+        @Override public Boolean load(Class<? extends Throwable> exceptionClass) {
+          return isRetryable(exceptionClass);
+        }
+      });
+
+  private boolean isRetryable(final Class<? extends Throwable> exceptionClass) {
+    if (retryableExceptions.contains(exceptionClass)) {
+      return true;
+    }
+    return Iterables.any(retryableExceptions, new Predicate<Class<? extends Exception>>() {
+      @Override public boolean apply(Class<? extends Exception> retryableExceptionClass) {
+        return retryableExceptionClass.isAssignableFrom(exceptionClass);
+      }
+    });
+  }
+
+  private static final Joiner STACK_TRACE_JOINER = Joiner.on('\n');
+
+  private static String combineStackTraces(List<Throwable> exceptions) {
+    if (exceptions.isEmpty()) {
+      return "none";
+    } else {
+      return STACK_TRACE_JOINER.join(Iterables.transform(exceptions,
+          new Function<Throwable, String>() {
+            private int index = 1;
+            @Override public String apply(Throwable exception) {
+              return String.format("[%d] %s",
+                  index++, Throwables.getStackTraceAsString(exception));
+            }
+          }));
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/thrift/callers/StatTrackingCaller.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/thrift/callers/StatTrackingCaller.java b/commons/src/main/java/org/apache/aurora/common/thrift/callers/StatTrackingCaller.java
new file mode 100644
index 0000000..514e665
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/thrift/callers/StatTrackingCaller.java
@@ -0,0 +1,103 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.thrift.callers;
+
+import java.lang.reflect.Method;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import javax.annotation.Nullable;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+
+import org.apache.aurora.common.thrift.TResourceExhaustedException;
+import org.apache.aurora.common.thrift.TTimeoutException;
+import org.apache.thrift.async.AsyncMethodCallback;
+
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+import org.apache.aurora.common.stats.StatsProvider;
+import org.apache.aurora.common.stats.StatsProvider.RequestTimer;
+
+/**
+ * A caller that exports statistics about calls made to the wrapped caller.
+ *
+ * @author William Farner
+ */
+public class StatTrackingCaller extends CallerDecorator {
+
+  private final StatsProvider statsProvider;
+  private final String serviceName;
+
+  private final LoadingCache<Method, RequestTimer> stats =
+      CacheBuilder.newBuilder().build(new CacheLoader<Method, RequestTimer>() {
+        @Override public RequestTimer load(Method method) {
+          // Thrift does not support overloads - so just the name disambiguates all calls.
+          return statsProvider.makeRequestTimer(serviceName + "_" + method.getName());
+        }
+      });
+
+  /**
+   * Creates a new stat tracking caller, which will export stats to the given {@link StatsProvider}.
+   *
+   * @param decoratedCaller The caller to decorate with a deadline.
+   * @param async Whether the caller is asynchronous.
+   * @param statsProvider The stat provider to export statistics to.
+   * @param serviceName The name of the service that methods are being called on.
+   */
+  public StatTrackingCaller(Caller decoratedCaller, boolean async, StatsProvider statsProvider,
+      String serviceName) {
+    super(decoratedCaller, async);
+
+    this.statsProvider = statsProvider;
+    this.serviceName = serviceName;
+  }
+
+  @Override
+  public Object call(Method method, Object[] args, @Nullable AsyncMethodCallback callback,
+      @Nullable Amount<Long, Time> connectTimeoutOverride) throws Throwable {
+    final RequestTimer requestStats = stats.get(method);
+    final long startTime = System.nanoTime();
+
+    ResultCapture capture = new ResultCapture() {
+      @Override public void success() {
+        requestStats.requestComplete(TimeUnit.NANOSECONDS.toMicros(
+            System.nanoTime() - startTime));
+      }
+
+      @Override public boolean fail(Throwable t) {
+        // TODO(John Sirois): the ruby client reconnects for timeouts too - this provides a natural
+        // backoff mechanism - consider how to plumb something similar.
+        if (t instanceof TTimeoutException || t instanceof TimeoutException) {
+          requestStats.incTimeouts();
+          return true;
+        }
+
+        // TODO(John Sirois): consider ditching reconnects since its nearly redundant with errors as
+        // it stands.
+        if (!(t instanceof TResourceExhaustedException)) {
+          requestStats.incReconnects();
+        }
+        // TODO(John Sirois): provide more detailed stats: track counts for distinct exceptions types,
+        // track retries-per-method, etc...
+        requestStats.incErrors();
+        return true;
+      }
+    };
+
+    return invoke(method, args, callback, capture, connectTimeoutOverride);
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/thrift/callers/ThriftCaller.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/thrift/callers/ThriftCaller.java b/commons/src/main/java/org/apache/aurora/common/thrift/callers/ThriftCaller.java
new file mode 100644
index 0000000..4e62940
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/thrift/callers/ThriftCaller.java
@@ -0,0 +1,157 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.thrift.callers;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+import org.apache.aurora.common.net.pool.Connection;
+import org.apache.aurora.common.net.pool.ObjectPool;
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+import org.apache.aurora.common.net.pool.ResourceExhaustedException;
+import org.apache.aurora.common.thrift.TResourceExhaustedException;
+import org.apache.aurora.common.thrift.TTimeoutException;
+import org.apache.aurora.common.net.loadbalancing.RequestTracker;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.apache.thrift.transport.TTransport;
+
+import javax.annotation.Nullable;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import java.util.logging.Logger;
+
+/**
+ * A caller that issues calls to a target that is assumed to be a client to a thrift service.
+ *
+ * @author William Farner
+ */
+public class ThriftCaller<T> implements Caller {
+  private static final Logger LOG = Logger.getLogger(ThriftCaller.class.getName());
+
+  private final ObjectPool<Connection<TTransport, InetSocketAddress>> connectionPool;
+  private final RequestTracker<InetSocketAddress> requestTracker;
+  private final Function<TTransport, T> clientFactory;
+  private final Amount<Long, Time> timeout;
+  private final boolean debug;
+
+  /**
+   * Creates a new thrift caller.
+   *
+   * @param connectionPool The connection pool to use.
+   * @param requestTracker The request tracker to nofify of request results.
+   * @param clientFactory Factory to use for building client object instances.
+   * @param timeout The timeout to use when requesting objects from the connection pool.
+   * @param debug Whether to use the caller in debug mode.
+   */
+  public ThriftCaller(ObjectPool<Connection<TTransport, InetSocketAddress>> connectionPool,
+      RequestTracker<InetSocketAddress> requestTracker, Function<TTransport, T> clientFactory,
+      Amount<Long, Time> timeout, boolean debug) {
+
+    this.connectionPool = connectionPool;
+    this.requestTracker = requestTracker;
+    this.clientFactory = clientFactory;
+    this.timeout = timeout;
+    this.debug = debug;
+  }
+
+  @Override
+  public Object call(Method method, Object[] args, @Nullable AsyncMethodCallback callback,
+      @Nullable Amount<Long, Time> connectTimeoutOverride) throws Throwable {
+
+    final Connection<TTransport, InetSocketAddress> connection = getConnection(connectTimeoutOverride);
+    final long startNanos = System.nanoTime();
+
+    ResultCapture capture = new ResultCapture() {
+      @Override public void success() {
+        try {
+          requestTracker.requestResult(connection.getEndpoint(),
+              RequestTracker.RequestResult.SUCCESS, System.nanoTime() - startNanos);
+        } finally {
+          connectionPool.release(connection);
+        }
+      }
+
+      @Override public boolean fail(Throwable t) {
+        if (debug) {
+          LOG.warning(String.format("Call to endpoint: %s failed: %s", connection, t));
+        }
+
+        try {
+          requestTracker.requestResult(connection.getEndpoint(),
+              RequestTracker.RequestResult.FAILED, System.nanoTime() - startNanos);
+        } finally {
+          connectionPool.remove(connection);
+        }
+        return true;
+      }
+    };
+
+    return invokeMethod(clientFactory.apply(connection.get()), method, args, callback, capture);
+  }
+
+  private static Object invokeMethod(Object target, Method method, Object[] args,
+      AsyncMethodCallback callback, final ResultCapture capture) throws Throwable {
+
+    // Swap the wrapped callback out for ours.
+    if (callback != null) {
+      callback = new WrappedMethodCallback(callback, capture);
+
+      List<Object> argsList = Lists.newArrayList(args);
+      argsList.add(callback);
+      args = argsList.toArray();
+    }
+
+    try {
+      Object result = method.invoke(target, args);
+      if (callback == null) capture.success();
+
+      return result;
+    } catch (InvocationTargetException t) {
+      // We allow this one to go to both sync and async captures.
+      if (callback != null) {
+        callback.onError((Exception) t.getCause());
+        return null;
+      } else {
+        capture.fail(t.getCause());
+        throw t.getCause();
+      }
+    }
+  }
+
+  private Connection<TTransport, InetSocketAddress> getConnection(
+      Amount<Long, Time> connectTimeoutOverride)
+      throws TResourceExhaustedException, TTimeoutException {
+    try {
+      Connection<TTransport, InetSocketAddress> connection;
+      if (connectTimeoutOverride != null) {
+        connection = connectionPool.get(connectTimeoutOverride);
+      } else {
+        connection = (timeout.getValue() > 0)
+            ? connectionPool.get(timeout) : connectionPool.get();
+      }
+
+      if (connection == null) {
+        throw new TResourceExhaustedException("no connection was available");
+      }
+      return connection;
+    } catch (ResourceExhaustedException e) {
+      throw new TResourceExhaustedException(e);
+    } catch (TimeoutException e) {
+      throw new TTimeoutException(e);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredNonblockingServerSocket.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredNonblockingServerSocket.java b/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredNonblockingServerSocket.java
new file mode 100644
index 0000000..a14f53a4
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredNonblockingServerSocket.java
@@ -0,0 +1,80 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.thrift.monitoring;
+
+import com.google.common.base.Preconditions;
+import org.apache.aurora.common.net.monitoring.ConnectionMonitor;
+import org.apache.thrift.transport.TNonblockingServerSocket;
+import org.apache.thrift.transport.TNonblockingSocket;
+import org.apache.thrift.transport.TTransportException;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Extension of TNonblockingServerSocket that allows for tracking of connected clients.
+ *
+ * @author William Farner
+ */
+public class TMonitoredNonblockingServerSocket extends TNonblockingServerSocket {
+  private final ConnectionMonitor monitor;
+
+  public TMonitoredNonblockingServerSocket(int port, ConnectionMonitor monitor)
+      throws TTransportException {
+    super(port);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  public TMonitoredNonblockingServerSocket(int port, int clientTimeout, ConnectionMonitor monitor)
+      throws TTransportException {
+    super(port, clientTimeout);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  public TMonitoredNonblockingServerSocket(InetSocketAddress bindAddr, ConnectionMonitor monitor)
+      throws TTransportException {
+    super(bindAddr);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  public TMonitoredNonblockingServerSocket(InetSocketAddress bindAddr, int clientTimeout,
+      ConnectionMonitor monitor) throws TTransportException {
+    super(bindAddr, clientTimeout);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  @Override
+  protected TNonblockingSocket acceptImpl() throws TTransportException {
+    /* TODO(William Farner): Finish implementing...may require an object proxy.
+    final TNonblockingSocket socket = super.acceptImpl();
+
+    TNonblockingSocket wrappedSocket = new TNonblockingSocket(socket.get) {
+      @Override public void close() {
+        super.close();
+        monitor.disconnected(this);
+      }
+    };
+
+    monitor.connected(wrappedSocket, socket.getSocket().getInetAddress());
+
+    return wrappedSocket;
+
+    */
+    return super.acceptImpl();
+  }
+
+  @Override
+  public void close() {
+    super.close();
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredProcessor.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredProcessor.java b/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredProcessor.java
new file mode 100644
index 0000000..a0d7d5f
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredProcessor.java
@@ -0,0 +1,62 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.thrift.monitoring;
+
+import com.google.common.base.Preconditions;
+import org.apache.aurora.common.net.loadbalancing.RequestTracker;
+import org.apache.thrift.TException;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.transport.TSocket;
+
+import java.net.InetSocketAddress;
+
+import static org.apache.aurora.common.net.loadbalancing.RequestTracker.RequestResult.*;
+
+/**
+ * A TProcessor that joins a wrapped TProcessor with a monitor.
+ *
+ * @author William Farner
+ */
+public class TMonitoredProcessor implements TProcessor {
+  private final TProcessor wrapped;
+  private final TMonitoredServerSocket monitoredServerSocket;
+  private final RequestTracker<InetSocketAddress> monitor;
+
+  public TMonitoredProcessor(TProcessor wrapped, TMonitoredServerSocket monitoredServerSocket,
+      RequestTracker<InetSocketAddress> monitor) {
+    this.wrapped = Preconditions.checkNotNull(wrapped);
+    this.monitoredServerSocket = Preconditions.checkNotNull(monitoredServerSocket);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  @Override
+  public boolean process(TProtocol in, TProtocol out) throws TException {
+    long startNanos = System.nanoTime();
+    boolean exceptionThrown = false;
+    try {
+      return wrapped.process(in, out);
+    } catch (TException e) {
+      exceptionThrown = true;
+      throw e;
+    } finally {
+      InetSocketAddress address = monitoredServerSocket.getAddress((TSocket) in.getTransport());
+      Preconditions.checkState(address != null,
+          "Address unknown for transport " + in.getTransport());
+
+      monitor.requestResult(address, exceptionThrown ? FAILED : SUCCESS,
+          System.nanoTime() - startNanos);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredServerSocket.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredServerSocket.java b/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredServerSocket.java
new file mode 100644
index 0000000..f4405c4
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/thrift/monitoring/TMonitoredServerSocket.java
@@ -0,0 +1,111 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.thrift.monitoring;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+import org.apache.aurora.common.net.monitoring.ConnectionMonitor;
+import org.apache.thrift.transport.TServerSocket;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransportException;
+
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Extension of TServerSocket that allows for tracking of connected clients.
+ *
+ * @author William Farner
+ */
+public class TMonitoredServerSocket extends TServerSocket {
+  private ConnectionMonitor<InetSocketAddress> monitor;
+
+  public TMonitoredServerSocket(ServerSocket serverSocket,
+      ConnectionMonitor<InetSocketAddress> monitor) {
+    super(serverSocket);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  public TMonitoredServerSocket(ServerSocket serverSocket, int clientTimeout,
+      ConnectionMonitor<InetSocketAddress> monitor) {
+    super(serverSocket, clientTimeout);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  public TMonitoredServerSocket(int port, ConnectionMonitor<InetSocketAddress> monitor)
+      throws TTransportException {
+    super(port);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  public TMonitoredServerSocket(int port, int clientTimeout,
+      ConnectionMonitor<InetSocketAddress> monitor) throws TTransportException {
+    super(port, clientTimeout);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  public TMonitoredServerSocket(InetSocketAddress bindAddr,
+      ConnectionMonitor<InetSocketAddress> monitor) throws TTransportException {
+    super(bindAddr);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  public TMonitoredServerSocket(InetSocketAddress bindAddr, int clientTimeout,
+      ConnectionMonitor<InetSocketAddress> monitor) throws TTransportException {
+    super(bindAddr, clientTimeout);
+    this.monitor = Preconditions.checkNotNull(monitor);
+  }
+
+  private final Map<TSocket, InetSocketAddress> addressMap =
+      Collections.synchronizedMap(Maps.<TSocket, InetSocketAddress>newHashMap());
+
+  public InetSocketAddress getAddress(TSocket socket) {
+    return addressMap.get(socket);
+  }
+
+  @Override
+  protected TSocket acceptImpl() throws TTransportException {
+    final TSocket socket = super.acceptImpl();
+    final InetSocketAddress remoteAddress =
+        (InetSocketAddress) socket.getSocket().getRemoteSocketAddress();
+
+    TSocket monitoredSocket = new TSocket(socket.getSocket()) {
+      boolean closed = false;
+
+      @Override public void close() {
+        try {
+          super.close();
+        } finally {
+          if (!closed) {
+            monitor.released(remoteAddress);
+            addressMap.remove(this);
+          }
+          closed = true;
+        }
+      }
+    };
+
+    addressMap.put(monitoredSocket, remoteAddress);
+
+    monitor.connected(remoteAddress);
+    return monitoredSocket;
+  }
+
+  @Override
+  public void close() {
+    super.close();
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/thrift/testing/MockTSocket.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/thrift/testing/MockTSocket.java b/commons/src/main/java/org/apache/aurora/common/thrift/testing/MockTSocket.java
new file mode 100644
index 0000000..2cec711
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/thrift/testing/MockTSocket.java
@@ -0,0 +1,45 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.thrift.testing;
+
+import org.apache.thrift.transport.TSocket;
+
+/**
+ * @author William Farner
+ */
+public class MockTSocket extends TSocket {
+  public static final String HOST = "dummyHost";
+  public static final int PORT = 1000;
+
+  private boolean connected = false;
+
+  public MockTSocket() {
+    super(HOST, PORT);
+  }
+
+  @Override
+  public void open() {
+    connected = true;
+    // TODO(William Farner): Allow for failure injection here by throwing TTransportException.
+  }
+
+  @Override
+  public boolean isOpen() {
+    return connected;
+  }
+
+  public void close() {
+    connected = false;
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/thrift/testing/TestThriftTypes.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/thrift/testing/TestThriftTypes.java b/commons/src/main/java/org/apache/aurora/common/thrift/testing/TestThriftTypes.java
new file mode 100644
index 0000000..377228f
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/thrift/testing/TestThriftTypes.java
@@ -0,0 +1,171 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.thrift.testing;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+import org.apache.thrift.TBase;
+import org.apache.thrift.TBaseHelper;
+import org.apache.thrift.TException;
+import org.apache.thrift.TFieldIdEnum;
+import org.apache.thrift.protocol.TField;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.protocol.TStruct;
+import org.apache.thrift.protocol.TType;
+
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * Hand-coded thrift types for use in tests.
+ *
+ * @author John Sirois
+ */
+public class TestThriftTypes {
+  public static class Field implements TFieldIdEnum {
+    private static final Map<Short, Field> FIELDS_BY_ID = Maps.newHashMap();
+    public static Field forId(int id) {
+      Field field = FIELDS_BY_ID.get((short) id);
+      Preconditions.checkArgument(field != null, "No Field with id: %s", id);
+      return field;
+    }
+
+    public static final Field NAME = new Field((short) 0, "name");
+    public static final Field VALUE = new Field((short) 1, "value");
+
+    private final short fieldId;
+    private final String fieldName;
+
+    private Field(short fieldId, String fieldName) {
+      this.fieldId = fieldId;
+      this.fieldName = fieldName;
+      FIELDS_BY_ID.put(fieldId, this);
+    }
+
+    @Override
+    public short getThriftFieldId() {
+      return fieldId;
+    }
+
+    @Override
+    public String getFieldName() {
+      return fieldName;
+    }
+  }
+
+  public static class Struct implements TBase<Struct, Field> {
+    private final Map<Field, Object> fields = Maps.newHashMap();
+
+    public Struct() {}
+
+    public Struct(String name, String value) {
+      fields.put(Field.NAME, name);
+      fields.put(Field.VALUE, value);
+    }
+
+    public String getName() {
+      Object name = getFieldValue(Field.NAME);
+      return name == null ? null : (String) name;
+    }
+
+    public String getValue() {
+      Object value = getFieldValue(Field.VALUE);
+      return value == null ? null : (String) value;
+    }
+
+    @Override
+    public void read(TProtocol tProtocol) throws TException {
+      tProtocol.readStructBegin();
+      TField field;
+      while((field = tProtocol.readFieldBegin()).type != TType.STOP) {
+        fields.put(fieldForId(field.id), tProtocol.readString());
+        tProtocol.readFieldEnd();
+      }
+      tProtocol.readStructEnd();
+    }
+
+    @Override
+    public void write(TProtocol tProtocol) throws TException {
+      tProtocol.writeStructBegin(new TStruct("Field"));
+      for (Entry<Field, Object> entry : fields.entrySet()) {
+        Field field = entry.getKey();
+        tProtocol.writeFieldBegin(
+            new TField(field.getFieldName(), TType.STRING, field.getThriftFieldId()));
+        tProtocol.writeString(entry.getValue().toString());
+        tProtocol.writeFieldEnd();
+      }
+      tProtocol.writeFieldStop();
+      tProtocol.writeStructEnd();
+    }
+
+    @Override
+    public boolean isSet(Field field) {
+      return fields.containsKey(field);
+    }
+
+    @Override
+    public Object getFieldValue(Field field) {
+      return fields.get(field);
+    }
+
+    @Override
+    public void setFieldValue(Field field, Object o) {
+      fields.put(field, o);
+    }
+
+    @Override
+    public TBase<Struct, Field> deepCopy() {
+      Struct struct = new Struct();
+      struct.fields.putAll(fields);
+      return struct;
+    }
+
+    @Override
+    public int compareTo(Struct other) {
+      if (!getClass().equals(other.getClass())) {
+        return getClass().getName().compareTo(other.getClass().getName());
+      }
+
+      int lastComparison;
+
+      lastComparison = Integer.valueOf(fields.size()).compareTo(other.fields.size());
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+
+      for (Map.Entry<Field, Object> entry : fields.entrySet()) {
+        Field field = entry.getKey();
+        lastComparison = Boolean.TRUE.compareTo(other.isSet(field));
+        if (lastComparison != 0) {
+          return lastComparison;
+        }
+        lastComparison = TBaseHelper.compareTo(entry.getValue(), other.getFieldValue(field));
+        if (lastComparison != 0) {
+          return lastComparison;
+        }
+      }
+      return 0;
+    }
+
+    @Override
+    public void clear() {
+      fields.clear();
+    }
+
+    @Override
+    public Field fieldForId(int fieldId) {
+      return Field.forId(fieldId);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/BackoffDecider.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/BackoffDecider.java b/commons/src/main/java/org/apache/aurora/common/util/BackoffDecider.java
new file mode 100644
index 0000000..e11a52c
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/BackoffDecider.java
@@ -0,0 +1,663 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.aurora.common.base.MorePreconditions;
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+import org.apache.aurora.common.stats.Stats;
+import org.apache.aurora.common.stats.StatsProvider;
+
+import javax.annotation.Nullable;
+import java.util.Deque;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Logger;
+
+/**
+ * Handles logic for deciding whether to back off from calls to a backend.
+ *
+ * This works by offering a guard method {@link #shouldBackOff()}, which instructs the caller
+ * whether they should avoid making the call.  The backoff logic will maintain statistics about
+ * the failure rate, and push into a backoff state (silent period) when the failure rate exceeds
+ * the configured threshold.  At the end of the quiet period, a recovery state will be entered,
+ * during which the decider will allow traffic to ramp back up to full capacity.
+ *
+ * The expected use case looks something like this:
+ *
+ * <pre>
+ * void sendRequestGuarded() {
+ *   if (!decider.shouldBackOff()) {
+ *     boolean success = sendRequestUnguarded();
+ *     if (success) {
+ *       decider.addSuccess();
+ *     } else {
+ *       decider.addFailure();
+ *     }
+ *   }
+ * }
+ * </pre>
+ *
+ * @author William Farner
+ */
+public class BackoffDecider {
+  private static final Logger LOG = Logger.getLogger(BackoffDecider.class.getName());
+
+  // The group that this decider is a part of.
+  private final Iterable<BackoffDecider> deciderGroup;
+
+  private final TimedStateMachine stateMachine;
+
+  private final String name;
+
+  private final double toleratedFailureRate;
+
+  @VisibleForTesting final RequestWindow requests;
+
+  // Used to calculate backoff durations when in backoff state.
+  private final BackoffStrategy strategy;
+
+  private final Amount<Long, Time> recoveryPeriod;
+  private long previousBackoffPeriodNs = 0;
+
+  // Used for random selection during recovery period.
+  private final Random random;
+
+  private final Clock clock;
+  private final AtomicLong backoffs;
+  private final RecoveryType recoveryType;
+
+  /**
+   * Different types of recovery mechanisms to use after exiting the backoff state.
+   */
+  public static enum RecoveryType {
+    // Randomly allows traffic to flow through, with a linearly-ascending probability.
+    RANDOM_LINEAR,
+    // Allows full traffic capacity to flow during the recovery period.
+    FULL_CAPACITY
+  }
+
+  private BackoffDecider(String name, int seedSize, double toleratedFailureRate,
+      @Nullable Iterable<BackoffDecider> deciderGroup, BackoffStrategy strategy,
+      @Nullable Amount<Long, Time> recoveryPeriod,
+      long requestWindowNs, int numBuckets, RecoveryType recoveryType, StatsProvider statsProvider,
+      Random random, Clock clock) {
+    MorePreconditions.checkNotBlank(name);
+    Preconditions.checkArgument(seedSize > 0);
+    Preconditions.checkArgument(toleratedFailureRate >= 0 && toleratedFailureRate < 1.0);
+    Preconditions.checkNotNull(strategy);
+    Preconditions.checkArgument(recoveryPeriod == null || recoveryPeriod.getValue() > 0);
+    Preconditions.checkArgument(requestWindowNs > 0);
+    Preconditions.checkArgument(numBuckets > 0);
+    Preconditions.checkNotNull(recoveryType);
+    Preconditions.checkNotNull(statsProvider);
+    Preconditions.checkNotNull(random);
+    Preconditions.checkNotNull(clock);
+
+    this.name = name;
+    this.toleratedFailureRate = toleratedFailureRate;
+    this.deciderGroup = deciderGroup;
+    this.strategy = strategy;
+    this.recoveryPeriod = recoveryPeriod;
+    this.recoveryType = recoveryType;
+
+    this.random = random;
+    this.clock = clock;
+
+    this.backoffs = statsProvider.makeCounter(name + "_backoffs");
+    this.requests = new RequestWindow(requestWindowNs, numBuckets, seedSize);
+
+    this.stateMachine = new TimedStateMachine(name);
+  }
+
+  /**
+   * Checks whether the caller should back off and if not then returns immediately; otherwise the
+   * method blocks until it is safe for the caller to proceed without backing off further based on
+   * all data available at the time of this call.
+   *
+   * @return the amount of time in nanoseconds spent awaiting backoff
+   * @throws InterruptedException if the calling thread was interrupted while backing off
+   */
+  public long awaitBackoff() throws InterruptedException {
+    if (shouldBackOff()) {
+      long backoffTimeMs = stateMachine.getStateRemainingMs();
+
+      if (backoffTimeMs > 0) {
+        // Wait without holding any external locks.
+        Object waitCondition = new Object();
+        synchronized (waitCondition) {
+          waitCondition.wait(backoffTimeMs);
+        }
+        return backoffTimeMs;
+      }
+    }
+    return 0;
+  }
+
+  /**
+   * Checks whether this decider instructs the caller that it should back off from the associated
+   * backend.  This is determined based on the response history for the backend as well as the
+   * backoff state of the decider group (if configured).
+   *
+   * @return {@code true} if the decider is in backoff mode, otherwise {@code false}.
+   */
+  @SuppressWarnings("fallthrough")
+  public synchronized boolean shouldBackOff() {
+
+    boolean preventRequest;
+    switch (stateMachine.getState()) {
+      case NORMAL:
+        preventRequest = false;
+        break;
+
+      case BACKOFF:
+        if (deciderGroup != null && allOthersBackingOff()) {
+          LOG.info("Backends in group with " + name + " down, forcing back up.");
+          stateMachine.transitionUnbounded(State.FORCED_NORMAL);
+          return false;
+        } else if (stateMachine.isStateExpired()) {
+          long recoveryPeriodNs = recoveryPeriod == null ? stateMachine.getStateDurationNs()
+              : recoveryPeriod.as(Time.NANOSECONDS);
+
+          // The silent period has expired, move to recovery state (and drop to its case block).
+          stateMachine.transition(State.RECOVERY, recoveryPeriodNs);
+          LOG.info(String.format("%s recovering for %s ms", name,
+              Amount.of(recoveryPeriodNs, Time.NANOSECONDS).as(Time.MILLISECONDS)));
+        } else {
+          preventRequest = true;
+          break;
+        }
+
+      case RECOVERY:
+        if (deciderGroup != null && allOthersBackingOff()) {
+          return false;
+        } else if (stateMachine.isStateExpired()) {
+          // We have reached the end of the recovery period, return to normal.
+          stateMachine.transitionUnbounded(State.NORMAL);
+          previousBackoffPeriodNs = 0;
+          preventRequest = false;
+        } else {
+          switch (recoveryType) {
+            case RANDOM_LINEAR:
+              // In the recovery period, allow request rate to return linearly to the full load.
+              preventRequest = random.nextDouble() > stateMachine.getStateFractionComplete();
+              break;
+            case FULL_CAPACITY:
+              preventRequest = false;
+              break;
+            default:
+              throw new IllegalStateException("Unhandled recovery type " + recoveryType);
+          }
+        }
+
+        break;
+
+      case FORCED_NORMAL:
+        if (!allOthersBackingOff()) {
+          // We were in forced normal state, but at least one other backend is up, try recovering.
+          stateMachine.transition(State.RECOVERY, stateMachine.getStateDurationNs());
+          preventRequest = false;
+        } else {
+          preventRequest = true;
+        }
+
+        break;
+
+      default:
+        LOG.severe("Unrecognized state: " + stateMachine.getState());
+        preventRequest = false;
+    }
+
+    if (preventRequest) {
+      backoffs.incrementAndGet();
+    }
+    return preventRequest;
+  }
+
+  private boolean allOthersBackingOff() {
+    // Search for another decider that is not backing off.
+    for (BackoffDecider decider : deciderGroup) {
+      State deciderState = decider.stateMachine.getState();
+      boolean inBackoffState = deciderState == State.BACKOFF || deciderState == State.FORCED_NORMAL;
+      if ((decider != this) && !inBackoffState) {
+        return false;
+      }
+    }
+
+    return true;
+  }
+
+  /**
+   * Records a failed request to the backend.
+   */
+  public void addFailure() {
+    addResult(false);
+  }
+
+  /**
+   * Records a successful request to the backend.
+   */
+  public void addSuccess() {
+    addResult(true);
+  }
+
+  /**
+   * Transitions the state to BACKOFF and logs a message appropriately if it is doing so because of high fail rate
+   * or by force.
+   *
+   * @param failRate rate of request failures on this host.
+   * @param force if {@code true}, forces the transition to BACKOFF. Typically used in cases when the host
+   * was not found to be alive by LiveHostChecker.
+   */
+  public synchronized void transitionToBackOff(double failRate, boolean force) {
+    long prevBackoffMs = Amount.of(previousBackoffPeriodNs, Time.NANOSECONDS)
+        .as(Time.MILLISECONDS);
+
+    long backoffPeriodNs = Amount.of(strategy.calculateBackoffMs(prevBackoffMs), Time.MILLISECONDS)
+        .as(Time.NANOSECONDS);
+    if (!force) {
+      LOG.info(String.format("%s failure rate at %g, backing off for %s ms", name,failRate,
+          Amount.of(backoffPeriodNs, Time.NANOSECONDS).as(Time.MILLISECONDS)));
+    } else {
+      LOG.info(String.format("%s forced to back off for %s ms", name,
+          Amount.of(backoffPeriodNs, Time.NANOSECONDS).as(Time.MILLISECONDS)));
+    }
+    stateMachine.transition(State.BACKOFF, backoffPeriodNs);
+    previousBackoffPeriodNs = backoffPeriodNs;
+  }
+
+  @SuppressWarnings("fallthrough")
+  private synchronized void addResult(boolean success) {
+    // Disallow statistics updating if we are in backoff state.
+    if (stateMachine.getState() == State.BACKOFF) {
+      return;
+    }
+
+    requests.addResult(success);
+    double failRate = requests.getFailureRate();
+    boolean highFailRate = requests.isSeeded() && (failRate > toleratedFailureRate);
+
+    switch (stateMachine.getState()) {
+      case NORMAL:
+        if (!highFailRate) {
+          // No-op.
+          break;
+        } else {
+          // Artificially move into recovery state (by falling through) with a zero-duration
+          // time window, to trigger the initial backoff period.
+          stateMachine.setStateDurationNs(0);
+        }
+
+      case RECOVERY:
+        if (highFailRate) {
+          // We were trying to recover, and the failure rate is still too high.  Go back to
+          // backoff state for a longer duration.
+          requests.reset();
+
+          // transition the state machine to BACKOFF state, due to high fail rate.
+          transitionToBackOff(failRate, false);
+        } else {
+          // Do nothing.  We only exit the recovery state by expiration.
+        }
+        break;
+
+      case FORCED_NORMAL:
+        if (!highFailRate) {
+          stateMachine.transition(State.RECOVERY, stateMachine.getStateDurationNs());
+        }
+        break;
+
+      case BACKOFF:
+        throw new IllegalStateException("Backoff state may only be exited by expiration.");
+    }
+  }
+
+  /**
+   * Creates a builder object.
+   *
+   * @param name Name for the backoff decider to build.
+   * @return A builder.
+   */
+  public static Builder builder(String name) {
+    return new Builder(name);
+  }
+
+  /**
+   * Builder class to configure a BackoffDecider.
+   *
+   * The builder allows for customization of many different parameters to the BackoffDecider, while
+   * defining defaults wherever possible.  The following defaults are used:
+   *
+   * <ul>
+   * <li> seed size - The number of requests to accumulate before a backoff will be considered.
+   * 100
+   *
+   * <li> tolerated failure rate - Maximum failure rate before backing off.
+   * 0.5
+   *
+   * <li> decider group - Group this decider is a part of, to prevent complete backend failure.
+   * null (disabled)
+   *
+   * <li> strategy - Used to calculate subsequent backoff durations.
+   * TruncatedBinaryBackoff, initial 100 ms, max 10s
+   *
+   * <li> recovery period - Fixed recovery period while ramping traffic back to full capacity..
+   * null (use last backoff period)
+   *
+   * <li> request window - Duration of the sliding window of requests to track statistics for.
+   * 10 seconds
+   *
+   * <li> num buckets - The number of time slices within the request window, for stat expiration.
+   *               The sliding request window advances in intervals of request window / num buckets.
+   * 100
+   *
+   * <li> recovery type - Defines behavior during the recovery period, and how traffic is permitted.
+   * random linear
+   *
+   * <li> stat provider - The stats provider to export statistics to.
+   * Stats.STATS_PROVIDER
+   * </ul>
+   *
+   */
+  public static class Builder {
+    private String name;
+    private int seedSize = 100;
+    private double toleratedFailureRate = 0.5;
+    private Set<BackoffDecider> deciderGroup = null;
+    private BackoffStrategy strategy = new TruncatedBinaryBackoff(
+        Amount.of(100L, Time.MILLISECONDS), Amount.of(10L, Time.SECONDS));
+    private Amount<Long, Time> recoveryPeriod = null;
+    private long requestWindowNs = Amount.of(10L, Time.SECONDS).as(Time.NANOSECONDS);
+    private int numBuckets = 100;
+    private RecoveryType recoveryType = RecoveryType.RANDOM_LINEAR;
+    private StatsProvider statsProvider = Stats.STATS_PROVIDER;
+    private Random random = Random.Util.newDefaultRandom();
+    private Clock clock = Clock.SYSTEM_CLOCK;
+
+    Builder(String name) {
+      this.name = name;
+    }
+
+    /**
+     * Sets the number of requests that must be accumulated before the error rate will be
+     * calculated.  This improves the genesis problem where the first few requests are errors,
+     * causing flapping in and out of backoff state.
+     *
+     * @param seedSize Request seed size.
+     * @return A reference to the builder.
+     */
+    public Builder withSeedSize(int seedSize) {
+      this.seedSize = seedSize;
+      return this;
+    }
+
+    /**
+     * Sets the tolerated failure rate for the decider.  If the rate is exceeded for the time
+     * window, the decider begins backing off.
+     *
+     * @param toleratedRate The tolerated failure rate (between 0.0 and 1.0, exclusive).
+     * @return A reference to the builder.
+     */
+    public Builder withTolerateFailureRate(double toleratedRate) {
+      this.toleratedFailureRate = toleratedRate;
+      return this;
+    }
+
+    /**
+     * Makes the decider a part of a group.  When a decider is a part of a group, it will monitor
+     * the other deciders to ensure that all deciders do not back off at once.
+     *
+     * @param deciderGroup Group to make this decider a part of.  More deciders may be added to the
+     *     group after this call is made.
+     * @return A reference to the builder.
+     */
+    public Builder groupWith(Set<BackoffDecider> deciderGroup) {
+      this.deciderGroup = deciderGroup;
+      return this;
+    }
+
+    /**
+     * Overrides the default backoff strategy.
+     *
+     * @param strategy Backoff strategy to use.
+     * @return A reference to the builder.
+     */
+    public Builder withStrategy(BackoffStrategy strategy) {
+      this.strategy = strategy;
+      return this;
+    }
+
+    /**
+     * Overrides the default recovery period behavior.  By default, the recovery period is equal
+     * to the previous backoff period (which is equivalent to setting the recovery period to null
+     * here).  A non-null value here will assign a fixed recovery period.
+     *
+     * @param recoveryPeriod Fixed recovery period.
+     * @return A reference to the builder.
+     */
+    public Builder withRecoveryPeriod(@Nullable Amount<Long, Time> recoveryPeriod) {
+      this.recoveryPeriod = recoveryPeriod;
+      return this;
+    }
+
+    /**
+     * Sets the time window over which to analyze failures.  Beyond the time window, request history
+     * is discarded (and ignored).
+     *
+     * @param requestWindow The analysis time window.
+     * @return A reference to the builder.
+     */
+    public Builder withRequestWindow(Amount<Long, Time> requestWindow) {
+      this.requestWindowNs = requestWindow.as(Time.NANOSECONDS);
+      return this;
+    }
+
+    /**
+     * Sets the number of time slices that the decider will use to partition aggregate statistics.
+     *
+     * @param numBuckets Bucket count.
+     * @return A reference to the builder.
+     */
+    public Builder withBucketCount(int numBuckets) {
+      this.numBuckets = numBuckets;
+      return this;
+    }
+
+    /**
+     * Sets the recovery mechanism to use when in the recovery period.
+     *
+     * @param recoveryType The recovery mechanism to use.
+     * @return A reference to the builder.
+     */
+    public Builder withRecoveryType(RecoveryType recoveryType) {
+      this.recoveryType = recoveryType;
+      return this;
+    }
+
+    /**
+     * Sets the stats provider that statistics should be exported to.
+     *
+     * @param statsProvider Stats provider to use.
+     * @return A reference to the builder.
+     */
+    public Builder withStatsProvider(StatsProvider statsProvider) {
+      this.statsProvider = statsProvider;
+      return this;
+    }
+
+    @VisibleForTesting public Builder withRandom(Random random) {
+      this.random = random;
+      return this;
+    }
+
+    @VisibleForTesting public Builder withClock(Clock clock) {
+      this.clock = clock;
+      return this;
+    }
+
+    /**
+     * Gets a reference to the built decider object.
+     * @return A decider object.
+     */
+    public BackoffDecider build() {
+      BackoffDecider decider =  new BackoffDecider(name, seedSize, toleratedFailureRate,
+          deciderGroup, strategy, recoveryPeriod, requestWindowNs, numBuckets, recoveryType,
+          statsProvider, random, clock);
+      if (deciderGroup != null) deciderGroup.add(decider);
+      return decider;
+    }
+  }
+
+  private class TimeSlice {
+    int requestCount = 0;
+    int failureCount = 0;
+    final long bucketStartNs;
+
+    public TimeSlice() {
+      bucketStartNs = clock.nowNanos();
+    }
+  }
+
+  class RequestWindow {
+    // These store the sum of the respective fields contained within buckets.  Doing so removes the
+    // need to accumulate the counts within the buckets every time the backoff state is
+    // recalculated.
+    @VisibleForTesting long totalRequests = 0;
+    @VisibleForTesting long totalFailures = 0;
+
+    private final long durationNs;
+    private final long bucketLengthNs;
+    private final int seedSize;
+
+    // Stores aggregate request/failure counts for time slices.
+    private final Deque<TimeSlice> buckets = Lists.newLinkedList();
+
+    RequestWindow(long durationNs, int bucketCount, int seedSize) {
+      this.durationNs = durationNs;
+      this.bucketLengthNs = durationNs / bucketCount;
+      buckets.addFirst(new TimeSlice());
+      this.seedSize = seedSize;
+    }
+
+    void reset() {
+      totalRequests = 0;
+      totalFailures = 0;
+      buckets.clear();
+      buckets.addFirst(new TimeSlice());
+    }
+
+    void addResult(boolean success) {
+      maybeShuffleBuckets();
+      buckets.peekFirst().requestCount++;
+      totalRequests++;
+
+      if (!success) {
+        buckets.peekFirst().failureCount++;
+        totalFailures++;
+      }
+    }
+
+    void maybeShuffleBuckets() {
+      // Check if the first bucket is still relevant.
+      if (clock.nowNanos() - buckets.peekFirst().bucketStartNs >= bucketLengthNs) {
+
+        // Remove old buckets.
+        while (!buckets.isEmpty()
+               && buckets.peekLast().bucketStartNs < clock.nowNanos() - durationNs) {
+          TimeSlice removed = buckets.removeLast();
+          totalRequests -= removed.requestCount;
+          totalFailures -= removed.failureCount;
+        }
+
+        buckets.addFirst(new TimeSlice());
+      }
+    }
+
+    boolean isSeeded() {
+      return totalRequests >= seedSize;
+    }
+
+    double getFailureRate() {
+      return totalRequests == 0 ? 0 : ((double) totalFailures) / totalRequests;
+    }
+  }
+
+  private static enum State {
+    NORMAL,        // All requests are being permitted.
+    BACKOFF,       // Quiet period while waiting for backend to recover/improve.
+    RECOVERY,      // Ramping period where an ascending fraction of requests is being permitted.
+    FORCED_NORMAL  // All other backends in the group are backing off, so this one is forced normal.
+  }
+  private class TimedStateMachine {
+    final StateMachine<State> stateMachine;
+
+    private long stateEndNs;
+    private long stateDurationNs;
+
+    TimedStateMachine(String name) {
+      stateMachine = StateMachine.<State>builder(name + "_backoff_state_machine")
+          .addState(State.NORMAL, State.BACKOFF, State.FORCED_NORMAL)
+          .addState(State.BACKOFF, State.RECOVERY, State.FORCED_NORMAL)
+          .addState(State.RECOVERY, State.NORMAL, State.BACKOFF, State.FORCED_NORMAL)
+          .addState(State.FORCED_NORMAL, State.RECOVERY)
+          .initialState(State.NORMAL)
+          .build();
+    }
+
+    State getState() {
+      return stateMachine.getState();
+    }
+
+    void transitionUnbounded(State state) {
+      stateMachine.transition(state);
+    }
+
+    void transition(State state, long durationNs) {
+      transitionUnbounded(state);
+      this.stateEndNs = clock.nowNanos() + durationNs;
+      this.stateDurationNs = durationNs;
+    }
+
+    long getStateDurationNs() {
+      return stateDurationNs;
+    }
+
+    long getStateDurationMs() {
+      return Amount.of(stateDurationNs, Time.NANOSECONDS).as(Time.MILLISECONDS);
+    }
+
+    void setStateDurationNs(long stateDurationNs) {
+      this.stateDurationNs = stateDurationNs;
+    }
+
+    long getStateRemainingNs() {
+      return stateEndNs - clock.nowNanos();
+    }
+
+    long getStateRemainingMs() {
+      return Amount.of(getStateRemainingNs(), Time.NANOSECONDS).as(Time.MILLISECONDS);
+    }
+
+    double getStateFractionComplete() {
+      return 1.0 - ((double) getStateRemainingNs()) / stateDurationNs;
+    }
+
+    boolean isStateExpired() {
+      return clock.nowNanos() > stateEndNs;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/BackoffHelper.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/BackoffHelper.java b/commons/src/main/java/org/apache/aurora/common/util/BackoffHelper.java
new file mode 100644
index 0000000..8f31ea7
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/BackoffHelper.java
@@ -0,0 +1,152 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.aurora.common.base.ExceptionalSupplier;
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+
+import java.util.logging.Logger;
+
+/**
+ * A utility for dealing with backoffs of retryable actions.
+ *
+ * <p>TODO(John Sirois): investigate synergies with BackoffDecider.
+ *
+ * @author John Sirois
+ */
+public class BackoffHelper {
+  private static final Logger LOG = Logger.getLogger(BackoffHelper.class.getName());
+
+  private static final Amount<Long,Time> DEFAULT_INITIAL_BACKOFF = Amount.of(1L, Time.SECONDS);
+  private static final Amount<Long,Time> DEFAULT_MAX_BACKOFF = Amount.of(1L, Time.MINUTES);
+
+  private final Clock clock;
+  private final BackoffStrategy backoffStrategy;
+
+  /**
+   * Creates a new BackoffHelper that uses truncated binary backoff starting at a 1 second backoff
+   * and maxing out at a 1 minute backoff.
+   */
+  public BackoffHelper() {
+    this(DEFAULT_INITIAL_BACKOFF, DEFAULT_MAX_BACKOFF);
+  }
+
+  /**
+   * Creates a new BackoffHelper that uses truncated binary backoff starting at the given
+   * {@code initialBackoff} and maxing out at the given {@code maxBackoff}.
+   *
+   * @param initialBackoff the initial amount of time to back off
+   * @param maxBackoff the maximum amount of time to back off
+   */
+  public BackoffHelper(Amount<Long, Time> initialBackoff, Amount<Long, Time> maxBackoff) {
+    this(new TruncatedBinaryBackoff(initialBackoff, maxBackoff));
+  }
+
+  /**
+   * Creates a new BackoffHelper that uses truncated binary backoff starting at the given
+   * {@code initialBackoff} and maxing out at the given {@code maxBackoff}. This will either:
+   * <ul>
+   *   <li>{@code stopAtMax == true} : throw {@code BackoffExpiredException} when maxBackoff is
+   *   reached</li>
+   *   <li>{@code stopAtMax == false} : continue backing off with maxBackoff</li>
+   * </ul>
+   *
+   * @param initialBackoff the initial amount of time to back off
+   * @param maxBackoff the maximum amount of time to back off
+   * @param stopAtMax if true, this will throw {@code BackoffStoppedException} when the max backoff is
+   * reached
+   */
+  public BackoffHelper(Amount<Long, Time> initialBackoff, Amount<Long, Time> maxBackoff,
+      boolean stopAtMax) {
+    this(new TruncatedBinaryBackoff(initialBackoff, maxBackoff, stopAtMax));
+  }
+
+  /**
+   * Creates a BackoffHelper that uses the given {@code backoffStrategy} to calculate backoffs
+   * between retries.
+   *
+   * @param backoffStrategy the backoff strategy to use
+   */
+  public BackoffHelper(BackoffStrategy backoffStrategy) {
+    this(Clock.SYSTEM_CLOCK, backoffStrategy);
+  }
+
+  @VisibleForTesting BackoffHelper(Clock clock, BackoffStrategy backoffStrategy) {
+    this.clock = Preconditions.checkNotNull(clock);
+    this.backoffStrategy = Preconditions.checkNotNull(backoffStrategy);
+  }
+
+  /**
+   * Executes the given task using the configured backoff strategy until the task succeeds as
+   * indicated by returning {@code true}.
+   *
+   * @param task the retryable task to execute until success
+   * @throws InterruptedException if interrupted while waiting for the task to execute successfully
+   * @throws BackoffStoppedException if the backoff stopped unsuccessfully
+   * @throws E if the task throws
+   */
+  public <E extends Exception> void doUntilSuccess(final ExceptionalSupplier<Boolean, E> task)
+      throws InterruptedException, BackoffStoppedException, E {
+    doUntilResult(new ExceptionalSupplier<Boolean, E>() {
+      @Override public Boolean get() throws E {
+        Boolean result = task.get();
+        return Boolean.TRUE.equals(result) ? result : null;
+      }
+    });
+  }
+
+  /**
+   * Executes the given task using the configured backoff strategy until the task succeeds as
+   * indicated by returning a non-null value.
+   *
+   * @param task the retryable task to execute until success
+   * @return the result of the successfully executed task
+   * @throws InterruptedException if interrupted while waiting for the task to execute successfully
+   * @throws BackoffStoppedException if the backoff stopped unsuccessfully
+   * @throws E if the task throws
+   */
+  public <T, E extends Exception> T doUntilResult(ExceptionalSupplier<T, E> task)
+      throws InterruptedException, BackoffStoppedException, E {
+    T result = task.get(); // give an immediate try
+    return (result != null) ? result : retryWork(task);
+  }
+
+  private <T, E extends Exception> T retryWork(ExceptionalSupplier<T, E> work)
+      throws E, InterruptedException, BackoffStoppedException {
+    long currentBackoffMs = 0;
+    while (backoffStrategy.shouldContinue(currentBackoffMs)) {
+      currentBackoffMs = backoffStrategy.calculateBackoffMs(currentBackoffMs);
+      LOG.fine("Operation failed, backing off for " + currentBackoffMs + "ms");
+      clock.waitFor(currentBackoffMs);
+
+      T result = work.get();
+      if (result != null) {
+        return result;
+      }
+    }
+    throw new BackoffStoppedException(String.format("Backoff stopped without succeeding."));
+  }
+
+  /**
+   * Occurs after the backoff strategy should stop.
+   */
+  public static class BackoffStoppedException extends RuntimeException {
+    public BackoffStoppedException(String msg) {
+      super(msg);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/BackoffStrategy.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/BackoffStrategy.java b/commons/src/main/java/org/apache/aurora/common/util/BackoffStrategy.java
new file mode 100644
index 0000000..d954762
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/BackoffStrategy.java
@@ -0,0 +1,37 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+/**
+ * Encapsulates a strategy for backing off from an operation that repeatedly fails.
+ */
+public interface BackoffStrategy {
+
+  /**
+   * Calculates the amount of time to backoff from an operation.
+   *
+   * @param lastBackoffMs the last used backoff in milliseconds where 0 signifies no backoff has
+   *     been performed yet
+   * @return the amount of time in milliseconds to back off before retrying the operation
+   */
+  long calculateBackoffMs(long lastBackoffMs);
+
+  /**
+   * Returns whether to continue backing off.
+   *
+   * @param lastBackoffMs the last used backoff in milliseconds
+   * @return whether to continue backing off
+   */
+  boolean shouldContinue(long lastBackoffMs);
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/BuildInfo.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/BuildInfo.java b/commons/src/main/java/org/apache/aurora/common/util/BuildInfo.java
new file mode 100644
index 0000000..c290d32
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/BuildInfo.java
@@ -0,0 +1,108 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import java.io.InputStream;
+import java.util.Properties;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.aurora.common.base.MorePreconditions;
+
+/**
+ * Handles loading of a build properties file, and provides keys to look up known values in the
+ * properties.
+ */
+public class BuildInfo {
+
+  private static final Logger LOG = Logger.getLogger(BuildInfo.class.getName());
+
+  private static final String DEFAULT_BUILD_PROPERTIES_PATH = "build.properties";
+
+  private final String resourcePath;
+
+  private Properties properties = null;
+
+  /**
+   * Creates a build info container that will use the default properties file path.
+   */
+  public BuildInfo() {
+    this(DEFAULT_BUILD_PROPERTIES_PATH);
+  }
+
+  /**
+   * Creates a build info container, reading from the given path.
+   *
+   * @param resourcePath The resource path to read build properties from.
+   */
+  public BuildInfo(String resourcePath) {
+    this.resourcePath = MorePreconditions.checkNotBlank(resourcePath);
+  }
+
+  @VisibleForTesting
+  public BuildInfo(Properties properties) {
+    this.resourcePath = null;
+    this.properties = properties;
+  }
+
+  private void fetchProperties() {
+    properties = new Properties();
+    LOG.info("Fetching build properties from " + resourcePath);
+    InputStream in = ClassLoader.getSystemResourceAsStream(resourcePath);
+    if (in == null) {
+      LOG.warning("Failed to fetch build properties from " + resourcePath);
+      return;
+    }
+
+    try {
+      properties.load(in);
+    } catch (Exception e) {
+      LOG.log(Level.WARNING, "Failed to load properties file " + resourcePath, e);
+    }
+  }
+
+  /**
+   * Fetches the properties stored in the resource location.
+   *
+   * @return The loaded properties, or a default properties object if there was a problem loading
+   *    the specified properties resource.
+   */
+  public Properties getProperties() {
+    if (properties == null) fetchProperties();
+    return properties;
+  }
+
+  /**
+   * Values of keys that are expected to exist in the loaded properties file.
+   */
+  public enum Key {
+    PATH("build.path"),
+    USER("build.user.name"),
+    MACHINE("build.machine"),
+    DATE("build.date"),
+    TIME("build.time"),
+    TIMESTAMP("build.timestamp"),
+    GIT_TAG("build.git.tag"),
+    GIT_REVISION("build.git.revision"),
+    GIT_REVISION_NUMBER("build.git.revision.number"),
+    GIT_BRANCHNAME("build.git.branchname");
+
+    public final String value;
+    private Key(String value) {
+      this.value = value;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/Clock.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/Clock.java b/commons/src/main/java/org/apache/aurora/common/util/Clock.java
new file mode 100644
index 0000000..87afb4e
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/Clock.java
@@ -0,0 +1,70 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import java.io.Serializable;
+
+/**
+ * An abstraction of the system clock.
+ *
+ * @author John Sirois
+ */
+public interface Clock {
+
+  /**
+   * A clock that returns the the actual time reported by the system.
+   * This clock is guaranteed to be serializable.
+   */
+  Clock SYSTEM_CLOCK = new SerializableClock() {
+    @Override public long nowMillis() {
+      return System.currentTimeMillis();
+    }
+    @Override public long nowNanos() {
+      return System.nanoTime();
+    }
+    @Override public void waitFor(long millis) throws InterruptedException {
+      Thread.sleep(millis);
+    }
+  };
+
+  /**
+   * Returns the current time in milliseconds since the epoch.
+   *
+   * @return The current time in milliseconds since the epoch.
+   * @see System#currentTimeMillis()
+   */
+  long nowMillis();
+
+  /**
+   * Returns the current time in nanoseconds.  Should be used only for relative timing.
+   * See {@code System.nanoTime()} for tips on using the value returned here.
+   *
+   * @return A measure of the current time in nanoseconds.
+   * @see System#nanoTime()
+   */
+  long nowNanos();
+
+  /**
+   * Waits for the given amount of time to pass on this clock before returning.
+   *
+   * @param millis the amount of time to wait in milliseconds
+   * @throws InterruptedException if this wait was interrupted
+   */
+  void waitFor(long millis) throws InterruptedException;
+}
+
+/**
+ * A typedef to support anonymous {@link Clock} implementations that are also {@link Serializable}.
+ */
+interface SerializableClock extends Clock, Serializable { }

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/CommandExecutor.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/CommandExecutor.java b/commons/src/main/java/org/apache/aurora/common/util/CommandExecutor.java
new file mode 100644
index 0000000..e591721
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/CommandExecutor.java
@@ -0,0 +1,42 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import org.apache.aurora.common.base.ExceptionalCommand;
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+
+/**
+ * Asynchronous executor of enqueued tasks in a rate limited manner.
+ *
+ * @author Srinivasan Rajagopal
+ */
+public interface CommandExecutor {
+
+  /**
+   * Enqueue a task to be executed with retry semantics defined.
+   *
+   * @param name Human readable name for this task.
+   * @param task task to execute.
+   * @param exceptionClass Concrete exception type.
+   * @param maxTries num of tries in case of failure.
+   * @param retryDelay interval between retries in case of failure.
+   */
+  <E extends Exception> void execute(
+      String name,
+      ExceptionalCommand<E> task,
+      Class<E> exceptionClass,
+      int maxTries,
+      Amount<Long, Time> retryDelay);
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/DateUtils.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/DateUtils.java b/commons/src/main/java/org/apache/aurora/common/util/DateUtils.java
new file mode 100644
index 0000000..eb20925
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/DateUtils.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import java.util.Calendar;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Utilities for working with java {@link Date}s.
+ *
+ * @author John Sirois
+ */
+public final class DateUtils {
+
+  public static Date now() {
+    return new Date();
+  }
+
+  public static long toUnixTime(Date date) {
+    return toUnixTime(date.getTime());
+  }
+
+  public static long nowUnixTime() {
+    return toUnixTime(System.currentTimeMillis());
+  }
+
+  public static long toUnixTime(long millisSinceEpoch) {
+    return TimeUnit.MILLISECONDS.toSeconds(millisSinceEpoch);
+  }
+
+  public static Date ago(int calendarField, int amount) {
+    return ago(now(), calendarField, amount);
+  }
+
+   public static Date ago(Date referenceDate, int calendarField, int amount) {
+    Calendar calendar = Calendar.getInstance();
+    calendar.setTime(referenceDate);
+    calendar.add(calendarField, -1 * amount);
+    return calendar.getTime();
+  }
+
+  private DateUtils() {
+    // utility
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/FileUtils.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/FileUtils.java b/commons/src/main/java/org/apache/aurora/common/util/FileUtils.java
new file mode 100644
index 0000000..9b23ee0
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/FileUtils.java
@@ -0,0 +1,51 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import java.io.File;
+
+/**
+ * Utilities for working with Files
+ *
+ * @author Florian Leibert
+ */
+public final class FileUtils {
+
+  private FileUtils() {
+  }
+
+  /**
+   * recursively deletes the path and all it's content and returns true if it succeeds
+   * Note that the content could be partially deleted and the method return false
+   *
+   * @param path the path to delete
+   * @return true if the path was deleted
+   */
+  public static boolean forceDeletePath(File path) {
+    if (path == null) {
+      return false;
+    }
+    if (path.exists() && path.isDirectory()) {
+      File[] files = path.listFiles();
+      for (File file : files) {
+        if (file.isDirectory()) {
+          forceDeletePath(file);
+        } else {
+          file.delete();
+        }
+      }
+    }
+    return path.delete();
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/LowResClock.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/LowResClock.java b/commons/src/main/java/org/apache/aurora/common/util/LowResClock.java
new file mode 100644
index 0000000..69fb9ed
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/LowResClock.java
@@ -0,0 +1,108 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+import java.io.Closeable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+
+/**
+ * Low resolution implementation of a {@link Clock},
+ * optimized for fast reads at the expense of precision.
+ * It works by caching the result of the system clock for a
+ * {@code resolution} amount of time.
+ */
+public class LowResClock implements Clock, Closeable {
+  private static final ScheduledExecutorService GLOBAL_SCHEDULER =
+      Executors.newScheduledThreadPool(1, new ThreadFactory() {
+        public Thread newThread(Runnable r) {
+          Thread t = new Thread(r, "LowResClock");
+          t.setDaemon(true);
+          return t;
+        }
+      });
+
+  private volatile long time;
+  private final ScheduledFuture<?> updaterHandler;
+  private final Clock underlying;
+
+  @VisibleForTesting
+  LowResClock(Amount<Long, Time> resolution, ScheduledExecutorService executor, Clock clock) {
+    long sleepTimeMs = resolution.as(Time.MILLISECONDS);
+    Preconditions.checkArgument(sleepTimeMs > 0);
+    underlying = clock;
+    Runnable ticker = new Runnable() {
+      @Override public void run() {
+        time = underlying.nowMillis();
+      }
+    };
+
+    // Ensure the constructing thread sees a LowResClock with a valid (low-res) time by executing a
+    // blocking call now.
+    ticker.run();
+
+    updaterHandler =
+        executor.scheduleAtFixedRate(ticker, sleepTimeMs, sleepTimeMs, TimeUnit.MILLISECONDS);
+  }
+
+
+  /**
+   * Construct a LowResClock which wraps the system clock.
+   * This constructor will also schedule a periodic task responsible for
+   * updating the time every {@code resolution}.
+   */
+  public LowResClock(Amount<Long, Time> resolution) {
+    this(resolution, GLOBAL_SCHEDULER, Clock.SYSTEM_CLOCK);
+  }
+
+  /**
+   * Terminate the underlying updater task.
+   * Any subsequent usage of the clock will throw an {@link IllegalStateException}.
+   */
+  public void close() {
+    updaterHandler.cancel(true);
+  }
+
+  @Override
+  public long nowMillis() {
+    checkNotClosed();
+    return time;
+  }
+
+  @Override
+  public long nowNanos() {
+    return nowMillis() * 1000 * 1000;
+  }
+
+  @Override
+  public void waitFor(long millis) throws InterruptedException {
+    checkNotClosed();
+    underlying.waitFor(millis);
+  }
+
+  private void checkNotClosed() {
+    if (updaterHandler.isCancelled()) {
+      throw new IllegalStateException("LowResClock invoked after being closed!");
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/aurora/blob/06ddaadb/commons/src/main/java/org/apache/aurora/common/util/ParsingUtil.java
----------------------------------------------------------------------
diff --git a/commons/src/main/java/org/apache/aurora/common/util/ParsingUtil.java b/commons/src/main/java/org/apache/aurora/common/util/ParsingUtil.java
new file mode 100644
index 0000000..0747e7a
--- /dev/null
+++ b/commons/src/main/java/org/apache/aurora/common/util/ParsingUtil.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.common.util;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.aurora.common.collections.Pair;
+
+/**
+ * Common methods for parsing configs.
+ *
+ * @author John Sirois
+ */
+public class ParsingUtil {
+  /**
+   * Parses a string as a range between one integer and another.  The integers must be separated by
+   * a hypen character (space padding is acceptable).  Additionally, the first integer
+   * (left-hand side) must be less than or equal to the second (right-hand side).
+   *
+   * @param rangeString The string to parse as an integer range.
+   * @return A pair of the parsed integers.
+   */
+  public static Pair<Integer, Integer> parseRange(String rangeString) {
+    if (rangeString == null) return null;
+
+    String[] startEnd = rangeString.split("-");
+    Preconditions.checkState(
+        startEnd.length == 2, "Shard range format: start-end (e.g. 1-4)");
+    int start;
+    int end;
+    try {
+      start = Integer.parseInt(startEnd[0].trim());
+      end = Integer.parseInt(startEnd[1].trim());
+    } catch (NumberFormatException e) {
+      throw new IllegalArgumentException("Failed to parse shard range.", e);
+    }
+
+    Preconditions.checkState(
+        start <= end, "The left-hand side of a shard range must be <= the right-hand side.");
+    return Pair.of(start, end);
+  }
+}


Mime
View raw message