feat(bigquery/storage/managedwriter): refine connection metrics (#8324)

This PR changes metrics instrumentation in two ways:

The AppendClientOpenView is now tagged with an error dimension, so that
failures to open a connection are clearer.  We use rpc status for the
value, with the expectation that non-rpc errors are tagged as Unknown.

A new metric (and view), AppendRequestReconnectsView provides additional
visibility into when errors during Send trigger reconnection of the
underlying connection.  We attempt to also attribute this to the origin
writer, so it may be tagged by error and source stream.

Towards: #8311

bigquery/storage/managedwriter/connection.go

@@ -136,9 +136,18 @@ func (cp *connectionPool) mergeCallOptions(co *connection) []gax.CallOption {
 func (cp *connectionPool) openWithRetry(co *connection) (storagepb.BigQueryWrite_AppendRowsClient, chan *pendingWrite, error) {
 	r := &unaryRetryer{}
 	for {
-		recordStat(cp.ctx, AppendClientOpenCount, 1)
 		arc, err := cp.open(co.ctx, cp.mergeCallOptions(co)...)
+		metricCtx := cp.ctx
+		if err == nil {
+			// accumulate AppendClientOpenCount for the success case.
+			recordStat(metricCtx, AppendClientOpenCount, 1)
+		}
 		if err != nil {
+			if tagCtx, tagErr := tag.New(cp.ctx, tag.Insert(keyError, grpcstatus.Code(err).String())); tagErr == nil {
+				metricCtx = tagCtx
+			}
+			// accumulate AppendClientOpenCount for the error case.
+			recordStat(metricCtx, AppendClientOpenCount, 1)
 			bo, shouldRetry := r.Retry(err)
 			if shouldRetry {
 				recordStat(cp.ctx, AppendClientOpenRetryCount, 1)
@@ -396,6 +405,14 @@ func (co *connection) lockingAppend(pw *pendingWrite) error {
 	}
 	if err != nil {
 		if shouldReconnect(err) {
+			metricCtx := co.ctx // start with the ctx that must be present
+			if pw.writer != nil {
+				metricCtx = pw.writer.ctx // the writer ctx bears the stream/origin tagging, so prefer it.
+			}
+			if tagCtx, tagErr := tag.New(metricCtx, tag.Insert(keyError, grpcstatus.Code(err).String())); tagErr == nil {
+				metricCtx = tagCtx
+			}
+			recordStat(metricCtx, AppendRequestReconnects, 1)
 			// if we think this connection is unhealthy, force a reconnect on the next send.
 			co.reconnect = true
 		}

bigquery/storage/managedwriter/instrumentation.go

@@ -61,6 +61,10 @@ var (
 	// It is EXPERIMENTAL and subject to change or removal without notice.
 	AppendRequestErrors = stats.Int64(statsPrefix+"append_request_errors", "Number of append requests that yielded immediate error", stats.UnitDimensionless)
 
+	// AppendRequestReconnects is a measure of the number of times that sending an append request triggered reconnect.
+	// It is EXPERIMENTAL and subject to change or removal without notice.
+	AppendRequestReconnects = stats.Int64(statsPrefix+"append_reconnections", "Number of append rows reconnections", stats.UnitDimensionless)
+
 	// AppendRequestRows is a measure of the number of append rows sent.
 	// It is EXPERIMENTAL and subject to change or removal without notice.
 	AppendRequestRows = stats.Int64(statsPrefix+"append_rows", "Number of append rows sent", stats.UnitDimensionless)
@@ -105,6 +109,10 @@ var (
 	// It is EXPERIMENTAL and subject to change or removal without notice.
 	AppendRequestErrorsView *view.View
 
+	// AppendRequestReconnectsView is a cumulative sum of AppendRequestReconnects.
+	// It is EXPERIMENTAL and subject to change or removal without notice.
+	AppendRequestReconnectsView *view.View
+
 	// AppendRequestRowsView is a cumulative sum of AppendRows.
 	// It is EXPERIMENTAL and subject to change or removal without notice.
 	AppendRequestRowsView *view.View
@@ -127,12 +135,13 @@ var (
 )
 
 func init() {
-	AppendClientOpenView = createSumView(stats.Measure(AppendClientOpenCount))
+	AppendClientOpenView = createSumView(stats.Measure(AppendClientOpenCount), keyError)
 	AppendClientOpenRetryView = createSumView(stats.Measure(AppendClientOpenRetryCount))
 
 	AppendRequestsView = createSumView(stats.Measure(AppendRequests), keyStream, keyDataOrigin)
 	AppendRequestBytesView = createSumView(stats.Measure(AppendRequestBytes), keyStream, keyDataOrigin)
 	AppendRequestErrorsView = createSumView(stats.Measure(AppendRequestErrors), keyStream, keyDataOrigin, keyError)
+	AppendRequestReconnectsView = createSumView(stats.Measure(AppendRequestReconnects), keyStream, keyDataOrigin, keyError)
 	AppendRequestRowsView = createSumView(stats.Measure(AppendRequestRows), keyStream, keyDataOrigin)
 
 	AppendResponsesView = createSumView(stats.Measure(AppendResponses), keyStream, keyDataOrigin)
@@ -147,6 +156,7 @@ func init() {
 		AppendRequestsView,
 		AppendRequestBytesView,
 		AppendRequestErrorsView,
+		AppendRequestReconnectsView,
 		AppendRequestRowsView,
 
 		AppendResponsesView,

bigquery/storage/managedwriter/integration_test.go

@@ -1066,7 +1066,7 @@ func testInstrumentation(ctx context.Context, t *testing.T, mwClient *Client, bq
 
 	// metric to key tag names
 	wantTags := map[string][]string{
-		"cloud.google.com/go/bigquery/storage/managedwriter/stream_open_count":       nil,
+		"cloud.google.com/go/bigquery/storage/managedwriter/stream_open_count":       {"error"},
 		"cloud.google.com/go/bigquery/storage/managedwriter/stream_open_retry_count": nil,
 		"cloud.google.com/go/bigquery/storage/managedwriter/append_requests":         {"streamID"},
 		"cloud.google.com/go/bigquery/storage/managedwriter/append_request_bytes":    {"streamID"},