// Copyright © 2014, 2015 Oracle and/or its affiliates. All rights reserved. // // MySQL Connector/NET is licensed under the terms of the GPLv2 // , like most // MySQL Connectors. There are special exceptions to the terms and // conditions of the GPLv2 as it is applied to this software, see the // FLOSS License Exception // . // // This program is free software; you can redistribute it and/or modify // it under the terms of the GNU General Public License as published // by the Free Software Foundation; version 2 of the License. // // This program is distributed in the hope that it will be useful, but // WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY // or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License // for more details. // // You should have received a copy of the GNU General Public License along // with this program; if not, write to the Free Software Foundation, Inc., // 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA using MySql.Data.MySqlClient; using MySql.Data.MySqlClient.Replication; using System; using System.Collections.Generic; using System.Data; using System.Globalization; using System.Linq; using System.Security.Cryptography; using System.Text; namespace MySql.Fabric { internal class FabricServerGroup : ReplicationServerGroup { private MySqlConnection fabricConnection; private string username; private string password; private string fabricInstance; private int ttl; // indexed by group_id private object _lockObject = new object(); internal Dictionary FabricGroups; // indexed by mapping_id internal Dictionary FabricShardTables; // Indexes by "schema.table" internal Dictionary FabricShardTablesPerTable; protected DateTime lastUpdate = DateTime.MinValue; internal string groupIdProperty; internal FabricServerModeEnum? modeProperty; // always in the format "schema.table" internal string tableProperty; internal object keyProperty; ReplicationServer lastServer = null; public FabricServerGroup(string name, int retryTime) : base(name, retryTime) { FabricGroups = new Dictionary(); FabricShardTables = new Dictionary(); FabricShardTablesPerTable = new Dictionary(); } internal protected override ReplicationServer GetServer(bool isMaster) { if (string.IsNullOrEmpty(groupIdProperty) && string.IsNullOrEmpty(tableProperty)) throw new MySqlFabricException(Properties.Resources.errorNotGroupNorTable); FabricServerModeEnum mode = modeProperty.HasValue ? modeProperty.Value : (isMaster ? FabricServerModeEnum.Read_Write : FabricServerModeEnum.Read_only); lock (_lockObject) { /* * Pick a server using this algorithm. * * */ if (string.IsNullOrEmpty(tableProperty)) { return GetServerByGroup(mode, groupIdProperty); } else { ReplicationServer server = GetServerByShard( mode ); // Ensure the database is the current shard db. string[] db = tableProperty.Split('.'); string conStr = server.ConnectionString; if( conStr.IndexOf( "database=" + db[ 0 ] + ";" ) == -1 ) { MySqlConnectionStringBuilder msb = new MySqlConnectionStringBuilder(conStr); msb.Database = db[0]; server.ConnectionString = msb.ToString(); } // return server return server; } } } internal protected override ReplicationServer GetServer(bool isMaster, MySqlConnectionStringBuilder settings) { if (fabricConnection == null) { if (servers.Count == 0) throw new MySqlFabricException(Properties.Resources.errorNoFabricSettings); username = settings.UserID; password = settings.Password; fabricConnection = new MySqlConnection(servers[0].ConnectionString); } bool hasChanged = false; if (groupIdProperty != settings.FabricGroup) hasChanged = true; groupIdProperty = settings.FabricGroup; if (modeProperty != (FabricServerModeEnum?)settings.FabricServerMode) hasChanged = true; modeProperty = (FabricServerModeEnum?)settings.FabricServerMode; if (tableProperty != settings.ShardingTable) hasChanged = true; tableProperty = settings.ShardingTable; if (keyProperty != settings.ShardingKey) hasChanged = true; keyProperty = settings.ShardingKey; if (hasChanged) { GetServerList(); lastServer = GetServer(isMaster); } return lastServer; } private ReplicationServer GetServerByShard(FabricServerModeEnum mode) { FabricShardTable shardTable = FabricShardTablesPerTable[tableProperty]; object key = keyProperty; if (shardTable.TypeShard == FabricShardIndexType.Hash) key = GetMd5Hash(key.ToString()); foreach( FabricShardIndex idx in shardTable.Indexes ) { if ( ShardKeyCompare( key, idx.LowerBound, idx.Type ) >= 0 ) { return GetServerByGroup(mode, idx.GroupId); } } if (shardTable.TypeShard == FabricShardIndexType.Hash) { return GetServerByGroup(mode, shardTable.Indexes.First().GroupId); } return null; } private ReplicationServer GetServerByGroup(FabricServerModeEnum mode, string groupId) { if (!FabricGroups.ContainsKey(groupId)) throw new MySqlFabricException(string.Format(Properties.Resources.errorGroupNotFound, groupId)); var serversInGroup = FabricGroups[groupId].Servers.Where(i => (i.Mode & mode) != 0 && (i.Status == FabricServerStatusEnum.Primary || i.Status == FabricServerStatusEnum.Secondary)).ToList(); if (serversInGroup.Count == 0) return null; double random_weight = new Random().NextDouble() * serversInGroup.Sum(i => i.Weight); double sum_weight = 0.0; foreach (FabricServer server in serversInGroup) { sum_weight += server.Weight; if (sum_weight > random_weight) return server.ReplicationServerInstance; } return serversInGroup.Last().ReplicationServerInstance; } private int ShardKeyCompare(object o1, object o2, FabricShardIndexType shardType) { if (shardType == FabricShardIndexType.Range) { int v; DateTime dt; if (int.TryParse(o1.ToString(), out v)) { return v - int.Parse(o2.ToString()); } else if (DateTime.TryParse(o1.ToString(), out dt)) { DateTime dt2 = DateTime.Parse(o2.ToString()); return DateTime.Compare(dt, dt2); } else { return string.Compare(o1.ToString(), o2.ToString()); } } else if (shardType == FabricShardIndexType.Hash) { return string.Compare(o1.ToString(), o2.ToString()); //return int.Parse(GetTruncatedHexString(o1.ToString()), NumberStyles.HexNumber) - //int.Parse(GetTruncatedHexString(o2.ToString()), NumberStyles.HexNumber); } else throw new NotSupportedException(Enum.GetName(typeof(FabricShardIndexType), FabricShardIndexType.List)); } internal protected override void HandleFailover(ReplicationServer server, Exception exception) { try { ExecuteCommand("threat report_error", server.Name); } catch (Exception ex) { MySqlTrace.LogError(-1, ex.ToString()); } GetServerList(); throw exception; } public void ResetCache() { lastUpdate = DateTime.MinValue; GetServerList(); } protected DataTable ExecuteCommand(string command, params string[] parameters) { DataTable table = null; using (fabricConnection) { fabricConnection.Open(); MySqlCommand mysqlCommand = new MySqlCommand(string.Format("CALL {0}({1})", command.Trim().Replace(' ', '.'), string.Join(",", parameters)), fabricConnection); using (MySqlDataReader reader = mysqlCommand.ExecuteReader()) { reader.Read(); if (fabricInstance == null) fabricInstance = reader.GetString(0); if (fabricInstance != reader.GetString(0)) throw new MySqlFabricException(Properties.Resources.errorFabricUuidChanged); ttl = reader.GetInt32(1); string message = reader.GetValue(2) as string; if (!string.IsNullOrEmpty(message)) throw new MySqlFabricException(message); if (reader.NextResult()) table = CreateTable(reader); } } return table; } protected void GetServerList() { if (lastUpdate.AddSeconds(ttl) > DateTime.Now) return; lock (_lockObject) { try { GetGroups(); GetShards(); lastUpdate = DateTime.Now; } catch (Exception ex) { MySqlTrace.LogError(-1, ex.ToString()); throw new MySqlFabricException(Properties.Resources.errorConnectFabricServer, ex); } } } private void GetShards() { // Gather shard_tables DataTable tblShards = ExecuteCommand("dump shard_tables"); FabricShardTables.Clear(); FabricShardTablesPerTable.Clear(); foreach (DataRow row in tblShards.Rows) { FabricShardTable shardTable = new FabricShardTable() { SchemaName = row["schema_name"] as string, TableName = row["table_name"] as string, ColumnName = row["column_name"] as string, MappingId = int.Parse(row["mapping_id"] as string) }; FabricShardTables.Add(shardTable.MappingId, shardTable); FabricShardTablesPerTable.Add(string.Format("{0}.{1}", shardTable.SchemaName, shardTable.TableName), shardTable); } // Gather shard_index'es DataTable tblShardIndexes = ExecuteCommand("dump shard_index"); foreach (DataRow rowIdx in tblShardIndexes.Rows) { FabricShardIndex Index = new FabricShardIndex() { GroupId = rowIdx["group_id"] as string, LowerBound = rowIdx["lower_bound"] as string, ShardId = int.Parse(rowIdx["shard_id"] as string), MappingId = int.Parse(rowIdx["mapping_id"] as string) }; FabricShardTable table = FabricShardTables[Index.MappingId]; table.Indexes.Add(Index); } // Gather global group DataTable tblGlobal = ExecuteCommand("dump shard_maps"); foreach (DataRow rowGlobal in tblGlobal.Rows) { int mappingId = int.Parse(rowGlobal["mapping_id"] as string); FabricShardTable tbl = FabricShardTables[mappingId]; tbl.GlobalGroupId = rowGlobal["global_group_id"] as string; tbl.TypeShard = (FabricShardIndexType)Enum.Parse(typeof(FabricShardIndexType), rowGlobal["type_name"] as string, true); tbl.Indexes.ForEach(i => i.Type = tbl.TypeShard); } foreach (KeyValuePair kvp in FabricShardTables) { kvp.Value.Indexes.Sort(delegate(FabricShardIndex x, FabricShardIndex y) { return ShardKeyCompare(x.LowerBound, y.LowerBound, x.Type) * -1; }); } } private void GetGroups() { DataTable tblGroups = ExecuteCommand("group lookup_groups"); FabricGroups.Clear(); foreach (DataRow row in tblGroups.Rows) { FabricGroup group = new FabricGroup() { GroupId = row["group_id"] as string, Description = row["description"] as string, FailureDetector = row["failure_detector"] as int?, MasterUuid = row["master_uuid"] as string }; FabricGroups.Add(group.GroupId, group); DataTable tblServers = ExecuteCommand("group lookup_servers", group.GroupId); foreach (DataRow rowSrv in tblServers.Rows) { string[] addressSrv = (rowSrv["address"] as string).Split(':'); string host = addressSrv[0]; int port = int.Parse(addressSrv[1]); FabricServer server = new FabricServer( new Guid(rowSrv["server_uuid"] as string), group.GroupId, host, port, (FabricServerModeEnum)Enum.Parse(typeof(FabricServerModeEnum), rowSrv["mode"] as string, true), (FabricServerStatusEnum)Enum.Parse(typeof(FabricServerStatusEnum), rowSrv["status"] as string, true), float.Parse(rowSrv["weight"] as string), username, password); group.Servers.Add(server); } } } protected DataTable CreateTable(MySqlDataReader reader) { DataTable table = new DataTable(); for (int i = 0; i < reader.FieldCount; i++) { table.Columns.Add(reader.GetName(i)); } while (reader.Read()) { DataRow row = table.NewRow(); table.Rows.Add(row); for (int i = 0; i < reader.FieldCount; i++) { row[i] = reader.GetValue(i); } } return table; } protected string GetMd5Hash(string input) { MD5 md5 = MD5.Create(); byte[] buffer = Encoding.UTF8.GetBytes(input); byte[] output = md5.ComputeHash(buffer); return BitConverter.ToString(output).Replace("-", ""); } protected string GetTruncatedHexString(string hexString) { int max_lenght = 7; if (hexString.Length > max_lenght) return hexString.Substring(0, max_lenght); return hexString; } } }