Refactor some notification service implements and directories (flink-…

…extended#371) * Refactor Notification Service
jiangxin369 · Sep 7, 2022 · 1e365e6 · 1e365e6
1 parent 69cad8d
commit 1e365e6
Show file tree

Hide file tree

Showing 126 changed files with 5,098 additions and 5,760 deletions.
diff --git a/ai_flow/cli/commands/task_manager_command.py b/ai_flow/cli/commands/task_manager_command.py
@@ -20,7 +20,7 @@
 import os
 import signal
 
-from notification_service.embedded_notification_client import EmbeddedNotificationClient
+from notification_service.client.embedded_notification_client import EmbeddedNotificationClient
 
 from ai_flow.common.configuration import config_constants
 from ai_flow.blob_manager.blob_manager_interface import BlobManagerFactory, BlobManagerConfig

diff --git a/ai_flow/model/condition.py b/ai_flow/model/condition.py
@@ -17,7 +17,7 @@
 import abc
 from typing import List
 
-from notification_service.event import EventKey, Event
+from notification_service.model.event import EventKey, Event
 
 from ai_flow.model.context import Context
 

diff --git a/ai_flow/model/internal/conditions.py b/ai_flow/model/internal/conditions.py
@@ -16,7 +16,7 @@
 # under the License.
 import json
 
-from notification_service.event import EventKey, Event
+from notification_service.model.event import EventKey, Event
 from typing import List
 
 from ai_flow.model.condition import Condition

diff --git a/ai_flow/model/internal/events.py b/ai_flow/model/internal/events.py
@@ -16,7 +16,7 @@
 # under the License.
 from enum import Enum
 import json
-from notification_service.event import Event, EventKey, DEFAULT_NAMESPACE
+from notification_service.model.event import Event, EventKey, DEFAULT_NAMESPACE
 
 from ai_flow.model.status import TaskStatus
 

diff --git a/ai_flow/model/operator.py b/ai_flow/model/operator.py
@@ -19,7 +19,7 @@
 
 from typing import Dict, Optional
 
-from notification_service.event import EventKey
+from notification_service.model.event import EventKey
 
 from ai_flow.model.action import TaskAction
 from ai_flow.model.condition import Condition

diff --git a/ai_flow/model/rule.py b/ai_flow/model/rule.py
@@ -14,7 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from notification_service.event import Event
+from notification_service.model.event import Event
 from typing import Optional
 
 from ai_flow.model.action import TaskAction

diff --git a/ai_flow/model/workflow.py b/ai_flow/model/workflow.py
@@ -16,7 +16,7 @@
 # under the License.
 from typing import Dict, List, Optional
 
-from notification_service.event import EventKey, Event
+from notification_service.model.event import EventKey
 
 from ai_flow.model.action import TaskAction
 from ai_flow.model.condition import Condition

diff --git a/ai_flow/rpc/client/aiflow_client.py b/ai_flow/rpc/client/aiflow_client.py
@@ -16,8 +16,8 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-from notification_service.embedded_notification_client import EmbeddedNotificationClient
-from notification_service.event import DEFAULT_NAMESPACE
+from notification_service.client.embedded_notification_client import EmbeddedNotificationClient
+from notification_service.model.event import DEFAULT_NAMESPACE
 
 from ai_flow.rpc.client.scheduler_client import SchedulerClient
 

diff --git a/ai_flow/rpc/service/scheduler_service.py b/ai_flow/rpc/service/scheduler_service.py
@@ -18,8 +18,8 @@
 import time
 from typing import List
 
-from notification_service.event import Event
-from notification_service.notification_client import ListenerProcessor
+from notification_service.model.event import Event
+from notification_service.client.notification_client import ListenerProcessor
 
 from ai_flow.common.env import get_aiflow_home
 from ai_flow.model.status import WORKFLOW_ALIVE_SET, WorkflowStatus, WORKFLOW_FINISHED_SET, TaskStatus

diff --git a/ai_flow/scheduler/dispatcher.py b/ai_flow/scheduler/dispatcher.py
@@ -17,7 +17,7 @@
 import json
 import logging
 
-from notification_service.event import Event
+from notification_service.model.event import Event
 from typing import List
 
 from ai_flow.common.exception.exceptions import AIFlowException

diff --git a/ai_flow/scheduler/rule_executor.py b/ai_flow/scheduler/rule_executor.py
@@ -16,7 +16,7 @@
 # under the License.
 import cloudpickle
 import logging
-from notification_service.event import Event
+from notification_service.model.event import Event
 from typing import Optional
 
 from ai_flow.metadata.metadata_manager import MetadataManager

diff --git a/ai_flow/scheduler/rule_extractor.py b/ai_flow/scheduler/rule_extractor.py
@@ -22,7 +22,7 @@
 from copy import deepcopy
 from typing import List, Dict, Set, Tuple
 
-from notification_service.event import Event, EventKey
+from notification_service.model.event import Event, EventKey
 
 from ai_flow.common.util.db_util.session import create_session
 from ai_flow.common.util.json_utils import is_valid_json

diff --git a/ai_flow/scheduler/scheduler.py b/ai_flow/scheduler/scheduler.py
@@ -16,7 +16,7 @@
 # under the License.
 import sys
 
-from notification_service.event import Event
+from notification_service.model.event import Event
 
 from ai_flow.common.configuration import config_constants
 from ai_flow.scheduler.dispatcher import Dispatcher

diff --git a/ai_flow/scheduler/scheduling_event_processor.py b/ai_flow/scheduler/scheduling_event_processor.py
@@ -20,7 +20,7 @@
 from typing import Union
 
 import cloudpickle
-from notification_service.event import Event
+from notification_service.model.event import Event
 
 from ai_flow.metadata.metadata_manager import MetadataManager
 from ai_flow.model.action import TaskAction

diff --git a/ai_flow/scheduler/scheduling_unit.py b/ai_flow/scheduler/scheduling_unit.py
@@ -16,7 +16,7 @@
 # under the License.
 from typing import Union, Tuple, Optional
 
-from notification_service.event import Event
+from notification_service.model.event import Event
 
 from ai_flow.scheduler.rule_wrapper import WorkflowExecutionRuleWrapper, WorkflowRuleWrapper
 

diff --git a/ai_flow/scheduler/worker.py b/ai_flow/scheduler/worker.py
@@ -19,7 +19,7 @@
 import queue
 import threading
 
-from notification_service.event import Event
+from notification_service.model.event import Event
 from ai_flow.common.util.db_util.session import create_session
 from ai_flow.metadata.metadata_manager import MetadataManager
 from ai_flow.model.action import TaskAction

diff --git a/ai_flow/task_executor/common/heartbeat_manager.py b/ai_flow/task_executor/common/heartbeat_manager.py
@@ -20,7 +20,7 @@
 from concurrent import futures
 
 import grpc
-from notification_service.embedded_notification_client import EmbeddedNotificationClient
+from notification_service.client.embedded_notification_client import EmbeddedNotificationClient
 
 from ai_flow.common.configuration.config_constants import NOTIFICATION_SERVER_URI
 from ai_flow.common.util.db_util.session import create_session

diff --git a/ai_flow/task_executor/common/task_executor_base.py b/ai_flow/task_executor/common/task_executor_base.py
@@ -20,7 +20,7 @@
 from queue import Empty
 from typing import Optional
 
-from notification_service.embedded_notification_client import EmbeddedNotificationClient
+from ai_flow.rpc.client.aiflow_client import get_notification_client
 
 from ai_flow.common.util.db_util.session import create_session
 from ai_flow.metadata.metadata_manager import MetadataManager
@@ -61,9 +61,7 @@ def schedule_task(self, command: TaskScheduleCommand):
 
     def start(self):
         logging.info("starting task executor.")
-        self.notification_client = EmbeddedNotificationClient(
-            server_uri=NOTIFICATION_SERVER_URI, namespace='task_status_change', sender='task_executor'
-        )
+        self.notification_client = get_notification_client(namespace='task_status_change', sender='task_executor')
         self.command_queue = PersistentQueue(maxsize=MAX_QUEUE_SIZE)
         self.command_processor.start()
         self.heartbeat_manager = HeartbeatManager()

diff --git a/...ion_service/notification_service/proxy.go → lib/notification_service/bin/proxy.go b/...ion_service/notification_service/proxy.go → lib/notification_service/bin/proxy.go
diff --git a/lib/notification_service/bin/start_notification_server.py b/lib/notification_service/bin/start_notification_server.py
@@ -24,7 +24,7 @@
 from notification_service.server import NotificationServerRunner
 from notification_service.server_config import NotificationServerConfig
 from notification_service.util import db
-from notification_service.util.config import create_server_config
+from notification_service.util.server_config import create_server_config
 
 
 def _prepare_args():

diff --git a/lib/notification_service/java/pom.xml b/lib/notification_service/java/pom.xml
@@ -27,7 +27,7 @@
     <version>0.4-SNAPSHOT</version>
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <grpc.version>1.27.2</grpc.version>
+        <grpc.version>1.35.0</grpc.version>
         <protobuf.version>3.11.0</protobuf.version>
         <protoc.version>3.11.0</protoc.version>
         <commons.lang3.version>3.9</commons.lang3.version>