add diaspora radio and is able to send resource info

parsl/executors/high_throughput/executor.py

@@ -323,7 +323,7 @@ def __init__(self,
             launch_cmd = DEFAULT_LAUNCH_CMD
         self.launch_cmd = launch_cmd
 
-    radio_mode = "htex"
+    radio_mode = "diaspora"
 
     def _warn_deprecated(self, old: str, new: str):
         warnings.warn(

parsl/monitoring/radios.py

@@ -1,3 +1,5 @@
+import datetime
+import json
 import os
 import socket
 import pickle
@@ -8,6 +10,7 @@
 
 from typing import Optional
 
+from parsl.monitoring.message_type import MessageType
 from parsl.serialize import serialize
 
 _db_manager_excepts: Optional[Exception]
@@ -21,6 +24,37 @@ class MonitoringRadio(metaclass=ABCMeta):
     def send(self, message: object) -> None:
         pass
 
+class DateTimeEncoder(json.JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj, (datetime.datetime, datetime.date, datetime.time)):
+            return obj.isoformat()
+        return super(DateTimeEncoder, self).default(obj)
+
+class DiasporaRadio(MonitoringRadio):
+    def __init__(self, monitoring_url: str, source_id: int, timeout: int = 10):
+        from diaspora_event_sdk import KafkaProducer
+        self.source_id = source_id
+        self.producer = KafkaProducer(value_serializer=DiasporaRadio.serialize)
+        logger.info("Diaspora-based monitoring channel initializing")
+
+    def send(self, message: object) -> None:
+        msg_type = message[0]
+        # TODO: make configurable
+        topic = "radio-test"
+        if 'run_id' in message[1]:
+            key = message[1]['run_id'].encode("utf-8")
+        else:
+            logger.info("set key as init")
+            key = b"init"
+        # logger.info(f"Sending message of type {key}:{msg_type} to topic {topic}, content {message[1]}")
+        self.producer.send(topic=topic, key=key, value=message[1])
+        logger.info(f"Sent message")
+        return
+
+    @staticmethod
+    def serialize(value):
+        return json.dumps(value, cls=DateTimeEncoder).encode("utf-8")
+
 
 class FilesystemRadio(MonitoringRadio):
     """A MonitoringRadio that sends messages over a shared filesystem.
@@ -173,3 +207,16 @@ def send(self, message: object) -> None:
             logging.error("Could not send message within timeout limit")
             return
         return
+
+
+def get_monitoring_radio(monitoring_url: str, source_id: int, radio_mode: str, run_dir: str) -> MonitoringRadio:
+    if radio_mode == "udp":
+        return UDPRadio(monitoring_url, source_id)
+    elif radio_mode == "htex":
+        return HTEXRadio(monitoring_url, source_id)
+    elif radio_mode == "filesystem":
+        return FilesystemRadio(monitoring_url=monitoring_url, source_id=source_id, run_dir=run_dir)
+    elif radio_mode == "diaspora":
+        return DiasporaRadio(monitoring_url, source_id)
+    else:
+        raise ValueError(f"Unknown radio mode {radio_mode}")

parsl/monitoring/remote.py

@@ -2,14 +2,14 @@
 import time
 import logging
 import datetime
+import parsl.monitoring.radios as radios
 from functools import wraps
 
 from parsl.multiprocessing import ForkProcess
 from multiprocessing import Event
 from parsl.process_loggers import wrap_with_logs
 
 from parsl.monitoring.message_type import MessageType
-from parsl.monitoring.radios import MonitoringRadio, UDPRadio, HTEXRadio, FilesystemRadio
 from typing import Any, Callable, Dict, List, Sequence, Tuple
 
 logger = logging.getLogger(__name__)
@@ -121,18 +121,8 @@ def send_first_last_message(try_id: int,
     import platform
     import os
 
-    radio: MonitoringRadio
-    if radio_mode == "udp":
-        radio = UDPRadio(monitoring_hub_url,
-                         source_id=task_id)
-    elif radio_mode == "htex":
-        radio = HTEXRadio(monitoring_hub_url,
-                          source_id=task_id)
-    elif radio_mode == "filesystem":
-        radio = FilesystemRadio(monitoring_url=monitoring_hub_url,
-                                source_id=task_id, run_dir=run_dir)
-    else:
-        raise RuntimeError(f"Unknown radio mode: {radio_mode}")
+    radio: radios.MonitoringRadio
+    radio = radios.get_monitoring_radio(monitoring_hub_url, task_id, radio_mode, run_dir)
 
     msg = (MessageType.RESOURCE_INFO,
            {'run_id': run_id,
@@ -177,18 +167,8 @@ def monitor(pid: int,
 
     setproctitle("parsl: task resource monitor")
 
-    radio: MonitoringRadio
-    if radio_mode == "udp":
-        radio = UDPRadio(monitoring_hub_url,
-                         source_id=task_id)
-    elif radio_mode == "htex":
-        radio = HTEXRadio(monitoring_hub_url,
-                          source_id=task_id)
-    elif radio_mode == "filesystem":
-        radio = FilesystemRadio(monitoring_url=monitoring_hub_url,
-                                source_id=task_id, run_dir=run_dir)
-    else:
-        raise RuntimeError(f"Unknown radio mode: {radio_mode}")
+    radio: radios.MonitoringRadio
+    radio = radios.get_monitoring_radio(monitoring_hub_url, task_id, radio_mode, run_dir)
 
     logging.debug("start of monitor")
 

parsl/tests/configs/htex_local_radio.py

@@ -0,0 +1,74 @@
+"""
+The aim of this configuration is to run a local htex
+in a similar manner to htex_local.py, but with lots of
+options different and more complicated than in that
+configuration, so that more code paths are executed
+than when testing only with htex_local.
+
+It does not matter too much *what* is different in this
+configuration; what matters is that the differences
+cause significantly different pieces of parsl code to be
+run - for example, by turning on monitoring, by allowing
+blocks to be started by a strategy, by using a different
+set of staging providers, by using timing parameters that
+will cause substantially different behaviour on whatever
+those timing parameters control.
+"""
+
+# imports for monitoring:
+from parsl.monitoring import MonitoringHub
+
+import os
+
+from parsl.providers import LocalProvider
+from parsl.channels import LocalChannel
+from parsl.launchers import SingleNodeLauncher
+
+from parsl.config import Config
+from parsl.executors import HighThroughputExecutor
+
+
+from parsl.data_provider.http import HTTPInTaskStaging
+from parsl.data_provider.ftp import FTPInTaskStaging
+from parsl.data_provider.file_noop import NoOpFileStaging
+
+working_dir = os.getcwd() + "/" + "test_htex_energy"
+
+
+def fresh_config():
+    return Config(
+        executors=[
+            HighThroughputExecutor(
+                address="127.0.0.1",
+                label="htex_Local",
+                working_dir=working_dir,
+                storage_access=[FTPInTaskStaging(), HTTPInTaskStaging(), NoOpFileStaging()],
+                worker_debug=True,
+                cores_per_worker=1,
+                heartbeat_period=2,
+                heartbeat_threshold=5,
+                poll_period=100,
+                provider=LocalProvider(
+                    channel=LocalChannel(),
+                    init_blocks=0,
+                    min_blocks=0,
+                    max_blocks=5,
+                    launcher=SingleNodeLauncher(),
+                ),
+                block_error_handler=False,
+            )
+        ],
+        strategy='simple',
+        app_cache=True, checkpoint_mode='task_exit',
+        retries=2,
+        monitoring=MonitoringHub(
+                        hub_address="localhost",
+                        hub_port=55055,
+                        monitoring_debug=False,
+                        resource_monitoring_interval=1,
+        ),
+        usage_tracking=True
+    )
+
+
+config = fresh_config()

parsl/tests/test_radio/diaspora_login.py

@@ -0,0 +1,6 @@
+from diaspora_event_sdk import Client as GlobusClient
+c = GlobusClient()
+print(c.retrieve_key())
+topic = "radio-test"
+print(c.register_topic(topic))
+print(c.list_topics())

parsl/tests/test_radio/test_basic.py

@@ -0,0 +1,47 @@
+import logging
+import os
+import parsl
+import pytest
+
+logger = logging.getLogger(__name__)
+
+
+@parsl.python_app
+def this_app():
+    # this delay needs to be several times the resource monitoring
+    # period configured in the test configuration, so that some
+    # messages are actually sent - there is no guarantee that any
+    # (non-first) resource message will be sent at all for a short app.
+    import time
+    time.sleep(3)
+
+    return 5
+
+@pytest.mark.local
+def test_energy_collection():
+    # this is imported here rather than at module level because
+    # it isn't available in a plain parsl install, so this module
+    # would otherwise fail to import and break even a basic test
+    # run.
+    import sqlalchemy
+    from sqlalchemy import text
+    from parsl.tests.configs.htex_local_radio import fresh_config
+
+    if os.path.exists("runinfo/monitoring.db"):
+        logger.info("Monitoring database already exists - deleting")
+        os.remove("runinfo/monitoring.db")
+
+    logger.info("loading parsl")
+    parsl.load(fresh_config())
+
+    logger.info("invoking and waiting for result")
+    assert this_app().result() == 5
+
+    logger.info("cleaning up parsl")
+    parsl.dfk().cleanup()
+    parsl.clear()
+
+    logger.info("all done")
+
+if __name__ == "__main__":
+    test_energy_collection()