telemetry_aggregator_influx.cc

Go to the documentation of this file.

#include "telemetry_aggregator_influx.h"

#include <netinet/in.h>
#include <pthread.h>
#include <stdint.h>
#include <sys/socket.h>
#include <sys/types.h>

#include <cstring>
#include <map>
#include <string>
#include <vector>

#include "util/logging.h"
#include "util/posix.h"
#include "util/string.h"

namespace perf {

TelemetryAggregatorInflux::TelemetryAggregatorInflux(
    Statistics *statistics,
    int send_rate_sec,
    OptionsManager *options_mgr,
    MountPoint *mount_point,
    const std::string &fqrn)
    : TelemetryAggregator(statistics, send_rate_sec, mount_point, fqrn)
    , influx_extra_fields_("")
    , influx_extra_tags_("")
    , socket_fd_(-1)
    , res_(NULL) {
  int params = 0;

  if (options_mgr->GetValue("CVMFS_INFLUX_HOST", &influx_host_)) {
    if (influx_host_.size() > 1) {
      params++;
    } else {
      LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogWarn,
               "No value given for CVMFS_INFLUX_HOST");
    }
  }

  std::string opt;
  if (options_mgr->GetValue("CVMFS_INFLUX_PORT", &opt)) {
    influx_port_ = static_cast<int>(String2Int64(opt.c_str()));
    if (influx_port_ > 0 && influx_port_ < 65536) {
      params++;
    } else {
      LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogWarn,
               "Invalid value for CVMFS_INFLUX_PORT [%s]", opt.c_str());
    }
  }

  if (options_mgr->GetValue("CVMFS_INFLUX_METRIC_NAME", &influx_metric_name_)) {
    params++;
  }

  if (!options_mgr->GetValue("CVMFS_INFLUX_EXTRA_TAGS", &influx_extra_tags_)) {
    influx_extra_tags_ = "";
  }

  if (!options_mgr->GetValue("CVMFS_INFLUX_EXTRA_FIELDS",
                             &influx_extra_fields_)) {
    influx_extra_fields_ = "";
  }

  if (params == 3) {
    is_zombie_ = false;
    LogCvmfs(kLogTelemetry, kLogDebug,
             "Enabling influx metrics. "
             "Send to [%s:%d] metric [%s]. Extra tags"
             "[%s]. Extra fields [%s].",
             influx_host_.c_str(), influx_port_, influx_metric_name_.c_str(),
             influx_extra_tags_.c_str(), influx_extra_fields_.c_str());
    TelemetryReturn ret = OpenSocket();
    if (ret != kTelemetrySuccess) {
      is_zombie_ = true;
      LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogWarn,
               "Not enabling influx metrics. Error while open socket. %d", ret);
    }
  } else {
    is_zombie_ = true;
    LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogWarn,
             "Not enabling influx metrics. Not all mandatory variables set: "
             "CVMFS_INFLUX_METRIC_NAME, CVMFS_INFLUX_HOST, CVMFS_INFLUX_PORT");
  }
}

TelemetryAggregatorInflux::~TelemetryAggregatorInflux() {
  if (socket_fd_ >= 0) {
    close(socket_fd_);
  }
  if (res_) {
    freeaddrinfo(res_);
  }
}

std::string TelemetryAggregatorInflux::MakePayload() {
  // measurement and tags
  std::string ret = influx_metric_name_ + "_absolute,repo=" + fqrn_;

  if (influx_extra_tags_ != "") {
    ret += "," + influx_extra_tags_;
  }

  // fields
  ret += " ";
  bool add_token = false;
  for (std::map<std::string, int64_t>::iterator it = counters_.begin(),
                                                iEnd = counters_.end();
       it != iEnd; it++) {
    if (it->second != 0) {
      if (add_token) {
        ret += ",";
      }
      ret += it->first + "=" + StringifyInt(it->second);
      add_token = true;
    }
  }
  if (influx_extra_fields_ != "") {
    if (add_token) {
      ret += ",";
    }
    ret += influx_extra_fields_;
    add_token = true;
  }

  // timestamp
  if (add_token) {
    ret += " ";
  }
  ret += StringifyUint(timestamp_);

  return ret;
}

std::string TelemetryAggregatorInflux::MakeDeltaPayload() {
  // measurement and tags
  std::string ret = "" + influx_metric_name_ + "_delta,repo=" + fqrn_;

  if (influx_extra_tags_ != "") {
    ret += "," + influx_extra_tags_;
  }

  // fields
  ret += " ";
  bool add_token = false;
  for (std::map<std::string, int64_t>::iterator it = counters_.begin(),
                                                iEnd = counters_.end();
       it != iEnd; it++) {
    int64_t value = it->second;
    if (value != 0) {
      int64_t old_value;
      try {
        old_value = old_counters_.at(it->first);
      } catch (const std::out_of_range &ex) {
        old_value = 0;
      }
      if (add_token) {
        ret += ",";
      }
      ret += it->first + "=" + StringifyInt(value - old_value);
      add_token = true;
    }
  }

  // timestamp
  if (add_token) {
    ret += " ";
  }
  ret += StringifyUint(timestamp_);

  return ret;
}

TelemetryReturn TelemetryAggregatorInflux::OpenSocket() {
  const char *hostname = influx_host_.c_str();
  struct addrinfo hints;
  int err;

  memset(&hints, 0, sizeof(hints));
  hints.ai_family = AF_INET;
  hints.ai_socktype = SOCK_DGRAM;

  err = getaddrinfo(influx_host_.c_str(), NULL, &hints, &res_);
  if (err != 0 || res_ == NULL) {
    LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogErr,
             "Failed to resolve influx server [%s]. errno=%d", hostname, errno);
    return kTelemetryFailHostAddress;
  }

  socket_fd_ = socket(AF_INET, SOCK_DGRAM, 0);
  if (socket_fd_ < 0) {
    LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogErr, "Failed to open socket");
    freeaddrinfo(res_);
    return kTelemetryFailSocket;
  }

  return kTelemetrySuccess;
}

TelemetryReturn TelemetryAggregatorInflux::SendToInflux(
    const std::string &payload) {
  struct sockaddr_in *dest_addr = NULL;
  dest_addr = reinterpret_cast<sockaddr_in *>(res_->ai_addr);
  dest_addr->sin_port = htons(influx_port_);

  ssize_t num_bytes_sent = sendto(
      socket_fd_,
      payload.data(),
      payload.size(),
      0,
      reinterpret_cast<struct sockaddr *>(dest_addr),
      sizeof(struct sockaddr_in));

  if (num_bytes_sent < 0) {
    LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogErr,
             "Failed to send to influx. errno=%d", errno);
    return kTelemetryFailSend;
  } else if (static_cast<size_t>(num_bytes_sent) != payload.size()) {
    LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogErr,
             "Incomplete send. Bytes transferred: %zd. Bytes expected %lu",
             num_bytes_sent, payload.size());
    return kTelemetryFailSend;
  }

  LogCvmfs(kLogTelemetry, kLogDebug, "INFLUX: POSTING [%s]", payload.c_str());
  return kTelemetrySuccess;
}


void TelemetryAggregatorInflux::PushMetrics() {
  // create payload
  std::string payload = MakePayload();
  std::string delta_payload = "";
  if (old_counters_.size() > 0) {
    delta_payload = MakeDeltaPayload();
    payload = payload + "\n" + delta_payload;
  }
  payload += "\n";

  // send to influx
  SendToInflux(payload);

  // current counter is now old counter
  counters_.swap(old_counters_);
}

}  // namespace perf