Merge pull request #17876 from LabNConsulting/chopps/backend-ds-notify

Handle datastore notifications correctly in backend clients (daemons)

11 files changed, 567 insertions, 200 deletions

diff --git a/lib/mgmt_be_client.c b/lib/mgmt_be_client.c
index 3a07a1d2d..806242ed5 100644
--- a/lib/mgmt_be_client.c
+++ b/lib/mgmt_be_client.c
@@ -1114,19 +1114,24 @@ static void be_client_handle_notify(struct mgmt_be_client *client, void *msgbuf,
 				    size_t msg_len)
 {
 	struct mgmt_msg_notify_data *notif_msg = msgbuf;
-	struct nb_node *nb_node;
-	struct lyd_node *dnode;
+	struct nb_node *nb_node, *nb_parent;
+	struct lyd_node *dnode = NULL;
 	const char *data = NULL;
 	const char *notif;
-	LY_ERR err;
+	bool is_yang_notify;
+	LY_ERR err = LY_SUCCESS;
 
 	debug_be_client("Received notification for client %s", client->name);
 
 	notif = mgmt_msg_native_xpath_data_decode(notif_msg, msg_len, data);
-	if (!notif || !data) {
+	if (!notif) {
 		log_err_be_client("Corrupt notify msg");
 		return;
 	}
+	if (!data && (notif_msg->op == NOTIFY_OP_DS_REPLACE || notif_msg->op == NOTIFY_OP_DS_PATCH)) {
+		log_err_be_client("Corrupt replace/patch notify msg: missing data");
+		return;
+	}
 
 	nb_node = nb_node_find(notif);
 	if (!nb_node) {
@@ -1134,20 +1139,41 @@ static void be_client_handle_notify(struct mgmt_be_client *client, void *msgbuf,
 		return;
 	}
 
-	if (!nb_node->cbs.notify) {
+	is_yang_notify = !!CHECK_FLAG(nb_node->snode->nodetype, LYS_NOTIF);
+
+	if (is_yang_notify && !nb_node->cbs.notify) {
 		debug_be_client("No notification callback for: %s", notif);
 		return;
 	}
 
-	err = yang_parse_notification(notif, notif_msg->result_type, data,
+	if (!nb_node->cbs.notify) {
+		/*
+		 * See if a parent has a callback, this is so backend's can
+		 * listen for changes on an entire datastore sub-tree.
+		 */
+		for (nb_parent = nb_node->parent; nb_parent; nb_parent = nb_node->parent)
+			if (nb_parent->cbs.notify)
+				break;
+		if (!nb_parent) {
+			debug_be_client("Including parents, no DS notification callback for: %s",
+					notif);
+			return;
+		}
+		nb_node = nb_parent;
+	}
+
+	if (data && is_yang_notify) {
+		err = yang_parse_notification(notif, notif_msg->result_type, data, &dnode);
+	} else if (data) {
+		err = yang_parse_data(notif, notif_msg->result_type, false, true, false, data,
 				      &dnode);
+	}
 	if (err) {
-		log_err_be_client("Can't parse notification data for: %s",
-				  notif);
+		log_err_be_client("Can't parse notification data for: %s", notif);
 		return;
 	}
 
-	nb_callback_notify(nb_node, notif, dnode);
+	nb_callback_notify(nb_node, notif_msg->op, notif, dnode);
 
 	lyd_free_all(dnode);
 }
diff --git a/lib/northbound.c b/lib/northbound.c
index 418cb246f..400309a75 100644
--- a/lib/northbound.c
+++ b/lib/northbound.c
@@ -1857,7 +1857,7 @@ int nb_callback_rpc(const struct nb_node *nb_node, const char *xpath,
 	return nb_node->cbs.rpc(&args);
 }
 
-void nb_callback_notify(const struct nb_node *nb_node, const char *xpath,
+void nb_callback_notify(const struct nb_node *nb_node, uint8_t op, const char *xpath,
 			struct lyd_node *dnode)
 {
 	struct nb_cb_notify_args args = {};
@@ -1865,6 +1865,7 @@ void nb_callback_notify(const struct nb_node *nb_node, const char *xpath,
 	DEBUGD(&nb_dbg_cbs_notify, "northbound notify: %s", xpath);
 
 	args.xpath = xpath;
+	args.op = op;
 	args.dnode = dnode;
 	nb_node->cbs.notify(&args);
 }
diff --git a/lib/northbound.h b/lib/northbound.h
index ce59bfd01..c31f007e7 100644
--- a/lib/northbound.h
+++ b/lib/northbound.h
@@ -305,6 +305,7 @@ struct nb_cb_rpc_args {
 struct nb_cb_notify_args {
 	/* XPath of the notification. */
 	const char *xpath;
+	uint8_t op;
 
 	/*
 	 * libyang data node representing the notification. If the notification
@@ -861,7 +862,7 @@ extern const void *nb_callback_lookup_next(const struct nb_node *nb_node,
 extern int nb_callback_rpc(const struct nb_node *nb_node, const char *xpath,
 			   const struct lyd_node *input, struct lyd_node *output,
 			   char *errmsg, size_t errmsg_len);
-extern void nb_callback_notify(const struct nb_node *nb_node, const char *xpath,
+extern void nb_callback_notify(const struct nb_node *nb_node, uint8_t op, const char *xpath,
 			       struct lyd_node *dnode);
 
 /*
diff --git a/lib/northbound_notif.c b/lib/northbound_notif.c
index b75c86561..10a81d05f 100644
--- a/lib/northbound_notif.c
+++ b/lib/northbound_notif.c
@@ -480,87 +480,96 @@ static struct op_changes_group *op_changes_group_next(void)
 /* Query for changes and notify */
 /* ---------------------------- */
 
+static void timer_walk_abort(struct nb_notif_walk_args *args);
 static void timer_walk_continue(struct event *event);
+static void timer_walk_done(struct nb_notif_walk_args *args);
+
+static struct op_change *__next_change(struct op_changes_group *group)
+{
+	struct op_change *next = RB_NEXT(op_changes, group->cur_change);
+
+	/* Remove and free current so retry works */
+	RB_REMOVE(op_changes, group->cur_changes, group->cur_change);
+	op_change_free(group->cur_change);
+	return next;
+}
+
+static struct op_changes_group *__next_group(struct op_changes_group *group)
+{
+	__dbg("done with oper-path collection for group");
+	op_changes_group_free(group);
+	return op_changes_group_next();
+}
 
 static enum nb_error oper_walk_done(const struct lyd_node *tree, void *arg, enum nb_error ret)
 {
 	struct nb_notif_walk_args *args = arg;
 	struct op_changes_group *group = args->group;
 	const char *path = group->cur_change->path;
-	const char *op = group->cur_changes == &group->adds ? "add" : "delete";
 
 	/* we don't send batches when yielding as we need completed edit in any patch */
 	assert(ret != NB_YIELD);
 
-	nb_notif_walk = NULL;
-
 	if (ret == NB_ERR_NOT_FOUND) {
 		__dbg("Path not found while walking oper tree: %s", path);
-		XFREE(MTYPE_NB_NOTIF_WALK_ARGS, args);
-		return ret;
-	}
-	/* Something else went wrong with the walk */
-	if (ret != NB_OK) {
+		ret = NB_OK;
+	} else if (ret != NB_OK) {
 error:
-		__log_err("Error notifying for datastore change on path: %s: %s", path,
-			  nb_err_name(ret));
-		XFREE(MTYPE_NB_NOTIF_WALK_ARGS, args);
-		/* XXX Need to inform mgmtd/front-ends things are out-of-sync */
-		return ret;
-	}
-
-	__dbg("done with oper-path collection for %s path: %s", op, path);
-
-	/* Do we need this? */
-	while (tree->parent)
-		tree = lyd_parent(tree);
-
-	/* Send the add (replace) notification */
-	if (mgmt_be_send_ds_replace_notification(path, tree)) {
-		ret = NB_ERR;
-		goto error;
+		__log_err("Error notifying for datastore path: %s: %s", path, nb_err_name(ret));
+
+		timer_walk_abort(args);
+		goto done;
+	} else {
+		__dbg("Done with oper-path collection for path: %s", path);
+
+		/* Do we need this? */
+		while (tree->parent)
+			tree = lyd_parent(tree);
+
+		/* Send the add (replace) notification */
+		if (mgmt_be_send_ds_replace_notification(path, tree)) {
+			__log_err("Error sending notification message for path: %s", path);
+			ret = NB_ERR;
+			goto error;
+		}
 	}
 
 	/*
-	 * Advance to next change (either dels or adds or both).
+	 * Advance to next change.
 	 */
 
-	group->cur_change = RB_NEXT(op_changes, group->cur_change);
+	group->cur_change = __next_change(group);
 	if (!group->cur_change) {
-		__dbg("done with oper-path collection for group");
-		op_changes_group_free(group);
-
-		group = op_changes_group_next();
-		args->group = group;
-		if (!group) {
-			__dbg("done with ALL oper-path collection for notification");
-			XFREE(MTYPE_NB_NOTIF_WALK_ARGS, args);
+		args->group = __next_group(group);
+		if (!args->group) {
+			timer_walk_done(args);
 			goto done;
 		}
 	}
 
+	/* Run next walk after giving other events a shot to run */
 	event_add_timer_msec(nb_notif_master, timer_walk_continue, args, 0, &nb_notif_timer);
 done:
 	/* Done with current walk and scheduled next one if there is more */
 	nb_notif_walk = NULL;
 
-	return NB_OK;
+	return ret;
 }
 
-static LY_ERR nb_notify_delete_changes(struct nb_notif_walk_args *args)
+static int nb_notify_delete_changes(struct nb_notif_walk_args *args)
 {
 	struct op_changes_group *group = args->group;
-	LY_ERR err;
 
 	group->cur_change = RB_MIN(op_changes, group->cur_changes);
 	while (group->cur_change) {
-		err = mgmt_be_send_ds_delete_notification(group->cur_change->path);
-		assert(err == LY_SUCCESS); /* XXX */
-
-		group->cur_change = RB_NEXT(op_changes, group->cur_change);
+		if (mgmt_be_send_ds_delete_notification(group->cur_change->path)) {
+			__log_err("Error sending delete notification message for path: %s",
+				  group->cur_change->path);
+			return 1;
+		}
+		group->cur_change = __next_change(group);
 	}
-
-	return LY_SUCCESS;
+	return 0;
 }
 
 static void timer_walk_continue(struct event *event)
@@ -568,15 +577,17 @@ static void timer_walk_continue(struct event *event)
 	struct nb_notif_walk_args *args = EVENT_ARG(event);
 	struct op_changes_group *group = args->group;
 	const char *path;
-	LY_ERR err;
+	int ret;
 
 	/*
 	 * Notify about deletes until we have add changes to collect.
 	 */
 	while (group->cur_changes == &group->dels) {
-		err = nb_notify_delete_changes(args);
-		assert(err == LY_SUCCESS);  /* XXX */
-		assert(!group->cur_change); /* we send all the deletes in one message */
+		ret = nb_notify_delete_changes(args);
+		if (ret) {
+			timer_walk_abort(args);
+			return;
+		}
 
 		/* after deletes advance to adds */
 		group->cur_changes = &group->adds;
@@ -584,14 +595,9 @@ static void timer_walk_continue(struct event *event)
 		if (group->cur_change)
 			break;
 
-		__dbg("done with oper-path change group");
-		op_changes_group_free(group);
-
-		group = op_changes_group_next();
-		args->group = group;
-		if (!group) {
-			__dbg("done with ALL oper-path changes");
-			XFREE(MTYPE_NB_NOTIF_WALK_ARGS, args);
+		args->group = __next_group(group);
+		if (!args->group) {
+			timer_walk_done(args);
 			return;
 		}
 	}
@@ -621,6 +627,22 @@ static void timer_walk_start(struct event *event)
 	timer_walk_continue(event);
 }
 
+static void timer_walk_abort(struct nb_notif_walk_args *args)
+{
+	__dbg("Failed notifying datastore changes, will retry");
+
+	__dbg("oper-state notify setting retry timer to fire in: %d msec ", NB_NOTIF_TIMER_MSEC);
+	event_add_timer_msec(nb_notif_master, timer_walk_continue, args, NB_NOTIF_TIMER_MSEC,
+			     &nb_notif_timer);
+}
+
+static void timer_walk_done(struct nb_notif_walk_args *args)
+{
+	__dbg("Finished notifying for all datastore changes");
+	assert(!args->group);
+	XFREE(MTYPE_NB_NOTIF_WALK_ARGS, args);
+}
+
 static void nb_notif_set_walk_timer(void)
 {
 	if (nb_notif_walk) {
@@ -659,19 +681,23 @@ void nb_notif_init(struct event_loop *tm)
 
 void nb_notif_terminate(void)
 {
-	struct nb_notif_walk_args *args;
+	struct nb_notif_walk_args *args = nb_notif_timer ? EVENT_ARG(nb_notif_timer) : NULL;
 	struct op_changes_group *group;
 
+	__dbg("terminating: timer: %p timer arg: %p walk %p", nb_notif_timer, args, nb_notif_walk);
+
 	EVENT_OFF(nb_notif_timer);
 
 	if (nb_notif_walk) {
-		nb_oper_cancel_walk(nb_notif_walk);
-		/* need to free the group that's in the walk */
+		/* Grab walk args from walk if active. */
 		args = nb_oper_walk_finish_arg(nb_notif_walk);
-		if (args)
-			op_changes_group_free(args->group);
+		nb_oper_cancel_walk(nb_notif_walk);
 		nb_notif_walk = NULL;
 	}
+	if (args) {
+		op_changes_group_free(args->group);
+		XFREE(MTYPE_NB_NOTIF_WALK_ARGS, args);
+	}
 
 	while ((group = op_changes_group_next()))
 		op_changes_group_free(group);
diff --git a/lib/yang.c b/lib/yang.c
index 2aa353925..dd48d8861 100644
--- a/lib/yang.c
+++ b/lib/yang.c
@@ -874,6 +874,60 @@ static void ly_zlog_cb(LY_LOG_LEVEL level, const char *msg, const char *data_pat
 		zlog(priority, "libyang: %s", msg);
 }
 
+LY_ERR yang_parse_data(const char *xpath, LYD_FORMAT format, bool as_subtree, bool is_oper,
+		       bool validate, const char *data, struct lyd_node **tree)
+{
+	struct ly_in *in = NULL;
+	struct lyd_node *subtree = NULL;
+	uint32_t parse_options = LYD_PARSE_STRICT | LYD_PARSE_ONLY;
+	uint32_t validate_options = LYD_VALIDATE_PRESENT;
+	LY_ERR err;
+
+	err = ly_in_new_memory(data, &in);
+	if (err != LY_SUCCESS)
+		return err;
+
+	if (as_subtree) {
+		struct lyd_node *parent;
+
+		/*
+		 * Create the subtree branch from root using the xpath. This
+		 * will be used below to parse the data rooted at the subtree --
+		 * a common YANG JSON technique (vs XML which starts all
+		 * data trees from the root).
+		 */
+		err = lyd_new_path2(NULL, ly_native_ctx, xpath, NULL, 0, 0, 0, &parent, &subtree);
+		if (err != LY_SUCCESS)
+			goto done;
+		err = lyd_find_path(parent, xpath, false, &subtree);
+		if (err != LY_SUCCESS)
+			goto done;
+	}
+
+	if (is_oper)
+		validate_options |= LYD_VALIDATE_OPERATIONAL;
+
+#ifdef LYD_VALIDATE_NOT_FINAL
+	if (!validate)
+		validate_options |= LYD_VALIDATE_NOT_FINAL;
+#endif
+
+	err = lyd_parse_data(ly_native_ctx, subtree, in, format, parse_options, validate_options,
+			     tree);
+	if (err == LY_SUCCESS && subtree)
+		*tree = subtree;
+done:
+	ly_in_free(in, 0);
+	if (err != LY_SUCCESS) {
+		if (*tree)
+			lyd_free_all(*tree);
+		else if (subtree)
+			lyd_free_all(subtree);
+		*tree = NULL;
+	}
+	return err;
+}
+
 LY_ERR yang_parse_notification(const char *xpath, LYD_FORMAT format,
 			       const char *data, struct lyd_node **notif)
 {
diff --git a/lib/yang.h b/lib/yang.h
index eed2fa8db..748f08903 100644
--- a/lib/yang.h
+++ b/lib/yang.h
@@ -681,6 +681,25 @@ extern struct ly_ctx *yang_ctx_new_setup(bool embedded_modules, bool explicit_co
  */
 extern void yang_debugging_set(bool enable);
 
+
+/*
+ * Parse YANG data.
+ *
+ * Args:
+ *	xpath: xpath of the data.
+ *	format: LYD_FORMAT of input data.
+ *	as_subtree: parse the data as starting at the subtree identified by xpath.
+ *	is_oper: parse as operational state allows for invalid (logs warning).
+ *	validate: validate the data (otherwise treat as non-final).
+ *	data: input data.
+ *	notif: pointer to the libyang data tree to store the parsed notification.
+ *	       If the notification is not on the top level of the yang model,
+ *	       the pointer to the notification node is still returned, but it's
+ *	       part of the full data tree with all its parents.
+ */
+LY_ERR yang_parse_data(const char *xpath, LYD_FORMAT format, bool as_subtree, bool is_oper,
+		       bool validate, const char *data, struct lyd_node **tree);
+
 /*
  * Parse a YANG notification.
  *
diff --git a/mgmtd/mgmt_be_adapter.c b/mgmtd/mgmt_be_adapter.c
index 1c32f936b..8f7049037 100644
--- a/mgmtd/mgmt_be_adapter.c
+++ b/mgmtd/mgmt_be_adapter.c
@@ -84,6 +84,13 @@ static const char *const zebra_oper_xpaths[] = {
 	NULL,
 };
 
+#ifdef HAVE_MGMTD_TESTC
+static const char *const mgmtd_testc_oper_xpaths[] = {
+	"/frr-backend:clients",
+	NULL,
+};
+#endif
+
 #ifdef HAVE_RIPD
 static const char *const ripd_config_xpaths[] = {
 	"/frr-filter:lib",
@@ -154,6 +161,9 @@ static const char *const *be_client_config_xpaths[MGMTD_BE_CLIENT_ID_MAX] = {
 };
 
 static const char *const *be_client_oper_xpaths[MGMTD_BE_CLIENT_ID_MAX] = {
+#ifdef HAVE_MGMTD_TESTC
+	[MGMTD_BE_CLIENT_ID_TESTC] = mgmtd_testc_oper_xpaths,
+#endif
 #ifdef HAVE_RIPD
 	[MGMTD_BE_CLIENT_ID_RIPD] = ripd_oper_xpaths,
 #endif
diff --git a/mgmtd/mgmt_fe_adapter.c b/mgmtd/mgmt_fe_adapter.c
index 96b7cbd59..8d5919880 100644
--- a/mgmtd/mgmt_fe_adapter.c
+++ b/mgmtd/mgmt_fe_adapter.c
@@ -107,32 +107,49 @@ static void mgmt_fe_free_ns_strings(struct ns_string_head *head)
 	ns_string_fini(head);
 }
 
-static void mgmt_fe_ns_string_remove_session(struct ns_string_head *head,
-					     struct mgmt_fe_session_ctx *session)
+static uint64_t mgmt_fe_ns_string_remove_session(struct ns_string_head *head,
+						 struct mgmt_fe_session_ctx *session)
 {
+	struct listnode *node;
 	struct ns_string *ns;
+	uint64_t clients = 0;
 
 	frr_each_safe (ns_string, head, ns) {
-		listnode_delete(ns->sessions, session);
+		node = listnode_lookup(ns->sessions, session);
+		if (!node)
+			continue;
+		list_delete_node(ns->sessions, node);
+		clients |= mgmt_be_interested_clients(ns->s, MGMT_BE_XPATH_SUBSCR_TYPE_OPER);
 		if (list_isempty(ns->sessions)) {
 			ns_string_del(head, ns);
 			mgmt_fe_free_ns_string(ns);
 		}
 	}
+
+	return clients;
 }
 
-static void mgmt_fe_add_ns_string(struct ns_string_head *head, const char *path, size_t plen,
-				  struct mgmt_fe_session_ctx *session)
+static uint64_t mgmt_fe_add_ns_string(struct ns_string_head *head, const char *path, size_t plen,
+				      struct mgmt_fe_session_ctx *session)
 {
 	struct ns_string *e, *ns;
+	uint64_t clients = 0;
 
 	ns = XCALLOC(MTYPE_MGMTD_XPATH, sizeof(*ns) + plen + 1);
 	strlcpy(ns->s, path, plen + 1);
+
 	e = ns_string_add(head, ns);
-	if (!e)
+	if (!e) {
 		ns->sessions = list_new();
-	if (!listnode_lookup(ns->sessions, session))
 		listnode_add(ns->sessions, session);
+		clients = mgmt_be_interested_clients(ns->s, MGMT_BE_XPATH_SUBSCR_TYPE_OPER);
+	} else {
+		XFREE(MTYPE_MGMTD_XPATH, ns);
+		if (!listnode_lookup(e->sessions, session))
+			listnode_add(e->sessions, session);
+	}
+
+	return clients;
 }
 
 char **mgmt_fe_get_all_selectors(void)
@@ -1653,7 +1670,7 @@ static void fe_adapter_handle_notify_select(struct mgmt_fe_session_ctx *session,
 	}
 
 	if (msg->replace) {
-		mgmt_fe_ns_string_remove_session(&mgmt_fe_ns_strings, session);
+		clients = mgmt_fe_ns_string_remove_session(&mgmt_fe_ns_strings, session);
 		// [ ] Keep a local tree to optimize sending selectors to BE?
 		// [*] Or just KISS and fanout the original message to BEs?
 		// mgmt_remove_add_notify_selectors(session->notify_xpaths, selectors);
@@ -1684,18 +1701,11 @@ static void fe_adapter_handle_notify_select(struct mgmt_fe_session_ctx *session,
 
 	/* Add the new selectors to the global tree */
 	darr_foreach_p (selectors, sp)
-		mgmt_fe_add_ns_string(&mgmt_fe_ns_strings, *sp, darr_strlen(*sp), session);
+		clients |= mgmt_fe_add_ns_string(&mgmt_fe_ns_strings, *sp, darr_strlen(*sp),
+						 session);
 
-	/* Check if any backends are interested in the new selectors. */
-	if (msg->replace) {
-		/* If we are replacing we'll send all the selectors again with replace flag */
-		clients = mgmt_be_interested_clients("/", MGMT_BE_XPATH_SUBSCR_TYPE_OPER);
-	} else {
-		darr_foreach_p (selectors, sp)
-			clients |= mgmt_be_interested_clients(*sp, MGMT_BE_XPATH_SUBSCR_TYPE_OPER);
-	}
 	if (!clients) {
-		__dbg("No backends provide oper for notify selectors: '%s' txn-id %Lu session-id: %Lu",
+		__dbg("No backends to newly notify for selectors: '%s' txn-id %Lu session-id: %Lu",
 		      selstr, session->txn_id, session->session_id);
 		goto done;
 	}
diff --git a/mgmtd/mgmt_testc.c b/mgmtd/mgmt_testc.c
index 8bb07ed06..ab8ea9a04 100644
--- a/mgmtd/mgmt_testc.c
+++ b/mgmtd/mgmt_testc.c
@@ -9,8 +9,10 @@
 #include <zebra.h>
 #include <lib/version.h>
 #include "darr.h"
+#include "debug.h"
 #include "libfrr.h"
 #include "mgmt_be_client.h"
+#include "mgmt_msg_native.h"
 #include "northbound.h"
 
 /* ---------------- */
@@ -43,15 +45,15 @@ struct zebra_privs_t __privs = {
 	.cap_num_i = 0,
 };
 
-#define OPTION_LISTEN	   2000
-#define OPTION_NOTIF_COUNT 2001
-#define OPTION_TIMEOUT	   2002
-const struct option longopts[] = {
-	{ "listen", no_argument, NULL, OPTION_LISTEN },
-	{ "notif-count", required_argument, NULL, OPTION_NOTIF_COUNT },
-	{ "timeout", required_argument, NULL, OPTION_TIMEOUT },
-	{ 0 }
-};
+#define OPTION_DATASTORE   2000
+#define OPTION_LISTEN	   2001
+#define OPTION_NOTIF_COUNT 2002
+#define OPTION_TIMEOUT	   2003
+const struct option longopts[] = { { "datastore", no_argument, NULL, OPTION_DATASTORE },
+				   { "listen", no_argument, NULL, OPTION_LISTEN },
+				   { "notify-count", required_argument, NULL, OPTION_NOTIF_COUNT },
+				   { "timeout", required_argument, NULL, OPTION_TIMEOUT },
+				   { 0 } };
 
 
 /* Master of threads. */
@@ -79,6 +81,20 @@ struct frr_signal_t __signals[] = {
 #define MGMTD_TESTC_VTY_PORT 2624
 
 /* clang-format off */
+static const struct frr_yang_module_info frr_if_info = {
+	.name = "frr-interface",
+	.ignore_cfg_cbs = true,
+	.nodes = {
+		{
+			.xpath = "/frr-interface:lib/interface",
+			.cbs.notify = async_notification,
+		},
+		{
+			.xpath = NULL,
+		}
+	}
+};
+
 static const struct frr_yang_module_info frr_ripd_info = {
 	.name = "frr-ripd",
 	.ignore_cfg_cbs = true,
@@ -98,6 +114,8 @@ static const struct frr_yang_module_info frr_ripd_info = {
 };
 
 static const struct frr_yang_module_info *const mgmt_yang_modules[] = {
+	&frr_backend_info,
+	&frr_if_info,
 	&frr_ripd_info,
 };
 
@@ -123,6 +141,7 @@ const char **__rpc_xpaths;
 struct mgmt_be_client_cbs __client_cbs = {};
 struct event *event_timeout;
 
+int f_datastore;
 int o_notif_count = 1;
 int o_timeout;
 
@@ -165,10 +184,56 @@ static void success(struct event *event)
 	quit(0);
 }
 
-static void async_notification(struct nb_cb_notify_args *args)
+static void __ds_notification(struct nb_cb_notify_args *args)
 {
-	zlog_notice("Received YANG notification");
+	uint8_t *output = NULL;
+
+	zlog_notice("Received YANG datastore notification: op %u", args->op);
+
+	if (args->op == NOTIFY_OP_NOTIFICATION) {
+		zlog_warn("ignoring non-datastore op notification: %s", args->xpath);
+		return;
+	}
+
+	/* datastore notification */
+	switch (args->op) {
+	case NOTIFY_OP_DS_REPLACE:
+		printfrr("#OP=REPLACE: %s\n", args->xpath);
+		break;
+	case NOTIFY_OP_DS_DELETE:
+		printfrr("#OP=DELETE: %s\n", args->xpath);
+		break;
+	case NOTIFY_OP_DS_PATCH:
+		printfrr("#OP=PATCH: %s\n", args->xpath);
+		break;
+	default:
+		printfrr("#OP=%u: unknown notify op\n", args->op);
+		quit(1);
+	}
 
+	if (args->dnode && args->op != NOTIFY_OP_DS_DELETE) {
+		output = yang_print_tree(args->dnode, LYD_JSON, LYD_PRINT_SHRINK);
+		if (output) {
+			printfrr("%s\n", output);
+			darr_free(output);
+		}
+	}
+	fflush(stdout);
+
+	if (o_notif_count && !--o_notif_count)
+		quit(0);
+}
+
+static void __notification(struct nb_cb_notify_args *args)
+{
+	zlog_notice("Received YANG notification: op: %u", args->op);
+
+	if (args->op != NOTIFY_OP_NOTIFICATION) {
+		zlog_warn("ignoring datastore notification: op: %u: path %s", args->op, args->xpath);
+		return;
+	}
+
+	/* bogus, we should print the actual data */
 	printf("{\"frr-ripd:authentication-failure\": {\"interface-name\": \"%s\"}}\n",
 	       yang_dnode_get_string(args->dnode, "interface-name"));
 
@@ -176,6 +241,14 @@ static void async_notification(struct nb_cb_notify_args *args)
 		quit(0);
 }
 
+static void async_notification(struct nb_cb_notify_args *args)
+{
+	if (f_datastore)
+		__ds_notification(args);
+	else
+		__notification(args);
+}
+
 static int rpc_callback(struct nb_cb_rpc_args *args)
 {
 	const char *vrf = NULL;
@@ -210,6 +283,9 @@ int main(int argc, char **argv)
 			break;
 
 		switch (opt) {
+		case OPTION_DATASTORE:
+			f_datastore = 1;
+			break;
 		case OPTION_LISTEN:
 			f_listen = 1;
 			break;
@@ -228,6 +304,9 @@ int main(int argc, char **argv)
 
 	master = frr_init();
 
+	mgmt_be_client_lib_vty_init();
+	mgmt_dbg_be_client.flags = DEBUG_MODE_ALL;
+
 	/*
 	 * Setup notification listen
 	 */
diff --git a/tests/topotests/mgmt_notif/test_ds_notify.py b/tests/topotests/mgmt_notif/test_ds_notify.py
new file mode 100644
index 000000000..1759bf8df
--- /dev/null
+++ b/tests/topotests/mgmt_notif/test_ds_notify.py
@@ -0,0 +1,238 @@
+# -*- coding: utf-8 eval: (blacken-mode 1) -*-
+#
+# January 14 2025, Christian Hopps <chopps@labn.net>
+#
+# Copyright (c) 2025, LabN Consulting, L.L.C.
+#
+"""
+Test YANG Datastore Notifications
+"""
+import json
+import logging
+import os
+import re
+import time
+
+import pytest
+from lib.topogen import Topogen
+from lib.topotest import json_cmp
+from munet.testing.util import waitline
+from oper import check_kernel_32
+
+pytestmark = [pytest.mark.ripd, pytest.mark.staticd, pytest.mark.mgmtd]
+
+CWD = os.path.dirname(os.path.realpath(__file__))
+FE_CLIENT = CWD + "/../lib/fe_client.py"
+
+
+@pytest.fixture(scope="module")
+def tgen(request):
+    "Setup/Teardown the environment and provide tgen argument to tests"
+
+    topodef = {
+        "s1": ("r1", "r2"),
+    }
+
+    tgen = Topogen(topodef, request.module.__name__)
+    tgen.start_topology()
+
+    router_list = tgen.routers()
+    for _, router in router_list.items():
+        router.load_frr_config("frr.conf")
+
+    tgen.start_router()
+    yield tgen
+    tgen.stop_topology()
+
+
+def get_op_and_json(output):
+    op = ""
+    path = ""
+    data = ""
+    for line in output.split("\n"):
+        if not line:
+            continue
+        if not op:
+            m = re.match("#OP=([A-Z]*): (.*)", line)
+            if m:
+                op = m.group(1)
+                path = m.group(2)
+                continue
+        data += line + "\n"
+    if not op:
+        assert False, f"No notifcation op present in:\n{output}"
+    return op, path, data
+
+
+def test_frontend_datastore_notification(tgen):
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    r1 = tgen.gears["r1"].net
+
+    check_kernel_32(r1, "11.11.11.11", 1, "")
+
+    rc, _, _ = r1.cmd_status(FE_CLIENT + " --help")
+
+    if rc:
+        pytest.skip("No protoc or present cannot run test")
+
+    # Start our FE client in the background
+    p = r1.popen(
+        [FE_CLIENT, "--datastore", "--listen=/frr-interface:lib/interface/state"]
+    )
+    assert waitline(p.stderr, "Connected", timeout=10)
+
+    r1.cmd_raises("ip link set r1-eth0 mtu 1200")
+
+    # {"frr-interface:lib":{"interface":[{"name":"r1-eth0","state":{"if-index":2,"mtu":1200,"mtu6":1200,"speed":10000,"metric":0,"phy-address":"ba:fd:de:b5:8b:90"}}]}}
+
+    try:
+        # Wait for FE client to exit
+        output, error = p.communicate(timeout=10)
+        op, path, data = get_op_and_json(output)
+
+        assert op == "REPLACE"
+        assert path.startswith("/frr-interface:lib/interface[name='r1-eth0']/state")
+
+        jsout = json.loads(data)
+        expected = json.loads(
+            '{"frr-interface:lib":{"interface":[{"name":"r1-eth0","state":{"mtu":1200}}]}}'
+        )
+        result = json_cmp(jsout, expected)
+        assert result is None
+    finally:
+        p.kill()
+        r1.cmd_raises("ip link set r1-eth0 mtu 1500")
+
+
+def test_backend_datastore_update(tgen):
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    r1 = tgen.gears["r1"].net
+
+    check_kernel_32(r1, "11.11.11.11", 1, "")
+
+    be_client_path = "/usr/lib/frr/mgmtd_testc"
+    rc, _, _ = r1.cmd_status(be_client_path + " --help")
+
+    if rc:
+        pytest.skip("No mgmtd_testc")
+
+    # Start our BE client in the background
+    p = r1.popen(
+        [
+            be_client_path,
+            "--timeout=20",
+            "--log=file:/dev/stderr",
+            "--datastore",
+            "--listen",
+            "/frr-interface:lib/interface",
+        ]
+    )
+    assert waitline(p.stderr, "Got SUBSCR_REPLY success 1", timeout=10)
+
+    r1.cmd_raises("ip link set r1-eth0 mtu 1200")
+    try:
+        expected = json.loads(
+            '{"frr-interface:lib":{"interface":[{"name":"r1-eth0","state":{"mtu":1200}}]}}'
+        )
+
+        output, error = p.communicate(timeout=10)
+        op, path, data = get_op_and_json(output)
+        jsout = json.loads(data)
+        result = json_cmp(jsout, expected)
+        assert result is None
+    finally:
+        p.kill()
+        r1.cmd_raises("ip link set r1-eth0 mtu 1500")
+
+
+def test_backend_datastore_add_delete(tgen):
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    r1 = tgen.gears["r1"].net
+
+    check_kernel_32(r1, "11.11.11.11", 1, "")
+
+    be_client_path = "/usr/lib/frr/mgmtd_testc"
+    rc, _, _ = r1.cmd_status(be_client_path + " --help")
+
+    if rc:
+        pytest.skip("No mgmtd_testc")
+
+    # Start our BE client in the background
+    p = r1.popen(
+        [
+            be_client_path,
+            "--timeout=20",
+            "--log=file:/dev/stderr",
+            "--notify-count=2",
+            "--datastore",
+            "--listen",
+            "/frr-interface:lib/interface",
+        ]
+    )
+    assert waitline(p.stderr, "Got SUBSCR_REPLY success 1", timeout=10)
+
+    r1.cmd_raises('vtysh -c "conf t" -c "int foobar"')
+    try:
+        assert waitline(
+            p.stdout,
+            re.escape('#OP=REPLACE: /frr-interface:lib/interface[name="foobar"]/state'),
+            timeout=2,
+        )
+
+        r1.cmd_raises('vtysh -c "conf t" -c "no int foobar"')
+        assert waitline(
+            p.stdout,
+            re.escape('#OP=DELETE: /frr-interface:lib/interface[name="foobar"]/state'),
+            timeout=2,
+        )
+    finally:
+        p.kill()
+        r1.cmd_raises('vtysh -c "conf t" -c "no int foobar"')
+
+
+def test_datastore_backend_filters(tgen):
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    r1 = tgen.gears["r1"].net
+
+    check_kernel_32(r1, "11.11.11.11", 1, "")
+
+    rc, _, _ = r1.cmd_status(FE_CLIENT + " --help")
+    if rc:
+        pytest.skip("No protoc or present cannot run test")
+
+    # Start our FE client in the background
+    p = r1.popen(
+        [FE_CLIENT, "--datastore", "--listen=/frr-interface:lib/interface/state"]
+    )
+    assert waitline(p.stderr, "Connected", timeout=10)
+    time.sleep(1)
+
+    try:
+        output = r1.cmd_raises(
+            'vtysh -c "show mgmt get-data /frr-backend:clients/client/state/notify-selectors"'
+        )
+        jsout = json.loads(output)
+
+        #
+        # Verify only zebra has the notify selector as it's the only provider currently
+        #
+        state = {"notify-selectors": ["/frr-interface:lib/interface/state"]}
+        expected = {
+            "frr-backend:clients": {"client": [{"name": "zebra", "state": state}]}
+        }
+
+        result = json_cmp(jsout, expected, exact=True)
+        assert result is None
+    except Exception as error:
+        logging.error("got exception: %s", error)
+        raise
+    finally:
+        p.kill()
diff --git a/tests/topotests/mgmt_notif/test_notif.py b/tests/topotests/mgmt_notif/test_notif.py
index 526f051e6..f3c7c8bc8 100644
--- a/tests/topotests/mgmt_notif/test_notif.py
+++ b/tests/topotests/mgmt_notif/test_notif.py
@@ -5,17 +5,13 @@
 #
 # Copyright (c) 2024, LabN Consulting, L.L.C.
 #
-
 """
-Test YANG Notifications
+Test Traditional YANG Notifications
 """
 import json
-import logging
 import os
-import re
 
 import pytest
-from lib.micronet import Timeout, comm_error
 from lib.topogen import Topogen
 from lib.topotest import json_cmp
 from oper import check_kernel_32
@@ -45,99 +41,6 @@ def tgen(request):
     tgen.stop_topology()
 
 
-def myreadline(f):
-    buf = ""
-    while True:
-        # logging.debug("READING 1 CHAR")
-        c = f.read(1)
-        if not c:
-            return buf if buf else None
-        buf += c
-        # logging.debug("READ CHAR: '%s'", c)
-        if c == "\n":
-            return buf
-
-
-def _wait_output(f, regex, maxwait=120):
-    timeout = Timeout(maxwait)
-    while not timeout.is_expired():
-        # line = p.stdout.readline()
-        line = myreadline(f)
-        if not line:
-            assert None, "EOF waiting for '{}'".format(regex)
-        line = line.rstrip()
-        if line:
-            logging.debug("GOT LINE: '%s'", line)
-        m = re.search(regex, line)
-        if m:
-            return m
-    assert None, "Failed to get output matching '{}' withint {} actual {}s".format(
-        regex, maxwait, timeout.elapsed()
-    )
-
-
-def get_op_and_json(output):
-    op = ""
-    path = ""
-    data = ""
-    for line in output.split("\n"):
-        if not line:
-            continue
-        if not op:
-            m = re.match("#OP=([A-Z]*): (.*)", line)
-            if m:
-                op = m.group(1)
-                path = m.group(2)
-                continue
-        data += line + "\n"
-    if not op:
-        assert False, f"No notifcation op present in:\n{output}"
-    return op, path, data
-
-
-def test_frontend_datastore_notification(tgen):
-    if tgen.routers_have_failure():
-        pytest.skip(tgen.errors)
-
-    r1 = tgen.gears["r1"].net
-
-    check_kernel_32(r1, "11.11.11.11", 1, "")
-
-    fe_client_path = CWD + "/../lib/fe_client.py"
-    rc, _, _ = r1.cmd_status(fe_client_path + " --help")
-
-    if rc:
-        pytest.skip("No protoc or present cannot run test")
-
-    # Start our FE client in the background
-    p = r1.popen(
-        [fe_client_path, "--datastore", "--listen=/frr-interface:lib/interface"]
-    )
-    _wait_output(p.stderr, "Connected", maxwait=10)
-
-    r1.cmd_raises("ip link set r1-eth0 mtu 1200")
-
-    # {"frr-interface:lib":{"interface":[{"name":"r1-eth0","state":{"if-index":2,"mtu":1200,"mtu6":1200,"speed":10000,"metric":0,"phy-address":"ba:fd:de:b5:8b:90"}}]}}
-
-    try:
-        # Wait for FE client to exit
-        output, error = p.communicate(timeout=10)
-        op, path, data = get_op_and_json(output)
-
-        assert op == "REPLACE"
-        assert path.startswith("/frr-interface:lib/interface[name='r1-eth0']/state")
-
-        jsout = json.loads(data)
-        expected = json.loads(
-            '{"frr-interface:lib":{"interface":[{"name":"r1-eth0","state":{"mtu":1200}}]}}'
-        )
-        result = json_cmp(jsout, expected)
-        assert result is None
-    finally:
-        p.kill()
-        r1.cmd_raises("ip link set r1-eth0 mtu 1500")
-
-
 def test_frontend_notification(tgen):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)
@@ -240,7 +143,7 @@ def test_frontend_all_notification(tgen):
         r1.cmd_raises("vtysh", stdin=conf)
 
 
-def test_backend_notification(tgen):
+def test_backend_yang_notification(tgen):
     if tgen.routers_have_failure():
         pytest.skip(tgen.errors)