granicus.if.org Git - icinga2/blob - lib/perfdata/gelfwriter.cpp

   1 /******************************************************************************
   2  * Icinga 2                                                                   *
   3  * Copyright (C) 2012-2017 Icinga Development Team (https://www.icinga.com/)  *
   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 "perfdata/gelfwriter.hpp"
  21 #include "perfdata/gelfwriter.tcpp"
  22 #include "icinga/service.hpp"
  23 #include "icinga/notification.hpp"
  24 #include "icinga/macroprocessor.hpp"
  25 #include "icinga/compatutility.hpp"
  26 #include "icinga/perfdatavalue.hpp"
  27 #include "base/tcpsocket.hpp"
  28 #include "base/configtype.hpp"
  29 #include "base/objectlock.hpp"
  30 #include "base/logger.hpp"
  31 #include "base/utility.hpp"
  32 #include "base/stream.hpp"
  33 #include "base/networkstream.hpp"
  34 #include "base/json.hpp"
  35 #include "base/context.hpp"
  36 #include <boost/algorithm/string/replace.hpp>
  37
  38 using namespace icinga;
  39
  40 REGISTER_TYPE(GelfWriter);
  41
  42 void GelfWriter::Start(bool runtimeCreated)
  43 {
  44         ObjectImpl<GelfWriter>::Start(runtimeCreated);
  45
  46         Log(LogInformation, "GelfWriter")
  47             << "'" << GetName() << "' started.";
  48
  49         m_ReconnectTimer = new Timer();
  50         m_ReconnectTimer->SetInterval(10);
  51         m_ReconnectTimer->OnTimerExpired.connect(boost::bind(&GelfWriter::ReconnectTimerHandler, this));
  52         m_ReconnectTimer->Start();
  53         m_ReconnectTimer->Reschedule(0);
  54
  55         // Send check results
  56         Service::OnNewCheckResult.connect(boost::bind(&GelfWriter::CheckResultHandler, this, _1, _2));
  57         // Send notifications
  58         Service::OnNotificationSentToUser.connect(boost::bind(&GelfWriter::NotificationToUserHandler, this, _1, _2, _3, _4, _5, _6, _7, _8));
  59         // Send state change
  60         Service::OnStateChange.connect(boost::bind(&GelfWriter::StateChangeHandler, this, _1, _2, _3));
  61 }
  62
  63 void GelfWriter::Stop(bool runtimeRemoved)
  64 {
  65         Log(LogInformation, "GelfWriter")
  66             << "'" << GetName() << "' stopped.";
  67
  68         ObjectImpl<GelfWriter>::Stop(runtimeRemoved);
  69 }
  70
  71 void GelfWriter::ReconnectTimerHandler(void)
  72 {
  73         if (m_Stream)
  74                 return;
  75
  76         TcpSocket::Ptr socket = new TcpSocket();
  77
  78         Log(LogNotice, "GelfWriter")
  79             << "Reconnecting to GELF endpoint '" << GetHost() << "' port '" << GetPort() << "'.";
  80
  81         try {
  82                 socket->Connect(GetHost(), GetPort());
  83         } catch (std::exception&) {
  84                 Log(LogCritical, "GelfWriter")
  85                     << "Can't connect to GELF endpoint '" << GetHost() << "' port '" << GetPort() << "'.";
  86                 return;
  87         }
  88
  89         m_Stream = new NetworkStream(socket);
  90 }
  91
  92 void GelfWriter::CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
  93 {
  94         CONTEXT("GELF Processing check result for '" + checkable->GetName() + "'");
  95
  96         Log(LogDebug, "GelfWriter")
  97             << "GELF Processing check result for '" << checkable->GetName() << "'";
  98
  99         Host::Ptr host;
 100         Service::Ptr service;
 101         tie(host, service) = GetHostService(checkable);
 102         double ts = Utility::GetTime();
 103
 104         Dictionary::Ptr fields = new Dictionary();
 105
 106         if (service) {
 107                 fields->Set("_service_name", service->GetShortName());
 108                 fields->Set("_service_state", Service::StateToString(service->GetState()));
 109                 fields->Set("_last_state", service->GetLastState());
 110                 fields->Set("_last_hard_state", service->GetLastHardState());
 111         } else {
 112                 fields->Set("_last_state", host->GetLastState());
 113                 fields->Set("_last_hard_state", host->GetLastHardState());
 114         }
 115
 116         fields->Set("_hostname", host->GetName());
 117         fields->Set("_type", "CHECK RESULT");
 118         fields->Set("_state", service ? Service::StateToString(service->GetState()) : Host::StateToString(host->GetState()));
 119
 120         fields->Set("_current_check_attempt", checkable->GetCheckAttempt());
 121         fields->Set("_max_check_attempts", checkable->GetMaxCheckAttempts());
 122
 123         fields->Set("_reachable", checkable->IsReachable());
 124
 125         if (cr) {
 126                 fields->Set("_latency", cr->CalculateLatency());
 127                 fields->Set("_execution_time", cr->CalculateExecutionTime());
 128                 fields->Set("short_message", CompatUtility::GetCheckResultOutput(cr));
 129                 fields->Set("full_message", cr->GetOutput());
 130                 fields->Set("_check_source", cr->GetCheckSource());
 131                 ts = cr->GetExecutionEnd();
 132         }
 133
 134         if (cr && GetEnableSendPerfdata()) {
 135                 Array::Ptr perfdata = cr->GetPerformanceData();
 136
 137                 if (perfdata) {
 138                         ObjectLock olock(perfdata);
 139                         for (const Value& val : perfdata) {
 140                                 PerfdataValue::Ptr pdv;
 141
 142                                 if (val.IsObjectType<PerfdataValue>())
 143                                         pdv = val;
 144                                 else {
 145                                         try {
 146                                                 pdv = PerfdataValue::Parse(val);
 147                                         } catch (const std::exception&) {
 148                                                 Log(LogWarning, "GelfWriter")
 149                                                     << "Ignoring invalid perfdata value: '" << val << "' for object '"
 150                                                     << checkable->GetName() << "'.";
 151                                         }
 152                                 }
 153
 154                                 String escaped_key = pdv->GetLabel();
 155                                 boost::replace_all(escaped_key, " ", "_");
 156                                 boost::replace_all(escaped_key, ".", "_");
 157                                 boost::replace_all(escaped_key, "\\", "_");
 158                                 boost::algorithm::replace_all(escaped_key, "::", ".");
 159
 160                                 fields->Set("_" + escaped_key, pdv->GetValue());
 161
 162                                 if (pdv->GetMin())
 163                                         fields->Set("_" + escaped_key + "_min", pdv->GetMin());
 164                                 if (pdv->GetMax())
 165                                         fields->Set("_" + escaped_key + "_max", pdv->GetMax());
 166                                 if (pdv->GetWarn())
 167                                         fields->Set("_" + escaped_key + "_warn", pdv->GetWarn());
 168                                 if (pdv->GetCrit())
 169                                         fields->Set("_" + escaped_key + "_crit", pdv->GetCrit());
 170                         }
 171                 }
 172         }
 173
 174         SendLogMessage(ComposeGelfMessage(fields, GetSource(), ts));
 175 }
 176
 177 void GelfWriter::NotificationToUserHandler(const Notification::Ptr& notification, const Checkable::Ptr& checkable,
 178     const User::Ptr& user, NotificationType notification_type, CheckResult::Ptr const& cr,
 179     const String& author, const String& comment_text, const String& command_name)
 180 {
 181         CONTEXT("GELF Processing notification to all users '" + checkable->GetName() + "'");
 182
 183         Log(LogDebug, "GelfWriter")
 184             << "GELF Processing notification for '" << checkable->GetName() << "'";
 185
 186         Host::Ptr host;
 187         Service::Ptr service;
 188         tie(host, service) = GetHostService(checkable);
 189         double ts = Utility::GetTime();
 190
 191         String notification_type_str = Notification::NotificationTypeToString(notification_type);
 192
 193         String author_comment = "";
 194
 195         if (notification_type == NotificationCustom || notification_type == NotificationAcknowledgement) {
 196                 author_comment = author + ";" + comment_text;
 197         }
 198
 199         String output;
 200
 201         if (cr) {
 202                 output = CompatUtility::GetCheckResultOutput(cr);
 203                 ts = cr->GetExecutionEnd();
 204         }
 205
 206         Dictionary::Ptr fields = new Dictionary();
 207
 208         if (service) {
 209                 fields->Set("_type", "SERVICE NOTIFICATION");
 210                 fields->Set("_service", service->GetShortName());
 211                 fields->Set("short_message", output);
 212         } else {
 213                 fields->Set("_type", "HOST NOTIFICATION");
 214                 fields->Set("short_message", "(" + CompatUtility::GetHostStateString(host) + ")");
 215         }
 216
 217         fields->Set("_state", service ? Service::StateToString(service->GetState()) : Host::StateToString(host->GetState()));
 218
 219         fields->Set("_hostname", host->GetName());
 220         fields->Set("_command", command_name);
 221         fields->Set("_notification_type", notification_type_str);
 222         fields->Set("_comment", author_comment);
 223
 224         SendLogMessage(ComposeGelfMessage(fields, GetSource(), ts));
 225 }
 226
 227 void GelfWriter::StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type)
 228 {
 229         CONTEXT("GELF Processing state change '" + checkable->GetName() + "'");
 230
 231         Log(LogDebug, "GelfWriter")
 232             << "GELF Processing state change for '" << checkable->GetName() << "'";
 233
 234         Host::Ptr host;
 235         Service::Ptr service;
 236         tie(host, service) = GetHostService(checkable);
 237         double ts = Utility::GetTime();
 238
 239         Dictionary::Ptr fields = new Dictionary();
 240
 241         fields->Set("_state", service ? Service::StateToString(service->GetState()) : Host::StateToString(host->GetState()));
 242         fields->Set("_type", "STATE CHANGE");
 243         fields->Set("_current_check_attempt", checkable->GetCheckAttempt());
 244         fields->Set("_max_check_attempts", checkable->GetMaxCheckAttempts());
 245         fields->Set("_hostname", host->GetName());
 246
 247         if (service) {
 248                 fields->Set("_service_name", service->GetShortName());
 249                 fields->Set("_service_state", Service::StateToString(service->GetState()));
 250                 fields->Set("_last_state", service->GetLastState());
 251                 fields->Set("_last_hard_state", service->GetLastHardState());
 252         } else {
 253                 fields->Set("_last_state", host->GetLastState());
 254                 fields->Set("_last_hard_state", host->GetLastHardState());
 255         }
 256
 257         if (cr) {
 258                 fields->Set("short_message", CompatUtility::GetCheckResultOutput(cr));
 259                 fields->Set("full_message", cr->GetOutput());
 260                 fields->Set("_check_source", cr->GetCheckSource());
 261                 ts = cr->GetExecutionEnd();
 262         }
 263
 264         SendLogMessage(ComposeGelfMessage(fields, GetSource(), ts));
 265 }
 266
 267 String GelfWriter::ComposeGelfMessage(const Dictionary::Ptr& fields, const String& source, double ts)
 268 {
 269         fields->Set("version", "1.1");
 270         fields->Set("host", source);
 271         fields->Set("timestamp", ts);
 272
 273         return JsonEncode(fields);
 274 }
 275
 276 void GelfWriter::SendLogMessage(const String& gelf)
 277 {
 278         std::ostringstream msgbuf;
 279         msgbuf << gelf;
 280         msgbuf << '\0';
 281
 282         String log = msgbuf.str();
 283
 284         ObjectLock olock(this);
 285
 286         if (!m_Stream)
 287                 return;
 288
 289         try {
 290                 //TODO remove
 291                 Log(LogDebug, "GelfWriter")
 292                     << "Sending '" << log << "'.";
 293                 m_Stream->Write(log.CStr(), log.GetLength());
 294         } catch (const std::exception& ex) {
 295                 Log(LogCritical, "GelfWriter")
 296                     << "Cannot write to TCP socket on host '" << GetHost() << "' port '" << GetPort() << "'.";
 297
 298                 m_Stream.reset();
 299         }
 300 }