Add gnmi test to verify subscribe API (sonic-net#13696)

### Description of PR

Summary:
Need new test to verify gnmi-native subscribe API.

### Approach
#### What is the motivation for this PR?
Need new test to verify gnmi-native subscribe API.

#### How did you do it?
Add test to verify subscribe polling mode with DEVICE_METADATA table in CONFIG_DB.
Add test to verify subscribe streaming sample mode with DEVICE_METADATA table in CONFIG_DB.
Add test to verify subscribe streaming onchange mode with DEVICE_METADATA table in CONFIG_DB.
Add test to read COUNTERS_QUEUE_NAME_MAP table in COUNTERS_DB.
Add test to read COUNTERS table in COUNTERS_DB.
Add test to verify subscribe polling mode with COUNTERS_PORT_NAME_MAP table in COUNTERS_DB.
Add test to verify subscribe streaming sample mode with COUNTERS_PORT_NAME_MAP table in COUNTERS_DB.
Add test to verify subscribe polling mode with COUNTERS table in COUNTERS_DB.
Add test to verify subscribe streaming sample mode with COUNTERS table in COUNTERS_DB.

#### How did you verify/test it?
Run gnmi end to end test

tests/gnmi/helper.py

@@ -234,6 +234,125 @@ def gnmi_get(duthost, ptfhost, path_list):
             raise Exception("error:" + msg)
 
 
+# py_gnmicli does not fully support POLLING mode
+# Use gnmi_cli instead
+def gnmi_subscribe_polling(duthost, ptfhost, path_list, interval_ms, count):
+    """
+    Send GNMI subscribe request with GNMI client
+
+    Args:
+        duthost: fixture for duthost
+        ptfhost: fixture for ptfhost
+        path_list: list for get path
+        interval_ms: interval, unit is ms
+        count: update count
+
+    Returns:
+        msg: gnmi client output
+    """
+    if path_list is None:
+        logger.error("path_list is None")
+        return "", ""
+    env = GNMIEnvironment(duthost, GNMIEnvironment.GNMI_MODE)
+    ip = duthost.mgmt_ip
+    port = env.gnmi_port
+    interval = interval_ms / 1000.0
+    # Run gnmi_cli in gnmi container as workaround
+    cmd = "docker exec %s gnmi_cli -client_types=gnmi -a %s:%s " % (env.gnmi_container, ip, port)
+    cmd += "-client_crt /etc/sonic/telemetry/gnmiclient.crt "
+    cmd += "-client_key /etc/sonic/telemetry/gnmiclient.key "
+    cmd += "-ca_crt /etc/sonic/telemetry/gnmiCA.pem "
+    cmd += "-logtostderr "
+    # Use sonic-db as default origin
+    cmd += '-origin=sonic-db '
+    cmd += '-query_type=polling '
+    cmd += '-polling_interval %us -count %u ' % (int(interval), count)
+    for path in path_list:
+        path = path.replace('sonic-db:', '')
+        cmd += '-q %s ' % (path)
+    output = duthost.shell(cmd, module_ignore_errors=True)
+    return output['stdout'], output['stderr']
+
+
+def gnmi_subscribe_streaming_sample(duthost, ptfhost, path_list, interval_ms, count):
+    """
+    Send GNMI subscribe request with GNMI client
+
+    Args:
+        duthost: fixture for duthost
+        ptfhost: fixture for ptfhost
+        path_list: list for get path
+        interval_ms: interval, unit is ms
+        count: update count
+
+    Returns:
+        msg: gnmi client output
+    """
+    if path_list is None:
+        logger.error("path_list is None")
+        return "", ""
+    env = GNMIEnvironment(duthost, GNMIEnvironment.GNMI_MODE)
+    ip = duthost.mgmt_ip
+    port = env.gnmi_port
+    cmd = 'python2 /root/gnxi/gnmi_cli_py/py_gnmicli.py '
+    cmd += '--timeout 30 '
+    cmd += '-t %s -p %u ' % (ip, port)
+    cmd += '-xo sonic-db '
+    cmd += '-rcert /root/gnmiCA.pem '
+    cmd += '-pkey /root/gnmiclient.key '
+    cmd += '-cchain /root/gnmiclient.crt '
+    cmd += '--encoding 4 '
+    cmd += '-m subscribe '
+    cmd += '--subscribe_mode 0 --submode 2 --create_connections 1 '
+    cmd += '--interval %u --update_count %u ' % (interval_ms, count)
+    cmd += '--xpath '
+    for path in path_list:
+        path = path.replace('sonic-db:', '')
+        cmd += " " + path
+    output = ptfhost.shell(cmd, module_ignore_errors=True)
+    msg = output['stdout'].replace('\\', '')
+    return msg, output['stderr']
+
+
+def gnmi_subscribe_streaming_onchange(duthost, ptfhost, path_list, count):
+    """
+    Send GNMI subscribe request with GNMI client
+
+    Args:
+        duthost: fixture for duthost
+        ptfhost: fixture for ptfhost
+        path_list: list for get path
+        count: update count
+
+    Returns:
+        msg: gnmi client output
+    """
+    if path_list is None:
+        logger.error("path_list is None")
+        return "", ""
+    env = GNMIEnvironment(duthost, GNMIEnvironment.GNMI_MODE)
+    ip = duthost.mgmt_ip
+    port = env.gnmi_port
+    cmd = 'python2 /root/gnxi/gnmi_cli_py/py_gnmicli.py '
+    cmd += '--timeout 30 '
+    cmd += '-t %s -p %u ' % (ip, port)
+    cmd += '-xo sonic-db '
+    cmd += '-rcert /root/gnmiCA.pem '
+    cmd += '-pkey /root/gnmiclient.key '
+    cmd += '-cchain /root/gnmiclient.crt '
+    cmd += '--encoding 4 '
+    cmd += '-m subscribe '
+    cmd += '--subscribe_mode 0 --submode 1 --create_connections 1 '
+    cmd += '--update_count %u ' % count
+    cmd += '--xpath '
+    for path in path_list:
+        path = path.replace('sonic-db:', '')
+        cmd += " " + path
+    output = ptfhost.shell(cmd, module_ignore_errors=True)
+    msg = output['stdout'].replace('\\', '')
+    return msg, output['stderr']
+
+
 def gnoi_reboot(duthost, method, delay, message):
     env = GNMIEnvironment(duthost, GNMIEnvironment.GNMI_MODE)
     ip = duthost.mgmt_ip

tests/gnmi/test_gnmi_configdb.py

@@ -1,8 +1,13 @@
 import json
 import logging
+import multiprocessing
 import pytest
+import re
+import time
 
 from .helper import gnmi_set, gnmi_get, gnoi_reboot
+from .helper import gnmi_subscribe_polling
+from .helper import gnmi_subscribe_streaming_sample, gnmi_subscribe_streaming_onchange
 from tests.common.helpers.assertions import pytest_assert
 from tests.common.utilities import wait_until
 from tests.common.platform.processes_utils import wait_critical_processes
@@ -105,6 +110,114 @@ def test_gnmi_configdb_incremental_02(duthosts, rand_one_dut_hostname, ptfhost):
         pytest.fail("Set request with invalid path")
 
 
+test_data_metadata = [
+    {
+        "name": "Subscribe table for DEVICE_METADATA",
+        "path": "/sonic-db:CONFIG_DB/localhost/DEVICE_METADATA"
+    },
+    {
+        "name": "Subscribe table key for DEVICE_METADATA",
+        "path": "/sonic-db:CONFIG_DB/localhost/DEVICE_METADATA/localhost"
+    },
+    {
+        "name": "Subscribe table field for DEVICE_METADATA",
+        "path": "/sonic-db:CONFIG_DB/localhost/DEVICE_METADATA/localhost/bgp_asn"
+    }
+]
+
+
+@pytest.mark.parametrize('test_data', test_data_metadata)
+def test_gnmi_configdb_polling_01(duthosts, rand_one_dut_hostname, ptfhost, test_data):
+    '''
+    Verify GNMI subscribe API, streaming onchange mode
+    Subscribe polling mode
+    '''
+    duthost = duthosts[rand_one_dut_hostname]
+    exp_cnt = 3
+    path_list = [test_data["path"]]
+    msg, _ = gnmi_subscribe_polling(duthost, ptfhost, path_list, 1000, exp_cnt)
+    assert msg.count("bgp_asn") >= exp_cnt, test_data["name"] + ": " + msg
+
+
+@pytest.mark.parametrize('test_data', test_data_metadata)
+def test_gnmi_configdb_streaming_sample_01(duthosts, rand_one_dut_hostname, ptfhost, test_data):
+    '''
+    Verify GNMI subscribe API, streaming onchange mode
+    Subscribe streaming sample mode
+    '''
+    duthost = duthosts[rand_one_dut_hostname]
+    exp_cnt = 5
+    path_list = [test_data["path"]]
+    msg, _ = gnmi_subscribe_streaming_sample(duthost, ptfhost, path_list, 0, exp_cnt)
+    assert msg.count("bgp_asn") >= exp_cnt, test_data["name"] + ": " + msg
+
+
+@pytest.mark.parametrize('test_data', test_data_metadata)
+def test_gnmi_configdb_streaming_onchange_01(duthosts, rand_one_dut_hostname, ptfhost, test_data):
+    '''
+    Verify GNMI subscribe API, streaming onchange mode
+    Subscribe streaming onchange mode
+    '''
+    duthost = duthosts[rand_one_dut_hostname]
+    run_flag = multiprocessing.Value('I', True)
+
+    # Update DEVICE_METADATA table to trigger onchange event
+    def worker(duthost, run_flag):
+        for i in range(100):
+            if not run_flag.value:
+                break
+            time.sleep(0.5)
+            cmd = "sonic-db-cli CONFIG_DB hdel \"DEVICE_METADATA|localhost\" bgp_asn "
+            duthost.shell(cmd, module_ignore_errors=True)
+            time.sleep(0.5)
+            cmd = "sonic-db-cli CONFIG_DB hset \"DEVICE_METADATA|localhost\" bgp_asn " + str(i+1000)
+            duthost.shell(cmd, module_ignore_errors=True)
+
+    client_task = multiprocessing.Process(target=worker, args=(duthost, run_flag,))
+    client_task.start()
+    exp_cnt = 5
+    path_list = [test_data["path"]]
+    msg, _ = gnmi_subscribe_streaming_onchange(duthost, ptfhost, path_list, exp_cnt*2)
+    run_flag.value = False
+    client_task.join()
+    assert msg.count("bgp_asn") >= exp_cnt, test_data["name"] + ": " + msg
+
+
+def test_gnmi_configdb_streaming_onchange_02(duthosts, rand_one_dut_hostname, ptfhost):
+    '''
+    Verify GNMI subscribe API, streaming onchange mode
+    Subscribe table, and verify gnmi output has table key
+    '''
+    duthost = duthosts[rand_one_dut_hostname]
+    run_flag = multiprocessing.Value('I', True)
+
+    # Update DEVICE_METADATA table to trigger onchange event
+    def worker(duthost, run_flag):
+        for i in range(100):
+            if not run_flag.value:
+                break
+            time.sleep(0.5)
+            cmd = "sonic-db-cli CONFIG_DB hset \"DEVICE_METADATA|localhost\" bgp_asn " + str(i+1000)
+            duthost.shell(cmd, module_ignore_errors=True)
+
+    client_task = multiprocessing.Process(target=worker, args=(duthost, run_flag,))
+    client_task.start()
+    exp_cnt = 3
+    path_list = ["/sonic-db:CONFIG_DB/localhost/DEVICE_METADATA"]
+    msg, _ = gnmi_subscribe_streaming_onchange(duthost, ptfhost, path_list, exp_cnt)
+    run_flag.value = False
+    client_task.join()
+
+    match_list = re.findall("json_ietf_val: \"({.*?})\"", msg)
+    assert len(match_list) >= exp_cnt, "Missing json_ietf_val in gnmi response: " + msg
+    for match in match_list:
+        result = json.loads(match)
+        # Verify table key
+        assert "localhost" in result, "Invalid result: " + match
+        # Verify table field
+        assert "bgp_asn" in result["localhost"], "Invalid result: " + match
+
+
 def test_gnmi_configdb_full_01(duthosts, rand_one_dut_hostname, ptfhost):
     '''
     Verify GNMI native write, full config for configDB