Introduce L3 agent extension for SNAT logging

This patch introduces SNAT logging extension for Neutron Packet Logging
framework.

Co-Authored-By: Kim Bao Long <longkb@vn.fujitsu.com>
Partial-Bug: #1752290
Change-Id: I882b8e75525f51ef8b5b032fc89843ae904844e7
This commit is contained in:
Nguyen Phuong An 2018-06-21 14:40:57 +07:00
parent 94a5026fb5
commit 6352b944a4
3 changed files with 90 additions and 0 deletions

View File

@ -0,0 +1,39 @@
# Copyright (c) 2018 Fujitsu Limited
# 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 neutron_lib.agent import l3_extension
from oslo_log import log as logging
from neutron.services.logapi.agent.l3 import base
from neutron.services.logapi.agent import log_extension as log_ext
from neutron.services.logapi.rpc import agent as agent_rpc
LOG = logging.getLogger(__name__)
SNAT_LOG_DRIVER = 'snat_log'
class SNATLoggingExtension(base.L3LoggingExtensionBase,
l3_extension.L3AgentExtension):
def initialize(self, connection, driver_type):
"""Initialize SNAT logging agent extension"""
snat_log_cls = self._load_driver_cls(
log_ext.LOGGING_DRIVERS_NAMESPACE, SNAT_LOG_DRIVER)
self.log_driver = snat_log_cls(self.agent_api)
self.resource_rpc = agent_rpc.LoggingApiStub()
self._register_rpc_consumers()
self.log_driver.initialize(self.resource_rpc)
LOG.info("Loaded SNAT logging driver")

View File

@ -0,0 +1,50 @@
# Copyright (c) 2018 Fujitsu Limited
# 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 mock
from neutron_lib import constants as lib_const
from neutron.agent.l3.extensions import snat_log
from neutron.api.rpc.callbacks.consumer import registry
from neutron.api.rpc.callbacks import resources
from neutron.api.rpc.handlers import resources_rpc
from neutron.tests.unit.services.logapi.agent.l3 import test_base
class SnatLogExtensionInitializeTestCase(test_base.L3LoggingExtBaseTestCase):
def setUp(self):
super(SnatLogExtensionInitializeTestCase, self).setUp()
self.snat_log_ext = snat_log.SNATLoggingExtension()
self.snat_log_ext.consume_api(self.agent_api)
@mock.patch.object(registry, 'register')
@mock.patch.object(resources_rpc, 'ResourcesPushRpcCallback')
def test_initialize_subscribed_to_rpc(self, rpc_mock, subscribe_mock):
call_to_patch = 'neutron.common.rpc.Connection'
with mock.patch(call_to_patch,
return_value=self.connection) as create_connection:
self.snat_log_ext.initialize(
self.connection, lib_const.L3_AGENT_MODE)
create_connection.assert_has_calls([mock.call()])
self.connection.create_consumer.assert_has_calls(
[mock.call(
resources_rpc.resource_type_versioned_topic(
resources.LOGGING_RESOURCE),
[rpc_mock()],
fanout=True)]
)
subscribe_mock.assert_called_with(
mock.ANY, resources.LOGGING_RESOURCE)

View File

@ -109,6 +109,7 @@ neutron.agent.l2.extensions =
neutron.agent.l3.extensions =
fip_qos = neutron.agent.l3.extensions.qos.fip:FipQosAgentExtension
port_forwarding = neutron.agent.l3.extensions.port_forwarding:PortForwardingAgentExtension
snat_log = neutron.agent.l3.extensions.snat_log:SNATLoggingExtension
neutron.services.logapi.drivers =
ovs = neutron.services.logapi.drivers.openvswitch.ovs_firewall_log:OVSFirewallLoggingDriver
neutron.qos.agent_drivers =