Problem rozwiązany ! Nie mogę uwierzyć, że spędziłem nad tym całe dwa dni... Patrzyłem całkowicie w złym kierunku.
Problem nie dotyczył niektórych konfiguracji sieci Dataflow lub GCP i o ile wiem...
to prawda.
Problem był oczywiście w moim kodzie:tylko problem został ujawniony tylko w środowisku rozproszonym. Popełniłem błąd, otwierając tunel z głównego procesora rurociągu, a nie z pracowników. Tak więc tunel SSH działał, ale nie między procesami roboczymi a serwerem docelowym, tylko między głównym potokiem a docelowym!
Aby to naprawić, musiałem zmienić moje żądanie DoFn, aby otoczyć wykonanie zapytania tunelem:
class TunnelledSQLSourceDoFn(sql.SQLSourceDoFn):
"""Wraps SQLSourceDoFn in a ssh tunnel"""
def __init__(self, *args, **kwargs):
self.dbport = kwargs["port"]
self.dbhost = kwargs["host"]
self.args = args
self.kwargs = kwargs
super().__init__(*args, **kwargs)
def process(self, query, *args, **kwargs):
# Remote side of the SSH Tunnel
remote_address = (self.dbhost, self.dbport)
ssh_tunnel = (self.kwargs['ssh_host'], self.kwargs['ssh_port'])
with open_tunnel(
ssh_tunnel,
ssh_username=self.kwargs["ssh_user"],
ssh_password=self.kwargs["ssh_password"],
remote_bind_address=remote_address,
set_keepalive=10.0
) as tunnel:
forwarded_port = tunnel.local_bind_port
self.kwargs["port"] = forwarded_port
source = sql.SQLSource(*self.args, **self.kwargs)
sql.SQLSouceInput._build_value(source, source.runtime_params)
logging.info("Processing - {}".format(query))
for records, schema in source.client.read(query):
for row in records:
yield source.client.row_as_dict(row, schema)
jak widać, musiałem nadpisać niektóre fragmenty biblioteki pysql_beam.
Na koniec każdy pracownik otwiera własny tunel dla każdego żądania. Prawdopodobnie można zoptymalizować to zachowanie, ale to wystarcza na moje potrzeby.
