Hot Reload for creds

Signed-off-by: Maxim Beryanov <msberyanov@gmail.com>

pom.xml

@@ -18,7 +18,7 @@
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.ibm.etcd</groupId>
    <artifactId>etcd-java</artifactId>
-   <version>0.0.25-SNAPSHOT</version>
+   <version>0.0.25</version>
    <packaging>jar</packaging>
 
    <name>etcd-java</name>

src/main/java/com/ibm/etcd/client/EtcdClient.java

@@ -95,9 +95,13 @@
 import io.netty.util.IllegalReferenceCountException;
 import io.netty.util.ReferenceCounted;
 import io.netty.util.concurrent.FastThreadLocalThread;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class EtcdClient implements KvStoreClient {
 
+    private static final Logger logger = LoggerFactory.getLogger(EtcdClient.class);
+
     private static final Key<String> TOKEN_KEY =
             Key.of("token", Metadata.ASCII_STRING_MARSHALLER);
 
@@ -111,6 +115,7 @@ public class EtcdClient implements KvStoreClient {
 
     public static final long DEFAULT_TIMEOUT_MS = 10_000L; // 10sec default
     public static final int DEFAULT_SESSION_TIMEOUT_SECS = 20; // 20sec default
+    public static final long AUTH_RETRY_TIMEOUT_MS = 100L; // 100ms default
 
     // (not intended to be strict hostname validation here)
     protected static final Pattern ADDR_PATT =
@@ -120,7 +125,7 @@ public class EtcdClient implements KvStoreClient {
 
     private final int sessionTimeoutSecs;
 
-    private final ByteString name, password;
+    private ByteString name, password;
 
     private final MultithreadEventLoopGroup internalExecutor;
     private final ScheduledExecutorService sharedInternalExecutor;
@@ -580,7 +585,7 @@ public boolean isClosed() {
     // ------ authentication logic
 
     private static final Set<Code> OTHER_AUTH_FAILURE_CODES = ImmutableSet.of(
-            Code.INVALID_ARGUMENT, Code.FAILED_PRECONDITION, Code.PERMISSION_DENIED, Code.UNKNOWN);
+            Code.INVALID_ARGUMENT, Code.FAILED_PRECONDITION, Code.PERMISSION_DENIED, Code.INTERNAL, Code.UNKNOWN);
 
     // Various different errors can imply a re-auth is required (sometimes non-obvious),
     // so we cover most related messages to be safe. This should not cause problems since
@@ -665,11 +670,10 @@ public void applyRequestMetadata(RequestInfo requestInfo,
                             failStatus = Status.UNAUTHENTICATED
                                     .withDescription("(Re)authentication RPC failed")
                                     .withCause(failure);
-                            authFailRetryTime = System.currentTimeMillis() + 5_000L;
+                            authFailRetryTime = System.currentTimeMillis() + AUTH_RETRY_TIMEOUT_MS;
                         } else {
                             failStatus = Status.fromThrowable(failure);
-                            // If this was a real auth failure, postpone further attempts a bit longer
-                            authFailRetryTime = System.currentTimeMillis() + 15_000L;
+                            authFailRetryTime = System.currentTimeMillis() + AUTH_RETRY_TIMEOUT_MS;
                         }
                         lastAuthFailStatus = failStatus;
                         // Augment with the RPC failure that triggered the re-authentication,
@@ -697,6 +701,8 @@ private static Metadata tokenHeader(AuthenticateResponse authResponse) {
     }
 
     private ListenableFuture<AuthenticateResponse> authenticate() {
+        logger.error("Auth token seems to be incorrect or uninitialized yet. Getting new token with current etcd creds...");
+
         AuthenticateRequest request = AuthenticateRequest.newBuilder()
                 .setNameBytes(name).setPasswordBytes(password).build();
         // no call creds for auth call
@@ -770,6 +776,12 @@ public PersistentLease getSessionLease() {
         return sl;
     }
 
+    @Override
+    public void updateCredentials(String name, String password) {
+        this.name = ByteString.copyFromUtf8(name);
+        this.password = ByteString.copyFromUtf8(password);
+    }
+
     public Executor getExecutor() {
         return grpc.getResponseExecutor();
     }

src/main/java/com/ibm/etcd/client/GrpcClient.java

@@ -264,16 +264,8 @@ private <ReqT,R> ListenableFuture<R> call(MethodDescriptor<ReqT,R> method,
                 // multiple retries disabled or deadline expired
                 return Futures.immediateFailedFuture(t);
             }
-            boolean reauth = false;
-            if (authProvider.requiresReauth(t)) {
-                if (afterReauth) {
-                    // if we have an auth failure immediately following a reauth, give up
-                    // (important to avoid infinite loop of auth failures)
-                    return Futures.immediateFailedFuture(t);
-                }
-                reauthenticate(baseCallOpts, t);
-                reauth = true;
-            } else if (!retry.retry(t, request)) {
+            boolean reauth = reauthIfRequired(t, baseCallOpts);;
+            if (!reauth && !retry.retry(t, request)) {
                 // retry predicate says no (non retryable request and/or error)
                 return Futures.immediateFailedFuture(t);
             }
@@ -442,8 +434,6 @@ final class ResilientBiDiStream<ReqT,RespT> {
         private boolean finished;
         private Throwable error;
 
-        private boolean lastAuthFailed;
-
         /**
          * 
          * @param method
@@ -636,7 +626,6 @@ public void beforeStart(ClientCallStreamObserver<ReqT> rs) {
             // called from grpc response thread
             @Override
             public void onNext(RespT value) {
-                lastAuthFailed = false;
                 respStream.onNext(value);
             }
             // called from grpc response thread
@@ -646,10 +635,9 @@ public void onError(Throwable t) {
                 if (finished) {
                     finalError = true;
                 } else {
-                    reauthed = !lastAuthFailed && reauthIfRequired(t, sentCallOptions);
+                    reauthed = reauthIfRequired(t, sentCallOptions);
                     finalError = !reauthed && !retryableStreamError(t);
                 }
-                lastAuthFailed = reauthed;
                 if (!finalError) {
                     int errCount = -1;
                     String msg;
@@ -703,7 +691,6 @@ public void onError(Throwable t) {
             // called from grpc response thread
             @Override
             public void onCompleted() {
-                lastAuthFailed = false;
                 if (!finished) {
                     logger.warn("Unexpected onCompleted received"
                             + " for stream of method " + method.getFullMethodName());

src/main/java/com/ibm/etcd/client/KvStoreClient.java

@@ -44,4 +44,5 @@ public interface KvStoreClient extends Closeable {
      */
     PersistentLease getSessionLease();
 
+    void updateCredentials(String name, String password);
 }