<s> from neutron_lib . api import validators <EOL> from neutron_lib import exceptions as n_exc <EOL> from oslo_config import cfg <EOL> from oslo_log import log as logging <EOL> import sqlalchemy as sa <EOL> from sqlalchemy import orm <EOL> from neutron . _i18n import _ , _LE <EOL> from neutron . common import utils <EOL> from neutron . db import db_base_plugin_v2 <EOL> from neutron . db import l3_db <EOL> from neutron . db import model_base <EOL> from neutron . db import models_v2 <EOL> from neutron . extensions import dns <EOL> from neutron . extensions import l3 <EOL> from neutron import manager <EOL> from neutron . plugins . common import constants as service_constants <EOL> from neutron . services . externaldns import driver <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class NetworkDNSDomain ( model_base . BASEV2 ) : <EOL> network_id = sa . Column ( sa . String ( <NUM_LIT> ) , <EOL> sa . ForeignKey ( '<STR_LIT>' , ondelete = "<STR_LIT>" ) , <EOL> primary_key = True , <EOL> index = True ) <EOL> dns_domain = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> network = orm . relationship ( models_v2 . Network , <EOL> backref = orm . backref ( "<STR_LIT>" , <EOL> lazy = '<STR_LIT>' , <EOL> uselist = False , <EOL> cascade = '<STR_LIT>' ) ) <EOL> class FloatingIPDNS ( model_base . BASEV2 ) : <EOL> __tablename__ = '<STR_LIT>' <EOL> floatingip_id = sa . Column ( sa . String ( <NUM_LIT> ) , <EOL> sa . ForeignKey ( '<STR_LIT>' , <EOL> ondelete = "<STR_LIT>" ) , <EOL> primary_key = True , <EOL> index = True ) <EOL> dns_name = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> dns_domain = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> published_dns_name = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> published_dns_domain = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> floatingip = orm . relationship ( l3_db . FloatingIP , <EOL> backref = orm . backref ( "<STR_LIT>" , <EOL> lazy = '<STR_LIT>' , <EOL> uselist = False , <EOL> cascade = '<STR_LIT>' ) ) <EOL> class PortDNS ( model_base . BASEV2 ) : <EOL> __tablename__ = '<STR_LIT>' <EOL> port_id = sa . Column ( sa . String ( <NUM_LIT> ) , <EOL> sa . ForeignKey ( '<STR_LIT>' , <EOL> ondelete = "<STR_LIT>" ) , <EOL> primary_key = True , <EOL> index = True ) <EOL> current_dns_name = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> current_dns_domain = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> previous_dns_name = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> previous_dns_domain = sa . Column ( sa . String ( <NUM_LIT:255> ) , <EOL> nullable = False ) <EOL> port = orm . relationship ( models_v2 . Port , <EOL> backref = orm . backref ( "<STR_LIT>" , <EOL> lazy = '<STR_LIT>' , <EOL> uselist = False , <EOL> cascade = '<STR_LIT>' ) ) <EOL> class DNSActionsData ( object ) : <EOL> def __init__ ( self , current_dns_name = None , current_dns_domain = None , <EOL> previous_dns_name = None , previous_dns_domain = None ) : <EOL> self . current_dns_name = current_dns_name <EOL> self . current_dns_domain = current_dns_domain <EOL> self . previous_dns_name = previous_dns_name <EOL> self . previous_dns_domain = previous_dns_domain <EOL> class DNSDbMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> _dns_driver = None <EOL> @ property <EOL> def dns_driver ( self ) : <EOL> if self . _dns_driver : <EOL> return self . _dns_driver <EOL> if not cfg . CONF . external_dns_driver : <EOL> return <EOL> try : <EOL> self . _dns_driver = driver . ExternalDNSService . get_instance ( ) <EOL> LOG . debug ( "<STR_LIT>" , <EOL> cfg . CONF . external_dns_driver ) <EOL> return self . _dns_driver <EOL> except ImportError : <EOL> LOG . exception ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> raise dns . ExternalDNSDriverNotFound ( <EOL> driver = cfg . CONF . external_dns_driver ) <EOL> def _extend_floatingip_dict_dns ( self , floatingip_res , floatingip_db ) : <EOL> floatingip_res [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> floatingip_res [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if floatingip_db . dns : <EOL> floatingip_res [ '<STR_LIT>' ] = floatingip_db . dns [ '<STR_LIT>' ] <EOL> floatingip_res [ '<STR_LIT>' ] = floatingip_db . dns [ '<STR_LIT>' ] <EOL> return floatingip_res <EOL> db_base_plugin_v2 . NeutronDbPluginV2 . register_dict_extend_funcs ( <EOL> l3 . FLOATINGIPS , [ '<STR_LIT>' ] ) <EOL> def _process_dns_floatingip_create_precommit ( self , context , <EOL> floatingip_data , req_data ) : <EOL> dns_domain = req_data . get ( dns . DNSDOMAIN ) <EOL> if not validators . is_attr_set ( dns_domain ) : <EOL> return <EOL> if not self . dns_driver : <EOL> return <EOL> dns_name = req_data [ dns . DNSNAME ] <EOL> self . _validate_floatingip_dns ( dns_name , dns_domain ) <EOL> current_dns_name , current_dns_domain = ( <EOL> self . _get_requested_state_for_external_dns_service_create ( <EOL> context , floatingip_data , req_data ) ) <EOL> dns_actions_data = None <EOL> if current_dns_name and current_dns_domain : <EOL> context . session . add ( FloatingIPDNS ( <EOL> floatingip_id = floatingip_data [ '<STR_LIT:id>' ] , <EOL> dns_name = req_data [ dns . DNSNAME ] , <EOL> dns_domain = req_data [ dns . DNSDOMAIN ] , <EOL> published_dns_name = current_dns_name , <EOL> published_dns_domain = current_dns_domain ) ) <EOL> dns_actions_data = DNSActionsData ( <EOL> current_dns_name = current_dns_name , <EOL> current_dns_domain = current_dns_domain ) <EOL> floatingip_data [ '<STR_LIT>' ] = dns_name <EOL> floatingip_data [ '<STR_LIT>' ] = dns_domain <EOL> return dns_actions_data <EOL> def _process_dns_floatingip_create_postcommit ( self , context , <EOL> floatingip_data , <EOL> dns_actions_data ) : <EOL> if not dns_actions_data : <EOL> return <EOL> self . _add_ips_to_external_dns_service ( <EOL> context , dns_actions_data . current_dns_domain , <EOL> dns_actions_data . current_dns_name , <EOL> [ floatingip_data [ '<STR_LIT>' ] ] ) <EOL> def _process_dns_floatingip_update_precommit ( self , context , <EOL> floatingip_data ) : <EOL> plugin = manager . NeutronManager . get_service_plugins ( ) . get ( <EOL> service_constants . L3_ROUTER_NAT ) <EOL> if not utils . is_extension_supported ( plugin , dns . Dns . get_alias ( ) ) : <EOL> return <EOL> if not self . dns_driver : <EOL> return <EOL> dns_data_db = context . session . query ( FloatingIPDNS ) . filter_by ( <EOL> floatingip_id = floatingip_data [ '<STR_LIT:id>' ] ) . one_or_none ( ) <EOL> if dns_data_db and dns_data_db [ '<STR_LIT>' ] : <EOL> return <EOL> current_dns_name , current_dns_domain = ( <EOL> self . _get_requested_state_for_external_dns_service_update ( <EOL> context , floatingip_data ) ) <EOL> if dns_data_db : <EOL> if ( dns_data_db [ '<STR_LIT>' ] != current_dns_name or <EOL> dns_data_db [ '<STR_LIT>' ] != current_dns_domain ) : <EOL> dns_actions_data = DNSActionsData ( <EOL> previous_dns_name = dns_data_db [ '<STR_LIT>' ] , <EOL> previous_dns_domain = dns_data_db [ '<STR_LIT>' ] ) <EOL> if current_dns_name and current_dns_domain : <EOL> dns_data_db [ '<STR_LIT>' ] = current_dns_name <EOL> dns_data_db [ '<STR_LIT>' ] = current_dns_domain <EOL> dns_actions_data . current_dns_name = current_dns_name <EOL> dns_actions_data . current_dns_domain = current_dns_domain <EOL> else : <EOL> context . session . delete ( dns_data_db ) <EOL> return dns_actions_data <EOL> else : <EOL> return <EOL> if current_dns_name and current_dns_domain : <EOL> context . session . add ( FloatingIPDNS ( <EOL> floatingip_id = floatingip_data [ '<STR_LIT:id>' ] , <EOL> dns_name = '<STR_LIT>' , <EOL> dns_domain = '<STR_LIT>' , <EOL> published_dns_name = current_dns_name , <EOL> published_dns_domain = current_dns_domain ) ) <EOL> return DNSActionsData ( current_dns_name = current_dns_name , <EOL> current_dns_domain = current_dns_domain ) <EOL> def _process_dns_floatingip_update_postcommit ( self , context , <EOL> floatingip_data , <EOL> dns_actions_data ) : <EOL> if not dns_actions_data : <EOL> return <EOL> if dns_actions_data . previous_dns_name : <EOL> self . _delete_floatingip_from_external_dns_service ( <EOL> context , dns_actions_data . previous_dns_domain , <EOL> dns_actions_data . previous_dns_name , <EOL> [ floatingip_data [ '<STR_LIT>' ] ] ) <EOL> if dns_actions_data . current_dns_name : <EOL> self . _add_ips_to_external_dns_service ( <EOL> context , dns_actions_data . current_dns_domain , <EOL> dns_actions_data . current_dns_name , <EOL> [ floatingip_data [ '<STR_LIT>' ] ] ) <EOL> def _process_dns_floatingip_delete ( self , context , floatingip_data ) : <EOL> plugin = manager . NeutronManager . get_service_plugins ( ) . get ( <EOL> service_constants . L3_ROUTER_NAT ) <EOL> if not utils . is_extension_supported ( plugin , dns . Dns . get_alias ( ) ) : <EOL> return <EOL> dns_data_db = context . session . query ( FloatingIPDNS ) . filter_by ( <EOL> floatingip_id = floatingip_data [ '<STR_LIT:id>' ] ) . one_or_none ( ) <EOL> if dns_data_db : <EOL> self . _delete_floatingip_from_external_dns_service ( <EOL> context , dns_data_db [ '<STR_LIT>' ] , <EOL> dns_data_db [ '<STR_LIT>' ] , <EOL> [ floatingip_data [ '<STR_LIT>' ] ] ) <EOL> def _validate_floatingip_dns ( self , dns_name , dns_domain ) : <EOL> if dns_domain and not dns_name : <EOL> msg = _ ( "<STR_LIT>" ) <EOL> raise n_exc . BadRequest ( resource = '<STR_LIT>' , msg = msg ) <EOL> if dns_name and not dns_domain : <EOL> msg = _ ( "<STR_LIT>" ) <EOL> raise n_exc . BadRequest ( resource = '<STR_LIT>' , msg = msg ) <EOL> def _get_internal_port_dns_data ( self , context , floatingip_data ) : <EOL> internal_port = context . session . query ( models_v2 . Port ) . filter_by ( <EOL> id = floatingip_data [ '<STR_LIT>' ] ) . one ( ) <EOL> dns_domain = None <EOL> if internal_port [ '<STR_LIT>' ] : <EOL> net_dns = context . session . query ( NetworkDNSDomain ) . filter_by ( <EOL> network_id = internal_port [ '<STR_LIT>' ] ) . one_or_none ( ) <EOL> if net_dns : <EOL> dns_domain = net_dns [ '<STR_LIT>' ] <EOL> return internal_port [ '<STR_LIT>' ] , dns_domain <EOL> def _delete_floatingip_from_external_dns_service ( self , context , dns_domain , <EOL> dns_name , records ) : <EOL> try : <EOL> self . dns_driver . delete_record_set ( context , dns_domain , dns_name , <EOL> records ) <EOL> except ( dns . DNSDomainNotFound , dns . DuplicateRecordSet ) as e : <EOL> LOG . exception ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> % { "<STR_LIT:name>" : dns_name , <EOL> "<STR_LIT>" : dns_domain , <EOL> "<STR_LIT:message>" : e . msg , <EOL> "<STR_LIT>" : '<STR_LIT:U+002CU+0020>' . join ( records ) } ) <EOL> def _get_requested_state_for_external_dns_service_create ( self , context , <EOL> floatingip_data , <EOL> req_data ) : <EOL> fip_dns_name = req_data [ dns . DNSNAME ] <EOL> if fip_dns_name : <EOL> return fip_dns_name , req_data [ dns . DNSDOMAIN ] <EOL> if floatingip_data [ '<STR_LIT>' ] : <EOL> return self . _get_internal_port_dns_data ( context , floatingip_data ) <EOL> return None , None <EOL> def _get_requested_state_for_external_dns_service_update ( self , context , <EOL> floatingip_data ) : <EOL> if floatingip_data [ '<STR_LIT>' ] : <EOL> return self . _get_internal_port_dns_data ( context , floatingip_data ) <EOL> return None , None <EOL> def _add_ips_to_external_dns_service ( self , context , dns_domain , dns_name , <EOL> records ) : <EOL> try : <EOL> self . dns_driver . create_record_set ( context , dns_domain , dns_name , <EOL> records ) <EOL> except ( dns . DNSDomainNotFound , dns . DuplicateRecordSet ) as e : <EOL> LOG . exception ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> % { "<STR_LIT:name>" : dns_name , <EOL> "<STR_LIT>" : dns_domain , <EOL> "<STR_LIT:message>" : e . msg } ) </s>
