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