Resolved merge conflict

Author: TetyanaYahodska

.github/CODEOWNERS

@@ -54,7 +54,7 @@
 
 # Cloud SDK Databases & Data Analytics teams
 # ---* Cloud Native DB
-/bigtable                        @GoogleCloudPlatform/java-samples-reviewers @yoshi-approver @GoogleCloudPlatform/cloud-samples-reviewers @GoogleCloudPlatform/cloud-native-db-dpes
+/bigtable                        @GoogleCloudPlatform/java-samples-reviewers @yoshi-approver @GoogleCloudPlatform/cloud-samples-reviewers @GoogleCloudPlatform/cloud-native-db-dpes @GoogleCloudPlatform/bigtable-eng
 /memorystore                     @GoogleCloudPlatform/java-samples-reviewers @yoshi-approver @GoogleCloudPlatform/cloud-samples-reviewers
 /spanner                         @GoogleCloudPlatform/java-samples-reviewers @yoshi-approver @GoogleCloudPlatform/cloud-samples-reviewers @GoogleCloudPlatform/api-spanner-java
 # ---* Cloud Storage

bigtable/bigtable-proxy/README.md

@@ -48,6 +48,7 @@ in a project your choosing. The metrics will be published under the namespace
 * `bigtableproxy.client.channel.count` Number of open channels
 * `bigtableproxy.client.call.max_outstanding_count` Maximum number of concurrent RPCs in a single
   minute window
+* `bigtableproxy.presence` Counts number of proxy processes (emit 1 per process).
 
 ## Requirements
 
@@ -63,14 +64,27 @@ in a project your choosing. The metrics will be published under the namespace
 # Build the binary
 mvn package
 
-# use the binary
+# unpack the binary on the proxy host
 unzip target/bigtable-proxy-0.0.1-SNAPSHOT-bin.zip
 cd bigtable-proxy-0.0.1-SNAPSHOT
+
+# Verify that the proxy has require permissions using an existing table. Please note that the table
+# data will not be modified, however a test metric will be written.
+./bigtable-verify.sh  \
+  --bigtable-project-id=$BIGTABLE_PROJECT_ID \
+  --bigtable-instance-id=$BIGTABLE_INSTANCE_ID \
+  --bigtable-table-id=$BIGTABLE_TABLE_ID \
+  --metrics-project-id=$METRICS_PROJECT_ID
+
+# Then start the proxy on the specified port. The proxy can forward requests for multiple
+# Bigtable projects/instances/tables. However it will export health metrics to a single project
+# specified by `metrics-project-id`. 
 ./bigtable-proxy.sh \
   --listen-port=1234 \
   --metrics-project-id=SOME_GCP_PROJECT
 
-export BIGTABLE_EMULATOR_HOST=1234
+# Start your application, and redirect the bigtable client to connect to the local proxy. 
+export BIGTABLE_EMULATOR_HOST="localhost:1234"
 path/to/application/with/bigtable/client
 ```
 

bigtable/bigtable-proxy/pom.xml

@@ -21,14 +21,19 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 
     <!-- dep versions -->
+    <!-- Overall -->
     <libraries-bom.version>26.50.0</libraries-bom.version>
-
+    <!-- OpenTelemetry -->
     <otel.version>1.44.1</otel.version>
+    <otel-contrib.version>1.41.0-alpha</otel-contrib.version>
+    <shared-resourcemapping.version>0.33.0</shared-resourcemapping.version>
     <exporter-metrics.version>0.33.0</exporter-metrics.version>
+    <!-- Utility -->
     <slf4j.version>2.0.16</slf4j.version>
     <logback.version>1.5.12</logback.version>
     <auto-value.version>1.11.0</auto-value.version>
     <picocli.version>4.7.6</picocli.version>
+    <!-- Test -->
     <junit.version>4.13.2</junit.version>
     <truth.version>1.4.4</truth.version>
   </properties>
@@ -125,6 +130,25 @@
       <artifactId>exporter-metrics</artifactId>
       <version>${exporter-metrics.version}</version>
     </dependency>
+    <!-- Workaround shared-configuration incorrectly marking google-cloud-core as a test dep -->
+    <dependency>
+      <groupId>com.google.cloud</groupId>
+      <artifactId>google-cloud-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.opentelemetry.contrib</groupId>
+      <artifactId>opentelemetry-gcp-resources</artifactId>
+      <version>${otel-contrib.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk-extension-autoconfigure-spi</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.google.cloud.opentelemetry</groupId>
+      <artifactId>shared-resourcemapping</artifactId>
+      <version>${shared-resourcemapping.version}</version>
+    </dependency>
 
     <!-- Logging -->
     <dependency>

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/Main.java

@@ -17,8 +17,7 @@
 package com.google.cloud.bigtable.examples.proxy;
 
 import com.google.cloud.bigtable.examples.proxy.commands.Serve;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import com.google.cloud.bigtable.examples.proxy.commands.Verify;
 import org.slf4j.bridge.SLF4JBridgeHandler;
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
@@ -27,10 +26,10 @@
  * Main entry point for proxy commands under {@link
  * com.google.cloud.bigtable.examples.proxy.commands}.
  */
-@Command(subcommands = {Serve.class}, name = "bigtable-proxy")
+@Command(
+    subcommands = {Serve.class, Verify.class},
+    name = "bigtable-proxy")
 public final class Main {
-  private static final Logger LOGGER = LoggerFactory.getLogger(Main.class);
-
   public static void main(String[] args) {
     SLF4JBridgeHandler.install();
     new CommandLine(new Main()).execute(args);

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/channelpool/DataChannel.java

@@ -17,22 +17,28 @@
 package com.google.cloud.bigtable.examples.proxy.channelpool;
 
 import com.google.bigtable.v2.BigtableGrpc;
-import com.google.bigtable.v2.BigtableGrpc.BigtableFutureStub;
 import com.google.bigtable.v2.PingAndWarmRequest;
 import com.google.bigtable.v2.PingAndWarmResponse;
+import com.google.cloud.bigtable.examples.proxy.metrics.CallLabels;
 import com.google.cloud.bigtable.examples.proxy.metrics.Metrics;
 import com.google.cloud.bigtable.examples.proxy.metrics.Tracer;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 import io.grpc.CallCredentials;
 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
+import io.grpc.ClientCall.Listener;
 import io.grpc.ConnectivityState;
 import io.grpc.Deadline;
 import io.grpc.ExperimentalApi;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
+import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
+import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
@@ -50,7 +56,7 @@ public class DataChannel extends ManagedChannel {
   private final ManagedChannel inner;
   private final Metrics metrics;
   private final ResourceCollector resourceCollector;
-  private final BigtableFutureStub warmingStub;
+  private final CallCredentials callCredentials;
   private final ScheduledFuture<?> antiIdleTask;
 
   private final AtomicBoolean closed = new AtomicBoolean();
@@ -65,6 +71,7 @@ public DataChannel(
       Metrics metrics) {
     this.resourceCollector = resourceCollector;
 
+    this.callCredentials = callCredentials;
     inner =
         ManagedChannelBuilder.forAddress(endpoint, port)
             .userAgent(userAgent)
@@ -76,8 +83,6 @@ public DataChannel(
     this.metrics = metrics;
 
     try {
-      warmingStub = BigtableGrpc.newFutureStub(inner).withCallCredentials(callCredentials);
-
       warm();
     } catch (RuntimeException e) {
       try {
@@ -107,10 +112,8 @@ private void warm() {
       return;
     }
 
-    BigtableFutureStub timedStub = warmingStub.withDeadline(Deadline.after(1, TimeUnit.MINUTES));
-
     List<ListenableFuture<PingAndWarmResponse>> futures =
-        requests.stream().map(timedStub::pingAndWarm).collect(Collectors.toList());
+        requests.stream().map(this::sendPingAndWarm).collect(Collectors.toList());
 
     int successCount = 0;
     int failures = 0;
@@ -148,6 +151,62 @@ private void warm() {
     }
   }
 
+  private ListenableFuture<PingAndWarmResponse> sendPingAndWarm(PingAndWarmRequest request) {
+    CallLabels callLabels =
+        CallLabels.create(
+            BigtableGrpc.getPingAndWarmMethod(),
+            Optional.of("bigtableproxy"),
+            Optional.of(request.getName()),
+            Optional.of(request.getAppProfileId()));
+    Tracer tracer = new Tracer(metrics, callLabels);
+
+    CallOptions callOptions =
+        CallOptions.DEFAULT
+            .withCallCredentials(callCredentials)
+            .withDeadline(Deadline.after(1, TimeUnit.MINUTES));
+    callOptions = tracer.injectIntoCallOptions(callOptions);
+
+    ClientCall<PingAndWarmRequest, PingAndWarmResponse> call =
+        inner.newCall(BigtableGrpc.getPingAndWarmMethod(), callOptions);
+
+    Metadata metadata = new Metadata();
+    metadata.put(
+        CallLabels.REQUEST_PARAMS,
+        String.format(
+            "name=%s&app_profile_id=%s",
+            URLEncoder.encode(request.getName(), StandardCharsets.UTF_8),
+            URLEncoder.encode(request.getAppProfileId(), StandardCharsets.UTF_8)));
+
+    SettableFuture<PingAndWarmResponse> f = SettableFuture.create();
+    call.start(
+        new Listener<>() {
+          @Override
+          public void onMessage(PingAndWarmResponse response) {
+            if (!f.set(response)) {
+              // TODO: set a metric
+              LOGGER.warn("PingAndWarm returned multiple responses");
+            }
+          }
+
+          @Override
+          public void onClose(Status status, Metadata trailers) {
+            tracer.onCallFinished(status);
+
+            if (status.isOk()) {
+              f.setException(new IllegalStateException("PingAndWarm was missing a response"));
+            } else {
+              f.setException(status.asRuntimeException());
+            }
+          }
+        },
+        metadata);
+    call.sendMessage(request);
+    call.halfClose();
+    call.request(Integer.MAX_VALUE);
+
+    return f;
+  }
+
   @Override
   public ManagedChannel shutdown() {
     if (closed.compareAndSet(false, true)) {
@@ -208,9 +267,13 @@ public void enterIdle() {
   @Override
   public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
       MethodDescriptor<RequestT, ResponseT> methodDescriptor, CallOptions callOptions) {
-    Optional.ofNullable(Tracer.extractTracerFromCallOptions(callOptions))
-        .map(Tracer::getCallLabels)
-        .ifPresent(resourceCollector::collect);
+    Tracer tracer =
+        Optional.ofNullable(Tracer.extractTracerFromCallOptions(callOptions))
+            .orElseThrow(
+                () ->
+                    new IllegalStateException(
+                        "DataChannel failed to extract Tracer from CallOptions"));
+    resourceCollector.collect(tracer.getCallLabels());
 
     return inner.newCall(methodDescriptor, callOptions);
   }

bigtable/bigtable-proxy/src/main/java/com/google/cloud/bigtable/examples/proxy/commands/Serve.java

@@ -106,6 +106,9 @@ void start() throws IOException {
         new InstrumentedCallCredentials(MoreCallCredentials.from(credentials));
 
     if (metrics == null) {
+      // InstrumentedCallCredentials expect to only be called when a Tracer is available in the
+      // CallOptions. This is only true for DataChannel pingAndWarm and things invoked by
+      // ProxyHandler. MetricsImpl does not do this, so it must get undecorated credentials.
       metrics = new MetricsImpl(credentials, metricsProjectId);
     }