userver: userver/storages/postgres/cluster.hpp Source File
Loading...
Searching...
No Matches
cluster.hpp
Go to the documentation of this file.
1#pragma once
2
3/// @file userver/storages/postgres/cluster.hpp
4/// @brief @copybrief storages::postgres::Cluster
5
6#include <memory>
7
8#include <userver/clients/dns/resolver_fwd.hpp>
9#include <userver/dynamic_config/source.hpp>
10#include <userver/engine/task/task_processor_fwd.hpp>
11#include <userver/engine/task/task_with_result.hpp>
12#include <userver/error_injection/settings_fwd.hpp>
13#include <userver/testsuite/postgres_control.hpp>
14#include <userver/testsuite/tasks.hpp>
15
16#include <userver/storages/postgres/cluster_types.hpp>
17#include <userver/storages/postgres/database.hpp>
18#include <userver/storages/postgres/detail/non_transaction.hpp>
19#include <userver/storages/postgres/notify.hpp>
20#include <userver/storages/postgres/options.hpp>
21#include <userver/storages/postgres/query.hpp>
22#include <userver/storages/postgres/statistics.hpp>
23#include <userver/storages/postgres/transaction.hpp>
24
25/// @page pg_topology uPg: Cluster topology discovery
26///
27/// @par Principles of PgaaS role determination
28/// - Every host except master is in recovery state from PostgreSQL's POV.
29/// This means the check 'select pg_is_in_recovery()' returns `false` for the
30/// master and `true` for every other host type.
31/// - Some hosts are in sync slave mode. This may be determined by executing
32/// 'show synchronous_standby_names' on the master.
33/// See
34/// https://www.postgresql.org/docs/current/runtime-config-replication.html#GUC-SYNCHRONOUS-STANDBY-NAMES
35/// for more information.
36///
37/// @par PgaaS sync slaves lag
38/// By default, PgaaS synchronous slaves are working with 'synchronous_commit'
39/// set to 'remote_apply'. Therefore, sync slave may be lagging behind the
40/// master and thus is not truly 'synchronous' from the reader's POV,
41/// but things may change with time.
42///
43/// @par Implementation
44/// Topology update runs every second.
45///
46/// Every host is assigned a connection with special ID (4100200300).
47/// Using this connection we check for host availability, writability
48/// (master detection) and perform RTT measurements.
49///
50/// After the initial check we know about master presence and RTT for each host.
51/// Master host is queried about synchronous replication status. We use this
52/// info to identify synchronous slaves and to detect "quorum commit" presence.
53///
54///
55/// ----------
56///
57/// @htmlonly <div class="bottom-nav"> @endhtmlonly
58/// ⇦ @ref pg_errors | @ref scripts/docs/en/userver/pg_connlimit_mode_auto.md ⇨
59/// @htmlonly </div> @endhtmlonly
60
61USERVER_NAMESPACE_BEGIN
62
63namespace components {
64class Postgres;
65} // namespace components
66
67namespace storages::postgres {
68
69namespace detail {
70
71class ClusterImpl;
72using ClusterImplPtr = std::unique_ptr<ClusterImpl>;
73
74} // namespace detail
75
76/// @ingroup userver_clients
77///
78/// @brief Interface for executing queries on a cluster of PostgreSQL servers
79///
80/// See @ref pg_user_row_types "Typed PostgreSQL results" for usage examples of
81/// the storages::postgres::ResultSet.
82///
83/// Usually retrieved from components::Postgres component.
84///
85/// @todo Add information about topology discovery
86class Cluster {
87 public:
88 /// Cluster constructor
89 /// @param dsns List of DSNs to connect to
90 /// @param resolver asynchronous DNS resolver
91 /// @param bg_task_processor task processor for blocking connection operations
92 /// @param cluster_settings struct with settings fields:
93 /// task_data_keys_settings - settings for per-handler command controls
94 /// topology_settings - settings for host discovery
95 /// pool_settings - settings for connection pools
96 /// conn_settings - settings for individual connections
97 /// @param default_cmd_ctls default command execution options
98 /// @param testsuite_pg_ctl command execution options customizer for testsuite
99 /// @param ei_settings error injection settings
100 /// @note When `max_connection_pool_size` is reached, and no idle connections
101 /// available, `PoolError` is thrown for every new connection
102 /// request
103 Cluster(DsnList dsns, clients::dns::Resolver* resolver,
104 engine::TaskProcessor& bg_task_processor,
105 const ClusterSettings& cluster_settings,
106 DefaultCommandControls&& default_cmd_ctls,
107 const testsuite::PostgresControl& testsuite_pg_ctl,
108 const error_injection::Settings& ei_settings,
109 testsuite::TestsuiteTasks& testsuite_tasks,
110 dynamic_config::Source config_source, int shard_number);
111 ~Cluster();
112
113 /// Get cluster statistics
114 ///
115 /// The statistics object is too big to fit on stack
117
118 /// @name Transaction start
119 /// @{
120
121 /// Start a transaction in any available connection depending on transaction
122 /// options.
123 ///
124 /// If the transaction is RW, will start transaction in a connection
125 /// to master. If the transaction is RO, will start trying connections
126 /// starting with slaves.
127 /// @throws ClusterUnavailable if no hosts are available
128 Transaction Begin(const TransactionOptions&, OptionalCommandControl = {});
129
130 /// Start a transaction in a connection with specified host selection rules.
131 ///
132 /// If the requested host role is not available, may fall back to another
133 /// host role, see ClusterHostType.
134 /// If the transaction is RW, only master connection can be used.
135 /// @throws ClusterUnavailable if no hosts are available
136 Transaction Begin(ClusterHostTypeFlags, const TransactionOptions&,
137 OptionalCommandControl = {});
138
139 /// Start a named transaction in any available connection depending on
140 /// transaction options.
141 ///
142 /// If the transaction is RW, will start transaction in a connection
143 /// to master. If the transaction is RO, will start trying connections
144 /// starting with slaves.
145 /// `name` is used to set command control in config at runtime.
146 /// @throws ClusterUnavailable if no hosts are available
147 Transaction Begin(std::string name, const TransactionOptions&);
148
149 /// Start a named transaction in a connection with specified host selection
150 /// rules.
151 ///
152 /// If the requested host role is not available, may fall back to another
153 /// host role, see ClusterHostType.
154 /// If the transaction is RW, only master connection can be used.
155 /// `name` is used to set command control in config at runtime.
156 /// @throws ClusterUnavailable if no hosts are available
157 Transaction Begin(std::string name, ClusterHostTypeFlags,
158 const TransactionOptions&);
159 /// @}
160
161 /// @name Single-statement query in an auto-commit transaction
162 /// @{
163
164 /// @brief Execute a statement at host of specified type.
165 /// @note You must specify at least one role from ClusterHostType here
166 ///
167 /// @snippet storages/postgres/tests/landing_test.cpp Exec sample
168 ///
169 /// @warning Do NOT create a query string manually by embedding arguments!
170 /// It leads to vulnerabilities and bad performance. Either pass arguments
171 /// separately, or use storages::postgres::ParameterScope.
172 template <typename... Args>
173 ResultSet Execute(ClusterHostTypeFlags, const Query& query,
174 const Args&... args);
175
176 /// @brief Execute a statement with specified host selection rules and command
177 /// control settings.
178 /// @note You must specify at least one role from ClusterHostType here
179 ///
180 /// @warning Do NOT create a query string manually by embedding arguments!
181 /// It leads to vulnerabilities and bad performance. Either pass arguments
182 /// separately, or use storages::postgres::ParameterScope.
183 template <typename... Args>
184 ResultSet Execute(ClusterHostTypeFlags, OptionalCommandControl,
185 const Query& query, const Args&... args);
186
187 /// @brief Execute a statement with stored arguments and specified host
188 /// selection rules.
189 ResultSet Execute(ClusterHostTypeFlags flags, const Query& query,
190 const ParameterStore& store);
191
192 /// @brief Execute a statement with stored arguments, specified host selection
193 /// rules and command control settings.
194 ResultSet Execute(ClusterHostTypeFlags flags,
195 OptionalCommandControl statement_cmd_ctl,
196 const Query& query, const ParameterStore& store);
197 /// @}
198
199 /// @brief Listen for notifications on channel
200 /// @warning Each NotifyScope owns a single connection taken from the pool,
201 /// which effectively decreases the number of usable connections
202 NotifyScope Listen(std::string_view channel, OptionalCommandControl = {});
203
204 /// Replaces globally updated command control with a static user-provided one
206
207 /// Returns current default command control
209
210 void SetHandlersCommandControl(
211 CommandControlByHandlerMap handlers_command_control);
212
213 void SetQueriesCommandControl(
214 CommandControlByQueryMap queries_command_control);
215
216 /// @cond
217 /// Updates default command control from global config (if not set by user)
218 void ApplyGlobalCommandControlUpdate(CommandControl);
219 /// @endcond
220
221 /// Replaces cluster connection settings.
222 ///
223 /// Connections with an old settings will be dropped and reestablished.
225
226 void SetPoolSettings(const PoolSettings& settings);
227
228 void SetTopologySettings(const TopologySettings& settings);
229
230 void SetStatementMetricsSettings(const StatementMetricsSettings& settings);
231
232 private:
233 detail::NonTransaction Start(ClusterHostTypeFlags, OptionalCommandControl);
234
235 OptionalCommandControl GetQueryCmdCtl(const std::string& query_name) const;
236 OptionalCommandControl GetHandlersCmdCtl(
237 OptionalCommandControl cmd_ctl) const;
238
239 detail::ClusterImplPtr pimpl_;
240};
241
242template <typename... Args>
243ResultSet Cluster::Execute(ClusterHostTypeFlags flags, const Query& query,
244 const Args&... args) {
245 return Execute(flags, OptionalCommandControl{}, query, args...);
246}
247
248template <typename... Args>
249ResultSet Cluster::Execute(ClusterHostTypeFlags flags,
250 OptionalCommandControl statement_cmd_ctl,
251 const Query& query, const Args&... args) {
252 if (!statement_cmd_ctl && query.GetName()) {
253 statement_cmd_ctl = GetQueryCmdCtl(query.GetName()->GetUnderlying());
254 }
255 statement_cmd_ctl = GetHandlersCmdCtl(statement_cmd_ctl);
256 auto ntrx = Start(flags, statement_cmd_ctl);
257 return ntrx.Execute(statement_cmd_ctl, query, args...);
258}
259
260} // namespace storages::postgres
261
262USERVER_NAMESPACE_END