[P4RT] Add meter counter data in read response, Fix error message to …

…use appropriate entity case, Don't check constraints for entries when the request is a DELETE request & Return read responses in batches. (sonic-net#470)



Co-authored-by: kishanps <[email protected]>

p4rt_app/p4runtime/p4runtime_impl.cc

@@ -96,6 +96,11 @@ std::string GetKeyErrorMessage(pdpi::IrEntity entity,
                                const std::string& extra) {
   switch (entity.entity_case()) {
     case pdpi::IrEntity::kTableEntry:
+      return absl::Substitute(
+          "[P4RT App] Table entry with the given key $0 in table '$1'", extra,
+          entity.table_entry().table_name());
+      break;
+    case pdpi::IrEntity::kPacketReplicationEngineEntry:
       return absl::Substitute(
           "[P4RT App] Packet Replication Engine entry with key of multicast "
           "group ID '$0' $1",
@@ -104,11 +109,6 @@ std::string GetKeyErrorMessage(pdpi::IrEntity entity,
               .multicast_group_id(),
           extra);
       break;
-    case pdpi::IrEntity::kPacketReplicationEngineEntry:
-      return absl::Substitute(
-          "[P4RT App] Table entry with the given key $0 in table '$1'", extra,
-          entity.table_entry().table_name());
-      break;
     default:
       break;
   }
@@ -300,10 +300,14 @@ absl::StatusOr<sonic::AppDbEntry> PiUpdateToAppDbEntry(
   }
 
   if (ir_entity->entity_case() == pdpi::IrEntity::kTableEntry) {
-    // If the constraints are not met then we should just report an error
-    // (i.e. do not try to handle the entry in lower layers).
-    RETURN_IF_ERROR(ValidateTableEntryConstraints(
-        pi_update.entity().table_entry(), constraint_info));
+    // Skip the constraint check for DELETE requests because existing entries
+    // already satisfy constraints, and the request may also omit actions.
+    if (pi_update.type() != p4::v1::Update::DELETE) {
+      // If the constraints are not met then we should just report an error
+      // (i.e. do not try to handle the entry in lower layers).
+      RETURN_IF_ERROR(ValidateTableEntryConstraints(
+          pi_update.entity().table_entry(), constraint_info));
+    }
 
     // Apply any custom translation that are needed on the switch side to
     // account for gNMI configs (e.g. port ID translation).
@@ -719,6 +723,9 @@ grpc::Status P4RuntimeImpl::Write(grpc::ServerContext* context,
 grpc::Status P4RuntimeImpl::Read(
     grpc::ServerContext* context, const p4::v1::ReadRequest* request,
     grpc::ServerWriter<p4::v1::ReadResponse>* response_writer) {
+  // Default max receive message size in GRPC is 4MB, setting the batch size to
+  // 2500 assuming each response message is less than ~1600 bytes max.
+  constexpr int kReadResponseBatchSize = 2500;
   absl::MutexLock l(&server_state_lock_);
 
 #ifdef __EXCEPTIONS
@@ -744,17 +751,20 @@ grpc::Status P4RuntimeImpl::Read(
                           "ReadResponse writer cannot be a nullptr.");
     }
 
-    auto response_status = ReadAllEntities(
-        *request, *ir_p4info_, entity_cache_, translate_port_ids_,
-        port_translation_map_, *cpu_queue_translator_, p4rt_table_);
-    if (!response_status.ok()) {
-      LOG(WARNING) << "Read failure: " << response_status.status();
+    auto responses_status = ReadAllEntitiesInBatches(
+        kReadResponseBatchSize, *request, *ir_p4info_, entity_cache_,
+        translate_port_ids_, port_translation_map_, *cpu_queue_translator_,
+        p4rt_table_);
+    if (!responses_status.ok()) {
+      LOG(WARNING) << "Read failure: " << responses_status.status();
       return grpc::Status(
           grpc::StatusCode::UNKNOWN,
-          absl::StrCat("Read failure: ", response_status.status().ToString()));
+          absl::StrCat("Read failure: ", responses_status.status().ToString()));
     }
 
-    response_writer->Write(response_status.value());
+    for (auto& response : responses_status.value()) {
+      response_writer->Write(response);
+    }
 
     absl::Duration read_execution_time = absl::Now() - read_start_time;
     read_total_requests_ += 1;

p4rt_app/p4runtime/p4runtime_read.cc

@@ -14,6 +14,7 @@
 #include "p4rt_app/p4runtime/p4runtime_read.h"
 
 #include <string>
+#include <vector>
 
 #include "absl/container/flat_hash_map.h"
 #include "absl/status/status.h"
@@ -62,6 +63,10 @@ absl::Status AppendAclCounterData(
   if (ir_table_entry.has_counter_data()) {
     *pi_table_entry.mutable_counter_data() = ir_table_entry.counter_data();
   }
+  if (ir_table_entry.has_meter_counter_data()) {
+    *pi_table_entry.mutable_meter_counter_data() =
+        ir_table_entry.meter_counter_data();
+  }
 
   return absl::OkStatus();
 }
@@ -113,13 +118,16 @@ absl::Status AppendTableEntryReads(
 
 }  // namespace
 
-absl::StatusOr<p4::v1::ReadResponse> ReadAllEntities(
-    const p4::v1::ReadRequest& request, const pdpi::IrP4Info& ir_p4_info,
+absl::StatusOr<std::vector<p4::v1::ReadResponse>> ReadAllEntitiesInBatches(
+    int batch_size, const p4::v1::ReadRequest& request,
+    const pdpi::IrP4Info& ir_p4_info,
     const absl::flat_hash_map<pdpi::EntityKey, p4::v1::Entity>& entity_cache,
     bool translate_port_ids,
     const boost::bimap<std::string, std::string>& port_translation_map,
     CpuQueueTranslator& cpu_queue_translator, sonic::P4rtTable& p4rt_table) {
-  p4::v1::ReadResponse response;
+  std::vector<p4::v1::ReadResponse> responses;
+  responses.push_back(p4::v1::ReadResponse{});
+  int i = 0;
   for (const auto& entity : request.entities()) {
     VLOG(1) << "Read request: " << entity.ShortDebugString();
     switch (entity.entity_case()) {
@@ -128,9 +136,13 @@ absl::StatusOr<p4::v1::ReadResponse> ReadAllEntities(
         for (const auto& [_, entry] : entity_cache) {
           if (entry.entity_case() == p4::v1::Entity::kTableEntry) {
             RETURN_IF_ERROR(AppendTableEntryReads(
-                response, entry.table_entry(), request.role(), ir_p4_info,
-                translate_port_ids, port_translation_map, cpu_queue_translator,
-                p4rt_table));
+                responses.back(), entry.table_entry(), request.role(),
+                ir_p4_info, translate_port_ids, port_translation_map,
+                cpu_queue_translator, p4rt_table));
+            if (++i >= batch_size) {
+              responses.push_back(p4::v1::ReadResponse{});
+              i = 0;
+            }
           }
         }
         break;
@@ -145,7 +157,7 @@ absl::StatusOr<p4::v1::ReadResponse> ReadAllEntities(
                << entity.ShortDebugString();
     }
   }
-  return response;
+  return responses;
 }
 
 }  // namespace p4rt_app

p4rt_app/p4runtime/p4runtime_read.h

@@ -27,15 +27,15 @@
 
 namespace p4rt_app {
 
-// Reads all table entries from a cache. For each ACL entry we also fetch
-// counter data from CounterDb.
-absl::StatusOr<p4::v1::ReadResponse> ReadAllEntities(
-    const p4::v1::ReadRequest& request, const pdpi::IrP4Info& ir_p4_info,
+// Reads all table entries from the cache and return in batches (of batch_size).
+// For each ACL entry we also fetch counter data from CounterDb.
+absl::StatusOr<std::vector<p4::v1::ReadResponse>> ReadAllEntitiesInBatches(
+    int batch_size, const p4::v1::ReadRequest& request,
+    const pdpi::IrP4Info& ir_p4_info,
     const absl::flat_hash_map<pdpi::EntityKey, p4::v1::Entity>& entity_cache,
     bool translate_port_ids,
     const boost::bimap<std::string, std::string>& port_translation_map,
     CpuQueueTranslator& cpu_queue_translator, sonic::P4rtTable& p4rt_table);
-
 }  // namespace p4rt_app
 
 #endif  // PINS_P4RT_APP_P4RUNTIME_P4RUNTIME_READ_H_

p4rt_app/sonic/app_db_manager.cc

@@ -13,7 +13,9 @@
 // limitations under the License.
 #include "p4rt_app/sonic/app_db_manager.h"
 
+#include <cstdint>
 #include <memory>
+#include <string>
 #include <type_traits>
 #include <utility>
 #include <vector>
@@ -23,8 +25,10 @@
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/strings/numbers.h"
+#include "absl/strings/str_cat.h"
 #include "absl/strings/str_format.h"
 #include "absl/strings/str_split.h"
+#include "absl/strings/string_view.h"
 #include "absl/strings/substitute.h"
 #include "glog/logging.h"
 #include "google/rpc/code.pb.h"
@@ -109,29 +113,55 @@ absl::StatusOr<std::string> CreateEntryForModify(
 absl::Status AppendCounterData(
     pdpi::IrTableEntry& table_entry,
     const std::vector<std::pair<std::string, std::string>>& counter_data) {
-  for (const auto& [field, data] : counter_data) {
-    // Update packet count only if data is present.
+  auto field_value_error = [&table_entry](absl::string_view field,
+                                          absl::string_view value) {
+    return absl::Substitute(
+        "Unexpected value '$0' for field '$1' in CountersDB for table entry: "
+        "$2",
+        value, field, table_entry.ShortDebugString());
+  };
+
+  for (const auto& [field, value] : counter_data) {
+    uint64_t counter_value = 0;
+    if (!absl::SimpleAtoi(value, &counter_value)) {
+      LOG(ERROR) << field_value_error(field, value);
+      continue;
+    }
+
+    // Update counter data if present.
     if (field == "packets") {
-      uint64_t packets = 0;
-      if (absl::SimpleAtoi(data, &packets)) {
-        table_entry.mutable_counter_data()->set_packet_count(packets);
-      } else {
-        LOG(ERROR) << "Unexpected packets value '" << data
-                   << "' in CountersDB for table entry: "
-                   << table_entry.ShortDebugString();
-      }
+      table_entry.mutable_counter_data()->set_packet_count(counter_value);
+    } else if (field == "bytes") {
+      table_entry.mutable_counter_data()->set_byte_count(counter_value);
     }
 
-    // Update byte count only if data is present.
-    if (field == "bytes") {
-      uint64_t bytes = 0;
-      if (absl::SimpleAtoi(data, &bytes)) {
-        table_entry.mutable_counter_data()->set_byte_count(bytes);
-      } else {
-        LOG(ERROR) << "Unexpected bytes value '" << data
-                   << "' in CountersDB for table entry: "
-                   << table_entry.ShortDebugString();
-      }
+    if (!table_entry.has_meter_config()) continue;
+
+    // Update meter counter data if present.
+    // Meter color counters are in the form of `color`_packets and
+    // `color`_bytes.
+    // Example: {red_packets 100}, {red_bytes 1000}.
+    if (field == "green_packets") {
+      table_entry.mutable_meter_counter_data()
+          ->mutable_green()
+          ->set_packet_count(counter_value);
+    } else if (field == "green_bytes") {
+      table_entry.mutable_meter_counter_data()->mutable_green()->set_byte_count(
+          counter_value);
+    } else if (field == "yellow_packets") {
+      table_entry.mutable_meter_counter_data()
+          ->mutable_yellow()
+          ->set_packet_count(counter_value);
+    } else if (field == "yellow_bytes") {
+      table_entry.mutable_meter_counter_data()
+          ->mutable_yellow()
+          ->set_byte_count(counter_value);
+    } else if (field == "red_packets") {
+      table_entry.mutable_meter_counter_data()->mutable_red()->set_packet_count(
+          counter_value);
+    } else if (field == "red_bytes") {
+      table_entry.mutable_meter_counter_data()->mutable_red()->set_byte_count(
+          counter_value);
     }
   }