fix: make client side metrics tag in sync with server (#2401)

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://togithub.com/googleapis/java-bigtable/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)
- [ ] Rollback plan is reviewed and LGTMed
- [ ] All new data plane features have a completed end to end testing plan

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the [samples format](
https://togithub.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).

Author: mutianf

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsTracer.java

@@ -79,9 +79,9 @@ class BuiltinMetricsTracer extends BigtableTracer {
   private final AtomicInteger requestLeft = new AtomicInteger(0);
 
   // Monitored resource labels
-  private String tableId = "unspecified";
+  private String tableId = "<unspecified>";
   private String zone = "global";
-  private String cluster = "unspecified";
+  private String cluster = "<unspecified>";
 
   private final AtomicLong totalClientBlockingTime = new AtomicLong(0);
 

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/metrics/Util.java

@@ -24,8 +24,10 @@
 import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.bigtable.v2.AuthorizedViewName;
 import com.google.bigtable.v2.CheckAndMutateRowRequest;
+import com.google.bigtable.v2.GenerateInitialChangeStreamPartitionsRequest;
 import com.google.bigtable.v2.MutateRowRequest;
 import com.google.bigtable.v2.MutateRowsRequest;
+import com.google.bigtable.v2.ReadChangeStreamRequest;
 import com.google.bigtable.v2.ReadModifyWriteRowRequest;
 import com.google.bigtable.v2.ReadRowsRequest;
 import com.google.bigtable.v2.ResponseParams;
@@ -127,14 +129,18 @@ static String extractTableId(Object request) {
     } else if (request instanceof ReadModifyWriteRowRequest) {
       tableName = ((ReadModifyWriteRowRequest) request).getTableName();
       authorizedViewName = ((ReadModifyWriteRowRequest) request).getAuthorizedViewName();
+    } else if (request instanceof GenerateInitialChangeStreamPartitionsRequest) {
+      tableName = ((GenerateInitialChangeStreamPartitionsRequest) request).getTableName();
+    } else if (request instanceof ReadChangeStreamRequest) {
+      tableName = ((ReadChangeStreamRequest) request).getTableName();
     }
-    if (tableName == null && authorizedViewName == null) return "undefined";
-    if (tableName.isEmpty() && authorizedViewName.isEmpty()) return "undefined";
-    if (!tableName.isEmpty()) {
+    if (tableName != null && !tableName.isEmpty()) {
       return TableName.parse(tableName).getTable();
-    } else {
+    }
+    if (authorizedViewName != null && !authorizedViewName.isEmpty()) {
       return AuthorizedViewName.parse(authorizedViewName).getTable();
     }
+    return "<unspecified>";
   }
 
   /**

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/StreamingMetricsMetadataIT.java

@@ -167,9 +167,9 @@ public void testFailure() {
 
     assertThat(pointData)
         .comparingElementsUsing(POINT_DATA_CLUSTER_ID_CONTAINS)
-        .contains("unspecified");
+        .contains("<unspecified>");
     assertThat(pointData).comparingElementsUsing(POINT_DATA_ZONE_ID_CONTAINS).contains("global");
-    assertThat(clusterAttributes).contains("unspecified");
+    assertThat(clusterAttributes).contains("<unspecified>");
     assertThat(zoneAttributes).contains("global");
   }
 }

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/UnaryMetricsMetadataIT.java

@@ -182,7 +182,7 @@ public void testFailure() throws Exception {
 
     assertThat(pointData)
         .comparingElementsUsing(POINT_DATA_CLUSTER_ID_CONTAINS)
-        .contains("unspecified");
+        .contains("<unspecified>");
     assertThat(pointData).comparingElementsUsing(POINT_DATA_ZONE_ID_CONTAINS).contains("global");
     List<String> clusterAttributes =
         pointData.stream()
@@ -193,7 +193,7 @@ public void testFailure() throws Exception {
             .map(pd -> pd.getAttributes().get(BuiltinMetricsConstants.ZONE_ID_KEY))
             .collect(Collectors.toList());
 
-    assertThat(clusterAttributes).contains("unspecified");
+    assertThat(clusterAttributes).contains("<unspecified>");
     assertThat(zoneAttributes).contains("global");
   }
 }

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/metrics/BuiltinMetricsTracerTest.java

@@ -395,7 +395,7 @@ public void testGfeMetrics() {
             .put(STATUS_KEY, "UNAVAILABLE")
             .put(TABLE_ID_KEY, TABLE)
             .put(ZONE_ID_KEY, "global")
-            .put(CLUSTER_ID_KEY, "unspecified")
+            .put(CLUSTER_ID_KEY, "<unspecified>")
             .put(METHOD_KEY, "Bigtable.ReadRows")
             .put(CLIENT_NAME_KEY, CLIENT_NAME)
             .build();
@@ -549,7 +549,7 @@ public void testMutateRowAttemptsTagValues() {
             .put(STATUS_KEY, "UNAVAILABLE")
             .put(TABLE_ID_KEY, TABLE)
             .put(ZONE_ID_KEY, "global")
-            .put(CLUSTER_ID_KEY, "unspecified")
+            .put(CLUSTER_ID_KEY, "<unspecified>")
             .put(METHOD_KEY, "Bigtable.MutateRow")
             .put(CLIENT_NAME_KEY, CLIENT_NAME)
             .put(STREAMING_KEY, false)
@@ -619,7 +619,7 @@ public void testMutateRowsRpcError() {
             .put(STATUS_KEY, "NOT_FOUND")
             .put(TABLE_ID_KEY, BAD_TABLE_ID)
             .put(ZONE_ID_KEY, "global")
-            .put(CLUSTER_ID_KEY, "unspecified")
+            .put(CLUSTER_ID_KEY, "<unspecified>")
             .put(METHOD_KEY, "Bigtable.MutateRows")
             .put(CLIENT_NAME_KEY, CLIENT_NAME)
             .put(STREAMING_KEY, false)
@@ -640,7 +640,7 @@ public void testReadRowsAttemptsTagValues() {
             .put(STATUS_KEY, "UNAVAILABLE")
             .put(TABLE_ID_KEY, TABLE)
             .put(ZONE_ID_KEY, "global")
-            .put(CLUSTER_ID_KEY, "unspecified")
+            .put(CLUSTER_ID_KEY, "<unspecified>")
             .put(METHOD_KEY, "Bigtable.ReadRows")
             .put(CLIENT_NAME_KEY, CLIENT_NAME)
             .put(STREAMING_KEY, true)
@@ -751,7 +751,7 @@ public void testPermanentFailure() {
             .toBuilder()
             .put(STATUS_KEY, "NOT_FOUND")
             .put(TABLE_ID_KEY, BAD_TABLE_ID)
-            .put(CLUSTER_ID_KEY, "unspecified")
+            .put(CLUSTER_ID_KEY, "<unspecified>")
             .put(ZONE_ID_KEY, "global")
             .put(STREAMING_KEY, true)
             .put(METHOD_KEY, "Bigtable.ReadRows")
@@ -776,7 +776,7 @@ public void testRemainingDeadline() {
             .put(TABLE_ID_KEY, TABLE)
             .put(METHOD_KEY, "Bigtable.ReadRows")
             .put(ZONE_ID_KEY, "global")
-            .put(CLUSTER_ID_KEY, "unspecified")
+            .put(CLUSTER_ID_KEY, "<unspecified>")
             .put(STREAMING_KEY, true)
             .put(CLIENT_NAME_KEY, CLIENT_NAME)
             .build();