/src/pdns/pdns/dnsdistdist/remote_logger.hh
Line | Count | Source |
1 | | /* |
2 | | * This file is part of PowerDNS or dnsdist. |
3 | | * Copyright -- PowerDNS.COM B.V. and its contributors |
4 | | * |
5 | | * This program is free software; you can redistribute it and/or modify |
6 | | * it under the terms of version 2 of the GNU General Public License as |
7 | | * published by the Free Software Foundation. |
8 | | * |
9 | | * In addition, for the avoidance of any doubt, permission is granted to |
10 | | * link this program with OpenSSL and to (re)distribute the binaries |
11 | | * produced as the result of such linking. |
12 | | * |
13 | | * This program is distributed in the hope that it will be useful, |
14 | | * but WITHOUT ANY WARRANTY; without even the implied warranty of |
15 | | * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
16 | | * GNU General Public License for more details. |
17 | | * |
18 | | * You should have received a copy of the GNU General Public License |
19 | | * along with this program; if not, write to the Free Software |
20 | | * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. |
21 | | */ |
22 | | #pragma once |
23 | | |
24 | | #include "config.h" |
25 | | |
26 | | #include <atomic> |
27 | | #include <thread> |
28 | | |
29 | | #include "iputils.hh" |
30 | | #include "circular_buffer.hh" // Keep, despite what clang-tidy days |
31 | | #include "lock.hh" |
32 | | #include "sstuff.hh" |
33 | | |
34 | | /* Writes can be submitted and they are atomically accepted. Either the whole write |
35 | | ends up in the buffer or nothing ends up in the buffer. |
36 | | In case nothing ends up in the buffer, an exception is thrown. |
37 | | Similarly, EOF leads to this treatment |
38 | | |
39 | | The filedescriptor can be in non-blocking mode. |
40 | | |
41 | | This class is not threadsafe. |
42 | | */ |
43 | | |
44 | | class CircularWriteBuffer |
45 | | { |
46 | | public: |
47 | | explicit CircularWriteBuffer(size_t size, uint8_t frame) : |
48 | | d_buffer(size), d_framesize(frame) |
49 | 0 | { |
50 | 0 | } |
51 | | |
52 | | [[nodiscard]] bool hasRoomFor(const std::string& str) const; |
53 | | [[nodiscard]] bool tooBig(const std::string& str) const; |
54 | | [[nodiscard]] bool isEmpty() const; |
55 | | bool write(const std::string& str); |
56 | | bool flush(int fileDesc); |
57 | | void clear(); |
58 | | |
59 | | private: |
60 | | boost::circular_buffer<char> d_buffer; |
61 | | uint8_t d_framesize; |
62 | | }; |
63 | | |
64 | | class RemoteLoggerInterface |
65 | | { |
66 | | public: |
67 | | enum class Result : uint8_t |
68 | | { |
69 | | Queued, |
70 | | PipeFull, |
71 | | TooLarge, |
72 | | OtherError |
73 | | }; |
74 | | static const std::string& toErrorString(Result result); |
75 | | |
76 | | RemoteLoggerInterface() = default; |
77 | | RemoteLoggerInterface(const RemoteLoggerInterface&) = delete; |
78 | | RemoteLoggerInterface(RemoteLoggerInterface&&) = delete; |
79 | | RemoteLoggerInterface& operator=(const RemoteLoggerInterface&) = delete; |
80 | | RemoteLoggerInterface& operator=(RemoteLoggerInterface&&) = delete; |
81 | | virtual ~RemoteLoggerInterface() = default; |
82 | | |
83 | | virtual Result queueData(const std::string& data) = 0; |
84 | | [[nodiscard]] virtual std::string address() const = 0; |
85 | | [[nodiscard]] virtual std::string toString() = 0; |
86 | | [[nodiscard]] virtual std::string name() const = 0; |
87 | 0 | [[nodiscard]] bool logQueries() const { return d_logQueries; } |
88 | 0 | [[nodiscard]] bool logResponses() const { return d_logResponses; } |
89 | 0 | [[nodiscard]] bool logNODs() const { return d_logNODs; } |
90 | 0 | [[nodiscard]] bool logUDRs() const { return d_logUDRs; } |
91 | 0 | void setLogQueries(bool flag) { d_logQueries = flag; } |
92 | 0 | void setLogResponses(bool flag) { d_logResponses = flag; } |
93 | 0 | void setLogNODs(bool flag) { d_logNODs = flag; } |
94 | 0 | void setLogUDRs(bool flag) { d_logUDRs = flag; } |
95 | | |
96 | | struct Stats |
97 | | { |
98 | | uint64_t d_queued{}; |
99 | | uint64_t d_pipeFull{}; |
100 | | uint64_t d_tooLarge{}; |
101 | | uint64_t d_otherError{}; |
102 | | |
103 | | Stats& operator+=(const Stats& rhs) |
104 | 0 | { |
105 | 0 | d_queued += rhs.d_queued; |
106 | 0 | d_pipeFull += rhs.d_pipeFull; |
107 | 0 | d_tooLarge += rhs.d_tooLarge; |
108 | 0 | d_otherError += rhs.d_otherError; |
109 | 0 | return *this; |
110 | 0 | } |
111 | | }; |
112 | | |
113 | | [[nodiscard]] virtual Stats getStats() = 0; |
114 | | |
115 | | private: |
116 | | bool d_logQueries{true}; |
117 | | bool d_logResponses{true}; |
118 | | bool d_logNODs{true}; |
119 | | bool d_logUDRs{false}; |
120 | | }; |
121 | | |
122 | | /* Thread safe. Will connect asynchronously on request. |
123 | | Runs a reconnection thread that also periodicall flushes. |
124 | | Note that the buffer only runs as long as there is a connection. |
125 | | If there is no connection we don't buffer a thing |
126 | | */ |
127 | | class RemoteLogger : public RemoteLoggerInterface |
128 | | { |
129 | | public: |
130 | | enum class FrameSize : uint8_t |
131 | | { |
132 | | Two, |
133 | | Four, |
134 | | }; |
135 | | RemoteLogger(const RemoteLogger&) = delete; |
136 | | RemoteLogger(RemoteLogger&&) = delete; |
137 | | RemoteLogger& operator=(const RemoteLogger&) = delete; |
138 | | RemoteLogger& operator=(RemoteLogger&&) = delete; |
139 | | RemoteLogger(const ComboAddress& remote, |
140 | | uint16_t timeout, // typically 2 |
141 | | uint64_t maxQueuedBytes, // typically 100 * maxQueuedEntries |
142 | | uint8_t reconnectWaitTime, // typically 1 |
143 | | bool asyncConnect, // typically false |
144 | | FrameSize frame, // typically FrameSize::Two |
145 | | time_t stalledWriteTimeoutSeconds); // typically 5 |
146 | | ~RemoteLogger() override; |
147 | | |
148 | | std::string address() const override |
149 | 0 | { |
150 | 0 | return d_remote.toStringWithPort(); |
151 | 0 | } |
152 | | |
153 | | [[nodiscard]] Result queueData(const std::string& data) override; |
154 | | [[nodiscard]] size_t maxSize() const |
155 | 0 | { |
156 | 0 | return d_framesize == FrameSize::Two ? std::numeric_limits<uint16_t>::max() : std::numeric_limits<uint32_t>::max(); |
157 | 0 | } |
158 | | [[nodiscard]] std::string name() const override |
159 | 0 | { |
160 | 0 | return "protobuf"; |
161 | 0 | } |
162 | | [[nodiscard]] std::string toString() override |
163 | 0 | { |
164 | 0 | auto runtime = d_runtime.lock(); |
165 | 0 | return d_remote.toStringWithPort() + " (" + std::to_string(runtime->d_stats.d_queued) + " processed, " + std::to_string(runtime->d_stats.d_pipeFull + runtime->d_stats.d_tooLarge + runtime->d_stats.d_otherError) + " dropped)"; |
166 | 0 | } |
167 | | |
168 | | [[nodiscard]] RemoteLoggerInterface::Stats getStats() override |
169 | 0 | { |
170 | 0 | return d_runtime.lock()->d_stats; |
171 | 0 | } |
172 | | |
173 | | void stop() |
174 | 0 | { |
175 | 0 | d_exiting = true; |
176 | 0 | } |
177 | | |
178 | | private: |
179 | | bool reconnect(); |
180 | | void maintenanceThread(); |
181 | | [[nodiscard]] bool connectionStalled(); |
182 | | |
183 | | struct RuntimeData |
184 | | { |
185 | | CircularWriteBuffer d_writer; |
186 | | std::unique_ptr<Socket> d_socket{nullptr}; |
187 | | RemoteLoggerInterface::Stats d_stats{}; |
188 | | }; |
189 | | |
190 | | ComboAddress d_remote; |
191 | | // if we have been trying to write for that long without any progress, |
192 | | // the connection is dead |
193 | | time_t d_stalledWriteTimeoutSeconds{}; |
194 | | time_t d_tryingToWriteSince{}; |
195 | | uint16_t d_timeout{}; |
196 | | uint8_t d_reconnectWaitTime{}; |
197 | | std::atomic<bool> d_exiting{false}; |
198 | | bool d_asyncConnect{false}; |
199 | | |
200 | | LockGuarded<RuntimeData> d_runtime; |
201 | | std::thread d_thread; |
202 | | FrameSize d_framesize{}; |
203 | | }; |