1 //
2 // Copyright 2022 gRPC authors.
3 //
4 // Licensed under the Apache License, Version 2.0 (the "License");
5 // you may not use this file except in compliance with the License.
6 // You may obtain a copy of the License at
7 //
8 // http://www.apache.org/licenses/LICENSE-2.0
9 //
10 // Unless required by applicable law or agreed to in writing, software
11 // distributed under the License is distributed on an "AS IS" BASIS,
12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 // See the License for the specific language governing permissions and
14 // limitations under the License.
15 //
16
17 #include "test/cpp/interop/istio_echo_server_lib.h"
18
19 #include <thread>
20
21 #include "absl/strings/str_cat.h"
22 #include "absl/strings/str_format.h"
23 #include "absl/strings/str_split.h"
24 #include "absl/synchronization/blocking_counter.h"
25
26 #include <grpcpp/client_context.h>
27 #include <grpcpp/grpcpp.h>
28
29 #include "src/core/lib/gprpp/host_port.h"
30 #include "src/proto/grpc/testing/istio_echo.pb.h"
31
32 using proto::EchoRequest;
33 using proto::EchoResponse;
34 using proto::EchoTestService;
35 using proto::ForwardEchoRequest;
36 using proto::ForwardEchoResponse;
37
38 namespace grpc {
39 namespace testing {
40 namespace {
41
42 const absl::string_view kRequestIdField = "x-request-id";
43 const absl::string_view kServiceVersionField = "ServiceVersion";
44 // const absl::string_view kServicePortField = "ServicePort";
45 const absl::string_view kStatusCodeField = "StatusCode";
46 // const absl::string_view kUrlField = "URL";
47 const absl::string_view kHostField = "Host";
48 const absl::string_view kHostnameField = "Hostname";
49 // const absl::string_view kMethodField = "Method";
50 const absl::string_view kRequestHeader = "RequestHeader";
51 // const absl::string_view kResponseHeader = "ResponseHeader";
52 // const absl::string_view kClusterField = "Cluster";
53 // const absl::string_view kIstioVersionField = "IstioVersion";
54 const absl::string_view kIpField = "IP"; // The Requester’s IP Address.
55
StringRefToStringView(const string_ref & r)56 absl::string_view StringRefToStringView(const string_ref& r) {
57 return absl::string_view(r.data(), r.size());
58 }
59
60 struct EchoCall {
61 grpc::ClientContext context;
62 proto::EchoResponse response;
63 Status status;
64 };
65
66 } // namespace
67
EchoTestServiceImpl(std::string hostname,std::string service_version,std::string forwarding_address)68 EchoTestServiceImpl::EchoTestServiceImpl(std::string hostname,
69 std::string service_version,
70 std::string forwarding_address)
71 : hostname_(std::move(hostname)),
72 service_version_(std::move(service_version)),
73 forwarding_address_(std::move(forwarding_address)) {
74 forwarding_stub_ = EchoTestService::NewStub(
75 CreateChannel(forwarding_address_, InsecureChannelCredentials()));
76 }
77
Echo(ServerContext * context,const EchoRequest * request,EchoResponse * response)78 Status EchoTestServiceImpl::Echo(ServerContext* context,
79 const EchoRequest* request,
80 EchoResponse* response) {
81 std::string s;
82 absl::StrAppend(&s, kHostField, "=",
83 StringRefToStringView(context->ExperimentalGetAuthority()),
84 "\n");
85 const std::multimap<string_ref, string_ref> metadata =
86 context->client_metadata();
87 for (const auto& kv : metadata) {
88 // Skip all binary headers.
89 if (kv.first.ends_with("-bin")) {
90 continue;
91 }
92 absl::StrAppend(&s, kRequestHeader, "=", StringRefToStringView(kv.first),
93 ":", StringRefToStringView(kv.second), "\n");
94 }
95 absl::string_view host;
96 absl::string_view port;
97 std::string peer = context->peer();
98 grpc_core::SplitHostPort(peer, &host, &port);
99 // This is not a complete list, but also not all fields are used. May
100 // need to add/remove fields later, if required by tests. Only keep the
101 // fields needed for now.
102 //
103 // absl::StrAppend(&s,kServicePortField,"=",this->port_,"\n");
104 // absl::StrAppend(&s,kClusterField,"=",this->cluster_,"\n");
105 // absl::StrAppend(&s,kIstioVersionField,"=",this->istio_version_,"\n");
106 absl::StrAppend(&s, kServiceVersionField, "=", this->service_version_, "\n");
107 absl::StrAppend(&s, kIpField, "=", host, "\n");
108 absl::StrAppend(&s, kStatusCodeField, "=", std::to_string(200), "\n");
109 absl::StrAppend(&s, kHostnameField, "=", this->hostname_, "\n");
110 absl::StrAppend(&s, "Echo=", request->message(), "\n");
111 response->set_message(s);
112 gpr_log(GPR_INFO, "Echo response:\n%s", s.c_str());
113 return Status::OK;
114 }
115
ForwardEcho(ServerContext * context,const ForwardEchoRequest * request,ForwardEchoResponse * response)116 Status EchoTestServiceImpl::ForwardEcho(ServerContext* context,
117 const ForwardEchoRequest* request,
118 ForwardEchoResponse* response) {
119 std::string raw_url = request->url();
120 size_t colon = raw_url.find_first_of(':');
121 std::string scheme;
122 if (colon == std::string::npos) {
123 return Status(
124 StatusCode::INVALID_ARGUMENT,
125 absl::StrFormat("No protocol configured for url %s", raw_url));
126 }
127 scheme = raw_url.substr(0, colon);
128 std::shared_ptr<Channel> channel;
129 if (scheme == "xds") {
130 // We can optionally add support for TLS creds, but we are primarily
131 // concerned with proxyless-grpc here.
132 gpr_log(GPR_INFO, "Creating channel to %s using xDS Creds",
133 raw_url.c_str());
134 channel =
135 CreateChannel(raw_url, XdsCredentials(InsecureChannelCredentials()));
136 } else if (scheme == "grpc") {
137 // We don't really want to test this but the istio test infrastructure needs
138 // this to be supported. If we ever decide to add support for this properly,
139 // we would need to add support for TLS creds here.
140 absl::string_view address = absl::StripPrefix(raw_url, "grpc://");
141 gpr_log(GPR_INFO, "Creating channel to %s", std::string(address).c_str());
142 channel = CreateChannel(std::string(address), InsecureChannelCredentials());
143 } else {
144 gpr_log(GPR_INFO, "Protocol %s not supported. Forwarding to %s",
145 scheme.c_str(), forwarding_address_.c_str());
146 ClientContext forwarding_ctx;
147 forwarding_ctx.set_deadline(context->deadline());
148 return forwarding_stub_->ForwardEcho(&forwarding_ctx, *request, response);
149 }
150 auto stub = EchoTestService::NewStub(channel);
151 auto count = request->count() == 0 ? 1 : request->count();
152 // Calculate the amount of time to sleep after each call.
153 std::chrono::duration<double> duration_per_query =
154 std::chrono::nanoseconds::zero();
155 if (request->qps() > 0) {
156 duration_per_query =
157 std::chrono::nanoseconds(std::chrono::seconds(1)) / request->qps();
158 }
159 std::vector<EchoCall> calls(count);
160 EchoRequest echo_request;
161 echo_request.set_message(request->message());
162 absl::BlockingCounter counter(count);
163 for (int i = 0; i < count; ++i) {
164 calls[i].context.AddMetadata(std::string(kRequestIdField),
165 std::to_string(i));
166 for (const auto& header : request->headers()) {
167 if (header.key() != kHostField) {
168 calls[i].context.AddMetadata(header.key(), header.value());
169 }
170 }
171 constexpr int kDefaultTimeout = 5 * 1000 * 1000 /* 5 seconds */;
172 std::chrono::system_clock::time_point deadline =
173 std::chrono::system_clock::now() +
174 std::chrono::microseconds(request->timeout_micros() > 0
175 ? request->timeout_micros()
176 : kDefaultTimeout);
177 calls[i].context.set_deadline(deadline);
178 stub->async()->Echo(&calls[i].context, &echo_request, &calls[i].response,
179 [&, index = i](Status s) {
180 calls[index].status = s;
181 counter.DecrementCount();
182 });
183 std::this_thread::sleep_for(duration_per_query);
184 }
185 // Wait for all calls to be done.
186 counter.Wait();
187 for (int i = 0; i < count; ++i) {
188 if (calls[i].status.ok()) {
189 std::string body;
190 // The test infrastructure might expect the entire struct instead of
191 // just the message.
192 absl::StrAppend(&body, absl::StrFormat("[%d] grpcecho.Echo(%s)\n", i,
193 request->message()));
194 auto contents =
195 absl::StrSplit(calls[i].response.message(), '\n', absl::SkipEmpty());
196 for (const auto& line : contents) {
197 absl::StrAppend(&body, absl::StrFormat("[%d body] %s\n", i, line));
198 }
199 response->add_output(body);
200 gpr_log(GPR_INFO, "Forward Echo response:%d\n%s", i, body.c_str());
201 } else {
202 gpr_log(GPR_ERROR, "RPC %d failed %d: %s", i,
203 calls[i].status.error_code(),
204 calls[i].status.error_message().c_str());
205 response->clear_output();
206 return calls[i].status;
207 }
208 }
209 return Status::OK;
210 }
211
212 } // namespace testing
213 } // namespace grpc
214