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 <grpcpp/client_context.h>
20 #include <grpcpp/grpcpp.h>
21
22 #include <thread>
23
24 #include "absl/log/log.h"
25 #include "absl/strings/str_cat.h"
26 #include "absl/strings/str_format.h"
27 #include "absl/strings/str_split.h"
28 #include "absl/synchronization/blocking_counter.h"
29 #include "src/core/util/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 LOG(INFO) << "Echo response:\n" << s;
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 LOG(INFO) << "Creating channel to " << raw_url << " using xDS Creds";
133 channel =
134 CreateChannel(raw_url, XdsCredentials(InsecureChannelCredentials()));
135 } else if (scheme == "grpc") {
136 // We don't really want to test this but the istio test infrastructure needs
137 // this to be supported. If we ever decide to add support for this properly,
138 // we would need to add support for TLS creds here.
139 absl::string_view address = absl::StripPrefix(raw_url, "grpc://");
140 LOG(INFO) << "Creating channel to " << address;
141 channel = CreateChannel(std::string(address), InsecureChannelCredentials());
142 } else {
143 LOG(INFO) << "Protocol " << scheme << " not supported. Forwarding to "
144 << forwarding_address_;
145 ClientContext forwarding_ctx;
146 forwarding_ctx.set_deadline(context->deadline());
147 return forwarding_stub_->ForwardEcho(&forwarding_ctx, *request, response);
148 }
149 auto stub = EchoTestService::NewStub(channel);
150 auto count = request->count() == 0 ? 1 : request->count();
151 // Calculate the amount of time to sleep after each call.
152 std::chrono::duration<double> duration_per_query =
153 std::chrono::nanoseconds::zero();
154 if (request->qps() > 0) {
155 duration_per_query =
156 std::chrono::nanoseconds(std::chrono::seconds(1)) / request->qps();
157 }
158 std::vector<EchoCall> calls(count);
159 EchoRequest echo_request;
160 echo_request.set_message(request->message());
161 absl::BlockingCounter counter(count);
162 for (int i = 0; i < count; ++i) {
163 calls[i].context.AddMetadata(std::string(kRequestIdField),
164 std::to_string(i));
165 for (const auto& header : request->headers()) {
166 if (header.key() != kHostField) {
167 calls[i].context.AddMetadata(header.key(), header.value());
168 }
169 }
170 constexpr int kDefaultTimeout = 5 * 1000 * 1000 /* 5 seconds */;
171 std::chrono::system_clock::time_point deadline =
172 std::chrono::system_clock::now() +
173 std::chrono::microseconds(request->timeout_micros() > 0
174 ? request->timeout_micros()
175 : kDefaultTimeout);
176 calls[i].context.set_deadline(deadline);
177 stub->async()->Echo(&calls[i].context, &echo_request, &calls[i].response,
178 [&, index = i](Status s) {
179 calls[index].status = s;
180 counter.DecrementCount();
181 });
182 std::this_thread::sleep_for(duration_per_query);
183 }
184 // Wait for all calls to be done.
185 counter.Wait();
186 for (int i = 0; i < count; ++i) {
187 if (calls[i].status.ok()) {
188 std::string body;
189 // The test infrastructure might expect the entire struct instead of
190 // just the message.
191 absl::StrAppend(&body, absl::StrFormat("[%d] grpcecho.Echo(%s)\n", i,
192 request->message()));
193 auto contents =
194 absl::StrSplit(calls[i].response.message(), '\n', absl::SkipEmpty());
195 for (const auto& line : contents) {
196 absl::StrAppend(&body, absl::StrFormat("[%d body] %s\n", i, line));
197 }
198 response->add_output(body);
199 LOG(INFO) << "Forward Echo response:" << i << "\n" << body;
200 } else {
201 LOG(ERROR) << "RPC " << i << " failed " << calls[i].status.error_code()
202 << ": " << calls[i].status.error_message();
203 response->clear_output();
204 return calls[i].status;
205 }
206 }
207 return Status::OK;
208 }
209
210 } // namespace testing
211 } // namespace grpc
212