lib: add new gRPC-based northbound plugin

This is an experimental plugin for now. Full documentation will
come later.

Signed-off-by: Renato Westphal <renato@opensourcerouting.org>
This commit is contained in:
Renato Westphal 2019-01-25 18:54:16 -02:00
parent 83981138fe
commit ec2ac5f28a
12 changed files with 1438 additions and 1 deletions

1
.gitignore vendored
View File

@ -49,6 +49,7 @@
*.pb.h
*.pb-c.h
*.pb-c.c
*.pb.cc
*_clippy.c
### dist

View File

@ -123,6 +123,7 @@ include zebra/subdir.am
include watchfrr/subdir.am
include qpb/subdir.am
include fpm/subdir.am
include grpc/subdir.am
include tools/subdir.am
include solaris/subdir.am
@ -198,6 +199,7 @@ EXTRA_DIST += \
doc/user/Makefile \
eigrpd/Makefile \
fpm/Makefile \
grpc/Makefile \
isisd/Makefile \
ldpd/Makefile \
lib/Makefile \

View File

@ -126,12 +126,15 @@ dnl Check CC and friends
dnl --------------------
dnl note orig_cflags is also used further down
orig_cflags="$CFLAGS"
orig_cxxflags="$CXXFLAGS"
AC_LANG([C])
AC_PROG_CC
AC_PROG_CPP
AC_PROG_CXX
AM_PROG_CC_C_O
dnl remove autoconf default "-g -O2"
CFLAGS="$orig_cflags"
CXXFLAGS="$orig_cxxflags"
AC_PROG_CC_C99
dnl NB: see C11 below
@ -447,6 +450,8 @@ AC_ARG_ENABLE([confd],
AS_HELP_STRING([--enable-confd=ARG], [enable confd integration]))
AC_ARG_ENABLE([sysrepo],
AS_HELP_STRING([--enable-sysrepo], [enable sysrepo integration]))
AC_ARG_ENABLE([grpc],
AS_HELP_STRING([--enable-grpc], [enable the gRPC northbound plugin]))
AC_ARG_ENABLE([zeromq],
AS_HELP_STRING([--enable-zeromq], [enable ZeroMQ handler (libfrrzmq)]))
AC_ARG_WITH([libpam],
@ -1678,6 +1683,25 @@ if test "$enable_sysrepo" = "yes"; then
fi
AM_CONDITIONAL([SYSREPO], [test "x$enable_sysrepo" = "xyes"])
dnl ---------------
dnl gRPC
dnl ---------------
if test "$enable_grpc" = "yes"; then
PKG_CHECK_MODULES([GRPC], [grpc grpc++ protobuf], [
AC_CHECK_PROGS([PROTOC], [protoc], [/bin/false])
if test "$PROTOC" = "/bin/false"; then
AC_MSG_FAILURE([grpc requested but protoc not found.])
fi
AC_DEFINE([HAVE_GRPC], [1], [Enable the gRPC northbound plugin])
GRPC=true
], [
GRPC=false
AC_MSG_ERROR([grpc/grpc++ were not found on your system.])
])
fi
AM_CONDITIONAL([GRPC], [test "x$enable_grpc" = "xyes"])
dnl ---------------
dnl math
dnl ---------------

10
grpc/Makefile Normal file
View File

@ -0,0 +1,10 @@
all: ALWAYS
@$(MAKE) -s -C .. grpc/libfrrgrpc_pb.la
%: ALWAYS
@$(MAKE) -s -C .. grpc/$@
Makefile:
#nothing
ALWAYS:
.PHONY: ALWAYS makefiles
.SUFFIXES:

412
grpc/frr-northbound.proto Normal file
View File

@ -0,0 +1,412 @@
//
// Copyright (C) 2019 NetDEF, Inc.
// Renato Westphal
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation; either version 2 of the License, or (at your option)
// any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along
// with this program; see the file COPYING; if not, write to the Free Software
// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
//
syntax = "proto3";
package frr;
// Service specification for the FRR northbound interface.
service Northbound {
// Retrieve the capabilities supported by the target.
rpc GetCapabilities(GetCapabilitiesRequest) returns (GetCapabilitiesResponse) {}
// Retrieve configuration data, state data or both from the target.
rpc Get(GetRequest) returns (stream GetResponse) {}
// Create a new candidate configuration and return a reference to it. The
// created candidate is a copy of the running configuration.
rpc CreateCandidate(CreateCandidateRequest) returns (CreateCandidateResponse) {}
// Delete a candidate configuration.
rpc DeleteCandidate(DeleteCandidateRequest) returns (DeleteCandidateResponse) {}
// Update a candidate configuration by rebasing the changes on top of the
// latest running configuration. Resolve conflicts automatically by giving
// preference to the changes done in the candidate configuration.
rpc UpdateCandidate(UpdateCandidateRequest) returns (UpdateCandidateResponse) {}
// Edit a candidate configuration. All changes are discarded if any error
// happens.
rpc EditCandidate(EditCandidateRequest) returns (EditCandidateResponse) {}
// Load configuration data into a candidate configuration. Both merge and
// replace semantics are supported.
rpc LoadToCandidate(LoadToCandidateRequest) returns (LoadToCandidateResponse) {}
// Create a new configuration transaction using a two-phase commit protocol.
rpc Commit(CommitRequest) returns (CommitResponse) {}
// List the metadata of all configuration transactions recorded in the
// transactions database.
rpc ListTransactions(ListTransactionsRequest) returns (stream ListTransactionsResponse) {}
// Fetch a configuration (identified by its transaction ID) from the
// transactions database.
rpc GetTransaction(GetTransactionRequest) returns (GetTransactionResponse) {}
// Lock the running configuration, preventing other users from changing it.
rpc LockConfig(LockConfigRequest) returns (LockConfigResponse) {}
// Unlock the running configuration.
rpc UnlockConfig(UnlockConfigRequest) returns (UnlockConfigResponse) {}
// Execute a YANG RPC.
rpc Execute(ExecuteRequest) returns (ExecuteResponse) {}
}
// ----------------------- Parameters and return types -------------------------
//
// RPC: GetCapabilities()
//
message GetCapabilitiesRequest {
// Empty.
}
message GetCapabilitiesResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// FRR version.
string frr_version = 1;
// Indicates whether FRR was compiled with support for configuration
// rollbacks or not (--enable-config-rollbacks).
bool rollback_support = 2;
// Supported schema modules.
repeated ModuleData supported_modules = 3;
// Supported encodings.
repeated Encoding supported_encodings = 4;
}
//
// RPC: Get()
//
message GetRequest {
// Type of elements within the data tree.
enum DataType {
// All data elements.
ALL = 0;
// Config elements.
CONFIG = 1;
// State elements.
STATE = 2;
}
// The type of data being requested.
DataType type = 1;
// Encoding to be used.
Encoding encoding = 2;
// Include implicit default nodes.
bool with_defaults = 3;
// Paths requested by the client.
repeated string path = 4;
}
message GetResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::INVALID_ARGUMENT: Invalid YANG data path.
// Timestamp in nanoseconds since Epoch.
int64 timestamp = 1;
// The requested data.
DataTree data = 2;
}
//
// RPC: CreateCandidate()
//
message CreateCandidateRequest {
// Empty.
}
message CreateCandidateResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::RESOURCE_EXHAUSTED: can't create candidate
// configuration.
// Handle to the new created candidate configuration.
uint32 candidate_id = 1;
}
//
// RPC: DeleteCandidate()
//
message DeleteCandidateRequest {
// Candidate configuration to delete.
uint32 candidate_id = 1;
}
message DeleteCandidateResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::NOT_FOUND: Candidate wasn't found.
}
//
// RPC: UpdateCandidate()
//
message UpdateCandidateRequest {
// Candidate configuration to update.
uint32 candidate_id = 1;
}
message UpdateCandidateResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::NOT_FOUND: Candidate wasn't found.
}
//
// RPC: EditCandidate()
//
message EditCandidateRequest {
// Candidate configuration that is going to be edited.
uint32 candidate_id = 1;
// Data elements to be created or updated.
repeated PathValue update = 2;
// Paths to be deleted from the data tree.
repeated PathValue delete = 3;
}
message EditCandidateResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::NOT_FOUND: Candidate wasn't found.
// - grpc::StatusCode::INVALID_ARGUMENT: An error occurred while editing the
// candidate configuration.
}
//
// RPC: LoadToCandidate()
//
message LoadToCandidateRequest {
enum LoadType {
// Merge the data tree into the candidate configuration.
MERGE = 0;
// Replace the candidate configuration by the provided data tree.
REPLACE = 1;
}
// Candidate configuration that is going to be edited.
uint32 candidate_id = 1;
// Load operation to apply.
LoadType type = 2;
// Configuration data.
DataTree config = 3;
}
message LoadToCandidateResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::INVALID_ARGUMENT: An error occurred while performing
// the load operation.
}
//
// RPC: Commit()
//
message CommitRequest {
enum Phase {
// Validate if the configuration changes are valid (phase 0).
VALIDATE = 0;
// Prepare resources to apply the configuration changes (phase 1).
PREPARE = 1;
// Release previously allocated resources (phase 2).
ABORT = 2;
// Apply the configuration changes (phase 2).
APPLY = 3;
// All of the above (VALIDATE + PREPARE + ABORT/APPLY).
//
// This option can't be used to implement network-wide transactions,
// since they require the manager entity to take into account the results
// of the preparation phase of multiple managed devices.
ALL = 4;
}
// Candidate configuration that is going to be committed.
uint32 candidate_id = 1;
// Transaction phase.
Phase phase = 2;
// Assign a comment to this commit.
string comment = 3;
}
message CommitResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::FAILED_PRECONDITION: misuse of the two-phase commit
// protocol.
// - grpc::StatusCode::INVALID_ARGUMENT: Validation error.
// - grpc::StatusCode::RESOURCE_EXHAUSTED: Failure to allocate resource.
// ID of the created configuration transaction (when the phase is APPLY
// or ALL).
uint32 transaction_id = 1;
}
//
// RPC: ListTransactions()
//
message ListTransactionsRequest {
// Empty.
}
message ListTransactionsResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// Transaction ID.
uint32 id = 1;
// Client that committed the transaction.
string client = 2;
// Date and time the transaction was committed.
string date = 3;
// Comment assigned to the transaction.
string comment = 4;
}
//
// RPC: GetTransaction()
//
message GetTransactionRequest {
// Transaction to retrieve.
uint32 transaction_id = 1;
// Encoding to be used.
Encoding encoding = 2;
// Include implicit default nodes.
bool with_defaults = 3;
}
message GetTransactionResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::NOT_FOUND: Transaction wasn't found in the transactions
// database.
DataTree config = 1;
}
//
// RPC: LockConfig()
//
message LockConfigRequest {
// Empty.
}
message LockConfigResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::FAILED_PRECONDITION: Running configuration is
// locked already.
}
//
// RPC: UnlockConfig()
//
message UnlockConfigRequest {
// Empty.
}
message UnlockConfigResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// - grpc::StatusCode::FAILED_PRECONDITION: Running configuration isn't
// locked.
}
//
// RPC: Execute()
//
message ExecuteRequest {
// Path of the YANG RPC or YANG Action.
string path = 1;
// Input parameters.
repeated PathValue input = 2;
}
message ExecuteResponse {
// Return values:
// - grpc::StatusCode::OK: Success.
// Output parameters.
repeated PathValue output = 1;
}
// -------------------------------- Definitions --------------------------------
// YANG module.
message ModuleData {
// Name of the YANG module;
string name = 1;
// Organization publishing the module.
string organization = 2;
// Latest revision of the module;
string revision = 3;
}
// Supported encodings for YANG instance data.
enum Encoding {
JSON = 0;
XML = 1;
}
// Path-value pair representing a data element.
message PathValue {
// YANG data path.
string path = 1;
// Data value.
string value = 2;
}
// YANG instance data.
message DataTree {
Encoding encoding = 1;
string data = 2;
}

30
grpc/subdir.am Normal file
View File

@ -0,0 +1,30 @@
if GRPC
lib_LTLIBRARIES += grpc/libfrrgrpc_pb.la
endif
grpc_libfrrgrpc_pb_la_LDFLAGS = -version-info 0:0:0
grpc_libfrrgrpc_pb_la_CPPFLAGS = $(AM_CPPFLAGS) $(GRPC_CXXFLAGS)
nodist_grpc_libfrrgrpc_pb_la_SOURCES = \
grpc/frr-northbound.pb.cc \
grpc/frr-northbound.grpc.pb.cc \
# end
CLEANFILES += \
grpc/frr-northbound.pb.cc \
grpc/frr-northbound.pb.h \
grpc/frr-northbound.grpc.pb.cc \
grpc/frr-northbound.grpc.pb.h \
# end
EXTRA_DIST += grpc/frr-northbound.proto
AM_V_PROTOC = $(am__v_PROTOC_$(V))
am__v_PROTOC_ = $(am__v_PROTOC_$(AM_DEFAULT_VERBOSITY))
am__v_PROTOC_0 = @echo " PROTOC" $@;
am__v_PROTOC_1 =
.proto.pb.cc:
$(AM_V_PROTOC)$(PROTOC) -I$(top_srcdir) --cpp_out=$(top_srcdir) $(top_srcdir)/$^
.proto.grpc.pb.cc:
$(AM_V_PROTOC)$(PROTOC) -I$(top_srcdir) --grpc_out=$(top_srcdir) --plugin=protoc-gen-grpc=`which grpc_cpp_plugin` $(top_srcdir)/$^

View File

@ -332,6 +332,12 @@ static struct log_ref ferr_lib_err[] = {
.description = "The northbound subsystem has detected that the libsysrepo library returned an error",
.suggestion = "Open an Issue with all relevant log files and restart FRR"
},
{
.code = EC_LIB_GRPC_INIT,
.title = "gRPC initialization error",
.description = "Upon startup FRR failed to properly initialize and startup the gRPC northbound plugin",
.suggestion = "Check if the gRPC libraries are installed correctly in the system.",
},
{
.code = EC_LIB_NB_CB_CONFIG_ABORT,
.title = "A northbound configuration callback has failed in the ABORT phase",

View File

@ -80,6 +80,7 @@ enum lib_log_refs {
EC_LIB_SYSREPO_INIT,
EC_LIB_SYSREPO_DATA_CONVERT,
EC_LIB_LIBSYSREPO,
EC_LIB_GRPC_INIT,
EC_LIB_ID_CONSISTENCY,
EC_LIB_ID_EXHAUST,
};

View File

@ -1827,6 +1827,8 @@ const char *nb_client_name(enum nb_client client)
return "ConfD";
case NB_CLIENT_SYSREPO:
return "Sysrepo";
case NB_CLIENT_GRPC:
return "gRPC";
default:
return "unknown";
}

View File

@ -418,6 +418,7 @@ enum nb_client {
NB_CLIENT_CLI,
NB_CLIENT_CONFD,
NB_CLIENT_SYSREPO,
NB_CLIENT_GRPC,
};
/* Northbound configuration. */

936
lib/northbound_grpc.cpp Normal file
View File

@ -0,0 +1,936 @@
//
// Copyright (C) 2019 NetDEF, Inc.
// Renato Westphal
//
// This program is free software; you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation; either version 2 of the License, or (at your option)
// any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along
// with this program; see the file COPYING; if not, write to the Free Software
// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
//
#include <zebra.h>
#include "log.h"
#include "libfrr.h"
#include "version.h"
#include "command.h"
#include "lib_errors.h"
#include "northbound.h"
#include "northbound_db.h"
#include <iostream>
#include <sstream>
#include <memory>
#include <string>
#include <grpcpp/grpcpp.h>
#include "grpc/frr-northbound.grpc.pb.h"
#define GRPC_DEFAULT_PORT 50051
/*
* NOTE: we can't use the FRR debugging infrastructure here since it uses
* atomics and C++ has a different atomics API. Enable gRPC debugging
* unconditionally until we figure out a way to solve this problem.
*/
static bool nb_dbg_client_grpc = 1;
static pthread_t grpc_pthread;
class NorthboundImpl final : public frr::Northbound::Service
{
public:
NorthboundImpl(void)
{
_nextCandidateId = 0;
}
~NorthboundImpl(void)
{
// Delete candidates.
for (auto it = _candidates.begin(); it != _candidates.end();
it++)
delete_candidate(&it->second);
}
grpc::Status
GetCapabilities(grpc::ServerContext *context,
frr::GetCapabilitiesRequest const *request,
frr::GetCapabilitiesResponse *response) override
{
if (nb_dbg_client_grpc)
zlog_debug("received RPC GetCapabilities()");
// Response: string frr_version = 1;
response->set_frr_version(FRR_VERSION);
// Response: bool rollback_support = 2;
#ifdef HAVE_CONFIG_ROLLBACKS
response->set_rollback_support(true);
#else
response->set_rollback_support(false);
#endif
// Response: repeated ModuleData supported_modules = 3;
struct yang_module *module;
RB_FOREACH (module, yang_modules, &yang_modules) {
auto m = response->add_supported_modules();
m->set_name(module->name);
if (module->info->rev_size)
m->set_revision(module->info->rev[0].date);
m->set_organization(module->info->org);
}
// Response: repeated Encoding supported_encodings = 4;
response->add_supported_encodings(frr::JSON);
response->add_supported_encodings(frr::XML);
return grpc::Status::OK;
}
grpc::Status Get(grpc::ServerContext *context,
frr::GetRequest const *request,
grpc::ServerWriter<frr::GetResponse> *writer) override
{
// Request: DataType type = 1;
int type = request->type();
// Request: Encoding encoding = 2;
frr::Encoding encoding = request->encoding();
// Request: bool with_defaults = 3;
bool with_defaults = request->with_defaults();
if (nb_dbg_client_grpc)
zlog_debug(
"received RPC Get(type: %u, encoding: %u, with_defaults: %u)",
type, encoding, with_defaults);
// Request: repeated string path = 4;
auto paths = request->path();
for (const std::string &path : paths) {
frr::GetResponse response;
grpc::Status status;
// Response: int64 timestamp = 1;
response.set_timestamp(time(NULL));
// Response: DataTree data = 2;
auto *data = response.mutable_data();
data->set_encoding(request->encoding());
status = get_path(data, path, type,
encoding2lyd_format(encoding),
with_defaults);
// Something went wrong...
if (!status.ok())
return status;
writer->Write(response);
}
if (nb_dbg_client_grpc)
zlog_debug("received RPC Get() end");
return grpc::Status::OK;
}
grpc::Status
CreateCandidate(grpc::ServerContext *context,
frr::CreateCandidateRequest const *request,
frr::CreateCandidateResponse *response) override
{
if (nb_dbg_client_grpc)
zlog_debug("received RPC CreateCandidate()");
struct candidate *candidate = create_candidate();
if (!candidate)
return grpc::Status(
grpc::StatusCode::RESOURCE_EXHAUSTED,
"Can't create candidate configuration");
// Response: uint32 candidate_id = 1;
response->set_candidate_id(candidate->id);
return grpc::Status::OK;
}
grpc::Status
DeleteCandidate(grpc::ServerContext *context,
frr::DeleteCandidateRequest const *request,
frr::DeleteCandidateResponse *response) override
{
// Request: uint32 candidate_id = 1;
uint32_t candidate_id = request->candidate_id();
if (nb_dbg_client_grpc)
zlog_debug(
"received RPC DeleteCandidate(candidate_id: %u)",
candidate_id);
struct candidate *candidate = get_candidate(candidate_id);
if (!candidate)
return grpc::Status(
grpc::StatusCode::NOT_FOUND,
"candidate configuration not found");
delete_candidate(candidate);
return grpc::Status::OK;
}
grpc::Status
UpdateCandidate(grpc::ServerContext *context,
frr::UpdateCandidateRequest const *request,
frr::UpdateCandidateResponse *response) override
{
// Request: uint32 candidate_id = 1;
uint32_t candidate_id = request->candidate_id();
if (nb_dbg_client_grpc)
zlog_debug(
"received RPC UpdateCandidate(candidate_id: %u)",
candidate_id);
struct candidate *candidate = get_candidate(candidate_id);
if (!candidate)
return grpc::Status(
grpc::StatusCode::NOT_FOUND,
"candidate configuration not found");
if (candidate->transaction)
return grpc::Status(
grpc::StatusCode::FAILED_PRECONDITION,
"candidate is in the middle of a transaction");
if (nb_candidate_update(candidate->config) != NB_OK)
return grpc::Status(
grpc::StatusCode::INTERNAL,
"failed to update candidate configuration");
return grpc::Status::OK;
}
grpc::Status
EditCandidate(grpc::ServerContext *context,
frr::EditCandidateRequest const *request,
frr::EditCandidateResponse *response) override
{
// Request: uint32 candidate_id = 1;
uint32_t candidate_id = request->candidate_id();
if (nb_dbg_client_grpc)
zlog_debug(
"received RPC EditCandidate(candidate_id: %u)",
candidate_id);
struct candidate *candidate = get_candidate(candidate_id);
if (!candidate)
return grpc::Status(
grpc::StatusCode::NOT_FOUND,
"candidate configuration not found");
// Create a copy of the candidate. For consistency, we need to
// ensure that either all changes are accepted or none are (in
// the event of an error).
struct nb_config *candidate_tmp =
nb_config_dup(candidate->config);
auto pvs = request->update();
for (const frr::PathValue &pv : pvs) {
if (yang_dnode_edit(candidate_tmp->dnode, pv.path(),
pv.value())
!= 0) {
nb_config_free(candidate_tmp);
return grpc::Status(
grpc::StatusCode::INVALID_ARGUMENT,
"Failed to update \"" + pv.path()
+ "\"");
}
}
pvs = request->delete_();
for (const frr::PathValue &pv : pvs) {
if (yang_dnode_delete(candidate_tmp->dnode, pv.path())
!= 0) {
nb_config_free(candidate_tmp);
return grpc::Status(
grpc::StatusCode::INVALID_ARGUMENT,
"Failed to remove \"" + pv.path()
+ "\"");
}
}
// No errors, accept all changes.
nb_config_replace(candidate->config, candidate_tmp, false);
return grpc::Status::OK;
}
grpc::Status
LoadToCandidate(grpc::ServerContext *context,
frr::LoadToCandidateRequest const *request,
frr::LoadToCandidateResponse *response) override
{
// Request: uint32 candidate_id = 1;
uint32_t candidate_id = request->candidate_id();
// Request: LoadType type = 2;
int load_type = request->type();
// Request: DataTree config = 3;
auto config = request->config();
if (nb_dbg_client_grpc)
zlog_debug(
"received RPC LoadToCandidate(candidate_id: %u)",
candidate_id);
struct candidate *candidate = get_candidate(candidate_id);
if (!candidate)
return grpc::Status(
grpc::StatusCode::NOT_FOUND,
"candidate configuration not found");
struct lyd_node *dnode = dnode_from_data_tree(&config, true);
if (!dnode)
return grpc::Status(
grpc::StatusCode::INTERNAL,
"Failed to parse the configuration");
struct nb_config *loaded_config = nb_config_new(dnode);
if (load_type == frr::LoadToCandidateRequest::REPLACE)
nb_config_replace(candidate->config, loaded_config,
false);
else if (nb_config_merge(candidate->config, loaded_config,
false)
!= NB_OK)
return grpc::Status(
grpc::StatusCode::INTERNAL,
"Failed to merge the loaded configuration");
return grpc::Status::OK;
}
grpc::Status Commit(grpc::ServerContext *context,
frr::CommitRequest const *request,
frr::CommitResponse *response) override
{
// Request: uint32 candidate_id = 1;
uint32_t candidate_id = request->candidate_id();
// Request: Phase phase = 2;
int phase = request->phase();
// Request: string comment = 3;
const std::string comment = request->comment();
if (nb_dbg_client_grpc)
zlog_debug("received RPC Commit(candidate_id: %u)",
candidate_id);
// Find candidate configuration.
struct candidate *candidate = get_candidate(candidate_id);
if (!candidate)
return grpc::Status(
grpc::StatusCode::NOT_FOUND,
"candidate configuration not found");
int ret = NB_OK;
uint32_t transaction_id = 0;
// Check for misuse of the two-phase commit protocol.
switch (phase) {
case frr::CommitRequest::PREPARE:
case frr::CommitRequest::ALL:
if (candidate->transaction)
return grpc::Status(
grpc::StatusCode::FAILED_PRECONDITION,
"pending transaction in progress");
break;
case frr::CommitRequest::ABORT:
case frr::CommitRequest::APPLY:
if (!candidate->transaction)
return grpc::Status(
grpc::StatusCode::FAILED_PRECONDITION,
"no transaction in progress");
break;
default:
break;
}
// Execute the user request.
switch (phase) {
case frr::CommitRequest::VALIDATE:
ret = nb_candidate_validate(candidate->config);
break;
case frr::CommitRequest::PREPARE:
ret = nb_candidate_commit_prepare(
candidate->config, NB_CLIENT_GRPC, NULL,
comment.c_str(), &candidate->transaction);
break;
case frr::CommitRequest::ABORT:
nb_candidate_commit_abort(candidate->transaction);
break;
case frr::CommitRequest::APPLY:
nb_candidate_commit_apply(candidate->transaction, true,
&transaction_id);
break;
case frr::CommitRequest::ALL:
ret = nb_candidate_commit(
candidate->config, NB_CLIENT_GRPC, NULL, true,
comment.c_str(), &transaction_id);
break;
}
// Map northbound error codes to gRPC error codes.
switch (ret) {
case NB_ERR_NO_CHANGES:
return grpc::Status(
grpc::StatusCode::ABORTED,
"No configuration changes detected");
case NB_ERR_LOCKED:
return grpc::Status(
grpc::StatusCode::UNAVAILABLE,
"There's already a transaction in progress");
case NB_ERR_VALIDATION:
return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
"Validation error");
case NB_ERR_RESOURCE:
return grpc::Status(
grpc::StatusCode::RESOURCE_EXHAUSTED,
"Failed do allocate resources");
case NB_ERR:
return grpc::Status(grpc::StatusCode::INTERNAL,
"Internal error");
default:
break;
}
// Response: uint32 transaction_id = 1;
if (transaction_id)
response->set_transaction_id(transaction_id);
return grpc::Status::OK;
}
grpc::Status
ListTransactions(grpc::ServerContext *context,
frr::ListTransactionsRequest const *request,
grpc::ServerWriter<frr::ListTransactionsResponse>
*writer) override
{
if (nb_dbg_client_grpc)
zlog_debug("received RPC ListTransactions()");
nb_db_transactions_iterate(list_transactions_cb, writer);
return grpc::Status::OK;
}
grpc::Status
GetTransaction(grpc::ServerContext *context,
frr::GetTransactionRequest const *request,
frr::GetTransactionResponse *response) override
{
struct nb_config *nb_config;
// Request: uint32 transaction_id = 1;
uint32_t transaction_id = request->transaction_id();
// Request: Encoding encoding = 2;
frr::Encoding encoding = request->encoding();
// Request: bool with_defaults = 3;
bool with_defaults = request->with_defaults();
if (nb_dbg_client_grpc)
zlog_debug(
"received RPC GetTransaction(transaction_id: %u, encoding: %u)",
transaction_id, encoding);
// Load configuration from the transactions database.
nb_config = nb_db_transaction_load(transaction_id);
if (!nb_config)
return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
"Transaction not found");
// Response: DataTree config = 1;
auto config = response->mutable_config();
config->set_encoding(encoding);
// Dump data using the requested format.
if (data_tree_from_dnode(config, nb_config->dnode,
encoding2lyd_format(encoding),
with_defaults)
!= 0) {
nb_config_free(nb_config);
return grpc::Status(grpc::StatusCode::INTERNAL,
"Failed to dump data");
}
nb_config_free(nb_config);
return grpc::Status::OK;
}
grpc::Status LockConfig(grpc::ServerContext *context,
frr::LockConfigRequest const *request,
frr::LockConfigResponse *response) override
{
if (nb_dbg_client_grpc)
zlog_debug("received RPC LockConfig()");
if (nb_running_lock(NB_CLIENT_GRPC, NULL))
return grpc::Status(
grpc::StatusCode::FAILED_PRECONDITION,
"running configuration is locked already");
return grpc::Status::OK;
}
grpc::Status UnlockConfig(grpc::ServerContext *context,
frr::UnlockConfigRequest const *request,
frr::UnlockConfigResponse *response) override
{
if (nb_dbg_client_grpc)
zlog_debug("received RPC UnlockConfig()");
if (nb_running_unlock(NB_CLIENT_GRPC, NULL))
return grpc::Status(
grpc::StatusCode::FAILED_PRECONDITION,
"failed to unlock the running configuration");
return grpc::Status::OK;
}
grpc::Status Execute(grpc::ServerContext *context,
frr::ExecuteRequest const *request,
frr::ExecuteResponse *response) override
{
struct nb_node *nb_node;
struct list *input_list;
struct list *output_list;
struct listnode *node;
struct yang_data *data;
const char *xpath;
// Request: string path = 1;
xpath = request->path().c_str();
if (nb_dbg_client_grpc)
zlog_debug("received RPC Execute(path: \"%s\")", xpath);
if (request->path().empty())
return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
"Data path is empty");
nb_node = nb_node_find(xpath);
if (!nb_node)
return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
"Unknown data path");
input_list = yang_data_list_new();
output_list = yang_data_list_new();
// Read input parameters.
auto input = request->input();
for (const frr::PathValue &pv : input) {
// Request: repeated PathValue input = 2;
data = yang_data_new(pv.path().c_str(),
pv.value().c_str());
listnode_add(input_list, data);
}
// Execute callback registered for this XPath.
if (nb_node->cbs.rpc(xpath, input_list, output_list) != NB_OK) {
flog_warn(EC_LIB_NB_CB_RPC,
"%s: rpc callback failed: %s", __func__,
xpath);
list_delete(&input_list);
list_delete(&output_list);
return grpc::Status(grpc::StatusCode::INTERNAL,
"RPC failed");
}
// Process output parameters.
for (ALL_LIST_ELEMENTS_RO(output_list, node, data)) {
// Response: repeated PathValue output = 1;
frr::PathValue *pv = response->add_output();
pv->set_path(data->xpath);
pv->set_value(data->value);
}
// Release memory.
list_delete(&input_list);
list_delete(&output_list);
return grpc::Status::OK;
}
private:
struct candidate {
uint32_t id;
struct nb_config *config;
struct nb_transaction *transaction;
};
std::map<uint32_t, struct candidate> _candidates;
uint32_t _nextCandidateId;
static int yang_dnode_edit(struct lyd_node *dnode,
const std::string &path,
const std::string &value)
{
ly_errno = LY_SUCCESS;
dnode = lyd_new_path(dnode, ly_native_ctx, path.c_str(),
(void *)value.c_str(),
(LYD_ANYDATA_VALUETYPE)0,
LYD_PATH_OPT_UPDATE);
if (!dnode && ly_errno != LY_SUCCESS) {
flog_warn(EC_LIB_LIBYANG, "%s: lyd_new_path() failed",
__func__);
return -1;
}
return 0;
}
static int yang_dnode_delete(struct lyd_node *dnode,
const std::string &path)
{
dnode = yang_dnode_get(dnode, path.c_str());
if (!dnode)
return -1;
lyd_free(dnode);
return 0;
}
static LYD_FORMAT encoding2lyd_format(enum frr::Encoding encoding)
{
switch (encoding) {
case frr::JSON:
return LYD_JSON;
case frr::XML:
return LYD_XML;
}
}
static int get_oper_data_cb(const struct lys_node *snode,
struct yang_translator *translator,
struct yang_data *data, void *arg)
{
struct lyd_node *dnode = static_cast<struct lyd_node *>(arg);
int ret = yang_dnode_edit(dnode, data->xpath, data->value);
yang_data_free(data);
return (ret == 0) ? NB_OK : NB_ERR;
}
static void list_transactions_cb(void *arg, int transaction_id,
const char *client_name,
const char *date, const char *comment)
{
grpc::ServerWriter<frr::ListTransactionsResponse> *writer =
static_cast<grpc::ServerWriter<
frr::ListTransactionsResponse> *>(arg);
frr::ListTransactionsResponse response;
// Response: uint32 id = 1;
response.set_id(transaction_id);
// Response: string client = 2;
response.set_client(client_name);
// Response: string date = 3;
response.set_date(date);
// Response: string comment = 4;
response.set_comment(comment);
writer->Write(response);
}
static int data_tree_from_dnode(frr::DataTree *dt,
const struct lyd_node *dnode,
LYD_FORMAT lyd_format,
bool with_defaults)
{
char *strp;
int options = 0;
SET_FLAG(options, LYP_FORMAT | LYP_WITHSIBLINGS);
if (with_defaults)
SET_FLAG(options, LYP_WD_ALL);
else
SET_FLAG(options, LYP_WD_TRIM);
if (lyd_print_mem(&strp, dnode, lyd_format, options) == 0) {
if (strp) {
dt->set_data(strp);
free(strp);
}
return 0;
}
return -1;
}
static struct lyd_node *dnode_from_data_tree(const frr::DataTree *dt,
bool config_only)
{
struct lyd_node *dnode;
int options;
if (config_only)
options = LYD_OPT_CONFIG;
else
options = LYD_OPT_DATA | LYD_OPT_DATA_NO_YANGLIB;
dnode = lyd_parse_mem(ly_native_ctx, dt->data().c_str(),
encoding2lyd_format(dt->encoding()),
options);
return dnode;
}
static struct lyd_node *get_dnode_config(const std::string &path)
{
struct lyd_node *dnode;
pthread_rwlock_rdlock(&running_config->lock);
{
dnode = yang_dnode_get(running_config->dnode,
path.empty() ? NULL
: path.c_str());
if (dnode)
dnode = yang_dnode_dup(dnode);
}
pthread_rwlock_unlock(&running_config->lock);
return dnode;
}
static struct lyd_node *get_dnode_state(const std::string &path)
{
struct lyd_node *dnode;
dnode = yang_dnode_new(ly_native_ctx, false);
if (nb_oper_data_iterate(path.c_str(), NULL, 0,
get_oper_data_cb, dnode)
!= NB_OK) {
yang_dnode_free(dnode);
return NULL;
}
return dnode;
}
static grpc::Status get_path(frr::DataTree *dt, const std::string &path,
int type, LYD_FORMAT lyd_format,
bool with_defaults)
{
struct lyd_node *dnode_config = NULL;
struct lyd_node *dnode_state = NULL;
struct lyd_node *dnode_final;
// Configuration data.
if (type == frr::GetRequest_DataType_ALL
|| type == frr::GetRequest_DataType_CONFIG) {
dnode_config = get_dnode_config(path);
if (!dnode_config)
return grpc::Status(
grpc::StatusCode::INVALID_ARGUMENT,
"Data path not found");
}
// Operational data.
if (type == frr::GetRequest_DataType_ALL
|| type == frr::GetRequest_DataType_STATE) {
dnode_state = get_dnode_state(path);
if (!dnode_state) {
if (dnode_config)
yang_dnode_free(dnode_config);
return grpc::Status(
grpc::StatusCode::INVALID_ARGUMENT,
"Failed to fetch operational data");
}
}
switch (type) {
case frr::GetRequest_DataType_ALL:
//
// Combine configuration and state data into a single
// dnode.
//
if (lyd_merge(dnode_state, dnode_config,
LYD_OPT_EXPLICIT)
!= 0) {
yang_dnode_free(dnode_state);
yang_dnode_free(dnode_config);
return grpc::Status(
grpc::StatusCode::INTERNAL,
"Failed to merge configuration and state data");
}
dnode_final = dnode_state;
break;
case frr::GetRequest_DataType_CONFIG:
dnode_final = dnode_config;
break;
case frr::GetRequest_DataType_STATE:
dnode_final = dnode_state;
break;
}
// Validate data to create implicit default nodes if necessary.
int validate_opts = 0;
if (type == frr::GetRequest_DataType_CONFIG)
validate_opts = LYD_OPT_CONFIG;
else
validate_opts = LYD_OPT_DATA | LYD_OPT_DATA_NO_YANGLIB;
lyd_validate(&dnode_final, validate_opts, ly_native_ctx);
// Dump data using the requested format.
int ret = data_tree_from_dnode(dt, dnode_final, lyd_format,
with_defaults);
yang_dnode_free(dnode_final);
if (ret != 0)
return grpc::Status(grpc::StatusCode::INTERNAL,
"Failed to dump data");
return grpc::Status::OK;
}
struct candidate *create_candidate(void)
{
uint32_t candidate_id = ++_nextCandidateId;
// Check for overflow.
// TODO: implement an algorithm for unique reusable IDs.
if (candidate_id == 0)
return NULL;
struct candidate *candidate = &_candidates[candidate_id];
candidate->id = candidate_id;
pthread_rwlock_rdlock(&running_config->lock);
{
candidate->config = nb_config_dup(running_config);
}
pthread_rwlock_unlock(&running_config->lock);
candidate->transaction = NULL;
return candidate;
}
void delete_candidate(struct candidate *candidate)
{
_candidates.erase(candidate->id);
nb_config_free(candidate->config);
if (candidate->transaction)
nb_candidate_commit_abort(candidate->transaction);
}
struct candidate *get_candidate(uint32_t candidate_id)
{
struct candidate *candidate;
if (_candidates.count(candidate_id) == 0)
return NULL;
return &_candidates[candidate_id];
}
};
static void *grpc_pthread_start(void *arg)
{
unsigned long *port = static_cast<unsigned long *>(arg);
NorthboundImpl service;
std::stringstream server_address;
server_address << "0.0.0.0:" << *port;
grpc::ServerBuilder builder;
builder.AddListeningPort(server_address.str(),
grpc::InsecureServerCredentials());
builder.RegisterService(&service);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
zlog_notice("gRPC server listening on %s",
server_address.str().c_str());
server->Wait();
return NULL;
}
static int frr_grpc_init(unsigned long *port)
{
/* Create a pthread for gRPC since it runs its own event loop. */
if (pthread_create(&grpc_pthread, NULL, grpc_pthread_start, port)) {
flog_err(EC_LIB_SYSTEM_CALL, "%s: error creating pthread: %s",
__func__, safe_strerror(errno));
return -1;
}
pthread_detach(grpc_pthread);
return 0;
}
static int frr_grpc_finish(void)
{
// TODO: cancel the gRPC pthreads gracefully.
return 0;
}
static int frr_grpc_module_late_init(struct thread_master *tm)
{
static unsigned long port = GRPC_DEFAULT_PORT;
const char *args = THIS_MODULE->load_args;
// Parse port number.
if (args) {
try {
port = std::stoul(args);
if (port < 1024)
throw std::invalid_argument(
"can't use privileged port");
if (port > UINT16_MAX)
throw std::invalid_argument(
"port number is too big");
} catch (std::exception &e) {
flog_err(EC_LIB_GRPC_INIT,
"%s: failed to parse port number: %s",
__func__, e.what());
goto error;
}
}
if (frr_grpc_init(&port) < 0)
goto error;
hook_register(frr_fini, frr_grpc_finish);
return 0;
error:
flog_err(EC_LIB_GRPC_INIT, "failed to initialize the gRPC module");
return -1;
}
static int frr_grpc_module_init(void)
{
hook_register(frr_late_init, frr_grpc_module_late_init);
return 0;
}
FRR_MODULE_SETUP(.name = "frr_grpc", .version = FRR_VERSION,
.description = "FRR gRPC northbound module",
.init = frr_grpc_module_init, )

View File

@ -302,6 +302,18 @@ lib_sysrepo_la_LDFLAGS = -avoid-version -module -shared -export-dynamic
lib_sysrepo_la_LIBADD = lib/libfrr.la $(SYSREPO_LIBS)
lib_sysrepo_la_SOURCES = lib/northbound_sysrepo.c
#
# gRPC northbound plugin
#
if GRPC
module_LTLIBRARIES += lib/grpc.la
endif
lib_grpc_la_CXXFLAGS = $(WERROR) $(GRPC_CFLAGS)
lib_grpc_la_LDFLAGS = -avoid-version -module -shared -export-dynamic
lib_grpc_la_LIBADD = lib/libfrr.la grpc/libfrrgrpc_pb.la $(GRPC_LIBS)
lib_grpc_la_SOURCES = lib/northbound_grpc.cpp
#
# CLI utilities
#
@ -346,7 +358,7 @@ am__v_CLIPPY_1 =
CLIPPY_DEPS = $(HOSTTOOLS)lib/clippy $(top_srcdir)/python/clidef.py
SUFFIXES = _clippy.c .proto .pb-c.c .pb-c.h .pb.h
SUFFIXES = _clippy.c .proto .pb-c.c .pb-c.h .pb.h .pb.cc .grpc.pb.cc
.c_clippy.c:
@{ test -x $(top_builddir)/$(HOSTTOOLS)lib/clippy || \
$(MAKE) -C $(top_builddir)/$(HOSTTOOLS) lib/clippy; }