/** * Copyright (C) 2018-present MongoDB, Inc. * * This program is free software: you can redistribute it and/or modify * it under the terms of the Server Side Public License, version 1, * as published by MongoDB, Inc. * * 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 * Server Side Public License for more details. * * You should have received a copy of the Server Side 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 Server Side 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_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kNetwork #include "mongo/platform/basic.h" #include "mongo/transport/service_entry_point_impl.h" #include #include "mongo/db/auth/restriction_environment.h" #include "mongo/db/service_context.h" #include "mongo/logv2/log.h" #include "mongo/transport/hello_metrics.h" #include "mongo/transport/service_state_machine.h" #include "mongo/transport/session.h" #include "mongo/util/processinfo.h" #include "mongo/util/scopeguard.h" #if !defined(_WIN32) #include #endif namespace mongo { bool shouldOverrideMaxConns(const transport::SessionHandle& session, const std::vector>& exemptions) { const auto& remoteAddr = session->remoteAddr(); const auto& localAddr = session->localAddr(); boost::optional remoteCIDR; if (remoteAddr.isValid() && remoteAddr.isIP()) { remoteCIDR = uassertStatusOK(CIDR::parse(remoteAddr.getAddr())); } for (const auto& exemption : exemptions) { // If this exemption is a CIDR range, then we check that the remote IP is in the // CIDR range if ((stdx::holds_alternative(exemption)) && (remoteCIDR)) { if (stdx::get(exemption).contains(*remoteCIDR)) { return true; } // Otherwise the exemption is a UNIX path and we should check the local path // (the remoteAddr == "anonymous unix socket") against the exemption string // // On Windows we don't check this at all and only CIDR ranges are supported #ifndef _WIN32 } else if ((stdx::holds_alternative(exemption)) && localAddr.isValid() && (localAddr.getAddr() == stdx::get(exemption))) { return true; #endif } } return false; } ServiceEntryPointImpl::ServiceEntryPointImpl(ServiceContext* svcCtx) : _svcCtx(svcCtx) { const auto supportedMax = [] { #ifdef _WIN32 return serverGlobalParams.maxConns; #else struct rlimit limit; verify(getrlimit(RLIMIT_NOFILE, &limit) == 0); size_t max = (size_t)(limit.rlim_cur * .8); LOGV2_DEBUG(22940, 1, "fd limit hard:{hard} soft:{soft} max conn: {conn}", "file descriptor and connection resource limits", "hard"_attr = limit.rlim_max, "soft"_attr = limit.rlim_cur, "conn"_attr = max); return std::min(max, serverGlobalParams.maxConns); #endif }(); // If we asked for more connections than supported, inform the user. if (supportedMax < serverGlobalParams.maxConns && serverGlobalParams.maxConns != DEFAULT_MAX_CONN) { LOGV2(22941, " --maxConns too high, can only handle {limit}", " --maxConns too high", "limit"_attr = supportedMax); } _maxNumConnections = supportedMax; if (serverGlobalParams.reservedAdminThreads) { _adminInternalPool = std::make_unique( _svcCtx, "admin/internal connections", serverGlobalParams.reservedAdminThreads); } } Status ServiceEntryPointImpl::start() { if (_adminInternalPool) return _adminInternalPool->start(); else return Status::OK(); } void ServiceEntryPointImpl::startSession(transport::SessionHandle session) { // Setup the restriction environment on the Session, if the Session has local/remote Sockaddrs const auto& remoteAddr = session->remoteAddr(); const auto& localAddr = session->localAddr(); invariant(remoteAddr.isValid() && localAddr.isValid()); auto restrictionEnvironment = std::make_unique(remoteAddr, localAddr); RestrictionEnvironment::set(session, std::move(restrictionEnvironment)); SSMListIterator ssmIt; const bool quiet = serverGlobalParams.quiet.load(); size_t connectionCount; auto transportMode = _svcCtx->getServiceExecutor()->transportMode(); auto ssm = ServiceStateMachine::create(_svcCtx, session, transportMode); auto usingMaxConnOverride = false; { stdx::lock_guard lk(_sessionsMutex); connectionCount = _sessions.size() + 1; if (connectionCount > _maxNumConnections) { usingMaxConnOverride = shouldOverrideMaxConns(session, serverGlobalParams.maxConnsOverride); } if (connectionCount <= _maxNumConnections || usingMaxConnOverride) { ssmIt = _sessions.emplace(_sessions.begin(), ssm); _currentConnections.store(connectionCount); _createdConnections.addAndFetch(1); } } // Checking if we successfully added a connection above. Separated from the lock so we don't log // while holding it. if (connectionCount > _maxNumConnections && !usingMaxConnOverride) { if (!quiet) { LOGV2(22942, "Connection refused because there are too many open connections", "connectionCount"_attr = connectionCount); } return; } else if (usingMaxConnOverride && _adminInternalPool) { ssm->setServiceExecutor(_adminInternalPool.get()); } if (!quiet) { LOGV2(22943, "Connection accepted", "remote"_attr = session->remote(), "connectionId"_attr = session->id(), "connectionCount"_attr = connectionCount); } ssm->setCleanupHook([this, ssmIt, quiet, session = std::move(session)] { size_t connectionCount; auto remote = session->remote(); { stdx::lock_guard lk(_sessionsMutex); _sessions.erase(ssmIt); connectionCount = _sessions.size(); _currentConnections.store(connectionCount); } _shutdownCondition.notify_one(); if (!quiet) { LOGV2(22944, "Connection ended", "remote"_attr = remote, "connectionId"_attr = session->id(), "connectionCount"_attr = connectionCount); } }); auto ownership = ServiceStateMachine::Ownership::kOwned; if (transportMode == transport::Mode::kSynchronous) { ownership = ServiceStateMachine::Ownership::kStatic; } ssm->start(ownership); } void ServiceEntryPointImpl::endAllSessions(transport::Session::TagMask tags) { // While holding the _sesionsMutex, loop over all the current connections, and if their tags // do not match the requested tags to skip, terminate the session. { stdx::unique_lock lk(_sessionsMutex); for (auto& ssm : _sessions) { ssm->terminateIfTagsDontMatch(tags); } } } bool ServiceEntryPointImpl::shutdown(Milliseconds timeout) { using logv2::LogComponent; stdx::unique_lock lk(_sessionsMutex); // Request that all sessions end, while holding the _sesionsMutex, loop over all the current // connections and terminate them for (auto& ssm : _sessions) { ssm->terminate(); } // Close all sockets and then wait for the number of active connections to reach zero with a // condition_variable that notifies in the session cleanup hook. If we haven't closed drained // all active operations within the deadline, just keep going with shutdown: the OS will do it // for us when the process terminates. auto timeSpent = Milliseconds(0); const auto checkInterval = std::min(Milliseconds(250), timeout); auto noWorkersLeft = [this] { return numOpenSessions() == 0; }; while (timeSpent < timeout && !_shutdownCondition.wait_for(lk, checkInterval.toSystemDuration(), noWorkersLeft)) { LOGV2(22945, "shutdown: still waiting on {workers} active workers to drain... ", "shutdown: still waiting on active workers to drain... ", "workers"_attr = numOpenSessions()); timeSpent += checkInterval; } bool result = noWorkersLeft(); if (result) { LOGV2(22946, "shutdown: no running workers found..."); } else { LOGV2( 22947, "shutdown: exhausted grace period for {workers} active workers to " "drain; continuing with shutdown...", "shutdown: exhausted grace period active workers to drain; continuing with shutdown...", "workers"_attr = numOpenSessions()); } return result; } void ServiceEntryPointImpl::appendStats(BSONObjBuilder* bob) const { size_t sessionCount = _currentConnections.load(); bob->append("current", static_cast(sessionCount)); bob->append("available", static_cast(_maxNumConnections - sessionCount)); bob->append("totalCreated", static_cast(_createdConnections.load())); if (auto sc = getGlobalServiceContext()) { bob->append("active", static_cast(sc->getActiveClientOperations())); bob->append("exhaustIsMaster", static_cast(HelloMetrics::get(sc)->getNumExhaustIsMaster())); bob->append("exhaustHello", static_cast(HelloMetrics::get(sc)->getNumExhaustHello())); bob->append("awaitingTopologyChanges", static_cast(HelloMetrics::get(sc)->getNumAwaitingTopologyChanges())); } if (_adminInternalPool) { BSONObjBuilder section(bob->subobjStart("adminConnections")); _adminInternalPool->appendStats(§ion); } } } // namespace mongo