Allow for automated call graph generation to keep contexts unique (#249)

* Allow for automated call graph generation to keep contexts unique within
StdCallbackContext for replay deduping. Now developers do not need to
provide a name when calling services. There is a unique call graph
context maintained for each service call made. The context is request
aware, so different requests made will have their own independent
context for dedupe

Dedupe identical requests

```java
   ProgressEvent<Model, StdCallbackContext> result =
      initiator.translateToServiceRequest(m -> createRepository)
         .makeServiceCall((r, c) ->
            c.injectCredentialsAndInvokeV2(
                r, c.client()::createRepository))
         .success();

   ProgressEvent<Model, StdCallbackContext> result_2 =
       // make same request call
       initiator.translateToServiceRequest(m -> createRepository)
           .makeServiceCall((r, c) ->
              c.injectCredentialsAndInvokeV2(
                r, c.client()::createRepository))
           .success();

   assertThat(result).isEqualsTo(result_2);

```

* Prevent ConcurrentModificationExceptions in stabilize calls if they
access the map and attempt to modify it

Co-authored-by: diwakar <diwakar@amazon.com>

Author: dchakrav-github

src/main/java/software/amazon/cloudformation/proxy/AmazonWebServicesClientProxy.java

@@ -40,6 +40,7 @@
 import software.amazon.awssdk.awscore.AwsResponse;
 import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
 import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.core.SdkClient;
 import software.amazon.awssdk.core.exception.NonRetryableException;
 import software.amazon.awssdk.core.exception.RetryableException;
 import software.amazon.awssdk.core.pagination.sync.SdkIterable;
@@ -169,6 +170,12 @@ public RequestMaker<ClientT, ModelT, CallbackT> initiate(String callGraph) {
             return new CallContext<>(callGraph, client, model, callback);
         }
 
+        @Override
+        public <
+            RequestT> Caller<RequestT, ClientT, ModelT, CallbackT> translateToServiceRequest(Function<ModelT, RequestT> maker) {
+            return initiate("").translateToServiceRequest(maker);
+        }
+
         @Override
         public ModelT getResourceModel() {
             return model;
@@ -192,6 +199,11 @@ public <NewModelT> Initiator<ClientT, NewModelT, CallbackT> rebindModel(NewModel
             Preconditions.checkNotNull(callback, "cxt can not be null");
             return new StdInitiator<>(client, model, callback);
         }
+
+        @Override
+        public Logger getLogger() {
+            return AmazonWebServicesClientProxy.this.loggerProxy;
+        }
     }
 
     @Override
@@ -234,6 +246,22 @@ class CallContext<ClientT, ModelT, CallbackT extends StdCallbackContext>
             RequestT> Caller<RequestT, ClientT, ModelT, CallbackT> translateToServiceRequest(Function<ModelT, RequestT> maker) {
             return new Caller<RequestT, ClientT, ModelT, CallbackT>() {
 
+                private final CallGraphNameGenerator<ModelT, RequestT, ClientT,
+                    CallbackT> generator = (incoming, model_, reqMaker, client_, context_) -> {
+                        final RequestT request = reqMaker.apply(model_);
+                        String objectHash = String.valueOf(Objects.hashCode(request));
+                        String serviceName = (client_ == null
+                            ? ""
+                            : (client_ instanceof SdkClient)
+                                ? ((SdkClient) client_).serviceName()
+                                : client_.getClass().getSimpleName());
+                        String requestName = request != null ? request.getClass().getSimpleName().replace("Request", "") : "";
+                        String callGraph = serviceName + ":" + requestName + "-" + (incoming != null ? incoming : "") + "-"
+                            + objectHash;
+                        context_.request(callGraph, (ignored -> request)).apply(model_);
+                        return callGraph;
+                    };
+
                 @Override
                 public Caller<RequestT, ClientT, ModelT, CallbackT> backoffDelay(Delay delay) {
                     CallContext.this.delay = delay;
@@ -315,6 +343,8 @@ public ProgressEvent<ModelT, CallbackT> done(Callback<RequestT, ResponseT, Clien
                             // stabilization
                             // lambdas. This ensures that we call demux as necessary.
                             //
+                            final String callGraph = generator.callGraph(CallContext.this.callGraph, model, maker,
+                                client.client(), context);
                             Delay delay = override.getDelay(callGraph, CallContext.this.delay);
                             Function<ModelT, RequestT> reqMaker = context.request(callGraph, maker);
                             BiFunction<RequestT, ProxyClient<ClientT>, ResponseT> resMaker = context.response(callGraph, caller);
@@ -377,6 +407,7 @@ public ProgressEvent<ModelT, CallbackT> done(Callback<RequestT, ResponseT, Clien
                                     long remainingTime = getRemainingTimeInMillis();
                                     long localWait = next.toMillis() + 2 * elapsed + 100;
                                     if (remainingTime > localWait) {
+                                        loggerProxy.log("Waiting for " + next.getSeconds() + " for call " + callGraph);
                                         Uninterruptibles.sleepUninterruptibly(next.getSeconds(), TimeUnit.SECONDS);
                                         continue;
                                     }

src/main/java/software/amazon/cloudformation/proxy/CallChain.java

@@ -53,7 +53,7 @@ public interface CallChain {
      * @param <ModelT> the model object being worked on
      * @param <CallbackT> the callback context
      */
-    interface Initiator<ClientT, ModelT, CallbackT extends StdCallbackContext> {
+    interface Initiator<ClientT, ModelT, CallbackT extends StdCallbackContext> extends RequestMaker<ClientT, ModelT, CallbackT> {
         /**
          * Each service call must be first initiated. Every call is provided a separate
          * name called call graph. This is essential from both a tracing perspective as
@@ -74,6 +74,11 @@ interface Initiator<ClientT, ModelT, CallbackT extends StdCallbackContext> {
          */
         CallbackT getCallbackContext();
 
+        /**
+         * @return logger associated to log messages
+         */
+        Logger getLogger();
+
         /**
          * Can rebind a new model to the call chain while retaining the client and
          * callback context

src/main/java/software/amazon/cloudformation/proxy/CallGraphNameGenerator.java

@@ -0,0 +1,22 @@
+/*
+* Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+*
+* Licensed under the Apache License, Version 2.0 (the "License").
+* You may not use this file except in compliance with the License.
+* A copy of the License is located at
+*
+*  http://aws.amazon.com/apache2.0
+*
+* or in the "license" file accompanying this file. This file 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 software.amazon.cloudformation.proxy;
+
+import java.util.function.Function;
+
+@FunctionalInterface
+public interface CallGraphNameGenerator<ModelT, RequestT, ClientT, CallbackT extends StdCallbackContext> {
+    String callGraph(String incoming, ModelT model, Function<ModelT, RequestT> reqMaker, ClientT client, CallbackT context);
+}

src/main/java/software/amazon/cloudformation/proxy/StdCallbackContext.java

@@ -32,9 +32,13 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.function.BiFunction;
 import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
@@ -272,13 +276,67 @@ public <ResponseT> ResponseT response(String callGraph) {
         return (ResponseT) callGraphs.get(callGraph + ".response");
     }
 
+    @SuppressWarnings("unchecked")
+    public <RequestT> RequestT findFirstRequestByContains(String contains) {
+        return (RequestT) findFirst((key) -> key.contains(contains) && key.endsWith(".request"));
+    }
+
+    @SuppressWarnings("unchecked")
+    public <RequestT> List<RequestT> findAllRequestByContains(String contains) {
+        return (List<RequestT>) findAll((key) -> key.contains(contains) && key.endsWith(".request"));
+    }
+
+    @SuppressWarnings("unchecked")
+    public <ResponseT> ResponseT findFirstResponseByContains(String contains) {
+        return (ResponseT) findFirst((key) -> key.contains(contains) && key.endsWith(".response"));
+    }
+
+    @SuppressWarnings("unchecked")
+    public <ResponseT> List<ResponseT> findAllResponseByContains(String contains) {
+        return (List<ResponseT>) findAll((key) -> key.contains(contains) && key.endsWith(".response"));
+    }
+
+    Object findFirst(Predicate<String> contains) {
+        Objects.requireNonNull(contains);
+        return callGraphs.entrySet().stream().filter(e -> contains.test(e.getKey())).findFirst().map(Map.Entry::getValue)
+            .orElse(null);
+
+    }
+
+    List<Object> findAll(Predicate<String> contains) {
+        Objects.requireNonNull(contains);
+        return callGraphs.entrySet().stream().filter(e -> contains.test(e.getKey())).map(Map.Entry::getValue)
+            .collect(Collectors.toList());
+    }
+
     <RequestT, ResponseT, ClientT, ModelT, CallbackT extends StdCallbackContext>
         CallChain.Callback<RequestT, ResponseT, ClientT, ModelT, CallbackT, Boolean>
         stabilize(String callGraph, CallChain.Callback<RequestT, ResponseT, ClientT, ModelT, CallbackT, Boolean> callback) {
         return (request1, response1, client, model, context) -> {
-            Boolean result = (Boolean) callGraphs.computeIfAbsent(callGraph + ".stabilize",
-                (ign) -> callback.invoke(request1, response1, client, model, context) ? Boolean.TRUE : null);
-            return result != null ? Boolean.TRUE : Boolean.FALSE;
+            final String key = callGraph + ".stabilize";
+            Boolean result = (Boolean) callGraphs.getOrDefault(key, Boolean.FALSE);
+            if (!result) {
+                //
+                // The StdCallbackContext can be shared. However the call to stabilize for a
+                // given content
+                // is usually confined to one thread. If for some reason we spread that across
+                // threads, the
+                // worst that can happen is a double compute for stabilize. This isn't the
+                // intended pattern.
+                // Why are we changing it from computeIfAbsent pattern? For the callback we send
+                // in the
+                // StdCallbackContext which can be used to add things into context. That will
+                // lead to
+                // ConcurrentModificationExceptions when the compute running added things into
+                // context when
+                // needed
+                //
+                result = callback.invoke(request1, response1, client, model, context);
+                if (result) {
+                    callGraphs.put(key, Boolean.TRUE);
+                }
+            }
+            return result;
         };
     }