2014-05-03 20:02:22 +02:00
/******************************************************************************
* Icinga 2 *
2016-01-12 08:29:59 +01:00
* Copyright ( C ) 2012 - 2016 Icinga Development Team ( https : //www.icinga.org/) *
2014-05-03 20:02:22 +02:00
* *
* This program is free software ; you can redistribute it and / or *
* modify it under the terms of the GNU General Public License *
* as published by the Free Software Foundation ; either version 2 *
* of the License , or ( at your option ) any later version . *
* *
* This program is distributed in the hope that it will be useful , *
* but WITHOUT ANY WARRANTY ; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE . See the *
* GNU General Public License for more details . *
* *
* You should have received a copy of the GNU General Public License *
* along with this program ; if not , write to the Free Software Foundation *
* Inc . , 51 Franklin St , Fifth Floor , Boston , MA 02110 - 1301 , USA . *
* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
2014-05-25 16:23:35 +02:00
# include "remote/apilistener.hpp"
2015-03-28 11:04:42 +01:00
# include "remote/apilistener.tcpp"
2015-06-22 11:11:21 +02:00
# include "remote/jsonrpcconnection.hpp"
2014-05-25 16:23:35 +02:00
# include "remote/endpoint.hpp"
2015-02-26 14:59:39 +01:00
# include "remote/jsonrpc.hpp"
2015-08-11 14:13:53 +02:00
# include "remote/apifunction.hpp"
2014-05-25 16:23:35 +02:00
# include "base/convert.hpp"
# include "base/netstring.hpp"
2014-10-26 19:59:49 +01:00
# include "base/json.hpp"
2015-08-15 20:28:05 +02:00
# include "base/configtype.hpp"
2014-10-19 14:21:12 +02:00
# include "base/logger.hpp"
2014-05-25 16:23:35 +02:00
# include "base/objectlock.hpp"
# include "base/stdiostream.hpp"
# include "base/application.hpp"
# include "base/context.hpp"
# include "base/statsfunction.hpp"
2014-12-15 10:16:06 +01:00
# include "base/exception.hpp"
2014-05-03 20:02:22 +02:00
# include <fstream>
using namespace icinga ;
REGISTER_TYPE ( ApiListener ) ;
boost : : signals2 : : signal < void ( bool ) > ApiListener : : OnMasterChanged ;
2015-11-24 15:25:55 +01:00
ApiListener : : Ptr ApiListener : : m_Instance ;
2014-05-03 20:02:22 +02:00
2015-09-21 11:44:58 +02:00
REGISTER_STATSFUNCTION ( ApiListener , & ApiListener : : StatsFunc ) ;
2014-05-03 20:02:22 +02:00
2015-08-11 14:13:53 +02:00
REGISTER_APIFUNCTION ( Hello , icinga , & ApiListener : : HelloAPIHandler ) ;
2015-03-02 09:56:09 +01:00
ApiListener : : ApiListener ( void )
2015-11-25 13:11:41 +01:00
: m_SyncQueue ( 0 , 4 ) , m_LogMessageCount ( 0 )
2016-06-14 08:19:13 +02:00
{
m_RelayQueue . SetName ( " ApiListener, RelayQueue " ) ;
m_SyncQueue . SetName ( " ApiListener, SyncQueue " ) ;
}
2015-03-02 09:56:09 +01:00
2014-05-03 20:02:22 +02:00
void ApiListener : : OnConfigLoaded ( void )
{
2015-11-30 15:31:11 +01:00
if ( m_Instance )
BOOST_THROW_EXCEPTION ( ScriptError ( " Only one ApiListener object is allowed. " , GetDebugInfo ( ) ) ) ;
m_Instance = this ;
2014-05-03 20:02:22 +02:00
/* set up SSL context */
2014-11-08 21:17:16 +01:00
boost : : shared_ptr < X509 > cert ;
2014-06-05 15:03:56 +02:00
try {
cert = GetX509Certificate ( GetCertPath ( ) ) ;
2014-08-25 08:35:35 +02:00
} catch ( const std : : exception & ) {
2015-09-22 17:58:12 +02:00
BOOST_THROW_EXCEPTION ( ScriptError ( " Cannot get certificate from cert path: ' "
+ GetCertPath ( ) + " '. " , GetDebugInfo ( ) ) ) ;
2014-06-05 15:03:56 +02:00
}
try {
SetIdentity ( GetCertificateCN ( cert ) ) ;
2014-08-25 08:35:35 +02:00
} catch ( const std : : exception & ) {
2015-09-22 17:58:12 +02:00
BOOST_THROW_EXCEPTION ( ScriptError ( " Cannot get certificate common name from cert path: ' "
+ GetCertPath ( ) + " '. " , GetDebugInfo ( ) ) ) ;
2014-06-05 15:03:56 +02:00
}
2014-10-20 10:09:57 +02:00
Log ( LogInformation , " ApiListener " )
< < " My API identity: " < < GetIdentity ( ) ;
2014-05-03 20:02:22 +02:00
2014-06-05 15:03:56 +02:00
try {
m_SSLContext = MakeSSLContext ( GetCertPath ( ) , GetKeyPath ( ) , GetCaPath ( ) ) ;
2014-08-25 08:35:35 +02:00
} catch ( const std : : exception & ) {
2015-09-22 17:58:12 +02:00
BOOST_THROW_EXCEPTION ( ScriptError ( " Cannot make SSL context for cert path: ' "
+ GetCertPath ( ) + " ' key path: ' " + GetKeyPath ( ) + " ' ca path: ' " + GetCaPath ( ) + " '. " , GetDebugInfo ( ) ) ) ;
2014-06-05 15:03:56 +02:00
}
2014-05-03 20:02:22 +02:00
2014-06-05 15:03:56 +02:00
if ( ! GetCrlPath ( ) . IsEmpty ( ) ) {
try {
AddCRLToSSLContext ( m_SSLContext , GetCrlPath ( ) ) ;
2014-08-25 08:35:35 +02:00
} catch ( const std : : exception & ) {
2015-09-22 17:58:12 +02:00
BOOST_THROW_EXCEPTION ( ScriptError ( " Cannot add certificate revocation list to SSL context for crl path: ' "
+ GetCrlPath ( ) + " '. " , GetDebugInfo ( ) ) ) ;
2014-06-05 15:03:56 +02:00
}
}
2016-03-25 21:25:19 +01:00
if ( ! GetCipherList ( ) . IsEmpty ( ) ) {
try {
SetCipherListToSSLContext ( m_SSLContext , GetCipherList ( ) ) ;
} catch ( const std : : exception & ) {
BOOST_THROW_EXCEPTION ( ScriptError ( " Cannot set cipher list to SSL context for cipher list: ' "
+ GetCipherList ( ) + " '. " , GetDebugInfo ( ) ) ) ;
}
}
2014-11-21 23:23:31 +01:00
}
2014-05-03 20:02:22 +02:00
2014-11-21 23:23:31 +01:00
void ApiListener : : OnAllConfigLoaded ( void )
{
2015-11-24 15:25:55 +01:00
m_LocalEndpoint = Endpoint : : GetByName ( GetIdentity ( ) ) ;
if ( ! m_LocalEndpoint )
2015-02-26 17:09:45 +01:00
BOOST_THROW_EXCEPTION ( ScriptError ( " Endpoint object for ' " + GetIdentity ( ) + " ' is missing. " , GetDebugInfo ( ) ) ) ;
2014-05-03 20:02:22 +02:00
}
/**
* Starts the component .
*/
2015-08-20 17:18:48 +02:00
void ApiListener : : Start ( bool runtimeCreated )
2014-05-03 20:02:22 +02:00
{
2014-11-21 23:23:31 +01:00
SyncZoneDirs ( ) ;
2015-08-20 17:18:48 +02:00
ObjectImpl < ApiListener > : : Start ( runtimeCreated ) ;
2014-05-03 20:02:22 +02:00
{
boost : : mutex : : scoped_lock ( m_LogLock ) ;
RotateLogFile ( ) ;
OpenLogFile ( ) ;
}
/* create the primary JSON-RPC listener */
2014-08-25 08:27:19 +02:00
if ( ! AddListener ( GetBindHost ( ) , GetBindPort ( ) ) ) {
2014-10-20 10:09:57 +02:00
Log ( LogCritical , " ApiListener " )
< < " Cannot add listener on host ' " < < GetBindHost ( ) < < " ' for port ' " < < GetBindPort ( ) < < " '. " ;
2015-03-12 11:43:04 +01:00
Application : : Exit ( EXIT_FAILURE ) ;
2014-08-04 16:34:17 +02:00
}
2014-05-03 20:02:22 +02:00
2014-11-08 21:17:16 +01:00
m_Timer = new Timer ( ) ;
2014-05-03 20:02:22 +02:00
m_Timer - > OnTimerExpired . connect ( boost : : bind ( & ApiListener : : ApiTimerHandler , this ) ) ;
m_Timer - > SetInterval ( 5 ) ;
m_Timer - > Start ( ) ;
m_Timer - > Reschedule ( 0 ) ;
2016-07-21 12:27:18 +02:00
m_ReconnectTimer = new Timer ( ) ;
m_ReconnectTimer - > OnTimerExpired . connect ( boost : : bind ( & ApiListener : : ApiReconnectTimerHandler , this ) ) ;
m_ReconnectTimer - > SetInterval ( 60 ) ;
m_ReconnectTimer - > Start ( ) ;
m_ReconnectTimer - > Reschedule ( 0 ) ;
2014-05-03 20:02:22 +02:00
OnMasterChanged ( true ) ;
}
ApiListener : : Ptr ApiListener : : GetInstance ( void )
{
2015-11-24 15:25:55 +01:00
return m_Instance ;
2014-05-03 20:02:22 +02:00
}
2014-11-08 21:17:16 +01:00
boost : : shared_ptr < SSL_CTX > ApiListener : : GetSSLContext ( void ) const
2014-05-03 20:02:22 +02:00
{
return m_SSLContext ;
}
Endpoint : : Ptr ApiListener : : GetMaster ( void ) const
{
Zone : : Ptr zone = Zone : : GetLocalZone ( ) ;
2014-08-04 16:34:17 +02:00
if ( ! zone )
return Endpoint : : Ptr ( ) ;
2014-05-03 20:02:22 +02:00
std : : vector < String > names ;
BOOST_FOREACH ( const Endpoint : : Ptr & endpoint , zone - > GetEndpoints ( ) )
2015-10-22 10:52:38 +02:00
if ( endpoint - > GetConnected ( ) | | endpoint - > GetName ( ) = = GetIdentity ( ) )
2014-05-03 20:02:22 +02:00
names . push_back ( endpoint - > GetName ( ) ) ;
std : : sort ( names . begin ( ) , names . end ( ) ) ;
return Endpoint : : GetByName ( * names . begin ( ) ) ;
}
bool ApiListener : : IsMaster ( void ) const
{
2014-08-04 16:34:17 +02:00
Endpoint : : Ptr master = GetMaster ( ) ;
if ( ! master )
return false ;
2015-11-24 15:25:55 +01:00
return master = = GetLocalEndpoint ( ) ;
2014-05-03 20:02:22 +02:00
}
/**
* Creates a new JSON - RPC listener on the specified port .
*
2014-08-25 08:27:19 +02:00
* @ param node The host the listener should be bound to .
2014-05-03 20:02:22 +02:00
* @ param service The port to listen on .
*/
2014-08-25 08:27:19 +02:00
bool ApiListener : : AddListener ( const String & node , const String & service )
2014-05-03 20:02:22 +02:00
{
ObjectLock olock ( this ) ;
2014-11-08 21:17:16 +01:00
boost : : shared_ptr < SSL_CTX > sslContext = m_SSLContext ;
2014-05-03 20:02:22 +02:00
2014-06-05 14:36:57 +02:00
if ( ! sslContext ) {
Log ( LogCritical , " ApiListener " , " SSL context is required for AddListener() " ) ;
2014-08-04 16:34:17 +02:00
return false ;
2014-06-05 14:36:57 +02:00
}
2014-05-03 20:02:22 +02:00
2014-10-20 10:09:57 +02:00
Log ( LogInformation , " ApiListener " )
< < " Adding new listener on port ' " < < service < < " ' " ;
2014-05-03 20:02:22 +02:00
2014-11-08 21:17:16 +01:00
TcpSocket : : Ptr server = new TcpSocket ( ) ;
2014-06-05 14:36:57 +02:00
try {
2014-08-25 08:27:19 +02:00
server - > Bind ( node , service , AF_UNSPEC ) ;
} catch ( const std : : exception & ) {
2014-10-20 10:09:57 +02:00
Log ( LogCritical , " ApiListener " )
< < " Cannot bind TCP socket for host ' " < < node < < " ' on port ' " < < service < < " '. " ;
2014-08-04 16:34:17 +02:00
return false ;
2014-06-05 14:36:57 +02:00
}
2014-05-03 20:02:22 +02:00
boost : : thread thread ( boost : : bind ( & ApiListener : : ListenerThreadProc , this , server ) ) ;
thread . detach ( ) ;
m_Servers . insert ( server ) ;
2014-08-04 16:34:17 +02:00
return true ;
2014-05-03 20:02:22 +02:00
}
void ApiListener : : ListenerThreadProc ( const Socket : : Ptr & server )
{
Utility : : SetThreadName ( " API Listener " ) ;
server - > Listen ( ) ;
for ( ; ; ) {
2014-06-05 16:17:53 +02:00
try {
Socket : : Ptr client = server - > Accept ( ) ;
2015-06-22 11:11:21 +02:00
boost : : thread thread ( boost : : bind ( & ApiListener : : NewClientHandler , this , client , String ( ) , RoleServer ) ) ;
thread . detach ( ) ;
2014-08-25 08:35:35 +02:00
} catch ( const std : : exception & ) {
2014-06-05 16:17:53 +02:00
Log ( LogCritical , " ApiListener " , " Cannot accept new connection. " ) ;
}
2014-05-03 20:02:22 +02:00
}
}
/**
2014-06-23 10:00:02 +02:00
* Creates a new JSON - RPC client and connects to the specified endpoint .
2014-05-03 20:02:22 +02:00
*
2014-06-23 10:00:02 +02:00
* @ param endpoint The endpoint .
2014-05-03 20:02:22 +02:00
*/
2014-06-23 10:00:02 +02:00
void ApiListener : : AddConnection ( const Endpoint : : Ptr & endpoint )
2014-05-22 13:45:42 +02:00
{
2014-05-03 20:02:22 +02:00
{
ObjectLock olock ( this ) ;
2014-11-08 21:17:16 +01:00
boost : : shared_ptr < SSL_CTX > sslContext = m_SSLContext ;
2014-05-03 20:02:22 +02:00
2014-06-05 14:36:57 +02:00
if ( ! sslContext ) {
2014-08-04 16:34:17 +02:00
Log ( LogCritical , " ApiListener " , " SSL context is required for AddConnection() " ) ;
2014-06-05 14:36:57 +02:00
return ;
}
2014-05-03 20:02:22 +02:00
}
2014-06-23 10:00:02 +02:00
String host = endpoint - > GetHost ( ) ;
String port = endpoint - > GetPort ( ) ;
2015-06-22 11:11:21 +02:00
Log ( LogInformation , " JsonRpcConnection " )
2014-10-20 10:09:57 +02:00
< < " Reconnecting to API endpoint ' " < < endpoint - > GetName ( ) < < " ' via host ' " < < host < < " ' and port ' " < < port < < " ' " ;
2014-08-22 15:39:34 +02:00
2014-11-08 21:17:16 +01:00
TcpSocket : : Ptr client = new TcpSocket ( ) ;
2014-05-03 20:02:22 +02:00
2014-05-22 13:45:42 +02:00
try {
2014-06-23 10:00:02 +02:00
endpoint - > SetConnecting ( true ) ;
client - > Connect ( host , port ) ;
2015-03-05 14:15:42 +01:00
NewClientHandler ( client , endpoint - > GetName ( ) , RoleClient ) ;
2014-06-23 10:00:02 +02:00
endpoint - > SetConnecting ( false ) ;
2014-05-22 13:45:42 +02:00
} catch ( const std : : exception & ex ) {
2014-06-23 10:00:02 +02:00
endpoint - > SetConnecting ( false ) ;
2014-08-04 13:35:12 +02:00
client - > Close ( ) ;
2014-06-23 10:00:02 +02:00
2014-10-20 10:09:57 +02:00
std : : ostringstream info ;
2014-06-23 10:00:02 +02:00
info < < " Cannot connect to host ' " < < host < < " ' on port ' " < < port < < " ' " ;
2014-06-05 14:36:57 +02:00
Log ( LogCritical , " ApiListener " , info . str ( ) ) ;
2014-10-20 10:09:57 +02:00
Log ( LogDebug , " ApiListener " )
< < info . str ( ) < < " \n " < < DiagnosticInformation ( ex ) ;
2014-05-22 13:45:42 +02:00
}
2014-05-03 20:02:22 +02:00
}
2015-08-11 12:08:54 +02:00
void ApiListener : : NewClientHandler ( const Socket : : Ptr & client , const String & hostname , ConnectionRole role )
{
try {
2015-08-11 12:56:30 +02:00
NewClientHandlerInternal ( client , hostname , role ) ;
2015-08-11 12:08:54 +02:00
} catch ( const std : : exception & ex ) {
Log ( LogCritical , " ApiListener " )
< < " Exception while handling new API client connection: " < < DiagnosticInformation ( ex ) ;
}
}
2014-05-03 20:02:22 +02:00
/**
* Processes a new client connection .
*
* @ param client The new client .
*/
2015-08-11 12:08:54 +02:00
void ApiListener : : NewClientHandlerInternal ( const Socket : : Ptr & client , const String & hostname , ConnectionRole role )
2014-05-03 20:02:22 +02:00
{
CONTEXT ( " Handling new API client connection " ) ;
TlsStream : : Ptr tlsStream ;
{
ObjectLock olock ( this ) ;
2014-06-05 15:03:56 +02:00
try {
2015-03-05 14:15:42 +01:00
tlsStream = new TlsStream ( client , hostname , role , m_SSLContext ) ;
2014-08-25 08:35:35 +02:00
} catch ( const std : : exception & ) {
2014-10-20 10:09:57 +02:00
Log ( LogCritical , " ApiListener " , " Cannot create TLS stream from client connection. " ) ;
2014-06-05 15:03:56 +02:00
return ;
}
2014-05-03 20:02:22 +02:00
}
2014-06-05 15:03:56 +02:00
try {
tlsStream - > Handshake ( ) ;
2015-06-22 11:11:21 +02:00
} catch ( const std : : exception & ex ) {
Log ( LogCritical , " ApiListener " , " Client TLS handshake failed " ) ;
2014-06-05 15:03:56 +02:00
return ;
}
2014-05-03 20:02:22 +02:00
2014-11-08 21:17:16 +01:00
boost : : shared_ptr < X509 > cert = tlsStream - > GetPeerCertificate ( ) ;
2014-06-05 15:03:56 +02:00
String identity ;
2015-06-22 11:11:21 +02:00
Endpoint : : Ptr endpoint ;
bool verify_ok = false ;
2014-06-05 15:03:56 +02:00
2016-07-25 09:31:38 +02:00
String conninfo ;
if ( role = = RoleClient )
conninfo = " to " ;
else
conninfo = " from " ;
conninfo + = " " + client - > GetPeerAddress ( ) ;
2015-06-22 11:11:21 +02:00
if ( cert ) {
try {
identity = GetCertificateCN ( cert ) ;
} catch ( const std : : exception & ) {
Log ( LogCritical , " ApiListener " )
< < " Cannot get certificate common name from cert path: ' " < < GetCertPath ( ) < < " '. " ;
return ;
}
2014-05-03 20:02:22 +02:00
2015-06-22 11:11:21 +02:00
verify_ok = tlsStream - > IsVerifyOK ( ) ;
2016-02-09 15:55:12 +01:00
if ( ! hostname . IsEmpty ( ) ) {
if ( identity ! = hostname ) {
Log ( LogWarning , " ApiListener " )
< < " Unexpected certificate common name while connecting to endpoint ' "
< < hostname < < " ': got ' " < < identity < < " ' " ;
return ;
} else if ( ! verify_ok ) {
Log ( LogWarning , " ApiListener " )
< < " Peer certificate for endpoint ' " < < hostname
< < " ' is not signed by the certificate authority. " ;
return ;
}
}
2014-05-08 15:00:09 +02:00
2015-06-22 11:11:21 +02:00
if ( verify_ok )
endpoint = Endpoint : : GetByName ( identity ) ;
2016-07-21 13:48:00 +02:00
{
Log log ( LogInformation , " ApiListener " ) ;
2016-07-25 09:31:38 +02:00
log < < " New client connection for identity ' " < < identity < < " ' " < < conninfo ;
2016-07-21 13:48:00 +02:00
if ( ! verify_ok )
2016-07-21 22:00:32 +02:00
log < < " (certificate validation failed: " < < tlsStream - > GetVerifyError ( ) < < " ) " ;
2016-07-21 13:48:00 +02:00
else if ( ! endpoint )
log < < " (no Endpoint object found for identity) " ;
}
2015-06-22 11:11:21 +02:00
} else {
Log ( LogInformation , " ApiListener " )
2016-07-25 09:31:38 +02:00
< < " New client connection " < < conninfo < < " (no client certificate) " ;
2015-06-22 11:11:21 +02:00
}
2014-05-03 20:02:22 +02:00
2015-06-22 11:11:21 +02:00
ClientType ctype ;
2014-05-03 20:02:22 +02:00
2015-06-22 11:11:21 +02:00
if ( role = = RoleClient ) {
Dictionary : : Ptr message = new Dictionary ( ) ;
message - > Set ( " jsonrpc " , " 2.0 " ) ;
message - > Set ( " method " , " icinga::Hello " ) ;
message - > Set ( " params " , new Dictionary ( ) ) ;
JsonRpc : : SendMessage ( tlsStream , message ) ;
ctype = ClientJsonRpc ;
} else {
tlsStream - > WaitForData ( 5 ) ;
2014-08-05 09:10:59 +02:00
2015-06-22 11:11:21 +02:00
if ( ! tlsStream - > IsDataAvailable ( ) ) {
2016-07-28 17:33:15 +02:00
Log ( LogWarning , " ApiListener " )
< < " No data received on new API connection for identity ' " < < identity < < " '. Ensure that the remote endpoints are properly configured in a cluster setup. " ;
2015-06-22 11:11:21 +02:00
return ;
}
char firstByte ;
tlsStream - > Peek ( & firstByte , 1 , false ) ;
2014-05-03 20:02:22 +02:00
2015-06-22 11:11:21 +02:00
if ( firstByte > = ' 0 ' & & firstByte < = ' 9 ' )
ctype = ClientJsonRpc ;
else
ctype = ClientHttp ;
}
if ( ctype = = ClientJsonRpc ) {
2015-09-30 08:41:09 +02:00
Log ( LogNotice , " ApiListener " , " New JSON-RPC client " ) ;
2015-06-22 11:11:21 +02:00
JsonRpcConnection : : Ptr aclient = new JsonRpcConnection ( identity , verify_ok , tlsStream , role ) ;
aclient - > Start ( ) ;
if ( endpoint ) {
2016-02-04 11:30:27 +01:00
bool needSync = ! endpoint - > GetConnected ( ) ;
2015-06-22 11:11:21 +02:00
endpoint - > AddClient ( aclient ) ;
2016-02-04 11:30:27 +01:00
m_SyncQueue . Enqueue ( boost : : bind ( & ApiListener : : SyncClient , this , aclient , endpoint , needSync ) ) ;
2015-06-22 11:11:21 +02:00
} else
AddAnonymousClient ( aclient ) ;
} else {
2015-09-30 08:41:09 +02:00
Log ( LogNotice , " ApiListener " , " New HTTP client " ) ;
2014-05-03 20:02:22 +02:00
2015-08-29 01:16:16 +02:00
HttpServerConnection : : Ptr aclient = new HttpServerConnection ( identity , verify_ok , tlsStream ) ;
2015-06-22 11:11:21 +02:00
aclient - > Start ( ) ;
AddHttpClient ( aclient ) ;
}
2014-05-03 20:02:22 +02:00
}
2016-02-04 11:30:27 +01:00
void ApiListener : : SyncClient ( const JsonRpcConnection : : Ptr & aclient , const Endpoint : : Ptr & endpoint , bool needSync )
2015-11-25 13:11:41 +01:00
{
try {
{
ObjectLock olock ( endpoint ) ;
endpoint - > SetSyncing ( true ) ;
}
2016-02-04 11:30:27 +01:00
/* Make sure that the config updates are synced
* before the logs are replayed .
*/
2015-11-25 13:11:41 +01:00
Log ( LogInformation , " ApiListener " )
2016-02-04 11:30:27 +01:00
< < " Sending config updates for endpoint ' " < < endpoint - > GetName ( ) < < " '. " ;
2015-11-25 13:11:41 +01:00
/* sync zone file config */
SendConfigUpdate ( aclient ) ;
/* sync runtime config */
SendRuntimeConfigObjects ( aclient ) ;
Log ( LogInformation , " ApiListener " )
2016-02-04 11:30:27 +01:00
< < " Finished sending config updates for endpoint ' " < < endpoint - > GetName ( ) < < " '. " ;
2016-02-08 13:15:24 +01:00
if ( ! needSync ) {
ObjectLock olock2 ( endpoint ) ;
endpoint - > SetSyncing ( false ) ;
2016-02-04 11:30:27 +01:00
return ;
2016-02-08 13:15:24 +01:00
}
2016-02-04 11:30:27 +01:00
Log ( LogInformation , " ApiListener " )
< < " Sending replay log for endpoint ' " < < endpoint - > GetName ( ) < < " '. " ;
2015-11-25 13:11:41 +01:00
ReplayLog ( aclient ) ;
2016-02-04 11:30:27 +01:00
2016-05-11 13:04:39 +02:00
if ( endpoint - > GetZone ( ) = = Zone : : GetLocalZone ( ) )
UpdateObjectAuthority ( ) ;
2016-02-04 11:30:27 +01:00
Log ( LogInformation , " ApiListener " )
< < " Finished sending replay log for endpoint ' " < < endpoint - > GetName ( ) < < " '. " ;
2015-11-25 13:11:41 +01:00
} catch ( const std : : exception & ex ) {
2016-02-08 13:15:24 +01:00
ObjectLock olock2 ( endpoint ) ;
endpoint - > SetSyncing ( false ) ;
2015-11-25 13:11:41 +01:00
Log ( LogCritical , " ApiListener " )
< < " Error while syncing endpoint ' " < < endpoint - > GetName ( ) < < " ': " < < DiagnosticInformation ( ex ) ;
}
}
2014-05-03 20:02:22 +02:00
void ApiListener : : ApiTimerHandler ( void )
{
double now = Utility : : GetTime ( ) ;
std : : vector < int > files ;
Utility : : Glob ( GetApiDir ( ) + " log/* " , boost : : bind ( & ApiListener : : LogGlobHandler , boost : : ref ( files ) , _1 ) , GlobFile ) ;
std : : sort ( files . begin ( ) , files . end ( ) ) ;
BOOST_FOREACH ( int ts , files ) {
bool need = false ;
2015-08-15 20:28:05 +02:00
BOOST_FOREACH ( const Endpoint : : Ptr & endpoint , ConfigType : : GetObjectsByType < Endpoint > ( ) ) {
2015-11-24 15:25:55 +01:00
if ( endpoint = = GetLocalEndpoint ( ) )
2014-05-03 20:02:22 +02:00
continue ;
if ( endpoint - > GetLogDuration ( ) > = 0 & & ts < now - endpoint - > GetLogDuration ( ) )
continue ;
if ( ts > endpoint - > GetLocalLogPosition ( ) ) {
need = true ;
break ;
}
}
if ( ! need ) {
String path = GetApiDir ( ) + " log/ " + Convert : : ToString ( ts ) ;
2014-10-20 10:09:57 +02:00
Log ( LogNotice , " ApiListener " )
< < " Removing old log file: " < < path ;
2014-05-03 20:02:22 +02:00
( void ) unlink ( path . CStr ( ) ) ;
}
}
2016-07-21 12:27:18 +02:00
BOOST_FOREACH ( const Endpoint : : Ptr & endpoint , ConfigType : : GetObjectsByType < Endpoint > ( ) ) {
if ( ! endpoint - > GetConnected ( ) )
continue ;
double ts = endpoint - > GetRemoteLogPosition ( ) ;
if ( ts = = 0 )
continue ;
Dictionary : : Ptr lparams = new Dictionary ( ) ;
lparams - > Set ( " log_position " , ts ) ;
Dictionary : : Ptr lmessage = new Dictionary ( ) ;
lmessage - > Set ( " jsonrpc " , " 2.0 " ) ;
lmessage - > Set ( " method " , " log::SetLogPosition " ) ;
lmessage - > Set ( " params " , lparams ) ;
double maxTs = 0 ;
BOOST_FOREACH ( const JsonRpcConnection : : Ptr & client , endpoint - > GetClients ( ) ) {
if ( client - > GetTimestamp ( ) > maxTs )
maxTs = client - > GetTimestamp ( ) ;
}
BOOST_FOREACH ( const JsonRpcConnection : : Ptr & client , endpoint - > GetClients ( ) ) {
if ( client - > GetTimestamp ( ) ! = maxTs )
client - > Disconnect ( ) ;
else
client - > SendMessage ( lmessage ) ;
}
Log ( LogNotice , " ApiListener " )
< < " Setting log position for identity ' " < < endpoint - > GetName ( ) < < " ': "
< < Utility : : FormatDateTime ( " %Y/%m/%d %H:%M:%S " , ts ) ;
}
}
void ApiListener : : ApiReconnectTimerHandler ( void )
{
2015-06-11 23:02:13 +02:00
Zone : : Ptr my_zone = Zone : : GetLocalZone ( ) ;
2015-08-15 20:28:05 +02:00
BOOST_FOREACH ( const Zone : : Ptr & zone , ConfigType : : GetObjectsByType < Zone > ( ) ) {
2015-10-15 17:15:19 +02:00
/* don't connect to global zones */
if ( zone - > GetGlobal ( ) )
continue ;
2015-06-11 23:02:13 +02:00
/* only connect to endpoints in a) the same zone b) our parent zone c) immediate child zones */
if ( my_zone ! = zone & & my_zone ! = zone - > GetParent ( ) & & zone ! = my_zone - > GetParent ( ) ) {
Log ( LogDebug , " ApiListener " )
2015-09-22 17:58:12 +02:00
< < " Not connecting to Zone ' " < < zone - > GetName ( )
< < " ' because it's not in the same zone, a parent or a child zone. " ;
2015-06-11 23:02:13 +02:00
continue ;
}
2014-05-03 20:02:22 +02:00
2015-06-11 23:02:13 +02:00
BOOST_FOREACH ( const Endpoint : : Ptr & endpoint , zone - > GetEndpoints ( ) ) {
/* don't connect to ourselves */
2015-11-24 15:25:55 +01:00
if ( endpoint = = GetLocalEndpoint ( ) ) {
2015-02-17 15:46:03 +01:00
Log ( LogDebug , " ApiListener " )
2015-06-11 23:02:13 +02:00
< < " Not connecting to Endpoint ' " < < endpoint - > GetName ( ) < < " ' because that's us. " ;
2014-05-03 20:02:22 +02:00
continue ;
2015-02-17 15:46:03 +01:00
}
2014-05-03 20:02:22 +02:00
2015-06-11 23:02:13 +02:00
/* don't try to connect to endpoints which don't have a host and port */
if ( endpoint - > GetHost ( ) . IsEmpty ( ) | | endpoint - > GetPort ( ) . IsEmpty ( ) ) {
Log ( LogDebug , " ApiListener " )
2015-09-22 17:58:12 +02:00
< < " Not connecting to Endpoint ' " < < endpoint - > GetName ( )
< < " ' because the host/port attributes are missing. " ;
2015-06-11 23:02:13 +02:00
continue ;
2014-05-08 15:00:09 +02:00
}
2014-05-03 20:02:22 +02:00
2015-06-11 23:02:13 +02:00
/* don't try to connect if there's already a connection attempt */
if ( endpoint - > GetConnecting ( ) ) {
2015-02-17 15:46:03 +01:00
Log ( LogDebug , " ApiListener " )
2015-09-22 17:58:12 +02:00
< < " Not connecting to Endpoint ' " < < endpoint - > GetName ( )
< < " ' because we're already trying to connect to it. " ;
2014-05-03 20:02:22 +02:00
continue ;
2015-02-17 15:46:03 +01:00
}
2014-05-03 20:02:22 +02:00
2015-06-11 23:02:13 +02:00
/* don't try to connect if we're already connected */
2015-10-22 10:52:38 +02:00
if ( endpoint - > GetConnected ( ) ) {
2015-06-11 23:02:13 +02:00
Log ( LogDebug , " ApiListener " )
2015-09-22 17:58:12 +02:00
< < " Not connecting to Endpoint ' " < < endpoint - > GetName ( )
< < " ' because we're already connected to it. " ;
2015-06-11 23:02:13 +02:00
continue ;
2014-05-08 15:00:09 +02:00
}
2015-06-11 23:02:13 +02:00
boost : : thread thread ( boost : : bind ( & ApiListener : : AddConnection , this , endpoint ) ) ;
thread . detach ( ) ;
2014-05-03 20:02:22 +02:00
}
}
2014-08-04 16:34:17 +02:00
Endpoint : : Ptr master = GetMaster ( ) ;
if ( master )
2014-10-20 10:09:57 +02:00
Log ( LogNotice , " ApiListener " )
< < " Current zone master: " < < master - > GetName ( ) ;
2014-05-03 20:02:22 +02:00
std : : vector < String > names ;
2015-08-15 20:28:05 +02:00
BOOST_FOREACH ( const Endpoint : : Ptr & endpoint , ConfigType : : GetObjectsByType < Endpoint > ( ) )
2015-10-22 10:52:38 +02:00
if ( endpoint - > GetConnected ( ) )
2014-05-03 20:02:22 +02:00
names . push_back ( endpoint - > GetName ( ) + " ( " + Convert : : ToString ( endpoint - > GetClients ( ) . size ( ) ) + " ) " ) ;
2014-10-20 10:09:57 +02:00
Log ( LogNotice , " ApiListener " )
< < " Connected endpoints: " < < Utility : : NaturalJoin ( names ) ;
2014-05-03 20:02:22 +02:00
}
2015-09-22 17:58:12 +02:00
void ApiListener : : RelayMessage ( const MessageOrigin : : Ptr & origin ,
const ConfigObject : : Ptr & secobj , const Dictionary : : Ptr & message , bool log )
2014-05-03 20:02:22 +02:00
{
2016-02-09 12:46:11 +01:00
if ( ! IsActive ( ) )
return ;
2015-12-10 16:54:43 +01:00
m_RelayQueue . Enqueue ( boost : : bind ( & ApiListener : : SyncRelayMessage , this , origin , secobj , message , log ) , PriorityNormal , true ) ;
2014-05-03 20:02:22 +02:00
}
2015-08-15 20:28:05 +02:00
void ApiListener : : PersistMessage ( const Dictionary : : Ptr & message , const ConfigObject : : Ptr & secobj )
2014-05-03 20:02:22 +02:00
{
double ts = message - > Get ( " ts " ) ;
ASSERT ( ts ! = 0 ) ;
2014-11-08 21:17:16 +01:00
Dictionary : : Ptr pmessage = new Dictionary ( ) ;
2014-05-03 20:02:22 +02:00
pmessage - > Set ( " timestamp " , ts ) ;
2014-10-26 19:59:49 +01:00
pmessage - > Set ( " message " , JsonEncode ( message ) ) ;
2016-06-15 11:26:35 +02:00
if ( secobj ) {
Dictionary : : Ptr secname = new Dictionary ( ) ;
secname - > Set ( " type " , secobj - > GetType ( ) - > GetName ( ) ) ;
secname - > Set ( " name " , secobj - > GetName ( ) ) ;
pmessage - > Set ( " secobj " , secname ) ;
}
2014-05-03 20:02:22 +02:00
boost : : mutex : : scoped_lock lock ( m_LogLock ) ;
if ( m_LogFile ) {
2014-10-26 19:59:49 +01:00
NetString : : WriteStringToStream ( m_LogFile , JsonEncode ( pmessage ) ) ;
2014-05-03 20:02:22 +02:00
m_LogMessageCount + + ;
SetLogMessageTimestamp ( ts ) ;
if ( m_LogMessageCount > 50000 ) {
CloseLogFile ( ) ;
RotateLogFile ( ) ;
OpenLogFile ( ) ;
}
}
}
2014-11-13 11:23:57 +01:00
void ApiListener : : SyncSendMessage ( const Endpoint : : Ptr & endpoint , const Dictionary : : Ptr & message )
{
ObjectLock olock ( endpoint ) ;
if ( ! endpoint - > GetSyncing ( ) ) {
Log ( LogNotice , " ApiListener " )
< < " Sending message to ' " < < endpoint - > GetName ( ) < < " ' " ;
2016-01-25 10:57:06 +01:00
double maxTs = 0 ;
BOOST_FOREACH ( const JsonRpcConnection : : Ptr & client , endpoint - > GetClients ( ) ) {
if ( client - > GetTimestamp ( ) > maxTs )
maxTs = client - > GetTimestamp ( ) ;
}
BOOST_FOREACH ( const JsonRpcConnection : : Ptr & client , endpoint - > GetClients ( ) ) {
if ( client - > GetTimestamp ( ) ! = maxTs )
continue ;
2014-11-13 11:23:57 +01:00
client - > SendMessage ( message ) ;
2016-01-25 10:57:06 +01:00
}
2014-11-13 11:23:57 +01:00
}
}
2016-01-27 13:39:31 +01:00
bool ApiListener : : RelayMessageOne ( const Zone : : Ptr & targetZone , const MessageOrigin : : Ptr & origin , const Dictionary : : Ptr & message , const Endpoint : : Ptr & currentMaster )
2014-05-03 20:02:22 +02:00
{
2016-01-27 12:18:16 +01:00
ASSERT ( targetZone ) ;
2014-05-03 20:02:22 +02:00
2016-01-27 12:18:16 +01:00
Zone : : Ptr myZone = Zone : : GetLocalZone ( ) ;
/* only relay the message to a) the same zone, b) the parent zone and c) direct child zones */
if ( targetZone ! = myZone & & targetZone ! = myZone - > GetParent ( ) & & targetZone - > GetParent ( ) ! = myZone )
return true ;
2014-05-03 20:02:22 +02:00
2016-01-27 13:39:31 +01:00
Endpoint : : Ptr myEndpoint = GetLocalEndpoint ( ) ;
2014-05-03 20:02:22 +02:00
std : : vector < Endpoint : : Ptr > skippedEndpoints ;
2016-01-27 12:18:16 +01:00
bool relayed = false , log_needed = false , log_done = false ;
BOOST_FOREACH ( const Endpoint : : Ptr & endpoint , targetZone - > GetEndpoints ( ) ) {
2015-10-01 09:17:23 +02:00
/* don't relay messages to ourselves */
2015-11-24 15:25:55 +01:00
if ( endpoint = = GetLocalEndpoint ( ) )
2014-05-03 20:02:22 +02:00
continue ;
2016-01-27 12:18:16 +01:00
log_needed = true ;
2015-10-01 09:17:23 +02:00
/* don't relay messages to disconnected endpoints */
2015-10-22 10:52:38 +02:00
if ( ! endpoint - > GetConnected ( ) ) {
2016-01-27 12:18:16 +01:00
if ( targetZone = = myZone )
log_done = false ;
2015-10-15 09:28:20 +02:00
2015-10-01 09:17:23 +02:00
continue ;
2015-10-15 09:28:20 +02:00
}
2015-10-01 09:17:23 +02:00
2016-01-27 12:18:16 +01:00
log_done = true ;
2015-10-01 09:17:23 +02:00
2015-10-15 09:28:20 +02:00
/* don't relay the message to the zone through more than one endpoint unless this is our own zone */
2016-01-27 12:18:16 +01:00
if ( relayed & & targetZone ! = myZone ) {
2014-05-03 20:02:22 +02:00
skippedEndpoints . push_back ( endpoint ) ;
continue ;
}
/* don't relay messages back to the endpoint which we got the message from */
2015-08-04 14:47:44 +02:00
if ( origin & & origin - > FromClient & & endpoint = = origin - > FromClient - > GetEndpoint ( ) ) {
2014-05-03 20:02:22 +02:00
skippedEndpoints . push_back ( endpoint ) ;
continue ;
}
/* don't relay messages back to the zone which we got the message from */
2016-01-27 12:18:16 +01:00
if ( origin & & origin - > FromZone & & targetZone = = origin - > FromZone ) {
2014-05-03 20:02:22 +02:00
skippedEndpoints . push_back ( endpoint ) ;
continue ;
}
/* only relay message to the master if we're not currently the master */
2016-01-27 13:39:31 +01:00
if ( currentMaster ! = myEndpoint & & currentMaster ! = endpoint ) {
2014-05-03 20:02:22 +02:00
skippedEndpoints . push_back ( endpoint ) ;
continue ;
}
2016-01-27 12:18:16 +01:00
relayed = true ;
2014-05-03 20:02:22 +02:00
2014-11-13 11:23:57 +01:00
SyncSendMessage ( endpoint , message ) ;
2014-05-03 20:02:22 +02:00
}
2016-01-27 12:18:16 +01:00
if ( ! skippedEndpoints . empty ( ) ) {
double ts = message - > Get ( " ts " ) ;
2015-10-01 09:17:23 +02:00
2016-01-27 12:18:16 +01:00
BOOST_FOREACH ( const Endpoint : : Ptr & endpoint , skippedEndpoints )
endpoint - > SetLocalLogPosition ( ts ) ;
}
return ! log_needed | | log_done ;
}
void ApiListener : : SyncRelayMessage ( const MessageOrigin : : Ptr & origin ,
const ConfigObject : : Ptr & secobj , const Dictionary : : Ptr & message , bool log )
{
double ts = Utility : : GetTime ( ) ;
message - > Set ( " ts " , ts ) ;
Log ( LogNotice , " ApiListener " )
< < " Relaying ' " < < message - > Get ( " method " ) < < " ' message " ;
if ( origin & & origin - > FromZone )
message - > Set ( " originZone " , origin - > FromZone - > GetName ( ) ) ;
Zone : : Ptr target_zone ;
if ( secobj ) {
if ( secobj - > GetReflectionType ( ) = = Zone : : TypeInstance )
target_zone = static_pointer_cast < Zone > ( secobj ) ;
else
target_zone = static_pointer_cast < Zone > ( secobj - > GetZone ( ) ) ;
}
if ( ! target_zone )
target_zone = Zone : : GetLocalZone ( ) ;
2016-01-27 13:39:31 +01:00
Endpoint : : Ptr master = GetMaster ( ) ;
bool need_log = ! RelayMessageOne ( target_zone , origin , message , master ) ;
2016-01-27 12:18:16 +01:00
BOOST_FOREACH ( const Zone : : Ptr & zone , target_zone - > GetAllParents ( ) ) {
2016-01-27 13:39:31 +01:00
if ( ! RelayMessageOne ( zone , origin , message , master ) )
2016-01-27 12:18:16 +01:00
need_log = true ;
}
if ( log & & need_log )
PersistMessage ( message , secobj ) ;
2014-05-03 20:02:22 +02:00
}
String ApiListener : : GetApiDir ( void )
{
return Application : : GetLocalStateDir ( ) + " /lib/icinga2/api/ " ;
}
/* must hold m_LogLock */
void ApiListener : : OpenLogFile ( void )
{
String path = GetApiDir ( ) + " log/current " ;
std : : fstream * fp = new std : : fstream ( path . CStr ( ) , std : : fstream : : out | std : : ofstream : : app ) ;
if ( ! fp - > good ( ) ) {
2014-10-20 10:09:57 +02:00
Log ( LogWarning , " ApiListener " )
< < " Could not open spool file: " < < path ;
2014-05-03 20:02:22 +02:00
return ;
}
2014-11-08 21:17:16 +01:00
m_LogFile = new StdioStream ( fp , true ) ;
2014-05-03 20:02:22 +02:00
m_LogMessageCount = 0 ;
SetLogMessageTimestamp ( Utility : : GetTime ( ) ) ;
}
/* must hold m_LogLock */
void ApiListener : : CloseLogFile ( void )
{
if ( ! m_LogFile )
return ;
m_LogFile - > Close ( ) ;
m_LogFile . reset ( ) ;
}
/* must hold m_LogLock */
void ApiListener : : RotateLogFile ( void )
{
double ts = GetLogMessageTimestamp ( ) ;
if ( ts = = 0 )
ts = Utility : : GetTime ( ) ;
String oldpath = GetApiDir ( ) + " log/current " ;
String newpath = GetApiDir ( ) + " log/ " + Convert : : ToString ( static_cast < int > ( ts ) + 1 ) ;
( void ) rename ( oldpath . CStr ( ) , newpath . CStr ( ) ) ;
}
void ApiListener : : LogGlobHandler ( std : : vector < int > & files , const String & file )
{
String name = Utility : : BaseName ( file ) ;
2015-09-29 11:02:08 +02:00
if ( name = = " current " )
return ;
2014-05-03 20:02:22 +02:00
int ts ;
try {
ts = Convert : : ToLong ( name ) ;
2015-09-29 11:02:08 +02:00
} catch ( const std : : exception & ) {
2014-05-03 20:02:22 +02:00
return ;
}
files . push_back ( ts ) ;
}
2015-06-22 11:11:21 +02:00
void ApiListener : : ReplayLog ( const JsonRpcConnection : : Ptr & client )
2014-05-03 20:02:22 +02:00
{
Endpoint : : Ptr endpoint = client - > GetEndpoint ( ) ;
2016-01-20 14:07:07 +01:00
if ( endpoint - > GetLogDuration ( ) = = 0 ) {
ObjectLock olock2 ( endpoint ) ;
endpoint - > SetSyncing ( false ) ;
2016-01-19 17:25:28 +01:00
return ;
2016-01-20 14:07:07 +01:00
}
2016-01-19 17:25:28 +01:00
2014-05-03 20:02:22 +02:00
CONTEXT ( " Replaying log for Endpoint ' " + endpoint - > GetName ( ) + " ' " ) ;
int count = - 1 ;
double peer_ts = endpoint - > GetLocalLogPosition ( ) ;
2015-02-26 14:59:39 +01:00
double logpos_ts = peer_ts ;
2014-05-03 20:02:22 +02:00
bool last_sync = false ;
2015-09-22 17:58:12 +02:00
2014-08-26 10:24:04 +02:00
Endpoint : : Ptr target_endpoint = client - > GetEndpoint ( ) ;
ASSERT ( target_endpoint ) ;
2015-09-22 17:58:12 +02:00
2014-08-26 10:24:04 +02:00
Zone : : Ptr target_zone = target_endpoint - > GetZone ( ) ;
2015-09-22 17:58:12 +02:00
2016-02-08 13:15:24 +01:00
if ( ! target_zone ) {
ObjectLock olock2 ( endpoint ) ;
endpoint - > SetSyncing ( false ) ;
2014-08-26 10:24:04 +02:00
return ;
2016-02-08 13:15:24 +01:00
}
2014-05-03 20:02:22 +02:00
for ( ; ; ) {
boost : : mutex : : scoped_lock lock ( m_LogLock ) ;
CloseLogFile ( ) ;
RotateLogFile ( ) ;
if ( count = = - 1 | | count > 50000 ) {
OpenLogFile ( ) ;
lock . unlock ( ) ;
} else {
last_sync = true ;
}
count = 0 ;
std : : vector < int > files ;
Utility : : Glob ( GetApiDir ( ) + " log/* " , boost : : bind ( & ApiListener : : LogGlobHandler , boost : : ref ( files ) , _1 ) , GlobFile ) ;
std : : sort ( files . begin ( ) , files . end ( ) ) ;
BOOST_FOREACH ( int ts , files ) {
String path = GetApiDir ( ) + " log/ " + Convert : : ToString ( ts ) ;
if ( ts < peer_ts )
continue ;
2014-10-20 10:09:57 +02:00
Log ( LogNotice , " ApiListener " )
< < " Replaying log: " < < path ;
2014-05-03 20:02:22 +02:00
2015-02-26 14:59:39 +01:00
std : : fstream * fp = new std : : fstream ( path . CStr ( ) , std : : fstream : : in | std : : fstream : : binary ) ;
2014-11-08 21:17:16 +01:00
StdioStream : : Ptr logStream = new StdioStream ( fp , true ) ;
2014-05-03 20:02:22 +02:00
String message ;
2015-02-14 16:34:36 +01:00
StreamReadContext src ;
2014-05-03 20:02:22 +02:00
while ( true ) {
Dictionary : : Ptr pmessage ;
try {
2015-02-14 18:48:33 +01:00
StreamReadStatus srs = NetString : : ReadStringFromStream ( logStream , & message , src ) ;
if ( srs = = StatusEof )
2014-05-03 20:02:22 +02:00
break ;
2015-02-14 18:48:33 +01:00
if ( srs ! = StatusNewItem )
continue ;
2014-10-26 19:59:49 +01:00
pmessage = JsonDecode ( message ) ;
2014-05-03 20:02:22 +02:00
} catch ( const std : : exception & ) {
2014-10-20 10:09:57 +02:00
Log ( LogWarning , " ApiListener " )
< < " Unexpected end-of-file for cluster log: " < < path ;
2014-05-03 20:02:22 +02:00
/* Log files may be incomplete or corrupted. This is perfectly OK. */
break ;
}
if ( pmessage - > Get ( " timestamp " ) < = peer_ts )
continue ;
2014-10-24 17:48:02 +02:00
Dictionary : : Ptr secname = pmessage - > Get ( " secobj " ) ;
2015-09-22 17:58:12 +02:00
2014-08-26 10:24:04 +02:00
if ( secname ) {
2015-08-15 20:28:05 +02:00
ConfigType : : Ptr dtype = ConfigType : : GetByName ( secname - > Get ( " type " ) ) ;
2015-09-22 17:58:12 +02:00
2014-08-26 10:24:04 +02:00
if ( ! dtype )
continue ;
2015-09-22 17:58:12 +02:00
2015-08-15 20:28:05 +02:00
ConfigObject : : Ptr secobj = dtype - > GetObject ( secname - > Get ( " name " ) ) ;
2015-09-22 17:58:12 +02:00
2014-08-26 10:24:04 +02:00
if ( ! secobj )
continue ;
2015-09-22 17:58:12 +02:00
2014-08-26 10:24:04 +02:00
if ( ! target_zone - > CanAccessObject ( secobj ) )
continue ;
}
2015-10-26 07:56:58 +01:00
try {
NetString : : WriteStringToStream ( client - > GetStream ( ) , pmessage - > Get ( " message " ) ) ;
count + + ;
} catch ( const std : : exception & ex ) {
Log ( LogWarning , " ApiListener " )
< < " Error while replaying log for endpoint ' " < < endpoint - > GetName ( ) < < " ': " < < DiagnosticInformation ( ex ) ;
break ;
}
2014-05-03 20:02:22 +02:00
peer_ts = pmessage - > Get ( " timestamp " ) ;
2015-02-26 14:59:39 +01:00
if ( ts > logpos_ts + 10 ) {
logpos_ts = ts ;
Dictionary : : Ptr lparams = new Dictionary ( ) ;
lparams - > Set ( " log_position " , logpos_ts ) ;
Dictionary : : Ptr lmessage = new Dictionary ( ) ;
lmessage - > Set ( " jsonrpc " , " 2.0 " ) ;
lmessage - > Set ( " method " , " log::SetLogPosition " ) ;
lmessage - > Set ( " params " , lparams ) ;
JsonRpc : : SendMessage ( client - > GetStream ( ) , lmessage ) ;
}
2014-05-03 20:02:22 +02:00
}
2014-05-09 10:23:54 +02:00
logStream - > Close ( ) ;
2014-05-03 20:02:22 +02:00
}
2015-09-11 14:09:46 +02:00
if ( count > 0 ) {
Log ( LogInformation , " ApiListener " )
< < " Replayed " < < count < < " messages. " ;
}
2014-10-20 10:09:57 +02:00
Log ( LogNotice , " ApiListener " )
< < " Replayed " < < count < < " messages. " ;
2014-05-03 20:02:22 +02:00
if ( last_sync ) {
{
ObjectLock olock2 ( endpoint ) ;
endpoint - > SetSyncing ( false ) ;
}
OpenLogFile ( ) ;
break ;
}
}
}
2015-02-13 11:28:43 +01:00
void ApiListener : : StatsFunc ( const Dictionary : : Ptr & status , const Array : : Ptr & perfdata )
2014-05-03 20:02:22 +02:00
{
std : : pair < Dictionary : : Ptr , Dictionary : : Ptr > stats ;
ApiListener : : Ptr listener = ApiListener : : GetInstance ( ) ;
if ( ! listener )
2015-02-07 22:36:17 +01:00
return ;
2014-05-03 20:02:22 +02:00
stats = listener - > GetStatus ( ) ;
2014-11-13 11:23:57 +01:00
ObjectLock olock ( stats . second ) ;
2014-09-17 15:38:39 +02:00
BOOST_FOREACH ( const Dictionary : : Pair & kv , stats . second )
perfdata - > Add ( " 'api_ " + kv . first + " '= " + Convert : : ToString ( kv . second ) ) ;
2014-05-03 20:02:22 +02:00
status - > Set ( " api " , stats . first ) ;
}
std : : pair < Dictionary : : Ptr , Dictionary : : Ptr > ApiListener : : GetStatus ( void )
{
2014-11-08 21:17:16 +01:00
Dictionary : : Ptr status = new Dictionary ( ) ;
Dictionary : : Ptr perfdata = new Dictionary ( ) ;
2014-05-03 20:02:22 +02:00
/* cluster stats */
status - > Set ( " identity " , GetIdentity ( ) ) ;
2015-09-23 18:09:46 +02:00
double allEndpoints = 0 ;
Array : : Ptr allNotConnectedEndpoints = new Array ( ) ;
Array : : Ptr allConnectedEndpoints = new Array ( ) ;
2014-05-03 20:02:22 +02:00
2015-07-31 16:19:58 +02:00
Zone : : Ptr my_zone = Zone : : GetLocalZone ( ) ;
2015-09-23 18:09:46 +02:00
Dictionary : : Ptr connectedZones = new Dictionary ( ) ;
2015-08-15 20:28:05 +02:00
BOOST_FOREACH ( const Zone : : Ptr & zone , ConfigType : : GetObjectsByType < Zone > ( ) ) {
2015-07-31 16:19:58 +02:00
/* only check endpoints in a) the same zone b) our parent zone c) immediate child zones */
if ( my_zone ! = zone & & my_zone ! = zone - > GetParent ( ) & & zone ! = my_zone - > GetParent ( ) ) {
Log ( LogDebug , " ApiListener " )
< < " Not checking connection to Zone ' " < < zone - > GetName ( ) < < " ' because it's not in the same zone, a parent or a child zone. " ;
2014-05-03 20:02:22 +02:00
continue ;
2015-07-31 16:19:58 +02:00
}
2014-05-03 20:02:22 +02:00
2015-09-23 18:09:46 +02:00
bool zoneConnected = false ;
int countZoneEndpoints = 0 ;
double zoneLag = 0 ;
Array : : Ptr zoneEndpoints = new Array ( ) ;
2015-07-31 16:19:58 +02:00
BOOST_FOREACH ( const Endpoint : : Ptr & endpoint , zone - > GetEndpoints ( ) ) {
2015-09-23 18:09:46 +02:00
zoneEndpoints - > Add ( endpoint - > GetName ( ) ) ;
2015-07-31 16:19:58 +02:00
if ( endpoint - > GetName ( ) = = GetIdentity ( ) )
continue ;
2014-05-03 20:02:22 +02:00
2015-09-25 14:23:42 +02:00
double eplag = CalculateZoneLag ( endpoint ) ;
2015-07-31 16:19:58 +02:00
2015-09-25 14:23:42 +02:00
if ( eplag > 0 & & eplag > zoneLag )
2015-09-23 18:09:46 +02:00
zoneLag = eplag ;
allEndpoints + + ;
countZoneEndpoints + + ;
2015-10-22 10:52:38 +02:00
if ( ! endpoint - > GetConnected ( ) ) {
2015-09-23 18:09:46 +02:00
allNotConnectedEndpoints - > Add ( endpoint - > GetName ( ) ) ;
} else {
allConnectedEndpoints - > Add ( endpoint - > GetName ( ) ) ;
zoneConnected = true ;
}
2015-07-31 16:19:58 +02:00
}
2015-09-23 18:09:46 +02:00
/* if there's only one endpoint inside the zone, we're not connected - that's us, fake it */
if ( zone - > GetEndpoints ( ) . size ( ) = = 1 & & countZoneEndpoints = = 0 )
zoneConnected = true ;
Dictionary : : Ptr zoneStats = new Dictionary ( ) ;
zoneStats - > Set ( " connected " , zoneConnected ) ;
zoneStats - > Set ( " client_log_lag " , zoneLag ) ;
zoneStats - > Set ( " endpoints " , zoneEndpoints ) ;
String parentZoneName ;
Zone : : Ptr parentZone = zone - > GetParent ( ) ;
if ( parentZone )
parentZoneName = parentZone - > GetName ( ) ;
zoneStats - > Set ( " parent_zone " , parentZoneName ) ;
connectedZones - > Set ( zone - > GetName ( ) , zoneStats ) ;
2014-05-03 20:02:22 +02:00
}
2015-09-23 18:09:46 +02:00
status - > Set ( " num_endpoints " , allEndpoints ) ;
status - > Set ( " num_conn_endpoints " , allConnectedEndpoints - > GetLength ( ) ) ;
status - > Set ( " num_not_conn_endpoints " , allNotConnectedEndpoints - > GetLength ( ) ) ;
status - > Set ( " conn_endpoints " , allConnectedEndpoints ) ;
status - > Set ( " not_conn_endpoints " , allNotConnectedEndpoints ) ;
status - > Set ( " zones " , connectedZones ) ;
2014-05-03 20:02:22 +02:00
2015-09-23 18:09:46 +02:00
perfdata - > Set ( " num_endpoints " , allEndpoints ) ;
perfdata - > Set ( " num_conn_endpoints " , Convert : : ToDouble ( allConnectedEndpoints - > GetLength ( ) ) ) ;
perfdata - > Set ( " num_not_conn_endpoints " , Convert : : ToDouble ( allNotConnectedEndpoints - > GetLength ( ) ) ) ;
2014-05-03 20:02:22 +02:00
return std : : make_pair ( status , perfdata ) ;
}
2014-05-08 15:12:56 +02:00
2015-09-25 14:23:42 +02:00
double ApiListener : : CalculateZoneLag ( const Endpoint : : Ptr & endpoint )
{
double remoteLogPosition = endpoint - > GetRemoteLogPosition ( ) ;
double eplag = Utility : : GetTime ( ) - remoteLogPosition ;
2015-10-22 10:52:38 +02:00
if ( ( endpoint - > GetSyncing ( ) | | ! endpoint - > GetConnected ( ) ) & & remoteLogPosition ! = 0 )
2015-09-25 14:23:42 +02:00
return eplag ;
return 0 ;
}
2015-06-22 11:11:21 +02:00
void ApiListener : : AddAnonymousClient ( const JsonRpcConnection : : Ptr & aclient )
2014-05-08 15:12:56 +02:00
{
ObjectLock olock ( this ) ;
m_AnonymousClients . insert ( aclient ) ;
}
2015-06-22 11:11:21 +02:00
void ApiListener : : RemoveAnonymousClient ( const JsonRpcConnection : : Ptr & aclient )
2014-05-08 15:12:56 +02:00
{
ObjectLock olock ( this ) ;
m_AnonymousClients . erase ( aclient ) ;
}
2015-06-22 11:11:21 +02:00
std : : set < JsonRpcConnection : : Ptr > ApiListener : : GetAnonymousClients ( void ) const
2014-05-08 15:12:56 +02:00
{
ObjectLock olock ( this ) ;
return m_AnonymousClients ;
2014-05-09 10:23:54 +02:00
}
2015-06-22 11:11:21 +02:00
2015-08-29 01:16:16 +02:00
void ApiListener : : AddHttpClient ( const HttpServerConnection : : Ptr & aclient )
2015-06-22 11:11:21 +02:00
{
ObjectLock olock ( this ) ;
m_HttpClients . insert ( aclient ) ;
}
2015-08-29 01:16:16 +02:00
void ApiListener : : RemoveHttpClient ( const HttpServerConnection : : Ptr & aclient )
2015-06-22 11:11:21 +02:00
{
ObjectLock olock ( this ) ;
m_HttpClients . erase ( aclient ) ;
}
2015-08-29 01:16:16 +02:00
std : : set < HttpServerConnection : : Ptr > ApiListener : : GetHttpClients ( void ) const
2015-06-22 11:11:21 +02:00
{
ObjectLock olock ( this ) ;
return m_HttpClients ;
}
2015-08-11 14:13:53 +02:00
2015-08-04 14:47:44 +02:00
Value ApiListener : : HelloAPIHandler ( const MessageOrigin : : Ptr & origin , const Dictionary : : Ptr & params )
2015-08-11 14:13:53 +02:00
{
return Empty ;
}
2015-11-24 15:25:55 +01:00
Endpoint : : Ptr ApiListener : : GetLocalEndpoint ( void ) const
{
return m_LocalEndpoint ;
}