[Thinkit] Replacing "Cpu0" with "CPU" after cpu port naming chang. Added destructor for packet listener.updating gnmi_helper.cc to support string values in gNMI GET response parsing

lib/BUILD.bazel

@@ -25,6 +25,7 @@ cc_library(
     hdrs = ["gpins_control_interface.h"],
     deps = [
         "//gutil:status",
+        "//gutil:testing",
         "//lib/gnmi:gnmi_helper",
         "//lib/p4rt:packet_listener",
         "//lib/validator:validator_lib",
@@ -36,6 +37,7 @@ cc_library(
         "//p4_pdpi/packetlib:packetlib_cc_proto",
         "//sai_p4/instantiations/google:instantiations",
         "//sai_p4/instantiations/google:sai_p4info_cc",
+        "//sai_p4/instantiations/google:sai_pd_cc_proto",
         "//tests/forwarding:util",
         "//thinkit:control_interface",
         "//thinkit:packet_generation_finalizer",

lib/gnmi/gnmi_helper.cc

@@ -16,6 +16,7 @@
 
 #include <memory>
 #include <string>
+#include <string_view>
 #include <utility>
 #include <vector>
 
@@ -135,6 +136,17 @@ absl::StatusOr<gnmi::GetRequest> BuildGnmiGetRequest(
   return req;
 }
 
+absl::StatusOr<std::string> ParseJsonResponse(absl::string_view val,
+                                              absl::string_view match_tag) {
+  const auto resp_json = json::parse(val);
+  const auto match_tag_json = resp_json.find(match_tag);
+  if (match_tag_json == resp_json.end()) {
+    return gutil::InternalErrorBuilder().LogError()
+           << match_tag << " not present in JSON response " << val;
+  }
+  return match_tag_json->dump();
+}
+
 absl::StatusOr<std::string> ParseGnmiGetResponse(
     const gnmi::GetResponse& response, absl::string_view match_tag) {
   if (response.notification_size() != 1)
@@ -146,19 +158,23 @@ absl::StatusOr<std::string> ParseGnmiGetResponse(
     return gutil::InternalErrorBuilder().LogError()
            << "Unexpected update size in response (should be 1): "
            << response.notification(0).update_size();
-
-  const auto resp_json =
-      json::parse(response.notification(0).update(0).val().json_ietf_val());
-  const auto match_tag_json = resp_json.find(match_tag);
-  if (match_tag_json == resp_json.end()) {
-    return gutil::InternalErrorBuilder().LogError()
-           << match_tag << " not present in JSON response "
-           << response.ShortDebugString();
+  switch (response.notification(0).update(0).val().value_case()) {
+    case gnmi::TypedValue::kStringVal:
+      return response.notification(0).update(0).val().string_val();
+    case gnmi::TypedValue::kJsonVal:
+      return ParseJsonResponse(
+          response.notification(0).update(0).val().json_val(), match_tag);
+    case gnmi::TypedValue::kJsonIetfVal:
+      return ParseJsonResponse(
+          response.notification(0).update(0).val().json_ietf_val(), match_tag);
+    default:
+      return gutil::InternalErrorBuilder().LogError()
+             << "Unexpected data type: "
+             << response.notification(0).update(0).val().value_case();
   }
-  return match_tag_json->dump();
 }
 
-absl::Status SetGnmiConfigPath(gnmi::gNMI::Stub* sut_gnmi_stub,
+absl::Status SetGnmiConfigPath(gnmi::gNMI::StubInterface* sut_gnmi_stub,
                                absl::string_view config_path,
                                GnmiSetType operation, absl::string_view value) {
   ASSIGN_OR_RETURN(gnmi::SetRequest request,
@@ -176,8 +192,8 @@ absl::Status SetGnmiConfigPath(gnmi::gNMI::Stub* sut_gnmi_stub,
   return absl::OkStatus();
 }
 
-absl::Status PushGnmiConfig(gnmi::gNMI::Stub& stub,
-                            absl::string_view chassis_name,
+absl::Status PushGnmiConfig(gnmi::gNMI::StubInterface& stub,
+                            const std::string& chassis_name,
                             const std::string& gnmi_config,
                             absl::uint128 election_id) {
   gnmi::SetRequest req;
@@ -203,8 +219,7 @@ absl::Status PushGnmiConfig(gnmi::gNMI::Stub& stub,
 
 absl::Status PushGnmiConfig(thinkit::Switch& chassis,
                             const std::string& gnmi_config) {
-  ASSIGN_OR_RETURN(std::unique_ptr<gnmi::gNMI::Stub> stub,
-                   chassis.CreateGnmiStub());
+  ASSIGN_OR_RETURN(auto stub, chassis.CreateGnmiStub());
   return pins_test::PushGnmiConfig(*stub, chassis.ChassisName(), gnmi_config);
 }
 
@@ -264,8 +279,8 @@ GetInterfaceToOperStatusMapOverGnmi(gnmi::gNMI::StubInterface& stub,
     }
     std::string name = std::string(StripQuotes(element_name_json->dump()));
 
-    // TODO: Remove once CpuX contains the oper-state subtree.
-    if (absl::StartsWith(name, "Cpu")) {
+    // TODO: Remove once CPU contains the oper-state subtree.
+    if (absl::StartsWith(name, "CPU")) {
       LOG(INFO) << "Skipping " << name << ".";
       continue;
     }

lib/gnmi/gnmi_helper.h

@@ -64,7 +64,7 @@ absl::StatusOr<gnmi::GetRequest> BuildGnmiGetRequest(
 absl::StatusOr<std::string> ParseGnmiGetResponse(
     const gnmi::GetResponse& response, absl::string_view match_tag);
 
-absl::Status SetGnmiConfigPath(gnmi::gNMI::Stub* sut_gnmi_stub,
+absl::Status SetGnmiConfigPath(gnmi::gNMI::StubInterface* sut_gnmi_stub,
                                absl::string_view config_path,
                                GnmiSetType operation, absl::string_view value);
 
@@ -98,7 +98,7 @@ absl::StatusOr<std::vector<absl::string_view>>
 GnmiGetElementFromTelemetryResponse(const gnmi::SubscribeResponse& response);
 
 absl::Status PushGnmiConfig(
-    gnmi::gNMI::Stub& stub, absl::string_view chassis_name,
+    gnmi::gNMI::StubInterface& stub, const std::string& chassis_name,
     const std::string& gnmi_config,
     absl::uint128 election_id = pdpi::TimeBasedElectionId());
 

lib/gnmi/gnmi_helper_test.cc

@@ -492,7 +492,7 @@ TEST(GetInterfaceOperStatusMap, SuccessfullyReturnsInterfaceOperStatusMap) {
                  update {
                    path { elem { name: "interfaces" } }
                    val {
-                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{\"interface\":[{\"name\":\"Cpu0\"},{\"name\":\"Ethernet0\",\"state\":{\"oper-status\":\"DOWN\"}}]}}"
+                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{\"interface\":[{\"name\":\"CPU\"},{\"name\":\"Ethernet0\",\"state\":{\"oper-status\":\"DOWN\"}}]}}"
                    }
                  }
                })pb")),
@@ -539,7 +539,7 @@ TEST(GetInterfacePortIdMap, PortIdNotFoundInState) {
                  update {
                    path { elem { name: "interfaces" } }
                    val {
-                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{\"interface\":[{\"name\":\"Cpu0\"},{\"name\":\"Ethernet0\",\"state\":{\"name\":\"Ethernet0\"}}]}}"
+                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{\"interface\":[{\"name\":\"CPU\"},{\"name\":\"Ethernet0\",\"state\":{\"name\":\"Ethernet0\"}}]}}"
                    }
                  }
                })pb")),
@@ -1010,5 +1010,122 @@ TEST(GetUpInterfaces, SuccessfullyGetsUpInterface) {
               testing::ContainerEq(std::vector<std::string>{"Ethernet0"}));
 }
 
+TEST(CheckParseGnmiGetResponse, FailDuetoResponseSize) {
+  gnmi::GetResponse response;
+  for (int i = 0; i < 2; i++) {
+    gnmi::Notification *notification = response.add_notification();
+    notification->set_timestamp(1620348032128305716);
+  }
+  EXPECT_THAT(
+      ParseGnmiGetResponse(response, "openconfig-interfaces:oper-status"),
+      StatusIs(absl::StatusCode::kInternal,
+               HasSubstr("Unexpected size in response")));
+}
+
+TEST(CheckParseGnmiGetResponse, FailDuetoUpdateSize) {
+  gnmi::GetResponse response;
+  gnmi::Notification *notification = response.add_notification();
+
+  for (int i = 0; i < 2; i++) {
+    notification->add_update();
+  }
+  EXPECT_THAT(
+      ParseGnmiGetResponse(response, "openconfig-interfaces:oper-status"),
+      StatusIs(absl::StatusCode::kInternal,
+               HasSubstr("Unexpected update size in response")));
+}
+
+TEST(CheckParseGnmiGetResponse, UnexpectedDataFormat) {
+  gnmi::GetResponse response;
+  constexpr char kOperstatus[] = "UP";
+  gnmi::Notification *notification = response.add_notification();
+  gnmi::Update *update = notification->add_update();
+
+  *update->mutable_path() = ConvertOCStringToPath(
+      "interfaces/interface[name=Ethernet8]/state/oper-status");
+  update->mutable_val()->set_ascii_val(kOperstatus);
+  LOG(INFO) << "response: " << response.DebugString();
+  EXPECT_THAT(
+      ParseGnmiGetResponse(response, "openconfig-interfaces:oper-status"),
+      StatusIs(absl::StatusCode::kInternal,
+               HasSubstr("Unexpected data type:")));
+}
+TEST(CheckParseGnmiGetResponse, FailDuetoMissingTag) {
+  gnmi::GetResponse response;
+  constexpr char kOperstatus[] =
+      R"({"openconfig-interfaces:status":"TESTING"})";
+  gnmi::Notification *notification = response.add_notification();
+  gnmi::Update *update = notification->add_update();
+  *update->mutable_path() =
+      ConvertOCStringToPath("interfaces/interface[name=Ethernet8]/state");
+  update->mutable_val()->set_json_ietf_val(kOperstatus);
+  LOG(INFO) << "response: " << response.DebugString();
+  EXPECT_THAT(
+      ParseGnmiGetResponse(response, "openconfig-interfaces:oper-status"),
+      StatusIs(absl::StatusCode::kInternal,
+               HasSubstr("not present in JSON response")));
+}
+TEST(CheckParseGnmiGetResponse, FailDuetoWrongResponse) {
+  gnmi::GetResponse response;
+  constexpr char kOperstatus[] =
+      R"({"openconfig-interfaces:oper-status":"TESTING"})";
+  gnmi::Notification *notification = response.add_notification();
+  gnmi::Update *update = notification->add_update();
+
+  *update->mutable_path() = ConvertOCStringToPath(
+      "interfaces/interface[name=Ethernet8]/state/oper-status");
+  update->mutable_val()->set_json_ietf_val(kOperstatus);
+  LOG(INFO) << "response: " << response.DebugString();
+  EXPECT_THAT(
+      ParseGnmiGetResponse(response, "openconfig-interfaces:oper-status"),
+      IsOkAndHolds(Not(HasSubstr("UP"))));
+}
+
+TEST(CheckParseGnmiGetResponse, JsonIetfResponseSuccess) {
+  gnmi::GetResponse response;
+  constexpr char kOperstatus[] =
+      R"({"openconfig-interfaces:oper-status":"UP"})";
+  gnmi::Notification *notification = response.add_notification();
+  gnmi::Update *update = notification->add_update();
+
+  *update->mutable_path() = ConvertOCStringToPath(
+      "interfaces/interface[name=Ethernet8]/state/oper-status");
+  update->mutable_val()->set_json_ietf_val(kOperstatus);
+  LOG(INFO) << "response: " << response.DebugString();
+  EXPECT_THAT(
+      ParseGnmiGetResponse(response, "openconfig-interfaces:oper-status"),
+      IsOkAndHolds(HasSubstr("UP")));
+}
+
+TEST(CheckParseGnmiGetResponse, JsonResponseSuccess) {
+  gnmi::GetResponse response;
+  constexpr char kOperstatus[] =
+      R"({"openconfig-interfaces:oper-status":"UP"})";
+  gnmi::Notification *notification = response.add_notification();
+  gnmi::Update *update = notification->add_update();
+
+  *update->mutable_path() = ConvertOCStringToPath(
+      "interfaces/interface[name=Ethernet8]/state/oper-status");
+  update->mutable_val()->set_json_val(kOperstatus);
+  LOG(INFO) << "response: " << response.DebugString();
+  EXPECT_THAT(
+      ParseGnmiGetResponse(response, "openconfig-interfaces:oper-status"),
+      IsOkAndHolds(HasSubstr("UP")));
+}
+
+TEST(CheckParseGnmiGetResponse, StringResponseSuccess) {
+  gnmi::GetResponse response;
+  constexpr char kOperstatus[] = "UP";
+  gnmi::Notification *notification = response.add_notification();
+  gnmi::Update *update = notification->add_update();
+
+  *update->mutable_path() = ConvertOCStringToPath(
+      "interfaces/interface[name=Ethernet8]/state/oper-status");
+  update->mutable_val()->set_string_val(kOperstatus);
+  LOG(INFO) << "response: " << response.DebugString();
+  EXPECT_THAT(
+      ParseGnmiGetResponse(response, "openconfig-interfaces:oper-status"),
+      IsOkAndHolds(HasSubstr("UP")));
+}
 }  // namespace
 }  // namespace pins_test

lib/gpins_control_interface.cc

@@ -191,8 +191,7 @@ GpinsControlInterface::GetBERTResult(
 
 absl::StatusOr<absl::flat_hash_set<std::string>>
 GpinsControlInterface::GetUpLinks(absl::Span<const std::string> interfaces) {
-  ASSIGN_OR_RETURN(std::unique_ptr<gnmi::gNMI::Stub> gnmi_stub,
-                   sut_->CreateGnmiStub());
+  ASSIGN_OR_RETURN(auto gnmi_stub, sut_->CreateGnmiStub());
   gnmi::GetResponse response;
   absl::flat_hash_set<std::string> up_links;
   for (const std::string& interface : interfaces) {

lib/p4rt/packet_listener.cc

@@ -20,11 +20,20 @@ PacketListener::PacketListener(
     const absl::flat_hash_map<std::string, std::string>*
         interface_port_id_to_name,
     thinkit::PacketCallback callback)
-    : session_(std::move(session)),
+    : session_(session),
+      time_to_exit_(false),
       receive_packet_thread_([this, ir_p4info, interface_port_id_to_name,
-                              callback = std::move(callback)]() {
+                              callback = std::move(
+                                  callback)]() ABSL_LOCKS_EXCLUDED(mutex_) {
         p4::v1::StreamMessageResponse pi_response;
         while (session_->StreamChannelRead(pi_response)) {
+          {
+            absl::MutexLock lock(&mutex_);
+            if (time_to_exit_) {
+              break;
+            }
+          }
+
           sai::StreamMessageResponse pd_response;
           if (!pdpi::PiStreamMessageResponseToPd(*ir_p4info, pi_response,
                                                  &pd_response)
@@ -37,6 +46,8 @@ PacketListener::PacketListener(
             return;
           }
           std::string port_id = pd_response.packet().metadata().ingress_port();
+
+          LOG(INFO) << "Packet received from port id: " << port_id;
           auto port_name = interface_port_id_to_name->find(port_id);
           if (port_name == interface_port_id_to_name->end()) {
             LOG(WARNING) << port_id << " not found.";

lib/p4rt/packet_listener.h

@@ -45,7 +45,7 @@ class PacketListener : public thinkit::PacketGenerationFinalizer {
                      interface_port_id_to_name,
                  thinkit::PacketCallback callback);
 
- ~PacketListener() ABSL_LOCKS_EXCLUDED(mutex_) {
+  ~PacketListener() ABSL_LOCKS_EXCLUDED(mutex_) {
     {
       absl::MutexLock lock(&mutex_);
       time_to_exit_ = true;

lib/validator/validator_lib.cc

@@ -38,12 +38,6 @@
 
 namespace pins_test {
 
-namespace {
-
-using Stub = ::p4::v1::P4Runtime::Stub;
-
-}  // namespace
-
 absl::Status Pingable(absl::string_view chassis_name, absl::Duration timeout) {
   constexpr char kPingCommand[] = R"(fping -t $0 $1; fping6 -t $0 $1)";
   FILE* in;
@@ -95,25 +89,20 @@ absl::Status SSHable(thinkit::Switch& thinkit_switch,
 
 // Checks if a P4Runtime session could be established.
 absl::Status P4rtAble(thinkit::Switch& thinkit_switch) {
-  ASSIGN_OR_RETURN(std::unique_ptr<Stub> p4runtime_stub,
-                   thinkit_switch.CreateP4RuntimeStub());
-  return pdpi::P4RuntimeSession::Create(std::move(p4runtime_stub),
-                                        thinkit_switch.DeviceId())
-      .status();
+  return pdpi::P4RuntimeSession::Create(thinkit_switch).status();
 }
 
 // Checks if a gNMI get all interface request can be sent and a response
 // received.
 absl::Status GnmiAble(thinkit::Switch& thinkit_switch, absl::Duration timeout) {
-  ASSIGN_OR_RETURN(std::unique_ptr<gnmi::gNMI::Stub> gnmi_stub,
-                   thinkit_switch.CreateGnmiStub());
+  ASSIGN_OR_RETURN(auto gnmi_stub, thinkit_switch.CreateGnmiStub());
   return pins_test::CanGetAllInterfaceOverGnmi(*gnmi_stub, timeout);
 }
 
 // Checks if a gNOI system get time request can be sent and a response
 // received.
 absl::Status GnoiAble(thinkit::Switch& thinkit_switch, absl::Duration timeout) {
-  ASSIGN_OR_RETURN(std::unique_ptr<gnoi::system::System::Stub> gnoi_system_stub,
+  ASSIGN_OR_RETURN(auto gnoi_system_stub,
                    thinkit_switch.CreateGnoiSystemStub());
   gnoi::system::TimeRequest request;
   gnoi::system::TimeResponse response;
@@ -127,8 +116,7 @@ absl::Status GnoiAble(thinkit::Switch& thinkit_switch, absl::Duration timeout) {
 absl::Status PortsUp(thinkit::Switch& thinkit_switch,
                      absl::Span<const std::string> interfaces,
                      absl::Duration timeout) {
-  ASSIGN_OR_RETURN(std::unique_ptr<gnmi::gNMI::Stub> gnmi_stub,
-                   thinkit_switch.CreateGnmiStub());
+  ASSIGN_OR_RETURN(auto gnmi_stub, thinkit_switch.CreateGnmiStub());
   if (interfaces.empty()) {
     return pins_test::CheckAllInterfaceOperStateOverGnmi(
         *gnmi_stub, /*interface_oper_state=*/"UP",
@@ -139,8 +127,7 @@ absl::Status PortsUp(thinkit::Switch& thinkit_switch,
 }
 
 absl::Status NoAlarms(thinkit::Switch& thinkit_switch, absl::Duration timeout) {
-  ASSIGN_OR_RETURN(std::unique_ptr<gnmi::gNMI::Stub> gnmi_stub,
-                   thinkit_switch.CreateGnmiStub());
+  ASSIGN_OR_RETURN(auto gnmi_stub, thinkit_switch.CreateGnmiStub());
   ASSIGN_OR_RETURN(std::vector<std::string> alarms,
                    pins_test::GetAlarms(*gnmi_stub));
   if (alarms.empty()) {

p4_pdpi/BUILD.bazel

@@ -336,6 +336,7 @@ cc_library(
         "//gutil:status",
         "//gutil:version",
         "//sai_p4/fixed:p4_roles",
+        "//thinkit:switch",
         "@com_github_google_glog//:glog",
         "@com_github_grpc_grpc//:grpc++",
         "@com_github_p4lang_p4runtime//:p4runtime_cc_grpc",