What is the purpose of the change

Support CLI tool for notification server

Brief change log

(for example:)

• Support CLI tool for notification server

Verifying this change

(Please pick either of the following options)

This change added tests.

What is the purpose of the change

Add notification cli server command #170

Brief change log

• Add notification cli server command

Verifying this change

This change added tests.

What is the purpose of the change

Fix #172

Brief change log

• Change the log_relative_path in remote log handler like S3, GCS, Azure when they init the context.
• Add get_provider_info.py to pass airflow test cases.
• Modified old test case in providers.

Verifying this change

This change has modified old test case to fit these new remote log handlers.

FileTaskHandler, a python log handler that handles and reads task instance logs, supports the current log mechanism which uses seq_num and try_number to name the log file. The remote log handler like S3TaskHandler should support the log mechanism.

What is the purpose of the change

(For example: This pull request makes user can stop workflow externally.)

Brief change log

(for example:)

• Send StopDagEvent to Airflow scheduler via notification service
• Once received StopDagEvent, Airflow scheduler kills all running dag runs

Verifying this change

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests.

What is the purpose of the change

(For example: This pull request makes user can stop workflow externally.)

Brief change log

(for example:)

• Send StopDagEvent to Airflow scheduler via notification service
• Once received StopDagEvent, Airflow scheduler kills all running dag runs

Verifying this change

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests.

Brief change log

(for example:)

• Add tutorial example and docs
• Each workflow execution should have its own working directory to avoid file overriding

Verifying this change

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

What is the purpose of the change

(For example: This pull request makes user can stop workflow externally.)

Brief change log

(for example:)

• Send StopDagEvent to Airflow scheduler via notification service
• Once received StopDagEvent, Airflow scheduler kills all running dag runs

Verifying this change

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests.

What is the purpose of the change

Refactor Notification Event

Verifying this change

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

What is the purpose of the change

Fix unittests about existed namespace

Verifying this change

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

What is the purpose of the change

Add docs about concepts

Verifying this change

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

Describe the bug

Your environment

Operating system

Database

Python version

To Reproduce

Steps to reproduce the behavior (if you can):

1. Submit a '...'
2. Click on '....'
3. See error

Expected behavior

Actual behavior

Screenshots

Additional context

Describe the bug

task1 action_on_task_status(task2, success) when task1 failed, task2 would not be scheduled, but the status of workflow execution is still running. I think the status shoud be failed

Your environment

Operating system

Database

Python version

To Reproduce

Steps to reproduce the behavior (if you can):

1. Submit a '...'
2. Click on '....'
3. See error

Expected behavior

Actual behavior

Screenshots

Additional context

Describe the feature

AIFlow support python3.6

Describe the solution you'd like

Describe alternatives you've considered

Additional context

When run aiflow with python3.6, notification server blocks wich following logs:

[2022-07-25 14:07:07,682 - server.py:68 [MainThread] - INFO: Notification server started.
[2022-07-25 14:07:18,780 - server.py:189 [Thread-1] - ERROR: Lock is not acquired.
Traceback (most recent call last):
  File "/root/venv_for_aiflow/lib64/python3.6/site-packages/notification_service/server.py", line 185, in _call_behavior_async
    return await behavior(argument, context), True
  File "/usr/lib64/python3.6/asyncio/coroutines.py", line 225, in coro
    res = yield from await_meth()
  File "/root/venv_for_aiflow/lib64/python3.6/site-packages/notification_service/service.py", line 221, in _list_all_events
    pass
  File "/usr/lib64/python3.6/asyncio/coroutines.py", line 212, in coro
    res = func(*args, **kw)
  File "/usr/lib64/python3.6/asyncio/locks.py", line 86, in __aexit__
    self.release()
  File "/usr/lib64/python3.6/asyncio/locks.py", line 207, in release
    raise RuntimeError('Lock is not acquired.')
RuntimeError: Lock is not acquired.

Describe the feature

Currently, the event is broadcasting, once the wanted event is sent, all workflow executions of this workflow would be triggered.

Describe the solution you'd like

As the scheduler dispatcher would inspect the context of each event to figure out if it contains the workflow execution id, we can inject the runtime context of each task execution to the user-defined event to make sure the event will only effect on the current workflow execution.

We need the following changes:

• Add a global variable _CURRENT_TASK_CONTEXT in context.py, it is used to store the runtime context of each task execution.
• To read and write the _CURRENT_TASK_CONTEXT, add get_runtime_task_context and set_runtime_task_context functions.
• Add a public API called wrap_execution_context to inject the runtime info to the context of the event before sending it.

_CURRENT_TASK_CONTEXT: TaskExecutionContext = None


def set_runtime_task_context(context: TaskExecutionContext):
    global _CURRENT_TASK_CONTEXT
    _CURRENT_TASK_CONTEXT = context


def get_runtime_task_context():
    return _CURRENT_TASK_CONTEXT


def wrap_execution_info_to_context(event: Event):
    """
  The event whose context is wrapped with workflow execution info would only be processed by specific workflow execution.
  """
    pass

How to use it?

def func():
    notification_client = get_notification_client()
    event = Event(event_key=EVENT_KEY, message='This is a custom message.')
    
    // wrap event with context
    wrap_execution_info_to_context(event)
    // send event
    notification_client.send_event(event)

with Workflow(name='workflow') as w1:
    task = PythonOperator(name='task', python_callable=func)

Describe alternatives you've considered

Additional context

Describe the bug

Cannot use one EmbeddedNotificationClient instance in multiple threads.

Your environment

Operating system

Database

Python version

To Reproduce

    def test_create_client_multiple_threads(self):
        import threading
        from notification_service.embedded_notification_client import EmbeddedNotificationClient
        from notification_service.event import Event, EventKey

        client = EmbeddedNotificationClient(server_uri="localhost:50052", namespace='default', sender='sender')

        def send_event():
            event = Event(event_key=EventKey('key1'), message='a')
            client.send_event(event)
            print(client.sequence_number)
        threads = []
        for i in range(100):
            thread = threading.Thread(target=send_event)
            threads.append(thread)
        for t in threads:
            t.start()
        for t in threads:
            t.join()

Expected behavior

100 events sent.

Actual behavior

less than 100 events sent.

Screenshots

Additional context