Repository: AgnostiqHQ/covalent (python)

Issue:
Make triggers compatible with self-hosted
### Environment

- **Covalent version**: 0.202.0rc0
- **Python version**: 3.8.13
- **Operating system**: MacOS Ventura M1


### What is happening?

When using the client machine as the Trigger server and self-hosted as the Dispatcher, there are two sets of errors that come up:

1. 
```
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/site-packages/watchdog/observers/api.py", line 205, in run
    self.dispatch_events(self.event_queue)
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/site-packages/watchdog/observers/api.py", line 381, in dispatch_events
    handler.dispatch(event)
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/site-packages/watchdog/events.py", line 272, in dispatch
    {
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/site-packages/covalent/triggers/dir_trigger.py", line 109, in proxy_trigger
    self.trigger()
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/site-packages/covalent/triggers/base.py", line 176, in trigger
    new_dispatch_id = self._do_redispatch()
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/site-packages/covalent/triggers/base.py", line 152, in _do_redispatch
    return redispatch(self.lattice_dispatch_id, self.dispatcher_addr, is_pending)()
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/site-packages/covalent/_dispatcher_plugins/local.py", line 253, in func
    body = get_redispatch_request_body(
  File "/Users/faiyaz/opt/anaconda3/envs/qa/lib/python3.8/site-packages/covalent/_dispatcher_plugins/local.py", line 62, in get_redispatch_request_body
    updates = {k: v.electron_object.as_transportable_dict for k, v in replace_electrons.items()}
AttributeError: 'bool' object has no attribute 'items'
```

2. The local Trigger is not registered but rather it's being set to the dispatcher address.

### How can we reproduce the issue?

Run the following workflows:

```
from pathlib import Path

import covalent as ct
from covalent.triggers import DirTrigger

import covalent as ct

dispatcher_address = "ec2-3-90-228-248.compute-1.amazonaws.com"
dispatcher_port = "48008"

ct.set_config("dispatcher.address", dispatcher_address)
ct.set_config("dispatcher.port", dispatcher_port)

Path("./tr_read").mkdir(exist_ok=True)
Path("./tr_write").mkdir(exist_ok=True)

read_file_path = str(Path("./tr_read/trigger_read.txt").resolve())
write_file_path = str(Path("./tr_write/trigger_write.txt").resolve())

dir_trigger = DirTrigger(str(Path("./tr_read").resolve()), event_names="modified", dispatcher_addr="http://ec2-3-90-228-248.compute-1.amazonaws.com:48008", triggers_server_addr="localhost:48008")

@ct.lattice(triggers=dir_trigger)
@ct.electron
def dir_workflow():
    with open(read_file_path, "r") as f:
        numbers = f.readlines()

    total_sum = sum(int(n) for n in numbers)

    with open(write_file_path, "a") as f:
        f.write(f"{total_sum}\n")

dispatch_id = ct.dispatch(dir_workflow)()
print(dispatch_id)
```

```
from pathlib import Path

import covalent as ct
from covalent.triggers import DirTrigger

import covalent as ct

dispatcher_address = "ec2-3-90-228-248.compute-1.amazonaws.com"
dispatcher_port = "48008"

ct.set_config("dispatcher.address", dispatcher_address)
ct.set_config("dispatcher.port", dispatcher_port)

Path("./tr_read").mkdir(exist_ok=True)
Path("./tr_write").mkdir(exist_ok=True)

read_file_path = str(Path("./tr_read/trigger_read.txt").resolve())
write_file_path = str(Path("./tr_write/trigger_write.txt").resolve())


@ct.lattice
@ct.electron
def dir_workflow():
    with open(read_file_path, "r") as f:
        numbers = f.readlines()

    total_sum = sum(int(n) for n in numbers)

    with open(write_file_path, "a") as f:
        f.write(f"{total_sum}\n")

dispatch_id = ct.dispatch(dir_workflow)()
print(dispatch_id)


Path("./tr_read").mkdir(exist_ok=True)
Path("./tr_write").mkdir(exist_ok=True)

read_file_path = str(Path("./tr_read/trigger_read.txt").resolve())
write_file_path = str(Path("./tr_write/trigger_write.txt").resolve())

dir_trigger = DirTrigger(str(Path("./tr_read").resolve()), event_names="modified", dispatcher_addr="http://ec2-3-90-228-248.compute-1.amazonaws.com:48008", triggers_server_addr="localhost:48008")

# Interacting with dispatcher server through API endpoints
dir_trigger.use_internal_funcs = False

# Attaching dispatch id of `my_workflow` to the trigger
dir_trigger.lattice_dispatch_id = dispatch_id

# Registering it to the triggers server
dir_trigger.register()
```

### What should happen?

The triggers should function as they do locally.

### Any suggestions?

Fix 1: Use keyword arguments when dispatching in the following:

```
    def _do_redispatch(self, is_pending: bool = False) -> str:
        """
        Perform a redispatch of the connected dispatch id and return a new one.

        Args:
            is_pending: Whether the connected dispatch id is pending

        Returns:
            new_dispatch_id: Dispatch id of the newly dispatched workflow
        """

        if self.use_internal_funcs:
            from covalent_dispatcher import run_redispatch

            return asyncio.run_coroutine_threadsafe(
                run_redispatch(self.lattice_dispatch_id, None, None, False, is_pending),
                self.event_loop,
            ).result()

        from .. import redispatch
```

Fix 2: Pass the trigger_server_addr in _register_triggers:

```
    @staticmethod
    def register_triggers(triggers_data: List[Dict], dispatch_id: str) -> None:
        """
        Register the given triggers to the Triggers server.
        Register also starts the `observe()` method of said trigger.
        This is done by calling `BaseTrigger._register` method
        with the given trigger dictionary and is equivalent to
        calling the trigger objects `register()` method.

        Args:
            triggers_data: List of trigger dictionaries to be registered
            dispatch_id: Lattice's dispatch id to be linked with given triggers

        Returns:
            None
        """

        for tr_dict in triggers_data:
            tr_dict["lattice_dispatch_id"] = dispatch_id
            BaseTrigger._register(tr_dict)
```

Pull request:
Fixed redispatch and trigger server address passing in base trigger
<!--
⚠️ Your pull request title should be short detailed and understandable for all.
⚠️ Increment the version number in the /VERSION file. If this is a bugfix, increment the patch version (the rightmost number), for example 0.18.2 becomes 0.18.3. If it's a new feature, increment the minor version (the middle number), for example 0.18.2 becomes 0.19.0.
⚠️ Add a note to /CHANGELOG.md with your version number and the date, summarizing the changes.
⚠️ If your pull request fixes an open issue, please link to the issue.
⚠️ Rebase the latest changes from the develop branch. Assuming origin points to https://github.com/AgnostiqHQ/covalent, you should run git rebase origin/develop.
-->

- [x] I have added the tests to cover my changes.
- [x] I have updated the documentation and CHANGELOG accordingly.
- [x] I have read the CONTRIBUTING document.

Resolves #1598 


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 87dc6045..a70d62af 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Result status comparison
 - Raise error on extra args/kwargs
+- Fixed redispatching and trigger server address passing in base trigger
 
 ## [0.221.0-rc.0] - 2023-04-17
 
diff --git a/covalent/triggers/base.py b/covalent/triggers/base.py
index fd98b483..93549bd2 100644
--- a/covalent/triggers/base.py
+++ b/covalent/triggers/base.py
@@ -81,16 +81,18 @@ class BaseTrigger:
         """
         Register this trigger to the Triggers server and start observing.
         """
-        self._register(self.to_dict(), self.triggers_server_addr)
+        self._register(self.to_dict())
 
     @staticmethod
-    def _register(trigger_data, triggers_server_addr=None) -> None:
+    def _register(trigger_data) -> None:
         """
         Register a trigger to the Triggers server given only its dictionary format and start observing.
 
         Args:
             trigger_data: Dictionary representation of a trigger
         """
+
+        triggers_server_addr = trigger_data.get("triggers_server_addr")
         if triggers_server_addr is None:
             triggers_server_addr = (
                 get_config("dispatcher.address") + ":" + str(get_config("dispatcher.port"))
@@ -149,7 +151,11 @@ class BaseTrigger:
 
         from .. import redispatch
 
-        return redispatch(self.lattice_dispatch_id, self.dispatcher_addr, is_pending)()
+        return redispatch(
+            dispatch_id=self.lattice_dispatch_id,
+            dispatcher_addr=self.dispatcher_addr,
+            is_pending=is_pending,
+        )
 
     def trigger(self) -> None:
         """


Test patch:
diff --git a/tests/covalent_tests/triggers/base_test.py b/tests/covalent_tests/triggers/base_test.py
index 6d98791c..bcf4af9f 100644
--- a/tests/covalent_tests/triggers/base_test.py
+++ b/tests/covalent_tests/triggers/base_test.py
@@ -33,7 +33,7 @@ def test_register(mocker):
     """
 
     mock_config = "mock-config"
-    mock_json_data = "mock-json-data"
+    mock_json_data = {"trigger_server_addr": "mock-json-data", "name": "mock-name"}
     mocker.patch("covalent.triggers.base.get_config", return_value=mock_config)
     mocker.patch("covalent.triggers.base.BaseTrigger.to_dict", return_value=mock_json_data)
     requests_mock = mocker.patch("covalent.triggers.base.requests")
@@ -120,7 +120,7 @@ def test_do_redispatch(mocker, use_internal_func, is_pending):
         mock_run_coro.assert_called_once()
         mock_fut_res.result.assert_called_once()
     else:
-        mock_redispatch = mocker.patch("covalent.redispatch")()
+        mock_redispatch = mocker.patch("covalent.redispatch")
         mock_redispatch.return_value = mock_redispatch_id
         redispatch_id = base_trigger._do_redispatch(is_pending)