1 /*
2 *
3 * Copyright 2023 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 <sys/types.h>
20
21 #include <chrono>
22 #include <condition_variable>
23 #include <iostream>
24 #include <memory>
25 #include <optional>
26 #include <string>
27
28 #include "absl/flags/flag.h"
29 #include "absl/flags/parse.h"
30 #include "absl/strings/str_join.h"
31 #include "absl/strings/str_split.h"
32 #include "absl/types/optional.h"
33 #include "opentelemetry/exporters/prometheus/exporter_factory.h"
34 #include "opentelemetry/exporters/prometheus/exporter_options.h"
35 #include "opentelemetry/sdk/metrics/meter_provider.h"
36
37 #include <grpcpp/ext/csm_observability.h>
38 #include <grpcpp/grpcpp.h>
39 #include <grpcpp/support/string_ref.h>
40
41 #ifdef BAZEL_BUILD
42 #include "examples/protos/helloworld.grpc.pb.h"
43 #else
44 #include "helloworld.grpc.pb.h"
45 #endif
46
47 ABSL_FLAG(std::string, target, "xds:///helloworld:50051", "Target string");
48 ABSL_FLAG(std::string, cookie_name, "GSSA", "Cookie name");
49 ABSL_FLAG(uint, delay_s, 5, "Delay between requests");
50
51 using grpc::Channel;
52 using grpc::ClientContext;
53 using grpc::Status;
54 using helloworld::Greeter;
55 using helloworld::HelloReply;
56 using helloworld::HelloRequest;
57
58 namespace {
59
60 struct Cookie {
61 std::string name;
62 std::string value;
63 std::set<std::string> attributes;
64
65 template <typename Sink>
AbslStringify(Sink & sink,const Cookie & cookie)66 friend void AbslStringify(Sink& sink, const Cookie& cookie) {
67 absl::Format(&sink, "(Cookie: %s, value: %s, attributes: {%s})",
68 cookie.name, cookie.value,
69 absl::StrJoin(cookie.attributes, ", "));
70 }
71 };
72
ParseCookie(absl::string_view header)73 Cookie ParseCookie(absl::string_view header) {
74 Cookie cookie;
75 std::pair<absl::string_view, absl::string_view> name_value =
76 absl::StrSplit(header, absl::MaxSplits('=', 1));
77 cookie.name = std::string(name_value.first);
78 std::pair<absl::string_view, absl::string_view> value_attrs =
79 absl::StrSplit(name_value.second, absl::MaxSplits(';', 1));
80 cookie.value = std::string(value_attrs.first);
81 for (absl::string_view segment : absl::StrSplit(value_attrs.second, ';')) {
82 cookie.attributes.emplace(absl::StripAsciiWhitespace(segment));
83 }
84 return cookie;
85 }
86
GetCookies(const std::multimap<grpc::string_ref,grpc::string_ref> & initial_metadata,absl::string_view cookie_name)87 std::vector<Cookie> GetCookies(
88 const std::multimap<grpc::string_ref, grpc::string_ref>& initial_metadata,
89 absl::string_view cookie_name) {
90 std::vector<Cookie> values;
91 auto pair = initial_metadata.equal_range("set-cookie");
92 for (auto it = pair.first; it != pair.second; ++it) {
93 const auto cookie = ParseCookie(it->second.data());
94 if (cookie.name == cookie_name) {
95 values.emplace_back(std::move(cookie));
96 }
97 }
98 return values;
99 }
100
101 class GreeterClient {
102 public:
GreeterClient(std::shared_ptr<Channel> channel,absl::string_view cookie_name)103 GreeterClient(std::shared_ptr<Channel> channel, absl::string_view cookie_name)
104 : stub_(Greeter::NewStub(channel)), cookie_name_(cookie_name) {}
105
106 // Assembles the client's payload, sends it and presents the response back
107 // from the server.
SayHello()108 void SayHello() {
109 // Data we are sending to the server.
110 HelloRequest request;
111 request.set_name("world");
112
113 // Container for the data we expect from the server.
114 HelloReply reply;
115
116 // Context for the client. It could be used to convey extra information to
117 // the server and/or tweak certain RPC behaviors.
118 ClientContext context;
119
120 // The actual RPC.
121 std::mutex mu;
122 std::condition_variable cv;
123 absl::optional<Status> status;
124 // Set the cookie header if we already got a cookie from the server
125 if (cookie_from_server_.has_value()) {
126 context.AddMetadata("cookie",
127 absl::StrFormat("%s=%s", cookie_from_server_->name,
128 cookie_from_server_->value));
129 }
130 std::unique_lock<std::mutex> lock(mu);
131 stub_->async()->SayHello(&context, &request, &reply, [&](Status s) {
132 std::lock_guard<std::mutex> lock(mu);
133 status = std::move(s);
134 cv.notify_one();
135 });
136 while (!status.has_value()) {
137 cv.wait(lock);
138 }
139 if (!status->ok()) {
140 std::cout << "RPC failed" << status->error_code() << ": "
141 << status->error_message() << std::endl;
142 return;
143 }
144 const std::multimap<grpc::string_ref, grpc::string_ref>&
145 server_initial_metadata = context.GetServerInitialMetadata();
146 // Update a cookie after a successful request
147 std::vector<Cookie> cookies =
148 GetCookies(server_initial_metadata, cookie_name_);
149 if (!cookies.empty()) {
150 cookie_from_server_.emplace(std::move(cookies.front()));
151 }
152 std::cout << "Greeter received: " << reply.message() << std::endl;
153 }
154
155 private:
156 std::unique_ptr<Greeter::Stub> stub_;
157 std::string cookie_name_;
158 absl::optional<Cookie> cookie_from_server_;
159 };
160
InitializeObservability()161 absl::StatusOr<grpc::CsmObservability> InitializeObservability() {
162 opentelemetry::exporter::metrics::PrometheusExporterOptions opts;
163 // default was "localhost:9464" which causes connection issue across GKE pods
164 opts.url = "0.0.0.0:9464";
165 auto prometheus_exporter =
166 opentelemetry::exporter::metrics::PrometheusExporterFactory::Create(opts);
167 auto meter_provider =
168 std::make_shared<opentelemetry::sdk::metrics::MeterProvider>();
169 meter_provider->AddMetricReader(std::move(prometheus_exporter));
170 return grpc::CsmObservabilityBuilder()
171 .SetMeterProvider(std::move(meter_provider))
172 .BuildAndRegister();
173 }
174
175 } // namespace
176
main(int argc,char ** argv)177 int main(int argc, char** argv) {
178 absl::ParseCommandLine(argc, argv);
179 // Setup the observability
180 auto observability = InitializeObservability();
181 if (!observability.ok()) {
182 std::cerr << "CsmObservability::Init() failed: "
183 << observability.status().ToString() << std::endl;
184 return static_cast<int>(observability.status().code());
185 }
186 GreeterClient greeter(grpc::CreateChannel(absl::GetFlag(FLAGS_target),
187 grpc::InsecureChannelCredentials()),
188 absl::GetFlag(FLAGS_cookie_name));
189 while (true) {
190 greeter.SayHello();
191 std::this_thread::sleep_for(
192 std::chrono::seconds(absl::GetFlag(FLAGS_delay_s)));
193 }
194 return 0;
195 }
196