Skip to content

Multi-cluster support: MySQL and PostgreSQL simultaneously#75

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
renecannao merged 1 commit into from
Apr 8, 2026
Merged

Multi-cluster support: MySQL and PostgreSQL simultaneously #75

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions go/db/generate_patches.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -617,4 +617,9 @@ var generateSQLPatches = []string{
database_instance
ADD COLUMN replication_group_primary_port smallint(5) unsigned NOT NULL DEFAULT 0 AFTER replication_group_primary_host
`,
`
ALTER TABLE
database_instance
ADD COLUMN provider_type varchar(20) CHARACTER SET ascii NOT NULL DEFAULT 'mysql' AFTER replication_group_primary_port
`,
}
55 changes: 54 additions & 1 deletion go/inst/analysis_dao.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -51,12 +51,54 @@
recentInstantAnalysis = cache.New(time.Duration(config.RecoveryPollSeconds*2)*time.Second, time.Second)
}

// hasPostgreSQLInstances checks whether the orchestrator backend has any
// instances with provider_type='postgresql'. This is used to decide whether
// PostgreSQL analysis should be run alongside MySQL analysis.
func hasPostgreSQLInstances(clusterName string) bool {
query := `SELECT 1 FROM database_instance WHERE provider_type = 'postgresql'`
args := sqlutils.Args()
if clusterName != "" {
query += ` AND cluster_name = ?`
args = append(args, clusterName)
}
query += ` LIMIT 1`
found := false
_ = db.QueryOrchestrator(query, args, func(m sqlutils.RowMap) error {
found = true
return nil
})
return found
}

// hasMySQLInstances checks whether the orchestrator backend has any
// instances with provider_type='mysql' (or empty, for backward compat).
func hasMySQLInstances(clusterName string) bool {
query := `SELECT 1 FROM database_instance WHERE provider_type IN ('mysql', '')`
args := sqlutils.Args()
if clusterName != "" {
query += ` AND cluster_name = ?`
args = append(args, clusterName)
}
query += ` LIMIT 1`
found := false
_ = db.QueryOrchestrator(query, args, func(m sqlutils.RowMap) error {
found = true
return nil
})
return found
}

// GetReplicationAnalysis will check for replication problems (dead master; unreachable master; etc)
func GetReplicationAnalysis(clusterName string, hints *ReplicationAnalysisHints) ([]ReplicationAnalysis, error) {
if config.Config.ProviderType == "postgresql" {
// When the global config is set to "postgresql" and there are no MySQL instances,
// use the pure PG path (backward compatibility).
if config.Config.ProviderType == "postgresql" && !hasMySQLInstances(clusterName) {
return GetPostgreSQLReplicationAnalysis(clusterName, hints)
}

// In multi-cluster mode, run both MySQL and PG analysis and merge results.
// First run MySQL analysis (below), then append PG results if any PG instances exist.

result := []ReplicationAnalysis{}

args := sqlutils.Args(config.Config.ReasonableLockedSemiSyncMasterSeconds, ValidSecondsFromSeenToLastAttemptedCheck(), config.Config.ReasonableReplicationLagSeconds, clusterName)
Expand DownExpand Up@@ -738,6 +780,17 @@
return result, log.Errore(err)
}
// TODO: result, err = getConcensusReplicationAnalysis(result)

// In multi-cluster mode, also run PostgreSQL analysis and merge results
if hasPostgreSQLInstances(clusterName) {
pgResult, pgErr := GetPostgreSQLReplicationAnalysis(clusterName, hints)
if pgErr != nil {
log.Errore(pgErr)

Check failure on line 788 in go/inst/analysis_dao.go

View workflow job for this annotation

GitHub Actions/ lint

Error return value of `log.Errore` is not checked (errcheck)
} else {
result = append(result, pgResult...)
}
}

return result, log.Errore(err)
}

Expand Down
3 changes: 3 additions & 0 deletions go/inst/instance.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -152,6 +152,9 @@ type Instance struct {

// Query string provider
QSP QueryStringProvider

// ProviderType indicates the database engine: "mysql" (default) or "postgresql"
ProviderType string
}

// NewInstance creates a new, empty instance
Expand Down
43 changes: 40 additions & 3 deletions go/inst/instance_dao.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -60,6 +60,32 @@ const (
var instanceReadChan = make(chan bool, backendDBConcurrency)
var instanceWriteChan = make(chan bool, backendDBConcurrency)
Comment on lines 60 to 61

@gemini-code-assistgemini-code-assistBotApr 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To avoid frequent and redundant database queries to the orchestrator backend during the discovery process, consider introducing a cache for the instance provider type. This is especially important as detectProviderType is called multiple times per discovery cycle for each instance.

Suggested change
varinstanceReadChan=make(chanbool, backendDBConcurrency)
varinstanceWriteChan=make(chanbool, backendDBConcurrency)
varinstanceReadChan=make(chanbool, backendDBConcurrency)
varinstanceWriteChan=make(chanbool, backendDBConcurrency)
varproviderTypeCache=cache.New(time.Hour, time.Hour)


// detectProviderType determines whether an instance is MySQL or PostgreSQL.
// It checks (in order):
// 1. The orchestrator backend DB for a previously stored provider_type
// 2. Port-based heuristic (5432 -> postgresql)
// 3. The global config.Config.ProviderType as fallback (backward compat)
func detectProviderType(instanceKey *InstanceKey) string {
// Check if we already know this instance's provider type from the backend DB
query := `SELECT provider_type FROM database_instance WHERE hostname = ? AND port = ?`
var providerType string
err := db.QueryOrchestrator(query, sqlutils.Args(instanceKey.Hostname, instanceKey.Port), func(m sqlutils.RowMap) error {
providerType = m.GetString("provider_type")
return nil
})
if err == nil && providerType != "" {
return providerType
}

// Port-based heuristic for new instances
if instanceKey.Port == 5432 {
return "postgresql"
}

// Fall back to global config (backward compatibility)
return config.Config.ProviderType
}
Comment on lines +68 to +87

@gemini-code-assistgemini-code-assistBotApr 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The detectProviderType function performs a synchronous database query every time it's called. Implementing caching here will significantly reduce the load on the orchestrator backend database during discovery. Additionally, the logic can be simplified to consolidate the return paths.

funcdetectProviderType(instanceKey*InstanceKey) string {
cacheKey:=instanceKey.StringCode()
ifval, found:=providerTypeCache.Get(cacheKey); found {
returnval.(string)
}
varproviderTypestring// Check if we already know this instance's provider type from the backend DBquery:=`SELECT provider_type FROM database_instance WHERE hostname = ? AND port = ?`err:=db.QueryOrchestrator(query, sqlutils.Args(instanceKey.Hostname, instanceKey.Port), func(m sqlutils.RowMap) error {
providerType=m.GetString("provider_type")
returnnil
})
iferr==nil&&providerType!="" {
// Found in DB
} elseifinstanceKey.Port==5432 {
providerType="postgresql"
} else {
providerType=config.Config.ProviderType
}
providerTypeCache.Set(cacheKey, providerType, cache.DefaultExpiration)
returnproviderType
}


// InstancesByCountReplicas is a sortable type for Instance
type InstancesByCountReplicas [](*Instance)

Expand DownExpand Up@@ -226,7 +252,8 @@ func logReadTopologyInstanceError(instanceKey *InstanceKey, hint string, err err
// server and writes the result synchronously to the orchestrator
// backend.
func ReadTopologyInstance(instanceKey *InstanceKey) (*Instance, error) {
if config.Config.ProviderType == "postgresql" {
providerType := detectProviderType(instanceKey)
if providerType == "postgresql" {
inst, err := ReadPostgreSQLTopologyInstance(instanceKey)
if err != nil {
// Mark the instance as checked-but-not-seen so analysis detects it as dead
Expand DownExpand Up@@ -360,7 +387,7 @@ func expectReplicationThreadsState(instance *Instance, instanceKey *InstanceKey,
// - timing information can be collected for the stages performed.
func ReadTopologyInstanceBufferable(instanceKey *InstanceKey, bufferWrites bool, latency *stopwatch.NamedStopwatch) (inst *Instance, skipped bool, err error) {
// PostgreSQL path: delegate to PG-specific discovery and skip MySQL logic entirely
if config.Config.ProviderType == "postgresql" {
if detectProviderType(instanceKey) == "postgresql" {
if latency != nil {
latency.Start("instance")
}
Expand DownExpand Up@@ -1222,7 +1249,7 @@ func ReadReplicationGroupPrimary(instance *Instance) (err error) {
func ReadInstanceClusterAttributes(instance *Instance) (err error) {
// PostgreSQL provider sets ClusterName directly during discovery.
// Skip the master-lookup logic which is MySQL-specific.
if config.Config.ProviderType == "postgresql" {
if instance.ProviderType == "postgresql" || detectProviderType(&instance.Key) == "postgresql" {

@gemini-code-assistgemini-code-assistBotApr 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Calling detectProviderType here is redundant if instance.ProviderType is already populated (which it should be after discovery or loading from the DB). Even with caching, it's better to avoid the extra function call and map lookup. You can also use this opportunity to ensure instance.ProviderType is initialized if it happens to be empty.

ifinstance.ProviderType=="" {
instance.ProviderType=detectProviderType(&instance.Key)
}
ifinstance.ProviderType=="postgresql" {

return nil
}
var masterOrGroupPrimaryInstanceKey InstanceKey
Expand DownExpand Up@@ -1462,6 +1489,10 @@ func readInstanceRow(m sqlutils.RowMap) *Instance {
Port: m.GetInt("replication_group_primary_port")}
_ = instance.ReplicationGroupMembers.ReadJson(m.GetString("replication_group_members"))
//instance.ReplicationGroup = m.GetString("replication_group_")
instance.ProviderType = m.GetString("provider_type")
if instance.ProviderType == "" {
instance.ProviderType = "mysql"
}

// problems
if !instance.IsLastCheckValid {
Expand DownExpand Up@@ -2780,6 +2811,7 @@ func mkInsertOdkuForInstances(instances []*Instance, instanceWasActuallyFound bo
"replication_group_members",
"replication_group_primary_host",
"replication_group_primary_port",
"provider_type",
}

var values = make([]string, len(columns))
Expand DownExpand Up@@ -2873,6 +2905,11 @@ func mkInsertOdkuForInstances(instances []*Instance, instanceWasActuallyFound bo
args = append(args, instance.ReplicationGroupMembers.ToJSONString())
args = append(args, instance.ReplicationGroupPrimaryInstanceKey.Hostname)
args = append(args, instance.ReplicationGroupPrimaryInstanceKey.Port)
providerType := instance.ProviderType
if providerType == "" {
providerType = "mysql"
}
args = append(args, providerType)
}

sql, err := mkInsertOdku("database_instance", columns, values, len(instances), insertIgnore)
Expand Down
1 change: 1 addition & 0 deletions go/inst/instance_dao_postgresql.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -57,6 +57,7 @@ func ReadPostgreSQLTopologyInstance(instanceKey *InstanceKey) (*Instance, error)
}
instance.Version = parsePostgreSQLVersion(version)
instance.FlavorName = "PostgreSQL"
instance.ProviderType = "postgresql"

// Check whether this instance is in recovery (standby) or is a primary
var inRecovery bool
Expand Down
Loading
Loading