Fixes #8396 "Postgres sync not respecting SSH tunneling" (#13391)

* Use db pool connection rather than raw details

connection pool is correctly setup with ssh tunneling information (if
any)

* Don't error when getting postgres enum types

this was throwing errors when we allowed our pooled connections to
close underneath us. But this particular step just isn't important
enough for that. If we don't have a connection available to us, let
the more robust query stuff fail it rather than this one off

* Set idleConnectionTestPeriod to 120 when ssh tunneling

The ssh tunnelling apparatus needs some work. The tunnels are created
and put in the spec details map but then this is tossed away when we
create the pool.

```clojure
(defn- create-pool!
  "Create a new C3P0 `ComboPooledDataSource` for connecting to the given `database`."
  [{:keys [id details], driver :engine, :as database}]
  {:pre [(map? database)]}
  (log/debug (u/format-color 'cyan (trs "Creating new connection pool for {0} database {1} ..." driver id)))
  (let [details-with-tunnel (ssh/include-ssh-tunnel details) ;; If the tunnel is disabled this returned unchanged
        spec                (connection-details->spec driver details-with-tunnel)
        properties          (data-warehouse-connection-pool-properties driver)]
    (connection-pool/connection-pool-spec spec (merge properties
                                                      (when (ssh/use-ssh-tunnel? details)
                                                        {"idleConnectionTestPeriod" 120})))))

```

this is swapped into an atom of {db_id -> pooleddatasource} but this
means we don't have the :tunnel-session and :tunnel-tracker info from
`start-ssh-tunnel`.

* Remove connection pool keys when set to nil

our connection pool would end up with {db_id nil ...} for no
reason. Add tests to ensure.

This change is largely driven by the codox testing
requirement. There's no good way to test that the ssh tunneling
options includes

(when (ssh/use-ssh-tunnel? details) {"idleConnectionTestPeriod" 120})

and the test coverage checker was unhappy

* Correct ns form in test file

* Remove try/catch around enum types

there's already error handling around this sync step so let it
happen. The issue was that this check didn't use the correct
connection to respect tunnelling and therefore would throw an error in
getting enum types and fail the database sync step. That is now done
correctly so if this step fails the other steps would as well and we
no longer need to handle an error here specifically

* Move ssh heartbeat into ssh.clj and not use jdbc

jdbc had a way to test idle connections but other non-jdbc datasources
obviously wouldn't benefit. Luckily our ssh connection can handle this
on its own

src/metabase/driver/postgres.clj

@@ -98,14 +98,14 @@
   [_]
   :monday)
 
-(defn- enum-types [driver database]
+(defn- enum-types [_driver database]
   (set
-   (map (comp keyword :typname)
-        (jdbc/query (sql-jdbc.conn/connection-details->spec driver (:details database))
-                    [(str "SELECT DISTINCT t.typname "
-                          "FROM pg_enum e "
-                          "LEFT JOIN pg_type t "
-                          "  ON t.oid = e.enumtypid")]))))
+    (map (comp keyword :typname)
+         (jdbc/query (sql-jdbc.conn/db->pooled-connection-spec database)
+                     [(str "SELECT DISTINCT t.typname "
+                           "FROM pg_enum e "
+                           "LEFT JOIN pg_type t "
+                           "  ON t.oid = e.enumtypid")]))))
 
 (def ^:private ^:dynamic *enum-types* nil)
 

src/metabase/driver/sql_jdbc/connection.clj

@@ -104,7 +104,9 @@
   more than one pool is ever open for a single database."
   [database-id pool-spec-or-nil]
   {:pre [(integer? database-id)]}
-  (let [[old-id->pool] (swap-vals! database-id->connection-pool assoc database-id pool-spec-or-nil)]
+  (let [[old-id->pool] (if pool-spec-or-nil
+                         (swap-vals! database-id->connection-pool assoc database-id pool-spec-or-nil)
+                         (swap-vals! database-id->connection-pool dissoc database-id))]
     ;; if we replaced a different pool with the new pool that is different from the old one, destroy the old pool
     (when-let [old-pool-spec (get old-id->pool database-id)]
       (when-not (identical? old-pool-spec pool-spec-or-nil)

src/metabase/util/ssh.clj

@@ -2,12 +2,16 @@
   (:require [clojure.tools.logging :as log]
             [metabase.util :as u])
   (:import java.io.ByteArrayInputStream
+           java.util.concurrent.TimeUnit
            org.apache.sshd.client.future.ConnectFuture
            org.apache.sshd.client.session.ClientSession
            org.apache.sshd.client.session.forward.PortForwardingTracker
            org.apache.sshd.client.SshClient
            [org.apache.sshd.common.config.keys FilePasswordProvider FilePasswordProvider$ResourceDecodeResult]
-           org.apache.sshd.common.session.SessionHolder
+           [org.apache.sshd.common.session
+            SessionHeartbeatController
+            SessionHeartbeatController$HeartbeatType
+            SessionHolder]
            org.apache.sshd.common.util.GenericUtils
            org.apache.sshd.common.util.io.resource.AbstractIoResource
            org.apache.sshd.common.util.net.SshdSocketAddress
@@ -50,6 +54,9 @@
         session                    (doto ^ClientSession (.getSession conn-status)
                                      (maybe-add-tunnel-password! tunnel-pass)
                                      (maybe-add-tunnel-private-key! tunnel-private-key tunnel-private-key-passphrase)
+                                     (.setSessionHeartbeat SessionHeartbeatController$HeartbeatType/IGNORE
+                                                           TimeUnit/SECONDS
+                                                           180)
                                      (.. auth (verify default-ssh-timeout)))
         tracker                    (.createLocalPortForwardingTracker session
                                                                       (SshdSocketAddress. "" 0)

test/metabase/driver/postgres_test.clj

@@ -377,7 +377,8 @@
   (create-enums-db!)
   (mt/with-temp Database [database {:engine :postgres, :details (enums-test-db-details)}]
     (sync-metadata/sync-db-metadata! database)
-    (f database)))
+    (f database)
+    (#'sql-jdbc.conn/set-pool! (u/id database) nil)))
 
 (deftest enums-test
   (mt/test-driver :postgres

test/metabase/driver/sql_jdbc/connection_test.clj

 (ns metabase.driver.sql-jdbc.connection-test
-  (:require [expectations :refer [expect]]
+  (:require [clojure.java.jdbc :as jdbc]
+            [clojure.test :refer :all]
+            [expectations :refer [expect]]
+            [metabase
+             [db :as mdb]
+             [test :as mt]
+             [util :as u]]
+            [metabase.driver.sql-jdbc.connection :as sql-jdbc.conn]
             [metabase.driver.util :as driver.u]
+            [metabase.models.database :refer [Database]]
             [metabase.test.data :as data]
             [metabase.test.util.log :as tu.log]))
 
@@ -28,3 +36,37 @@
   false
   (tu.log/suppress-output
     (driver.u/can-connect-with-details? :postgres {:host "google.com", :port 80})))
+
+(deftest db->pooled-connection-spec-test
+  (mt/test-driver :h2
+    (testing "creating and removing specs works"
+      ;; need to create a new, nonexistent h2 db
+      (binding [mdb/*allow-potentailly-unsafe-connections* true]
+        (let [destroyed?       (atom false)
+              original-destroy @#'sql-jdbc.conn/destroy-pool!
+              spec             (sql-jdbc.conn/connection-details->spec
+                                   :h2
+                                 (mt/dbdef->connection-details :h2 :server {:database-name "connection_test"}))]
+          (with-redefs [sql-jdbc.conn/destroy-pool! (fn [id destroyed-spec]
+                                                      (original-destroy id destroyed-spec)
+                                                      (reset! destroyed? true))]
+            (jdbc/with-db-connection [conn spec]
+              (jdbc/execute! spec ["CREATE TABLE birds (name varchar)"])
+              (jdbc/execute! spec ["INSERT INTO birds values ('rasta'),('lucky')"])
+              (mt/with-temp Database [database {:engine :h2 :details spec}]
+                (testing "database id is not in our connection map initially"
+                  ;; deref'ing a var to get the atom. looks weird
+                  (is (not (contains? @@#'sql-jdbc.conn/database-id->connection-pool
+                                      (u/id database)))))
+                (testing "when getting a pooled connection it is now in our connection map"
+                  (let [stored-spec (sql-jdbc.conn/db->pooled-connection-spec database)
+                        birds       (jdbc/query stored-spec ["SELECT * FROM birds"])]
+                    (is (seq birds))
+                    (is (contains? @@#'sql-jdbc.conn/database-id->connection-pool
+                                   (u/id database)))))
+                (testing "and is no longer in our connection map after cleanup"
+                  (#'sql-jdbc.conn/set-pool! (u/id database) nil)
+                  (is (not (contains? @@#'sql-jdbc.conn/database-id->connection-pool
+                                      (u/id database)))))
+                (testing "the pool has been destroyed"
+                  (is @destroyed?))))))))))