// // // Copyright 2024 gRPC authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. // // // Explicitly define HAVE_ABSEIL to avoid conflict with OTel's Abseil // version. Refer // https://github.com/open-telemetry/opentelemetry-cpp/issues/1042. #ifndef HAVE_ABSEIL #define HAVE_ABSEIL #endif #include #include #include "opentelemetry/sdk/metrics/view/instrument_selector_factory.h" #include "opentelemetry/sdk/metrics/view/meter_selector_factory.h" #include "opentelemetry/sdk/metrics/view/view_factory.h" #include #include #include #ifdef BAZEL_BUILD #include "examples/cpp/otel/util.h" #include "examples/protos/helloworld.grpc.pb.h" #else #include "helloworld.grpc.pb.h" #include "util.h" #endif using grpc::CallbackServerContext; using grpc::Channel; using grpc::ClientContext; using grpc::Server; using grpc::ServerBuilder; using grpc::ServerUnaryReactor; using grpc::Status; using helloworld::Greeter; using helloworld::HelloReply; using helloworld::HelloRequest; void AddLatencyView(opentelemetry::sdk::metrics::MeterProvider* provider, const std::string& name, const std::string& unit) { auto histogram_config = std::make_shared< opentelemetry::sdk::metrics::HistogramAggregationConfig>(); histogram_config->boundaries_ = { 0, 0.00001, 0.00005, 0.0001, 0.0003, 0.0006, 0.0008, 0.001, 0.002, 0.003, 0.004, 0.005, 0.006, 0.008, 0.01, 0.013, 0.016, 0.02, 0.025, 0.03, 0.04, 0.05, 0.065, 0.08, 0.1, 0.13, 0.16, 0.2, 0.25, 0.3, 0.4, 0.5, 0.65, 0.8, 1, 2, 5, 10, 20, 50, 100}; provider->AddView( opentelemetry::sdk::metrics::InstrumentSelectorFactory::Create( opentelemetry::sdk::metrics::InstrumentType::kHistogram, name, unit), opentelemetry::sdk::metrics::MeterSelectorFactory::Create( "grpc-c++", grpc::Version(), ""), opentelemetry::sdk::metrics::ViewFactory::Create( name, "", unit, opentelemetry::sdk::metrics::AggregationType::kHistogram, std::move(histogram_config))); } namespace { class GreeterClient { public: GreeterClient(std::shared_ptr channel) : stub_(Greeter::NewStub(channel)) {} // Assembles the client's payload, sends it and presents the response back // from the server. std::string SayHello(const std::string& user) { // Data we are sending to the server. HelloRequest request; request.set_name(user); // Container for the data we expect from the server. HelloReply reply; // Context for the client. It could be used to convey extra information to // the server and/or tweak certain RPC behaviors. ClientContext context; // The actual RPC. std::mutex mu; std::condition_variable cv; bool done = false; Status status; stub_->async()->SayHello(&context, &request, &reply, [&mu, &cv, &done, &status](Status s) { status = std::move(s); std::lock_guard lock(mu); done = true; cv.notify_one(); }); std::unique_lock lock(mu); while (!done) { cv.wait(lock); } // Act upon its status. if (status.ok()) { return reply.message(); } else { std::cout << status.error_code() << ": " << status.error_message() << std::endl; return "RPC failed"; } } private: std::unique_ptr stub_; }; // Logic and data behind the server's behavior. class GreeterServiceImpl final : public Greeter::CallbackService { ServerUnaryReactor* SayHello(CallbackServerContext* context, const HelloRequest* request, HelloReply* reply) override { std::string prefix("Hello "); reply->set_message(prefix + request->name()); ServerUnaryReactor* reactor = context->DefaultReactor(); reactor->Finish(Status::OK); return reactor; } }; } // namespace void RunServer(uint16_t port) { std::string server_address = absl::StrFormat("0.0.0.0:%d", port); GreeterServiceImpl service; grpc::EnableDefaultHealthCheckService(true); grpc::reflection::InitProtoReflectionServerBuilderPlugin(); ServerBuilder builder; // Listen on the given address without any authentication mechanism. builder.AddListeningPort(server_address, grpc::InsecureServerCredentials()); // Register "service" as the instance through which we'll communicate with // clients. In this case it corresponds to an *synchronous* service. builder.RegisterService(&service); // Finally assemble the server. std::unique_ptr server(builder.BuildAndStart()); std::cout << "Server listening on " << server_address << std::endl; // Wait for the server to shutdown. Note that some other thread must be // responsible for shutting down the server for this call to ever return. server->Wait(); } void RunClient(const std::string& target_str) { // Instantiate the client. It requires a channel, out of which the actual RPCs // are created. This channel models a connection to an endpoint specified by // the argument "--target=" which is the only expected argument. grpc::ChannelArguments args; // Continuously send RPCs every second. while (true) { GreeterClient greeter(grpc::CreateCustomChannel( target_str, grpc::InsecureChannelCredentials(), args)); std::string user("world"); std::string reply = greeter.SayHello(user); std::cout << "Greeter received: " << reply << std::endl; std::this_thread::sleep_for(std::chrono::seconds(1)); } }