From 51aeaeaeabfb428101a817da66d4f839172f668b Mon Sep 17 00:00:00 2001 From: Seva Date: Fri, 21 Jun 2019 12:49:47 -0400 Subject: [PATCH] dhtproxy: use logger everywhere & use same format --- include/opendht/http.h | 8 ++--- src/dht_proxy_client.cpp | 46 ++++++++++++++++++++------- src/dht_proxy_server.cpp | 69 +++++++++++++++++++++++++++------------- src/http.cpp | 27 ++++++++-------- tests/dhtproxytester.cpp | 6 ++-- 5 files changed, 102 insertions(+), 54 deletions(-) diff --git a/include/opendht/http.h b/include/opendht/http.h index b982f16f..419a24fb 100644 --- a/include/opendht/http.h +++ b/include/opendht/http.h @@ -165,25 +165,25 @@ public: template void trace(Builder && msg_builder){ if (m_logger) - m_logger->d("[restinio] %s", msg_builder().c_str()); + m_logger->d("[proxy:server] %s", msg_builder().c_str()); } template void info(Builder && msg_builder){ if (m_logger) - m_logger->d("[restinio] %s", msg_builder().c_str()); + m_logger->d("[proxy:server] %s", msg_builder().c_str()); } template void warn(Builder && msg_builder){ if (m_logger) - m_logger->w("[restinio] %s", msg_builder().c_str()); + m_logger->w("[proxy:server] %s", msg_builder().c_str()); } template void error(Builder && msg_builder){ if (m_logger) - m_logger->e("[restinio] %s", msg_builder().c_str()); + m_logger->e("[proxy:server] %s", msg_builder().c_str()); } private: diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 5dacdf03..4b51e9c9 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -291,6 +291,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Value::Filter filter; std::atomic_bool ok {true}; std::atomic_bool stop {false}; + std::shared_ptr logger; }; auto context = std::make_shared(); context->filter = w.empty() ? f : f.chain(w.getFilter()); @@ -315,6 +316,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, context->stop = true; }); }; + if (logger_) + context->logger = logger_; auto parser = std::make_shared(); http_parser_init(parser.get(), HTTP_RESPONSE); @@ -325,7 +328,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, parser_s->on_status = [](http_parser *parser, const char *at, size_t length) -> int { auto context = static_cast(parser->data); if (parser->status_code != 200){ - std::cerr << "Error in get status_code=" << parser->status_code << std::endl; + if (context->logger) + context->logger->e("[proxy:client] get status error: %i", parser->status_code); context->ok = true; } return 0; @@ -348,7 +352,8 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, context->cb({value}); } } catch(const std::exception& e) { - std::cerr << "Error in get parsing: " << e.what() << std::endl; + if (context->logger) + context->logger->e("[proxy:client] get body parsing error: %s", e.what()); context->ok = false; return 1; } @@ -360,7 +365,9 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, if (context->donecb) context->donecb(context->ok); } catch(const std::exception& e) { - std::cerr << "Error in get parsing: " << e.what() << std::endl; + if (context->logger) + context->logger->e("[proxy:client] get message complete parsing error: %i", + parser->status_code); return 1; } return 0; @@ -456,6 +463,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ struct GetContext { DoneCallbackSimple donecb; // wrapper std::atomic_bool ok {false}; + std::shared_ptr logger; }; auto context = std::make_shared(); // keeping context data alive @@ -468,6 +476,8 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ } loopSignal_(); }; + if (logger_) + context->logger = logger_; auto parser = std::make_shared(); http_parser_init(parser.get(), HTTP_RESPONSE); @@ -480,7 +490,8 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ if (parser->status_code == 200){ context->ok = true; } else { - std::cerr << "Error in get status_code=" << parser->status_code << std::endl; + if (context->logger) + context->logger->e("[proxy:client] put status error: %i", parser->status_code); } return 0; }; @@ -490,7 +501,8 @@ DhtProxyClient::doPut(const InfoHash& key, Sp val, DoneCallback cb, time_ if (context->donecb) context->donecb(context->ok); } catch(const std::exception& e) { - std::cerr << "Error in get parsing: " << e.what() << std::endl; + if (context->logger) + context->logger->e("[proxy:client] put message complete error: %s", e.what()); return 1; } return 0; @@ -621,12 +633,15 @@ DhtProxyClient::handleProxyStatus(const asio::error_code &ec, std::shared_ptr infoState; SockAddr resolvedProxy; std::function proxyInfo; + std::shared_ptr logger; }; auto context = std::make_shared(); context->infoState = infoState; context->resolvedProxy = resolvedProxy; context->proxyInfo = std::bind(&DhtProxyClient::onProxyInfos, this, std::placeholders::_1, std::placeholders::_2); + if (logger_) + context->logger = logger_; // initialize the parser auto parser = std::make_shared(); @@ -639,7 +654,8 @@ DhtProxyClient::handleProxyStatus(const asio::error_code &ec, parser_s->on_status = [](http_parser *parser, const char *at, size_t length) -> int { auto context = static_cast(parser->data); if (parser->status_code != 200){ - std::cerr << "Error in get status_code=" << parser->status_code << std::endl; + if (context->logger) + context->logger->e("[proxy:client] proxy status error: %i", parser->status_code); context->ok = true; } return 0; @@ -666,7 +682,8 @@ DhtProxyClient::handleProxyStatus(const asio::error_code &ec, context->proxyInfo(proxyInfos, family); } catch (const std::exception& e) { - std::cerr << "Error in get parsing: " << e.what() << std::endl; + if (context->logger) + context->logger->e("[proxy:client] proxy status body error: %s", e.what()); context->ok = false; return 1; } @@ -916,14 +933,14 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header, logger_->d(request.c_str()); struct ListenContext { - Logger *logger; + std::shared_ptr logger; ValueCallback cb; // wrapper Value::Filter filter; std::shared_ptr state; }; auto context = std::make_shared(); if (logger_) - context->logger = logger_.get(); + context->logger = logger_; // keeping context data alive context->cb = [context, cb](const std::vector>& values, bool expired){ return cb(values, expired); @@ -940,7 +957,8 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header, parser_s->on_status = [](http_parser *parser, const char *at, size_t length) -> int { auto context = static_cast(parser->data); if (parser->status_code != 200){ - std::cerr << "Error in listen status_code=" << parser->status_code << std::endl; + if (context->logger) + context->logger->e("[proxy:client] listen status error: %i", parser->status_code); context->state->ok = false; } return 0; @@ -1019,9 +1037,12 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken) // define context struct UnsubscribeContext { InfoHash key; + std::shared_ptr logger; }; auto context = std::make_shared(); context->key = key; + if (logger_) + context->logger = logger_; // define parser auto parser = std::make_shared(); http_parser_init(parser.get(), HTTP_RESPONSE); @@ -1032,8 +1053,9 @@ DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken) parser_s->on_status = [](http_parser *parser, const char *at, size_t length) -> int { auto context = static_cast(parser->data); if (parser->status_code != 200){ - std::cerr << "[search " << context->key.to_c_str() << "] " << - "cancel listen failed status_code=" << parser->status_code << std::endl; + if (context->logger) + context->logger->e("[proxy:client] [search %s] cancel listen failed: %i", + context->key.to_c_str(), parser->status_code); } return 0; }; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index a9400df6..e87d31f9 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -59,12 +59,15 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, if (not dht_) throw std::invalid_argument("A DHT instance must be provided"); - std::cout << "Running DHT proxy server on port " << port << std::endl; + if (logger_) + logger_->d("[proxy:server] [init] running on %i", port); if (not pushServer.empty()){ #ifdef OPENDHT_PUSH_NOTIFICATIONS - std::cout << "Using push notification server: " << pushServer << std::endl; + if (logger_) + logger_->d("[proxy:server] [init] using push server %s", pushServer.c_str()); #else - std::cerr << "Push server defined but built OpenDHT built without push notification support" << std::endl; + if (logger_) + logger_->e("[proxy:server] [init] opendht built without push notification support"); #endif } @@ -92,7 +95,8 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, httpServer_->io_context().run(); } catch(const std::exception &ex){ - std::cerr << "Error starting RESTinio: " << ex.what() << std::endl; + if (logger_) + logger_->e("[proxy:server] error starting: %s", ex.what()); } }); dht->forwardAllMessages(true); @@ -143,13 +147,13 @@ void DhtProxyServer::stop() { if (logger_) - logger_->d("[restinio] closing http server async operations"); + logger_->d("[proxy:server] closing http server async operations"); httpServer_->io_context().reset(); httpServer_->io_context().stop(); if (httpServerThread_.joinable()) httpServerThread_.join(); if (logger_) - logger_->d("[restinio] http server closed"); + logger_->d("[proxy:server] http server closed"); } void @@ -183,7 +187,7 @@ DhtProxyServer::asyncPrintStats() auto json = nodeInfo_.toJson(); auto str = Json::writeString(jsonBuilder_, json); if (logger_) - logger_->d("[stats] %s", str.c_str()); + logger_->d("[proxy:server] [stats] %s", str.c_str()); } printStatsTimer_->expires_at(printStatsTimer_->expiry() + PRINT_STATS_PERIOD); printStatsTimer_->async_wait(std::bind(&DhtProxyServer::asyncPrintStats, this)); @@ -499,10 +503,10 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, listener.expireNotifyTimer->async_wait( [this, infoHash, pushToken, isAndroid, clientId](const asio::error_code &ec){ if (logger_) - logger_->d("[subscribe] sending refresh %s", infoHash.toString().c_str()); + logger_->d("[proxy:server] [subscribe] sending refresh %s", infoHash.toString().c_str()); if (ec){ if (logger_) - logger_->d("[subscribe] error sending refresh: %s", ec.message().c_str()); + logger_->d("[proxy:server] [subscribe] error sending refresh: %s", ec.message().c_str()); } Json::Value json; json["timeout"] = infoHash.toString(); @@ -638,12 +642,22 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso auto parser = std::make_shared(); http_parser_init(parser.get(), HTTP_RESPONSE); + struct PushContext { + std::shared_ptr logger; + }; + auto context = std::make_shared(); + if (logger_) + context->logger = logger_; + parser->data = static_cast(context.get()); + auto parser_s = std::make_shared(); http_parser_settings_init(parser_s.get()); parser_s->on_status = []( http_parser * parser, const char * at, size_t length ) -> int { + auto context = static_cast(parser->data); if (parser->status_code == 200) return 0; - std::cerr << "Error in SendPushNotification status_code=" << parser->status_code << std::endl; + if (context->logger) + context->logger->e("[proxy:server] [notification] error send push: %i", parser->status_code); return 1; }; auto request = httpClient_->create_request(header, header_fields, @@ -656,7 +670,8 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso void DhtProxyServer::cancelPut(const InfoHash& key, Value::Id vid) { - std::cout << "cancelPut " << key << " " << vid << std::endl; + if (logger_) + logger_->d("[proxy:server] [put] cancel put %s %i", key.toString(), vid); auto sPuts = puts_.find(key); if (sPuts == puts_.end()) return; @@ -705,8 +720,9 @@ DhtProxyServer::put(restinio::request_handle_t request, if (reader->parse(char_data, char_data + request->body().size(), &root, &err)){ auto value = std::make_shared(root); bool permanent = root.isMember("permanent"); - std::cout << "Got put " << infoHash << " " << *value << - " " << (permanent ? "permanent" : "") << std::endl; + if (logger_) + logger_->d("[proxy:server] [put] %s %s %s", infoHash.toString(), + value->toString(), (permanent ? "permanent" : "")); if (permanent){ std::string pushToken, clientId, platform; auto& pVal = root["permanent"]; @@ -725,9 +741,12 @@ DhtProxyServer::put(restinio::request_handle_t request, auto &ctx = httpServer_->io_context(); pput.expireTimer = std::make_unique(ctx, timeout); pput.expireTimer->async_wait([this, infoHash, vid](const asio::error_code &ec){ - std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl; - if (ec) - std::cout << "Permanent put error: " << ec.message() << std::endl; + if (logger_) + logger_->d("[proxy:server] [put] permanent expired: %s %i", infoHash.toString(), vid); + if (ec){ + if (logger_) + logger_->e("[proxy:server] [put] error in permanent: %s", ec.message().c_str()); + } cancelPut(infoHash, vid); }); #ifdef OPENDHT_PUSH_NOTIFICATIONS @@ -738,9 +757,12 @@ DhtProxyServer::put(restinio::request_handle_t request, pput.expireNotifyTimer->async_wait( [this, infoHash, vid, pushToken, clientId, isAndroid](const asio::error_code &ec) { - std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl; - if (ec) - std::cout << "Permanent put refresh error: " << ec.message() << std::endl; + if (logger_) + logger_->d("[proxy:server] [put] refresh: %s %i", infoHash.toString(), vid); + if (ec){ + if (logger_) + logger_->e("[proxy:server] [put] error in refresh: %s", ec.message().c_str()); + } Json::Value json; json["timeout"] = infoHash.toString(); json["to"] = clientId; @@ -776,7 +798,8 @@ DhtProxyServer::put(restinio::request_handle_t request, return response.done(); } } catch (const std::exception& e){ - std::cout << "Error performing put: " << e.what() << std::endl; + if (logger_) + logger_->d("[proxy:server] [put] error: %s", e.what()); auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -839,7 +862,8 @@ RequestStatus DhtProxyServer::putSigned(restinio::request_handle_t request, return response.done(); } } catch (const std::exception& e){ - std::cout << "Error performing put: " << e.what() << std::endl; + if (logger_) + logger_->d("[proxy:server] [put] error: %s", e.what()); auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -906,7 +930,8 @@ DhtProxyServer::putEncrypted(restinio::request_handle_t request, return response.done(); } } catch (const std::exception& e){ - std::cout << "Error performing put: " << e.what() << std::endl; + if (logger_) + logger_->d("[proxy:server] [put] error: %s", e.what()); auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(RESP_MSG_INTERNAL_SERVER_ERRROR); diff --git a/src/http.cpp b/src/http.cpp index 58980719..3c31b810 100644 --- a/src/http.cpp +++ b/src/http.cpp @@ -69,12 +69,12 @@ ConnectionListener::state_changed(const restinio::connection_state::notice_t &no if (listeners_->find(id) != listeners_->end()){ if (notice.cause() == restinio::connection_state::cause_t::closed){ if (logger_) - logger_->d("[restinio] [connection:%li] cancelling listener", id); + logger_->d("[proxy:server] [connection:%li] cancelling listener", id); dht_->cancelListen(listeners_->at(id).hash, std::move(listeners_->at(id).token)); listeners_->erase(id); if (logger_) - logger_->d("[restinio] %li listeners are connected", listeners_->size()); + logger_->d("[proxy:server] %li listeners are connected", listeners_->size()); } } } @@ -133,7 +133,7 @@ Client::create_connection() auto conn = std::make_shared(connId_, std::move(asio::ip::tcp::socket{resolver_.get_io_context()})); if (logger_) - logger_->d("[connection:%i] created", conn->id()); + logger_->d("[proxy:client] [connection:%i] created", conn->id()); connId_++; return conn; } @@ -156,7 +156,7 @@ Client::close_connection(const uint16_t conn_id) // remove from active requests requests_.erase(conn_id); if (logger_) - logger_->d("[connection:%i] closed", conn_id); + logger_->d("[proxy:client] [connection:%i] closed", conn_id); } std::string @@ -236,12 +236,12 @@ Client::handle_resolve(const asio::error_code &ec, { if (ec){ if (logger_) - logger_->e("[connection:%i] error resolving", conn->id()); + logger_->e("[proxy:client] [connection:%i] error resolving", conn->id()); conn->close(); return; } if (logger_){ - logger_->d("[connection:%i] resolved host=%s service=%s", conn->id(), + logger_->d("[proxy:client] [connection:%i] resolved host=%s service=%s", conn->id(), endpoint_it->host_name().c_str(), endpoint_it->service_name().c_str()); } asio::ip::tcp::endpoint endpoint = *endpoint_it; @@ -257,14 +257,14 @@ Client::handle_connect(const asio::error_code &ec, { if (ec){ if (logger_) - logger_->e("[connection:%i] error opening: %s", + logger_->e("[proxy:client] [connection:%i] error opening: %s", conn->id(), ec.message().c_str()); close(conn->id()); return; } else if (endpoint_it != asio::ip::tcp::resolver::iterator()){ if (logger_) - logger_->e("[connection:%i] error connecting, trying next endpoint", conn->id()); + logger_->e("[proxy:client] [connection:%i] error connecting, trying next endpoint", conn->id()); conn->socket_.close(); // connect to next one asio::ip::tcp::endpoint endpoint = *endpoint_it; @@ -283,13 +283,13 @@ Client::handle_request(const asio::error_code &ec, std::shared_ptr c { if (ec and ec != asio::error::eof){ if (logger_) - logger_->e("[connection:%i] error handling request: %s", + logger_->e("[proxy:client] [connection:%i] error handling request: %s", conn->id(), ec.message().c_str()); close_connection(conn->id()); return; } if (logger_) - logger_->d("[connection:%i] request write", conn->id()); + logger_->d("[proxy:client] [connection:%i] request write", conn->id()); // read response asio::async_read_until(conn->socket_, conn->response_, "\r\n\r\n", @@ -301,7 +301,7 @@ Client::handle_response(const asio::error_code &ec, std::shared_ptr { if (ec && ec != asio::error::eof){ if (logger_) - logger_->e("[connection:%i] error handling response: %s", + logger_->e("[proxy:client] [connection:%i] error handling response: %s", conn->id(), ec.message().c_str()); return; } @@ -310,7 +310,7 @@ Client::handle_response(const asio::error_code &ec, std::shared_ptr return; } if (logger_) - logger_->d("[connection:%i] response read", conn->id()); + logger_->d("[proxy:client] [connection:%i] response read", conn->id()); // read the response buffer std::ostringstream str_s; @@ -325,7 +325,8 @@ Client::handle_response(const asio::error_code &ec, std::shared_ptr if (HPE_OK != req.parser->http_errno && HPE_PAUSED != req.parser->http_errno){ if (logger_){ auto err = HTTP_PARSER_ERRNO(req.parser.get()); - logger_->e("[connection:%i] error parsing: %s", conn->id(), http_errno_name(err)); + logger_->e("[proxy:client] [connection:%i] error parsing: %s", + conn->id(), http_errno_name(err)); } } asio::async_read(conn->socket_, conn->response_, asio::transfer_at_least(1), diff --git a/tests/dhtproxytester.cpp b/tests/dhtproxytester.cpp index f050f946..c0914178 100644 --- a/tests/dhtproxytester.cpp +++ b/tests/dhtproxytester.cpp @@ -52,13 +52,13 @@ DhtProxyTester::setUp() { void DhtProxyTester::tearDown() { - logger->d("[dhtproxytester] stopping peer node"); + logger->d("[tester:proxy] stopping peer node"); nodePeer.join(); nodeClient->join(); - logger->d("[dhtproxytester] stopping proxy server"); + logger->d("[tester:proxy] stopping proxy server"); serverProxy->stop(); serverProxy = nullptr; - logger->d("[dhtproxytester] stopping proxy node"); + logger->d("[tester:proxy] stopping proxy node"); nodeProxy->join(); }