Refine and commit all temporary metrics from DanShaders:ton:more-metrics

adnl/CMakeLists.txt

@@ -81,7 +81,7 @@ if (NOT TON_ONLY_TONLIB)
 add_library(adnl STATIC ${ADNL_SOURCE})
 
 target_include_directories(adnl PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/..>)
-target_link_libraries(adnl PUBLIC tdactor ton_crypto tl_api tdnet tddb keys keyring )
+target_link_libraries(adnl PUBLIC tdactor ton_crypto tl_api tdnet tddb keys keyring metrics)
 
 add_executable(adnl-proxy ${ADNL_PROXY_SOURCE})
 target_include_directories(adnl-proxy PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/..>)

adnl/adnl-network-manager.cpp

@@ -30,6 +30,26 @@ td::actor::ActorOwn<AdnlNetworkManager> AdnlNetworkManager::create(td::uint16 po
   return td::actor::create_actor<AdnlNetworkManagerImpl>("NetworkManager", port);
 }
 
+void AdnlNetworkManagerImpl::start_up() {
+  alarm_timestamp() = td::Timestamp::in(60.0);
+  add_collector(metrics_collector_.get());
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, udp_ingress_bytes_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, udp_ingress_packets_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, udp_ingress_drops_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector,
+                          udp_ingress_proxy_control_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector,
+                          udp_proxy_ingress_bytes_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, udp_egress_bytes_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, udp_egress_packets_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, udp_egress_drops_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector,
+                          udp_proxy_egress_bytes_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector,
+                          udp_proxy_egress_packets_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, listening_sockets_);
+}
+
 AdnlNetworkManagerImpl::OutDesc *AdnlNetworkManagerImpl::choose_out_iface(td::uint8 cat, td::uint32 priority) {
   auto it = out_desc_.upper_bound(priority);
   while (true) {
@@ -108,19 +128,25 @@ void AdnlNetworkManagerImpl::add_proxy_addr(td::IPAddress addr, td::uint16 local
 }
 
 void AdnlNetworkManagerImpl::receive_udp_message(td::UdpMessage message, size_t idx) {
+  udp_ingress_packets_->add(1);
+  udp_ingress_bytes_->add(message.data.size());
   if (!callback_) {
+    udp_ingress_drop_no_callback_->add(1);
     LOG(ERROR) << this << ": dropping IN message [?->?]: peer table unitialized";
     return;
   }
   if (message.error.is_error()) {
+    udp_ingress_drop_error_->add(1);
     VLOG(ADNL_WARNING) << this << ": dropping ERROR message: " << message.error;
     return;
   }
   if (message.data.size() < 32) {
+    udp_ingress_drop_too_short_->add(1);
     VLOG(ADNL_WARNING) << this << ": received too small proxy packet of size " << message.data.size();
     return;
   }
   if (message.data.size() >= get_mtu() + 128) {
+    udp_ingress_drop_too_huge_->add(1);
     VLOG(ADNL_NOTICE) << this << ": received huge packet of size " << message.data.size();
   }
   CHECK(idx < udp_sockets_.size());
@@ -138,38 +164,46 @@ void AdnlNetworkManagerImpl::receive_udp_message(td::UdpMessage message, size_t
       CHECK(proxy_iface.is_proxy());
       auto R = in_desc_[it->second].proxy->decrypt(std::move(message.data));
       if (R.is_error()) {
+        udp_ingress_drop_bad_proxy_decrypt_->add(1);
         VLOG(ADNL_WARNING) << this << ": failed to decrypt proxy mesage: " << R.move_as_error();
         return;
       }
       auto packet = R.move_as_ok();
+      udp_proxy_ingress_bytes_->add(packet.data.size());
       if (packet.flags & 1) {
         message.address.init_host_port(td::IPAddress::ipv4_to_str(packet.ip), packet.port).ensure();
       } else {
         message.address = td::IPAddress{};
       }
       if ((packet.flags & 6) == 6) {
         if (packet.seqno <= 0 || packet.adnl_start_time < 0) {
+          udp_ingress_drop_bad_proxy_seqno_->add(1);
           VLOG(ADNL_WARNING) << this << ": dropping proxy packet: invalid start_time/seqno";
           return;
         }
         if (proxy_iface.received.packet_is_delivered(packet.adnl_start_time, packet.seqno)) {
+          udp_ingress_drop_bad_proxy_seqno_->add(1);
           VLOG(ADNL_WARNING) << this << ": dropping duplicate proxy packet";
           return;
         }
       }
       if (packet.flags & 8) {
         if (packet.date < td::Clocks::system() - 60 || packet.date > td::Clocks::system() + 60) {
+          udp_ingress_drop_bad_proxy_time_->add(1);
           VLOG(ADNL_WARNING) << this << ": dropping proxy packet: bad time " << packet.date;
           return;
         }
       }
       if (!(packet.flags & (1 << 16))) {
+        udp_ingress_drop_bad_proxy_outflag_->add(1);
         VLOG(ADNL_WARNING) << this << ": dropping proxy packet: packet has outbound flag";
         return;
       }
       if (packet.flags & (1 << 17)) {
+        udp_ingress_proxy_control_->add(1);
         auto F = fetch_tl_object<ton_api::adnl_ProxyControlPacket>(std::move(packet.data), true);
         if (F.is_error()) {
+          udp_ingress_drop_bad_control_->add(1);
           VLOG(ADNL_WARNING) << this << ": dropping proxy packet: bad control packet";
           return;
         }
@@ -205,6 +239,7 @@ void AdnlNetworkManagerImpl::receive_udp_message(td::UdpMessage message, size_t
   }
   if (!from_proxy) {
     if (socket.in_desc == std::numeric_limits<size_t>::max()) {
+      udp_ingress_drop_no_in_desc_->add(1);
       VLOG(ADNL_WARNING) << this << ": received bad packet to proxy-only listenung port";
       return;
     }
@@ -226,12 +261,14 @@ void AdnlNetworkManagerImpl::send_udp_packet(AdnlNodeIdShort src_id, AdnlNodeIdS
                                              td::uint32 priority, td::BufferSlice data) {
   auto it = adnl_id_2_cat_.find(src_id);
   if (it == adnl_id_2_cat_.end()) {
+    udp_egress_drop_unknown_src_->add(1);
     VLOG(ADNL_WARNING) << this << ": dropping OUT message [" << src_id << "->" << dst_id << "]: unknown src";
     return;
   }
 
   auto out = choose_out_iface(it->second, priority);
   if (!out) {
+    udp_egress_drop_no_route_->add(1);
     VLOG(ADNL_WARNING) << this << ": dropping OUT message [" << src_id << "->" << dst_id << "]: no out rules";
     return;
   }
@@ -246,12 +283,15 @@ void AdnlNetworkManagerImpl::send_udp_packet(AdnlNodeIdShort src_id, AdnlNodeIdS
 
     CHECK(M.data.size() <= get_mtu());
 
+    udp_egress_packets_->add(1);
+    udp_egress_bytes_->add(M.data.size());
     td::actor::send_closure(socket.server, &td::UdpServer::send, std::move(M));
   } else {
     AdnlProxy::Packet p;
     p.flags = 7;
     p.ip = dst_addr.get_ipv4();
     p.port = static_cast<td::uint16>(dst_addr.get_port());
+    auto inner_size = data.size();
     p.data = std::move(data);
     p.adnl_start_time = Adnl::adnl_start_time();
     p.seqno = ++v.out_seqno;
@@ -262,6 +302,10 @@ void AdnlNetworkManagerImpl::send_udp_packet(AdnlNodeIdShort src_id, AdnlNodeIdS
     M.address = v.proxy_addr;
     M.data = std::move(enc);
 
+    udp_egress_packets_->add(1);
+    udp_egress_bytes_->add(M.data.size());
+    udp_proxy_egress_packets_->add(1);
+    udp_proxy_egress_bytes_->add(inner_size);
     td::actor::send_closure(socket.server, &td::UdpServer::send, std::move(M));
   }
 }
@@ -282,6 +326,8 @@ void AdnlNetworkManagerImpl::proxy_register(OutDesc &desc) {
   M.address = desc.proxy_addr;
   M.data = std::move(enc);
 
+  udp_egress_packets_->add(1);
+  udp_egress_bytes_->add(M.data.size());
   auto &socket = udp_sockets_[desc.socket_idx];
   td::actor::send_closure(socket.server, &td::UdpServer::send, std::move(M));
 }

adnl/adnl-network-manager.h

@@ -27,6 +27,8 @@
 #include "adnl-node-id.hpp"
 #include "adnl-proxy-types.h"
 
+#include "metrics/metrics-collectors.h"
+
 namespace td {
 class UdpServer;
 }
@@ -54,7 +56,7 @@ class AdnlNetworkConnection : public td::actor::Actor {
   virtual ~AdnlNetworkConnection() = default;
 };
 
-class AdnlNetworkManager : public td::actor::Actor {
+class AdnlNetworkManager : public td::actor::Actor, public virtual metrics::CollectorWrapper {
  public:
   //using ConnHandle = td::uint64;
   class Callback {

adnl/adnl-network-manager.hpp

@@ -19,7 +19,9 @@
 #pragma once
 
 #include <map>
+#include <memory>
 
+#include "metrics/metrics-collectors.h"
 #include "td/actor/PromiseFuture.h"
 #include "td/actor/actor.h"
 #include "td/net/TcpListener.h"
@@ -106,9 +108,7 @@ class AdnlNetworkManagerImpl : public AdnlNetworkManager {
   }
 
   void alarm() override;
-  void start_up() override {
-    alarm_timestamp() = td::Timestamp::in(60.0);
-  }
+  void start_up() override;
 
   void add_in_addr(InDesc desc, size_t socket_idx) {
     for (size_t idx = 0; idx < in_desc_.size(); idx++) {
@@ -155,6 +155,51 @@ class AdnlNetworkManagerImpl : public AdnlNetworkManager {
   td::uint64 received_messages_ = 0;
   td::uint64 sent_messages_ = 0;
 
+  using CounterPtr = std::shared_ptr<metrics::AtomicCounter<td::uint64>>;
+  metrics::MultiCollector::Own metrics_collector_ = metrics::MultiCollector::create("adnl_net");
+  metrics::Labeled<std::string, metrics::AtomicCounter<td::uint64>>::Ptr udp_ingress_drops_ =
+      metrics::Labeled<std::string, metrics::AtomicCounter<td::uint64>>::make(
+          "reason", "udp_ingress_drops_total", "UDP ingress packets dropped, by reason.");
+  metrics::Labeled<std::string, metrics::AtomicCounter<td::uint64>>::Ptr udp_egress_drops_ =
+      metrics::Labeled<std::string, metrics::AtomicCounter<td::uint64>>::make(
+          "reason", "udp_egress_drops_total", "ADNL outbound packets dropped, by reason.");
+  metrics::AtomicCounter<td::uint64>::Ptr udp_ingress_bytes_ = metrics::AtomicCounter<td::uint64>::make(
+      "udp_ingress_bytes_total", "Total UDP bytes received on ADNL sockets (wire-level).");
+  metrics::AtomicCounter<td::uint64>::Ptr udp_ingress_packets_ = metrics::AtomicCounter<td::uint64>::make(
+      "udp_ingress_packets_total", "Total UDP packets received on ADNL sockets.");
+  metrics::AtomicCounter<td::uint64>::Ptr udp_ingress_proxy_control_ =
+      metrics::AtomicCounter<td::uint64>::make("udp_ingress_proxy_control_total", "ADNL proxy control packets received.");
+  metrics::AtomicCounter<td::uint64>::Ptr udp_proxy_ingress_bytes_ = metrics::AtomicCounter<td::uint64>::make(
+      "udp_proxy_ingress_bytes_total", "Total payload bytes received via ADNL proxy (after decrypt).");
+  metrics::AtomicCounter<td::uint64>::Ptr udp_egress_bytes_ = metrics::AtomicCounter<td::uint64>::make(
+      "udp_egress_bytes_total", "Total UDP bytes sent on ADNL sockets (wire-level).");
+  metrics::AtomicCounter<td::uint64>::Ptr udp_egress_packets_ = metrics::AtomicCounter<td::uint64>::make(
+      "udp_egress_packets_total", "Total UDP packets sent on ADNL sockets.");
+  metrics::AtomicCounter<td::uint64>::Ptr udp_proxy_egress_bytes_ = metrics::AtomicCounter<td::uint64>::make(
+      "udp_proxy_egress_bytes_total", "Total inner payload bytes wrapped through ADNL proxy.");
+  metrics::AtomicCounter<td::uint64>::Ptr udp_proxy_egress_packets_ = metrics::AtomicCounter<td::uint64>::make(
+      "udp_proxy_egress_packets_total", "Total packets wrapped through ADNL proxy.");
+  metrics::LambdaGauge::Ptr listening_sockets_ = metrics::LambdaGauge::make(
+      "listening_sockets",
+      [this] {
+        return std::vector<metrics::Sample>{
+            metrics::Sample{.label_set = {}, .value = static_cast<double>(udp_sockets_.size())}};
+      },
+      "Number of UDP sockets owned by the ADNL network manager.");
+
+  CounterPtr udp_ingress_drop_no_callback_ = udp_ingress_drops_->label("no_callback");
+  CounterPtr udp_ingress_drop_error_ = udp_ingress_drops_->label("error");
+  CounterPtr udp_ingress_drop_too_short_ = udp_ingress_drops_->label("too_short");
+  CounterPtr udp_ingress_drop_too_huge_ = udp_ingress_drops_->label("too_huge");
+  CounterPtr udp_ingress_drop_bad_proxy_decrypt_ = udp_ingress_drops_->label("bad_proxy_decrypt");
+  CounterPtr udp_ingress_drop_bad_proxy_seqno_ = udp_ingress_drops_->label("bad_proxy_seqno");
+  CounterPtr udp_ingress_drop_bad_proxy_time_ = udp_ingress_drops_->label("bad_proxy_time");
+  CounterPtr udp_ingress_drop_bad_proxy_outflag_ = udp_ingress_drops_->label("bad_proxy_outflag");
+  CounterPtr udp_ingress_drop_bad_control_ = udp_ingress_drops_->label("bad_control");
+  CounterPtr udp_ingress_drop_no_in_desc_ = udp_ingress_drops_->label("no_in_desc");
+  CounterPtr udp_egress_drop_unknown_src_ = udp_egress_drops_->label("unknown_src");
+  CounterPtr udp_egress_drop_no_route_ = udp_egress_drops_->label("no_route");
+
   std::vector<UdpSocketDesc> udp_sockets_;
   std::map<td::uint16, size_t> port_2_socket_;
 

adnl/adnl-peer-table.cpp

@@ -16,7 +16,6 @@
 
     Copyright 2017-2020 Telegram Systems LLP
 */
-#include "absl/functional/internal/any_invocable.h"
 #include "td/db/RocksDb.h"
 #include "td/utils/Random.h"
 #include "td/utils/crypto.h"
@@ -50,7 +49,9 @@ td::actor::ActorOwn<Adnl> Adnl::create(std::string db, td::actor::ActorId<keyrin
 }
 
 void AdnlPeerTableImpl::receive_packet(td::IPAddress addr, AdnlCategoryMask cat_mask, td::BufferSlice data) {
+  inbound_packets_->add(1);
   if (data.size() < 32) {
+    inbound_drop_too_short_->add(1);
     VLOG(ADNL_WARNING) << this << ": dropping IN message [?->?]: message too short: len=" << data.size();
     return;
   }
@@ -61,6 +62,7 @@ void AdnlPeerTableImpl::receive_packet(td::IPAddress addr, AdnlCategoryMask cat_
   auto it = local_ids_.find(dst);
   if (it != local_ids_.end()) {
     if (!cat_mask.test(it->second.cat)) {
+      inbound_drop_cat_mismatch_->add(1);
       VLOG(ADNL_WARNING) << this << ": dropping IN message [?->" << dst << "]: category mismatch";
       return;
     }
@@ -72,13 +74,15 @@ void AdnlPeerTableImpl::receive_packet(td::IPAddress addr, AdnlCategoryMask cat_
   auto it2 = channels_.find(dst_chan_id);
   if (it2 != channels_.end()) {
     if (!cat_mask.test(it2->second.second)) {
+      inbound_drop_cat_mismatch_->add(1);
       VLOG(ADNL_WARNING) << this << ": dropping IN message to channel [?->" << dst << "]: category mismatch";
       return;
     }
     td::actor::send_closure(it2->second.first, &AdnlChannel::receive, addr, std::move(data));
     return;
   }
 
+  inbound_drop_unknown_dst_->add(1);
   VLOG(ADNL_DEBUG) << this << ": dropping IN message [?->" << dst << "]: unknown dst " << dst
                    << " (len=" << (data.size() + 32) << ")";
 }
@@ -121,6 +125,8 @@ AdnlPeerTableImpl::PeerPair *AdnlPeerTableImpl::get_peer_pair_if_exists(AdnlNode
 }
 
 void AdnlPeerTableImpl::receive_decrypted_packet(AdnlNodeIdShort dst, AdnlPacket packet, td::uint64 serialized_size) {
+  decrypt_packets_->add(1);
+  decrypt_bytes_->add(serialized_size);
   packet.run_basic_checks().ensure();
 
   if (!packet.inited_from_short()) {
@@ -187,6 +193,7 @@ void AdnlPeerTableImpl::send_message_in(AdnlNodeIdShort src, AdnlNodeIdShort dst
                                         td::uint32 flags) {
   auto it2 = local_ids_.find(src);
   if (it2 == local_ids_.end()) {
+    app_send_drop_unknown_src_->add(1);
     LOG(ERROR) << this << ": dropping OUT message [" << src << "->" << dst << "]: unknown src";
     return;
   }
@@ -201,26 +208,31 @@ void AdnlPeerTableImpl::send_message_in(AdnlNodeIdShort src, AdnlNodeIdShort dst
 void AdnlPeerTableImpl::answer_query(AdnlNodeIdShort src, AdnlNodeIdShort dst, AdnlQueryId query_id,
                                      td::BufferSlice data) {
   if (data.size() > get_mtu()) {
+    app_send_drop_too_big_->add(1);
     LOG(ERROR) << this << ": dropping OUT message [" << src << "->" << dst
                << "]: message too big: size=" << data.size();
     return;
   }
+  app_metrics_->record_send("answer", data.as_slice());
   send_message_in(src, dst, adnlmessage::AdnlMessageAnswer{query_id, std::move(data)}, 0);
 }
 
 void AdnlPeerTableImpl::send_query(AdnlNodeIdShort src, AdnlNodeIdShort dst, std::string name,
                                    td::Promise<td::BufferSlice> promise, td::Timestamp timeout, td::BufferSlice data) {
   if (data.size() > huge_packet_max_size()) {
+    app_send_drop_too_big_->add(1);
     VLOG(ADNL_WARNING) << "dropping too big packet [" << src << "->" << dst << "]: size=" << data.size();
     VLOG(ADNL_WARNING) << "DUMP: " << td::buffer_to_hex(data.as_slice().truncate(128));
     return;
   }
 
   auto it2 = local_ids_.find(src);
   if (it2 == local_ids_.end()) {
+    app_send_drop_unknown_src_->add(1);
     LOG(ERROR) << this << ": dropping OUT message [" << src << "->" << dst << "]: unknown src";
     return;
   }
+  app_metrics_->record_send("query", data.as_slice());
   auto &peer_info = peers_[dst];
 
   td::actor::send_closure(get_peer_pair(dst, peer_info, src, it2->second), &AdnlPeerPair::send_query, name,
@@ -354,6 +366,18 @@ void AdnlPeerTableImpl::unregister_channel(AdnlChannelIdShort id) {
 }
 
 void AdnlPeerTableImpl::start_up() {
+  add_collector(metrics_collector_.get());
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, app_metrics_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, app_send_dropped_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, inbound_packets_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, inbound_dropped_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, decrypt_packets_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, decrypt_bytes_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, local_ids_gauge_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, peers_gauge_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, peer_pairs_gauge_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, channels_gauge_);
+  td::actor::send_closure(metrics_collector_.get(), &metrics::MultiCollector::add_sync_collector, static_nodes_gauge_);
 }
 
 void AdnlPeerTableImpl::write_new_addr_list_to_db(AdnlNodeIdShort local_id, AdnlNodeIdShort peer_id, AdnlDbItem node,
@@ -384,13 +408,15 @@ AdnlPeerTableImpl::AdnlPeerTableImpl(std::string db_root, td::actor::ActorId<key
 void AdnlPeerTableImpl::deliver(AdnlNodeIdShort src, AdnlNodeIdShort dst, td::BufferSlice data) {
   auto it = local_ids_.find(dst);
   if (it != local_ids_.end()) {
+    app_metrics_->record_deliver("message", data.as_slice());
     td::actor::send_closure(it->second.local_id, &AdnlLocalId::deliver, src, std::move(data));
   }
 }
 void AdnlPeerTableImpl::deliver_query(AdnlNodeIdShort src, AdnlNodeIdShort dst, td::BufferSlice data,
                                       td::Promise<td::BufferSlice> promise) {
   auto it = local_ids_.find(dst);
   if (it != local_ids_.end()) {
+    app_metrics_->record_deliver("query", data.as_slice());
     td::actor::send_closure(it->second.local_id, &AdnlLocalId::deliver_query, src, std::move(data), std::move(promise));
   } else {
     LOG(WARNING) << "deliver query: unknown dst " << dst;