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