Merge pull request #740 from oracle/bad_client_pooling

Clean up fix for failed clients being returned to the pool

Author: rjeberhard

operator/src/main/java/oracle/kubernetes/operator/Watcher.java

@@ -1,4 +1,4 @@
-// Copyright 2017, 2018 Oracle Corporation and/or its affiliates.  All rights reserved.
+// Copyright 2017, 2019 Oracle Corporation and/or its affiliates.  All rights reserved.
 // Licensed under the Universal Permissive License v 1.0 as shown at
 // http://oss.oracle.com/licenses/upl.
 
@@ -126,13 +126,7 @@ private void watchForEvents() {
                 .withResourceVersion(resourceVersion.toString())
                 .withTimeoutSeconds(tuning.watchLifetime))) {
       while (watch.hasNext()) {
-        Watch.Response<T> item;
-        try {
-          item = watch.next();
-        } catch (Throwable e) {
-          watch.discardClient();
-          throw e;
-        }
+        Watch.Response<T> item = watch.next();
 
         if (isStopping()) setIsDraining(true);
         if (isDraining()) continue;

operator/src/main/java/oracle/kubernetes/operator/builders/WatchI.java

@@ -1,4 +1,4 @@
-// Copyright 2018 Oracle Corporation and/or its affiliates.  All rights reserved.
+// Copyright 2018, 2019 Oracle Corporation and/or its affiliates.  All rights reserved.
 // Licensed under the Universal Permissive License v 1.0 as shown at
 // http://oss.oracle.com/licenses/upl.
 
@@ -14,6 +14,4 @@
  * @param <T> the generic object type
  */
 public interface WatchI<T>
-    extends Iterable<Watch.Response<T>>, Iterator<Watch.Response<T>>, java.io.Closeable {
-  default void discardClient() {}
-}
+    extends Iterable<Watch.Response<T>>, Iterator<Watch.Response<T>>, java.io.Closeable {}

operator/src/main/java/oracle/kubernetes/operator/builders/WatchImpl.java

@@ -1,4 +1,4 @@
-// Copyright 2018 Oracle Corporation and/or its affiliates.  All rights reserved.
+// Copyright 2018, 2019 Oracle Corporation and/or its affiliates.  All rights reserved.
 // Licensed under the Universal Permissive License v 1.0 as shown at
 // http://oss.oracle.com/licenses/upl.
 
@@ -28,12 +28,7 @@ public class WatchImpl<T> implements WatchI<T> {
   @Override
   public void close() throws IOException {
     impl.close();
-    pool.recycle(client);
-  }
-
-  @Override
-  public void discardClient() {
-    client = pool.take();
+    if (client != null) pool.recycle(client);
   }
 
   @Override
@@ -49,6 +44,11 @@ public boolean hasNext() {
 
   @Override
   public Watch.Response<T> next() {
-    return impl.next();
+    try {
+      return impl.next();
+    } catch (Exception e) {
+      client = null;
+      throw e;
+    }
   }
 }

operator/src/main/java/oracle/kubernetes/operator/helpers/ClientPool.java

@@ -1,4 +1,4 @@
-// Copyright 2017, Oracle Corporation and/or its affiliates.  All rights reserved.
+// Copyright 2017, 2019 Oracle Corporation and/or its affiliates.  All rights reserved.
 // Licensed under the Universal Permissive License v 1.0 as shown at
 // http://oss.oracle.com/licenses/upl.
 
@@ -17,7 +17,7 @@
 
 public class ClientPool extends Pool<ApiClient> {
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
-  private static final ClientPool SINGLETON = new ClientPool();
+  private static ClientPool SINGLETON = new ClientPool();
 
   private static final ClientFactory FACTORY = new DefaultClientFactory();
 
@@ -27,8 +27,6 @@ public static ClientPool getInstance() {
 
   private final AtomicBoolean isFirst = new AtomicBoolean(true);
 
-  private ClientPool() {}
-
   @Override
   protected ApiClient create() {
     return getApiClient();

operator/src/main/java/oracle/kubernetes/operator/helpers/Pool.java

@@ -1,10 +1,10 @@
-// Copyright 2017, 2018, Oracle Corporation and/or its affiliates.  All rights reserved.
+// Copyright 2017, 2019 Oracle Corporation and/or its affiliates.  All rights reserved.
 // Licensed under the Universal Permissive License v 1.0 as shown at
 // http://oss.oracle.com/licenses/upl.
 
 package oracle.kubernetes.operator.helpers;
 
-import java.lang.ref.WeakReference;
+import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import oracle.kubernetes.operator.logging.LoggingFacade;
 import oracle.kubernetes.operator.logging.LoggingFactory;
@@ -14,7 +14,7 @@ public abstract class Pool<T> {
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
 
   // volatile since multiple threads may access queue reference
-  private volatile WeakReference<ConcurrentLinkedQueue<T>> queue;
+  private volatile Queue<T> queue = new ConcurrentLinkedQueue<>();
 
   /**
    * Gets a new object from the pool. If no object is available in the pool, this method creates a
@@ -36,20 +36,8 @@ public final T take() {
     return instance;
   }
 
-  private ConcurrentLinkedQueue<T> getQueue() {
-    WeakReference<ConcurrentLinkedQueue<T>> referenceQueue = queue;
-    if (referenceQueue != null) {
-      ConcurrentLinkedQueue<T> returnQueue = referenceQueue.get();
-      if (returnQueue != null) {
-        return returnQueue;
-      }
-    }
-
-    // overwrite the queue
-    ConcurrentLinkedQueue<T> d = new ConcurrentLinkedQueue<>();
-    queue = new WeakReference<>(d);
-
-    return d;
+  protected Queue<T> getQueue() {
+    return queue;
   }
 
   /**
@@ -72,9 +60,4 @@ public final void recycle(T instance) {
    * @return Created instance
    */
   protected abstract T create();
-
-  /** Drains pool of all entries; useful for unit-testing */
-  public void drain() {
-    getQueue().clear();
-  }
 }

operator/src/test/java/oracle/kubernetes/operator/builders/WatchBuilderTest.java

@@ -1,4 +1,4 @@
-// Copyright 2018, Oracle Corporation and/or its affiliates.  All rights reserved.
+// Copyright 2018, 2019 Oracle Corporation and/or its affiliates.  All rights reserved.
 // Licensed under the Universal Permissive License v 1.0 as shown at
 // http://oss.oracle.com/licenses/upl.
 
@@ -8,30 +8,38 @@
 import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
 import static oracle.kubernetes.operator.LabelConstants.CREATEDBYOPERATOR_LABEL;
 import static oracle.kubernetes.operator.LabelConstants.DOMAINUID_LABEL;
-import static oracle.kubernetes.operator.builders.EventMatcher.*;
+import static oracle.kubernetes.operator.builders.EventMatcher.addEvent;
+import static oracle.kubernetes.operator.builders.EventMatcher.deleteEvent;
+import static oracle.kubernetes.operator.builders.EventMatcher.errorEvent;
+import static oracle.kubernetes.operator.builders.EventMatcher.modifyEvent;
 import static oracle.kubernetes.operator.builders.WatchBuilderTest.JsonServletAction.withResponses;
 import static oracle.kubernetes.operator.builders.WatchBuilderTest.ParameterValidation.parameter;
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.*;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.junit.Assert.fail;
 
 import com.meterware.pseudoserver.HttpUserAgentTest;
 import com.meterware.pseudoserver.PseudoServlet;
 import com.meterware.pseudoserver.WebResource;
 import com.meterware.simplestub.Memento;
 import com.meterware.simplestub.StaticStubSupport;
-import com.squareup.okhttp.Call;
 import io.kubernetes.client.ApiClient;
-import io.kubernetes.client.ApiException;
 import io.kubernetes.client.models.V1ObjectMeta;
 import io.kubernetes.client.models.V1Pod;
 import io.kubernetes.client.models.V1Service;
-import java.io.IOException;
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
-import java.util.function.BiFunction;
+import java.util.Queue;
 import oracle.kubernetes.TestUtils;
-import oracle.kubernetes.operator.helpers.Pool;
+import oracle.kubernetes.operator.helpers.ClientPool;
 import oracle.kubernetes.weblogic.domain.v2.Domain;
 import org.junit.After;
 import org.junit.Before;
@@ -61,12 +69,12 @@ public class WatchBuilderTest extends HttpUserAgentTest {
   @Before
   public void setUp() throws Exception {
     mementos.add(TestUtils.silenceOperatorLogger());
-    mementos.add(TestServerWatchFactory.install(getHostPath()));
+    mementos.add(ClientPoolStub.install(getHostPath()));
     validationErrors = new ArrayList<>();
   }
 
   @After
-  public void tearDown() throws Exception {
+  public void tearDown() {
     for (Memento memento : mementos) memento.revert();
     if (!validationErrors.isEmpty()) throw validationErrors.get(0);
   }
@@ -85,7 +93,35 @@ public void whenDomainWatchReceivesAddResponse_returnItFromIterator() throws Exc
     assertThat(domainWatch, contains(addEvent(domain)));
   }
 
-  @SuppressWarnings("unchecked")
+  @Test
+  public void afterWatchClosed_returnClientToPool() throws Exception {
+    Domain domain =
+        new Domain()
+            .withApiVersion(API_VERSION)
+            .withKind("Domain")
+            .withMetadata(createMetaData("domain1", NAMESPACE));
+    defineHttpResponse(DOMAIN_RESOURCE, withResponses(createAddedResponse(domain)));
+
+    try (WatchI<Domain> domainWatch = new WatchBuilder().createDomainWatch(NAMESPACE)) {
+      domainWatch.next();
+    }
+
+    assertThat(ClientPoolStub.getPooledClients(), not(empty()));
+  }
+
+  @Test
+  public void afterWatchError_closeDoesNotReturnClientToPool() throws Exception {
+    defineHttpResponse(DOMAIN_RESOURCE, withResponses());
+
+    try (WatchI<Domain> domainWatch = new WatchBuilder().createDomainWatch(NAMESPACE)) {
+      domainWatch.next();
+      fail("Should have thrown an exception");
+    } catch (Throwable ignore) {
+    }
+
+    assertThat(ClientPoolStub.getPooledClients(), is(empty()));
+  }
+
   @Test
   public void whenDomainWatchReceivesModifyAndDeleteResponses_returnBothFromIterator()
       throws Exception {
@@ -239,7 +275,7 @@ private JsonServlet(JsonServletAction... actions) {
     }
 
     @Override
-    public WebResource getGetResponse() throws IOException {
+    public WebResource getGetResponse() {
       if (requestNum >= actions.size())
         return new WebResource("Unexpected Request #" + requestNum, HTTP_UNAVAILABLE);
 
@@ -251,6 +287,7 @@ public WebResource getGetResponse() throws IOException {
     }
   }
 
+  @SuppressWarnings("SameParameterValue")
   private V1ObjectMeta createMetaData(String name, String namespace) {
     return new V1ObjectMeta()
         .name(name)
@@ -274,45 +311,38 @@ private <T> String createDeletedResponse(T object) {
     return WatchEvent.createDeleteEvent(object).toJson();
   }
 
+  @SuppressWarnings("SameParameterValue")
   private String createErrorResponse(int statusCode) {
     return WatchEvent.createErrorEvent(statusCode).toJson();
   }
 
-  static class TestServerWatchFactory extends WatchBuilder.WatchFactoryImpl {
+  static class ClientPoolStub extends ClientPool {
+    private String basePath;
+    private static Queue<ApiClient> queue;
+
     static Memento install(String basePath) throws NoSuchFieldException {
-      return StaticStubSupport.install(
-          WatchBuilder.class, "FACTORY", new TestServerWatchFactory(basePath));
+      queue = new ArrayDeque<>();
+      return StaticStubSupport.install(ClientPool.class, "SINGLETON", new ClientPoolStub(basePath));
     }
 
-    private String basePath;
+    static Collection<ApiClient> getPooledClients() {
+      return Collections.unmodifiableCollection(queue);
+    }
 
-    private TestServerWatchFactory(String basePath) {
+    ClientPoolStub(String basePath) {
       this.basePath = basePath;
     }
 
     @Override
-    public <T> WatchI<T> createWatch(
-        Pool<ApiClient> pool,
-        CallParams callParams,
-        Class<?> responseBodyType,
-        BiFunction<ApiClient, CallParams, Call> function)
-        throws ApiException {
-      Pool<ApiClient> testPool =
-          new Pool<ApiClient>() {
-
-            @Override
-            protected ApiClient create() {
-              Memento memento = TestUtils.silenceOperatorLogger();
-              try {
-                ApiClient client = pool.take();
-                client.setBasePath(basePath);
-                return client;
-              } finally {
-                memento.revert();
-              }
-            }
-          };
-      return super.createWatch(testPool, callParams, responseBodyType, function);
+    protected ApiClient create() {
+      ApiClient apiClient = super.create();
+      apiClient.setBasePath(basePath);
+      return apiClient;
+    }
+
+    @Override
+    protected Queue<ApiClient> getQueue() {
+      return queue;
     }
   }
 }

operator/src/test/java/oracle/kubernetes/operator/helpers/ClientPoolTest.java

@@ -0,0 +1,46 @@
+// Copyright 2019 Oracle Corporation and/or its affiliates.  All rights reserved.
+// Licensed under the Universal Permissive License v 1.0 as shown at
+// http://oss.oracle.com/licenses/upl.
+
+package oracle.kubernetes.operator.helpers;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.junit.Assert.*;
+
+import com.meterware.simplestub.Memento;
+import io.kubernetes.client.ApiClient;
+import java.util.ArrayList;
+import java.util.List;
+import oracle.kubernetes.TestUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ClientPoolTest {
+
+  private List<Memento> mementos = new ArrayList<>();
+
+  @Before
+  public void setUp() {
+    mementos.add(TestUtils.silenceOperatorLogger());
+  }
+
+  @After
+  public void tearDown() {
+    for (Memento memento : mementos) memento.revert();
+  }
+
+  @Test
+  public void onTake_returnApiClient() {
+    assertThat(ClientPool.getInstance().take(), instanceOf(ApiClient.class));
+  }
+
+  @Test
+  public void afterRecycle_takeReturnsSameClient() {
+    ApiClient apiClient = ClientPool.getInstance().take();
+    ClientPool.getInstance().recycle(apiClient);
+
+    assertThat(ClientPool.getInstance().take(), sameInstance(apiClient));
+  }
+}