Notes on flowcept vs academy

[benclifford]

Describe the Request

I implemented a rough integration of flowcept / academy configurable observabilty #295.

The key changes are:

i) in academy, some more meaningful log events around action start/finish. I'll make these into a separate PR because they aren't part of #295 configurability.

ii) a log config+handler that understands how to gateway those events into flowcept. That looks like this:

import logging
import uuid
from academy.logging import LogConfig
from flowcept import Flowcept, FlowceptTask

class FlowceptLogging(LogConfig):

    def __init__(self):
        self.uid = uuid.uuid4()

    def init_logging(self):

        flowcept = Flowcept(workflow_name=str(self.uid))
        flowcept.start()

        h = FlowceptHandler()
        h.setLevel(logging.DEBUG)
        action_logger = logging.getLogger("academy.handle")
        action_logger.addHandler(h)
        action_logger.level = min(action_logger.level, h.level)

        def uninit():
            flowcept.stop()

        return uninit

class FlowceptHandler(logging.Handler):

  def __init__(self):
      super().__init__()
      self.flowcept_tasks: dict[str, FlowceptTask] = {}

  def emit(self, record):
      if "academy.action_invocation" in record.__dict__ \
         and "academy.action_state" in record.__dict__:
          action_invocation = record.__dict__["academy.action_invocation"]
          print("*" * 79)
          match record.__dict__["academy.action_state"]:
            case "start":
              assert action_invocation not in self.flowcept_tasks
              ft = FlowceptTask(task_id = action_invocation,
                                activity_id = record.__dict__["academy.action"])
              self.flowcept_tasks[action_invocation] = ft
            case "completed":
              ft = self.flowcept_tasks[action_invocation]
              ft.end()
              del self.flowcept_tasks[action_invocation]

This looks for log events that have an action_invocation and action_state attribute, and reports those into flowcept. These log events aren't merged into academy master yet, but indicate the start and end of action invocations.

I am unsure around the right place to be initializing a Flowcept object in a distributed environment: should it be initialized in every process, or only once at the start of a workflow?

And, for FlowceptTask, one process might be invoking an action, but flowcept has other features like resource usage which presumably need some presence on the executing side process?

Sample Code