Mitaka 21st march base latency optimized

gbpservice/neutron/services/servicechain/plugins/ncp/node_drivers/nfp_node_driver.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 
 import eventlet
+from eventlet import greenpool
 from keystoneclient import exceptions as k_exceptions
 from keystoneclient.v2_0 import client as keyclient
 from neutron._i18n import _LE
@@ -38,6 +39,7 @@
 from gbpservice.nfp.common import constants as nfp_constants
 from gbpservice.nfp.common import topics as nfp_rpc_topics
 
+from gbpservice.neutron.services.grouppolicy.common import constants as gp_constants
 
 NFP_NODE_DRIVER_OPTS = [
     cfg.BoolOpt('is_service_admin_owned',
@@ -59,7 +61,6 @@
 
 LOG = logging.getLogger(__name__)
 
-
 class InvalidServiceType(exc.NodeCompositionPluginBadRequest):
     message = _("The NFP Node driver only supports the services "
                 "VPN, Firewall and LB in a Service Chain")
@@ -227,6 +228,9 @@ class NFPNodeDriver(driver_base.NodeDriverBase):
     def __init__(self):
         super(NFPNodeDriver, self).__init__()
         self._lbaas_plugin = None
+        self.thread_pool = greenpool.GreenPool(10)
+        self.active_threads = []
+        self.sc_node_count = 0
 
     @property
     def name(self):
@@ -336,8 +340,18 @@ def create(self, context):
         self._set_node_instance_network_function_map(
             context.plugin_session, context.current_node['id'],
             context.instance['id'], network_function_id)
-        self._wait_for_network_function_operation_completion(
-            context, network_function_id, operation='create')
+
+        # Check for NF status in a separate thread
+        gth = self.thread_pool.spawn(self._wait_for_network_function_operation_completion,
+                context, network_function_id, operation='create')
+
+        self.active_threads.append(gth)
+
+        # At last wait for the threads to complete, success/failure/timeout
+        if len(self.active_threads) == self.sc_node_count:
+            for gth in self.active_threads:
+                gth.wait()
+            self.active_threads = []
 
     def update(self, context):
         context._plugin_context = self._get_resource_owner_context(
@@ -591,23 +605,49 @@ def _get_service_targets(self, context):
                           {'service_type': service_type})
                 raise Exception("Service Targets are not created for the Node")
 
-        service_target_info = {'provider_ports': [], 'provider_pts': [],
-                               'consumer_ports': [], 'consumer_pts': []}
+        service_target_info = {
+            'provider_ports': [],
+            'provider_subnet': None,
+            'provider_pts': [],
+            'provider_pt_objs': [],
+            'provider_ptg': [],
+            'consumer_ports': [],
+            'consumer_subnet': None,
+            'consumer_pts': [],
+            'consumer_pt_objs': [],
+            'consumer_ptg': []}
+
         for service_target in provider_service_targets:
             policy_target = context.gbp_plugin.get_policy_target(
                 context.plugin_context, service_target.policy_target_id)
+            policy_target_group = context.gbp_plugin.get_policy_target_group(
+                context.plugin_context, policy_target['policy_target_group_id'])
             port = context.core_plugin.get_port(
                 context.plugin_context, policy_target['port_id'])
+            port['ip_address'] = port['fixed_ips'][0]['ip_address']
+            subnet = context.core_plugin.get_subnet(
+                context.plugin_context, port['fixed_ips'][0]['subnet_id'])
             service_target_info['provider_ports'].append(port)
+            service_target_info['provider_subnet'] = subnet
             service_target_info['provider_pts'].append(policy_target['id'])
+            service_target_info['provider_pt_objs'].append(policy_target)
+            service_target_info['provider_ptg'].append(policy_target_group)
 
         for service_target in consumer_service_targets:
             policy_target = context.gbp_plugin.get_policy_target(
                 context.plugin_context, service_target.policy_target_id)
+            policy_target_group = context.gbp_plugin.get_policy_target_group(
+                context.plugin_context, policy_target['policy_target_group_id'])
             port = context.core_plugin.get_port(
                 context.plugin_context, policy_target['port_id'])
+            port['ip_address'] = port['fixed_ips'][0]['ip_address']
+            subnet = context.core_plugin.get_subnet(
+                context.plugin_context, port['fixed_ips'][0]['subnet_id'])
             service_target_info['consumer_ports'].append(port)
+            service_target_info['consumer_subnet'] = subnet
             service_target_info['consumer_pts'].append(policy_target['id'])
+            service_target_info['consumer_pt_objs'].append(policy_target)
+            service_target_info['consumer_ptg'].append(policy_target_group)
 
         return service_target_info
 
@@ -619,6 +659,7 @@ def _is_node_order_in_spec_supported(self, context):
         for spec in current_specs:
             node_list.extend(spec['nodes'])
 
+        self.sc_node_count = len(node_list)
         for node_id in node_list:
             node_info = context.sc_plugin.get_servicechain_node(
                 context.plugin_context, node_id)
@@ -641,9 +682,65 @@ def _is_node_order_in_spec_supported(self, context):
             raise InvalidNodeOrderInChain(
                     node_order=allowed_chain_combinations)
 
+    def _get_consumers_for_provider(self, context, provider):
+        '''
+        {
+            consuming_ptgs_details: [{'ptg': <>, 'subnets': <>}]
+            consuming_eps_details: []
+        }
+        '''
+
+        consuming_ptgs_details = []
+        consuming_eps_details = []
+
+        provided_prs_id = provider['provided_policy_rule_sets'][0]
+        provided_prs = context.gbp_plugin.get_policy_rule_set(
+            context.plugin_context, provided_prs_id)
+        consuming_ptg_ids = provided_prs['consuming_policy_target_groups']
+        consuming_ep_ids = provided_prs['consuming_external_policies']
+
+        consuming_ptgs = context.gbp_plugin.get_policy_target_groups(
+                context.plugin_context, filters={'id':consuming_ptg_ids})
+        consuming_eps_details = context.gbp_plugin.get_external_policies(
+                context.plugin_context, filters={'id': consuming_ep_ids})
+
+        for ptg in consuming_ptgs:
+            subnet_ids = ptg['subnets']
+            subnets = context.core_plugin.get_subnets(context.plugin_context, filters={'id':subnet_ids})
+            consuming_ptgs_details.append({'ptg':ptg, 'subnets':subnets})
+
+        return consuming_ptgs_details, consuming_eps_details
+
+
     def _create_network_function(self, context):
+        """
+        nfp_create_nf_data :-
+
+        {'resource_owner_context': <>,
+         'service_chain_instance': <>,
+         'service_chain_node': <>,
+         'service_profile': <>,
+         'service_config': context.current_node.get('config'),
+         'provider': {'pt':<>, 'ptg':<>, 'port':<>, 'subnet':<>},
+         'consumer': {'pt':<>, 'ptg':<>, 'port':<>, 'subnet':<>},
+         'management': {'pt':<>, 'ptg':<>, 'port':<>, 'subnet':<>},
+         'management_ptg_id': <>,
+         'network_function_mode': nfp_constants.GBP_MODE,
+         'tenant_id': <>,
+         'consuming_ptgs_details': [],
+         'consuming_eps_details': []
+        }
+
+        """
+        nfp_create_nf_data = {}
+
         sc_instance = context.instance
         service_targets = self._get_service_targets(context)
+
+        consuming_ptgs_details, consuming_eps_details = \
+            self._get_consumers_for_provider(context,
+                service_targets['provider_ptg'][0])
+
         if context.current_profile['service_type'] == pconst.LOADBALANCER:
             config_param_values = sc_instance.get('config_param_values', {})
             if config_param_values:
@@ -661,35 +758,58 @@ def _create_network_function(self, context):
                 context.core_plugin.update_port(
                     context.plugin_context, provider_port['id'], port)
 
-        port_info = []
-        if service_targets.get('provider_pts'):
-            # Device case, for Base mode ports won't be available.
-            port_info = [
-                {
-                    'id': service_targets['provider_pts'][0],
-                    'port_model': nfp_constants.GBP_PORT,
-                    'port_classification': nfp_constants.PROVIDER,
-                }
-            ]
-        if service_targets.get('consumer_ports'):
-            port_info.append({
-                'id': service_targets['consumer_pts'][0],
-                'port_model': nfp_constants.GBP_PORT,
-                'port_classification': nfp_constants.CONSUMER,
-            })
-        network_function = {
-            'tenant_id': context.provider['tenant_id'],
-            'service_chain_id': sc_instance['id'],
-            'service_id': context.current_node['id'],
-            'service_profile_id': context.current_profile['id'],
-            'management_ptg_id': sc_instance['management_ptg_id'],
+        provider = {
+            'pt': service_targets.get('provider_pt_objs', [None])[0],
+            'ptg': service_targets.get('provider_ptg', [None])[0],
+            'port': service_targets.get('provider_ports', [None])[0],
+            'subnet': service_targets.get('provider_subnet', None),
+            'port_model': nfp_constants.GBP_PORT,
+            'port_classification': nfp_constants.PROVIDER}
+
+        consumer_pt = None
+        consumer_ptg = None
+        consumer_ports = None
+
+        if service_targets['consumer_pt_objs']:
+            consumer_pt = service_targets.get('consumer_pt_objs', [None])[0]
+        if service_targets['consumer_ptg']:
+            consumer_ptg = service_targets.get('consumer_ptg', [None])[0]
+        if service_targets['consumer_ports']:
+            consumer_ports = service_targets.get('consumer_ports', [None])[0]
+
+        consumer = {
+            'pt': consumer_pt,
+            'ptg': consumer_ptg,
+            'port': consumer_ports,
+            'subnet': service_targets.get('consumer_subnet', None),
+            'port_model': nfp_constants.GBP_PORT,
+            'port_classification': nfp_constants.CONSUMER}
+
+        management = {
+            'pt': None,
+            'ptg': None,
+            'port': None,
+            'subnet': None,
+            'port_model': nfp_constants.GBP_NETWORK,
+            'port_classification': nfp_constants.MANAGEMENT}
+
+        nfp_create_nf_data = {
+            'resource_owner_context': context._plugin_context.to_dict(),
+            'service_chain_instance': sc_instance,
+            'service_chain_node': context.current_node,
+            'service_profile': context.current_profile,
             'service_config': context.current_node.get('config'),
-            'port_info': port_info,
+            'provider': provider,
+            'consumer': consumer,
+            'management': management,
+            'management_ptg_id': sc_instance['management_ptg_id'],
             'network_function_mode': nfp_constants.GBP_MODE,
-        }
+            'tenant_id': context.provider['tenant_id'],
+            'consuming_ptgs_details': consuming_ptgs_details,
+            'consuming_eps_details': consuming_eps_details}
 
         return self.nfp_notifier.create_network_function(
-            context.plugin_context, network_function=network_function)['id']
+            context.plugin_context, network_function=nfp_create_nf_data)['id']
 
     def _set_node_instance_network_function_map(
         self, session, sc_node_id, sc_instance_id, network_function_id):

gbpservice/nfp/bin/nfp_configurator.ini

@@ -12,10 +12,10 @@ kombu_reconnect_delay=1.0
 rabbit_use_ssl=False
 
 rabbit_virtual_host=/
-workers=1
+workers=2
 modules_dir=gbpservice.nfp.configurator.modules
 reportstate_interval=10
-periodic_interval=9
+periodic_interval=2
 
 log_forward_ip_address=
 log_forward_port=514

gbpservice/nfp/bin/nfp_orch_agent.ini

@@ -1,5 +1,5 @@
 [DEFAULT]
-workers=1
+workers=2
 debug=False
 kombu_reconnect_delay=1.0
 rabbit_use_ssl=False

gbpservice/nfp/bin/nfp_proxy_agent.ini

@@ -6,4 +6,4 @@ rabbit_use_ssl=False
 rabbit_virtual_host=/
 modules_dir=gbpservice.nfp.proxy_agent.modules
 backend=unix_rest
-periodic_interval=10
+periodic_interval=2

gbpservice/nfp/bin/proxy.ini

@@ -5,7 +5,8 @@ max_connections=10
 rest_server_address= 11.0.0.3
 ##for docker ##
 rest_server_port= 8070
-worker_threads=40
+#[Note: worker threads should not be less than connect_max_wait_timeout/{periodic_interval or spacing for pull_notification}]
+worker_threads=100
 connect_max_wait_timeout=120
 idle_max_wait_timeout=120
 idle_min_wait_timeout=0.1

gbpservice/nfp/config_orchestrator/agent/otc_service_events.py

@@ -67,7 +67,7 @@ def _delete_service(self, context, resource):
                                                "DELETE",
                                                network_function_event=True)
 
-    @core_pt.poll_event_desc(event='SERVICE_CREATE_PENDING', spacing=5)
+    @core_pt.poll_event_desc(event='SERVICE_CREATE_PENDING', spacing=2)
     def create_sevice_pending_event(self, ev):
         event_data = ev.data
         ctxt = n_context.Context.from_dict(event_data['context'])

gbpservice/nfp/configurator/agents/generic_config.py

@@ -21,6 +21,9 @@
 from gbpservice.nfp.core import event as nfp_event
 from gbpservice.nfp.core import poll as nfp_poll
 
+STOP_POLLING = {'poll': False}
+CONTINUE_POLLING = {'poll': True}
+
 LOG = nfp_logging.getLogger(__name__)
 
 """Implements APIs invoked by configurator for processing RPC messages.
@@ -65,6 +68,7 @@ def _send_event(self, context, resource_data, event_id, event_key=None):
         arg_dict = {'context': context,
                     'resource_data': resource_data}
         ev = self.sc.new_event(id=event_id, data=arg_dict, key=event_key)
+
         self.sc.post_event(ev)
 
     def configure_interfaces(self, context, resource_data):
@@ -268,8 +272,9 @@ def _process_event(self, ev):
             if (resource_data.get('periodicity') == gen_cfg_const.INITIAL and
                     result == common_const.SUCCESS):
                 notification_data = self._prepare_notification_data(ev, result)
-                self.sc.poll_event_done(ev)
+                # self.sc.poll_event_done(ev)
                 self.notify._notification(notification_data)
+                return STOP_POLLING
             elif resource_data.get('periodicity') == gen_cfg_const.FOREVER:
                 if result == common_const.FAILED:
                     """If health monitoring fails continuously for 5 times
@@ -282,8 +287,9 @@ def _process_event(self, ev):
                         notification_data = self._prepare_notification_data(
                                                                     ev,
                                                                     result)
-                        self.sc.poll_event_done(ev)
+                        # self.sc.poll_event_done(ev)
                         self.notify._notification(notification_data)
+                        return STOP_POLLING
                 elif result == common_const.SUCCESS:
                     """set fail_count to 0 if it had failed earlier even once
                     """
@@ -293,8 +299,9 @@ def _process_event(self, ev):
                that particular service vm's health monitor
             """
             notification_data = self._prepare_notification_data(ev, result)
-            self.sc.poll_event_done(ev)
+            # self.sc.poll_event_done(ev)
             self.notify._notification(notification_data)
+            return STOP_POLLING
         else:
             """For other events, irrespective of result send notification"""
             notification_data = self._prepare_notification_data(ev, result)
@@ -357,7 +364,7 @@ def poll_event_cancel(self, ev):
 
     @nfp_poll.poll_event_desc(
                             event=gen_cfg_const.EVENT_CONFIGURE_HEALTHMONITOR,
-                            spacing=5)
+                            spacing=2)
     def handle_configure_healthmonitor(self, ev):
         """Decorator method called for poll event CONFIGURE_HEALTHMONITOR
            Finally it Enqueues response into notification queue.
@@ -367,7 +374,7 @@ def handle_configure_healthmonitor(self, ev):
         Returns: None
 
         """
-        self._process_event(ev)
+        return self._process_event(ev)
 
 
 def events_init(sc, drivers, rpcmgr):

gbpservice/nfp/configurator/api/config.py

@@ -48,7 +48,8 @@
         'logfile': {
             'class': 'logging.FileHandler',
             'filename': '/var/log/nfp/nfp_pecan.log',
-            'level': 'INFO'
+            'level': 'INFO',
+            'formatter': 'simple'
         }
     },
     'formatters': {

gbpservice/nfp/configurator/lib/generic_config_constants.py

@@ -22,4 +22,4 @@
 MAX_FAIL_COUNT = 12  # 5 secs delay * 12 = 60 secs
 INITIAL = 'initial'
 FOREVER = 'forever'
-INITIAL_HM_RETRIES = 24  # 5 secs delay * 24 = 120 secs
+INITIAL_HM_RETRIES = 90  # 5 secs delay * 24 = 120 secs