[Python CSM] Add example for CSM O11Y (#36829)

Verified that when following User Guide, we're able to collect metrics
and export to GCP.
Images are available at:
*
`us-docker.pkg.dev/grpc-testing/examples/csm-o11y-example-python-client:latest`
*
`us-docker.pkg.dev/grpc-testing/examples/csm-o11y-example-python-server:latest`
<!--

If you know who should review your pull request, please assign it to
that
person, otherwise the pull request would get assigned randomly.

If your pull request is for a specific language, please add the
appropriate
lang label.

-->

examples/python/observability/csm/BUILD.bazel

@@ -0,0 +1,46 @@
+# Copyright 2024 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.
+
+load("@grpc_python_dependencies//:requirements.bzl", "requirement")
+
+# TODO(xuanwn): Instaed of using Bazel build, we should pip install all dependencies
+# once we have a released version of grpcio-csm-observability.
+
+py_binary(
+    name = "csm_greeter_client",
+    srcs = ["csm_greeter_client.py"],
+    python_version = "PY3",
+    deps = [
+        "//src/proto/grpc/testing:py_messages_proto",
+        "//src/proto/grpc/testing:py_test_proto",
+        "//src/proto/grpc/testing:test_py_pb2_grpc",
+        "//src/python/grpcio/grpc:grpcio",
+        "//src/python/grpcio_csm_observability/grpc_csm_observability:csm_observability",
+        requirement("opentelemetry-exporter-prometheus"),
+    ],
+)
+
+py_binary(
+    name = "csm_greeter_server",
+    srcs = ["csm_greeter_server.py"],
+    python_version = "PY3",
+    deps = [
+        "//src/proto/grpc/testing:py_messages_proto",
+        "//src/proto/grpc/testing:py_test_proto",
+        "//src/proto/grpc/testing:test_py_pb2_grpc",
+        "//src/python/grpcio/grpc:grpcio",
+        "//src/python/grpcio_csm_observability/grpc_csm_observability:csm_observability",
+        requirement("opentelemetry-exporter-prometheus"),
+    ],
+)

examples/python/observability/csm/Dockerfile.client

@@ -0,0 +1,26 @@
+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 tools/bazel build -c dbg //examples/python/observability/csm:csm_greeter_client
+RUN cp -rL /workdir/bazel-bin/examples/python/observability/csm/csm_greeter_client* /artifacts/
+
+FROM python:3.9-slim-bookworm
+
+RUN apt-get update -y \
+  && apt-get install -y python3 \
+  && apt-get -y autoremove \
+  && apt-get install -y curl \
+  && rm -rf /var/lib/apt/lists/*
+
+RUN ln -s /usr/bin/python3 /usr/bin/python
+
+COPY --from=0 /artifacts ./
+
+ENTRYPOINT ["/csm_greeter_client"]

examples/python/observability/csm/Dockerfile.server

@@ -0,0 +1,26 @@
+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 tools/bazel build -c dbg //examples/python/observability/csm:csm_greeter_server
+RUN cp -rL /workdir/bazel-bin/examples/python/observability/csm/csm_greeter_server* /artifacts/
+
+FROM python:3.9-slim-bookworm
+
+RUN apt-get update -y \
+    && apt-get install -y python3 \
+    && apt-get -y autoremove \
+    && apt-get install -y curl \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN ln -s /usr/bin/python3 /usr/bin/python
+
+COPY --from=0 /artifacts ./
+
+ENTRYPOINT ["/csm_greeter_server"]

examples/python/observability/csm/README.md

@@ -0,0 +1,33 @@
+# gRPC Python CSM Hello World Example
+
+This CSM example builds on the [Python xDS Example](https://github.com/grpc/grpc/tree/master/examples/python/xds) and changes the gRPC client and server to accept configuration from an xDS control plane and test CSM observability.
+
+## Configuration
+
+The client takes the following command-line arguments -
+* `--target` - By default, the client tries to connect to the target "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.
+* `--secure_mode` - Whether to use xDS to retrieve server credentials. Default value is False.
+* `--prometheus_endpoint` - Endpoint used for prometheus. Default value is localhost:9464.
+
+
+The server takes the following command-line arguments -
+* `--port` - Port on which the Hello World service is run. Defaults to 50051.
+* `--secure_mode` - Whether to use xDS to retrieve server credentials. Default value is False.
+* `--server_id` - The server ID to return in responses.
+* `--prometheus_endpoint` - Endpoint used for prometheus. Default value is `localhost:9464`.
+
+## Building
+
+From the gRPC workspace folder:
+
+Client:
+```
+docker build -f examples/python/observability/csm/Dockerfile.client -t "us-docker.pkg.dev/grpc-testing/examples/csm-o11y-example-python-client" .
+```
+
+Server:
+```
+docker build -f examples/python/observability/csm/Dockerfile.server -t "us-docker.pkg.dev/grpc-testing/examples/csm-o11y-example-python-server" .
+```
+
+And then push the tagged image using `docker push`.

examples/python/observability/csm/csm_greeter_client.py

@@ -0,0 +1,101 @@
+# Copyright 2024 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.
+
+import argparse
+import logging
+import time
+
+import grpc
+from grpc_csm_observability import CsmOpenTelemetryPlugin
+from opentelemetry.exporter.prometheus import PrometheusMetricReader
+from opentelemetry.sdk.metrics import MeterProvider
+from prometheus_client import start_http_server
+
+from src.proto.grpc.testing import messages_pb2
+from src.proto.grpc.testing import test_pb2_grpc
+
+logger = logging.getLogger()
+console_handler = logging.StreamHandler()
+formatter = logging.Formatter(fmt="%(asctime)s: %(levelname)-8s %(message)s")
+console_handler.setFormatter(formatter)
+logger.addHandler(console_handler)
+
+
+def _run(target: int, secure_mode: bool, prometheus_endpoint: int):
+    csm_plugin = _prepare_csm_observability_plugin(prometheus_endpoint)
+    csm_plugin.register_global()
+    if secure_mode:
+        fallback_creds = grpc.experimental.insecure_channel_credentials()
+        channel_creds = grpc.xds_channel_credentials(fallback_creds)
+        channel = grpc.secure_channel(target, channel_creds)
+    else:
+        channel = grpc.insecure_channel(target)
+    with channel:
+        stub = test_pb2_grpc.TestServiceStub(channel)
+        # Continuously send RPCs every second.
+        while True:
+            request = messages_pb2.SimpleRequest()
+            logger.info("Sending request to server")
+            stub.UnaryCall(request)
+            time.sleep(1)
+    csm_plugin.deregister_global()
+
+
+def _prepare_csm_observability_plugin(
+    prometheus_endpoint: int,
+) -> CsmOpenTelemetryPlugin:
+    # Start Prometheus client
+    start_http_server(port=prometheus_endpoint, addr="0.0.0.0")
+    reader = PrometheusMetricReader()
+    meter_provider = MeterProvider(metric_readers=[reader])
+    csm_plugin = CsmOpenTelemetryPlugin(
+        meter_provider=meter_provider,
+    )
+    return csm_plugin
+
+
+def bool_arg(arg: str) -> bool:
+    if arg.lower() in ("true", "yes", "y"):
+        return True
+    elif arg.lower() in ("false", "no", "n"):
+        return False
+    else:
+        raise argparse.ArgumentTypeError(f"Could not parse '{arg}' as a bool.")
+
+
+if __name__ == "__main__":
+    logging.basicConfig()
+    logger.setLevel(logging.INFO)
+    parser = argparse.ArgumentParser(
+        description="Run Python CSM Observability Test client."
+    )
+    parser.add_argument(
+        "--target",
+        default="xds:///helloworld:50051",
+        help="The address of the server.",
+    )
+    parser.add_argument(
+        "--secure_mode",
+        default="False",
+        type=bool_arg,
+        help="If specified, uses xDS credentials to connect to the server.",
+    )
+    parser.add_argument(
+        "--prometheus_endpoint",
+        type=int,
+        default=9464,
+        help="Port for servers besides test server.",
+    )
+    args = parser.parse_args()
+    _run(args.target, args.secure_mode, args.prometheus_endpoint)

examples/python/observability/csm/csm_greeter_server.py

@@ -0,0 +1,149 @@
+# Copyright 2024 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.
+
+import argparse
+from concurrent import futures
+import logging
+import socket
+
+import grpc
+from grpc_csm_observability import CsmOpenTelemetryPlugin
+from opentelemetry.exporter.prometheus import PrometheusMetricReader
+from opentelemetry.sdk.metrics import MeterProvider
+from prometheus_client import start_http_server
+
+from src.proto.grpc.testing import messages_pb2
+from src.proto.grpc.testing import test_pb2_grpc
+
+_LISTEN_HOST = "0.0.0.0"
+_THREAD_POOL_SIZE = 256
+
+logger = logging.getLogger()
+console_handler = logging.StreamHandler()
+formatter = logging.Formatter(fmt="%(asctime)s: %(levelname)-8s %(message)s")
+console_handler.setFormatter(formatter)
+logger.addHandler(console_handler)
+
+
+class TestService(test_pb2_grpc.TestServiceServicer):
+    def __init__(self, server_id, hostname):
+        self._server_id = server_id
+        self._hostname = hostname
+
+    def UnaryCall(
+        self, request: messages_pb2.SimpleRequest, context: grpc.ServicerContext
+    ) -> messages_pb2.SimpleResponse:
+        context.send_initial_metadata((("hostname", self._hostname),))
+        if request.response_size > 0:
+            response = messages_pb2.SimpleResponse(
+                payload=messages_pb2.Payload(body=b"0" * request.response_size)
+            )
+        else:
+            response = messages_pb2.SimpleResponse()
+        response.server_id = self._server_id
+        response.hostname = self._hostname
+        logger.info("Sending response to client")
+        return response
+
+
+def _run(
+    port: int,
+    secure_mode: bool,
+    server_id: str,
+    prometheus_endpoint: int,
+) -> None:
+    csm_plugin = _prepare_csm_observability_plugin(prometheus_endpoint)
+    csm_plugin.register_global()
+    server = grpc.server(
+        futures.ThreadPoolExecutor(max_workers=_THREAD_POOL_SIZE)
+    )
+    _configure_test_server(server, port, secure_mode, server_id)
+    server.start()
+    logger.info("Test server listening on port %d", port)
+    server.wait_for_termination()
+    csm_plugin.deregister_global()
+
+
+def _prepare_csm_observability_plugin(
+    prometheus_endpoint: int,
+) -> CsmOpenTelemetryPlugin:
+    # Start Prometheus client
+    start_http_server(port=prometheus_endpoint, addr="0.0.0.0")
+    reader = PrometheusMetricReader()
+    meter_provider = MeterProvider(metric_readers=[reader])
+    csm_plugin = CsmOpenTelemetryPlugin(
+        meter_provider=meter_provider,
+    )
+    return csm_plugin
+
+
+def _configure_test_server(
+    server: grpc.Server, port: int, secure_mode: bool, server_id: str
+) -> None:
+    test_pb2_grpc.add_TestServiceServicer_to_server(
+        TestService(server_id, socket.gethostname()), server
+    )
+    listen_address = f"{_LISTEN_HOST}:{port}"
+    if not secure_mode:
+        server.add_insecure_port(listen_address)
+    else:
+        logger.info("Running with xDS Server credentials")
+        server_fallback_creds = grpc.insecure_server_credentials()
+        server_creds = grpc.xds_server_credentials(server_fallback_creds)
+        server.add_secure_port(listen_address, server_creds)
+
+
+def bool_arg(arg: str) -> bool:
+    if arg.lower() in ("true", "yes", "y"):
+        return True
+    elif arg.lower() in ("false", "no", "n"):
+        return False
+    else:
+        raise argparse.ArgumentTypeError(f"Could not parse '{arg}' as a bool.")
+
+
+if __name__ == "__main__":
+    logging.basicConfig()
+    logger.setLevel(logging.INFO)
+    parser = argparse.ArgumentParser(
+        description="Run Python CSM Observability Test server."
+    )
+    parser.add_argument(
+        "--port", type=int, default=50051, help="Port for test server."
+    )
+    parser.add_argument(
+        "--secure_mode",
+        type=bool_arg,
+        default="False",
+        help="If specified, uses xDS to retrieve server credentials.",
+    )
+    parser.add_argument(
+        "--server_id",
+        type=str,
+        default="python_server",
+        help="The server ID to return in responses.",
+    )
+    parser.add_argument(
+        "--prometheus_endpoint",
+        type=int,
+        default=9464,
+        help="Port for servers besides test server.",
+    )
+    args = parser.parse_args()
+    _run(
+        args.port,
+        args.secure_mode,
+        args.server_id,
+        args.prometheus_endpoint,
+    )