granicus.if.org Git - icinga2/blob - lib/livestatus/livestatuslistener.cpp

   1 /* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */
   2
   3 #include "livestatus/livestatuslistener.hpp"
   4 #include "livestatus/livestatuslistener-ti.cpp"
   5 #include "base/utility.hpp"
   6 #include "base/perfdatavalue.hpp"
   7 #include "base/objectlock.hpp"
   8 #include "base/configtype.hpp"
   9 #include "base/logger.hpp"
  10 #include "base/exception.hpp"
  11 #include "base/tcpsocket.hpp"
  12 #include "base/unixsocket.hpp"
  13 #include "base/networkstream.hpp"
  14 #include "base/application.hpp"
  15 #include "base/function.hpp"
  16 #include "base/statsfunction.hpp"
  17 #include "base/convert.hpp"
  18
  19 using namespace icinga;
  20
  21 REGISTER_TYPE(LivestatusListener);
  22
  23 static int l_ClientsConnected = 0;
  24 static int l_Connections = 0;
  25 static boost::mutex l_ComponentMutex;
  26
  27 REGISTER_STATSFUNCTION(LivestatusListener, &LivestatusListener::StatsFunc);
  28
  29 void LivestatusListener::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
  30 {
  31         DictionaryData nodes;
  32
  33         for (const LivestatusListener::Ptr& livestatuslistener : ConfigType::GetObjectsByType<LivestatusListener>()) {
  34                 nodes.emplace_back(livestatuslistener->GetName(), new Dictionary({
  35                         { "connections", l_Connections }
  36                 }));
  37
  38                 perfdata->Add(new PerfdataValue("livestatuslistener_" + livestatuslistener->GetName() + "_connections", l_Connections));
  39         }
  40
  41         status->Set("livestatuslistener", new Dictionary(std::move(nodes)));
  42 }
  43
  44 /**
  45  * Starts the component.
  46  */
  47 void LivestatusListener::Start(bool runtimeCreated)
  48 {
  49         ObjectImpl<LivestatusListener>::Start(runtimeCreated);
  50
  51         Log(LogInformation, "LivestatusListener")
  52                 << "'" << GetName() << "' started.";
  53
  54         if (GetSocketType() == "tcp") {
  55                 TcpSocket::Ptr socket = new TcpSocket();
  56
  57                 try {
  58                         socket->Bind(GetBindHost(), GetBindPort(), AF_UNSPEC);
  59                 } catch (std::exception&) {
  60                         Log(LogCritical, "LivestatusListener")
  61                                 << "Cannot bind TCP socket on host '" << GetBindHost() << "' port '" << GetBindPort() << "'.";
  62                         return;
  63                 }
  64
  65                 m_Listener = socket;
  66
  67                 m_Thread = std::thread(std::bind(&LivestatusListener::ServerThreadProc, this));
  68
  69                 Log(LogInformation, "LivestatusListener")
  70                         << "Created TCP socket listening on host '" << GetBindHost() << "' port '" << GetBindPort() << "'.";
  71         }
  72         else if (GetSocketType() == "unix") {
  73 #ifndef _WIN32
  74                 UnixSocket::Ptr socket = new UnixSocket();
  75
  76                 try {
  77                         socket->Bind(GetSocketPath());
  78                 } catch (std::exception&) {
  79                         Log(LogCritical, "LivestatusListener")
  80                                 << "Cannot bind UNIX socket to '" << GetSocketPath() << "'.";
  81                         return;
  82                 }
  83
  84                 /* group must be able to write */
  85                 mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
  86
  87                 if (chmod(GetSocketPath().CStr(), mode) < 0) {
  88                         Log(LogCritical, "LivestatusListener")
  89                                 << "chmod() on unix socket '" << GetSocketPath() << "' failed with error code " << errno << ", \"" << Utility::FormatErrorNumber(errno) << "\"";
  90                         return;
  91                 }
  92
  93                 m_Listener = socket;
  94
  95                 m_Thread = std::thread(std::bind(&LivestatusListener::ServerThreadProc, this));
  96
  97                 Log(LogInformation, "LivestatusListener")
  98                         << "Created UNIX socket in '" << GetSocketPath() << "'.";
  99 #else
 100                 /* no UNIX sockets on windows */
 101                 Log(LogCritical, "LivestatusListener", "Unix sockets are not supported on Windows.");
 102                 return;
 103 #endif
 104         }
 105 }
 106
 107 void LivestatusListener::Stop(bool runtimeRemoved)
 108 {
 109         ObjectImpl<LivestatusListener>::Stop(runtimeRemoved);
 110
 111         Log(LogInformation, "LivestatusListener")
 112                 << "'" << GetName() << "' stopped.";
 113
 114         m_Listener->Close();
 115
 116         if (m_Thread.joinable())
 117                 m_Thread.join();
 118 }
 119
 120 int LivestatusListener::GetClientsConnected()
 121 {
 122         boost::mutex::scoped_lock lock(l_ComponentMutex);
 123
 124         return l_ClientsConnected;
 125 }
 126
 127 int LivestatusListener::GetConnections()
 128 {
 129         boost::mutex::scoped_lock lock(l_ComponentMutex);
 130
 131         return l_Connections;
 132 }
 133
 134 void LivestatusListener::ServerThreadProc()
 135 {
 136         m_Listener->Listen();
 137
 138         try {
 139                 for (;;) {
 140                         timeval tv = { 0, 500000 };
 141
 142                         if (m_Listener->Poll(true, false, &tv)) {
 143                                 Socket::Ptr client = m_Listener->Accept();
 144                                 Log(LogNotice, "LivestatusListener", "Client connected");
 145                                 Utility::QueueAsyncCallback(std::bind(&LivestatusListener::ClientHandler, this, client), LowLatencyScheduler);
 146                         }
 147
 148                         if (!IsActive())
 149                                 break;
 150                 }
 151         } catch (std::exception&) {
 152                 Log(LogCritical, "LivestatusListener", "Cannot accept new connection.");
 153         }
 154
 155         m_Listener->Close();
 156 }
 157
 158 void LivestatusListener::ClientHandler(const Socket::Ptr& client)
 159 {
 160         {
 161                 boost::mutex::scoped_lock lock(l_ComponentMutex);
 162                 l_ClientsConnected++;
 163                 l_Connections++;
 164         }
 165
 166         Stream::Ptr stream = new NetworkStream(client);
 167
 168         StreamReadContext context;
 169
 170         for (;;) {
 171                 String line;
 172
 173                 std::vector<String> lines;
 174
 175                 for (;;) {
 176                         StreamReadStatus srs = stream->ReadLine(&line, context);
 177
 178                         if (srs == StatusEof)
 179                                 break;
 180
 181                         if (srs != StatusNewItem)
 182                                 continue;
 183
 184                         if (line.GetLength() > 0)
 185                                 lines.push_back(line);
 186                         else
 187                                 break;
 188                 }
 189
 190                 if (lines.empty())
 191                         break;
 192
 193                 LivestatusQuery::Ptr query = new LivestatusQuery(lines, GetCompatLogPath());
 194                 if (!query->Execute(stream))
 195                         break;
 196         }
 197
 198         {
 199                 boost::mutex::scoped_lock lock(l_ComponentMutex);
 200                 l_ClientsConnected--;
 201         }
 202 }
 203
 204
 205 void LivestatusListener::ValidateSocketType(const Lazy<String>& lvalue, const ValidationUtils& utils)
 206 {
 207         ObjectImpl<LivestatusListener>::ValidateSocketType(lvalue, utils);
 208
 209         if (lvalue() != "unix" && lvalue() != "tcp")
 210                 BOOST_THROW_EXCEPTION(ValidationError(this, { "socket_type" }, "Socket type '" + lvalue() + "' is invalid."));
 211 }