SERVER-25466 Add MessageCompressor stats to server status

8 files changed, 115 insertions, 17 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 556f8e43878..6d0e7666745 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -72,6 +72,7 @@ env.Library(
         '$BUILD_DIR/mongo/s/coreshard',
         '$BUILD_DIR/mongo/s/write_ops/batch_write_types',
         '$BUILD_DIR/mongo/scripting/scripting_common',
+        '$BUILD_DIR/mongo/transport/message_compressor',
         '$BUILD_DIR/mongo/transport/transport_layer_common',
         '$BUILD_DIR/mongo/util/cmdline_utils/cmdline_utils',
         '$BUILD_DIR/mongo/util/foundation',
diff --git a/src/mongo/db/commands/server_status.cpp b/src/mongo/db/commands/server_status.cpp
index 1872745bf24..531d477db5f 100644
--- a/src/mongo/db/commands/server_status.cpp
+++ b/src/mongo/db/commands/server_status.cpp
@@ -47,6 +47,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/platform/process_id.h"
+#include "mongo/transport/message_compressor_registry.h"
 #include "mongo/transport/transport_layer.h"
 #include "mongo/util/log.h"
 #include "mongo/util/net/hostname_canonicalization_worker.h"
@@ -287,6 +288,7 @@ public:
     BSONObj generateSection(OperationContext* txn, const BSONElement& configElement) const {
         BSONObjBuilder b;
         networkCounter.append(b);
+        appendMessageCompressionStats(&b);
         return b.obj();
     }
 
diff --git a/src/mongo/db/stats/counters.cpp b/src/mongo/db/stats/counters.cpp
index 61b5be912dd..9f61240dc69 100644
--- a/src/mongo/db/stats/counters.cpp
+++ b/src/mongo/db/stats/counters.cpp
@@ -135,26 +135,46 @@ BSONObj OpCounters::getObj() const {
     return b.obj();
 }
 
-void NetworkCounter::hit(long long bytesIn, long long bytesOut) {
+void NetworkCounter::hitPhysical(long long bytesIn, long long bytesOut) {
     const int64_t MAX = 1ULL << 60;
 
     // don't care about the race as its just a counter
-    bool overflow = _bytesIn.loadRelaxed() > MAX || _bytesOut.loadRelaxed() > MAX;
+    bool overflow = _physicalBytesIn.loadRelaxed() > MAX || _physicalBytesOut.loadRelaxed() > MAX;
 
     if (overflow) {
-        _bytesIn.store(bytesIn);
-        _bytesOut.store(bytesOut);
+        _physicalBytesIn.store(bytesIn);
+        _physicalBytesOut.store(bytesOut);
+    } else {
+        _physicalBytesIn.fetchAndAdd(bytesIn);
+        _physicalBytesOut.fetchAndAdd(bytesOut);
+    }
+}
+
+void NetworkCounter::hitLogical(long long bytesIn, long long bytesOut) {
+    const int64_t MAX = 1ULL << 60;
+
+    // don't care about the race as its just a counter
+    bool overflow = _logicalBytesIn.loadRelaxed() > MAX || _logicalBytesOut.loadRelaxed() > MAX;
+
+    if (overflow) {
+        _logicalBytesIn.store(bytesIn);
+        _logicalBytesOut.store(bytesOut);
+        // The requests field only gets incremented here (and not in hitPhysical) because the
+        // hitLogical and hitPhysical are each called for each operation. Incrementing it in both
+        // functions would double-count the number of operations.
         _requests.store(1);
     } else {
-        _bytesIn.fetchAndAdd(bytesIn);
-        _bytesOut.fetchAndAdd(bytesOut);
+        _logicalBytesIn.fetchAndAdd(bytesIn);
+        _logicalBytesOut.fetchAndAdd(bytesOut);
         _requests.fetchAndAdd(1);
     }
 }
 
 void NetworkCounter::append(BSONObjBuilder& b) {
-    b.append("bytesIn", static_cast<long long>(_bytesIn.loadRelaxed()));
-    b.append("bytesOut", static_cast<long long>(_bytesOut.loadRelaxed()));
+    b.append("bytesIn", static_cast<long long>(_logicalBytesIn.loadRelaxed()));
+    b.append("bytesOut", static_cast<long long>(_logicalBytesOut.loadRelaxed()));
+    b.append("physicalBytesIn", static_cast<long long>(_physicalBytesIn.loadRelaxed()));
+    b.append("physicalBytesOut", static_cast<long long>(_physicalBytesOut.loadRelaxed()));
     b.append("numRequests", static_cast<long long>(_requests.loadRelaxed()));
 }
 
diff --git a/src/mongo/db/stats/counters.h b/src/mongo/db/stats/counters.h
index fe86aa16d67..435542beb37 100644
--- a/src/mongo/db/stats/counters.h
+++ b/src/mongo/db/stats/counters.h
@@ -95,14 +95,20 @@ extern OpCounters replOpCounters;
 
 class NetworkCounter {
 public:
-    NetworkCounter() : _bytesIn(0), _bytesOut(0), _requests(0) {}
-    void hit(long long bytesIn, long long bytesOut);
+    // Increment the counters for the number of bytes read directly off the wire
+    void hitPhysical(long long bytesIn, long long bytesOut);
+    // Increment the counters for the number of bytes passed out of the TransportLayer to the
+    // server
+    void hitLogical(long long bytesIn, long long bytesOut);
     void append(BSONObjBuilder& b);
 
 private:
-    AtomicInt64 _bytesIn;
-    AtomicInt64 _bytesOut;
-    AtomicInt64 _requests;
+    AtomicInt64 _physicalBytesIn{0};
+    AtomicInt64 _physicalBytesOut{0};
+    AtomicInt64 _logicalBytesIn{0};
+    AtomicInt64 _logicalBytesOut{0};
+
+    AtomicInt64 _requests{0};
 };
 
 extern NetworkCounter networkCounter;
diff --git a/src/mongo/transport/SConscript b/src/mongo/transport/SConscript
index 83fb5729e01..1291d6d2a54 100644
--- a/src/mongo/transport/SConscript
+++ b/src/mongo/transport/SConscript
@@ -106,6 +106,7 @@ env.Library(
     target='message_compressor',
     source=[
         'message_compressor_manager.cpp',
+        'message_compressor_metrics.cpp',
         'message_compressor_registry.cpp',
         'message_compressor_snappy.cpp',
     ],
@@ -126,3 +127,4 @@ env.CppUnitTest(
         'message_compressor',
     ]
 )
+
diff --git a/src/mongo/transport/message_compressor_metrics.cpp b/src/mongo/transport/message_compressor_metrics.cpp
new file mode 100644
index 00000000000..462ba66a4b6
--- /dev/null
+++ b/src/mongo/transport/message_compressor_metrics.cpp
@@ -0,0 +1,66 @@
+/**
+ *    Copyright (C) 2016 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 <http://www.gnu.org/licenses/>.
+ *
+ *    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.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/transport/message_compressor_registry.h"
+
+namespace mongo {
+namespace {
+const auto kBytesIn = "bytesIn"_sd;
+const auto kBytesOut = "bytesOut"_sd;
+}  // namespace
+
+void appendMessageCompressionStats(BSONObjBuilder* b) {
+    auto& registry = MessageCompressorRegistry::get();
+    const auto& names = registry.getCompressorNames();
+    if (names.empty()) {
+        return;
+    }
+    BSONObjBuilder compressionSection(b->subobjStart("compression"));
+
+    for (auto&& name : names) {
+        auto&& compressor = registry.getCompressor(name);
+        BSONObjBuilder base(compressionSection.subobjStart(name));
+
+        BSONObjBuilder compressed(base.subobjStart("compressed"));
+        compressed << kBytesIn << compressor->getCompressedBytesIn() << kBytesOut
+                   << compressor->getCompressedBytesOut();
+        compressed.doneFast();
+
+        BSONObjBuilder decompressed(base.subobjStart("decompressed"));
+        decompressed << kBytesIn << compressor->getDecompressedBytesIn() << kBytesOut
+                     << compressor->getDecompressedBytesOut();
+        decompressed.doneFast();
+        base.doneFast();
+    }
+    compressionSection.doneFast();
+}
+
+}  // namespace mongo
diff --git a/src/mongo/transport/message_compressor_registry.h b/src/mongo/transport/message_compressor_registry.h
index 9d8549ed0e3..e6fa63148c8 100644
--- a/src/mongo/transport/message_compressor_registry.h
+++ b/src/mongo/transport/message_compressor_registry.h
@@ -119,4 +119,5 @@ private:
 
 Status addMessageCompressionOptions(moe::OptionSection* options, bool forShell);
 Status storeMessageCompressionOptions(const moe::Environment& params);
+void appendMessageCompressionStats(BSONObjBuilder* b);
 }  // namespace mongo
diff --git a/src/mongo/transport/transport_layer_legacy.cpp b/src/mongo/transport/transport_layer_legacy.cpp
index 3c004eb793f..5fc0657da35 100644
--- a/src/mongo/transport/transport_layer_legacy.cpp
+++ b/src/mongo/transport/transport_layer_legacy.cpp
@@ -105,12 +105,14 @@ Ticket TransportLayerLegacy::sourceMessage(Session& session, Message* message, D
             return {ErrorCodes::HostUnreachable, "Recv failed"};
         }
 
+        networkCounter.hitPhysical(message->size(), 0);
         if (message->operation() == dbCompressed) {
             auto swm = compressorMgr.decompressMessage(*message);
             if (!swm.isOK())
                 return swm.getStatus();
             *message = swm.getValue();
         }
+        networkCounter.hitLogical(message->size(), 0);
         return Status::OK();
     };
 
@@ -149,11 +151,13 @@ Ticket TransportLayerLegacy::sinkMessage(Session& session,
     auto& compressorMgr = session.getCompressorManager();
     auto sinkCb = [&message, &compressorMgr](AbstractMessagingPort* amp) -> Status {
         try {
+            networkCounter.hitLogical(0, message.size());
             auto swm = compressorMgr.compressMessage(message);
             if (!swm.isOK())
                 return swm.getStatus();
             const auto& compressedMessage = swm.getValue();
             amp->say(compressedMessage);
+            networkCounter.hitPhysical(0, compressedMessage.size());
 
             return Status::OK();
         } catch (const SocketException& e) {
@@ -259,13 +263,9 @@ Status TransportLayerLegacy::_runTicket(Ticket ticket) {
         amp = conn->second.amp.get();
     }
 
-    amp->clearCounters();
-
     auto legacyTicket = checked_cast<LegacyTicket*>(getTicketImpl(ticket));
     auto res = legacyTicket->_fill(amp);
 
-    networkCounter.hit(amp->getBytesIn(), amp->getBytesOut());
-
     {
         stdx::lock_guard<stdx::mutex> lk(_connectionsMutex);