Merge pull request snabbco#961 from Igalia/lwaftr-get-state

Add snabb get-state multiprocess support

src/apps/config/leader.lua

@@ -682,11 +682,14 @@ function Leader:rpc_get_state (args)
       local printer = path_printer_for_schema_by_name(
          self.schema_name, args.path, false, args.format, args.print_default)
       local states = {}
+      local state_reader = self.support.compute_state_reader(self.schema_name)
       for _, follower in pairs(self.followers) do
-         table.insert(states, state.read_state(self.schema_name, follower.pid))
+         local follower_config = self.support.configuration_for_follower(
+            follower, self.current_configuration)
+         table.insert(states, state_reader(follower.pid, follower_config))
       end
-      -- FIXME: How to combine states?  Add counters?
-      local state = printer(states[1], yang.string_output_file())
+      local state = printer(self.support.process_states(states),
+                            yang.string_output_file())
       return { state = state }
    end
    local success, response = pcall(getter)

src/apps/config/support.lua

@@ -190,6 +190,20 @@ local function add_restarts(actions, app_graph, to_restart)
    return actions
 end
 
+local function configuration_for_follower(follower, configuration)
+   return configuration
+end
+
+local function compute_state_reader(schema_name)
+   return function(pid)
+      local reader = state.state_reader_from_schema_by_name(schema_name)
+      return reader(state.counters_for_pid(pid))
+   end
+end
+
+local function process_states(states)
+   return states[1]
+end
 
 generic_schema_config_support = {
    compute_config_actions = function(
@@ -201,6 +215,9 @@ generic_schema_config_support = {
          in_place_dependencies, app_graph, schema_name, verb, path, arg)
       return compute_mutable_objects_embedded_in_app_initargs(app_graph)
    end,
+   compute_state_reader = compute_state_reader,
+   configuration_for_follower = configuration_for_follower,
+   process_states = process_states,
    compute_apps_to_restart_after_configuration_update =
       compute_apps_to_restart_after_configuration_update,
    translators = {}

src/apps/config/support/snabb-softwire-v2.lua

@@ -6,6 +6,7 @@ local corelib = require('core.lib')
 local equal = require('core.lib').equal
 local dirname = require('core.lib').dirname
 local data = require('lib.yang.data')
+local state = require('lib.yang.state')
 local ipv4_ntop = require('lib.yang.util').ipv4_ntop
 local ipv6 = require('lib.protocol.ipv6')
 local yang = require('lib.yang.yang')
@@ -599,13 +600,77 @@ local function ietf_softwire_translator ()
    return ret
 end
 
+local function configuration_for_follower(follower, configuration)
+   return follower.graph.apps.lwaftr.arg
+end
+
+local function compute_state_reader(schema_name)
+   -- The schema has two lists which we want to look in.
+   local schema = yang.load_schema_by_name(schema_name)
+   local grammar = data.data_grammar_from_schema(schema, false)
+
+   local instance_list_gmr = grammar.members["softwire-config"].members.instance
+   local instance_state_gmr = instance_list_gmr.values["softwire-state"]
+
+   local base_reader = state.state_reader_from_grammar(grammar)
+   local instance_state_reader = state.state_reader_from_grammar(instance_state_gmr)
+
+   return function(pid, data)
+      local counters = state.counters_for_pid(pid)
+      local ret = base_reader(counters)
+      ret.softwire_config.instance = {}
+
+      for device, instance in pairs(data.softwire_config.instance) do
+         local instance_state = instance_state_reader(counters)
+         ret.softwire_config.instance[device] = {}
+         ret.softwire_config.instance[device].softwire_state = instance_state
+      end
+
+      return ret
+   end
+end
+
+local function process_states(states)
+   -- We need to create a summation of all the states as well as adding all the
+   -- instance specific state data to create a total in software-state.
+
+   local unified = {
+      softwire_config = {instance = {}},
+      softwire_state = {}
+   }
+
+   local function total_counter(name, softwire_stats, value)
+      if softwire_stats[name] == nil then
+         return value
+      else
+         return softwire_stats[name] + value
+      end
+   end
+
+   for _, inst_config in ipairs(states) do
+      local name, instance = next(inst_config.softwire_config.instance)
+      unified.softwire_config.instance[name] = instance
+
+      for name, value in pairs(instance.softwire_state) do
+         unified.softwire_state[name] = total_counter(
+            name, unified.softwire_state, value)
+      end
+   end
+
+   return unified
+end
+
+
 function get_config_support()
    return {
       compute_config_actions = compute_config_actions,
       update_mutable_objects_embedded_in_app_initargs =
          update_mutable_objects_embedded_in_app_initargs,
       compute_apps_to_restart_after_configuration_update =
          compute_apps_to_restart_after_configuration_update,
+      compute_state_reader = compute_state_reader,
+      process_states = process_states,
+      configuration_for_follower = configuration_for_follower,
       translators = { ['ietf-softwire'] = ietf_softwire_translator () }
    }
 end

src/lib/yang/state.lua

@@ -38,7 +38,11 @@ local function find_counters(pid)
    return apps
 end
 
-local function state_reader_from_grammar(production, maybe_keyword)
+function counters_for_pid(pid)
+   return flatten(find_counters(pid))
+end
+
+function state_reader_from_grammar(production, maybe_keyword)
    local visitor = {}
    local function visit(keyword, production)
       return assert(visitor[production.type])(keyword, production)
@@ -104,11 +108,6 @@ function state_reader_from_schema_by_name(schema_name)
 end
 state_reader_from_schema_by_name = util.memoize(state_reader_from_schema_by_name)
 
-function read_state(schema_name, pid)
-   local reader = state_reader_from_schema_by_name(schema_name)
-   return reader(flatten(find_counters(pid)))
-end
-
 function selftest ()
    print("selftest: lib.yang.state")
    local simple_router_schema_src = [[module snabb-simple-router {

src/program/lwaftr/counters.lua

@@ -18,7 +18,8 @@ function counter_names ()
 end
 
 function read_counters (pid)
-   local s = state.read_state('snabb-softwire-v2', pid or S.getpid())
+   local reader = state.state_reader_from_schema_by_name('snabb-softwire-v2')
+   local s = reader(state.counters_for_pid(pid or S.getpid()))
    local ret = {}
    for k, id in pairs(counter_names()) do
       ret[k] = s.softwire_state[id]

src/program/lwaftr/tests/subcommands/config_test.py

@@ -148,14 +148,12 @@ def tearDown(self):
         return super().tearDown()
 
     def test_start_empty(self):
-        """ Lwaftr can be started with no instances """
         config = str(DATA_DIR / "empty.conf")
         pid = self.start_daemon(config)
         self.assertEqual(len(self.instances[pid]), 0)
 
 
-    def test_snabb_config(self):
-        """ New instance can be started by snabb config """
+    def test_added_instances_startup(self):
         config = str(DATA_DIR / "icmp_on_fail.conf")
         pid = self.start_daemon(config)
         initial_instance_amount = len(self.instances[pid])
@@ -198,8 +196,7 @@ def test_snabb_config(self):
             len(self.instances[pid]), (initial_instance_amount + 1)
         )
 
-    def test_snabb_config(self):
-        """ Removed instances are stopped by snabb config """
+    def test_removed_instances_shutdown(self):
         config = str(DATA_DIR / "icmp_on_fail.conf")
         pid = self.start_daemon(config)
         initial_instance_amount = len(self.instances[pid])
@@ -221,6 +218,66 @@ def test_snabb_config(self):
             len(self.instances[pid]), (initial_instance_amount - 1)
         )
 
+    def test_snabb_get_state_summation(self):
+        config = str(DATA_DIR / "icmp_on_fail_multiproc.conf")
+        pid = self.start_daemon(config)
+
+        get_state_cmd = list(self.config_args)
+        get_state_cmd[2] = "get-state"
+        get_state_cmd.insert(3, "-f")
+        get_state_cmd.insert(4, "xpath")
+        get_state_cmd.append("/")
+
+        state = self.run_cmd(get_state_cmd).decode(ENC)
+        state = [line for line in state.split("\n") if line]
+
+        # Build two dictionaries, one of each instance counter (a total)
+        # and one of just the values in the global "softwire-state"
+        summed = {}
+        instance = {}
+        for line in state:
+            if "softwire-state" not in line:
+                continue
+
+            path = [elem for elem in line.split("/") if elem]
+            cname = path[-1].split()[0]
+            cvalue = int(path[-1].split()[1])
+
+            if path[0].startswith("instance"):
+                instance[cname] = instance.get(cname, 0) + cvalue
+            elif len(path) < 3:
+                summed[cname] = cvalue
+
+        # Now assert they're the same :)
+        for name, value in summed.items():
+            self.assertEqual(value, instance[name])
+
+    def test_snabb_get_state_lists_instances(self):
+        config = str(DATA_DIR / "icmp_on_fail_multiproc.conf")
+        pid = self.start_daemon(config)
+
+        get_state_cmd = list(self.config_args)
+        get_state_cmd[2] = "get-state"
+        get_state_cmd.insert(3, "-f")
+        get_state_cmd.insert(4, "xpath")
+        get_state_cmd.append("/")
+
+        state = self.run_cmd(get_state_cmd).decode(ENC)
+        state = [line for line in state.split("\n") if line]
+
+        instances = set()
+        for line in state:
+            path = [elem for elem in line.split("/") if elem]
+            if not path[0].startswith("instance"):
+                continue
+
+            device_name = path[0][path[0].find("=")+1:-1]
+            instances.add(device_name)
+
+        self.assertTrue(len(instances) == 2)
+        self.assertTrue("test" in instances)
+        self.assertTrue("test1" in instances)
+
 
 class TestConfigListen(BaseTestCase):
     """