Add requested replicas field to ReplicaSet and Deployment resources.

Summary: Since usually users want to know how many replicas are requested of each resource which manager replicas, add a `requested_replicas` field to rs and deployments which is taken from the spec object. This will give more visibility to the users about the state of the object.

Test Plan: test that requested_replicas field is added to all steps of metadata pipeline

Reviewers: philkuz, vihang, michelle

Reviewed By: philkuz

Signed-off-by: Taras Priadka <tpriadka@pixielabs.ai>

Differential Revision: https://phab.corp.pixielabs.ai/D11993

GitOrigin-RevId: 7a6b61d

Author: TarasPriadka

src/carnot/funcs/metadata/metadata_ops.h

@@ -698,6 +698,7 @@ inline types::StringValue ReplicaSetInfoToStatus(const px::md::ReplicaSetInfo* r
   int ready_replicas = 0;
   int available_replicas = 0;
   int observed_generation = 0;
+  int requested_replicas = 0;
   std::string conditions = "";
 
   if (rs_info != nullptr) {
@@ -706,6 +707,7 @@ inline types::StringValue ReplicaSetInfoToStatus(const px::md::ReplicaSetInfo* r
     ready_replicas = rs_info->ready_replicas();
     available_replicas = rs_info->available_replicas();
     observed_generation = rs_info->observed_generation();
+    requested_replicas = rs_info->requested_replicas();
 
     auto rs_conditions = rs_info->conditions();
     for (const auto& condition : rs_info->conditions()) {
@@ -721,6 +723,7 @@ inline types::StringValue ReplicaSetInfoToStatus(const px::md::ReplicaSetInfo* r
   d.AddMember("fully_labeled_replicas", fully_labeled_replicas, d.GetAllocator());
   d.AddMember("ready_replicas", ready_replicas, d.GetAllocator());
   d.AddMember("available_replicas", available_replicas, d.GetAllocator());
+  d.AddMember("requested_replicas", requested_replicas, d.GetAllocator());
   d.AddMember("observed_generation", observed_generation, d.GetAllocator());
   d.AddMember("conditions", internal::StringRef(conditions), d.GetAllocator());
   rapidjson::StringBuffer sb;
@@ -774,6 +777,8 @@ inline types::StringValue DeploymentInfoToStatus(const px::md::DeploymentInfo* d
   int available_replicas = 0;
   int unavailable_replicas = 0;
   int observed_generation = 0;
+  int requested_replicas = 0;
+
   std::string conditions = "";
 
   if (dep_info != nullptr) {
@@ -783,6 +788,7 @@ inline types::StringValue DeploymentInfoToStatus(const px::md::DeploymentInfo* d
     available_replicas = dep_info->available_replicas();
     unavailable_replicas = dep_info->unavailable_replicas();
     observed_generation = dep_info->observed_generation();
+    requested_replicas = dep_info->requested_replicas();
 
     auto rs_conditions = dep_info->conditions();
     for (const auto& condition : dep_info->conditions()) {
@@ -799,6 +805,7 @@ inline types::StringValue DeploymentInfoToStatus(const px::md::DeploymentInfo* d
   d.AddMember("ready_replicas", ready_replicas, d.GetAllocator());
   d.AddMember("available_replicas", available_replicas, d.GetAllocator());
   d.AddMember("unavailable_replicas", unavailable_replicas, d.GetAllocator());
+  d.AddMember("requested_replicas", requested_replicas, d.GetAllocator());
   d.AddMember("observed_generation", observed_generation, d.GetAllocator());
   d.AddMember("conditions", internal::StringRef(conditions), d.GetAllocator());
   rapidjson::StringBuffer sb;

src/carnot/funcs/metadata/metadata_ops_test.cc

@@ -899,8 +899,9 @@ TEST_F(MetadataOpsTest, replicaset_id_to_status_test) {
 
   md::ReplicaSetConditions conditions_ready{{"ready", md::ConditionStatus::kTrue}};
   md::ReplicaSetConditions conditions_terminated{{"Terminating", md::ConditionStatus::kTrue}};
-  md::ReplicaSetInfo rs_info_ready("", "", "", 5, 5, 3, 3, 5, conditions_ready, 101, 0);
-  md::ReplicaSetInfo rs_info_terminating("", "", "", 0, 0, 0, 0, 5, conditions_terminated, 101, 0);
+  md::ReplicaSetInfo rs_info_ready("", "", "", 5, 5, 3, 3, 5, 5, conditions_ready, 101, 0);
+  md::ReplicaSetInfo rs_info_terminating("", "", "", 0, 0, 0, 0, 5, 0, conditions_terminated, 101,
+                                         0);
 
   udf_tester.ForInput("rs0_uid").Expect(ReplicaSetInfoToStatus(&rs_info_ready));
   udf_tester.ForInput("terminating_rs0_uid").Expect(ReplicaSetInfoToStatus(&rs_info_terminating));
@@ -1018,8 +1019,9 @@ TEST_F(MetadataOpsTest, replicaset_name_to_status_test) {
 
   md::ReplicaSetConditions conditionsReady{{"ready", md::ConditionStatus::kTrue}};
   md::ReplicaSetConditions conditionsTerminating{{"Terminating", md::ConditionStatus::kTrue}};
-  md::ReplicaSetInfo rs_info_ready("", "", "", 5, 5, 3, 3, 5, conditionsReady, 101, 0);
-  md::ReplicaSetInfo rs_info_terminating("", "", "", 0, 0, 0, 0, 5, conditionsTerminating, 101, 0);
+  md::ReplicaSetInfo rs_info_ready("", "", "", 5, 5, 3, 3, 5, 5, conditionsReady, 101, 0);
+  md::ReplicaSetInfo rs_info_terminating("", "", "", 0, 0, 0, 0, 5, 0, conditionsTerminating, 101,
+                                         0);
 
   udf_tester.ForInput("pl/rs0").Expect(ReplicaSetInfoToStatus(&rs_info_ready));
   udf_tester.ForInput("pl/terminating_rs").Expect(ReplicaSetInfoToStatus(&rs_info_terminating));
@@ -1120,11 +1122,11 @@ TEST_F(MetadataOpsTest, deployment_id_to_status_test) {
   md::DeploymentConditions conditions_terminating{
       {md::DeploymentConditionType::kAvailable, md::ConditionStatus::kFalse}};
 
-  md::DeploymentInfo dep_info_ready("", "", "", 5, 5, 4, 3, 3, 2, conditions_ready, 101, 0);
-  md::DeploymentInfo dep_info_terminating("", "", "", 2, 6, 5, 3, 3, 2, conditions_terminating, 123,
-                                          0);
-  md::DeploymentInfo dep_info_terminated("", "", "", 2, 0, 0, 0, 0, 0, conditions_terminating, 123,
-                                         150);
+  md::DeploymentInfo dep_info_ready("", "", "", 5, 5, 4, 3, 3, 2, 5, conditions_ready, 101, 0);
+  md::DeploymentInfo dep_info_terminating("", "", "", 2, 6, 5, 3, 3, 2, 0, conditions_terminating,
+                                          123, 0);
+  md::DeploymentInfo dep_info_terminated("", "", "", 2, 0, 0, 0, 0, 0, 0, conditions_terminating,
+                                         123, 150);
 
   udf_tester.ForInput("deployment_uid").Expect(DeploymentInfoToStatus(&dep_info_ready));
   udf_tester.ForInput("terminating_deployment_uid")
@@ -1202,11 +1204,11 @@ TEST_F(MetadataOpsTest, deployment_name_to_status_test) {
   md::DeploymentConditions conditions_terminating{
       {md::DeploymentConditionType::kAvailable, md::ConditionStatus::kFalse}};
 
-  md::DeploymentInfo dep_info_ready("", "", "", 5, 5, 4, 3, 3, 2, conditions_ready, 101, 0);
-  md::DeploymentInfo dep_info_terminating("", "", "", 2, 6, 5, 3, 3, 2, conditions_terminating, 123,
-                                          0);
-  md::DeploymentInfo dep_info_terminated("", "", "", 2, 0, 0, 0, 0, 0, conditions_terminating, 123,
-                                         150);
+  md::DeploymentInfo dep_info_ready("", "", "", 5, 5, 4, 3, 3, 2, 5, conditions_ready, 101, 0);
+  md::DeploymentInfo dep_info_terminating("", "", "", 2, 6, 5, 3, 3, 2, 0, conditions_terminating,
+                                          123, 0);
+  md::DeploymentInfo dep_info_terminated("", "", "", 2, 0, 0, 0, 0, 0, 0, conditions_terminating,
+                                         123, 150);
 
   udf_tester.ForInput("pl/deployment1").Expect(DeploymentInfoToStatus(&dep_info_ready));
   udf_tester.ForInput("pl/terminating_deployment1")