Skip to content

Commit 6cd86fa

Browse files
renecannaorenecannao
authored
Merge pull request #79 from ProxySQL/issue77-named-channels-phase2
Named channels phases 4-6: API, failover, docs & tests
2 parents 56293ce + e08fb45 commit 6cd86fa

5 files changed

Lines changed: 342 additions & 2 deletions

File tree

‎docs/named-channels.md‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
# Named Replication Channels
2+
3+
MySQL 5.7+ supports multi-source replication, where a single replica can replicate from multiple masters simultaneously. Each replication connection is identified by a **named channel**. Orchestrator supports discovery, management, and failover of instances that use named replication channels.
4+
5+
## Overview
6+
7+
In a traditional single-source topology, a replica has exactly one `SHOW SLAVE STATUS` row. With multi-source replication, each channel appears as a separate row in `SHOW SLAVE STATUS`, each with its own IO thread, SQL thread, binlog coordinates, and lag.
8+
9+
Orchestrator handles multi-source replicas by:
10+
11+
1. Discovering all replication channels on each instance.
12+
2. Selecting one channel as the **managed channel** for topology purposes.
13+
3. Using channel-aware SQL operations (`FOR CHANNEL`) during failover and topology changes.
14+
4. Preserving non-managed channels during promotion and recovery.
15+
16+
## Discovery
17+
18+
When orchestrator reads an instance's replication status via `SHOW SLAVE STATUS`, it parses all rows. Each row becomes a `ChannelStatus` entry stored in the instance's `ReplicationChannels` slice.
19+
20+
For single-source instances (0 or 1 channels), behavior is identical to previous versions. For multi-source instances (2+ channels), orchestrator selects a **canonical channel** to represent the instance's primary replication relationship:
21+
22+
1. The default channel (empty name `""`) is preferred.
23+
2. Otherwise, the first non-Group-Replication-internal channel is selected.
24+
3. As a fallback, the first channel is used.
25+
26+
The selected channel's status (IO/SQL thread state, binlog coordinates, lag, master key, etc.) populates the instance's top-level replication fields. This ensures backward compatibility with all existing topology logic.
27+
28+
## Managed Channel Name
29+
30+
The `ManagedChannelName` field on an instance indicates which channel orchestrator manages for topology operations. When this field is empty, all operations use standard SQL without a `FOR CHANNEL` clause (single-source behavior).
31+
32+
When `ManagedChannelName` is set (multi-source instances), orchestrator appends `FOR CHANNEL '<name>'` to replication commands, ensuring only the managed channel is affected. Other channels remain untouched.
33+
34+
## Channel-Aware Operations
35+
36+
The following operations are channel-aware:
37+
38+
- `STOP SLAVE` / `STOP REPLICA` -- stops only the managed channel
39+
- `START SLAVE` / `START REPLICA` -- starts only the managed channel
40+
- `CHANGE MASTER TO` -- targets only the managed channel
41+
- `RESET SLAVE` / `RESET REPLICA` -- resets only the managed channel
42+
43+
All of these append `FOR CHANNEL '<name>'` when a channel name is specified.
44+
45+
## Failover with Multi-Source Replicas
46+
47+
### Dead Master Recovery
48+
49+
When a master dies and orchestrator initiates recovery (`recoverDeadMaster`), replicas are regrouped using GTID or Pseudo-GTID. The underlying `StopReplication`, `ChangeMasterTo`, and `StartReplication` calls all respect the `ManagedChannelName`, so only the dead master's channel is affected on multi-source replicas. Other channels (replicating from other masters) continue operating normally.
50+
51+
### Candidate Selection
52+
53+
A multi-source replica where the dead master is one of its replication channels is a valid promotion candidate. During promotion, only the managed channel is modified; all other channels are preserved.
54+
55+
### Graceful Master Takeover
56+
57+
`GracefulMasterTakeover` uses `ChangeMasterToForChannel` and `StartReplicationForChannel` with the managed channel name. This ensures that when the demoted master is reconfigured to replicate from the promoted instance, only the relevant channel is set up, and any other channels on the demoted master remain intact.
58+
59+
## API Endpoints
60+
61+
### V1 API
62+
63+
- `GET /api/instance-channels/{host}/{port}` -- Returns the `ReplicationChannels` slice as JSON for the given instance. Each entry includes channel name, master key, IO/SQL thread state, binlog coordinates, lag, and error information.
64+
65+
- `GET /api/instance/{host}/{port}` -- The standard instance endpoint now includes `ReplicationChannels` and `ManagedChannelName` in its JSON response.
66+
67+
### V2 API
68+
69+
- `GET /api/v2/instances/{host}/{port}/channels` -- Returns channels in the V2 response envelope (`{"status": "ok", "data": [...]}`).
70+
71+
## Group Replication Channels
72+
73+
Group Replication uses internal channels named `group_replication_applier` and `group_replication_recovery`. These are automatically detected and excluded from canonical channel selection. Orchestrator will not select a GR internal channel as the managed channel unless no other channels exist.
74+
75+
## Limitations
76+
77+
- Orchestrator manages exactly one channel per instance for topology purposes. Manual management of other channels is expected.
78+
- Channel-aware operations require MySQL 5.7+ or MariaDB 10.1+ (which support the `FOR CHANNEL` syntax).
79+
- The backend database stores channel information in the `database_instance_channels` table. Ensure schema migrations have been applied.
80+
- Multi-source replicas where multiple channels point to the same cluster may cause unexpected behavior in topology analysis.

‎go/http/api.go‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -251,6 +251,23 @@ func (this *HttpAPI) Instance(w http.ResponseWriter, r *http.Request) {
251251
renderJSON(w, http.StatusOK, instance)
252252
}
253253

254+
// InstanceChannels returns the replication channels for a given instance.
255+
// For multi-source replicas, this returns all named replication channels and their status.
256+
func (this *HttpAPI) InstanceChannels(w http.ResponseWriter, r *http.Request) {
257+
instanceKey, err := this.getInstanceKey(chi.URLParam(r, "host"), chi.URLParam(r, "port"))
258+
259+
if err != nil {
260+
Respond(w, &APIResponse{Code: ERROR, Message: err.Error()})
261+
return
262+
}
263+
instance, found, err := inst.ReadInstance(&instanceKey)
264+
if (!found) || (err != nil) {
265+
Respond(w, &APIResponse{Code: ERROR, Message: fmt.Sprintf("Cannot read instance: %+v", instanceKey)})
266+
return
267+
}
268+
renderJSON(w, http.StatusOK, instance.ReplicationChannels)
269+
}
270+
254271
// AsyncDiscover issues an asynchronous read on an instance. This is
255272
// useful for bulk loads of a new set of instances and will not block
256273
// if the instance is slow to respond or not reachable.
@@ -3926,6 +3943,7 @@ func (this *HttpAPI) RegisterRequests(router chi.Router) {
39263943
this.registerAPIRequest(router, "masters", this.Masters)
39273944
this.registerAPIRequest(router, "master/{clusterHint}", this.ClusterMaster)
39283945
this.registerAPIRequest(router, "instance-replicas/{host}/{port}", this.InstanceReplicas)
3946+
this.registerAPIRequest(router, "instance-channels/{host}/{port}", this.InstanceChannels)
39293947
this.registerAPIRequest(router, "all-instances", this.AllInstances)
39303948
this.registerAPIRequest(router, "downtimed", this.Downtimed)
39313949
this.registerAPIRequest(router, "downtimed/{clusterHint}", this.Downtimed)

‎go/http/apiv2.go‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ func RegisterV2Routes(r chi.Router) {
8080

8181
// Instance endpoints
8282
r.Get("/instances/{host}/{port}", V2Instance)
83+
r.Get("/instances/{host}/{port}/channels", V2InstanceChannels)
8384

8485
// Recovery endpoints
8586
r.Get("/recoveries", V2Recoveries)
@@ -219,6 +220,34 @@ func V2Status(w http.ResponseWriter, r *http.Request) {
219220
respondOK(w, health)
220221
}
221222

223+
// V2InstanceChannels returns the replication channels for a specific MySQL instance.
224+
// For multi-source replicas, this returns all named replication channels and their status.
225+
func V2InstanceChannels(w http.ResponseWriter, r *http.Request) {
226+
host := chi.URLParam(r, "host")
227+
port := chi.URLParam(r, "port")
228+
229+
instanceKey, err := inst.NewResolveInstanceKeyStrings(host, port)
230+
if err != nil {
231+
respondError(w, http.StatusBadRequest, "INVALID_INSTANCE", fmt.Sprintf("Invalid instance key: %v", err))
232+
return
233+
}
234+
instanceKey, err = inst.FigureInstanceKey(instanceKey, nil)
235+
if err != nil {
236+
respondError(w, http.StatusBadRequest, "INVALID_INSTANCE", fmt.Sprintf("Cannot resolve instance: %v", err))
237+
return
238+
}
239+
instance, found, err := inst.ReadInstance(instanceKey)
240+
if err != nil {
241+
respondError(w, http.StatusInternalServerError, "INSTANCE_READ_ERROR", fmt.Sprintf("Failed to read instance: %v", err))
242+
return
243+
}
244+
if !found {
245+
respondNotFound(w, fmt.Sprintf("Instance not found: %s:%s", host, port))
246+
return
247+
}
248+
respondOK(w, instance.ReplicationChannels)
249+
}
250+
222251
// V2ProxySQLServers returns all servers from ProxySQL's runtime_mysql_servers table.
223252
func V2ProxySQLServers(w http.ResponseWriter, r *http.Request) {
224253
hook := proxysql.GetHook()

‎go/inst/channel_test.go‎

Lines changed: 211 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,211 @@
1+
package inst
2+
3+
import (
4+
"database/sql"
5+
"testing"
6+
7+
test "github.com/proxysql/golib/tests"
8+
)
9+
10+
func TestSelectCanonicalChannelIndexEmpty(t *testing.T) {
11+
channels := []ChannelStatus{}
12+
idx := selectCanonicalChannelIndex(channels)
13+
test.S(t).ExpectEquals(idx, -1)
14+
}
15+
16+
func TestSelectCanonicalChannelIndexSingleDefault(t *testing.T) {
17+
channels := []ChannelStatus{
18+
{ChannelName: "", MasterKey: InstanceKey{Hostname: "master1", Port: 3306}},
19+
}
20+
idx := selectCanonicalChannelIndex(channels)
21+
test.S(t).ExpectEquals(idx, 0)
22+
}
23+
24+
func TestSelectCanonicalChannelIndexPrefersDefault(t *testing.T) {
25+
channels := []ChannelStatus{
26+
{ChannelName: "channel_a", MasterKey: InstanceKey{Hostname: "master1", Port: 3306}},
27+
{ChannelName: "", MasterKey: InstanceKey{Hostname: "master2", Port: 3306}},
28+
{ChannelName: "channel_b", MasterKey: InstanceKey{Hostname: "master3", Port: 3306}},
29+
}
30+
idx := selectCanonicalChannelIndex(channels)
31+
test.S(t).ExpectEquals(idx, 1)
32+
}
33+
34+
func TestSelectCanonicalChannelIndexSkipsGRChannels(t *testing.T) {
35+
channels := []ChannelStatus{
36+
{ChannelName: "group_replication_applier", MasterKey: InstanceKey{Hostname: "master1", Port: 3306}},
37+
{ChannelName: "group_replication_recovery", MasterKey: InstanceKey{Hostname: "master1", Port: 3306}},
38+
{ChannelName: "my_channel", MasterKey: InstanceKey{Hostname: "master2", Port: 3306}},
39+
}
40+
idx := selectCanonicalChannelIndex(channels)
41+
test.S(t).ExpectEquals(idx, 2)
42+
}
43+
44+
func TestSelectCanonicalChannelIndexAllGR(t *testing.T) {
45+
channels := []ChannelStatus{
46+
{ChannelName: "group_replication_applier", MasterKey: InstanceKey{Hostname: "master1", Port: 3306}},
47+
{ChannelName: "group_replication_recovery", MasterKey: InstanceKey{Hostname: "master1", Port: 3306}},
48+
}
49+
// When only GR channels exist, falls back to index 0
50+
idx := selectCanonicalChannelIndex(channels)
51+
test.S(t).ExpectEquals(idx, 0)
52+
}
53+
54+
func TestSelectCanonicalChannelIndexMultipleNamed(t *testing.T) {
55+
channels := []ChannelStatus{
56+
{ChannelName: "channel_a", MasterKey: InstanceKey{Hostname: "master1", Port: 3306}},
57+
{ChannelName: "channel_b", MasterKey: InstanceKey{Hostname: "master2", Port: 3306}},
58+
}
59+
// No default channel, no GR channels: picks the first one
60+
idx := selectCanonicalChannelIndex(channels)
61+
test.S(t).ExpectEquals(idx, 0)
62+
}
63+
64+
func TestIsGRInternalChannel(t *testing.T) {
65+
cs := ChannelStatus{ChannelName: "group_replication_applier"}
66+
test.S(t).ExpectTrue(cs.IsGRInternalChannel())
67+
68+
cs2 := ChannelStatus{ChannelName: "group_replication_recovery"}
69+
test.S(t).ExpectTrue(cs2.IsGRInternalChannel())
70+
71+
cs3 := ChannelStatus{ChannelName: "my_channel"}
72+
test.S(t).ExpectFalse(cs3.IsGRInternalChannel())
73+
74+
cs4 := ChannelStatus{ChannelName: ""}
75+
test.S(t).ExpectFalse(cs4.IsGRInternalChannel())
76+
}
77+
78+
func TestForChannelClause(t *testing.T) {
79+
test.S(t).ExpectEquals(forChannelClause(""), "")
80+
test.S(t).ExpectEquals(forChannelClause("my_channel"), " FOR CHANNEL 'my_channel'")
81+
test.S(t).ExpectEquals(forChannelClause("group_replication_applier"), " FOR CHANNEL 'group_replication_applier'")
82+
}
83+
84+
func TestSingleSourceBehaviorUnchanged(t *testing.T) {
85+
// Verify that an instance with no replication channels behaves identically
86+
// to pre-channel-support behavior
87+
instance := NewInstance()
88+
instance.Key = InstanceKey{Hostname: "replica1", Port: 3306}
89+
instance.MasterKey = InstanceKey{Hostname: "master1", Port: 3306}
90+
instance.ReadBinlogCoordinates = BinlogCoordinates{LogFile: "mysql-bin.000001", LogPos: 100}
91+
instance.Version = "5.7.35"
92+
93+
// No channels set
94+
test.S(t).ExpectEquals(len(instance.ReplicationChannels), 0)
95+
test.S(t).ExpectEquals(instance.ManagedChannelName, "")
96+
test.S(t).ExpectTrue(instance.IsReplica())
97+
}
98+
99+
func TestMultiSourceInstanceManagedChannel(t *testing.T) {
100+
// Verify that when ReplicationChannels has multiple entries and we pick canonical,
101+
// the ManagedChannelName reflects the chosen channel
102+
channels := []ChannelStatus{
103+
{
104+
ChannelName: "channel_a",
105+
MasterKey: InstanceKey{Hostname: "master1", Port: 3306},
106+
ReplicationIOThreadRunning: true,
107+
ReplicationSQLThreadRunning: true,
108+
SecondsBehindMaster: sql.NullInt64{Int64: 0, Valid: true},
109+
},
110+
{
111+
ChannelName: "channel_b",
112+
MasterKey: InstanceKey{Hostname: "master2", Port: 3306},
113+
ReplicationIOThreadRunning: true,
114+
ReplicationSQLThreadRunning: true,
115+
SecondsBehindMaster: sql.NullInt64{Int64: 5, Valid: true},
116+
},
117+
}
118+
119+
idx := selectCanonicalChannelIndex(channels)
120+
test.S(t).ExpectEquals(idx, 0)
121+
122+
// The canonical channel should be channel_a (first non-GR channel)
123+
ch := channels[idx]
124+
test.S(t).ExpectEquals(ch.ChannelName, "channel_a")
125+
test.S(t).ExpectEquals(ch.MasterKey.Hostname, "master1")
126+
}
127+
128+
func TestChannelAwareSQLGeneration(t *testing.T) {
129+
qsp := GetQueryStringProvider("5.7.35")
130+
131+
// Without channel name -- no FOR CHANNEL clause
132+
stopSQL := qsp.StopReplicaForChannel("")
133+
test.S(t).ExpectTrue(len(stopSQL) > 0)
134+
test.S(t).ExpectEquals(stopSQL, qsp.stop_slave())
135+
136+
startSQL := qsp.StartReplicaForChannel("")
137+
test.S(t).ExpectEquals(startSQL, qsp.start_slave())
138+
139+
// With channel name -- should include FOR CHANNEL clause
140+
stopSQLCh := qsp.StopReplicaForChannel("my_channel")
141+
test.S(t).ExpectTrue(len(stopSQLCh) > len(stopSQL))
142+
expectedStop := qsp.stop_slave() + " FOR CHANNEL 'my_channel'"
143+
test.S(t).ExpectEquals(stopSQLCh, expectedStop)
144+
145+
startSQLCh := qsp.StartReplicaForChannel("my_channel")
146+
expectedStart := qsp.start_slave() + " FOR CHANNEL 'my_channel'"
147+
test.S(t).ExpectEquals(startSQLCh, expectedStart)
148+
149+
resetSQLCh := qsp.ResetReplicaForChannel("my_channel")
150+
expectedReset := qsp.reset_slave() + " FOR CHANNEL 'my_channel'"
151+
test.S(t).ExpectEquals(resetSQLCh, expectedReset)
152+
}
153+
154+
func TestChannelAwareSQLGeneration84(t *testing.T) {
155+
// Test with MySQL 8.4+ which uses "stop replica" / "start replica" syntax
156+
qsp := GetQueryStringProvider("8.4.0")
157+
158+
stopSQL := qsp.StopReplicaForChannel("my_channel")
159+
test.S(t).ExpectEquals(stopSQL, "stop replica FOR CHANNEL 'my_channel'")
160+
161+
startSQL := qsp.StartReplicaForChannel("my_channel")
162+
test.S(t).ExpectEquals(startSQL, "start replica FOR CHANNEL 'my_channel'")
163+
164+
resetSQL := qsp.ResetReplicaForChannel("my_channel")
165+
test.S(t).ExpectEquals(resetSQL, "reset replica FOR CHANNEL 'my_channel'")
166+
}
167+
168+
func TestChannelAwareIOSQLThreadOperations(t *testing.T) {
169+
qsp := GetQueryStringProvider("5.7.35")
170+
171+
// IO thread operations
172+
stopIO := qsp.StopReplicaIOThreadForChannel("ch1")
173+
test.S(t).ExpectEquals(stopIO, "stop slave io_thread FOR CHANNEL 'ch1'")
174+
175+
startIO := qsp.StartReplicaIOThreadForChannel("ch1")
176+
test.S(t).ExpectEquals(startIO, "start slave io_thread FOR CHANNEL 'ch1'")
177+
178+
// SQL thread operations
179+
stopSQLThread := qsp.StopReplicaSQLThreadForChannel("ch1")
180+
test.S(t).ExpectEquals(stopSQLThread, "stop slave sql_thread FOR CHANNEL 'ch1'")
181+
182+
startSQLThread := qsp.StartReplicaSQLThreadForChannel("ch1")
183+
test.S(t).ExpectEquals(startSQLThread, "start slave sql_thread FOR CHANNEL 'ch1'")
184+
185+
// Without channel name -- no FOR CHANNEL
186+
stopIODefault := qsp.StopReplicaIOThreadForChannel("")
187+
test.S(t).ExpectEquals(stopIODefault, "stop slave io_thread")
188+
}
189+
190+
func TestSelectCanonicalChannelWithDefaultAndGR(t *testing.T) {
191+
// When default channel exists alongside GR channels, prefer default
192+
channels := []ChannelStatus{
193+
{ChannelName: "group_replication_applier"},
194+
{ChannelName: ""},
195+
{ChannelName: "group_replication_recovery"},
196+
}
197+
idx := selectCanonicalChannelIndex(channels)
198+
test.S(t).ExpectEquals(idx, 1)
199+
}
200+
201+
func TestSelectCanonicalChannelWithNamedAndGR(t *testing.T) {
202+
// When no default channel but named + GR channels exist, prefer the named one
203+
channels := []ChannelStatus{
204+
{ChannelName: "group_replication_applier"},
205+
{ChannelName: "group_replication_recovery"},
206+
{ChannelName: "custom_repl"},
207+
{ChannelName: "another_repl"},
208+
}
209+
idx := selectCanonicalChannelIndex(channels)
210+
test.S(t).ExpectEquals(idx, 2)
211+
}

‎go/logic/topology_recovery.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2228,7 +2228,9 @@ func GracefulMasterTakeover(clusterName string, designatedKey *inst.InstanceKey,
22282228
if topologyRecovery.RecoveryType == MasterRecoveryGTID {
22292229
gtidHint = inst.GTIDHintForce
22302230
}
2231-
clusterMaster, err = inst.ChangeMasterTo(&clusterMaster.Key, &designatedInstance.Key, promotedMasterCoordinates, false, gtidHint)
2231+
// Use channel-aware operations to preserve other replication channels on multi-source replicas
2232+
managedChannel := clusterMaster.ManagedChannelName
2233+
clusterMaster, err = inst.ChangeMasterToForChannel(&clusterMaster.Key, &designatedInstance.Key, promotedMasterCoordinates, false, gtidHint, managedChannel)
22322234
if !clusterMaster.SelfBinlogCoordinates.Equals(demotedMasterSelfBinlogCoordinates) {
22332235
log.Errorf("GracefulMasterTakeover: sanity problem. Demoted master's coordinates changed from %+v to %+v while supposed to have been frozen", *demotedMasterSelfBinlogCoordinates, clusterMaster.SelfBinlogCoordinates)
22342236
}
@@ -2250,7 +2252,7 @@ func GracefulMasterTakeover(clusterName string, designatedKey *inst.InstanceKey,
22502252
}
22512253
}
22522254
if auto {
2253-
_, startReplicationErr := inst.StartReplication(&clusterMaster.Key)
2255+
_, startReplicationErr := inst.StartReplicationForChannel(&clusterMaster.Key, managedChannel)
22542256
if err == nil {
22552257
err = startReplicationErr
22562258
}

0 commit comments

Comments
 (0)