Skip to content
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Commit b224828

Browse files
antmatantmat
committedJun 22, 2017
fix: adapt elliptics inside plugins(again)
1 parent a1119d0 commit b224828

9 files changed

+115
-120
lines changed
 

‎CMakeLists.txt

+2
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ OPTION(CACHE "Build LRU Cache plugin" ON)
1414
OPTION(CHRONO "Build Chrono Service plugin" ON)
1515
OPTION(DOCKER "Build Docker Isolation plugin" ON)
1616
OPTION(ELASTICSEARCH "Build ElasticSearch Service plugin" ON)
17+
OPTION(ELLIPTICS "Build Elliptics Storage Service plugin" ON)
1718
OPTION(GRAPHITE "Build Graphite Service plugin" ON)
1819
OPTION(IPVS "Build IPVS Gateway Service plugin" ON)
1920
OPTION(LOGGING "Build logging v2 service plugin" ON)
@@ -60,6 +61,7 @@ SET(PLUGINS
6061
chrono
6162
docker
6263
elasticsearch
64+
elliptics
6365
graphite
6466
ipvs
6567
logging

‎debian/control

+20
Original file line numberDiff line numberDiff line change
@@ -287,3 +287,23 @@ Architecture: any
287287
Depends: ${shlibs:Depends}, ${misc:Depends}
288288
Description: Vicodyn gateway (dev files)
289289
Vicodyn gateway (dev files)
290+
291+
Package: libcocaine-plugin-elliptics3
292+
Architecture: any
293+
Depends: ${shlibs:Depends}, ${misc:Depends}, elliptics-client2.27 (>= 2.27.2.5), libcocaine-core3 (>= 0.12.14)
294+
Conflicts: libcocaine-plugin-elliptics
295+
Replaces: libcocaine-plugin-elliptics
296+
Description: Distributed hash table storage (cocaine plugin)
297+
Elliptics network is a fault tolerant distributed hash table object storage.
298+
299+
Package: libcocaine-plugin-elliptics-dbg
300+
Architecture: any
301+
Depends: ${misc:Depends}, libcocaine-plugin-elliptics3 (= ${binary:Version})
302+
Description: Distributed hash table storage (cocaine plugin debug files)
303+
Elliptics network is a fault tolerant distributed hash table object storage.
304+
305+
Package: libcocaine-plugin-elliptics-dev
306+
Architecture: any
307+
Depends: ${shlibs:Depends}, ${misc:Depends},
308+
Description: Distributed hash table storage (cocaine plugin includes)
309+
Elliptics network is a fault tolerant distributed hash table object storage.

‎debian/libcocaine-plugin-elliptics-dev.install

+2
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
usr/include/cocaine/services/elliptics.hpp
2+
usr/include/cocaine/elliptics_logger.hpp

‎debian/libcocaine-plugin-elliptics3.install

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
usr/lib/cocaine/elliptics-extensions.cocaine-plugin.*

‎elliptics/CMakeLists.txt

+1-11
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,8 @@
11
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
22

3-
PROJECT(COCAINE-PLUGIN-ELLIPTICS)
4-
5-
IF("${CMAKE_SHARED_LIBRARY_SUFFIX}" STREQUAL "")
6-
SET(COCAINE_PLUGIN_SUFFIX ".cocaine-plugin")
7-
ELSE()
8-
SET(COCAINE_PLUGIN_SUFFIX ".cocaine-plugin${CMAKE_SHARED_LIBRARY_SUFFIX}")
9-
ENDIF()
10-
113
include_directories(
124
${PROJECT_SOURCE_DIR}/foreign/asio/asio/include
13-
${PROJECT_SOURCE_DIR}/foreign/blackhole/src
14-
${PROJECT_SOURCE_DIR}/include
5+
${PROJECT_SOURCE_DIR}/elliptics/include
156
)
167

178
ADD_LIBRARY(elliptics-extensions MODULE
@@ -24,7 +15,6 @@ TARGET_LINK_LIBRARIES(elliptics-extensions
2415
cocaine-core
2516
cocaine-io-util
2617
elliptics_client
27-
elliptics_cpp
2818
${Boost_LIBRARIES})
2919

3020
SET_TARGET_PROPERTIES(elliptics-extensions PROPERTIES

‎elliptics/include/cocaine/elliptics_logger.hpp

+52
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
#pragma once
2+
3+
#include <blackhole/logger.hpp>
4+
namespace cocaine {
5+
6+
class elliptics_logger_t: public blackhole::logger_t {
7+
std::unique_ptr<blackhole::logger_t> inner;
8+
9+
public:
10+
struct scope_t {
11+
scope_t() {
12+
prev_trace = trace_t::current();
13+
// We intentionally zero everything, as elliptics client writes itself trace_id attribute
14+
trace_t::current() = trace_t(0, 0, 0, dnet_logger_get_trace_bit(), {});
15+
}
16+
~scope_t() {
17+
trace_t::current() = prev_trace;
18+
}
19+
trace_t prev_trace;
20+
};
21+
22+
elliptics_logger_t(std::unique_ptr<blackhole::logger_t> log):
23+
inner(std::move(log))
24+
{}
25+
26+
auto map_severity(blackhole::severity_t severity) -> blackhole::severity_t {
27+
if(severity) {
28+
severity = severity - 1;
29+
}
30+
return severity;
31+
}
32+
33+
auto log(blackhole::severity_t severity, const blackhole::message_t& message) -> void {
34+
scope_t scope;
35+
return inner->log(map_severity(severity), message);
36+
}
37+
auto log(blackhole::severity_t severity, const blackhole::message_t& message, blackhole::attribute_pack& pack) -> void {
38+
scope_t scope;
39+
return inner->log(map_severity(severity), message, pack);
40+
}
41+
auto log(blackhole::severity_t severity, const blackhole::lazy_message_t& message, blackhole::attribute_pack& pack) -> void {
42+
scope_t scope;
43+
return inner->log(map_severity(severity), message, pack);
44+
}
45+
46+
auto manager() -> blackhole::scope::manager_t& {
47+
scope_t scope;
48+
return inner->manager();
49+
}
50+
};
51+
52+
} // namespace cocaine

‎elliptics/include/cocaine/storage.hpp

+4-16
Original file line numberDiff line numberDiff line change
@@ -36,18 +36,9 @@ class elliptics_service_t;
3636

3737
namespace cocaine { namespace storage {
3838

39-
/*
40-
* Wraps the v1.0 logger into the v0.2 interface to be usable within Elliptics.
41-
*/
42-
class log_adapter_t : public ioremap::elliptics::logger_base {
43-
public:
44-
log_adapter_t(std::shared_ptr<logging::logger_t> wrapped, ioremap::elliptics::log_level severity);
45-
};
46-
4739
class elliptics_storage_t : public api::storage_t {
4840
public:
4941
typedef api::storage_t category_type;
50-
typedef std::shared_ptr<logging::logger_t> log_ptr;
5142
typedef std::map<dnet_raw_id, std::string, ioremap::elliptics::dnet_raw_id_less_than<>> key_name_map;
5243

5344
elliptics_storage_t(context_t &context, const std::string &name, const dynamic_t &args);
@@ -71,20 +62,17 @@ class elliptics_storage_t : public api::storage_t {
7162
const std::vector<std::string> &blobs);
7263
ioremap::elliptics::async_read_result async_read_latest(const std::string &collection, const std::string &key);
7364

74-
static std::vector<std::string> convert_list_result(const ioremap::elliptics::sync_find_indexes_result &result);
75-
7665
private:
77-
std::string id(const std::string &collection,
78-
const std::string &key)
79-
{
66+
std::string id(const std::string &collection, const std::string &key) {
8067
return collection + '\0' + key;
8168
}
69+
std::unique_ptr<logging::logger_t> make_elliptics_logger(const std::string& name) const;
70+
ioremap::elliptics::session prepare_session(const std::string collection, int timeout);
8271

8372
private:
8473
context_t &m_context;
85-
log_ptr m_log;
74+
std::unique_ptr<logging::logger_t> m_log;
8675

87-
log_adapter_t m_log_adapter;
8876
// Perform read latest operation on read request.
8977
bool m_read_latest;
9078
dnet_config m_config;

‎elliptics/src/service.cpp

-9
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,6 @@
2323

2424
#include "cocaine/tuple.hpp"
2525

26-
#define debug() if (1) {} else std::cerr
27-
//#define debug() std::cerr << __PRETTY_FUNCTION__ << ": " << __LINE__ << " "
28-
2926
namespace cocaine {
3027

3128
namespace ph = std::placeholders;
@@ -38,8 +35,6 @@ elliptics_service_t::elliptics_service_t(context_t &context, asio::io_service &r
3835
, dispatch<io::elliptics_tag>(name)
3936
, m_storage(api::storage(context, args.as_object().at("source", "core").as_string()))
4037
, m_elliptics(dynamic_cast<storage::elliptics_storage_t *>(m_storage.get())) {
41-
debug() << m_elliptics << std::endl;
42-
4338
if (!m_elliptics) {
4439
throw std::system_error(-1, std::generic_category(),
4540
"to be able to use elliptics service, storage must be also elliptics");
@@ -56,7 +51,6 @@ elliptics_service_t::elliptics_service_t(context_t &context, asio::io_service &r
5651
}
5752

5853
deferred<std::string> elliptics_service_t::read(const std::string &collection, const std::string &key) {
59-
debug() << "read, collection: " << collection << ", key: " << key << std::endl;
6054
deferred<std::string> promise;
6155

6256
m_elliptics->async_read(collection, key).connect(std::bind(&on_read_completed,
@@ -66,7 +60,6 @@ deferred<std::string> elliptics_service_t::read(const std::string &collection, c
6660
}
6761

6862
deferred<std::string> elliptics_service_t::read_latest(const std::string &collection, const std::string &key) {
69-
debug() << "read_latest, collection: " << collection << ", key: " << key << std::endl;
7063
deferred<std::string> promise;
7164

7265
m_elliptics->async_read_latest(collection, key).connect(std::bind(&on_read_completed,
@@ -77,7 +70,6 @@ deferred<std::string> elliptics_service_t::read_latest(const std::string &collec
7770

7871
deferred<void> elliptics_service_t::write(const std::string &collection, const std::string &key,
7972
const std::string &blob, const std::vector<std::string> &tags) {
80-
debug() << "write, collection: " << collection << ", key: " << key << std::endl;
8173
deferred<void> promise;
8274

8375
m_elliptics->async_write(collection, key, blob, tags).connect(std::bind(&on_write_completed,
@@ -97,7 +89,6 @@ deferred<std::vector<std::string>> elliptics_service_t::find(const std::string &
9789
}
9890

9991
deferred<void> elliptics_service_t::remove(const std::string &collection, const std::string &key) {
100-
debug() << "remove, collection: " << collection << ", key: " << key << std::endl;
10192
deferred<void> promise;
10293

10394
m_elliptics->async_remove(collection, key).connect(std::bind(&on_remove_completed,

‎elliptics/src/storage.cpp

+33-84
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,9 @@
1818
* along with Elliptics. If not, see <http://www.gnu.org/licenses/>.
1919
*/
2020

21-
#include <cocaine/storage.hpp>
21+
#include "cocaine/storage.hpp"
22+
23+
#include "cocaine/elliptics_logger.hpp"
2224

2325
#include <cocaine/context.hpp>
2426
#include <cocaine/dynamic.hpp>
@@ -27,6 +29,7 @@
2729
#include <blackhole/formatter/string.hpp>
2830
#include <blackhole/v1/attribute.hpp>
2931
#include <blackhole/v1/logger.hpp>
32+
#include <cocaine/storage.hpp>
3033

3134
namespace cocaine { namespace storage {
3235

@@ -37,46 +40,6 @@ using namespace cocaine::logging;
3740
using namespace cocaine::storage;
3841
namespace ell = ioremap::elliptics;
3942

40-
auto map_severity(blackhole::severity_t severity) -> blackhole::severity_t {
41-
if(severity) {
42-
severity -= 1;
43-
}
44-
return severity;
45-
}
46-
47-
48-
class frontend_t : public blackhole::base_frontend_t {
49-
public:
50-
frontend_t(std::shared_ptr<logging::logger_t> log, ell::log_level severity)
51-
: log(std::move(log))
52-
, severity(severity)
53-
, formatter("%(message)s %(...::)s") {}
54-
55-
virtual void handle(const blackhole::log::record_t& record) {
56-
const auto level =
57-
record.extract<dnet_log_level>(blackhole::keyword::severity<dnet_log_level>().name());
58-
59-
if (level < severity) {
60-
return;
61-
}
62-
63-
const auto mapped = convert(level);
64-
65-
log->log(static_cast<int>(mapped), formatter.format(record));
66-
}
67-
68-
private:
69-
std::shared_ptr<logging::logger_t> log;
70-
ell::log_level severity;
71-
blackhole::formatter::string_t formatter;
72-
};
73-
74-
log_adapter_t::log_adapter_t(std::shared_ptr<logging::logger_t> log, ell::log_level level)
75-
: ell::logger_base() {
76-
verbosity(DNET_LOG_DEBUG);
77-
add_frontend(std::unique_ptr<frontend_t>(new frontend_t(log, level)));
78-
}
79-
8043
namespace {
8144

8245
dnet_config parse_json_config(const dynamic_t::object_t& args) {
@@ -94,16 +57,18 @@ dnet_config parse_json_config(const dynamic_t::object_t& args) {
9457

9558
}
9659

60+
std::unique_ptr<logging::logger_t> elliptics_storage_t::make_elliptics_logger(const std::string& name) const {
61+
auto logger_name = format("storage/{}/elliptics_client", name);
62+
return std::unique_ptr<logging::logger_t>(new elliptics_logger_t(m_context.log(logger_name)));
63+
}
64+
9765
elliptics_storage_t::elliptics_storage_t(context_t &context, const std::string &name, const dynamic_t &args)
9866
: category_type(context, name, args)
9967
, m_context(context)
100-
, m_log(context.log(name)) // TODO: It was with attributes: {{"storage", "elliptics"}}.
101-
, // XXX: dynamic_t from cocaine can't convert int to uint, and DNET_LOG_INFO being an enum value is int
102-
m_log_adapter(m_log, static_cast<ioremap::elliptics::log_level>(
103-
args.as_object().at("verbosity", uint(DNET_LOG_INFO)).as_uint()))
68+
, m_log(context.log(name, {{"storage", "elliptics"}}))
10469
, m_read_latest(args.as_object().at("read_latest", false).as_bool())
10570
, m_config(parse_json_config(args.as_object()))
106-
, m_node(ell::logger(m_log_adapter, blackhole::log::attributes_t{{"storage", {"elliptics"}}}), m_config)
71+
, m_node(make_elliptics_logger(name), m_config)
10772
, m_session(m_node) {
10873
dynamic_t::array_t nodes = args.as_object().at("nodes").as_array();
10974

@@ -224,29 +189,31 @@ void elliptics_storage_t::remove(const std::string &collection, const std::strin
224189
});
225190
}
226191

227-
ell::async_read_result elliptics_storage_t::async_read(const std::string &collection, const std::string &key)
228-
{
229-
using namespace std::placeholders;
230-
231-
COCAINE_LOG_DEBUG(m_log, "reading the '{}' object, collection: '{}'", key, collection);
232-
192+
ell::session elliptics_storage_t::prepare_session(const std::string collection, int timeout) {
233193
ell::session session = m_session.clone();
234194
session.set_namespace(collection.data(), collection.size());
235-
session.set_timeout(m_timeouts.read);
195+
session.set_timeout(timeout);
196+
auto trace = trace_t::current();
197+
if(!trace.empty()) {
198+
session.set_trace_id(trace.get_trace_id());
199+
}
200+
if(trace.verbose()) {
201+
session.set_trace_bit(true);
202+
}
203+
return session;
204+
}
236205

206+
ell::async_read_result elliptics_storage_t::async_read(const std::string &collection, const std::string &key)
207+
{
208+
COCAINE_LOG_DEBUG(m_log, "reading the '{}' object, collection: '{}'", key, collection);
209+
ell::session session = prepare_session(collection, m_timeouts.read);
237210
return session.read_data(key, 0, 0);
238211
}
239212

240213
ell::async_read_result elliptics_storage_t::async_read_latest(const std::string &collection, const std::string &key)
241214
{
242-
using namespace std::placeholders;
243-
244215
COCAINE_LOG_DEBUG(m_log, "reading the '{}' object, collection: '{}'", key, collection);
245-
246-
ell::session session = m_session.clone();
247-
session.set_namespace(collection.data(), collection.size());
248-
session.set_timeout(m_timeouts.read);
249-
216+
ell::session session = prepare_session(collection, m_timeouts.read);
250217
return session.read_latest(key, 0, 0);
251218
}
252219

@@ -258,38 +225,27 @@ ell::async_write_result elliptics_storage_t::async_write(const std::string &coll
258225

259226
COCAINE_LOG_DEBUG(m_log, "writing the '{}' object, collection: '{}'", key, collection);
260227

261-
ell::session session = m_session.clone();
262-
session.set_namespace(collection.data(), collection.size());
228+
ell::session session = prepare_session(collection, m_timeouts.write);
263229
session.set_filter(ioremap::elliptics::filters::all_with_ack);
264-
session.set_timeout(m_timeouts.write);
265230
session.set_checker(m_success_copies_num);
266-
267231
return session.write_data(key, blob, 0);
268232
}
269233

270234
ell::async_remove_result elliptics_storage_t::async_remove(const std::string &collection, const std::string &key)
271235
{
272-
using namespace std::placeholders;
273-
274236
COCAINE_LOG_DEBUG(m_log, "removing the '{}' object, collection: '{}'", key, collection);
275237

276-
ell::session session = m_session.clone();
277-
session.set_namespace(collection.data(), collection.size());
278-
session.set_timeout(m_timeouts.remove);
238+
ell::session session = prepare_session(collection, m_timeouts.remove);
279239
session.set_checker(m_success_copies_num);
280-
281240
return session.remove(key);
282241
}
283242

284243
ioremap::elliptics::async_read_result elliptics_storage_t::async_cache_read(const std::string &collection, const std::string &key)
285244
{
286245
COCAINE_LOG_DEBUG(m_log, "cache reading the '{}' object, collection: '{}'", key, collection);
287246

288-
ell::session session = m_session.clone();
289-
session.set_namespace(collection.data(), collection.size());
247+
ell::session session = prepare_session(collection, m_timeouts.read);
290248
session.set_ioflags(DNET_IO_FLAGS_CACHE | DNET_IO_FLAGS_CACHE_ONLY);
291-
session.set_timeout(m_timeouts.read);
292-
293249
return session.read_data(key, 0, 0);
294250
}
295251

@@ -298,12 +254,9 @@ ioremap::elliptics::async_write_result elliptics_storage_t::async_cache_write(co
298254
{
299255
COCAINE_LOG_DEBUG(m_log, "cache writing the '{}' object, collection: '{}'", key, collection);
300256

301-
ell::session session = m_session.clone();
302-
session.set_namespace(collection.data(), collection.size());
257+
ell::session session = prepare_session(collection, m_timeouts.write);
303258
session.set_ioflags(DNET_IO_FLAGS_CACHE | DNET_IO_FLAGS_CACHE_ONLY);
304-
session.set_timeout(m_timeouts.write);
305259
session.set_checker(m_success_copies_num);
306-
307260
return session.write_cache(key, blob, timeout);
308261
}
309262

@@ -312,9 +265,7 @@ std::pair<ioremap::elliptics::async_read_result, elliptics_storage_t::key_name_m
312265
{
313266
COCAINE_LOG_DEBUG(m_log, "bulk reading, collection: '{}'", collection);
314267

315-
ell::session session = m_session.clone();
316-
session.set_namespace(collection.data(), collection.size());
317-
session.set_timeout(m_timeouts.read);
268+
ell::session session = prepare_session(collection, m_timeouts.read);
318269

319270
key_name_map keys_map;
320271
dnet_raw_id id;
@@ -332,10 +283,8 @@ ioremap::elliptics::async_write_result elliptics_storage_t::async_bulk_write(con
332283
{
333284
COCAINE_LOG_DEBUG(m_log, "bulk writing, collection: '{}'", collection);
334285

335-
ell::session session = m_session.clone();
336-
session.set_namespace(collection.data(), collection.size());
286+
ell::session session = prepare_session(collection, m_timeouts.write);
337287
session.set_filter(ell::filters::all);
338-
session.set_timeout(m_timeouts.write);
339288
session.set_checker(m_success_copies_num);
340289

341290
std::vector<dnet_io_attr> ios;

0 commit comments

Comments
 (0)
Please sign in to comment.