Hi Jakub,
I think I did the necessary things but it is not working for me. What I see is I am sending notification without any issue in package logs. But I can not see that I am sending in netconf logs. Also I can subscribe the stream, but as expected it can not get any notification. I have a very simple notification but may be I am doing a mistake when I am preparing the payload. Can you have a look at my code. I am preparing XML at get_nso_sri_sync_notif_message function.
module rfs-ctn-custom-notifications {
yang-version 1.1;
namespace "http://ses.com/rfs-ctn-custom-notifications";
prefix rfs-ctn-custom-notifications;
...
notification sri-event-base-sync {
leaf device-id {
description
"Name of the device that has a sub-interface resource change.";
type string;
}
}
...
}
"""Main ctn-nso-sri-notification code."""
import socket
from datetime import datetime, timezone
from typing import List
import _ncs
import ncs
from ncs.dp import Action
NOTIFCTX = None
OPER = {
1: "MOP_CREATED",
2: "MOP_DELETED",
3: "MOP_MODIFIED",
4: "MOP_VALUE_SET",
5: "MOP_MOVED_AFTER",
6: "MOP_ATTR_SET",
}
NS_HASH = _ncs.str2hash("http://ses.com/rfs-ctn-custom-notifications")
def get_device_name_from_kpath(kpath: str) -> str:
"""Get device name from kpath."""
kpath = str(kpath)
device = kpath[kpath.find("{") + 1:kpath.find("}")]
return device
def send_nso_sri_sync_notification(device: str, log: ncs.log.Log) -> None:
"""Send netconf-notification to CFS node."""
try:
notif_socket = socket.socket()
notif_daemon = _ncs.dp.init_daemon("netconf-notification-sri-event-base-sync")
_ncs.dp.connect(dx=notif_daemon, sock=notif_socket, type=_ncs.dp.CONTROL_SOCKET, ip="127.0.0.1", port=_ncs.PORT)
live_ctx = _ncs.dp.register_notification_stream(notif_daemon, None, notif_socket, "sri-event-base-sync")
log.info("Generating notification message...")
notif_message = get_nso_sri_sync_notif_message(device)
log.info(f"Notification message is set to={notif_message}")
send_notifation(live_ctx, notif_message, log)
except Exception as err: # pylint: disable=broad-except
log.error(err)
finally:
notif_socket.close()
def get_nso_sri_sync_notif_message(device_name: str):
"""Create notification payload."""
notif_values = [get_device_tag_value(device_name)]
return create_nso_sri_sync_notif_message(notif_values)
def get_device_tag_value(device_name) -> _ncs.TagValue:
"""Create xml tag and value."""
device_tag_hash = _ncs.str2hash("device-id")
device_tag = _ncs.XmlTag(NS_HASH, device_tag_hash)
device_val = _ncs.Value(device_name, _ncs.C_BUF)
return _ncs.TagValue(xmltag=device_tag, v=device_val)
def create_nso_sri_sync_notif_message(values) -> List[_ncs.TagValue]:
"""Create sri-event-base-sync notification payload xml."""
notification_tag_hash = _ncs.str2hash("sri-event-base-sync")
xml_begin = _ncs.Value((notification_tag_hash, NS_HASH), _ncs.C_XMLBEGIN)
xml_end = _ncs.Value((notification_tag_hash, NS_HASH), _ncs.C_XMLEND)
notif_tag = _ncs.XmlTag(NS_HASH, notification_tag_hash)
start = _ncs.TagValue(xmltag=notif_tag, v=xml_begin)
end = _ncs.TagValue(xmltag=notif_tag, v=xml_end)
return [start] + values + [end]
def send_notifation(ctx, msg, log) -> None:
"""Send netconf notification stream."""
current_time = datetime.now(timezone.utc)
notif_time = _ncs.DateTime(current_time.year, current_time.month, current_time.day, current_time.hour,
current_time.minute, current_time.second, current_time.microsecond,
int(current_time.strftime("%z")[:3]), int(current_time.strftime("%z")[3:]))
log.info(f"Sending notification with={ctx=}, {notif_time=}, {msg=}...")
_ncs.dp.notification_send(ctx, notif_time, msg)
log.info("Notification sent...")
# ---------------
# Action CALLBACK
# ---------------
class CtnNsoSriNotification(Action):
"""SRI notification action based on changes in device resources."""
def iterator(self, kp, op, oldv, newv):
"""Diff-iterate function."""
self.log.info(f"kp={kp}, op={OPER[op]}, newv={newv}")
return ncs.ITER_RECURSE
@Action.action
def cb_action(self, uinfo, name, kp, input, output, trans):
"""Create and send notification to cfs nso."""
self.log.info('action name: ', name)
_ncs.dp.action_set_timeout(uinfo, 1800)
self.log.info(f"Triggering kicker {input.kicker_id}, for path {input.path}, tid: {input.tid}")
with ncs.maapi.Maapi() as _maapi:
trans = _maapi.attach(input.tid)
trans.diff_iterate(self.iterator, ncs.ITER_WANT_ATTR)
_maapi.detach(input.tid)
send_nso_sri_sync_notification(get_device_name_from_kpath(kp), self.log)
class Main(ncs.application.Application):
"""ctn-nso-sri-notification main class."""
def setup(self):
"""Register service and actions."""
self.log.info("Main RUNNING")
self.register_action("sri-event-base-sync-action", CtnNsoSriNotification)
def teardown(self):
"""Teardown service and actions."""
self.log.info("Main FINISHED")
<INFO> 17-Mar-2024::12:53:35.430 rfs-ctn-custom-notifications ncs-dp-103617-rfs-ctn-custom-notifications:main:0-1-usid-75-sri-event-base-sync-action: - Generating notification message...
<INFO> 17-Mar-2024::12:53:35.431 rfs-ctn-custom-notifications ncs-dp-103617-rfs-ctn-custom-notifications:main:0-1-usid-75-sri-event-base-sync-action: - Notification message is set to=[_ncs.TagValue(tag=1870939232, ns=0), _ncs.TagValue(tag=669279939, ns=0), _ncs.TagValue(tag=1870939232, ns=0)]
<INFO> 17-Mar-2024::12:53:35.431 rfs-ctn-custom-notifications ncs-dp-103617-rfs-ctn-custom-notifications:main:0-1-usid-75-sri-event-base-sync-action: - Sending notification with=ctx=_ncs.dp.NotificationCtxRef, notif_time=_ncs.DateTime(year=2024, month=3, day=17, hour=12, min=53, sec=35, micro=431065, timezone=0, timezone_minutes=0), msg=[_ncs.TagValue(tag=1870939232, ns=0), _ncs.TagValue(tag=669279939, ns=0), _ncs.TagValue(tag=1870939232, ns=0)]...
<INFO> 17-Mar-2024::12:53:35.431 rfs-ctn-custom-notifications ncs-dp-103617-rfs-ctn-custom-notifications:main:0-1-usid-75-sri-event-base-sync-action: - Notification sent...
