[examples] Update SSA example

examples/cpp/csm/Dockerfile.client

@@ -22,7 +22,7 @@ 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 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

examples/cpp/csm/README.md

@@ -6,6 +6,30 @@ This CSM example builds on the [Hello World Example](https://github.com/grpc/grp
 
 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.
+* cookie_name - session affinity cookie name. Defaults to "GSSA"
+* delay_s - delay (in seconds) between the RPCs. Default value is 5
+
 
 The server takes the following command-line arguments -
 * port - Port on which the Hello World service is run. Defaults to 50051.
+
+## Building
+
+From the gRPC workspace folder:
+
+Client:
+```
+docker build -f examples/cpp/csm/Dockerfile.client
+```
+Server:
+```
+docker build -f examples/cpp/csm/Dockerfile.server
+```
+
+To push to a registry, add a tag to the image either by adding a `-t` flag to `docker build` command above or run:
+
+```
+docker image tag ${sha from build command above} ${tag}
+```
+
+And then push the tagged image using `docker push`

examples/cpp/csm/csm_greeter_client.cc

@@ -16,16 +16,20 @@
  *
  */
 
+#include <sys/types.h>
+
+#include <chrono>
 #include <condition_variable>
 #include <iostream>
 #include <memory>
-#include <mutex>
+#include <optional>
 #include <string>
 
 #include "absl/flags/flag.h"
 #include "absl/flags/parse.h"
 #include "absl/strings/str_join.h"
 #include "absl/strings/str_split.h"
+#include "absl/types/optional.h"
 #include "opentelemetry/exporters/prometheus/exporter_factory.h"
 #include "opentelemetry/exporters/prometheus/exporter_options.h"
 #include "opentelemetry/sdk/metrics/meter_provider.h"
@@ -41,6 +45,8 @@
 #endif
 
 ABSL_FLAG(std::string, target, "xds:///helloworld:50051", "Target string");
+ABSL_FLAG(std::string, cookie_name, "GSSA", "Cookie name");
+ABSL_FLAG(uint, delay_s, 5, "Delay between requests");
 
 using grpc::Channel;
 using grpc::ClientContext;
@@ -49,15 +55,13 @@ using helloworld::Greeter;
 using helloworld::HelloReply;
 using helloworld::HelloRequest;
 
+namespace {
+
 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})",
@@ -66,49 +70,45 @@ struct Cookie {
   }
 };
 
-class GreeterClient {
+Cookie ParseCookie(absl::string_view header) {
- protected:
+  Cookie cookie;
-  static Cookie ParseCookie(absl::string_view header) {
+  std::pair<absl::string_view, absl::string_view> name_value =
-    Cookie cookie;
+      absl::StrSplit(header, absl::MaxSplits('=', 1));
-    std::pair<absl::string_view, absl::string_view> name_value =
+  cookie.name = std::string(name_value.first);
-        absl::StrSplit(header, absl::MaxSplits('=', 1));
+  std::pair<absl::string_view, absl::string_view> value_attrs =
-    cookie.name = std::string(name_value.first);
+      absl::StrSplit(name_value.second, absl::MaxSplits(';', 1));
-    std::pair<absl::string_view, absl::string_view> value_attrs =
+  cookie.value = std::string(value_attrs.first);
-        absl::StrSplit(name_value.second, absl::MaxSplits(';', 1));
+  for (absl::string_view segment : absl::StrSplit(value_attrs.second, ';')) {
-    cookie.value = std::string(value_attrs.first);
+    cookie.attributes.emplace(absl::StripAsciiWhitespace(segment));
-    for (absl::string_view segment : absl::StrSplit(value_attrs.second, ';')) {
-      cookie.attributes.emplace(absl::StripAsciiWhitespace(segment));
-    }
-    return cookie;
   }
+  return cookie;
+}
 
-  static std::vector<Cookie> GetCookies(
+std::vector<Cookie> GetCookies(
-      const std::multimap<grpc::string_ref, grpc::string_ref>&
+    const std::multimap<grpc::string_ref, grpc::string_ref>& initial_metadata,
-          server_initial_metadata,
+    absl::string_view cookie_name) {
-      absl::string_view cookie_name) {
+  std::vector<Cookie> values;
-    std::vector<Cookie> values;
+  auto pair = initial_metadata.equal_range("set-cookie");
-    auto pair = server_initial_metadata.equal_range("set-cookie");
+  for (auto it = pair.first; it != pair.second; ++it) {
-    for (auto it = pair.first; it != pair.second; ++it) {
+    const auto cookie = ParseCookie(it->second.data());
-      gpr_log(GPR_INFO, "set-cookie header: %s", it->second.data());
+    if (cookie.name == cookie_name) {
-      const auto cookie = ParseCookie(it->second.data());
+      values.emplace_back(std::move(cookie));
-      if (cookie.name == cookie_name) {
-        values.emplace_back(cookie);
-      }
     }
-    return values;
   }
+  return values;
+}
 
+class GreeterClient {
  public:
-  GreeterClient(std::shared_ptr<Channel> channel)
+  GreeterClient(std::shared_ptr<Channel> channel, absl::string_view cookie_name)
-      : stub_(Greeter::NewStub(channel)) {}
+      : stub_(Greeter::NewStub(channel)), cookie_name_(cookie_name) {}
 
   // Assembles the client's payload, sends it and presents the response back
   // from the server.
-  std::string SayHello(const std::string& user, Cookie* cookieFromServer,
+  void SayHello() {
-                       const Cookie* cookieToServer) {
     // Data we are sending to the server.
     HelloRequest request;
-    request.set_name(user);
+    request.set_name("world");
 
     // Container for the data we expect from the server.
     HelloReply reply;
@@ -120,59 +120,45 @@ class GreeterClient {
     // The actual RPC.
     std::mutex mu;
     std::condition_variable cv;
-    bool done = false;
+    absl::optional<Status> status;
-    Status status;
+    // Set the cookie header if we already got a cookie from the server
-    if (cookieToServer != NULL) {
+    if (cookie_from_server_.has_value()) {
-      std::pair<std::string, std::string> cookieHeader =
+      context.AddMetadata("cookie",
-          cookieToServer->Header();
+                          absl::StrFormat("%s=%s", cookie_from_server_->name,
-      context.AddMetadata(cookieHeader.first, cookieHeader.second);
+                                          cookie_from_server_->value));
     }
-    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) {
+    stub_->async()->SayHello(&context, &request, &reply, [&](Status s) {
+      std::lock_guard<std::mutex> lock(mu);
+      status = std::move(s);
+      cv.notify_one();
+    });
+    while (!status.has_value()) {
       cv.wait(lock);
     }
-
+    if (!status->ok()) {
-    // Act upon its status.
+      std::cout << "RPC failed" << status->error_code() << ": "
-    if (status.ok()) {
+                << status->error_message() << std::endl;
-      if (cookieFromServer != NULL) {
+      return;
-        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";
     }
+    const std::multimap<grpc::string_ref, grpc::string_ref>&
+        server_initial_metadata = context.GetServerInitialMetadata();
+    // Update a cookie after a successful request
+    std::vector<Cookie> cookies =
+        GetCookies(server_initial_metadata, cookie_name_);
+    if (!cookies.empty()) {
+      cookie_from_server_.emplace(std::move(cookies.front()));
+    }
+    std::cout << "Greeter received: " << reply.message() << std::endl;
   }
 
  private:
   std::unique_ptr<Greeter::Stub> stub_;
+  std::string cookie_name_;
+  absl::optional<Cookie> cookie_from_server_;
 };
 
-static void sayHello(GreeterClient& greeter, Cookie* cookieFromServer,
+absl::StatusOr<grpc::experimental::CsmObservability> InitializeObservability() {
-                     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";
@@ -181,21 +167,29 @@ int main(int argc, char** argv) {
   auto meter_provider =
       std::make_shared<opentelemetry::sdk::metrics::MeterProvider>();
   meter_provider->AddMetricReader(std::move(prometheus_exporter));
-  auto observability = grpc::experimental::CsmObservabilityBuilder()
+  return grpc::experimental::CsmObservabilityBuilder()
-                           .SetMeterProvider(std::move(meter_provider))
+      .SetMeterProvider(std::move(meter_provider))
-                           .BuildAndRegister();
+      .BuildAndRegister();
+}
+
+}  // namespace
+
+int main(int argc, char** argv) {
+  absl::ParseCommandLine(argc, argv);
+  // Setup the observability
+  auto observability = InitializeObservability();
   if (!observability.ok()) {
     std::cerr << "CsmObservability::Init() failed: "
               << observability.status().ToString() << std::endl;
     return static_cast<int>(observability.status().code());
   }
-  GreeterClient greeter(grpc::CreateChannel(
+  GreeterClient greeter(grpc::CreateChannel(absl::GetFlag(FLAGS_target),
-      absl::GetFlag(FLAGS_target), grpc::InsecureChannelCredentials()));
+                                            grpc::InsecureChannelCredentials()),
-
+                        absl::GetFlag(FLAGS_cookie_name));
-  Cookie session_cookie;
-  sayHello(greeter, &session_cookie, NULL);
   while (true) {
-    sayHello(greeter, NULL, &session_cookie);
+    greeter.SayHello();
+    std::this_thread::sleep_for(
+        std::chrono::seconds(absl::GetFlag(FLAGS_delay_s)));
   }
   return 0;
 }