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

   1 /* Icinga 2 | (c) 2012 Icinga GmbH | GPLv2+ */
   2
   3 #include "perfdata/opentsdbwriter.hpp"
   4 #include "perfdata/opentsdbwriter-ti.cpp"
   5 #include "icinga/service.hpp"
   6 #include "icinga/checkcommand.hpp"
   7 #include "icinga/macroprocessor.hpp"
   8 #include "icinga/icingaapplication.hpp"
   9 #include "icinga/compatutility.hpp"
  10 #include "base/tcpsocket.hpp"
  11 #include "base/configtype.hpp"
  12 #include "base/objectlock.hpp"
  13 #include "base/logger.hpp"
  14 #include "base/convert.hpp"
  15 #include "base/utility.hpp"
  16 #include "base/perfdatavalue.hpp"
  17 #include "base/application.hpp"
  18 #include "base/stream.hpp"
  19 #include "base/networkstream.hpp"
  20 #include "base/exception.hpp"
  21 #include "base/statsfunction.hpp"
  22 #include <boost/algorithm/string.hpp>
  23 #include <boost/algorithm/string/replace.hpp>
  24
  25 using namespace icinga;
  26
  27 REGISTER_TYPE(OpenTsdbWriter);
  28
  29 REGISTER_STATSFUNCTION(OpenTsdbWriter, &OpenTsdbWriter::StatsFunc);
  30
  31 /*
  32  * Enable HA capabilities once the config object is loaded.
  33  */
  34 void OpenTsdbWriter::OnConfigLoaded()
  35 {
  36         ObjectImpl<OpenTsdbWriter>::OnConfigLoaded();
  37
  38         if (!GetEnableHa()) {
  39                 Log(LogDebug, "OpenTsdbWriter")
  40                         << "HA functionality disabled. Won't pause connection: " << GetName();
  41
  42                 SetHAMode(HARunEverywhere);
  43         } else {
  44                 SetHAMode(HARunOnce);
  45         }
  46 }
  47
  48 /**
  49  * Feature stats interface
  50  *
  51  * @param status Key value pairs for feature stats
  52  */
  53 void OpenTsdbWriter::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr&)
  54 {
  55         DictionaryData nodes;
  56
  57         for (const OpenTsdbWriter::Ptr& opentsdbwriter : ConfigType::GetObjectsByType<OpenTsdbWriter>()) {
  58                 nodes.emplace_back(opentsdbwriter->GetName(), new Dictionary({
  59                         { "connected", opentsdbwriter->GetConnected() }
  60                 }));
  61         }
  62
  63         status->Set("opentsdbwriter", new Dictionary(std::move(nodes)));
  64 }
  65
  66 /**
  67  * Resume is equivalent to Start, but with HA capabilities to resume at runtime.
  68  */
  69 void OpenTsdbWriter::Resume()
  70 {
  71         ObjectImpl<OpenTsdbWriter>::Resume();
  72
  73         Log(LogInformation, "OpentsdbWriter")
  74                 << "'" << GetName() << "' resumed.";
  75
  76         m_ReconnectTimer = new Timer();
  77         m_ReconnectTimer->SetInterval(10);
  78         m_ReconnectTimer->OnTimerExpired.connect(std::bind(&OpenTsdbWriter::ReconnectTimerHandler, this));
  79         m_ReconnectTimer->Start();
  80         m_ReconnectTimer->Reschedule(0);
  81
  82         Service::OnNewCheckResult.connect(std::bind(&OpenTsdbWriter::CheckResultHandler, this, _1, _2));
  83 }
  84
  85 /**
  86  * Pause is equivalent to Stop, but with HA capabilities to resume at runtime.
  87  */
  88 void OpenTsdbWriter::Pause()
  89 {
  90         m_ReconnectTimer.reset();
  91
  92         Log(LogInformation, "OpentsdbWriter")
  93                 << "'" << GetName() << "' paused.";
  94
  95         m_Stream->close();
  96
  97         SetConnected(false);
  98
  99         ObjectImpl<OpenTsdbWriter>::Pause();
 100 }
 101
 102 /**
 103  * Reconnect handler called by the timer.
 104  * Handles TLS
 105  */
 106 void OpenTsdbWriter::ReconnectTimerHandler()
 107 {
 108         if (IsPaused())
 109                 return;
 110
 111         SetShouldConnect(true);
 112
 113         if (GetConnected())
 114                 return;
 115
 116         Log(LogNotice, "OpenTsdbWriter")
 117                 << "Reconnect to OpenTSDB TSD on host '" << GetHost() << "' port '" << GetPort() << "'.";
 118
 119         /*
 120          * We're using telnet as input method. Future PRs may change this into using the HTTP API.
 121          * http://opentsdb.net/docs/build/html/user_guide/writing/index.html#telnet
 122          */
 123
 124         m_Stream = std::make_shared<AsioTcpStream>(IoEngine::Get().GetIoContext());
 125
 126         try {
 127                 icinga::Connect(m_Stream->lowest_layer(), GetHost(), GetPort());
 128         } catch (const std::exception& ex) {
 129                 Log(LogWarning, "OpenTsdbWriter")
 130                         << "Can't connect to OpenTSDB on host '" << GetHost() << "' port '" << GetPort() << ".'";
 131         }
 132
 133         SetConnected(true);
 134 }
 135
 136 /**
 137  * Registered check result handler processing data.
 138  * Calculates tags from the config.
 139  *
 140  * @param checkable Host/service object
 141  * @param cr Check result
 142  */
 143 void OpenTsdbWriter::CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
 144 {
 145         if (IsPaused())
 146                 return;
 147
 148         CONTEXT("Processing check result for '" + checkable->GetName() + "'");
 149
 150         if (!IcingaApplication::GetInstance()->GetEnablePerfdata() || !checkable->GetEnablePerfdata())
 151                 return;
 152
 153         Service::Ptr service = dynamic_pointer_cast<Service>(checkable);
 154         Host::Ptr host;
 155
 156         if (service)
 157                 host = service->GetHost();
 158         else
 159                 host = static_pointer_cast<Host>(checkable);
 160
 161         String metric;
 162         std::map<String, String> tags;
 163
 164         String escaped_hostName = EscapeTag(host->GetName());
 165         tags["host"] = escaped_hostName;
 166
 167         double ts = cr->GetExecutionEnd();
 168
 169         if (service) {
 170                 String serviceName = service->GetShortName();
 171                 String escaped_serviceName = EscapeMetric(serviceName);
 172                 metric = "icinga.service." + escaped_serviceName;
 173
 174                 SendMetric(checkable, metric + ".state", tags, service->GetState(), ts);
 175         } else {
 176                 metric = "icinga.host";
 177                 SendMetric(checkable, metric + ".state", tags, host->GetState(), ts);
 178         }
 179
 180         SendMetric(checkable, metric + ".state_type", tags, checkable->GetStateType(), ts);
 181         SendMetric(checkable, metric + ".reachable", tags, checkable->IsReachable(), ts);
 182         SendMetric(checkable, metric + ".downtime_depth", tags, checkable->GetDowntimeDepth(), ts);
 183         SendMetric(checkable, metric + ".acknowledgement", tags, checkable->GetAcknowledgement(), ts);
 184
 185         SendPerfdata(checkable, metric, tags, cr, ts);
 186
 187         metric = "icinga.check";
 188
 189         if (service) {
 190                 tags["type"] = "service";
 191                 String serviceName = service->GetShortName();
 192                 String escaped_serviceName = EscapeTag(serviceName);
 193                 tags["service"] = escaped_serviceName;
 194         } else {
 195                 tags["type"] = "host";
 196         }
 197
 198         SendMetric(checkable, metric + ".current_attempt", tags, checkable->GetCheckAttempt(), ts);
 199         SendMetric(checkable, metric + ".max_check_attempts", tags, checkable->GetMaxCheckAttempts(), ts);
 200         SendMetric(checkable, metric + ".latency", tags, cr->CalculateLatency(), ts);
 201         SendMetric(checkable, metric + ".execution_time", tags, cr->CalculateExecutionTime(), ts);
 202 }
 203
 204 /**
 205  * Parse and send performance data metrics to OpenTSDB
 206  *
 207  * @param checkable Host/service object
 208  * @param metric Full metric name
 209  * @param tags Tag key pairs
 210  * @param cr Check result containing performance data
 211  * @param ts Timestamp when the check result was received
 212  */
 213 void OpenTsdbWriter::SendPerfdata(const Checkable::Ptr& checkable, const String& metric,
 214         const std::map<String, String>& tags, const CheckResult::Ptr& cr, double ts)
 215 {
 216         Array::Ptr perfdata = cr->GetPerformanceData();
 217
 218         if (!perfdata)
 219                 return;
 220
 221         CheckCommand::Ptr checkCommand = checkable->GetCheckCommand();
 222
 223         ObjectLock olock(perfdata);
 224         for (const Value& val : perfdata) {
 225                 PerfdataValue::Ptr pdv;
 226
 227                 if (val.IsObjectType<PerfdataValue>())
 228                         pdv = val;
 229                 else {
 230                         try {
 231                                 pdv = PerfdataValue::Parse(val);
 232                         } catch (const std::exception&) {
 233                                 Log(LogWarning, "OpenTsdbWriter")
 234                                         << "Ignoring invalid perfdata for checkable '"
 235                                         << checkable->GetName() << "' and command '"
 236                                         << checkCommand->GetName() << "' with value: " << val;
 237                                 continue;
 238                         }
 239                 }
 240
 241                 String escaped_key = EscapeMetric(pdv->GetLabel());
 242                 boost::algorithm::replace_all(escaped_key, "::", ".");
 243
 244                 SendMetric(checkable, metric + "." + escaped_key, tags, pdv->GetValue(), ts);
 245
 246                 if (pdv->GetCrit())
 247                         SendMetric(checkable, metric + "." + escaped_key + "_crit", tags, pdv->GetCrit(), ts);
 248                 if (pdv->GetWarn())
 249                         SendMetric(checkable, metric + "." + escaped_key + "_warn", tags, pdv->GetWarn(), ts);
 250                 if (pdv->GetMin())
 251                         SendMetric(checkable, metric + "." + escaped_key + "_min", tags, pdv->GetMin(), ts);
 252                 if (pdv->GetMax())
 253                         SendMetric(checkable, metric + "." + escaped_key + "_max", tags, pdv->GetMax(), ts);
 254         }
 255 }
 256
 257 /**
 258  * Send given metric to OpenTSDB
 259  *
 260  * @param checkable Host/service object
 261  * @param metric Full metric name
 262  * @param tags Tag key pairs
 263  * @param value Floating point metric value
 264  * @param ts Timestamp where the metric was received from the check result
 265  */
 266 void OpenTsdbWriter::SendMetric(const Checkable::Ptr& checkable, const String& metric,
 267         const std::map<String, String>& tags, double value, double ts)
 268 {
 269         String tags_string = "";
 270
 271         for (const Dictionary::Pair& tag : tags) {
 272                 tags_string += " " + tag.first + "=" + Convert::ToString(tag.second);
 273         }
 274
 275         std::ostringstream msgbuf;
 276         /*
 277          * must be (http://opentsdb.net/docs/build/html/user_guide/query/timeseries.html)
 278          * put <metric> <timestamp> <value> <tagk1=tagv1[ tagk2=tagv2 ...tagkN=tagvN]>
 279          * "tags" must include at least one tag, we use "host=HOSTNAME"
 280          */
 281         msgbuf << "put " << metric << " " << static_cast<long>(ts) << " " << Convert::ToString(value) << " " << tags_string;
 282
 283         Log(LogDebug, "OpenTsdbWriter")
 284                 << "Checkable '" << checkable->GetName() << "' adds to metric list: '" << msgbuf.str() << "'.";
 285
 286         /* do not send \n to debug log */
 287         msgbuf << "\n";
 288         String put = msgbuf.str();
 289
 290         ObjectLock olock(this);
 291
 292         if (!GetConnected())
 293                 return;
 294
 295         try {
 296                 Log(LogDebug, "OpenTsdbWriter")
 297                         << "Checkable '" << checkable->GetName() << "' sending message '" << put << "'.";
 298
 299                 boost::asio::write(*m_Stream, boost::asio::buffer(msgbuf.str()));
 300                 m_Stream->flush();
 301         } catch (const std::exception& ex) {
 302                 Log(LogCritical, "OpenTsdbWriter")
 303                         << "Cannot write to TCP socket on host '" << GetHost() << "' port '" << GetPort() << "'.";
 304         }
 305 }
 306
 307 /**
 308  * Escape tags for OpenTSDB
 309  * http://opentsdb.net/docs/build/html/user_guide/query/timeseries.html#precisions-on-metrics-and-tags
 310  *
 311  * @param str Tag name
 312  * @return Escaped tag
 313  */
 314 String OpenTsdbWriter::EscapeTag(const String& str)
 315 {
 316         String result = str;
 317
 318         boost::replace_all(result, " ", "_");
 319         boost::replace_all(result, "\\", "_");
 320
 321         return result;
 322 }
 323
 324 /**
 325  * Escape metric name for OpenTSDB
 326  * http://opentsdb.net/docs/build/html/user_guide/query/timeseries.html#precisions-on-metrics-and-tags
 327  *
 328  * @param str Metric name
 329  * @return Escaped metric
 330  */
 331 String OpenTsdbWriter::EscapeMetric(const String& str)
 332 {
 333         String result = str;
 334
 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 }