Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Bugfix] Compatibility with cocaine/cocaine-core#191. #64

Merged
merged 1 commit into from
Apr 7, 2015
Merged

[Bugfix] Compatibility with cocaine/cocaine-core#191. #64

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 46 additions & 53 deletions ipvs/gateway.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,8 +82,9 @@ class ipvs_t::remote_t {
COCAINE_DECLARE_NONCOPYABLE(remote_t)

struct info_t {
const std::string name;
metadata_t meta;
std::string service;
std::size_t version;
std::vector<tcp::endpoint> endpoints;
};

ipvs_t *const parent;
Expand All @@ -96,42 +97,40 @@ class ipvs_t::remote_t {
std::multimap<std::string, ipvs_dest_t> backends;

public:
remote_t(ipvs_t* parent, const std::string& name, unsigned int version, const graph_root_t& graph);
remote_t(ipvs_t* parent, const partition_t& name);
~remote_t();

// Observers

auto
reduce() const -> metadata_t;
reduce() const -> std::vector<tcp::endpoint>;

// Modifiers

void
size_t
insert(const std::string& uuid, const std::vector<tcp::endpoint>& endpoints);

void
size_t
remove(const std::string& uuid);

private:
void
format_address(union nf_inet_addr& target, const tcp::endpoint& endpoint);
};

ipvs_t::remote_t::remote_t(ipvs_t *const parent_, const std::string& name_, unsigned int version,
const graph_root_t& graph)
:
ipvs_t::remote_t::remote_t(ipvs_t *const parent_, const partition_t& name_):
parent(parent_),
info({name_, metadata_t({}, version, graph)})
info({std::get<0>(name_), std::get<1>(name_), {}})
{
ipvs_service_t handle;

scoped_attributes_t attributes(*parent->m_log, {
attribute::make("service", info.name)
attribute::make("service", info.service),
attribute::make("version", info.version)
});

auto port = parent->m_context.mapper.assign(info.name + ":virtual");
auto& endpoints = std::get<0>(info.meta);

auto port = parent->m_context.mapper.assign(cocaine::format("%s@%d:virtual", info.service,
info.version));
COCAINE_LOG_DEBUG(parent->m_log, "publishing virtual service");

for(auto it = parent->m_endpoints.begin(); it != parent->m_endpoints.end(); ++it) {
Expand Down Expand Up @@ -167,7 +166,7 @@ ipvs_t::remote_t::remote_t(ipvs_t *const parent_, const std::string& name_, unsi
}

// Store the endpoint.
endpoints.push_back(endpoint);
info.endpoints.push_back(endpoint);

// Store the kernel service handle.
services[handle.af] = handle;
Expand All @@ -177,17 +176,18 @@ ipvs_t::remote_t::remote_t(ipvs_t *const parent_, const std::string& name_, unsi
COCAINE_LOG_ERROR(parent->m_log, "no valid endpoints found for virtual service");

// Force disconnect the remote node.
throw std::system_error(EADDRNOTAVAIL, std::generic_category());
throw std::system_error(EADDRNOTAVAIL, std::system_category());
}

COCAINE_LOG_INFO(parent->m_log, "virtual service published on port %d with %d endpoint(s)",
port, endpoints.size()
port, info.endpoints.size()
);
}

ipvs_t::remote_t::~remote_t() {
COCAINE_LOG_DEBUG(parent->m_log, "cleaning up virtual service")(
"service", info.name
"service", info.service,
"version", info.version
);

for(auto it = backends.begin(); it != backends.end(); ++it) {
Expand Down Expand Up @@ -225,28 +225,28 @@ struct is_serving {
} // namespace

auto
ipvs_t::remote_t::reduce() const -> metadata_t {
ipvs_t::remote_t::reduce() const -> std::vector<tcp::endpoint> {
if(backends.empty()) {
throw std::system_error(error::service_not_available);
}

auto endpoints = std::vector<tcp::endpoint>();
auto builder = std::back_inserter(endpoints);
std::vector<tcp::endpoint> endpoints;

std::copy_if(std::get<0>(info.meta).begin(), std::get<0>(info.meta).end(), builder,
std::copy_if(info.endpoints.begin(), info.endpoints.end(), std::back_inserter(endpoints),
is_serving{backends}
);

return metadata_t { endpoints, std::get<1>(info.meta), std::get<2>(info.meta) };
return endpoints;
}

void
size_t
ipvs_t::remote_t::insert(const std::string& uuid, const std::vector<tcp::endpoint>& endpoints) {
ipvs_dest_t handle;

scoped_attributes_t attributes(*parent->m_log, {
attribute::make("service", info.name),
attribute::make("uuid", uuid)
attribute::make("service", info.service),
attribute::make("uuid", uuid),
attribute::make("version", info.version)
});

std::map<int, ipvs_dest_t> handles;
Expand Down Expand Up @@ -289,25 +289,24 @@ ipvs_t::remote_t::insert(const std::string& uuid, const std::vector<tcp::endpoin
COCAINE_LOG_ERROR(parent->m_log, "no valid endpoints found for destination");

// Force disconnect the remote node.
throw std::system_error(EHOSTUNREACH, std::generic_category());
throw std::system_error(EHOSTUNREACH, std::system_category());
}

for(auto it = handles.begin(); it != handles.end(); ++it) {
backends.insert({uuid, it->second});
}

COCAINE_LOG_INFO(parent->m_log, "destination registered with %d endpoint(s)", handles.size());

return backends.size();
}

void
size_t
ipvs_t::remote_t::remove(const std::string& uuid) {
if(!backends.count(uuid)) {
return;
}

COCAINE_LOG_INFO(parent->m_log, "removing destination with %d endpoint(s)", backends.count(uuid))(
"service", info.name,
"uuid", uuid
"service", info.service,
"uuid", uuid,
"version", info.version
);

for(auto it = backends.lower_bound(uuid); it != backends.upper_bound(uuid); ++it) {
Expand All @@ -319,7 +318,7 @@ ipvs_t::remote_t::remove(const std::string& uuid) {
}
}

backends.erase(uuid);
backends.erase(uuid); return backends.size();
}

void
Expand Down Expand Up @@ -419,44 +418,38 @@ ipvs_t::~ipvs_t() {
}

auto
ipvs_t::resolve(const std::string& name) const -> metadata_t {
ipvs_t::resolve(const partition_t& name) const -> std::vector<tcp::endpoint> {
auto ptr = m_remotes.synchronize();

if(!ptr->count(name)) {
throw std::system_error(error::service_not_available);
}

COCAINE_LOG_DEBUG(m_log, "providing service using virtual node")(
"service", name
);
COCAINE_LOG_DEBUG(m_log, "providing service using virtual node");

return ptr->at(name)->reduce();
}

void
ipvs_t::consume(const std::string& uuid, const std::string& name, const metadata_t& info) {
auto endpoints = std::vector<tcp::endpoint>();
auto graph = graph_root_t();
auto version = 0;

std::tie(endpoints, version, graph) = info;

size_t
ipvs_t::consume(const std::string& uuid,
const partition_t& name, const std::vector<tcp::endpoint>& endpoints)
{
auto ptr = m_remotes.synchronize();

if(!ptr->count(name)) {
(*ptr)[name] = std::make_unique<remote_t>(this, name, version, graph);
(*ptr)[name] = std::make_unique<remote_t>(this, name);
}

ptr->at(name)->insert(uuid, endpoints);
return ptr->at(name)->insert(uuid, endpoints);
}

void
ipvs_t::cleanup(const std::string& uuid, const std::string& name) {
size_t
ipvs_t::cleanup(const std::string& uuid, const partition_t& name) {
auto ptr = m_remotes.synchronize();

if(!ptr->count(name)) {
return;
return 0;
}

ptr->at(name)->remove(uuid);
return ptr->at(name)->remove(uuid);
}
18 changes: 11 additions & 7 deletions ipvs/gateway.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,8 @@

namespace cocaine { namespace gateway {

class ipvs_config_t {
class ipvs_config_t
{
public:
std::string scheduler;
unsigned int weight;
Expand All @@ -36,6 +37,8 @@ class ipvs_t:
{
class remote_t;

typedef std::map<partition_t, std::unique_ptr<remote_t>> remote_map_t;

context_t& m_context;

const std::unique_ptr<logging::log_t> m_log;
Expand All @@ -45,7 +48,7 @@ class ipvs_t:
std::vector<asio::ip::address> m_endpoints;

// Keeps track of IPVS configuration.
synchronized<std::map<std::string, std::unique_ptr<remote_t>>> m_remotes;
synchronized<remote_map_t> m_remotes;

public:
ipvs_t(context_t& context, const std::string& name, const dynamic_t& args);
Expand All @@ -55,15 +58,16 @@ class ipvs_t:

virtual
auto
resolve(const std::string& name) const -> metadata_t;
resolve(const partition_t& name) const -> std::vector<asio::ip::tcp::endpoint>;

virtual
void
consume(const std::string& uuid, const std::string& name, const metadata_t& info);
size_t
consume(const std::string& uuid,
const partition_t& name, const std::vector<asio::ip::tcp::endpoint>& endpoints);

virtual
void
cleanup(const std::string& uuid, const std::string& name);
size_t
cleanup(const std::string& uuid, const partition_t& name);
};

}} // namespace cocaine::gateway
Expand Down