routes peer checks to api

flow/connectors/bigquery/bigquery.go

@@ -205,6 +205,11 @@ func (c *BigQueryConnector) Close() error {
 
 // ConnectionActive returns true if the connection is active.
 func (c *BigQueryConnector) ConnectionActive() bool {
+	_, err := c.client.Dataset(c.datasetID).Metadata(c.ctx)
+	if err != nil {
+		return false
+	}
+
 	return c.client != nil
 }
 

flow/connectors/core.go

@@ -223,7 +223,12 @@ func GetConnector(ctx context.Context, peer *protos.Peer) (Connector, error) {
 			return nil, fmt.Errorf("missing snowflake config for %s peer %s", peer.Type.String(), peer.Name)
 		}
 		return connsnowflake.NewSnowflakeConnector(ctx, sfConfig)
-
+	case protos.DBType_S3:
+		s3Config := peer.GetS3Config()
+		if s3Config == nil {
+			return nil, fmt.Errorf("missing s3 config for %s peer %s", peer.Type.String(), peer.Name)
+		}
+		return conns3.NewS3Connector(ctx, s3Config)
 	case protos.DBType_SQLSERVER:
 		sqlServerConfig := peer.GetSqlserverConfig()
 		if sqlServerConfig == nil {

flow/connectors/eventhub/eventhub.go

@@ -66,7 +66,7 @@ func (c *EventHubConnector) Close() error {
 }
 
 func (c *EventHubConnector) ConnectionActive() bool {
-	return true
+	return c.pgMetadata.Ping()
 }
 
 func (c *EventHubConnector) InitializeTableSchema(req map[string]*protos.TableSchema) error {

flow/connectors/external_metadata/store.go

@@ -52,6 +52,16 @@ func NewPostgresMetadataStore(ctx context.Context, pgConfig *protos.PostgresConf
 	}, nil
 }
 
+func (p *PostgresMetadataStore) Ping() bool {
+	err := p.pool.Ping(p.ctx)
+	if err != nil {
+		log.Errorf("failed to connect to metadata db: %v", err)
+		return false
+	}
+
+	return true
+}
+
 func (p *PostgresMetadataStore) Close() error {
 	if p.pool != nil {
 		p.pool.Close()

flow/connectors/s3/s3.go

@@ -141,8 +141,9 @@ func ValidCheck(s3Client *s3.S3, bucketURL string, metadataDB *metadataStore.Pos
 }
 
 func (c *S3Connector) ConnectionActive() bool {
-	_, err := c.client.ListBuckets(nil)
-	return err == nil
+	_, listErr := c.client.ListBuckets(nil)
+	pinged := c.pgMetadata.Ping()
+	return listErr == nil && pinged
 }
 
 func (c *S3Connector) NeedsSetupMetadataTables() bool {