Merge "NSX Policy: reinitialize connection cluster"

gbpservice/neutron/services/grouppolicy/drivers/vmware/nsx_policy/nsx_policy_mapping.py

@@ -13,6 +13,10 @@
 from oslo_config import cfg
 from oslo_log import log as logging
 
+from neutron_lib.callbacks import events
+from neutron_lib.callbacks import registry
+from neutron_lib.callbacks import resources
+
 from vmware_nsx.db import db as nsx_db
 from vmware_nsx.plugins.nsx_v3 import utils as nsx_utils
 
@@ -131,20 +135,28 @@ class NsxPolicyMappingDriver(api.ResourceMappingDriver):
 
         return v3.NsxPolicyLib(nsxlib_config)
 
-    def get_nsxmanager_client(self):
+    def get_nsxmanager_lib(self):
         """Prepare agent for NSX Manager API calls"""
-        nsxlib = nsx_utils.get_nsxlib_wrapper()
-
-        return nsxlib.client
+        return nsx_utils.get_nsxlib_wrapper()
 
     def initialize(self):
         super(NsxPolicyMappingDriver, self).initialize()
         self._gbp_plugin = None
         self.nsx_policy = self.get_nsxpolicy_lib()
+        # reinitialize the cluster upon fork for api workers to ensure each
+        # process has its own keepalive loops + state
+        registry.subscribe(
+            self.nsx_policy.reinitialize_cluster,
+            resources.PROCESS, events.AFTER_INIT)
         self.policy_api = self.nsx_policy.policy_api
 
-        nsx_manager_client = self.get_nsxmanager_client()
-        self.nsx_port = nsx_resources.LogicalPort(nsx_manager_client)
+        self.nsx_manager = self.get_nsxmanager_lib()
+        registry.subscribe(
+            self.nsx_manager.reinitialize_cluster,
+            resources.PROCESS, events.AFTER_INIT)
+
+        self.nsx_port = nsx_resources.LogicalPort(self.nsx_manager.client)
+
         self._verify_enforcement_point()
 
         # TODO(annak): add validation for core plugin (can only be nsxv3)
@@ -408,6 +420,7 @@ class NsxPolicyMappingDriver(api.ResourceMappingDriver):
         # Translate neutron port id to nsx port id
         _net_id, nsx_port_id = nsx_db.get_nsx_switch_and_port_id(
                  context._plugin_context.session, port_id)
+
         self.nsx_port.update(nsx_port_id, None,
                              tags_update=[{'scope': 'gbp',
                                            'tag': tag}])