lib/northbound_grpc.cpp

   1 //
   2 // Copyright (C) 2019  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 "libfrr.h"
  24 #include "version.h"
  25 #include "command.h"
  26 #include "lib_errors.h"
  27 #include "northbound.h"
  28 #include "northbound_db.h"
  29
  30 #include <iostream>
  31 #include <sstream>
  32 #include <memory>
  33 #include <string>
  34
  35 #include <grpcpp/grpcpp.h>
  36 #include "grpc/frr-northbound.grpc.pb.h"
  37
  38 #define GRPC_DEFAULT_PORT 50051
  39
  40 /*
  41  * NOTE: we can't use the FRR debugging infrastructure here since it uses
  42  * atomics and C++ has a different atomics API. Enable gRPC debugging
  43  * unconditionally until we figure out a way to solve this problem.
  44  */
  45 static bool nb_dbg_client_grpc = 1;
  46
  47 static pthread_t grpc_pthread;
  48
  49 class NorthboundImpl final : public frr::Northbound::Service
  50 {
  51       public:
  52         NorthboundImpl(void)
  53         {
  54                 _nextCandidateId = 0;
  55         }
  56
  57         ~NorthboundImpl(void)
  58         {
  59                 // Delete candidates.
  60                 for (auto it = _candidates.begin(); it != _candidates.end();
  61                      it++)
  62                         delete_candidate(&it->second);
  63         }
  64
  65         grpc::Status
  66         GetCapabilities(grpc::ServerContext *context,
  67                         frr::GetCapabilitiesRequest const *request,
  68                         frr::GetCapabilitiesResponse *response) override
  69         {
  70                 if (nb_dbg_client_grpc)
  71                         zlog_debug("received RPC GetCapabilities()");
  72
  73                 // Response: string frr_version = 1;
  74                 response->set_frr_version(FRR_VERSION);
  75
  76                 // Response: bool rollback_support = 2;
  77 #ifdef HAVE_CONFIG_ROLLBACKS
  78                 response->set_rollback_support(true);
  79 #else
  80                 response->set_rollback_support(false);
  81 #endif
  82
  83                 // Response: repeated ModuleData supported_modules = 3;
  84                 struct yang_module *module;
  85                 RB_FOREACH (module, yang_modules, &yang_modules) {
  86                         auto m = response->add_supported_modules();
  87
  88                         m->set_name(module->name);
  89                         if (module->info->rev_size)
  90                                 m->set_revision(module->info->rev[0].date);
  91                         m->set_organization(module->info->org);
  92                 }
  93
  94                 // Response: repeated Encoding supported_encodings = 4;
  95                 response->add_supported_encodings(frr::JSON);
  96                 response->add_supported_encodings(frr::XML);
  97
  98                 return grpc::Status::OK;
  99         }
 100
 101         grpc::Status Get(grpc::ServerContext *context,
 102                          frr::GetRequest const *request,
 103                          grpc::ServerWriter<frr::GetResponse> *writer) override
 104         {
 105                 // Request: DataType type = 1;
 106                 int type = request->type();
 107                 // Request: Encoding encoding = 2;
 108                 frr::Encoding encoding = request->encoding();
 109                 // Request: bool with_defaults = 3;
 110                 bool with_defaults = request->with_defaults();
 111
 112                 if (nb_dbg_client_grpc)
 113                         zlog_debug(
 114                                 "received RPC Get(type: %u, encoding: %u, with_defaults: %u)",
 115                                 type, encoding, with_defaults);
 116
 117                 // Request: repeated string path = 4;
 118                 auto paths = request->path();
 119                 for (const std::string &path : paths) {
 120                         frr::GetResponse response;
 121                         grpc::Status status;
 122
 123                         // Response: int64 timestamp = 1;
 124                         response.set_timestamp(time(NULL));
 125
 126                         // Response: DataTree data = 2;
 127                         auto *data = response.mutable_data();
 128                         data->set_encoding(request->encoding());
 129                         status = get_path(data, path, type,
 130                                           encoding2lyd_format(encoding),
 131                                           with_defaults);
 132
 133                         // Something went wrong...
 134                         if (!status.ok())
 135                                 return status;
 136
 137                         writer->Write(response);
 138                 }
 139
 140                 if (nb_dbg_client_grpc)
 141                         zlog_debug("received RPC Get() end");
 142
 143                 return grpc::Status::OK;
 144         }
 145
 146         grpc::Status
 147         CreateCandidate(grpc::ServerContext *context,
 148                         frr::CreateCandidateRequest const *request,
 149                         frr::CreateCandidateResponse *response) override
 150         {
 151                 if (nb_dbg_client_grpc)
 152                         zlog_debug("received RPC CreateCandidate()");
 153
 154                 struct candidate *candidate = create_candidate();
 155                 if (!candidate)
 156                         return grpc::Status(
 157                                 grpc::StatusCode::RESOURCE_EXHAUSTED,
 158                                 "Can't create candidate configuration");
 159
 160                 // Response: uint32 candidate_id = 1;
 161                 response->set_candidate_id(candidate->id);
 162
 163                 return grpc::Status::OK;
 164         }
 165
 166         grpc::Status
 167         DeleteCandidate(grpc::ServerContext *context,
 168                         frr::DeleteCandidateRequest const *request,
 169                         frr::DeleteCandidateResponse *response) override
 170         {
 171                 // Request: uint32 candidate_id = 1;
 172                 uint32_t candidate_id = request->candidate_id();
 173
 174                 if (nb_dbg_client_grpc)
 175                         zlog_debug(
 176                                 "received RPC DeleteCandidate(candidate_id: %u)",
 177                                 candidate_id);
 178
 179                 struct candidate *candidate = get_candidate(candidate_id);
 180                 if (!candidate)
 181                         return grpc::Status(
 182                                 grpc::StatusCode::NOT_FOUND,
 183                                 "candidate configuration not found");
 184
 185                 delete_candidate(candidate);
 186
 187                 return grpc::Status::OK;
 188         }
 189
 190         grpc::Status
 191         UpdateCandidate(grpc::ServerContext *context,
 192                         frr::UpdateCandidateRequest const *request,
 193                         frr::UpdateCandidateResponse *response) override
 194         {
 195                 // Request: uint32 candidate_id = 1;
 196                 uint32_t candidate_id = request->candidate_id();
 197
 198                 if (nb_dbg_client_grpc)
 199                         zlog_debug(
 200                                 "received RPC UpdateCandidate(candidate_id: %u)",
 201                                 candidate_id);
 202
 203                 struct candidate *candidate = get_candidate(candidate_id);
 204                 if (!candidate)
 205                         return grpc::Status(
 206                                 grpc::StatusCode::NOT_FOUND,
 207                                 "candidate configuration not found");
 208
 209                 if (candidate->transaction)
 210                         return grpc::Status(
 211                                 grpc::StatusCode::FAILED_PRECONDITION,
 212                                 "candidate is in the middle of a transaction");
 213
 214                 if (nb_candidate_update(candidate->config) != NB_OK)
 215                         return grpc::Status(
 216                                 grpc::StatusCode::INTERNAL,
 217                                 "failed to update candidate configuration");
 218
 219                 return grpc::Status::OK;
 220         }
 221
 222         grpc::Status
 223         EditCandidate(grpc::ServerContext *context,
 224                       frr::EditCandidateRequest const *request,
 225                       frr::EditCandidateResponse *response) override
 226         {
 227                 // Request: uint32 candidate_id = 1;
 228                 uint32_t candidate_id = request->candidate_id();
 229
 230                 if (nb_dbg_client_grpc)
 231                         zlog_debug(
 232                                 "received RPC EditCandidate(candidate_id: %u)",
 233                                 candidate_id);
 234
 235                 struct candidate *candidate = get_candidate(candidate_id);
 236                 if (!candidate)
 237                         return grpc::Status(
 238                                 grpc::StatusCode::NOT_FOUND,
 239                                 "candidate configuration not found");
 240
 241                 // Create a copy of the candidate. For consistency, we need to
 242                 // ensure that either all changes are accepted or none are (in
 243                 // the event of an error).
 244                 struct nb_config *candidate_tmp =
 245                         nb_config_dup(candidate->config);
 246
 247                 auto pvs = request->update();
 248                 for (const frr::PathValue &pv : pvs) {
 249                         if (yang_dnode_edit(candidate_tmp->dnode, pv.path(),
 250                                             pv.value())
 251                             != 0) {
 252                                 nb_config_free(candidate_tmp);
 253                                 return grpc::Status(
 254                                         grpc::StatusCode::INVALID_ARGUMENT,
 255                                         "Failed to update \"" + pv.path()
 256                                                 + "\"");
 257                         }
 258                 }
 259
 260                 pvs = request->delete_();
 261                 for (const frr::PathValue &pv : pvs) {
 262                         if (yang_dnode_delete(candidate_tmp->dnode, pv.path())
 263                             != 0) {
 264                                 nb_config_free(candidate_tmp);
 265                                 return grpc::Status(
 266                                         grpc::StatusCode::INVALID_ARGUMENT,
 267                                         "Failed to remove \"" + pv.path()
 268                                                 + "\"");
 269                         }
 270                 }
 271
 272                 // No errors, accept all changes.
 273                 nb_config_replace(candidate->config, candidate_tmp, false);
 274
 275                 return grpc::Status::OK;
 276         }
 277
 278         grpc::Status
 279         LoadToCandidate(grpc::ServerContext *context,
 280                         frr::LoadToCandidateRequest const *request,
 281                         frr::LoadToCandidateResponse *response) override
 282         {
 283                 // Request: uint32 candidate_id = 1;
 284                 uint32_t candidate_id = request->candidate_id();
 285                 // Request: LoadType type = 2;
 286                 int load_type = request->type();
 287                 // Request: DataTree config = 3;
 288                 auto config = request->config();
 289
 290                 if (nb_dbg_client_grpc)
 291                         zlog_debug(
 292                                 "received RPC LoadToCandidate(candidate_id: %u)",
 293                                 candidate_id);
 294
 295                 struct candidate *candidate = get_candidate(candidate_id);
 296                 if (!candidate)
 297                         return grpc::Status(
 298                                 grpc::StatusCode::NOT_FOUND,
 299                                 "candidate configuration not found");
 300
 301                 struct lyd_node *dnode = dnode_from_data_tree(&config, true);
 302                 if (!dnode)
 303                         return grpc::Status(
 304                                 grpc::StatusCode::INTERNAL,
 305                                 "Failed to parse the configuration");
 306
 307                 struct nb_config *loaded_config = nb_config_new(dnode);
 308
 309                 if (load_type == frr::LoadToCandidateRequest::REPLACE)
 310                         nb_config_replace(candidate->config, loaded_config,
 311                                           false);
 312                 else if (nb_config_merge(candidate->config, loaded_config,
 313                                          false)
 314                          != NB_OK)
 315                         return grpc::Status(
 316                                 grpc::StatusCode::INTERNAL,
 317                                 "Failed to merge the loaded configuration");
 318
 319                 return grpc::Status::OK;
 320         }
 321
 322         grpc::Status Commit(grpc::ServerContext *context,
 323                             frr::CommitRequest const *request,
 324                             frr::CommitResponse *response) override
 325         {
 326                 // Request: uint32 candidate_id = 1;
 327                 uint32_t candidate_id = request->candidate_id();
 328                 // Request: Phase phase = 2;
 329                 int phase = request->phase();
 330                 // Request: string comment = 3;
 331                 const std::string comment = request->comment();
 332
 333                 if (nb_dbg_client_grpc)
 334                         zlog_debug("received RPC Commit(candidate_id: %u)",
 335                                    candidate_id);
 336
 337                 // Find candidate configuration.
 338                 struct candidate *candidate = get_candidate(candidate_id);
 339                 if (!candidate)
 340                         return grpc::Status(
 341                                 grpc::StatusCode::NOT_FOUND,
 342                                 "candidate configuration not found");
 343
 344                 int ret = NB_OK;
 345                 uint32_t transaction_id = 0;
 346
 347                 // Check for misuse of the two-phase commit protocol.
 348                 switch (phase) {
 349                 case frr::CommitRequest::PREPARE:
 350                 case frr::CommitRequest::ALL:
 351                         if (candidate->transaction)
 352                                 return grpc::Status(
 353                                         grpc::StatusCode::FAILED_PRECONDITION,
 354                                         "pending transaction in progress");
 355                         break;
 356                 case frr::CommitRequest::ABORT:
 357                 case frr::CommitRequest::APPLY:
 358                         if (!candidate->transaction)
 359                                 return grpc::Status(
 360                                         grpc::StatusCode::FAILED_PRECONDITION,
 361                                         "no transaction in progress");
 362                         break;
 363                 default:
 364                         break;
 365                 }
 366
 367                 // Execute the user request.
 368                 switch (phase) {
 369                 case frr::CommitRequest::VALIDATE:
 370                         ret = nb_candidate_validate(candidate->config);
 371                         break;
 372                 case frr::CommitRequest::PREPARE:
 373                         ret = nb_candidate_commit_prepare(
 374                                 candidate->config, NB_CLIENT_GRPC, NULL,
 375                                 comment.c_str(), &candidate->transaction);
 376                         break;
 377                 case frr::CommitRequest::ABORT:
 378                         nb_candidate_commit_abort(candidate->transaction);
 379                         break;
 380                 case frr::CommitRequest::APPLY:
 381                         nb_candidate_commit_apply(candidate->transaction, true,
 382                                                   &transaction_id);
 383                         break;
 384                 case frr::CommitRequest::ALL:
 385                         ret = nb_candidate_commit(
 386                                 candidate->config, NB_CLIENT_GRPC, NULL, true,
 387                                 comment.c_str(), &transaction_id);
 388                         break;
 389                 }
 390
 391                 // Map northbound error codes to gRPC error codes.
 392                 switch (ret) {
 393                 case NB_ERR_NO_CHANGES:
 394                         return grpc::Status(
 395                                 grpc::StatusCode::ABORTED,
 396                                 "No configuration changes detected");
 397                 case NB_ERR_LOCKED:
 398                         return grpc::Status(
 399                                 grpc::StatusCode::UNAVAILABLE,
 400                                 "There's already a transaction in progress");
 401                 case NB_ERR_VALIDATION:
 402                         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
 403                                             "Validation error");
 404                 case NB_ERR_RESOURCE:
 405                         return grpc::Status(
 406                                 grpc::StatusCode::RESOURCE_EXHAUSTED,
 407                                 "Failed do allocate resources");
 408                 case NB_ERR:
 409                         return grpc::Status(grpc::StatusCode::INTERNAL,
 410                                             "Internal error");
 411                 default:
 412                         break;
 413                 }
 414
 415                 // Response: uint32 transaction_id = 1;
 416                 if (transaction_id)
 417                         response->set_transaction_id(transaction_id);
 418
 419                 return grpc::Status::OK;
 420         }
 421
 422         grpc::Status
 423         ListTransactions(grpc::ServerContext *context,
 424                          frr::ListTransactionsRequest const *request,
 425                          grpc::ServerWriter<frr::ListTransactionsResponse>
 426                                  *writer) override
 427         {
 428                 if (nb_dbg_client_grpc)
 429                         zlog_debug("received RPC ListTransactions()");
 430
 431                 nb_db_transactions_iterate(list_transactions_cb, writer);
 432
 433                 return grpc::Status::OK;
 434         }
 435
 436         grpc::Status
 437         GetTransaction(grpc::ServerContext *context,
 438                        frr::GetTransactionRequest const *request,
 439                        frr::GetTransactionResponse *response) override
 440         {
 441                 struct nb_config *nb_config;
 442
 443                 // Request: uint32 transaction_id = 1;
 444                 uint32_t transaction_id = request->transaction_id();
 445                 // Request: Encoding encoding = 2;
 446                 frr::Encoding encoding = request->encoding();
 447                 // Request: bool with_defaults = 3;
 448                 bool with_defaults = request->with_defaults();
 449
 450                 if (nb_dbg_client_grpc)
 451                         zlog_debug(
 452                                 "received RPC GetTransaction(transaction_id: %u, encoding: %u)",
 453                                 transaction_id, encoding);
 454
 455                 // Load configuration from the transactions database.
 456                 nb_config = nb_db_transaction_load(transaction_id);
 457                 if (!nb_config)
 458                         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
 459                                             "Transaction not found");
 460
 461                 // Response: DataTree config = 1;
 462                 auto config = response->mutable_config();
 463                 config->set_encoding(encoding);
 464
 465                 // Dump data using the requested format.
 466                 if (data_tree_from_dnode(config, nb_config->dnode,
 467                                          encoding2lyd_format(encoding),
 468                                          with_defaults)
 469                     != 0) {
 470                         nb_config_free(nb_config);
 471                         return grpc::Status(grpc::StatusCode::INTERNAL,
 472                                             "Failed to dump data");
 473                 }
 474
 475                 nb_config_free(nb_config);
 476
 477                 return grpc::Status::OK;
 478         }
 479
 480         grpc::Status LockConfig(grpc::ServerContext *context,
 481                                 frr::LockConfigRequest const *request,
 482                                 frr::LockConfigResponse *response) override
 483         {
 484                 if (nb_dbg_client_grpc)
 485                         zlog_debug("received RPC LockConfig()");
 486
 487                 if (nb_running_lock(NB_CLIENT_GRPC, NULL))
 488                         return grpc::Status(
 489                                 grpc::StatusCode::FAILED_PRECONDITION,
 490                                 "running configuration is locked already");
 491
 492                 return grpc::Status::OK;
 493         }
 494
 495         grpc::Status UnlockConfig(grpc::ServerContext *context,
 496                                   frr::UnlockConfigRequest const *request,
 497                                   frr::UnlockConfigResponse *response) override
 498         {
 499                 if (nb_dbg_client_grpc)
 500                         zlog_debug("received RPC UnlockConfig()");
 501
 502                 if (nb_running_unlock(NB_CLIENT_GRPC, NULL))
 503                         return grpc::Status(
 504                                 grpc::StatusCode::FAILED_PRECONDITION,
 505                                 "failed to unlock the running configuration");
 506
 507                 return grpc::Status::OK;
 508         }
 509
 510         grpc::Status Execute(grpc::ServerContext *context,
 511                              frr::ExecuteRequest const *request,
 512                              frr::ExecuteResponse *response) override
 513         {
 514                 struct nb_node *nb_node;
 515                 struct list *input_list;
 516                 struct list *output_list;
 517                 struct listnode *node;
 518                 struct yang_data *data;
 519                 const char *xpath;
 520
 521                 // Request: string path = 1;
 522                 xpath = request->path().c_str();
 523
 524                 if (nb_dbg_client_grpc)
 525                         zlog_debug("received RPC Execute(path: \"%s\")", xpath);
 526
 527                 if (request->path().empty())
 528                         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
 529                                             "Data path is empty");
 530
 531                 nb_node = nb_node_find(xpath);
 532                 if (!nb_node)
 533                         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
 534                                             "Unknown data path");
 535
 536                 input_list = yang_data_list_new();
 537                 output_list = yang_data_list_new();
 538
 539                 // Read input parameters.
 540                 auto input = request->input();
 541                 for (const frr::PathValue &pv : input) {
 542                         // Request: repeated PathValue input = 2;
 543                         data = yang_data_new(pv.path().c_str(),
 544                                              pv.value().c_str());
 545                         listnode_add(input_list, data);
 546                 }
 547
 548                 // Execute callback registered for this XPath.
 549                 if (nb_node->cbs.rpc(xpath, input_list, output_list) != NB_OK) {
 550                         flog_warn(EC_LIB_NB_CB_RPC,
 551                                   "%s: rpc callback failed: %s", __func__,
 552                                   xpath);
 553                         list_delete(&input_list);
 554                         list_delete(&output_list);
 555                         return grpc::Status(grpc::StatusCode::INTERNAL,
 556                                             "RPC failed");
 557                 }
 558
 559                 // Process output parameters.
 560                 for (ALL_LIST_ELEMENTS_RO(output_list, node, data)) {
 561                         // Response: repeated PathValue output = 1;
 562                         frr::PathValue *pv = response->add_output();
 563                         pv->set_path(data->xpath);
 564                         pv->set_value(data->value);
 565                 }
 566
 567                 // Release memory.
 568                 list_delete(&input_list);
 569                 list_delete(&output_list);
 570
 571                 return grpc::Status::OK;
 572         }
 573
 574       private:
 575         struct candidate {
 576                 uint32_t id;
 577                 struct nb_config *config;
 578                 struct nb_transaction *transaction;
 579         };
 580         std::map<uint32_t, struct candidate> _candidates;
 581         uint32_t _nextCandidateId;
 582
 583         static int yang_dnode_edit(struct lyd_node *dnode,
 584                                    const std::string &path,
 585                                    const std::string &value)
 586         {
 587                 ly_errno = LY_SUCCESS;
 588                 dnode = lyd_new_path(dnode, ly_native_ctx, path.c_str(),
 589                                      (void *)value.c_str(),
 590                                      (LYD_ANYDATA_VALUETYPE)0,
 591                                      LYD_PATH_OPT_UPDATE);
 592                 if (!dnode && ly_errno != LY_SUCCESS) {
 593                         flog_warn(EC_LIB_LIBYANG, "%s: lyd_new_path() failed",
 594                                   __func__);
 595                         return -1;
 596                 }
 597
 598                 return 0;
 599         }
 600
 601         static int yang_dnode_delete(struct lyd_node *dnode,
 602                                      const std::string &path)
 603         {
 604                 dnode = yang_dnode_get(dnode, path.c_str());
 605                 if (!dnode)
 606                         return -1;
 607
 608                 lyd_free(dnode);
 609
 610                 return 0;
 611         }
 612
 613         static LYD_FORMAT encoding2lyd_format(enum frr::Encoding encoding)
 614         {
 615                 switch (encoding) {
 616                 case frr::JSON:
 617                         return LYD_JSON;
 618                 case frr::XML:
 619                         return LYD_XML;
 620                 }
 621         }
 622
 623         static int get_oper_data_cb(const struct lys_node *snode,
 624                                     struct yang_translator *translator,
 625                                     struct yang_data *data, void *arg)
 626         {
 627                 struct lyd_node *dnode = static_cast<struct lyd_node *>(arg);
 628                 int ret = yang_dnode_edit(dnode, data->xpath, data->value);
 629                 yang_data_free(data);
 630
 631                 return (ret == 0) ? NB_OK : NB_ERR;
 632         }
 633
 634         static void list_transactions_cb(void *arg, int transaction_id,
 635                                          const char *client_name,
 636                                          const char *date, const char *comment)
 637         {
 638                 grpc::ServerWriter<frr::ListTransactionsResponse> *writer =
 639                         static_cast<grpc::ServerWriter<
 640                                 frr::ListTransactionsResponse> *>(arg);
 641                 frr::ListTransactionsResponse response;
 642
 643                 // Response: uint32 id = 1;
 644                 response.set_id(transaction_id);
 645
 646                 // Response: string client = 2;
 647                 response.set_client(client_name);
 648
 649                 // Response: string date = 3;
 650                 response.set_date(date);
 651
 652                 // Response: string comment = 4;
 653                 response.set_comment(comment);
 654
 655                 writer->Write(response);
 656         }
 657
 658         static int data_tree_from_dnode(frr::DataTree *dt,
 659                                         const struct lyd_node *dnode,
 660                                         LYD_FORMAT lyd_format,
 661                                         bool with_defaults)
 662         {
 663                 char *strp;
 664                 int options = 0;
 665
 666                 SET_FLAG(options, LYP_FORMAT | LYP_WITHSIBLINGS);
 667                 if (with_defaults)
 668                         SET_FLAG(options, LYP_WD_ALL);
 669                 else
 670                         SET_FLAG(options, LYP_WD_TRIM);
 671
 672                 if (lyd_print_mem(&strp, dnode, lyd_format, options) == 0) {
 673                         if (strp) {
 674                                 dt->set_data(strp);
 675                                 free(strp);
 676                         }
 677                         return 0;
 678                 }
 679
 680                 return -1;
 681         }
 682
 683         static struct lyd_node *dnode_from_data_tree(const frr::DataTree *dt,
 684                                                      bool config_only)
 685         {
 686                 struct lyd_node *dnode;
 687                 int options;
 688
 689                 if (config_only)
 690                         options = LYD_OPT_CONFIG;
 691                 else
 692                         options = LYD_OPT_DATA | LYD_OPT_DATA_NO_YANGLIB;
 693
 694                 dnode = lyd_parse_mem(ly_native_ctx, dt->data().c_str(),
 695                                       encoding2lyd_format(dt->encoding()),
 696                                       options);
 697
 698                 return dnode;
 699         }
 700
 701         static struct lyd_node *get_dnode_config(const std::string &path)
 702         {
 703                 struct lyd_node *dnode;
 704
 705                 pthread_rwlock_rdlock(&running_config->lock);
 706                 {
 707                         dnode = yang_dnode_get(running_config->dnode,
 708                                                path.empty() ? NULL
 709                                                             : path.c_str());
 710                         if (dnode)
 711                                 dnode = yang_dnode_dup(dnode);
 712                 }
 713                 pthread_rwlock_unlock(&running_config->lock);
 714
 715                 return dnode;
 716         }
 717
 718         static struct lyd_node *get_dnode_state(const std::string &path)
 719         {
 720                 struct lyd_node *dnode;
 721
 722                 dnode = yang_dnode_new(ly_native_ctx, false);
 723                 if (nb_oper_data_iterate(path.c_str(), NULL, 0,
 724                                          get_oper_data_cb, dnode)
 725                     != NB_OK) {
 726                         yang_dnode_free(dnode);
 727                         return NULL;
 728                 }
 729
 730                 return dnode;
 731         }
 732
 733         static grpc::Status get_path(frr::DataTree *dt, const std::string &path,
 734                                      int type, LYD_FORMAT lyd_format,
 735                                      bool with_defaults)
 736         {
 737                 struct lyd_node *dnode_config = NULL;
 738                 struct lyd_node *dnode_state = NULL;
 739                 struct lyd_node *dnode_final;
 740
 741                 // Configuration data.
 742                 if (type == frr::GetRequest_DataType_ALL
 743                     || type == frr::GetRequest_DataType_CONFIG) {
 744                         dnode_config = get_dnode_config(path);
 745                         if (!dnode_config)
 746                                 return grpc::Status(
 747                                         grpc::StatusCode::INVALID_ARGUMENT,
 748                                         "Data path not found");
 749                 }
 750
 751                 // Operational data.
 752                 if (type == frr::GetRequest_DataType_ALL
 753                     || type == frr::GetRequest_DataType_STATE) {
 754                         dnode_state = get_dnode_state(path);
 755                         if (!dnode_state) {
 756                                 if (dnode_config)
 757                                         yang_dnode_free(dnode_config);
 758                                 return grpc::Status(
 759                                         grpc::StatusCode::INVALID_ARGUMENT,
 760                                         "Failed to fetch operational data");
 761                         }
 762                 }
 763
 764                 switch (type) {
 765                 case frr::GetRequest_DataType_ALL:
 766                         //
 767                         // Combine configuration and state data into a single
 768                         // dnode.
 769                         //
 770                         if (lyd_merge(dnode_state, dnode_config,
 771                                       LYD_OPT_EXPLICIT)
 772                             != 0) {
 773                                 yang_dnode_free(dnode_state);
 774                                 yang_dnode_free(dnode_config);
 775                                 return grpc::Status(
 776                                         grpc::StatusCode::INTERNAL,
 777                                         "Failed to merge configuration and state data");
 778                         }
 779
 780                         dnode_final = dnode_state;
 781                         break;
 782                 case frr::GetRequest_DataType_CONFIG:
 783                         dnode_final = dnode_config;
 784                         break;
 785                 case frr::GetRequest_DataType_STATE:
 786                         dnode_final = dnode_state;
 787                         break;
 788                 }
 789
 790                 // Validate data to create implicit default nodes if necessary.
 791                 int validate_opts = 0;
 792                 if (type == frr::GetRequest_DataType_CONFIG)
 793                         validate_opts = LYD_OPT_CONFIG;
 794                 else
 795                         validate_opts = LYD_OPT_DATA | LYD_OPT_DATA_NO_YANGLIB;
 796                 lyd_validate(&dnode_final, validate_opts, ly_native_ctx);
 797
 798                 // Dump data using the requested format.
 799                 int ret = data_tree_from_dnode(dt, dnode_final, lyd_format,
 800                                                with_defaults);
 801                 yang_dnode_free(dnode_final);
 802                 if (ret != 0)
 803                         return grpc::Status(grpc::StatusCode::INTERNAL,
 804                                             "Failed to dump data");
 805
 806                 return grpc::Status::OK;
 807         }
 808
 809         struct candidate *create_candidate(void)
 810         {
 811                 uint32_t candidate_id = ++_nextCandidateId;
 812
 813                 // Check for overflow.
 814                 // TODO: implement an algorithm for unique reusable IDs.
 815                 if (candidate_id == 0)
 816                         return NULL;
 817
 818                 struct candidate *candidate = &_candidates[candidate_id];
 819                 candidate->id = candidate_id;
 820                 pthread_rwlock_rdlock(&running_config->lock);
 821                 {
 822                         candidate->config = nb_config_dup(running_config);
 823                 }
 824                 pthread_rwlock_unlock(&running_config->lock);
 825                 candidate->transaction = NULL;
 826
 827                 return candidate;
 828         }
 829
 830         void delete_candidate(struct candidate *candidate)
 831         {
 832                 _candidates.erase(candidate->id);
 833                 nb_config_free(candidate->config);
 834                 if (candidate->transaction)
 835                         nb_candidate_commit_abort(candidate->transaction);
 836         }
 837
 838         struct candidate *get_candidate(uint32_t candidate_id)
 839         {
 840                 struct candidate *candidate;
 841
 842                 if (_candidates.count(candidate_id) == 0)
 843                         return NULL;
 844
 845                 return &_candidates[candidate_id];
 846         }
 847 };
 848
 849 static void *grpc_pthread_start(void *arg)
 850 {
 851         unsigned long *port = static_cast<unsigned long *>(arg);
 852         NorthboundImpl service;
 853         std::stringstream server_address;
 854
 855         server_address << "0.0.0.0:" << *port;
 856
 857         grpc::ServerBuilder builder;
 858         builder.AddListeningPort(server_address.str(),
 859                                  grpc::InsecureServerCredentials());
 860         builder.RegisterService(&service);
 861
 862         std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
 863
 864         zlog_notice("gRPC server listening on %s",
 865                     server_address.str().c_str());
 866
 867         server->Wait();
 868
 869         return NULL;
 870 }
 871
 872 static int frr_grpc_init(unsigned long *port)
 873 {
 874         /* Create a pthread for gRPC since it runs its own event loop. */
 875         if (pthread_create(&grpc_pthread, NULL, grpc_pthread_start, port)) {
 876                 flog_err(EC_LIB_SYSTEM_CALL, "%s: error creating pthread: %s",
 877                          __func__, safe_strerror(errno));
 878                 return -1;
 879         }
 880         pthread_detach(grpc_pthread);
 881
 882         return 0;
 883 }
 884
 885 static int frr_grpc_finish(void)
 886 {
 887         // TODO: cancel the gRPC pthreads gracefully.
 888
 889         return 0;
 890 }
 891
 892 static int frr_grpc_module_late_init(struct thread_master *tm)
 893 {
 894         static unsigned long port = GRPC_DEFAULT_PORT;
 895         const char *args = THIS_MODULE->load_args;
 896
 897         // Parse port number.
 898         if (args) {
 899                 try {
 900                         port = std::stoul(args);
 901                         if (port < 1024)
 902                                 throw std::invalid_argument(
 903                                         "can't use privileged port");
 904                         if (port > UINT16_MAX)
 905                                 throw std::invalid_argument(
 906                                         "port number is too big");
 907                 } catch (std::exception &e) {
 908                         flog_err(EC_LIB_GRPC_INIT,
 909                                  "%s: failed to parse port number: %s",
 910                                  __func__, e.what());
 911                         goto error;
 912                 }
 913         }
 914
 915         if (frr_grpc_init(&port) < 0)
 916                 goto error;
 917
 918         hook_register(frr_fini, frr_grpc_finish);
 919
 920         return 0;
 921
 922 error:
 923         flog_err(EC_LIB_GRPC_INIT, "failed to initialize the gRPC module");
 924         return -1;
 925 }
 926
 927 static int frr_grpc_module_init(void)
 928 {
 929         hook_register(frr_late_init, frr_grpc_module_late_init);
 930
 931         return 0;
 932 }
 933
 934 FRR_MODULE_SETUP(.name = "frr_grpc", .version = FRR_VERSION,
 935                  .description = "FRR gRPC northbound module",
 936                  .init = frr_grpc_module_init, )