1 //
2 //
3 // Copyright 2024 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 // Explicitly define HAVE_ABSEIL to avoid conflict with OTel's Abseil
20 // version. Refer
21 // https://github.com/open-telemetry/opentelemetry-cpp/issues/1042.
22 #ifndef HAVE_ABSEIL
23 #define HAVE_ABSEIL
24 #endif
25
26 #include <grpcpp/ext/proto_server_reflection_plugin.h>
27 #include <grpcpp/grpcpp.h>
28 #include <grpcpp/health_check_service_interface.h>
29 #include <grpcpp/xds_server_builder.h>
30
31 #include <condition_variable>
32 #include <mutex>
33
34 #include "opentelemetry/sdk/metrics/view/instrument_selector_factory.h"
35 #include "opentelemetry/sdk/metrics/view/meter_selector_factory.h"
36 #include "opentelemetry/sdk/metrics/view/view_factory.h"
37
38 #ifdef BAZEL_BUILD
39 #include "examples/cpp/otel/util.h"
40 #include "examples/protos/helloworld.grpc.pb.h"
41 #else
42 #include "helloworld.grpc.pb.h"
43 #include "util.h"
44 #endif
45
46 using grpc::CallbackServerContext;
47 using grpc::Channel;
48 using grpc::ClientContext;
49 using grpc::Server;
50 using grpc::ServerBuilder;
51 using grpc::ServerUnaryReactor;
52 using grpc::Status;
53 using helloworld::Greeter;
54 using helloworld::HelloReply;
55 using helloworld::HelloRequest;
56
AddLatencyView(opentelemetry::sdk::metrics::MeterProvider * provider,const std::string & name,const std::string & unit)57 void AddLatencyView(opentelemetry::sdk::metrics::MeterProvider* provider,
58 const std::string& name, const std::string& unit) {
59 auto histogram_config = std::make_shared<
60 opentelemetry::sdk::metrics::HistogramAggregationConfig>();
61 histogram_config->boundaries_ = {
62 0, 0.00001, 0.00005, 0.0001, 0.0003, 0.0006, 0.0008, 0.001, 0.002,
63 0.003, 0.004, 0.005, 0.006, 0.008, 0.01, 0.013, 0.016, 0.02,
64 0.025, 0.03, 0.04, 0.05, 0.065, 0.08, 0.1, 0.13, 0.16,
65 0.2, 0.25, 0.3, 0.4, 0.5, 0.65, 0.8, 1, 2,
66 5, 10, 20, 50, 100};
67 provider->AddView(
68 opentelemetry::sdk::metrics::InstrumentSelectorFactory::Create(
69 opentelemetry::sdk::metrics::InstrumentType::kHistogram, name, unit),
70 opentelemetry::sdk::metrics::MeterSelectorFactory::Create(
71 "grpc-c++", grpc::Version(), ""),
72 opentelemetry::sdk::metrics::ViewFactory::Create(
73 name, "", unit,
74 opentelemetry::sdk::metrics::AggregationType::kHistogram,
75 std::move(histogram_config)));
76 }
77
78 namespace {
79
80 class GreeterClient {
81 public:
GreeterClient(std::shared_ptr<Channel> channel)82 GreeterClient(std::shared_ptr<Channel> channel)
83 : stub_(Greeter::NewStub(channel)) {}
84
85 // Assembles the client's payload, sends it and presents the response back
86 // from the server.
SayHello(const std::string & user)87 std::string SayHello(const std::string& user) {
88 // Data we are sending to the server.
89 HelloRequest request;
90 request.set_name(user);
91
92 // Container for the data we expect from the server.
93 HelloReply reply;
94
95 // Context for the client. It could be used to convey extra information to
96 // the server and/or tweak certain RPC behaviors.
97 ClientContext context;
98
99 // The actual RPC.
100 std::mutex mu;
101 std::condition_variable cv;
102 bool done = false;
103 Status status;
104 stub_->async()->SayHello(&context, &request, &reply,
105 [&mu, &cv, &done, &status](Status s) {
106 status = std::move(s);
107 std::lock_guard<std::mutex> lock(mu);
108 done = true;
109 cv.notify_one();
110 });
111
112 std::unique_lock<std::mutex> lock(mu);
113 while (!done) {
114 cv.wait(lock);
115 }
116
117 // Act upon its status.
118 if (status.ok()) {
119 return reply.message();
120 } else {
121 std::cout << status.error_code() << ": " << status.error_message()
122 << std::endl;
123 return "RPC failed";
124 }
125 }
126
127 private:
128 std::unique_ptr<Greeter::Stub> stub_;
129 };
130
131 // Logic and data behind the server's behavior.
132 class GreeterServiceImpl final : public Greeter::CallbackService {
SayHello(CallbackServerContext * context,const HelloRequest * request,HelloReply * reply)133 ServerUnaryReactor* SayHello(CallbackServerContext* context,
134 const HelloRequest* request,
135 HelloReply* reply) override {
136 std::string prefix("Hello ");
137 reply->set_message(prefix + request->name());
138
139 ServerUnaryReactor* reactor = context->DefaultReactor();
140 reactor->Finish(Status::OK);
141 return reactor;
142 }
143 };
144
145 } // namespace
146
RunServer(uint16_t port)147 void RunServer(uint16_t port) {
148 std::string server_address = absl::StrFormat("0.0.0.0:%d", port);
149 GreeterServiceImpl service;
150
151 grpc::EnableDefaultHealthCheckService(true);
152 grpc::reflection::InitProtoReflectionServerBuilderPlugin();
153 ServerBuilder builder;
154 // Listen on the given address without any authentication mechanism.
155 builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
156 // Register "service" as the instance through which we'll communicate with
157 // clients. In this case it corresponds to an *synchronous* service.
158 builder.RegisterService(&service);
159 // Finally assemble the server.
160 std::unique_ptr<Server> server(builder.BuildAndStart());
161 std::cout << "Server listening on " << server_address << std::endl;
162
163 // Wait for the server to shutdown. Note that some other thread must be
164 // responsible for shutting down the server for this call to ever return.
165 server->Wait();
166 }
167
RunXdsEnabledServer(uint16_t port)168 void RunXdsEnabledServer(uint16_t port) {
169 std::string server_address = absl::StrFormat("0.0.0.0:%d", port);
170 GreeterServiceImpl service;
171
172 grpc::EnableDefaultHealthCheckService(true);
173 grpc::reflection::InitProtoReflectionServerBuilderPlugin();
174 grpc::XdsServerBuilder builder;
175 // Listen on the given address without any authentication mechanism.
176 builder.AddListeningPort(
177 server_address,
178 grpc::XdsServerCredentials(grpc::InsecureServerCredentials()));
179 // Register "service" as the instance through which we'll communicate with
180 // clients. In this case it corresponds to an *synchronous* service.
181 builder.RegisterService(&service);
182 // Finally assemble the server.
183 std::unique_ptr<Server> server(builder.BuildAndStart());
184 std::cout << "Server listening on " << server_address << std::endl;
185
186 // Wait for the server to shutdown. Note that some other thread must be
187 // responsible for shutting down the server for this call to ever return.
188 server->Wait();
189 }
190
RunClient(const std::string & target_str)191 void RunClient(const std::string& target_str) {
192 // Instantiate the client. It requires a channel, out of which the actual RPCs
193 // are created. This channel models a connection to an endpoint specified by
194 // the argument "--target=" which is the only expected argument.
195 grpc::ChannelArguments args;
196 GreeterClient greeter(grpc::CreateCustomChannel(
197 target_str, grpc::XdsCredentials(grpc::InsecureChannelCredentials()),
198 args));
199 // Continuously send RPCs every second.
200 while (true) {
201 std::string user("world");
202 std::string reply = greeter.SayHello(user);
203 std::cout << "Greeter received: " << reply << std::endl;
204 std::this_thread::sleep_for(std::chrono::seconds(1));
205 }
206 }
207