<commit_before>/******************************************************************************
 *                    _   _____   __________                                  *
 *                   | | / / _ | / __/_  __/     Visibility                   *
 *                   | |/ / __ |_\ \  / /          Across                     *
 *                   |___/_/ |_/___/ /_/       Space and Time                 *
 *                                                                            *
 * This file is part of VAST. It is subject to the license terms in the       *
 * LICENSE file found in the top-level directory of this distribution and at  *
 * http://vast.io/license. No part of VAST, including this file, may be       *
 * copied, modified, propagated, or distributed except according to the terms *
 * contained in the LICENSE file.                                             *
 ******************************************************************************/

#include <cmath>
#include <iomanip>
#include <ios>
#include <locale>

#include "vast/logger.hpp"

#include <caf/all.hpp>

#include "vast/concept/printable/stream.hpp"
#include "vast/concept/printable/vast/filesystem.hpp"
#include "vast/error.hpp"

#include "vast/system/accountant.hpp"

#include "vast/detail/coding.hpp"
#include "vast/detail/fill_status_map.hpp"

namespace vast {
namespace system {

namespace {

using accountant_actor = accountant_type::stateful_base<accountant_state>;
constexpr std::chrono::seconds overview_delay(5);

void init(accountant_actor* self, const path& filename) {
  if (!exists(filename.parent())) {
    auto t = mkdir(filename.parent());
    if (!t) {
      VAST_ERROR(self, to_string(t.error()));
      self->quit(t.error());
      return;
    }
  }
  VAST_DEBUG(self, "opens log file:", filename.trim(-4));
  auto& file = self->state.file;
  file.open(filename.str());
  if (!file.is_open()) {
    VAST_ERROR(self, "failed to open file:", filename);
    auto e = make_error(ec::filesystem_error, "failed to open file:", filename);
    self->quit(e);
    return;
  }
  file << "host\tpid\taid\tkey\tvalue\n";
  if (!file)
    self->quit(make_error(ec::filesystem_error));
  VAST_DEBUG(self, "kicks off flush loop");
  self->send(self, flush_atom::value);
#if VAST_LOG_LEVEL >= CAF_LOG_LEVEL_INFO
  self->delayed_send(self, overview_delay, telemetry_atom::value);
#endif
}

template <class T>
void record(accountant_actor* self, const std::string& key, T x) {
  using namespace std::chrono;
  auto aid = self->current_sender()->id();
  auto node = self->current_sender()->node();
  auto& st = self->state;
  for (auto byte : node.host_id())
    st.file << static_cast<int>(byte);
  st.file << std::dec << '\t' << node.process_id() << '\t' << aid << '\t'
          << st.actor_map[aid] << '\t' << key << '\t' << std::setprecision(6)
          << x << '\n';
  // Flush after at most 10 seconds.
  if (!st.flush_pending) {
    st.flush_pending = true;
    self->delayed_send(self, 10s, flush_atom::value);
  }
}

void record(accountant_actor* self, const std::string& key, timespan x) {
  using namespace std::chrono;
  auto us = duration_cast<microseconds>(x).count();
  record(self, key, us);
}

void record(accountant_actor* self, const std::string& key, timestamp x) {
  using namespace std::chrono;
  record(self, key, x.time_since_epoch());
}

// Calculate rate in seconds resolution from nanosecond duration.
double calc_rate(const measurement& m) {
  if (m.duration.count() > 0)
    return m.events * 1'000'000'000 / m.duration.count();
  else
    return std::numeric_limits<double>::quiet_NaN();
}

} // namespace <anonymous>

void accountant_state::command_line_heartbeat() {
  auto logger = caf::logger::current_logger();
  if (logger && logger->verbosity() >= CAF_LOG_LEVEL_INFO
      && accumulator.node.events > 0) {
    std::ostringstream oss;
    oss.imbue(std::locale(""));
    auto node_rate = std::round(calc_rate(accumulator.node));
    oss << "ingested " << accumulator.node.events << " events"
        << " at a rate of " << node_rate << " events/sec";
    VAST_INFO_ANON(oss.str());
  }
  accumulator = {};
}

accountant_type::behavior_type accountant(accountant_actor* self,
                                          const path& filename) {
  using namespace std::chrono;
  init(self, filename);
  self->set_exit_handler(
    [=](const caf::exit_msg& msg) {
      self->state.file.flush();
      self->quit(msg.reason);
    }
  );
  self->set_down_handler(
    [=](const caf::down_msg& msg) {
      VAST_DEBUG(self, "received DOWN from", msg.source);
      self->state.actor_map.erase(msg.source.id());
    }
  );
  return {[=](announce_atom, const std::string& name) {
            self->state.actor_map[self->current_sender()->id()] = name;
            self->monitor(self->current_sender());
          },
          [=](const std::string& key, const std::string& value) {
            VAST_TRACE(self, "received", key, "from", self->current_sender());<commit_msg>Use a single contiguous string in ostream output<commit_after>/******************************************************************************
 *                    _   _____   __________                                  *
 *                   | | / / _ | / __/_  __/     Visibility                   *
 *                   | |/ / __ |_\ \  / /          Across                     *
 *                   |___/_/ |_/___/ /_/       Space and Time                 *
 *                                                                            *
 * This file is part of VAST. It is subject to the license terms in the       *
 * LICENSE file found in the top-level directory of this distribution and at  *
 * http://vast.io/license. No part of VAST, including this file, may be       *
 * copied, modified, propagated, or distributed except according to the terms *
 * contained in the LICENSE file.                                             *
 ******************************************************************************/

#include <cmath>
#include <iomanip>
#include <ios>
#include <locale>

#include "vast/logger.hpp"

#include <caf/all.hpp>

#include "vast/concept/printable/stream.hpp"
#include "vast/concept/printable/vast/filesystem.hpp"
#include "vast/error.hpp"

#include "vast/system/accountant.hpp"

#include "vast/detail/coding.hpp"
#include "vast/detail/fill_status_map.hpp"

namespace vast {
namespace system {

namespace {

using accountant_actor = accountant_type::stateful_base<accountant_state>;
constexpr std::chrono::seconds overview_delay(5);

void init(accountant_actor* self, const path& filename) {
  if (!exists(filename.parent())) {
    auto t = mkdir(filename.parent());
    if (!t) {
      VAST_ERROR(self, to_string(t.error()));
      self->quit(t.error());
      return;
    }
  }
  VAST_DEBUG(self, "opens log file:", filename.trim(-4));
  auto& file = self->state.file;
  file.open(filename.str());
  if (!file.is_open()) {
    VAST_ERROR(self, "failed to open file:", filename);
    auto e = make_error(ec::filesystem_error, "failed to open file:", filename);
    self->quit(e);
    return;
  }
  file << "host\tpid\taid\tkey\tvalue\n";
  if (!file)
    self->quit(make_error(ec::filesystem_error));
  VAST_DEBUG(self, "kicks off flush loop");
  self->send(self, flush_atom::value);
#if VAST_LOG_LEVEL >= CAF_LOG_LEVEL_INFO
  self->delayed_send(self, overview_delay, telemetry_atom::value);
#endif
}

template <class T>
void record(accountant_actor* self, const std::string& key, T x) {
  using namespace std::chrono;
  auto aid = self->current_sender()->id();
  auto node = self->current_sender()->node();
  auto& st = self->state;
  for (auto byte : node.host_id())
    st.file << static_cast<int>(byte);
  st.file << std::dec << '\t' << node.process_id() << '\t' << aid << '\t'
          << st.actor_map[aid] << '\t' << key << '\t' << std::setprecision(6)
          << x << '\n';
  // Flush after at most 10 seconds.
  if (!st.flush_pending) {
    st.flush_pending = true;
    self->delayed_send(self, 10s, flush_atom::value);
  }
}

void record(accountant_actor* self, const std::string& key, timespan x) {
  using namespace std::chrono;
  auto us = duration_cast<microseconds>(x).count();
  record(self, key, us);
}

void record(accountant_actor* self, const std::string& key, timestamp x) {
  using namespace std::chrono;
  record(self, key, x.time_since_epoch());
}

// Calculate rate in seconds resolution from nanosecond duration.
double calc_rate(const measurement& m) {
  if (m.duration.count() > 0)
    return m.events * 1'000'000'000 / m.duration.count();
  else
    return std::numeric_limits<double>::quiet_NaN();
}

} // namespace <anonymous>

void accountant_state::command_line_heartbeat() {
  auto logger = caf::logger::current_logger();
  if (logger && logger->verbosity() >= CAF_LOG_LEVEL_INFO
      && accumulator.node.events > 0) {
    std::ostringstream oss;
    oss.imbue(std::locale(""));
    auto node_rate = std::round(calc_rate(accumulator.node));
    oss << "ingested " << accumulator.node.events << " events at a rate of "
        << node_rate << " events/sec";
    VAST_INFO_ANON(oss.str());
  }
  accumulator = {};
}

accountant_type::behavior_type accountant(accountant_actor* self,
                                          const path& filename) {
  using namespace std::chrono;
  init(self, filename);
  self->set_exit_handler(
    [=](const caf::exit_msg& msg) {
      self->state.file.flush();
      self->quit(msg.reason);
    }
  );
  self->set_down_handler(
    [=](const caf::down_msg& msg) {
      VAST_DEBUG(self, "received DOWN from", msg.source);
      self->state.actor_map.erase(msg.source.id());
    }
  );
  return {[=](announce_atom, const std::string& name) {
            self->state.actor_map[self->current_sender()->id()] = name;
            self->monitor(self->current_sender());
          },
          [=](const std::string& key, const std::string& value) {
            VAST_TRACE(self, "received", key, "from", self->current_sender());
{'fraction_non_alphanumeric': 0.17922363037620995, 'fraction_numerical': 0.003792036722881948, 'mean_word_length': 2.8635312259059367, 'pattern_counts': {'":': 0, '<': 107, '<?xml version=': 0, '>': 81, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6417909', 'n_tokens_mistral': 3182, 'n_tokens_neox': 3063, 'n_words': 690}