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