From d60f3cfd0809d145d5a3e75fee5b9ed84f23e998 Mon Sep 17 00:00:00 2001 From: stulzq Date: Mon, 16 Nov 2020 17:33:14 +0800 Subject: [PATCH] Client HA --- .../Connections/HA/CanalClientRunningInfo.cs | 16 +++ ...nningInfo.cs => CanalServerRunningInfo.cs} | 2 +- .../Connections/HA/ClusterCanalConnection.cs | 121 +++++++++++++++--- .../Simple/SimpleCanalConnection.cs | 9 +- 4 files changed, 125 insertions(+), 23 deletions(-) create mode 100644 src/CanalSharp/Connections/HA/CanalClientRunningInfo.cs rename src/CanalSharp/Connections/HA/{CanalRunningInfo.cs => CanalServerRunningInfo.cs} (77%) diff --git a/src/CanalSharp/Connections/HA/CanalClientRunningInfo.cs b/src/CanalSharp/Connections/HA/CanalClientRunningInfo.cs new file mode 100644 index 0000000..3f339fd --- /dev/null +++ b/src/CanalSharp/Connections/HA/CanalClientRunningInfo.cs @@ -0,0 +1,16 @@ +using Newtonsoft.Json; + +namespace CanalSharp.Connections +{ + public class CanalClientRunningInfo + { + [JsonProperty("active")] + public bool Active { get; set; } + + [JsonProperty("address")] + public string Address { get; set; } + + [JsonProperty("clientId")] + public string ClientId { get; set; } + } +} \ No newline at end of file diff --git a/src/CanalSharp/Connections/HA/CanalRunningInfo.cs b/src/CanalSharp/Connections/HA/CanalServerRunningInfo.cs similarity index 77% rename from src/CanalSharp/Connections/HA/CanalRunningInfo.cs rename to src/CanalSharp/Connections/HA/CanalServerRunningInfo.cs index 502cdfe..2e01e94 100644 --- a/src/CanalSharp/Connections/HA/CanalRunningInfo.cs +++ b/src/CanalSharp/Connections/HA/CanalServerRunningInfo.cs @@ -1,6 +1,6 @@ namespace CanalSharp.Connections { - public class CanalRunningInfo + public class CanalServerRunningInfo { public bool Active { get; set; } public string Address { get; set; } diff --git a/src/CanalSharp/Connections/HA/ClusterCanalConnection.cs b/src/CanalSharp/Connections/HA/ClusterCanalConnection.cs index 5881d31..8053da1 100644 --- a/src/CanalSharp/Connections/HA/ClusterCanalConnection.cs +++ b/src/CanalSharp/Connections/HA/ClusterCanalConnection.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Text; using System.Threading.Tasks; using CanalSharp.Connections.Enums; @@ -7,36 +8,44 @@ using JetBrains.Annotations; using Microsoft.Extensions.Logging; using Newtonsoft.Json; using NZookeeper; +using NZookeeper.ACL; using NZookeeper.Enums; +using NZookeeper.Node; +using org.apache.zookeeper; namespace CanalSharp.Connections { /// /// Support canal server cluster and client cluster. /// - public class ClusterCanalConnection : ICanalConnection + public class ClusterCanalConnection { private readonly ClusterCanalOptions _options; private readonly ILoggerFactory _loggerFactory; private readonly ILogger _logger; + private readonly TaskCompletionSource _completionSource; // ReSharper disable once InconsistentNaming - private readonly string ZK_RUNNING_NODE; + private readonly string ZK_SERVER_RUNNING_NODE; + // ReSharper disable once InconsistentNaming + private readonly string ZK_CLIENT_RUNNING_NODE; - // For reconnection + // For reconnect private string _lastSubFilter; private bool _serverRunningNodeReCreated; private SimpleCanalConnection _currentConn; private ZkConnection _zk; + private CanalClientRunningInfo _clientRunningInfo; public ClusterCanalConnection([NotNull] ClusterCanalOptions options, ILoggerFactory loggerFactory) { _options = options; _loggerFactory = loggerFactory; _logger = loggerFactory.CreateLogger(); - ZK_RUNNING_NODE = $"/otter/canal/destinations/{_options.Destination}/running"; + _completionSource = new TaskCompletionSource(); + ZK_SERVER_RUNNING_NODE = $"/otter/canal/destinations/{_options.Destination}/running"; + ZK_CLIENT_RUNNING_NODE = $"/otter/canal/destinations/{_options.Destination}/{_options.ClientId}/running"; } - public async Task ReConnectAsync() { await DisConnectAsync(); @@ -72,17 +81,78 @@ namespace CanalSharp.Connections public async Task ConnectAsync() { - await ConnectToZkAsync(); //get canal address from zk - var nodeData = await _zk.GetDataAsync(ZK_RUNNING_NODE); - var runningInfo = JsonConvert.DeserializeObject(Encoding.UTF8.GetString(nodeData)); + await ConnectToZkAsync(); + var nodeData = await _zk.GetDataAsync(ZK_SERVER_RUNNING_NODE); + var runningInfo = JsonConvert.DeserializeObject(Encoding.UTF8.GetString(nodeData)); _logger.LogInformation($"get canal address from zookeeper success: {runningInfo.Address}"); //connect to canal _currentConn = new SimpleCanalConnection(CopyOptions(runningInfo), _loggerFactory.CreateLogger()); await _currentConn.ConnectAsync(); - _serverRunningNodeReCreated = false; + + var localIp = _currentConn.GetLocalEndPoint().ToString(); + _clientRunningInfo = new CanalClientRunningInfo() + { Active = true, Address = localIp, ClientId = _options.ClientId }; + _ = GetZkLockAsync(_clientRunningInfo); + await _completionSource.Task; + + _logger.LogInformation("Ready to use!"); + } + + public async Task GetZkLockAsync(CanalClientRunningInfo runningInfo, bool waiting = false) + { + var times = 0; + while (waiting && times < 60) + { + await Task.Delay(1000); + times++; + _logger.LogWarning($"Waiting for get lock {times}-60..."); + } + + if (await _zk.NodeExistsAsync(ZK_CLIENT_RUNNING_NODE)) + { + _logger.LogInformation($"Node {ZK_CLIENT_RUNNING_NODE} exits, get Zookeeper lock failed. Other instances are running."); + _logger.LogWarning("Waiting..."); + } + else + { + try + { + var clientNodeData = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(runningInfo)); + + var parentNode = ZK_CLIENT_RUNNING_NODE.Replace("/running", ""); + if (!await _zk.NodeExistsAsync(parentNode)) + await _zk.CreateNodeAsync(parentNode, null, + new List() { new Acl(AclPerm.All, AclScheme.World, AclId.World()) }, NodeType.Persistent); + + await _zk.CreateNodeAsync(ZK_CLIENT_RUNNING_NODE, clientNodeData, + new List() { new Acl(AclPerm.All, AclScheme.World, AclId.World()) }, NodeType.Ephemeral); + await _zk.CreateNodeAsync(parentNode + "/" + runningInfo.Address, null, + new List() { new Acl(AclPerm.All, AclScheme.World, AclId.World()) }, NodeType.Ephemeral); + await _zk.NodeExistsAsync(ZK_CLIENT_RUNNING_NODE); + _completionSource.SetResult(0); + _logger.LogInformation("Get zookeeper lock success."); + } + catch (KeeperException.NodeExistsException e) + { + if (e.getPath() != ZK_CLIENT_RUNNING_NODE) + { + _logger.LogError(e, "Error during get lock."); + Environment.Exit(-1); + } + + _logger.LogInformation( + $"Node {ZK_CLIENT_RUNNING_NODE} exits, get Zookeeper lock failed. Other instances are running."); + _logger.LogWarning("Waiting..."); + } + catch (Exception e) + { + _logger.LogError(e, "Exception in lock"); + } + } + } private async Task ConnectToZkAsync() @@ -105,7 +175,7 @@ namespace CanalSharp.Connections var times = 0; while (times < 60) { - if (await _zk.NodeExistsAsync(ZK_RUNNING_NODE)) + if (await _zk.NodeExistsAsync(ZK_SERVER_RUNNING_NODE)) { return; } @@ -113,25 +183,36 @@ namespace CanalSharp.Connections await Task.Delay(1000); times++; - _logger.LogWarning($"Can not find node {ZK_RUNNING_NODE} on Zookeeper. Retrying... {times}"); + _logger.LogWarning($"Can not find node {ZK_SERVER_RUNNING_NODE} on Zookeeper. Retrying... {times}"); } - throw new CanalConnectionException($"Can not find node {ZK_RUNNING_NODE} on Zookeeper."); + throw new CanalConnectionException($"Can not find node {ZK_SERVER_RUNNING_NODE} on Zookeeper."); } private async Task Zk_OnWatch(ZkWatchEventArgs args) { - if (args.Path == ZK_RUNNING_NODE && args.EventType == WatchEventType.NodeCreated) + if (args.Path == ZK_SERVER_RUNNING_NODE && args.EventType == WatchEventType.NodeCreated) { _serverRunningNodeReCreated = true; - _logger.LogInformation($"Zookeeper node {ZK_RUNNING_NODE} Created"); - } - else if (args.Path == ZK_RUNNING_NODE && args.EventType == WatchEventType.NodeDeleted) - { - _logger.LogInformation($"Zookeeper node {ZK_RUNNING_NODE} Deleted"); + _logger.LogInformation($"Server node {ZK_SERVER_RUNNING_NODE} Created"); } - await _zk.NodeExistsAsync(ZK_RUNNING_NODE); + if (args.Path == ZK_SERVER_RUNNING_NODE && args.EventType == WatchEventType.NodeDeleted) + { + _logger.LogInformation($"Server node {ZK_SERVER_RUNNING_NODE} Deleted"); + } + + if (args.Path == ZK_CLIENT_RUNNING_NODE && args.EventType == WatchEventType.NodeDeleted) + { + _logger.LogInformation($"Client node {ZK_SERVER_RUNNING_NODE} Deleted"); + _ = GetZkLockAsync(_clientRunningInfo, true); + } + + if (_zk.Connected) + { + await _zk.NodeExistsAsync(ZK_SERVER_RUNNING_NODE); + await _zk.NodeExistsAsync(ZK_CLIENT_RUNNING_NODE); + } } public Task SubscribeAsync(string filter = ".*\\..*") @@ -202,7 +283,7 @@ namespace CanalSharp.Connections return _currentConn.GetWithoutAckAsync(fetchSize, timeout); } - private SimpleCanalOptions CopyOptions(CanalRunningInfo runningInfo) + private SimpleCanalOptions CopyOptions(CanalServerRunningInfo runningInfo) { var tmpArr = runningInfo.Address.Split(":"); var op = new SimpleCanalOptions(tmpArr[0], int.Parse(tmpArr[1]), _options.ClientId) diff --git a/src/CanalSharp/Connections/Simple/SimpleCanalConnection.cs b/src/CanalSharp/Connections/Simple/SimpleCanalConnection.cs index 0a3b6c6..7d9d185 100644 --- a/src/CanalSharp/Connections/Simple/SimpleCanalConnection.cs +++ b/src/CanalSharp/Connections/Simple/SimpleCanalConnection.cs @@ -1,5 +1,6 @@ using System.Diagnostics.CodeAnalysis; using System.Linq; +using System.Net; using System.Text; using System.Threading.Tasks; using BeetleX; @@ -15,7 +16,7 @@ namespace CanalSharp.Connections /// /// Simple connection, suitable for business scenarios with small data traffic. High Availability is not supported. /// - public class SimpleCanalConnection : ICanalConnection + public class SimpleCanalConnection { private readonly SimpleCanalOptions _options; private readonly ILogger _logger; @@ -45,7 +46,6 @@ namespace CanalSharp.Connections // _client.Socket.ReceiveTimeout = _options.SoTimeout; //Handshake ValidatePackageType(p, PacketType.Handshake, 1); - var handshake = Handshake.Parser.ParseFrom(p.Body); var seed = handshake.Seeds; @@ -336,6 +336,11 @@ namespace CanalSharp.Connections return result; } + internal EndPoint GetLocalEndPoint() + { + return _client.Socket.LocalEndPoint; + } + private void ValidatePackageType(Packet p, PacketType expectPacketType, int expectPacketVersion) { if (p.Type != expectPacketType)