-
Notifications
You must be signed in to change notification settings - Fork 1.5k
/
plugin.py
2533 lines (2298 loc) · 122 KB
/
plugin.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Copyright (c) 2013 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.
from eventlet import greenthread
import netaddr
from netaddr.strategy import eui48
from neutron_lib.agent import constants as agent_consts
from neutron_lib.agent import topics
from neutron_lib.api.definitions import address_scope
from neutron_lib.api.definitions import agent as agent_apidef
from neutron_lib.api.definitions import agent_resources_synced
from neutron_lib.api.definitions import allowedaddresspairs as addr_apidef
from neutron_lib.api.definitions import availability_zone as az_def
from neutron_lib.api.definitions import availability_zone_filter
from neutron_lib.api.definitions import default_subnetpools
from neutron_lib.api.definitions import dhcpagentscheduler
from neutron_lib.api.definitions import empty_string_filtering
from neutron_lib.api.definitions import external_net
from neutron_lib.api.definitions import extra_dhcp_opt as edo_ext
from neutron_lib.api.definitions import filter_validation as filter_apidef
from neutron_lib.api.definitions import ip_allocation as ipalloc_apidef
from neutron_lib.api.definitions import ip_substring_port_filtering
from neutron_lib.api.definitions import multiprovidernet
from neutron_lib.api.definitions import network as net_def
from neutron_lib.api.definitions import network_availability_zone
from neutron_lib.api.definitions import network_mtu as mtu_apidef
from neutron_lib.api.definitions import network_mtu_writable as mtuw_apidef
from neutron_lib.api.definitions import port as port_def
from neutron_lib.api.definitions import port_mac_address_regenerate
from neutron_lib.api.definitions import port_security as psec
from neutron_lib.api.definitions import portbindings
from neutron_lib.api.definitions import portbindings_extended as pbe_ext
from neutron_lib.api.definitions import provider_net
from neutron_lib.api.definitions import rbac_security_groups as rbac_sg_apidef
from neutron_lib.api.definitions import security_groups_port_filtering
from neutron_lib.api.definitions import subnet as subnet_def
from neutron_lib.api.definitions import subnet_onboard as subnet_onboard_def
from neutron_lib.api.definitions import vlantransparent as vlan_apidef
from neutron_lib.api import extensions
from neutron_lib.api import validators
from neutron_lib.api.validators import availability_zone as az_validator
from neutron_lib.callbacks import events
from neutron_lib.callbacks import exceptions
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants as const
from neutron_lib.db import api as db_api
from neutron_lib.db import model_query
from neutron_lib.db import resource_extend
from neutron_lib.db import utils as db_utils
from neutron_lib import exceptions as exc
from neutron_lib.exceptions import allowedaddresspairs as addr_exc
from neutron_lib.exceptions import port_security as psec_exc
from neutron_lib.objects import utils as obj_utils
from neutron_lib.plugins import constants as plugin_constants
from neutron_lib.plugins import directory
from neutron_lib.plugins.ml2 import api
from neutron_lib.plugins import utils as p_utils
from neutron_lib import rpc as n_rpc
from neutron_lib.services.qos import constants as qos_consts
from oslo_config import cfg
from oslo_db import exception as os_db_exception
from oslo_log import helpers as log_helpers
from oslo_log import log
from oslo_serialization import jsonutils
from oslo_utils import excutils
from oslo_utils import importutils
from oslo_utils import uuidutils
import sqlalchemy
from sqlalchemy import or_
from sqlalchemy.orm import exc as sa_exc
from neutron._i18n import _
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
from neutron.api.rpc.handlers import dhcp_rpc
from neutron.api.rpc.handlers import dvr_rpc
from neutron.api.rpc.handlers import metadata_rpc
from neutron.api.rpc.handlers import resources_rpc
from neutron.api.rpc.handlers import securitygroups_rpc
from neutron.common import utils
from neutron.db import address_scope_db
from neutron.db import agents_db
from neutron.db import agentschedulers_db
from neutron.db import allowedaddresspairs_db as addr_pair_db
from neutron.db import db_base_plugin_v2
from neutron.db import dvr_mac_db
from neutron.db import external_net_db
from neutron.db import extradhcpopt_db
from neutron.db.models import securitygroup as sg_models
from neutron.db import models_v2
from neutron.db import provisioning_blocks
from neutron.db.quota import driver # noqa
from neutron.db import securitygroups_rpc_base as sg_db_rpc
from neutron.db import segments_db
from neutron.db import subnet_service_type_mixin
from neutron.db import vlantransparent_db
from neutron.extensions import filter_validation
from neutron.extensions import vlantransparent
from neutron.ipam import exceptions as ipam_exc
from neutron.objects import base as base_obj
from neutron.objects import ports as ports_obj
from neutron.plugins.ml2.common import exceptions as ml2_exc
from neutron.plugins.ml2 import db
from neutron.plugins.ml2 import driver_context
from neutron.plugins.ml2.drivers import mech_agent
from neutron.plugins.ml2.extensions import qos as qos_ext
from neutron.plugins.ml2 import managers
from neutron.plugins.ml2 import models
from neutron.plugins.ml2 import ovo_rpc
from neutron.plugins.ml2 import rpc
from neutron.quota import resource_registry
from neutron.services.segments import plugin as segments_plugin
LOG = log.getLogger(__name__)
MAX_BIND_TRIES = 10
SERVICE_PLUGINS_REQUIRED_DRIVERS = {
'qos': [qos_ext.QOS_EXT_DRIVER_ALIAS]
}
def _ml2_port_result_filter_hook(query, filters):
values = filters and filters.get(portbindings.HOST_ID, [])
if not values:
return query
bind_criteria = models.PortBinding.host.in_(values)
return query.filter(models_v2.Port.port_bindings.any(bind_criteria))
@resource_extend.has_resource_extenders
@registry.has_registry_receivers
class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
dvr_mac_db.DVRDbMixin,
external_net_db.External_net_db_mixin,
sg_db_rpc.SecurityGroupServerRpcMixin,
agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
addr_pair_db.AllowedAddressPairsMixin,
vlantransparent_db.Vlantransparent_db_mixin,
extradhcpopt_db.ExtraDhcpOptMixin,
address_scope_db.AddressScopeDbMixin,
subnet_service_type_mixin.SubnetServiceTypeMixin):
"""Implement the Neutron L2 abstractions using modules.
Ml2Plugin is a Neutron plugin based on separately extensible sets
of network types and mechanisms for connecting to networks of
those types. The network types and mechanisms are implemented as
drivers loaded via Python entry points. Networks can be made up of
multiple segments (not yet fully implemented).
"""
# This attribute specifies whether the plugin supports or not
# bulk/pagination/sorting operations. Name mangling is used in
# order to ensure it is qualified by class
__native_bulk_support = True
__native_pagination_support = True
__native_sorting_support = True
# This attribute specifies whether the plugin supports or not
# filter validations. Name mangling is used in
# order to ensure it is qualified by class
__filter_validation_support = True
# List of supported extensions
_supported_extension_aliases = [provider_net.ALIAS,
external_net.ALIAS, portbindings.ALIAS,
"quotas", "security-group",
rbac_sg_apidef.ALIAS,
agent_apidef.ALIAS,
dhcpagentscheduler.ALIAS,
multiprovidernet.ALIAS,
addr_apidef.ALIAS,
edo_ext.ALIAS, "subnet_allocation",
mtu_apidef.ALIAS,
mtuw_apidef.ALIAS,
vlan_apidef.ALIAS,
address_scope.ALIAS,
az_def.ALIAS,
network_availability_zone.ALIAS,
availability_zone_filter.ALIAS,
default_subnetpools.ALIAS,
"subnet-service-types",
ip_substring_port_filtering.ALIAS,
security_groups_port_filtering.ALIAS,
empty_string_filtering.ALIAS,
filter_apidef.ALIAS,
port_mac_address_regenerate.ALIAS,
pbe_ext.ALIAS,
agent_resources_synced.ALIAS,
subnet_onboard_def.ALIAS]
# List of agent types for which all binding_failed ports should try to be
# rebound when agent revive
_rebind_on_revive_agent_types = [const.AGENT_TYPE_OVS]
@property
def supported_extension_aliases(self):
if not hasattr(self, '_aliases'):
aliases = self._supported_extension_aliases[:]
aliases += self.extension_manager.extension_aliases()
sg_rpc.disable_security_group_extension_by_config(aliases)
vlantransparent._disable_extension_by_config(aliases)
filter_validation._disable_extension_by_config(aliases)
self._aliases = aliases
return self._aliases
def __new__(cls, *args, **kwargs):
model_query.register_hook(
models_v2.Port,
"ml2_port_bindings",
query_hook=None,
filter_hook=None,
result_filters=_ml2_port_result_filter_hook)
return super(Ml2Plugin, cls).__new__(cls, *args, **kwargs)
@resource_registry.tracked_resources(
network=models_v2.Network,
port=models_v2.Port,
subnet=models_v2.Subnet,
subnetpool=models_v2.SubnetPool,
security_group=sg_models.SecurityGroup,
security_group_rule=sg_models.SecurityGroupRule)
def __init__(self):
# First load drivers, then initialize DB, then initialize drivers
self.type_manager = managers.TypeManager()
self.extension_manager = managers.ExtensionManager()
self.mechanism_manager = managers.MechanismManager()
super(Ml2Plugin, self).__init__()
self.type_manager.initialize()
self.extension_manager.initialize()
self.mechanism_manager.initialize()
self._setup_dhcp()
self._start_rpc_notifiers()
self.add_agent_status_check_worker(self.agent_health_check)
self.add_workers(self.mechanism_manager.get_workers())
self._verify_service_plugins_requirements()
LOG.info("Modular L2 Plugin initialization complete")
def _setup_rpc(self):
"""Initialize components to support agent communication."""
self.endpoints = [
rpc.RpcCallbacks(self.notifier, self.type_manager),
securitygroups_rpc.SecurityGroupServerRpcCallback(),
dvr_rpc.DVRServerRpcCallback(),
dhcp_rpc.DhcpRpcCallback(),
agents_db.AgentExtRpcCallback(),
metadata_rpc.MetadataRpcCallback(),
resources_rpc.ResourcesPullRpcCallback()
]
def _setup_dhcp(self):
"""Initialize components to support DHCP."""
self.network_scheduler = importutils.import_object(
cfg.CONF.network_scheduler_driver
)
self.add_periodic_dhcp_agent_status_check()
def _verify_service_plugins_requirements(self):
for service_plugin in cfg.CONF.service_plugins:
extension_drivers = SERVICE_PLUGINS_REQUIRED_DRIVERS.get(
service_plugin, []
)
for extension_driver in extension_drivers:
if extension_driver not in self.extension_manager.names():
raise ml2_exc.ExtensionDriverNotFound(
driver=extension_driver, service_plugin=service_plugin
)
@registry.receives(resources.PORT,
[provisioning_blocks.PROVISIONING_COMPLETE])
def _port_provisioned(self, rtype, event, trigger, context, object_id,
**kwargs):
port_id = object_id
port = db.get_port(context, port_id)
port_binding = p_utils.get_port_binding_by_status_and_host(
getattr(port, 'port_bindings', []), const.ACTIVE)
if not port or not port_binding:
LOG.debug("Port %s was deleted so its status cannot be updated.",
port_id)
return
if port_binding.vif_type in (portbindings.VIF_TYPE_BINDING_FAILED,
portbindings.VIF_TYPE_UNBOUND):
# NOTE(kevinbenton): we hit here when a port is created without
# a host ID and the dhcp agent notifies that its wiring is done
LOG.debug("Port %s cannot update to ACTIVE because it "
"is not bound.", port_id)
return
else:
# port is bound, but we have to check for new provisioning blocks
# one last time to detect the case where we were triggered by an
# unbound port and the port became bound with new provisioning
# blocks before 'get_port' was called above
if provisioning_blocks.is_object_blocked(context, port_id,
resources.PORT):
LOG.debug("Port %s had new provisioning blocks added so it "
"will not transition to active.", port_id)
return
if not port.admin_state_up:
LOG.debug("Port %s is administratively disabled so it will "
"not transition to active.", port_id)
return
self.update_port_status(context, port_id, const.PORT_STATUS_ACTIVE)
@log_helpers.log_method_call
def _start_rpc_notifiers(self):
"""Initialize RPC notifiers for agents."""
self.ovo_notifier = ovo_rpc.OVOServerRpcInterface()
self.notifier = rpc.AgentNotifierApi(topics.AGENT)
self.agent_notifiers[const.AGENT_TYPE_DHCP] = (
dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
)
@log_helpers.log_method_call
def start_rpc_listeners(self):
"""Start the RPC loop to let the plugin communicate with agents."""
self._setup_rpc()
self.topic = topics.PLUGIN
self.conn = n_rpc.Connection()
self.conn.create_consumer(self.topic, self.endpoints, fanout=False)
self.conn.create_consumer(
topics.SERVER_RESOURCE_VERSIONS,
[resources_rpc.ResourcesPushToServerRpcCallback()],
fanout=True)
# process state reports despite dedicated rpc workers
self.conn.create_consumer(topics.REPORTS,
[agents_db.AgentExtRpcCallback()],
fanout=False)
return self.conn.consume_in_threads()
def start_rpc_state_reports_listener(self):
self.conn_reports = n_rpc.Connection()
self.conn_reports.create_consumer(topics.REPORTS,
[agents_db.AgentExtRpcCallback()],
fanout=False)
return self.conn_reports.consume_in_threads()
def _filter_nets_provider(self, context, networks, filters):
return [network
for network in networks
if self.type_manager.network_matches_filters(network, filters)
]
def _check_mac_update_allowed(self, orig_port, port, binding):
unplugged_types = (portbindings.VIF_TYPE_BINDING_FAILED,
portbindings.VIF_TYPE_UNBOUND)
new_mac = port.get('mac_address')
mac_change = (new_mac is not None and
orig_port['mac_address'] != new_mac)
if (mac_change and binding.vif_type not in unplugged_types):
raise exc.PortBound(port_id=orig_port['id'],
vif_type=binding.vif_type,
old_mac=orig_port['mac_address'],
new_mac=port['mac_address'])
return mac_change
def _reset_mac_for_direct_physical(self, orig_port, port, binding):
# when unbinding direct-physical port we need to free
# physical device MAC address so that other ports may reuse it
if (binding.vnic_type == portbindings.VNIC_DIRECT_PHYSICAL and
port.get('device_id') == '' and
port.get('device_owner') == '' and
orig_port['device_id'] != ''):
port['mac_address'] = self._generate_macs()[0]
return True
else:
return False
@registry.receives(resources.AGENT, [events.AFTER_UPDATE])
def _retry_binding_revived_agents(self, resource, event, trigger,
payload=None):
context = payload.context
host = payload.metadata.get('host')
agent = payload.desired_state
agent_status = agent.get('agent_status')
agent_type = agent.get('agent_type')
if (agent_status != agent_consts.AGENT_REVIVED or
not agent.get('admin_state_up') or
agent_type not in self._rebind_on_revive_agent_types):
return
ports = ports_obj.Port.get_ports_by_binding_type_and_host(
context, portbindings.VIF_TYPE_BINDING_FAILED, host)
for port in ports:
binding = self._get_binding_for_host(port.bindings, host)
if not binding:
LOG.debug('No bindings found for port %(port_id)s '
'on host %(host)s',
{'port_id': port.id, 'host': host})
continue
port_dict = self._make_port_dict(port.db_obj)
network = self.get_network(context, port.network_id)
try:
levels = db.get_binding_level_objs(
context, port.id, binding.host)
# TODO(slaweq): use binding OVO instead of binding.db_obj when
# ML2 plugin will switch to use Port Binding OVO everywhere
mech_context = driver_context.PortContext(
self, context, port_dict, network, binding.db_obj, levels)
self._bind_port_if_needed(mech_context)
except Exception as e:
LOG.warning('Attempt to bind port %(port_id)s after agent '
'%(agent_type)s on host %(host)s revived failed. '
'Error: %(error)s',
{'port_id': port.id,
'agent_type': agent_type,
'host': host,
'error': e})
def _clear_port_binding(self, mech_context, binding, port, original_host):
binding.vif_type = portbindings.VIF_TYPE_UNBOUND
binding.vif_details = ''
db.clear_binding_levels(mech_context._plugin_context, port['id'],
original_host)
mech_context._clear_binding_levels()
def _process_port_binding_attributes(self, binding, attrs):
changes = False
host = const.ATTR_NOT_SPECIFIED
if attrs and portbindings.HOST_ID in attrs:
host = attrs.get(portbindings.HOST_ID) or ''
original_host = binding.host
if validators.is_attr_set(host) and original_host != host:
binding.host = host
changes = True
vnic_type = attrs and attrs.get(portbindings.VNIC_TYPE)
if (validators.is_attr_set(vnic_type) and
binding.vnic_type != vnic_type):
binding.vnic_type = vnic_type
changes = True
# treat None as clear of profile.
profile = None
if attrs and portbindings.PROFILE in attrs:
profile = attrs.get(portbindings.PROFILE) or {}
if profile not in (None, const.ATTR_NOT_SPECIFIED,
self._get_profile(binding)):
binding.profile = jsonutils.dumps(profile)
if len(binding.profile) > models.BINDING_PROFILE_LEN:
msg = _("binding:profile value too large")
raise exc.InvalidInput(error_message=msg)
changes = True
return changes, original_host
def _process_port_binding(self, mech_context, attrs):
plugin_context = mech_context._plugin_context
binding = mech_context._binding
port = mech_context.current
changes, original_host = self._process_port_binding_attributes(binding,
attrs)
# Unbind the port if needed.
if changes:
self._clear_port_binding(mech_context, binding, port,
original_host)
port['status'] = const.PORT_STATUS_DOWN
super(Ml2Plugin, self).update_port(
mech_context._plugin_context, port['id'],
{port_def.RESOURCE_NAME:
{'status': const.PORT_STATUS_DOWN}})
if port['device_owner'] == const.DEVICE_OWNER_DVR_INTERFACE:
self._clear_port_binding(mech_context, binding, port,
original_host)
binding.host = ''
self._update_port_dict_binding(port, binding)
binding.persist_state_to_session(plugin_context.session)
return changes
@db_api.retry_db_errors
def _bind_port_if_needed(self, context, allow_notify=False,
need_notify=False, allow_commit=True):
if not context.network.network_segments:
LOG.debug("Network %s has no segments, skipping binding",
context.network.current['id'])
return context
for count in range(1, MAX_BIND_TRIES + 1):
if count > 1:
# yield for binding retries so that we give other threads a
# chance to do their work
greenthread.sleep(0)
# multiple attempts shouldn't happen very often so we log each
# attempt after the 1st.
LOG.info("Attempt %(count)s to bind port %(port)s",
{'count': count, 'port': context.current['id']})
bind_context, need_notify, try_again = self._attempt_binding(
context, need_notify)
if count == MAX_BIND_TRIES or not try_again:
if self._should_bind_port(context) and allow_commit:
# At this point, we attempted to bind a port and reached
# its final binding state. Binding either succeeded or
# exhausted all attempts, thus no need to try again.
# Now, the port and its binding state should be committed.
context, need_notify, try_again = (
self._commit_port_binding(context, bind_context,
need_notify, try_again))
else:
context = bind_context
if not try_again:
if allow_notify and need_notify:
self._notify_port_updated(context)
return context
LOG.error("Failed to commit binding results for %(port)s "
"after %(max)s tries",
{'port': context.current['id'], 'max': MAX_BIND_TRIES})
return context
def _should_bind_port(self, context):
return (context._binding.host and context._binding.vif_type
in (portbindings.VIF_TYPE_UNBOUND,
portbindings.VIF_TYPE_BINDING_FAILED))
def _attempt_binding(self, context, need_notify):
try_again = False
if self._should_bind_port(context):
bind_context = self._bind_port(context)
if bind_context.vif_type != portbindings.VIF_TYPE_BINDING_FAILED:
# Binding succeeded. Suggest notifying of successful binding.
need_notify = True
else:
# Current attempt binding failed, try to bind again.
try_again = True
context = bind_context
return context, need_notify, try_again
def _bind_port(self, orig_context):
# Construct a new PortContext from the one from the previous
# transaction.
port = orig_context.current
orig_binding = orig_context._binding
new_binding = models.PortBinding(
host=orig_binding.host,
vnic_type=orig_binding.vnic_type,
profile=orig_binding.profile,
vif_type=portbindings.VIF_TYPE_UNBOUND,
vif_details=''
)
self._update_port_dict_binding(port, new_binding)
new_context = driver_context.PortContext(
self, orig_context._plugin_context, port,
orig_context.network.current, new_binding, None,
original_port=orig_context.original)
# Attempt to bind the port and return the context with the
# result.
self.mechanism_manager.bind_port(new_context)
return new_context
def _commit_port_binding(self, orig_context, bind_context,
need_notify, try_again,
update_binding_levels=True):
port_id = orig_context.current['id']
plugin_context = orig_context._plugin_context
orig_binding = orig_context._binding
new_binding = bind_context._binding
# TODO(yamahata): revise what to be passed or new resource
# like PORTBINDING should be introduced?
# It would be addressed during EventPayload conversion.
registry.notify(resources.PORT, events.BEFORE_UPDATE, self,
context=plugin_context, port=orig_context.current,
original_port=orig_context.current,
orig_binding=orig_binding, new_binding=new_binding)
# After we've attempted to bind the port, we begin a
# transaction, get the current port state, and decide whether
# to commit the binding results.
with db_api.CONTEXT_WRITER.using(plugin_context):
# Get the current port state and build a new PortContext
# reflecting this state as original state for subsequent
# mechanism driver update_port_*commit() calls.
try:
port_db = self._get_port(plugin_context, port_id)
cur_binding = p_utils.get_port_binding_by_status_and_host(
port_db.port_bindings, const.ACTIVE)
except exc.PortNotFound:
port_db, cur_binding = None, None
if not port_db or not cur_binding:
# The port has been deleted concurrently, so just
# return the unbound result from the initial
# transaction that completed before the deletion.
LOG.debug("Port %s has been deleted concurrently", port_id)
return orig_context, False, False
# Since the mechanism driver bind_port() calls must be made
# outside a DB transaction locking the port state, it is
# possible (but unlikely) that the port's state could change
# concurrently while these calls are being made. If another
# thread or process succeeds in binding the port before this
# thread commits its results, the already committed results are
# used. If attributes such as binding:host_id, binding:profile,
# or binding:vnic_type are updated concurrently, the try_again
# flag is returned to indicate that the commit was unsuccessful.
oport = self._make_port_dict(port_db)
port = self._make_port_dict(port_db)
network = bind_context.network.current
if port['device_owner'] == const.DEVICE_OWNER_DVR_INTERFACE:
# REVISIT(rkukura): The PortBinding instance from the
# ml2_port_bindings table, returned as cur_binding
# from port_db.port_binding above, is
# currently not used for DVR distributed ports, and is
# replaced here with the DistributedPortBinding instance from
# the ml2_distributed_port_bindings table specific to the host
# on which the distributed port is being bound. It
# would be possible to optimize this code to avoid
# fetching the PortBinding instance in the DVR case,
# and even to avoid creating the unused entry in the
# ml2_port_bindings table. But the upcoming resolution
# for bug 1367391 will eliminate the
# ml2_distributed_port_bindings table, use the
# ml2_port_bindings table to store non-host-specific
# fields for both distributed and non-distributed
# ports, and introduce a new ml2_port_binding_hosts
# table for the fields that need to be host-specific
# in the distributed case. Since the PortBinding
# instance will then be needed, it does not make sense
# to optimize this code to avoid fetching it.
cur_binding = db.get_distributed_port_binding_by_host(
plugin_context, port_id, orig_binding.host)
cur_context_binding = cur_binding
if new_binding.status == const.INACTIVE:
cur_context_binding = (
p_utils.get_port_binding_by_status_and_host(
port_db.port_bindings, const.INACTIVE,
host=new_binding.host))
cur_context = driver_context.PortContext(
self, plugin_context, port, network, cur_context_binding, None,
original_port=oport)
# Commit our binding results only if port has not been
# successfully bound concurrently by another thread or
# process and no binding inputs have been changed.
commit = ((cur_binding.vif_type in
[portbindings.VIF_TYPE_UNBOUND,
portbindings.VIF_TYPE_BINDING_FAILED]) and
orig_binding.host == cur_binding.host and
orig_binding.vnic_type == cur_binding.vnic_type and
orig_binding.profile == cur_binding.profile)
if commit:
# Update the port's binding state with our binding
# results.
if new_binding.status == const.INACTIVE:
cur_context_binding.status = const.ACTIVE
cur_binding.status = const.INACTIVE
else:
cur_context_binding.vif_type = new_binding.vif_type
cur_context_binding.vif_details = new_binding.vif_details
if update_binding_levels:
db.clear_binding_levels(plugin_context, port_id,
cur_binding.host)
db.set_binding_levels(plugin_context,
bind_context._binding_levels)
# refresh context with a snapshot of updated state
cur_context._binding = driver_context.InstanceSnapshot(
cur_context_binding)
cur_context._binding_levels = bind_context._binding_levels
# Update PortContext's port dictionary to reflect the
# updated binding state.
self._update_port_dict_binding(port, cur_context_binding)
# Update the port status if requested by the bound driver.
if (bind_context._binding_levels and
bind_context._new_port_status):
port_db.status = bind_context._new_port_status
port['status'] = bind_context._new_port_status
# Call the mechanism driver precommit methods, commit
# the results, and call the postcommit methods.
self.mechanism_manager.update_port_precommit(cur_context)
else:
# Try to populate the PortContext with the current binding
# levels so that the RPC notification won't get suppressed.
# This is to avoid leaving ports stuck in a DOWN state.
# For more information see bug:
# https://bugs.launchpad.net/neutron/+bug/1755810
LOG.warning("Concurrent port binding operations failed on "
"port %s", port_id)
levels = db.get_binding_level_objs(plugin_context, port_id,
cur_binding.host)
for level in levels:
cur_context._push_binding_level(level)
# refresh context with a snapshot of the current binding state
cur_context._binding = driver_context.InstanceSnapshot(
cur_binding)
if commit:
# Continue, using the port state as of the transaction that
# just finished, whether that transaction committed new
# results or discovered concurrent port state changes.
# Also, Trigger notification for successful binding commit.
kwargs = {
'context': plugin_context,
'port': self._make_port_dict(port_db), # ensure latest state
'mac_address_updated': False,
'original_port': oport,
}
registry.notify(resources.PORT, events.AFTER_UPDATE,
self, **kwargs)
self.mechanism_manager.update_port_postcommit(cur_context)
need_notify = True
try_again = False
else:
try_again = True
return cur_context, need_notify, try_again
def _update_port_dict_binding(self, port, binding):
port[portbindings.VNIC_TYPE] = binding.vnic_type
port[portbindings.PROFILE] = self._get_profile(binding)
if port['device_owner'] == const.DEVICE_OWNER_DVR_INTERFACE:
port[portbindings.HOST_ID] = ''
port[portbindings.VIF_TYPE] = portbindings.VIF_TYPE_DISTRIBUTED
port[portbindings.VIF_DETAILS] = {}
else:
port[portbindings.HOST_ID] = binding.host
port[portbindings.VIF_TYPE] = binding.vif_type
port[portbindings.VIF_DETAILS] = self._get_vif_details(binding)
def _get_vif_details(self, binding):
if binding.vif_details:
try:
return jsonutils.loads(binding.vif_details)
except Exception:
LOG.error("Serialized vif_details DB value '%(value)s' "
"for port %(port)s is invalid",
{'value': binding.vif_details,
'port': binding.port_id})
return {}
def _get_profile(self, binding):
if binding.profile:
try:
return jsonutils.loads(binding.profile)
except Exception:
LOG.error("Serialized profile DB value '%(value)s' for "
"port %(port)s is invalid",
{'value': binding.profile,
'port': binding.port_id})
return {}
@staticmethod
@resource_extend.extends([port_def.COLLECTION_NAME])
def _ml2_extend_port_dict_binding(port_res, port_db):
plugin = directory.get_plugin()
if isinstance(port_db, ports_obj.Port):
bindings = port_db.bindings
else:
bindings = port_db.port_bindings
port_binding = p_utils.get_port_binding_by_status_and_host(
bindings, const.ACTIVE)
# None when called during unit tests for other plugins.
if port_binding:
plugin._update_port_dict_binding(port_res, port_binding)
# ML2's resource extend functions allow extension drivers that extend
# attributes for the resources to add those attributes to the result.
@staticmethod
@resource_extend.extends([net_def.COLLECTION_NAME])
def _ml2_md_extend_network_dict(result, netdb):
plugin = directory.get_plugin()
session = plugin._object_session_or_new_session(netdb)
plugin.extension_manager.extend_network_dict(session, netdb, result)
@staticmethod
@resource_extend.extends([port_def.COLLECTION_NAME])
def _ml2_md_extend_port_dict(result, portdb):
plugin = directory.get_plugin()
session = plugin._object_session_or_new_session(portdb)
plugin.extension_manager.extend_port_dict(session, portdb, result)
@staticmethod
@resource_extend.extends([subnet_def.COLLECTION_NAME])
def _ml2_md_extend_subnet_dict(result, subnetdb):
plugin = directory.get_plugin()
session = plugin._object_session_or_new_session(subnetdb)
plugin.extension_manager.extend_subnet_dict(session, subnetdb, result)
@staticmethod
def _object_session_or_new_session(sql_obj):
session = sqlalchemy.inspect(sql_obj).session
if not session:
session = db_api.get_reader_session()
return session
def _notify_port_updated(self, mech_context):
port = mech_context.current
segment = mech_context.bottom_bound_segment
if not segment:
# REVISIT(rkukura): This should notify agent to unplug port
network = mech_context.network.current
LOG.debug("In _notify_port_updated(), no bound segment for "
"port %(port_id)s on network %(network_id)s",
{'port_id': port['id'], 'network_id': network['id']})
return
self.notifier.port_update(mech_context._plugin_context, port,
segment[api.NETWORK_TYPE],
segment[api.SEGMENTATION_ID],
segment[api.PHYSICAL_NETWORK])
def _update_segmentation_id(self, context, network, net_data):
"""Update segmentation ID in a single provider network"""
segments = segments_db.get_networks_segments(
context, [network['id']])[network['id']]
if len(segments) > 1:
msg = _('Provider network attributes can be updated only in '
'provider networks with a single segment.')
raise exc.InvalidInput(error_message=msg)
vif_types = [portbindings.VIF_TYPE_UNBOUND,
portbindings.VIF_TYPE_BINDING_FAILED]
for mech_driver in self.mechanism_manager.ordered_mech_drivers:
if (isinstance(mech_driver.obj,
mech_agent.AgentMechanismDriverBase) and
provider_net.SEGMENTATION_ID in mech_driver.obj.
provider_network_attribute_updates_supported()):
agent_type = mech_driver.obj.agent_type
agents = self.get_agents(
context, filters={'agent_type': [agent_type]})
for agent in agents:
vif_types.append(mech_driver.obj.get_vif_type(
context, agent, segments[0]))
filter_obj = obj_utils.NotIn(vif_types)
filters = {portbindings.VIF_TYPE:
filter_obj.filter(models.PortBinding.vif_type),
'network_id': [network['id']]}
if super(Ml2Plugin, self).get_ports_count(context,
filters=filters):
msg = (_('Provider network attribute %(attr)s cannot be updated '
'if any port in the network has not the following '
'%(vif_field)s: %(vif_types)s') %
{'attr': provider_net.SEGMENTATION_ID,
'vif_field': portbindings.VIF_TYPE,
'vif_types': ', '.join(vif_types)})
raise exc.InvalidInput(error_message=msg)
self.type_manager.update_network_segment(context, network,
net_data, segments[0])
def _update_provider_network_attributes(self, context, network, net_data):
"""Raise exception if provider network attrs update are not supported.
This function will raise an exception if the provider network attribute
update is not supported.
"""
provider_net_attrs = (set(provider_net.ATTRIBUTES) -
{provider_net.SEGMENTATION_ID})
requested_provider_net_attrs = set(net_data) & provider_net_attrs
for attr in requested_provider_net_attrs:
if (validators.is_attr_set(net_data.get(attr)) and
net_data.get(attr) != network[attr]):
msg = (_('Plugin does not support updating the following '
'provider network attributes: %s') %
', '.join(provider_net_attrs))
raise exc.InvalidInput(error_message=msg)
if net_data.get(provider_net.SEGMENTATION_ID):
self._update_segmentation_id(context, network, net_data)
def _delete_objects(self, context, resource, objects):
delete_op = getattr(self, 'delete_%s' % resource)
for obj in objects:
try:
delete_op(context, obj['result']['id'])
except KeyError:
LOG.exception("Could not find %s to delete.",
resource)
except Exception:
LOG.exception("Could not delete %(res)s %(id)s.",
{'res': resource,
'id': obj['result']['id']})
def _create_bulk_ml2(self, resource, context, request_items):
objects = []
collection = "%ss" % resource
items = request_items[collection]
obj_before_create = getattr(self, '_before_create_%s' % resource)
for item in items:
obj_before_create(context, item)
with db_api.CONTEXT_WRITER.using(context):
obj_creator = getattr(self, '_create_%s_db' % resource)
for item in items:
try:
attrs = item[resource]
result, mech_context = obj_creator(context, item)
objects.append({'mech_context': mech_context,
'result': result,
'attributes': attrs})
except Exception as e:
with excutils.save_and_reraise_exception():
utils.attach_exc_details(
e, ("An exception occurred while creating "
"the %(resource)s:%(item)s"),
{'resource': resource, 'item': item})
postcommit_op = getattr(self, '_after_create_%s' % resource)
for obj in objects:
try:
postcommit_op(context, obj['result'], obj['mech_context'])
except Exception:
with excutils.save_and_reraise_exception():
resource_ids = [res['result']['id'] for res in objects]
LOG.exception("ML2 _after_create_%(res)s "
"failed for %(res)s: "
"'%(failed_id)s'. Deleting "
"%(res)ss %(resource_ids)s",
{'res': resource,
'failed_id': obj['result']['id'],
'resource_ids': ', '.join(resource_ids)})
# _after_handler will have deleted the object that threw
to_delete = [o for o in objects if o != obj]
self._delete_objects(context, resource, to_delete)
return objects
def _get_network_mtu(self, network_db, validate=True):
mtus = []
try:
segments = network_db['segments']
except KeyError:
segments = [network_db]
for s in segments:
segment_type = s.get('network_type')
if segment_type is None:
continue
try:
type_driver = self.type_manager.drivers[segment_type].obj
except KeyError:
# NOTE(ihrachys) This can happen when type driver is not loaded
# for an existing segment, or simply when the network has no
# segments at the specific time this is computed.
# In the former case, while it's probably an indication of
# a bad setup, it's better to be safe than sorry here. Also,
# several unit tests use non-existent driver types that may
# trigger the exception here.
if segment_type and s['segmentation_id']:
LOG.warning(
"Failed to determine MTU for segment "
"%(segment_type)s:%(segment_id)s; network "
"%(network_id)s MTU calculation may be not "
"accurate",
{
'segment_type': segment_type,
'segment_id': s['segmentation_id'],
'network_id': network_db['id'],
}
)
else:
mtu = type_driver.get_mtu(s['physical_network'])
# Some drivers, like 'local', may return None; the assumption
# then is that for the segment type, MTU has no meaning or
# unlimited, and so we should then ignore those values.
if mtu:
mtus.append(mtu)
max_mtu = min(mtus) if mtus else p_utils.get_deployment_physnet_mtu()
net_mtu = network_db.get('mtu')
if validate:
# validate that requested mtu conforms to allocated segments
if net_mtu and max_mtu and max_mtu < net_mtu:
msg = _("Requested MTU is too big, maximum is %d") % max_mtu
raise exc.InvalidInput(error_message=msg)
# if mtu is not set in database, use the maximum possible
return net_mtu or max_mtu
def _before_create_network(self, context, network):
net_data = network[net_def.RESOURCE_NAME]
registry.notify(resources.NETWORK, events.BEFORE_CREATE, self,
context=context, network=net_data)