|
|
|
@@ -667,6 +667,421 @@ func TestTunnelUpdateRecoversFromAddressInUseContract(t *testing.T) {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func TestTunnelUpdateChangesEntryNodeButLeavesOldForwardRuntimeContract(t *testing.T) {
|
|
|
|
|
secret := "contract-jwt-secret"
|
|
|
|
|
router, r := setupContractRouter(t, secret)
|
|
|
|
|
server := httptest.NewServer(router)
|
|
|
|
|
defer server.Close()
|
|
|
|
|
|
|
|
|
|
adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
|
|
|
|
|
if err != nil {
|
|
|
|
|
t.Fatalf("generate admin token: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
now := time.Now().UnixMilli()
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
|
|
|
|
|
VALUES(2, 'issue281_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
|
|
|
|
|
`, now, now).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert user: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
|
|
|
|
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
|
|
|
`, "issue281-tunnel", 1.0, 2, "tls", 99999, now, now, 1, nil, 0).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert tunnel: %v", err)
|
|
|
|
|
}
|
|
|
|
|
tunnelID := mustLastInsertID(t, r, "issue281-tunnel")
|
|
|
|
|
|
|
|
|
|
insertNode := func(name, secretValue, ip, portRange string, inx int) int64 {
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx)
|
|
|
|
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
|
|
|
`, name, secretValue, ip, ip, "", portRange, "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", inx).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert node %s: %v", name, err)
|
|
|
|
|
}
|
|
|
|
|
return mustLastInsertID(t, r, name)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
oldEntryNodeID := insertNode("issue281-old-entry", "issue281-old-entry-secret", "10.51.0.1", "51000-51010", 0)
|
|
|
|
|
newEntryNodeID := insertNode("issue281-new-entry", "issue281-new-entry-secret", "10.51.0.2", "52000-52010", 1)
|
|
|
|
|
exitNodeID := insertNode("issue281-exit", "issue281-exit-secret", "10.51.0.3", "53000-53010", 2)
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
|
|
|
|
|
VALUES(?, 1, ?, 51001, 'round', 1, 'tls')
|
|
|
|
|
`, tunnelID, oldEntryNodeID).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert old entry chain_tunnel: %v", err)
|
|
|
|
|
}
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
|
|
|
|
|
VALUES(?, 3, ?, 53001, 'round', 1, 'tls')
|
|
|
|
|
`, tunnelID, exitNodeID).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert exit chain_tunnel: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
|
|
|
|
|
VALUES(281, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)
|
|
|
|
|
`, tunnelID).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert user_tunnel: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx)
|
|
|
|
|
VALUES(2, 'issue281_user', 'issue281-forward', ?, '8.8.8.8:53', 'fifo', 0, 0, ?, ?, 1, 0)
|
|
|
|
|
`, tunnelID, now, now).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert forward: %v", err)
|
|
|
|
|
}
|
|
|
|
|
forwardID := mustLastInsertID(t, r, "issue281-forward")
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, oldEntryNodeID, 51001).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert forward_port: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
forwardBase := fmt.Sprintf("%d_%d_%d", forwardID, 2, 281)
|
|
|
|
|
|
|
|
|
|
var commandMu sync.Mutex
|
|
|
|
|
oldEntryDeleteNames := make([]string, 0)
|
|
|
|
|
newEntryUpdateNames := make([]string, 0)
|
|
|
|
|
|
|
|
|
|
recordForwardServiceNames := func(data json.RawMessage, list *[]string) {
|
|
|
|
|
var serviceList []map[string]interface{}
|
|
|
|
|
if err := json.Unmarshal(data, &serviceList); err == nil {
|
|
|
|
|
for _, service := range serviceList {
|
|
|
|
|
name, _ := service["name"].(string)
|
|
|
|
|
if strings.HasPrefix(strings.TrimSpace(name), forwardBase) {
|
|
|
|
|
*list = append(*list, name)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
var payload map[string]interface{}
|
|
|
|
|
if err := json.Unmarshal(data, &payload); err != nil {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
if rawServices, ok := payload["services"].([]interface{}); ok {
|
|
|
|
|
for _, raw := range rawServices {
|
|
|
|
|
name, _ := raw.(string)
|
|
|
|
|
if strings.HasPrefix(strings.TrimSpace(name), forwardBase) {
|
|
|
|
|
*list = append(*list, name)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
stopOldEntry := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-old-entry-secret", func(cmdType string, data json.RawMessage) (bool, string) {
|
|
|
|
|
commandMu.Lock()
|
|
|
|
|
defer commandMu.Unlock()
|
|
|
|
|
if strings.EqualFold(strings.TrimSpace(cmdType), "DeleteService") {
|
|
|
|
|
recordForwardServiceNames(data, &oldEntryDeleteNames)
|
|
|
|
|
}
|
|
|
|
|
return false, ""
|
|
|
|
|
})
|
|
|
|
|
defer stopOldEntry()
|
|
|
|
|
|
|
|
|
|
stopNewEntry := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-new-entry-secret", func(cmdType string, data json.RawMessage) (bool, string) {
|
|
|
|
|
commandMu.Lock()
|
|
|
|
|
defer commandMu.Unlock()
|
|
|
|
|
if strings.EqualFold(strings.TrimSpace(cmdType), "UpdateService") || strings.EqualFold(strings.TrimSpace(cmdType), "AddService") {
|
|
|
|
|
recordForwardServiceNames(data, &newEntryUpdateNames)
|
|
|
|
|
}
|
|
|
|
|
return false, ""
|
|
|
|
|
})
|
|
|
|
|
defer stopNewEntry()
|
|
|
|
|
|
|
|
|
|
stopExit := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-exit-secret", func(cmdType string, data json.RawMessage) (bool, string) {
|
|
|
|
|
return false, ""
|
|
|
|
|
})
|
|
|
|
|
defer stopExit()
|
|
|
|
|
|
|
|
|
|
waitNodeStatus(t, r, oldEntryNodeID, 1)
|
|
|
|
|
waitNodeStatus(t, r, newEntryNodeID, 1)
|
|
|
|
|
waitNodeStatus(t, r, exitNodeID, 1)
|
|
|
|
|
|
|
|
|
|
payload := map[string]interface{}{
|
|
|
|
|
"id": tunnelID,
|
|
|
|
|
"name": "issue281-tunnel",
|
|
|
|
|
"type": 2,
|
|
|
|
|
"flow": 99999,
|
|
|
|
|
"trafficRatio": 1.0,
|
|
|
|
|
"status": 1,
|
|
|
|
|
"inNodeId": []map[string]interface{}{
|
|
|
|
|
{"nodeId": newEntryNodeID, "protocol": "tls", "strategy": "round"},
|
|
|
|
|
},
|
|
|
|
|
"chainNodes": []interface{}{},
|
|
|
|
|
"outNodeId": []map[string]interface{}{
|
|
|
|
|
{"nodeId": exitNodeID, "protocol": "tls", "strategy": "round", "port": 53001},
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
body, err := json.Marshal(payload)
|
|
|
|
|
if err != nil {
|
|
|
|
|
t.Fatalf("marshal payload: %v", err)
|
|
|
|
|
}
|
|
|
|
|
req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/update", bytes.NewReader(body))
|
|
|
|
|
req.Header.Set("Authorization", adminToken)
|
|
|
|
|
req.Header.Set("Content-Type", "application/json")
|
|
|
|
|
res := httptest.NewRecorder()
|
|
|
|
|
router.ServeHTTP(res, req)
|
|
|
|
|
assertCode(t, res, 0)
|
|
|
|
|
|
|
|
|
|
nodeAfter, portAfter := mustQueryInt64Int(t, r, `SELECT node_id, port FROM forward_port WHERE forward_id = ? LIMIT 1`, forwardID)
|
|
|
|
|
if nodeAfter != newEntryNodeID || portAfter != 51001 {
|
|
|
|
|
t.Fatalf("expected forward_port rebound to node=%d port=51001, got node=%d port=%d", newEntryNodeID, nodeAfter, portAfter)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
commandMu.Lock()
|
|
|
|
|
defer commandMu.Unlock()
|
|
|
|
|
if len(newEntryUpdateNames) == 0 {
|
|
|
|
|
t.Fatalf("expected new entry node to receive forward runtime sync for %s", forwardBase)
|
|
|
|
|
}
|
|
|
|
|
if len(oldEntryDeleteNames) == 0 {
|
|
|
|
|
t.Fatalf("expected old entry node to receive forward DeleteService cleanup for %s, got none", forwardBase)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func TestTunnelUpdateEntryTransitionsCleanupForwardRuntimeContract(t *testing.T) {
|
|
|
|
|
secret := "contract-jwt-secret"
|
|
|
|
|
router, r := setupContractRouter(t, secret)
|
|
|
|
|
server := httptest.NewServer(router)
|
|
|
|
|
defer server.Close()
|
|
|
|
|
|
|
|
|
|
adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
|
|
|
|
|
if err != nil {
|
|
|
|
|
t.Fatalf("generate admin token: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
now := time.Now().UnixMilli()
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, created_time, updated_time, status)
|
|
|
|
|
VALUES(2, 'issue281_transition_user', 'pwd', 1, 2727251700000, 99999, 0, 0, 1, 99999, ?, ?, 1)
|
|
|
|
|
`, now, now).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert user: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
|
|
|
|
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
|
|
|
`, "issue281-transition-tunnel", 1.0, 2, "tls", 99999, now, now, 1, nil, 0).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert tunnel: %v", err)
|
|
|
|
|
}
|
|
|
|
|
tunnelID := mustLastInsertID(t, r, "issue281-transition-tunnel")
|
|
|
|
|
|
|
|
|
|
insertNode := func(name, secretValue, ip, portRange string, inx int) int64 {
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx)
|
|
|
|
|
VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
|
|
|
`, name, secretValue, ip, ip, "", portRange, "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", inx).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert node %s: %v", name, err)
|
|
|
|
|
}
|
|
|
|
|
return mustLastInsertID(t, r, name)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
entryA := insertNode("issue281-transition-entry-a", "issue281-transition-entry-a-secret", "10.52.0.1", "54000-54010", 0)
|
|
|
|
|
entryB := insertNode("issue281-transition-entry-b", "issue281-transition-entry-b-secret", "10.52.0.2", "55000-55010", 1)
|
|
|
|
|
entryC := insertNode("issue281-transition-entry-c", "issue281-transition-entry-c-secret", "10.52.0.3", "56000-56010", 2)
|
|
|
|
|
exitNodeID := insertNode("issue281-transition-exit", "issue281-transition-exit-secret", "10.52.0.4", "57000-57010", 3)
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
|
|
|
|
|
VALUES(?, 1, ?, 54001, 'round', 1, 'tls')
|
|
|
|
|
`, tunnelID, entryA).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert initial entry chain_tunnel: %v", err)
|
|
|
|
|
}
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
|
|
|
|
|
VALUES(?, 3, ?, 57001, 'round', 1, 'tls')
|
|
|
|
|
`, tunnelID, exitNodeID).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert exit chain_tunnel: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO user_tunnel(id, user_id, tunnel_id, speed_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
|
|
|
|
|
VALUES(282, 2, ?, NULL, 999, 99999, 0, 0, 1, 2727251700000, 1)
|
|
|
|
|
`, tunnelID).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert user_tunnel: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`
|
|
|
|
|
INSERT INTO forward(user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx)
|
|
|
|
|
VALUES(2, 'issue281_transition_user', 'issue281-transition-forward', ?, '1.1.1.1:443', 'fifo', 0, 0, ?, ?, 1, 0)
|
|
|
|
|
`, tunnelID, now, now).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert forward: %v", err)
|
|
|
|
|
}
|
|
|
|
|
forwardID := mustLastInsertID(t, r, "issue281-transition-forward")
|
|
|
|
|
|
|
|
|
|
if err := r.DB().Exec(`INSERT INTO forward_port(forward_id, node_id, port) VALUES(?, ?, ?)`, forwardID, entryA, 54001).Error; err != nil {
|
|
|
|
|
t.Fatalf("insert forward_port: %v", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
forwardBase := fmt.Sprintf("%d_%d_%d", forwardID, 2, 282)
|
|
|
|
|
recorder := newForwardRuntimeCommandRecorder(forwardBase)
|
|
|
|
|
|
|
|
|
|
stopEntryA := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-transition-entry-a-secret", recorder.handler("entry-a"))
|
|
|
|
|
defer stopEntryA()
|
|
|
|
|
stopEntryB := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-transition-entry-b-secret", recorder.handler("entry-b"))
|
|
|
|
|
defer stopEntryB()
|
|
|
|
|
stopEntryC := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-transition-entry-c-secret", recorder.handler("entry-c"))
|
|
|
|
|
defer stopEntryC()
|
|
|
|
|
stopExit := startMockNodeSessionWithCommandRecorder(t, server.URL, "issue281-transition-exit-secret", recorder.handler("exit"))
|
|
|
|
|
defer stopExit()
|
|
|
|
|
|
|
|
|
|
waitNodeStatus(t, r, entryA, 1)
|
|
|
|
|
waitNodeStatus(t, r, entryB, 1)
|
|
|
|
|
waitNodeStatus(t, r, entryC, 1)
|
|
|
|
|
waitNodeStatus(t, r, exitNodeID, 1)
|
|
|
|
|
|
|
|
|
|
updateTunnelEntries := func(entries []map[string]interface{}) {
|
|
|
|
|
payload := map[string]interface{}{
|
|
|
|
|
"id": tunnelID,
|
|
|
|
|
"name": "issue281-transition-tunnel",
|
|
|
|
|
"type": 2,
|
|
|
|
|
"flow": 99999,
|
|
|
|
|
"trafficRatio": 1.0,
|
|
|
|
|
"status": 1,
|
|
|
|
|
"inNodeId": entries,
|
|
|
|
|
"chainNodes": []interface{}{},
|
|
|
|
|
"outNodeId": []map[string]interface{}{
|
|
|
|
|
{"nodeId": exitNodeID, "protocol": "tls", "strategy": "round", "port": 57001},
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
body, err := json.Marshal(payload)
|
|
|
|
|
if err != nil {
|
|
|
|
|
t.Fatalf("marshal payload: %v", err)
|
|
|
|
|
}
|
|
|
|
|
req := httptest.NewRequest(http.MethodPost, "/api/v1/tunnel/update", bytes.NewReader(body))
|
|
|
|
|
req.Header.Set("Authorization", adminToken)
|
|
|
|
|
req.Header.Set("Content-Type", "application/json")
|
|
|
|
|
res := httptest.NewRecorder()
|
|
|
|
|
router.ServeHTTP(res, req)
|
|
|
|
|
assertCode(t, res, 0)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
updateTunnelEntries([]map[string]interface{}{
|
|
|
|
|
{"nodeId": entryA, "protocol": "tls", "strategy": "round"},
|
|
|
|
|
{"nodeId": entryB, "protocol": "tls", "strategy": "round"},
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
afterMulti := mustQueryNodePorts(t, r, `SELECT node_id, port FROM forward_port WHERE forward_id = ? ORDER BY id ASC`, forwardID)
|
|
|
|
|
if len(afterMulti) != 2 || afterMulti[entryA] != 54001 || afterMulti[entryB] != 54001 {
|
|
|
|
|
t.Fatalf("expected forward_port on entryA+entryB with port 54001, got %v", afterMulti)
|
|
|
|
|
}
|
|
|
|
|
if recorder.syncCount("entry-b") == 0 {
|
|
|
|
|
t.Fatalf("expected entry-b to receive forward runtime sync for %s", forwardBase)
|
|
|
|
|
}
|
|
|
|
|
if recorder.deleteCount("entry-a") != 0 {
|
|
|
|
|
t.Fatalf("expected no cleanup on retained entry-a during single->multi transition, got %v", recorder.deleteNames("entry-a"))
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
updateTunnelEntries([]map[string]interface{}{
|
|
|
|
|
{"nodeId": entryC, "protocol": "tls", "strategy": "round"},
|
|
|
|
|
})
|
|
|
|
|
|
|
|
|
|
afterSingle := mustQueryNodePorts(t, r, `SELECT node_id, port FROM forward_port WHERE forward_id = ? ORDER BY id ASC`, forwardID)
|
|
|
|
|
if len(afterSingle) != 1 || afterSingle[entryC] != 54001 {
|
|
|
|
|
t.Fatalf("expected forward_port on entryC with port 54001, got %v", afterSingle)
|
|
|
|
|
}
|
|
|
|
|
if recorder.deleteCount("entry-a") == 0 {
|
|
|
|
|
t.Fatalf("expected cleanup on removed entry-a during multi->single transition, got %v", recorder.deleteNames("entry-a"))
|
|
|
|
|
}
|
|
|
|
|
if recorder.deleteCount("entry-b") == 0 {
|
|
|
|
|
t.Fatalf("expected cleanup on removed entry-b during multi->single transition, got %v", recorder.deleteNames("entry-b"))
|
|
|
|
|
}
|
|
|
|
|
if recorder.syncCount("entry-c") == 0 {
|
|
|
|
|
t.Fatalf("expected entry-c to receive forward runtime sync for %s", forwardBase)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
type forwardRuntimeCommandRecorder struct {
|
|
|
|
|
prefix string
|
|
|
|
|
|
|
|
|
|
mu sync.Mutex
|
|
|
|
|
deletes map[string][]string
|
|
|
|
|
syncNames map[string][]string
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func newForwardRuntimeCommandRecorder(prefix string) *forwardRuntimeCommandRecorder {
|
|
|
|
|
return &forwardRuntimeCommandRecorder{
|
|
|
|
|
prefix: strings.TrimSpace(prefix),
|
|
|
|
|
deletes: make(map[string][]string),
|
|
|
|
|
syncNames: make(map[string][]string),
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (r *forwardRuntimeCommandRecorder) handler(node string) func(string, json.RawMessage) (bool, string) {
|
|
|
|
|
return func(cmdType string, data json.RawMessage) (bool, string) {
|
|
|
|
|
names := collectForwardServiceNames(data, r.prefix)
|
|
|
|
|
if len(names) == 0 {
|
|
|
|
|
return false, ""
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
r.mu.Lock()
|
|
|
|
|
defer r.mu.Unlock()
|
|
|
|
|
|
|
|
|
|
if strings.EqualFold(strings.TrimSpace(cmdType), "DeleteService") {
|
|
|
|
|
r.deletes[node] = append(r.deletes[node], names...)
|
|
|
|
|
}
|
|
|
|
|
if strings.EqualFold(strings.TrimSpace(cmdType), "UpdateService") || strings.EqualFold(strings.TrimSpace(cmdType), "AddService") {
|
|
|
|
|
r.syncNames[node] = append(r.syncNames[node], names...)
|
|
|
|
|
}
|
|
|
|
|
return false, ""
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (r *forwardRuntimeCommandRecorder) deleteCount(node string) int {
|
|
|
|
|
r.mu.Lock()
|
|
|
|
|
defer r.mu.Unlock()
|
|
|
|
|
return len(r.deletes[node])
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (r *forwardRuntimeCommandRecorder) syncCount(node string) int {
|
|
|
|
|
r.mu.Lock()
|
|
|
|
|
defer r.mu.Unlock()
|
|
|
|
|
return len(r.syncNames[node])
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (r *forwardRuntimeCommandRecorder) deleteNames(node string) []string {
|
|
|
|
|
r.mu.Lock()
|
|
|
|
|
defer r.mu.Unlock()
|
|
|
|
|
return append([]string(nil), r.deletes[node]...)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func collectForwardServiceNames(data json.RawMessage, prefix string) []string {
|
|
|
|
|
prefix = strings.TrimSpace(prefix)
|
|
|
|
|
if prefix == "" {
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
names := make([]string, 0)
|
|
|
|
|
var serviceList []map[string]interface{}
|
|
|
|
|
if err := json.Unmarshal(data, &serviceList); err == nil {
|
|
|
|
|
for _, service := range serviceList {
|
|
|
|
|
name, _ := service["name"].(string)
|
|
|
|
|
if strings.HasPrefix(strings.TrimSpace(name), prefix) {
|
|
|
|
|
names = append(names, name)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
return names
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
var payload map[string]interface{}
|
|
|
|
|
if err := json.Unmarshal(data, &payload); err != nil {
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
if rawServices, ok := payload["services"].([]interface{}); ok {
|
|
|
|
|
for _, raw := range rawServices {
|
|
|
|
|
name, _ := raw.(string)
|
|
|
|
|
if strings.HasPrefix(strings.TrimSpace(name), prefix) {
|
|
|
|
|
names = append(names, name)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
return names
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func startMockNodeSessionWithCommandFailures(t *testing.T, baseURL string, nodeSecret string, failCommands map[string]string) func() {
|
|
|
|
|
t.Helper()
|
|
|
|
|
|
|
|
|
|