Updates to fix ZMQ with ports

Author: awicenec

daliuge-common/dlg/utils.py

@@ -141,7 +141,7 @@ def zmq_safe(host_or_addr):
     # The catch-all IP address, ZMQ needs a *
     if host_or_addr == "0.0.0.0":
         return "*"
-
+    host_or_addr = host_or_addr.split(":")[0]
     # Return otherwise always an IP address
     return socket.gethostbyname(host_or_addr)
 

daliuge-engine/dlg/graph_loader.py

@@ -221,7 +221,10 @@ def loadDropSpecs(dropSpecList):
                 # relationship list but doesn't exist in the list of DROPs
                 for oid in dropSpec[rel]:
                     oid = list(oid.keys())[0] if isinstance(oid, dict) else oid
-                    dropSpecs[oid]
+                    if oid in dropSpecs:
+                        dropSpecs[oid]
+                    else:
+                        continue
 
             # N-1 relationships
             elif rel in __TOONE:
@@ -277,7 +280,10 @@ def createGraphFromDropSpecList(dropSpecList, session=None):
                 link = __TOMANY[attr]
                 for rel in dropSpec[attr]:
                     oid = list(rel.keys())[0] if isinstance(rel, dict) else rel
-                    lhDrop = drops[oid]
+                    if oid in drops:
+                        lhDrop = drops[oid]
+                    else:
+                        continue
                     relFuncName = LINKTYPE_1TON_APPEND_METHOD[link]
                     try:
                         relFunc = getattr(drop, relFuncName)

daliuge-engine/dlg/manager/node_manager.py

@@ -435,6 +435,8 @@ def add_node_subscriptions(self, sessionId, relationships):
                 host, events_port, _ = nodesub
 
             # TODO: we also have to unsubscribe from them at some point
+            host = host.split(":")[0]
+            logger.debug("Sending subscription to %s", f"{host}:{events_port}")
             self.subscribe(host, events_port)
 
     def has_method(self, sessionId, uid, mname):

daliuge-engine/dlg/manager/session.py

@@ -287,6 +287,7 @@ def addGraphSpec(self, graphSpec):
         self.status = SessionStates.BUILDING
 
         # This will check the consistency of each dropSpec
+        logger.debug("Trying to add graphSpec: %s", [[x['oid'],x['node']] for x in graphSpec])
         graphSpecDict, self._graphreprodata = graph_loader.loadDropSpecs(
             graphSpec
         )

daliuge-engine/dlg/rpc.py

@@ -55,6 +55,8 @@ class RPCClientBase(RPCObject):
 
     def get_drop_attribute(self, hostname, port, session_id, uid, name):
 
+        hostname = hostname.split(":")[0]
+
         logger.debug(
             "Getting attribute %s for drop %s of session %s at %s:%d",
             name,
@@ -63,6 +65,7 @@ def get_drop_attribute(self, hostname, port, session_id, uid, name):
             hostname,
             port,
         )
+        hostname = hostname.split(":")[0]
 
         client, closer = self.get_rpc_client(hostname, port)
 
@@ -93,7 +96,7 @@ class RPCServerBase(RPCObject):
     """Base class for all RPC server"""
 
     def __init__(self, host, port):
-        self._rpc_host = host
+        self._rpc_host = host.split(":")[0]
         self._rpc_port = port
 
 
@@ -137,10 +140,11 @@ def shutdown(self):
 
     def get_client_for_endpoint(self, host, port):
 
+        host = host.split(":")[0]
         endpoint = (host, port)
 
         with self._zrpcclient_acquisition_lock:
-            if endpoint in self._zrpcclients:
+            if endpoint in self._zrpcclinents:
                 return self._zrpcclients[endpoint]
 
             # We start the new client on its own thread so it uses gevent, etc.
@@ -179,6 +183,7 @@ def has_method(self, session_id, uid, name):
             return client
 
     def run_zrpcclient(self, host, port, req_queue):
+        host = host.split(":")[0]
         client = zerorpc.Client("tcp://%s:%d" % (host, port), context=self._context)
 
         forwarder = gevent.spawn(self.forward_requests, req_queue, client)
@@ -210,6 +215,7 @@ def queue_request(self, client, req):
         async_result.rawlink(lambda x: self.process_response(req, x))
 
     def get_rpc_client(self, hostname, port):
+        # hostname = hostname.split(":")[0] 
         client = self.get_client_for_endpoint(hostname, port)
         # No closing function since clients are long-lived
         return client, lambda: None