granicus.if.org Git - icinga2/blob - lib/remote/apilistener.cpp

   1 /******************************************************************************
   2  * Icinga 2                                                                   *
   3  * Copyright (C) 2012-2014 Icinga Development Team (http://www.icinga.org)    *
   4  *                                                                            *
   5  * This program is free software; you can redistribute it and/or              *
   6  * modify it under the terms of the GNU General Public License                *
   7  * as published by the Free Software Foundation; either version 2             *
   8  * of the License, or (at your option) any later version.                     *
   9  *                                                                            *
  10  * This program is distributed in the hope that it will be useful,            *
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the              *
  13  * GNU General Public License for more details.                               *
  14  *                                                                            *
  15  * You should have received a copy of the GNU General Public License          *
  16  * along with this program; if not, write to the Free Software Foundation     *
  17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
  18  ******************************************************************************/
  19
  20 #include "remote/apilistener.hpp"
  21 #include "remote/apiclient.hpp"
  22 #include "remote/endpoint.hpp"
  23 #include "base/convert.hpp"
  24 #include "base/netstring.hpp"
  25 #include "base/dynamictype.hpp"
  26 #include "base/logger_fwd.hpp"
  27 #include "base/objectlock.hpp"
  28 #include "base/stdiostream.hpp"
  29 #include "base/application.hpp"
  30 #include "base/context.hpp"
  31 #include "base/statsfunction.hpp"
  32 #include <fstream>
  33
  34 using namespace icinga;
  35
  36 REGISTER_TYPE(ApiListener);
  37
  38 boost::signals2::signal<void(bool)> ApiListener::OnMasterChanged;
  39
  40 REGISTER_STATSFUNCTION(ApiListenerStats, &ApiListener::StatsFunc);
  41
  42 void ApiListener::OnConfigLoaded(void)
  43 {
  44         /* set up SSL context */
  45         shared_ptr<X509> cert = make_shared<X509>();
  46         try {
  47                 cert = GetX509Certificate(GetCertPath());
  48         } catch (std::exception&) {
  49                 Log(LogCritical, "ApiListener", "Cannot get certificate from cert path: '" + GetCertPath() + "'.");
  50                 return;
  51         }
  52
  53         try {
  54                 SetIdentity(GetCertificateCN(cert));
  55         } catch (std::exception&) {
  56                 Log(LogCritical, "ApiListener", "Cannot get certificate common name from cert path: '" + GetCertPath() + "'.");
  57                 return;
  58         }
  59
  60         Log(LogInformation, "ApiListener", "My API identity: " + GetIdentity());
  61
  62         try {
  63                 m_SSLContext = MakeSSLContext(GetCertPath(), GetKeyPath(), GetCaPath());
  64         } catch (std::exception&) {
  65                 Log(LogCritical, "ApiListener", "Cannot make SSL context for cert path: '" + GetCertPath() + "' key path: '" + GetKeyPath() + "' ca path: '" + GetCaPath() + "'.");
  66                 return;
  67         }
  68
  69         if (!GetCrlPath().IsEmpty()) {
  70                 try {
  71                         AddCRLToSSLContext(m_SSLContext, GetCrlPath());
  72                 } catch (std::exception&) {
  73                         Log(LogCritical, "ApiListener", "Cannot add certificate revocation list to SSL context for crl path: '" + GetCrlPath() + "'.");
  74                         return;
  75                 }
  76         }
  77
  78         if (!Endpoint::GetByName(GetIdentity())) {
  79                 Log(LogCritical, "ApiListener", "Endpoint object for '" + GetIdentity() + "' is missing.");
  80                 return;
  81         }
  82
  83         SyncZoneDirs();
  84 }
  85
  86 /**
  87  * Starts the component.
  88  */
  89 void ApiListener::Start(void)
  90 {
  91         if (std::distance(DynamicType::GetObjects<ApiListener>().first, DynamicType::GetObjects<ApiListener>().second) > 1) {
  92                 Log(LogCritical, "ApiListener", "Only one ApiListener object is allowed.");
  93                 return;
  94         }
  95
  96         DynamicObject::Start();
  97
  98         {
  99                 boost::mutex::scoped_lock(m_LogLock);
 100                 RotateLogFile();
 101                 OpenLogFile();
 102         }
 103
 104         /* create the primary JSON-RPC listener */
 105         if (!AddListener(GetBindPort())) {
 106                 Log(LogCritical, "ApiListener", "Cannot add listener for port '" + Convert::ToString(GetBindPort()) + "'.");
 107                 Application::Exit(EXIT_FAILURE);
 108         }
 109
 110         m_Timer = make_shared<Timer>();
 111         m_Timer->OnTimerExpired.connect(boost::bind(&ApiListener::ApiTimerHandler, this));
 112         m_Timer->SetInterval(5);
 113         m_Timer->Start();
 114         m_Timer->Reschedule(0);
 115
 116         OnMasterChanged(true);
 117 }
 118
 119 ApiListener::Ptr ApiListener::GetInstance(void)
 120 {
 121         BOOST_FOREACH(const ApiListener::Ptr& listener, DynamicType::GetObjects<ApiListener>())
 122                 return listener;
 123
 124         return ApiListener::Ptr();
 125 }
 126
 127 shared_ptr<SSL_CTX> ApiListener::GetSSLContext(void) const
 128 {
 129         return m_SSLContext;
 130 }
 131
 132 Endpoint::Ptr ApiListener::GetMaster(void) const
 133 {
 134         Zone::Ptr zone = Zone::GetLocalZone();
 135
 136         if (!zone)
 137                 return Endpoint::Ptr();
 138
 139         std::vector<String> names;
 140
 141         BOOST_FOREACH(const Endpoint::Ptr& endpoint, zone->GetEndpoints())
 142                 if (endpoint->IsConnected() || endpoint->GetName() == GetIdentity())
 143                         names.push_back(endpoint->GetName());
 144
 145         std::sort(names.begin(), names.end());
 146
 147         return Endpoint::GetByName(*names.begin());
 148 }
 149
 150 bool ApiListener::IsMaster(void) const
 151 {
 152         Endpoint::Ptr master = GetMaster();
 153
 154         if (!master)
 155                 return false;
 156
 157         return master->GetName() == GetIdentity();
 158 }
 159
 160 /**
 161  * Creates a new JSON-RPC listener on the specified port.
 162  *
 163  * @param service The port to listen on.
 164  */
 165 bool ApiListener::AddListener(const String& service)
 166 {
 167         ObjectLock olock(this);
 168
 169         shared_ptr<SSL_CTX> sslContext = m_SSLContext;
 170
 171         if (!sslContext) {
 172                 Log(LogCritical, "ApiListener", "SSL context is required for AddListener()");
 173                 return false;
 174         }
 175
 176         std::ostringstream s;
 177         s << "Adding new listener: port " << service;
 178         Log(LogInformation, "ApiListener", s.str());
 179
 180         TcpSocket::Ptr server = make_shared<TcpSocket>();
 181
 182         try {
 183                 server->Bind(service, AF_UNSPEC);
 184         } catch(std::exception&) {
 185                 Log(LogCritical, "ApiListener", "Cannot bind tcp socket on '" + service + "'.");
 186                 return false;
 187         }
 188
 189         boost::thread thread(boost::bind(&ApiListener::ListenerThreadProc, this, server));
 190         thread.detach();
 191
 192         m_Servers.insert(server);
 193
 194         return true;
 195 }
 196
 197 void ApiListener::ListenerThreadProc(const Socket::Ptr& server)
 198 {
 199         Utility::SetThreadName("API Listener");
 200
 201         server->Listen();
 202
 203         for (;;) {
 204                 try {
 205                         Socket::Ptr client = server->Accept();
 206                         Utility::QueueAsyncCallback(boost::bind(&ApiListener::NewClientHandler, this, client, RoleServer));
 207                 } catch (std::exception&) {
 208                         Log(LogCritical, "ApiListener", "Cannot accept new connection.");
 209                 }
 210         }
 211 }
 212
 213 /**
 214  * Creates a new JSON-RPC client and connects to the specified endpoint.
 215  *
 216  * @param endpoint The endpoint.
 217  */
 218 void ApiListener::AddConnection(const Endpoint::Ptr& endpoint)
 219 {
 220         {
 221                 ObjectLock olock(this);
 222
 223                 shared_ptr<SSL_CTX> sslContext = m_SSLContext;
 224
 225                 if (!sslContext) {
 226                         Log(LogCritical, "ApiListener", "SSL context is required for AddConnection()");
 227                         return;
 228                 }
 229         }
 230
 231         String host = endpoint->GetHost();
 232         String port = endpoint->GetPort();
 233
 234         TcpSocket::Ptr client = make_shared<TcpSocket>();
 235
 236         try {
 237                 endpoint->SetConnecting(true);
 238                 client->Connect(host, port);
 239                 NewClientHandler(client, RoleClient);
 240                 endpoint->SetConnecting(false);
 241         } catch (const std::exception& ex) {
 242                 endpoint->SetConnecting(false);
 243                 client->Close();
 244
 245                 std::ostringstream info, debug;
 246                 info << "Cannot connect to host '" << host << "' on port '" << port << "'";
 247                 debug << info.str() << std::endl << DiagnosticInformation(ex);
 248                 Log(LogCritical, "ApiListener", info.str());
 249                 Log(LogDebug, "ApiListener", debug.str());
 250         }
 251 }
 252
 253 /**
 254  * Processes a new client connection.
 255  *
 256  * @param client The new client.
 257  */
 258 void ApiListener::NewClientHandler(const Socket::Ptr& client, ConnectionRole role)
 259 {
 260         CONTEXT("Handling new API client connection");
 261
 262         TlsStream::Ptr tlsStream;
 263
 264         {
 265                 ObjectLock olock(this);
 266                 try {
 267                         tlsStream = make_shared<TlsStream>(client, role, m_SSLContext);
 268                 } catch (std::exception&) {
 269                         Log(LogCritical, "ApiListener", "Cannot create tls stream from client connection.");
 270                         return;
 271                 }
 272         }
 273
 274         try {
 275                 tlsStream->Handshake();
 276         } catch (std::exception) {
 277                 Log(LogCritical, "ApiListener", "Client TLS handshake failed.");
 278                 return;
 279         }
 280
 281         shared_ptr<X509> cert = tlsStream->GetPeerCertificate();
 282         String identity;
 283
 284         try {
 285                 identity = GetCertificateCN(cert);
 286         } catch (std::exception&) {
 287                 Log(LogCritical, "ApiListener", "Cannot get certificate common name from cert path: '" + GetCertPath() + "'.");
 288                 return;
 289         }
 290
 291         Log(LogInformation, "ApiListener", "New client connection for identity '" + identity + "'");
 292
 293         Endpoint::Ptr endpoint = Endpoint::GetByName(identity);
 294
 295         bool need_sync = false;
 296
 297         if (endpoint)
 298                 need_sync = !endpoint->IsConnected();
 299
 300         ApiClient::Ptr aclient = make_shared<ApiClient>(identity, tlsStream, role);
 301         aclient->Start();
 302
 303         if (endpoint) {
 304                 if (need_sync) {
 305                         {
 306                                 ObjectLock olock(endpoint);
 307
 308                                 endpoint->SetSyncing(true);
 309                         }
 310
 311                         ReplayLog(aclient);
 312                 }
 313
 314                 SendConfigUpdate(aclient);
 315
 316                 endpoint->AddClient(aclient);
 317         } else
 318                 AddAnonymousClient(aclient);
 319 }
 320
 321 void ApiListener::ApiTimerHandler(void)
 322 {
 323         double now = Utility::GetTime();
 324
 325         std::vector<int> files;
 326         Utility::Glob(GetApiDir() + "log/*", boost::bind(&ApiListener::LogGlobHandler, boost::ref(files), _1), GlobFile);
 327         std::sort(files.begin(), files.end());
 328
 329         BOOST_FOREACH(int ts, files) {
 330                 bool need = false;
 331
 332                 BOOST_FOREACH(const Endpoint::Ptr& endpoint, DynamicType::GetObjects<Endpoint>()) {
 333                         if (endpoint->GetName() == GetIdentity())
 334                                 continue;
 335
 336                         if (endpoint->GetLogDuration() >= 0 && ts < now - endpoint->GetLogDuration())
 337                                 continue;
 338
 339                         if (ts > endpoint->GetLocalLogPosition()) {
 340                                 need = true;
 341                                 break;
 342                         }
 343                 }
 344
 345                 if (!need) {
 346                         String path = GetApiDir() + "log/" + Convert::ToString(ts);
 347                         Log(LogNotice, "ApiListener", "Removing old log file: " + path);
 348                         (void)unlink(path.CStr());
 349                 }
 350         }
 351
 352         if (IsMaster()) {
 353                 Zone::Ptr my_zone = Zone::GetLocalZone();
 354
 355                 BOOST_FOREACH(const Zone::Ptr& zone, DynamicType::GetObjects<Zone>()) {
 356                         /* only connect to endpoints in a) the same zone b) our parent zone c) immediate child zones */
 357                         if (my_zone != zone && my_zone != zone->GetParent() && zone != my_zone->GetParent())
 358                                 continue;
 359
 360                         bool connected = false;
 361
 362                         BOOST_FOREACH(const Endpoint::Ptr& endpoint, zone->GetEndpoints()) {
 363                                 if (endpoint->IsConnected()) {
 364                                         connected = true;
 365                                         break;
 366                                 }
 367                         }
 368
 369                         /* don't connect to an endpoint if we already have a connection to the zone */
 370                         if (connected)
 371                                 continue;
 372
 373                         BOOST_FOREACH(const Endpoint::Ptr& endpoint, zone->GetEndpoints()) {
 374                                 /* don't connect to ourselves */
 375                                 if (endpoint->GetName() == GetIdentity())
 376                                         continue;
 377
 378                                 /* don't try to connect to endpoints which don't have a host and port */
 379                                 if (endpoint->GetHost().IsEmpty() || endpoint->GetPort().IsEmpty())
 380                                         continue;
 381
 382                                 /* don't try to connect if there's already a connection attempt */
 383                                 if (endpoint->GetConnecting())
 384                                         continue;
 385
 386                                 Utility::QueueAsyncCallback(boost::bind(&ApiListener::AddConnection, this, endpoint));
 387                         }
 388                 }
 389         }
 390
 391         BOOST_FOREACH(const Endpoint::Ptr& endpoint, DynamicType::GetObjects<Endpoint>()) {
 392                 if (!endpoint->IsConnected())
 393                         continue;
 394
 395                 double ts = endpoint->GetRemoteLogPosition();
 396
 397                 if (ts == 0)
 398                         continue;
 399
 400                 Dictionary::Ptr lparams = make_shared<Dictionary>();
 401                 lparams->Set("log_position", ts);
 402
 403                 Dictionary::Ptr lmessage = make_shared<Dictionary>();
 404                 lmessage->Set("jsonrpc", "2.0");
 405                 lmessage->Set("method", "log::SetLogPosition");
 406                 lmessage->Set("params", lparams);
 407
 408                 BOOST_FOREACH(const ApiClient::Ptr& client, endpoint->GetClients())
 409                         client->SendMessage(lmessage);
 410
 411                 Log(LogNotice, "ApiListener", "Setting log position for identity '" + endpoint->GetName() + "': " +
 412                         Utility::FormatDateTime("%Y/%m/%d %H:%M:%S", ts));
 413         }
 414
 415         Endpoint::Ptr master = GetMaster();
 416
 417         if (master)
 418                 Log(LogNotice, "ApiListener", "Current zone master: " + master->GetName());
 419
 420         std::vector<String> names;
 421         BOOST_FOREACH(const Endpoint::Ptr& endpoint, DynamicType::GetObjects<Endpoint>())
 422                 if (endpoint->IsConnected())
 423                         names.push_back(endpoint->GetName() + " (" + Convert::ToString(endpoint->GetClients().size()) + ")");
 424
 425         Log(LogNotice, "ApiListener", "Connected endpoints: " + Utility::NaturalJoin(names));
 426 }
 427
 428 void ApiListener::RelayMessage(const MessageOrigin& origin, const DynamicObject::Ptr& secobj, const Dictionary::Ptr& message, bool log)
 429 {
 430         m_RelayQueue.Enqueue(boost::bind(&ApiListener::SyncRelayMessage, this, origin, secobj, message, log));
 431 }
 432
 433 void ApiListener::PersistMessage(const Dictionary::Ptr& message)
 434 {
 435         double ts = message->Get("ts");
 436
 437         ASSERT(ts != 0);
 438
 439         Dictionary::Ptr pmessage = make_shared<Dictionary>();
 440         pmessage->Set("timestamp", ts);
 441
 442         pmessage->Set("message", JsonSerialize(message));
 443
 444         boost::mutex::scoped_lock lock(m_LogLock);
 445         if (m_LogFile) {
 446                 NetString::WriteStringToStream(m_LogFile, JsonSerialize(pmessage));
 447                 m_LogMessageCount++;
 448                 SetLogMessageTimestamp(ts);
 449
 450                 if (m_LogMessageCount > 50000) {
 451                         CloseLogFile();
 452                         RotateLogFile();
 453                         OpenLogFile();
 454                 }
 455         }
 456 }
 457
 458 void ApiListener::SyncRelayMessage(const MessageOrigin& origin, const DynamicObject::Ptr& secobj, const Dictionary::Ptr& message, bool log)
 459 {
 460         double ts = Utility::GetTime();
 461         message->Set("ts", ts);
 462
 463         Log(LogNotice, "ApiListener", "Relaying '" + message->Get("method") + "' message");
 464
 465         if (log)
 466                 m_LogQueue.Enqueue(boost::bind(&ApiListener::PersistMessage, this, message));
 467
 468         if (origin.FromZone)
 469                 message->Set("originZone", origin.FromZone->GetName());
 470
 471         bool is_master = IsMaster();
 472         Endpoint::Ptr master = GetMaster();
 473         Zone::Ptr my_zone = Zone::GetLocalZone();
 474
 475         std::vector<Endpoint::Ptr> skippedEndpoints;
 476         std::set<Zone::Ptr> finishedZones;
 477
 478         BOOST_FOREACH(const Endpoint::Ptr& endpoint, DynamicType::GetObjects<Endpoint>()) {
 479                 /* don't relay messages to ourselves or disconnected endpoints */
 480                 if (endpoint->GetName() == GetIdentity() || !endpoint->IsConnected())
 481                         continue;
 482
 483                 Zone::Ptr target_zone = endpoint->GetZone();
 484
 485                 /* don't relay the message to the zone through more than one endpoint */
 486                 if (finishedZones.find(target_zone) != finishedZones.end()) {
 487                         skippedEndpoints.push_back(endpoint);
 488                         continue;
 489                 }
 490
 491                 /* don't relay messages back to the endpoint which we got the message from */
 492                 if (origin.FromClient && endpoint == origin.FromClient->GetEndpoint()) {
 493                         skippedEndpoints.push_back(endpoint);
 494                         continue;
 495                 }
 496
 497                 /* don't relay messages back to the zone which we got the message from */
 498                 if (origin.FromZone && target_zone == origin.FromZone) {
 499                         skippedEndpoints.push_back(endpoint);
 500                         continue;
 501                 }
 502
 503                 /* only relay message to the master if we're not currently the master */
 504                 if (!is_master && master != endpoint) {
 505                         skippedEndpoints.push_back(endpoint);
 506                         continue;
 507                 }
 508
 509                 /* only relay the message to a) the same zone, b) the parent zone and c) direct child zones */
 510                 if (target_zone != my_zone && target_zone != my_zone->GetParent() &&
 511                     secobj->GetZone() != target_zone->GetName()) {
 512                         skippedEndpoints.push_back(endpoint);
 513                         continue;
 514                 }
 515
 516                 /* only relay messages to zones which have access to the object */
 517                 if (!target_zone->CanAccessObject(secobj))
 518                         continue;
 519
 520                 finishedZones.insert(target_zone);
 521
 522                 {
 523                         ObjectLock olock(endpoint);
 524
 525                         if (!endpoint->GetSyncing()) {
 526                                 Log(LogNotice, "ApiListener", "Sending message to '" + endpoint->GetName() + "'");
 527
 528                                 BOOST_FOREACH(const ApiClient::Ptr& client, endpoint->GetClients())
 529                                         client->SendMessage(message);
 530                         }
 531                 }
 532         }
 533
 534         BOOST_FOREACH(const Endpoint::Ptr& endpoint, skippedEndpoints)
 535                 endpoint->SetLocalLogPosition(ts);
 536 }
 537
 538 String ApiListener::GetApiDir(void)
 539 {
 540         return Application::GetLocalStateDir() + "/lib/icinga2/api/";
 541 }
 542
 543 /* must hold m_LogLock */
 544 void ApiListener::OpenLogFile(void)
 545 {
 546         String path = GetApiDir() + "log/current";
 547
 548         std::fstream *fp = new std::fstream(path.CStr(), std::fstream::out | std::ofstream::app);
 549
 550         if (!fp->good()) {
 551                 Log(LogWarning, "ApiListener", "Could not open spool file: " + path);
 552                 return;
 553         }
 554
 555         m_LogFile = make_shared<StdioStream>(fp, true);
 556         m_LogMessageCount = 0;
 557         SetLogMessageTimestamp(Utility::GetTime());
 558 }
 559
 560 /* must hold m_LogLock */
 561 void ApiListener::CloseLogFile(void)
 562 {
 563         if (!m_LogFile)
 564                 return;
 565
 566         m_LogFile->Close();
 567         m_LogFile.reset();
 568 }
 569
 570 /* must hold m_LogLock */
 571 void ApiListener::RotateLogFile(void)
 572 {
 573         double ts = GetLogMessageTimestamp();
 574
 575         if (ts == 0)
 576                 ts = Utility::GetTime();
 577
 578         String oldpath = GetApiDir() + "log/current";
 579         String newpath = GetApiDir() + "log/" + Convert::ToString(static_cast<int>(ts)+1);
 580         (void) rename(oldpath.CStr(), newpath.CStr());
 581 }
 582
 583 void ApiListener::LogGlobHandler(std::vector<int>& files, const String& file)
 584 {
 585         String name = Utility::BaseName(file);
 586
 587         int ts;
 588
 589         try {
 590                 ts = Convert::ToLong(name);
 591         }
 592         catch (const std::exception&) {
 593                 return;
 594         }
 595
 596         files.push_back(ts);
 597 }
 598
 599 void ApiListener::ReplayLog(const ApiClient::Ptr& client)
 600 {
 601         Endpoint::Ptr endpoint = client->GetEndpoint();
 602
 603         CONTEXT("Replaying log for Endpoint '" + endpoint->GetName() + "'");
 604
 605         int count = -1;
 606         double peer_ts = endpoint->GetLocalLogPosition();
 607         bool last_sync = false;
 608
 609         for (;;) {
 610                 boost::mutex::scoped_lock lock(m_LogLock);
 611
 612                 CloseLogFile();
 613                 RotateLogFile();
 614
 615                 if (count == -1 || count > 50000) {
 616                         OpenLogFile();
 617                         lock.unlock();
 618                 } else {
 619                         last_sync = true;
 620                 }
 621
 622                 count = 0;
 623
 624                 std::vector<int> files;
 625                 Utility::Glob(GetApiDir() + "log/*", boost::bind(&ApiListener::LogGlobHandler, boost::ref(files), _1), GlobFile);
 626                 std::sort(files.begin(), files.end());
 627
 628                 BOOST_FOREACH(int ts, files) {
 629                         String path = GetApiDir() + "log/" + Convert::ToString(ts);
 630
 631                         if (ts < peer_ts)
 632                                 continue;
 633
 634                         Log(LogNotice, "ApiListener", "Replaying log: " + path);
 635
 636                         std::fstream *fp = new std::fstream(path.CStr(), std::fstream::in);
 637                         StdioStream::Ptr logStream = make_shared<StdioStream>(fp, true);
 638
 639                         String message;
 640                         while (true) {
 641                                 Dictionary::Ptr pmessage;
 642
 643                                 try {
 644                                         if (!NetString::ReadStringFromStream(logStream, &message))
 645                                                 break;
 646
 647                                         pmessage = JsonDeserialize(message);
 648                                 } catch (const std::exception&) {
 649                                         Log(LogWarning, "ApiListener", "Unexpected end-of-file for cluster log: " + path);
 650
 651                                         /* Log files may be incomplete or corrupted. This is perfectly OK. */
 652                                         break;
 653                                 }
 654
 655                                 if (pmessage->Get("timestamp") <= peer_ts)
 656                                         continue;
 657
 658                                 NetString::WriteStringToStream(client->GetStream(), pmessage->Get("message"));
 659                                 count++;
 660
 661                                 peer_ts = pmessage->Get("timestamp");
 662                         }
 663
 664                         logStream->Close();
 665                 }
 666
 667                 Log(LogNotice, "ApiListener", "Replayed " + Convert::ToString(count) + " messages.");
 668
 669                 if (last_sync) {
 670                         {
 671                                 ObjectLock olock2(endpoint);
 672                                 endpoint->SetSyncing(false);
 673                         }
 674
 675                         OpenLogFile();
 676
 677                         break;
 678                 }
 679         }
 680 }
 681
 682 Value ApiListener::StatsFunc(Dictionary::Ptr& status, Dictionary::Ptr& perfdata)
 683 {
 684         Dictionary::Ptr nodes = make_shared<Dictionary>();
 685         std::pair<Dictionary::Ptr, Dictionary::Ptr> stats;
 686
 687         ApiListener::Ptr listener = ApiListener::GetInstance();
 688
 689         if (!listener)
 690                 return 0;
 691
 692         stats = listener->GetStatus();
 693
 694         BOOST_FOREACH(Dictionary::Pair const& kv, stats.second)
 695                 perfdata->Set("api_" + kv.first, kv.second);
 696
 697         status->Set("api", stats.first);
 698
 699         return 0;
 700 }
 701
 702 std::pair<Dictionary::Ptr, Dictionary::Ptr> ApiListener::GetStatus(void)
 703 {
 704         Dictionary::Ptr status = make_shared<Dictionary>();
 705         Dictionary::Ptr perfdata = make_shared<Dictionary>();
 706
 707         /* cluster stats */
 708         status->Set("identity", GetIdentity());
 709
 710         double count_endpoints = 0;
 711         Array::Ptr not_connected_endpoints = make_shared<Array>();
 712         Array::Ptr connected_endpoints = make_shared<Array>();
 713
 714         BOOST_FOREACH(const Endpoint::Ptr& endpoint, DynamicType::GetObjects<Endpoint>()) {
 715                 if (endpoint->GetName() == GetIdentity())
 716                         continue;
 717
 718                 count_endpoints++;
 719
 720                 if (!endpoint->IsConnected())
 721                         not_connected_endpoints->Add(endpoint->GetName());
 722                 else
 723                         connected_endpoints->Add(endpoint->GetName());
 724         }
 725
 726         status->Set("num_endpoints", count_endpoints);
 727         status->Set("num_conn_endpoints", connected_endpoints->GetLength());
 728         status->Set("num_not_conn_endpoints", not_connected_endpoints->GetLength());
 729         status->Set("conn_endpoints", connected_endpoints);
 730         status->Set("not_conn_endpoints", not_connected_endpoints);
 731
 732         perfdata->Set("num_endpoints", count_endpoints);
 733         perfdata->Set("num_conn_endpoints", Convert::ToDouble(connected_endpoints->GetLength()));
 734         perfdata->Set("num_not_conn_endpoints", Convert::ToDouble(not_connected_endpoints->GetLength()));
 735
 736         return std::make_pair(status, perfdata);
 737 }
 738
 739 void ApiListener::AddAnonymousClient(const ApiClient::Ptr& aclient)
 740 {
 741         ObjectLock olock(this);
 742         m_AnonymousClients.insert(aclient);
 743 }
 744
 745 void ApiListener::RemoveAnonymousClient(const ApiClient::Ptr& aclient)
 746 {
 747         ObjectLock olock(this);
 748         m_AnonymousClients.erase(aclient);
 749 }
 750
 751 std::set<ApiClient::Ptr> ApiListener::GetAnonymousClients(void) const
 752 {
 753         ObjectLock olock(this);
 754         return m_AnonymousClients;
 755 }