Problém je vyřešen ! Nemůžu uvěřit, že jsem nad tím strávil celé dva dny... Díval jsem se úplně špatným směrem.
Problém nebyl s nějakou konfigurací sítě Dataflow nebo GCP, a pokud mohu říci...
je pravda.
Problém byl samozřejmě v mém kódu:problém byl odhalen pouze v distribuovaném prostředí. Udělal jsem chybu, že jsem tunel otevřel z hlavního potrubního procesoru místo dělníků. Tunel SSH byl spuštěn, ale ne mezi pracovníky a cílovým serverem, pouze mezi hlavním potrubím a cílem!
Abych to vyřešil, musel jsem změnit svůj požadující DoFn tak, aby zalomil provádění dotazu 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 vidíte, musel jsem přepsat některé části knihovny pysql_beam.
Nakonec každý pracovník otevře svůj vlastní tunel pro každý požadavek. Pravděpodobně je možné toto chování optimalizovat, ale pro mé potřeby to stačí.