'agent_type': q_const.AGENT_TYPE_OVS,
'start_flag': True}
+ if tunnel_types:
+ self.enable_tunneling = True
+ else:
+ self.enable_tunneling = False
+
# Validate agent configurations
self._check_agent_configurations()
self.polling_interval = polling_interval
self.minimize_polling = minimize_polling
self.ovsdb_monitor_respawn_interval = ovsdb_monitor_respawn_interval
-
- if tunnel_types:
- self.enable_tunneling = True
- else:
- self.enable_tunneling = False
self.local_ip = local_ip
self.tunnel_count = 0
self.vxlan_udp_port = cfg.CONF.AGENT.vxlan_udp_port
rpc_api.client.timeout = timeout
def _check_agent_configurations(self):
- if self.enable_distributed_routing and not self.l2_pop:
- raise ValueError(_("DVR cannot be enabled without "
- "L2 population."))
+ if (self.enable_distributed_routing and self.enable_tunneling
+ and not self.l2_pop):
+ raise ValueError(_("DVR deployments for VXLAN/GRE underlays "
+ "require L2-pop to be enabled, in both the "
+ "Agent and Server side."))
def _ofport_set_to_str(ofport_set):
self.assertFalse(set_ovs_db_func.called)
self.assertFalse(delete_flows_func.called)
- def test_check_agent_configurations_raises(self):
+ def test_check_agent_configurations_for_dvr_raises(self):
self.agent.enable_distributed_routing = True
+ self.agent.enable_tunneling = True
self.agent.l2_pop = False
self.assertRaises(ValueError,
self.agent._check_agent_configurations)
- def test_check_agent_configurations(self):
+ def test_check_agent_configurations_for_dvr(self):
self.agent.enable_distributed_routing = True
+ self.agent.enable_tunneling = True
self.agent.l2_pop = True
self.assertIsNone(self.agent._check_agent_configurations())
+ def test_check_agent_configurations_for_dvr_with_vlan(self):
+ self.agent.enable_distributed_routing = True
+ self.agent.enable_tunneling = False
+ self.agent.l2_pop = False
+ self.assertIsNone(self.agent._check_agent_configurations())
+
def test_port_bound_deletes_flows_for_valid_ofport(self):
self._mock_port_bound(ofport=1, new_local_vlan=1)