Merge pull request #2964 from zhoukangsheng/support_health_check_option

Support channel level application health check options

Author: Huixxi

docs/cn/client.md

@@ -291,7 +291,7 @@ channel.Init("http://...", "random:min_working_instances=6 hold_seconds=10", &op
 | ------------------------- | ----- | ---------------------------------------- | ----------------------- |
 | health_check_interval （R） | 3     | seconds between consecutive health-checkings | src/brpc/socket_map.cpp |
 
-在默认的配置下，一旦server被连接上，它会恢复为可用状态；brpc还提供了应用层健康检查的机制，框架会发送一个HTTP GET请求到该server，请求路径通过-health\_check\_path设置（默认为空），只有当server返回200时，它才会恢复。在两种健康检查机制下，都可通过-health\_check\_timeout\_ms设置超时（默认500ms）。如果在隔离过程中，server从命名服务中删除了，brpc也会停止连接尝试。
+在默认的配置下，一旦server被连接上，它会恢复为可用状态,可通过-health\_check\_timeout\_ms设置超时（默认500ms）；brpc还提供了应用层健康检查的机制，框架会发送一个HTTP GET请求到该server，只有当server返回200时，它才会恢复，在这种机制下，既可通过-health\_check\_path（默认为空）和-health\_check\_timeout\_ms（默认500ms）分别设置全局的健康检查请求路径和超时，也可通过ChannelOptions中的hc_option成员变量来对不同的channel设置不同的请求路径和超时，ChannelOptions设置的健康检查参数优先级要高于gflag参数。如果在隔离过程中，server从命名服务中删除了，brpc也会停止连接尝试。
 
 # 发起访问
 

src/brpc/channel.cpp

@@ -42,6 +42,13 @@ namespace brpc {
 
 DECLARE_bool(enable_rpcz);
 DECLARE_bool(usercode_in_pthread);
+DEFINE_string(health_check_path, "", "Http path of health check call."
+    "By default health check succeeds if the server is connectable."
+    "If this flag is set, health check is not completed until a http "
+    "call to the path succeeds within -health_check_timeout_ms(to make "
+    "sure the server functions well).");
+DEFINE_int32(health_check_timeout_ms, 500, "The timeout for both establishing "
+    "the connection and the http call to -health_check_path over the connection");
 
 ChannelOptions::ChannelOptions()
     : connect_timeout_ms(200)
@@ -70,7 +77,8 @@ ChannelSSLOptions* ChannelOptions::mutable_ssl_options() {
 static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
     if (opt.auth == NULL &&
         !opt.has_ssl_options() &&
-        opt.connection_group.empty()) {
+        opt.connection_group.empty() &&
+        opt.hc_option.health_check_path.empty()) {
         // Returning zeroized result by default is more intuitive for users.
         return ChannelSignature();
     }
@@ -90,6 +98,12 @@ static ChannelSignature ComputeChannelSignature(const ChannelOptions& opt) {
             buf.append("|auth=");
             buf.append((char*)&opt.auth, sizeof(opt.auth));
         }
+        if (!opt.hc_option.health_check_path.empty()) {
+            buf.append("|health_check_path=");
+            buf.append(opt.hc_option.health_check_path);
+            buf.append("|health_check_timeout_ms=");
+            buf.append(std::to_string(opt.hc_option.health_check_timeout_ms));
+        }
         if (opt.has_ssl_options()) {
             const ChannelSSLOptions& ssl = opt.ssl_options();
             buf.push_back('|');
@@ -173,7 +187,10 @@ int Channel::InitChannelOptions(const ChannelOptions* options) {
         LOG(ERROR) << "Channel does not support the protocol";
         return -1;
     }
-
+    if (_options.hc_option.health_check_path.empty()) {
+        _options.hc_option.health_check_path = FLAGS_health_check_path;
+        _options.hc_option.health_check_timeout_ms = FLAGS_health_check_timeout_ms;
+    }
     if (_options.use_rdma) {
 #if BRPC_WITH_RDMA
         if (!OptionsAvailableForRdma(&_options)) {
@@ -349,7 +366,7 @@ int Channel::InitSingle(const butil::EndPoint& server_addr_and_port,
         return -1;
     }
     if (SocketMapInsert(SocketMapKey(server_addr_and_port, sig),
-                        &_server_id, ssl_ctx, _options.use_rdma) != 0) {
+                        &_server_id, ssl_ctx, _options.use_rdma, _options.hc_option) != 0) {
         LOG(ERROR) << "Fail to insert into SocketMap";
         return -1;
     }
@@ -388,6 +405,7 @@ int Channel::Init(const char* ns_url,
     ns_opt.log_succeed_without_server = _options.log_succeed_without_server;
     ns_opt.use_rdma = _options.use_rdma;
     ns_opt.channel_signature = ComputeChannelSignature(_options);
+    ns_opt.hc_option =  _options.hc_option;
     if (CreateSocketSSLContext(_options, &ns_opt.ssl_ctx) != 0) {
         return -1;
     }

src/brpc/channel.h

@@ -36,6 +36,7 @@
 #include "brpc/retry_policy.h"
 #include "brpc/backup_request_policy.h"
 #include "brpc/naming_service_filter.h"
+#include "brpc/health_check_option.h"
 
 namespace brpc {
 
@@ -143,6 +144,10 @@ struct ChannelOptions {
     // Default: ""
     std::string connection_group;
 
+    // Set the health check param according to the channel granularity. 
+    // Its priority is higher than FLAGS_health_check_path and FLAGS_health_check_timeout_ms.
+    // When it is not set, FLAGS_health_check_path and FLAGS_health_check_timeout_ms will take effect.
+    HealthCheckOption hc_option;
 private:
     // SSLOptions is large and not often used, allocate it on heap to
     // prevent ChannelOptions from being bloated in most cases.

src/brpc/details/health_check.cpp

@@ -31,14 +31,6 @@ namespace brpc {
 // Declared at socket.cpp
 extern SocketVarsCollector* g_vars;
 
-DEFINE_string(health_check_path, "", "Http path of health check call."
-        "By default health check succeeds if the server is connectable."
-        "If this flag is set, health check is not completed until a http "
-        "call to the path succeeds within -health_check_timeout_ms(to make "
-        "sure the server functions well).");
-DEFINE_int32(health_check_timeout_ms, 500, "The timeout for both establishing "
-        "the connection and the http call to -health_check_path over the connection");
-
 class HealthCheckChannel : public brpc::Channel {
 public:
     HealthCheckChannel() {}
@@ -65,6 +57,7 @@ class OnAppHealthCheckDone : public google::protobuf::Closure {
     SocketId id;
     int64_t interval_s;
     int64_t last_check_time_ms;
+    HealthCheckOption hc_option;
 };
 
 class HealthCheckManager {
@@ -81,15 +74,16 @@ void HealthCheckManager::StartCheck(SocketId id, int64_t check_interval_s) {
                  << " was abandoned during health checking";
         return;
     }
-    LOG(INFO) << "Checking path=" << ptr->remote_side() << FLAGS_health_check_path;
+    LOG(INFO) << "Checking path=" << ptr->remote_side() << ptr->health_check_path();
     OnAppHealthCheckDone* done = new OnAppHealthCheckDone;
     done->id = id;
     done->interval_s = check_interval_s;
+    done->hc_option = ptr->_hc_option;
     brpc::ChannelOptions options;
     options.protocol = PROTOCOL_HTTP;
     options.max_retry = 0;
     options.timeout_ms =
-        std::min((int64_t)FLAGS_health_check_timeout_ms, check_interval_s * 1000);
+        std::min((int64_t)(done->hc_option.health_check_timeout_ms), check_interval_s * 1000);
     if (done->channel.Init(id, &options) != 0) {
         LOG(WARNING) << "Fail to init health check channel to SocketId=" << id;
         ptr->_ninflight_app_health_check.fetch_sub(
@@ -103,7 +97,7 @@ void HealthCheckManager::StartCheck(SocketId id, int64_t check_interval_s) {
 void* HealthCheckManager::AppCheck(void* arg) {
     OnAppHealthCheckDone* done = static_cast<OnAppHealthCheckDone*>(arg);
     done->cntl.Reset();
-    done->cntl.http_request().uri() = FLAGS_health_check_path;
+    done->cntl.http_request().uri() = done->hc_option.health_check_path;
     ControllerPrivateAccessor(&done->cntl).set_health_check_call();
     done->last_check_time_ms = butil::gettimeofday_ms();
     done->channel.CallMethod(NULL, &done->cntl, NULL, NULL, done);
@@ -121,14 +115,14 @@ void OnAppHealthCheckDone::Run() {
     }
     if (!cntl.Failed() || ptr->Failed()) {
         LOG_IF(INFO, !cntl.Failed()) << "Succeeded to call "
-            << ptr->remote_side() << FLAGS_health_check_path;
+            << ptr->remote_side() << hc_option.health_check_path;
         // if ptr->Failed(), previous SetFailed would trigger next round
         // of hc, just return here.
         ptr->_ninflight_app_health_check.fetch_sub(
                     1, butil::memory_order_relaxed);
         return;
     }
-    RPC_VLOG << "Fail to check path=" << FLAGS_health_check_path
+    RPC_VLOG << "Fail to check path=" << hc_option.health_check_path
         << ", " << cntl.ErrorText();
 
     int64_t sleep_time_ms =
@@ -206,14 +200,14 @@ bool HealthCheckTask::OnTriggeringTask(timespec* next_abstime) {
         hc = ptr->CheckHealth();
     }
     if (hc == 0) {
-        if (!FLAGS_health_check_path.empty()) {
+        if (!ptr->health_check_path().empty()) {
             ptr->_ninflight_app_health_check.fetch_add(
                     1, butil::memory_order_relaxed);
         }
         // See comments above.
         ptr->Revive(2/*note*/);
         ptr->_hc_count = 0;
-        if (!FLAGS_health_check_path.empty()) {
+        if (!ptr->health_check_path().empty()) {
             HealthCheckManager::StartCheck(_id, ptr->_health_check_interval_s);
         }
         ptr->AfterHCCompleted();

src/brpc/details/naming_service_thread.cpp

@@ -126,7 +126,7 @@ void NamingServiceThread::Actions::ResetServers(
         //       to pick those Sockets with the right settings during OnAddedServers
         const SocketMapKey key(_added[i], _owner->_options.channel_signature);
         CHECK_EQ(0, SocketMapInsert(key, &tagged_id.id, _owner->_options.ssl_ctx,
-                                    _owner->_options.use_rdma));
+                                    _owner->_options.use_rdma, _owner->_options.hc_option));
         _added_sockets.push_back(tagged_id);
     }
 

src/brpc/details/naming_service_thread.h

@@ -50,6 +50,7 @@ struct GetNamingServiceThreadOptions {
     bool succeed_without_server;
     bool log_succeed_without_server;
     bool use_rdma;
+    HealthCheckOption hc_option;
     ChannelSignature channel_signature;
     std::shared_ptr<SocketSSLContext> ssl_ctx;
 };

src/brpc/health_check_option.h

@@ -0,0 +1,34 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef  BRPC_HEALTH_CHECK_OPTION_H
+#define  BRPC_HEALTH_CHECK_OPTION_H
+
+#include <string>
+
+namespace brpc {
+
+struct HealthCheckOption {
+    // Http path of health check call
+    std::string health_check_path;
+    // The timeout for both establishing the connection and the http call to health_check_path over the connection
+    int32_t health_check_timeout_ms{500};
+};
+
+}  // namespace brpc
+
+#endif  // BRPC_HEALTH_CHECK_OPTION_H

src/brpc/socket.cpp

@@ -99,7 +99,6 @@ DEFINE_int32(connect_timeout_as_unreachable, 3,
              "times *continuously*, the error is changed to ENETUNREACH which "
              "fails the main socket as well when this socket is pooled.");
 
-DECLARE_int32(health_check_timeout_ms);
 DECLARE_bool(usercode_in_coroutine);
 
 static bool validate_connect_timeout_as_unreachable(const char*, int32_t v) {
@@ -743,6 +742,7 @@ int Socket::OnCreated(const SocketOptions& options) {
     reset_parsing_context(options.initial_parsing_context);
     _correlation_id = 0;
     _health_check_interval_s = options.health_check_interval_s;
+    _hc_option = options.hc_option;
     _is_hc_related_ref_held = false;
     _hc_started.store(false, butil::memory_order_relaxed);
     _ninprocess.store(1, butil::memory_order_relaxed);
@@ -2593,7 +2593,7 @@ int Socket::CheckHealth() {
         LOG(INFO) << "Checking " << *this;
     }
     const timespec duetime =
-        butil::milliseconds_from_now(FLAGS_health_check_timeout_ms);
+        butil::milliseconds_from_now(_hc_option.health_check_timeout_ms);
     const int connected_fd = Connect(&duetime, NULL, NULL);
     if (connected_fd >= 0) {
         ::close(connected_fd);

src/brpc/socket.h

@@ -41,6 +41,7 @@
 #include "brpc/http_method.h"
 #include "brpc/event_dispatcher.h"
 #include "brpc/versioned_ref_with_id.h"
+#include "brpc/health_check_option.h"
 
 namespace brpc {
 namespace policy {
@@ -285,6 +286,7 @@ struct SocketOptions {
     int tcp_user_timeout_ms{ -1};
     // Tag of this socket
     bthread_tag_t bthread_tag{bthread_self_tag()};
+    HealthCheckOption hc_option;
 };
 
 // Abstractions on reading from and writing into file descriptors.
@@ -412,6 +414,10 @@ friend void DereferenceSocket(Socket*);
     // Initialized by SocketOptions.health_check_interval_s.
     int health_check_interval() const { return _health_check_interval_s; }
 
+    const std::string& health_check_path() const { return _hc_option.health_check_path; }
+
+    int32_t health_check_timeout_ms() const {return _hc_option.health_check_timeout_ms; }
+
     // True if health checking is enabled.
     bool HCEnabled() const {
         // This fence makes sure that we see change of
@@ -980,6 +986,7 @@ friend void DereferenceSocket(Socket*);
     int _tcp_user_timeout_ms;
 
     HttpMethod _http_request_method;
+    HealthCheckOption _hc_option;
 };
 
 } // namespace brpc

src/brpc/socket_map.cpp

@@ -91,8 +91,9 @@ SocketMap* get_or_new_client_side_socket_map() {
 
 int SocketMapInsert(const SocketMapKey& key, SocketId* id,
                     const std::shared_ptr<SocketSSLContext>& ssl_ctx,
-                    bool use_rdma) {
-    return get_or_new_client_side_socket_map()->Insert(key, id, ssl_ctx, use_rdma);
+                    bool use_rdma,
+                    const HealthCheckOption& hc_option) {
+    return get_or_new_client_side_socket_map()->Insert(key, id, ssl_ctx, use_rdma, hc_option);
 }    
 
 int SocketMapFind(const SocketMapKey& key, SocketId* id) {
@@ -225,7 +226,8 @@ void SocketMap::ShowSocketMapInBvarIfNeed() {
 
 int SocketMap::Insert(const SocketMapKey& key, SocketId* id,
                       const std::shared_ptr<SocketSSLContext>& ssl_ctx,
-                      bool use_rdma) {
+                      bool use_rdma,
+                      const HealthCheckOption& hc_option) {
     ShowSocketMapInBvarIfNeed();
 
     std::unique_lock<butil::Mutex> mu(_mutex);
@@ -249,6 +251,7 @@ int SocketMap::Insert(const SocketMapKey& key, SocketId* id,
     opt.remote_side = key.peer.addr;
     opt.initial_ssl_ctx = ssl_ctx;
     opt.use_rdma = use_rdma;
+    opt.hc_option = hc_option;
     if (_options.socket_creator->CreateSocket(opt, &tmp_id) != 0) {
         PLOG(FATAL) << "Fail to create socket to " << key.peer;
         return -1;