pass down metadata in other rpcs

Author: daniel-sanche

google/cloud/bigtable/data/_async/_mutate_rows.py

@@ -187,6 +187,7 @@ async def _run_attempt(self):
                 ),
                 timeout=next(self.timeout_generator),
                 retry=None,
+                metadata=[self._operation_metric.interceptor_metadata],
             )
             async for result_list in result_generator:
                 for result in result_list.entries:

google/cloud/bigtable/data/_async/_read_rows.py

@@ -156,6 +156,7 @@ def _read_rows_attempt(self) -> CrossSync.Iterable[Row]:
             self.request,
             timeout=next(self.attempt_timeout_gen),
             retry=None,
+            metadata=[self._operation_metric.interceptor_metadata],
         )
         chunked_stream = self.chunk_stream(gapic_stream)
         return self.merge_rows(chunked_stream)

google/cloud/bigtable/data/_async/client.py

@@ -1489,6 +1489,7 @@ async def mutate_row(
                 ),
                 timeout=attempt_timeout,
                 retry=None,
+                metadata=[operation_metric.interceptor_metadata],
             )
             return await CrossSync.retry_target(
                 target,