Adding stats collection

This changelist adds methods to record essential stats for each gRPC connection and expose them.

PiperOrigin-RevId: 615700708

Author: GPINs Team

server/gnpsi_service_impl.cc

@@ -2,18 +2,60 @@
 
 #include <memory>
 #include <string>
+#include <vector>
 
 #include "absl/base/thread_annotations.h"
 #include "glog/logging.h"
 #include "absl/status/status.h"
+#include "absl/strings/match.h"
+#include "absl/strings/numbers.h"
 #include "absl/strings/str_cat.h"
+#include "absl/strings/string_view.h"
+#include "absl/strings/strip.h"
 #include "absl/synchronization/mutex.h"
 #include "absl/time/clock.h"
 #include "absl/time/time.h"
 #include "proto/gnpsi/gnpsi.pb.h"
 
 namespace gnpsi {
 
+absl::Status GnpsiConnection::InitializeStats() {
+  std::string uri = this->GetPeerName();
+  LOG(INFO) << "uri: " << uri;
+  absl::string_view ip;
+  int port;
+  if (absl::StartsWith(uri, kIpv4Indicator)) {
+    absl::string_view path =
+        absl::StripPrefix(uri, absl::StrCat(kIpv4Indicator, ":"));
+    if (int colon = path.find(':'); colon != absl::string_view::npos) {
+      ip = path.substr(0, colon);
+      if (!absl::SimpleAtoi(path.substr(colon + 1), &port)) {
+        return absl::InternalError(
+            "Error retrieving port information from uri string");
+      }
+      this->stats_ = GnpsiStats(ip, port);
+      return absl::OkStatus();
+    }
+    return absl::NotFoundError("Port not found in uri string");
+  }
+  if (absl::StartsWith(uri, kIpv6Indicator)) {
+    absl::string_view path =
+        absl::StripPrefix(uri, absl::StrCat(kIpv6Indicator, ":%5B"));
+    if (int closing_bracket = path.find("%5D");
+        closing_bracket != absl::string_view::npos) {
+      ip = path.substr(0, closing_bracket);
+      if (!absl::SimpleAtoi(path.substr(closing_bracket + 4), &port)) {
+        return absl::InternalError(
+            "Error retrieving port information from uri string");
+      }
+      this->stats_ = GnpsiStats(ip, port);
+      return absl::OkStatus();
+    }
+    return absl::NotFoundError("Port not found in uri string");
+  }
+  return absl::InvalidArgumentError("The passed URI format is not supported");
+}
+
 void GnpsiConnection::WaitUntilClosed() {
   absl::MutexLock l(&mu_);
   auto stream_disconnected = [this]() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_) {
@@ -71,6 +113,10 @@ absl::Status GnpsiServiceImpl::AddConnection(GnpsiConnection* connection) {
                      client_max_number_, " clients."));
   }
   LOG(INFO) << "Add " << connection->GetPeerName() << " to client list.";
+  if (absl::Status status = connection->InitializeStats(); !status.ok()) {
+    LOG(ERROR) << "Error while creating stats object for peer - "
+               << status.message();
+  }
   gnpsi_connections_.push_back(connection);
   return absl::OkStatus();
 }
@@ -113,13 +159,25 @@ void GnpsiServiceImpl::SendSamplePacket(
         connection->SendResponse(response)) {
       VLOG(1) << "Successfully sent sample packet to "
               << connection->GetPeerName() << ".";
+      connection->IncrementDatagramCount();
+      connection->IncrementBytesSampled(sample_packet.size());
     } else {
       // If it fails to send response to any client, close this connection.
       LOG(ERROR) << "Failed to send sample packet to "
                  << connection->GetPeerName() << ".";
+      connection->IncrementWriteErrorCount();
       connection->CloseStream();
     }
   }
 }
 
+std::vector<GnpsiStats> GnpsiServiceImpl::GetStats() {
+  absl::MutexLock l(&mu_);
+  std::vector<GnpsiStats> stats;
+  for (auto it = gnpsi_connections_.begin(), end = gnpsi_connections_.end();
+       it != end; it++) {
+    stats.push_back((*it)->GetConnectionStats());
+  }
+  return stats;
+}
 }  // namespace gnpsi

server/gnpsi_service_impl.h

@@ -1,9 +1,11 @@
 #ifndef OPENCONFIG_GNPSI_SERVER_GNPSI_SERVICE_IMPL_H_
 #define OPENCONFIG_GNPSI_SERVER_GNPSI_SERVICE_IMPL_H_
 
+#include <cstdint>
 #include <string>
 #include <vector>
 
+#include "absl/status/status.h"
 #include "absl/synchronization/mutex.h"
 #include "grpcpp/server_context.h"
 #include "grpcpp/support/status.h"
@@ -18,6 +20,25 @@ using ::grpc::ServerWriterInterface;
 using ::grpc::Status;
 using ::grpc::StatusCode;
 
+inline constexpr absl::string_view kIpv4Indicator = "ipv4";
+inline constexpr absl::string_view kIpv6Indicator = "ipv6";
+
+struct GnpsiStats {
+  GnpsiStats() : datagram_count(0), bytes_sampled(0), error_count(0) {}
+  GnpsiStats(absl::string_view ip, int port)
+      : collector_ip(ip),
+        collector_port(port),
+        datagram_count(0),
+        bytes_sampled(0),
+        error_count(0) {}
+
+  std::string collector_ip;
+  int collector_port;
+  int datagram_count;
+  uint64_t bytes_sampled;
+  int error_count;
+};
+
 // Interface to gNPSI sender method
 class GnpsiSenderInterface {
  public:
@@ -26,6 +47,7 @@ class GnpsiSenderInterface {
       const std::string& sample_packet,
       SFlowMetadata::Version version = SFlowMetadata::V5) = 0;
   virtual void DrainConnections() = 0;
+  virtual std::vector<GnpsiStats> GetStats() = 0;
 };
 
 // A connection between a client and gNPSI server. This class is thread-safe.
@@ -35,7 +57,9 @@ class GnpsiConnection {
                            ServerWriterInterface<Sample>* writer)
       : context_(context), writer_(writer), is_stream_closed_(false) {}
 
-  std::string GetPeerName() const { return context_->peer(); }
+  virtual ~GnpsiConnection() = default;
+
+  virtual std::string GetPeerName() const { return context_->peer(); }
 
   bool IsContextCancelled() const { return context_->IsCancelled(); }
 
@@ -52,13 +76,35 @@ class GnpsiConnection {
   // or context is cancelled.
   void WaitUntilClosed() ABSL_LOCKS_EXCLUDED(mu_);
 
+  // Initialize the stats object with the required connection related values.
+  // IPV4 Uri = ipv4:a.b.c.d:e will set connection_ip = a.b.c.d and
+  // connection_port = e. IPV6 Uri = ipv6:[a:b:c:d::]:e will set connection_ip =
+  // a:b:c:d:: and connection_port = e.
+  absl::Status InitializeStats();
+
+  // Increment the count of datagram sent to this connection by 1.
+  void IncrementDatagramCount() { this->stats_.datagram_count++; }
+
+  // Increment the bytes sampled by size of packet.
+  void IncrementBytesSampled(uint64_t packet_size) {
+    this->stats_.bytes_sampled += packet_size;
+  }
+
+  // Increment the count of write errors for this connection by 1.
+  void IncrementWriteErrorCount() { this->stats_.error_count++; }
+
+  // Returns the number of datagrams sent to this connection.
+  GnpsiStats GetConnectionStats() { return this->stats_; }
+
  private:
   ServerContext* context_;
   ServerWriterInterface<::gnpsi::Sample>* writer_;
   // Lock for protecting is_stream_closed_.
   absl::Mutex mu_;
   // When set to true, it means stream is broken.
   bool is_stream_closed_ ABSL_GUARDED_BY(mu_);
+  // Maintains the stats for this connections.
+  GnpsiStats stats_;
 };
 
 // Implementation of gNPSI server.
@@ -85,6 +131,9 @@ class GnpsiServiceImpl : public ::gnpsi::gNPSI::Service,
   // Closes all current conections and blocks any new incoming connections.
   void DrainConnections() ABSL_LOCKS_EXCLUDED(mu_) override;
 
+  // Returns stats per connection collected by the server.
+  std::vector<GnpsiStats> GetStats() ABSL_LOCKS_EXCLUDED(mu_) override;
+
  private:
   int client_max_number_;
   // Lock for protecting member fields.

server/mock_gnpsi_service_impl.h

@@ -1,6 +1,7 @@
 #ifndef OPENCONFIG_GNPSI_SERVER_MOCK_GNPSI_SERVICE_IMPL_H_
 #define OPENCONFIG_GNPSI_SERVER_MOCK_GNPSI_SERVICE_IMPL_H_
 
+#include <vector>
 #include "gmock/gmock.h"
 #include "server/gnpsi_service_impl.h"
 
@@ -13,6 +14,7 @@ class MockGnpsiServiceImpl : public GnpsiSenderInterface {
                SFlowMetadata::Version version),
               (override));
   MOCK_METHOD(void, DrainConnections, (), (override));
+  MOCK_METHOD(std::vector<GnpsiStats>, GetStats, (), (override));
 };
 }  // namespace gnpsi