<commit_before># -*- coding: utf-8 -*-
#
# Copyright (c) 2015 Metaswitch Networks
# 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.

# Etcd-based transport for the Calico/OpenStack Plugin.

# Standard Python library imports.
import etcd
import eventlet
import eventlet.event
import json
import re
import weakref

# OpenStack imports.
from oslo.config import cfg
from neutron.openstack.common import log

# Calico imports.
from calico.datamodel_v1 import (READY_KEY, CONFIG_DIR, TAGS_KEY_RE, HOST_DIR,
                                 key_for_endpoint, PROFILE_DIR,
                                 key_for_profile, key_for_profile_rules,
                                 key_for_profile_tags, key_for_config)

# Register Calico-specific options.
calico_opts = [
    cfg.StrOpt('etcd_host', default='localhost',
               help="The hostname or IP of the etcd node/proxy"),
    cfg.IntOpt('etcd_port', default=4001,
               help="The port to use for the etcd node/proxy"),
]
cfg.CONF.register_opts(calico_opts, 'calico')

OPENSTACK_ENDPOINT_RE = re.compile(
    r'^' + HOST_DIR +
    r'/(?P<hostname>[^/]+)/.*openstack.*/endpoint/(?P<endpoint_id>[^/]+)')

json_decoder = json.JSONDecoder()

PERIODIC_RESYNC_INTERVAL_SECS = 30

LOG = log.getLogger(__name__)


class CalicoTransportEtcd(object):
    """Calico transport implementation based on etcd."""

    def __init__(self, driver):
        # Explicitly store the driver as a weakreference. This prevents
        # the reference loop between transport and driver keeping the objects
        # alive.
        self.driver = weakref.proxy(driver)

        # Prepare client for accessing etcd data.
        self.client = etcd.Client(host=cfg.CONF.calico.etcd_host,
                                  port=cfg.CONF.calico.etcd_port)

        # Spawn a green thread for periodically resynchronizing etcd against
        # the OpenStack database.
        eventlet.spawn(self.periodic_resync_thread)

    def periodic_resync_thread(self):
        while True:
            LOG.info("Calico plugin doing periodic resync.")
            try:
                # Write non-default config that Felices need.
                self.provide_felix_config()

                # Resynchronize endpoint data.
                self.resync_endpoints()

                # Resynchronize security group data.
                self.resync_security_groups()

            except:
                LOG.exception("Exception in periodic resync thread")

            # Sleep until time for next resync.
            LOG.info("Calico plugin finished periodic resync.  "
                     "Next resync in %s seconds.",
                     PERIODIC_RESYNC_INTERVAL_SECS)
            eventlet.sleep(PERIODIC_RESYNC_INTERVAL_SECS)

    def resync_endpoints(self):
        pass

    def resync_security_groups(self):
        pass


    def write_profile_to_etcd(self, profile_id):
        self.client.write(key_for_profile_rules(profile_id),
                          json.dumps(self.profile_rules(profile_id)))
        self.client.write(key_for_profile_tags(profile_id),
                          json.dumps(self.profile_tags(profile_id)))

    def profile_rules(self, profile_id):
        inbound = []
        outbound = []
        for sgid in self.profile_tags(profile_id):
            # Be tolerant of a security group not being here. Allow up to 20
            # attempts to get it, waiting a few hundred ms in between: we might
            # just be racing slightly ahead of a security group update.
            rules = None
            retries = 20
            while rules is None:
                try:
                    with self._sgs_semaphore:
                        rules = self.sgs[sgid]['security_group_rules']
                except KeyError:
                    LOG.warning("Missing info for SG %s: waiting.", sgid)
                    retries -= 1

                    if not retries:
                        LOG.error("Gave up waiting for SG %s", sgid)
                        raise

                    # Wait for 200ms
                    eventlet.sleep(0.2)


            for rule in rules:
                LOG.info("Neutron rule  %s : %s", profile_id, rule)
                etcd_rule = _neutron_rule_to_etcd_rule(rule)
                if rule['direction'] == 'ingress':
                    inbound.append(etcd_rule)
                else:
                    outbound.append(etcd_rule)

        return {'inbound_rules': inbound, 'outbound_rules': outbound}

    def profile_tags(self, profile_id):
        return profile_id.split('_')

    def endpoint_created(self, port, profile):
        """
        Write appropriate data to etcd for an endpoint creation event.
        """
        # First, write etcd data for the new endpoint.
        # TODO: Write this function.
        self.write_port_to_etcd(port)

        # Next, write the security profile.
        # TODO: Fix this function to do the right thing.
        self.write_profile_to_etcd(profile)

    def endpoint_updated(self, port, profile):
        """
        Write data to etcd for an endpoint updated event.
        """
        # Do the same as for endpoint_created.
        self.endpoint_created(port, profile)

    def endpoint_deleted(self, port):
        """
        Delete data from etcd for an endpoint deleted event.
        """
        # TODO: What do we do about profiles here?
        # Delete the etcd key for this endpoint.
        key = self.port_etcd_key(port)
        try:
            self.client.delete(key)
        except etcd.EtcdKeyNotFound:
            # Already gone, treat as success.
            LOG.debug("Key %s, which we were deleting, disappeared", key)

    def security_group_updated(self, sg):
        # Update the data that we're keeping for this security group.
        with self._sgs_semaphore:
            self.sgs[sg['id']] = sg

        # Identify all the needed profiles that incorporate this security
        # group, and rewrite their data.
        # Take the profile lock so that no-one can modify needed_profiles
        # while we iterate over it.  (This is probably unneeded since there
        # aren't any yield points in the loop but better safe than sorry.)
        profiles_to_rewrite = set()
        with self.profile_semaphore:
            for profile_id in self.needed_profiles:
                if sg['id'] in self.profile_tags(profile_id):
                    # Write etcd data for this profile.
                    profiles_to_rewrite.add(profile_id)
        for profile_id in profiles_to_rewrite:
            self.write_profile_to_etcd(profile_id)

    def provide_felix_config(self):
        """Specify the prefix of the TAP interfaces that Felix should
        look for and work with.  This config setting does not have a
        default value, because different cloud systems will do
        different things.  Here we provide the prefix that Neutron
        uses.
        """
        # First read the config values, so as to avoid unnecessary
        # writes.
        prefix = None
        ready = None
        iface_pfx_key = key_for_config('InterfacePrefix')
        try:
            prefix = self.client.read(iface_pfx_key).value
            ready = self.client.read(READY_KEY).value
        except etcd.EtcdKeyNotFound:
            LOG.info('%s values are missing', CONFIG_DIR)

        # Now write the values that need writing.
        if prefix != 'tap':
            LOG.info('%s -> tap', iface_pfx_key)<commit_msg>Write port to etcd<commit_after># -*- coding: utf-8 -*-
#
# Copyright (c) 2015 Metaswitch Networks
# 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.

# Etcd-based transport for the Calico/OpenStack Plugin.

# Standard Python library imports.
import etcd
import eventlet
import eventlet.event
import json
import re
import weakref

# OpenStack imports.
from oslo.config import cfg
from neutron.openstack.common import log

# Calico imports.
from calico.datamodel_v1 import (READY_KEY, CONFIG_DIR, TAGS_KEY_RE, HOST_DIR,
                                 key_for_endpoint, PROFILE_DIR,
                                 key_for_profile, key_for_profile_rules,
                                 key_for_profile_tags, key_for_config)

# Register Calico-specific options.
calico_opts = [
    cfg.StrOpt('etcd_host', default='localhost',
               help="The hostname or IP of the etcd node/proxy"),
    cfg.IntOpt('etcd_port', default=4001,
               help="The port to use for the etcd node/proxy"),
]
cfg.CONF.register_opts(calico_opts, 'calico')

OPENSTACK_ENDPOINT_RE = re.compile(
    r'^' + HOST_DIR +
    r'/(?P<hostname>[^/]+)/.*openstack.*/endpoint/(?P<endpoint_id>[^/]+)')

json_decoder = json.JSONDecoder()

PERIODIC_RESYNC_INTERVAL_SECS = 30

LOG = log.getLogger(__name__)


class CalicoTransportEtcd(object):
    """Calico transport implementation based on etcd."""

    def __init__(self, driver):
        # Explicitly store the driver as a weakreference. This prevents
        # the reference loop between transport and driver keeping the objects
        # alive.
        self.driver = weakref.proxy(driver)

        # Prepare client for accessing etcd data.
        self.client = etcd.Client(host=cfg.CONF.calico.etcd_host,
                                  port=cfg.CONF.calico.etcd_port)

        # Spawn a green thread for periodically resynchronizing etcd against
        # the OpenStack database.
        eventlet.spawn(self.periodic_resync_thread)

    def periodic_resync_thread(self):
        while True:
            LOG.info("Calico plugin doing periodic resync.")
            try:
                # Write non-default config that Felices need.
                self.provide_felix_config()

                # Resynchronize endpoint data.
                self.resync_endpoints()

                # Resynchronize security group data.
                self.resync_security_groups()

            except:
                LOG.exception("Exception in periodic resync thread")

            # Sleep until time for next resync.
            LOG.info("Calico plugin finished periodic resync.  "
                     "Next resync in %s seconds.",
                     PERIODIC_RESYNC_INTERVAL_SECS)
            eventlet.sleep(PERIODIC_RESYNC_INTERVAL_SECS)

    def resync_endpoints(self):
        pass

    def resync_security_groups(self):
        pass


    def write_profile_to_etcd(self, profile_id):
        self.client.write(key_for_profile_rules(profile_id),
                          json.dumps(self.profile_rules(profile_id)))
        self.client.write(key_for_profile_tags(profile_id),
                          json.dumps(self.profile_tags(profile_id)))

    def profile_rules(self, profile_id):
        inbound = []
        outbound = []
        for sgid in self.profile_tags(profile_id):
            # Be tolerant of a security group not being here. Allow up to 20
            # attempts to get it, waiting a few hundred ms in between: we might
            # just be racing slightly ahead of a security group update.
            rules = None
            retries = 20
            while rules is None:
                try:
                    with self._sgs_semaphore:
                        rules = self.sgs[sgid]['security_group_rules']
                except KeyError:
                    LOG.warning("Missing info for SG %s: waiting.", sgid)
                    retries -= 1

                    if not retries:
                        LOG.error("Gave up waiting for SG %s", sgid)
                        raise

                    # Wait for 200ms
                    eventlet.sleep(0.2)


            for rule in rules:
                LOG.info("Neutron rule  %s : %s", profile_id, rule)
                etcd_rule = _neutron_rule_to_etcd_rule(rule)
                if rule['direction'] == 'ingress':
                    inbound.append(etcd_rule)
                else:
                    outbound.append(etcd_rule)

        return {'inbound_rules': inbound, 'outbound_rules': outbound}

    def profile_tags(self, profile_id):
        return profile_id.split('_')

    def endpoint_created(self, port, profile):
        """
        Write appropriate data to etcd for an endpoint creation event.
        """
        # First, write etcd data for the new endpoint.
        # TODO: Write this function.
        self.write_port_to_etcd(port, profile.id)

        # Next, write the security profile.
        # TODO: Fix this function to do the right thing.
        self.write_profile_to_etcd(profile)

    def endpoint_updated(self, port, profile):
        """
        Write data to etcd for an endpoint updated event.
        """
        # Do the same as for endpoint_created.
        self.endpoint_created(port, profile)

    def endpoint_deleted(self, port):
        """
        Delete data from etcd for an endpoint deleted event.
        """
        # TODO: What do we do about profiles here?
        # Delete the etcd key for this endpoint.
        key = self.port_etcd_key(port)
        try:
            self.client.delete(key)
        except etcd.EtcdKeyNotFound:
            # Already gone, treat as success.
            LOG.debug("Key %s, which we were deleting, disappeared", key)

    def security_group_updated(self, sg):
        # Update the data that we're keeping for this security group.
        with self._sgs_semaphore:
            self.sgs[sg['id']] = sg

        # Identify all the needed profiles that incorporate this security
        # group, and rewrite their data.
        # Take the profile lock so that no-one can modify needed_profiles
        # while we iterate over it.  (This is probably unneeded since there
        # aren't any yield points in the loop but better safe than sorry.)
        profiles_to_rewrite = set()
        with self.profile_semaphore:
            for profile_id in self.needed_profiles:
                if sg['id'] in self.profile_tags(profile_id):
                    # Write etcd data for this profile.
                    profiles_to_rewrite.add(profile_id)
        for profile_id in profiles_to_rewrite:
            self.write_profile_to_etcd(profile_id)

    def write_port_to_etcd(self, port, profile_id):
        """
        Writes a given port dictionary to etcd.
        """
        data = port_etcd_data(port, profile_id)
        self.client.write(port_etcd_key(port), json.dumps(data))


    def provide_felix_config(self):
        """Specify the prefix of the TAP interfaces that Felix should
        look for and work with.  This config setting does not have a
        default value, because different cloud systems will do
        different things.  Here we provide the prefix that Neutron
        uses.
        """
        # First read the config values, so as to avoid unnecessary
        # writes.
        prefix = None
        ready = None
        iface_pfx_key = key_for_config('InterfacePrefix')
        try:
            prefix = self.client.read(iface_pfx_key).value
            ready = self.client.read(READY_KEY).value
        except etcd.EtcdKeyNotFound:
            LOG.info('%s values are missing', CONFIG_DIR)

        # Now write the values that need writing.
        if prefix != 'tap':
            LOG.info('%s -> tap', iface_pfx_key)
{'fraction_non_alphanumeric': 0.07673449374846776, 'fraction_numerical': 0.003187055650894827, 'mean_word_length': 2.0630749014454666, 'pattern_counts': {'":': 0, '<': 7, '<?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': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3364676', 'n_tokens_mistral': 4598, 'n_tokens_neox': 4394, 'n_words': 1387}