mir/kea
2
0
Fork 0
mirror of https://gitlab.isc.org/isc-projects/kea synced 2025-08-31 14:05:33 +00:00
Code Issues Releases Wiki Activity

[master] Merge branch 'trac5318'

Browse Source
This commit is contained in:
Marcin Siodelski
2017-07-05 16:32:14 +02:00
parent de3fcb076e 7c296923c6
commit 8531a65521
12 changed files with 924 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
doc/guide/ctrl-channel.xml
View File

@@ -60,6 +60,20 @@
it on its own.
</para>
<para>Control connections over both HTTP and unix domain sockets are
guarded with timeouts. The default timeout value is set to 10s
and is not configurable. The timeout configuration will be
implemented in the future.
</para>
<note>
<simpara>Kea 1.2.0 release and earlier had a limitation of 64kB
on the maximum size of a command and a response sent over the unix
domain socket. This limitation has been removed in Kea 1.3.0
release.
</simpara>
</note>
<section id="ctrl-channel-syntax">
<title>Data Syntax</title>
<para>Communication over the control channel is conducted using JSON

300
src/bin/dhcp4/tests/ctrl_dhcp4_srv_unittest.cc
View File

@@ -6,6 +6,7 @@
#include <config.h>
#include <asiolink/interval_timer.h>
#include <asiolink/io_service.h>
#include <cc/command_interpreter.h>
#include <config/command_mgr.h>
@@ -27,9 +28,12 @@
#include <boost/scoped_ptr.hpp>
#include <gtest/gtest.h>
#include <cstdlib>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <sstream>
#include <thread>
#include <arpa/inet.h>
#include <unistd.h>
@@ -47,6 +51,32 @@ using namespace isc::test;
namespace {
/// @brief Simple RAII class which stops IO service upon destruction
/// of the object.
class IOServiceWork {
public:
/// @brief Constructor.
///
/// @param io_service Pointer to the IO service to be stopped.
IOServiceWork(const IOServicePtr& io_service)
: io_service_(io_service) {
}
/// @brief Destructor.
///
/// Stops IO service.
~IOServiceWork() {
io_service_->stop();
}
private:
/// @brief Pointer to the IO service to be stopped upon destruction.
IOServicePtr io_service_;
};
class NakedControlledDhcpv4Srv: public ControlledDhcpv4Srv {
// "Naked" DHCPv4 server, exposes internal fields
public:
@@ -58,6 +88,9 @@ public:
using Dhcpv4Srv::receivePacket;
};
/// @brief Default control connection timeout.
const size_t DEFAULT_CONNECTION_TIMEOUT = 10;
/// @brief Fixture class intended for testin control channel in the DHCPv4Srv
class CtrlChannelDhcpv4SrvTest : public ::testing::Test {
public:
@@ -87,6 +120,8 @@ public:
StatsMgr::instance().removeAll();
CommandMgr::instance().closeCommandSocket();
CommandMgr::instance().deregisterAll();
CommandMgr::instance().setConnectionTimeout(DEFAULT_CONNECTION_TIMEOUT);
server_.reset();
};
@@ -298,6 +333,54 @@ public:
ADD_FAILURE() << "Invalid expected status: " << exp_status;
}
}
/// @brief Handler for long command.
///
/// It checks whether the received command is equal to the one specified
/// as an argument.
///
/// @param expected_command String representing an expected command.
/// @param command_name Command name received by the handler.
/// @param arguments Command arguments received by the handler.
///
/// @returns Success answer.
static ConstElementPtr
longCommandHandler(const std::string& expected_command,
const std::string& command_name,
const ConstElementPtr& arguments) {
// The handler is called with a command name and the structure holding
// command arguments. We have to rebuild the command from those
// two arguments so as it can be compared against expected_command.
ElementPtr entire_command = Element::createMap();
entire_command->set("command", Element::create(command_name));
entire_command->set("arguments", (arguments));
// The rebuilt command will have a different order of parameters so
// let's parse expected_command back to JSON to guarantee that
// both structures are built using the same order.
EXPECT_EQ(Element::fromJSON(expected_command)->str(),
entire_command->str());
return (createAnswer(0, "long command received ok"));
}
/// @brief Command handler which generates long response
///
/// This handler generates a large response (over 400kB). It includes
/// a list of randomly generated strings to make sure that the test
/// can catch out of order delivery.
static ConstElementPtr longResponseHandler(const std::string&,
const ConstElementPtr&) {
// By seeding the generator with the constant value we will always
// get the same sequence of generated strings.
std::srand(1);
ElementPtr arguments = Element::createList();
for (unsigned i = 0; i < 40000; ++i) {
std::ostringstream s;
s << std::setw(10) << std::rand();
arguments->add(Element::create(s.str()));
}
return (createAnswer(0, arguments));
}
};
TEST_F(CtrlChannelDhcpv4SrvTest, commands) {
@@ -433,7 +516,7 @@ TEST_F(CtrlChannelDhcpv4SrvTest, controlChannelNegative) {
sendUnixCommand("utter nonsense", response);
EXPECT_EQ("{ \"result\": 1, "
"\"text\": \"error: unexpected character u in <string>:1:2\" }",
"\"text\": \"invalid first character u\" }",
response);
}
@@ -712,7 +795,7 @@ TEST_F(CtrlChannelDhcpv4SrvTest, configSet) {
// Should fail with a syntax error
EXPECT_EQ("{ \"result\": 1, "
"\"text\": \"subnet configuration failed: mandatory 'subnet' "
"parameter is missing for a subnet being configured (<string>:20:17)\" }",
"parameter is missing for a subnet being configured (<wire>:19:17)\" }",
response);
// Check that the config was not lost
@@ -911,7 +994,7 @@ TEST_F(CtrlChannelDhcpv4SrvTest, configTest) {
// Should fail with a syntax error
EXPECT_EQ("{ \"result\": 1, "
"\"text\": \"subnet configuration failed: mandatory 'subnet' "
"parameter is missing for a subnet being configured (<string>:20:17)\" }",
"parameter is missing for a subnet being configured (<wire>:19:17)\" }",
response);
// Check that the config was not lost
@@ -952,7 +1035,7 @@ TEST_F(CtrlChannelDhcpv4SrvTest, configTest) {
// Clean up after the test.
CfgMgr::instance().clear();
}
// Tests if config-write can be called without any parameters.
TEST_F(CtrlChannelDhcpv4SrvTest, writeConfigNoFilename) {
createUnixChannelServer();
@@ -1110,4 +1193,213 @@ TEST_F(CtrlChannelDhcpv4SrvTest, concurrentConnections) {
ASSERT_NO_THROW(getIOService()->poll());
}
// This test verifies that the server can receive and process a large command.
TEST_F(CtrlChannelDhcpv4SrvTest, longCommand) {
std::ostringstream command;
// This is the desired size of the command sent to the server (1MB). The
// actual size sent will be slightly greater than that.
const size_t command_size = 1024 * 1000;
while (command.tellp() < command_size) {
// We're sending command 'foo' with arguments being a list of
// strings. If this is the first transmission, send command name
// and open the arguments list. Also insert the first argument
// so as all subsequent arguments can be prefixed with a comma.
if (command.tellp() == 0) {
command << "{ \"command\": \"foo\", \"arguments\": [ \"begin\"";
} else {
// Generate a random number and insert it into the stream as
// 10 digits long string.
std::ostringstream arg;
arg << setw(10) << std::rand();
// Append the argument in the command.
command << ", \"" << arg.str() << "\"\n";
// If we have hit the limit of the command size, close braces to
// get appropriate JSON.
if (command.tellp() > command_size) {
command << "] }";
}
}
}
ASSERT_NO_THROW(
CommandMgr::instance().registerCommand("foo",
boost::bind(&CtrlChannelDhcpv4SrvTest::longCommandHandler,
command.str(), _1, _2));
);
createUnixChannelServer();
std::string response;
std::thread th([this, &response, &command]() {
// IO service will be stopped automatically when this object goes
// out of scope and is destroyed. This is useful because we use
// asserts which may break the thread in various exit points.
IOServiceWork work(getIOService());
// Create client which we will use to send command to the server.
boost::scoped_ptr<UnixControlClient> client(new UnixControlClient());
ASSERT_TRUE(client);
// Connect to the server. This will trigger acceptor handler on the
// server side and create a new connection.
ASSERT_TRUE(client->connectToServer(socket_path_));
// Initially the remaining_string holds the entire command and we
// will be erasing the portions that we have sent.
std::string remaining_data = command.str();
while (!remaining_data.empty()) {
// Send the command in chunks of 1024 bytes.
const size_t l = remaining_data.size() < 1024 ? remaining_data.size() : 1024;
ASSERT_TRUE(client->sendCommand(remaining_data.substr(0, l)));
remaining_data.erase(0, l);
}
// Set timeout to 5 seconds to allow the time for the server to send
// a response.
const unsigned int timeout = 5;
ASSERT_TRUE(client->getResponse(response, timeout));
// We're done. Close the connection to the server.
client->disconnectFromServer();
});
// Run the server until the command has been processed and response
// received.
getIOService()->run();
// Wait for the thread to complete.
th.join();
EXPECT_EQ("{ \"result\": 0, \"text\": \"long command received ok\" }",
response);
}
// This test verifies that the server can send long response to the client.
TEST_F(CtrlChannelDhcpv4SrvTest, longResponse) {
// We need to generate large response. The simplest way is to create
// a command and a handler which will generate some static response
// of a desired size.
ASSERT_NO_THROW(
CommandMgr::instance().registerCommand("foo",
boost::bind(&CtrlChannelDhcpv4SrvTest::longResponseHandler, _1, _2));
);
createUnixChannelServer();
// The UnixControlClient doesn't have any means to check that the entire
// response has been received. What we want to do is to generate a
// reference response using our command handler and then compare
// what we have received over the unix domain socket with this reference
// response to figure out when to stop receiving.
std::string reference_response = longResponseHandler("foo", ConstElementPtr())->str();
// In this stream we're going to collect out partial responses.
std::ostringstream response;
// The client is synchronous so it is useful to run it in a thread.
std::thread th([this, &response, reference_response]() {
// IO service will be stopped automatically when this object goes
// out of scope and is destroyed. This is useful because we use
// asserts which may break the thread in various exit points.
IOServiceWork work(getIOService());
// Remember the response size so as we know when we should stop
// receiving.
const size_t long_response_size = reference_response.size();
// Create the client and connect it to the server.
boost::scoped_ptr<UnixControlClient> client(new UnixControlClient());
ASSERT_TRUE(client);
ASSERT_TRUE(client->connectToServer(socket_path_));
// Send the stub command.
std::string command = "{ \"command\": \"foo\", \"arguments\": { } }";
ASSERT_TRUE(client->sendCommand(command));
// Keep receiving response data until we have received the full answer.
while (response.tellp() < long_response_size) {
std::string partial;
const unsigned int timeout = 5;
ASSERT_TRUE(client->getResponse(partial, 5));
response << partial;
}
// We have received the entire response, so close the connection and
// stop the IO service.
client->disconnectFromServer();
});
// Run the server until the entire response has been received.
getIOService()->run();
// Wait for the thread to complete.
th.join();
// Make sure we have received correct response.
EXPECT_EQ(reference_response, response.str());
}
// This test verifies that the server signals timeout if the transmission
// takes too long.
TEST_F(CtrlChannelDhcpv4SrvTest, connectionTimeout) {
createUnixChannelServer();
// Set connection timeout to 2s to prevent long waiting time for the
// timeout during this test.
const unsigned short timeout = 2;
CommandMgr::instance().setConnectionTimeout(timeout);
// Server's response will be assigned to this variable.
std::string response;
// It is useful to create a thread and run the server and the client
// at the same time and independently.
std::thread th([this, &response]() {
// IO service will be stopped automatically when this object goes
// out of scope and is destroyed. This is useful because we use
// asserts which may break the thread in various exit points.
IOServiceWork work(getIOService());
// Create the client and connect it to the server.
boost::scoped_ptr<UnixControlClient> client(new UnixControlClient());
ASSERT_TRUE(client);
ASSERT_TRUE(client->connectToServer(socket_path_));
// Send partial command. The server will be waiting for the remaining
// part to be sent and will eventually signal a timeout.
std::string command = "{ \"command\": \"foo\" ";
ASSERT_TRUE(client->sendCommand(command));
// Let's wait up to 15s for the server's response. The response
// should arrive sooner assuming that the timeout mechanism for
// the server is working properly.
const unsigned int timeout = 15;
ASSERT_TRUE(client->getResponse(response, timeout));
// Explicitly close the client's connection.
client->disconnectFromServer();
});
// Run the server until stopped.
getIOService()->run();
// Wait for the thread to return.
th.join();
// Check that the server has signalled a timeout.
EXPECT_EQ("{ \"result\": 1, \"text\": \"Connection over control channel"
" timed out\" }", response);
}
} // End of anonymous namespace

296
src/bin/dhcp6/tests/ctrl_dhcp6_srv_unittest.cc
View File

@@ -26,11 +26,16 @@
#include <boost/scoped_ptr.hpp>
#include <gtest/gtest.h>
#include <iomanip>
#include <sstream>
#include <sys/select.h>
#include <sys/stat.h>
#include <sys/ioctl.h>
#include <cstdlib>
#include <thread>
using namespace std;
using namespace isc::asiolink;
using namespace isc::config;
@@ -43,7 +48,31 @@ using namespace isc::test;
namespace {
/// @brief Simple RAII class which stops IO service upon destruction
/// of the object.
class IOServiceWork {
public:
/// @brief Constructor.
///
/// @param io_service Pointer to the IO service to be stopped.
IOServiceWork(const IOServicePtr& io_service)
: io_service_(io_service) {
}
/// @brief Destructor.
///
/// Stops IO service.
~IOServiceWork() {
io_service_->stop();
}
private:
/// @brief Pointer to the IO service to be stopped upon destruction.
IOServicePtr io_service_;
};
class NakedControlledDhcpv6Srv: public ControlledDhcpv6Srv {
// "Naked" DHCPv6 server, exposes internal fields
@@ -56,6 +85,9 @@ public:
using Dhcpv6Srv::receivePacket;
};
/// @brief Default control connection timeout.
const size_t DEFAULT_CONNECTION_TIMEOUT = 10;
class CtrlDhcpv6SrvTest : public BaseServerTest {
public:
CtrlDhcpv6SrvTest()
@@ -66,6 +98,9 @@ public:
virtual ~CtrlDhcpv6SrvTest() {
LeaseMgrFactory::destroy();
StatsMgr::instance().removeAll();
CommandMgr::instance().deregisterAll();
CommandMgr::instance().setConnectionTimeout(DEFAULT_CONNECTION_TIMEOUT);
reset();
};
@@ -307,6 +342,54 @@ public:
ADD_FAILURE() << "Invalid expected status: " << exp_status;
}
}
/// @brief Handler for long command.
///
/// It checks whether the received command is equal to the one specified
/// as an argument.
///
/// @param expected_command String representing an expected command.
/// @param command_name Command name received by the handler.
/// @param arguments Command arguments received by the handler.
///
/// @returns Success answer.
static ConstElementPtr
longCommandHandler(const std::string& expected_command,
const std::string& command_name,
const ConstElementPtr& arguments) {
// The handler is called with a command name and the structure holding
// command arguments. We have to rebuild the command from those
// two arguments so as it can be compared against expected_command.
ElementPtr entire_command = Element::createMap();
entire_command->set("command", Element::create(command_name));
entire_command->set("arguments", (arguments));
// The rebuilt command will have a different order of parameters so
// let's parse expected_command back to JSON to guarantee that
// both structures are built using the same order.
EXPECT_EQ(Element::fromJSON(expected_command)->str(),
entire_command->str());
return (createAnswer(0, "long command received ok"));
}
/// @brief Command handler which generates long response
///
/// This handler generates a large response (over 400kB). It includes
/// a list of randomly generated strings to make sure that the test
/// can catch out of order delivery.
static ConstElementPtr longResponseHandler(const std::string&,
const ConstElementPtr&) {
// By seeding the generator with the constant value we will always
// get the same sequence of generated strings.
std::srand(1);
ElementPtr arguments = Element::createList();
for (unsigned i = 0; i < 40000; ++i) {
std::ostringstream s;
s << std::setw(10) << std::rand();
arguments->add(Element::create(s.str()));
}
return (createAnswer(0, arguments));
}
};
@@ -485,7 +568,7 @@ TEST_F(CtrlChannelDhcpv6SrvTest, configSet) {
// Should fail with a syntax error
EXPECT_EQ("{ \"result\": 1, "
"\"text\": \"subnet configuration failed: mandatory 'subnet' parameter is missing for a subnet being configured (<string>:21:17)\" }",
"\"text\": \"subnet configuration failed: mandatory 'subnet' parameter is missing for a subnet being configured (<wire>:20:17)\" }",
response);
// Check that the config was not lost
@@ -631,7 +714,7 @@ TEST_F(CtrlChannelDhcpv6SrvTest, configTest) {
// Should fail with a syntax error
EXPECT_EQ("{ \"result\": 1, "
"\"text\": \"subnet configuration failed: mandatory 'subnet' parameter "
"is missing for a subnet being configured (<string>:21:17)\" }",
"is missing for a subnet being configured (<wire>:20:17)\" }",
response);
// Check that the config was not lost
@@ -738,7 +821,7 @@ TEST_F(CtrlChannelDhcpv6SrvTest, controlChannelNegative) {
sendUnixCommand("utter nonsense", response);
EXPECT_EQ("{ \"result\": 1, "
"\"text\": \"error: unexpected character u in <string>:1:2\" }",
"\"text\": \"invalid first character u\" }",
response);
}
@@ -1131,5 +1214,212 @@ TEST_F(CtrlChannelDhcpv6SrvTest, concurrentConnections) {
ASSERT_NO_THROW(getIOService()->poll());
}
// This test verifies that the server can receive and process a large command.
TEST_F(CtrlChannelDhcpv6SrvTest, longCommand) {
std::ostringstream command;
// This is the desired size of the command sent to the server (1MB). The
// actual size sent will be slightly greater than that.
const size_t command_size = 1024 * 1000;
while (command.tellp() < command_size) {
// We're sending command 'foo' with arguments being a list of
// strings. If this is the first transmission, send command name
// and open the arguments list. Also insert the first argument
// so as all subsequent arguments can be prefixed with a comma.
if (command.tellp() == 0) {
command << "{ \"command\": \"foo\", \"arguments\": [ \"begin\"";
} else {
// Generate a random number and insert it into the stream as
// 10 digits long string.
std::ostringstream arg;
arg << setw(10) << std::rand();
// Append the argument in the command.
command << ", \"" << arg.str() << "\"\n";
// If we have hit the limit of the command size, close braces to
// get appropriate JSON.
if (command.tellp() > command_size) {
command << "] }";
}
}
}
ASSERT_NO_THROW(
CommandMgr::instance().registerCommand("foo",
boost::bind(&CtrlChannelDhcpv6SrvTest::longCommandHandler,
command.str(), _1, _2));
);
createUnixChannelServer();
std::string response;
std::thread th([this, &response, &command]() {
// IO service will be stopped automatically when this object goes
// out of scope and is destroyed. This is useful because we use
// asserts which may break the thread in various exit points.
IOServiceWork work(getIOService());
// Create client which we will use to send command to the server.
boost::scoped_ptr<UnixControlClient> client(new UnixControlClient());
ASSERT_TRUE(client);
// Connect to the server. This will trigger acceptor handler on the
// server side and create a new connection.
ASSERT_TRUE(client->connectToServer(socket_path_));
// Initially the remaining_string holds the entire command and we
// will be erasing the portions that we have sent.
std::string remaining_data = command.str();
while (!remaining_data.empty()) {
// Send the command in chunks of 1024 bytes.
const size_t l = remaining_data.size() < 1024 ? remaining_data.size() : 1024;
ASSERT_TRUE(client->sendCommand(remaining_data.substr(0, l)));
remaining_data.erase(0, l);
}
// Set timeout to 5 seconds to allow the time for the server to send
// a response.
const unsigned int timeout = 5;
ASSERT_TRUE(client->getResponse(response, timeout));
// We're done. Close the connection to the server.
client->disconnectFromServer();
});
// Run the server until the command has been processed and response
// received.
getIOService()->run();
// Wait for the thread to complete.
th.join();
EXPECT_EQ("{ \"result\": 0, \"text\": \"long command received ok\" }",
response);
}
// This test verifies that the server can send long response to the client.
TEST_F(CtrlChannelDhcpv6SrvTest, longResponse) {
// We need to generate large response. The simplest way is to create
// a command and a handler which will generate some static response
// of a desired size.
ASSERT_NO_THROW(
CommandMgr::instance().registerCommand("foo",
boost::bind(&CtrlChannelDhcpv6SrvTest::longResponseHandler, _1, _2));
);
createUnixChannelServer();
// The UnixControlClient doesn't have any means to check that the entire
// response has been received. What we want to do is to generate a
// reference response using our command handler and then compare
// what we have received over the unix domain socket with this reference
// response to figure out when to stop receiving.
std::string reference_response = longResponseHandler("foo", ConstElementPtr())->str();
// In this stream we're going to collect out partial responses.
std::ostringstream response;
// The client is synchronous so it is useful to run it in a thread.
std::thread th([this, &response, reference_response]() {
// IO service will be stopped automatically when this object goes
// out of scope and is destroyed. This is useful because we use
// asserts which may break the thread in various exit points.
IOServiceWork work(getIOService());
// Remember the response size so as we know when we should stop
// receiving.
const size_t long_response_size = reference_response.size();
// Create the client and connect it to the server.
boost::scoped_ptr<UnixControlClient> client(new UnixControlClient());
ASSERT_TRUE(client);
ASSERT_TRUE(client->connectToServer(socket_path_));
// Send the stub command.
std::string command = "{ \"command\": \"foo\", \"arguments\": { } }";
ASSERT_TRUE(client->sendCommand(command));
// Keep receiving response data until we have received the full answer.
while (response.tellp() < long_response_size) {
std::string partial;
const unsigned int timeout = 5;
ASSERT_TRUE(client->getResponse(partial, 5));
response << partial;
}
// We have received the entire response, so close the connection and
// stop the IO service.
client->disconnectFromServer();
});
// Run the server until the entire response has been received.
getIOService()->run();
// Wait for the thread to complete.
th.join();
// Make sure we have received correct response.
EXPECT_EQ(reference_response, response.str());
}
// This test verifies that the server signals timeout if the transmission
// takes too long.
TEST_F(CtrlChannelDhcpv6SrvTest, connectionTimeout) {
createUnixChannelServer();
// Set connection timeout to 2s to prevent long waiting time for the
// timeout during this test.
const unsigned short timeout = 2;
CommandMgr::instance().setConnectionTimeout(timeout);
// Server's response will be assigned to this variable.
std::string response;
// It is useful to create a thread and run the server and the client
// at the same time and independently.
std::thread th([this, &response]() {
// IO service will be stopped automatically when this object goes
// out of scope and is destroyed. This is useful because we use
// asserts which may break the thread in various exit points.
IOServiceWork work(getIOService());
// Create the client and connect it to the server.
boost::scoped_ptr<UnixControlClient> client(new UnixControlClient());
ASSERT_TRUE(client);
ASSERT_TRUE(client->connectToServer(socket_path_));
// Send partial command. The server will be waiting for the remaining
// part to be sent and will eventually signal a timeout.
std::string command = "{ \"command\": \"foo\" ";
ASSERT_TRUE(client->sendCommand(command));
// Let's wait up to 15s for the server's response. The response
// should arrive sooner assuming that the timeout mechanism for
// the server is working properly.
const unsigned int timeout = 15;
ASSERT_TRUE(client->getResponse(response, timeout));
// Explicitly close the client's connection.
client->disconnectFromServer();
});
// Run the server until stopped.
getIOService()->run();
// Wait for the thread to return.
th.join();
// Check that the server has signalled a timeout.
EXPECT_EQ("{ \"result\": 1, \"text\": \"Connection over control channel"
" timed out\" }", response);
}
} // End of anonymous namespace

40
src/lib/asiolink/unix_domain_socket.cc
View File

@@ -145,6 +145,12 @@ public:
const boost::system::error_code& ec,
size_t length);
/// @brief Disables read and write operations on the socket.
void shutdown();
/// @brief Cancels asynchronous operations on the socket.
void cancel();
/// @brief Closes the socket.
void close();
@@ -244,9 +250,31 @@ UnixDomainSocketImpl::receiveHandler(const UnixDomainSocket::Handler& remote_han
remote_handler(ec, length);
}
void
UnixDomainSocketImpl::shutdown() {
boost::system::error_code ec;
static_cast<void>(socket_.shutdown(stream_protocol::socket::shutdown_both, ec));
if (ec) {
isc_throw(UnixDomainSocketError, ec.message());
}
}
void
UnixDomainSocketImpl::cancel() {
boost::system::error_code ec;
static_cast<void>(socket_.cancel(ec));
if (ec) {
isc_throw(UnixDomainSocketError, ec.message());
}
}
void
UnixDomainSocketImpl::close() {
static_cast<void>(socket_.close());
boost::system::error_code ec;
static_cast<void>(socket_.close(ec));
if (ec) {
isc_throw(UnixDomainSocketError, ec.message());
}
}
UnixDomainSocket::UnixDomainSocket(IOService& io_service)
@@ -312,6 +340,16 @@ UnixDomainSocket::asyncReceive(void* data, const size_t length,
impl_->asyncReceive(data, length, handler);
}
void
UnixDomainSocket::shutdown() {
impl_->shutdown();
}
void
UnixDomainSocket::cancel() {
impl_->cancel();
}
void
UnixDomainSocket::close() {
impl_->close();

12
src/lib/asiolink/unix_domain_socket.h
View File

@@ -104,7 +104,19 @@ public:
/// error is signalled.
void asyncReceive(void* data, const size_t length, const Handler& handler);
/// @brief Disables read and write operations on the socket.
///
/// @throw UnixDomainSocketError if an error occurs during shutdown.
void shutdown();
/// @brief Cancels scheduled asynchronous operations on the socket.
///
/// @throw UnixDomainSocketError if an error occurs during cancel operation.
void cancel();
/// @brief Closes the socket.
///
/// @throw UnixDomainSocketError if an error occurs during closure.
void close();
/// @brief Returns reference to the underlying ASIO socket.

2
src/lib/cc/json_feed.cc
View File

@@ -141,7 +141,7 @@ JSONFeed::defineStates() {
void
JSONFeed::feedFailure(const std::string& error_msg) {
error_message_ = error_msg + " : " + getContextStr();
error_message_ = error_msg;
transition(FEED_FAILED_ST, FEED_FAILED_EVT);
}

4
src/lib/config/command-socket.dox
View File

@@ -163,9 +163,9 @@ or HTTPS connection):
The @ref isc::config::CommandMgr is implemented using boost ASIO and uses
asynchronous calls to accept new connections and receive commands from the
controlling clients. ASIO uses IO service object to run asynchronous calls.
Thus, before the server can use the @ref isc::dhcp::CommandMgr it must
Thus, before the server can use the @ref isc::config::CommandMgr it must
provide it with a common instance of the @ref isc::asiolink::IOService
object using @ref isc::dhcp::CommandMgr::setIOService. The server's
object using @ref isc::config::CommandMgr::setIOService. The server's
main loop must contain calls to @ref isc::asiolink::IOService::run or
@ref isc::asiolink::IOService::poll or their variants to invoke Command
Manager's handlers as required for processing control requests.

290
src/lib/config/command_mgr.cc
View File

@@ -5,6 +5,7 @@
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
#include <asiolink/asio_wrapper.h>
#include <asiolink/interval_timer.h>
#include <asiolink/io_service.h>
#include <asiolink/unix_domain_socket.h>
#include <asiolink/unix_domain_socket_acceptor.h>
@@ -12,6 +13,7 @@
#include <config/command_mgr.h>
#include <cc/data.h>
#include <cc/command_interpreter.h>
#include <cc/json_feed.h>
#include <dhcp/iface_mgr.h>
#include <config/config_log.h>
#include <boost/bind.hpp>
@@ -26,6 +28,12 @@ using namespace isc::data;
namespace {
/// @brief Maximum size of the data chunk sent/received over the socket.
const size_t BUF_SIZE = 8192;
/// @brief Default connection timeout in seconds.
const unsigned short DEFAULT_CONNECTION_TIMEOUT = 10;
class ConnectionPool;
/// @brief Represents a single connection over control socket.
@@ -40,26 +48,40 @@ public:
/// This constructor registers a socket of this connection in the Interface
/// Manager to cause the blocking call to @c select() to return as soon as
/// a transmission over the control socket is received.
Connection(const boost::shared_ptr<UnixDomainSocket>& socket,
ConnectionPool& connection_pool)
: socket_(socket), connection_pool_(connection_pool),
///
/// @param io_service IOService object used to handle the asio operations
/// @param socket Pointer to the object representing a socket which is used
/// for data transmission.
/// @param connection_pool Reference to the connection pool to which this
/// connection belongs.
/// @param timeout Connection timeout (in seconds).
Connection(const IOServicePtr& io_service,
const boost::shared_ptr<UnixDomainSocket>& socket,
ConnectionPool& connection_pool,
const unsigned short timeout)
: socket_(socket), timeout_timer_(*io_service), timeout_(timeout),
buf_(), response_(), connection_pool_(connection_pool), feed_(),
response_in_progress_(false) {
LOG_INFO(command_logger, COMMAND_SOCKET_CONNECTION_OPENED)
.arg(socket_->getNative());
// Callback value of 0 is used to indicate that callback function is
// not installed.
isc::dhcp::IfaceMgr::instance().addExternalSocket(socket_->getNative(), 0);
// Initialize state model for receiving and preparsing commands.
feed_.initModel();
// Start timer for detecting timeouts.
timeout_timer_.setup(boost::bind(&Connection::timeoutHandler, this),
timeout_ * 1000, IntervalTimer::ONE_SHOT);
}
/// @brief Start asynchronous read over the unix domain socket.
/// @brief Destructor.
///
/// This method doesn't block. Once the transmission is received over the
/// socket, the @c Connection::receiveHandler callback is invoked to
/// process received data.
void start() {
socket_->asyncReceive(&buf_[0], sizeof(buf_),
boost::bind(&Connection::receiveHandler,
shared_from_this(), _1, _2));
/// Cancels timeout timer if one is scheduled.
~Connection() {
timeout_timer_.cancel();
}
/// @brief Close current connection.
@@ -75,28 +97,99 @@ public:
isc::dhcp::IfaceMgr::instance().deleteExternalSocket(socket_->getNative());
socket_->close();
timeout_timer_.cancel();
}
}
/// @brief Gracefully terminates current connection.
///
/// This method should be called prior to closing the socket to initiate
/// graceful shutdown.
void terminate();
/// @brief Start asynchronous read over the unix domain socket.
///
/// This method doesn't block. Once the transmission is received over the
/// socket, the @c Connection::receiveHandler callback is invoked to
/// process received data.
void doReceive() {
socket_->asyncReceive(&buf_[0], sizeof(buf_),
boost::bind(&Connection::receiveHandler,
shared_from_this(), _1, _2));
}
/// @brief Starts asynchronous send over the unix domain socket.
///
/// This method doesn't block. Once the send operation (that covers the whole
/// data if it's small or first BUF_SIZE bytes if its large) is completed, the
/// @c Connection::sendHandler callback is invoked. That handler will either
/// close the connection gracefully if all data has been sent, or will
/// call @ref doSend() again to send the next chunk of data.
void doSend() {
size_t chunk_size = (response_.size() < BUF_SIZE) ? response_.size() : BUF_SIZE;
socket_->asyncSend(&response_[0], chunk_size,
boost::bind(&Connection::sendHandler, shared_from_this(), _1, _2));
}
/// @brief Handler invoked when the data is received over the control
/// socket.
///
/// It collects received data into the @c isc::config::JSONFeed object and
/// schedules additional asynchronous read of data if this object signals
/// that command is incomplete. When the entire command is received, the
/// handler processes this command and asynchronously responds to the
/// controlling client.
//
///
/// @param ec Error code.
/// @param bytes_transferred Number of bytes received.
void receiveHandler(const boost::system::error_code& ec,
size_t bytes_transferred);
/// @brief Handler invoked when the data is sent over the control socket.
///
/// If there are still data to be sent, another asynchronous send is
/// scheduled. When the entire command is sent, the connection is shutdown
/// and closed.
///
/// @param ec Error code.
/// @param bytes_transferred Number of bytes sent.
void sendHandler(const boost::system::error_code& ec,
size_t bytes_trasferred);
/// @brief Handler invoked when timeout has occurred.
///
/// Asynchronously sends a response to the client indicating that the
/// timeout has occurred.
void timeoutHandler();
private:
/// @brief Pointer to the socket used for transmission.
boost::shared_ptr<UnixDomainSocket> socket_;
/// @brief Interval timer used to detect connection timeouts.
IntervalTimer timeout_timer_;
/// @brief Connection timeout (in seconds)
unsigned short timeout_;
/// @brief Buffer used for received data.
std::array<char, 65535> buf_;
std::array<char, BUF_SIZE> buf_;
/// @brief Response created by the server.
std::string response_;
/// @brief Reference to the pool of connections.
ConnectionPool& connection_pool_;
/// @brief State model used to receive data over the connection and detect
/// when the command ends.
JSONFeed feed_;
/// @brief Boolean flag indicating if the request to stop connection is a
/// result of server reconfiguration.
bool response_in_progress_;
@@ -114,7 +207,7 @@ public:
///
/// @param connection Pointer to the new connection object.
void start(const ConnectionPtr& connection) {
connection->start();
connection->doReceive();
connections_.insert(connection);
}
@@ -122,8 +215,13 @@ public:
///
/// @param connection Pointer to the new connection object.
void stop(const ConnectionPtr& connection) {
connection->stop();
connections_.erase(connection);
try {
connection->stop();
connections_.erase(connection);
} catch (const std::exception& ex) {
LOG_ERROR(command_logger, COMMAND_SOCKET_CONNECTION_CLOSE_FAIL)
.arg(ex.what());
}
}
/// @brief Stops all connections which are allowed to stop.
@@ -135,10 +233,6 @@ public:
connections_.clear();
}
size_t getConnectionsNum() const {
return (connections_.size());
}
private:
/// @brief Pool of connections.
@@ -146,6 +240,16 @@ private:
};
void
Connection::terminate() {
try {
socket_->shutdown();
} catch (const std::exception& ex) {
LOG_ERROR(command_logger, COMMAND_SOCKET_CONNECTION_SHUTDOWN_FAIL)
.arg(ex.what());
}
}
void
Connection::receiveHandler(const boost::system::error_code& ec,
@@ -156,15 +260,13 @@ Connection::receiveHandler(const boost::system::error_code& ec,
// connection pool.
LOG_INFO(command_logger, COMMAND_SOCKET_CLOSED_BY_FOREIGN_HOST)
.arg(socket_->getNative());
connection_pool_.stop(shared_from_this());
} else if (ec.value() != boost::asio::error::operation_aborted) {
LOG_ERROR(command_logger, COMMAND_SOCKET_READ_FAIL)
.arg(ec.value()).arg(socket_->getNative());
}
/// @todo: Should we close the connection, similar to what is already
/// being done for bytes_transferred == 0.
connection_pool_.stop(shared_from_this());
return;
} else if (bytes_transferred == 0) {
@@ -176,21 +278,35 @@ Connection::receiveHandler(const boost::system::error_code& ec,
LOG_DEBUG(command_logger, DBG_COMMAND, COMMAND_SOCKET_READ)
.arg(bytes_transferred).arg(socket_->getNative());
ConstElementPtr cmd, rsp;
ConstElementPtr rsp;
try {
// Received some data over the socket. Append them to the JSON feed
// to see if we have reached the end of command.
feed_.postBuffer(&buf_[0], bytes_transferred);
feed_.poll();
// If we haven't yet received the full command, continue receiving.
if (feed_.needData()) {
doReceive();
return;
}
// Try to interpret it as JSON.
std::string sbuf(&buf_[0], bytes_transferred);
cmd = Element::fromJSON(sbuf, true);
// Received entire command. Parse the command into JSON.
if (feed_.feedOk()) {
ConstElementPtr cmd = feed_.toElement();
response_in_progress_ = true;
response_in_progress_ = true;
// If successful, then process it as a command.
rsp = CommandMgr::instance().processCommand(cmd);
// If successful, then process it as a command.
rsp = CommandMgr::instance().processCommand(cmd);
response_in_progress_ = false;
response_in_progress_ = false;
} else {
// Failed to parse command as JSON or process the received command.
// This exception will be caught below and the error response will
// be sent.
isc_throw(BadValue, feed_.getErrorMessage());
}
} catch (const Exception& ex) {
LOG_WARN(command_logger, COMMAND_PROCESS_ERROR1).arg(ex.what());
@@ -203,35 +319,76 @@ Connection::receiveHandler(const boost::system::error_code& ec,
rsp = createAnswer(CONTROL_RESULT_ERROR,
"internal server error: no response generated");
} else {
// Let's convert JSON response to text. Note that at this stage
// the rsp pointer is always set.
response_ = rsp->str();
doSend();
return;
}
// Let's convert JSON response to text. Note that at this stage
// the rsp pointer is always set.
std::string txt = rsp->str();
size_t len = txt.length();
if (len > 65535) {
// Hmm, our response is too large. Let's send the first
// 64KB and hope for the best.
LOG_ERROR(command_logger, COMMAND_SOCKET_RESPONSE_TOOLARGE).arg(len);
len = 65535;
}
try {
// Send the data back over socket.
socket_->write(txt.c_str(), len);
} catch (const std::exception& ex) {
// Response transmission failed. Since the response failed, it doesn't
// make sense to send any status codes. Let's log it and be done with
// it.
LOG_ERROR(command_logger, COMMAND_SOCKET_WRITE_FAIL)
.arg(len).arg(socket_->getNative()).arg(ex.what());
}
// Close the connection if we have sent the entire response.
connection_pool_.stop(shared_from_this());
}
void
Connection::sendHandler(const boost::system::error_code& ec,
size_t bytes_transferred) {
if (ec) {
// If an error occurred, log this error and stop the connection.
if (ec.value() != boost::asio::error::operation_aborted) {
LOG_ERROR(command_logger, COMMAND_SOCKET_WRITE_FAIL)
.arg(socket_->getNative()).arg(ec.message());
}
} else {
// No error. We are in a process of sending a response. Need to
// remove the chunk that we have managed to sent with the previous
// attempt.
response_.erase(0, bytes_transferred);
LOG_DEBUG(command_logger, DBG_COMMAND, COMMAND_SOCKET_WRITE)
.arg(bytes_transferred).arg(response_.size())
.arg(socket_->getNative());
// Check if there is any data left to be sent and sent it.
if (!response_.empty()) {
doSend();
return;
}
// Gracefully shutdown the connection and close the socket if
// we have sent the whole response.
terminate();
}
// All data sent or an error has occurred. Close the connection.
connection_pool_.stop(shared_from_this());
}
void
Connection::timeoutHandler() {
LOG_INFO(command_logger, COMMAND_SOCKET_CONNECTION_TIMEOUT)
.arg(socket_->getNative());
try {
socket_->cancel();
} catch (const std::exception& ex) {
LOG_ERROR(command_logger, COMMAND_SOCKET_CONNECTION_CANCEL_FAIL)
.arg(socket_->getNative())
.arg(ex.what());
}
ConstElementPtr rsp = createAnswer(CONTROL_RESULT_ERROR, "Connection over"
" control channel timed out");
response_ = rsp->str();
doSend();
}
}
namespace isc {
@@ -244,7 +401,7 @@ public:
/// @brief Constructor.
CommandMgrImpl()
: io_service_(), acceptor_(), socket_(), socket_name_(),
connection_pool_() {
connection_pool_(), timeout_(DEFAULT_CONNECTION_TIMEOUT) {
}
/// @brief Opens acceptor service allowing the control clients to connect.
@@ -274,6 +431,9 @@ public:
/// @brief Pool of connections.
ConnectionPool connection_pool_;
/// @brief Connection timeout
unsigned short timeout_;
};
void
@@ -333,8 +493,14 @@ CommandMgrImpl::doAccept() {
acceptor_->asyncAccept(*socket_, [this](const boost::system::error_code& ec) {
if (!ec) {
// New connection is arriving. Start asynchronous transmission.
ConnectionPtr connection(new Connection(socket_, connection_pool_));
ConnectionPtr connection(new Connection(io_service_, socket_,
connection_pool_,
timeout_));
connection_pool_.start(connection);
} else if (ec.value() != boost::asio::error::operation_aborted) {
LOG_ERROR(command_logger, COMMAND_SOCKET_ACCEPT_FAIL)
.arg(acceptor_->getNative()).arg(ec.message());
}
// Unless we're stopping the service, start accepting connections again.
@@ -385,5 +551,11 @@ CommandMgr::setIOService(const IOServicePtr& io_service) {
impl_->io_service_ = io_service;
}
void
CommandMgr::setConnectionTimeout(const unsigned short timeout) {
impl_->timeout_ = timeout;
}
}; // end of isc::config
}; // end of isc

5
src/lib/config/command_mgr.h
View File

@@ -54,6 +54,11 @@ public:
/// @param io_service Pointer to the IO service.
void setIOService(const asiolink::IOServicePtr& io_service);
/// @brief Override default connection timeout.
///
/// @param timeout New connection timeout in seconds.
void setConnectionTimeout(const unsigned short timeout);
/// @brief Opens control socket with parameters specified in socket_info
///
/// Currently supported types are:

51
src/lib/config/config_messages.mes
View File

@@ -59,26 +59,34 @@ information may be provided by the system as second parameter.
This is an information message indicating that the command connection has been
closed by a command control client.
% COMMAND_SOCKET_CONNECTION_CANCEL_FAIL Failed to cancel read operation on socket %1: %2
This error message is issued to indicate an error to cancel asynchronous read
of the control command over the control socket. The cancel operation is performed
when the timeout occurs during communication with a client. The error message
includes details about the reason for failure.
% COMMAND_SOCKET_CONNECTION_CLOSED Closed socket %1 for existing command connection
This is an informational message that the socket created for handling
client's connection is closed. This usually means that the client disconnected,
but may also mean a timeout.
% COMMAND_SOCKET_CONNECTION_OPENED Opened socket %1 for incoming command connection on socket %2
% COMMAND_SOCKET_CONNECTION_CLOSE_FAIL Failed to close command connection: %1
This error message is issued when an error occurred when closing a
command connection and/or removing it from the connections pool. The
detailed error is provided as an argument.
% COMMAND_SOCKET_CONNECTION_OPENED Opened socket %1 for incoming command connection
This is an informational message that a new incoming command connection was
detected and a dedicated socket was opened for that connection.
% COMMAND_SOCKET_DUP_WARN Failed to duplicate socket for response: %1
This debug message indicates that the commandReader was unable to duplicate
the connection socket prior to executing the command. This is most likely a
system resource issue. The command should still be processed and the response
sent, unless the command caused the command channel to be closed (e.g. a
reconfiguration command).
% COMMAND_SOCKET_CONNECTION_SHUTDOWN_FAIL Encountered error %1 while trying to gracefully shutdown socket
This message indicates an error while trying to gracefully shutdown command
connection. The type of the error is included in the message.
% COMMAND_SOCKET_FAIL_NONBLOCK Failed to set non-blocking mode for socket %1 created for incoming connection on socket %2: %3
This error message indicates that the server failed to set non-blocking mode
on just created socket. That socket was created for accepting specific
incoming connection. Additional information may be provided as third parameter.
% COMMAND_SOCKET_CONNECTION_TIMEOUT Timeout occurred for connection over socket %1
This is an informational message that indicates that the timeout has
occurred for one of the command channel connections. The response
sent by the server indicates a timeout and is then closed.
% COMMAND_SOCKET_READ Received %1 bytes over command socket %2
This debug message indicates that specified number of bytes was received
@@ -88,27 +96,10 @@ over command socket identified by specified file descriptor.
This error message indicates that an error was encountered while
reading from command socket.
% COMMAND_SOCKET_RESPONSE_TOOLARGE Server's response was larger (%1) than supported 64KB
This error message indicates that the server received a command and generated
an answer for it, but that response was larger than supported 64KB. Server
will attempt to send the first 64KB of the response. Depending on the nature
of this response, this may indicate a software or configuration error. Future
Kea versions are expected to have better support for large responses.
% COMMAND_SOCKET_UNIX_CLOSE Command socket closed: UNIX, fd=%1, path=%2
This informational message indicates that the daemon closed a command
processing socket. This was a UNIX socket. It was opened with the file
descriptor and path specified.
% COMMAND_SOCKET_UNIX_OPEN Command socket opened: UNIX, fd=%1, path=%2
This informational message indicates that the daemon opened a command
processing socket. This is a UNIX socket. It was opened with the file
descriptor and path specified.
% COMMAND_SOCKET_WRITE Sent response of %1 bytes over command socket %2
% COMMAND_SOCKET_WRITE Sent response of %1 bytes (%2 bytes left to send) over command socket %3
This debug message indicates that the specified number of bytes was sent
over command socket identifier by the specified file descriptor.
% COMMAND_SOCKET_WRITE_FAIL Error while writing %1 bytes to command socket %2 : %3
% COMMAND_SOCKET_WRITE_FAIL Error while writing to command socket %1 : %2
This error message indicates that an error was encountered while
attempting to send a response to the command socket.

10
src/lib/testutils/unix_control_client.cc
View File

@@ -84,18 +84,18 @@ bool UnixControlClient::sendCommand(const std::string& command) {
return (true);
}
bool UnixControlClient::getResponse(std::string& response) {
bool UnixControlClient::getResponse(std::string& response,
const unsigned int timeout_sec) {
// Receive response
char buf[65536];
memset(buf, 0, sizeof(buf));
switch (selectCheck()) {
switch (selectCheck(timeout_sec)) {
case -1: {
const char* errmsg = strerror(errno);
ADD_FAILURE() << "getResponse - select failed: " << errmsg;
return (false);
}
case 0:
ADD_FAILURE() << "No response data sent";
return (false);
default:
@@ -120,7 +120,7 @@ bool UnixControlClient::getResponse(std::string& response) {
return (true);
}
int UnixControlClient::selectCheck() {
int UnixControlClient::selectCheck(const unsigned int timeout_sec) {
int maxfd = 0;
fd_set read_fds;
@@ -131,7 +131,7 @@ int UnixControlClient::selectCheck() {
maxfd = socket_fd_;
struct timeval select_timeout;
select_timeout.tv_sec = 0;
select_timeout.tv_sec = static_cast<time_t>(timeout_sec);
select_timeout.tv_usec = 0;
return (select(maxfd + 1, &read_fds, NULL, NULL, &select_timeout));

7
src/lib/testutils/unix_control_client.h
View File

@@ -44,13 +44,16 @@ public:
/// @brief Reads the response text from the open Control Channel
/// @param response variable into which the received response should be
/// placed.
/// @param timeout_sec Timeout for receiving response in seconds.
/// @return true if data was successfully read from the socket,
/// false otherwise
bool getResponse(std::string& response);
bool getResponse(std::string& response, const unsigned int timeout_sec = 0);
/// @brief Uses select to poll the Control Channel for data waiting
///
/// @param timeout_sec Select timeout in seconds
/// @return -1 on error, 0 if no data is available, 1 if data is ready
int selectCheck();
int selectCheck(const unsigned int timeout_sec);
/// @brief Retains the fd of the open socket
int socket_fd_;