services: implement a lb policy agnostic call metric recorder for backend applications (#6004)

* services: implement a lb policy agnostic call metric recorder for backend applications.

* Renamed snapshot() to finalizeAndDump() and make it package-private with an internal accessor class.

* Added Javadoc link for Context.

* Added ExperimentalApi annotation.

* Added since annotations.

services/src/main/java/io/grpc/services/CallMetricRecorder.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2019 The gRPC Authors
+ *
+ * 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 io.grpc.services;
+
+import io.grpc.Context;
+import io.grpc.ExperimentalApi;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.concurrent.ThreadSafe;
+
+/**
+ * Utility to record call metrics for load-balancing. One instance per call.
+ */
+@ExperimentalApi("https://github.com/grpc/grpc-java/issues/6012")
+@ThreadSafe
+public final class CallMetricRecorder {
+  private static final CallMetricRecorder NOOP = new CallMetricRecorder().disable();
+
+  static final Context.Key<CallMetricRecorder> CONTEXT_KEY =
+      Context.key("io.grpc.services.CallMetricRecorder");
+
+  private final AtomicReference<ConcurrentHashMap<String, Double>> metrics =
+      new AtomicReference<>();
+  private volatile boolean disabled;
+
+  CallMetricRecorder() {
+  }
+
+  /**
+   * Returns the call metric recorder attached to the current {@link Context}.  If there is none,
+   * returns a no-op recorder.
+   *
+   * <p><strong>IMPORTANT:</strong>It returns the recorder specifically for the current RPC call.
+   * <b>DO NOT</b> save the returned object or share it between different RPC calls.
+   *
+   * <p><strong>IMPORTANT:</strong>It must be called under the {@link Context} under which the RPC
+   * handler was called.  If it is called from a different thread, the Context must be propagated to
+   * the same thread, e.g., with {@link Context#wrap(Runnable)}.
+   *
+   * @since 1.23.0
+   */
+  public static CallMetricRecorder getCurrent() {
+    CallMetricRecorder recorder = CONTEXT_KEY.get();
+    return recorder != null ? recorder : NOOP;
+  }
+
+  /**
+   * Records a call metric measurement.  If RPC has already finished, this method is no-op.
+   *
+   * <p>A latter record will overwrite its former name-sakes.
+   *
+   * @return this recorder object
+   * @since 1.23.0
+   */
+  public CallMetricRecorder recordCallMetric(String name, double value) {
+    if (disabled) {
+      return this;
+    }
+    if (metrics.get() == null) {
+      // The chance of race of creation of the map should be very small, so it should be fine
+      // to create these maps that might be discarded.
+      metrics.compareAndSet(null, new ConcurrentHashMap<String, Double>());
+    }
+    metrics.get().put(name, value);
+    return this;
+  }
+
+  /**
+   * Returns all save metric values. No more metric values will be recorded after this method is
+   * called. Calling this method multiple times returns the same collection of metric values.
+   *
+   * @return a map containing all saved metric name-value pairs.
+   */
+  Map<String, Double> finalizeAndDump() {
+    disabled = true;
+    Map<String, Double> savedMetrics = metrics.get();
+    if (savedMetrics == null) {
+      return Collections.emptyMap();
+    }
+    return Collections.unmodifiableMap(savedMetrics);
+  }
+
+  /**
+   * Turn this recorder into a no-op one.
+   */
+  private CallMetricRecorder disable() {
+    disabled = true;
+    return this;
+  }
+}

services/src/main/java/io/grpc/services/InternalCallMetricRecorder.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2019 The gRPC Authors
+ *
+ * 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 io.grpc.services;
+
+import io.grpc.Internal;
+import java.util.Map;
+
+/**
+ * Internal {@link CallMetricRecorder} accessor.  This is intended for usage internal to the gRPC
+ * team.  If you *really* think you need to use this, contact the gRPC team first.
+ */
+@Internal
+public final class InternalCallMetricRecorder {
+
+  // Prevent instantiation.
+  private InternalCallMetricRecorder() {
+  }
+
+  public static Map<String, Double> finalizeAndDump(CallMetricRecorder recorder) {
+    return recorder.finalizeAndDump();
+  }
+}

services/src/test/java/io/grpc/services/CallMetricRecorderTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2019 The gRPC Authors
+ *
+ * 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 io.grpc.services;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import java.util.Map;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link BinlogHelper}. */
+@RunWith(JUnit4.class)
+public class CallMetricRecorderTest {
+
+  private final CallMetricRecorder recorder = new CallMetricRecorder();
+
+  @Test
+  public void dumpGivesEmptyResultWhenNoSavedMetricValues() {
+    assertThat(recorder.finalizeAndDump()).isEmpty();
+  }
+
+  @Test
+  public void dumpDumpsAllSavedMetricValues() {
+    recorder.recordCallMetric("ssd", 154353.423);
+    recorder.recordCallMetric("cpu", 0.1367);
+    recorder.recordCallMetric("mem", 1437.34);
+
+    Map<String, Double> dump = recorder.finalizeAndDump();
+    assertThat(dump)
+        .containsExactly("ssd", 154353.423, "cpu", 0.1367, "mem", 1437.34);
+  }
+
+  @Test
+  public void noMetricsRecordedAfterSnapshot() {
+    Map<String, Double> initDump = recorder.finalizeAndDump();
+    recorder.recordCallMetric("cpu", 154353.423);
+    assertThat(recorder.finalizeAndDump()).isEqualTo(initDump);
+  }
+}