[CSM] Add CSM Example client and server images (#34447)

Joint effort with @yashykt and @sanjaypujare

---------

Co-authored-by: Yash Tibrewal <yashkt@google.com>
Co-authored-by: Sanjay Pujare <sanjaypujare@users.noreply.github.com>

examples/cpp/csm/BUILD

@@ -0,0 +1,47 @@
+# Copyright 2023 the 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.
+
+licenses(["notice"])
+
+cc_binary(
+    name = "csm_greeter_client",
+    srcs = ["csm_greeter_client.cc"],
+    defines = ["BAZEL_BUILD"],
+    deps = [
+        "//:grpc++",
+        "//:grpcpp_csm_observability",
+        "//examples/protos:helloworld_cc_grpc",
+        "@com_google_absl//absl/flags:flag",
+        "@com_google_absl//absl/flags:parse",
+        "@io_opentelemetry_cpp//exporters/prometheus:prometheus_exporter",
+        "@io_opentelemetry_cpp//sdk/src/metrics",
+    ],
+)
+
+cc_binary(
+    name = "csm_greeter_server",
+    srcs = ["csm_greeter_server.cc"],
+    defines = ["BAZEL_BUILD"],
+    deps = [
+        "//:grpc++",
+        "//:grpc++_reflection",
+        "//:grpcpp_admin",
+        "//:grpcpp_csm_observability",
+        "//examples/protos:helloworld_cc_grpc",
+        "@com_google_absl//absl/flags:flag",
+        "@com_google_absl//absl/flags:parse",
+        "@io_opentelemetry_cpp//exporters/prometheus:prometheus_exporter",
+        "@io_opentelemetry_cpp//sdk/src/metrics",
+    ],
+)

examples/cpp/csm/Dockerfile.client

@@ -0,0 +1,39 @@
+# Copyright 2023 The 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.
+
+FROM python:3.9-slim-bookworm
+
+RUN apt-get update -y && apt-get upgrade -y && apt-get install -y build-essential clang curl
+
+WORKDIR /workdir
+
+RUN ln -s /usr/bin/python3 /usr/bin/python
+RUN mkdir /artifacts
+
+COPY . .
+RUN OVERRIDE_BAZEL_VERSION=5.4.0 tools/bazel build //examples/cpp/csm:csm_greeter_client
+RUN cp -rL /workdir/bazel-bin/examples/cpp/csm/csm_greeter_client /artifacts/
+
+FROM python:3.9-slim-bookworm
+
+RUN apt-get update \
+    && apt-get -y upgrade \
+    && apt-get -y autoremove \
+    && apt-get install -y curl
+
+COPY --from=0 /artifacts ./
+
+ENV GRPC_EXPERIMENTAL_XDS_ENABLE_OVERRIDE_HOST=true
+
+ENTRYPOINT ["/csm_greeter_client"]

examples/cpp/csm/Dockerfile.server

@@ -0,0 +1,39 @@
+# Copyright 2023 The 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.
+
+FROM python:3.9-slim-bookworm
+
+RUN apt-get update -y && apt-get upgrade -y && apt-get install -y build-essential clang curl
+
+WORKDIR /workdir
+
+RUN ln -s /usr/bin/python3 /usr/bin/python
+RUN mkdir /artifacts
+
+COPY . .
+RUN OVERRIDE_BAZEL_VERSION=5.4.0 tools/bazel build //examples/cpp/csm:csm_greeter_server
+RUN cp -rL /workdir/bazel-bin/examples/cpp/csm/csm_greeter_server /artifacts/
+
+FROM python:3.9-slim-bookworm
+
+RUN apt-get update \
+    && apt-get -y upgrade \
+    && apt-get -y autoremove \
+    && apt-get install -y curl
+
+COPY --from=0 /artifacts ./
+
+ENV GRPC_EXPERIMENTAL_XDS_ENABLE_OVERRIDE_HOST=true
+
+ENTRYPOINT ["/csm_greeter_server"]

examples/cpp/csm/README.md

@@ -0,0 +1,11 @@
+# gRPC C++ CSM Hello World Example
+
+This CSM example builds on the [Hello World Example](https://github.com/grpc/grpc/tree/master/examples/cpp/helloworld) and changes the gRPC client and server to accept configuration from an xDS control plane and test SSA and CSM observability
+
+## Configuration
+
+The client takes the following command-line arguments -
+* target - By default, the client tries to connect to the xDS "xds:///helloworld:50051" and gRPC would use xDS to resolve this target and connect to the server backend. This can be overriden to change the target.
+
+The server takes the following command-line arguments -
+* port - Port on which the Hello World service is run. Defaults to 50051.

examples/cpp/csm/csm_greeter_client.cc

@@ -0,0 +1,200 @@
+/*
+ *
+ * Copyright 2023 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.
+ *
+ */
+
+#include <condition_variable>
+#include <iostream>
+#include <memory>
+#include <mutex>
+#include <string>
+
+#include "absl/flags/flag.h"
+#include "absl/flags/parse.h"
+#include "absl/strings/str_split.h"
+#include "opentelemetry/exporters/prometheus/exporter_factory.h"
+#include "opentelemetry/exporters/prometheus/exporter_options.h"
+#include "opentelemetry/sdk/metrics/meter_provider.h"
+
+#include <grpcpp/ext/csm_observability.h>
+#include <grpcpp/grpcpp.h>
+#include <grpcpp/support/string_ref.h>
+
+#ifdef BAZEL_BUILD
+#include "examples/protos/helloworld.grpc.pb.h"
+#else
+#include "helloworld.grpc.pb.h"
+#endif
+
+ABSL_FLAG(std::string, target, "xds:///helloworld:50051", "Target string");
+
+using grpc::Channel;
+using grpc::ClientContext;
+using grpc::Status;
+using helloworld::Greeter;
+using helloworld::HelloReply;
+using helloworld::HelloRequest;
+
+struct Cookie {
+  std::string name;
+  std::string value;
+  std::set<std::string> attributes;
+
+  std::pair<std::string, std::string> Header() const {
+    return std::make_pair("cookie", absl::StrFormat("%s=%s", name, value));
+  }
+
+  template <typename Sink>
+  friend void AbslStringify(Sink& sink, const Cookie& cookie) {
+    absl::Format(&sink, "(Cookie: %s, value: %s, attributes: {%s})",
+                 cookie.name, cookie.value,
+                 absl::StrJoin(cookie.attributes, ", "));
+  }
+};
+
+class GreeterClient {
+ protected:
+  static Cookie ParseCookie(absl::string_view header) {
+    Cookie cookie;
+    std::pair<absl::string_view, absl::string_view> name_value =
+        absl::StrSplit(header, absl::MaxSplits('=', 1));
+    cookie.name = std::string(name_value.first);
+    std::pair<absl::string_view, absl::string_view> value_attrs =
+        absl::StrSplit(name_value.second, absl::MaxSplits(';', 1));
+    cookie.value = std::string(value_attrs.first);
+    for (absl::string_view segment : absl::StrSplit(value_attrs.second, ';')) {
+      cookie.attributes.emplace(absl::StripAsciiWhitespace(segment));
+    }
+    return cookie;
+  }
+
+  static std::vector<Cookie> GetCookies(
+      const std::multimap<grpc::string_ref, grpc::string_ref>&
+          server_initial_metadata,
+      absl::string_view cookie_name) {
+    std::vector<Cookie> values;
+    auto pair = server_initial_metadata.equal_range("set-cookie");
+    for (auto it = pair.first; it != pair.second; ++it) {
+      gpr_log(GPR_INFO, "set-cookie header: %s", it->second.data());
+      const auto cookie = ParseCookie(it->second.data());
+      if (cookie.name == cookie_name) {
+        values.emplace_back(cookie);
+      }
+    }
+    return values;
+  }
+
+ public:
+  GreeterClient(std::shared_ptr<Channel> 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, Cookie* cookieFromServer,
+                       const Cookie* cookieToServer) {
+    // 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;
+    if (cookieToServer != NULL) {
+      std::pair<std::string, std::string> cookieHeader =
+          cookieToServer->Header();
+      context.AddMetadata(cookieHeader.first, cookieHeader.second);
+    }
+    stub_->async()->SayHello(&context, &request, &reply,
+                             [&mu, &cv, &done, &status](Status s) {
+                               status = std::move(s);
+                               std::lock_guard<std::mutex> lock(mu);
+                               done = true;
+                               cv.notify_one();
+                             });
+
+    std::unique_lock<std::mutex> lock(mu);
+    while (!done) {
+      cv.wait(lock);
+    }
+
+    // Act upon its status.
+    if (status.ok()) {
+      if (cookieFromServer != NULL) {
+        const std::multimap<grpc::string_ref, grpc::string_ref>&
+            server_initial_metadata = context.GetServerInitialMetadata();
+        std::vector<Cookie> cookies =
+            GetCookies(server_initial_metadata, "GSSA");
+        if (!cookies.empty()) {
+          *cookieFromServer = cookies.front();
+        }
+      }
+      return reply.message();
+    } else {
+      std::cout << status.error_code() << ": " << status.error_message()
+                << std::endl;
+      return "RPC failed";
+    }
+  }
+
+ private:
+  std::unique_ptr<Greeter::Stub> stub_;
+};
+
+static void sayHello(GreeterClient& greeter, Cookie* cookieFromServer,
+                     const Cookie* cookieToServer) {
+  std::string user("world");
+  std::string reply = greeter.SayHello(user, cookieFromServer, cookieToServer);
+  std::cout << "Greeter received: " << reply << std::endl;
+  std::this_thread::sleep_for(std::chrono::seconds(5));
+}
+
+int main(int argc, char** argv) {
+  absl::ParseCommandLine(argc, argv);
+  opentelemetry::exporter::metrics::PrometheusExporterOptions opts;
+  // default was "localhost:9464" which causes connection issue across GKE pods
+  opts.url = "0.0.0.0:9464";
+  auto prometheus_exporter =
+      opentelemetry::exporter::metrics::PrometheusExporterFactory::Create(opts);
+  auto meter_provider =
+      std::make_shared<opentelemetry::sdk::metrics::MeterProvider>();
+  meter_provider->AddMetricReader(std::move(prometheus_exporter));
+  auto observability = grpc::experimental::CsmObservabilityBuilder()
+                           .SetMeterProvider(std::move(meter_provider))
+                           .BuildAndRegister();
+  if (!observability.ok()) {
+    std::cerr << "CsmObservability::Init() failed: "
+              << observability.status().ToString() << std::endl;
+    return static_cast<int>(observability.status().code());
+  }
+  GreeterClient greeter(grpc::CreateChannel(
+      absl::GetFlag(FLAGS_target), grpc::InsecureChannelCredentials()));
+
+  Cookie session_cookie;
+  sayHello(greeter, &session_cookie, NULL);
+  while (true) {
+    sayHello(greeter, NULL, &session_cookie);
+  }
+  return 0;
+}

examples/cpp/csm/csm_greeter_server.cc

@@ -0,0 +1,126 @@
+/*
+ *
+ * Copyright 2023 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.
+ *
+ */
+
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include "absl/flags/flag.h"
+#include "absl/flags/parse.h"
+#include "absl/strings/str_cat.h"
+#include "opentelemetry/exporters/prometheus/exporter_factory.h"
+#include "opentelemetry/exporters/prometheus/exporter_options.h"
+#include "opentelemetry/sdk/metrics/meter_provider.h"
+
+#include <grpcpp/ext/admin_services.h>
+#include <grpcpp/ext/csm_observability.h>
+#include <grpcpp/ext/proto_server_reflection_plugin.h>
+#include <grpcpp/grpcpp.h>
+#include <grpcpp/health_check_service_interface.h>
+#include <grpcpp/xds_server_builder.h>
+
+#include "src/core/lib/iomgr/gethostname.h"
+
+#ifdef BAZEL_BUILD
+#include "examples/protos/helloworld.grpc.pb.h"
+#else
+#include "helloworld.grpc.pb.h"
+#endif
+
+ABSL_FLAG(int32_t, port, 50051, "Server port for service.");
+
+using grpc::CallbackServerContext;
+using grpc::Server;
+using grpc::ServerBuilder;
+using grpc::ServerUnaryReactor;
+using grpc::Status;
+using helloworld::Greeter;
+using helloworld::HelloReply;
+using helloworld::HelloRequest;
+
+// 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 from ");
+    prefix += my_name + " ";
+    reply->set_message(prefix + request->name());
+
+    ServerUnaryReactor* reactor = context->DefaultReactor();
+    reactor->Finish(Status::OK);
+    return reactor;
+  }
+
+ public:
+  GreeterServiceImpl(const std::string& my_hostname) : my_name(my_hostname) {}
+
+ private:
+  const std::string my_name;
+};
+
+void RunServer(const char* hostname) {
+  grpc::EnableDefaultHealthCheckService(true);
+  grpc::reflection::InitProtoReflectionServerBuilderPlugin();
+  int port = absl::GetFlag(FLAGS_port);
+  grpc::XdsServerBuilder xds_builder;
+  std::unique_ptr<Server> xds_enabled_server;
+
+  std::string my_hostname(hostname);
+  GreeterServiceImpl service(my_hostname);
+  // Register "service" as the instance through which we'll communicate with
+  // clients. In this case it corresponds to an *synchronous* service.
+  xds_builder.RegisterService(&service);
+  // Listen on the given address with XdsServerCredentials and a fallback of
+  // InsecureServerCredentials
+  xds_builder.AddListeningPort(absl::StrCat("0.0.0.0:", port),
+                               grpc::InsecureServerCredentials());
+  xds_enabled_server = xds_builder.BuildAndStart();
+  gpr_log(GPR_INFO, "Server starting on 0.0.0.0:%d", port);
+
+  // 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.
+  xds_enabled_server->Wait();
+}
+
+int main(int argc, char** argv) {
+  absl::ParseCommandLine(argc, argv);
+  opentelemetry::exporter::metrics::PrometheusExporterOptions opts;
+  // default was "localhost:9464" which causes connection issue across GKE pods
+  opts.url = "0.0.0.0:9464";
+  auto prometheus_exporter =
+      opentelemetry::exporter::metrics::PrometheusExporterFactory::Create(opts);
+  auto meter_provider =
+      std::make_shared<opentelemetry::sdk::metrics::MeterProvider>();
+  meter_provider->AddMetricReader(std::move(prometheus_exporter));
+  auto observability = grpc::experimental::CsmObservabilityBuilder()
+                           .SetMeterProvider(std::move(meter_provider))
+                           .BuildAndRegister();
+  if (!observability.ok()) {
+    std::cerr << "CsmObservability::Init() failed: "
+              << observability.status().ToString() << std::endl;
+    return static_cast<int>(observability.status().code());
+  }
+  const char* hostname = grpc_gethostname();
+  if (hostname == nullptr) {
+    std::cout << "Failed to get hostname, terminating" << std::endl;
+    return 1;
+  }
+  RunServer(hostname);
+  return 0;
+}