<commit_before># Copyright 2012 OpenStack Foundation
# 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 os
import re

import eventlet
import netaddr
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
import six

from neutron._i18n import _, _LE, _LW
from neutron.agent.common import utils
from neutron.common import exceptions as n_exc
from neutron.common import utils as common_utils
from neutron.privileged.agent.linux import ip_lib as privileged

LOG = logging.getLogger(__name__)

OPTS = [
    cfg.BoolOpt('ip_lib_force_root',
                default=False,
                help=_('Force ip_lib calls to use the root helper')),
]

IP_NONLOCAL_BIND = 'net.ipv4.ip_nonlocal_bind'

LOOPBACK_DEVNAME = 'lo'
GRE_TUNNEL_DEVICE_NAMES = ['gre0', 'gretap0']

SYS_NET_PATH = '/sys/class/net'
DEFAULT_GW_PATTERN = re.compile(r"via (\S+)")
METRIC_PATTERN = re.compile(r"metric (\S+)")
DEVICE_NAME_PATTERN = re.compile(r"(\d+?): (\S+?):.*")


def remove_interface_suffix(interface):
    """Remove a possible "<if>@<endpoint>" suffix from an interface' name.

    This suffix can appear in some kernel versions, and intends on specifying,
    for example, a veth's pair. However, this interface name is useless to us
    as further 'ip' commands require that the suffix be removed.
    """
    # If '@' is not present, this will do nothing.
    return interface.partition("@")[0]


class AddressNotReady(exceptions.NeutronException):
    message = _("Failure waiting for address %(address)s to "
                "become ready: %(reason)s")


class SubProcessBase(object):
    def __init__(self, namespace=None,
                 log_fail_as_error=True):
        self.namespace = namespace
        self.log_fail_as_error = log_fail_as_error
        try:
            self.force_root = cfg.CONF.ip_lib_force_root
        except cfg.NoSuchOptError:
            # Only callers that need to force use of the root helper
            # need to register the option.
            self.force_root = False

    def _run(self, options, command, args):
        if self.namespace:
            return self._as_root(options, command, args)
        elif self.force_root:
            # Force use of the root helper to ensure that commands
            # will execute in dom0 when running under XenServer/XCP.
            return self._execute(options, command, args, run_as_root=True,
                                 log_fail_as_error=self.log_fail_as_error)
        else:
            return self._execute(options, command, args,
                                 log_fail_as_error=self.log_fail_as_error)

    def _as_root(self, options, command, args, use_root_namespace=False):
        namespace = self.namespace if not use_root_namespace else None

        return self._execute(options, command, args, run_as_root=True,
                             namespace=namespace,
                             log_fail_as_error=self.log_fail_as_error)

    @classmethod
    def _execute(cls, options, command, args, run_as_root=False,
                 namespace=None, log_fail_as_error=True):
        opt_list = ['-%s' % o for o in options]
        ip_cmd = add_namespace_to_cmd(['ip'], namespace)
        cmd = ip_cmd + opt_list + [command] + list(args)
        return utils.execute(cmd, run_as_root=run_as_root,
                             log_fail_as_error=log_fail_as_error)

    def set_log_fail_as_error(self, fail_with_error):
        self.log_fail_as_error = fail_with_error

    def get_log_fail_as_error(self):
        return self.log_fail_as_error


class IPWrapper(SubProcessBase):
    def __init__(self, namespace=None):
        super(IPWrapper, self).__init__(namespace=namespace)
        self.netns = IpNetnsCommand(self)

    def device(self, name):
        return IPDevice(name, namespace=self.namespace)

    def get_devices(self, exclude_loopback=False, exclude_gre_devices=False):
        retval = []
        if self.namespace:
            # we call out manually because in order to avoid screen scraping
            # iproute2 we use find to see what is in the sysfs directory, as
            # suggested by <NAME> (iproute2 dev).
            try:
                cmd = ['ip', 'netns', 'exec', self.namespace,
                       'find', SYS_NET_PATH, '-maxdepth', '1',
                       '-type', 'l', '-printf', '%f ']
                output = utils.execute(
                    cmd,
                    run_as_root=True,
                    log_fail_as_error=self.log_fail_as_error).split()
            except RuntimeError:
                # We could be racing with a cron job deleting namespaces.
                # Just return a empty list if the namespace is deleted.
                with excutils.save_and_reraise_exception() as ctx:
                    if not self.netns.exists(self.namespace):
                        ctx.reraise = False
                        return []
        else:
            output = (<commit_msg>Adding debug trace to IPWrapper.get_devices()<commit_after># Copyright 2012 OpenStack Foundation
# 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 os
import re

import eventlet
import netaddr
from neutron_lib import constants
from neutron_lib import exceptions
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
import six

from neutron._i18n import _, _LE, _LW
from neutron.agent.common import utils
from neutron.common import exceptions as n_exc
from neutron.common import utils as common_utils
from neutron.privileged.agent.linux import ip_lib as privileged

LOG = logging.getLogger(__name__)

OPTS = [
    cfg.BoolOpt('ip_lib_force_root',
                default=False,
                help=_('Force ip_lib calls to use the root helper')),
]

IP_NONLOCAL_BIND = 'net.ipv4.ip_nonlocal_bind'

LOOPBACK_DEVNAME = 'lo'
GRE_TUNNEL_DEVICE_NAMES = ['gre0', 'gretap0']

SYS_NET_PATH = '/sys/class/net'
DEFAULT_GW_PATTERN = re.compile(r"via (\S+)")
METRIC_PATTERN = re.compile(r"metric (\S+)")
DEVICE_NAME_PATTERN = re.compile(r"(\d+?): (\S+?):.*")


def remove_interface_suffix(interface):
    """Remove a possible "<if>@<endpoint>" suffix from an interface' name.

    This suffix can appear in some kernel versions, and intends on specifying,
    for example, a veth's pair. However, this interface name is useless to us
    as further 'ip' commands require that the suffix be removed.
    """
    # If '@' is not present, this will do nothing.
    return interface.partition("@")[0]


class AddressNotReady(exceptions.NeutronException):
    message = _("Failure waiting for address %(address)s to "
                "become ready: %(reason)s")


class SubProcessBase(object):
    def __init__(self, namespace=None,
                 log_fail_as_error=True):
        self.namespace = namespace
        self.log_fail_as_error = log_fail_as_error
        try:
            self.force_root = cfg.CONF.ip_lib_force_root
        except cfg.NoSuchOptError:
            # Only callers that need to force use of the root helper
            # need to register the option.
            self.force_root = False

    def _run(self, options, command, args):
        if self.namespace:
            return self._as_root(options, command, args)
        elif self.force_root:
            # Force use of the root helper to ensure that commands
            # will execute in dom0 when running under XenServer/XCP.
            return self._execute(options, command, args, run_as_root=True,
                                 log_fail_as_error=self.log_fail_as_error)
        else:
            return self._execute(options, command, args,
                                 log_fail_as_error=self.log_fail_as_error)

    def _as_root(self, options, command, args, use_root_namespace=False):
        namespace = self.namespace if not use_root_namespace else None

        return self._execute(options, command, args, run_as_root=True,
                             namespace=namespace,
                             log_fail_as_error=self.log_fail_as_error)

    @classmethod
    def _execute(cls, options, command, args, run_as_root=False,
                 namespace=None, log_fail_as_error=True):
        opt_list = ['-%s' % o for o in options]
        ip_cmd = add_namespace_to_cmd(['ip'], namespace)
        cmd = ip_cmd + opt_list + [command] + list(args)
        return utils.execute(cmd, run_as_root=run_as_root,
                             log_fail_as_error=log_fail_as_error)

    def set_log_fail_as_error(self, fail_with_error):
        self.log_fail_as_error = fail_with_error

    def get_log_fail_as_error(self):
        return self.log_fail_as_error


class IPWrapper(SubProcessBase):
    def __init__(self, namespace=None):
        super(IPWrapper, self).__init__(namespace=namespace)
        self.netns = IpNetnsCommand(self)

    def device(self, name):
        return IPDevice(name, namespace=self.namespace)

    def get_devices(self, exclude_loopback=False, exclude_gre_devices=False):
        retval = []
        if self.namespace:
            # we call out manually because in order to avoid screen scraping
            # iproute2 we use find to see what is in the sysfs directory, as
            # suggested by <NAME> (iproute2 dev).
            try:
                cmd = ['ip', 'netns', 'exec', self.namespace,
                       'find', SYS_NET_PATH, '-maxdepth', '1',
                       '-type', 'l', '-printf', '%f ']
                output_str = utils.execute(
                    cmd,
                    run_as_root=True,
                    log_fail_as_error=self.log_fail_as_error)
                # NOTE(dalvarez): Logging the output of this call due to
                # bug1654287.
                LOG.debug('get_devices(): %s', output_str)
                output = output_str.split()
            except RuntimeError:
                # We could be racing with a cron job deleting namespaces.
                # Just return a empty list if the namespace is deleted.
                with excutils.save_and_reraise_exception() as ctx:
                    if not self.netns.exists(self.namespace):
                        ctx.reraise = False
                        return []
        else:
            output = (
{'fraction_non_alphanumeric': 0.08190108733777622, 'fraction_numerical': 0.0037706068046299544, 'mean_word_length': 2.4728989037758833, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1757200', 'n_tokens_mistral': 3295, 'n_tokens_neox': 3191, 'n_words': 981}