CORTX-28971: update service fids along with process fids

A motr client restart is considered as a permanent failure
and thus, if an existing motr client restarts its existing
fid is updated by Hare. A motr process comprises of motr
services associated to it. Interested motr modules subscribe
to change in the service states, also, motr confc (configuration
cache) maintains a hierarchy of processes and services, thus
if a process fid changes without updating the service fids,
it will not be possible to map a given process to its corresponding
services. Thus, It is important to update the services fids
along with its respective process fid.

Solution:
- Allocate service fids along with its corresponding process fid.
- Make fid allocation routine generic to work for any motr
  configuration object.

Signed-off-by: Mandar Sawant <[email protected]>

Author: Mandar Sawant

hax/hax/motr/__init__.py

@@ -199,7 +199,7 @@ def send_entrypoint_request_reply(self, message: EntrypointRequest):
             util = self.consul_util
             # Disabling dynamic fids allocation until dtm is ready to consume.
             # if util.is_proc_client(process_fid) and message.is_first_request:
-            #     util.alloc_next_process_fid(process_fid)
+            #     util.update_process_fid(process_fid)
 
             # When stopping, there's a possibility that hax may receive
             # an entrypoint request from motr land. In order to unblock
@@ -327,13 +327,9 @@ def _proc_fids_to_eps(proc_skip_list: List[Fid]) -> List[str]:
             # fid.
             # if (st.fid.container == ObjT.PROCESS.value and
             #         self.consul_util.is_proc_client(st.fid)):
-            #     proc_full_fid = self.consul_util.get_process_full_fid(st.fid)
+            #      proc_full_fid = self.consul_util.get_obj_full_fid(st.fid)
             #     st.fid = proc_full_fid
             note = HaNoteStruct(st.fid.to_c(), st.status.to_ha_note_status())
-            # if (st.fid.container == ObjT.PROCESS.value and
-            #         self.consul_util.is_proc_client(st.fid)):
-            #     proc_full_fid = self.consul_util.get_process_full_fid(st.fid)
-            #     st.fid = proc_full_fid
             notes.append(note)
 
             # For process failure, we report failure for the corresponding
@@ -531,7 +527,8 @@ def _generate_sub_services(self,
         LOG.debug('Process fid=%s encloses %s services as follows: %s', fid,
                   len(service_list), service_list)
         service_notes = [
-            HaNoteStruct(no_id=x.fid.to_c(), no_state=new_state)
+            HaNoteStruct(no_id=cns.get_obj_full_fid(x.fid).to_c(),
+                         no_state=new_state)
             for x in service_list
         ]
         if notify_devices:

hax/hax/types.py

@@ -206,7 +206,8 @@ def for_json(self):
 
 ObjTMaskMap = {
     # here for object key - higher 32 bits is masked for dynamic part.
-    ObjT.PROCESS: Fid(QW_F, DW_F)
+    ObjT.PROCESS: Fid(QW_F, DW_F),
+    ObjT.SERVICE: Fid(QW_F, DW_F)
 }
 
 

hax/hax/util.py

@@ -37,9 +37,9 @@
 from hax.common import HaxGlobalState
 from hax.exception import HAConsistencyException, InterruptedException
 from hax.types import (ByteCountStats, ConfHaProcess, Fid, FsStatsWithTime,
-                       ObjT, ObjHealth, ObjTMaskMap, Profile, PverInfo,
-                       PverState, m0HaProcessEvent, m0HaProcessType,
-                       KeyDelete, HaNoteStruct, m0HaObjState)
+                       FidTypeToObjT, ObjT, ObjHealth, ObjTMaskMap,
+                       Profile, PverInfo, PverState, m0HaProcessEvent,
+                       m0HaProcessType, KeyDelete, HaNoteStruct, m0HaObjState)
 
 from hax.consul.cache import (uses_consul_cache, invalidates_consul_cache,
                               supports_consul_cache)
@@ -1728,7 +1728,7 @@ def get_process_status(self,
                            fid: Fid,
                            proc_node=None,
                            kv_cache=None) -> MotrConsulProcInfo:
-        proc_base_fid = self.get_process_base_fid(fid)
+        proc_base_fid = self.get_base_fid(fid)
         key = f'processes/{proc_base_fid}'
         status = self.kv.kv_get(key, kv_cache=kv_cache, allow_null=True)
         if status:
@@ -1742,7 +1742,7 @@ def get_process_status_local(self,
                                  fid: Fid,
                                  proc_node=None,
                                  kv_cache=None) -> MotrConsulProcInfo:
-        proc_base_fid = self.get_process_base_fid(fid)
+        proc_base_fid = self.get_base_fid(fid)
         this_node = self.get_local_nodename()
         key = f'{this_node}/processes/{proc_base_fid}'
         status = self.kv.kv_get(key, kv_cache=kv_cache, allow_null=True)
@@ -1753,12 +1753,12 @@ def get_process_status_local(self,
             return MotrConsulProcInfo('Unknown', 'Unknown')
 
     @repeat_if_fails()
-    def get_process_full_fid(self, proc_base_fid: Fid) -> Optional[Fid]:
-        proc_fid = self.kv.kv_get(str(proc_base_fid), recurse=False)
-        if proc_fid is not None:
-            pfid: Fid = Fid.parse(json.loads(proc_fid['Value']))
-            return pfid
-        return proc_base_fid
+    def get_obj_full_fid(self, base_fid: Fid) -> Optional[Fid]:
+        full_fid = self.kv.kv_get(str(base_fid), recurse=False)
+        if full_fid is not None:
+            fid: Fid = Fid.parse(json.loads(full_fid['Value']))
+            return fid
+        return base_fid
 
     def is_proc_local(self, pfid: Fid) -> bool:
         local_node = self.get_local_nodename()
@@ -1924,7 +1924,7 @@ def get_service_health(self,
     @uses_consul_cache
     def get_process_node(self, proc_fid: Fid, kv_cache=None) -> str:
         try:
-            proc_base_fid = self.get_process_base_fid(proc_fid)
+            proc_base_fid = self.get_base_fid(proc_fid)
             fidk = proc_base_fid.key
             # 'node/<node_name>/process/<process_fidk>/service/type'
             # node_items = self.kv.kv_get('m0conf/nodes',
@@ -2171,7 +2171,7 @@ def set_process_state(self,
             ObjHealth.STOPPED: 'stopped',
             ObjHealth.RECOVERING: 'dtm_recovering'
         }
-        proc_base_fid = self.get_process_base_fid(process_fid)
+        proc_base_fid = self.get_base_fid(process_fid)
 
         # Example key is as follows
         # m0conf/nodes/0x6e00000000000001:0x3/processes/0x7200000000000001:
@@ -2281,62 +2281,75 @@ def set_motr_processes_status(self, fid, status, bAdd=False):
                       motr_processes_status)
 
     @repeat_if_fails()
-    def process_dynamic_fidk_lock(self) -> bool:
+    def obj_dynamic_fidk_lock(self, objt: ObjT) -> bool:
         # Acquire lock to update last_updated_base_fidk.
         # This will block until lock is acquired.
         # Will break if any other exception than
         # HAConsistencyException occurs.
         try:
-            while not self.kv.kv_put('fidk_update_lock', 'true', cas=0):
+            while not self.kv.kv_put(f'{objt.name}/fidk_update_lock',
+                                     'true', cas=0):
                 sleep(1)
             return True
         except Exception:
             return False
 
     @repeat_if_fails()
-    def process_dynamic_fidk_unlock(self):
+    def obj_dynamic_fidk_unlock(self, objt: ObjT):
         # Release fidk_update_lock.
         # This will block until released.
         try:
             keys: List[KeyDelete] = [
-                KeyDelete(name='fidk_update_lock', recurse=True),
+                KeyDelete(name=f'{objt.name}/fidk_update_lock', recurse=True),
             ]
             while not self.kv.kv_delete_in_transaction(keys):
                 sleep(1)
         except Exception:
             raise RuntimeError('Unreachable')
 
     @repeat_if_fails()
-    def get_process_next_dynamic_fidk_lock(self) -> int:
-        if self.process_dynamic_fidk_lock():
-            fidk = self.kv.kv_get('last_dynamic_fid_key/process',
+    def get_obj_next_dynamic_fidk_lock(self, objt: ObjT) -> int:
+        if self.obj_dynamic_fidk_lock(objt):
+            fidk = self.kv.kv_get(f'last_dynamic_fid_key/{objt.name.lower()}',
                                   recurse=False)
             new_fidk = int(json.loads(fidk['Value'])) + 1
             # Update dynamic fid key.
-            while not self.kv.kv_put('last_dynamic_fid_key/process',
-                                     json.dumps(str(new_fidk))):
+            while not self.kv.kv_put(
+                          f'last_dynamic_fid_key/{objt.name.lower()}',
+                          json.dumps(str(new_fidk))):
                 sleep(1)
-            self.process_dynamic_fidk_unlock()
+            self.obj_dynamic_fidk_unlock(objt)
         return new_fidk
 
     @repeat_if_fails()
-    def alloc_next_process_fid(self, process_fid: Fid) -> Fid:
-        next_fidk = self.get_process_next_dynamic_fidk_lock()
-        fid_mask: Fid = ObjTMaskMap[ObjT.PROCESS]
-        fid_cont = process_fid.container
-        fid_key = process_fid.key + ((fid_mask.key * next_fidk) + next_fidk)
-        new_proc_fid = Fid(fid_cont, fid_key)
-        base_fid = self.get_process_base_fid(new_proc_fid)
+    def alloc_next_obj_fid(self, obj_fid: Fid) -> Fid:
+        objt: ObjT = FidTypeToObjT[obj_fid.container]
+        next_fidk = self.get_obj_next_dynamic_fidk_lock(objt)
+        fid_mask: Fid = ObjTMaskMap[objt]
+        fid_cont = obj_fid.container
+        fid_key = obj_fid.key + ((fid_mask.key * next_fidk) + next_fidk)
+        new_obj_fid = Fid(fid_cont, fid_key)
+        base_fid = self.get_base_fid(new_obj_fid)
         # Save base fid to actualy fid mapping in Consul.
         while not self.kv.kv_put(f'{base_fid}',
-                                 json.dumps(str(new_proc_fid))):
+                                 json.dumps(str(new_obj_fid))):
             sleep(1)
-        return new_proc_fid
-
-    def get_process_base_fid(self, proc_fid: Fid) -> Fid:
-        fid_mask: Fid = ObjTMaskMap[ObjT.PROCESS]
-        base_fid = Fid(proc_fid.container,
-                       (proc_fid.key & fid_mask.key))
+        return new_obj_fid
+
+    def update_process_fid(self, proc_fid: Fid) -> List[Fid]:
+        new_fids: List[Fid] = []
+        # First allocate new process fid.
+        new_fids.append(self.alloc_next_obj_fid(proc_fid))
+        service_list = self.get_services_by_parent_process(proc_fid)
+        for svc in service_list:
+            new_fids.append(self.alloc_next_obj_fid(svc.fid))
+        return new_fids
+
+    def get_base_fid(self, obj_fid: Fid) -> Fid:
+        objt: ObjT = FidTypeToObjT[obj_fid.container]
+        fid_mask: Fid = ObjTMaskMap[objt]
+        base_fid = Fid(obj_fid.container,
+                       (obj_fid.key & fid_mask.key))
         return base_fid
 
     @repeat_if_fails()

hax/test/integration/test_motr.py

@@ -808,6 +808,13 @@ def my_get(key: str, recurse: bool = False, **kwds):
                     "name": "localhost",
                     "state": "M0_NC_UNKNOWN"
                 }))
+        elif key == '0x7300000000000001:0x10':
+            return new_kv('0x7300000000000001:0x10',
+                          json.dumps('0x7300000000000001:0x10'))
+        elif key == '0x7300000000000001:0x11':
+            return new_kv('0x7300000000000001:0x11',
+                          json.dumps('0x7300000000000001:0x11'))
+
         raise RuntimeError(f'Unexpected call: key={key}, recurse={recurse}')
 
     return my_get

hax/test/test_failure.py

@@ -104,6 +104,7 @@ def test_process_failure(self):
         consul_util.get_node_encl_fid = Mock(return_value=encl_fid)
         consul_util.get_node_ctrl_fids = Mock(return_value=[ctrl_fid])
         consul_util.get_ioservice_ctrl_fid = Mock(return_value=ctrl_fid)
+        consul_util.get_obj_full_fid = Mock(return_value=service_fid)
 
         # These failure indications are here to trigger specific code paths for
         # node failure. Additional tests can cover different scenarios (e.g.