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 "src/core/load_balancing/grpclb/load_balancer_api.h"
20
21 #include <grpc/support/port_platform.h>
22 #include <grpc/support/time.h>
23 #include <string.h>
24
25 #include <algorithm>
26
27 #include "absl/log/log.h"
28 #include "google/protobuf/duration.upb.h"
29 #include "google/protobuf/timestamp.upb.h"
30 #include "src/core/util/memory.h"
31 #include "src/proto/grpc/lb/v1/load_balancer.upb.h"
32 #include "upb/base/string_view.h"
33
34 namespace grpc_core {
35
operator ==(const GrpcLbServer & other) const36 bool GrpcLbServer::operator==(const GrpcLbServer& other) const {
37 if (ip_size != other.ip_size) return false;
38 int r = memcmp(ip_addr, other.ip_addr, ip_size);
39 if (r != 0) return false;
40 if (port != other.port) return false;
41 r = strncmp(load_balance_token, other.load_balance_token,
42 sizeof(load_balance_token));
43 if (r != 0) return false;
44 return drop == other.drop;
45 }
46
47 namespace {
48
grpc_grpclb_request_encode(const grpc_lb_v1_LoadBalanceRequest * request,upb_Arena * arena)49 grpc_slice grpc_grpclb_request_encode(
50 const grpc_lb_v1_LoadBalanceRequest* request, upb_Arena* arena) {
51 size_t buf_length;
52 char* buf =
53 grpc_lb_v1_LoadBalanceRequest_serialize(request, arena, &buf_length);
54 return grpc_slice_from_copied_buffer(buf, buf_length);
55 }
56
57 } // namespace
58
GrpcLbRequestCreate(absl::string_view lb_service_name,upb_Arena * arena)59 grpc_slice GrpcLbRequestCreate(absl::string_view lb_service_name,
60 upb_Arena* arena) {
61 grpc_lb_v1_LoadBalanceRequest* req = grpc_lb_v1_LoadBalanceRequest_new(arena);
62 grpc_lb_v1_InitialLoadBalanceRequest* initial_request =
63 grpc_lb_v1_LoadBalanceRequest_mutable_initial_request(req, arena);
64 size_t name_len = std::min(lb_service_name.size(),
65 size_t{GRPC_GRPCLB_SERVICE_NAME_MAX_LENGTH});
66 grpc_lb_v1_InitialLoadBalanceRequest_set_name(
67 initial_request,
68 upb_StringView_FromDataAndSize(lb_service_name.data(), name_len));
69 return grpc_grpclb_request_encode(req, arena);
70 }
71
72 namespace {
73
google_protobuf_Timestamp_assign(google_protobuf_Timestamp * timestamp,const gpr_timespec & value)74 void google_protobuf_Timestamp_assign(google_protobuf_Timestamp* timestamp,
75 const gpr_timespec& value) {
76 google_protobuf_Timestamp_set_seconds(timestamp, value.tv_sec);
77 google_protobuf_Timestamp_set_nanos(timestamp, value.tv_nsec);
78 }
79
80 } // namespace
81
GrpcLbLoadReportRequestCreate(int64_t num_calls_started,int64_t num_calls_finished,int64_t num_calls_finished_with_client_failed_to_send,int64_t num_calls_finished_known_received,const GrpcLbClientStats::DroppedCallCounts * drop_token_counts,upb_Arena * arena)82 grpc_slice GrpcLbLoadReportRequestCreate(
83 int64_t num_calls_started, int64_t num_calls_finished,
84 int64_t num_calls_finished_with_client_failed_to_send,
85 int64_t num_calls_finished_known_received,
86 const GrpcLbClientStats::DroppedCallCounts* drop_token_counts,
87 upb_Arena* arena) {
88 grpc_lb_v1_LoadBalanceRequest* req = grpc_lb_v1_LoadBalanceRequest_new(arena);
89 grpc_lb_v1_ClientStats* req_stats =
90 grpc_lb_v1_LoadBalanceRequest_mutable_client_stats(req, arena);
91 google_protobuf_Timestamp_assign(
92 grpc_lb_v1_ClientStats_mutable_timestamp(req_stats, arena),
93 gpr_now(GPR_CLOCK_REALTIME));
94 grpc_lb_v1_ClientStats_set_num_calls_started(req_stats, num_calls_started);
95 grpc_lb_v1_ClientStats_set_num_calls_finished(req_stats, num_calls_finished);
96 grpc_lb_v1_ClientStats_set_num_calls_finished_with_client_failed_to_send(
97 req_stats, num_calls_finished_with_client_failed_to_send);
98 grpc_lb_v1_ClientStats_set_num_calls_finished_known_received(
99 req_stats, num_calls_finished_known_received);
100 if (drop_token_counts != nullptr) {
101 for (size_t i = 0; i < drop_token_counts->size(); ++i) {
102 const GrpcLbClientStats::DropTokenCount& cur = (*drop_token_counts)[i];
103 grpc_lb_v1_ClientStatsPerToken* cur_msg =
104 grpc_lb_v1_ClientStats_add_calls_finished_with_drop(req_stats, arena);
105 const size_t token_len = strlen(cur.token.get());
106 char* token = reinterpret_cast<char*>(upb_Arena_Malloc(arena, token_len));
107 memcpy(token, cur.token.get(), token_len);
108 grpc_lb_v1_ClientStatsPerToken_set_load_balance_token(
109 cur_msg, upb_StringView_FromDataAndSize(token, token_len));
110 grpc_lb_v1_ClientStatsPerToken_set_num_calls(cur_msg, cur.count);
111 }
112 }
113 return grpc_grpclb_request_encode(req, arena);
114 }
115
116 namespace {
117
ParseServerList(const grpc_lb_v1_LoadBalanceResponse & response,std::vector<GrpcLbServer> * server_list)118 bool ParseServerList(const grpc_lb_v1_LoadBalanceResponse& response,
119 std::vector<GrpcLbServer>* server_list) {
120 // Determine the number of servers.
121 const grpc_lb_v1_ServerList* server_list_msg =
122 grpc_lb_v1_LoadBalanceResponse_server_list(&response);
123 if (server_list_msg == nullptr) return false;
124 size_t server_count = 0;
125 const grpc_lb_v1_Server* const* servers =
126 grpc_lb_v1_ServerList_servers(server_list_msg, &server_count);
127 // Populate servers.
128 if (server_count > 0) {
129 server_list->reserve(server_count);
130 for (size_t i = 0; i < server_count; ++i) {
131 GrpcLbServer& cur = *server_list->emplace(server_list->end());
132 upb_StringView address = grpc_lb_v1_Server_ip_address(servers[i]);
133 if (address.size == 0) {
134 ; // Nothing to do because cur->ip_address is an empty string.
135 } else if (address.size <= GRPC_GRPCLB_SERVER_IP_ADDRESS_MAX_SIZE) {
136 cur.ip_size = static_cast<int32_t>(address.size);
137 memcpy(cur.ip_addr, address.data, address.size);
138 }
139 cur.port = grpc_lb_v1_Server_port(servers[i]);
140 upb_StringView token = grpc_lb_v1_Server_load_balance_token(servers[i]);
141 if (token.size == 0) {
142 ; // Nothing to do because cur->load_balance_token is an empty string.
143 } else if (token.size <= GRPC_GRPCLB_SERVER_LOAD_BALANCE_TOKEN_MAX_SIZE) {
144 memcpy(cur.load_balance_token, token.data, token.size);
145 } else {
146 LOG(ERROR) << "grpc_lb_v1_LoadBalanceResponse has too long token. len="
147 << token.size;
148 }
149 cur.drop = grpc_lb_v1_Server_drop(servers[i]);
150 }
151 }
152 return true;
153 }
154
ParseDuration(const google_protobuf_Duration * duration_pb)155 Duration ParseDuration(const google_protobuf_Duration* duration_pb) {
156 return Duration::FromSecondsAndNanoseconds(
157 google_protobuf_Duration_seconds(duration_pb),
158 google_protobuf_Duration_nanos(duration_pb));
159 }
160
161 } // namespace
162
GrpcLbResponseParse(const grpc_slice & serialized_response,upb_Arena * arena,GrpcLbResponse * result)163 bool GrpcLbResponseParse(const grpc_slice& serialized_response,
164 upb_Arena* arena, GrpcLbResponse* result) {
165 grpc_lb_v1_LoadBalanceResponse* response =
166 grpc_lb_v1_LoadBalanceResponse_parse(
167 reinterpret_cast<const char*>(
168 GRPC_SLICE_START_PTR(serialized_response)),
169 GRPC_SLICE_LENGTH(serialized_response), arena);
170 // Handle serverlist responses.
171 if (ParseServerList(*response, &result->serverlist)) {
172 result->type = result->SERVERLIST;
173 return true;
174 }
175 // Handle initial responses.
176 auto* initial_response =
177 grpc_lb_v1_LoadBalanceResponse_initial_response(response);
178 if (initial_response != nullptr) {
179 result->type = result->INITIAL;
180 const google_protobuf_Duration* client_stats_report_interval =
181 grpc_lb_v1_InitialLoadBalanceResponse_client_stats_report_interval(
182 initial_response);
183 if (client_stats_report_interval != nullptr) {
184 result->client_stats_report_interval =
185 ParseDuration(client_stats_report_interval);
186 }
187 return true;
188 }
189 // Handle fallback.
190 if (grpc_lb_v1_LoadBalanceResponse_has_fallback_response(response)) {
191 result->type = result->FALLBACK;
192 return true;
193 }
194 // Unknown response type.
195 return false;
196 }
197
198 } // namespace grpc_core
199