Developing agents with NDK in Python#
This guide explains how to consume the NDK service when developers write the agents using Python1.
This guide provides code snippets for several operations that a typical agent needs to perform according to the NDK Service Operations Flow chapter.
Where applicable, the chapters on this page will refer to the NDK Architecture section to provide more context on the operations.
In addition to the publicly available protobuf files, which define the NDK Service, Nokia also provides generated Python bindings for data access classes of NDK the
nokia/srlinux-ndk-py repo. The generated module enables developers of NDK agents to immediately start writing NDK applications without the need to generate the Python package themselves.
Establish gRPC channel with NDK manager and instantiate an NDK client#
To call service methods, a developer first needs to create a gRPC channel to communicate with the NDK manager application running on SR Linux.
This is done by passing the NDK server address -
localhost:50053 - to
grpc.Dial() as follows:
Once the gRPC channel is setup, we need to instantiate a client (often called stub) to perform RPCs. The
sdk_common_pb2_grpc.SdkMgrServiceStub method returns a
Register the agent with the NDK manager#
Agent must be first registered with SR Linux by calling the
AgentRegister method available on the returned
SdkMgrService interface. The initial agent state is created during the registration process.
During registration, SR Linux will be expecting a list of tuples with the
agent_name item and value of the agent's name as the other item of the tuple. The agent name is defined in the agent's YAML file.
from ndk.sdk_service_pb2 import AgentRegistrationRequest from ndk.sdk_common_pb2 import SdkMgrStatus register_request = AgentRegistrationRequest() register_request.agent_liveliness = keepalive_interval # Optional response = sdk_mgr_client.AgentRegister(request=register_request, metadata=metadata) if response.status == SdkMgrStatus.kSdkMgrSuccess: # Agent has been registered successfully pass else: # Agent registration failed error string available as response.error_str pass
AgentRegister method returns a
AgentRegistrationResponse object containing the status of the request as a
SdkMgrStatus object, error message (if request failed) as a string and the app id as a integer.
Register notification streams#
Create subscription stream#
A subscription stream needs to be created first before any of the subscription types can be added.
SdkMgrService first creates the subscription stream by executing
NotificationRegister method with a
NotificationRegisterRequest only field
op set to a value of
NotificationRegisterRequest.Create. This effectively creates a stream which is identified with a
stream_id returned inside the
stream_id must be associated when subscribing/unsubscribing to certain types of router notifications.
from ndk.sdk_service_pb2 import NotificationRegisterRequest request = NotificationRegisterRequest(op=NotificationRegisterRequest.Create) response = sdk_mgr_client.NotificationRegister(request=request, metadata=metadata) if response.status == sdk_status.kSdkMgrSuccess: # Notification Register successful stream_id = response.stream_id pass else: # Notification Register failed, error string available as response.error_str pass
stream_id will be used in the Streaming notifications section.
Add notification subscriptions#
stream_id is acquired, a client can register notifications of a particular type to be delivered over that stream.
Different types of notifications types can be subscribed to by calling the same
NotificationRegister method with a
op field set to
NotificationRegisterRequest.AddSubscription and the correct name argument for the configuration type being added (
NotificationRegisterRequest fields for the named arguments).
from ndk.config_service_pb2 import ConfigSubscriptionRequest request = NotificationRegisterRequest( stream_id=stream_id, op=NotificationRegisterRequest.AddSubscription, config=ConfigSubscriptionRequest(), ) response = sdk_mgr_client.NotificationRegister(request=request, metadata=metadata) if response.status == sdk_status.kSdkMgrSuccess: # Successful registration pass else: # Registration failed, error string available as response.error_str pass
It is possible to register for multiple different types of notifications at the same time by passing different subscription requests to the same
Actual streaming of notifications is a task for another service -
SdkNotificationService. This service requires developers to create its own client, which is done with
SdkNotificationService has a single method
NotificationStream that is used to start streaming notifications.
NotificationsStream is a server-side streaming RPC which means that SR Linux (server) will send back multiple event notification responses after getting the agent's (client) request.
stream_id that was returned in the Create subscription stream is used to tell the server to included the notifications that were created between when the
SdkNotificationService was created and when its
NotificationsStream method is invoked.
Handle the streamed notifications#
Handling notifications starts with reading the incoming notification messages and detecting which type this notification is exactly. When the type is known the client reads the fields of a certain notification. Here is a method that checks for all notification types and delegates handling to helper methods.
from ndk.sdk_service_pb2 import Notification def handle_notification(notification: Notification) -> None: # Field names are available on the Notification documentation page if notification.HasField("config"): handle_ConfigNotification(notification.config) if notification.HasField("intf"): handle_InterfaceNotification(notification.intf) if notification.HasField("nw_inst"): handle_NetworkInstanceNotification(notification.nw_inst) if notification.HasField("lldp_neighbor"): handle_LldpNeighborNotification(notification.lldp_neighbor) if notification.HasField("bfd_session"): handle_BfdSessionNotification(notification.bfd_session) if notification.HasField("route"): handle_IpRouteNotification(notification.route) if notification.HasField("appid"): handle_AppIdentNotification(notification.appid) if notification.HasField("nhg"): handle_NextHopGroupNotification(notification.nhg)
Notification object has a
HasField() method that allows to check if the field contains a notification. Once it is confirmed that
XXXXX field is present we can access it as attribute of the notification (
notification.XXXXX) this will return a notification of the associated type (for example accessing
notification.config returns a
It is essential to verify if the notification has a given field with the
HasField() method as accessing an invalid field will give an empty notification. The value will not be
None and the accessing the invalid field will not throw an Exception.
Agent needs to handle SIGTERM signal that is sent when a user invokes
stop command via SR Linux CLI. The following is the required steps to cleanly stop the agent:
- Remove any agent's state if it was set using
TelemetryDeletemethod of a Telemetry client.
- Delete notification subscriptions stream using
- Invoke use
AgentUnRegister()method of the
- Close gRPC channel with the
To debug an agent, the developers can analyze the log messages that the agent produced. If the agent's logging facility used stdout/stderr to write log messages, then these messages will be found at
The default SR Linux debug messages are found in the messages directory
/var/log/srlinux/buffer/messages; check them when something went wrong within the SR Linux system (agent registration failed, IDB server warning messages, etc.).