mirror of
https://github.com/zerotier/ZeroTierOne.git
synced 2025-05-08 22:53:43 +02:00
361 lines
11 KiB
C++
361 lines
11 KiB
C++
/**************************************************************************
|
|
Copyright (c) 2017 sewenew
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*************************************************************************/
|
|
|
|
#include "sentinel.h"
|
|
#include <cassert>
|
|
#include <thread>
|
|
#include <random>
|
|
#include <algorithm>
|
|
#include "redis.h"
|
|
#include "errors.h"
|
|
|
|
namespace sw {
|
|
|
|
namespace redis {
|
|
|
|
class Sentinel::Iterator {
|
|
public:
|
|
Iterator(std::list<Connection> &healthy_sentinels,
|
|
std::list<ConnectionOptions> &broken_sentinels) :
|
|
_healthy_sentinels(healthy_sentinels),
|
|
_broken_sentinels(broken_sentinels) {
|
|
reset();
|
|
}
|
|
|
|
Connection& next();
|
|
|
|
void reset();
|
|
|
|
private:
|
|
std::list<Connection> &_healthy_sentinels;
|
|
|
|
std::size_t _healthy_size = 0;
|
|
|
|
std::list<ConnectionOptions> &_broken_sentinels;
|
|
|
|
std::size_t _broken_size = 0;
|
|
};
|
|
|
|
Connection& Sentinel::Iterator::next() {
|
|
while (_healthy_size > 0) {
|
|
assert(_healthy_sentinels.size() >= _healthy_size);
|
|
|
|
--_healthy_size;
|
|
|
|
auto &connection = _healthy_sentinels.front();
|
|
if (connection.broken()) {
|
|
_broken_sentinels.push_front(connection.options());
|
|
++_broken_size;
|
|
|
|
_healthy_sentinels.pop_front();
|
|
} else {
|
|
_healthy_sentinels.splice(_healthy_sentinels.end(),
|
|
_healthy_sentinels,
|
|
_healthy_sentinels.begin());
|
|
|
|
return _healthy_sentinels.back();
|
|
}
|
|
}
|
|
|
|
while (_broken_size > 0) {
|
|
assert(_broken_sentinels.size() >= _broken_size);
|
|
|
|
--_broken_size;
|
|
|
|
try {
|
|
const auto &opt = _broken_sentinels.front();
|
|
Connection connection(opt);
|
|
_healthy_sentinels.push_back(std::move(connection));
|
|
|
|
_broken_sentinels.pop_front();
|
|
|
|
return _healthy_sentinels.back();
|
|
} catch (const Error &e) {
|
|
// Failed to connect to sentinel.
|
|
_broken_sentinels.splice(_broken_sentinels.end(),
|
|
_broken_sentinels,
|
|
_broken_sentinels.begin());
|
|
}
|
|
}
|
|
|
|
throw StopIterError();
|
|
}
|
|
|
|
void Sentinel::Iterator::reset() {
|
|
_healthy_size = _healthy_sentinels.size();
|
|
_broken_size = _broken_sentinels.size();
|
|
}
|
|
|
|
Sentinel::Sentinel(const SentinelOptions &sentinel_opts) :
|
|
_broken_sentinels(_parse_options(sentinel_opts)),
|
|
_sentinel_opts(sentinel_opts) {
|
|
if (_sentinel_opts.connect_timeout == std::chrono::milliseconds(0)
|
|
|| _sentinel_opts.socket_timeout == std::chrono::milliseconds(0)) {
|
|
throw Error("With sentinel, connection timeout and socket timeout cannot be 0");
|
|
}
|
|
}
|
|
|
|
Connection Sentinel::master(const std::string &master_name, const ConnectionOptions &opts) {
|
|
std::lock_guard<std::mutex> lock(_mutex);
|
|
|
|
Iterator iter(_healthy_sentinels, _broken_sentinels);
|
|
std::size_t retries = 0;
|
|
while (true) {
|
|
try {
|
|
auto &sentinel = iter.next();
|
|
|
|
auto master = _get_master_addr_by_name(sentinel, master_name);
|
|
if (!master) {
|
|
// Try the next sentinel.
|
|
continue;
|
|
}
|
|
|
|
auto connection = _connect_redis(*master, opts);
|
|
if (_get_role(connection) != Role::MASTER) {
|
|
// Retry the whole process at most SentinelOptions::max_retry times.
|
|
++retries;
|
|
if (retries > _sentinel_opts.max_retry) {
|
|
throw Error("Failed to get master from sentinel");
|
|
}
|
|
|
|
std::this_thread::sleep_for(_sentinel_opts.retry_interval);
|
|
|
|
// Restart the iteration.
|
|
iter.reset();
|
|
continue;
|
|
}
|
|
|
|
return connection;
|
|
} catch (const StopIterError &e) {
|
|
throw;
|
|
} catch (const Error &e) {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
|
|
Connection Sentinel::slave(const std::string &master_name, const ConnectionOptions &opts) {
|
|
std::lock_guard<std::mutex> lock(_mutex);
|
|
|
|
Iterator iter(_healthy_sentinels, _broken_sentinels);
|
|
std::size_t retries = 0;
|
|
while (true) {
|
|
try {
|
|
auto &sentinel = iter.next();
|
|
|
|
auto slaves = _get_slave_addr_by_name(sentinel, master_name);
|
|
if (slaves.empty()) {
|
|
// Try the next sentinel.
|
|
continue;
|
|
}
|
|
|
|
// Normally slaves list is NOT very large, so there won't be a performance problem.
|
|
auto slave_iter = std::find(slaves.begin(),
|
|
slaves.end(),
|
|
Node{opts.host, opts.port});
|
|
if (slave_iter != slaves.end() && slave_iter != slaves.begin()) {
|
|
// The given node is still a valid slave. Try it first.
|
|
std::swap(*(slaves.begin()), *slave_iter);
|
|
}
|
|
|
|
for (const auto &slave : slaves) {
|
|
try {
|
|
auto connection = _connect_redis(slave, opts);
|
|
if (_get_role(connection) != Role::SLAVE) {
|
|
// Retry the whole process at most SentinelOptions::max_retry times.
|
|
++retries;
|
|
if (retries > _sentinel_opts.max_retry) {
|
|
throw Error("Failed to get slave from sentinel");
|
|
}
|
|
|
|
std::this_thread::sleep_for(_sentinel_opts.retry_interval);
|
|
|
|
// Restart the iteration.
|
|
iter.reset();
|
|
break;
|
|
}
|
|
|
|
return connection;
|
|
} catch (const Error &e) {
|
|
// Try the next slave.
|
|
continue;
|
|
}
|
|
}
|
|
} catch (const StopIterError &e) {
|
|
throw;
|
|
} catch (const Error &e) {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
|
|
Optional<Node> Sentinel::_get_master_addr_by_name(Connection &connection, const StringView &name) {
|
|
connection.send("SENTINEL GET-MASTER-ADDR-BY-NAME %b", name.data(), name.size());
|
|
|
|
auto reply = connection.recv();
|
|
|
|
assert(reply);
|
|
|
|
auto master = reply::parse<Optional<std::pair<std::string, std::string>>>(*reply);
|
|
if (!master) {
|
|
return {};
|
|
}
|
|
|
|
int port = 0;
|
|
try {
|
|
port = std::stoi(master->second);
|
|
} catch (const std::exception &) {
|
|
throw ProtoError("Master port is invalid: " + master->second);
|
|
}
|
|
|
|
return Optional<Node>{Node{master->first, port}};
|
|
}
|
|
|
|
std::vector<Node> Sentinel::_get_slave_addr_by_name(Connection &connection,
|
|
const StringView &name) {
|
|
try {
|
|
connection.send("SENTINEL SLAVES %b", name.data(), name.size());
|
|
|
|
auto reply = connection.recv();
|
|
|
|
assert(reply);
|
|
|
|
auto slaves = _parse_slave_info(*reply);
|
|
|
|
// Make slave list random.
|
|
std::mt19937 gen(std::random_device{}());
|
|
std::shuffle(slaves.begin(), slaves.end(), gen);
|
|
|
|
return slaves;
|
|
} catch (const ReplyError &e) {
|
|
// Unknown master name.
|
|
return {};
|
|
}
|
|
}
|
|
|
|
std::vector<Node> Sentinel::_parse_slave_info(redisReply &reply) const {
|
|
using SlaveInfo = std::unordered_map<std::string, std::string>;
|
|
|
|
auto slaves = reply::parse<std::vector<SlaveInfo>>(reply);
|
|
|
|
std::vector<Node> nodes;
|
|
for (const auto &slave : slaves) {
|
|
auto flags_iter = slave.find("flags");
|
|
auto ip_iter = slave.find("ip");
|
|
auto port_iter = slave.find("port");
|
|
if (flags_iter == slave.end() || ip_iter == slave.end() || port_iter == slave.end()) {
|
|
throw ProtoError("Invalid slave info");
|
|
}
|
|
|
|
// This slave is down, e.g. 's_down,slave,disconnected'
|
|
if (flags_iter->second != "slave") {
|
|
continue;
|
|
}
|
|
|
|
int port = 0;
|
|
try {
|
|
port = std::stoi(port_iter->second);
|
|
} catch (const std::exception &) {
|
|
throw ProtoError("Slave port is invalid: " + port_iter->second);
|
|
}
|
|
|
|
nodes.push_back(Node{ip_iter->second, port});
|
|
}
|
|
|
|
return nodes;
|
|
}
|
|
|
|
Connection Sentinel::_connect_redis(const Node &node, ConnectionOptions opts) {
|
|
opts.host = node.host;
|
|
opts.port = node.port;
|
|
|
|
return Connection(opts);
|
|
}
|
|
|
|
Role Sentinel::_get_role(Connection &connection) {
|
|
connection.send("INFO REPLICATION");
|
|
auto reply = connection.recv();
|
|
|
|
assert(reply);
|
|
auto info = reply::parse<std::string>(*reply);
|
|
|
|
auto start = info.find("role:");
|
|
if (start == std::string::npos) {
|
|
throw ProtoError("Invalid INFO REPLICATION reply");
|
|
}
|
|
start += 5;
|
|
auto stop = info.find("\r\n", start);
|
|
if (stop == std::string::npos) {
|
|
throw ProtoError("Invalid INFO REPLICATION reply");
|
|
}
|
|
|
|
auto role = info.substr(start, stop - start);
|
|
if (role == "master") {
|
|
return Role::MASTER;
|
|
} else if (role == "slave") {
|
|
return Role::SLAVE;
|
|
} else {
|
|
throw Error("Invalid role: " + role);
|
|
}
|
|
}
|
|
|
|
std::list<ConnectionOptions> Sentinel::_parse_options(const SentinelOptions &opts) const {
|
|
std::list<ConnectionOptions> options;
|
|
for (const auto &node : opts.nodes) {
|
|
ConnectionOptions opt;
|
|
opt.host = node.first;
|
|
opt.port = node.second;
|
|
opt.password = opts.password;
|
|
opt.keep_alive = opts.keep_alive;
|
|
opt.connect_timeout = opts.connect_timeout;
|
|
opt.socket_timeout = opts.socket_timeout;
|
|
|
|
options.push_back(opt);
|
|
}
|
|
|
|
return options;
|
|
}
|
|
|
|
SimpleSentinel::SimpleSentinel(const std::shared_ptr<Sentinel> &sentinel,
|
|
const std::string &master_name,
|
|
Role role) :
|
|
_sentinel(sentinel),
|
|
_master_name(master_name),
|
|
_role(role) {
|
|
if (!_sentinel) {
|
|
throw Error("Sentinel cannot be null");
|
|
}
|
|
|
|
if (_role != Role::MASTER && _role != Role::SLAVE) {
|
|
throw Error("Role must be Role::MASTER or Role::SLAVE");
|
|
}
|
|
}
|
|
|
|
Connection SimpleSentinel::create(const ConnectionOptions &opts) {
|
|
assert(_sentinel);
|
|
|
|
if (_role == Role::MASTER) {
|
|
return _sentinel->master(_master_name, opts);
|
|
}
|
|
|
|
assert(_role == Role::SLAVE);
|
|
|
|
return _sentinel->slave(_master_name, opts);
|
|
}
|
|
|
|
}
|
|
|
|
}
|