analyzer/report_master/report_scheduler.cc - cobalt - Git at Google

 // Copyright 2017 The Fuchsia Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 #include "analyzer/report_master/report_scheduler.h"

 #include <iomanip>
 #include <string>
 #include <utility>

 #include "config/encodings.pb.h"
 #include "config/metrics.pb.h"
 #include "gflags/gflags.h"
 #include "glog/logging.h"
 #include "util/clock.h"
 #include "util/datetime_util.h"
 #include "util/log_based_metrics.h"

 namespace cobalt {
 namespace analyzer {

 using config::ReportRegistry;
 using util::SystemClock;
 using util::TimeToDayIndex;

 DEFINE_uint32(daily_report_makeup_days, 30,
               "The number of days in the past that the ReportMaster should "
               "look to find missed scheduled reports to make up. Must be less "
               "than 100 or we will CHECK fail.");

 // Stackdriver metric constants
 namespace {
 const char kStartReportNowFailure[] = "report-scheduler-start-report-failure";
 const char kProcessReportFailure[] =
     "report-scheduler-process-report-failure-for-";

 const std::string ProcessReportFailMetricName(const std::string& suffix) {
   std::ostringstream stream;
   stream << kProcessReportFailure;
   stream << suffix;
   return stream.str();
 }

 }  // namespace

 namespace {
 // Returns a human-readable respresentation of the report config ID.
 // Used in forming error messages.
 // TODO(rudominer) This function has been copied multiple times throughout the
 // code. We should centralize it in a utility.
 std::string IdString(const ReportConfig& report_config) {
   std::ostringstream stream;
   stream << "(" << report_config.customer_id() << ","
          << report_config.project_id() << "," << report_config.id() << ")";
   return stream.str();
 }

 // Builds a string of the form YYYYMMDD to represent the date corresponding to
 // the given day_index. This is the standard format used at Google for the
 // suffix of a file name for files containing data for dated tables.
 std::string DateSuffix(uint32_t day_index) {
   util::CalendarDate cd = util::DayIndexToCalendarDate(day_index);
   std::ostringstream stream;
   stream << std::setfill('0') << std::setw(4) << cd.year << std::setw(2)
          << cd.month << std::setw(2) << cd.day_of_month;
   return stream.str();
 }

 }  // namespace

 ReportScheduler::ReportScheduler(
     std::shared_ptr<config::AnalyzerConfigManager> config_manager,
     std::shared_ptr<store::ReportStore> report_store,
     std::shared_ptr<ReportStarterInterface> report_starter,
     std::chrono::milliseconds sleep_interval)
     : clock_(new SystemClock()),
       config_manager_(config_manager),
       report_starter_(report_starter),
       report_history_(new ReportHistoryCache(
           CurrentDayIndex() - FLAGS_daily_report_makeup_days, report_store)),
       sleep_interval_(sleep_interval),
       shut_down_(false) {
   CHECK_LT(FLAGS_daily_report_makeup_days, 100);
 }

 ReportScheduler::~ReportScheduler() {
   {
     std::lock_guard<std::mutex> lock(mutex_);
     shut_down_ = true;
     if (!scheduler_thread_.joinable()) {
       return;
     }
   }
   scheduler_thread_.join();
 }

 void ReportScheduler::Start() {
   std::lock_guard<std::mutex> lock(mutex_);
   std::thread t([this] { this->Run(); });
   scheduler_thread_ = std::move(t);
 }

 void ReportScheduler::Run() {
   while (!shut_down_) {
     Sleep();
     if (shut_down_) {
       return;
     }
     ProcessReports();
   }
 }

 void ReportScheduler::Sleep() {
   // Note: We invoke the real system clock here, not clock_->now().
   // This is because even in a test we want to use the real system clock to
   // compute wakeup_time because std::condition_varaible::wait_until() always
   // uses the real system clock. A test is able to control the sleep time by
   // setting the value of sleep_interval_.
   auto wakeup_time = std::chrono::system_clock::now() + sleep_interval_;
   VLOG(3) << "ReportScheduler sleeping for " << sleep_interval_.count() << "ms";
   std::unique_lock<std::mutex> lock(mutex_);
   // Sleep until wakeup_time or shut_down_ = true.
   shut_down_notifier_.wait_until(lock, wakeup_time,
                                  [this] { return this->shut_down_.load(); });
 }

 void ReportScheduler::ProcessReports() {
   static const int kTimeoutSeconds = 60;
   config_manager_->Update(kTimeoutSeconds);
   uint32_t current_day_index = CurrentDayIndex();
   std::shared_ptr<config::ReportRegistry> report_registry =
       config_manager_->GetCurrent()->report_registry();
   for (const ReportConfig& report_config : *report_registry) {
     if (shut_down_) {
       return;
     }
     ProcessOneReport(report_config, current_day_index);
   }
 }

 void ReportScheduler::ProcessOneReport(const ReportConfig& report_config,
                                        uint32_t current_day_index) {
   LOG(INFO) << "ReportScheduler processing report_config "
             << IdString(report_config);
   if (!report_config.has_scheduling()) {
     LOG(INFO) << "Skpping report_config " << IdString(report_config)
               << " because it has no SchedulingConfig.";
     return;
   }
   switch (report_config.scheduling().aggregation_epoch_type()) {
     case DAY:
       ProcessDailyReport(report_config, current_day_index);
       return;

     case WEEK:
       ProcessWeeklyReport(report_config, current_day_index);
       return;

     case MONTH:
       ProcessMonthlyReport(report_config, current_day_index);
       return;

     default: {
       LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("one"))
           << "Unrecognized aggregatoin_epoch_type: "
           << report_config.scheduling().aggregation_epoch_type()
           << "In ReportConfig " << IdString(report_config);
       return;
     }
   }
 }

 void ReportScheduler::ProcessDailyReport(const ReportConfig& report_config,
                                          uint32_t current_day_index) {
   // Look back a number of days equal to the maximum of daily_report_makeup_days
   // and report_finalization_days.
   auto scheduling = report_config.scheduling();
   if (report_config.scheduling().report_finalization_days() > 20) {
     LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("daily"))
         << "Invalid ReportConfig: " << IdString(report_config)
         << " report_finalization_days too large: "
         << report_config.scheduling().report_finalization_days();
     return;
   }

   uint32_t finalization_days = scheduling.report_finalization_days();
   uint32_t lookback_days = FLAGS_daily_report_makeup_days >= finalization_days
                                ? FLAGS_daily_report_makeup_days
                                : finalization_days;
   uint32_t period_start =
       (current_day_index >= lookback_days ? (current_day_index - lookback_days)
                                           : 0u);
   VLOG(4) << "ReportScheduler considering days in the interval ["
           << period_start << ", " << current_day_index << "]";
   for (uint32_t day_index = period_start; day_index <= current_day_index;
        day_index++) {
     if (shut_down_) {
       return;
     }
     if (ShouldStartDailyReportNow(report_config, day_index,
                                   current_day_index)) {
       StartReportNow(report_config, day_index, day_index);
     } else {
       VLOG(4) << "ShouldStartDailyReportNow() returned false for report_config "
               << IdString(report_config) << " day_index=" << day_index
               << " current_day_index=" << current_day_index;
     }
   }
 }

 void ReportScheduler::ProcessWeeklyReport(const ReportConfig& report_config,
                                           uint32_t current_day_index) {
   LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("weekly"))
       << "Scheduling of weekly reports is not yet implemented. ReportConfig: "
       << IdString(report_config);
   return;
 }

 void ReportScheduler::ProcessMonthlyReport(const ReportConfig& report_config,
                                            uint32_t current_day_index) {
   LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("monthly"))
       << "Scheduling of monthly reports is not yet implemented. ReportConfig: "
       << IdString(report_config);
   return;
 }

 bool ReportScheduler::ShouldStartDailyReportNow(
     const ReportConfig& report_config, uint32_t day_index,
     uint32_t current_day_index) {
   if (day_index > current_day_index) {
     LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("daily"))
         << "Unexpected condition: " << day_index
         << " = day_index > current_day_index = " << current_day_index
         << " for ReportConfig " << IdString(report_config);
     return false;
   }
   if (day_index > current_day_index -
                       report_config.scheduling().report_finalization_days()) {
     // We want to generate the report repeatedly during the report finalization
     // period, but we don't want to start it again now if we previously started
     // it and that hasn't completed.
     return !report_history_->InProgress(report_config, day_index, day_index);
   }
   // After the report finalization period we only want to run the report once.
   // If it was ever successfully completed don't run it again. Also if we
   // previously started the report and that attempt hasn't finished yet,
   // don't start it again.
   return !(report_history_->CompletedSuccessfullyOrInProgress(
       report_config, day_index, day_index));
 }

 void ReportScheduler::StartReportNow(const ReportConfig& report_config,
                                      uint32_t first_day_index,
                                      uint32_t last_day_index) {
   const std::string export_name =
       ReportExportName(report_config, first_day_index, last_day_index);
   // We want to store the rows of the report in the report store in all cases
   // except when it is a RAW_DUMP report.
   bool in_store = (report_config.report_type() != RAW_DUMP);
   ReportId report_id;
   LOG(INFO) << "ReportScheduler starting report " << IdString(report_config)
             << " [" << first_day_index << ", " << last_day_index << "]";
   auto status = report_starter_->StartReport(report_config, first_day_index,
                                              last_day_index, export_name,
                                              in_store, &report_id);
   if (!status.ok()) {
     LOG_STACKDRIVER_COUNT_METRIC(ERROR, kStartReportNowFailure)
         << "ReportScheduler was unable to start a report for ReportConfig "
         << IdString(report_config) << " first_day_index=" << first_day_index
         << " last_day_index=" << last_day_index
         << " error code=" << status.error_code()
         << " error message=" << status.error_message();
     return;
   }
   report_history_->RecordStart(report_config, first_day_index, last_day_index,
                                report_id);
 }

 std::string ReportScheduler::ReportExportName(const ReportConfig& report_config,
                                               uint32_t first_day_index,
                                               uint32_t last_day_index) {
   std::ostringstream stream;
   stream << "report_" << report_config.customer_id() << "_"
          << report_config.project_id() << "_" << report_config.id() << "_"
          << DateSuffix(first_day_index);
   if (last_day_index != first_day_index) {
     stream << "_" << DateSuffix(last_day_index);
   }
   return stream.str();
 }

 uint32_t ReportScheduler::CurrentDayIndex() {
   CHECK(clock_);
   std::time_t current_time =
       std::chrono::system_clock::to_time_t(clock_->now());
   return TimeToDayIndex(current_time, Metric::UTC);
 }

 ReportStarter::ReportStarter(ReportMasterService* report_master_service)
     : report_master_service_(report_master_service) {}

 grpc::Status ReportStarter::StartReport(const ReportConfig& report_config,
                                         uint32_t first_day_index,
                                         uint32_t last_day_index,
                                         const std::string& export_name,
                                         bool in_store,
                                         ReportId* report_id_out) {
   StartReportRequest start_request;
   start_request.set_customer_id(report_config.customer_id());
   start_request.set_project_id(report_config.project_id());
   start_request.set_report_config_id(report_config.id());
   start_request.set_first_day_index(first_day_index);
   start_request.set_last_day_index(last_day_index);
   StartReportResponse response_not_used;
   // This is not a one-off report generation. Rather it is scheduled.
   bool one_off = false;
   return report_master_service_->StartReportNoAuth(
       &start_request, one_off, export_name, in_store, report_id_out,
       &response_not_used);
 }

 }  // namespace analyzer
 }  // namespace cobalt
	// Copyright 2017 The Fuchsia Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	#include "analyzer/report_master/report_scheduler.h"

	#include <iomanip>
	#include <string>
	#include <utility>

	#include "config/encodings.pb.h"
	#include "config/metrics.pb.h"
	#include "gflags/gflags.h"
	#include "glog/logging.h"
	#include "util/clock.h"
	#include "util/datetime_util.h"
	#include "util/log_based_metrics.h"

	namespace cobalt {
	namespace analyzer {

	using config::ReportRegistry;
	using util::SystemClock;
	using util::TimeToDayIndex;

	DEFINE_uint32(daily_report_makeup_days, 30,
	"The number of days in the past that the ReportMaster should "
	"look to find missed scheduled reports to make up. Must be less "
	"than 100 or we will CHECK fail.");

	// Stackdriver metric constants
	namespace {
	const char kStartReportNowFailure[] = "report-scheduler-start-report-failure";
	const char kProcessReportFailure[] =
	"report-scheduler-process-report-failure-for-";

	const std::string ProcessReportFailMetricName(const std::string& suffix) {
	std::ostringstream stream;
	stream << kProcessReportFailure;
	stream << suffix;
	return stream.str();
	}

	} // namespace

	namespace {
	// Returns a human-readable respresentation of the report config ID.
	// Used in forming error messages.
	// TODO(rudominer) This function has been copied multiple times throughout the
	// code. We should centralize it in a utility.
	std::string IdString(const ReportConfig& report_config) {
	std::ostringstream stream;
	stream << "(" << report_config.customer_id() << ","
	<< report_config.project_id() << "," << report_config.id() << ")";
	return stream.str();
	}

	// Builds a string of the form YYYYMMDD to represent the date corresponding to
	// the given day_index. This is the standard format used at Google for the
	// suffix of a file name for files containing data for dated tables.
	std::string DateSuffix(uint32_t day_index) {
	util::CalendarDate cd = util::DayIndexToCalendarDate(day_index);
	std::ostringstream stream;
	stream << std::setfill('0') << std::setw(4) << cd.year << std::setw(2)
	<< cd.month << std::setw(2) << cd.day_of_month;
	return stream.str();
	}

	} // namespace

	ReportScheduler::ReportScheduler(
	std::shared_ptr<config::AnalyzerConfigManager> config_manager,
	std::shared_ptr<store::ReportStore> report_store,
	std::shared_ptr<ReportStarterInterface> report_starter,
	std::chrono::milliseconds sleep_interval)
	: clock_(new SystemClock()),
	config_manager_(config_manager),
	report_starter_(report_starter),
	report_history_(new ReportHistoryCache(
	CurrentDayIndex() - FLAGS_daily_report_makeup_days, report_store)),
	sleep_interval_(sleep_interval),
	shut_down_(false) {
	CHECK_LT(FLAGS_daily_report_makeup_days, 100);
	}

	ReportScheduler::~ReportScheduler() {
	{
	std::lock_guard<std::mutex> lock(mutex_);
	shut_down_ = true;
	if (!scheduler_thread_.joinable()) {
	return;
	}
	}
	scheduler_thread_.join();
	}

	void ReportScheduler::Start() {
	std::lock_guard<std::mutex> lock(mutex_);
	std::thread t([this] { this->Run(); });
	scheduler_thread_ = std::move(t);
	}

	void ReportScheduler::Run() {
	while (!shut_down_) {
	Sleep();
	if (shut_down_) {
	return;
	}
	ProcessReports();
	}
	}

	void ReportScheduler::Sleep() {
	// Note: We invoke the real system clock here, not clock_->now().
	// This is because even in a test we want to use the real system clock to
	// compute wakeup_time because std::condition_varaible::wait_until() always
	// uses the real system clock. A test is able to control the sleep time by
	// setting the value of sleep_interval_.
	auto wakeup_time = std::chrono::system_clock::now() + sleep_interval_;
	VLOG(3) << "ReportScheduler sleeping for " << sleep_interval_.count() << "ms";
	std::unique_lock<std::mutex> lock(mutex_);
	// Sleep until wakeup_time or shut_down_ = true.
	shut_down_notifier_.wait_until(lock, wakeup_time,
	[this] { return this->shut_down_.load(); });
	}

	void ReportScheduler::ProcessReports() {
	static const int kTimeoutSeconds = 60;
	config_manager_->Update(kTimeoutSeconds);
	uint32_t current_day_index = CurrentDayIndex();
	std::shared_ptr<config::ReportRegistry> report_registry =
	config_manager_->GetCurrent()->report_registry();
	for (const ReportConfig& report_config : *report_registry) {
	if (shut_down_) {
	return;
	}
	ProcessOneReport(report_config, current_day_index);
	}
	}

	void ReportScheduler::ProcessOneReport(const ReportConfig& report_config,
	uint32_t current_day_index) {
	LOG(INFO) << "ReportScheduler processing report_config "
	<< IdString(report_config);
	if (!report_config.has_scheduling()) {
	LOG(INFO) << "Skpping report_config " << IdString(report_config)
	<< " because it has no SchedulingConfig.";
	return;
	}
	switch (report_config.scheduling().aggregation_epoch_type()) {
	case DAY:
	ProcessDailyReport(report_config, current_day_index);
	return;

	case WEEK:
	ProcessWeeklyReport(report_config, current_day_index);
	return;

	case MONTH:
	ProcessMonthlyReport(report_config, current_day_index);
	return;

	default: {
	LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("one"))
	<< "Unrecognized aggregatoin_epoch_type: "
	<< report_config.scheduling().aggregation_epoch_type()
	<< "In ReportConfig " << IdString(report_config);
	return;
	}
	}
	}

	void ReportScheduler::ProcessDailyReport(const ReportConfig& report_config,
	uint32_t current_day_index) {
	// Look back a number of days equal to the maximum of daily_report_makeup_days
	// and report_finalization_days.
	auto scheduling = report_config.scheduling();
	if (report_config.scheduling().report_finalization_days() > 20) {
	LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("daily"))
	<< "Invalid ReportConfig: " << IdString(report_config)
	<< " report_finalization_days too large: "
	<< report_config.scheduling().report_finalization_days();
	return;
	}

	uint32_t finalization_days = scheduling.report_finalization_days();
	uint32_t lookback_days = FLAGS_daily_report_makeup_days >= finalization_days
	? FLAGS_daily_report_makeup_days
	: finalization_days;
	uint32_t period_start =
	(current_day_index >= lookback_days ? (current_day_index - lookback_days)
	: 0u);
	VLOG(4) << "ReportScheduler considering days in the interval ["
	<< period_start << ", " << current_day_index << "]";
	for (uint32_t day_index = period_start; day_index <= current_day_index;
	day_index++) {
	if (shut_down_) {
	return;
	}
	if (ShouldStartDailyReportNow(report_config, day_index,
	current_day_index)) {
	StartReportNow(report_config, day_index, day_index);
	} else {
	VLOG(4) << "ShouldStartDailyReportNow() returned false for report_config "
	<< IdString(report_config) << " day_index=" << day_index
	<< " current_day_index=" << current_day_index;
	}
	}
	}

	void ReportScheduler::ProcessWeeklyReport(const ReportConfig& report_config,
	uint32_t current_day_index) {
	LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("weekly"))
	<< "Scheduling of weekly reports is not yet implemented. ReportConfig: "
	<< IdString(report_config);
	return;
	}

	void ReportScheduler::ProcessMonthlyReport(const ReportConfig& report_config,
	uint32_t current_day_index) {
	LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("monthly"))
	<< "Scheduling of monthly reports is not yet implemented. ReportConfig: "
	<< IdString(report_config);
	return;
	}

	bool ReportScheduler::ShouldStartDailyReportNow(
	const ReportConfig& report_config, uint32_t day_index,
	uint32_t current_day_index) {
	if (day_index > current_day_index) {
	LOG_STACKDRIVER_COUNT_METRIC(ERROR, ProcessReportFailMetricName("daily"))
	<< "Unexpected condition: " << day_index
	<< " = day_index > current_day_index = " << current_day_index
	<< " for ReportConfig " << IdString(report_config);
	return false;
	}
	if (day_index > current_day_index -
	report_config.scheduling().report_finalization_days()) {
	// We want to generate the report repeatedly during the report finalization
	// period, but we don't want to start it again now if we previously started
	// it and that hasn't completed.
	return !report_history_->InProgress(report_config, day_index, day_index);
	}
	// After the report finalization period we only want to run the report once.
	// If it was ever successfully completed don't run it again. Also if we
	// previously started the report and that attempt hasn't finished yet,
	// don't start it again.
	return !(report_history_->CompletedSuccessfullyOrInProgress(
	report_config, day_index, day_index));
	}

	void ReportScheduler::StartReportNow(const ReportConfig& report_config,
	uint32_t first_day_index,
	uint32_t last_day_index) {
	const std::string export_name =
	ReportExportName(report_config, first_day_index, last_day_index);
	// We want to store the rows of the report in the report store in all cases
	// except when it is a RAW_DUMP report.
	bool in_store = (report_config.report_type() != RAW_DUMP);
	ReportId report_id;
	LOG(INFO) << "ReportScheduler starting report " << IdString(report_config)
	<< " [" << first_day_index << ", " << last_day_index << "]";
	auto status = report_starter_->StartReport(report_config, first_day_index,
	last_day_index, export_name,
	in_store, &report_id);
	if (!status.ok()) {
	LOG_STACKDRIVER_COUNT_METRIC(ERROR, kStartReportNowFailure)
	<< "ReportScheduler was unable to start a report for ReportConfig "
	<< IdString(report_config) << " first_day_index=" << first_day_index
	<< " last_day_index=" << last_day_index
	<< " error code=" << status.error_code()
	<< " error message=" << status.error_message();
	return;
	}
	report_history_->RecordStart(report_config, first_day_index, last_day_index,
	report_id);
	}

	std::string ReportScheduler::ReportExportName(const ReportConfig& report_config,
	uint32_t first_day_index,
	uint32_t last_day_index) {
	std::ostringstream stream;
	stream << "report_" << report_config.customer_id() << "_"
	<< report_config.project_id() << "_" << report_config.id() << "_"
	<< DateSuffix(first_day_index);
	if (last_day_index != first_day_index) {
	stream << "_" << DateSuffix(last_day_index);
	}
	return stream.str();
	}

	uint32_t ReportScheduler::CurrentDayIndex() {
	CHECK(clock_);
	std::time_t current_time =
	std::chrono::system_clock::to_time_t(clock_->now());
	return TimeToDayIndex(current_time, Metric::UTC);
	}

	ReportStarter::ReportStarter(ReportMasterService* report_master_service)
	: report_master_service_(report_master_service) {}

	grpc::Status ReportStarter::StartReport(const ReportConfig& report_config,
	uint32_t first_day_index,
	uint32_t last_day_index,
	const std::string& export_name,
	bool in_store,
	ReportId* report_id_out) {
	StartReportRequest start_request;
	start_request.set_customer_id(report_config.customer_id());
	start_request.set_project_id(report_config.project_id());
	start_request.set_report_config_id(report_config.id());
	start_request.set_first_day_index(first_day_index);
	start_request.set_last_day_index(last_day_index);
	StartReportResponse response_not_used;
	// This is not a one-off report generation. Rather it is scheduled.
	bool one_off = false;
	return report_master_service_->StartReportNoAuth(
	&start_request, one_off, export_name, in_store, report_id_out,
	&response_not_used);
	}

	} // namespace analyzer
	} // namespace cobalt