Merge pull request #4945 from shawkins/iss4931

fix #4931: using coarse locking to prevent odd concurrent behavior

CHANGELOG.md

@@ -19,6 +19,7 @@
 * Fix #4891: address vertx not completely reading exec streams
 * Fix #4899: BuildConfigs.instantiateBinary().fromFile() does not time out
 * Fix #4908: using the response headers in the vertx response
+* Fix #4931: using coarse grain locking for all mock server operations
 * Fix #4947: typo in HttpClient.Factory scoring system logic
 * Fix #4928: allows non-okhttp clients to handle invalid status
 

junit/kubernetes-server-mock/src/main/java/io/fabric8/kubernetes/client/server/mock/KubernetesCrudDispatcher.java

@@ -18,6 +18,7 @@
 import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
 import io.fabric8.kubernetes.client.Watcher.Action;
 import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
+import io.fabric8.kubernetes.client.server.mock.crud.KubernetesCrudDispatcherException;
 import io.fabric8.kubernetes.client.server.mock.crud.KubernetesCrudDispatcherHandler;
 import io.fabric8.kubernetes.client.server.mock.crud.KubernetesCrudPersistence;
 import io.fabric8.kubernetes.client.server.mock.crud.PatchHandler;
@@ -47,8 +48,8 @@
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicLong;
-
-import static io.fabric8.kubernetes.client.server.mock.crud.KubernetesCrudDispatcherHandler.process;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 public class KubernetesCrudDispatcher extends CrudDispatcher implements KubernetesCrudPersistence, CustomResourceAware {
 
@@ -61,6 +62,7 @@ public class KubernetesCrudDispatcher extends CrudDispatcher implements Kubernet
   private final KubernetesCrudDispatcherHandler postHandler;
   private final KubernetesCrudDispatcherHandler putHandler;
   private final KubernetesCrudDispatcherHandler patchHandler;
+  private final ReadWriteLock lock = new ReentrantReadWriteLock();
 
   public KubernetesCrudDispatcher() {
     this(Collections.emptyList());
@@ -81,6 +83,17 @@ public KubernetesCrudDispatcher(List<CustomResourceDefinitionContext> crdContext
     crdContexts.stream().forEach(this::expectCustomResource);
   }
 
+  MockResponse process(RecordedRequest request, KubernetesCrudDispatcherHandler handler) {
+    lock.writeLock().lock();
+    try {
+      return handler.handle(request);
+    } catch (KubernetesCrudDispatcherException e) {
+      return new MockResponse().setResponseCode(e.getCode()).setBody(e.toStatusBody());
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+
   /**
    * Adds the specified object to the in-memory db.
    *
@@ -111,10 +124,15 @@ public MockResponse handleUpdate(RecordedRequest request) {
    */
   @Override
   public MockResponse handleGet(String path) {
-    if (detectWatchMode(path)) {
-      return handleWatch(path);
+    lock.readLock().lock();
+    try {
+      if (detectWatchMode(path)) {
+        return handleWatch(path);
+      }
+      return handle(path, null);
+    } finally {
+      lock.readLock().unlock();
     }
-    return handle(path, null);
   }
 
   private interface EventProcessor {
@@ -126,17 +144,15 @@ private MockResponse handle(String path, EventProcessor eventProcessor) {
     List<String> items = new ArrayList<>();
     AttributeSet query = attributeExtractor.fromPath(path);
 
-    synchronized (map) {
-      new ArrayList<>(map.entrySet()).stream()
-          .filter(entry -> entry.getKey().matches(query))
-          .forEach(entry -> {
-            LOGGER.debug("Entry found for query {} : {}", query, entry);
-            items.add(entry.getValue());
-            if (eventProcessor != null) {
-              eventProcessor.processEvent(path, query, entry.getKey());
-            }
-          });
-    }
+    new ArrayList<>(map.entrySet()).stream()
+        .filter(entry -> entry.getKey().matches(query))
+        .forEach(entry -> {
+          LOGGER.debug("Entry found for query {} : {}", query, entry);
+          items.add(entry.getValue());
+          if (eventProcessor != null) {
+            eventProcessor.processEvent(path, query, entry.getKey());
+          }
+        });
 
     if (query.containsKey(KubernetesAttributesExtractor.NAME)) {
       if (!items.isEmpty()) {
@@ -179,26 +195,30 @@ public MockResponse handlePatch(RecordedRequest request) {
    */
   @Override
   public MockResponse handleDelete(String path) {
-    return handle(path, (p, pathAttributes, oldAttributes) -> {
-      String jsonStringOfResource = map.get(oldAttributes);
-      /*
-       * Potential performance improvement: The resource is unmarshalled and marshalled in other places (e.g., when creating a
-       * WatchEvent later).
-       * This could be avoided by storing the unmarshalled object (instead of a String) in the map.
-       */
-      final GenericKubernetesResource resource = Serialization.unmarshal(jsonStringOfResource, GenericKubernetesResource.class);
-      if (resource.getFinalizers().isEmpty()) {
-        // No finalizers left, actually remove the resource.
-        processEvent(path, pathAttributes, oldAttributes, null);
-        return;
-      } else if (!resource.isMarkedForDeletion()) {
-        // Mark the resource as deleted, but don't remove it yet (wait for finalizer-removal).
-        resource.getMetadata().setDeletionTimestamp(LocalDateTime.now().toString());
-        String updatedResource = Serialization.asJson(resource);
-        processEvent(path, pathAttributes, oldAttributes, updatedResource);
-      }
-      // else: if the resource is already marked for deletion and still has finalizers, do nothing.
-    });
+    lock.writeLock().lock();
+    try {
+      return handle(path, this::processDelete);
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+
+  private void processDelete(String path, AttributeSet pathAttributes, AttributeSet oldAttributes) {
+    String jsonStringOfResource = map.get(oldAttributes);
+    final GenericKubernetesResource resource = Serialization.unmarshal(jsonStringOfResource, GenericKubernetesResource.class);
+    if (resource.getFinalizers().isEmpty()) {
+      // No finalizers left, actually remove the resource.
+      processEvent(path, pathAttributes, oldAttributes, null, null);
+      return;
+    }
+    if (!resource.isMarkedForDeletion()) {
+      // Mark the resource as deleted, but don't remove it yet (wait for finalizer-removal).
+      resource.getMetadata().setDeletionTimestamp(LocalDateTime.now().toString());
+      String updatedResource = Serialization.asJson(resource);
+      processEvent(path, pathAttributes, oldAttributes, resource, updatedResource);
+      return;
+    }
+    // else: if the resource is already marked for deletion and still has finalizers, do nothing.
   }
 
   @Override
@@ -213,11 +233,9 @@ public AttributeSet getKey(String path) {
 
   @Override
   public Map.Entry<AttributeSet, String> findResource(AttributeSet attributes) {
-    synchronized (map) {
-      return map.entrySet().stream()
-          .filter(entry -> entry.getKey().matches(attributes))
-          .findFirst().orElse(null);
-    }
+    return map.entrySet().stream()
+        .filter(entry -> entry.getKey().matches(attributes))
+        .findFirst().orElse(null);
   }
 
   @Override
@@ -226,11 +244,16 @@ public boolean isStatusSubresourceEnabledForResource(String path) {
   }
 
   @Override
-  public void processEvent(String path, AttributeSet pathAttributes, AttributeSet oldAttributes, String newState) {
+  public void processEvent(String path, AttributeSet pathAttributes, AttributeSet oldAttributes,
+      GenericKubernetesResource resource, String newState) {
     String existing = map.remove(oldAttributes);
     AttributeSet newAttributes = null;
     if (newState != null) {
-      newAttributes = kubernetesAttributesExtractor.fromResource(newState);
+      if (resource != null) {
+        newAttributes = kubernetesAttributesExtractor.extract(resource);
+      } else {
+        newAttributes = kubernetesAttributesExtractor.fromResource(newState);
+      }
       // corner case - we need to get the plural from the path
       if (!newAttributes.containsKey(KubernetesAttributesExtractor.PLURAL)) {
         newAttributes = AttributeSet.merge(pathAttributes, newAttributes);
@@ -269,13 +292,9 @@ public MockResponse handleWatch(String path) {
       query = query.add(new Attribute("name", resourceName));
     }
     WatchEventsListener watchEventListener = new WatchEventsListener(context, query, watchEventListeners, LOGGER,
-        watch -> {
-          synchronized (map) {
-            map.entrySet().stream()
-                .filter(entry -> watch.attributeMatches(entry.getKey()))
-                .forEach(entry -> watch.sendWebSocketResponse(entry.getValue(), Action.ADDED));
-          }
-        });
+        watch -> map.entrySet().stream()
+            .filter(entry -> watch.attributeMatches(entry.getKey()))
+            .forEach(entry -> watch.sendWebSocketResponse(entry.getValue(), Action.ADDED)));
     watchEventListeners.add(watchEventListener);
     mockResponse.setSocketPolicy(SocketPolicy.KEEP_OPEN);
     return mockResponse.withWebSocketUpgrade(watchEventListener);

junit/kubernetes-server-mock/src/main/java/io/fabric8/kubernetes/client/server/mock/crud/KubernetesCrudDispatcherHandler.java

@@ -113,14 +113,6 @@ default GenericKubernetesResource validateRequestBody(String requestBody) throws
     return resource;
   }
 
-  static MockResponse process(RecordedRequest request, KubernetesCrudDispatcherHandler handler) {
-    try {
-      return handler.handle(request);
-    } catch (KubernetesCrudDispatcherException e) {
-      return new MockResponse().setResponseCode(e.getCode()).setBody(e.toStatusBody());
-    }
-  }
-
   static boolean isStatusPath(String path) {
     return path.endsWith("/" + STATUS);
   }

junit/kubernetes-server-mock/src/main/java/io/fabric8/kubernetes/client/server/mock/crud/KubernetesCrudPersistence.java

@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectReader;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
 import io.fabric8.kubernetes.client.server.mock.Resetable;
 import io.fabric8.kubernetes.client.utils.Serialization;
 import io.fabric8.mockwebserver.crud.AttributeSet;
@@ -47,7 +48,8 @@ public interface KubernetesCrudPersistence extends Resetable {
 
   boolean isStatusSubresourceEnabledForResource(String path);
 
-  void processEvent(String path, AttributeSet pathAttributes, AttributeSet oldAttributes, String newState);
+  void processEvent(String path, AttributeSet pathAttributes, AttributeSet oldAttributes, GenericKubernetesResource resource,
+      String newState);
 
   default JsonNode asNode(Map.Entry<AttributeSet, String> resource) throws KubernetesCrudDispatcherException {
     return asNode(resource.getValue());

junit/kubernetes-server-mock/src/main/java/io/fabric8/kubernetes/client/server/mock/crud/PatchHandler.java

@@ -89,9 +89,10 @@ public MockResponse handle(String path, String contentType, String requestBody)
     if (deserializedResource.isMarkedForDeletion() && deserializedResource.getFinalizers().isEmpty()) {
       // Delete the resource.
       updatedAsString = null;
+      deserializedResource = null;
     }
 
-    persistence.processEvent(path, query, currentResourceEntry.getKey(), updatedAsString);
+    persistence.processEvent(path, query, currentResourceEntry.getKey(), deserializedResource, updatedAsString);
     return new MockResponse().setResponseCode(HTTP_ACCEPTED).setBody(Utils.getNonNullOrElse(updatedAsString, ""));
   }
 

junit/kubernetes-server-mock/src/main/java/io/fabric8/kubernetes/client/server/mock/crud/PostHandler.java

@@ -62,7 +62,7 @@ public MockResponse handle(String path, String contentType, String requestBody)
       resource.getAdditionalProperties().remove(STATUS);
     }
     final String response = Serialization.asJson(resource);
-    persistence.processEvent(path, attributes, null, response);
+    persistence.processEvent(path, attributes, null, resource, response);
     return new MockResponse().setResponseCode(HttpURLConnection.HTTP_CREATED).setBody(response);
   }
 

junit/kubernetes-server-mock/src/main/java/io/fabric8/kubernetes/client/server/mock/crud/PutHandler.java

@@ -47,7 +47,7 @@ public MockResponse handle(String path, String contentType, String requestBody)
 
     // Delete the resource if it is marked for deletion and has no finalizers.
     if (resource.isMarkedForDeletion() && resource.getFinalizers().isEmpty()) {
-      persistence.processEvent(path, attributes, currentResourceEntry.getKey(), null);
+      persistence.processEvent(path, attributes, currentResourceEntry.getKey(), null, null);
       return new MockResponse().setResponseCode(HttpURLConnection.HTTP_OK);
     }
 
@@ -72,7 +72,7 @@ public MockResponse handle(String path, String contentType, String requestBody)
     }
     persistence.touchResourceVersion(currentResource, updatedResource);
     final String response = Serialization.asJson(updatedResource);
-    persistence.processEvent(path, attributes, currentResourceEntry.getKey(), response);
+    persistence.processEvent(path, attributes, currentResourceEntry.getKey(), null, response);
     return new MockResponse().setResponseCode(HttpURLConnection.HTTP_OK).setBody(response);
   }
 }