feat(bigtable): Async refresh dry run in parallel with sync refresh

bigtable/bigtable.go

@@ -32,6 +32,7 @@ import (
 	btopt "cloud.google.com/go/bigtable/internal/option"
 	"cloud.google.com/go/internal/trace"
 	gax "github.com/googleapis/gax-go/v2"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"google.golang.org/api/option"
 	"google.golang.org/api/option/internaloption"
@@ -95,6 +96,18 @@ func NewClient(ctx context.Context, project, instance string, opts ...option.Cli
 
 // NewClientWithConfig creates a new client with the given config.
 func NewClientWithConfig(ctx context.Context, project, instance string, config ClientConfig, opts ...option.ClientOption) (*Client, error) {
+	metricsProvider := config.MetricsProvider
+	if emulatorAddr := os.Getenv("BIGTABLE_EMULATOR_HOST"); emulatorAddr != "" {
+		// Do not emit metrics when emulator is being used
+		metricsProvider = NoopMetricsProvider{}
+	}
+
+	// Create a OpenTelemetry metrics configuration
+	metricsTracerFactory, err := newBuiltinMetricsTracerFactory(ctx, project, instance, config.AppProfile, metricsProvider, opts...)
+	if err != nil {
+		return nil, err
+	}
+
 	o, err := btopt.DefaultClientOptions(prodAddr, mtlsProdAddr, Scope, clientUserAgent)
 	if err != nil {
 		return nil, err
@@ -112,21 +125,24 @@ func NewClientWithConfig(ctx context.Context, project, instance string, config C
 	// Allow non-default service account in DirectPath.
 	o = append(o, internaloption.AllowNonDefaultServiceAccount(true))
 	o = append(o, opts...)
-	connPool, err := gtransport.DialPool(ctx, o...)
-	if err != nil {
-		return nil, fmt.Errorf("dialing: %w", err)
-	}
 
-	metricsProvider := config.MetricsProvider
-	if emulatorAddr := os.Getenv("BIGTABLE_EMULATOR_HOST"); emulatorAddr != "" {
-		// Do not emit metrics when emulator is being used
-		metricsProvider = NoopMetricsProvider{}
-	}
+	asyncRefreshMetricAttrs := metricsTracerFactory.clientAttributes
+	asyncRefreshMetricAttrs = append(asyncRefreshMetricAttrs,
+		attribute.String(metricLabelKeyTag, "async_refresh_dry_run"),
 
-	// Create a OpenTelemetry metrics configuration
-	metricsTracerFactory, err := newBuiltinMetricsTracerFactory(ctx, project, instance, config.AppProfile, metricsProvider, opts...)
+		// Table, cluster and zone are unknown at this point
+		// Use default values
+		attribute.String(monitoredResLabelKeyTable, ""),
+		attribute.String(monitoredResLabelKeyCluster, ""),
+		attribute.String(monitoredResLabelKeyZone, "global"),
+	)
+	o = append(o, internaloption.EnableAsyncRefreshDryRun(func() {
+		metricsTracerFactory.debugTags.Add(context.Background(), 1,
+			metric.WithAttributes(asyncRefreshMetricAttrs...))
+	}))
+	connPool, err := gtransport.DialPool(ctx, o...)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("dialing: %w", err)
 	}
 
 	return &Client{

bigtable/metrics.go

@@ -50,6 +50,7 @@ const (
 	metricLabelKeyAppProfile         = "app_profile"
 	metricLabelKeyMethod             = "method"
 	metricLabelKeyStatus             = "status"
+	metricLabelKeyTag                = "tag"
 	metricLabelKeyStreamingOperation = "streaming"
 	metricLabelKeyClientName         = "client_name"
 	metricLabelKeyClientUID          = "client_uid"
@@ -59,6 +60,7 @@ const (
 	metricNameAttemptLatencies   = "attempt_latencies"
 	metricNameServerLatencies    = "server_latencies"
 	metricNameRetryCount         = "retry_count"
+	metricNameDebugTags          = "debug_tags"
 
 	// Metric units
 	metricUnitMS    = "ms"
@@ -68,7 +70,7 @@ const (
 // These are effectively constant, but for testing purposes they are mutable
 var (
 	// duration between two metric exports
-	defaultSamplePeriod = 5 * time.Minute
+	defaultSamplePeriod = time.Minute
 
 	metricsErrorPrefix = "bigtable-metrics: "
 
@@ -79,7 +81,7 @@ var (
 		800.0, 1000.0, 2000.0, 5000.0, 10000.0, 20000.0, 50000.0, 100000.0, 200000.0,
 		400000.0, 800000.0, 1600000.0, 3200000.0}
 
-	// All the built-in metrics have same attributes except 'status' and 'streaming'
+	// All the built-in metrics have same attributes except 'tag', 'status' and 'streaming'
 	// These attributes need to be added to only few of the metrics
 	metricsDetails = map[string]metricInfo{
 		metricNameOperationLatencies: {
@@ -109,6 +111,12 @@ var (
 			},
 			recordedPerAttempt: true,
 		},
+		metricNameDebugTags: {
+			additionalAttrs: []string{
+				metricLabelKeyTag,
+			},
+			recordedPerAttempt: true,
+		},
 	}
 
 	// Generates unique client ID in the format go-<random UUID>@<hostname>
@@ -125,6 +133,7 @@ var (
 	// createExporterOptions takes Bigtable client options and returns exporter options
 	// Overwritten in tests
 	createExporterOptions = func(btOpts ...option.ClientOption) []option.ClientOption {
+		btOpts = append(btOpts, option.WithGRPCConnectionPool(4))
 		return btOpts
 	}
 )
@@ -148,6 +157,7 @@ type builtinMetricsTracerFactory struct {
 	serverLatencies    metric.Float64Histogram
 	attemptLatencies   metric.Float64Histogram
 	retryCount         metric.Int64Counter
+	debugTags          metric.Int64Counter
 }
 
 func newBuiltinMetricsTracerFactory(ctx context.Context, project, instance, appProfile string, metricsProvider MetricsProvider, opts ...option.ClientOption) (*builtinMetricsTracerFactory, error) {
@@ -253,6 +263,16 @@ func (tf *builtinMetricsTracerFactory) createInstruments(meter metric.Meter) err
 		metric.WithDescription("The number of additional RPCs sent after the initial attempt."),
 		metric.WithUnit(metricUnitCount),
 	)
+	if err != nil {
+		return err
+	}
+
+	// Create debug_tags
+	tf.debugTags, err = meter.Int64Counter(
+		metricNameDebugTags,
+		metric.WithDescription("The number of additional RPCs sent after the initial attempt."),
+		metric.WithUnit(metricUnitCount),
+	)
 	return err
 }
 
@@ -271,6 +291,7 @@ type builtinMetricsTracer struct {
 	instrumentServerLatencies    metric.Float64Histogram
 	instrumentAttemptLatencies   metric.Float64Histogram
 	instrumentRetryCount         metric.Int64Counter
+	instrumentDebugTags          metric.Int64Counter
 
 	tableName   string
 	method      string
@@ -363,6 +384,7 @@ func (tf *builtinMetricsTracerFactory) createBuiltinMetricsTracer(ctx context.Co
 		instrumentServerLatencies:    tf.serverLatencies,
 		instrumentAttemptLatencies:   tf.attemptLatencies,
 		instrumentRetryCount:         tf.retryCount,
+		instrumentDebugTags:          tf.debugTags,
 
 		tableName:   tableName,
 		isStreaming: isStreaming,