[POC] test build VS image for pops with 1024

cfgmgr/portmgr.cpp

@@ -85,13 +85,15 @@ bool PortMgr::isPortStateOk(const string &alias)
         auto state_opt = swss::fvsGetValue(temp, "state", true);
         if (!state_opt)
         {
+            SWSS_LOG_WARN("Port %s is not ready", alias.c_str());
             return false;
         }
 
-        SWSS_LOG_INFO("Port %s is ready", alias.c_str());
+        SWSS_LOG_WARN("Port %s is ready", alias.c_str());
         return true;
     }
 
+    SWSS_LOG_WARN("Port %s get failed", alias.c_str());
     return false;
 }
 
@@ -192,17 +194,21 @@ void PortMgr::doTask(Consumer &consumer)
                 }
             }
 
+            if (!portOk)
+            {
+                field_values.emplace_back("mtu", mtu);
+                field_values.emplace_back("admin_status", admin_status);
+            }
+
             if (field_values.size())
             {
                 writeConfigToAppDb(alias, field_values);
             }
 
             if (!portOk)
             {
-                SWSS_LOG_INFO("Port %s is not ready, pending...", alias.c_str());
+                SWSS_LOG_WARN("Port %s is not ready, pending...", alias.c_str());
 
-                writeConfigToAppDb(alias, "mtu", mtu);
-                writeConfigToAppDb(alias, "admin_status", admin_status);
                 /* Retry setting these params after the netdev is created */
                 field_values.clear();
                 field_values.emplace_back("mtu", mtu);
@@ -215,18 +221,18 @@ void PortMgr::doTask(Consumer &consumer)
             if (!mtu.empty())
             {
                 setPortMtu(alias, mtu);
-                SWSS_LOG_NOTICE("Configure %s MTU to %s", alias.c_str(), mtu.c_str());
+                SWSS_LOG_WARN("Configure %s MTU to %s", alias.c_str(), mtu.c_str());
             }
 
             if (!admin_status.empty())
             {
                 setPortAdminStatus(alias, admin_status == "up");
-                SWSS_LOG_NOTICE("Configure %s admin status to %s", alias.c_str(), admin_status.c_str());
+                SWSS_LOG_WARN("Configure %s admin status to %s", alias.c_str(), admin_status.c_str());
             }
         }
         else if (op == DEL_COMMAND)
         {
-            SWSS_LOG_NOTICE("Delete Port: %s", alias.c_str());
+            SWSS_LOG_WARN("Delete Port: %s", alias.c_str());
             m_appPortTable.del(alias);
             m_portList.erase(alias);
         }

orchagent/orch.cpp

@@ -242,14 +242,20 @@ void Consumer::execute()
     // ConsumerBase::execute_impl<swss::ConsumerTableBase>();
     SWSS_LOG_ENTER();
 
-    size_t update_size = 0;
     auto table = static_cast<swss::ConsumerTableBase *>(getSelectable());
-    do
+    std::deque<KeyOpFieldsValuesTuple> entries;
+    table->pops(entries);
+
+    // debug log
+    SWSS_LOG_WARN("[HUA] Consumer::execute %s pops %d", table->getTableName().c_str(), (int)(entries.size()));
+    for (auto& entry: entries)
     {
-        std::deque<KeyOpFieldsValuesTuple> entries;
-        table->pops(entries);
-        update_size = addToSync(entries);
-    } while (update_size != 0);
+        // the timestamp of swss.rec difficult to match with syslog, add this info here
+        SWSS_LOG_WARN("[HUA] Consumer::execute entry: %s", dumpTuple(entry).c_str());
+    }
+
+    // add to sync
+    addToSync(entries);
 
     drain();
 }

tests/dash/dash_db.py

@@ -2,6 +2,7 @@
 from dvslib.dvs_common import wait_for_result
 import typing
 import pytest
+import time
 
 from dash_api.appliance_pb2 import *
 from dash_api.vnet_pb2 import *
@@ -58,9 +59,11 @@ def __setitem__(self, key: str, pairs: typing.Union[dict, list, tuple]):
         for k, v in pairs:
             pairs_str.append((to_string(k), to_string(v)))
         self.set(key, pairs_str)
+        time.sleep(3)
 
     def __delitem__(self, key: str):
         self.delete(str(key))
+        time.sleep(3)
 
 
 class Table(swsscommon.Table):

tests/dash/test_dash_acl.py

@@ -21,6 +21,10 @@
 
 import pytest
 
+import logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
 DVS_ENV = ["HWSKU=DPU-2P"]
 NUM_PORTS = 2
 
@@ -87,10 +91,12 @@ def __setitem__(self, key: str, pairs: Union[dict, list, tuple]):
             pairs_str.append((to_string(k), to_string(v)))
         self.table.set(key, pairs_str)
         self.keys.add(key)
+        time.sleep(3)
 
     def __delitem__(self, key: str):
         self.table.delete(str(key))
         self.keys.discard(key)
+        time.sleep(3)
 
     def get_keys(self):
         return self.keys
@@ -242,6 +248,7 @@ def remove_acl_rule(self, group_id, rule_id):
     def create_acl_group(self, group_id, ip_version):
         pb = AclGroup()
         pb.ip_version = IpVersion.IP_VERSION_IPV4
+        logger.warning("create_acl_group: {}='{}'".format(str(group_id), pb.SerializeToString()))
         self.app_dash_acl_group_table[str(group_id)] = {"pb": pb.SerializeToString()}
 
     def remove_acl_group(self, group_id):

tests/mock_tests/consumer_ut.cpp

@@ -10,6 +10,25 @@ namespace consumer_test
 {
     using namespace std;
 
+    class TestOrch : public Orch
+    {
+    public:
+        TestOrch(swss::DBConnector *db, string tableName)
+            :Orch(db, tableName),
+            m_notification_count(0)
+        {
+        }
+
+        void doTask(Consumer& consumer)
+        {
+            std::cout << "TestOrch::doTask " << consumer.m_toSync.size() << std::endl;
+            m_notification_count += consumer.m_toSync.size();
+            consumer.m_toSync.clear();
+        }
+
+        long m_notification_count;
+    };
+
     struct ConsumerTest : public ::testing::Test
     {
         shared_ptr<swss::DBConnector> m_app_db;
@@ -322,4 +341,31 @@ namespace consumer_test
         validate_syncmap(consumer->m_toSync, 1, key, exp_kofv);
 
     }
+
+    TEST_F(ConsumerTest, ConsumerPops_notification_count)
+    {
+        int consumer_pops_batch_size = 10;
+        TestOrch test_orch(m_config_db.get(), "CFG_TEST_TABLE");
+        Consumer test_consumer(
+                new swss::ConsumerStateTable(m_config_db.get(), "CFG_TEST_TABLE", consumer_pops_batch_size, 1), &test_orch, "CFG_TEST_TABLE");
+        swss::ProducerStateTable producer_table(m_config_db.get(), "CFG_TEST_TABLE");
+
+        m_config_db->flushdb();
+        for (int notification_count = 0; notification_count< consumer_pops_batch_size*2; notification_count++)
+        {
+            std::vector<FieldValueTuple> fields;
+            FieldValueTuple t("test_field", "test_value");
+            fields.push_back(t);
+            producer_table.set(std::to_string(notification_count), fields);
+
+            cout << "ConsumerPops_notification_count:: add key: " << notification_count << endl;
+        }
+
+        // consumer should pops consumer_pops_batch_size notifications 
+        test_consumer.execute();
+        ASSERT_EQ(test_orch.m_notification_count, consumer_pops_batch_size);
+
+        test_consumer.execute();
+        ASSERT_EQ(test_orch.m_notification_count, consumer_pops_batch_size*2);
+    }
 }

tests/mock_tests/mock_consumerstatetable.cpp

@@ -7,4 +7,34 @@ namespace swss
         TableName_KeySet(tableName)
     {
     }
+
+    void ConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const std::string& /*prefix*/)
+    {
+        int count = 0;
+        swss::Table table(getDbConnector(), getTableName());
+        std::vector<std::string> keys;
+        table.getKeys(keys);
+        for (const auto &key: keys)
+        {
+            // pop with batch size
+            if (count < POP_BATCH_SIZE)
+            {
+                count++;
+            }
+            else
+            {
+                break;
+            }
+
+            KeyOpFieldsValuesTuple kco;
+            kfvKey(kco) = key;
+            kfvOp(kco) = SET_COMMAND;
+            if (!table.get(key, kfvFieldsValues(kco)))
+            {
+                continue;
+            }
+            table.del(key);
+            vkco.push_back(kco);
+        }
+    }
 }