lib/northbound_sysrepo.c

   1 /*
   2  * Copyright (C) 2018  NetDEF, Inc.
   3  *                     Renato Westphal
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the Free
   7  * Software Foundation; either version 2 of the License, or (at your option)
   8  * any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful, but WITHOUT
  11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  13  * more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; see the file COPYING; if not, write to the Free Software
  17  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
  18  */
  19
  20 #include <zebra.h>
  21
  22 #include "log.h"
  23 #include "lib_errors.h"
  24 #include "command.h"
  25 #include "memory.h"
  26 #include "libfrr.h"
  27 #include "version.h"
  28 #include "northbound.h"
  29
  30 #include <sysrepo.h>
  31 #include <sysrepo/values.h>
  32 #include <sysrepo/xpath.h>
  33
  34 DEFINE_MTYPE_STATIC(LIB, SYSREPO, "Sysrepo module")
  35
  36 static struct thread_master *master;
  37 static struct list *sysrepo_threads;
  38 static sr_session_ctx_t *session;
  39 static sr_conn_ctx_t *connection;
  40
  41 static int frr_sr_read_cb(struct thread *thread);
  42 static int frr_sr_write_cb(struct thread *thread);
  43 static int frr_sr_finish(void);
  44
  45 /* Convert FRR YANG data value to sysrepo YANG data value. */
  46 static int yang_data_frr2sr(struct yang_data *frr_data, sr_val_t *sr_data)
  47 {
  48         struct nb_node *nb_node;
  49         const struct lys_node *snode;
  50         struct lys_node_container *scontainer;
  51         struct lys_node_leaf *sleaf;
  52         struct lys_node_leaflist *sleaflist;
  53         LY_DATA_TYPE type;
  54
  55         sr_val_set_xpath(sr_data, frr_data->xpath);
  56
  57         nb_node = nb_node_find(frr_data->xpath);
  58         if (!nb_node) {
  59                 flog_warn(EC_LIB_YANG_UNKNOWN_DATA_PATH,
  60                           "%s: unknown data path: %s", __func__,
  61                           frr_data->xpath);
  62                 return -1;
  63         }
  64
  65         snode = nb_node->snode;
  66         switch (snode->nodetype) {
  67         case LYS_CONTAINER:
  68                 scontainer = (struct lys_node_container *)snode;
  69                 if (!scontainer->presence)
  70                         return -1;
  71                 sr_data->type = SR_CONTAINER_PRESENCE_T;
  72                 return 0;
  73         case LYS_LIST:
  74                 sr_data->type = SR_LIST_T;
  75                 return 0;
  76         case LYS_LEAF:
  77                 sleaf = (struct lys_node_leaf *)snode;
  78                 type = sleaf->type.base;
  79                 break;
  80         case LYS_LEAFLIST:
  81                 sleaflist = (struct lys_node_leaflist *)snode;
  82                 type = sleaflist->type.base;
  83                 break;
  84         default:
  85                 return -1;
  86         }
  87
  88         switch (type) {
  89         case LY_TYPE_BINARY:
  90                 sr_val_set_str_data(sr_data, SR_BINARY_T, frr_data->value);
  91                 break;
  92         case LY_TYPE_BITS:
  93                 sr_val_set_str_data(sr_data, SR_BITS_T, frr_data->value);
  94                 break;
  95         case LY_TYPE_BOOL:
  96                 sr_data->type = SR_BOOL_T;
  97                 sr_data->data.bool_val = yang_str2bool(frr_data->value);
  98                 break;
  99         case LY_TYPE_DEC64:
 100                 sr_data->type = SR_DECIMAL64_T;
 101                 sr_data->data.decimal64_val =
 102                         yang_str2dec64(frr_data->xpath, frr_data->value);
 103                 break;
 104         case LY_TYPE_EMPTY:
 105                 sr_data->type = SR_LEAF_EMPTY_T;
 106                 break;
 107         case LY_TYPE_ENUM:
 108                 sr_val_set_str_data(sr_data, SR_ENUM_T, frr_data->value);
 109                 break;
 110         case LY_TYPE_IDENT:
 111                 sr_val_set_str_data(sr_data, SR_IDENTITYREF_T, frr_data->value);
 112                 break;
 113         case LY_TYPE_INST:
 114                 sr_val_set_str_data(sr_data, SR_INSTANCEID_T, frr_data->value);
 115                 break;
 116         case LY_TYPE_INT8:
 117                 sr_data->type = SR_INT8_T;
 118                 sr_data->data.int8_val = yang_str2int8(frr_data->value);
 119                 break;
 120         case LY_TYPE_INT16:
 121                 sr_data->type = SR_INT16_T;
 122                 sr_data->data.int16_val = yang_str2int16(frr_data->value);
 123                 break;
 124         case LY_TYPE_INT32:
 125                 sr_data->type = SR_INT32_T;
 126                 sr_data->data.int32_val = yang_str2int32(frr_data->value);
 127                 break;
 128         case LY_TYPE_INT64:
 129                 sr_data->type = SR_INT64_T;
 130                 sr_data->data.int64_val = yang_str2int64(frr_data->value);
 131                 break;
 132         case LY_TYPE_STRING:
 133                 sr_val_set_str_data(sr_data, SR_STRING_T, frr_data->value);
 134                 break;
 135         case LY_TYPE_UINT8:
 136                 sr_data->type = SR_UINT8_T;
 137                 sr_data->data.uint8_val = yang_str2uint8(frr_data->value);
 138                 break;
 139         case LY_TYPE_UINT16:
 140                 sr_data->type = SR_UINT16_T;
 141                 sr_data->data.uint16_val = yang_str2uint16(frr_data->value);
 142                 break;
 143         case LY_TYPE_UINT32:
 144                 sr_data->type = SR_UINT32_T;
 145                 sr_data->data.uint32_val = yang_str2uint32(frr_data->value);
 146                 break;
 147         case LY_TYPE_UINT64:
 148                 sr_data->type = SR_UINT64_T;
 149                 sr_data->data.uint64_val = yang_str2uint64(frr_data->value);
 150                 break;
 151         default:
 152                 return -1;
 153         }
 154
 155         return 0;
 156 }
 157
 158 static int frr_sr_process_change(struct nb_config *candidate,
 159                                  sr_change_oper_t sr_op, sr_val_t *sr_old_val,
 160                                  sr_val_t *sr_new_val)
 161 {
 162         struct nb_node *nb_node;
 163         enum nb_operation nb_op;
 164         sr_val_t *sr_data;
 165         const char *xpath;
 166         char value_str[YANG_VALUE_MAXLEN];
 167         struct yang_data *data;
 168         int ret;
 169
 170         sr_data = sr_new_val ? sr_new_val : sr_old_val;
 171         assert(sr_data);
 172
 173         xpath = sr_data->xpath;
 174
 175         /* Non-presence container - nothing to do. */
 176         if (sr_data->type == SR_CONTAINER_T)
 177                 return NB_OK;
 178
 179         nb_node = nb_node_find(xpath);
 180         if (!nb_node) {
 181                 flog_warn(EC_LIB_YANG_UNKNOWN_DATA_PATH,
 182                           "%s: unknown data path: %s", __func__, xpath);
 183                 return NB_ERR;
 184         }
 185
 186         /* Map operation values. */
 187         switch (sr_op) {
 188         case SR_OP_CREATED:
 189         case SR_OP_MODIFIED:
 190                 if (nb_operation_is_valid(NB_OP_CREATE, nb_node->snode))
 191                         nb_op = NB_OP_CREATE;
 192                 else if (nb_operation_is_valid(NB_OP_MODIFY, nb_node->snode)) {
 193                         nb_op = NB_OP_MODIFY;
 194                 } else
 195                         /* Ignore list keys modifications. */
 196                         return NB_OK;
 197                 break;
 198         case SR_OP_DELETED:
 199                 /*
 200                  * When a list is deleted or one of its keys is changed, we are
 201                  * notified about the removal of all of its leafs, even the ones
 202                  * that are non-optional. We need to ignore these notifications.
 203                  */
 204                 if (!nb_operation_is_valid(NB_OP_DELETE, nb_node->snode))
 205                         return NB_OK;
 206
 207                 nb_op = NB_OP_DELETE;
 208                 break;
 209         case SR_OP_MOVED:
 210                 nb_op = NB_OP_MOVE;
 211                 break;
 212         default:
 213                 flog_err(EC_LIB_DEVELOPMENT,
 214                          "%s: unexpected operation %u [xpath %s]", __func__,
 215                          sr_op, xpath);
 216                 return NB_ERR;
 217         }
 218
 219         sr_val_to_buff(sr_data, value_str, sizeof(value_str));
 220         data = yang_data_new(xpath, value_str);
 221
 222         ret = nb_candidate_edit(candidate, nb_node, nb_op, xpath, NULL, data);
 223         yang_data_free(data);
 224         if (ret != NB_OK) {
 225                 flog_warn(
 226                         EC_LIB_NB_CANDIDATE_EDIT_ERROR,
 227                         "%s: failed to edit candidate configuration: operation [%s] xpath [%s]",
 228                         __func__, nb_operation_name(nb_op), xpath);
 229                 return NB_ERR;
 230         }
 231
 232         return NB_OK;
 233 }
 234
 235 /* Callback for changes in the running configuration. */
 236 static int frr_sr_config_change_cb(sr_session_ctx_t *session,
 237                                    const char *module_name,
 238                                    sr_notif_event_t sr_ev, void *private_ctx)
 239 {
 240         sr_change_iter_t *it;
 241         int ret;
 242         sr_change_oper_t sr_op;
 243         sr_val_t *sr_old_val, *sr_new_val;
 244         char xpath[XPATH_MAXLEN];
 245         struct nb_config *candidate;
 246
 247         /*
 248          * Ignore SR_EV_ABORT and SR_EV_APPLY. We'll leverage the northbound
 249          * layer itself to abort or apply the configuration changes when a
 250          * transaction is created.
 251          */
 252         if (sr_ev != SR_EV_ENABLED && sr_ev != SR_EV_VERIFY)
 253                 return SR_ERR_OK;
 254
 255         snprintf(xpath, sizeof(xpath), "/%s:*", module_name);
 256         ret = sr_get_changes_iter(session, xpath, &it);
 257         if (ret != SR_ERR_OK) {
 258                 flog_err(EC_LIB_LIBSYSREPO,
 259                          "%s: sr_get_changes_iter() failed for xpath %s",
 260                          __func__, xpath);
 261                 return ret;
 262         }
 263
 264         candidate = nb_config_dup(running_config);
 265
 266         while ((ret = sr_get_change_next(session, it, &sr_op, &sr_old_val,
 267                                          &sr_new_val))
 268                == SR_ERR_OK) {
 269                 ret = frr_sr_process_change(candidate, sr_op, sr_old_val,
 270                                             sr_new_val);
 271                 sr_free_val(sr_old_val);
 272                 sr_free_val(sr_new_val);
 273                 if (ret != NB_OK)
 274                         break;
 275         }
 276
 277         sr_free_change_iter(it);
 278         if (ret != NB_OK && ret != SR_ERR_NOT_FOUND) {
 279                 nb_config_free(candidate);
 280                 return SR_ERR_INTERNAL;
 281         }
 282
 283         /* Commit changes. */
 284         ret = nb_candidate_commit(candidate, NB_CLIENT_SYSREPO, true, NULL,
 285                                   NULL);
 286         nb_config_free(candidate);
 287
 288         /* Map northbound return code to sysrepo return code. */
 289         switch (ret) {
 290         case NB_OK:
 291         case NB_ERR_NO_CHANGES:
 292                 return SR_ERR_OK;
 293         case NB_ERR_LOCKED:
 294                 return SR_ERR_LOCKED;
 295         case NB_ERR_RESOURCE:
 296                 return SR_ERR_NOMEM;
 297         default:
 298                 return SR_ERR_VALIDATION_FAILED;
 299         }
 300 }
 301
 302 static int frr_sr_state_data_iter_cb(const struct lys_node *snode,
 303                                      struct yang_translator *translator,
 304                                      struct yang_data *data, void *arg)
 305 {
 306         struct list *elements = arg;
 307
 308         listnode_add(elements, data);
 309
 310         return NB_OK;
 311 }
 312
 313 /* Callback for state retrieval. */
 314 static int frr_sr_state_cb(const char *xpath, sr_val_t **values,
 315                            size_t *values_cnt, uint64_t request_id,
 316                            void *private_ctx)
 317 {
 318         struct list *elements;
 319         struct yang_data *data;
 320         struct listnode *node;
 321         sr_val_t *v;
 322         int ret, count, i = 0;
 323
 324         elements = yang_data_list_new();
 325         if (nb_oper_data_iterate(xpath, NULL, NB_OPER_DATA_ITER_NORECURSE,
 326                                  frr_sr_state_data_iter_cb, elements)
 327             != NB_OK) {
 328                 flog_warn(EC_LIB_NB_OPERATIONAL_DATA,
 329                           "%s: failed to obtain operational data [xpath %s]",
 330                           __func__, xpath);
 331                 goto exit;
 332         }
 333
 334         if (list_isempty(elements))
 335                 goto exit;
 336
 337         count = listcount(elements);
 338         ret = sr_new_values(count, &v);
 339         if (ret != SR_ERR_OK) {
 340                 flog_err(EC_LIB_LIBSYSREPO, "%s: sr_new_values(): %s", __func__,
 341                          sr_strerror(ret));
 342                 goto exit;
 343         }
 344
 345         for (ALL_LIST_ELEMENTS_RO(elements, node, data)) {
 346                 if (yang_data_frr2sr(data, &v[i++]) != 0) {
 347                         flog_err(EC_LIB_SYSREPO_DATA_CONVERT,
 348                                  "%s: failed to convert data to sysrepo format",
 349                                  __func__);
 350                 }
 351         }
 352
 353         *values = v;
 354         *values_cnt = count;
 355
 356         list_delete(&elements);
 357
 358         return SR_ERR_OK;
 359
 360 exit:
 361         list_delete(&elements);
 362         *values = NULL;
 363         values_cnt = 0;
 364
 365         return SR_ERR_OK;
 366 }
 367
 368 static int frr_sr_config_rpc_cb(const char *xpath, const sr_val_t *sr_input,
 369                                 const size_t input_cnt, sr_val_t **sr_output,
 370                                 size_t *sr_output_cnt, void *private_ctx)
 371 {
 372         struct nb_node *nb_node;
 373         struct list *input;
 374         struct list *output;
 375         struct yang_data *data;
 376         size_t cb_output_cnt;
 377         int ret = SR_ERR_OK;
 378
 379         nb_node = nb_node_find(xpath);
 380         if (!nb_node) {
 381                 flog_warn(EC_LIB_YANG_UNKNOWN_DATA_PATH,
 382                           "%s: unknown data path: %s", __func__, xpath);
 383                 return SR_ERR_INTERNAL;
 384         }
 385
 386         input = yang_data_list_new();
 387         output = yang_data_list_new();
 388
 389         /* Process input. */
 390         for (size_t i = 0; i < input_cnt; i++) {
 391                 char value_str[YANG_VALUE_MAXLEN];
 392
 393                 sr_val_to_buff(&sr_input[i], value_str, sizeof(value_str));
 394
 395                 data = yang_data_new(xpath, value_str);
 396                 listnode_add(input, data);
 397         }
 398
 399         /* Execute callback registered for this XPath. */
 400         if (nb_node->cbs.rpc(xpath, input, output) != NB_OK) {
 401                 flog_warn(EC_LIB_NB_CB_RPC, "%s: rpc callback failed: %s",
 402                           __func__, xpath);
 403                 ret = SR_ERR_OPERATION_FAILED;
 404                 goto exit;
 405         }
 406
 407         /* Process output. */
 408         if (listcount(output) > 0) {
 409                 sr_val_t *values = NULL;
 410                 struct listnode *node;
 411                 int i = 0;
 412
 413                 cb_output_cnt = listcount(output);
 414                 ret = sr_new_values(cb_output_cnt, &values);
 415                 if (ret != SR_ERR_OK) {
 416                         flog_err(EC_LIB_LIBSYSREPO, "%s: sr_new_values(): %s",
 417                                  __func__, sr_strerror(ret));
 418                         goto exit;
 419                 }
 420
 421                 for (ALL_LIST_ELEMENTS_RO(output, node, data)) {
 422                         if (yang_data_frr2sr(data, &values[i++]) != 0) {
 423                                 flog_err(
 424                                         EC_LIB_SYSREPO_DATA_CONVERT,
 425                                         "%s: failed to convert data to Sysrepo format",
 426                                         __func__);
 427                                 ret = SR_ERR_INTERNAL;
 428                                 sr_free_values(values, cb_output_cnt);
 429                                 goto exit;
 430                         }
 431                 }
 432
 433                 *sr_output = values;
 434                 *sr_output_cnt = cb_output_cnt;
 435         }
 436
 437 exit:
 438         /* Release memory. */
 439         list_delete(&input);
 440         list_delete(&output);
 441
 442         return ret;
 443 }
 444
 445 static int frr_sr_notification_send(const char *xpath, struct list *arguments)
 446 {
 447         sr_val_t *values = NULL;
 448         size_t values_cnt = 0;
 449         int ret;
 450
 451         if (arguments && listcount(arguments) > 0) {
 452                 struct yang_data *data;
 453                 struct listnode *node;
 454                 int i = 0;
 455
 456                 values_cnt = listcount(arguments);
 457                 ret = sr_new_values(values_cnt, &values);
 458                 if (ret != SR_ERR_OK) {
 459                         flog_err(EC_LIB_LIBSYSREPO, "%s: sr_new_values(): %s",
 460                                  __func__, sr_strerror(ret));
 461                         return NB_ERR;
 462                 }
 463
 464                 for (ALL_LIST_ELEMENTS_RO(arguments, node, data)) {
 465                         if (yang_data_frr2sr(data, &values[i++]) != 0) {
 466                                 flog_err(
 467                                         EC_LIB_SYSREPO_DATA_CONVERT,
 468                                         "%s: failed to convert data to sysrepo format",
 469                                         __func__);
 470                                 sr_free_values(values, values_cnt);
 471                                 return NB_ERR;
 472                         }
 473                 }
 474         }
 475
 476         ret = sr_event_notif_send(session, xpath, values, values_cnt,
 477                                   SR_EV_NOTIF_DEFAULT);
 478         if (ret != SR_ERR_OK) {
 479                 flog_err(EC_LIB_LIBSYSREPO,
 480                          "%s: sr_event_notif_send() failed for xpath %s",
 481                          __func__, xpath);
 482                 return NB_ERR;
 483         }
 484
 485         return NB_OK;
 486 }
 487
 488 /* Code to integrate the sysrepo client into FRR main event loop. */
 489 struct sysrepo_thread {
 490         struct thread *thread;
 491         sr_fd_event_t event;
 492         int fd;
 493 };
 494
 495 static struct sysrepo_thread *frr_sr_fd_lookup(sr_fd_event_t event, int fd)
 496 {
 497         struct sysrepo_thread *sr_thread;
 498         struct listnode *node;
 499
 500         for (ALL_LIST_ELEMENTS_RO(sysrepo_threads, node, sr_thread)) {
 501                 if (sr_thread->event == event && sr_thread->fd == fd)
 502                         return sr_thread;
 503         }
 504
 505         return NULL;
 506 }
 507
 508 static void frr_sr_fd_add(int event, int fd)
 509 {
 510         struct sysrepo_thread *sr_thread;
 511
 512         if (frr_sr_fd_lookup(event, fd) != NULL)
 513                 return;
 514
 515         sr_thread = XCALLOC(MTYPE_SYSREPO, sizeof(*sr_thread));
 516         sr_thread->event = event;
 517         sr_thread->fd = fd;
 518         listnode_add(sysrepo_threads, sr_thread);
 519
 520         switch (event) {
 521         case SR_FD_INPUT_READY:
 522                 thread_add_read(master, frr_sr_read_cb, NULL, fd,
 523                                 &sr_thread->thread);
 524                 break;
 525         case SR_FD_OUTPUT_READY:
 526                 thread_add_write(master, frr_sr_write_cb, NULL, fd,
 527                                  &sr_thread->thread);
 528                 break;
 529         default:
 530                 return;
 531         }
 532 }
 533
 534 static void frr_sr_fd_free(struct sysrepo_thread *sr_thread)
 535 {
 536         THREAD_OFF(sr_thread->thread);
 537         XFREE(MTYPE_SYSREPO, sr_thread);
 538 }
 539
 540 static void frr_sr_fd_del(int event, int fd)
 541 {
 542         struct sysrepo_thread *sr_thread;
 543
 544         sr_thread = frr_sr_fd_lookup(event, fd);
 545         if (!sr_thread)
 546                 return;
 547
 548         listnode_delete(sysrepo_threads, sr_thread);
 549         frr_sr_fd_free(sr_thread);
 550 }
 551
 552 static void frr_sr_fd_update(sr_fd_change_t *fd_change_set,
 553                              size_t fd_change_set_cnt)
 554 {
 555         for (size_t i = 0; i < fd_change_set_cnt; i++) {
 556                 int fd = fd_change_set[i].fd;
 557                 int event = fd_change_set[i].events;
 558
 559                 if (event != SR_FD_INPUT_READY && event != SR_FD_OUTPUT_READY)
 560                         continue;
 561
 562                 switch (fd_change_set[i].action) {
 563                 case SR_FD_START_WATCHING:
 564                         frr_sr_fd_add(event, fd);
 565                         break;
 566                 case SR_FD_STOP_WATCHING:
 567                         frr_sr_fd_del(event, fd);
 568                         break;
 569                 default:
 570                         break;
 571                 }
 572         }
 573 }
 574
 575 static int frr_sr_read_cb(struct thread *thread)
 576 {
 577         int fd = THREAD_FD(thread);
 578         sr_fd_change_t *fd_change_set = NULL;
 579         size_t fd_change_set_cnt = 0;
 580         int ret;
 581
 582         ret = sr_fd_event_process(fd, SR_FD_INPUT_READY, &fd_change_set,
 583                                   &fd_change_set_cnt);
 584         if (ret != SR_ERR_OK) {
 585                 flog_err(EC_LIB_LIBSYSREPO, "%s: sr_fd_event_process(): %s",
 586                          __func__, sr_strerror(ret));
 587                 return -1;
 588         }
 589
 590         thread = NULL;
 591         thread_add_read(master, frr_sr_read_cb, NULL, fd, &thread);
 592
 593         frr_sr_fd_update(fd_change_set, fd_change_set_cnt);
 594         free(fd_change_set);
 595
 596         return 0;
 597 }
 598
 599 static int frr_sr_write_cb(struct thread *thread)
 600 {
 601         int fd = THREAD_FD(thread);
 602         sr_fd_change_t *fd_change_set = NULL;
 603         size_t fd_change_set_cnt = 0;
 604         int ret;
 605
 606         ret = sr_fd_event_process(fd, SR_FD_OUTPUT_READY, &fd_change_set,
 607                                   &fd_change_set_cnt);
 608         if (ret != SR_ERR_OK) {
 609                 flog_err(EC_LIB_LIBSYSREPO, "%s: sr_fd_event_process(): %s",
 610                          __func__, sr_strerror(ret));
 611                 return -1;
 612         }
 613
 614         frr_sr_fd_update(fd_change_set, fd_change_set_cnt);
 615         free(fd_change_set);
 616
 617         return 0;
 618 }
 619
 620 static void frr_sr_subscribe_config(struct yang_module *module)
 621 {
 622         int ret;
 623
 624         ret = sr_module_change_subscribe(
 625                 session, module->name, frr_sr_config_change_cb, NULL, 0,
 626                 SR_SUBSCR_DEFAULT | SR_SUBSCR_EV_ENABLED,
 627                 &module->sr_subscription);
 628         if (ret != SR_ERR_OK)
 629                 flog_err(EC_LIB_LIBSYSREPO, "sr_module_change_subscribe(): %s",
 630                          sr_strerror(ret));
 631 }
 632
 633 static int frr_sr_subscribe_state(const struct lys_node *snode, void *arg)
 634 {
 635         struct yang_module *module = arg;
 636         struct nb_node *nb_node;
 637         int ret;
 638
 639         if (!CHECK_FLAG(snode->flags, LYS_CONFIG_R))
 640                 return YANG_ITER_CONTINUE;
 641         /* We only need to subscribe to the root of the state subtrees. */
 642         if (snode->parent && CHECK_FLAG(snode->parent->flags, LYS_CONFIG_R))
 643                 return YANG_ITER_CONTINUE;
 644
 645         nb_node = snode->priv;
 646         if (debug_northbound)
 647                 zlog_debug("%s: providing data to '%s'", __func__,
 648                            nb_node->xpath);
 649
 650         ret = sr_dp_get_items_subscribe(
 651                 session, nb_node->xpath, frr_sr_state_cb, NULL,
 652                 SR_SUBSCR_CTX_REUSE, &module->sr_subscription);
 653         if (ret != SR_ERR_OK)
 654                 flog_err(EC_LIB_LIBSYSREPO, "sr_dp_get_items_subscribe(): %s",
 655                          sr_strerror(ret));
 656
 657         return YANG_ITER_CONTINUE;
 658 }
 659
 660 static int frr_sr_subscribe_rpc(const struct lys_node *snode, void *arg)
 661 {
 662         struct yang_module *module = arg;
 663         struct nb_node *nb_node;
 664         int ret;
 665
 666         if (snode->nodetype != LYS_RPC)
 667                 return YANG_ITER_CONTINUE;
 668
 669         nb_node = snode->priv;
 670         if (debug_northbound)
 671                 zlog_debug("%s: providing RPC to '%s'", __func__,
 672                            nb_node->xpath);
 673
 674         ret = sr_rpc_subscribe(session, nb_node->xpath, frr_sr_config_rpc_cb,
 675                                NULL, SR_SUBSCR_CTX_REUSE,
 676                                &module->sr_subscription);
 677         if (ret != SR_ERR_OK)
 678                 flog_err(EC_LIB_LIBSYSREPO, "sr_rpc_subscribe(): %s",
 679                          sr_strerror(ret));
 680
 681         return YANG_ITER_CONTINUE;
 682 }
 683
 684 static int frr_sr_subscribe_action(const struct lys_node *snode, void *arg)
 685 {
 686         struct yang_module *module = arg;
 687         struct nb_node *nb_node;
 688         int ret;
 689
 690         if (snode->nodetype != LYS_ACTION)
 691                 return YANG_ITER_CONTINUE;
 692
 693         nb_node = snode->priv;
 694         if (debug_northbound)
 695                 zlog_debug("%s: providing action to '%s'", __func__,
 696                            nb_node->xpath);
 697
 698         ret = sr_action_subscribe(session, nb_node->xpath, frr_sr_config_rpc_cb,
 699                                   NULL, SR_SUBSCR_CTX_REUSE,
 700                                   &module->sr_subscription);
 701         if (ret != SR_ERR_OK)
 702                 flog_err(EC_LIB_LIBSYSREPO, "sr_action_subscribe(): %s",
 703                          sr_strerror(ret));
 704
 705         return YANG_ITER_CONTINUE;
 706 }
 707
 708 /* FRR's Sysrepo initialization. */
 709 static int frr_sr_init(const char *program_name)
 710 {
 711         struct yang_module *module;
 712         int sysrepo_fd, ret;
 713
 714         sysrepo_threads = list_new();
 715
 716         ret = sr_fd_watcher_init(&sysrepo_fd, NULL);
 717         if (ret != SR_ERR_OK) {
 718                 flog_err(EC_LIB_SYSREPO_INIT, "%s: sr_fd_watcher_init(): %s",
 719                          __func__, sr_strerror(ret));
 720                 goto cleanup;
 721         }
 722
 723         /* Connect to Sysrepo. */
 724         ret = sr_connect(program_name, SR_CONN_DEFAULT, &connection);
 725         if (ret != SR_ERR_OK) {
 726                 flog_err(EC_LIB_SYSREPO_INIT, "%s: sr_connect(): %s", __func__,
 727                          sr_strerror(ret));
 728                 goto cleanup;
 729         }
 730
 731         /* Start session. */
 732         ret = sr_session_start(connection, SR_DS_RUNNING, SR_SESS_DEFAULT,
 733                                &session);
 734         if (ret != SR_ERR_OK) {
 735                 flog_err(EC_LIB_SYSREPO_INIT, "%s: sr_session_start(): %s",
 736                          __func__, sr_strerror(ret));
 737                 goto cleanup;
 738         }
 739
 740         /* Perform subscriptions. */
 741         RB_FOREACH (module, yang_modules, &yang_modules) {
 742                 frr_sr_subscribe_config(module);
 743                 yang_snodes_iterate_module(module->info, frr_sr_subscribe_state,
 744                                            0, module);
 745                 yang_snodes_iterate_module(module->info, frr_sr_subscribe_rpc,
 746                                            0, module);
 747                 yang_snodes_iterate_module(module->info,
 748                                            frr_sr_subscribe_action, 0, module);
 749         }
 750
 751         hook_register(nb_notification_send, frr_sr_notification_send);
 752
 753         frr_sr_fd_add(SR_FD_INPUT_READY, sysrepo_fd);
 754
 755         return 0;
 756
 757 cleanup:
 758         frr_sr_finish();
 759
 760         return -1;
 761 }
 762
 763 static int frr_sr_finish(void)
 764 {
 765         struct yang_module *module;
 766
 767         RB_FOREACH (module, yang_modules, &yang_modules) {
 768                 if (!module->sr_subscription)
 769                         continue;
 770                 sr_unsubscribe(session, module->sr_subscription);
 771         }
 772
 773         if (session)
 774                 sr_session_stop(session);
 775         if (connection)
 776                 sr_disconnect(connection);
 777
 778         sysrepo_threads->del = (void (*)(void *))frr_sr_fd_free;
 779         list_delete(&sysrepo_threads);
 780         sr_fd_watcher_cleanup();
 781
 782         return 0;
 783 }
 784
 785 static int frr_sr_module_late_init(struct thread_master *tm)
 786 {
 787         master = tm;
 788
 789         if (frr_sr_init(frr_get_progname()) < 0) {
 790                 flog_err(EC_LIB_SYSREPO_INIT,
 791                          "failed to initialize the Sysrepo module");
 792                 return -1;
 793         }
 794
 795         hook_register(frr_fini, frr_sr_finish);
 796
 797         return 0;
 798 }
 799
 800 static int frr_sr_module_init(void)
 801 {
 802         hook_register(frr_late_init, frr_sr_module_late_init);
 803
 804         return 0;
 805 }
 806
 807 FRR_MODULE_SETUP(.name = "frr_sysrepo", .version = FRR_VERSION,
 808                  .description = "FRR sysrepo integration module",
 809                  .init = frr_sr_module_init, )