C++实现基于etcd的gRPC Resolver遇到的问题及代码疑问

头文件缺失问题 #

编译安装gRPC v1.41.0版本后，/usr/local/include/grpc++/目录下找不到resolver.h和balancer.h，但这两个文件存在于gRPC源码的grpc/src/core/lib/resolver/路径中。

Resolver实现相关问题 #

1. 无法将endpoint字符串转换为gRPC的grpc_core::Resolver::Result结构体
2. 不知道如何将转换后的Result设置到gRPC中

#pragma once
#include <grpc/grpc.h>
#include <grpcpp/grpcpp.h>
#include <grpcpp/client_context.h>
#include <grpcpp/create_channel.h>
#include <grpcpp/support/channel_arguments.h>

#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>

#include "etcd-cpp-apiv3/Client.hpp"
#include "etcd-cpp-apiv3/Response.hpp"
#include "etcd-cpp-apiv3/Watcher.hpp"


class EtcdResolver : public grpc_core::Resolver {
public:
    explicit EtcdResolver(const std::string& etcd_address, const std::string& service_name, 
                            const std::string& runtime, const std::string& ca, const std::string& cert, const std::string& prikey)
    {
        etcd::Client* etcd = etcd::Client::WithSSL(etcd_address, ca, cert, prikey);
        if (etcd == nullptr)
        {
            printf("create xetcd client fail\n");
            exit(1);
        }
        std::shared_ptr<etcd::Client> shared_client(etcd);
        client_ = shared_client;
        runtime_ = runtime;
        etcd_address_ = etcd_address;
        service_name_ = service_name;
        ca_ = ca;
        cert_ = cert;
        prikey_ = prikey;
    };

    ~EtcdResolver(){};

    void StartLocked()
    {
        std::vector<std::string> endpoints = FirstGetEndpointsFromEtcd();
        grpc::ChannelArguments args;
        args.SetServiceConfigJSON("{\"loadBalancingConfig\":[{\"round_robin\":{}}]}");
    
        for (const auto& endpoint : endpoints) {
            // how to convert string to grpc_core::Resolver::Result ?
        }
        UpdateStateLocked(result_);

        etcd::Watcher watcher(*client_.get(), service_key_, std::bind(&EtcdResolver::OnWatch, this, std::placeholders::_1), true);
        std::shared_ptr<etcd::Watcher> shared_watcher(&watcher);
        watcher_ = shared_watcher;
    };

    void RequestReresolution(){};

    void ResetBackoff(){};

    void UpdateStateLocked(grpc_core::Resolver::Result result)
    {
        // how to set the result into grpc?
    };

private:
    std::vector<std::string> FirstGetEndpointsFromEtcd()
    {
        std::vector<std::string> endpoints;
        service_key_ = std::string("services_prefix/") + runtime_ + "/" + service_name_ + "/";
        auto resp = client_->ls(service_key_).get();
        if (resp.error_code() != 0) {
            throw std::runtime_error(resp.error_message());
        }
        for (const auto& val : resp.values()) {
            Json::Reader reader;
            Json::Value value;
            if (reader.parse(val.as_string(), value)){
                endpoints.emplace_back(value["addr"].asString());
            }
        }
        return endpoints;
    };

    void OnWatch(etcd::Response const & resp)
    {
        if (!resp.is_ok()) return;
        std::cout << "OnWatch triggered" << std::endl;
        auto resp2 = client_->ls(service_key_).get();
        if (resp2.error_code() != 0) {
            throw std::runtime_error(resp2.error_message());
        }
        std::unique_lock<std::mutex> lock(mutex_);
        std::vector<std::string> endpoints;
        for (const auto& val : resp2.values()) {
            Json::Reader reader;
            Json::Value value;
            if (reader.parse(val.as_string(), value)){
                endpoints.emplace_back(value["addr"].asString());
            }
        }
        for (const auto& endpoint : endpoints) {
            // how to convert string to grpc_core::Resolver::Result ?
        }
        std::unique_lock<std::mutex> unlock(mutex_);
        UpdateStateLocked(result_);
    };

private:
    std::string runtime_;
    std::string etcd_address_;
    std::string service_name_;
    std::shared_ptr<etcd::Client> client_;
    std::shared_ptr<etcd::Watcher> watcher_;
    grpc_core::Resolver::Result result_;
    std::mutex mutex_;

    std::string ca_;
    std::string cert_;
    std::string prikey_;
    std::string service_key_;
};

阅读gRPC源码后写出以下转换代码，但不确定正确性：

// convert string to grpc_core::Resolver::Result
grpc_resolved_address resolved_address;
memset(&resolved_address, 0, sizeof(grpc_resolved_address));

absl::StatusOr<grpc_core::URI> uri = grpc_core::URI::Parse(endpoint);
grpc_parse_uri(uri, &resolved_address);
result_.addresses.emplace_back(grpc_core::ServerAddress(resolved_address, nullptr));

头文件缺失问题 #

resolver.h和balancer.h属于gRPC core层内部头文件，默认安装不会导出到/usr/local/include/grpc++/。编译自定义Resolver时，需要手动添加源码路径到头文件搜索列表：

• 若用CMake，添加：

  include_directories(${GRPC_SOURCE_DIR}/src/core/lib/resolver)
  

• 若直接用g++编译，添加参数：

  -I/path/to/grpc/src/core/lib/resolver
  

Endpoint转Resolver::Result的正确方式 #

你写的转换代码参数有误，正确做法如下：

#include <grpc_core/uri.h>
#include <grpc_core/server_address.h>

grpc_core::Resolver::Result result;
absl::StatusOr<grpc_core::URI> uri = grpc_core::URI::Parse(endpoint);
if (!uri.ok()) {
    // 处理解析失败逻辑
    return;
}
grpc_resolved_address resolved_addr;
if (!grpc_resolve_uri(uri->ToString().c_str(), nullptr, &resolved_addr)) {
    // 处理解析失败逻辑
    return;
}
// 第二个参数为负载均衡属性，可传nullptr或自定义属性
result.addresses.emplace_back(grpc_core::ServerAddress(&resolved_addr, nullptr));

将Result设置到gRPC中 #

grpc_core::Resolver父类提供了UpdateState方法用于更新状态，同时你的代码中Watcher创建存在栈对象销毁问题，需修正：

1. 修改UpdateStateLocked方法：

void UpdateStateLocked(grpc_core::Resolver::Result result)
{
    // 调用父类方法传入Result
    UpdateState(std::move(result));
}

2. 修正StartLocked中的Watcher创建与Result转换：

void StartLocked()
{
    std::vector<std::string> endpoints = FirstGetEndpointsFromEtcd();
    grpc_core::Resolver::Result result;
    // 批量转换endpoint
    for (const auto& endpoint : endpoints) {
        absl::StatusOr<grpc_core::URI> uri = grpc_core::URI::Parse(endpoint);
        if (!uri.ok()) continue;
        grpc_resolved_address resolved_addr;
        if (!grpc_resolve_uri(uri->ToString().c_str(), nullptr, &resolved_addr)) continue;
        result.addresses.emplace_back(grpc_core::ServerAddress(&resolved_addr, nullptr));
    }
    // 设置服务配置
    result.service_config = grpc_core::ServiceConfig::Create(
        "{\"loadBalancingConfig\":[{\"round_robin\":{}}]}", nullptr);
    UpdateState(std::move(result));

    // 动态分配Watcher，避免栈对象提前销毁
    watcher_ = std::make_shared<etcd::Watcher>(*client_.get(), service_key_, 
        std::bind(&EtcdResolver::OnWatch, this, std::placeholders::_1), true);
}

3. 修正OnWatch中的锁使用：

void OnWatch(etcd::Response const & resp)
{
    if (!resp.is_ok()) return;
    std::cout << "OnWatch triggered" << std::endl;
    auto resp2 = client_->ls(service_key_).get();
    if (resp2.error_code() != 0) {
        throw std::runtime_error(resp2.error_message());
    }
    std::lock_guard<std::mutex> lock(mutex_);
    grpc_core::Resolver::Result result;
    for (const auto& val : resp2.values()) {
        Json::Reader reader;
        Json::Value value;
        if (!reader.parse(val.as_string(), value)) continue;
        std::string endpoint = value["addr"].asString();
        absl::StatusOr<grpc_core::URI> uri = grpc_core::URI::Parse(endpoint);
        if (!uri.ok()) continue;
        grpc_resolved_address resolved_addr;
        if (!grpc_resolve_uri(uri->ToString().c_str(), nullptr, &resolved_addr)) continue;
        result.addresses.emplace_back(grpc_core::ServerAddress(&resolved_addr, nullptr));
    }
    result.service_config = grpc_core::ServiceConfig::Create(
        "{\"loadBalancingConfig\":[{\"round_robin\":{}}]}", nullptr);
    UpdateState(std::move(result));
}

内容的提问来源于stack exchange，提问作者peng alan