feat postgresql: set application_name in pg connection

commit_hash:2f2343a69963b45718bb2a11838b32d76c0ee708

Author: romankoshelev

postgresql/include/userver/storages/postgres/options.hpp

@@ -296,6 +296,8 @@ struct ConnectionSettings {
     /// Helps keep track of the changes in settings
     SettingsVersion version{0U};
 
+    std::optional<std::string> application_name{};
+
     bool operator==(const ConnectionSettings& rhs) const {
         return !RequiresConnectionReset(rhs) && recent_errors_threshold == rhs.recent_errors_threshold;
     }
@@ -308,7 +310,7 @@ struct ConnectionSettings {
                ignore_unused_query_params != rhs.ignore_unused_query_params ||
                max_prepared_cache_size != rhs.max_prepared_cache_size || pipeline_mode != rhs.pipeline_mode ||
                max_ttl != rhs.max_ttl || discard_on_connect != rhs.discard_on_connect ||
-               omit_describe_mode != rhs.omit_describe_mode;
+               omit_describe_mode != rhs.omit_describe_mode || application_name != rhs.application_name;
     }
 };
 

postgresql/src/storages/postgres/component.yaml

@@ -146,3 +146,6 @@ properties:
           - auto
           - manual
         description: how to learn the `max_pool_size`
+    application_name:
+        type: string
+        description: will be displayed in postgres statistics and logs

postgresql/src/storages/postgres/detail/connection_impl.cpp

@@ -315,7 +315,13 @@ void ConnectionImpl::AsyncConnect(const Dsn& dsn, engine::Deadline deadline) {
     // some allowance.
     auto timeout = std::chrono::duration_cast<std::chrono::milliseconds>(deadline.TimeLeft());
     deadline = testsuite_pg_ctl_.MakeExecuteDeadline(timeout);
-    conn_wrapper_.AsyncConnect(dsn, deadline, scope);
+    const std::string client_encoding = "UTF8";
+    const PGConnectionWrapper::ConnectParams start_params{
+        /* .dsn = */ dsn,
+        /* .application_name = */ settings_.application_name,
+        /* .client_encoding = */ client_encoding,
+    };
+    conn_wrapper_.AsyncConnect(start_params, deadline, scope);
     conn_wrapper_.FillSpanTags(span, {timeout, GetStatementTimeout()});
 
     const auto session_id = ExecuteCommandNoPrepare(kGetSessionIdQuery, deadline).AsSingleRow<std::string>();
@@ -325,8 +331,16 @@ void ConnectionImpl::AsyncConnect(const Dsn& dsn, engine::Deadline deadline) {
     // We cannot handle exceptions here, so we let them got to the caller
     if (settings_.discard_on_connect == ConnectionSettings::kDiscardAll) {
         ExecuteCommandNoPrepare("DISCARD ALL", deadline);
+        SetParameter("client_encoding", client_encoding, Connection::ParameterScope::kSession, deadline);
+        if (start_params.application_name) {
+            SetParameter(
+                "application_name",
+                *start_params.application_name,
+                Connection::ParameterScope::kSession,
+                deadline
+            );
+        }
     }
-    SetParameter("client_encoding", "UTF8", Connection::ParameterScope::kSession, deadline);
     RefreshReplicaState(deadline);
     SetConnectionStatementTimeout(GetDefaultCommandControl().statement_timeout_ms, deadline);
     if (settings_.user_types != ConnectionSettings::kPredefinedTypesOnly) {

postgresql/src/storages/postgres/detail/pg_connection_wrapper.cpp

@@ -1,5 +1,7 @@
 #include <storages/postgres/detail/pg_connection_wrapper.hpp>
 
+#include <boost/container/small_vector.hpp>
+
 #include <pg_config.h>
 
 #ifndef USERVER_NO_LIBPQ_PATCHES
@@ -297,49 +299,65 @@ engine::Task PGConnectionWrapper::Cancel() {
     });
 }
 
-void PGConnectionWrapper::AsyncConnect(const Dsn& dsn, Deadline deadline, tracing::ScopeTime& scope) {
-    PGCW_LOG_DEBUG() << "Connecting to " << DsnCutPassword(dsn);
+void PGConnectionWrapper::AsyncConnect(const ConnectParams& params, Deadline deadline, tracing::ScopeTime& scope) {
+    PGCW_LOG_DEBUG() << "Connecting to " << DsnCutPassword(params.dsn);
 
-    auto options = OptionsFromDsn(dsn);
+    auto options = OptionsFromDsn(params.dsn);
     log_extra_.Extend(tracing::kDatabaseInstance, std::move(options.dbname));
     log_extra_.Extend(tracing::kPeerAddress, std::move(options.host) + ':' + options.port);
 
     scope.Reset(scopes::kLibpqConnect);
-    StartAsyncConnect(dsn);
+    StartAsyncConnect(params);
     scope.Reset(scopes::kLibpqWaitConnectFinish);
-    WaitConnectionFinish(deadline, dsn);
-    PGCW_LOG_DEBUG() << "Connected to " << DsnCutPassword(dsn);
+    WaitConnectionFinish(deadline, params.dsn);
+    PGCW_LOG_DEBUG() << "Connected to " << DsnCutPassword(params.dsn);
 }
 
-void PGConnectionWrapper::StartAsyncConnect(const Dsn& dsn) {
+void PGConnectionWrapper::StartAsyncConnect(const ConnectParams& params) {
     if (conn_) {
         PGCW_LOG_LIMITED_ERROR()
             << "Attempt to connect a connection that is already connected" << logging::LogExtra::Stacktrace();
-        throw ConnectionFailed{dsn, "Already connected"};
+        throw ConnectionFailed{params.dsn, "Already connected"};
     }
 
     // PQconnectStart() may access /etc/hosts, ~/.pgpass, /etc/passwd, etc.
-    engine::CriticalAsyncNoSpan(bg_task_processor_, [&dsn, this] {
-        conn_ = PQconnectStart(dsn.GetUnderlying().c_str());
+    engine::CriticalAsyncNoSpan(bg_task_processor_, [&params, this] {
+        boost::container::small_vector<const char*, 8> keywords{"dbname"};
+        boost::container::small_vector<const char*, 8> values{params.dsn.GetUnderlying().c_str()};
+
+        if (params.application_name) {
+            keywords.push_back("application_name");
+            values.push_back(params.application_name->c_str());
+        }
+
+        if (params.client_encoding) {
+            keywords.push_back("client_encoding");
+            values.push_back(params.client_encoding->c_str());
+        }
+
+        keywords.push_back(nullptr);
+        values.push_back(nullptr);
+
+        conn_ = PQconnectStartParams(keywords.data(), values.data(), true);
     }).Get();
 
     if (!conn_) {
         // The only reason the pointer cannot be null is that libpq failed
         // to allocate memory for the structure
         PGCW_LOG_LIMITED_ERROR() << "libpq failed to allocate a PGconn structure" << logging::LogExtra::Stacktrace();
-        throw ConnectionFailed{dsn, "Failed to allocate PGconn structure"};
+        throw ConnectionFailed{params.dsn, "Failed to allocate PGconn structure"};
     }
 
     const auto status = PQstatus(conn_);
     if (CONNECTION_BAD == status) {
-        const std::string msg = MsgForStatus(status);
+        const std::string msg = fmt::format("{}: {}", MsgForStatus(status), PQerrorMessage(conn_));
         PGCW_LOG_WARNING() << msg;
-        CloseWithError(ConnectionFailed{dsn, msg});
+        CloseWithError(ConnectionFailed{params.dsn, msg});
     } else {
         PGCW_LOG_TRACE() << MsgForStatus(status);
     }
 
-    RefreshSocket(dsn);
+    RefreshSocket(params.dsn);
 
     // set this as early as possible to avoid dumping notices to stderr
     PQsetNoticeReceiver(conn_, &NoticeReceiver, this);

postgresql/src/storages/postgres/detail/pg_connection_wrapper.hpp

@@ -30,6 +30,13 @@ class PGConnectionWrapper {
     using Duration = Deadline::TimePoint::clock::duration;
     using ResultHandle = detail::ResultWrapper::ResultHandle;
 
+    // https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-PARAMKEYWORDS
+    struct ConnectParams final {
+        Dsn dsn;
+        std::optional<std::string> application_name;
+        std::optional<std::string> client_encoding;
+    };
+
     PGConnectionWrapper(
         engine::TaskProcessor& tp,
         concurrent::BackgroundTaskStorageCore& bts,
@@ -53,7 +60,7 @@ class PGConnectionWrapper {
     ///
     /// Start asynchronous connection and wait for it's completion (suspending
     /// current coroutine)
-    void AsyncConnect(const Dsn& dsn, Deadline deadline, tracing::ScopeTime&);
+    void AsyncConnect(const ConnectParams& params, Deadline deadline, tracing::ScopeTime&);
 
     /// @brief Causes a connection to enter pipeline mode.
     ///
@@ -154,7 +161,7 @@ class PGConnectionWrapper {
 private:
     PGTransactionStatusType GetTransactionStatus() const;
 
-    void StartAsyncConnect(const Dsn& dsn);
+    void StartAsyncConnect(const ConnectParams& params);
 
     /// @throws ConnectionTimeoutError if was awakened by the deadline
     void WaitConnectionFinish(Deadline deadline, const Dsn& dsn);

postgresql/src/storages/postgres/postgres_config.cpp

@@ -7,6 +7,7 @@
 #include <storages/postgres/experiments.hpp>
 #include <userver/storages/postgres/component.hpp>
 #include <userver/storages/postgres/exceptions.hpp>
+#include <userver/utils/userver_info.hpp>
 
 #include <userver/formats/common/items.hpp>
 
@@ -97,6 +98,8 @@ ConnectionSettings ParseConnectionSettings(const ConfigType& config) {
             ? ConnectionSettings::kDiscardAll
             : ConnectionSettings::kDiscardNone;
     settings.deadline_propagation_enabled = config["deadline-propagation-enabled"].template As<bool>(true);
+    settings.application_name =
+        config["application_name"].template As<std::string>(USERVER_NAMESPACE::utils::GetUserverIdentifier());
 
     return settings;
 }