Check if its pgconn.ConnectError{} instead of doing ping.

Author: 0xgouda

internal/sinks/ca.crt

@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDPzCCAiegAwIBAgIUeENQlQFVH5h7HszFJLLWo+KCQwQwDQYJKoZIhvcNAQEL
+BQAwEjEQMA4GA1UEAwwHcGd3YXRjaDAeFw0yNTA2MTEwNTI1MDJaFw0zNTA2MDkw
+NTI1MDJaMBIxEDAOBgNVBAMMB3Bnd2F0Y2gwggEiMA0GCSqGSIb3DQEBAQUAA4IB
+DwAwggEKAoIBAQDtTW+kyvb3Y5OaYlriKp8HkHt95lfOgxQNZQfiREfEyLWU59bx
+0ZIvFmejK38Qc0dlca9d+5tEkxotsbggJflLljfmnzhxsuZpr8SjmDd1m8XSo0IA
+oDlVbKO6SZMlsyq3QrAOYAjG1LTPlATqvGAOs9NfFonjwoPXCjIwSfa+wexe5dRD
+gJ114AKXw3ck5ZQ4Pw+w5ylgNSfVl548WY9DSOA+6HlZ17MYA1qmMOTwKae5fmsc
+xlRPoIV3EJrKas7VlbebDOXOSXDV+9aMW6ox1xanUUDUgabzkzntfmuOttaIX1g1
+nSMuHa7EEQF7lxgdg9OU8i/jygdlGcgBUYjtAgMBAAGjgYwwgYkwDAYDVR0TBAUw
+AwEB/zAdBgNVHQ4EFgQUKGJs7OXINd2WL6X2meH90eINoJQwTQYDVR0jBEYwRIAU
+KGJs7OXINd2WL6X2meH90eINoJShFqQUMBIxEDAOBgNVBAMMB3Bnd2F0Y2iCFHhD
+UJUBVR+Yex7MxSSy1qPigkMEMAsGA1UdDwQEAwIBBjANBgkqhkiG9w0BAQsFAAOC
+AQEABdY/4rsgMu+sCqEdacNzHqAz9X1ew37y1UONngm/7LPqbQrzzg/fBvOOJLcd
+IzMJPtpdwokPYOW29jw/hY4R1RWr8012zc8Z0GsuDR7I/Z2Hww7tzYhf1H5mjy1d
+eQDhHNpsSb5pHLoPft5O0sT/0WqAlKWPb2KmSoAio8jE2BSUTK3ZgE0yJIikONon
+HCWOlNCWx+RsyPoRnQqbpVa+SmGBqpiyHchpZ8sFPe+pgPu+8u921lJ0PRvmfp7L
+4YZIaM8LQAV8FWk2VLXmsqYUJYYLAXCG6Unkx1oIOtq1AyAoXHCl/3hKbCeXIrgA
+Cs5qN+ZUHRdKff5gFpraKtHKkw==
+-----END CERTIFICATE-----

internal/sinks/postgres.go

@@ -438,8 +438,8 @@ func (pgw *PostgresWriter) flush(msgs []metrics.MeasurementEnvelope) {
 		rowsBatched += n
 		if err != nil {
 			logger.Error(err)
-			if err := pgw.sinkDb.Ping(pgw.ctx); err != nil {
-				logger.WithError(err).Error("Sink db is not reachable, dropping cached measurements")
+			if _, ok := err.(*pgconn.ConnectError); ok {
+				logger.Errorf("Sink DB not reachable, dropping %d cached measurements", len(msgs))
 				break
 			}
 			if PgError, ok := err.(*pgconn.PgError); ok {

internal/sinks/postgres_test.go

@@ -1083,19 +1083,25 @@ func TestEnsureMetricDbnameTime_IdempotentAcrossRestarts(t *testing.T) {
 }
 
 func TestFlush_CopyFromFailsAndPingSucceeds(t *testing.T) {
-	a := assert.New(t)
-	conn, err := pgxmock.NewPool()
-	a.NoError(err)
-	defer conn.Close()
-
-	pgw := &PostgresWriter{
-		ctx:                      ctx,
-		sinkDb:                   conn,
-		opts:                     &CmdOpts{PartitionInterval: "1 hour"},
-		partitionMapMetricDbname: make(map[string]map[string]ExistingPartitionInfo),
-		metricSchema:             DbStorageSchemaPostgres,
+	r := require.New(t)
+
+	pgContainer, pgTearDown, err := testutil.SetupPostgresContainer()
+	r.NoError(err)
+
+	connStr, err := pgContainer.ConnectionString(ctx, "sslmode=disable")
+	r.NoError(err)
+
+	opts := &CmdOpts{
+		PartitionInterval:   "1 day",
+		RetentionInterval:   "7 days",
+		MaintenanceInterval: "12 hours",
+		BatchingDelay:       time.Second,
 	}
 
+	pgw, err := NewPostgresWriter(ctx, connStr, opts)
+	r.NoError(err)
+	pgTearDown()
+
 	now := time.Now()
 	msgs := []metrics.MeasurementEnvelope{
 		{
@@ -1107,13 +1113,5 @@ func TestFlush_CopyFromFailsAndPingSucceeds(t *testing.T) {
 		},
 	}
 
-	conn.ExpectQuery("(?i)SELECT.*ensure_partition_metric_dbname_time").
-		WithArgs("test_metric", "test_db", pgxmock.AnyArg(), "1 hour").
-		WillReturnRows(pgxmock.NewRows([]string{"start_time", "end_time"}).AddRow(now.Add(-time.Hour), now.Add(time.Hour)))
-	conn.ExpectCopyFrom(pgx.Identifier{"test_metric"}, targetColumns[:]).WillReturnError(errors.New("copy failed"))
-	conn.ExpectPing().WillReturnError(nil)
-
 	pgw.flush(msgs)
-
-	a.NoError(conn.ExpectationsWereMet())
 }