Merge pull request #17796 from LabNConsulting/chopps/datastore-notifications

operational-state (datastore) change notifications

lib/darr.h

@@ -578,7 +578,7 @@ void *__darr_resize(void *a, uint count, size_t esize, struct memtype *mt);
 		darr_ensure_cap_mt(D, __dlen + __slen + 1, MTYPE_DARR_STR);                        \
 		if (darr_len(D) == 0)                                                              \
 			*darr_append(D) = 0;                                                       \
-		memcpy(darr_last(D), (S), __slen + 1);                         \
+		memcpy(&(D)[darr_strlen(D)] /* darr_last(D) clangSA :( */, (S), __slen + 1);       \
 		_darr_len(D) += __slen;                                                            \
 		D;                                                                                 \
 	})

lib/if.c

@@ -29,6 +29,10 @@
 #include "admin_group.h"
 #include "lib/if_clippy.c"
 
+
+/* Set by the owner (zebra). */
+bool if_notify_oper_changes;
+
 DEFINE_MTYPE_STATIC(LIB, IF, "Interface");
 DEFINE_MTYPE_STATIC(LIB, IFDESC, "Intf Desc");
 DEFINE_MTYPE_STATIC(LIB, CONNECTED, "Connected");
@@ -208,6 +212,104 @@ void if_down_via_zapi(struct interface *ifp)
 	hook_call(if_down, ifp);
 }
 
+void if_update_state_metric(struct interface *ifp, uint32_t metric)
+{
+	if (ifp->metric == metric)
+		return;
+	ifp->metric = metric;
+	if (ifp->state && if_notify_oper_changes)
+		nb_op_updatef(ifp->state, "metric", "%u", ifp->metric);
+}
+
+void if_update_state_mtu(struct interface *ifp, uint mtu)
+{
+	if (ifp->mtu == mtu)
+		return;
+	ifp->mtu = mtu;
+	if (ifp->state && if_notify_oper_changes)
+		nb_op_updatef(ifp->state, "mtu", "%u", ifp->mtu);
+}
+
+void if_update_state_mtu6(struct interface *ifp, uint mtu)
+{
+	if (ifp->mtu6 == mtu)
+		return;
+	ifp->mtu6 = mtu;
+	if (ifp->state && if_notify_oper_changes)
+		nb_op_updatef(ifp->state, "mtu6", "%u", ifp->mtu);
+}
+
+void if_update_state_hw_addr(struct interface *ifp, const uint8_t *hw_addr, uint len)
+{
+	if (len == (uint)ifp->hw_addr_len && (len == 0 || !memcmp(hw_addr, ifp->hw_addr, len)))
+		return;
+	memcpy(ifp->hw_addr, hw_addr, len);
+	ifp->hw_addr_len = len;
+	if (ifp->state && if_notify_oper_changes)
+		nb_op_updatef(ifp->state, "phy-address", "%pEA", (struct ethaddr *)ifp->hw_addr);
+}
+
+void if_update_state_speed(struct interface *ifp, uint32_t speed)
+{
+	if (ifp->speed == speed)
+		return;
+	ifp->speed = speed;
+	if (ifp->state && if_notify_oper_changes)
+		nb_op_updatef(ifp->state, "speed", "%u", ifp->speed);
+}
+
+void if_update_state(struct interface *ifp)
+{
+	struct lyd_node *state = ifp->state;
+
+	if (!state || !if_notify_oper_changes)
+		return;
+
+	/*
+	 * Remove top level container update when we have patch support, for now
+	 * this keeps us from generating 6 separate REPLACE messages though.
+	 */
+	// nb_op_update(state, ".", NULL);
+	nb_op_updatef(state, "if-index", "%d", ifp->ifindex);
+	nb_op_updatef(state, "mtu", "%u", ifp->mtu);
+	nb_op_updatef(state, "mtu6", "%u", ifp->mtu);
+	nb_op_updatef(state, "speed", "%u", ifp->speed);
+	nb_op_updatef(state, "metric", "%u", ifp->metric);
+	nb_op_updatef(state, "phy-address", "%pEA", (struct ethaddr *)ifp->hw_addr);
+}
+
+static void if_update_state_remove(struct interface *ifp)
+{
+	if (!if_notify_oper_changes || ifp->name[0] == 0)
+		return;
+
+	if (vrf_is_backend_netns())
+		nb_op_update_delete_pathf(NULL, "/frr-interface:lib/interface[name=\"%s:%s\"]/state",
+					  ifp->vrf->name, ifp->name);
+	else
+		nb_op_update_delete_pathf(NULL, "/frr-interface:lib/interface[name=\"%s\"]/state",
+					  ifp->name);
+	if (ifp->state) {
+		lyd_free_all(ifp->state);
+		ifp->state = NULL;
+	}
+}
+
+static void if_update_state_add(struct interface *ifp)
+{
+	if (!if_notify_oper_changes || ifp->name[0] == 0)
+		return;
+
+	if (vrf_is_backend_netns())
+		ifp->state = nb_op_update_pathf(NULL,
+						"/frr-interface:lib/interface[name=\"%s:%s\"]/state",
+						NULL, ifp->vrf->name, ifp->name);
+	else
+		ifp->state = nb_op_update_pathf(NULL,
+						"/frr-interface:lib/interface[name=\"%s\"]/state",
+						NULL, ifp->name);
+}
+
 static struct interface *if_create_name(const char *name, struct vrf *vrf)
 {
 	struct interface *ifp;
@@ -216,7 +318,11 @@ static struct interface *if_create_name(const char *name, struct vrf *vrf)
 
 	if_set_name(ifp, name);
 
+	if (if_notify_oper_changes && ifp->state)
+		if_update_state(ifp);
+
 	hook_call(if_add, ifp);
+
 	return ifp;
 }
 
@@ -228,8 +334,10 @@ void if_update_to_new_vrf(struct interface *ifp, vrf_id_t vrf_id)
 	/* remove interface from old master vrf list */
 	old_vrf = ifp->vrf;
 
-	if (ifp->name[0] != '\0')
+	if (ifp->name[0] != '\0') {
 		IFNAME_RB_REMOVE(old_vrf, ifp);
+		if_update_state_remove(ifp);
+	}
 
 	if (ifp->ifindex != IFINDEX_INTERNAL)
 		IFINDEX_RB_REMOVE(old_vrf, ifp);
@@ -237,8 +345,11 @@ void if_update_to_new_vrf(struct interface *ifp, vrf_id_t vrf_id)
 	vrf = vrf_get(vrf_id, NULL);
 	ifp->vrf = vrf;
 
-	if (ifp->name[0] != '\0')
+	if (ifp->name[0] != '\0') {
 		IFNAME_RB_INSERT(vrf, ifp);
+		if_update_state_add(ifp);
+		if_update_state(ifp);
+	}
 
 	if (ifp->ifindex != IFINDEX_INTERNAL)
 		IFINDEX_RB_INSERT(vrf, ifp);
@@ -280,6 +391,8 @@ void if_delete(struct interface **ifp)
 
 	XFREE(MTYPE_IFDESC, ptr->desc);
 
+	if_update_state_remove(ptr);
+
 	XFREE(MTYPE_IF, ptr);
 	*ifp = NULL;
 }
@@ -630,6 +743,9 @@ int if_set_index(struct interface *ifp, ifindex_t ifindex)
 
 	ifp->ifindex = ifindex;
 
+	if (if_notify_oper_changes)
+		nb_op_updatef(ifp->state, "if-index", "%d", ifp->ifindex);
+
 	if (ifp->ifindex != IFINDEX_INTERNAL) {
 		/*
 		 * This should never happen, since we checked if there was
@@ -648,13 +764,17 @@ static void if_set_name(struct interface *ifp, const char *name)
 	if (if_cmp_name_func(ifp->name, name) == 0)
 		return;
 
-	if (ifp->name[0] != '\0')
+	if (ifp->name[0] != '\0') {
 		IFNAME_RB_REMOVE(ifp->vrf, ifp);
+		if_update_state_remove(ifp);
+	}
 
 	strlcpy(ifp->name, name, sizeof(ifp->name));
 
-	if (ifp->name[0] != '\0')
+	if (ifp->name[0] != '\0') {
 		IFNAME_RB_INSERT(ifp->vrf, ifp);
+		if_update_state_add(ifp);
+	}
 }
 
 /* Does interface up ? */

lib/if.h

@@ -297,6 +297,8 @@ struct interface {
 
 	struct vrf *vrf;
 
+	struct lyd_node *state;
+
 	/*
 	 * Has the end users entered `interface XXXX` from the cli in some
 	 * fashion?
@@ -633,6 +635,14 @@ extern void if_up_via_zapi(struct interface *ifp);
 extern void if_down_via_zapi(struct interface *ifp);
 extern void if_destroy_via_zapi(struct interface *ifp);
 
+extern void if_update_state(struct interface *ifp);
+extern void if_update_state_metric(struct interface *ifp, uint32_t metric);
+extern void if_update_state_mtu(struct interface *ifp, uint mtu);
+extern void if_update_state_mtu6(struct interface *ifp, uint mtu);
+extern void if_update_state_hw_addr(struct interface *ifp, const uint8_t *hw_addr, uint len);
+extern void if_update_state_speed(struct interface *ifp, uint32_t speed);
+
+extern bool if_notify_oper_changes;
 extern const struct frr_yang_module_info frr_interface_info;
 extern const struct frr_yang_module_info frr_interface_cli_info;
 

lib/mgmt_be_client.c

@@ -322,8 +322,7 @@ static int __send_notification(struct mgmt_be_client *client, const char *xpath,
 	LY_ERR err;
 	int ret = 0;
 
-	assert(tree);
+	assert(op != NOTIFY_OP_NOTIFICATION || xpath || tree);
-
 	debug_be_client("%s: sending %sYANG %snotification: %s", __func__,
 			op == NOTIFY_OP_DS_DELETE    ? "delete "
 			: op == NOTIFY_OP_DS_REPLACE ? "replace "
@@ -1154,6 +1153,22 @@ static void be_client_handle_notify(struct mgmt_be_client *client, void *msgbuf,
 	lyd_free_all(dnode);
 }
 
+/*
+ * Process a notify select msg
+ */
+static void be_client_handle_notify_select(struct mgmt_be_client *client, void *msgbuf,
+					   size_t msg_len)
+{
+	struct mgmt_msg_notify_select *msg = msgbuf;
+	const char **selectors = NULL;
+
+	debug_be_client("Received notify-select for client %s", client->name);
+
+	if (msg_len >= sizeof(*msg))
+		selectors = mgmt_msg_native_strings_decode(msg, msg_len, msg->selectors);
+	nb_notif_set_filters(selectors, msg->replace);
+}
+
 /*
  * Handle a native encoded message
  *
@@ -1175,6 +1190,9 @@ static void be_client_handle_native_msg(struct mgmt_be_client *client,
 	case MGMT_MSG_CODE_NOTIFY:
 		be_client_handle_notify(client, msg, msg_len);
 		break;
+	case MGMT_MSG_CODE_NOTIFY_SELECT:
+		be_client_handle_notify_select(client, msg, msg_len);
+		break;
 	default:
 		log_err_be_client("unknown native message txn-id %" PRIu64
 				  " req-id %" PRIu64 " code %u to client %s",

lib/mgmt_msg_native.c

@@ -14,7 +14,8 @@ DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_ERROR, "native error msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_GET_TREE, "native get tree msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_TREE_DATA, "native tree data msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_GET_DATA, "native get data msg");
-DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_NOTIFY, "native get data msg");
+DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_NOTIFY, "native notify msg");
+DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_NOTIFY_SELECT, "native notify select msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_EDIT, "native edit msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_EDIT_REPLY, "native edit reply msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_RPC, "native RPC msg");

lib/mgmt_msg_native.h

@@ -159,6 +159,7 @@ DECLARE_MTYPE(MSG_NATIVE_GET_TREE);
 DECLARE_MTYPE(MSG_NATIVE_TREE_DATA);
 DECLARE_MTYPE(MSG_NATIVE_GET_DATA);
 DECLARE_MTYPE(MSG_NATIVE_NOTIFY);
+DECLARE_MTYPE(MSG_NATIVE_NOTIFY_SELECT);
 DECLARE_MTYPE(MSG_NATIVE_EDIT);
 DECLARE_MTYPE(MSG_NATIVE_EDIT_REPLY);
 DECLARE_MTYPE(MSG_NATIVE_RPC);

lib/northbound.c

@@ -2754,10 +2754,15 @@ void nb_init(struct event_loop *tm,
 
 	/* Initialize oper-state */
 	nb_oper_init(tm);
+
+	/* Initialize notification-state */
+	nb_notif_init(tm);
 }
 
 void nb_terminate(void)
 {
+	nb_notif_terminate();
+
 	nb_oper_terminate();
 
 	/* Terminate the northbound CLI. */

lib/northbound.h

@@ -1512,6 +1512,15 @@ extern void nb_oper_cancel_walk(void *walk);
  */
 extern void nb_oper_cancel_all_walks(void);
 
+/**
+ * nb_oper_walk_finish_arg() - return the finish arg for this walk
+ */
+extern void *nb_oper_walk_finish_arg(void *walk);
+/**
+ * nb_oper_walk_cb_arg() - return the callback arg for this walk
+ */
+extern void *nb_oper_walk_cb_arg(void *walk);
+
 /*
  * Validate if the northbound callback operation is valid for the given node.
  *
@@ -1744,6 +1753,80 @@ extern void nb_oper_init(struct event_loop *loop);
 extern void nb_oper_terminate(void);
 extern bool nb_oper_is_yang_lib_query(const char *xpath);
 
+
+/**
+ * nb_op_update() - Create new state data.
+ * @tree: subtree @path is relative to or NULL in which case @path must be
+ *	  absolute.
+ * @path: The path of the state node to create.
+ * @value: The canonical value of the state.
+ *
+ * Return: The new libyang node.
+ */
+extern struct lyd_node *nb_op_update(struct lyd_node *tree, const char *path, const char *value);
+
+/**
+ * nb_op_update_delete() - Delete state data.
+ * @tree: subtree @path is relative to or NULL in which case @path must be
+ *	  absolute.
+ * @path: The path of the state node to delete, or NULL if @tree should just be
+ *	  deleted.
+ */
+extern void nb_op_update_delete(struct lyd_node *tree, const char *path);
+
+/**
+ * nb_op_update_pathf() - Create new state data.
+ * @tree: subtree @path_fmt is relative to or NULL in which case @path_fmt must
+ *        be absolute.
+ * @path_fmt: The path format string of the state node to create.
+ * @value: The canonical value of the state.
+ * @...: The values to substitute into @path_fmt.
+ *
+ * Return: The new libyang node.
+ */
+extern struct lyd_node *nb_op_update_pathf(struct lyd_node *tree, const char *path_fmt,
+					   const char *value, ...) PRINTFRR(2, 4);
+extern struct lyd_node *nb_op_update_vpathf(struct lyd_node *tree, const char *path_fmt,
+					    const char *value, va_list ap);
+/**
+ * nb_op_update_delete_pathf() - Delete state data.
+ * @tree: subtree @path_fmt is relative to or NULL in which case @path_fmt must
+ *	  be absolute.
+ * @path: The path of the state node to delete.
+ * @...: The values to substitute into @path_fmt.
+ */
+extern void nb_op_update_delete_pathf(struct lyd_node *tree, const char *path_fmt, ...)
+	PRINTFRR(2, 3);
+extern void nb_op_update_delete_vpathf(struct lyd_node *tree, const char *path_fmt, va_list ap);
+
+/**
+ * nb_op_updatef() - Create new state data.
+ * @tree: subtree @path is relative to or NULL in which case @path must be
+ *	  absolute.
+ * @path: The path of the state node to create.
+ * @val_fmt: The value format string to set the canonical value of the state.
+ * @...: The values to substitute into @val_fmt.
+ *
+ * Return: The new libyang node.
+ */
+extern struct lyd_node *nb_op_updatef(struct lyd_node *tree, const char *path, const char *val_fmt,
+				      ...) PRINTFRR(3, 4);
+
+extern struct lyd_node *nb_op_vupdatef(struct lyd_node *tree, const char *path, const char *val_fmt,
+				       va_list ap);
+
+/**
+ * nb_notif_set_filters() - add or replace notification filters
+ * @selectors: darr array of selector (filter) xpath strings, can be NULL if
+ *	       @replace is true. nb_notif_set_filters takes ownership of this
+ *	       array and the contained darr strings.
+ * @replace: true to replace existing set otherwise append.
+ */
+extern void nb_notif_set_filters(const char **selectors, bool replace);
+
+extern void nb_notif_init(struct event_loop *loop);
+extern void nb_notif_terminate(void);
+
 #ifdef __cplusplus
 }
 #endif

lib/northbound_notif.c

@@ -0,0 +1,680 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ * December 1 2024, Christian Hopps <chopps@labn.net>
+ *
+ * Copyright (c) 2024, LabN Consulting, L.L.C.
+ *
+ */
+#include <zebra.h>
+#include "debug.h"
+#include "lib_errors.h"
+#include "typesafe.h"
+#include "northbound.h"
+#include "mgmt_be_client.h"
+
+#define __dbg(fmt, ...)	    DEBUGD(&nb_dbg_notif, "NB_OP_CHANGE: %s: " fmt, __func__, ##__VA_ARGS__)
+#define __log_err(fmt, ...) zlog_err("NB_OP_CHANGE: %s: ERROR: " fmt, __func__, ##__VA_ARGS__)
+
+#define NB_NOTIF_TIMER_MSEC (10) /* 10msec */
+
+/*
+ * ADDS:
+ * - Less specific:
+ *   - Any new add will cause more specific pending adds to be dropped and equal
+ *     or more specific deletes to be dropped.
+ * - More specific:
+ *   - Ignore any new add that is the same or more specific than an existing add.
+ *   - A new add that is more specific than a delete should change the delete
+ *     into an add query (since adds are reported as a replace).
+ *
+ * DELETES:
+ * - Less specific:
+ *   - Any new delete will cause more specific pending deletes to be dropped and
+ *     equal or more specific adds to be dropped.
+ * - More specific:
+ *   - Ignore new deletes that are the same or more specific than existing
+ *     deletes.
+ *   - A new delete that is more specific than an add can be dropped since we
+ *     use replacement methodology for the add.
+ *
+ * One thing we have to pay close attention to is that the state is going to be
+ * queried when the notification sent, not when we are told of the change.
+ */
+
+DEFINE_MTYPE_STATIC(LIB, OP_CHANGE, "NB Oper Change");
+DEFINE_MTYPE_STATIC(LIB, OP_CHANGES_GROUP, "NB Oper Changes Group");
+DEFINE_MTYPE_STATIC(LIB, NB_NOTIF_WALK_ARGS, "NB Notify Oper Walk");
+
+struct op_change {
+	RB_ENTRY(op_change) link;
+	char path[];
+};
+
+/*
+ * RB tree for op_change
+ */
+static int op_change_cmp(const struct op_change *e1, const struct op_change *e2);
+RB_HEAD(op_changes, op_change);
+RB_PROTOTYPE(op_changes, op_change, link, op_change_cmp)
+RB_GENERATE(op_changes, op_change, link, op_change_cmp)
+
+struct op_changes nb_notif_adds = RB_INITIALIZER(&nb_notif_adds);
+struct op_changes nb_notif_dels = RB_INITIALIZER(&nb_notif_dels);
+struct event_loop *nb_notif_master;
+struct event *nb_notif_timer;
+void *nb_notif_walk;
+
+const char **nb_notif_filters;
+
+/*
+ * We maintain a queue of change lists one entry per query and notification send
+ * action
+ */
+PREDECL_LIST(op_changes_queue);
+struct op_changes_group {
+	struct op_changes_queue_item item;
+	struct op_changes adds;
+	struct op_changes dels;
+	struct op_changes *cur_changes; /* used when walking */
+	struct op_change *cur_change;	/* "    "     " */
+};
+
+DECLARE_LIST(op_changes_queue, struct op_changes_group, item);
+static struct op_changes_queue_head op_changes_queue;
+
+struct nb_notif_walk_args {
+	struct op_changes_group *group;
+	struct lyd_node *tree;
+};
+
+static void nb_notif_set_walk_timer(void);
+
+
+static int pathncmp(const char *s1, const char *s2, size_t n)
+{
+	size_t i = 0;
+
+	while (i < n && *s1 && *s2) {
+		char c1 = *s1;
+		char c2 = *s2;
+
+		if ((c1 == '\'' && c2 == '\"') || (c1 == '\"' && c2 == '\'')) {
+			s1++;
+			s2++;
+			i++;
+			continue;
+		}
+		if (c1 != c2)
+			return (unsigned char)c1 - (unsigned char)c2;
+		s1++;
+		s2++;
+		i++;
+	}
+	if (i < n)
+		return (unsigned char)*s1 - (unsigned char)*s2;
+	return 0;
+}
+
+static int pathcmp(const char *s1, const char *s2)
+{
+	while (*s1 && *s2) {
+		char c1 = *s1;
+		char c2 = *s2;
+
+		if ((c1 == '\'' && c2 == '\"') || (c1 == '\"' && c2 == '\'')) {
+			s1++;
+			s2++;
+			continue;
+		}
+		if (c1 != c2)
+			return (unsigned char)c1 - (unsigned char)c2;
+		s1++;
+		s2++;
+	}
+	return (unsigned char)*s1 - (unsigned char)*s2;
+}
+
+
+static int op_change_cmp(const struct op_change *e1, const struct op_change *e2)
+{
+	return pathcmp(e1->path, e2->path);
+}
+
+static struct op_change *op_change_alloc(const char *path)
+{
+	struct op_change *note;
+	size_t ssize = strlen(path) + 1;
+
+	note = XMALLOC(MTYPE_OP_CHANGE, sizeof(*note) + ssize);
+	memset(note, 0, sizeof(*note));
+	strlcpy(note->path, path, ssize);
+
+	return note;
+}
+
+static void op_change_free(struct op_change *note)
+{
+	XFREE(MTYPE_OP_CHANGE, note);
+}
+
+/**
+ * op_changes_group_push() - Save the current set of changes on the queue.
+ *
+ * This function will save the current set of changes on the queue and
+ * initialize a new set of changes.
+ */
+static void op_changes_group_push(void)
+{
+	struct op_changes_group *changes;
+
+	if (RB_EMPTY(op_changes, &nb_notif_adds) && RB_EMPTY(op_changes, &nb_notif_dels))
+		return;
+
+	__dbg("pushing current oper changes onto queue");
+
+	changes = XCALLOC(MTYPE_OP_CHANGES_GROUP, sizeof(*changes));
+	changes->adds = nb_notif_adds;
+	changes->dels = nb_notif_dels;
+	op_changes_queue_add_tail(&op_changes_queue, changes);
+
+	RB_INIT(op_changes, &nb_notif_adds);
+	RB_INIT(op_changes, &nb_notif_dels);
+}
+
+static void op_changes_group_free(struct op_changes_group *group)
+{
+	struct op_change *e, *next;
+
+	RB_FOREACH_SAFE (e, op_changes, &group->adds, next) {
+		RB_REMOVE(op_changes, &group->adds, e);
+		op_change_free(e);
+	}
+	RB_FOREACH_SAFE (e, op_changes, &group->dels, next) {
+		RB_REMOVE(op_changes, &group->dels, e);
+		op_change_free(e);
+	}
+	XFREE(MTYPE_OP_CHANGES_GROUP, group);
+}
+
+static struct op_change *__find_less_specific(struct op_changes *head, struct op_change *note)
+{
+	struct op_change *e;
+	size_t plen;
+
+	/*
+	 * RB_NFIND finds equal or greater (more specific) than the key,
+	 * so the previous node will be a less specific or no match that
+	 * sorts earlier. We want to find when we are a more specific
+	 * match.
+	 */
+	e = RB_NFIND(op_changes, head, note);
+	if (e)
+		e = RB_PREV(op_changes, e);
+	else
+		e = RB_MAX(op_changes, head);
+	if (!e)
+		return NULL;
+	plen = strlen(e->path);
+	if (pathncmp(e->path, note->path, plen))
+		return NULL;
+	/* equal would have been returned from RB_NFIND() then we went RB_PREV */
+	assert(strlen(note->path) != plen);
+	return e;
+}
+
+static void __drop_eq_or_more_specific(struct op_changes *head, const char *path, int plen,
+				       struct op_change *next)
+{
+	struct op_change *e;
+
+	for (e = next; e != NULL; e = next) {
+		/* if the prefix no longer matches we are done */
+		if (pathncmp(path, e->path, plen))
+			break;
+		__dbg("dropping more specific %s: %s", head == &nb_notif_adds ? "add" : "delete",
+		      e->path);
+		next = RB_NEXT(op_changes, e);
+		RB_REMOVE(op_changes, head, e);
+		op_change_free(e);
+	}
+}
+
+static void __op_change_add_del(const char *path, struct op_changes *this_head,
+				struct op_changes *other_head)
+{
+	/* find out if this has been subsumed or will subsume */
+
+	const char *op = this_head == &nb_notif_adds ? "add" : "delete";
+	struct op_change *note = op_change_alloc(path);
+	struct op_change *next, *e;
+	int plen;
+
+	__dbg("processing oper %s change path: %s", op, path);
+
+	/*
+	 * See if we are already covered by a more general `op`.
+	 */
+	e = __find_less_specific(this_head, note);
+	if (e) {
+		__dbg("%s path already covered by: %s", op, e->path);
+		op_change_free(note);
+		return;
+	}
+
+	/*
+	 * Handle having a less-specific `other op`.
+	 */
+	e = __find_less_specific(other_head, note);
+	if (e) {
+		if (this_head == &nb_notif_dels) {
+			/*
+			 * If we have a less-specific add then drop this
+			 * more-specific delete as the add-replace will remove
+			 * this missing state.
+			 */
+			__dbg("delete path already covered add-replace: %s", e->path);
+		} else {
+			/*
+			 * If we have a less-specific delete, convert the delete
+			 * to an add, and drop this more-specific add. The new
+			 * less-specific add will pick up the more specific add
+			 * during the walk and as adds are processed as replaces
+			 * any other existing state that was to be deleted will
+			 * still be deleted (unless it also returns) by the replace.
+			 */
+			__dbg("add covered, converting covering delete to add-replace: %s", e->path);
+			RB_REMOVE(op_changes, other_head, e);
+			__op_change_add_del(e->path, &nb_notif_adds, &nb_notif_dels);
+			op_change_free(e);
+		}
+		op_change_free(note);
+		return;
+	}
+
+	e = RB_INSERT(op_changes, this_head, note);
+	if (e) {
+		__dbg("path already in %s tree: %s", op, path);
+		op_change_free(note);
+		return;
+	}
+
+	__dbg("scanning for subsumed or subsuming: %s", path);
+
+	plen = strlen(path);
+
+	next = RB_NEXT(op_changes, note);
+	__drop_eq_or_more_specific(this_head, path, plen, next);
+
+	/* Drop exact match or more specific `other op` */
+	next = RB_NFIND(op_changes, other_head, note);
+	__drop_eq_or_more_specific(other_head, path, plen, next);
+
+	nb_notif_set_walk_timer();
+}
+
+static void nb_notif_add(const char *path)
+{
+	__op_change_add_del(path, &nb_notif_adds, &nb_notif_dels);
+}
+
+
+static void nb_notif_delete(const char *path)
+{
+	__op_change_add_del(path, &nb_notif_dels, &nb_notif_adds);
+}
+
+struct lyd_node *nb_op_update(struct lyd_node *tree, const char *path, const char *value)
+{
+	struct lyd_node *dnode;
+	const char *abs_path = NULL;
+
+	__dbg("updating path: %s with value: %s", path, value);
+
+	dnode = yang_state_new(tree, path, value);
+
+	if (path[0] == '/')
+		abs_path = path;
+	else
+		abs_path = lyd_path(dnode, LYD_PATH_STD, NULL, 0);
+
+	nb_notif_add(abs_path);
+
+	if (abs_path != path)
+		free((char *)abs_path);
+
+	return dnode;
+}
+
+void nb_op_update_delete(struct lyd_node *tree, const char *path)
+{
+	char *abs_path = NULL;
+
+	__dbg("deleting path: %s", path);
+
+	if (path && path[0] == '/')
+		abs_path = (char *)path;
+	else {
+		assert(tree);
+		abs_path = lyd_path(tree, LYD_PATH_STD, NULL, 0);
+		assert(abs_path);
+		if (path) {
+			char *tmp = darr_strdup(abs_path);
+
+			free(abs_path);
+			abs_path = tmp;
+			if (*darr_last(abs_path) != '/')
+				darr_in_strcat(abs_path, "/");
+			assert(abs_path); /* silence bad CLANG NULL warning */
+			darr_in_strcat(abs_path, path);
+		}
+	}
+
+	yang_state_delete(tree, path);
+
+	nb_notif_delete(abs_path);
+
+	if (abs_path != path) {
+		if (path)
+			darr_free(abs_path);
+		else
+			free(abs_path);
+	}
+}
+
+PRINTFRR(2, 0)
+struct lyd_node *nb_op_update_vpathf(struct lyd_node *tree, const char *path_fmt, const char *value,
+				     va_list ap)
+{
+	struct lyd_node *dnode;
+	char *path;
+
+	path = darr_vsprintf(path_fmt, ap);
+	dnode = nb_op_update(tree, path, value);
+	darr_free(path);
+
+	return dnode;
+}
+
+struct lyd_node *nb_op_update_pathf(struct lyd_node *tree, const char *path_fmt, const char *value,
+				    ...)
+{
+	struct lyd_node *dnode;
+	va_list ap;
+
+	va_start(ap, value);
+	dnode = nb_op_update_vpathf(tree, path_fmt, value, ap);
+	va_end(ap);
+
+	return dnode;
+}
+
+PRINTFRR(2, 0)
+void nb_op_update_delete_vpathf(struct lyd_node *tree, const char *path_fmt, va_list ap)
+{
+	char *path;
+
+	path = darr_vsprintf(path_fmt, ap);
+	nb_op_update_delete(tree, path);
+	darr_free(path);
+}
+
+void nb_op_update_delete_pathf(struct lyd_node *tree, const char *path_fmt, ...)
+{
+	va_list ap;
+
+	va_start(ap, path_fmt);
+	nb_op_update_delete_vpathf(tree, path_fmt, ap);
+	va_end(ap);
+}
+
+
+PRINTFRR(3, 0)
+struct lyd_node *nb_op_vupdatef(struct lyd_node *tree, const char *path, const char *val_fmt,
+				va_list ap)
+{
+	struct lyd_node *dnode;
+	char *value;
+
+	value = darr_vsprintf(val_fmt, ap);
+	dnode = nb_op_update(tree, path, value);
+	darr_free(value);
+
+	return dnode;
+}
+
+
+struct lyd_node *nb_op_updatef(struct lyd_node *tree, const char *path, const char *val_fmt, ...)
+{
+	struct lyd_node *dnode;
+	va_list ap;
+
+	va_start(ap, val_fmt);
+	dnode = nb_op_vupdatef(tree, path, val_fmt, ap);
+	va_end(ap);
+
+	return dnode;
+}
+
+static struct op_changes_group *op_changes_group_next(void)
+{
+	struct op_changes_group *group;
+
+	group = op_changes_queue_pop(&op_changes_queue);
+	if (!group) {
+		op_changes_group_push();
+		group = op_changes_queue_pop(&op_changes_queue);
+	}
+	if (!group)
+		return NULL;
+	group->cur_changes = &group->dels;
+	group->cur_change = RB_MIN(op_changes, group->cur_changes);
+	if (!group->cur_change) {
+		group->cur_changes = &group->adds;
+		group->cur_change = RB_MIN(op_changes, group->cur_changes);
+		assert(group->cur_change);
+	}
+	return group;
+}
+
+/* ---------------------------- */
+/* Query for changes and notify */
+/* ---------------------------- */
+
+static void timer_walk_continue(struct event *event);
+
+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) {
+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;
+	}
+
+	/*
+	 * Advance to next change (either dels or adds or both).
+	 */
+
+	group->cur_change = RB_NEXT(op_changes, group->cur_change);
+	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);
+			goto done;
+		}
+	}
+
+	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;
+}
+
+static LY_ERR 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);
+	}
+
+	return LY_SUCCESS;
+}
+
+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;
+
+	/*
+	 * 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 */
+
+		/* after deletes advance to adds */
+		group->cur_changes = &group->adds;
+		group->cur_change = RB_MIN(op_changes, group->cur_changes);
+		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);
+			return;
+		}
+	}
+
+	path = group->cur_change->path;
+	__dbg("starting next oper-path replace walk for path: %s", path);
+	nb_notif_walk = nb_oper_walk(path, NULL, 0, false, NULL, NULL, oper_walk_done, args);
+}
+
+static void timer_walk_start(struct event *event)
+{
+	struct op_changes_group *group;
+	struct nb_notif_walk_args *args;
+
+	__dbg("oper-state change notification timer fires");
+
+	group = op_changes_group_next();
+	if (!group) {
+		__dbg("no oper changes to notify");
+		return;
+	}
+
+	args = XCALLOC(MTYPE_NB_NOTIF_WALK_ARGS, sizeof(*args));
+	args->group = group;
+
+	EVENT_ARG(event) = args;
+	timer_walk_continue(event);
+}
+
+static void nb_notif_set_walk_timer(void)
+{
+	if (nb_notif_walk) {
+		__dbg("oper-state walk already in progress.");
+		return;
+	}
+	if (event_is_scheduled(nb_notif_timer)) {
+		__dbg("oper-state notification timer already set.");
+		return;
+	}
+
+	__dbg("oper-state notification setting timer to fire in: %d msec ", NB_NOTIF_TIMER_MSEC);
+	event_add_timer_msec(nb_notif_master, timer_walk_start, NULL, NB_NOTIF_TIMER_MSEC,
+			     &nb_notif_timer);
+}
+
+void nb_notif_set_filters(const char **selectors, bool replace)
+{
+	const char **csp;
+
+	if (replace) {
+		darr_free_free(nb_notif_filters);
+		nb_notif_filters = selectors;
+		return;
+	}
+	darr_foreach_p (selectors, csp)
+		*darr_append(nb_notif_filters) = *csp;
+	darr_free(selectors);
+}
+
+void nb_notif_init(struct event_loop *tm)
+{
+	nb_notif_master = tm;
+	op_changes_queue_init(&op_changes_queue);
+}
+
+void nb_notif_terminate(void)
+{
+	struct nb_notif_walk_args *args;
+	struct op_changes_group *group;
+
+	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 */
+		args = nb_oper_walk_finish_arg(nb_notif_walk);
+		if (args)
+			op_changes_group_free(args->group);
+		nb_notif_walk = NULL;
+	}
+
+	while ((group = op_changes_group_next()))
+		op_changes_group_free(group);
+
+	darr_free_free(nb_notif_filters);
+}

lib/northbound_oper.c

@@ -35,6 +35,7 @@
  * We must also process containers with lookup-next descendants last.
  */
 
+DEFINE_MTYPE_STATIC(LIB, NB_STATE, "Northbound State");
 DEFINE_MTYPE_STATIC(LIB, NB_YIELD_STATE, "NB Yield State");
 DEFINE_MTYPE_STATIC(LIB, NB_NODE_INFOS, "NB Node Infos");
 
@@ -1833,6 +1834,20 @@ bool nb_oper_is_yang_lib_query(const char *xpath)
 	return strlen(xpath) > liblen;
 }
 
+void *nb_oper_walk_finish_arg(void *walk)
+{
+	struct nb_op_yield_state *ys = walk;
+
+	return ys->finish_arg;
+}
+
+void *nb_oper_walk_cb_arg(void *walk)
+{
+	struct nb_op_yield_state *ys = walk;
+
+	return ys->cb_arg;
+}
+
 void *nb_oper_walk(const char *xpath, struct yang_translator *translator,
 		   uint32_t flags, bool should_batch, nb_oper_data_cb cb,
 		   void *cb_arg, nb_oper_data_finish_cb finish, void *finish_arg)

lib/subdir.am

@@ -84,6 +84,7 @@ lib_libfrr_la_SOURCES = \
 	lib/northbound.c \
 	lib/northbound_cli.c \
 	lib/northbound_db.c \
+	lib/northbound_notif.c \
 	lib/northbound_oper.c \
 	lib/ntop.c \
 	lib/openbsd-tree.c \

lib/vrf.c

@@ -22,6 +22,9 @@
 #include "northbound.h"
 #include "northbound_cli.h"
 
+/* Set by the owner (zebra). */
+bool vrf_notify_oper_changes;
+
 /* default VRF name value used when VRF backend is not NETNS */
 #define VRF_DEFAULT_NAME_INTERNAL "default"
 
@@ -105,6 +108,19 @@ int vrf_switchback_to_initial(void)
 	return ret;
 }
 
+static void vrf_update_state(struct vrf *vrf)
+{
+	if (!vrf->state || !vrf_notify_oper_changes)
+		return;
+
+	/*
+	 * Remove top level container update when we have patch support, for now
+	 * this keeps us from generating 2 separate REPLACE messages though.
+	 */
+	nb_op_updatef(vrf->state, "id", "%u", vrf->vrf_id);
+	nb_op_update(vrf->state, "active", CHECK_FLAG(vrf->status, VRF_ACTIVE) ? "true" : "false");
+}
+
 /* Get a VRF. If not found, create one.
  * Arg:
  *   name   - The name of the vrf.  May be NULL if unknown.
@@ -155,16 +171,32 @@ struct vrf *vrf_get(vrf_id_t vrf_id, const char *name)
 
 	/* Set name */
 	if (name && vrf->name[0] != '\0' && strcmp(name, vrf->name)) {
-		/* update the vrf name */
+		/* vrf name has changed */
+		if (vrf_notify_oper_changes) {
+			nb_op_update_delete_pathf(NULL, "/frr-vrf:lib/vrf[name=\"%s\"]", vrf->name);
+			lyd_free_all(vrf->state);
+		}
 		RB_REMOVE(vrf_name_head, &vrfs_by_name, vrf);
-		strlcpy(vrf->data.l.netns_name,
+		strlcpy(vrf->data.l.netns_name, name, NS_NAMSIZ);
-			name, NS_NAMSIZ);
 		strlcpy(vrf->name, name, sizeof(vrf->name));
 		RB_INSERT(vrf_name_head, &vrfs_by_name, vrf);
+		/* New state with new name */
+		if (vrf_notify_oper_changes)
+			vrf->state = nb_op_update_pathf(NULL, "/frr-vrf:lib/vrf[name=\"%s\"]/state",
+							NULL, vrf->name);
 	} else if (name && vrf->name[0] == '\0') {
 		strlcpy(vrf->name, name, sizeof(vrf->name));
 		RB_INSERT(vrf_name_head, &vrfs_by_name, vrf);
+
+		/* We have a name now so we can have state */
+		if (vrf_notify_oper_changes)
+			vrf->state = nb_op_update_pathf(NULL, "/frr-vrf:lib/vrf[name=\"%s\"]/state",
+							NULL, vrf->name);
 	}
+	/* Update state before hook call */
+	if (vrf->state)
+		vrf_update_state(vrf);
+
 	if (new &&vrf_master.vrf_new_hook)
 		(*vrf_master.vrf_new_hook)(vrf);
 
@@ -208,6 +240,7 @@ struct vrf *vrf_update(vrf_id_t new_vrf_id, const char *name)
 		vrf->vrf_id = new_vrf_id;
 		RB_INSERT(vrf_id_head, &vrfs_by_id, vrf);
 
+		vrf_update_state(vrf);
 	} else {
 
 		/*
@@ -254,6 +287,11 @@ void vrf_delete(struct vrf *vrf)
 	if (vrf->name[0] != '\0')
 		RB_REMOVE(vrf_name_head, &vrfs_by_name, vrf);
 
+	if (vrf_notify_oper_changes) {
+		nb_op_update_delete_pathf(NULL, "/frr-vrf:lib/vrf[name=\"%s\"]", vrf->name);
+		lyd_free_all(vrf->state);
+	}
+
 	XFREE(MTYPE_VRF, vrf);
 }
 
@@ -282,6 +320,8 @@ int vrf_enable(struct vrf *vrf)
 
 	SET_FLAG(vrf->status, VRF_ACTIVE);
 
+	vrf_update_state(vrf);
+
 	if (vrf_master.vrf_enable_hook)
 		(*vrf_master.vrf_enable_hook)(vrf);
 
@@ -307,6 +347,8 @@ void vrf_disable(struct vrf *vrf)
 
 	UNSET_FLAG(vrf->status, VRF_ACTIVE);
 
+	vrf_update_state(vrf);
+
 	if (debug_vrf)
 		zlog_debug("VRF %s(%u) is to be disabled.", vrf->name,
 			   vrf->vrf_id);

lib/vrf.h

@@ -80,6 +80,8 @@ struct vrf {
 	/* Back pointer to namespace context */
 	void *ns_ctxt;
 
+	struct lyd_node *state;
+
 	QOBJ_FIELDS;
 };
 RB_HEAD(vrf_id_head, vrf);
@@ -299,6 +301,7 @@ extern void vrf_disable(struct vrf *vrf);
 extern int vrf_enable(struct vrf *vrf);
 extern void vrf_delete(struct vrf *vrf);
 
+extern bool vrf_notify_oper_changes;
 extern const struct frr_yang_module_info frr_vrf_info;
 extern const struct frr_yang_module_info frr_vrf_cli_info;
 

lib/yang.c

@@ -14,6 +14,7 @@
 #include <libyang/version.h>
 #include "northbound.h"
 #include "frrstr.h"
+#include "darr.h"
 
 #include "lib/config_paths.h"
 
@@ -680,6 +681,116 @@ void yang_dnode_rpc_output_add(struct lyd_node *output, const char *xpath,
 	assert(err == LY_SUCCESS);
 }
 
+struct lyd_node *yang_state_new(struct lyd_node *tree, const char *path, const char *value)
+{
+	struct lyd_node *dnode, *parent;
+	LY_ERR err;
+
+	err = lyd_new_path2(tree, ly_native_ctx, path, value, 0, 0, LYD_NEW_PATH_UPDATE, &parent,
+			    &dnode);
+	assert(err == LY_SUCCESS);
+
+	/*
+	 * If the node exists and isn't updated returned dnode will be NULL, so
+	 * we need to find it. But even if returned it can be the first newly
+	 * created node (could be container of path) not the actual path dnode.
+	 * So we always find.
+	 */
+	err = lyd_find_path(tree ?: parent, path, false, &dnode);
+	assert(err == LY_SUCCESS);
+
+	return dnode;
+}
+
+void yang_state_delete(struct lyd_node *tree, const char *path)
+{
+	LY_ERR err;
+
+	if (!tree)
+		return;
+
+	if (path) {
+		err = lyd_find_path(tree, path, false, &tree);
+		if (err != LY_SUCCESS) {
+			zlog_info("State %s has already been deleted", path);
+			return;
+		}
+	}
+	lyd_free_tree(tree);
+}
+
+PRINTFRR(2, 0)
+struct lyd_node *yang_state_new_vpathf(struct lyd_node *tree, const char *path_fmt,
+				       const char *value, va_list ap)
+{
+	struct lyd_node *dnode;
+	char *path;
+
+	path = darr_vsprintf(path_fmt, ap);
+	dnode = yang_state_new(tree, path, value);
+	darr_free(path);
+
+	return dnode;
+}
+
+struct lyd_node *yang_state_new_pathf(struct lyd_node *tree, const char *path_fmt,
+				      const char *value, ...)
+{
+	struct lyd_node *dnode;
+	va_list ap;
+
+	va_start(ap, value);
+	dnode = yang_state_new_vpathf(tree, path_fmt, value, ap);
+	va_end(ap);
+
+	return dnode;
+}
+
+PRINTFRR(2, 0)
+void yang_state_delete_vpathf(struct lyd_node *tree, const char *path_fmt, va_list ap)
+{
+	char *path;
+
+	path = darr_vsprintf(path_fmt, ap);
+	yang_state_delete(tree, path);
+	darr_free(path);
+}
+
+void yang_state_delete_pathf(struct lyd_node *tree, const char *path_fmt, ...)
+{
+	va_list ap;
+
+	va_start(ap, path_fmt);
+	yang_state_delete_vpathf(tree, path_fmt, ap);
+	va_end(ap);
+}
+
+PRINTFRR(3, 0)
+struct lyd_node *yang_state_vnewf(struct lyd_node *tree, const char *path, const char *val_fmt,
+				  va_list ap)
+{
+	struct lyd_node *dnode;
+	char *value;
+
+	value = darr_vsprintf(val_fmt, ap);
+	dnode = yang_state_new(tree, path, value);
+	darr_free(value);
+
+	return dnode;
+}
+
+struct lyd_node *yang_state_newf(struct lyd_node *tree, const char *path, const char *val_fmt, ...)
+{
+	struct lyd_node *dnode;
+	va_list ap;
+
+	va_start(ap, val_fmt);
+	dnode = yang_state_vnewf(tree, path, val_fmt, ap);
+	va_end(ap);
+
+	return dnode;
+}
+
 struct yang_data *yang_data_new(const char *xpath, const char *value)
 {
 	struct yang_data *data;

lib/yang.h

@@ -535,6 +535,66 @@ extern struct lyd_node *yang_dnode_dup(const struct lyd_node *dnode);
  */
 extern void yang_dnode_free(struct lyd_node *dnode);
 
+/**
+ * yang_state_new() - Create new state data.
+ * @tree: subtree @path is relative to or NULL in which case @path must be
+ *	  absolute.
+ * @path: The path of the state node to create.
+ * @value: The canonical value of the state.
+ *
+ * Return: The new libyang node.
+ */
+extern struct lyd_node *yang_state_new(struct lyd_node *tree, const char *path, const char *value);
+
+/**
+ * yang_state_delete() - Delete state data.
+ * @tree: subtree @path is relative to or NULL in which case @path must be
+ *	  absolute.
+ * @path: The path of the state node to delete, or NULL if @tree should just be
+ *	  deleted.
+ */
+extern void yang_state_delete(struct lyd_node *tree, const char *path);
+
+/**
+ * yang_state_new_pathf() - Create new state data.
+ * @tree: subtree @path_fmt is relative to or NULL in which case @path_fmt must
+ *        be absolute.
+ * @path_fmt: The path format string of the state node to create.
+ * @value: The canonical value of the state.
+ * @...: The values to substitute into @path_fmt.
+ *
+ * Return: The new libyang node.
+ */
+extern struct lyd_node *yang_state_new_pathf(struct lyd_node *tree, const char *path_fmt,
+					     const char *value, ...) PRINTFRR(2, 4);
+extern struct lyd_node *yang_state_new_vpathf(struct lyd_node *tree, const char *path_fmt,
+					      const char *value, va_list ap);
+/**
+ * yang_state_delete_pathf() - Delete state data.
+ * @tree: subtree @path_fmt is relative to or NULL in which case @path_fmt must
+ *	  be absolute.
+ * @path: The path of the state node to delete.
+ * @...: The values to substitute into @path_fmt.
+ */
+extern void yang_state_delete_pathf(struct lyd_node *tree, const char *path_fmt, ...) PRINTFRR(2, 3);
+extern void yang_state_delete_vpathf(struct lyd_node *tree, const char *path_fmt, va_list ap);
+
+/**
+ * yang_state_newf() - Create new state data.
+ * @tree: subtree @path is relative to or NULL in which case @path must be
+ *	  absolute.
+ * @path: The path of the state node to create.
+ * @val_fmt: The value format string to set the canonical value of the state.
+ * @...: The values to substitute into @val_fmt.
+ *
+ * Return: The new libyang node.
+ */
+extern struct lyd_node *yang_state_newf(struct lyd_node *tree, const char *path,
+					const char *val_fmt, ...) PRINTFRR(3, 4);
+
+extern struct lyd_node *yang_state_vnewf(struct lyd_node *tree, const char *path,
+					 const char *val_fmt, va_list ap);
+
 /*
  * Add a libyang data node to an RPC/action output container.
  *

mgmtd/mgmt_be_adapter.c

@@ -320,7 +320,7 @@ static void mgmt_be_xpath_map_init(void)
 
 	__dbg("Total Cfg XPath Maps: %u", darr_len(be_cfg_xpath_map));
 	__dbg("Total Oper XPath Maps: %u", darr_len(be_oper_xpath_map));
-	__dbg("Total Noitf XPath Maps: %u", darr_len(be_notif_xpath_map));
+	__dbg("Total Notif XPath Maps: %u", darr_len(be_notif_xpath_map));
 	__dbg("Total RPC XPath Maps: %u", darr_len(be_rpc_xpath_map));
 }
 
@@ -651,13 +651,17 @@ int mgmt_be_send_native(enum mgmt_be_client_id id, void *msg)
 	return mgmt_msg_native_send_msg(adapter->conn, msg, false);
 }
 
+/*
+ * Send notification to back-ends that subscribed for them.
+ */
 static void mgmt_be_adapter_send_notify(struct mgmt_msg_notify_data *msg,
 					size_t msglen)
 {
 	struct mgmt_be_client_adapter *adapter;
 	struct mgmt_be_xpath_map *map;
-	struct nb_node *nb_node;
+	struct nb_node *nb_node = NULL;
 	const char *notif;
+	bool is_root;
 	uint id, len;
 
 	if (!darr_len(be_notif_xpath_map))
@@ -669,28 +673,34 @@ static void mgmt_be_adapter_send_notify(struct mgmt_msg_notify_data *msg,
 		return;
 	}
 
+	is_root = !strcmp(notif, "/");
+	if (!is_root) {
 		nb_node = nb_node_find(notif);
 		if (!nb_node) {
 			__log_err("No schema found for notification: %s", notif);
 			return;
 		}
+	}
 
 	darr_foreach_p (be_notif_xpath_map, map) {
+		if (!is_root) {
 			len = strlen(map->xpath_prefix);
 			if (strncmp(map->xpath_prefix, nb_node->xpath, len) &&
 			    strncmp(map->xpath_prefix, notif, len))
 				continue;
-
+		}
 		FOREACH_BE_CLIENT_BITS (id, map->clients) {
 			adapter = mgmt_be_get_adapter_by_id(id);
 			if (!adapter)
 				continue;
+
 			msg_conn_send_msg(adapter->conn, MGMT_MSG_VERSION_NATIVE,
 					  msg, msglen, NULL, false);
 		}
 	}
 }
 
+
 /*
  * Handle a native encoded message
  */
@@ -735,6 +745,9 @@ static void be_adapter_handle_native_msg(struct mgmt_be_client_adapter *adapter,
 		mgmt_txn_notify_rpc_reply(adapter, rpc_msg, msg_len);
 		break;
 	case MGMT_MSG_CODE_NOTIFY:
+		/*
+		 * Handle notify message from a back-end client
+		 */
 		notify_msg = (typeof(notify_msg))msg;
 		__dbg("Got NOTIFY from '%s'", adapter->name);
 		mgmt_be_adapter_send_notify(notify_msg, msg_len);

mgmtd/mgmt_fe_adapter.c

@@ -9,6 +9,7 @@
 
 #include <zebra.h>
 #include "darr.h"
+#include "frrstr.h"
 #include "sockopt.h"
 #include "network.h"
 #include "libfrr.h"
@@ -31,6 +32,7 @@
 #define FOREACH_ADAPTER_IN_LIST(adapter)                                       \
 	frr_each_safe (mgmt_fe_adapters, &mgmt_fe_adapters, (adapter))
 
+
 enum mgmt_session_event {
 	MGMTD_FE_SESSION_CFG_TXN_CLNUP = 1,
 	MGMTD_FE_SESSION_SHOW_TXN_CLNUP,
@@ -55,6 +57,22 @@ DECLARE_LIST(mgmt_fe_sessions, struct mgmt_fe_session_ctx, list_linkage);
 #define FOREACH_SESSION_IN_LIST(adapter, session)                              \
 	frr_each_safe (mgmt_fe_sessions, &(adapter)->fe_sessions, (session))
 
+/*
+ * A tree for storing unique notify-select strings.
+ */
+PREDECL_RBTREE_UNIQ(ns_string);
+struct ns_string {
+	struct ns_string_item link;
+	struct list *sessions;
+	char s[];
+};
+static uint32_t ns_string_compare(const struct ns_string *ns1, const struct ns_string *ns2);
+DECLARE_RBTREE_UNIQ(ns_string, struct ns_string, link, ns_string_compare);
+
+/* ---------------- */
+/* Global variables */
+/* ---------------- */
+
 static struct event_loop *mgmt_loop;
 static struct msg_server mgmt_fe_server = {.fd = -1};
 
@@ -63,6 +81,72 @@ static struct mgmt_fe_adapters_head mgmt_fe_adapters;
 static struct hash *mgmt_fe_sessions;
 static uint64_t mgmt_fe_next_session_id;
 
+static struct ns_string_head mgmt_fe_ns_strings;
+
+/* ------------------------------ */
+/* Notify select string functions */
+/* ------------------------------ */
+
+static uint32_t ns_string_compare(const struct ns_string *ns1, const struct ns_string *ns2)
+{
+	return strcmp(ns1->s, ns2->s);
+}
+
+static void mgmt_fe_free_ns_string(struct ns_string *ns)
+{
+	list_delete(&ns->sessions);
+	XFREE(MTYPE_MGMTD_XPATH, ns);
+}
+
+static void mgmt_fe_free_ns_strings(struct ns_string_head *head)
+{
+	struct ns_string *ns;
+
+	while ((ns = ns_string_pop(head)))
+		mgmt_fe_free_ns_string(ns);
+	ns_string_fini(head);
+}
+
+static void mgmt_fe_ns_string_remove_session(struct ns_string_head *head,
+					     struct mgmt_fe_session_ctx *session)
+{
+	struct ns_string *ns;
+
+	frr_each_safe (ns_string, head, ns) {
+		listnode_delete(ns->sessions, session);
+		if (list_isempty(ns->sessions)) {
+			ns_string_del(head, ns);
+			mgmt_fe_free_ns_string(ns);
+		}
+	}
+}
+
+static void 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;
+
+	ns = XCALLOC(MTYPE_MGMTD_XPATH, sizeof(*ns) + plen + 1);
+	strlcpy(ns->s, path, plen + 1);
+	e = ns_string_add(head, ns);
+	if (!e)
+		ns->sessions = list_new();
+	if (!listnode_lookup(ns->sessions, session))
+		listnode_add(ns->sessions, session);
+}
+
+char **mgmt_fe_get_all_selectors(void)
+{
+	char **selectors = NULL;
+	struct ns_string *ns;
+
+	frr_each (ns_string, &mgmt_fe_ns_strings, ns)
+		*darr_append(selectors) = darr_strdup(ns->s);
+
+	return selectors;
+}
+
+
 /* Forward declarations */
 static void
 mgmt_fe_session_register_event(struct mgmt_fe_session_ctx *session,
@@ -190,6 +274,7 @@ static void mgmt_fe_cleanup_session(struct mgmt_fe_session_ctx **sessionp)
 		assert(session->adapter->refcount > 1);
 		mgmt_fe_adapter_unlock(&session->adapter);
 	}
+	mgmt_fe_ns_string_remove_session(&mgmt_fe_ns_strings, session);
 	darr_free_free(session->notify_xpaths);
 	hash_release(mgmt_fe_sessions, session);
 	XFREE(MTYPE_MGMTD_FE_SESSION, session);
@@ -1542,32 +1627,90 @@ static void fe_adapter_handle_edit(struct mgmt_fe_session_ctx *session,
  * @__msg: the message data.
  * @msg_len: the length of the message data.
  */
-static void fe_adapter_handle_notify_select(struct mgmt_fe_session_ctx *session,
+static void fe_adapter_handle_notify_select(struct mgmt_fe_session_ctx *session, void *__msg,
-					    void *__msg, size_t msg_len)
+					    size_t msg_len)
 {
 	struct mgmt_msg_notify_select *msg = __msg;
 	uint64_t req_id = msg->req_id;
 	const char **selectors = NULL;
 	const char **new;
+	const char **sp;
+	char *selstr = NULL;
+	uint64_t clients = 0;
+	uint ret;
 
 	if (msg_len >= sizeof(*msg)) {
-		selectors = mgmt_msg_native_strings_decode(msg, msg_len,
+		selectors = mgmt_msg_native_strings_decode(msg, msg_len, msg->selectors);
-							   msg->selectors);
 		if (!selectors) {
-			fe_adapter_send_error(session, req_id, false, -EINVAL,
+			fe_adapter_send_error(session, req_id, false, -EINVAL, "Invalid message");
-					      "Invalid message");
 			return;
 		}
 	}
+	if (DEBUG_MODE_CHECK(&mgmt_debug_fe, DEBUG_MODE_ALL)) {
+		selstr = frrstr_join(selectors, darr_len(selectors), ", ");
+		if (!selstr)
+			selstr = XSTRDUP(MTYPE_TMP, "");
+	}
+
 	if (msg->replace) {
+		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);
 		darr_free_free(session->notify_xpaths);
 		session->notify_xpaths = selectors;
 	} else if (selectors) {
-		new = darr_append_nz(session->notify_xpaths,
+		// [ ] Keep a local tree to optimize sending selectors to BE?
-				     darr_len(selectors));
+		// [*] Or just KISS and fanout the original message to BEs?
+		// mgmt_remove_add_notify_selectors(session->notify_xpaths, selectors);
+		new = darr_append_nz(session->notify_xpaths, darr_len(selectors));
 		memcpy(new, selectors, darr_len(selectors) * sizeof(*selectors));
-		darr_free(selectors);
+	} else {
+		__log_err("Invalid msg from session-id: %Lu: no selectors present in non-replace msg",
+			  session->session_id);
+		darr_free_free(selectors);
+		selectors = NULL;
+		goto done;
 	}
+
+
+	if (session->notify_xpaths && DEBUG_MODE_CHECK(&mgmt_debug_fe, DEBUG_MODE_ALL)) {
+		const char **sel = session->notify_xpaths;
+		char *s = frrstr_join(sel, darr_len(sel), ", ");
+		__dbg("New NOTIF %d selectors '%s' (replace: %d) txn-id: %Lu for session-id: %Lu",
+		      darr_len(sel), s, msg->replace, session->cfg_txn_id, session->session_id);
+		XFREE(MTYPE_TMP, s);
+	}
+
+	/* 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);
+
+	/* 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",
+		      selstr, session->txn_id, session->session_id);
+		goto done;
+	}
+
+	/* We don't use a transaction for this, just send the message */
+	ret = mgmt_txn_send_notify_selectors(req_id, clients, msg->replace ? NULL : selectors);
+	if (ret) {
+		fe_adapter_send_error(session, req_id, false, -EINPROGRESS,
+				      "Failed to create a NOTIFY_SELECT transaction");
+	}
+done:
+	if (session->notify_xpaths != selectors)
+		darr_free(selectors);
+	if (selstr)
+		XFREE(MTYPE_TMP, selstr);
 }
 
 /**
@@ -1758,10 +1901,11 @@ void mgmt_fe_adapter_send_notify(struct mgmt_msg_notify_data *msg, size_t msglen
 {
 	struct mgmt_fe_client_adapter *adapter;
 	struct mgmt_fe_session_ctx *session;
-	struct nb_node *nb_node;
+	struct nb_node *nb_node = NULL;
-	const char **xpath_prefix;
+	struct listnode *node;
+	struct ns_string *ns;
 	const char *notif;
-	bool sendit;
+	bool is_root;
 	uint len;
 
 	assert(msg->refer_id == 0);
@@ -1772,6 +1916,8 @@ void mgmt_fe_adapter_send_notify(struct mgmt_msg_notify_data *msg, size_t msglen
 		return;
 	}
 
+	is_root = !strcmp(notif, "/");
+	if (!is_root) {
 		/*
 		 * We need the nb_node to obtain a path which does not include any
 		 * specific list entry selectors
@@ -1781,27 +1927,37 @@ void mgmt_fe_adapter_send_notify(struct mgmt_msg_notify_data *msg, size_t msglen
 			__log_err("No schema found for notification: %s", notif);
 			return;
 		}
+	}
 
+	frr_each (ns_string, &mgmt_fe_ns_strings, ns) {
+		if (!is_root) {
+			len = strlen(ns->s);
+			if (strncmp(ns->s, notif, len) && strncmp(ns->s, nb_node->xpath, len))
+				continue;
+		}
+		for (ALL_LIST_ELEMENTS_RO(ns->sessions, node, session)) {
+			msg->refer_id = session->session_id;
+			(void)fe_adapter_send_native_msg(session->adapter, msg, msglen, false);
+		}
+	}
+
+	/*
+	 * Send all YANG defined notifications to all sesisons with *no*
+	 * selectors as well (i.e., original NETCONF/RESTCONF notification
+	 * scheme).
+	 */
+	if (!is_root && CHECK_FLAG(nb_node->snode->nodetype, LYS_NOTIF)) {
 		FOREACH_ADAPTER_IN_LIST (adapter) {
 			FOREACH_SESSION_IN_LIST (adapter, session) {
-			/* If no selectors then always send */
+				if (session->notify_xpaths)
-			sendit = !session->notify_xpaths;
+					continue;
-			darr_foreach_p (session->notify_xpaths, xpath_prefix) {
-				len = strlen(*xpath_prefix);
-				if (!strncmp(*xpath_prefix, notif, len) ||
-				    !strncmp(*xpath_prefix, nb_node->xpath,
-					     len)) {
-					sendit = true;
-					break;
-				}
-			}
-			if (sendit) {
 				msg->refer_id = session->session_id;
 				(void)fe_adapter_send_native_msg(adapter, msg,
 								 msglen, false);
 			}
 		}
 	}
+
 	msg->refer_id = 0;
 }
 
@@ -1810,9 +1966,10 @@ void mgmt_fe_adapter_lock(struct mgmt_fe_client_adapter *adapter)
 	adapter->refcount++;
 }
 
-extern void mgmt_fe_adapter_unlock(struct mgmt_fe_client_adapter **adapter)
+void mgmt_fe_adapter_unlock(struct mgmt_fe_client_adapter **adapter)
 {
 	struct mgmt_fe_client_adapter *a = *adapter;
+
 	assert(a && a->refcount);
 
 	if (!--a->refcount) {
@@ -1840,6 +1997,8 @@ void mgmt_fe_adapter_init(struct event_loop *tm)
 		hash_create(mgmt_fe_session_hash_key, mgmt_fe_session_hash_cmp,
 			    "MGMT Frontend Sessions");
 
+	ns_string_init(&mgmt_fe_ns_strings);
+
 	snprintf(server_path, sizeof(server_path), MGMTD_FE_SOCK_NAME);
 
 	if (msg_server_init(&mgmt_fe_server, server_path, tm,
@@ -1869,10 +2028,13 @@ void mgmt_fe_adapter_destroy(void)
 
 	msg_server_cleanup(&mgmt_fe_server);
 
+
 	/* Deleting the adapters will delete all the sessions */
 	FOREACH_ADAPTER_IN_LIST (adapter)
 		mgmt_fe_adapter_delete(adapter);
 
+	mgmt_fe_free_ns_strings(&mgmt_fe_ns_strings);
+
 	hash_clean_and_free(&mgmt_fe_sessions, mgmt_fe_abort_if_session);
 }
 
@@ -1885,8 +2047,7 @@ struct msg_conn *mgmt_fe_create_adapter(int conn_fd, union sockunion *from)
 
 	adapter = mgmt_fe_find_adapter_by_fd(conn_fd);
 	if (!adapter) {
-		adapter = XCALLOC(MTYPE_MGMTD_FE_ADPATER,
+		adapter = XCALLOC(MTYPE_MGMTD_FE_ADPATER, sizeof(struct mgmt_fe_client_adapter));
-				sizeof(struct mgmt_fe_client_adapter));
 		snprintf(adapter->name, sizeof(adapter->name), "Unknown-FD-%d",
 			 conn_fd);
 

mgmtd/mgmt_fe_adapter.h

@@ -225,6 +225,13 @@ extern int mgmt_fe_adapter_txn_error(uint64_t txn_id, uint64_t req_id,
 				     const char *errstr);
 
 
+/**
+ * mgmt_fe_get_all_selectors() - Get all selectors for all frontend adapters.
+ *
+ * Returns: A darr array of all selectors for all frontend adapters.
+ */
+extern char **mgmt_fe_get_all_selectors(void);
+
 /* Fetch frontend client session set-config stats */
 extern struct mgmt_setcfg_stats *
 mgmt_fe_get_session_setcfg_stats(uint64_t session_id);

mgmtd/mgmt_history.c

@@ -177,6 +177,7 @@ static bool mgmt_history_dump_cmt_record_index(void)
 		return false;
 	}
 
+	assert(cnt <= 10); /* silence bad CLANG SA warning */
 	ret = fwrite(&cmt_info_set, sizeof(struct mgmt_cmt_info_t), cnt, fp);
 	fclose(fp);
 	if (ret != cnt) {

mgmtd/mgmt_txn.c

@@ -237,6 +237,7 @@ struct mgmt_txn_ctx {
 	struct event *clnup;
 
 	/* List of backend adapters involved in this transaction */
+	/* XXX reap this */
 	struct mgmt_txn_badapters_head be_adapters;
 
 	int refcount;
@@ -2651,6 +2652,52 @@ int mgmt_txn_send_rpc(uint64_t txn_id, uint64_t req_id, uint64_t clients,
 	return 0;
 }
 
+int mgmt_txn_send_notify_selectors(uint64_t req_id, uint64_t clients, const char **selectors)
+{
+	struct mgmt_msg_notify_select *msg;
+	char **all_selectors = NULL;
+	uint64_t id;
+	int ret;
+	uint i;
+
+	msg = mgmt_msg_native_alloc_msg(struct mgmt_msg_notify_select, 0,
+					MTYPE_MSG_NATIVE_NOTIFY_SELECT);
+	msg->refer_id = MGMTD_TXN_ID_NONE;
+	msg->req_id = req_id;
+	msg->code = MGMT_MSG_CODE_NOTIFY_SELECT;
+	msg->replace = selectors == NULL;
+
+	if (selectors == NULL) {
+		/* Get selectors for all sessions */
+		all_selectors = mgmt_fe_get_all_selectors();
+		selectors = (const char **)all_selectors;
+	}
+
+	darr_foreach_i (selectors, i)
+		mgmt_msg_native_add_str(msg, selectors[i]);
+
+	assert(clients);
+	FOREACH_BE_CLIENT_BITS (id, clients) {
+		/* make sure the backend is running/connected */
+		if (!mgmt_be_get_adapter_by_id(id))
+			continue;
+		ret = mgmt_be_send_native(id, msg);
+		if (ret) {
+			__log_err("Could not send notify-select message to backend client %s",
+				  mgmt_be_client_id2name(id));
+			continue;
+		}
+
+		__dbg("Sent notify-select req to backend client %s", mgmt_be_client_id2name(id));
+	}
+	mgmt_msg_native_free_msg(msg);
+
+	if (all_selectors)
+		darr_free_free(all_selectors);
+
+	return 0;
+}
+
 /*
  * Error reply from the backend client.
  */

mgmtd/mgmt_txn.h

@@ -297,6 +297,16 @@ extern int mgmt_txn_send_rpc(uint64_t txn_id, uint64_t req_id, uint64_t clients,
 			     LYD_FORMAT result_type, const char *xpath,
 			     const char *data, size_t data_len);
 
+/**
+ * mgmt_txn_send_notify_selectors() - Send NOTIFY SELECT request.
+ * @req_id: FE client request identifier.
+ * @clients: Bitmask of clients to send RPC to.
+ * @selectors: Array of selectors or NULL to resend all selectors to BE clients.
+ *
+ * Returns 0 on success.
+ */
+extern int mgmt_txn_send_notify_selectors(uint64_t req_id, uint64_t clients, const char **selectors);
+
 /*
  * Notifiy backend adapter on connection.
  */

tests/topotests/conftest.py

@@ -31,7 +31,7 @@ from lib import topolog, topotest
 
 try:
     # Used by munet native tests
-    from munet.testing.fixtures import unet  # pylint: disable=all # noqa
+    from munet.testing.fixtures import stepf, unet  # pylint: disable=all # noqa
 
     @pytest.fixture(scope="module")
     def rundir_module(pytestconfig):

tests/topotests/mgmt_notif/r1/frr.conf

@@ -4,7 +4,7 @@ log file frr.log
 no debug memstats-at-exit
 
 debug northbound notifications
-debug northbound libyang
+!! debug northbound libyang
 debug northbound events
 debug northbound callbacks
 

tests/topotests/mgmt_notif/r2/frr.conf

@@ -16,7 +16,7 @@ ip route 22.22.22.22/32 lo
 
 interface r2-eth0
  ip address 1.1.1.2/24
- ip rip authentication string bar
+ ip rip authentication string foo
  ip rip authentication mode text
 exit
 

tests/topotests/mgmt_notif/test_notif.py

@@ -10,9 +10,12 @@
 Test 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
@@ -42,6 +45,99 @@ 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)
@@ -50,30 +146,98 @@ def test_frontend_notification(tgen):
 
     check_kernel_32(r1, "11.11.11.11", 1, "")
 
-    fe_client_path = CWD + "/../lib/fe_client.py --verbose"
+    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")
 
-    # The first notifications is a frr-ripd:authentication-type-failure
+    # Update config to non-matching authentication.
-    # So we filter to avoid that, all the rest are frr-ripd:authentication-failure
+    conf = """
-    # making our test deterministic
+    conf t
+    interface r1-eth0
+    ip rip authentication string bar
+    """
+    r1.cmd_raises("vtysh", stdin=conf)
+
+    try:
         output = r1.cmd_raises(
             fe_client_path + " --listen /frr-ripd:authentication-failure"
         )
-    jsout = json.loads(output)
 
+        jsout = json.loads(output)
         expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
         result = json_cmp(jsout, expected)
         assert result is None
 
-    output = r1.cmd_raises(fe_client_path + " --use-protobuf --listen")
+        output = r1.cmd_raises(
+            fe_client_path + " --use-protobuf --listen /frr-ripd:authentication-failure"
+        )
         jsout = json.loads(output)
-
         expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
         result = json_cmp(jsout, expected)
         assert result is None
+    finally:
+        # Update config to matching authentication.
+        conf = """
+        conf t
+        interface r1-eth0
+        ip rip authentication string foo
+        """
+        r1.cmd_raises("vtysh", stdin=conf)
+
+
+def test_frontend_all_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")
+
+    # Update config to non-matching authentication.
+    conf = """
+    conf t
+    interface r1-eth0
+    ip rip authentication string bar
+    """
+    r1.cmd_raises("vtysh", stdin=conf)
+
+    try:
+        # The first notifications is a frr-ripd:authentication-type-failure
+        # All the rest are frr-ripd:authentication-failure so we check for both.
+        output = r1.cmd_raises(fe_client_path + " --listen /")
+        jsout = json.loads(output)
+        expected = {
+            "frr-ripd:authentication-type-failure": {"interface-name": "r1-eth0"}
+        }
+        result = json_cmp(jsout, expected)
+        if result is not None:
+            expected = {
+                "frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}
+            }
+            result = json_cmp(jsout, expected)
+        assert result is None
+
+        output = r1.cmd_raises(fe_client_path + " --use-protobuf --listen /")
+        jsout = json.loads(output)
+        expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
+        result = json_cmp(jsout, expected)
+        assert result is None
+    finally:
+        # Update config to matching authentication.
+        conf = """
+        conf t
+        interface r1-eth0
+        ip rip authentication string foo
+        """
+        r1.cmd_raises("vtysh", stdin=conf)
 
 
 def test_backend_notification(tgen):
@@ -90,12 +254,28 @@ def test_backend_notification(tgen):
     if rc:
         pytest.skip("No mgmtd_testc")
 
+    # Update config to non-matching authentication.
+    conf = """
+    conf t
+    interface r1-eth0
+    ip rip authentication string bar
+    """
+    r1.cmd_raises("vtysh", stdin=conf)
+
+    try:
         output = r1.cmd_raises(
-        be_client_path + " --timeout 20 --log file:mgmt_testc.log --listen /frr-ripd"
+            be_client_path
+            + " --timeout 20 --log file:mgmt_testc.log --listen /frr-ripd"
         )
-
         jsout = json.loads(output)
-
         expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
         result = json_cmp(jsout, expected)
         assert result is None
+    finally:
+        # Update config to matching authentication.
+        conf = """
+        conf t
+        interface r1-eth0
+        ip rip authentication string foo
+        """
+        r1.cmd_raises("vtysh", stdin=conf)

zebra/interface.c

@@ -81,7 +81,7 @@ static void if_zebra_speed_update(struct event *thread)
 	if (new_speed != ifp->speed) {
 		zlog_info("%s: %s old speed: %u new speed: %u", __func__,
 			  ifp->name, ifp->speed, new_speed);
-		ifp->speed = new_speed;
+		if_update_state_speed(ifp, new_speed);
 		if_add_update(ifp);
 		changed = true;
 	}
@@ -1563,17 +1563,20 @@ static inline void zebra_if_set_ziftype(struct interface *ifp,
 static void interface_update_hw_addr(struct zebra_dplane_ctx *ctx,
 				     struct interface *ifp)
 {
-	int i;
+	uint8_t hw_addr[INTERFACE_HWADDR_MAX];
+	uint i, hw_addr_len;
 
-	ifp->hw_addr_len = dplane_ctx_get_ifp_hw_addr_len(ctx);
+	hw_addr_len = dplane_ctx_get_ifp_hw_addr_len(ctx);
-	memcpy(ifp->hw_addr, dplane_ctx_get_ifp_hw_addr(ctx), ifp->hw_addr_len);
+	memcpy(hw_addr, dplane_ctx_get_ifp_hw_addr(ctx), hw_addr_len);
 
-	for (i = 0; i < ifp->hw_addr_len; i++)
+	for (i = 0; i < hw_addr_len; i++)
-		if (ifp->hw_addr[i] != 0)
+		if (hw_addr[i] != 0)
 			break;
 
-	if (i == ifp->hw_addr_len)
+	if (i == hw_addr_len)
-		ifp->hw_addr_len = 0;
+		hw_addr_len = 0;
+
+	if_update_state_hw_addr(ifp, hw_addr, hw_addr_len);
 }
 
 static void interface_update_l2info(struct zebra_dplane_ctx *ctx,
@@ -1984,9 +1987,10 @@ static void zebra_if_dplane_ifp_handling(struct zebra_dplane_ctx *ctx)
 			/* Update interface information. */
 			set_ifindex(ifp, ifindex, zns);
 			ifp->flags = flags;
-			ifp->mtu6 = ifp->mtu = mtu;
+			if_update_state_mtu(ifp, mtu);
-			ifp->metric = 0;
+			if_update_state_mtu6(ifp, mtu);
-			ifp->speed = kernel_get_speed(ifp, NULL);
+			if_update_state_metric(ifp, 0);
+			if_update_state_speed(ifp, kernel_get_speed(ifp, NULL));
 			ifp->ptm_status = ZEBRA_PTM_STATUS_UNKNOWN;
 			ifp->txqlen = dplane_ctx_get_intf_txqlen(ctx);
 
@@ -2036,6 +2040,7 @@ static void zebra_if_dplane_ifp_handling(struct zebra_dplane_ctx *ctx)
 						IS_ZEBRA_IF_BRIDGE_VLAN_AWARE(
 							zif));
 			}
+			// if_update_state(ifp);
 		} else if (ifp->vrf->vrf_id != vrf_id) {
 			/* VRF change for an interface. */
 			if (IS_ZEBRA_DEBUG_KERNEL)
@@ -2058,8 +2063,9 @@ static void zebra_if_dplane_ifp_handling(struct zebra_dplane_ctx *ctx)
 					   (unsigned long long)flags);
 
 			set_ifindex(ifp, ifindex, zns);
-			ifp->mtu6 = ifp->mtu = mtu;
+			if_update_state_mtu(ifp, mtu);
-			ifp->metric = 0;
+			if_update_state_mtu6(ifp, mtu);
+			if_update_state_metric(ifp, 0);
 			ifp->txqlen = dplane_ctx_get_intf_txqlen(ctx);
 
 			/*

zebra/main.c

@@ -356,6 +356,9 @@ int main(int argc, char **argv)
 
 	zserv_path = NULL;
 
+	if_notify_oper_changes = true;
+	vrf_notify_oper_changes = true;
+
 	vrf_configure_backend(VRF_BACKEND_VRF_LITE);
 
 	frr_preinit(&zebra_di, argc, argv);