add pgport to command line args and to port allocator (#7528)

Author: rekby

ydb/core/config/init/init_impl.h

@@ -315,6 +315,8 @@ struct TCommonAppOptions {
     TString GRpcPublicHost = "";
     ui32 GRpcPublicPort = 0;
     ui32 GRpcsPublicPort = 0;
+    TString PGWireAddress = "";
+    ui32 PGWirePort = 0;
     TVector<TString> GRpcPublicAddressesV4;
     TVector<TString> GRpcPublicAddressesV6;
     TString GRpcPublicTargetNameOverride = "";
@@ -326,7 +328,7 @@ struct TCommonAppOptions {
     bool SysLogEnabled = false;
     bool TcpEnabled = false;
     bool SuppressVersionCheck = false;
-    EWorkload Workload = EWorkload::Hybrid; 
+    EWorkload Workload = EWorkload::Hybrid;
 
     void RegisterCliOptions(NLastGetopt::TOpts& opts) {
         opts.AddLongOption("cluster-name", "which cluster this node belongs to")
@@ -383,6 +385,8 @@ struct TCommonAppOptions {
         opts.AddLongOption("grpc-public-host", "set public gRPC host for discovery").RequiredArgument("HOST").StoreResult(&GRpcPublicHost);
         opts.AddLongOption("grpc-public-port", "set public gRPC port for discovery").RequiredArgument("PORT").StoreResult(&GRpcPublicPort);
         opts.AddLongOption("grpcs-public-port", "set public gRPC SSL port for discovery").RequiredArgument("PORT").StoreResult(&GRpcsPublicPort);
+        opts.AddLongOption("pgwire-address", "set host for listen postgres protocol").RequiredArgument("ADDR").StoreResult(&PGWireAddress);
+        opts.AddLongOption("pgwire-port", "set port for listen postgres protocol").OptionalArgument("PORT").StoreResult(&PGWirePort);
         opts.AddLongOption("grpc-public-address-v4", "set public ipv4 address for discovery").RequiredArgument("ADDR").EmplaceTo(&GRpcPublicAddressesV4);
         opts.AddLongOption("grpc-public-address-v6", "set public ipv6 address for discovery").RequiredArgument("ADDR").EmplaceTo(&GRpcPublicAddressesV6);
         opts.AddLongOption("grpc-public-target-name-override", "set public hostname override for TLS in discovery").RequiredArgument("HOST").StoreResult(&GRpcPublicTargetNameOverride);
@@ -496,7 +500,7 @@ struct TCommonAppOptions {
             if (offset) {
                 connectorConfig.MutableEndpoint()->Setport(InterconnectPort + offset) ;
 
-                // Assign default hostname 'localhost', because 
+                // Assign default hostname 'localhost', because
                 // connector is usually deployed to the same host as the dynamic node.
                 if (connectorConfig.GetEndpoint().host().Empty()) {
                     connectorConfig.MutableEndpoint()->Sethost("localhost");
@@ -597,6 +601,12 @@ struct TCommonAppOptions {
             }
             ConfigUpdateTracer.AddUpdate(NKikimrConsole::TConfigItem::GRpcConfigItem, TConfigItemInfo::EUpdateKind::UpdateExplicitly);
         }
+        if (PGWireAddress) {
+            appConfig.MutableLocalPgWireConfig()->SetAddress(PGWireAddress);
+        }
+        if (PGWirePort) {
+            appConfig.MutableLocalPgWireConfig()->SetListeningPort(PGWirePort);
+        }
         for (const auto& addr : GRpcPublicAddressesV4) {
             appConfig.MutableGRpcConfig()->AddPublicAddressesV4(addr);
         }
@@ -658,7 +668,7 @@ struct TCommonAppOptions {
                     ApplyDontStartGrpcProxy(*appConfig.MutableGRpcConfig(), ConfigUpdateTracer);
                     break;
                 case EWorkload::Hybrid:
-                    // default, do nothing 
+                    // default, do nothing
                     break;
             }
         }

ydb/tests/functional/postgresql/test_postgres.py

@@ -59,8 +59,6 @@ def execute_binary(binary_name, cmd, stdin_string=None):
 class BasePostgresTest(object):
     @classmethod
     def setup_class(cls):
-        cls.pm = yatest.common.network.PortManager()
-        cls.pgport = cls.pm.get_port()
         cls.cluster = kikimr_cluster_factory(KikimrConfigGenerator(
             additional_log_configs={
                 'LOCAL_PGWIRE': LogLevels.DEBUG,
@@ -70,14 +68,14 @@ def setup_class(cls):
                 'KQP_PROXY': LogLevels.DEBUG
             },
             extra_feature_flags=['enable_table_pg_types', 'enable_temp_tables'],
-            pgwire_port=cls.pgport
         ))
         cls.cluster.start()
 
+        cls.pgport = cls.cluster.nodes[1].pgwire_port
+
     @classmethod
     def teardown_class(cls):
         cls.cluster.stop()
-        cls.pm.release()
 
 
 class TestPostgresSuite(BasePostgresTest):

ydb/tests/library/harness/kikimr_config.py

@@ -165,7 +165,6 @@ def __init__(
             default_user_sid=None,
             pg_compatible_expirement=False,
             generic_connector_config=None,  # typing.Optional[TGenericConnectorConfig]
-            pgwire_port=None,
     ):
         if extra_feature_flags is None:
             extra_feature_flags = []
@@ -262,10 +261,6 @@ def __init__(
             self.yaml_config["local_pg_wire_config"] = {}
             self.yaml_config["local_pg_wire_config"]["listening_port"] = os.getenv('PGWIRE_LISTENING_PORT')
 
-        if pgwire_port:
-            self.yaml_config["local_pg_wire_config"] = {}
-            self.yaml_config["local_pg_wire_config"]["listening_port"] = pgwire_port
-
         if disable_iterator_reads:
             self.yaml_config["table_service_config"]["enable_kqp_scan_query_source_read"] = False
 
@@ -400,10 +395,8 @@ def __init__(
             if not "local_pg_wire_config" in self.yaml_config:
                 self.yaml_config["local_pg_wire_config"] = {}
 
-            ydb_pg_port=5432
-            if 'YDB_PG_PORT' in os.environ:
-                ydb_pg_port = os.environ['YDB_PG_PORT']
-            self.yaml_config['local_pg_wire_config']['listening_port'] = ydb_pg_port
+            ydb_pgwire_port=self.port_allocator.get_node_port_allocator(node_id).pgwire_port
+            self.yaml_config['local_pg_wire_config']['listening_port'] = ydb_pgwire_port
 
             # https://github.com/ydb-platform/ydb/issues/5152
             # self.yaml_config["table_service_config"]["enable_pg_consts_to_params"] = True

ydb/tests/library/harness/kikimr_port_allocator.py

@@ -84,6 +84,7 @@ def __init__(self, port_manager):
         self.__grpc_ssl_port = None
         self.__ext_port = None
         self.__public_http_port = None
+        self.__pgwire_port = None
 
     @property
     def mon_port(self):
@@ -127,6 +128,12 @@ def ext_port(self):
             self.__ext_port = self.__port_manager.get_port()
         return self.__ext_port
 
+    @property
+    def pgwire_port(self):
+        if self.__pgwire_port is None:
+            self.__pgwire_port = self.__port_manager.get_port()
+        return self.__pgwire_port
+
     @property
     def public_http_port(self):
         if self.__public_http_port is None:
@@ -162,7 +169,7 @@ def release_ports(self):
 class KikimrFixedNodePortAllocator(KikimrNodePortAllocatorInterface):
 
     def __init__(self, base_port_offset, mon_port=8765, grpc_port=2135, mbus_port=2134, ic_port=19001, sqs_port=8771, grpc_ssl_port=2137,
-                 ext_port=2237, public_http_port=8766):
+                 ext_port=2237, public_http_port=8766, pgwire_port=5432):
         super(KikimrFixedNodePortAllocator, self).__init__()
 
         self.base_port_offset = base_port_offset
@@ -192,6 +199,10 @@ def __init__(self, base_port_offset, mon_port=8765, grpc_port=2135, mbus_port=21
             self.__public_http_port = int(os.getenv('PUBLIC_HTTP_PORT'))
         else:
             self.__public_http_port = public_http_port
+        if os.getenv('YDB_PGWIRE_PORT') is not None:
+            self.__pgwire_port = int(os.getenv('YDB_PGWIRE_PORT'))
+        else:
+            self.__pgwire_port = pgwire_port
 
     @property
     def mon_port(self):
@@ -224,6 +235,10 @@ def ext_port(self):
     def public_http_port(self):
         return self.__public_http_port + self.base_port_offset
 
+    @property
+    def pgwire_port(self):
+        return self.__pgwire_port + self.base_port_offset
+
 
 class KikimrFixedPortAllocator(KikimrPortAllocatorInterface):
     def __init__(self,

ydb/tests/library/harness/kikimr_runner.py

@@ -68,6 +68,7 @@ def __init__(self, node_id, config_path, port_allocator, cluster_name, configura
         self.mon_port = port_allocator.mon_port
         self.ic_port = port_allocator.ic_port
         self.grpc_ssl_port = port_allocator.grpc_ssl_port
+        self.pgwire_port = port_allocator.pgwire_port
         self.sqs_port = None
         if configurator.sqs_service_enabled:
             self.sqs_port = port_allocator.sqs_port
@@ -177,6 +178,7 @@ def __make_run_command(self):
                 "--grpc-port=%s" % self.grpc_port,
                 "--mon-port=%d" % self.mon_port,
                 "--ic-port=%d" % self.ic_port,
+                "--pgwire-port=%d" % self.pgwire_port,
             ]
         )