summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/ingestor.cpp32
-rw-r--r--src/ingestor.hpp5
2 files changed, 25 insertions, 12 deletions
diff --git a/src/ingestor.cpp b/src/ingestor.cpp
index f965d1d..4cd7859 100644
--- a/src/ingestor.cpp
+++ b/src/ingestor.cpp
@@ -164,20 +164,30 @@ namespace WebStat {
void
Ingestor::ingestLogLine(DB::Connection * dbconn, const std::string_view line)
{
+ auto rememberNewEntityIds = [this](const auto & ids) {
+ existingEntities.insert_range(ids | std::views::take_while(&std::optional<Crc32Value>::has_value)
+ | std::views::transform([](auto && value) {
+ return *value;
+ }));
+ };
if (auto result = scanLogLine(line)) {
linesParsed++;
const auto values = crc32ScanValues(result->values());
- std::optional<DB::TransactionScope> dbtx;
- if (const auto newEnts = newEntities(values); newEnts.front()) {
- dbtx.emplace(*dbconn);
- storeEntities(dbconn, newEnts);
+ NewEntityIds ids;
+ {
+ std::optional<DB::TransactionScope> dbtx;
+ if (const auto newEnts = newEntities(values); newEnts.front()) {
+ dbtx.emplace(*dbconn);
+ ids = storeEntities(dbconn, newEnts);
+ }
+ storeLogLine(dbconn, values);
}
- storeLogLine(dbconn, values);
+ rememberNewEntityIds(ids);
}
else {
linesDiscarded++;
const auto unparsableLine = toEntity(line, EntityType::UnparsableLine);
- storeEntities(dbconn, {unparsableLine});
+ rememberNewEntityIds(storeEntities(dbconn, {unparsableLine}));
}
}
@@ -270,15 +280,16 @@ namespace WebStat {
return rtn;
}
- void
+ Ingestor::NewEntityIds
Ingestor::storeEntities(DB::Connection * dbconn, const std::span<const std::optional<Entity>> values) const
{
static constexpr std::array ENTITY_TYPE_VALUES {
"host", "virtual_host", "path", "query_string", "referrer", "user_agent", "unparsable_line"};
auto insert = dbconn->modify(SQL::ENTITY_INSERT, SQL::ENTITY_INSERT_OPTS);
- std::ranges::for_each(
- values | std::views::take_while(&std::optional<Entity>::has_value), [this, &insert](auto && entity) {
+ NewEntityIds ids;
+ std::ranges::transform(values | std::views::take_while(&std::optional<Entity>::has_value), ids.begin(),
+ [this, &insert](auto && entity) {
const auto & [entityId, type, value] = *entity;
bindMany(insert, 0, entityId, ENTITY_TYPE_VALUES[std::to_underlying(type)], value);
if (insert->execute() > 0) {
@@ -293,8 +304,9 @@ namespace WebStat {
break;
}
}
- existingEntities.emplace(std::get<0>(*entity));
+ return std::get<0>(*entity);
});
+ return ids;
}
template<typename... T>
diff --git a/src/ingestor.hpp b/src/ingestor.hpp
index a20071e..8be7360 100644
--- a/src/ingestor.hpp
+++ b/src/ingestor.hpp
@@ -60,13 +60,15 @@ namespace WebStat {
size_t linesParsed = 0;
size_t linesDiscarded = 0;
size_t linesParked = 0;
+ mutable std::flat_set<Crc32Value> existingEntities;
using JobLastRunTime = std::chrono::system_clock::time_point;
JobLastRunTime lastRunIngestParkedLines;
private:
static constexpr size_t MAX_NEW_ENTITIES = 6;
- void storeEntities(DB::Connection *, std::span<const std::optional<Entity>>) const;
+ using NewEntityIds = std::array<std::optional<Crc32Value>, MAX_NEW_ENTITIES>;
+ NewEntityIds storeEntities(DB::Connection *, std::span<const std::optional<Entity>>) const;
using NewEntities = std::array<std::optional<Entity>, MAX_NEW_ENTITIES>;
template<typename... T> NewEntities newEntities(const std::tuple<T...> &) const;
void handleCurlOperations();
@@ -75,7 +77,6 @@ namespace WebStat {
void jobIngestParkedLine(const std::filesystem::path &, uintmax_t size);
using CurlOperations = std::map<CURL *, std::unique_ptr<CurlOperation>>;
- mutable std::flat_set<Crc32Value> existingEntities;
uint32_t hostnameId;
CurlMultiPtr curl;
mutable CurlOperations curlOperations;