Make database thread pool size configurable (close #60)

Author: spenes

config/config.kinesis.reference.hocon

@@ -51,6 +51,9 @@
       # Maximum number of connections database pool is allowed to reach
       # Default is 10
       "maxConnections": 10
+      # Size of the thread pool for blocking database operations
+      # Default is value of "maxConnections"
+      "threadPoolSize": 10
     }
 
     # Events that fail validation are written to specified stream.

config/config.local.reference.hocon

@@ -30,6 +30,9 @@
       # Maximum number of connections database pool is allowed to reach
       # Default is 10
       "maxConnections": 10
+      # Size of the thread pool for blocking database operations
+      # Default is value of "maxConnections"
+      "threadPoolSize": 10
     }
 
     # Events that fail validation are written to specified stream.

config/config.pubsub.reference.hocon

@@ -37,6 +37,9 @@
       # Maximum number of connections database pool is allowed to reach
       # Default is 10
       "maxConnections": 10
+      # Size of the thread pool for blocking database operations
+      # Default is value of "maxConnections"
+      "threadPoolSize": 10
     }
 
     # Events that fail validation are written to specified stream.

modules/common/src/main/scala/com/snowplowanalytics/snowplow/postgres/config/DBConfig.scala

@@ -24,7 +24,8 @@ case class DBConfig(host: String,
                     password: String, // TODO: can be EC2 store
                     sslMode: String,
                     schema: String,
-                    maxConnections: Option[Int]
+                    maxConnections: Int,
+                    threadPoolSize: Option[Int]
 ) {
   def getJdbc: JdbcUri =
     JdbcUri(host, port, database, sslMode.toLowerCase().replace('_', '-'))
@@ -46,7 +47,7 @@ object DBConfig {
     config.setJdbcUrl(dbConfig.getJdbc.toString)
     config.setUsername(dbConfig.username)
     config.setPassword(dbConfig.password)
-    dbConfig.maxConnections.foreach(config.setMaximumPoolSize)
+    config.setMaximumPoolSize(dbConfig.maxConnections)
     config
   }
 

modules/common/src/main/scala/com/snowplowanalytics/snowplow/postgres/resources.scala

@@ -38,7 +38,7 @@ object resources {
   def initialize[F[_]: Concurrent: Clock: ContextShift](postgres: DBConfig, iglu: Client[F, Json]) =
     for {
       blocker <- Blocker[F]
-      xa <- resources.getTransactor[F](DBConfig.hikariConfig(postgres), blocker)
+      xa <- resources.getTransactor[F](DBConfig.hikariConfig(postgres), blocker, postgres.threadPoolSize)
       state <- Resource.eval(initializeState(postgres.schema, iglu, xa))
     } yield (blocker, xa, state)
 
@@ -58,14 +58,8 @@ object resources {
     } yield state
 
   /** Get a HikariCP transactor */
-  def getTransactor[F[_]: Async: ContextShift](config: HikariConfig, be: Blocker): Resource[F, HikariTransactor[F]] = {
-    val threadPoolSize = {
-      // This could be made configurable, but these are sensible defaults and unlikely to be critical for tuning throughput.
-      // Exceeding availableProcessors could lead to unnecessary context switching.
-      // Exceeding the connection pool size is unnecessary, because that is limit of the app's parallelism.
-      val maxPoolSize = if (config.getMaximumPoolSize > 0) config.getMaximumPoolSize else 10
-      Math.min(maxPoolSize, Runtime.getRuntime.availableProcessors)
-    }
+  def getTransactor[F[_]: Async: ContextShift](config: HikariConfig, be: Blocker, threadPoolSizeOpt: Option[Int] = None): Resource[F, HikariTransactor[F]] = {
+    val threadPoolSize = threadPoolSizeOpt.getOrElse(config.getMaximumPoolSize)
     logger.debug(s"Using thread pool of size $threadPoolSize for Hikari transactor")
 
     for {

modules/loader/src/test/scala/com.snowplowanalytics.snowplow.postgres/config/CliSpec.scala

@@ -54,7 +54,8 @@ class CliSpec extends Specification {
             "mysecretpassword",
             "REQUIRE",
             "atomic",
-            Some(10)
+            10,
+            None
           ),
           LoaderConfig.StreamSink.Noop
         ),
@@ -91,6 +92,7 @@ class CliSpec extends Specification {
             "mysecretpassword",
             "REQUIRE",
             "atomic",
+            10,
             Some(10)
           ),
           LoaderConfig.StreamSink.Kinesis(
@@ -131,7 +133,8 @@ class CliSpec extends Specification {
             "mysecretpassword",
             "REQUIRE",
             "atomic",
-            Some(10)
+            10,
+            None
           ),
           LoaderConfig.StreamSink.Noop
         ),
@@ -165,6 +168,7 @@ class CliSpec extends Specification {
             "mysecretpassword",
             "REQUIRE",
             "atomic",
+            10,
             Some(10)
           ),
           LoaderConfig.StreamSink.PubSub(
@@ -202,7 +206,8 @@ class CliSpec extends Specification {
             "mysecretpassword",
             "REQUIRE",
             "atomic",
-            Some(10)
+            10,
+            None
           ),
           LoaderConfig.StreamSink.Noop
         ),
@@ -232,6 +237,7 @@ class CliSpec extends Specification {
             "mysecretpassword",
             "REQUIRE",
             "atomic",
+            10,
             Some(10)
           ),
           LoaderConfig.StreamSink.Local(PathInfo(Path.fromString("./tmp/bad").get, PathType.Relative))
@@ -262,7 +268,8 @@ class CliSpec extends Specification {
             "mysecretpassword",
             "REQUIRE",
             "atomic",
-            Some(10)
+            10,
+            None
           ),
           LoaderConfig.StreamSink.Noop
         ),

modules/loader/src/test/scala/com.snowplowanalytics.snowplow.postgres/config/LocalSourceSpec.scala

@@ -54,6 +54,7 @@ class LocalSourceSpec extends Database {
             "mysecretpassword",
             "allow",
             "public",
+            10,
             None
           ),
           LoaderConfig.StreamSink.Noop