Skip to content

run_grpc_services

drunc.grpc_testing_tools.run_grpc_services

Functions

run_child_controller_server(controller_max_workers, server_port, root_port, child_name, log_file, server_options=None, client_options=None, ready_event=None, stop_event=None)

Run ChildController server with RootController client connection.

Source code in drunc/grpc_testing_tools/run_grpc_services.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def run_child_controller_server(
    controller_max_workers: int,
    server_port: int,
    root_port: int,
    child_name: str,
    log_file: str,
    server_options: Optional[List[Tuple[str, Any]]] = None,
    client_options: Optional[List[Tuple[str, Any]]] = None,
    ready_event=None,
    stop_event=None,
) -> None:
    """Run ChildController server with RootController client connection."""
    from drunc.grpc_testing_tools.child_controller import (
        ChildControllerServiceImpl,
    )
    from drunc.grpc_testing_tools.test_services_pb2_grpc import (
        add_ChildControllerServiceServicer_to_server,
    )

    run_grpc_server(
        server_name=child_name,
        servicer_instance=ChildControllerServiceImpl(child_name),
        add_servicer_func=add_ChildControllerServiceServicer_to_server,
        max_workers=controller_max_workers,
        server_port=server_port,
        log_file=log_file,
        server_options=server_options,
        upstream_connection={
            "host": "localhost",
            "port": root_port,
            "options": client_options or [],
        },
        ready_event=ready_event,
        stop_event=stop_event,
    )

run_grpc_server(server_name, servicer_instance, add_servicer_func, max_workers, server_port, log_file, server_options=None, upstream_connection=None, ready_event=None, stop_event=None)

Generic gRPC server runner that handles the common server lifecycle.

Source code in drunc/grpc_testing_tools/run_grpc_services.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def run_grpc_server(
    server_name: str,
    servicer_instance: Any,
    add_servicer_func: Any,
    max_workers: int,
    server_port: int,
    log_file: str,
    server_options: Optional[List[Tuple[str, Any]]] = None,
    upstream_connection: Optional[Dict[str, Any]] = None,
    ready_event=None,
    stop_event=None,
) -> None:
    """Generic gRPC server runner that handles the common server lifecycle."""

    stderr_observer(log_file)
    stdout_observer(log_file)

    from grpc import insecure_channel
    from grpc import server as grpc_server

    shutdown_requested = False

    def signal_handler(signum: int, frame) -> None:
        """Handle SIGTERM and SIGINT for graceful server shutdown."""
        nonlocal shutdown_requested
        shutdown_requested = True
        if stop_event:
            stop_event.set()
        print(f"{server_name} server received signal {signum}, shutting down...")

    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

    server = grpc_server(
        futures.ThreadPoolExecutor(max_workers=max_workers),
        options=server_options or [],
    )

    add_servicer_func(servicer_instance, server)
    port = server.add_insecure_port(f"[::]:{server_port}")

    upstream_channel = None
    try:
        server.start()

        if upstream_connection:
            upstream_host = upstream_connection["host"]
            upstream_port = upstream_connection["port"]
            client_options = upstream_connection.get("options", [])

            upstream_channel = insecure_channel(
                f"{upstream_host}:{upstream_port}", options=client_options
            )
            print(
                f"{server_name} server started on port {port}, connected to upstream on {upstream_port}"
            )
        else:
            print(f"{server_name} server started on port {port}")

        if ready_event:
            ready_event.set()

        if stop_event:
            while not shutdown_requested and not stop_event.is_set():
                time.sleep(0.1)
        else:
            while not shutdown_requested:
                time.sleep(0.1)

    except Exception as e:
        print(f"{server_name} server error: {e}")
    finally:
        print(f"Shutting down {server_name} server...")
        if upstream_channel:
            upstream_channel.close()
        server.stop(grace=SERVER_GRACE_PERIOD)

run_process_manager_server(manager_max_workers, server_port, log_file, server_options=None, ready_event=None, stop_event=None, lifetime_manager_type=ProcessManagerTypes.SSH_PARAMIKO)

Run Manager server process with output logging.

Source code in drunc/grpc_testing_tools/run_grpc_services.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def run_process_manager_server(
    manager_max_workers: int,
    server_port: int,
    log_file: str,
    server_options: Optional[List[Tuple[str, Any]]] = None,
    ready_event=None,
    stop_event=None,
    lifetime_manager_type: ProcessManagerTypes = ProcessManagerTypes.SSH_PARAMIKO,
) -> None:
    """Run Manager server process with output logging."""
    from drunc.grpc_testing_tools.process_manager import ManagerServiceImpl
    from drunc.grpc_testing_tools.test_services_pb2_grpc import (
        add_ManagerServiceServicer_to_server,
    )

    run_grpc_server(
        server_name="Manager",
        servicer_instance=ManagerServiceImpl(
            lifetime_manager_type=lifetime_manager_type
        ),
        add_servicer_func=add_ManagerServiceServicer_to_server,
        max_workers=manager_max_workers,
        server_port=server_port,
        log_file=log_file,
        server_options=server_options,
        upstream_connection=None,
        ready_event=ready_event,
        stop_event=stop_event,
    )

run_root_controller_server(controller_max_workers, server_port, manager_port, log_file, server_options=None, client_options=None, ready_event=None, stop_event=None)

Run RootController server with Manager client connection.

Source code in drunc/grpc_testing_tools/run_grpc_services.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def run_root_controller_server(
    controller_max_workers: int,
    server_port: int,
    manager_port: int,
    log_file: str,
    server_options: Optional[List[Tuple[str, Any]]] = None,
    client_options: Optional[List[Tuple[str, Any]]] = None,
    ready_event=None,
    stop_event=None,
) -> None:
    """Run RootController server with Manager client connection."""
    from drunc.grpc_testing_tools.root_controller import RootControllerServiceImpl
    from drunc.grpc_testing_tools.test_services_pb2_grpc import (
        add_RootControllerServiceServicer_to_server,
    )

    run_grpc_server(
        server_name="RootController",
        servicer_instance=RootControllerServiceImpl(),
        add_servicer_func=add_RootControllerServiceServicer_to_server,
        max_workers=controller_max_workers,
        server_port=server_port,
        log_file=log_file,
        server_options=server_options,
        upstream_connection={
            "host": "localhost",
            "port": manager_port,
            "options": client_options or [],
        },
        ready_event=ready_event,
        stop_event=stop_event,
    )