StatisticsSenderService.cc

Go to the documentation of this file.

#include "Utilities/StorageFactory/interface/StatisticsSenderService.h"
#include "Utilities/StorageFactory/interface/StorageAccount.h"
#include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
#include "FWCore/Catalog/interface/SiteLocalConfig.h"
#include "FWCore/ServiceRegistry/interface/Service.h"
#include "FWCore/Utilities/src/Guid.h"

#include <string>
#include <cmath>

#include <unistd.h>
#include <fcntl.h>

#include <openssl/x509.h>
#include <openssl/pem.h>

#define UPDATE_STATISTIC(x) \
    m_ ## x = x;

#define UPDATE_AND_OUTPUT_STATISTIC(x) \
    os << "\"" #x "\":" << (x-m_ ## x) << ", "; \
    UPDATE_STATISTIC(x)

// Simple hack to define HOST_NAME_MAX on Mac.
// Allows arrays to be statically allocated
#ifndef HOST_NAME_MAX
#define HOST_NAME_MAX 128
#endif

#define JOB_UNIQUE_ID_ENV "CRAB_UNIQUE_JOB_ID"
#define JOB_UNIQUE_ID_ENV_V2 "DashboardJobId"

using namespace edm::storage;

StatisticsSenderService::FileStatistics::FileStatistics() :
  m_read_single_operations(0),
  m_read_single_bytes(0),
  m_read_single_square(0),
  m_read_vector_operations(0),
  m_read_vector_bytes(0),
  m_read_vector_square(0),
  m_read_vector_count_sum(0),
  m_read_vector_count_square(0),
  m_start_time(time(NULL))
{}

void
StatisticsSenderService::FileStatistics::fillUDP(std::ostringstream &os) {
  const StorageAccount::StorageStats &stats = StorageAccount::summary();
  ssize_t read_single_operations = 0;
  ssize_t read_single_bytes = 0;
  ssize_t read_single_square = 0;
  ssize_t read_vector_operations = 0;
  ssize_t read_vector_bytes = 0;
  ssize_t read_vector_square = 0;
  ssize_t read_vector_count_sum = 0;
  ssize_t read_vector_count_square = 0;
  auto token = StorageAccount::tokenForStorageClassName("tstoragefile");
  for (StorageAccount::StorageStats::const_iterator i = stats.begin (); i != stats.end(); ++i) {
    if (i->first == token.value()) {
      continue;
    }
    for (StorageAccount::OperationStats::const_iterator j = i->second.begin(); j != i->second.end(); ++j) {
      if (j->first == static_cast<int>(StorageAccount::Operation::readv)) {
        read_vector_operations += j->second.attempts;
        read_vector_bytes += j->second.amount;
        read_vector_count_square += j->second.vector_square;
        read_vector_square += j->second.amount_square;
        read_vector_count_sum += j->second.vector_count;
      } else if (j->first == static_cast<int>(StorageAccount::Operation::read)) {
        read_single_operations += j->second.attempts;
        read_single_bytes += j->second.amount;
        read_single_square += j->second.amount_square;
      }
    }
  }
  int64_t single_op_count = read_single_operations - m_read_single_operations;
  if (single_op_count > 0) {
    double single_sum = read_single_bytes-m_read_single_bytes;
    double single_average = single_sum/static_cast<double>(single_op_count);
    os << "\"read_single_sigma\":" << sqrt((static_cast<double>(read_single_square-m_read_single_square) - single_average*single_average*single_op_count)/static_cast<double>(single_op_count)) << ", ";
    os << "\"read_single_average\":" << single_average << ", ";
  }
  m_read_single_square = read_single_square;
  int64_t vector_op_count = read_vector_operations - m_read_vector_operations;
  if (vector_op_count > 0) {
    double vector_average = static_cast<double>(read_vector_bytes-m_read_vector_bytes)/static_cast<double>(vector_op_count);
    os << "\"read_vector_average\":" << vector_average << ", ";
    os << "\"read_vector_sigma\":" << sqrt((static_cast<double>(read_vector_square-m_read_vector_square) - vector_average*vector_average*vector_op_count)/static_cast<double>(vector_op_count)) << ", ";
    double vector_count_average = static_cast<double>(read_vector_count_sum-m_read_vector_count_sum)/static_cast<double>(vector_op_count);
    os << "\"read_vector_count_average\":" << vector_count_average << ", ";
    os << "\"read_vector_count_sigma\":" << sqrt((static_cast<double>(read_vector_count_square-m_read_vector_count_square) - vector_count_average*vector_count_average*vector_op_count)/static_cast<double>(vector_op_count)) << ", ";
  }
  m_read_vector_square = read_vector_square;
  m_read_vector_count_square = read_vector_count_square;
  m_read_vector_count_sum = read_vector_count_sum;

  os << "\"read_bytes\":" << (read_vector_bytes + read_single_bytes - m_read_vector_bytes - m_read_single_bytes) << ", ";
  os << "\"read_bytes_at_close\":" << (read_vector_bytes + read_single_bytes - m_read_vector_bytes - m_read_single_bytes) << ", ";

  // See top of file for macros; not complex, just avoiding copy/paste
  UPDATE_AND_OUTPUT_STATISTIC(read_single_operations)
  UPDATE_AND_OUTPUT_STATISTIC(read_single_bytes)
  UPDATE_AND_OUTPUT_STATISTIC(read_vector_operations)
  UPDATE_AND_OUTPUT_STATISTIC(read_vector_bytes)

  os << "\"start_time\":" << m_start_time << ", ";
  m_start_time = time(NULL);
  // NOTE: last entry doesn't have the trailing comma.
  os << "\"end_time\":" << m_start_time;
}

StatisticsSenderService::StatisticsSenderService(edm::ParameterSet const& /*pset*/, edm::ActivityRegistry& ar) :
  m_clienthost("unknown"),
  m_clientdomain("unknown"),
  m_serverhost("unknown"),
  m_serverdomain("unknown"),
  m_filelfn("unknown"),
  m_filestats(),
  m_guid(Guid().toString()),
  m_counter(0),
  m_size(-1),
  m_userdn("unknown")
{
  determineHostnames();
  ar.watchPreCloseFile(this, &StatisticsSenderService::filePreCloseEvent);
  if (!getX509Subject(m_userdn)) {
    m_userdn = "unknown";
  }
}

const char *
StatisticsSenderService::getJobID() {
  const char * id = getenv(JOB_UNIQUE_ID_ENV);
  // Dashboard developers requested that we migrate to this environment variable.
  return id ? id : getenv(JOB_UNIQUE_ID_ENV_V2);
}

void
StatisticsSenderService::setCurrentServer(const std::string &servername) {
  size_t dot_pos = servername.find(".");
  std::string serverhost;
  std::string serverdomain;
  if (dot_pos == std::string::npos) {
    serverhost = servername.substr(0, servername.find(":"));
    serverdomain = "unknown";
  } else {
    serverhost = servername.substr(0, dot_pos);
    serverdomain = servername.substr(dot_pos+1, servername.find(":")-dot_pos-1);
    if (serverdomain.empty()) {
      serverdomain = "unknown";
    }
  }
  {
    std::lock_guard<std::mutex> sentry(m_servermutex);
    m_serverhost = std::move(serverhost);
    m_serverdomain = std::move(serverdomain);
  }
}

void
StatisticsSenderService::setSize(size_t size) {
  m_size = size;
}

void
StatisticsSenderService::filePreCloseEvent(std::string const& lfn, bool usedFallback) {
  m_filelfn = lfn;

  edm::Service<edm::SiteLocalConfig> pSLC;
  if (!pSLC.isAvailable()) {
    return;
  }

  const struct addrinfo * addresses = pSLC->statisticsDestination();
  if (!addresses) {
    return;
  }

  std::set<std::string> const * info = pSLC->statisticsInfo();
  if (info && info->size() && (m_userdn != "unknown") && (
      (info->find("dn") == info->end()) ||
      (info->find("nodn") != info->end()))
     )
  {
    m_userdn = "not reported";
  }

  std::string results;
  fillUDP(pSLC->siteName(), usedFallback, results);

  for (const struct addrinfo *address = addresses; address != nullptr; address = address->ai_next) {
    int sock = socket(address->ai_family, address->ai_socktype, address->ai_protocol);
    if (sock < 0) {
      continue;
    }
    auto close_del = [](int* iSocket) { close(*iSocket); };
    std::unique_ptr<int,decltype(close_del)> guard(&sock, close_del);
    if (sendto(sock, results.c_str(), results.size(), 0, address->ai_addr, address->ai_addrlen) >= 0) {
      break; 
    }
  }

  m_counter++;
}

void
StatisticsSenderService::determineHostnames(void) {
  char tmpName[HOST_NAME_MAX];
  if (gethostname(tmpName, HOST_NAME_MAX) != 0) {
    // Sigh, no way to log errors from here.
    m_clienthost = "unknown";
  } else {
    m_clienthost = tmpName;
  }
  size_t dot_pos = m_clienthost.find(".");
  if (dot_pos == std::string::npos) {
    m_clientdomain = "unknown";
  } else {
    m_clientdomain = m_clienthost.substr(dot_pos+1, m_clienthost.size()-dot_pos-1);
    m_clienthost = m_clienthost.substr(0, dot_pos);
  }
}

void
StatisticsSenderService::fillUDP(const std::string& siteName, bool usedFallback, std::string &udpinfo) {
  std::ostringstream os;

  // Header - same for all IO accesses
  os << "{";
  if (!siteName.empty()) {
    os << "\"site_name\":\"" << siteName << "\", ";
  }
  if (usedFallback) {
    os << "\"fallback\": true, ";
  }
  std::string serverhost;
  std::string serverdomain;
  {
    std::lock_guard<std::mutex> sentry(m_servermutex);
    serverhost = m_serverhost;
    serverdomain = m_serverdomain;
  }
  
  os << "\"user_dn\":\"" << m_userdn << "\", ";
  os << "\"client_host\":\"" << m_clienthost << "\", ";
  os << "\"client_domain\":\"" << m_clientdomain << "\", ";
  os << "\"server_host\":\"" << serverhost << "\", ";
  os << "\"server_domain\":\"" << serverdomain << "\", ";
  os << "\"unique_id\":\"" << m_guid << "-" << m_counter << "\", ";
  os << "\"file_lfn\":\"" << m_filelfn << "\", ";
  // Dashboard devs requested that we send out no app_info if a job ID
  // is not present in the environment.
  const char * jobId = getJobID();
  if (jobId) {
    os << "\"app_info\":\"" << jobId << "\", ";
  }

  if (m_size >= 0) {
    os << "\"file_size\":" << m_size << ", ";
  }

  m_filestats.fillUDP(os);

  os << "}";
  udpinfo = os.str();
}

/*
 * Pull the X509 user subject from the environment.
 * Based on initial code from the Frontier client:
 *   http://cdcvs.fnal.gov/cgi-bin/public-cvs/cvsweb-public.cgi/~checkout~/frontier/client/frontier.c?rev=1.57&content-type=text/plain
 * This was further extended by walking up the returned chain similar to the Globus function
 *   globus_gsi_cert_utils-6.6/library/globus_gsi_cert_utils.c:globus_gsi_cert_utils_get_eec
 *   globus_gsi_credential-3.5/library/globus_gsi_credential.c:globus_gsi_cred_read_proxy_bio
 */

/* 
 * Given a stack of x509 proxies, take a guess at the EEC.
 * Assumes the proxies are in reverse sorted order and looks for the first
 * proxy which is not a substring of the prior proxy.
 * THIS DOES NOT VERIFY THE RESULTS, and is a best-effort GUESS.
 * Again, DO NOT REUSE THIS CODE THINKING IT VERIFIES THE CHAIN!
 */
static X509 * findEEC(STACK_OF(X509) * certstack) {
  int depth = sk_X509_num(certstack);
  if (depth == 0) {
    return nullptr;
  }
  int idx = depth-1;
  char *priorsubject = nullptr;
  char *subject = nullptr;
  X509 *x509cert = sk_X509_value(certstack, idx);
  for (; x509cert && idx>0; idx--) {
    subject = X509_NAME_oneline(X509_get_subject_name(x509cert),0,0);
    if (subject && priorsubject && (strncmp(subject, priorsubject, strlen(subject)) != 0)) {
      break;
    }
    x509cert = sk_X509_value(certstack, idx);
    if (subject) {
      OPENSSL_free(subject);
      subject = nullptr;
    }
  }
  if (subject) {
    OPENSSL_free(subject);
    subject = nullptr;
  }
  return x509cert;
}

static bool
getX509SubjectFromFile(const std::string &filename, std::string &result) {
  BIO *biof = nullptr;
  STACK_OF(X509) *certs = nullptr;
  char *subject = nullptr;
  unsigned char *data = nullptr;
  char *header = nullptr;
  char *name = nullptr;
  long len = 0U;

  if((biof = BIO_new_file(filename.c_str(), "r")))  {

    certs = sk_X509_new_null();
    bool encountered_error = false;
    while ((!encountered_error) && (!BIO_eof(biof)) && PEM_read_bio(biof, &name, &header, &data, &len)) {
      if (strcmp(name, PEM_STRING_X509) == 0 || strcmp(name, PEM_STRING_X509_OLD) == 0) {
        X509 * tmp_cert = nullptr;
        // See WARNINGS section in http://www.openssl.org/docs/crypto/d2i_X509.html
        // Without this cmsRun crashes on a mac with a valid grid proxy.
        const unsigned char *p;
        p=data;
        tmp_cert = d2i_X509(&tmp_cert, &p, len);
        if (tmp_cert) {
          sk_X509_push(certs, tmp_cert);
        } else {
          encountered_error = true;
        }
      } // Note we ignore any proxy key in the file.
      if (data) { OPENSSL_free(data); data = nullptr;}
      if (header) { OPENSSL_free(header); header = nullptr;}
      if (name) { OPENSSL_free(name); name = nullptr;}
    }
    X509 *x509cert = nullptr;
    if (!encountered_error && sk_X509_num(certs)) {
      x509cert = findEEC(certs);
    }
    if (x509cert) {
      subject = X509_NAME_oneline(X509_get_subject_name(x509cert),0,0);
    }
    // Note we do not free x509cert directly, as it's still owned by the certs stack.
    if (certs) {
      sk_X509_pop_free(certs, X509_free);
      x509cert = nullptr;
    }
    BIO_free(biof);
    if (subject) {
      result = subject;
      OPENSSL_free(subject);
     return true;
    }
  }
  return false;
}

bool
StatisticsSenderService::getX509Subject(std::string &result) {
  char *filename = getenv("X509_USER_PROXY");
  if (filename && getX509SubjectFromFile(filename, result)) {
    return true;
  }
  std::stringstream ss;
  ss << "/tmp/x509up_u" << geteuid();
  return getX509SubjectFromFile(ss.str(), result);
}