Merge pull request #15355 from idryzhov/fix-mgmtd-notif

Fix and rework YANG notifications

lib/mgmt_be_client.c

@@ -963,22 +963,29 @@ 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 mgmt_be_client_notification_cb *cb;
-	const char *notif;
-	uint i;
+	struct nb_node *nb_node;
+	char notif[XPATH_MAXLEN];
+	struct lyd_node *dnode;
+	LY_ERR err;
 
 	debug_be_client("Received notification for client %s", client->name);
 
-	/* "{\"modname:notification-name\": ...}" */
-	notif = (const char *)notif_msg->result + 2;
+	err = yang_parse_notification(notif_msg->result_type,
+				      (char *)notif_msg->result, &dnode);
+	if (err)
+		return;
 
-	for (i = 0; i < client->cbs.nnotify_cbs; i++) {
-		cb = &client->cbs.notify_cbs[i];
-		if (strncmp(cb->xpath, notif, strlen(cb->xpath)))
-			continue;
-		cb->callback(client, client->user_data, cb,
-			     (const char *)notif_msg->result);
+	lysc_path(dnode->schema, LYSC_PATH_DATA, notif, sizeof(notif));
+
+	nb_node = nb_node_find(notif);
+	if (!nb_node || !nb_node->cbs.notify) {
+		debug_be_client("No notification callback for %s", notif);
+		goto cleanup;
 	}
+
+	nb_callback_notify(nb_node, notif, dnode);
+cleanup:
+	lyd_free_all(dnode);
 }
 
 /*
@@ -1049,8 +1056,6 @@ int mgmt_be_send_subscr_req(struct mgmt_be_client *client_ctx,
 {
 	Mgmtd__BeMessage be_msg;
 	Mgmtd__BeSubscribeReq subscr_req;
-	const char **notif_xpaths = NULL;
-	int ret;
 
 	mgmtd__be_subscribe_req__init(&subscr_req);
 	subscr_req.client_name = client_ctx->name;
@@ -1060,16 +1065,8 @@ int mgmt_be_send_subscr_req(struct mgmt_be_client *client_ctx,
 	subscr_req.oper_xpaths = oper_xpaths;
 
 	/* See if we should register for notifications */
-	subscr_req.n_notif_xpaths = client_ctx->cbs.nnotify_cbs;
-	if (client_ctx->cbs.nnotify_cbs) {
-		struct mgmt_be_client_notification_cb *cb, *ecb;
-
-		cb = client_ctx->cbs.notify_cbs;
-		ecb = cb + client_ctx->cbs.nnotify_cbs;
-		for (; cb < ecb; cb++)
-			*darr_append(notif_xpaths) = cb->xpath;
-	}
-	subscr_req.notif_xpaths = (char **)notif_xpaths;
+	subscr_req.n_notif_xpaths = client_ctx->cbs.nnotif_xpaths;
+	subscr_req.notif_xpaths = (char **)client_ctx->cbs.notif_xpaths;
 
 	mgmtd__be_message__init(&be_msg);
 	be_msg.message_case = MGMTD__BE_MESSAGE__MESSAGE_SUBSCR_REQ;
@@ -1079,9 +1076,7 @@ int mgmt_be_send_subscr_req(struct mgmt_be_client *client_ctx,
 			subscr_req.client_name, subscr_req.n_config_xpaths,
 			subscr_req.n_oper_xpaths, subscr_req.n_notif_xpaths);
 
-	ret = mgmt_be_client_send_msg(client_ctx, &be_msg);
-	darr_free(notif_xpaths);
-	return ret;
+	return mgmt_be_client_send_msg(client_ctx, &be_msg);
 }
 
 static int _notify_conenct_disconnect(struct msg_client *msg_client,

lib/mgmt_be_client.h

@@ -73,16 +73,8 @@ struct mgmt_be_client_cbs {
 			   struct mgmt_be_client_txn_ctx *txn_ctx,
 			   bool destroyed);
 
-	struct mgmt_be_client_notification_cb *notify_cbs;
-	uint nnotify_cbs;
-};
-
-struct mgmt_be_client_notification_cb {
-	const char *xpath; /* the notification */
-	uint8_t format;	   /* currently only LYD_JSON supported */
-	void (*callback)(struct mgmt_be_client *client, uintptr_t usr_data,
-			 struct mgmt_be_client_notification_cb *this,
-			 const char *notif_data);
+	const char **notif_xpaths;
+	uint nnotif_xpaths;
 };
 
 /***************************************************************

lib/northbound.c

@@ -284,6 +284,8 @@ static unsigned int nb_node_validate_cbs(const struct nb_node *nb_node)
 				     !!nb_node->cbs.lookup_entry, false);
 	error += nb_node_validate_cb(nb_node, NB_CB_RPC, !!nb_node->cbs.rpc,
 				     false);
+	error += nb_node_validate_cb(nb_node, NB_CB_NOTIFY,
+				     !!nb_node->cbs.notify, true);
 
 	return error;
 }
@@ -1605,6 +1607,18 @@ 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,
+			struct lyd_node *dnode)
+{
+	struct nb_cb_notify_args args = {};
+
+	DEBUGD(&nb_dbg_cbs_notify, "northbound notify: %s", xpath);
+
+	args.xpath = xpath;
+	args.dnode = dnode;
+	nb_node->cbs.notify(&args);
+}
+
 /*
  * Call the northbound configuration callback associated to a given
  * configuration change.
@@ -1653,6 +1667,7 @@ static int nb_callback_configuration(struct nb_context *context,
 	case NB_CB_GET_KEYS:
 	case NB_CB_LOOKUP_ENTRY:
 	case NB_CB_RPC:
+	case NB_CB_NOTIFY:
 		yang_dnode_get_path(dnode, xpath, sizeof(xpath));
 		flog_err(EC_LIB_DEVELOPMENT,
 			 "%s: unknown operation (%u) [xpath %s]", __func__,
@@ -2047,6 +2062,10 @@ bool nb_cb_operation_is_valid(enum nb_cb_operation operation,
 			return false;
 		}
 		return true;
+	case NB_CB_NOTIFY:
+		if (snode->nodetype != LYS_NOTIF)
+			return false;
+		return true;
 	default:
 		return false;
 	}
@@ -2279,6 +2298,8 @@ const char *nb_cb_operation_name(enum nb_cb_operation operation)
 		return "lookup_entry";
 	case NB_CB_RPC:
 		return "rpc";
+	case NB_CB_NOTIFY:
+		return "notify";
 	}
 
 	assert(!"Reached end of function we should never hit");

lib/northbound.h

@@ -99,6 +99,7 @@ enum nb_cb_operation {
 	NB_CB_GET_KEYS,
 	NB_CB_LOOKUP_ENTRY,
 	NB_CB_RPC,
+	NB_CB_NOTIFY,
 };
 
 union nb_resource {
@@ -286,6 +287,18 @@ struct nb_cb_rpc_args {
 	size_t errmsg_len;
 };
 
+struct nb_cb_notify_args {
+	/* XPath of the notification. */
+	const char *xpath;
+
+	/*
+	 * libyang data node representing the notification. If the notification
+	 * is not top-level, it still points to the notification node, but it's
+	 * part of the full data tree with all its parents.
+	 */
+	struct lyd_node *dnode;
+};
+
 /*
  * Set of configuration callbacks that can be associated to a northbound node.
  */
@@ -509,6 +522,17 @@ struct nb_callbacks {
 	 */
 	int (*rpc)(struct nb_cb_rpc_args *args);
 
+	/*
+	 * Notification callback.
+	 *
+	 * The callback is called when a YANG notification is received.
+	 *
+	 * args
+	 *    Refer to the documentation comments of nb_cb_notify_args for
+	 *    details.
+	 */
+	void (*notify)(struct nb_cb_notify_args *args);
+
 	/*
 	 * Optional callback to compare the data nodes when printing
 	 * the CLI commands associated with them.
@@ -786,6 +810,7 @@ DECLARE_HOOK(nb_client_debug_set_all, (uint32_t flags, bool set), (flags, set));
 extern struct debug nb_dbg_cbs_config;
 extern struct debug nb_dbg_cbs_state;
 extern struct debug nb_dbg_cbs_rpc;
+extern struct debug nb_dbg_cbs_notify;
 extern struct debug nb_dbg_notif;
 extern struct debug nb_dbg_events;
 extern struct debug nb_dbg_libyang;
@@ -814,6 +839,8 @@ 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 list *input, struct list *output,
 			   char *errmsg, size_t errmsg_len);
+extern void nb_callback_notify(const struct nb_node *nb_node, const char *xpath,
+			       struct lyd_node *dnode);
 
 /*
  * Create a northbound node for all YANG schema nodes.

lib/northbound_cli.c

@@ -25,6 +25,7 @@
 struct debug nb_dbg_cbs_config = {0, "Northbound callbacks: configuration"};
 struct debug nb_dbg_cbs_state = {0, "Northbound callbacks: state"};
 struct debug nb_dbg_cbs_rpc = {0, "Northbound callbacks: RPCs"};
+struct debug nb_dbg_cbs_notify = {0, "Northbound callbacks: notifications"};
 struct debug nb_dbg_notif = {0, "Northbound notifications"};
 struct debug nb_dbg_events = {0, "Northbound events"};
 struct debug nb_dbg_libyang = {0, "libyang debugging"};
@@ -1772,13 +1773,15 @@ DEFPY (rollback_config,
 /* Debug CLI commands. */
 static struct debug *nb_debugs[] = {
 	&nb_dbg_cbs_config, &nb_dbg_cbs_state, &nb_dbg_cbs_rpc,
-	&nb_dbg_notif,      &nb_dbg_events,    &nb_dbg_libyang,
+	&nb_dbg_cbs_notify, &nb_dbg_notif,     &nb_dbg_events,
+	&nb_dbg_libyang,
 };
 
 static const char *const nb_debugs_conflines[] = {
 	"debug northbound callbacks configuration",
 	"debug northbound callbacks state",
 	"debug northbound callbacks rpc",
+	"debug northbound callbacks notify",
 	"debug northbound notifications",
 	"debug northbound events",
 	"debug northbound libyang",
@@ -1803,7 +1806,7 @@ DEFPY (debug_nb,
        debug_nb_cmd,
        "[no] debug northbound\
           [<\
-	    callbacks$cbs [{configuration$cbs_cfg|state$cbs_state|rpc$cbs_rpc}]\
+	    callbacks$cbs [{configuration$cbs_cfg|state$cbs_state|rpc$cbs_rpc|notify$cbs_notify}]\
 	    |notifications$notifications\
 	    |events$events\
 	    |libyang$libyang\
@@ -1816,20 +1819,23 @@ DEFPY (debug_nb,
        "State\n"
        "RPC\n"
        "Notifications\n"
+       "Notifications\n"
        "Events\n"
        "libyang debugging\n")
 {
 	uint32_t mode = DEBUG_NODE2MODE(vty->node);
 
 	if (cbs) {
-		bool none = (!cbs_cfg && !cbs_state && !cbs_rpc);
+		bool none = (!cbs_cfg && !cbs_state && !cbs_rpc && !cbs_notify);
 
 		if (none || cbs_cfg)
 			DEBUG_MODE_SET(&nb_dbg_cbs_config, mode, !no);
 		if (none || cbs_state)
 			DEBUG_MODE_SET(&nb_dbg_cbs_state, mode, !no);
 		if (none || cbs_rpc)
 			DEBUG_MODE_SET(&nb_dbg_cbs_rpc, mode, !no);
+		if (none || cbs_notify)
+			DEBUG_MODE_SET(&nb_dbg_cbs_notify, mode, !no);
 	}
 	if (notifications)
 		DEBUG_MODE_SET(&nb_dbg_notif, mode, !no);

lib/yang.c

@@ -714,6 +714,52 @@ static void ly_log_cb(LY_LOG_LEVEL level, const char *msg, const char *path)
 		zlog(priority, "libyang: %s", msg);
 }
 
+LY_ERR yang_parse_notification(LYD_FORMAT format, const char *data,
+			       struct lyd_node **notif)
+{
+	struct lyd_node *tree, *dnode;
+	struct ly_in *in = NULL;
+	bool found = false;
+	LY_ERR err;
+
+	err = ly_in_new_memory(data, &in);
+	if (err) {
+		zlog_err("Failed to initialize ly_in: %s", ly_last_errmsg());
+		return err;
+	}
+
+	err = lyd_parse_op(ly_native_ctx, NULL, in, format, LYD_TYPE_NOTIF_YANG,
+			   &tree, NULL);
+	if (err) {
+		zlog_err("Failed to parse notification: %s", ly_last_errmsg());
+		ly_in_free(in, 0);
+		return err;
+	}
+
+	/*
+	 * Notification can be a child of some data node, so traverse the tree
+	 * until we find the notification.
+	 */
+	LYD_TREE_DFS_BEGIN (tree, dnode) {
+		if (dnode->schema->nodetype == LYS_NOTIF) {
+			found = true;
+			break;
+		}
+		LYD_TREE_DFS_END(tree, dnode);
+	}
+
+	if (!found) {
+		zlog_err("Notification not found in the parsed tree");
+		lyd_free_all(tree);
+		ly_in_free(in, 0);
+		return LY_ENOTFOUND;
+	}
+
+	*notif = dnode;
+
+	return LY_SUCCESS;
+}
+
 static ssize_t yang_print_darr(void *arg, const void *buf, size_t count)
 {
 	uint8_t *dst = darr_append_n(*(uint8_t **)arg, count);

lib/yang.h

@@ -607,6 +607,19 @@ extern struct ly_ctx *yang_ctx_new_setup(bool embedded_modules,
  */
 extern void yang_debugging_set(bool enable);
 
+/*
+ * Parse a YANG notification.
+ *
+ * Args:
+ *	format: LYD_FORMAT of input data.
+ *	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.
+ */
+extern LY_ERR yang_parse_notification(LYD_FORMAT format, const char *data,
+				      struct lyd_node **notif);
 
 /*
  * "Print" the yang tree in `root` into dynamic sized array.

mgmtd/mgmt_be_adapter.c

@@ -592,14 +592,22 @@ static void mgmt_be_adapter_send_notify(struct mgmt_msg_notify_data *msg,
 {
 	struct mgmt_be_client_adapter *adapter;
 	struct mgmt_be_xpath_map *map;
-	const char *notif;
+	char notif[XPATH_MAXLEN];
+	struct lyd_node *dnode;
+	LY_ERR err;
 	uint id;
 
 	if (!darr_len(be_notif_xpath_map))
 		return;
 
-	/* "{\"modname:notification-name\": ...}" */
-	notif = (const char *)msg->result + 2;
+	err = yang_parse_notification(msg->result_type, (char *)msg->result,
+				      &dnode);
+	if (err)
+		return;
+
+	lysc_path(dnode->schema, LYSC_PATH_DATA, notif, sizeof(notif));
+
+	lyd_free_all(dnode);
 
 	darr_foreach_p (be_notif_xpath_map, map) {
 		if (strncmp(map->xpath_prefix, notif, strlen(map->xpath_prefix)))