add async notification support in active-active topo; refactor code for ycable tasks for change events

[vdahiya12]

Signed-off-by: vaibhav-dahiya vdahiya@microsoft.com
This PR adds an enhancement to ycabled when if and whenever a notification arrives from SoC or server that it is going to be serviced. The notification can come at anytime and hence it must be listened to or awaited by the client at all times. For this we use asyncio lib and avail the async functionality in python.

the added code defines a GracefulRestartClient class that can be used to send requests to a gRPC server for graceful restart. The class has three async methods: send_request, receive_response, and notify_graceful_restart_start.

send_request method retrieves tor from request_queue, creates a request with the ToR, and sends it to the server using the stub's NotifyGracefulRestartStart method. It then reads the response from the server and prints the details of the response.

receive_response method retrieves response from response_queue, prints/puts the the response in DB, sleeps for the period mentioned in the response, and then puts the tor of the response back in the request_queue

We follow the existing ycabled pattern and instantiate the task_worker which contains all these tasks in a seperate thread

Description

Motivation and Context

How Has This Been Tested?

UT and using this server to validate

import asyncio
import grpc
import os
from concurrent import futures
from proto_out import linkmgr_grpc_driver_pb2 as linkmgr_pb2
from proto_out import linkmgr_grpc_driver_pb2_grpc as linkmgr_pb2_grpc os.environ["GRPC_TRACE"] = "http" class GracefulRestartServicer(linkmgr_pb2_grpc.GracefulRestartServicer):     async def NotifyGracefulRestartStart(self, request, context):
        # Dummy values
        msgtype = linkmgr_pb2.SERVICE_BEGIN
        notifytype = linkmgr_pb2.CONTROL_PLANE
        guid = "abc123"
        period = 60         # Yield a stream of responses
        for i in range(5):
            response = linkmgr_pb2.GracefulAdminResponse(
                msgtype=msgtype,
                notifytype=notifytype,
                guid=guid,
                period=period
            )
            yield response
            await asyncio.sleep(1)  # delay between responses async def serve():
    server = grpc.aio.server()
    linkmgr_pb2_grpc.add_GracefulRestartServicer_to_server(
        GracefulRestartServicer(), server)
    server.add_insecure_port('[::]:50052')
    await server.start()
    await server.wait_for_termination() if __name__ == '__main__':
    asyncio.run(serve())

   

Additional Information (Optional)

[prgeor]

@vdahiya12 merge conflict

[prgeor]

@vdahiya12 description says "GracefulRestartClient" class, i didnot find in the code

[zjswhhh]

If I understand it correctly, the graceful restart notification is implemented by looping below?
fetch ToR side from request queue -> send request to server -> process response -> sleep if there is a restart period -> put ToR in request queue -> ...

[zjswhhh]

Will response ever be written to DBs for other processes to consume? Or the only action item is to sleep gRPC client for now?

[vdahiya12]

Currently the plan is to log it, if its going to be used later on for more purposes, we will put it to DB

[zjswhhh]

Will server respond when there isn't a graceful restart?

How is 20s determined here? If ToR doesn't get graceful restart notification before the scheduled restart, will SoC go ahead and restart?

[vdahiya12]

Discussed offline, as mentioned 20seconds sleep is when there is no soc connectivity, otherwise once the notification for service start comes, we would sleep for advertised period.

[zjswhhh]

Discussed offline. @prgeor can you also review this PR?