Implement pipeline logging with bounded channels and orchestrator integration

This commit is contained in:
Aaron Po
2026-05-14 21:08:17 -04:00
parent f93b14897b
commit 74f11b57e2
16 changed files with 361 additions and 61 deletions

View File

@@ -155,13 +155,13 @@ target_sources(${PROJECT_NAME} PRIVATE
src/application_options/parse_arguments.cc src/application_options/parse_arguments.cc
) )
# --- biergarten_data_generator --- # --- biergarten_pipeline_orchestrator ---
target_sources(${PROJECT_NAME} PRIVATE target_sources(${PROJECT_NAME} PRIVATE
src/biergarten_data_generator/log_results.cc src/biergarten_pipeline_orchestrator/log_results.cc
src/biergarten_data_generator/biergarten_data_generator.cc src/biergarten_pipeline_orchestrator/biergarten_data_generator.cc
src/biergarten_data_generator/generate_breweries.cc src/biergarten_pipeline_orchestrator/generate_breweries.cc
src/biergarten_data_generator/run.cc src/biergarten_pipeline_orchestrator/run.cc
src/biergarten_data_generator/query_cities_with_countries.cc src/biergarten_pipeline_orchestrator/query_cities_with_countries.cc
) )
# --- web_client --- # --- web_client ---
@@ -202,17 +202,23 @@ target_sources(${PROJECT_NAME} PRIVATE
# --- services: sqlite --- # --- services: sqlite ---
target_sources(${PROJECT_NAME} PRIVATE target_sources(${PROJECT_NAME} PRIVATE
src/services/sqlite/process_record.cc src/services/sqlite/process_record.cc
src/services/sqlite/sqlite_export_service.cc src/services/sqlite/sqlite_export_service.cc
src/services/sqlite/finalize.cc src/services/sqlite/finalize.cc
src/services/sqlite/initialize.cc src/services/sqlite/initialize.cc
src/services/sqlite/helpers/sqlite_connection_helpers.cc src/services/sqlite/helpers/sqlite_connection_helpers.cc
src/services/sqlite/helpers/sqlite_statement_helpers.cc src/services/sqlite/helpers/sqlite_statement_helpers.cc
)
# --- services: logging ---
target_sources(${PROJECT_NAME} PRIVATE
"src/services/logging/channel_logger.cc"
src/services/logging/log_consumer.cc
) )
# --- services (top-level) --- # --- services (top-level) ---
target_sources(${PROJECT_NAME} PRIVATE target_sources(${PROJECT_NAME} PRIVATE
src/services/prompt_directory.cc src/services/prompt_directory.cc
) )
# 6. Include Directories, Link Libraries & Compile Definitions # 6. Include Directories, Link Libraries & Compile Definitions

View File

@@ -2,7 +2,7 @@
#define BIERGARTEN_PIPELINE_INCLUDES_BIERGARTEN_DATA_GENERATOR_H_ #define BIERGARTEN_PIPELINE_INCLUDES_BIERGARTEN_DATA_GENERATOR_H_
/** /**
* @file biergarten_data_generator.h * @file biergarten_pipeline_orchestrator.h
* @brief Core orchestration class for pipeline data generation. * @brief Core orchestration class for pipeline data generation.
*/ */
@@ -21,7 +21,7 @@
* This class encapsulates the core logic for generating brewery data. * This class encapsulates the core logic for generating brewery data.
* It handles location loading, city enrichment, and brewery generation. * It handles location loading, city enrichment, and brewery generation.
*/ */
class BiergartenDataGenerator { class BiergartenPipelineOrchestrator {
public: public:
/** /**
* @brief Construct a BiergartenDataGenerator with injected dependencies. * @brief Construct a BiergartenDataGenerator with injected dependencies.
@@ -30,10 +30,11 @@ class BiergartenDataGenerator {
* @param generator Brewery and user data generator. * @param generator Brewery and user data generator.
* @param exporter Storage backend for generated brewery data. * @param exporter Storage backend for generated brewery data.
*/ */
BiergartenDataGenerator(std::unique_ptr<IEnrichmentService> context_service, BiergartenPipelineOrchestrator(
std::unique_ptr<DataGenerator> generator, std::unique_ptr<IEnrichmentService> context_service,
std::unique_ptr<IExportService> exporter, std::unique_ptr<DataGenerator> generator,
const ApplicationOptions& application_options); std::unique_ptr<IExportService> exporter,
const ApplicationOptions& application_options);
/** /**
* @brief Run the data generation pipeline. * @brief Run the data generation pipeline.
@@ -57,7 +58,7 @@ class BiergartenDataGenerator {
/// @brief Storage backend for generated brewery records. /// @brief Storage backend for generated brewery records.
std::unique_ptr<IExportService> exporter_; std::unique_ptr<IExportService> exporter_;
const ApplicationOptions application_options_; ApplicationOptions application_options_;
/** /**
* @brief Load locations from JSON and sample cities. * @brief Load locations from JSON and sample cities.

View File

@@ -2,17 +2,17 @@
// Created by aaronpo on 29/04/2026. // Created by aaronpo on 29/04/2026.
// //
#ifndef CONCURRENCY_INTRO_CPP_INCLUDES_BOUNDEDCHANNEL_H_ #ifndef BIERGARTEN_PIPELINE_INCLUDES_CONCURRENCY_BOUNDED_CHANNEL_H_
#define CONCURRENCY_INTRO_CPP_INCLUDES_BOUNDEDCHANNEL_H_ #define BIERGARTEN_PIPELINE_INCLUDES_CONCURRENCY_BOUNDED_CHANNEL_H_
#include <condition_variable> #include <condition_variable>
#include <cstddef>
#include <mutex> #include <mutex>
#include <optional> #include <optional>
#include <queue> #include <queue>
#include <cstddef>
/** /**
* @file BoundedChannel.h * @file bounded_channel.h
* @brief A thread-safe, bounded multi-producer/multi-consumer channel. * @brief A thread-safe, bounded multi-producer/multi-consumer channel.
*/ */
@@ -26,7 +26,6 @@
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
template <typename T> template <typename T>
class BoundedChannel { class BoundedChannel {
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
// Internal state — all access must be guarded by mutex_. // Internal state — all access must be guarded by mutex_.
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
@@ -44,7 +43,6 @@ class BoundedChannel {
bool closed_ = false; bool closed_ = false;
public: public:
/** /**
* @brief Construct a bounded channel with the given capacity. * @brief Construct a bounded channel with the given capacity.
* @param capacity Maximum number of items the channel may hold. * @param capacity Maximum number of items the channel may hold.
@@ -68,10 +66,10 @@ class BoundedChannel {
/** /**
* @brief Close the channel and unblock all waiting threads. Idempotent. * @brief Close the channel and unblock all waiting threads. Idempotent.
*/ */
void close(); void Close();
}; };
// Include the template implementation // Include the template implementation
#include "BoundedChannel.tcc" #include "bounded_channel.tcc"
#endif // CONCURRENCY_INTRO_CPP_INCLUDES_BOUNDEDCHANNEL_H_ #endif // BIERGARTEN_PIPELINE_INCLUDES_CONCURRENCY_BOUNDED_CHANNEL_H_

View File

@@ -1,4 +1,4 @@
#include "BoundedChannel.h" #include "bounded_channel.h"
template <typename T> template <typename T>
void BoundedChannel<T>::Send(T item) { void BoundedChannel<T>::Send(T item) {
@@ -42,7 +42,7 @@ std::optional<T> BoundedChannel<T>::Receive() {
} }
template <typename T> template <typename T>
void BoundedChannel<T>::close() { void BoundedChannel<T>::Close() {
// Acquire exclusive ownership of the mutex to ensure visibility of the flag. // Acquire exclusive ownership of the mutex to ensure visibility of the flag.
std::unique_lock lock(mutex_); std::unique_lock lock(mutex_);

View File

@@ -0,0 +1,37 @@
/**
* @file services/logging/channel_logger.h
* @brief Channel-backed implementation of the Logger interface.
*
* ChannelLogger constructs LogEntry values and forwards them to a
* BoundedChannel<LogEntry> for asynchronous consumption by LogConsumer.
* The channel is injected by reference; ChannelLogger does not own it.
*/
#ifndef BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_CHANNEL_LOGGER_H_
#define BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_CHANNEL_LOGGER_H_
#include <string_view>
#include "concurrency/bounded_channel.h"
#include "services/logging/log_entry.h"
#include "services/logging/logger.h"
class ChannelLogger final : public ILogger {
public:
explicit ChannelLogger(BoundedChannel<LogEntry>& channel);
ChannelLogger(const ChannelLogger&) = delete;
ChannelLogger& operator=(const ChannelLogger&) = delete;
ChannelLogger(ChannelLogger&&) = delete;
ChannelLogger& operator=(ChannelLogger&&) = delete;
~ChannelLogger() override = default;
void Log(LogLevel level, PipelinePhase phase,
std::string_view message) override;
private:
BoundedChannel<LogEntry>& channel_;
};
#endif // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_CHANNEL_LOGGER_H_

View File

@@ -0,0 +1,44 @@
/**
* @file services/logging/log_consumer.h
* @brief Dedicated log drain worker for the pipeline logging channel.
*
* LogConsumer runs on its own thread, draining LogEntry items from a
* BoundedChannel<LogEntry> and forwarding them to spdlog. Exits cleanly
* when the channel is closed.
*/
#ifndef BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_CONSUMER_H_
#define BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_CONSUMER_H_
#include <spdlog/spdlog.h>
#include <string>
#include "concurrency/bounded_channel.h"
#include "services/logging/log_entry.h"
class LogConsumer {
public:
explicit LogConsumer(BoundedChannel<LogEntry>& channel);
LogConsumer(const LogConsumer&) = delete;
LogConsumer& operator=(const LogConsumer&) = delete;
LogConsumer(LogConsumer&&) = delete;
LogConsumer& operator=(LogConsumer&&) = delete;
/**
* @brief Drains the channel until it is closed.
*
* Intended to be run on a dedicated std::thread. Exits when:
* - The channel is closed and the queue is fully drained.
*/
void Run();
private:
BoundedChannel<LogEntry>& channel_;
static spdlog::level::level_enum ToSpdlogLevel(LogLevel level);
static std::string ToString(PipelinePhase phase);
};
#endif // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_CONSUMER_H_

View File

@@ -0,0 +1,41 @@
/**
* @file services/logging/log_entry.h
* @brief POD struct representing a single log event in the pipeline.
*
* LogEntry is produced by PipelineLogger and consumed by LogWorker via
* BoundedChannel<LogEntry>. All fields are value types so entries are
* safely movable across the channel without shared ownership.
*/
#ifndef BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_ENTRY_H_
#define BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_ENTRY_H_
#include <chrono>
#include <string>
enum class LogLevel {
Debug,
Info,
Warn,
Error,
};
enum class PipelinePhase {
Startup,
UserGeneration,
BreweryAndBeerGeneration,
CheckinGeneration,
RatingGeneration,
FollowGeneration,
Teardown,
};
struct LogEntry {
std::chrono::system_clock::time_point timestamp =
std::chrono::system_clock::now();
LogLevel level;
PipelinePhase phase;
std::string message;
};
#endif // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOG_ENTRY_H_

View File

@@ -0,0 +1,31 @@
/**
* @file services/logging/logger.h
* @brief Abstract interface for pipeline logging.
*
* Kept intentionally narrow. Components that need to log depend on Logger,
* not on PipelineLogger or any channel type.
*/
#ifndef BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOGGER_H_
#define BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOGGER_H_
#include <optional>
#include <string>
#include <string_view>
#include "services/logging/log_entry.h"
class ILogger {
public:
ILogger() = default;
ILogger(const ILogger&) = delete;
ILogger& operator=(const ILogger&) = delete;
ILogger(ILogger&&) = delete;
ILogger& operator=(ILogger&&) = delete;
virtual ~ILogger() = default;
virtual void Log(LogLevel level, PipelinePhase phase,
std::string_view message) = 0;
};
#endif // BIERGARTEN_PIPELINE_INCLUDES_SERVICES_LOGGING_LOGGER_H_

View File

@@ -1,5 +1,5 @@
/** /**
* @file biergarten_data_generator/biergarten_data_generator.cc * @file biergarten_pipeline_orchestrator/biergarten_pipeline_orchestrator.cc
* @brief BiergartenDataGenerator constructor implementation. * @brief BiergartenDataGenerator constructor implementation.
*/ */
@@ -7,7 +7,7 @@
#include <utility> #include <utility>
BiergartenDataGenerator::BiergartenDataGenerator( BiergartenPipelineOrchestrator::BiergartenPipelineOrchestrator(
std::unique_ptr<IEnrichmentService> context_service, std::unique_ptr<IEnrichmentService> context_service,
std::unique_ptr<DataGenerator> generator, std::unique_ptr<DataGenerator> generator,
std::unique_ptr<IExportService> exporter, std::unique_ptr<IExportService> exporter,

View File

@@ -1,5 +1,5 @@
/** /**
* @file biergarten_data_generator/generate_breweries.cc * @file biergarten_pipeline_orchestrator/generate_breweries.cc
* @brief BiergartenDataGenerator::GenerateBreweries() implementation. * @brief BiergartenDataGenerator::GenerateBreweries() implementation.
*/ */
@@ -7,7 +7,7 @@
#include "biergarten_data_generator.h" #include "biergarten_data_generator.h"
void BiergartenDataGenerator::GenerateBreweries( void BiergartenPipelineOrchestrator::GenerateBreweries(
std::span<const EnrichedCity> cities) { std::span<const EnrichedCity> cities) {
spdlog::info("\n=== SAMPLE BREWERY GENERATION ==="); spdlog::info("\n=== SAMPLE BREWERY GENERATION ===");

View File

@@ -1,5 +1,5 @@
/** /**
* @file biergarten_data_generator/log_results.cc * @file biergarten_pipeline_orchestrator/log_results.cc
* @brief BiergartenDataGenerator::LogResults() implementation. * @brief BiergartenDataGenerator::LogResults() implementation.
*/ */
@@ -7,7 +7,7 @@
#include "biergarten_data_generator.h" #include "biergarten_data_generator.h"
void BiergartenDataGenerator::LogResults() const { void BiergartenPipelineOrchestrator::LogResults() const {
spdlog::info("\n=== GENERATED DATA DUMP ==="); spdlog::info("\n=== GENERATED DATA DUMP ===");
size_t index = 1; size_t index = 1;
for (const auto& [location, brewery] : generated_breweries_) { for (const auto& [location, brewery] : generated_breweries_) {

View File

@@ -1,5 +1,5 @@
/** /**
* @file biergarten_data_generator/query_cities_with_countries.cc * @file biergarten_pipeline_orchestrator/query_cities_with_countries.cc
* @brief BiergartenDataGenerator::QueryCitiesWithCountries() implementation. * @brief BiergartenDataGenerator::QueryCitiesWithCountries() implementation.
*/ */
@@ -13,7 +13,7 @@
#include "biergarten_data_generator.h" #include "biergarten_data_generator.h"
#include "json_handling/json_loader.h" #include "json_handling/json_loader.h"
std::vector<Location> BiergartenDataGenerator::QueryCitiesWithCountries() { std::vector<Location> BiergartenPipelineOrchestrator::QueryCitiesWithCountries() {
spdlog::info("\n=== GEOGRAPHIC DATA OVERVIEW ==="); spdlog::info("\n=== GEOGRAPHIC DATA OVERVIEW ===");
const std::filesystem::path locations_path = "locations.json"; const std::filesystem::path locations_path = "locations.json";

View File

@@ -1,5 +1,5 @@
/** /**
* @file biergarten_data_generator/run.cc * @file biergarten_pipeline_orchestrator/run.cc
* @brief BiergartenDataGenerator::Run() implementation. * @brief BiergartenDataGenerator::Run() implementation.
*/ */
@@ -9,7 +9,7 @@
#include "biergarten_data_generator.h" #include "biergarten_data_generator.h"
bool BiergartenDataGenerator::Run() { bool BiergartenPipelineOrchestrator::Run() {
try { try {
exporter_->Initialize(); exporter_->Initialize();

View File

@@ -12,8 +12,10 @@
#include <memory> #include <memory>
#include <optional> #include <optional>
#include <string> #include <string>
#include <thread>
#include "biergarten_data_generator.h" #include "biergarten_data_generator.h"
#include "concurrency/bounded_channel.h"
#include "data_generation/llama_generator.h" #include "data_generation/llama_generator.h"
#include "data_generation/mock_generator.h" #include "data_generation/mock_generator.h"
#include "data_generation/prompt_formatting/gemma4_jinja_prompt_formatter.h" #include "data_generation/prompt_formatting/gemma4_jinja_prompt_formatter.h"
@@ -25,12 +27,22 @@
#include "services/enrichment/enrichment_service.h" #include "services/enrichment/enrichment_service.h"
#include "services/enrichment/mock_enrichment.h" #include "services/enrichment/mock_enrichment.h"
#include "services/enrichment/wikipedia_service.h" #include "services/enrichment/wikipedia_service.h"
#include "services/logging/channel_logger.h"
#include "services/logging/log_consumer.h"
#include "services/logging/log_entry.h"
#include "services/logging/logger.h"
#include "services/prompting/prompt_directory.h" #include "services/prompting/prompt_directory.h"
#include "web_client/http_web_client.h" #include "web_client/http_web_client.h"
namespace di = boost::di; namespace di = boost::di;
static constexpr size_t kLogMaxCount = 512;
int main(const int argc, char** argv) { int main(const int argc, char** argv) {
auto log_channel = std::make_shared<BoundedChannel<LogEntry>>(kLogMaxCount);
ChannelLogger channel_logger(*log_channel);
LogConsumer log_worker(*log_channel);
std::thread log_thread([&log_worker] { log_worker.Run(); });
try { try {
Timer timer; Timer timer;
spdlog::set_pattern("[%Y-%m-%d %H:%M:%S.%e] [%^%l%$] %v"); spdlog::set_pattern("[%Y-%m-%d %H:%M:%S.%e] [%^%l%$] %v");
@@ -46,6 +58,8 @@ int main(const int argc, char** argv) {
ParseArguments(argc, argv); ParseArguments(argc, argv);
if (!parsed_options.has_value()) { if (!parsed_options.has_value()) {
log_channel->Close();
log_thread.join();
return 0; return 0;
} }
@@ -54,66 +68,97 @@ int main(const int argc, char** argv) {
const auto sampling = const auto sampling =
options.generator.sampling.value_or(SamplingOptions{}); options.generator.sampling.value_or(SamplingOptions{});
// -----------------------------------------------------------------------
// Prompt directory
// Conditionally constructed before the injector; moved into LlamaGenerator.
// -----------------------------------------------------------------------
std::unique_ptr<IPromptDirectory> prompt_directory; std::unique_ptr<IPromptDirectory> prompt_directory;
if (!options.generator.use_mocked) { if (!options.generator.use_mocked) {
try { try {
prompt_directory = prompt_directory =
std::make_unique<PromptDirectory>(options.pipeline.prompt_dir); std::make_unique<PromptDirectory>(options.pipeline.prompt_dir);
} catch (const std::exception& dir_error) { } catch (const std::exception& dir_error) {
spdlog::error("[Startup] Invalid --prompt-dir: {}", dir_error.what()); channel_logger.Log(
LogLevel::Error, PipelinePhase::Startup,
std::string("Invalid --prompt-dir: ") + dir_error.what());
log_channel->Close();
log_thread.join();
return 1; return 1;
} }
} }
// -----------------------------------------------------------------------
// Dependency injection
// -----------------------------------------------------------------------
const auto injector = di::make_injector( const auto injector = di::make_injector(
di::bind<ApplicationOptions>().to(options), di::bind<ApplicationOptions>().to(options),
di::bind<std::string>().to(model_path), di::bind<std::string>().to(model_path),
di::bind<WebClient>().to<HttpWebClient>(), di::bind<WebClient>().to<HttpWebClient>(),
di::bind<IExportService>().to<SqliteExportService>(), di::bind<IExportService>().to<SqliteExportService>(),
di::bind<IPromptFormatter>().to<Gemma4JinjaPromptFormatter>(), di::bind<IPromptFormatter>().to<Gemma4JinjaPromptFormatter>(),
di::bind<ILogger>().to(
[log_channel](const auto&) -> std::unique_ptr<ILogger> {
return std::make_unique<ChannelLogger>(*log_channel);
}),
di::bind<IEnrichmentService>().to( di::bind<IEnrichmentService>().to(
[options](const auto& inj) -> std::unique_ptr<IEnrichmentService> { [options](const auto& inj) -> std::unique_ptr<IEnrichmentService> {
if (options.generator.use_mocked) { if (options.generator.use_mocked) {
return std::make_unique<MockEnrichmentService>(); return std::make_unique<MockEnrichmentService>();
} }
return std::make_unique<WikipediaEnrichmentService>( return std::make_unique<WikipediaEnrichmentService>(
inj.template create<std::unique_ptr<WebClient>>()); inj.template create<std::unique_ptr<WebClient>>());
}), }),
di::bind<DataGenerator>().to( di::bind<DataGenerator>().to(
[options, model_path, sampling, &prompt_directory]( [&options, &model_path, &sampling, &prompt_directory,
&channel_logger](
const auto& inj) -> std::unique_ptr<DataGenerator> { const auto& inj) -> std::unique_ptr<DataGenerator> {
if (options.generator.use_mocked) { if (options.generator.use_mocked) {
spdlog::info( channel_logger.Log(
"[Generator] Using MockGenerator (no model path provided)"); LogLevel::Info, PipelinePhase::Startup,
"Using MockGenerator (no model path provided)");
return std::make_unique<MockGenerator>(); return std::make_unique<MockGenerator>();
} }
channel_logger.Log(
spdlog::info( LogLevel::Info, PipelinePhase::Startup,
"[Generator] Using LlamaGenerator: {} (temperature={}, " "Using LlamaGenerator: " + model_path +
"top-p={}, top-k={}, n_ctx={}, seed={})", " (temperature=" + std::to_string(sampling.temperature) +
model_path, sampling.temperature, sampling.top_p, ", top-p=" + std::to_string(sampling.top_p) +
sampling.top_k, sampling.n_ctx, sampling.seed); ", top-k=" + std::to_string(sampling.top_k) +
", n_ctx=" + std::to_string(sampling.n_ctx) +
", seed=" + std::to_string(sampling.seed) + ")");
return std::make_unique<LlamaGenerator>( return std::make_unique<LlamaGenerator>(
options, model_path, options, model_path,
inj.template create<std::unique_ptr<IPromptFormatter>>(), inj.template create<std::unique_ptr<IPromptFormatter>>(),
std::move(prompt_directory)); std::move(prompt_directory));
}) }));
); // -----------------------------------------------------------------------
// Pipeline execution
// -----------------------------------------------------------------------
const auto orchestrator =
injector.create<std::unique_ptr<BiergartenPipelineOrchestrator>>();
const auto generator = if (!orchestrator->Run()) {
injector.create<std::unique_ptr<BiergartenDataGenerator>>(); channel_logger.Log(LogLevel::Error, PipelinePhase::Teardown,
"Pipeline execution failed");
if (!generator->Run()) { log_channel->Close();
spdlog::error("Pipeline execution failed"); log_thread.join();
return 1; return 1;
} }
spdlog::info("Pipeline executed successfully in {} ms", timer.Elapsed()); channel_logger.Log(LogLevel::Info, PipelinePhase::Teardown,
"Pipeline executed successfully in " +
std::to_string(timer.Elapsed()) + " ms");
log_channel->Close();
log_thread.join();
return 0; return 0;
} catch (const std::exception& exception) { } catch (const std::exception& exception) {
// Channel may be in an unknown state; fall back to spdlog directly.
spdlog::critical("Unhandled fatal error in main: {}", exception.what()); spdlog::critical("Unhandled fatal error in main: {}", exception.what());
log_channel->Close();
log_thread.join();
return 1; return 1;
} }
} }

View File

@@ -0,0 +1,23 @@
/**
* @file services/logging/channel_logger.cc
*/
#include <chrono>
#include <optional>
#include <string>
#include <string_view>
#include "concurrency/bounded_channel.h"
#include "services/logging/channel_logger.h"
#include "services/logging/log_entry.h"
ChannelLogger::ChannelLogger(BoundedChannel<LogEntry>& channel)
: channel_(channel) {}
void ChannelLogger::Log(LogLevel level, PipelinePhase phase,
const std::string_view message) {
channel_.Send(LogEntry{.timestamp = std::chrono::system_clock::now(),
.level = level,
.phase = phase,
.message = std::string(message)});
}

View File

@@ -0,0 +1,74 @@
/**
* @file services/logging/log_consumer.cc
* @brief Dedicated log drain worker implementation.
*
* LogConsumer drains LogEntry items from a BoundedChannel and forwards them
* to spdlog for final output.
*/
#include "services/logging/log_consumer.h"
#include <spdlog/spdlog.h>
#include <string>
#include "concurrency/bounded_channel.h"
#include "services/logging/log_entry.h"
LogConsumer::LogConsumer(BoundedChannel<LogEntry>& channel)
: channel_(channel) {}
void LogConsumer::Run() {
while (true) {
auto entry = channel_.Receive();
if (!entry.has_value()) {
// Channel is closed and drained.
break;
}
const LogEntry& log_entry = entry.value();
auto logger = spdlog::default_logger();
const std::string formatted_message = [&] {
std::string msg = std::string(log_entry.message);
msg += " [phase=" + ToString(log_entry.phase) + "]";
return msg;
}();
logger->log(ToSpdlogLevel(log_entry.level), formatted_message);
}
}
spdlog::level::level_enum LogConsumer::ToSpdlogLevel(LogLevel level) {
switch (level) {
case LogLevel::Debug:
return spdlog::level::debug;
case LogLevel::Info:
return spdlog::level::info;
case LogLevel::Warn:
return spdlog::level::warn;
case LogLevel::Error:
return spdlog::level::err;
}
return spdlog::level::info;
}
std::string LogConsumer::ToString(PipelinePhase phase) {
switch (phase) {
case PipelinePhase::Startup:
return "Startup";
case PipelinePhase::UserGeneration:
return "UserGeneration";
case PipelinePhase::BreweryAndBeerGeneration:
return "BreweryAndBeerGeneration";
case PipelinePhase::CheckinGeneration:
return "CheckinGeneration";
case PipelinePhase::RatingGeneration:
return "RatingGeneration";
case PipelinePhase::FollowGeneration:
return "FollowGeneration";
case PipelinePhase::Teardown:
return "Teardown";
}
return "Unknown";
}