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         pthread_rwlock_rdlock(&running_config->lock);
 260         {
 261                 candidate = nb_config_dup(running_config);
 262         }
 263         pthread_rwlock_unlock(&running_config->lock);
 264
 265         while ((ret = sr_get_change_next(session, it, &sr_op, &sr_old_val,
 266                                          &sr_new_val))
 267                == SR_ERR_OK) {
 268                 ret = frr_sr_process_change(candidate, sr_op, sr_old_val,
 269                                             sr_new_val);
 270                 sr_free_val(sr_old_val);
 271                 sr_free_val(sr_new_val);
 272                 if (ret != NB_OK)
 273                         break;
 274         }
 275
 276         sr_free_change_iter(it);
 277         if (ret != NB_OK && ret != SR_ERR_NOT_FOUND) {
 278                 nb_config_free(candidate);
 279                 return SR_ERR_INTERNAL;
 280         }
 281
 282         transaction = NULL;
 283         if (startup_config) {
 284                 /*
 285                  * sysrepod sends the entire startup configuration using a
 286                  * single event (SR_EV_ENABLED). This means we need to perform
 287                  * the full two-phase commit protocol in one go here.
 288                  */
 289                 ret = nb_candidate_commit(candidate, NB_CLIENT_SYSREPO, NULL,
 290                                           true, NULL, NULL);
 291         } else {
 292                 /*
 293                  * Validate the configuration changes and allocate all resources
 294                  * required to apply them.
 295                  */
 296                 ret = nb_candidate_commit_prepare(candidate, NB_CLIENT_SYSREPO,
 297                                                   NULL, NULL, &transaction);
 298         }
 299
 300         /* Map northbound return code to sysrepo return code. */
 301         switch (ret) {
 302         case NB_OK:
 303                 return SR_ERR_OK;
 304         case NB_ERR_NO_CHANGES:
 305                 nb_config_free(candidate);
 306                 return SR_ERR_OK;
 307         case NB_ERR_LOCKED:
 308                 return SR_ERR_LOCKED;
 309         case NB_ERR_RESOURCE:
 310                 return SR_ERR_NOMEM;
 311         default:
 312                 return SR_ERR_VALIDATION_FAILED;
 313         }
 314 }
 315
 316 static int frr_sr_config_change_cb_apply(sr_session_ctx_t *session,
 317                                          const char *module_name)
 318 {
 319         /* Apply the transaction. */
 320         if (transaction) {
 321                 struct nb_config *candidate = transaction->config;
 322
 323                 nb_candidate_commit_apply(transaction, true, NULL);
 324                 nb_config_free(candidate);
 325         }
 326
 327         return SR_ERR_OK;
 328 }
 329
 330 static int frr_sr_config_change_cb_abort(sr_session_ctx_t *session,
 331                                          const char *module_name)
 332 {
 333         /* Abort the transaction. */
 334         if (transaction) {
 335                 struct nb_config *candidate = transaction->config;
 336
 337                 nb_candidate_commit_abort(transaction);
 338                 nb_config_free(candidate);
 339         }
 340
 341         return SR_ERR_OK;
 342 }
 343
 344 /* Callback for changes in the running configuration. */
 345 static int frr_sr_config_change_cb(sr_session_ctx_t *session,
 346                                    const char *module_name,
 347                                    sr_notif_event_t sr_ev, void *private_ctx)
 348 {
 349         switch (sr_ev) {
 350         case SR_EV_ENABLED:
 351                 return frr_sr_config_change_cb_verify(session, module_name,
 352                                                       true);
 353         case SR_EV_VERIFY:
 354                 return frr_sr_config_change_cb_verify(session, module_name,
 355                                                       false);
 356         case SR_EV_APPLY:
 357                 return frr_sr_config_change_cb_apply(session, module_name);
 358         case SR_EV_ABORT:
 359                 return frr_sr_config_change_cb_abort(session, module_name);
 360         default:
 361                 flog_err(EC_LIB_LIBSYSREPO, "%s: unknown sysrepo event: %u",
 362                          __func__, sr_ev);
 363                 return SR_ERR_INTERNAL;
 364         }
 365 }
 366
 367 static int frr_sr_state_data_iter_cb(const struct lys_node *snode,
 368                                      struct yang_translator *translator,
 369                                      struct yang_data *data, void *arg)
 370 {
 371         struct list *elements = arg;
 372
 373         listnode_add(elements, data);
 374
 375         return NB_OK;
 376 }
 377
 378 /* Callback for state retrieval. */
 379 static int frr_sr_state_cb(const char *xpath, sr_val_t **values,
 380                            size_t *values_cnt, uint64_t request_id,
 381                            const char *original_xpath, void *private_ctx)
 382 {
 383         struct list *elements;
 384         struct yang_data *data;
 385         struct listnode *node;
 386         sr_val_t *v;
 387         int ret, count, i = 0;
 388
 389         elements = yang_data_list_new();
 390         if (nb_oper_data_iterate(xpath, NULL, NB_OPER_DATA_ITER_NORECURSE,
 391                                  frr_sr_state_data_iter_cb, elements)
 392             != NB_OK) {
 393                 flog_warn(EC_LIB_NB_OPERATIONAL_DATA,
 394                           "%s: failed to obtain operational data [xpath %s]",
 395                           __func__, xpath);
 396                 goto exit;
 397         }
 398
 399         if (list_isempty(elements))
 400                 goto exit;
 401
 402         count = listcount(elements);
 403         ret = sr_new_values(count, &v);
 404         if (ret != SR_ERR_OK) {
 405                 flog_err(EC_LIB_LIBSYSREPO, "%s: sr_new_values(): %s", __func__,
 406                          sr_strerror(ret));
 407                 goto exit;
 408         }
 409
 410         for (ALL_LIST_ELEMENTS_RO(elements, node, data)) {
 411                 if (yang_data_frr2sr(data, &v[i++]) != 0) {
 412                         flog_err(EC_LIB_SYSREPO_DATA_CONVERT,
 413                                  "%s: failed to convert data to sysrepo format",
 414                                  __func__);
 415                 }
 416         }
 417
 418         *values = v;
 419         *values_cnt = count;
 420
 421         list_delete(&elements);
 422
 423         return SR_ERR_OK;
 424
 425 exit:
 426         list_delete(&elements);
 427         *values = NULL;
 428         values_cnt = 0;
 429
 430         return SR_ERR_OK;
 431 }
 432
 433 static int frr_sr_config_rpc_cb(const char *xpath, const sr_val_t *sr_input,
 434                                 const size_t input_cnt, sr_val_t **sr_output,
 435                                 size_t *sr_output_cnt, void *private_ctx)
 436 {
 437         struct nb_node *nb_node;
 438         struct list *input;
 439         struct list *output;
 440         struct yang_data *data;
 441         size_t cb_output_cnt;
 442         int ret = SR_ERR_OK;
 443
 444         nb_node = nb_node_find(xpath);
 445         if (!nb_node) {
 446                 flog_warn(EC_LIB_YANG_UNKNOWN_DATA_PATH,
 447                           "%s: unknown data path: %s", __func__, xpath);
 448                 return SR_ERR_INTERNAL;
 449         }
 450
 451         input = yang_data_list_new();
 452         output = yang_data_list_new();
 453
 454         /* Process input. */
 455         for (size_t i = 0; i < input_cnt; i++) {
 456                 char value_str[YANG_VALUE_MAXLEN];
 457
 458                 sr_val_to_buff(&sr_input[i], value_str, sizeof(value_str));
 459
 460                 data = yang_data_new(xpath, value_str);
 461                 listnode_add(input, data);
 462         }
 463
 464         /* Execute callback registered for this XPath. */
 465         if (nb_callback_rpc(nb_node, xpath, input, output) != NB_OK) {
 466                 flog_warn(EC_LIB_NB_CB_RPC, "%s: rpc callback failed: %s",
 467                           __func__, xpath);
 468                 ret = SR_ERR_OPERATION_FAILED;
 469                 goto exit;
 470         }
 471
 472         /* Process output. */
 473         if (listcount(output) > 0) {
 474                 sr_val_t *values = NULL;
 475                 struct listnode *node;
 476                 int i = 0;
 477
 478                 cb_output_cnt = listcount(output);
 479                 ret = sr_new_values(cb_output_cnt, &values);
 480                 if (ret != SR_ERR_OK) {
 481                         flog_err(EC_LIB_LIBSYSREPO, "%s: sr_new_values(): %s",
 482                                  __func__, sr_strerror(ret));
 483                         goto exit;
 484                 }
 485
 486                 for (ALL_LIST_ELEMENTS_RO(output, node, data)) {
 487                         if (yang_data_frr2sr(data, &values[i++]) != 0) {
 488                                 flog_err(
 489                                         EC_LIB_SYSREPO_DATA_CONVERT,
 490                                         "%s: failed to convert data to Sysrepo format",
 491                                         __func__);
 492                                 ret = SR_ERR_INTERNAL;
 493                                 sr_free_values(values, cb_output_cnt);
 494                                 goto exit;
 495                         }
 496                 }
 497
 498                 *sr_output = values;
 499                 *sr_output_cnt = cb_output_cnt;
 500         }
 501
 502 exit:
 503         /* Release memory. */
 504         list_delete(&input);
 505         list_delete(&output);
 506
 507         return ret;
 508 }
 509
 510 static int frr_sr_notification_send(const char *xpath, struct list *arguments)
 511 {
 512         sr_val_t *values = NULL;
 513         size_t values_cnt = 0;
 514         int ret;
 515
 516         if (arguments && listcount(arguments) > 0) {
 517                 struct yang_data *data;
 518                 struct listnode *node;
 519                 int i = 0;
 520
 521                 values_cnt = listcount(arguments);
 522                 ret = sr_new_values(values_cnt, &values);
 523                 if (ret != SR_ERR_OK) {
 524                         flog_err(EC_LIB_LIBSYSREPO, "%s: sr_new_values(): %s",
 525                                  __func__, sr_strerror(ret));
 526                         return NB_ERR;
 527                 }
 528
 529                 for (ALL_LIST_ELEMENTS_RO(arguments, node, data)) {
 530                         if (yang_data_frr2sr(data, &values[i++]) != 0) {
 531                                 flog_err(
 532                                         EC_LIB_SYSREPO_DATA_CONVERT,
 533                                         "%s: failed to convert data to sysrepo format",
 534                                         __func__);
 535                                 sr_free_values(values, values_cnt);
 536                                 return NB_ERR;
 537                         }
 538                 }
 539         }
 540
 541         ret = sr_event_notif_send(session, xpath, values, values_cnt,
 542                                   SR_EV_NOTIF_DEFAULT);
 543         if (ret != SR_ERR_OK) {
 544                 flog_err(EC_LIB_LIBSYSREPO,
 545                          "%s: sr_event_notif_send() failed for xpath %s",
 546                          __func__, xpath);
 547                 return NB_ERR;
 548         }
 549
 550         return NB_OK;
 551 }
 552
 553 /* Code to integrate the sysrepo client into FRR main event loop. */
 554 struct sysrepo_thread {
 555         struct thread *thread;
 556         sr_fd_event_t event;
 557         int fd;
 558 };
 559
 560 static struct sysrepo_thread *frr_sr_fd_lookup(sr_fd_event_t event, int fd)
 561 {
 562         struct sysrepo_thread *sr_thread;
 563         struct listnode *node;
 564
 565         for (ALL_LIST_ELEMENTS_RO(sysrepo_threads, node, sr_thread)) {
 566                 if (sr_thread->event == event && sr_thread->fd == fd)
 567                         return sr_thread;
 568         }
 569
 570         return NULL;
 571 }
 572
 573 static void frr_sr_fd_add(int event, int fd)
 574 {
 575         struct sysrepo_thread *sr_thread;
 576
 577         if (frr_sr_fd_lookup(event, fd) != NULL)
 578                 return;
 579
 580         sr_thread = XCALLOC(MTYPE_SYSREPO, sizeof(*sr_thread));
 581         sr_thread->event = event;
 582         sr_thread->fd = fd;
 583         listnode_add(sysrepo_threads, sr_thread);
 584
 585         switch (event) {
 586         case SR_FD_INPUT_READY:
 587                 thread_add_read(master, frr_sr_read_cb, NULL, fd,
 588                                 &sr_thread->thread);
 589                 break;
 590         case SR_FD_OUTPUT_READY:
 591                 thread_add_write(master, frr_sr_write_cb, NULL, fd,
 592                                  &sr_thread->thread);
 593                 break;
 594         default:
 595                 return;
 596         }
 597 }
 598
 599 static void frr_sr_fd_free(struct sysrepo_thread *sr_thread)
 600 {
 601         THREAD_OFF(sr_thread->thread);
 602         XFREE(MTYPE_SYSREPO, sr_thread);
 603 }
 604
 605 static void frr_sr_fd_del(int event, int fd)
 606 {
 607         struct sysrepo_thread *sr_thread;
 608
 609         sr_thread = frr_sr_fd_lookup(event, fd);
 610         if (!sr_thread)
 611                 return;
 612
 613         listnode_delete(sysrepo_threads, sr_thread);
 614         frr_sr_fd_free(sr_thread);
 615 }
 616
 617 static void frr_sr_fd_update(sr_fd_change_t *fd_change_set,
 618                              size_t fd_change_set_cnt)
 619 {
 620         for (size_t i = 0; i < fd_change_set_cnt; i++) {
 621                 int fd = fd_change_set[i].fd;
 622                 int event = fd_change_set[i].events;
 623
 624                 if (event != SR_FD_INPUT_READY && event != SR_FD_OUTPUT_READY)
 625                         continue;
 626
 627                 switch (fd_change_set[i].action) {
 628                 case SR_FD_START_WATCHING:
 629                         frr_sr_fd_add(event, fd);
 630                         break;
 631                 case SR_FD_STOP_WATCHING:
 632                         frr_sr_fd_del(event, fd);
 633                         break;
 634                 default:
 635                         break;
 636                 }
 637         }
 638 }
 639
 640 static int frr_sr_read_cb(struct thread *thread)
 641 {
 642         int fd = THREAD_FD(thread);
 643         sr_fd_change_t *fd_change_set = NULL;
 644         size_t fd_change_set_cnt = 0;
 645         int ret;
 646
 647         ret = sr_fd_event_process(fd, SR_FD_INPUT_READY, &fd_change_set,
 648                                   &fd_change_set_cnt);
 649         if (ret != SR_ERR_OK) {
 650                 flog_err(EC_LIB_LIBSYSREPO, "%s: sr_fd_event_process(): %s",
 651                          __func__, sr_strerror(ret));
 652                 return -1;
 653         }
 654
 655         thread = NULL;
 656         thread_add_read(master, frr_sr_read_cb, NULL, fd, &thread);
 657
 658         frr_sr_fd_update(fd_change_set, fd_change_set_cnt);
 659         free(fd_change_set);
 660
 661         return 0;
 662 }
 663
 664 static int frr_sr_write_cb(struct thread *thread)
 665 {
 666         int fd = THREAD_FD(thread);
 667         sr_fd_change_t *fd_change_set = NULL;
 668         size_t fd_change_set_cnt = 0;
 669         int ret;
 670
 671         ret = sr_fd_event_process(fd, SR_FD_OUTPUT_READY, &fd_change_set,
 672                                   &fd_change_set_cnt);
 673         if (ret != SR_ERR_OK) {
 674                 flog_err(EC_LIB_LIBSYSREPO, "%s: sr_fd_event_process(): %s",
 675                          __func__, sr_strerror(ret));
 676                 return -1;
 677         }
 678
 679         frr_sr_fd_update(fd_change_set, fd_change_set_cnt);
 680         free(fd_change_set);
 681
 682         return 0;
 683 }
 684
 685 static void frr_sr_subscribe_config(struct yang_module *module)
 686 {
 687         int ret;
 688
 689         ret = sr_module_change_subscribe(
 690                 session, module->name, frr_sr_config_change_cb, NULL, 0,
 691                 SR_SUBSCR_DEFAULT | SR_SUBSCR_EV_ENABLED,
 692                 &module->sr_subscription);
 693         if (ret != SR_ERR_OK)
 694                 flog_err(EC_LIB_LIBSYSREPO, "sr_module_change_subscribe(): %s",
 695                          sr_strerror(ret));
 696 }
 697
 698 static int frr_sr_subscribe_state(const struct lys_node *snode, void *arg)
 699 {
 700         struct yang_module *module = arg;
 701         struct nb_node *nb_node;
 702         int ret;
 703
 704         if (!CHECK_FLAG(snode->flags, LYS_CONFIG_R))
 705                 return YANG_ITER_CONTINUE;
 706         /* We only need to subscribe to the root of the state subtrees. */
 707         if (snode->parent && CHECK_FLAG(snode->parent->flags, LYS_CONFIG_R))
 708                 return YANG_ITER_CONTINUE;
 709
 710         nb_node = snode->priv;
 711
 712         DEBUGD(&nb_dbg_client_sysrepo, "%s: providing data to '%s'", __func__,
 713                nb_node->xpath);
 714
 715         ret = sr_dp_get_items_subscribe(
 716                 session, nb_node->xpath, frr_sr_state_cb, NULL,
 717                 SR_SUBSCR_CTX_REUSE, &module->sr_subscription);
 718         if (ret != SR_ERR_OK)
 719                 flog_err(EC_LIB_LIBSYSREPO, "sr_dp_get_items_subscribe(): %s",
 720                          sr_strerror(ret));
 721
 722         return YANG_ITER_CONTINUE;
 723 }
 724
 725 static int frr_sr_subscribe_rpc(const struct lys_node *snode, void *arg)
 726 {
 727         struct yang_module *module = arg;
 728         struct nb_node *nb_node;
 729         int ret;
 730
 731         if (snode->nodetype != LYS_RPC)
 732                 return YANG_ITER_CONTINUE;
 733
 734         nb_node = snode->priv;
 735
 736         DEBUGD(&nb_dbg_client_sysrepo, "%s: providing RPC to '%s'", __func__,
 737                nb_node->xpath);
 738
 739         ret = sr_rpc_subscribe(session, nb_node->xpath, frr_sr_config_rpc_cb,
 740                                NULL, SR_SUBSCR_CTX_REUSE,
 741                                &module->sr_subscription);
 742         if (ret != SR_ERR_OK)
 743                 flog_err(EC_LIB_LIBSYSREPO, "sr_rpc_subscribe(): %s",
 744                          sr_strerror(ret));
 745
 746         return YANG_ITER_CONTINUE;
 747 }
 748
 749 static int frr_sr_subscribe_action(const struct lys_node *snode, void *arg)
 750 {
 751         struct yang_module *module = arg;
 752         struct nb_node *nb_node;
 753         int ret;
 754
 755         if (snode->nodetype != LYS_ACTION)
 756                 return YANG_ITER_CONTINUE;
 757
 758         nb_node = snode->priv;
 759
 760         DEBUGD(&nb_dbg_client_sysrepo, "%s: providing action to '%s'", __func__,
 761                nb_node->xpath);
 762
 763         ret = sr_action_subscribe(session, nb_node->xpath, frr_sr_config_rpc_cb,
 764                                   NULL, SR_SUBSCR_CTX_REUSE,
 765                                   &module->sr_subscription);
 766         if (ret != SR_ERR_OK)
 767                 flog_err(EC_LIB_LIBSYSREPO, "sr_action_subscribe(): %s",
 768                          sr_strerror(ret));
 769
 770         return YANG_ITER_CONTINUE;
 771 }
 772
 773 /* CLI commands. */
 774 DEFUN (debug_nb_sr,
 775        debug_nb_sr_cmd,
 776        "[no] debug northbound client sysrepo",
 777        NO_STR
 778        DEBUG_STR
 779        "Northbound debugging\n"
 780        "Northbound client\n"
 781        "Sysrepo\n")
 782 {
 783         uint32_t mode = DEBUG_NODE2MODE(vty->node);
 784         bool no = strmatch(argv[0]->text, "no");
 785
 786         DEBUG_MODE_SET(&nb_dbg_client_sysrepo, mode, !no);
 787
 788         return CMD_SUCCESS;
 789 }
 790
 791 static int frr_sr_debug_config_write(struct vty *vty)
 792 {
 793         if (DEBUG_MODE_CHECK(&nb_dbg_client_sysrepo, DEBUG_MODE_CONF))
 794                 vty_out(vty, "debug northbound client sysrepo\n");
 795
 796         return 0;
 797 }
 798
 799 static int frr_sr_debug_set_all(uint32_t flags, bool set)
 800 {
 801         DEBUG_FLAGS_SET(&nb_dbg_client_sysrepo, flags, set);
 802
 803         /* If all modes have been turned off, don't preserve options. */
 804         if (!DEBUG_MODE_CHECK(&nb_dbg_client_sysrepo, DEBUG_MODE_ALL))
 805                 DEBUG_CLEAR(&nb_dbg_client_sysrepo);
 806
 807         return 0;
 808 }
 809
 810 static void frr_sr_cli_init(void)
 811 {
 812         hook_register(nb_client_debug_config_write, frr_sr_debug_config_write);
 813         hook_register(nb_client_debug_set_all, frr_sr_debug_set_all);
 814
 815         install_element(ENABLE_NODE, &debug_nb_sr_cmd);
 816         install_element(CONFIG_NODE, &debug_nb_sr_cmd);
 817 }
 818
 819 /* FRR's Sysrepo initialization. */
 820 static int frr_sr_init(const char *program_name)
 821 {
 822         struct yang_module *module;
 823         int sysrepo_fd, ret;
 824
 825         sysrepo_threads = list_new();
 826
 827         ret = sr_fd_watcher_init(&sysrepo_fd, NULL);
 828         if (ret != SR_ERR_OK) {
 829                 flog_err(EC_LIB_SYSREPO_INIT, "%s: sr_fd_watcher_init(): %s",
 830                          __func__, sr_strerror(ret));
 831                 goto cleanup;
 832         }
 833
 834         /* Connect to Sysrepo. */
 835         ret = sr_connect(program_name, SR_CONN_DEFAULT, &connection);
 836         if (ret != SR_ERR_OK) {
 837                 flog_err(EC_LIB_SYSREPO_INIT, "%s: sr_connect(): %s", __func__,
 838                          sr_strerror(ret));
 839                 goto cleanup;
 840         }
 841
 842         /* Start session. */
 843         ret = sr_session_start(connection, SR_DS_RUNNING, SR_SESS_DEFAULT,
 844                                &session);
 845         if (ret != SR_ERR_OK) {
 846                 flog_err(EC_LIB_SYSREPO_INIT, "%s: sr_session_start(): %s",
 847                          __func__, sr_strerror(ret));
 848                 goto cleanup;
 849         }
 850
 851         /* Perform subscriptions. */
 852         RB_FOREACH (module, yang_modules, &yang_modules) {
 853                 frr_sr_subscribe_config(module);
 854                 yang_snodes_iterate_module(module->info, frr_sr_subscribe_state,
 855                                            0, module);
 856                 yang_snodes_iterate_module(module->info, frr_sr_subscribe_rpc,
 857                                            0, module);
 858                 yang_snodes_iterate_module(module->info,
 859                                            frr_sr_subscribe_action, 0, module);
 860         }
 861
 862         hook_register(nb_notification_send, frr_sr_notification_send);
 863
 864         frr_sr_fd_add(SR_FD_INPUT_READY, sysrepo_fd);
 865
 866         return 0;
 867
 868 cleanup:
 869         frr_sr_finish();
 870
 871         return -1;
 872 }
 873
 874 static int frr_sr_finish(void)
 875 {
 876         struct yang_module *module;
 877
 878         RB_FOREACH (module, yang_modules, &yang_modules) {
 879                 if (!module->sr_subscription)
 880                         continue;
 881                 sr_unsubscribe(session, module->sr_subscription);
 882         }
 883
 884         if (session)
 885                 sr_session_stop(session);
 886         if (connection)
 887                 sr_disconnect(connection);
 888
 889         sysrepo_threads->del = (void (*)(void *))frr_sr_fd_free;
 890         list_delete(&sysrepo_threads);
 891         sr_fd_watcher_cleanup();
 892
 893         return 0;
 894 }
 895
 896 static int frr_sr_module_late_init(struct thread_master *tm)
 897 {
 898         master = tm;
 899
 900         if (frr_sr_init(frr_get_progname()) < 0) {
 901                 flog_err(EC_LIB_SYSREPO_INIT,
 902                          "failed to initialize the Sysrepo module");
 903                 return -1;
 904         }
 905
 906         hook_register(frr_fini, frr_sr_finish);
 907         frr_sr_cli_init();
 908
 909         return 0;
 910 }
 911
 912 static int frr_sr_module_init(void)
 913 {
 914         hook_register(frr_late_init, frr_sr_module_late_init);
 915
 916         return 0;
 917 }
 918
 919 FRR_MODULE_SETUP(.name = "frr_sysrepo", .version = FRR_VERSION,
 920                  .description = "FRR sysrepo integration module",
 921                  .init = frr_sr_module_init, )