1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
#pragma once
#include "curlOp.hpp"
#include "logTypes.hpp"
#include "settings.hpp"
#include <c++11Helpers.h>
#include <connectionPool.h>
#include <connection_fwd.h>
#include <cstdio>
#include <expected>
#include <flat_map>
#include <future>
#include <scn/scan.h>
#include <span>
#include <sys/utsname.h>
namespace WebStat {
using namespace std::chrono;
using namespace std::chrono_literals;
struct IngestorSettings : Settings {
// NOLINTBEGIN(readability-magic-numbers)
std::string dbConnStr = "dbname=webstat user=webstat";
std::string userAgentAPI = "https://useragentstring.com";
std::filesystem::path fallbackDir = "/var/log/webstat";
unsigned int dbMax = 4;
unsigned int dbKeep = 2;
size_t maxBatchSize = 1;
size_t maxBatches = 5;
minutes checkJobsAfter = 1min;
minutes freqIngestParkedLines = 30min;
minutes freqPurgeOldLogs = 6h;
unsigned int purgeDaysToKeep = 61; // ~2 months
unsigned int purgeDeleteMax = 10'000;
minutes purgeDeleteMaxTime = 5min;
seconds purgeDeletePause = 3s;
// NOLINTEND(readability-magic-numbers)
};
class Ingestor {
public:
using LineBatch = std::vector<std::string>;
using LinesView = std::span<const std::string>;
Ingestor(const utsname &, IngestorSettings);
Ingestor(const utsname &, DB::ConnectionPoolPtr, IngestorSettings);
virtual ~Ingestor();
SPECIAL_MEMBERS_DELETE(Ingestor);
using ScanResult = decltype(scn::scan<std::string_view, std::string_view, uint64_t, std::string_view,
QuotedString, QueryString, std::string_view, unsigned short, unsigned int, unsigned int, CLFString,
CLFString, CLFString>(std::declval<std::string_view>(), ""));
using ScanValues = std::remove_cvref_t<decltype(std::declval<WebStat::Ingestor::ScanResult>()->values())>;
[[nodiscard]] static ScanResult scanLogLine(std::string_view);
void ingestLog(std::FILE *);
void tryIngestQueuedLogLines();
void ingestLogLines(DB::Connection *, LinesView lines);
std::expected<std::filesystem::path, int> parkQueuedLogLines();
void runJobsAsNeeded();
unsigned int jobIngestParkedLines();
unsigned int jobPurgeOldLogs();
template<typename... T> void storeLogLine(DB::Connection *, const std::tuple<T...> &) const;
IngestorSettings settings;
struct Stats {
size_t linesRead;
size_t linesParsed;
size_t linesParseFailed;
size_t logsInserted;
size_t entitiesInserted;
constexpr bool operator==(const Ingestor::Stats &) const = default;
};
protected:
static Ingestor * currentIngestor;
DB::ConnectionPoolPtr dbpool;
mutable Stats stats {};
std::flat_map<EntityHash, EntityId> existingEntities;
LineBatch queuedLines;
bool terminated = false;
struct Job {
using LastRunTime = std::chrono::system_clock::time_point;
using Impl = unsigned int (Ingestor::*)();
explicit Job(Impl jobImpl) : impl(jobImpl) { }
const Impl impl;
LastRunTime lastRun {LastRunTime::clock::now()};
std::optional<std::future<unsigned int>> currentRun;
};
Job::LastRunTime lastCheckedJobs {Job::LastRunTime::clock::now()};
Job ingestParkedLines;
Job purgeOldLogs;
private:
template<typename... T> static std::vector<Entity *> entities(std::tuple<T...> &);
void fillKnownEntities(std::span<Entity *>) const;
void storeNewEntities(DB::Connection *, std::span<Entity *>) const;
void storeNewEntity(DB::Connection *, Entity &) const;
void onNewUserAgent(const Entity &) const;
void handleCurlOperations();
void logStats() const;
void clearStats();
void finalizeJob(Job &, minutes freq, Job::LastRunTime::clock::time_point now);
void finishAllJobs();
void jobIngestParkedLines(const std::filesystem::path &);
size_t jobIngestParkedLines(FILE *, size_t count);
static void sigtermHandler(int);
void terminate(int);
static void sigusr1Handler(int);
static void sigusr2Handler(int);
[[gnu::format(printf, 3, 4)]] virtual void log(int level, const char * msgfmt, ...) const = 0;
using CurlOperations = std::map<CURL *, std::unique_ptr<CurlOperation>>;
EntityId hostnameId;
CurlMultiPtr curl;
mutable CurlOperations curlOperations;
std::thread::id mainThread;
};
}
|
