[PLAT-14531] Connection manager for k8s

Summary:
Adds support to enable/disable connection pooling when creating a k8s universe or later via the configure YSQL API.

How to enable/disable connection pooling and the API payloads are the same as in this diff: https://phorge.dev.yugabyte.com/D37470

Note:
We can only use the hardcoded YSQL (5433) and internal YSQL ports (6433) for now. We don't have support to customise ports for k8s universes yet.

Test Plan:
Manually tested the following flows:
1. Created a normal k8s universe with no YSQL auth, enabled connection pooling via the configure YSQL API.
2. Created a universe with connection pooling enabled, ensured connection manger is up and running.
3. Created a universe with YSQL auth and connection pooling, ensured it works.
4. Created. a universe with no YSQL auth and no connection pooling, enabled connection pooling and YSQL auth via the configure YSQL API. Ensured it works correctly.

Apart from this,
Run UTs.
Run itests.

Reviewers: anijhawan, sneelakantan, vkumar

Reviewed By: anijhawan

Subscribers: svc_phabricator, yugaware

Differential Revision: https://phorge.dev.yugabyte.com/D38409

managed/src/main/java/com/yugabyte/yw/commissioner/HealthChecker.java

@@ -750,10 +750,12 @@ public void checkSingleUniverse(CheckSingleUniverseParams params) {
           nodeInfo.setSslProtocol(tserverGflags.get("ssl_protocols"));
         }
         if (nodeInfo.enableYSQL && nodeDetails.isYsqlServer) {
-          nodeInfo.setYsqlPort(nodeDetails.ysqlServerRpcPort);
+          nodeInfo.setYsqlPort(
+              params.universe.getUniverseDetails().communicationPorts.ysqlServerRpcPort);
           nodeInfo.setYsqlServerHttpPort(nodeDetails.ysqlServerHttpPort);
           if (nodeInfo.enableConnectionPooling) {
-            nodeInfo.setInternalYsqlPort(nodeDetails.internalYsqlServerRpcPort);
+            nodeInfo.setInternalYsqlPort(
+                params.universe.getUniverseDetails().communicationPorts.internalYsqlServerRpcPort);
           }
         }
         if (nodeInfo.enableYCQL && nodeDetails.isYqlServer) {

managed/src/main/java/com/yugabyte/yw/commissioner/tasks/subtasks/CreateTable.java

@@ -96,9 +96,10 @@ private void createPgSqlTable() {
     boolean tableCreated = false;
     int attempt = 0;
     Instant timeout = Instant.now().plusSeconds(TOTAL_ATTEMPTS_DURATION_SEC);
+    int internalYsqlServerRpcPort =
+        universe.getUniverseDetails().communicationPorts.internalYsqlServerRpcPort;
     while (Instant.now().isBefore(timeout) || attempt < MIN_RETRY_COUNT) {
       NodeDetails randomTServer = CommonUtils.getARandomLiveTServer(universe);
-      int internalYsqlServerRpcPort = randomTServer.internalYsqlServerRpcPort;
       ShellResponse response =
           nodeUniverseManager.runYsqlCommand(
               randomTServer,

managed/src/main/java/com/yugabyte/yw/commissioner/tasks/subtasks/KubernetesCommandExecutor.java

@@ -87,6 +87,9 @@ public class KubernetesCommandExecutor extends UniverseTaskBase {
   private static final List<CommandType> skipNamespaceCommands =
       Arrays.asList(CommandType.POD_INFO, CommandType.COPY_PACKAGE, CommandType.YBC_ACTION);
 
+  public static final int DEFAULT_YSQL_SERVER_RPC_PORT = 5433;
+  public static final int DEFAULT_INTERNAL_YSQL_SERVER_RPC_PORT = 6433;
+
   public enum CommandType {
     CREATE_NAMESPACE,
     APPLY_SECRET,
@@ -1155,6 +1158,28 @@ private String generateHelmOverride() {
         .enableYSQL) { // In the UI, we can choose not to show these entries for read replica.
       tserverGFlags.put("enable_ysql", "false");
     }
+
+    if (primaryClusterIntent.enableYSQL) {
+      // For now, set a default value for the ysql server rpc port.
+      // TO DO:
+      // If the user passed in a custom value, use it as an override.
+      int ysqlServerRpcPort = DEFAULT_YSQL_SERVER_RPC_PORT;
+      if (primaryClusterIntent.enableConnectionPooling) {
+        // For now, set a default value for the internal ysql server rpc port.
+        // TO DO:
+        // Check if the user passed in a value for the internal ysql server rpc port.
+        // If given, use it as an override.
+        int internalYsqlServerRpcPort = DEFAULT_INTERNAL_YSQL_SERVER_RPC_PORT;
+        tserverGFlags.put("enable_ysql_conn_mgr", "true");
+        tserverGFlags.put("allowed_preview_flags_csv", "enable_ysql_conn_mgr");
+        tserverGFlags.put("ysql_conn_mgr_port", String.valueOf(ysqlServerRpcPort));
+        tserverGFlags.put(
+            "pgsql_proxy_bind_address",
+            (primaryClusterIntent.enableIPV6 ? "[::]" : "0.0.0.0")
+                + String.valueOf(internalYsqlServerRpcPort));
+      }
+    }
+
     if (!primaryClusterIntent.enableYCQL) {
       tserverGFlags.put("start_cql_proxy", "false");
     }

managed/src/main/java/com/yugabyte/yw/commissioner/tasks/upgrade/ConfigureDBApisKubernetes.java

@@ -81,6 +81,7 @@ public void run() {
           createUpdateDBApiDetailsTask(
                   taskParams().enableYSQL,
                   taskParams().enableYSQLAuth,
+                  taskParams().enableConnectionPooling,
                   taskParams().enableYCQL,
                   taskParams().enableYCQLAuth)
               .setSubTaskGroupType(getTaskSubGroupType());
@@ -116,6 +117,7 @@ protected void syncTaskParamsToUserIntent() {
               cluster -> {
                 cluster.userIntent.enableYSQL = taskParams().enableYSQL;
                 cluster.userIntent.enableYSQLAuth = taskParams().enableYSQLAuth;
+                cluster.userIntent.enableConnectionPooling = taskParams().enableConnectionPooling;
                 cluster.userIntent.enableYCQL = taskParams().enableYCQL;
                 cluster.userIntent.enableYCQLAuth = taskParams().enableYCQLAuth;
               });

managed/src/main/java/com/yugabyte/yw/controllers/handlers/UniverseCRUDHandler.java

@@ -32,6 +32,7 @@
 import com.yugabyte.yw.commissioner.tasks.UniverseTaskBase;
 import com.yugabyte.yw.commissioner.tasks.UniverseTaskBase.ServerType;
 import com.yugabyte.yw.commissioner.tasks.XClusterConfigTaskBase;
+import com.yugabyte.yw.commissioner.tasks.subtasks.KubernetesCommandExecutor;
 import com.yugabyte.yw.common.AppConfigHelper;
 import com.yugabyte.yw.common.ImageBundleUtil;
 import com.yugabyte.yw.common.KubernetesManagerFactory;
@@ -857,9 +858,18 @@ public UniverseResp createUniverse(Customer customer, UniverseDefinitionTaskPara
               BAD_REQUEST, "YSQL RPC port cannot be the same as internal YSQL RPC port");
         }
 
-        if (Common.CloudType.kubernetes.equals(userIntent.providerType)) {
-          throw new PlatformServiceException(
-              BAD_REQUEST, "Connection pooling is not yet supported for kubernetes universes.");
+        if (userIntent.providerType.equals(Common.CloudType.kubernetes)) {
+          if (taskParams.communicationPorts.ysqlServerRpcPort
+              != KubernetesCommandExecutor.DEFAULT_YSQL_SERVER_RPC_PORT) {
+            throw new PlatformServiceException(
+                BAD_REQUEST, "Custom YSQL RPC port is not yet supported for Kubernetes universes.");
+          }
+          if (taskParams.communicationPorts.internalYsqlServerRpcPort
+              != KubernetesCommandExecutor.DEFAULT_INTERNAL_YSQL_SERVER_RPC_PORT) {
+            throw new PlatformServiceException(
+                BAD_REQUEST,
+                "Custom Internal YSQL RPC port is not yet supported for Kubernetes universes.");
+          }
         }
       }
 

managed/src/main/java/com/yugabyte/yw/controllers/handlers/UniverseYbDbAdminHandler.java

@@ -20,7 +20,7 @@
 import com.typesafe.config.Config;
 import com.yugabyte.yw.commissioner.Commissioner;
 import com.yugabyte.yw.commissioner.Common;
-import com.yugabyte.yw.commissioner.Common.CloudType;
+import com.yugabyte.yw.commissioner.tasks.subtasks.KubernetesCommandExecutor;
 import com.yugabyte.yw.common.ConfigHelper;
 import com.yugabyte.yw.common.PlatformServiceException;
 import com.yugabyte.yw.common.Util;
@@ -243,10 +243,23 @@ public UUID configureYSQL(
                 CONNECTION_POOLING_STABLE_VERSION, CONNECTION_POOLING_PREVIEW_VERSION));
       }
 
-      if (CloudType.kubernetes.equals(
-          universe.getUniverseDetails().getPrimaryCluster().userIntent.providerType)) {
-        throw new PlatformServiceException(
-            BAD_REQUEST, "Connection pooling is not yet supported for kubernetes universes.");
+      if (universe
+          .getUniverseDetails()
+          .getPrimaryCluster()
+          .userIntent
+          .providerType
+          .equals(Common.CloudType.kubernetes)) {
+        if (requestParams.communicationPorts.ysqlServerRpcPort
+            != KubernetesCommandExecutor.DEFAULT_YSQL_SERVER_RPC_PORT) {
+          throw new PlatformServiceException(
+              BAD_REQUEST, "Custom YSQL RPC port is not yet supported for Kubernetes universes.");
+        }
+        if (requestParams.communicationPorts.internalYsqlServerRpcPort
+            != KubernetesCommandExecutor.DEFAULT_INTERNAL_YSQL_SERVER_RPC_PORT) {
+          throw new PlatformServiceException(
+              BAD_REQUEST,
+              "Custom Internal YSQL RPC port is not yet supported for Kubernetes universes.");
+        }
       }
     }
     // Verify request params

managed/src/main/java/com/yugabyte/yw/forms/UniverseTaskParams.java

@@ -131,6 +131,7 @@ public static void mergeCommunicationPorts(
         CommunicationPorts ports, ConfigureDBApiParams params) {
       ports.ysqlServerHttpPort = params.communicationPorts.ysqlServerHttpPort;
       ports.ysqlServerRpcPort = params.communicationPorts.ysqlServerRpcPort;
+      ports.internalYsqlServerRpcPort = params.communicationPorts.internalYsqlServerRpcPort;
       ports.yqlServerHttpPort = params.communicationPorts.yqlServerHttpPort;
       ports.yqlServerRpcPort = params.communicationPorts.yqlServerRpcPort;
     }