Use different application names for db connection to show context

Adds a "context" parameter to database connection setup describing what
this connection is for. This is something like "middle.query" or
"out.pgsql". It will be added to the application name, so querying
pg_stat_activity will show it like "osm2pgsql.middle.query/C2" etc.

Author: joto

src/db-copy.cpp

@@ -128,7 +128,7 @@ db_copy_thread_t::thread_t::thread_t(connection_params_t connection_params,
 void db_copy_thread_t::thread_t::operator()()
 {
     try {
-        m_conn = std::make_unique<pg_conn_t>(m_connection_params);
+        m_conn = std::make_unique<pg_conn_t>(m_connection_params, "copy");
 
         // Let commits happen faster by delaying when they actually occur.
         m_conn->exec("SET synchronous_commit = off");

src/expire-output.cpp

@@ -58,7 +58,7 @@ std::size_t expire_output_t::output_tiles_to_table(
 {
     auto const qn = qualified_name(m_schema, m_table);
 
-    pg_conn_t connection{connection_params};
+    pg_conn_t connection{connection_params, "expire"};
 
     auto const result = connection.exec("SELECT * FROM {} LIMIT 1", qn);
 

src/flex-table.cpp

@@ -245,7 +245,8 @@ void table_connection_t::connect(connection_params_t const &connection_params)
 {
     assert(!m_db_connection);
 
-    m_db_connection = std::make_unique<pg_conn_t>(connection_params);
+    m_db_connection =
+        std::make_unique<pg_conn_t>(connection_params, "out.flex.table");
     m_db_connection->exec("SET synchronous_commit = off");
 }
 

src/gen/osm2pgsql-gen.cpp

@@ -205,7 +205,7 @@ void run_tile_gen(connection_params_t const &connection_params,
 
     log_debug("Started generalizer thread for '{}'.",
               master_generalizer->strategy());
-    pg_conn_t db_connection{connection_params};
+    pg_conn_t db_connection{connection_params, "gen.tile"};
     std::string const strategy{master_generalizer->strategy()};
     auto generalizer = create_generalizer(strategy, &db_connection, &params);
 
@@ -286,7 +286,7 @@ class genproc_t
         write_to_debug_log(params, "Params (config):");
 
         log_debug("Connecting to database...");
-        pg_conn_t db_connection{m_connection_params};
+        pg_conn_t db_connection{m_connection_params, "gen.proc"};
 
         log_debug("Creating generalizer...");
         auto generalizer =
@@ -368,7 +368,7 @@ class genproc_t
             queries.emplace_back("COMMIT");
         }
 
-        pg_conn_t const db_connection{m_connection_params};
+        pg_conn_t const db_connection{m_connection_params, "gen.sql"};
 
         if (m_append && !if_has_rows.empty()) {
             auto const result = db_connection.exec(if_has_rows);
@@ -592,7 +592,7 @@ void genproc_t::run()
     }
 
     if (!m_append) {
-        pg_conn_t const db_connection{m_connection_params};
+        pg_conn_t const db_connection{m_connection_params, "gen.index"};
         for (auto const &table : m_tables) {
             if (table.id_type() == flex_table_index_type::tile &&
                 (table.always_build_id_index() || m_updatable)) {
@@ -702,7 +702,7 @@ int main(int argc, char *argv[])
 
         log_debug("Checking database capabilities...");
         {
-            pg_conn_t const db_connection{connection_params};
+            pg_conn_t const db_connection{connection_params, "gen.check"};
             init_database_capabilities(db_connection);
         }
 

src/middle-pgsql.cpp

@@ -168,7 +168,7 @@ void middle_pgsql_t::table_desc::build_index(
 
     // Use a temporary connection here because we might run in a separate
     // thread context.
-    pg_conn_t const db_connection{connection_params};
+    pg_conn_t const db_connection{connection_params, "middle.index"};
 
     log_info("Building index on table '{}'", name());
     for (auto const &query : m_create_fw_dep_indexes) {
@@ -1300,7 +1300,7 @@ middle_query_pgsql_t::middle_query_pgsql_t(
     std::shared_ptr<node_locations_t> cache,
     std::shared_ptr<node_persistent_cache> persistent_cache,
     middle_pgsql_options const &options)
-: m_sql_conn(connection_params), m_cache(std::move(cache)),
+: m_sql_conn(connection_params, "middle.query"), m_cache(std::move(cache)),
   m_persistent_cache(std::move(persistent_cache)), m_store_options(options)
 {
     // Disable JIT and parallel workers as they are known to cause
@@ -1649,7 +1649,7 @@ middle_pgsql_t::middle_pgsql_t(std::shared_ptr<thread_pool_t> thread_pool,
 : middle_t(std::move(thread_pool)), m_options(options),
   m_cache(std::make_unique<node_locations_t>(
       static_cast<std::size_t>(options->cache) * 1024UL * 1024UL)),
-  m_db_connection(m_options->connection_params),
+  m_db_connection(m_options->connection_params, "middle.main"),
   m_copy_thread(std::make_shared<db_copy_thread_t>(options->connection_params)),
   m_db_copy(m_copy_thread), m_append(options->append)
 {

src/osm2pgsql.cpp

@@ -86,7 +86,7 @@ static file_info run(options_t const &options)
 
 static void check_db(options_t const &options)
 {
-    pg_conn_t const db_connection{options.connection_params};
+    pg_conn_t const db_connection{options.connection_params, "check"};
 
     init_database_capabilities(db_connection);
 

src/output-flex.cpp

@@ -1241,7 +1241,7 @@ create_expire_tables(std::vector<expire_output_t> const &expire_outputs,
         return;
     }
 
-    pg_conn_t const connection{connection_params};
+    pg_conn_t const connection{connection_params, "out.flex.expire"};
     for (auto const &expire_output : expire_outputs) {
         if (!expire_output.table().empty()) {
             expire_output.create_output_table(connection);

src/output-gazetteer.cpp

@@ -49,7 +49,7 @@ void output_gazetteer_t::start()
     if (!get_options()->append) {
         int const srid = get_options()->projection->target_srs();
 
-        pg_conn_t const conn{get_options()->connection_params};
+        pg_conn_t const conn{get_options()->connection_params, "out.gazetteer"};
 
         /* Drop any existing table */
         conn.exec("DROP TABLE IF EXISTS place CASCADE");

src/pgsql.cpp

@@ -29,7 +29,7 @@ std::size_t pg_result_t::affected_rows() const noexcept
 std::atomic<std::uint32_t> pg_conn_t::connection_id{0};
 
 static PGconn *open_connection(connection_params_t const &connection_params,
-                               std::uint32_t id)
+                               std::string_view context, std::uint32_t id)
 {
     std::vector<char const *> keywords;
     std::vector<char const *> values;
@@ -39,7 +39,7 @@ static PGconn *open_connection(connection_params_t const &connection_params,
         values.push_back(v.c_str());
     }
 
-    std::string const app_name{fmt::format("osm2pgsql/C{}", id)};
+    std::string const app_name{fmt::format("osm2pgsql.{}/C{}", context, id)};
     keywords.push_back("fallback_application_name");
     values.push_back(app_name.c_str());
 
@@ -49,23 +49,25 @@ static PGconn *open_connection(connection_params_t const &connection_params,
     return PQconnectdbParams(keywords.data(), values.data(), 1);
 }
 
-pg_conn_t::pg_conn_t(connection_params_t const &connection_params)
+pg_conn_t::pg_conn_t(connection_params_t const &connection_params,
+                     std::string_view context)
 : m_connection_id(connection_id.fetch_add(1))
 {
-    m_conn.reset(open_connection(connection_params, m_connection_id));
+    m_conn.reset(open_connection(connection_params, context, m_connection_id));
 
     if (!m_conn) {
-        throw std::runtime_error{"Connecting to database failed."};
+        throw fmt_error("Connecting to database failed (context={}).", context);
     }
 
     if (PQstatus(m_conn.get()) != CONNECTION_OK) {
-        throw fmt_error("Connecting to database failed: {}.", error_msg());
+        throw fmt_error("Connecting to database failed (context={}): {}.",
+                        context, error_msg());
     }
 
     if (get_logger().log_sql()) {
         auto const results = exec("SELECT pg_backend_pid()");
-        log_sql("(C{}) New database connection (backend_pid={})",
-                m_connection_id, results.get(0, 0));
+        log_sql("(C{}) New database connection (context={}, backend_pid={})",
+                m_connection_id, context, results.get(0, 0));
     }
 
     // PostgreSQL sends notices in many different contexts which aren't that

src/pgsql.hpp

@@ -152,7 +152,8 @@ class binary_param : public std::string_view
 class pg_conn_t
 {
 public:
-    explicit pg_conn_t(connection_params_t const &connection_params);
+    explicit pg_conn_t(connection_params_t const &connection_params,
+                       std::string_view context);
 
     /**
      * Run the specified SQL command.

src/properties.cpp

@@ -92,7 +92,7 @@ void properties_t::set_string(std::string property, std::string value,
     auto const &inserted = *(r.first);
     log_debug("  Storing {}='{}'", inserted.first, inserted.second);
 
-    pg_conn_t const db_connection{m_connection_params};
+    pg_conn_t const db_connection{m_connection_params, "prop.set"};
     db_connection.exec(
         "PREPARE set_property(text, text) AS"
         " INSERT INTO {} (property, value) VALUES ($1, $2)"
@@ -119,7 +119,7 @@ void properties_t::store()
     auto const table = table_name();
 
     log_info("Storing properties to table '{}'.", table);
-    pg_conn_t const db_connection{m_connection_params};
+    pg_conn_t const db_connection{m_connection_params, "prop.store"};
 
     if (m_has_properties_table) {
         db_connection.exec("TRUNCATE {}", table);
@@ -153,7 +153,7 @@ bool properties_t::load()
     auto const table = table_name();
     log_info("Loading properties from table '{}'.", table);
 
-    pg_conn_t const db_connection{m_connection_params};
+    pg_conn_t const db_connection{m_connection_params, "prop.load"};
     auto const result = db_connection.exec("SELECT * FROM {}", table);
 
     for (int i = 0; i < result.num_tuples(); ++i) {

src/table.cpp

@@ -66,7 +66,7 @@ void table_t::sync() { m_copy.sync(); }
 
 void table_t::connect()
 {
-    m_sql_conn = std::make_unique<pg_conn_t>(m_connection_params);
+    m_sql_conn = std::make_unique<pg_conn_t>(m_connection_params, "out.pgsql");
     //let commits happen faster by delaying when they actually occur
     m_sql_conn->exec("SET synchronous_commit = off");
 }

tests/common-pg.hpp

@@ -38,7 +38,7 @@ class conn_t : public pg_conn_t
 {
 public:
     conn_t(connection_params_t const &connection_params)
-    : pg_conn_t(connection_params)
+    : pg_conn_t(connection_params, "test")
     {
     }