refactor(inkless): add read/write pg connection configs

docs/inkless/configs.rst

@@ -265,6 +265,110 @@ Under ``inkless.control.plane.``
 
 
 
+-----------------
+PostgresControlPlaneConfig - read overrides
+-----------------
+Under ``inkless.control.plane..read.``
+
+``connection.string``
+  PostgreSQL connection string
+
+  * Type: string
+  * Valid Values: non-empty string
+  * Importance: high
+
+``username``
+  Username
+
+  * Type: string
+  * Valid Values: non-empty string
+  * Importance: high
+
+``password``
+  Password
+
+  * Type: password
+  * Default: null
+  * Importance: high
+
+``file.merge.lock.period.ms``
+  The period of time when the file merge job is locked (assumed being performed).
+
+  * Type: long
+  * Default: 3600000 (1 hour)
+  * Valid Values: [1,...]
+  * Importance: medium
+
+``file.merge.size.threshold.bytes``
+  The total minimum volume of files to be merged together.
+
+  * Type: long
+  * Default: 104857600 (100 mebibytes)
+  * Valid Values: [1,...]
+  * Importance: medium
+
+``max.connections``
+  Maximum number of connections to the database
+
+  * Type: int
+  * Default: 10
+  * Valid Values: [1,...]
+  * Importance: medium
+
+
+
+-----------------
+PostgresControlPlaneConfig - write overrides
+-----------------
+Under ``inkless.control.plane..write.``
+
+``connection.string``
+  PostgreSQL connection string
+
+  * Type: string
+  * Valid Values: non-empty string
+  * Importance: high
+
+``username``
+  Username
+
+  * Type: string
+  * Valid Values: non-empty string
+  * Importance: high
+
+``password``
+  Password
+
+  * Type: password
+  * Default: null
+  * Importance: high
+
+``file.merge.lock.period.ms``
+  The period of time when the file merge job is locked (assumed being performed).
+
+  * Type: long
+  * Default: 3600000 (1 hour)
+  * Valid Values: [1,...]
+  * Importance: medium
+
+``file.merge.size.threshold.bytes``
+  The total minimum volume of files to be merged together.
+
+  * Type: long
+  * Default: 104857600 (100 mebibytes)
+  * Valid Values: [1,...]
+  * Importance: medium
+
+``max.connections``
+  Maximum number of connections to the database
+
+  * Type: int
+  * Default: 10
+  * Valid Values: [1,...]
+  * Importance: medium
+
+
+
 -----------------
 AzureBlobStorageConfig
 -----------------

storage/inkless/src/main/java/io/aiven/inkless/control_plane/ControlPlane.java

@@ -85,5 +85,6 @@ static ControlPlane create(final InklessConfig config, final Time time) {
 
     boolean isSafeToDeleteFile(String objectKeyPath);
 
+    // used for testing purposes only
     List<GetLogInfoResponse> getLogInfo(List<GetLogInfoRequest> requests);
 }

storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/HikariMetricsTracker.java

@@ -18,19 +18,14 @@
 package io.aiven.inkless.control_plane.postgres;
 
 import org.apache.kafka.common.MetricNameTemplate;
-import org.apache.kafka.common.metrics.JmxReporter;
-import org.apache.kafka.common.metrics.KafkaMetricsContext;
-import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.stats.Avg;
 import org.apache.kafka.common.metrics.stats.Max;
-import org.apache.kafka.common.utils.Time;
 
 import com.zaxxer.hikari.metrics.IMetricsTracker;
 import com.zaxxer.hikari.metrics.PoolStats;
 
-import java.util.List;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Supplier;
 
@@ -43,7 +38,6 @@
 import static io.aiven.inkless.control_plane.postgres.HikariMetricsRegistry.CONNECTION_USAGE_MILLIS;
 import static io.aiven.inkless.control_plane.postgres.HikariMetricsRegistry.IDLE_CONNECTIONS_COUNT;
 import static io.aiven.inkless.control_plane.postgres.HikariMetricsRegistry.MAX_CONNECTIONS_COUNT;
-import static io.aiven.inkless.control_plane.postgres.HikariMetricsRegistry.METRIC_CONTEXT;
 import static io.aiven.inkless.control_plane.postgres.HikariMetricsRegistry.MIN_CONNECTIONS_COUNT;
 import static io.aiven.inkless.control_plane.postgres.HikariMetricsRegistry.PENDING_THREADS_COUNT;
 import static io.aiven.inkless.control_plane.postgres.HikariMetricsRegistry.TOTAL_CONNECTIONS_COUNT;
@@ -62,12 +56,8 @@ public class HikariMetricsTracker implements IMetricsTracker {
     private final Sensor pendingThreadsCountSensor;
     private final Sensor connectionTimeoutCountSensor;
 
-    public HikariMetricsTracker(final String poolName, final PoolStats poolStats) {
-        final JmxReporter reporter = new JmxReporter();
-        this.metrics = new Metrics(
-            new MetricConfig(), List.of(reporter), Time.SYSTEM,
-            new KafkaMetricsContext(METRIC_CONTEXT)
-        );
+    public HikariMetricsTracker(final Metrics metrics, final String poolName, final PoolStats poolStats) {
+        this.metrics = metrics;
         this.metricsRegistry = new HikariMetricsRegistry(poolName);
 
         activeConnectionsCountSensor = registerSensor(metrics, metricsRegistry.activeConnectionsCountMetricName, ACTIVE_CONNECTIONS_COUNT, () -> (long) poolStats.getActiveConnections());

storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/PostgresConnectionConfig.java

@@ -0,0 +1,96 @@
+/*
+ * Inkless
+ * Copyright (C) 2024 - 2025 Aiven OY
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package io.aiven.inkless.control_plane.postgres;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.types.Password;
+
+import java.util.Map;
+
+import io.aiven.inkless.control_plane.AbstractControlPlaneConfig;
+
+public class PostgresConnectionConfig extends AbstractControlPlaneConfig {
+    public static final String CONNECTION_STRING_CONFIG = "connection.string";
+    private static final String CONNECTION_STRING_DOC = "PostgreSQL connection string";
+
+    public static final String USERNAME_CONFIG = "username";
+    private static final String USERNAME_DOC = "Username";
+
+    public static final String PASSWORD_CONFIG = "password";
+    private static final String PASSWORD_DOC = "Password";
+
+    public static final String MAX_CONNECTIONS_CONFIG = "max.connections";
+    private static final String MAX_CONNECTIONS_DOC = "Maximum number of connections to the database";
+
+    public static ConfigDef configDef() {
+        return baseConfigDef()
+            .define(
+                CONNECTION_STRING_CONFIG,
+                ConfigDef.Type.STRING,
+                ConfigDef.NO_DEFAULT_VALUE,
+                new ConfigDef.NonEmptyString(),
+                ConfigDef.Importance.HIGH,
+                CONNECTION_STRING_DOC
+            )
+            .define(
+                USERNAME_CONFIG,
+                ConfigDef.Type.STRING,
+                ConfigDef.NO_DEFAULT_VALUE,
+                new ConfigDef.NonEmptyString(),
+                ConfigDef.Importance.HIGH,
+                USERNAME_DOC
+            )
+            .define(
+                PASSWORD_CONFIG,
+                ConfigDef.Type.PASSWORD,
+                null,
+                null,  // can be empty
+                ConfigDef.Importance.HIGH,
+                PASSWORD_DOC
+            )
+            .define(
+                MAX_CONNECTIONS_CONFIG,
+                ConfigDef.Type.INT,
+                10,
+                ConfigDef.Range.atLeast(1),
+                ConfigDef.Importance.MEDIUM,
+                MAX_CONNECTIONS_DOC
+            );
+    }
+
+    public PostgresConnectionConfig(final ConfigDef configDef, final Map<?, ?> originals) {
+        super(configDef, originals);
+    }
+
+    public String connectionString() {
+        return getString(CONNECTION_STRING_CONFIG);
+    }
+
+    public String username() {
+        return getString(USERNAME_CONFIG);
+    }
+
+    public String password() {
+        final Password configValue = getPassword(PASSWORD_CONFIG);
+        return configValue == null ? null : configValue.value();
+    }
+
+    public int maxConnections() {
+        return getInt(MAX_CONNECTIONS_CONFIG);
+    }
+}