fix: an export that runs out of its own deadline is never a permanent error; robust e2e export test; one-line warning

e2e_metrics_test.go

@@ -180,10 +180,25 @@ func TestMetricsE2EExport(t *testing.T) {
 	if quotaProject != "" {
 		opts = append(opts, cloudsqlconn.WithQuotaProject(quotaProject))
 	}
+	// Export on the periodic path, which the SDK gives a 30 s export
+	// timeout, and keep the Dialer open until the data is visible in Cloud
+	// Monitoring. Relying on the final flush in Close is not robust: that
+	// flush is bounded by cloudsqlconn.MetricShutdownTimeoutForTest, and the
+	// first export from a cold identity has to mint a token first (on a CI
+	// runner that is GitHub OIDC -> STS -> IAM), which has been seen to take
+	// longer than the bound. Exports after the first land inside the
+	// Monitoring API's minimum write interval and are dropped, which the
+	// connector treats as benign (debug-only).
+	prevInterval := tel.DefaultExportInterval
+	tel.DefaultExportInterval = 5 * time.Second
+	t.Cleanup(func() { tel.DefaultExportInterval = prevInterval })
+
 	d, err := cloudsqlconn.NewDialer(ctx, opts...)
 	if err != nil {
 		t.Fatal(err)
 	}
+	closeDialer := sync.OnceValue(d.Close)
+	t.Cleanup(func() { _ = closeDialer() })
 	clientUID := cloudsqlconn.DialerIDForTest(d)
 
 	c, err := d.Dial(ctx, postgresConnName)
@@ -194,10 +209,6 @@ func TestMetricsE2EExport(t *testing.T) {
 	if err := c.Close(); err != nil {
 		t.Fatal(err)
 	}
-	// Close flushes the exporters synchronously (bounded by cloudsqlconn.MetricShutdownTimeoutForTest).
-	if err := d.Close(); err != nil {
-		t.Fatal(err)
-	}
 
 	// If the identity cannot write or read metrics, say exactly what to grant
 	// and skip rather than fail: the rest of the e2e suite is still valid.
@@ -207,7 +218,7 @@ func TestMetricsE2EExport(t *testing.T) {
 	if os.Getenv("CI") == "true" {
 		report = t.Fatalf
 	}
-	if failed := logger.find("built-in metrics export failed"); len(failed) > 0 {
+	reportExportFailure := func(failed []string) {
 		principal := "<the e2e service account>"
 		if p := cloudsqlconn.DetectPrincipalForTest(ctx); p != "" {
 			principal = p
@@ -223,9 +234,6 @@ func TestMetricsE2EExport(t *testing.T) {
 			"  gcloud projects add-iam-policy-binding %s --member=serviceAccount:%s --role=roles/serviceusage.serviceUsageConsumer\n"+
 			"debug: %s", cn.Project(), cn.Project(), principal, cn.Project(), principal, billing, billing, principal, failed[0])
 	}
-	if w := warn.String(); w != "" {
-		t.Fatalf("unexpected standard-logger output during a healthy export: %q", w)
-	}
 
 	// Poll Cloud Monitoring for this process's client_uid. Unlike the write,
 	// timeSeries.list requires billing on the project it is billed to, and
@@ -298,8 +306,22 @@ func TestMetricsE2EExport(t *testing.T) {
 		if all {
 			break
 		}
+		if failed := logger.find("built-in metrics export failed"); len(failed) > 0 {
+			reportExportFailure(failed)
+		}
 		time.Sleep(10 * time.Second)
 	}
+	// Everything has been exported; Close's final flush is dropped as too
+	// frequent, which must stay silent on the standard logger.
+	if err := closeDialer(); err != nil {
+		t.Fatal(err)
+	}
+	if failed := logger.find("built-in metrics export failed"); len(failed) > 0 {
+		reportExportFailure(failed)
+	}
+	if w := warn.String(); w != "" {
+		t.Fatalf("unexpected standard-logger output during a healthy export: %q", w)
+	}
 	for name, ok := range want {
 		if !ok {
 			t.Errorf("metric %s did not arrive in Cloud Monitoring within 3 minutes", name)

internal/tel/tel.go

@@ -766,6 +766,9 @@ type sharedClientExporter struct {
 	mu             sync.Mutex
 	lastErr        string
 	suspendedUntil time.Time
+	// exportedOnce records that at least one export succeeded (first
+	// success is debug-logged once).
+	exportedOnce bool
 	// backoff is the current (un-jittered) suspension length; zero until the
 	// first failure, reset to zero on success.
 	backoff time.Duration
@@ -788,6 +791,13 @@ func (e *sharedClientExporter) Export(ctx context.Context, rm *metricdata.Resour
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	if err == nil {
+		if !e.exportedOnce {
+			// One debug line per instance when the pipeline is known to
+			// work, so a user with a debug logger can tell metrics are
+			// flowing without checking Cloud Monitoring.
+			e.logger.Debugf(ctx, "[%v] built-in metrics: first export succeeded", e.name)
+			e.exportedOnce = true
+		}
 		if e.lastErr != "" {
 			e.logger.Debugf(ctx, "[%v] built-in metrics export recovered", e.name)
 			e.lastErr = ""
@@ -815,7 +825,15 @@ func (e *sharedClientExporter) Export(ctx context.Context, rm *metricdata.Resour
 			"metric's minimum write interval (30s) of the previous export; the last partial interval is not recorded: %v", e.name, err)
 		return nil
 	}
-	permanent := isPermanentExportError(err)
+	// If the export ran out of its own time budget (the SDK's export timeout
+	// for periodic exports, the Shutdown bound for the final flush), the
+	// Monitoring API never gave a verdict, whatever code the error carries.
+	// A slow first token fetch, for example, surfaces as Unauthenticated
+	// ("per-RPC creds failed ... context deadline exceeded"). Never treat
+	// that as permanent: it would suspend export for the permanent-error
+	// backoff and print a misleading "grant roles/monitoring.metricWriter"
+	// warning to a user whose IAM is fine. Back off as transient instead.
+	permanent := ctx.Err() == nil && isPermanentExportError(err)
 	initial, maxBackoff := TransientExportErrorInitialBackoff, TransientExportErrorMaxBackoff
 	if permanent {
 		initial, maxBackoff = PermanentExportErrorInitialBackoff, PermanentExportErrorMaxBackoff
@@ -831,9 +849,12 @@ func (e *sharedClientExporter) Export(ctx context.Context, rm *metricdata.Resour
 	e.suspendedUntil = time.Now().Add(jitter(e.backoff))
 	if permanent {
 		permanentErrorOnce.Do(func() {
-			warnLogger.Printf("cloudsqlconn: built-in metrics export failed for %s: %v. %s "+
+			// gRPC status errors can span several lines ("error details: ...");
+			// keep the warning to one physical line so log collectors treat
+			// it as a single entry.
+			warnLogger.Printf("cloudsqlconn: built-in metrics export failed for %s: %s. %s "+
 				"Retries back off exponentially up to %v; further failures are reported via the debug logger only.",
-				e.name, err, e.remediationHint(err), PermanentExportErrorMaxBackoff)
+				e.name, singleLine(err), e.remediationHint(err), PermanentExportErrorMaxBackoff)
 		})
 	}
 	if msg := err.Error(); msg != e.lastErr {
@@ -905,6 +926,12 @@ func isServiceDisabledError(err error) bool {
 		strings.Contains(st.Message(), "Cloud Monitoring API has not been used")
 }
 
+// singleLine renders err on one line: embedded newlines (as in gRPC status
+// details) are collapsed to single spaces.
+func singleLine(err error) string {
+	return strings.Join(strings.Fields(err.Error()), " ")
+}
+
 // serviceDisabledProject returns the project the Monitoring API reports as
 // having the service disabled (the consumer the call was billed to), or
 // fallback when the error does not carry it.
@@ -957,9 +984,8 @@ func isInsufficientScopeError(err error) bool {
 
 // isCanceledExport reports whether err means the export was canceled by its
 // own context (context.Canceled, or gRPC Canceled), as happens to an
-// in-flight export when the reader shuts down. Deadline errors are not
-// included: a Monitoring API call that times out is a real transient failure
-// and must back off like any other.
+// in-flight export when the reader shuts down. An export that ran out of its
+// own deadline is not "canceled": Export backs it off as transient.
 func isCanceledExport(err error) bool {
 	return errors.Is(err, context.Canceled) || grpcstatus.Code(err) == codes.Canceled
 }

internal/tel/tel_test.go

@@ -218,7 +218,7 @@ func (m *mockServer) Verify(t *testing.T, wantProjectName, wantResourceType, wan
 
 }
 
-func setupMockServer(t *testing.T) (*mockServer, *grpc.ClientConn, func()) {
+func setupMockServer(t *testing.T, dialOpts ...grpc.DialOption) (*mockServer, *grpc.ClientConn, func()) {
 	t.Helper()
 
 	s := grpc.NewServer()
@@ -230,9 +230,10 @@ func setupMockServer(t *testing.T) (*mockServer, *grpc.ClientConn, func()) {
 		_ = s.Serve(lis)
 	}()
 
-	conn, err := grpc.NewClient("passthrough://bufnet",
+	conn, err := grpc.NewClient("passthrough://bufnet", append([]grpc.DialOption{
 		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return lis.Dial() }),
-		grpc.WithTransportCredentials(insecure.NewCredentials()))
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	}, dialOpts...)...)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -722,14 +723,17 @@ func TestExportFailuresAreQuietAndBackedOff(t *testing.T) {
 		t.Fatalf("export did not resume after backoff; server saw %d requests", got)
 	}
 	deadline = time.Now().Add(time.Second)
-	for time.Now().Before(deadline) && len(logger.snapshot()) < 1 {
+	for time.Now().Before(deadline) && len(logger.snapshot()) < 2 {
 		time.Sleep(10 * time.Millisecond)
 	}
 	// Give any erroneous extra lines a chance to appear before asserting.
 	time.Sleep(3 * tel.DefaultExportInterval)
 	lines = logger.snapshot()
-	if len(lines) != 1 || !strings.Contains(lines[0], "recovered") {
-		t.Fatalf("want exactly one debug line (recovery) after the failure line, got %q", lines)
+	// The first success is logged once ("first export succeeded"), then the
+	// recovery once; the further successful exports in the 3 intervals
+	// above must add nothing.
+	if len(lines) != 2 || !strings.Contains(lines[0], "first export succeeded") || !strings.Contains(lines[1], "recovered") {
+		t.Fatalf("want exactly two debug lines (first success, recovery) after the failure line, got %q", lines)
 	}
 	if w := warnBuf.String(); strings.Count(w, "cloudsqlconn: built-in metrics export failed") != 1 {
 		t.Fatalf("recovery must not add standard-logger output; got %q", w)
@@ -1286,6 +1290,55 @@ func TestCanceledExportIsNotAFailure(t *testing.T) {
 	}
 }
 
+// TestExportDeadlineIsTransient reproduces a slow first token fetch: the
+// export's own deadline expires while per-RPC credentials are still being
+// minted, and gRPC reports it as Unauthenticated. Because the Monitoring API
+// never gave a verdict, it must be backed off as transient: no permanent
+// suspension and no "grant roles/monitoring.metricWriter" warning on the
+// standard logger. (A real Unauthenticated, with time left on the context,
+// stays permanent; see TestIsPermanentExportError and
+// TestRemediationHintMatchesCause.)
+func TestExportDeadlineIsTransient(t *testing.T) {
+	oldInterval := tel.DefaultExportInterval
+	tel.DefaultExportInterval = time.Hour
+	t.Cleanup(func() { tel.DefaultExportInterval = oldInterval })
+	var warnBuf lockedBuffer
+	restore := tel.SetWarnLoggerForTest(log.New(&warnBuf, "", 0))
+	t.Cleanup(restore)
+	logger := &recordingLogger{}
+
+	slowCreds := func(ctx context.Context, _ string, _, _ any, _ *grpc.ClientConn, _ grpc.UnaryInvoker, _ ...grpc.CallOption) error {
+		<-ctx.Done()
+		return grpcstatus.Errorf(codes.Unauthenticated,
+			"transport: per-RPC creds failed due to error: credentials: invalid response when retrieving subject token: %v", ctx.Err())
+	}
+	_, conn, cleanup := setupMockServer(t, grpc.WithUnaryInterceptor(slowCreds))
+	t.Cleanup(cleanup)
+	mClient, err := monitoring.NewMetricClient(t.Context(), option.WithGRPCConn(conn), option.WithTokenSource(nullTokenSource{}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	mr := tel.NewMetricRecorder(ctx, logger, mClient, tel.Config{
+		Enabled: true, Version: "1.2.3", ClientUID: "uid", ResourceContainer: "p", ResourceID: "p:i", Region: "r",
+	})
+	mr.RecordOpenConnectionCount(ctx, tel.Attributes{IPType: "public", AttemptStatus: tel.ConnectSuccess})
+	sctx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
+	defer cancel()
+	_ = mr.Shutdown(sctx) // the final flush runs out of its deadline while "minting a token"
+
+	lines := logger.snapshot()
+	if len(lines) != 1 || !strings.Contains(lines[0], "retrying with backoff") || !strings.Contains(lines[0], "Unauthenticated") {
+		t.Fatalf("want exactly one transient (retrying with backoff) debug line for the Unauthenticated error, got %q", lines)
+	}
+	if strings.Contains(lines[0], "suspending export") {
+		t.Fatalf("a timed-out export must not be treated as permanent, got %q", lines[0])
+	}
+	if w := warnBuf.String(); w != "" {
+		t.Fatalf("a timed-out export must not reach the standard logger, got %q", w)
+	}
+}
+
 // TestRemediationHintMatchesCause verifies that the once-per-process warning
 // names the actual fix for each permanent-error cause, not always the IAM
 // role: a disabled Monitoring API, a universe-domain mismatch and an
@@ -1357,3 +1410,46 @@ func TestConnectorVersionValue(t *testing.T) {
 		}
 	}
 }
+
+// TestPermanentErrorWarningIsOneLine verifies that the once-per-process
+// warning is a single physical line even when the gRPC error carries
+// multi-line details, so log collectors see one entry.
+func TestPermanentErrorWarningIsOneLine(t *testing.T) {
+	oldInterval := tel.DefaultExportInterval
+	tel.DefaultExportInterval = 10 * time.Millisecond
+	t.Cleanup(func() { tel.DefaultExportInterval = oldInterval })
+	var warnBuf lockedBuffer
+	restore := tel.SetWarnLoggerForTest(log.New(&warnBuf, "", 0))
+	t.Cleanup(restore)
+
+	mock, conn, cleanup := setupMockServer(t)
+	t.Cleanup(cleanup)
+	st, err := grpcstatus.New(codes.PermissionDenied, "Permission monitoring.timeSeries.create denied (or the resource may not exist).").
+		WithDetails(&errdetails.DebugInfo{Detail: "Permission monitoring.timeSeries.create denied on projects/123: no extra explanation needed."})
+	if err != nil {
+		t.Fatal(err)
+	}
+	mock.setFailureStatus(st)
+	mClient, err := monitoring.NewMetricClient(t.Context(), option.WithGRPCConn(conn), option.WithTokenSource(nullTokenSource{}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	mr := tel.NewMetricRecorder(ctx, nullLogger{t}, mClient, tel.Config{
+		Enabled: true, Version: "1.2.3", ClientUID: "uid", ResourceContainer: "proj", ResourceID: "proj:inst", Region: "r",
+	})
+	mr.RecordOpenConnectionCount(ctx, tel.Attributes{IPType: "public", AttemptStatus: tel.ConnectSuccess})
+	time.Sleep(100 * time.Millisecond)
+	_ = mr.Shutdown(ctx)
+
+	w := strings.TrimRight(warnBuf.String(), "\n")
+	if w == "" {
+		t.Fatal("expected a warning")
+	}
+	if n := strings.Count(w, "\n"); n != 0 {
+		t.Fatalf("warning spans %d extra line(s): %q", n, w)
+	}
+	if !strings.Contains(w, "DebugInfo") || !strings.Contains(w, "grant roles/monitoring.metricWriter") {
+		t.Fatalf("warning lost content: %q", w)
+	}
+}