<commit_before># Copyright 2015 Mirantis, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import signal
import sys
import time

from oslo_config import cfg
from oslo_log import log as logging
from oslo_service import loopingcall
import oslo_messaging

from neutron import context
from neutron.agent import rpc as agent_rpc
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.agent.common import polling
# from neutron.agent.linux import ip_lib
from neutron.common import config as common_config
from neutron.common import constants as n_const
from neutron.common import utils
from neutron.common import topics
from neutron.i18n import _LE, _LI
from neutron.plugins.common import constants

from mech_vmware_dvs import exceptions
from mech_vmware_dvs import util
from mech_vmware_dvs import constants as dvs_const
from mech_vmware_dvs.agentDVS import agentAPI

LOG = logging.getLogger(__name__)
cfg.CONF.import_group('AGENT', 'mech_vmware_dvs.agentDVS.vmware_conf')


class DVSPluginApi(agent_rpc.PluginApi):
    pass


class DVSAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin, agentAPI.ExtendAPI):

    target = oslo_messaging.Target(version='1.2')

    def __init__(self, vsphere_hostname, vsphere_login, vsphere_password,
                 bridge_mappings, polling_interval, veth_mtu=None,
                 minimize_polling=False, quitting_rpc_timeout=None):
        super(DVSAgent, self).__init__()
        self.veth_mtu = veth_mtu

        self.agent_state = {
            'binary': 'neutron-dvs-agent',
            'host': cfg.CONF.host,
            'topic': n_const.L2_AGENT_TOPIC,
            'configurations': {'bridge_mappings': bridge_mappings,
                               'vsphere_hostname': vsphere_hostname,
                               'log_agent_heartbeats':
                                   cfg.CONF.AGENT.log_agent_heartbeats},
            'agent_type': 'DVS agent',
            'start_flag': True}

        self.setup_rpc()
        report_interval = cfg.CONF.AGENT.report_interval
        if report_interval:
            heartbeat = loopingcall.FixedIntervalLoopingCall(
                self._report_state)
            heartbeat.start(interval=report_interval)

        self.polling_interval = polling_interval
        self.minimize_polling = minimize_polling
        # Security group agent support
        self.sg_agent = sg_rpc.SecurityGroupAgentRpc(
            self.context, self.sg_plugin_rpc, defer_refresh_firewall=True)
        self.run_daemon_loop = True
        self.iter_num = 0
        self.fullsync = True
        # The initialization is complete; we can start receiving messages
        self.connection.consume_in_threads()

        self.quitting_rpc_timeout = quitting_rpc_timeout
        self.network_map = util.create_network_map_from_config(
            cfg.CONF.ML2_VMWARE)
        self.updated_ports = set()
        self.deleted_ports = set()
        self.known_ports = set()
        dvs_ports = self._get_dvs_ports()
        self.added_ports = dvs_ports - self.known_ports
        self.booked_ports = set()

    @util.wrap_retry
    def create_network_precommit(self, current, segment):
        try:
            dvs = self._lookup_dvs_for_context(segment)
        except (exceptions.NoDVSForPhysicalNetwork,
                exceptions.NotSupportedNetworkType) as e:
            LOG.info(_LI('Network %(id)s not created. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.InvalidNetwork:
            pass
        else:
            dvs.create_network(current, segment)

    @util.wrap_retry
    def delete_network_postcommit(self, current, segment):
        try:
            dvs = self._lookup_dvs_for_context(segment)
        except (exceptions.NoDVSForPhysicalNetwork,
                exceptions.NotSupportedNetworkType) as e:
            LOG.info(_LI('Network %(id)s not deleted. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.InvalidNetwork:
            pass
        else:
            dvs.delete_network(current)

    @util.wrap_retry
    def update_network_precommit(self, current, segment, original):
        try:
            dvs = self._lookup_dvs_for_context(segment)
        except (exceptions.NoDVSForPhysicalNetwork,
                exceptions.NotSupportedNetworkType) as e:
            LOG.info(_LI('Network %(id)s not updated. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.InvalidNetwork:
            pass
        else:
            dvs.update_network(current, original)

    @util.wrap_retry
    def book_port(self, current, network_segments, network_current):
        physnet = network_current['provider:physical_network']
        dvs = None
        dvs_segment = None
        for segment in network_segments:
            if segment['physical_network'] == physnet:
                dvs = self._lookup_dvs_for_context(segment)
                dvs_segment = segment
        if dvs:
            port = dvs.book_port(network_current, current['id'], dvs_segment)
            self.booked_ports.add(current['id'])
            return port
        return None

    @util.wrap_retry
    def update_port_postcommit(self, current, original, segment, sg_info):
        try:
            dvs = self._lookup_dvs_for_context(segment)
            if current['id'] in self.booked_ports:
                self.added_ports.add(current['id'])
                self.booked_ports.discard(current['id'])
        except exceptions.NotSupportedNetworkType as e:
            LOG.info(_LI('Port %(id)s not updated. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.NoDVSForPhysicalNetwork:
            raise exceptions.InvalidSystemState(details=_(
                'Port %(port_id)s belong to VMWare VM, but there is '
                'no mapping from network to DVS.') % {'port_id': current['id']}
            )
        else:
            self._update_admin_state_up(dvs, original, current)
            # TODO SlOPS: update security groups on girect call

    @util.wrap_retry
    def delete_port_postcommit(self, current, original, segment, sg_info):
        try:
            dvs = self._lookup_dvs_for_context(segment)
        except exceptions.NotSupportedNetworkType as e:
            LOG.info(_LI('Port %(id)s not deleted. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.NoDVSForPhysicalNetwork:
            raise exceptions.InvalidSystemState(details=_(
                'Port %(port_id)s belong to VMWare VM, but there is '
                'no mapping from network to DVS.') % {'port_id': current['id']}
            )
        else:
            # TODO SlOPS: update security groups on girect call
            dvs.release_port(current)

    def _lookup_dvs_for_context(self, segment):
        if segment['network_type'] == constants.TYPE_VLAN:
            physical_network = segment['physical_network']
            try:
                return self.network_map[physical_network]
            except KeyError:
                LOG.debug('No dvs mapped for physical '
                          'network: %s' % physical_network)
                raise exceptions.NoDVSForPhysicalNetwork(
                    physical_network=physical_network)
        else:
            raise exceptions.NotSupportedNetworkType(
                network_type=segment['network_type'])

    def _update_admin_state_up(self, dvs, original, current):
        try:
            original_admin_state_up = original['admin_state_up']
        except KeyError:
            pass
        else:
            current_admin_state_up = current['admin_state_up']
            perform = current_admin_state_up != original_admin_state_up
            if perform:
                dvs.switch_port_blocked_state(current)

    def _report_state(self):
        try:
            agent_status = self.state_rpc.report_state(self.context,
                                                       self.agent_state,
                                                       True)
            if agent_status == n_const.AGENT_REVIVED:
                LOG.info(_LI('Agent has just revived. Do a full sync.'))
                self.fullsync = True
            self.agent_state.pop('start_flag', None)
        except Exception:
            LOG.exception(_LE("Failed reporting state!"))

    def setup_rpc(self):
        self.agent_id = 'dvs-agent-%s' % cfg.CONF.host
        self.topic = topics.AGENT
        self.plugin_rpc = DVSPluginApi(topics.PLUGIN)
        # self.agentside_rpc = ServerAPI()
        self.sg_plugin_rpc = sg_rpc.SecurityGroupServerRpcApi(topics.PLUGIN)
        self.state_rpc = agent_rpc.PluginReportStateAPI(topics.REPORTS)

        # RPC network init
        self.context = context.get_admin_context_without_session()
        # Handle updates from service
        self.endpoints = [self]
        # Define the listening consumers for the agent
        consumers = [[topics.PORT, topics.UPDATE],
                     [topics.PORT, topics.DELETE],
                     [topics.NETWORK, topics.DELETE],
                     [topics.SECURITY_GROUP, topics.UPDATE],
                     [dvs_const.DVS, topics.UPDATE]]
        self.connection = agent_rpc.create_consumers(self.endpoints,
                                                     self.topic,
                                                     consumers,
                                                     start_listening=False)

    def _handle_sigterm(self, signum, frame):
        LOG.debug("Agent caught SIGTERM, quitting daemon loop.")
        self.run_daemon_loop = False

    def daemon_loop(self):
        with polling.get_polling_manager() as pm:
            self.rpc_loop(polling_manager=pm)

    def rpc_loop(self, polling_manager=None):
        if not polling_manager:
            polling_manager = polling.get_polling_manager(
                minimize_polling=False)
        while self.run_daemon_loop:
            start = time.time()
            port_stats = {'regular': {'added': 0,
                                      'updated': 0,
                                      'removed': 0}}
            if self.fullsync:
                LOG.info(_LI("Agent out of sync with plugin!"))
                self.fullsync = False
                polling_manager.force_polling()
            if self._agent_has_updates(polling_manager):
                LOG.debug("Agent rpc_loop - update")
                self.process_ports()
                port_stats['regular']['added'] = len(self.added_ports)
                port_stats['regular']['updated'] = len(self.updated_ports)
                port_stats['regular']['removed'] = len(self.deleted_ports)<commit_msg>Remove trash ports from vsphere when agent starts<commit_after># Copyright 2015 Mirantis, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import signal
import sys
import time
import itertools

from oslo_config import cfg
from oslo_log import log as logging
from oslo_service import loopingcall
import oslo_messaging

from neutron import context
from neutron.agent import rpc as agent_rpc
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.agent.common import polling
# from neutron.agent.linux import ip_lib
from neutron.common import config as common_config
from neutron.common import constants as n_const
from neutron.common import utils
from neutron.common import topics
from neutron.i18n import _LE, _LI
from neutron.plugins.common import constants

from mech_vmware_dvs import exceptions
from mech_vmware_dvs import util
from mech_vmware_dvs import constants as dvs_const
from mech_vmware_dvs.agentDVS import agentAPI

LOG = logging.getLogger(__name__)
cfg.CONF.import_group('AGENT', 'mech_vmware_dvs.agentDVS.vmware_conf')


class DVSPluginApi(agent_rpc.PluginApi):
    pass


class DVSAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin, agentAPI.ExtendAPI):

    target = oslo_messaging.Target(version='1.2')

    def __init__(self, vsphere_hostname, vsphere_login, vsphere_password,
                 bridge_mappings, polling_interval, veth_mtu=None,
                 minimize_polling=False, quitting_rpc_timeout=None):
        super(DVSAgent, self).__init__()
        self.veth_mtu = veth_mtu

        self.agent_state = {
            'binary': 'neutron-dvs-agent',
            'host': cfg.CONF.host,
            'topic': n_const.L2_AGENT_TOPIC,
            'configurations': {'bridge_mappings': bridge_mappings,
                               'vsphere_hostname': vsphere_hostname,
                               'log_agent_heartbeats':
                                   cfg.CONF.AGENT.log_agent_heartbeats},
            'agent_type': 'DVS agent',
            'start_flag': True}

        self.setup_rpc()
        report_interval = cfg.CONF.AGENT.report_interval
        if report_interval:
            heartbeat = loopingcall.FixedIntervalLoopingCall(
                self._report_state)
            heartbeat.start(interval=report_interval)

        self.polling_interval = polling_interval
        self.minimize_polling = minimize_polling
        # Security group agent support
        self.sg_agent = sg_rpc.SecurityGroupAgentRpc(
            self.context, self.sg_plugin_rpc, defer_refresh_firewall=True)
        self.run_daemon_loop = True
        self.iter_num = 0
        self.fullsync = True
        # The initialization is complete; we can start receiving messages
        self.connection.consume_in_threads()

        self.quitting_rpc_timeout = quitting_rpc_timeout
        self.network_map = util.create_network_map_from_config(
            cfg.CONF.ML2_VMWARE)
        self.updated_ports = set()
        self.deleted_ports = set()
        self.known_ports = set()
        self.added_ports = set()
        self.booked_ports = set()

    @util.wrap_retry
    def create_network_precommit(self, current, segment):
        try:
            dvs = self._lookup_dvs_for_context(segment)
        except (exceptions.NoDVSForPhysicalNetwork,
                exceptions.NotSupportedNetworkType) as e:
            LOG.info(_LI('Network %(id)s not created. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.InvalidNetwork:
            pass
        else:
            dvs.create_network(current, segment)

    @util.wrap_retry
    def delete_network_postcommit(self, current, segment):
        try:
            dvs = self._lookup_dvs_for_context(segment)
        except (exceptions.NoDVSForPhysicalNetwork,
                exceptions.NotSupportedNetworkType) as e:
            LOG.info(_LI('Network %(id)s not deleted. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.InvalidNetwork:
            pass
        else:
            dvs.delete_network(current)

    @util.wrap_retry
    def update_network_precommit(self, current, segment, original):
        try:
            dvs = self._lookup_dvs_for_context(segment)
        except (exceptions.NoDVSForPhysicalNetwork,
                exceptions.NotSupportedNetworkType) as e:
            LOG.info(_LI('Network %(id)s not updated. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.InvalidNetwork:
            pass
        else:
            dvs.update_network(current, original)

    @util.wrap_retry
    def book_port(self, current, network_segments, network_current):
        physnet = network_current['provider:physical_network']
        dvs = None
        dvs_segment = None
        for segment in network_segments:
            if segment['physical_network'] == physnet:
                dvs = self._lookup_dvs_for_context(segment)
                dvs_segment = segment
        if dvs:
            port = dvs.book_port(network_current, current['id'], dvs_segment)
            self.booked_ports.add(current['id'])
            return port
        return None

    @util.wrap_retry
    def update_port_postcommit(self, current, original, segment, sg_info):
        try:
            dvs = self._lookup_dvs_for_context(segment)
            if current['id'] in self.booked_ports:
                self.added_ports.add(current['id'])
                self.booked_ports.discard(current['id'])
        except exceptions.NotSupportedNetworkType as e:
            LOG.info(_LI('Port %(id)s not updated. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.NoDVSForPhysicalNetwork:
            raise exceptions.InvalidSystemState(details=_(
                'Port %(port_id)s belong to VMWare VM, but there is '
                'no mapping from network to DVS.') % {'port_id': current['id']}
            )
        else:
            self._update_admin_state_up(dvs, original, current)
            # TODO SlOPS: update security groups on girect call

    @util.wrap_retry
    def delete_port_postcommit(self, current, original, segment, sg_info):
        try:
            dvs = self._lookup_dvs_for_context(segment)
        except exceptions.NotSupportedNetworkType as e:
            LOG.info(_LI('Port %(id)s not deleted. Reason: %(reason)s') % {
                'id': current['id'],
                'reason': e.message})
        except exceptions.NoDVSForPhysicalNetwork:
            raise exceptions.InvalidSystemState(details=_(
                'Port %(port_id)s belong to VMWare VM, but there is '
                'no mapping from network to DVS.') % {'port_id': current['id']}
            )
        else:
            # TODO SlOPS: update security groups on girect call
            dvs.release_port(current)

    def _lookup_dvs_for_context(self, segment):
        if segment['network_type'] == constants.TYPE_VLAN:
            physical_network = segment['physical_network']
            try:
                return self.network_map[physical_network]
            except KeyError:
                LOG.debug('No dvs mapped for physical '
                          'network: %s' % physical_network)
                raise exceptions.NoDVSForPhysicalNetwork(
                    physical_network=physical_network)
        else:
            raise exceptions.NotSupportedNetworkType(
                network_type=segment['network_type'])

    def _update_admin_state_up(self, dvs, original, current):
        try:
            original_admin_state_up = original['admin_state_up']
        except KeyError:
            pass
        else:
            current_admin_state_up = current['admin_state_up']
            perform = current_admin_state_up != original_admin_state_up
            if perform:
                dvs.switch_port_blocked_state(current)

    def _report_state(self):
        try:
            agent_status = self.state_rpc.report_state(self.context,
                                                       self.agent_state,
                                                       True)
            if agent_status == n_const.AGENT_REVIVED:
                LOG.info(_LI('Agent has just revived. Do a full sync.'))
                self.fullsync = True
            self.agent_state.pop('start_flag', None)
        except Exception:
            LOG.exception(_LE("Failed reporting state!"))

    def setup_rpc(self):
        self.agent_id = 'dvs-agent-%s' % cfg.CONF.host
        self.topic = topics.AGENT
        self.plugin_rpc = DVSPluginApi(topics.PLUGIN)
        # self.agentside_rpc = ServerAPI()
        self.sg_plugin_rpc = sg_rpc.SecurityGroupServerRpcApi(topics.PLUGIN)
        self.state_rpc = agent_rpc.PluginReportStateAPI(topics.REPORTS)

        # RPC network init
        self.context = context.get_admin_context_without_session()
        # Handle updates from service
        self.endpoints = [self]
        # Define the listening consumers for the agent
        consumers = [[topics.PORT, topics.UPDATE],
                     [topics.PORT, topics.DELETE],
                     [topics.NETWORK, topics.DELETE],
                     [topics.SECURITY_GROUP, topics.UPDATE],
                     [dvs_const.DVS, topics.UPDATE]]
        self.connection = agent_rpc.create_consumers(self.endpoints,
                                                     self.topic,
                                                     consumers,
                                                     start_listening=False)

    def _handle_sigterm(self, signum, frame):
        LOG.debug("Agent caught SIGTERM, quitting daemon loop.")
        self.run_daemon_loop = False

    def _clean_up_vsphere_extra_ports(self, connected_ports):
        LOG.debug("Cleanup vsphere extra ports...")
        vsphere_not_connected_ports_maps = {}
        for phys_net, dvs in self.network_map.items():
            all_dvs_ports = set([p.config.name for p in dvs.get_ports(False)])
            not_connected_dvs_ports = all_dvs_ports - connected_ports
            vsphere_not_connected_ports_maps.update(
                {port_id: phys_net for port_id in not_connected_dvs_ports})

        if not vsphere_not_connected_ports_maps:
            return

        devices_details_list = (
            self.plugin_rpc.get_devices_details_list_and_failed_devices(
                self.context,
                vsphere_not_connected_ports_maps.keys(),
                self.agent_id,
                cfg.CONF.host))
        neutron_ports = set([
            p['port_id'] for p in itertools.chain(
                devices_details_list['devices'],
                devices_details_list['failed_devices']) if p.get('port_id')
        ])

        for port_id, phys_net in vsphere_not_connected_ports_maps.items():
            if port_id not in neutron_ports:
                dvs = self.network_map[phys_net]
                dvs.release_port({'id': port_id})

    def daemon_loop(self):
        with polling.get_polling_manager() as pm:
            self.rpc_loop(polling_manager=pm)

    def rpc_loop(self, polling_manager=None):
        if not polling_manager:
            polling_manager = polling.get_polling_manager(
                minimize_polling=False)
        while self.run_daemon_loop:
            start = time.time()
            port_stats = {'regular': {'added': 0,
                                      'updated': 0,
                                      'removed': 0}}
            if self.fullsync:
                LOG.info(_LI("Agent out of sync with plugin!"))
                connected_ports = self._get_dvs_ports()
                self.added_ports = connected_ports - self.known_ports
                self._clean_up_vsphere_extra_ports(connected_ports)
                self.fullsync = False
                polling_manager.force_polling()
            if self._agent_has_updates(polling_manager):
                LOG.debug("Agent rpc_loop - update")
                self.process_ports()
                port_stats['regular']['added'] = len(self.added_ports)
                port_stats['regular']['updated'] = len(self.updated_ports)
                port_stats['regular']['removed'] = len(self.deleted_ports)
{'fraction_non_alphanumeric': 0.0842468284271462, 'fraction_numerical': 0.0014780145338095824, 'mean_word_length': 2.1312508034451727, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '924921', 'n_tokens_mistral': 6845, 'n_tokens_neox': 6618, 'n_words': 1487}