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

   1 /******************************************************************************
   2  * Icinga 2                                                                   *
   3  * Copyright (C) 2012-2018 Icinga Development Team (https://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/graphitewriter.hpp"
  21 #include "perfdata/graphitewriter-ti.cpp"
  22 #include "icinga/service.hpp"
  23 #include "icinga/macroprocessor.hpp"
  24 #include "icinga/icingaapplication.hpp"
  25 #include "base/tcpsocket.hpp"
  26 #include "base/configtype.hpp"
  27 #include "base/objectlock.hpp"
  28 #include "base/logger.hpp"
  29 #include "base/convert.hpp"
  30 #include "base/utility.hpp"
  31 #include "base/perfdatavalue.hpp"
  32 #include "base/application.hpp"
  33 #include "base/stream.hpp"
  34 #include "base/networkstream.hpp"
  35 #include "base/exception.hpp"
  36 #include "base/statsfunction.hpp"
  37 #include <boost/algorithm/string.hpp>
  38 #include <boost/algorithm/string/replace.hpp>
  39 #include <utility>
  40
  41 using namespace icinga;
  42
  43 REGISTER_TYPE(GraphiteWriter);
  44
  45 REGISTER_STATSFUNCTION(GraphiteWriter, &GraphiteWriter::StatsFunc);
  46
  47 void GraphiteWriter::OnConfigLoaded()
  48 {
  49         ObjectImpl<GraphiteWriter>::OnConfigLoaded();
  50
  51         m_WorkQueue.SetName("GraphiteWriter, " + GetName());
  52
  53         if (!GetEnableHa()) {
  54                 Log(LogDebug, "GraphiteWriter")
  55                         << "HA functionality disabled. Won't pause connection: " << GetName();
  56
  57                 SetHAMode(HARunEverywhere);
  58         } else {
  59                 SetHAMode(HARunOnce);
  60         }
  61 }
  62
  63 void GraphiteWriter::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
  64 {
  65         DictionaryData nodes;
  66
  67         for (const GraphiteWriter::Ptr& graphitewriter : ConfigType::GetObjectsByType<GraphiteWriter>()) {
  68                 size_t workQueueItems = graphitewriter->m_WorkQueue.GetLength();
  69                 double workQueueItemRate = graphitewriter->m_WorkQueue.GetTaskCount(60) / 60.0;
  70
  71                 nodes.emplace_back(graphitewriter->GetName(), new Dictionary({
  72                         { "work_queue_items", workQueueItems },
  73                         { "work_queue_item_rate", workQueueItemRate },
  74                         { "connected", graphitewriter->GetConnected() }
  75                 }));
  76
  77                 perfdata->Add(new PerfdataValue("graphitewriter_" + graphitewriter->GetName() + "_work_queue_items", workQueueItems));
  78                 perfdata->Add(new PerfdataValue("graphitewriter_" + graphitewriter->GetName() + "_work_queue_item_rate", workQueueItemRate));
  79         }
  80
  81         status->Set("graphitewriter", new Dictionary(std::move(nodes)));
  82 }
  83
  84 void GraphiteWriter::Resume()
  85 {
  86         ObjectImpl<GraphiteWriter>::Resume();
  87
  88         Log(LogInformation, "GraphiteWriter")
  89                 << "'" << GetName() << "' resumed.";
  90
  91         /* Register exception handler for WQ tasks. */
  92         m_WorkQueue.SetExceptionCallback(std::bind(&GraphiteWriter::ExceptionHandler, this, _1));
  93
  94         /* Timer for reconnecting */
  95         m_ReconnectTimer = new Timer();
  96         m_ReconnectTimer->SetInterval(10);
  97         m_ReconnectTimer->OnTimerExpired.connect(std::bind(&GraphiteWriter::ReconnectTimerHandler, this));
  98         m_ReconnectTimer->Start();
  99         m_ReconnectTimer->Reschedule(0);
 100
 101         /* Register event handlers. */
 102         Checkable::OnNewCheckResult.connect(std::bind(&GraphiteWriter::CheckResultHandler, this, _1, _2));
 103 }
 104
 105 void GraphiteWriter::Pause()
 106 {
 107         Log(LogInformation, "GraphiteWriter")
 108                 << "'" << GetName() << "' paused.";
 109
 110         m_WorkQueue.Join();
 111
 112         ObjectImpl<GraphiteWriter>::Pause();
 113 }
 114
 115 void GraphiteWriter::AssertOnWorkQueue()
 116 {
 117         ASSERT(m_WorkQueue.IsWorkerThread());
 118 }
 119
 120 void GraphiteWriter::ExceptionHandler(boost::exception_ptr exp)
 121 {
 122         Log(LogCritical, "GraphiteWriter", "Exception during Graphite operation: Verify that your backend is operational!");
 123
 124         Log(LogDebug, "GraphiteWriter")
 125                 << "Exception during Graphite operation: " << DiagnosticInformation(std::move(exp));
 126
 127         if (GetConnected()) {
 128                 m_Stream->Close();
 129
 130                 SetConnected(false);
 131         }
 132 }
 133
 134 void GraphiteWriter::Reconnect()
 135 {
 136         AssertOnWorkQueue();
 137
 138         if (IsPaused()) {
 139                 SetConnected(false);
 140                 return;
 141         }
 142
 143         double startTime = Utility::GetTime();
 144
 145         CONTEXT("Reconnecting to Graphite '" + GetName() + "'");
 146
 147         SetShouldConnect(true);
 148
 149         if (GetConnected())
 150                 return;
 151
 152         TcpSocket::Ptr socket = new TcpSocket();
 153
 154         Log(LogNotice, "GraphiteWriter")
 155                 << "Reconnecting to Graphite on host '" << GetHost() << "' port '" << GetPort() << "'.";
 156
 157         try {
 158                 socket->Connect(GetHost(), GetPort());
 159         } catch (const std::exception& ex) {
 160                 Log(LogCritical, "GraphiteWriter")
 161                         << "Can't connect to Graphite on host '" << GetHost() << "' port '" << GetPort() << "'.";
 162                 throw ex;
 163         }
 164
 165         m_Stream = new NetworkStream(socket);
 166
 167         SetConnected(true);
 168
 169         Log(LogInformation, "GraphiteWriter")
 170                 << "Finished reconnecting to Graphite in " << std::setw(2) << Utility::GetTime() - startTime << " second(s).";
 171 }
 172
 173 void GraphiteWriter::ReconnectTimerHandler()
 174 {
 175         m_WorkQueue.Enqueue(std::bind(&GraphiteWriter::Reconnect, this), PriorityNormal);
 176 }
 177
 178 void GraphiteWriter::Disconnect()
 179 {
 180         AssertOnWorkQueue();
 181
 182         if (!GetConnected())
 183                 return;
 184
 185         m_Stream->Close();
 186
 187         SetConnected(false);
 188 }
 189
 190 void GraphiteWriter::CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
 191 {
 192         if (IsPaused())
 193                 return;
 194
 195         m_WorkQueue.Enqueue(std::bind(&GraphiteWriter::CheckResultHandlerInternal, this, checkable, cr));
 196 }
 197
 198 void GraphiteWriter::CheckResultHandlerInternal(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
 199 {
 200         AssertOnWorkQueue();
 201
 202         CONTEXT("Processing check result for '" + checkable->GetName() + "'");
 203
 204         if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata())
 205                 return;
 206
 207         Host::Ptr host;
 208         Service::Ptr service;
 209         tie(host, service) = GetHostService(checkable);
 210
 211         MacroProcessor::ResolverList resolvers;
 212         if (service)
 213                 resolvers.emplace_back("service", service);
 214         resolvers.emplace_back("host", host);
 215         resolvers.emplace_back("icinga", IcingaApplication::GetInstance());
 216
 217         String prefix;
 218
 219         if (service) {
 220                 prefix = MacroProcessor::ResolveMacros(GetServiceNameTemplate(), resolvers, cr, nullptr, std::bind(&GraphiteWriter::EscapeMacroMetric, _1));
 221         } else {
 222                 prefix = MacroProcessor::ResolveMacros(GetHostNameTemplate(), resolvers, cr, nullptr, std::bind(&GraphiteWriter::EscapeMacroMetric, _1));
 223         }
 224
 225         String prefixPerfdata = prefix + ".perfdata";
 226         String prefixMetadata = prefix + ".metadata";
 227
 228         double ts = cr->GetExecutionEnd();
 229
 230         if (GetEnableSendMetadata()) {
 231                 if (service) {
 232                         SendMetric(prefixMetadata, "state", service->GetState(), ts);
 233                 } else {
 234                         SendMetric(prefixMetadata, "state", host->GetState(), ts);
 235                 }
 236
 237                 SendMetric(prefixMetadata, "current_attempt", checkable->GetCheckAttempt(), ts);
 238                 SendMetric(prefixMetadata, "max_check_attempts", checkable->GetMaxCheckAttempts(), ts);
 239                 SendMetric(prefixMetadata, "state_type", checkable->GetStateType(), ts);
 240                 SendMetric(prefixMetadata, "reachable", checkable->IsReachable(), ts);
 241                 SendMetric(prefixMetadata, "downtime_depth", checkable->GetDowntimeDepth(), ts);
 242                 SendMetric(prefixMetadata, "acknowledgement", checkable->GetAcknowledgement(), ts);
 243                 SendMetric(prefixMetadata, "latency", cr->CalculateLatency(), ts);
 244                 SendMetric(prefixMetadata, "execution_time", cr->CalculateExecutionTime(), ts);
 245         }
 246
 247         SendPerfdata(prefixPerfdata, cr, ts);
 248 }
 249
 250 void GraphiteWriter::SendPerfdata(const String& prefix, const CheckResult::Ptr& cr, double ts)
 251 {
 252         Array::Ptr perfdata = cr->GetPerformanceData();
 253
 254         if (!perfdata)
 255                 return;
 256
 257         ObjectLock olock(perfdata);
 258         for (const Value& val : perfdata) {
 259                 PerfdataValue::Ptr pdv;
 260
 261                 if (val.IsObjectType<PerfdataValue>())
 262                         pdv = val;
 263                 else {
 264                         try {
 265                                 pdv = PerfdataValue::Parse(val);
 266                         } catch (const std::exception&) {
 267                                 Log(LogWarning, "GraphiteWriter")
 268                                         << "Ignoring invalid perfdata value: " << val;
 269                                 continue;
 270                         }
 271                 }
 272
 273                 String escapedKey = EscapeMetricLabel(pdv->GetLabel());
 274
 275                 SendMetric(prefix, escapedKey + ".value", pdv->GetValue(), ts);
 276
 277                 if (GetEnableSendThresholds()) {
 278                         if (pdv->GetCrit())
 279                                 SendMetric(prefix, escapedKey + ".crit", pdv->GetCrit(), ts);
 280                         if (pdv->GetWarn())
 281                                 SendMetric(prefix, escapedKey + ".warn", pdv->GetWarn(), ts);
 282                         if (pdv->GetMin())
 283                                 SendMetric(prefix, escapedKey + ".min", pdv->GetMin(), ts);
 284                         if (pdv->GetMax())
 285                                 SendMetric(prefix, escapedKey + ".max", pdv->GetMax(), ts);
 286                 }
 287         }
 288 }
 289
 290 void GraphiteWriter::SendMetric(const String& prefix, const String& name, double value, double ts)
 291 {
 292         std::ostringstream msgbuf;
 293         msgbuf << prefix << "." << name << " " << Convert::ToString(value) << " " << static_cast<long>(ts);
 294
 295         Log(LogDebug, "GraphiteWriter")
 296                 << "Add to metric list:'" << msgbuf.str() << "'.";
 297
 298         // do not send \n to debug log
 299         msgbuf << "\n";
 300         String metric = msgbuf.str();
 301
 302         boost::mutex::scoped_lock lock(m_StreamMutex);
 303
 304         if (!GetConnected())
 305                 return;
 306
 307         try {
 308                 m_Stream->Write(metric.CStr(), metric.GetLength());
 309         } catch (const std::exception& ex) {
 310                 Log(LogCritical, "GraphiteWriter")
 311                         << "Cannot write to TCP socket on host '" << GetHost() << "' port '" << GetPort() << "'.";
 312
 313                 throw ex;
 314         }
 315 }
 316
 317 String GraphiteWriter::EscapeMetric(const String& str)
 318 {
 319         String result = str;
 320
 321         //don't allow '.' in metric prefixes
 322         boost::replace_all(result, " ", "_");
 323         boost::replace_all(result, ".", "_");
 324         boost::replace_all(result, "\\", "_");
 325         boost::replace_all(result, "/", "_");
 326
 327         return result;
 328 }
 329
 330 String GraphiteWriter::EscapeMetricLabel(const String& str)
 331 {
 332         String result = str;
 333
 334         //allow to pass '.' in perfdata labels
 335         boost::replace_all(result, " ", "_");
 336         boost::replace_all(result, "\\", "_");
 337         boost::replace_all(result, "/", "_");
 338         boost::replace_all(result, "::", ".");
 339
 340         return result;
 341 }
 342
 343 Value GraphiteWriter::EscapeMacroMetric(const Value& value)
 344 {
 345         if (value.IsObjectType<Array>()) {
 346                 Array::Ptr arr = value;
 347                 ArrayData result;
 348
 349                 ObjectLock olock(arr);
 350                 for (const Value& arg : arr) {
 351                         result.push_back(EscapeMetric(arg));
 352                 }
 353
 354                 return Utility::Join(new Array(std::move(result)), '.');
 355         } else
 356                 return EscapeMetric(value);
 357 }
 358
 359 void GraphiteWriter::ValidateHostNameTemplate(const Lazy<String>& lvalue, const ValidationUtils& utils)
 360 {
 361         ObjectImpl<GraphiteWriter>::ValidateHostNameTemplate(lvalue, utils);
 362
 363         if (!MacroProcessor::ValidateMacroString(lvalue()))
 364                 BOOST_THROW_EXCEPTION(ValidationError(this, { "host_name_template" }, "Closing $ not found in macro format string '" + lvalue() + "'."));
 365 }
 366
 367 void GraphiteWriter::ValidateServiceNameTemplate(const Lazy<String>& lvalue, const ValidationUtils& utils)
 368 {
 369         ObjectImpl<GraphiteWriter>::ValidateServiceNameTemplate(lvalue, utils);
 370
 371         if (!MacroProcessor::ValidateMacroString(lvalue()))
 372                 BOOST_THROW_EXCEPTION(ValidationError(this, { "service_name_template" }, "Closing $ not found in macro format string '" + lvalue() + "'."));
 373 }