From b2d8bd06318e1fddf4f1579084bbda4fb556c176 Mon Sep 17 00:00:00 2001 From: Jonathan Reams Date: Tue, 20 Feb 2018 14:33:42 -0500 Subject: SERVER-33300 Integrate TransportLayer with DBClient --- src/mongo/SConscript | 2 + src/mongo/client/SConscript | 2 + src/mongo/client/connection_pool.cpp | 4 +- src/mongo/client/dbclient.cpp | 254 ++++++++++++--------- src/mongo/client/dbclientinterface.h | 31 +-- src/mongo/db/client.h | 1 - src/mongo/db/repl/oplogreader.cpp | 3 +- src/mongo/dbtests/repltests.cpp | 11 + src/mongo/s/SConscript | 1 + src/mongo/shell/dbshell.cpp | 13 ++ src/mongo/shell/shell_options.cpp | 1 + src/mongo/shell/shell_options.h | 1 + src/mongo/tools/bridge.cpp | 76 +++--- src/mongo/transport/SConscript | 13 ++ src/mongo/transport/asio_utils.h | 93 ++++++++ src/mongo/transport/mock_session.h | 4 + src/mongo/transport/session.h | 10 + src/mongo/transport/session_asio.h | 170 ++++++++++---- src/mongo/transport/transport_layer.h | 11 + src/mongo/transport/transport_layer_asio.cpp | 224 +++++++++++++++--- src/mongo/transport/transport_layer_asio.h | 30 ++- .../transport/transport_layer_egress_init.cpp | 58 +++++ src/mongo/transport/transport_layer_manager.cpp | 13 ++ src/mongo/transport/transport_layer_manager.h | 8 + src/mongo/transport/transport_layer_mock.cpp | 13 ++ src/mongo/transport/transport_layer_mock.h | 8 + src/mongo/unittest/SConscript | 3 + src/mongo/unittest/integration_test_main.cpp | 2 + 28 files changed, 818 insertions(+), 242 deletions(-) create mode 100644 src/mongo/transport/transport_layer_egress_init.cpp (limited to 'src/mongo') diff --git a/src/mongo/SConscript b/src/mongo/SConscript index b1030f3a380..55ff0660453 100644 --- a/src/mongo/SConscript +++ b/src/mongo/SConscript @@ -466,6 +466,7 @@ if not has_option('noshell') and usemozjs: 'scripting/scripting', 'shell/mongojs', 'transport/message_compressor', + 'transport/transport_layer_manager', 'util/net/network', 'util/options_parser/options_parser_init', 'util/processinfo', @@ -508,6 +509,7 @@ if not has_option('noshell') and usemozjs: "$BUILD_DIR/third_party/shim_pcrecpp", "shell_core", "db/server_options_core", + "db/service_context_noop_init", "client/clientdriver", "$BUILD_DIR/mongo/util/password", ], diff --git a/src/mongo/client/SConscript b/src/mongo/client/SConscript index 3a48533419e..92785bdae90 100644 --- a/src/mongo/client/SConscript +++ b/src/mongo/client/SConscript @@ -47,6 +47,7 @@ env.CppUnitTest( ], LIBDEPS=[ 'clientdriver', + '$BUILD_DIR/mongo/transport/transport_layer_egress_init', ] ) @@ -282,6 +283,7 @@ env.CppUnitTest( '$BUILD_DIR/mongo/db/auth/authorization_manager_mock_init', '$BUILD_DIR/mongo/db/service_context_noop_init', '$BUILD_DIR/mongo/transport/transport_layer', + '$BUILD_DIR/mongo/transport/transport_layer_egress_init', '$BUILD_DIR/mongo/util/net/network', '$BUILD_DIR/mongo/util/version_impl', ], diff --git a/src/mongo/client/connection_pool.cpp b/src/mongo/client/connection_pool.cpp index 965ce7d790c..dbccb711d8d 100644 --- a/src/mongo/client/connection_pool.cpp +++ b/src/mongo/client/connection_pool.cpp @@ -105,7 +105,7 @@ void ConnectionPool::closeAllInUseConnections() { stdx::lock_guard lk(_mutex); for (ConnectionList::iterator iter = _inUseConnections.begin(); iter != _inUseConnections.end(); ++iter) { - iter->conn->port().shutdown(); + iter->conn->shutdown(); } } @@ -189,7 +189,7 @@ ConnectionPool::ConnectionList::iterator ConnectionPool::acquireConnection( conn->setSoTimeout(durationCount(timeout) / 1000.0); uassertStatusOK(conn->connect(target, StringData())); - conn->port().setTag(conn->port().getTag() | _messagingPortTags); + conn->setTags(_messagingPortTags); if (isInternalAuthSet()) { conn->auth(getInternalUserAuthParams()); diff --git a/src/mongo/client/dbclient.cpp b/src/mongo/client/dbclient.cpp index 2d05e518b7b..d7f6c164552 100644 --- a/src/mongo/client/dbclient.cpp +++ b/src/mongo/client/dbclient.cpp @@ -70,12 +70,12 @@ #include "mongo/util/debug_util.h" #include "mongo/util/fail_point_service.h" #include "mongo/util/log.h" -#include "mongo/util/net/message_port.h" +#include "mongo/util/net/sock.h" #include "mongo/util/net/socket_exception.h" #include "mongo/util/net/ssl_manager.h" #include "mongo/util/net/ssl_options.h" #include "mongo/util/password_digest.h" -#include "mongo/util/represent_as.h" +#include "mongo/util/scopeguard.h" #include "mongo/util/time_support.h" #include "mongo/util/version.h" @@ -851,7 +851,7 @@ Status DBClientConnection::connect(const HostAndPort& serverAddress, StringData auto swIsMasterReply = initWireVersion(this, _applicationName); if (!swIsMasterReply.isOK()) { - _failed = true; + _markFailed(kSetFlag); return swIsMasterReply.status; } @@ -922,8 +922,7 @@ Status DBClientConnection::connect(const HostAndPort& serverAddress, StringData auto validationStatus = _hook(swIsMasterReply); if (!validationStatus.isOK()) { // Disconnect and mark failed. - _failed = true; - _port.reset(); + _markFailed(kReleaseSession); return validationStatus; } } @@ -931,27 +930,9 @@ Status DBClientConnection::connect(const HostAndPort& serverAddress, StringData return Status::OK(); } -namespace { -const auto kMaxMillisCount = Milliseconds::max().count(); -} // namespace - Status DBClientConnection::connectSocketOnly(const HostAndPort& serverAddress) { _serverAddress = serverAddress; - _failed = true; - - // We need to construct a SockAddr so we can resolve the address. - SockAddr osAddr{serverAddress.host().c_str(), - serverAddress.port(), - static_cast(IPv6Enabled() ? AF_UNSPEC : AF_INET)}; - - if (!osAddr.isValid()) { - return Status(ErrorCodes::InvalidOptions, - str::stream() << "couldn't initialize connection to host " - << serverAddress.host() - << ", address is invalid"); - } - - _port.reset(new MessagingPort(_so_timeout, _logLevel)); + _markFailed(kReleaseSession); if (serverAddress.host().empty()) { return Status(ErrorCodes::InvalidOptions, @@ -959,47 +940,45 @@ Status DBClientConnection::connectSocketOnly(const HostAndPort& serverAddress) { << ", host is empty"); } - if (osAddr.getAddr() == "0.0.0.0") { + if (serverAddress.host() == "0.0.0.0") { return Status(ErrorCodes::InvalidOptions, str::stream() << "couldn't connect to server " << _serverAddress.toString() << ", address resolved to 0.0.0.0"); } - _resolvedAddress = osAddr.getAddr(); - - if (!_port->connect(osAddr)) { - return Status(ErrorCodes::HostUnreachable, - str::stream() << "couldn't connect to server " << _serverAddress.toString() - << ", connection attempt failed"); - } - + transport::ConnectSSLMode sslMode = transport::kGlobalSSLMode; #ifdef MONGO_CONFIG_SSL // Prefer to get SSL mode directly from our URI, but if it is not set, fall back to // checking global SSL params. DBClientConnections create through the shell will have a // meaningful URI set, but DBClientConnections created from within the server may not. - int sslMode; auto options = _uri.getOptions(); auto iter = options.find("ssl"); if (iter != options.end()) { if (iter->second == "true") { - sslMode = SSLParams::SSLMode_requireSSL; + sslMode = transport::kEnableSSL; } else { - sslMode = SSLParams::SSLMode_disabled; + sslMode = transport::kDisableSSL; } - } else { - sslMode = sslGlobalParams.sslMode.load(); } - if (sslMode == SSLParams::SSLMode_preferSSL || sslMode == SSLParams::SSLMode_requireSSL) { - uassert(40312, "SSL is not enabled; cannot create an SSL connection", sslManager()); - if (!_port->secure(sslManager(), serverAddress.host())) { - return Status(ErrorCodes::SSLHandshakeFailed, "Failed to initialize SSL on connection"); - } - } #endif + auto tl = getGlobalServiceContext()->getTransportLayer(); + auto sws = tl->connect(serverAddress, sslMode, _socketTimeout.value_or(Milliseconds{5000})); + if (!sws.isOK()) { + return Status(ErrorCodes::HostUnreachable, + str::stream() << "couldn't connect to server " << _serverAddress.toString() + << ", connection attempt failed: " + << sws.getStatus().toString()); + } + + _session = std::move(sws.getValue()); + _sessionCreationMicros = curTimeMicros64(); + _lastConnectivityCheck = Date_t::now(); + _session->setTimeout(_socketTimeout); + _session->setTags(_tagMask); _failed = false; - LOG(1) << "connected to server " << toString() << endl; + LOG(1) << "connected to server " << toString(); return Status::OK(); } @@ -1040,13 +1019,62 @@ rpc::UniqueReply DBClientConnection::parseCommandReplyMessage(const std::string& return DBClientBase::parseCommandReplyMessage(host, std::move(replyMsg)); } catch (const DBException& ex) { if (ErrorCodes::isConnectionFatalMessageParseError(ex.code())) { - _port->shutdown(); - _failed = true; + _markFailed(kEndSession); } throw; } } +void DBClientConnection::_markFailed(FailAction action) { + _failed = true; + if (_session) { + if (action == kEndSession) { + _session->end(); + } else if (action == kReleaseSession) { + _session.reset(); + } + } +} + +bool DBClientConnection::isStillConnected() { + // This method tries to figure out whether the connection is still open, but with several + // caveats. + + // If we don't have a _session then we may have hit an error, or we may just not have + // connected yet - the _failed flag should indicate which. + // + // Otherwise, return false if we know we've had an error (_failed is true) + if (!_session) { + return !_failed; + } else if (_failed) { + return false; + } + + // Checking whether the socket actually has an error by calling _session->isConnected() + // is actually pretty expensive, so we cache the result for 5 seconds + auto now = getGlobalServiceContext()->getFastClockSource()->now(); + if (now - _lastConnectivityCheck < Seconds{5}) { + return true; + } + + _lastConnectivityCheck = now; + + // This will poll() the underlying socket and do a 1 byte recv to see if the connection + // has been closed. + return _session->isConnected(); +} + +void DBClientConnection::setTags(transport::Session::TagMask tags) { + _tagMask = tags; + if (!_session) + return; + _session->setTags(tags); +} + +void DBClientConnection::shutdown() { + _markFailed(kEndSession); +} + void DBClientConnection::_checkConnection() { if (!_failed) return; @@ -1062,7 +1090,7 @@ void DBClientConnection::_checkConnection() { _failed = false; auto connectStatus = connect(_serverAddress, _applicationName); if (!connectStatus.isOK()) { - _failed = true; + _markFailed(kSetFlag); LOG(_logLevel) << "reconnect " << toString() << " failed " << errmsg << endl; if (connectStatus == ErrorCodes::IncompatibleCatalogManager) { uassertStatusOK(connectStatus); // Will always throw @@ -1087,24 +1115,29 @@ void DBClientConnection::_checkConnection() { } void DBClientConnection::setSoTimeout(double timeout) { - _so_timeout = timeout; - if (_port) { - // `timeout` is in seconds. - auto ms = representAs(std::floor(timeout * 1000)).value_or(kMaxMillisCount); - _port->setTimeout(ms > kMaxMillisCount ? Milliseconds::max() : Milliseconds(ms)); + Milliseconds::rep timeoutMs = std::floor(timeout * 1000); + if (timeout <= 0) { + _socketTimeout = boost::none; + } else if (timeoutMs >= Milliseconds::max().count()) { + _socketTimeout = Milliseconds::max(); + } else { + _socketTimeout = Milliseconds{timeoutMs}; + } + + if (_session) { + _session->setTimeout(_socketTimeout); } } uint64_t DBClientConnection::getSockCreationMicroSec() const { - if (_port) { - return _port->getSockCreationMicroSec(); + if (_session) { + return _sessionCreationMicros; } else { return INVALID_SOCK_CREATION_TIME; } } -const uint64_t DBClientBase::INVALID_SOCK_CREATION_TIME = - static_cast(0xFFFFFFFFFFFFFFFFULL); +const uint64_t DBClientBase::INVALID_SOCK_CREATION_TIME = std::numeric_limits::max(); unique_ptr DBClientBase::query(const string& ns, Query query, @@ -1206,8 +1239,7 @@ unsigned long long DBClientConnection::query(stdx::functionshutdown(); + _markFailed(kEndSession); throw; } @@ -1378,7 +1410,6 @@ DBClientConnection::DBClientConnection(bool _autoReconnect, : _failed(false), autoReconnect(_autoReconnect), autoReconnectBackoff(1000, 2000), - _so_timeout(so_timeout), _hook(hook), _uri(std::move(uri)) { _numConnections.fetchAndAdd(1); @@ -1386,74 +1417,73 @@ DBClientConnection::DBClientConnection(bool _autoReconnect, void DBClientConnection::say(Message& toSend, bool isRetry, string* actualServer) { checkConnection(); - try { - toSend.header().setId(nextMessageId()); - toSend.header().setResponseToMsgId(0); - auto swm = _compressorManager.compressMessage(toSend); - uassertStatusOK(swm.getStatus()); - port().say(swm.getValue()); - } catch (const DBException&) { - _failed = true; - _port->shutdown(); - throw; - } + auto killSessionOnError = MakeGuard([this] { _markFailed(kEndSession); }); + + toSend.header().setId(nextMessageId()); + toSend.header().setResponseToMsgId(0); + uassertStatusOK( + _session->sinkMessage(uassertStatusOK(_compressorManager.compressMessage(toSend)))); + killSessionOnError.Dismiss(); } bool DBClientConnection::recv(Message& m, int lastRequestId) { - if (!port().recv(m)) { - _failed = true; + auto killSessionOnError = MakeGuard([this] { _markFailed(kEndSession); }); + auto swm = _session->sourceMessage(); + if (!swm.isOK()) { return false; } - try { - uassert(40570, - "Response ID did not match the sent message ID.", - m.header().getResponseToMsgId() == lastRequestId); - - if (m.operation() == dbCompressed) { - m = uassertStatusOK(_compressorManager.decompressMessage(m)); - } + m = std::move(swm.getValue()); + uassert(40570, + "Response ID did not match the sent message ID.", + m.header().getResponseToMsgId() == lastRequestId); - return true; - } catch (const DBException&) { - _failed = true; - _port->shutdown(); - throw; + if (m.operation() == dbCompressed) { + m = uassertStatusOK(_compressorManager.decompressMessage(m)); } + + killSessionOnError.Dismiss(); + return true; } bool DBClientConnection::call(Message& toSend, Message& response, bool assertOk, string* actualServer) { - /* todo: this is very ugly messagingport::call returns an error code AND can throw - an exception. we should make it return void and just throw an exception anytime - it fails - */ checkConnection(); - try { - toSend.header().setId(nextMessageId()); - toSend.header().setResponseToMsgId(0); - auto swm = _compressorManager.compressMessage(toSend); - uassertStatusOK(swm.getStatus()); - - if (!port().call(swm.getValue(), response)) { - _failed = true; - if (assertOk) - uasserted(10278, - str::stream() << "dbclient error communicating with server: " - << getServerAddress()); - return false; - } + auto killSessionOnError = MakeGuard([this] { _markFailed(kEndSession); }); + auto maybeThrow = [&](const auto& errStatus) { + if (assertOk) + uasserted(10278, + str::stream() << "dbclient error communicating with server " + << getServerAddress() + << ": " + << redact(errStatus)); + return false; + }; - if (response.operation() == dbCompressed) { - response = uassertStatusOK(_compressorManager.decompressMessage(response)); - } - } catch (const DBException&) { - _failed = true; - _port->shutdown(); - throw; + toSend.header().setId(nextMessageId()); + toSend.header().setResponseToMsgId(0); + auto swm = _compressorManager.compressMessage(toSend); + uassertStatusOK(swm.getStatus()); + + auto sinkStatus = _session->sinkMessage(swm.getValue()); + if (!sinkStatus.isOK()) { + return maybeThrow(sinkStatus); + } + + swm = _session->sourceMessage(); + if (swm.isOK()) { + response = std::move(swm.getValue()); + } else { + return maybeThrow(swm.getStatus()); } + + if (response.operation() == dbCompressed) { + response = uassertStatusOK(_compressorManager.decompressMessage(response)); + } + + killSessionOnError.Dismiss(); return true; } @@ -1504,7 +1534,7 @@ void DBClientConnection::handleNotMasterResponse(const BSONObj& replyBody, << _parentReplSetName}); } - _failed = true; + _markFailed(kSetFlag); } AtomicInt32 DBClientConnection::_numConnections; diff --git a/src/mongo/client/dbclientinterface.h b/src/mongo/client/dbclientinterface.h index 86848f9ed93..457839a9fba 100644 --- a/src/mongo/client/dbclientinterface.h +++ b/src/mongo/client/dbclientinterface.h @@ -45,8 +45,9 @@ #include "mongo/rpc/unique_message.h" #include "mongo/stdx/functional.h" #include "mongo/transport/message_compressor_manager.h" +#include "mongo/transport/session.h" +#include "mongo/transport/transport_layer.h" #include "mongo/util/mongoutils/str.h" -#include "mongo/util/net/abstract_message_port.h" #include "mongo/util/net/message.h" #include "mongo/util/net/op_msg.h" @@ -969,9 +970,11 @@ public: return _failed; } - bool isStillConnected() { - return _port ? _port->isStillConnected() : true; - } + bool isStillConnected(); + + void setTags(transport::Session::TagMask tag); + + void shutdown(); void setWireVersions(int minWireVersion, int maxWireVersion) { _minWireVersion = minWireVersion; @@ -986,11 +989,6 @@ public: return _maxWireVersion; } - AbstractMessagingPort& port() { - verify(_port); - return *_port; - } - std::string toString() const { std::stringstream ss; ss << _serverAddress; @@ -1020,7 +1018,7 @@ public: } void setSoTimeout(double timeout); double getSoTimeout() const { - return _so_timeout; + return _socketTimeout.value_or(Milliseconds{0}).count() / 1000.0; } virtual bool lazySupported() const { @@ -1037,7 +1035,7 @@ public: */ void setParentReplSetName(const std::string& replSetName); - uint64_t getSockCreationMicroSec() const; + uint64_t getSockCreationMicroSec() const override; MessageCompressorManager& getCompressorManager() { return _compressorManager; @@ -1065,9 +1063,13 @@ protected: virtual void _auth(const BSONObj& params); - std::unique_ptr _port; + transport::SessionHandle _session; + boost::optional _socketTimeout; + transport::Session::TagMask _tagMask = transport::Session::kEmptyTagMask; + uint64_t _sessionCreationMicros = INVALID_SOCK_CREATION_TIME; + Date_t _lastConnectivityCheck; - bool _failed; + bool _failed = false; const bool autoReconnect; Backoff autoReconnectBackoff; @@ -1078,7 +1080,6 @@ protected: void _checkConnection(); std::map authCache; - double _so_timeout; static AtomicInt32 _numConnections; @@ -1089,6 +1090,8 @@ private: * returned. */ void handleNotMasterResponse(const BSONObj& replyBody, StringData errorMsgFieldName); + enum FailAction { kSetFlag, kEndSession, kReleaseSession }; + void _markFailed(FailAction action); // Contains the string for the replica set name of the host this is connected to. // Should be empty if this connection is not pointing to a replica set member. diff --git a/src/mongo/db/client.h b/src/mongo/db/client.h index 86629690af2..40e77ac04ab 100644 --- a/src/mongo/db/client.h +++ b/src/mongo/db/client.h @@ -52,7 +52,6 @@ namespace mongo { -class AbstractMessagingPort; class Collection; class OperationContext; diff --git a/src/mongo/db/repl/oplogreader.cpp b/src/mongo/db/repl/oplogreader.cpp index 659e061c554..9f5f01021bf 100644 --- a/src/mongo/db/repl/oplogreader.cpp +++ b/src/mongo/db/repl/oplogreader.cpp @@ -82,8 +82,7 @@ bool OplogReader::connect(const HostAndPort& host) { error() << errmsg << endl; return false; } - _conn->port().setTag(_conn->port().getTag() | - executor::NetworkInterface::kMessagingPortKeepOpen); + _conn->setTags(executor::NetworkInterface::kMessagingPortKeepOpen); _host = host; } return true; diff --git a/src/mongo/dbtests/repltests.cpp b/src/mongo/dbtests/repltests.cpp index 6306c4ce0db..4e4937b2e71 100644 --- a/src/mongo/dbtests/repltests.cpp +++ b/src/mongo/dbtests/repltests.cpp @@ -52,6 +52,7 @@ #include "mongo/db/repl/replication_coordinator_mock.h" #include "mongo/db/repl/sync_tail.h" #include "mongo/dbtests/dbtests.h" +#include "mongo/transport/transport_layer_asio.h" #include "mongo/util/log.h" using namespace mongo::repl; @@ -109,6 +110,15 @@ public: if (mongo::storageGlobalParams.engine == "mobile") { return; } + + transport::TransportLayerASIO::Options opts; + opts.mode = transport::TransportLayerASIO::Options::kEgress; + auto sc = getGlobalServiceContext(); + + sc->setTransportLayer(std::make_unique(opts, nullptr)); + ASSERT_OK(sc->getTransportLayer()->setup()); + ASSERT_OK(sc->getTransportLayer()->start()); + ReplSettings replSettings; replSettings.setOplogSizeBytes(10 * 1024 * 1024); replSettings.setMaster(true); @@ -154,6 +164,7 @@ public: ->setFollowerMode(repl::MemberState::RS_PRIMARY) .ignore(); + getGlobalServiceContext()->getTransportLayer()->shutdown(); } catch (...) { FAIL("Exception while cleaning up test"); diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript index 722599088bb..83c784ff0be 100644 --- a/src/mongo/s/SConscript +++ b/src/mongo/s/SConscript @@ -409,6 +409,7 @@ env.CppUnitTest( LIBDEPS=[ '$BUILD_DIR/mongo/db/service_context_noop_init', '$BUILD_DIR/mongo/dbtests/mocklib', + '$BUILD_DIR/mongo/transport/transport_layer_egress_init', '$BUILD_DIR/mongo/util/net/network', 'client/sharding_connection_hook', 'sharding_legacy_api', diff --git a/src/mongo/shell/dbshell.cpp b/src/mongo/shell/dbshell.cpp index eef4859804a..c6e4724efc6 100644 --- a/src/mongo/shell/dbshell.cpp +++ b/src/mongo/shell/dbshell.cpp @@ -57,6 +57,7 @@ #include "mongo/shell/shell_utils.h" #include "mongo/shell/shell_utils_launcher.h" #include "mongo/stdx/utility.h" +#include "mongo/transport/transport_layer_asio.h" #include "mongo/util/exit.h" #include "mongo/util/file.h" #include "mongo/util/log.h" @@ -741,6 +742,18 @@ int _main(int argc, char* argv[], char** envp) { mongo::runGlobalInitializersOrDie(argc, argv, envp); + // TODO This should use a TransportLayerManager or TransportLayerFactory + auto serviceContext = getGlobalServiceContext(); + transport::TransportLayerASIO::Options opts; + opts.enableIPv6 = shellGlobalParams.enableIPv6; + opts.mode = transport::TransportLayerASIO::Options::kEgress; + + serviceContext->setTransportLayer( + std::make_unique(opts, nullptr)); + auto tlPtr = serviceContext->getTransportLayer(); + uassertStatusOK(tlPtr->setup()); + uassertStatusOK(tlPtr->start()); + // hide password from ps output for (int i = 0; i < (argc - 1); ++i) { if (!strcmp(argv[i], "-p") || !strcmp(argv[i], "--password")) { diff --git a/src/mongo/shell/shell_options.cpp b/src/mongo/shell/shell_options.cpp index 537b61c87cf..8e8ca5f4b3e 100644 --- a/src/mongo/shell/shell_options.cpp +++ b/src/mongo/shell/shell_options.cpp @@ -281,6 +281,7 @@ Status storeMongoShellOptions(const moe::Environment& params, #endif if (params.count("ipv6")) { mongo::enableIPv6(); + shellGlobalParams.enableIPv6 = true; } if (params.count("verbose")) { logger::globalLogDomain()->setMinimumLoggedSeverity(logger::LogSeverity::Debug(1)); diff --git a/src/mongo/shell/shell_options.h b/src/mongo/shell/shell_options.h index 4092c6a74df..1fe68e571a4 100644 --- a/src/mongo/shell/shell_options.h +++ b/src/mongo/shell/shell_options.h @@ -63,6 +63,7 @@ struct ShellGlobalParams { bool norc; bool nojit = true; bool javascriptProtection = true; + bool enableIPv6 = false; std::string script; diff --git a/src/mongo/tools/bridge.cpp b/src/mongo/tools/bridge.cpp index 5f9eac1e4e0..375c88bbc1b 100644 --- a/src/mongo/tools/bridge.cpp +++ b/src/mongo/tools/bridge.cpp @@ -49,6 +49,7 @@ #include "mongo/stdx/thread.h" #include "mongo/tools/bridge_commands.h" #include "mongo/tools/mongobridge_options.h" +#include "mongo/transport/transport_layer_asio.h" #include "mongo/util/assert_util.h" #include "mongo/util/exit.h" #include "mongo/util/log.h" @@ -91,33 +92,33 @@ public: : _mp(mp), _settingsMutex(settingsMutex), _settings(settings), _prng(seed) {} void operator()() { - DBClientConnection dest; - - { + transport::SessionHandle dest = []() -> transport::SessionHandle { HostAndPort destAddr{mongoBridgeGlobalParams.destUri}; const Seconds kConnectTimeout(30); - Timer connectTimer; - while (true) { - // DBClientConnection::connectSocketOnly() is used instead of - // DBClientConnection::connect() to avoid sending an isMaster command when the - // connection is established. We'd otherwise trigger a socket timeout when - // forwarding an _isSelf command because dest's replication subsystem hasn't been - // initialized yet and so it cannot respond to the isMaster command. - auto status = dest.connectSocketOnly(destAddr); - if (status.isOK()) { - break; - } - Seconds elapsed{connectTimer.seconds()}; - if (elapsed >= kConnectTimeout) { - warning() << "Unable to establish connection to " - << mongoBridgeGlobalParams.destUri << " after " << elapsed - << " seconds: " << status; - log() << "end connection " << _mp->remote().toString(); - _mp->shutdown(); - return; + auto now = getGlobalServiceContext()->getFastClockSource()->now(); + const auto connectExpiration = now + kConnectTimeout; + while (now < connectExpiration) { + auto tl = getGlobalServiceContext()->getTransportLayer(); + auto sws = + tl->connect(destAddr, transport::kGlobalSSLMode, connectExpiration - now); + auto status = sws.getStatus(); + if (!status.isOK()) { + warning() << "Unable to establish connection to " << destAddr << ": " << status; + now = getGlobalServiceContext()->getFastClockSource()->now(); + } else { + return std::move(sws.getValue()); } + sleepmillis(500); } + + return nullptr; + }(); + + if (!dest) { + log() << "end connection " << _mp->remote(); + _mp->shutdown(); + return; } bool receivingFirstMessage = true; @@ -231,8 +232,11 @@ public: request.operation() == dbCommand || request.operation() == dbMsg)) { // TODO dbMsg moreToCome // Forward the message to 'dest' and receive its reply in 'response'. - response.reset(); - dest.port().call(request, response); + uassertStatusOK(dest->sinkMessage(request)); + response = uassertStatusOK(dest->sourceMessage()); + uassert(50727, + "Response ID did not match the sent message ID.", + response.header().getResponseToMsgId() == request.header().getId()); // If there's nothing to respond back to '_mp' with, then close the connection. if (response.empty()) { @@ -280,15 +284,14 @@ public: MsgData::View header = response.header(); QueryResult::View qr = header.view2ptr(); if (qr.getCursorId()) { - response.reset(); - dest.port().recv(response); + response = uassertStatusOK(dest->sourceMessage()); _mp->say(response); } else { exhaust = false; } } } else { - dest.port().say(request); + uassertStatusOK(dest->sinkMessage(request)); } } catch (const DBException& ex) { error() << "Caught DBException in Forwarder: " << ex << ", end connection " @@ -409,6 +412,25 @@ int bridgeMain(int argc, char** argv, char** envp) { runGlobalInitializersOrDie(argc, argv, envp); startSignalProcessingThread(LogFileStatus::kNoLogFileToRotate); + auto serviceContext = getGlobalServiceContext(); + transport::TransportLayerASIO::Options opts; + opts.mode = mongo::transport::TransportLayerASIO::Options::kEgress; + + serviceContext->setTransportLayer( + std::make_unique(opts, nullptr)); + auto tl = serviceContext->getTransportLayer(); + if (!tl->setup().isOK()) { + log() << "Error setting up transport layer"; + return EXIT_NET_ERROR; + } + + if (!tl->start().isOK()) { + log() << "Error starting transport layer"; + return EXIT_NET_ERROR; + } + + serviceContext->notifyStartupComplete(); + listener = stdx::make_unique(); listener->setupSockets(); listener->initAndListen(); diff --git a/src/mongo/transport/SConscript b/src/mongo/transport/SConscript index 8527328b08c..f5cce983eac 100644 --- a/src/mongo/transport/SConscript +++ b/src/mongo/transport/SConscript @@ -67,6 +67,19 @@ tlEnv.Library( ], ) +# This library will initialize an egress transport layer in a mongo initializer +# for C++ tests that require networking. +env.Library( + target='transport_layer_egress_init', + source=[ + 'transport_layer_egress_init.cpp', + ], + LIBDEPS_PRIVATE=[ + 'transport_layer', + '$BUILD_DIR/mongo/db/service_context_noop_init', + ] +) + tlEnv.CppUnitTest( target='transport_layer_asio_test', source=[ diff --git a/src/mongo/transport/asio_utils.h b/src/mongo/transport/asio_utils.h index 89e7821f5b8..e5647a19264 100644 --- a/src/mongo/transport/asio_utils.h +++ b/src/mongo/transport/asio_utils.h @@ -30,9 +30,14 @@ #include "mongo/base/status.h" #include "mongo/base/system_error.h" +#include "mongo/util/errno_util.h" #include "mongo/util/net/hostandport.h" #include "mongo/util/net/sockaddr.h" +#ifndef _WIN32 +#include +#endif // ndef _WIN32 + #include namespace mongo { @@ -60,6 +65,9 @@ inline Status errorCodeToStatus(const std::error_code& ec) { if (ec == asio::error::try_again || ec == asio::error::would_block) { #endif return {ErrorCodes::NetworkTimeout, "Socket operation timed out"}; + } else if (ec == asio::error::eof || ec == asio::error::connection_reset || + ec == asio::error::network_reset) { + return {ErrorCodes::HostUnreachable, "Connection was closed"}; } // If the ec.category() is a mongoErrorCategory() then this error was propogated from @@ -73,5 +81,90 @@ inline Status errorCodeToStatus(const std::error_code& ec) { return {errorCode, ec.message()}; } +/* + * The ASIO implementation of poll (i.e. socket.wait()) cannot poll for a mask of events, and + * doesn't support timeouts. + * + * This wraps up ::select/::poll for Windows/POSIX for a single socket and handles EINTR on POSIX + * + * - On timeout: it returns Status(ErrorCodes::NetworkTimeout) + * - On poll returning with an event: it returns the EventsMask for the socket, the caller must + * check whether it matches the expected events mask. + * - On error: it returns a Status(ErrorCodes::InternalError) + */ +template +StatusWith pollASIOSocket(Socket& socket, EventsMask mask, Milliseconds timeout) { +#ifdef _WIN32 + fd_set readfds; + fd_set writefds; + fd_set errfds; + + FD_ZERO(&readfds); + FD_ZERO(&writefds); + FD_ZERO(&errfds); + + auto fd = socket.native_handle(); + if (mask & POLLIN) { + FD_SET(fd, &readfds); + } + if (mask & POLLOUT) { + FD_SET(fd, &writefds); + } + FD_SET(fd, &errfds); + + timeval timeoutTv{}; + auto timeoutUs = duration_cast(timeout); + if (timeoutUs >= Seconds{1}) { + auto timeoutSec = duration_cast(timeoutUs); + timeoutTv.tv_sec = timeoutSec.count(); + timeoutUs -= timeoutSec; + } + timeoutTv.tv_usec = timeoutUs.count(); + int result = ::select(1, &readfds, &writefds, &errfds, &timeoutTv); + if (result == SOCKET_ERROR) { + auto errDesc = errnoWithDescription(WSAGetLastError()); + return {ErrorCodes::InternalError, errDesc}; + } + int revents = (FD_ISSET(fd, &readfds) ? POLLIN : 0) | (FD_ISSET(fd, &writefds) ? POLLOUT : 0) | + (FD_ISSET(fd, &errfds) ? POLLERR : 0); +#else + pollfd pollItem; + pollItem.fd = socket.native_handle(); + pollItem.events = mask; + + int result; + boost::optional expiration; + if (timeout.count() > 0) { + expiration = Date_t::now() + timeout; + } + do { + Milliseconds curTimeout; + if (expiration) { + curTimeout = *expiration - Date_t::now(); + if (curTimeout.count() <= 0) { + result = 0; + break; + } + } else { + curTimeout = timeout; + } + result = ::poll(&pollItem, 1, curTimeout.count()); + } while (result == -1 && errno == EINTR); + + if (result == -1) { + int errCode = errno; + return {ErrorCodes::InternalError, errnoWithDescription(errCode)}; + } + int revents = pollItem.revents; +#endif + + if (result == 0) { + return {ErrorCodes::NetworkTimeout, "Timed out waiting for poll"}; + } else { + return revents; + } +} + + } // namespace transport } // namespace mongo diff --git a/src/mongo/transport/mock_session.h b/src/mongo/transport/mock_session.h index 411e57bd0ec..c9d519f6288 100644 --- a/src/mongo/transport/mock_session.h +++ b/src/mongo/transport/mock_session.h @@ -105,6 +105,10 @@ public: void setTimeout(boost::optional) override {} + bool isConnected() override { + return true; + } + explicit MockSession(TransportLayer* tl) : _tl(checked_cast(tl)), _remote(), _local() {} explicit MockSession(HostAndPort remote, HostAndPort local, TransportLayer* tl) diff --git a/src/mongo/transport/session.h b/src/mongo/transport/session.h index db80aa0814d..c9438269ec7 100644 --- a/src/mongo/transport/session.h +++ b/src/mongo/transport/session.h @@ -122,6 +122,16 @@ public: */ virtual void setTimeout(boost::optional timeout) = 0; + /** + * This will return whether calling sourceMessage()/sinkMessage() will fail with an EOF error. + * + * Implementations may actually perform some I/O or call syscalls to determine this, rather + * than just checking a flag. + * + * This must not be called while the session is currently sourcing or sinking a message. + */ + virtual bool isConnected() = 0; + virtual const HostAndPort& remote() const = 0; virtual const HostAndPort& local() const = 0; diff --git a/src/mongo/transport/session_asio.h b/src/mongo/transport/session_asio.h index f18dfac6fe1..a1100eed535 100644 --- a/src/mongo/transport/session_asio.h +++ b/src/mongo/transport/session_asio.h @@ -94,7 +94,7 @@ public: std::error_code ec; getSocket().cancel(); getSocket().shutdown(GenericSocket::shutdown_both, ec); - if (ec) { + if ((ec) && (ec != asio::error::not_connected)) { error() << "Error shutting down socket: " << ec.message(); } } @@ -163,6 +163,120 @@ public: _configuredTimeout = timeout; } + bool isConnected() override { + // socket.is_open() only returns whether the socket is a valid file descriptor and + // if we haven't marked this socket as closed already. + if (!getSocket().is_open()) + return false; + + auto swPollEvents = pollASIOSocket(getSocket(), POLLIN, Milliseconds{0}); + if (!swPollEvents.isOK()) { + if (swPollEvents != ErrorCodes::NetworkTimeout) { + warning() << "Failed to poll socket for connectivity check: " + << swPollEvents.getStatus(); + return false; + } + return true; + } + + auto revents = swPollEvents.getValue(); + if (revents & POLLIN) { + char testByte; + int size = ::recv(getSocket().native_handle(), &testByte, sizeof(testByte), MSG_PEEK); + if (size == sizeof(testByte)) { + return true; + } else if (size == -1) { + auto errDesc = errnoWithDescription(errno); + warning() << "Failed to check socket connectivity: " << errDesc; + } + // If size == 0 then we got disconnected and we should return false. + } + + return false; + } + +protected: + friend class TransportLayerASIO; + +#ifdef MONGO_CONFIG_SSL + template + void handshakeSSLForEgress(HostAndPort target, HandshakeCb onComplete) { + if (!_tl->_egressSSLContext) { + return onComplete( + {ErrorCodes::SSLHandshakeFailed, "SSL requested but SSL support is disabled"}); + } + + _sslSocket.emplace(std::move(_socket), *_tl->_egressSSLContext); + auto handshakeCompleteCb = + [ this, target = std::move(target), onComplete = std::move(onComplete) ]( + const std::error_code& ec) { + _ranHandshake = true; + if (ec) { + onComplete(errorCodeToStatus(ec)); + return; + } + + auto sslManager = getSSLManager(); + auto swPeerInfo = sslManager->parseAndValidatePeerCertificate( + _sslSocket->native_handle(), target.host()); + if (!swPeerInfo.isOK()) { + onComplete(swPeerInfo.getStatus()); + return; + } + + if (swPeerInfo.getValue()) { + SSLPeerInfo::forSession(shared_from_this()) = std::move(*swPeerInfo.getValue()); + } + + onComplete(Status::OK()); + }; + if (_blockingMode == Sync) { + std::error_code ec; + _sslSocket->handshake(asio::ssl::stream_base::client, ec); + handshakeCompleteCb(ec); + } else { + return _sslSocket->async_handshake(asio::ssl::stream_base::client, + std::move(handshakeCompleteCb)); + } + } +#endif + + void ensureSync() { + asio::error_code ec; + if (_blockingMode != Sync) { + getSocket().non_blocking(false, ec); + fassertStatusOK(40490, errorCodeToStatus(ec)); + _blockingMode = Sync; + } + + if (_socketTimeout != _configuredTimeout) { + // Change boost::none (which means no timeout) into a zero value for the socket option, + // which also means no timeout. + auto timeout = _configuredTimeout.value_or(Milliseconds{0}); + getSocket().set_option(ASIOSocketTimeoutOption(timeout), ec); + uassertStatusOK(errorCodeToStatus(ec)); + + getSocket().set_option(ASIOSocketTimeoutOption(timeout), ec); + uassertStatusOK(errorCodeToStatus(ec)); + + _socketTimeout = _configuredTimeout; + } + } + + void ensureAsync() { + if (_blockingMode == Async) + return; + + // Socket timeouts currently only effect synchronous calls, so make sure the caller isn't + // expecting a socket timeout when they do an async operation. + invariant(!_configuredTimeout); + + asio::error_code ec; + getSocket().non_blocking(true, ec); + fassertStatusOK(50706, errorCodeToStatus(ec)); + _blockingMode = Async; + } + private: template class ASIOSocketTimeoutOption { @@ -215,14 +329,6 @@ private: return _socket; } - bool isOpen() const { -#ifdef MONGO_CONFIG_SSL - return _sslSocket ? _sslSocket->lowest_layer().is_open() : _socket.is_open(); -#else - return _socket.is_open(); -#endif - } - template void sourceMessageImpl(Callback&& cb) { static constexpr auto kHeaderSize = sizeof(MSGHEADER::Value); @@ -300,7 +406,7 @@ private: return; } - maybeHandshakeSSL(buffers, std::move(postHandshakeCb)); + maybeHandshakeSSLForIngress(buffers, std::move(postHandshakeCb)); }; return opportunisticRead(_socket, buffers, std::move(handshakeRecvCb)); @@ -312,6 +418,7 @@ private: template void write(const ConstBufferSequence& buffers, CompleteHandler&& handler) { #ifdef MONGO_CONFIG_SSL + _ranHandshake = true; if (_sslSocket) { return opportunisticWrite(*_sslSocket, buffers, std::forward(handler)); } @@ -319,42 +426,6 @@ private: return opportunisticWrite(_socket, buffers, std::forward(handler)); } - void ensureSync() { - asio::error_code ec; - if (_blockingMode != Sync) { - getSocket().non_blocking(false, ec); - fassertStatusOK(40490, errorCodeToStatus(ec)); - _blockingMode = Sync; - } - - if (_socketTimeout != _configuredTimeout) { - // Change boost::none (which means no timeout) into a zero value for the socket option, - // which also means no timeout. - auto timeout = _configuredTimeout.value_or(Milliseconds{0}); - getSocket().set_option(ASIOSocketTimeoutOption(timeout), ec); - uassertStatusOK(errorCodeToStatus(ec)); - - getSocket().set_option(ASIOSocketTimeoutOption(timeout), ec); - uassertStatusOK(errorCodeToStatus(ec)); - - _socketTimeout = _configuredTimeout; - } - } - - void ensureAsync() { - if (_blockingMode == Async) - return; - - // Socket timeouts currently only effect synchronous calls, so make sure the caller isn't - // expecting a socket timeout when they do an async operation. - invariant(!_configuredTimeout); - - asio::error_code ec; - getSocket().non_blocking(true, ec); - fassertStatusOK(50706, errorCodeToStatus(ec)); - _blockingMode = Async; - } - template void opportunisticRead(Stream& stream, const MutableBufferSequence& buffers, @@ -411,7 +482,7 @@ private: #ifdef MONGO_CONFIG_SSL template - void maybeHandshakeSSL(const MutableBufferSequence& buffer, HandshakeCb onComplete) { + void maybeHandshakeSSLForIngress(const MutableBufferSequence& buffer, HandshakeCb onComplete) { invariant(asio::buffer_size(buffer) >= sizeof(MSGHEADER::Value)); MSGHEADER::ConstView headerView(asio::buffer_cast(buffer)); auto responseTo = headerView.getResponseToMsgId(); @@ -424,15 +495,14 @@ private: // protocol message needs to be 0 or -1. Otherwise the connection is either sending // garbage or a TLS Hello packet which will be caught by the TLS handshake. if (responseTo != 0 && responseTo != -1) { - if (!_tl->_sslContext) { + if (!_tl->_ingressSSLContext) { return onComplete( {ErrorCodes::SSLHandshakeFailed, "SSL handshake received but server is started without SSL support"}, false); } - _sslSocket.emplace(std::move(_socket), *_tl->_sslContext); - + _sslSocket.emplace(std::move(_socket), *_tl->_ingressSSLContext); auto handshakeCompleteCb = [ this, onComplete = std::move(onComplete) ]( const std::error_code& ec, size_t size) mutable { auto& sslPeerInfo = SSLPeerInfo::forSession(shared_from_this()); diff --git a/src/mongo/transport/transport_layer.h b/src/mongo/transport/transport_layer.h index 053aa5f6c15..29b4a7bb72c 100644 --- a/src/mongo/transport/transport_layer.h +++ b/src/mongo/transport/transport_layer.h @@ -37,6 +37,8 @@ namespace mongo { namespace transport { +enum ConnectSSLMode { kGlobalSSLMode, kEnableSSL, kDisableSSL }; + /** * The TransportLayer moves Messages between transport::Endpoints and the database. * This class owns an Acceptor that generates new endpoints from which it can @@ -63,6 +65,15 @@ public: virtual ~TransportLayer() = default; + virtual StatusWith connect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout) = 0; + + virtual void asyncConnect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout, + std::function)> callback) = 0; + /** * Start the TransportLayer. After this point, the TransportLayer will begin accepting active * sessions from new transport::Endpoints. diff --git a/src/mongo/transport/transport_layer_asio.cpp b/src/mongo/transport/transport_layer_asio.cpp index 8e437ef949e..1f5d45e053b 100644 --- a/src/mongo/transport/transport_layer_asio.cpp +++ b/src/mongo/transport/transport_layer_asio.cpp @@ -36,9 +36,6 @@ #include #include "mongo/config.h" -#ifdef MONGO_CONFIG_SSL -#include "mongo/util/net/ssl.hpp" -#endif #include "mongo/base/system_error.h" #include "mongo/db/server_options.h" @@ -53,6 +50,10 @@ #include "mongo/util/net/ssl_manager.h" #include "mongo/util/net/ssl_options.h" +#ifdef MONGO_CONFIG_SSL +#include "mongo/util/net/ssl.hpp" +#endif + // session_asio.h has some header dependencies that require it to be the last header. #include "mongo/transport/session_asio.h" @@ -74,7 +75,8 @@ TransportLayerASIO::TransportLayerASIO(const TransportLayerASIO::Options& opts, : _workerIOContext(std::make_shared()), _acceptorIOContext(stdx::make_unique()), #ifdef MONGO_CONFIG_SSL - _sslContext(nullptr), + _ingressSSLContext(nullptr), + _egressSSLContext(nullptr), #endif _sep(sep), _listenerOptions(opts) { @@ -82,24 +84,164 @@ TransportLayerASIO::TransportLayerASIO(const TransportLayerASIO::Options& opts, TransportLayerASIO::~TransportLayerASIO() = default; +StatusWith TransportLayerASIO::connect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout) { + std::error_code ec; + GenericSocket sock(*_workerIOContext); +#ifndef _WIN32 + if (mongoutils::str::contains(peer.host(), '/')) { + invariant(!peer.hasPort()); + auto res = + _doSyncConnect(asio::local::stream_protocol::endpoint(peer.host()), peer, timeout); + if (!res.isOK()) { + return res.getStatus(); + } else { + return static_cast(std::move(res.getValue())); + } + } +#endif + + using Resolver = asio::ip::tcp::resolver; + Resolver resolver(*_workerIOContext); + std::string portNumberStr = std::to_string(peer.port()); + auto doResolve = [&](auto resolverFlags) -> StatusWith { + // If IPv6 is disabled, then we should specify that we only want IPv4 addresses, otherwise + // we should do a normal AF_UNSPEC resolution to get both IPv4/IPv6 + Resolver::iterator resolverIt; + if (_listenerOptions.enableIPv6) { + resolverIt = resolver.resolve(peer.host(), portNumberStr, resolverFlags, ec); + } else { + resolverIt = resolver.resolve( + asio::ip::tcp::v4(), peer.host(), portNumberStr, resolverFlags, ec); + } + + if (ec) { + return {ErrorCodes::HostNotFound, + str::stream() << "Could not find address for " << peer.host() << ": " + << ec.message()}; + } else if (resolverIt == Resolver::iterator()) { + return {ErrorCodes::HostNotFound, + str::stream() << "Could not find address for " << peer.host()}; + } + + return resolverIt; + }; + + // We always want to resolve the "service" (port number) as a numeric. + // + // We intentionally don't set the Resolver::address_configured flag because it might prevent us + // from connecting to localhost on hosts with only a loopback interface (see SERVER-1579). + const auto resolverFlags = Resolver::numeric_service; + + // We resolve in two steps, the first step tries to resolve the hostname as an IP address - + // that way if there's a DNS timeout, we can still connect to IP addresses quickly. + // (See SERVER-1709) + // + // Then, if the numeric (IP address) lookup failed, we fall back to DNS or return the error + // from the resolver. + auto swResolverIt = doResolve(resolverFlags | Resolver::numeric_host); + if (!swResolverIt.isOK()) { + if (swResolverIt == ErrorCodes::HostNotFound) { + swResolverIt = doResolve(resolverFlags); + if (!swResolverIt.isOK()) { + return swResolverIt.getStatus(); + } + } else { + return swResolverIt.getStatus(); + } + } + + auto& resolverIt = swResolverIt.getValue(); + auto sws = _doSyncConnect(resolverIt->endpoint(), peer, timeout); + if (!sws.isOK()) { + return sws.getStatus(); + } + + auto session = std::move(sws.getValue()); + session->ensureSync(); + +#ifndef MONGO_CONFIG_SSL + if (sslMode == kEnableSSL) { + return {ErrorCodes::InvalidSSLConfiguration, "SSL requested but not supported"}; + } +#else + auto globalSSLMode = _sslMode(); + if (sslMode == kEnableSSL || + (sslMode == kGlobalSSLMode && ((globalSSLMode == SSLParams::SSLMode_preferSSL) || + (globalSSLMode == SSLParams::SSLMode_requireSSL)))) { + Status sslStatus = Status::OK(); + auto onComplete = [&sslStatus](Status status) { sslStatus = status; }; + session->handshakeSSLForEgress(peer, std::move(onComplete)); + if (!sslStatus.isOK()) { + return sslStatus; + } + } +#endif + + return static_cast(std::move(session)); +} + +template +StatusWith TransportLayerASIO::_doSyncConnect( + Endpoint endpoint, const HostAndPort& peer, const Milliseconds& timeout) { + GenericSocket sock(*_workerIOContext); + std::error_code ec; + sock.open(endpoint.protocol()); + sock.non_blocking(true); + + auto now = Date_t::now(); + auto expiration = now + timeout; + do { + auto curTimeout = expiration - now; + sock.connect(endpoint, curTimeout.toSystemDuration(), ec); + if (ec) { + now = Date_t::now(); + } + // We loop below if ec == interrupted to deal with EINTR failures, otherwise we handle + // the error/timeout below. + } while (ec == asio::error::interrupted && now < expiration); + + if (ec) { + return errorCodeToStatus(ec); + } else if (now >= expiration) { + return {ErrorCodes::NetworkTimeout, str::stream() << "Timed out connecting to " << peer}; + } + + sock.non_blocking(false); + return std::make_shared(this, std::move(sock)); +} + +void TransportLayerASIO::asyncConnect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout, + std::function)> callback) { + MONGO_UNREACHABLE; +} + Status TransportLayerASIO::setup() { std::vector listenAddrs; - if (_listenerOptions.ipList.empty()) { + if (_listenerOptions.ipList.empty() && _listenerOptions.isIngress()) { listenAddrs = {"127.0.0.1"}; if (_listenerOptions.enableIPv6) { listenAddrs.emplace_back("::1"); } - } else { + } else if (!_listenerOptions.ipList.empty()) { boost::split( listenAddrs, _listenerOptions.ipList, boost::is_any_of(","), boost::token_compress_on); } #ifndef _WIN32 - if (_listenerOptions.useUnixSockets) { + if (_listenerOptions.useUnixSockets && _listenerOptions.isIngress()) { listenAddrs.emplace_back(makeUnixSockPath(_listenerOptions.port)); } #endif + if (!(_listenerOptions.isIngress()) && !listenAddrs.empty()) { + return {ErrorCodes::BadValue, + "Cannot bind to listening sockets with ingress networking is disabled"}; + } + _listenerPort = _listenerOptions.port; for (auto& ip : listenAddrs) { @@ -177,20 +319,32 @@ Status TransportLayerASIO::setup() { } } - if (_acceptors.empty()) { + if (_acceptors.empty() && _listenerOptions.isIngress()) { return Status(ErrorCodes::SocketException, "No available addresses/ports to bind to"); } #ifdef MONGO_CONFIG_SSL const auto& sslParams = getSSLGlobalParams(); + auto sslManager = getSSLManager(); - if (_sslMode() != SSLParams::SSLMode_disabled) { - _sslContext = stdx::make_unique(asio::ssl::context::sslv23); + if (_sslMode() != SSLParams::SSLMode_disabled && _listenerOptions.isIngress()) { + _ingressSSLContext = stdx::make_unique(asio::ssl::context::sslv23); Status status = - getSSLManager()->initSSLContext(_sslContext->native_handle(), - sslParams, - SSLManagerInterface::ConnectionDirection::kIncoming); + sslManager->initSSLContext(_ingressSSLContext->native_handle(), + sslParams, + SSLManagerInterface::ConnectionDirection::kIncoming); + if (!status.isOK()) { + return status; + } + } + + if (_listenerOptions.isEgress() && sslManager) { + _egressSSLContext = stdx::make_unique(asio::ssl::context::sslv23); + Status status = + sslManager->initSSLContext(_egressSSLContext->native_handle(), + sslParams, + SSLManagerInterface::ConnectionDirection::kOutgoing); if (!status.isOK()) { return status; } @@ -204,31 +358,35 @@ Status TransportLayerASIO::start() { stdx::lock_guard lk(_mutex); _running.store(true); - _listenerThread = stdx::thread([this] { - setThreadName("listener"); - while (_running.load()) { - asio::io_context::work work(*_acceptorIOContext); - try { - _acceptorIOContext->run(); - } catch (...) { - severe() << "Uncaught exception in the listener: " << exceptionToStatus(); - fassertFailed(40491); - } + if (_listenerOptions.isIngress()) { + for (auto& acceptor : _acceptors) { + acceptor.second.listen(serverGlobalParams.listenBacklog); + _acceptConnection(acceptor.second); } - }); - for (auto& acceptor : _acceptors) { - acceptor.second.listen(serverGlobalParams.listenBacklog); - _acceptConnection(acceptor.second); - } + _listenerThread = stdx::thread([this] { + setThreadName("listener"); + while (_running.load()) { + asio::io_context::work work(*_acceptorIOContext); + try { + _acceptorIOContext->run(); + } catch (...) { + severe() << "Uncaught exception in the listener: " << exceptionToStatus(); + fassertFailed(40491); + } + } + }); - const char* ssl = ""; + const char* ssl = ""; #ifdef MONGO_CONFIG_SSL - if (_sslMode() != SSLParams::SSLMode_disabled) { - ssl = " ssl"; - } + if (_sslMode() != SSLParams::SSLMode_disabled) { + ssl = " ssl"; + } #endif - log() << "waiting for connections on port " << _listenerPort << ssl; + log() << "waiting for connections on port " << _listenerPort << ssl; + } else { + invariant(_acceptors.empty()); + } return Status::OK(); } diff --git a/src/mongo/transport/transport_layer_asio.h b/src/mongo/transport/transport_layer_asio.h index 62c0e02edae..da2bdab547e 100644 --- a/src/mongo/transport/transport_layer_asio.h +++ b/src/mongo/transport/transport_layer_asio.h @@ -77,6 +77,19 @@ public: explicit Options(const ServerGlobalParams* params); Options() = default; + constexpr static auto kIngress = 0x1; + constexpr static auto kEgress = 0x10; + + int mode = kIngress | kEgress; + + bool isIngress() const { + return mode & kIngress; + } + + bool isEgress() const { + return mode & kEgress; + } + int port = ServerGlobalParams::DefaultDBPort; // port to bind to std::string ipList; // addresses to bind to #ifndef _WIN32 @@ -92,6 +105,14 @@ public: virtual ~TransportLayerASIO(); + StatusWith connect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout) final; + void asyncConnect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout, + std::function)> callback) final; + Status setup() final; Status start() final; @@ -111,6 +132,12 @@ private: using GenericAcceptor = asio::basic_socket_acceptor; void _acceptConnection(GenericAcceptor& acceptor); + + template + StatusWith _doSyncConnect(Endpoint endpoint, + const HostAndPort& peer, + const Milliseconds& timeout); + #ifdef MONGO_CONFIG_SSL SSLParams::SSLModes _sslMode() const; #endif @@ -144,7 +171,8 @@ private: std::unique_ptr _acceptorIOContext; #ifdef MONGO_CONFIG_SSL - std::unique_ptr _sslContext; + std::unique_ptr _ingressSSLContext; + std::unique_ptr _egressSSLContext; #endif std::vector> _acceptors; diff --git a/src/mongo/transport/transport_layer_egress_init.cpp b/src/mongo/transport/transport_layer_egress_init.cpp new file mode 100644 index 00000000000..20a4fe5a47a --- /dev/null +++ b/src/mongo/transport/transport_layer_egress_init.cpp @@ -0,0 +1,58 @@ +/** + * Copyright 2018 MongoDB Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License, version 3, + * as published by the Free Software Foundation. + * + * 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 Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * As a special exception, the copyright holders give permission to link the + * code of portions of this program with the OpenSSL library under certain + * conditions as described in each individual source file and distribute + * linked combinations including the program with the OpenSSL library. You + * must comply with the GNU Affero General Public License in all respects + * for all of the code used other than as permitted herein. If you modify + * file(s) with this exception, you may extend this exception to your + * version of the file(s), but you are not obligated to do so. If you do not + * wish to do so, delete this exception statement from your version. If you + * delete this exception statement from all source files in the program, + * then also delete it in the license file. + */ + +#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kDefault + +#include "mongo/platform/basic.h" + +#include "mongo/base/init.h" +#include "mongo/db/service_context.h" +#include "mongo/transport/transport_layer_asio.h" + +namespace mongo { +namespace { +// Linking with this file will configure an egress-only TransportLayer on a ServiceContextNoop. +// Use this for unit/integration tests that require only egress networking. +MONGO_INITIALIZER_WITH_PREREQUISITES(ConfigureEgressTransportLayer, ("SetGlobalEnvironment")) +(InitializerContext* context) { + auto sc = getGlobalServiceContext(); + invariant(!sc->getTransportLayer()); + + transport::TransportLayerASIO::Options opts; + opts.mode = transport::TransportLayerASIO::Options::kEgress; + sc->setTransportLayer(std::make_unique(opts, nullptr)); + auto status = sc->getTransportLayer()->setup(); + if (!status.isOK()) { + return status; + } + + return sc->getTransportLayer()->start(); +} + +} // namespace +} // namespace diff --git a/src/mongo/transport/transport_layer_manager.cpp b/src/mongo/transport/transport_layer_manager.cpp index 28bcf351e51..f53250e91bd 100644 --- a/src/mongo/transport/transport_layer_manager.cpp +++ b/src/mongo/transport/transport_layer_manager.cpp @@ -59,6 +59,19 @@ void TransportLayerManager::_foreach(Callable&& cb) const { } } +StatusWith TransportLayerManager::connect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout) { + return _tls.front()->connect(peer, sslMode, timeout); +} + +void TransportLayerManager::asyncConnect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout, + std::function)> callback) { + MONGO_UNREACHABLE; +} + // TODO Right now this and setup() leave TLs started if there's an error. In practice the server // exits with an error and this isn't an issue, but we should make this more robust. Status TransportLayerManager::start() { diff --git a/src/mongo/transport/transport_layer_manager.h b/src/mongo/transport/transport_layer_manager.h index 7c80935f00a..70f54b10652 100644 --- a/src/mongo/transport/transport_layer_manager.h +++ b/src/mongo/transport/transport_layer_manager.h @@ -57,6 +57,14 @@ public: : _tls(std::move(tls)) {} TransportLayerManager(); + StatusWith connect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout) override; + void asyncConnect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout, + std::function)> callback) override; + Status start() override; void shutdown() override; Status setup() override; diff --git a/src/mongo/transport/transport_layer_mock.cpp b/src/mongo/transport/transport_layer_mock.cpp index ec1a28e6777..0fc6ce7964c 100644 --- a/src/mongo/transport/transport_layer_mock.cpp +++ b/src/mongo/transport/transport_layer_mock.cpp @@ -62,6 +62,19 @@ bool TransportLayerMock::owns(Session::Id id) { return _sessions.count(id) > 0; } +StatusWith TransportLayerMock::connect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout) { + MONGO_UNREACHABLE; +} + +void TransportLayerMock::asyncConnect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout, + std::function)> callback) { + MONGO_UNREACHABLE; +} + Status TransportLayerMock::setup() { return Status::OK(); } diff --git a/src/mongo/transport/transport_layer_mock.h b/src/mongo/transport/transport_layer_mock.h index 7a861df129a..06a9cd3f37c 100644 --- a/src/mongo/transport/transport_layer_mock.h +++ b/src/mongo/transport/transport_layer_mock.h @@ -53,6 +53,14 @@ public: SessionHandle get(Session::Id id); bool owns(Session::Id id); + StatusWith connect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout) override; + void asyncConnect(HostAndPort peer, + ConnectSSLMode sslMode, + Milliseconds timeout, + std::function)> callback) override; + Status setup() override; Status start() override; void shutdown() override; diff --git a/src/mongo/unittest/SConscript b/src/mongo/unittest/SConscript index 60315dcc5bd..e00502d94f7 100644 --- a/src/mongo/unittest/SConscript +++ b/src/mongo/unittest/SConscript @@ -32,6 +32,9 @@ env.Library(target="integration_test_main", '$BUILD_DIR/mongo/client/connection_string', '$BUILD_DIR/mongo/util/options_parser/options_parser_init', ], + LIBDEPS_PRIVATE=[ + '$BUILD_DIR/mongo/transport/transport_layer_egress_init', + ], ) bmEnv = env.Clone() diff --git a/src/mongo/unittest/integration_test_main.cpp b/src/mongo/unittest/integration_test_main.cpp index 69743f6c4e2..aaaf12f8c23 100644 --- a/src/mongo/unittest/integration_test_main.cpp +++ b/src/mongo/unittest/integration_test_main.cpp @@ -36,6 +36,8 @@ #include "mongo/base/initializer.h" #include "mongo/client/connection_string.h" +#include "mongo/db/service_context.h" +#include "mongo/transport/transport_layer_asio.h" #include "mongo/unittest/unittest.h" #include "mongo/util/log.h" #include "mongo/util/options_parser/environment.h" -- cgit v1.2.1