marsultor/minecraft-watchdog
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

should be done now, wrote a new batched logger for the last thing

Browse Source
This commit is contained in:
Mars Ultor
2025-12-24 20:11:34 -06:00
parent a80b1a2293
commit 519f37e385
1 changed files with 91 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
src/processing_utils.cpp
View File

@@ -8,7 +8,7 @@
#include <variant> #include <variant>
#include "database_utils.hpp" #include "database_utils.hpp"
namespace processing_utils { namespace processing_utils {
constexpr std::size_t MAX_BATCH_ROWS = 500;
bool parse_date_from_filename(std::string_view path, bool parse_date_from_filename(std::string_view path,
int& year, int& month, int& day) int& year, int& month, int& day)
{ {
@@ -204,10 +204,12 @@ parse_chat(std::string_view line)
activity_events.insert_as_batch(client); chat_events.insert_as_batch(client); client_events.insert_as_batch(client); activity_events.insert_as_batch(client); chat_events.insert_as_batch(client); client_events.insert_as_batch(client);
current_line.clear(); current_line.clear();
} }
void process_blocking(std::string path, void process_blocking(std::string path,
std::shared_ptr<clickhouse::Client> client, std::shared_ptr<clickhouse::Client> client,
std::string target_user_id) std::string target_user_id)
{ {
constexpr uint64_t MAX_BATCH_ROWS = 500;
spdlog::info("Processing (blocking) file {}", path); spdlog::info("Processing (blocking) file {}", path);
std::ifstream input_file(path); std::ifstream input_file(path);
@@ -216,68 +218,123 @@ parse_chat(std::string_view line)
return; return;
} }
// Seek to the end like tail -f // tail -f semantics
input_file.seekg(0, std::ios::end); input_file.seekg(0, std::ios::end);
std::string current_line; std::string current_line;
db_utils::ClientEventsBatch client_events{};
db_utils::ChatEventsBatch chat_events{}; db_utils::ClientEventsBatch client_events;
db_utils::ActivityEventsBatch activity_events{}; db_utils::ChatEventsBatch chat_events;
db_utils::ActivityEventsBatch activity_events;
uint64_t row_count = 0;
while (true) { while (true) {
// Read all available lines bool read_any = false;
while (std::getline(input_file, current_line)) {
if (current_line.empty()) continue;
std::time_t log_timestamp = extract_timestamp_utc(path, current_line); while (std::getline(input_file, current_line)) {
read_any = true;
if (current_line.empty())
continue;
std::time_t ts = extract_timestamp_utc(path, current_line);
if (ts == static_cast<std::time_t>(-1))
continue;
switch (classify_line(current_line)) { switch (classify_line(current_line)) {
case Route::CLIENT_START: case Route::CLIENT_START:
client_events.add_row(target_user_id, log_timestamp, client_events.add_row(
std::variant<std::string, uint8_t>{"start"}, path); target_user_id, ts,
std::variant<std::string, uint8_t>{"start"},
path
);
++row_count;
break; break;
case Route::CLIENT_END: case Route::CLIENT_END:
client_events.add_row(target_user_id, log_timestamp, client_events.add_row(
std::variant<std::string, uint8_t>{"end"}, path); target_user_id, ts,
std::variant<std::string, uint8_t>{"end"},
path
);
++row_count;
break; break;
case Route::ACT_COMMAND: case Route::ACT_COMMAND:
activity_events.add_row(target_user_id, log_timestamp, "command", 1.0f); activity_events.add_row(
target_user_id, ts,
"command", 1.0f
);
++row_count;
break; break;
case Route::ACT_DEATH: case Route::ACT_DEATH:
activity_events.add_row(target_user_id, log_timestamp, activity_events.add_row(
std::variant<std::string, uint8_t>{"death"}, 0.5f); target_user_id, ts,
std::variant<std::string, uint8_t>{"death"},
0.5f
);
++row_count;
break; break;
case Route::ACT_DIMENSION: case Route::ACT_DIMENSION:
activity_events.add_row(target_user_id, log_timestamp, activity_events.add_row(
std::variant<std::string, uint8_t>{"dimension"}, 0.3f); target_user_id, ts,
break; std::variant<std::string, uint8_t>{"dimension"},
case Route::ACT_OTHER: 0.3f
activity_events.add_row(target_user_id, log_timestamp, );
std::variant<std::string, uint8_t>{"other"}, 0.2f); ++row_count;
break; break;
case Route::CHAT: { case Route::CHAT: {
auto chat_pair = parse_chat(current_line); auto chat = parse_chat(current_line);
chat_events.add_row(target_user_id, log_timestamp, if (!chat.second.empty()) {
chat_pair.first, chat_pair.second); chat_events.add_row(
target_user_id, ts,
chat.first, chat.second
);
++row_count;
}
break; break;
} }
default: default:
break; break;
} }
// HARD FLUSH CONDITION
if (row_count >= MAX_BATCH_ROWS) {
activity_events.insert_as_batch(client);
chat_events.insert_as_batch(client);
client_events.insert_as_batch(client);
// Destroy and recreate — ONLY valid reset
client_events = db_utils::ClientEventsBatch{};
chat_events = db_utils::ChatEventsBatch{};
activity_events = db_utils::ActivityEventsBatch{};
row_count = 0;
}
} }
// Insert accumulated batches into ClickHouse // Flush partial batch if we read anything
activity_events.insert_as_batch(client); if (read_any && row_count > 0) {
chat_events.insert_as_batch(client); activity_events.insert_as_batch(client);
client_events.insert_as_batch(client); chat_events.insert_as_batch(client);
client_events.insert_as_batch(client);
current_line.clear(); client_events = db_utils::ClientEventsBatch{};
chat_events = db_utils::ChatEventsBatch{};
activity_events = db_utils::ActivityEventsBatch{};
// Clear EOF state so we can continue reading row_count = 0;
input_file.clear(); }
// Sleep a bit before polling for new data input_file.clear(); // clear EOF only
std::this_thread::sleep_for(std::chrono::milliseconds(200)); std::this_thread::sleep_for(std::chrono::milliseconds(200));
} }
} }
} }