src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.cc

   1 /*
   2  *
   3  * Copyright 2016 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <grpc/support/port_platform.h>
  20
  21 #include "src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h"
  22 #include "src/core/lib/gpr/useful.h"
  23
  24 #include "google/protobuf/duration.upb.h"
  25 #include "google/protobuf/timestamp.upb.h"
  26
  27 #include <grpc/support/alloc.h>
  28
  29 namespace grpc_core {
  30
  31 bool GrpcLbServer::operator==(const GrpcLbServer& other) const {
  32   if (ip_size != other.ip_size) return false;
  33   int r = memcmp(ip_addr, other.ip_addr, ip_size);
  34   if (r != 0) return false;
  35   if (port != other.port) return false;
  36   r = strncmp(load_balance_token, other.load_balance_token,
  37               sizeof(load_balance_token));
  38   if (r != 0) return false;
  39   return drop == other.drop;
  40 }
  41
  42 namespace {
  43
  44 grpc_slice grpc_grpclb_request_encode(
  45     const grpc_lb_v1_LoadBalanceRequest* request, upb_arena* arena) {
  46   size_t buf_length;
  47   char* buf =
  48       grpc_lb_v1_LoadBalanceRequest_serialize(request, arena, &buf_length);
  49   return grpc_slice_from_copied_buffer(buf, buf_length);
  50 }
  51
  52 }  // namespace
  53
  54 grpc_slice GrpcLbRequestCreate(const char* lb_service_name, upb_arena* arena) {
  55   grpc_lb_v1_LoadBalanceRequest* req = grpc_lb_v1_LoadBalanceRequest_new(arena);
  56   grpc_lb_v1_InitialLoadBalanceRequest* initial_request =
  57       grpc_lb_v1_LoadBalanceRequest_mutable_initial_request(req, arena);
  58   size_t name_len =
  59       GPR_MIN(strlen(lb_service_name), GRPC_GRPCLB_SERVICE_NAME_MAX_LENGTH);
  60   grpc_lb_v1_InitialLoadBalanceRequest_set_name(
  61       initial_request, upb_strview_make(lb_service_name, name_len));
  62   return grpc_grpclb_request_encode(req, arena);
  63 }
  64
  65 namespace {
  66
  67 void google_protobuf_Timestamp_assign(google_protobuf_Timestamp* timestamp,
  68                                       const gpr_timespec& value) {
  69   google_protobuf_Timestamp_set_seconds(timestamp, value.tv_sec);
  70   google_protobuf_Timestamp_set_nanos(timestamp, value.tv_nsec);
  71 }
  72
  73 }  // namespace
  74
  75 grpc_slice GrpcLbLoadReportRequestCreate(
  76     int64_t num_calls_started, int64_t num_calls_finished,
  77     int64_t num_calls_finished_with_client_failed_to_send,
  78     int64_t num_calls_finished_known_received,
  79     const GrpcLbClientStats::DroppedCallCounts* drop_token_counts,
  80     upb_arena* arena) {
  81   grpc_lb_v1_LoadBalanceRequest* req = grpc_lb_v1_LoadBalanceRequest_new(arena);
  82   grpc_lb_v1_ClientStats* req_stats =
  83       grpc_lb_v1_LoadBalanceRequest_mutable_client_stats(req, arena);
  84   google_protobuf_Timestamp_assign(
  85       grpc_lb_v1_ClientStats_mutable_timestamp(req_stats, arena),
  86       gpr_now(GPR_CLOCK_REALTIME));
  87   grpc_lb_v1_ClientStats_set_num_calls_started(req_stats, num_calls_started);
  88   grpc_lb_v1_ClientStats_set_num_calls_finished(req_stats, num_calls_finished);
  89   grpc_lb_v1_ClientStats_set_num_calls_finished_with_client_failed_to_send(
  90       req_stats, num_calls_finished_with_client_failed_to_send);
  91   grpc_lb_v1_ClientStats_set_num_calls_finished_known_received(
  92       req_stats, num_calls_finished_known_received);
  93   if (drop_token_counts != nullptr) {
  94     for (size_t i = 0; i < drop_token_counts->size(); ++i) {
  95       const GrpcLbClientStats::DropTokenCount& cur = (*drop_token_counts)[i];
  96       grpc_lb_v1_ClientStatsPerToken* cur_msg =
  97           grpc_lb_v1_ClientStats_add_calls_finished_with_drop(req_stats, arena);
  98       const size_t token_len = strlen(cur.token.get());
  99       char* token = reinterpret_cast<char*>(upb_arena_malloc(arena, token_len));
 100       memcpy(token, cur.token.get(), token_len);
 101       grpc_lb_v1_ClientStatsPerToken_set_load_balance_token(
 102           cur_msg, upb_strview_make(token, token_len));
 103       grpc_lb_v1_ClientStatsPerToken_set_num_calls(cur_msg, cur.count);
 104     }
 105   }
 106   return grpc_grpclb_request_encode(req, arena);
 107 }
 108
 109 namespace {
 110
 111 bool ParseServerList(const grpc_lb_v1_LoadBalanceResponse& response,
 112                      std::vector<GrpcLbServer>* server_list) {
 113   // Determine the number of servers.
 114   const grpc_lb_v1_ServerList* server_list_msg =
 115       grpc_lb_v1_LoadBalanceResponse_server_list(&response);
 116   if (server_list_msg == nullptr) return false;
 117   size_t server_count = 0;
 118   const grpc_lb_v1_Server* const* servers =
 119       grpc_lb_v1_ServerList_servers(server_list_msg, &server_count);
 120   // Populate servers.
 121   if (server_count > 0) {
 122     server_list->reserve(server_count);
 123     for (size_t i = 0; i < server_count; ++i) {
 124       GrpcLbServer& cur = *server_list->emplace(server_list->end());
 125       upb_strview address = grpc_lb_v1_Server_ip_address(servers[i]);
 126       if (address.size == 0) {
 127         ;  // Nothing to do because cur->ip_address is an empty string.
 128       } else if (address.size <= GRPC_GRPCLB_SERVER_IP_ADDRESS_MAX_SIZE) {
 129         cur.ip_size = static_cast<int32_t>(address.size);
 130         memcpy(cur.ip_addr, address.data, address.size);
 131       }
 132       cur.port = grpc_lb_v1_Server_port(servers[i]);
 133       upb_strview token = grpc_lb_v1_Server_load_balance_token(servers[i]);
 134       if (token.size == 0) {
 135         ;  // Nothing to do because cur->load_balance_token is an empty string.
 136       } else if (token.size <= GRPC_GRPCLB_SERVER_LOAD_BALANCE_TOKEN_MAX_SIZE) {
 137         memcpy(cur.load_balance_token, token.data, token.size);
 138       } else {
 139         gpr_log(GPR_ERROR,
 140                 "grpc_lb_v1_LoadBalanceResponse has too long token. len=%zu",
 141                 token.size);
 142       }
 143       cur.drop = grpc_lb_v1_Server_drop(servers[i]);
 144     }
 145   }
 146   return true;
 147 }
 148
 149 grpc_millis grpc_grpclb_duration_to_millis(
 150     const google_protobuf_Duration* duration_pb) {
 151   return static_cast<grpc_millis>(
 152       (google_protobuf_Duration_seconds(duration_pb) * GPR_MS_PER_SEC) +
 153       (google_protobuf_Duration_nanos(duration_pb) / GPR_NS_PER_MS));
 154 }
 155
 156 }  // namespace
 157
 158 bool GrpcLbResponseParse(const grpc_slice& serialized_response,
 159                          upb_arena* arena, GrpcLbResponse* result) {
 160   grpc_lb_v1_LoadBalanceResponse* response =
 161       grpc_lb_v1_LoadBalanceResponse_parse(
 162           reinterpret_cast<const char*>(
 163               GRPC_SLICE_START_PTR(serialized_response)),
 164           GRPC_SLICE_LENGTH(serialized_response), arena);
 165   // Handle serverlist responses.
 166   if (ParseServerList(*response, &result->serverlist)) {
 167     result->type = result->SERVERLIST;
 168     return true;
 169   }
 170   // Handle initial responses.
 171   auto* initial_response =
 172       grpc_lb_v1_LoadBalanceResponse_initial_response(response);
 173   if (initial_response != nullptr) {
 174     result->type = result->INITIAL;
 175     const google_protobuf_Duration* client_stats_report_interval =
 176         grpc_lb_v1_InitialLoadBalanceResponse_client_stats_report_interval(
 177             initial_response);
 178     if (client_stats_report_interval != nullptr) {
 179       result->client_stats_report_interval =
 180           grpc_grpclb_duration_to_millis(client_stats_report_interval);
 181     }
 182     return true;
 183   }
 184   // Handle fallback.
 185   if (grpc_lb_v1_LoadBalanceResponse_has_fallback_response(response)) {
 186     result->type = result->FALLBACK;
 187     return true;
 188   }
 189   // Unknown response type.
 190   return false;
 191 }
 192
 193 }  // namespace grpc_core