natas_hw
/
grpc
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[Python Observability] Building package and add to run_test (#34207) 

### Changes in this PR
* Refactor and remove some Core/C++ dependencies to simplify Python Observability package build process.
* Refactored code to read config at Python layer.
* Enable observability build from source.
* Add observability to run_test.
  * Currently it's only enabled in Linux.
* Add error handler in run_test loaders.
  * Current framework will always visit modules in test directory then decide which tests to skip.
  * Since we're not building Observability for MacOS and Windows this step will fail with error `No module named 'grpc_observability'`.
  * After the change we'll just skip those modules.
    * We still have `_sanity_test` to make sure all tests are loaded correctly for each platform.
* Remov OC dependency as we're migrating to OTel.
  * Also removed trace from testing.
  * Note that trace propagation function was also removed because of this.

### Testing
* Passed existing tests.
* Tested locally, able to build observability from source using `GRPC_PYTHON_BUILD_WITH_CYTHON=1 pip install .`.

<!--

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.

-->

Closes #34207

PiperOrigin-RevId: 590258014
This commit is contained in:
Xuan Wang 2023-12-12 10:40:59 -08:00 committed by Copybara-Service
parent 2cc1550e77
commit 7f2ecdb3f8
32 changed files with 1247 additions and 428 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
BUILD
View File

@ -2308,29 +2308,6 @@ grpc_cc_library(
],
)
grpc_cc_library(
name = "grpc_rpc_encoding",
srcs = [
"src/cpp/ext/filters/census/rpc_encoding.cc",
],
hdrs = [
"src/cpp/ext/filters/census/rpc_encoding.h",
],
external_deps = [
"absl/base",
"absl/base:core_headers",
"absl/base:endian",
"absl/meta:type_traits",
"absl/status",
"absl/strings",
"absl/time",
],
language = "c++",
tags = ["nofixdeps"],
visibility = ["@grpc:grpc_python_observability"],
deps = ["gpr_platform"],
)
grpc_cc_library(
name = "grpc_opencensus_plugin",
srcs = [

3
requirements.bazel.txt
View File

@ -3,6 +3,7 @@ coverage==4.5.4
cython==0.29.21
protobuf>=3.5.0.post1, < 4.0dev
wheel==0.38.1
google-auth==1.24.0
oauth2client==4.1.0
requests==2.25.1
urllib3==1.26.5
@ -13,7 +14,5 @@ gevent==22.08.0
zope.event==4.5.0
setuptools==44.1.1
xds-protos==0.0.11
opencensus==0.10.0
opencensus-ext-stackdriver==0.8.0
absl-py==1.4.0
googleapis-common-protos==1.61.0

5
src/python/grpcio_observability/.gitignore vendored
View File

@ -1,6 +1,7 @@
build/
include/
grpc_root/
third_party/
*.egg-info/
*.c
*.cpp
*.egg-info
*.so

8
src/python/grpcio_observability/MANIFEST.in Normal file
View File

@ -0,0 +1,8 @@
graft src/python/grpcio_observability/grpcio_observability.egg-info
graft grpc_observability
graft grpc_root
graft third_party
include _parallel_compile_patch.py
include grpc_version.py
include observability_lib_deps.py
include README.rst

77
src/python/grpcio_observability/_parallel_compile_patch.py Normal file
View File

@ -0,0 +1,77 @@
# 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.
"""Patches the compile() to allow enable parallel compilation of C/C++.
build_ext has lots of C/C++ files and normally them one by one.
Enabling parallel build helps a lot.
"""
import os
try:
BUILD_EXT_COMPILER_JOBS = int(
os.environ["GRPC_PYTHON_BUILD_EXT_COMPILER_JOBS"]
)
except KeyError:
import multiprocessing
BUILD_EXT_COMPILER_JOBS = multiprocessing.cpu_count()
# monkey-patch for parallel compilation
# TODO(xuanwn): Use a template for this file.
def _parallel_compile(
self,
sources,
output_dir=None,
macros=None,
include_dirs=None,
debug=0,
extra_preargs=None,
extra_postargs=None,
depends=None,
):
# setup the same way as distutils.ccompiler.CCompiler
# https://github.com/python/cpython/blob/31368a4f0e531c19affe2a1becd25fc316bc7501/Lib/distutils/ccompiler.py#L564
macros, objects, extra_postargs, pp_opts, build = self._setup_compile(
output_dir, macros, include_dirs, sources, depends, extra_postargs
)
cc_args = self._get_cc_args(pp_opts, debug, extra_preargs)
def _compile_single_file(obj):
try:
src, ext = build[obj]
except KeyError:
return
self._compile(obj, src, ext, cc_args, extra_postargs, pp_opts)
# run compilation of individual files in parallel
import multiprocessing.pool
multiprocessing.pool.ThreadPool(BUILD_EXT_COMPILER_JOBS).map(
_compile_single_file, objects
)
return objects
def monkeypatch_compile_maybe():
"""
Monkeypatching is dumb, but the build speed gain is worth it.
After python 3.12, we won't find distutils if SETUPTOOLS_USE_DISTUTILS=stdlib.
"""
use_distutils = os.environ.get("SETUPTOOLS_USE_DISTUTILS", "")
if BUILD_EXT_COMPILER_JOBS > 1 and use_distutils != "stdlib":
import distutils.ccompiler # pylint: disable=wrong-import-position
distutils.ccompiler.CCompiler.compile = _parallel_compile

17
src/python/grpcio_observability/grpc_observability/BUILD.bazel
View File

@ -12,12 +12,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
load("@grpc_python_dependencies//:requirements.bzl", "requirement")
load("//bazel:cython_library.bzl", "pyx_library")
package(default_visibility = ["//visibility:public"])
# TODO(xuanwn): We also need support Python-native build
package(default_visibility = ["//visibility:private"])
cc_library(
name = "observability",
@ -25,6 +22,7 @@ cc_library(
"client_call_tracer.cc",
"observability_util.cc",
"python_census_context.cc",
"rpc_encoding.cc",
"sampler.cc",
"server_call_tracer.cc",
],
@ -33,15 +31,13 @@ cc_library(
"constants.h",
"observability_util.h",
"python_census_context.h",
"rpc_encoding.h",
"sampler.h",
"server_call_tracer.h",
],
includes = ["."],
deps = [
#TODO(xuanwn): Confirm only referenced code is inlcuded in shared object library
"//:grpc",
"//:grpc_rpc_encoding",
"//src/cpp/ext/gcp:observability_config",
"//:grpc_base",
],
)
@ -65,7 +61,7 @@ py_library(
"_gcp_observability.py",
"_measures.py",
"_observability.py",
"_open_census_exporter.py",
"_observability_config.py",
"_views.py",
],
imports = [
@ -78,8 +74,5 @@ py_library(
],
deps = [
":cyobservability",
"//src/python/grpcio/grpc:grpcio",
requirement("opencensus"),
requirement("opencensus-ext-stackdriver"),
],
)

29
src/python/grpcio_observability/grpc_observability/_cyobservability.pyx
View File

@ -22,7 +22,7 @@ import os
from threading import Thread
from typing import List, Mapping, Tuple, Union
import _observability
from grpc_observability import _observability
# Time we wait for batch exporting census data
# TODO(xuanwn): change interval to a more appropriate number
@ -72,7 +72,6 @@ class MetricsName(enum.Enum):
CLIENT_RETRIES_PER_CALL = _CyMetricsName.CY_CLIENT_RETRIES_PER_CALL
CLIENT_TRANSPARENT_RETRIES_PER_CALL = _CyMetricsName.CY_CLIENT_TRANSPARENT_RETRIES_PER_CALL
CLIENT_RETRY_DELAY_PER_CALL = _CyMetricsName.CY_CLIENT_RETRY_DELAY_PER_CALL
CLIENT_TRANSPORT_LATENCY = _CyMetricsName.CY_CLIENT_TRANSPORT_LATENCY
SERVER_SENT_MESSAGES_PER_RPC = _CyMetricsName.CY_SERVER_SENT_MESSAGES_PER_RPC
SERVER_SENT_BYTES_PER_RPC = _CyMetricsName.CY_SERVER_SENT_BYTES_PER_RPC
SERVER_RECEIVED_MESSAGES_PER_RPC = _CyMetricsName.CY_SERVER_RECEIVED_MESSAGES_PER_RPC
@ -101,28 +100,16 @@ def _start_exporting_thread(object exporter) -> None:
GLOBAL_EXPORT_THREAD = Thread(target=_export_census_data, args=(exporter,))
GLOBAL_EXPORT_THREAD.start()
def activate_config(object py_config) -> None:
py_config: "_observability_config.GcpObservabilityConfig"
def set_gcp_observability_config(object py_config) -> bool:
py_config: _gcp_observability.GcpObservabilityPythonConfig
py_labels = {}
sampling_rate = 0.0
cdef cGcpObservabilityConfig c_config = ReadAndActivateObservabilityConfig()
if not c_config.is_valid:
return False
for label in c_config.labels:
py_labels[_decode(label.key)] = _decode(label.value)
if PythonCensusTracingEnabled():
sampling_rate = c_config.cloud_trace.sampling_rate
if (py_config.tracing_enabled):
EnablePythonCensusTracing(True);
# Save sampling rate to global sampler.
ProbabilitySampler.Get().SetThreshold(sampling_rate)
ProbabilitySampler.Get().SetThreshold(py_config.sampling_rate)
py_config.set_configuration(_decode(c_config.project_id), sampling_rate, py_labels,
PythonCensusTracingEnabled(), PythonCensusStatsEnabled())
return True
if (py_config.stats_enabled):
EnablePythonCensusStats(True);
def create_client_call_tracer(bytes method_name, bytes trace_id,

98
src/python/grpcio_observability/grpc_observability/_gcp_observability.py
View File

@ -13,20 +13,15 @@
# limitations under the License.
from __future__ import annotations
from dataclasses import dataclass
from dataclasses import field
import logging
import threading
import time
from typing import Any, Mapping, Optional
from typing import Any
import grpc
from grpc_observability import _cyobservability # pytype: disable=pyi-error
from grpc_observability._open_census_exporter import CENSUS_UPLOAD_INTERVAL_SECS
from grpc_observability._open_census_exporter import OpenCensusExporter
from opencensus.trace import execution_context
from opencensus.trace import span_context as span_context_module
from opencensus.trace import trace_options as trace_options_module
# pytype: disable=pyi-error
from grpc_observability import _cyobservability
from grpc_observability import _observability_config
_LOGGER = logging.getLogger(__name__)
@ -56,42 +51,6 @@ GRPC_STATUS_CODE_TO_STRING = {
grpc.StatusCode.DATA_LOSS: "DATA_LOSS",
}
GRPC_SPAN_CONTEXT = "grpc_span_context"
@dataclass
class GcpObservabilityPythonConfig:
_singleton = None
_lock: threading.RLock = threading.RLock()
project_id: str = ""
stats_enabled: bool = False
tracing_enabled: bool = False
labels: Optional[Mapping[str, str]] = field(default_factory=dict)
sampling_rate: Optional[float] = 0.0
@staticmethod
def get():
with GcpObservabilityPythonConfig._lock:
if GcpObservabilityPythonConfig._singleton is None:
GcpObservabilityPythonConfig._singleton = (
GcpObservabilityPythonConfig()
)
return GcpObservabilityPythonConfig._singleton
def set_configuration(
self,
project_id: str,
sampling_rate: Optional[float] = 0.0,
labels: Optional[Mapping[str, str]] = None,
tracing_enabled: bool = False,
stats_enabled: bool = False,
) -> None:
self.project_id = project_id
self.stats_enabled = stats_enabled
self.tracing_enabled = tracing_enabled
self.labels = labels
self.sampling_rate = sampling_rate
# pylint: disable=no-self-use
class GCPOpenCensusObservability(grpc._observability.ObservabilityPlugin):
@ -108,25 +67,22 @@ class GCPOpenCensusObservability(grpc._observability.ObservabilityPlugin):
exporter: Exporter used to export data.
"""
config: GcpObservabilityPythonConfig
config: _observability_config.GcpObservabilityConfig
exporter: "grpc_observability.Exporter"
use_open_census_exporter: bool
def __init__(self, exporter: "grpc_observability.Exporter" = None):
self.exporter = None
self.config = GcpObservabilityPythonConfig.get()
self.use_open_census_exporter = False
config_valid = _cyobservability.set_gcp_observability_config(
self.config
)
if not config_valid:
raise ValueError("Invalid configuration")
self.config = None
try:
self.config = _observability_config.read_config()
_cyobservability.activate_config(self.config)
except Exception as e: # pylint: disable=broad-except
raise ValueError(f"Reading configuration failed with: {e}")
if exporter:
self.exporter = exporter
else:
self.exporter = OpenCensusExporter(self.config)
self.use_open_census_exporter = True
raise ValueError(f"Please provide an exporter!")
if self.config.tracing_enabled:
self.set_tracing(True)
@ -156,9 +112,6 @@ class GCPOpenCensusObservability(grpc._observability.ObservabilityPlugin):
# TODO(xuanwn): explicit synchronization
# https://github.com/grpc/grpc/issues/33262
time.sleep(_cyobservability.CENSUS_EXPORT_BATCH_INTERVAL_SECS)
if self.use_open_census_exporter:
# Sleep so StackDriver can upload data to GCP.
time.sleep(CENSUS_UPLOAD_INTERVAL_SECS)
self.set_tracing(False)
self.set_stats(False)
_cyobservability.observability_deinit()
@ -167,20 +120,10 @@ class GCPOpenCensusObservability(grpc._observability.ObservabilityPlugin):
def create_client_call_tracer(
self, method_name: bytes
) -> ClientCallTracerCapsule:
grpc_span_context = execution_context.get_opencensus_attr(
GRPC_SPAN_CONTEXT
trace_id = b"TRACE_ID"
capsule = _cyobservability.create_client_call_tracer(
method_name, trace_id
)
if grpc_span_context:
trace_id = grpc_span_context.trace_id.encode("utf8")
parent_span_id = grpc_span_context.span_id.encode("utf8")
capsule = _cyobservability.create_client_call_tracer(
method_name, trace_id, parent_span_id
)
else:
trace_id = span_context_module.generate_trace_id().encode("utf8")
capsule = _cyobservability.create_client_call_tracer(
method_name, trace_id
)
return capsule
def create_server_call_tracer_factory(
@ -197,14 +140,7 @@ class GCPOpenCensusObservability(grpc._observability.ObservabilityPlugin):
def save_trace_context(
self, trace_id: str, span_id: str, is_sampled: bool
) -> None:
trace_options = trace_options_module.TraceOptions(0)
trace_options.set_enabled(is_sampled)
span_context = span_context_module.SpanContext(
trace_id=trace_id,
span_id=span_id,
trace_options=trace_options,
)
execution_context.set_opencensus_attr(GRPC_SPAN_CONTEXT, span_context)
pass
def record_rpc_latency(
self, method: str, rpc_latency: float, status_code: grpc.StatusCode

129
src/python/grpcio_observability/grpc_observability/_observability_config.py Normal file
View File

@ -0,0 +1,129 @@
# 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.
"""Helper to read observability config."""
from dataclasses import dataclass
from dataclasses import field
import json
import os
from typing import Mapping, Optional
GRPC_GCP_OBSERVABILITY_CONFIG_FILE_ENV = "GRPC_GCP_OBSERVABILITY_CONFIG_FILE"
GRPC_GCP_OBSERVABILITY_CONFIG_ENV = "GRPC_GCP_OBSERVABILITY_CONFIG"
@dataclass
class GcpObservabilityConfig:
project_id: str = ""
stats_enabled: bool = False
tracing_enabled: bool = False
labels: Optional[Mapping[str, str]] = field(default_factory=dict)
sampling_rate: Optional[float] = 0.0
def load_from_string_content(self, config_contents: str) -> None:
"""Loads the configuration from a string.
Args:
config_contents: The configuration string.
Raises:
ValueError: If the configuration is invalid.
"""
try:
config_json = json.loads(config_contents)
except json.decoder.JSONDecodeError:
raise ValueError("Failed to load Json configuration.")
if config_json and not isinstance(config_json, dict):
raise ValueError("Found invalid configuration.")
self.project_id = config_json.get("project_id", "")
self.labels = config_json.get("labels", {})
self.stats_enabled = "cloud_monitoring" in config_json.keys()
self.tracing_enabled = "cloud_trace" in config_json.keys()
tracing_config = config_json.get("cloud_trace", {})
self.sampling_rate = tracing_config.get("sampling_rate", 0.0)
def read_config() -> GcpObservabilityConfig:
"""Reads the GCP observability config from the environment variables.
Returns:
The GCP observability config.
Raises:
ValueError: If the configuration is invalid.
"""
config_contents = _get_gcp_observability_config_contents()
config = GcpObservabilityConfig()
config.load_from_string_content(config_contents)
if not config.project_id:
# Get project ID from GCP environment variables since project ID was not
# set it in the GCP observability config.
config.project_id = _get_gcp_project_id_from_env_var()
if not config.project_id:
# Could not find project ID from GCP environment variables either.
raise ValueError("GCP Project ID not found.")
return config
def _get_gcp_project_id_from_env_var() -> Optional[str]:
"""Gets the project ID from the GCP environment variables.
Returns:
The project ID, or an empty string if the project ID could not be found.
"""
project_id = ""
project_id = os.getenv("GCP_PROJECT")
if project_id:
return project_id
project_id = os.getenv("GCLOUD_PROJECT")
if project_id:
return project_id
project_id = os.getenv("GOOGLE_CLOUD_PROJECT")
if project_id:
return project_id
return project_id
def _get_gcp_observability_config_contents() -> str:
"""Get the contents of the observability config from environment variable or file.
Returns:
The content from environment variable.
Raises:
ValueError: If no configuration content was found.
"""
contents_str = ""
# First try get config from GRPC_GCP_OBSERVABILITY_CONFIG_FILE_ENV.
config_path = os.getenv(GRPC_GCP_OBSERVABILITY_CONFIG_FILE_ENV)
if config_path:
with open(config_path, "r") as f:
contents_str = f.read()
# Next, try GRPC_GCP_OBSERVABILITY_CONFIG_ENV env var.
if not contents_str:
contents_str = os.getenv(GRPC_GCP_OBSERVABILITY_CONFIG_ENV)
if not contents_str:
raise ValueError("Configuration content not found.")
return contents_str

11
src/python/grpcio_observability/grpc_observability/_open_census_exporter.py
View File

@ -14,10 +14,11 @@
from datetime import datetime
import os
from typing import Any, List, Mapping, Optional, Tuple
from typing import List, Mapping, Optional, Tuple
from google.rpc import code_pb2
from grpc_observability import _observability # pytype: disable=pyi-error
from grpc_observability import _observability_config
from grpc_observability import _views
from opencensus.common.transports import async_
from opencensus.ext.stackdriver import stats_exporter
@ -38,8 +39,6 @@ from opencensus.trace import time_event
from opencensus.trace import trace_options
from opencensus.trace import tracer
_gcp_observability = Any # grpc_observability.py imports this module.
# 60s is the default time for open census to call export.
CENSUS_UPLOAD_INTERVAL_SECS = int(
os.environ.get("GRPC_PYTHON_CENSUS_EXPORT_UPLOAD_INTERVAL_SECS", 20)
@ -61,16 +60,14 @@ class StackDriverAsyncTransport(async_.AsyncTransport):
class OpenCensusExporter(_observability.Exporter):
config: "_gcp_observability.GcpObservabilityPythonConfig"
config: _observability_config.GcpObservabilityConfig
default_labels: Optional[Mapping[str, str]]
project_id: str
tracer: Optional[tracer.Tracer]
stats_recorder: Optional[StatsRecorder]
view_manager: Optional[ViewManager]
def __init__(
self, config: "_gcp_observability.GcpObservabilityPythonConfig"
):
def __init__(self, config: _observability_config.GcpObservabilityConfig):
self.config = config.get()
self.default_labels = self.config.labels
self.project_id = self.config.project_id

27
src/python/grpcio_observability/grpc_observability/client_call_tracer.cc
View File

@ -12,24 +12,21 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#include "src/python/grpcio_observability/grpc_observability/client_call_tracer.h"
#include "client_call_tracer.h"
#include <constants.h>
#include <observability_util.h>
#include <python_census_context.h>
#include <stddef.h>
#include <algorithm>
#include <vector>
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "absl/time/clock.h"
#include "constants.h"
#include "observability_util.h"
#include "python_census_context.h"
#include <grpc/slice.h>
#include "src/core/lib/experiments/experiments.h"
#include "src/core/lib/gprpp/sync.h"
#include "src/core/lib/slice/slice.h"
namespace grpc_observability {
@ -56,7 +53,7 @@ void PythonOpenCensusCallTracer::GenerateContext() {}
void PythonOpenCensusCallTracer::RecordAnnotation(
absl::string_view annotation) {
if (!context_.SpanContext().IsSampled()) {
if (!context_.GetSpanContext().IsSampled()) {
return;
}
context_.AddSpanAnnotation(annotation);
@ -64,7 +61,7 @@ void PythonOpenCensusCallTracer::RecordAnnotation(
void PythonOpenCensusCallTracer::RecordAnnotation(
const Annotation& annotation) {
if (!context_.SpanContext().IsSampled()) {
if (!context_.GetSpanContext().IsSampled()) {
return;
}
@ -93,7 +90,7 @@ PythonOpenCensusCallTracer::~PythonOpenCensusCallTracer() {
if (tracing_enabled_) {
context_.EndSpan();
if (IsSampled()) {
RecordSpan(context_.Span().ToCensusData());
RecordSpan(context_.GetSpan().ToCensusData());
}
}
}
@ -101,7 +98,7 @@ PythonOpenCensusCallTracer::~PythonOpenCensusCallTracer() {
PythonCensusContext
PythonOpenCensusCallTracer::CreateCensusContextForCallAttempt() {
auto context = PythonCensusContext(absl::StrCat("Attempt.", method_),
&(context_.Span()), context_.Labels());
&(context_.GetSpan()), context_.Labels());
return context;
}
@ -274,11 +271,11 @@ void PythonOpenCensusCallTracer::PythonOpenCensusCallAttemptTracer::RecordEnd(
if (parent_->tracing_enabled_) {
if (status_code_ != absl::StatusCode::kOk) {
context_.Span().SetStatus(StatusCodeToString(status_code_));
context_.GetSpan().SetStatus(StatusCodeToString(status_code_));
}
context_.EndSpan();
if (IsSampled()) {
RecordSpan(context_.Span().ToCensusData());
RecordSpan(context_.GetSpan().ToCensusData());
}
}
@ -289,7 +286,7 @@ void PythonOpenCensusCallTracer::PythonOpenCensusCallAttemptTracer::RecordEnd(
void PythonOpenCensusCallTracer::PythonOpenCensusCallAttemptTracer::
RecordAnnotation(absl::string_view annotation) {
if (!context_.SpanContext().IsSampled()) {
if (!context_.GetSpanContext().IsSampled()) {
return;
}
context_.AddSpanAnnotation(annotation);
@ -297,7 +294,7 @@ void PythonOpenCensusCallTracer::PythonOpenCensusCallAttemptTracer::
void PythonOpenCensusCallTracer::PythonOpenCensusCallAttemptTracer::
RecordAnnotation(const Annotation& annotation) {
if (!context_.SpanContext().IsSampled()) {
if (!context_.GetSpanContext().IsSampled()) {
return;
}

25
src/python/grpcio_observability/grpc_observability/client_call_tracer.h
View File

@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef GRPC_PYRHON_OPENCENSUS_CLIENT_CALL_TRACER_H
#define GRPC_PYRHON_OPENCENSUS_CLIENT_CALL_TRACER_H
#ifndef GRPC_PYTHON_OPENCENSUS_CLIENT_CALL_TRACER_H
#define GRPC_PYTHON_OPENCENSUS_CLIENT_CALL_TRACER_H
#include <stdint.h>
@ -24,16 +24,11 @@
#include "absl/strings/escaping.h"
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "python_census_context.h"
#include <grpc/support/time.h>
#include "src/core/lib/channel/call_tracer.h"
#include "src/core/lib/gprpp/sync.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/slice/slice_buffer.h"
#include "src/core/lib/transport/metadata_batch.h"
#include "src/core/lib/transport/transport.h"
#include "src/python/grpcio_observability/grpc_observability/python_census_context.h"
namespace grpc_observability {
@ -46,15 +41,15 @@ class PythonOpenCensusCallTracer : public grpc_core::ClientCallTracer {
bool is_transparent_retry);
std::string TraceId() override {
return absl::BytesToHexString(
absl::string_view(context_.SpanContext().TraceId()));
absl::string_view(context_.GetSpanContext().TraceId()));
}
std::string SpanId() override {
return absl::BytesToHexString(
absl::string_view(context_.SpanContext().SpanId()));
absl::string_view(context_.GetSpanContext().SpanId()));
}
bool IsSampled() override { return context_.SpanContext().IsSampled(); }
bool IsSampled() override { return context_.GetSpanContext().IsSampled(); }
void RecordSendInitialMetadata(
grpc_metadata_batch* send_initial_metadata) override;
@ -102,15 +97,15 @@ class PythonOpenCensusCallTracer : public grpc_core::ClientCallTracer {
std::string TraceId() override {
return absl::BytesToHexString(
absl::string_view(context_.SpanContext().TraceId()));
absl::string_view(context_.GetSpanContext().TraceId()));
}
std::string SpanId() override {
return absl::BytesToHexString(
absl::string_view(context_.SpanContext().SpanId()));
absl::string_view(context_.GetSpanContext().SpanId()));
}
bool IsSampled() override { return context_.SpanContext().IsSampled(); }
bool IsSampled() override { return context_.GetSpanContext().IsSampled(); }
void GenerateContext();
PythonOpenCensusCallAttemptTracer* StartNewAttempt(
@ -139,4 +134,4 @@ class PythonOpenCensusCallTracer : public grpc_core::ClientCallTracer {
} // namespace grpc_observability
#endif // GRPC_PYRHON_OPENCENSUS_CLIENT_CALL_TRACER_H
#endif // GRPC_PYTHON_OPENCENSUS_CLIENT_CALL_TRACER_H

63
src/python/grpcio_observability/grpc_observability/observability_util.cc
View File

@ -12,10 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#include "src/python/grpcio_observability/grpc_observability/observability_util.h"
#include <constants.h>
#include <python_census_context.h>
#include "observability_util.h"
#include <chrono>
#include <cstdlib>
@ -25,13 +22,13 @@
#include "absl/status/statusor.h"
#include "absl/strings/string_view.h"
#include "absl/types/optional.h"
#include "client_call_tracer.h"
#include "constants.h"
#include "python_census_context.h"
#include "server_call_tracer.h"
#include <grpc/support/log.h>
#include "src/cpp/ext/gcp/observability_config.h"
#include "src/python/grpcio_observability/grpc_observability/client_call_tracer.h"
#include "src/python/grpcio_observability/grpc_observability/server_call_tracer.h"
namespace grpc_observability {
std::queue<CensusData>* g_census_data_buffer;
@ -128,56 +125,6 @@ void AddCensusDataToBuffer(const CensusData& data) {
}
}
GcpObservabilityConfig ReadAndActivateObservabilityConfig() {
auto config = grpc::internal::GcpObservabilityConfig::ReadFromEnv();
if (!config.ok()) {
return GcpObservabilityConfig();
}
if (!config->cloud_trace.has_value() &&
!config->cloud_monitoring.has_value() &&
!config->cloud_logging.has_value()) {
return GcpObservabilityConfig(true);
}
if (config->cloud_trace.has_value()) {
EnablePythonCensusTracing(true);
}
if (config->cloud_monitoring.has_value()) {
EnablePythonCensusStats(true);
}
std::vector<Label> labels;
std::string project_id = config->project_id;
CloudMonitoring cloud_monitoring_config = CloudMonitoring();
CloudTrace cloud_trace_config = CloudTrace();
CloudLogging cloud_logging_config = CloudLogging();
if (config->cloud_trace.has_value() || config->cloud_monitoring.has_value()) {
labels.reserve(config->labels.size());
// Insert in user defined labels from the GCP Observability config.
for (const auto& label : config->labels) {
labels.emplace_back(label.first, label.second);
}
if (config->cloud_trace.has_value()) {
double sampleRate = config->cloud_trace->sampling_rate;
cloud_trace_config = CloudTrace(sampleRate);
}
if (config->cloud_monitoring.has_value()) {
cloud_monitoring_config = CloudMonitoring();
}
}
// Clound logging
if (config->cloud_logging.has_value()) {
// TODO(xuanwn): Read cloud logging config
}
return GcpObservabilityConfig(cloud_monitoring_config, cloud_trace_config,
cloud_logging_config, project_id, labels);
}
absl::string_view StatusCodeToString(grpc_status_code code) {
switch (code) {
case GRPC_STATUS_OK:

42
src/python/grpcio_observability/grpc_observability/observability_util.h
View File

@ -26,12 +26,11 @@
#include <vector>
#include "absl/strings/string_view.h"
#include "constants.h"
#include "python_census_context.h"
#include <grpc/status.h>
#include "src/python/grpcio_observability/grpc_observability/constants.h"
#include "src/python/grpcio_observability/grpc_observability/python_census_context.h"
namespace grpc_observability {
struct CensusData {
@ -46,41 +45,6 @@ struct CensusData {
CensusData(const SpanCensusData& sd) : type(kSpanData), span_data(sd) {}
};
struct CloudMonitoring {
CloudMonitoring() {}
};
struct CloudTrace {
float sampling_rate = 0.0;
CloudTrace() {}
CloudTrace(double sr) : sampling_rate(sr) {}
};
struct CloudLogging {
CloudLogging() {}
};
struct GcpObservabilityConfig {
CloudMonitoring cloud_monitoring;
CloudTrace cloud_trace;
CloudLogging cloud_logging;
std::string project_id;
std::vector<Label> labels;
bool is_valid;
GcpObservabilityConfig() : is_valid(false) {}
GcpObservabilityConfig(bool valid) : is_valid(true) {}
GcpObservabilityConfig(CloudMonitoring cloud_monitoring,
CloudTrace cloud_trace, CloudLogging cloud_logging,
const std::string& project_id,
const std::vector<Label>& labels)
: cloud_monitoring(cloud_monitoring),
cloud_trace(cloud_trace),
cloud_logging(cloud_logging),
project_id(project_id),
labels(labels),
is_valid(true) {}
};
// extern is required for Cython
extern std::queue<CensusData>* g_census_data_buffer;
extern std::mutex g_census_data_buffer_mutex;
@ -105,8 +69,6 @@ void RecordDoubleMetric(MetricsName name, double value,
void RecordSpan(const SpanCensusData& span_census_data);
GcpObservabilityConfig ReadAndActivateObservabilityConfig();
absl::string_view StatusCodeToString(grpc_status_code code);
} // namespace grpc_observability

17
src/python/grpcio_observability/grpc_observability/python_census_context.cc
View File

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#include "src/python/grpcio_observability/grpc_observability/python_census_context.h"
#include "python_census_context.h"
#include <string.h>
@ -23,9 +23,9 @@
#include "absl/numeric/int128.h"
#include "absl/random/random.h"
#include "absl/strings/escaping.h"
#include "rpc_encoding.h"
#include "src/core/lib/transport/transport.h"
#include "src/cpp/ext/filters/census/rpc_encoding.h"
namespace grpc_observability {
@ -90,10 +90,10 @@ void ToGrpcTraceBinHeader(const PythonCensusContext& ctx, uint8_t* out) {
uint8_t trace_options_rep_[kSizeTraceOptions];
std::string trace_id =
absl::HexStringToBytes(absl::string_view(ctx.SpanContext().TraceId()));
absl::HexStringToBytes(absl::string_view(ctx.GetSpanContext().TraceId()));
std::string span_id =
absl::HexStringToBytes(absl::string_view(ctx.SpanContext().SpanId()));
trace_options_rep_[0] = ctx.SpanContext().IsSampled() ? 1 : 0;
absl::HexStringToBytes(absl::string_view(ctx.GetSpanContext().SpanId()));
trace_options_rep_[0] = ctx.GetSpanContext().IsSampled() ? 1 : 0;
memcpy(reinterpret_cast<uint8_t*>(&out[kTraceIdOfs + 1]), trace_id.c_str(),
kSizeTraceID);
@ -156,14 +156,13 @@ size_t StatsContextSerialize(size_t /*max_tags_len*/, grpc_slice* /*tags*/) {
size_t ServerStatsDeserialize(const char* buf, size_t buf_size,
uint64_t* server_elapsed_time) {
return grpc::internal::RpcServerStatsEncoding::Decode(
absl::string_view(buf, buf_size), server_elapsed_time);
return RpcServerStatsEncoding::Decode(absl::string_view(buf, buf_size),
server_elapsed_time);
}
size_t ServerStatsSerialize(uint64_t server_elapsed_time, char* buf,
size_t buf_size) {
return grpc::internal::RpcServerStatsEncoding::Encode(server_elapsed_time,
buf, buf_size);
return RpcServerStatsEncoding::Encode(server_elapsed_time, buf, buf_size);
}
uint64_t GetIncomingDataSize(const grpc_call_final_info* final_info) {

16
src/python/grpcio_observability/grpc_observability/python_census_context.h
View File

@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef GRPC_PYRHON_OPENCENSUS_H
#define GRPC_PYRHON_OPENCENSUS_H
#ifndef GRPC_PYTHON_OPENCENSUS_H
#define GRPC_PYTHON_OPENCENSUS_H
#include <stddef.h>
#include <stdint.h>
@ -27,12 +27,12 @@
#include "absl/strings/strip.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "constants.h"
#include "sampler.h"
#include <grpc/slice.h>
#include "src/core/lib/channel/channel_stack.h"
#include "src/python/grpcio_observability/grpc_observability/constants.h"
#include "src/python/grpcio_observability/grpc_observability/sampler.h"
namespace grpc_observability {
@ -235,9 +235,9 @@ class PythonCensusContext {
PythonCensusContext(absl::string_view name, const Span* parent)
: span_(Span::StartSpan(name, parent)), labels_({}) {}
Span& Span() { return span_; }
Span& GetSpan() { return span_; }
std::vector<Label>& Labels() { return labels_; } // Only used for metrics
const SpanContext& SpanContext() const { return span_.Context(); }
const SpanContext& GetSpanContext() const { return span_.Context(); }
void AddSpanAttribute(absl::string_view key, absl::string_view attribute) {
span_.AddAttribute(key, attribute);
@ -249,7 +249,7 @@ class PythonCensusContext {
void IncreaseChildSpanCount() { span_.IncreaseChildSpanCount(); }
void EndSpan() { Span().End(); }
void EndSpan() { GetSpan().End(); }
private:
grpc_observability::Span span_;
@ -324,4 +324,4 @@ uint64_t GetOutgoingDataSize(const grpc_call_final_info* final_info);
} // namespace grpc_observability
#endif // GRPC_PYRHON_OPENCENSUS_H
#endif // GRPC_PYTHON_OPENCENSUS_H

29
src/python/grpcio_observability/grpc_observability/rpc_encoding.cc Normal file
View File

@ -0,0 +1,29 @@
// 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 <grpc/support/port_platform.h>
#include "rpc_encoding.h"
// TODO(xuanwn): Reuse c++ rpc_encoding file.
namespace grpc_observability {
constexpr size_t RpcServerStatsEncoding::kRpcServerStatsSize;
constexpr size_t RpcServerStatsEncoding::kEncodeDecodeFailure;
constexpr size_t RpcServerStatsEncoding::kVersionIdSize;
constexpr size_t RpcServerStatsEncoding::kFieldIdSize;
constexpr size_t RpcServerStatsEncoding::kVersionIdOffset;
constexpr size_t RpcServerStatsEncoding::kVersionId;
} // namespace grpc_observability

108
src/python/grpcio_observability/grpc_observability/rpc_encoding.h Normal file
View File

@ -0,0 +1,108 @@
// 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.
#ifndef GRPC_PYTHON_OPENCENSUS_RPC_ENCODING_H
#define GRPC_PYTHON_OPENCENSUS_RPC_ENCODING_H
#include <grpc/support/port_platform.h>
#include <stdint.h>
#include <string.h>
#include "absl/base/internal/endian.h"
#include "absl/strings/string_view.h"
namespace grpc_observability {
// TODO(xuanwn): Reuse c++ rpc_encoding file.
// RpcServerStatsEncoding encapsulates the logic for encoding and decoding of
// rpc server stats messages. Rpc server stats consists of a uint64_t time
// value (server latency in nanoseconds).
class RpcServerStatsEncoding {
public:
// Size of encoded RPC server stats.
static constexpr size_t kRpcServerStatsSize = 10;
// Error value.
static constexpr size_t kEncodeDecodeFailure = 0;
// Deserializes rpc server stats from the incoming 'buf' into *time. Returns
// number of bytes decoded. If the buffer is of insufficient size (it must be
// at least kRpcServerStatsSize bytes) or the encoding version or field ID are
// unrecognized, *time will be set to 0 and it will return
// kEncodeDecodeFailure. Inlined for performance reasons.
static size_t Decode(absl::string_view buf, uint64_t* time) {
if (buf.size() < kRpcServerStatsSize) {
*time = 0;
return kEncodeDecodeFailure;
}
uint8_t version = buf[kVersionIdOffset];
uint32_t fieldID = buf[kServerElapsedTimeOffset];
if (version != kVersionId || fieldID != kServerElapsedTimeField) {
*time = 0;
return kEncodeDecodeFailure;
}
*time = absl::little_endian::Load64(
&buf[kServerElapsedTimeOffset + kFieldIdSize]);
return kRpcServerStatsSize;
}
// Serializes rpc server stats into the provided buffer. It returns the
// number of bytes written to the buffer. If the buffer is smaller than
// kRpcServerStatsSize bytes it will return kEncodeDecodeFailure. Inlined for
// performance reasons.
static size_t Encode(uint64_t time, char* buf, size_t buf_size) {
if (buf_size < kRpcServerStatsSize) {
return kEncodeDecodeFailure;
}
buf[kVersionIdOffset] = kVersionId;
buf[kServerElapsedTimeOffset] = kServerElapsedTimeField;
absl::little_endian::Store64(&buf[kServerElapsedTimeOffset + kFieldIdSize],
time);
return kRpcServerStatsSize;
}
private:
// Size of Version ID.
static constexpr size_t kVersionIdSize = 1;
// Size of Field ID.
static constexpr size_t kFieldIdSize = 1;
// Offset and value for currently supported version ID.
static constexpr size_t kVersionIdOffset = 0;
static constexpr size_t kVersionId = 0;
enum FieldIdValue {
kServerElapsedTimeField = 0,
};
enum FieldSize {
kServerElapsedTimeSize = 8,
};
enum FieldIdOffset {
kServerElapsedTimeOffset = kVersionIdSize,
};
RpcServerStatsEncoding() = delete;
RpcServerStatsEncoding(const RpcServerStatsEncoding&) = delete;
RpcServerStatsEncoding(RpcServerStatsEncoding&&) = delete;
RpcServerStatsEncoding operator=(const RpcServerStatsEncoding&) = delete;
RpcServerStatsEncoding operator=(RpcServerStatsEncoding&&) = delete;
};
} // namespace grpc_observability
#endif // GRPC_PYTHON_OPENCENSUS_RPC_ENCODING_H

2
src/python/grpcio_observability/grpc_observability/sampler.cc
View File

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#include "src/python/grpcio_observability/grpc_observability/sampler.h"
#include "sampler.h"
#include <cmath>
#include <cstdint>

23
src/python/grpcio_observability/grpc_observability/server_call_tracer.cc
View File

@ -12,12 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#include "src/python/grpcio_observability/grpc_observability/server_call_tracer.h"
// TODO(xuanwn): clean up includes
#include <grpc/support/port_platform.h>
#include <constants.h>
#include "server_call_tracer.h"
#include <stdint.h>
#include <string.h>
@ -33,6 +31,9 @@
#include "absl/strings/string_view.h"
#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "constants.h"
#include "observability_util.h"
#include "python_census_context.h"
#include "src/core/lib/channel/call_tracer.h"
#include "src/core/lib/channel/channel_stack.h"
@ -41,8 +42,6 @@
#include "src/core/lib/slice/slice.h"
#include "src/core/lib/slice/slice_buffer.h"
#include "src/core/lib/transport/metadata_batch.h"
#include "src/python/grpcio_observability/grpc_observability/observability_util.h"
#include "src/python/grpcio_observability/grpc_observability/python_census_context.h"
namespace grpc_observability {
@ -94,15 +93,15 @@ class PythonOpenCensusServerCallTracer : public grpc_core::ServerCallTracer {
std::string TraceId() override {
return absl::BytesToHexString(
absl::string_view(context_.SpanContext().TraceId()));
absl::string_view(context_.GetSpanContext().TraceId()));
}
std::string SpanId() override {
return absl::BytesToHexString(
absl::string_view(context_.SpanContext().SpanId()));
absl::string_view(context_.GetSpanContext().SpanId()));
}
bool IsSampled() override { return context_.SpanContext().IsSampled(); }
bool IsSampled() override { return context_.GetSpanContext().IsSampled(); }
// Please refer to `grpc_transport_stream_op_batch_payload` for details on
// arguments.
@ -150,14 +149,14 @@ class PythonOpenCensusServerCallTracer : public grpc_core::ServerCallTracer {
void RecordEnd(const grpc_call_final_info* final_info) override;
void RecordAnnotation(absl::string_view annotation) override {
if (!context_.SpanContext().IsSampled()) {
if (!context_.GetSpanContext().IsSampled()) {
return;
}
context_.AddSpanAnnotation(annotation);
}
void RecordAnnotation(const Annotation& annotation) override {
if (!context_.SpanContext().IsSampled()) {
if (!context_.GetSpanContext().IsSampled()) {
return;
}
@ -247,7 +246,7 @@ void PythonOpenCensusServerCallTracer::RecordEnd(
if (PythonCensusTracingEnabled()) {
context_.EndSpan();
if (IsSampled()) {
RecordSpan(context_.Span().ToCensusData());
RecordSpan(context_.GetSpan().ToCensusData());
}
}

6
src/python/grpcio_observability/grpc_observability/server_call_tracer.h
View File

@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#ifndef GRPC_PYRHON_OPENCENSUS_SERVER_CALL_TRACER_H
#define GRPC_PYRHON_OPENCENSUS_SERVER_CALL_TRACER_H
#ifndef GRPC_PYTHON_OPENCENSUS_SERVER_CALL_TRACER_H
#define GRPC_PYTHON_OPENCENSUS_SERVER_CALL_TRACER_H
#include <grpc/support/port_platform.h>
@ -43,4 +43,4 @@ inline absl::string_view GetMethod(const grpc_core::Slice& path) {
} // namespace grpc_observability
#endif // GRPC_PYRHON_OPENCENSUS_SERVER_CALL_TRACER_H
#endif // GRPC_PYTHON_OPENCENSUS_SERVER_CALL_TRACER_H

17
src/python/grpcio_observability/grpc_version.py Normal file
View File

@ -0,0 +1,17 @@
# 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.
# AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio_observability/grpc_version.py.template`!!!
VERSION = '1.61.0.dev0'

215
src/python/grpcio_observability/make_grpcio_observability.py Executable file
View File

@ -0,0 +1,215 @@
#!/usr/bin/env python3
# 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.
import errno
import os
import os.path
import pprint
import shutil
import subprocess
import sys
import traceback
# the template for the content of observability_lib_deps.py
DEPS_FILE_CONTENT = """
# 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.
# AUTO-GENERATED BY make_grpcio_observability.py!
CC_FILES={cc_files}
CC_INCLUDES={cc_includes}
"""
# maps bazel reference to actual path
BAZEL_REFERENCE_LINK = [
("@com_google_absl//", "third_party/abseil-cpp/"),
("//src", "grpc_root/src"),
]
ABSL_INCLUDE = (os.path.join("third_party", "abseil-cpp"),)
# will be added to include path when building grpcio_observability
EXTENSION_INCLUDE_DIRECTORIES = ABSL_INCLUDE
CC_INCLUDES = list(EXTENSION_INCLUDE_DIRECTORIES)
# the target directory is relative to the grpcio_observability package root.
GRPCIO_OBSERVABILITY_ROOT_PREFIX = "src/python/grpcio_observability/"
# Pairs of (source, target) directories to copy
# from the grpc repo root to the grpcio_observability build root.
COPY_FILES_SOURCE_TARGET_PAIRS = [
("include", "grpc_root/include"),
("third_party/abseil-cpp/absl", "third_party/abseil-cpp/absl"),
("src/core/lib", "grpc_root/src/core/lib"),
(
"src/core/ext/filters/client_channel/lb_policy",
"grpc_root/src/core/ext/filters/client_channel/lb_policy",
),
]
# grpc repo root
GRPC_ROOT = os.path.abspath(
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "..")
)
# the file to generate
GRPC_PYTHON_OBSERVABILITY_LIB_DEPS = os.path.join(
GRPC_ROOT,
"src",
"python",
"grpcio_observability",
"observability_lib_deps.py",
)
# the script to run for getting dependencies
BAZEL_DEPS = os.path.join(
GRPC_ROOT, "tools", "distrib", "python", "bazel_deps.sh"
)
# the bazel target to scrape to get list of sources for the build
BAZEL_DEPS_QUERIES = [
"//src/core:slice",
]
def _bazel_query(query):
"""Runs 'bazel query' to collect source file info."""
print('Running "bazel query %s"' % query)
output = subprocess.check_output([BAZEL_DEPS, query])
return output.decode("ascii").splitlines()
def _pretty_print_list(items):
"""Pretty print python list"""
formatted = pprint.pformat(items, indent=4)
# add newline after opening bracket (and fix indent of the next line)
if formatted.startswith("["):
formatted = formatted[0] + "\n " + formatted[1:]
# add newline before closing bracket
if formatted.endswith("]"):
formatted = formatted[:-1] + "\n" + formatted[-1]
return formatted
def _bazel_name_to_file_path(name):
"""Transform bazel reference to source file name."""
for link in BAZEL_REFERENCE_LINK:
if name.startswith(link[0]):
filepath = link[1] + name[len(link[0]) :].replace(":", "/")
return filepath
return None
def _generate_deps_file_content():
"""Returns the data structure with dependencies of protoc as python code."""
cc_files_output = []
for query in BAZEL_DEPS_QUERIES:
cc_files_output += _bazel_query(query)
# Collect .cc files (that will be later included in the native extension build)
cc_files = set()
for name in cc_files_output:
if name.endswith(".cc"):
filepath = _bazel_name_to_file_path(name)
if filepath:
cc_files.add(filepath)
deps_file_content = DEPS_FILE_CONTENT.format(
cc_files=_pretty_print_list(sorted(list(cc_files))),
cc_includes=_pretty_print_list(CC_INCLUDES),
)
return deps_file_content
def _copy_source_tree(source, target):
"""Copies source directory to a given target directory."""
print("Copying contents of %s to %s" % (source, target))
for source_dir, _, files in os.walk(source):
target_dir = os.path.abspath(
os.path.join(target, os.path.relpath(source_dir, source))
)
try:
os.makedirs(target_dir)
except OSError as error:
if error.errno != errno.EEXIST:
raise
for relative_file in files:
source_file = os.path.abspath(
os.path.join(source_dir, relative_file)
)
target_file = os.path.abspath(
os.path.join(target_dir, relative_file)
)
shutil.copyfile(source_file, target_file)
def main():
os.chdir(GRPC_ROOT)
# Step 1:
# In order to be able to build the grpcio_observability package, we need the source
# code for the plugins and its dependencies to be available under the build root of
# the grpcio_observability package.
# So we simply copy all the necessary files where the build will expect them to be.
for source, target in COPY_FILES_SOURCE_TARGET_PAIRS:
# convert the slashes in the relative path to platform-specific path dividers.
# All paths are relative to GRPC_ROOT
source_abs = os.path.join(GRPC_ROOT, os.path.join(*source.split("/")))
# for targets, add grpcio_observability root prefix
target = GRPCIO_OBSERVABILITY_ROOT_PREFIX + target
target_abs = os.path.join(GRPC_ROOT, os.path.join(*target.split("/")))
_copy_source_tree(source_abs, target_abs)
print(
"The necessary source files were copied under the grpcio_observability package root."
)
# Step 2:
# Extract build metadata from bazel build (by running "bazel query")
# and populate the observability_lib_deps.py file with python-readable data structure
# that will be used by grpcio_observability's setup.py.
try:
print('Invoking "bazel query" to gather the dependencies.')
observability_lib_deps_content = _generate_deps_file_content()
except Exception as error:
# We allow this script to succeed even if we couldn't get the dependencies,
# as then we can assume that even without a successful bazel run the
# dependencies currently in source control are 'good enough'.
sys.stderr.write("Got non-fatal error:\n")
traceback.print_exc(file=sys.stderr)
return
# If we successfully got the dependencies, truncate and rewrite the deps file.
with open(GRPC_PYTHON_OBSERVABILITY_LIB_DEPS, "w") as deps_file:
deps_file.write(observability_lib_deps_content)
print('File "%s" updated.' % GRPC_PYTHON_OBSERVABILITY_LIB_DEPS)
print("Done.")
if __name__ == "__main__":
main()

193
src/python/grpcio_observability/observability_lib_deps.py Normal file
View File

@ -0,0 +1,193 @@
# 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.
# AUTO-GENERATED BY make_grpcio_observability.py!
CC_FILES = [
"grpc_root/src/core/lib/config/config_vars.cc",
"grpc_root/src/core/lib/config/config_vars_non_generated.cc",
"grpc_root/src/core/lib/config/load_config.cc",
"grpc_root/src/core/lib/debug/trace.cc",
"grpc_root/src/core/lib/event_engine/thread_local.cc",
"grpc_root/src/core/lib/gpr/alloc.cc",
"grpc_root/src/core/lib/gpr/android/log.cc",
"grpc_root/src/core/lib/gpr/atm.cc",
"grpc_root/src/core/lib/gpr/iphone/cpu.cc",
"grpc_root/src/core/lib/gpr/linux/cpu.cc",
"grpc_root/src/core/lib/gpr/linux/log.cc",
"grpc_root/src/core/lib/gpr/log.cc",
"grpc_root/src/core/lib/gpr/msys/tmpfile.cc",
"grpc_root/src/core/lib/gpr/posix/cpu.cc",
"grpc_root/src/core/lib/gpr/posix/log.cc",
"grpc_root/src/core/lib/gpr/posix/string.cc",
"grpc_root/src/core/lib/gpr/posix/sync.cc",
"grpc_root/src/core/lib/gpr/posix/time.cc",
"grpc_root/src/core/lib/gpr/posix/tmpfile.cc",
"grpc_root/src/core/lib/gpr/string.cc",
"grpc_root/src/core/lib/gpr/sync.cc",
"grpc_root/src/core/lib/gpr/sync_abseil.cc",
"grpc_root/src/core/lib/gpr/time.cc",
"grpc_root/src/core/lib/gpr/time_precise.cc",
"grpc_root/src/core/lib/gpr/windows/cpu.cc",
"grpc_root/src/core/lib/gpr/windows/log.cc",
"grpc_root/src/core/lib/gpr/windows/string.cc",
"grpc_root/src/core/lib/gpr/windows/string_util.cc",
"grpc_root/src/core/lib/gpr/windows/sync.cc",
"grpc_root/src/core/lib/gpr/windows/time.cc",
"grpc_root/src/core/lib/gpr/windows/tmpfile.cc",
"grpc_root/src/core/lib/gpr/wrap_memcpy.cc",
"grpc_root/src/core/lib/gprpp/crash.cc",
"grpc_root/src/core/lib/gprpp/examine_stack.cc",
"grpc_root/src/core/lib/gprpp/fork.cc",
"grpc_root/src/core/lib/gprpp/host_port.cc",
"grpc_root/src/core/lib/gprpp/linux/env.cc",
"grpc_root/src/core/lib/gprpp/mpscq.cc",
"grpc_root/src/core/lib/gprpp/posix/env.cc",
"grpc_root/src/core/lib/gprpp/posix/stat.cc",
"grpc_root/src/core/lib/gprpp/posix/thd.cc",
"grpc_root/src/core/lib/gprpp/strerror.cc",
"grpc_root/src/core/lib/gprpp/tchar.cc",
"grpc_root/src/core/lib/gprpp/time_util.cc",
"grpc_root/src/core/lib/gprpp/windows/env.cc",
"grpc_root/src/core/lib/gprpp/windows/stat.cc",
"grpc_root/src/core/lib/gprpp/windows/thd.cc",
"grpc_root/src/core/lib/slice/slice.cc",
"grpc_root/src/core/lib/slice/slice_refcount.cc",
"grpc_root/src/core/lib/slice/slice_string_helpers.cc",
"third_party/abseil-cpp/absl/base/internal/cycleclock.cc",
"third_party/abseil-cpp/absl/base/internal/low_level_alloc.cc",
"third_party/abseil-cpp/absl/base/internal/raw_logging.cc",
"third_party/abseil-cpp/absl/base/internal/spinlock.cc",
"third_party/abseil-cpp/absl/base/internal/spinlock_wait.cc",
"third_party/abseil-cpp/absl/base/internal/strerror.cc",
"third_party/abseil-cpp/absl/base/internal/sysinfo.cc",
"third_party/abseil-cpp/absl/base/internal/thread_identity.cc",
"third_party/abseil-cpp/absl/base/internal/throw_delegate.cc",
"third_party/abseil-cpp/absl/base/internal/unscaledcycleclock.cc",
"third_party/abseil-cpp/absl/base/log_severity.cc",
"third_party/abseil-cpp/absl/container/internal/hashtablez_sampler.cc",
"third_party/abseil-cpp/absl/container/internal/hashtablez_sampler_force_weak_definition.cc",
"third_party/abseil-cpp/absl/container/internal/raw_hash_set.cc",
"third_party/abseil-cpp/absl/crc/crc32c.cc",
"third_party/abseil-cpp/absl/crc/internal/cpu_detect.cc",
"third_party/abseil-cpp/absl/crc/internal/crc.cc",
"third_party/abseil-cpp/absl/crc/internal/crc_cord_state.cc",
"third_party/abseil-cpp/absl/crc/internal/crc_memcpy_fallback.cc",
"third_party/abseil-cpp/absl/crc/internal/crc_memcpy_x86_64.cc",
"third_party/abseil-cpp/absl/crc/internal/crc_non_temporal_memcpy.cc",
"third_party/abseil-cpp/absl/crc/internal/crc_x86_arm_combined.cc",
"third_party/abseil-cpp/absl/debugging/internal/address_is_readable.cc",
"third_party/abseil-cpp/absl/debugging/internal/demangle.cc",
"third_party/abseil-cpp/absl/debugging/internal/elf_mem_image.cc",
"third_party/abseil-cpp/absl/debugging/internal/vdso_support.cc",
"third_party/abseil-cpp/absl/debugging/stacktrace.cc",
"third_party/abseil-cpp/absl/debugging/symbolize.cc",
"third_party/abseil-cpp/absl/flags/commandlineflag.cc",
"third_party/abseil-cpp/absl/flags/flag.cc",
"third_party/abseil-cpp/absl/flags/internal/commandlineflag.cc",
"third_party/abseil-cpp/absl/flags/internal/flag.cc",
"third_party/abseil-cpp/absl/flags/internal/private_handle_accessor.cc",
"third_party/abseil-cpp/absl/flags/internal/program_name.cc",
"third_party/abseil-cpp/absl/flags/marshalling.cc",
"third_party/abseil-cpp/absl/flags/reflection.cc",
"third_party/abseil-cpp/absl/flags/usage_config.cc",
"third_party/abseil-cpp/absl/hash/internal/city.cc",
"third_party/abseil-cpp/absl/hash/internal/hash.cc",
"third_party/abseil-cpp/absl/hash/internal/low_level_hash.cc",
"third_party/abseil-cpp/absl/numeric/int128.cc",
"third_party/abseil-cpp/absl/profiling/internal/exponential_biased.cc",
"third_party/abseil-cpp/absl/random/discrete_distribution.cc",
"third_party/abseil-cpp/absl/random/gaussian_distribution.cc",
"third_party/abseil-cpp/absl/random/internal/pool_urbg.cc",
"third_party/abseil-cpp/absl/random/internal/randen.cc",
"third_party/abseil-cpp/absl/random/internal/randen_detect.cc",
"third_party/abseil-cpp/absl/random/internal/randen_hwaes.cc",
"third_party/abseil-cpp/absl/random/internal/randen_round_keys.cc",
"third_party/abseil-cpp/absl/random/internal/randen_slow.cc",
"third_party/abseil-cpp/absl/random/internal/seed_material.cc",
"third_party/abseil-cpp/absl/random/seed_gen_exception.cc",
"third_party/abseil-cpp/absl/random/seed_sequences.cc",
"third_party/abseil-cpp/absl/status/status.cc",
"third_party/abseil-cpp/absl/status/status_payload_printer.cc",
"third_party/abseil-cpp/absl/status/statusor.cc",
"third_party/abseil-cpp/absl/strings/ascii.cc",
"third_party/abseil-cpp/absl/strings/charconv.cc",
"third_party/abseil-cpp/absl/strings/cord.cc",
"third_party/abseil-cpp/absl/strings/cord_analysis.cc",
"third_party/abseil-cpp/absl/strings/cord_buffer.cc",
"third_party/abseil-cpp/absl/strings/escaping.cc",
"third_party/abseil-cpp/absl/strings/internal/charconv_bigint.cc",
"third_party/abseil-cpp/absl/strings/internal/charconv_parse.cc",
"third_party/abseil-cpp/absl/strings/internal/cord_internal.cc",
"third_party/abseil-cpp/absl/strings/internal/cord_rep_btree.cc",
"third_party/abseil-cpp/absl/strings/internal/cord_rep_btree_navigator.cc",
"third_party/abseil-cpp/absl/strings/internal/cord_rep_btree_reader.cc",
"third_party/abseil-cpp/absl/strings/internal/cord_rep_consume.cc",
"third_party/abseil-cpp/absl/strings/internal/cord_rep_crc.cc",
"third_party/abseil-cpp/absl/strings/internal/cord_rep_ring.cc",
"third_party/abseil-cpp/absl/strings/internal/cordz_functions.cc",
"third_party/abseil-cpp/absl/strings/internal/cordz_handle.cc",
"third_party/abseil-cpp/absl/strings/internal/cordz_info.cc",
"third_party/abseil-cpp/absl/strings/internal/damerau_levenshtein_distance.cc",
"third_party/abseil-cpp/absl/strings/internal/escaping.cc",
"third_party/abseil-cpp/absl/strings/internal/memutil.cc",
"third_party/abseil-cpp/absl/strings/internal/ostringstream.cc",
"third_party/abseil-cpp/absl/strings/internal/str_format/arg.cc",
"third_party/abseil-cpp/absl/strings/internal/str_format/bind.cc",
"third_party/abseil-cpp/absl/strings/internal/str_format/extension.cc",
"third_party/abseil-cpp/absl/strings/internal/str_format/float_conversion.cc",
"third_party/abseil-cpp/absl/strings/internal/str_format/output.cc",
"third_party/abseil-cpp/absl/strings/internal/str_format/parser.cc",
"third_party/abseil-cpp/absl/strings/internal/stringify_sink.cc",
"third_party/abseil-cpp/absl/strings/internal/utf8.cc",
"third_party/abseil-cpp/absl/strings/match.cc",
"third_party/abseil-cpp/absl/strings/numbers.cc",
"third_party/abseil-cpp/absl/strings/str_cat.cc",
"third_party/abseil-cpp/absl/strings/str_replace.cc",
"third_party/abseil-cpp/absl/strings/str_split.cc",
"third_party/abseil-cpp/absl/strings/string_view.cc",
"third_party/abseil-cpp/absl/strings/substitute.cc",
"third_party/abseil-cpp/absl/synchronization/barrier.cc",
"third_party/abseil-cpp/absl/synchronization/blocking_counter.cc",
"third_party/abseil-cpp/absl/synchronization/internal/create_thread_identity.cc",
"third_party/abseil-cpp/absl/synchronization/internal/futex_waiter.cc",
"third_party/abseil-cpp/absl/synchronization/internal/graphcycles.cc",
"third_party/abseil-cpp/absl/synchronization/internal/kernel_timeout.cc",
"third_party/abseil-cpp/absl/synchronization/internal/per_thread_sem.cc",
"third_party/abseil-cpp/absl/synchronization/internal/pthread_waiter.cc",
"third_party/abseil-cpp/absl/synchronization/internal/sem_waiter.cc",
"third_party/abseil-cpp/absl/synchronization/internal/stdcpp_waiter.cc",
"third_party/abseil-cpp/absl/synchronization/internal/waiter_base.cc",
"third_party/abseil-cpp/absl/synchronization/internal/win32_waiter.cc",
"third_party/abseil-cpp/absl/synchronization/mutex.cc",
"third_party/abseil-cpp/absl/synchronization/notification.cc",
"third_party/abseil-cpp/absl/time/civil_time.cc",
"third_party/abseil-cpp/absl/time/clock.cc",
"third_party/abseil-cpp/absl/time/duration.cc",
"third_party/abseil-cpp/absl/time/format.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/civil_time_detail.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/time_zone_fixed.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/time_zone_format.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/time_zone_if.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/time_zone_impl.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/time_zone_info.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/time_zone_libc.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/time_zone_lookup.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/time_zone_posix.cc",
"third_party/abseil-cpp/absl/time/internal/cctz/src/zone_info_source.cc",
"third_party/abseil-cpp/absl/time/time.cc",
"third_party/abseil-cpp/absl/types/bad_optional_access.cc",
"third_party/abseil-cpp/absl/types/bad_variant_access.cc",
]
CC_INCLUDES = ["third_party/abseil-cpp"]

297
src/python/grpcio_observability/setup.py Normal file
View File

@ -0,0 +1,297 @@
# 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.
import os
import os.path
import platform
import re
import shlex
import subprocess
from subprocess import PIPE
import sys
import sysconfig
import pkg_resources
import setuptools
from setuptools import Extension
from setuptools.command import build_ext
PYTHON_STEM = os.path.realpath(os.path.dirname(__file__))
README_PATH = os.path.join(PYTHON_STEM, "README.rst")
os.chdir(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.abspath("."))
import _parallel_compile_patch
import observability_lib_deps
import grpc_version
_parallel_compile_patch.monkeypatch_compile_maybe()
CLASSIFIERS = [
"Private :: Do Not Upload",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
"License :: OSI Approved :: Apache Software License",
]
O11Y_CC_SRCS = [
"server_call_tracer.cc",
"client_call_tracer.cc",
"observability_util.cc",
"python_census_context.cc",
"sampler.cc",
"rpc_encoding.cc",
]
def _env_bool_value(env_name, default):
"""Parses a bool option from an environment variable"""
return os.environ.get(env_name, default).upper() not in ["FALSE", "0", ""]
# Environment variable to determine whether or not the Cython extension should
# *use* Cython or use the generated C files. Note that this requires the C files
# to have been generated by building first *with* Cython support.
BUILD_WITH_CYTHON = _env_bool_value("GRPC_PYTHON_BUILD_WITH_CYTHON", "False")
# Export this variable to force building the python extension with a statically linked libstdc++.
# At least on linux, this is normally not needed as we can build manylinux-compatible wheels on linux just fine
# without statically linking libstdc++ (which leads to a slight increase in the wheel size).
# This option is useful when crosscompiling wheels for aarch64 where
# it's difficult to ensure that the crosscompilation toolchain has a high-enough version
# of GCC (we require >=5.1) but still uses old-enough libstdc++ symbols.
# TODO(jtattermusch): remove this workaround once issues with crosscompiler version are resolved.
BUILD_WITH_STATIC_LIBSTDCXX = _env_bool_value(
"GRPC_PYTHON_BUILD_WITH_STATIC_LIBSTDCXX", "False"
)
def check_linker_need_libatomic():
"""Test if linker on system needs libatomic."""
code_test = (
b"#include <atomic>\n"
+ b"int main() { return std::atomic<int64_t>{}; }"
)
cxx = shlex.split(os.environ.get("CXX", "c++"))
cpp_test = subprocess.Popen(
cxx + ["-x", "c++", "-std=c++14", "-"],
stdin=PIPE,
stdout=PIPE,
stderr=PIPE,
)
cpp_test.communicate(input=code_test)
if cpp_test.returncode == 0:
return False
# Double-check to see if -latomic actually can solve the problem.
# https://github.com/grpc/grpc/issues/22491
cpp_test = subprocess.Popen(
cxx + ["-x", "c++", "-std=c++14", "-", "-latomic"],
stdin=PIPE,
stdout=PIPE,
stderr=PIPE,
)
cpp_test.communicate(input=code_test)
return cpp_test.returncode == 0
class BuildExt(build_ext.build_ext):
"""Custom build_ext command."""
def get_ext_filename(self, ext_name):
# since python3.5, python extensions' shared libraries use a suffix that corresponds to the value
# of sysconfig.get_config_var('EXT_SUFFIX') and contains info about the architecture the library targets.
# E.g. on x64 linux the suffix is ".cpython-XYZ-x86_64-linux-gnu.so"
# When crosscompiling python wheels, we need to be able to override this suffix
# so that the resulting file name matches the target architecture and we end up with a well-formed
# wheel.
filename = build_ext.build_ext.get_ext_filename(self, ext_name)
orig_ext_suffix = sysconfig.get_config_var("EXT_SUFFIX")
new_ext_suffix = os.getenv("GRPC_PYTHON_OVERRIDE_EXT_SUFFIX")
if new_ext_suffix and filename.endswith(orig_ext_suffix):
filename = filename[: -len(orig_ext_suffix)] + new_ext_suffix
return filename
# There are some situations (like on Windows) where CC, CFLAGS, and LDFLAGS are
# entirely ignored/dropped/forgotten by distutils and its Cygwin/MinGW support.
# We use these environment variables to thus get around that without locking
# ourselves in w.r.t. the multitude of operating systems this ought to build on.
# We can also use these variables as a way to inject environment-specific
# compiler/linker flags. We assume GCC-like compilers and/or MinGW as a
# reasonable default.
EXTRA_ENV_COMPILE_ARGS = os.environ.get("GRPC_PYTHON_CFLAGS", None)
EXTRA_ENV_LINK_ARGS = os.environ.get("GRPC_PYTHON_LDFLAGS", None)
if EXTRA_ENV_COMPILE_ARGS is None:
EXTRA_ENV_COMPILE_ARGS = "-std=c++14"
if "win32" in sys.platform:
# We need to statically link the C++ Runtime, only the C runtime is
# available dynamically
EXTRA_ENV_COMPILE_ARGS += " /MT"
elif "linux" in sys.platform or "darwin" in sys.platform:
EXTRA_ENV_COMPILE_ARGS += " -fno-wrapv -frtti -fvisibility=hidden"
if EXTRA_ENV_LINK_ARGS is None:
EXTRA_ENV_LINK_ARGS = ""
if "linux" in sys.platform or "darwin" in sys.platform:
EXTRA_ENV_LINK_ARGS += " -lpthread"
if check_linker_need_libatomic():
EXTRA_ENV_LINK_ARGS += " -latomic"
# This enables the standard link-time optimizer, which help us prevent some undefined symbol errors by
# remove some unused symbols from .so file.
# Note that it does not work for MSCV on windows.
if "win32" not in sys.platform:
EXTRA_ENV_COMPILE_ARGS += " -flto"
EXTRA_COMPILE_ARGS = shlex.split(EXTRA_ENV_COMPILE_ARGS)
EXTRA_LINK_ARGS = shlex.split(EXTRA_ENV_LINK_ARGS)
if BUILD_WITH_STATIC_LIBSTDCXX:
EXTRA_LINK_ARGS.append("-static-libstdc++")
CC_FILES = [
os.path.normpath(cc_file) for cc_file in observability_lib_deps.CC_FILES
]
CC_INCLUDES = [
os.path.normpath(include_dir)
for include_dir in observability_lib_deps.CC_INCLUDES
]
DEFINE_MACROS = (("_WIN32_WINNT", 0x600),)
if "win32" in sys.platform:
DEFINE_MACROS += (
("WIN32_LEAN_AND_MEAN", 1),
("CARES_STATICLIB", 1),
("GRPC_ARES", 0),
("NTDDI_VERSION", 0x06000000),
# avoid https://github.com/abseil/abseil-cpp/issues/1425
("NOMINMAX", 1),
)
if "64bit" in platform.architecture()[0]:
DEFINE_MACROS += (("MS_WIN64", 1),)
else:
# For some reason, this is needed to get access to inet_pton/inet_ntop
# on msvc, but only for 32 bits
DEFINE_MACROS += (("NTDDI_VERSION", 0x06000000),)
elif "linux" in sys.platform or "darwin" in sys.platform:
DEFINE_MACROS += (("HAVE_PTHREAD", 1),)
# Fix for Cython build issue in aarch64.
# It's required to define this macro before include <inttypes.h>.
# <inttypes.h> was included in core/lib/channel/call_tracer.h.
# This macro should already be defined in grpc/grpc.h through port_platform.h,
# but we're still having issue in aarch64, so we manually define the macro here.
# TODO(xuanwn): Figure out what's going on in the aarch64 build so we can support
# gcc + Bazel.
DEFINE_MACROS += (("__STDC_FORMAT_MACROS", None),)
# Use `-fvisibility=hidden` will hide cython init symbol, we need that symbol exported
# in order to import cython module.
if "linux" in sys.platform or "darwin" in sys.platform:
pymodinit = 'extern "C" __attribute__((visibility ("default"))) PyObject*'
DEFINE_MACROS += (("PyMODINIT_FUNC", pymodinit),)
# By default, Python3 distutils enforces compatibility of
# c plugins (.so files) with the OSX version Python was built with.
# We need OSX 10.10, the oldest which supports C++ thread_local.
if "darwin" in sys.platform:
mac_target = sysconfig.get_config_var("MACOSX_DEPLOYMENT_TARGET")
if mac_target and (
pkg_resources.parse_version(mac_target)
< pkg_resources.parse_version("10.10.0")
):
os.environ["MACOSX_DEPLOYMENT_TARGET"] = "10.10"
os.environ["_PYTHON_HOST_PLATFORM"] = re.sub(
r"macosx-[0-9]+\.[0-9]+-(.+)",
r"macosx-10.10-\1",
sysconfig.get_platform(),
)
def extension_modules():
if BUILD_WITH_CYTHON:
cython_module_files = [
os.path.join("grpc_observability", "_cyobservability.pyx")
]
else:
cython_module_files = [
os.path.join("grpc_observability", "_cyobservability.cpp")
]
plugin_include = [
".",
"grpc_root",
os.path.join("grpc_root", "include"),
] + CC_INCLUDES
plugin_sources = CC_FILES
O11Y_CC_PATHS = (
os.path.join("grpc_observability", f) for f in O11Y_CC_SRCS
)
plugin_sources += O11Y_CC_PATHS
plugin_sources += cython_module_files
plugin_ext = Extension(
name="grpc_observability._cyobservability",
sources=plugin_sources,
include_dirs=plugin_include,
language="c++",
define_macros=list(DEFINE_MACROS),
extra_compile_args=list(EXTRA_COMPILE_ARGS),
extra_link_args=list(EXTRA_LINK_ARGS),
)
extensions = [plugin_ext]
if BUILD_WITH_CYTHON:
from Cython import Build
return Build.cythonize(
extensions, compiler_directives={"language_level": "3"}
)
else:
return extensions
PACKAGES = setuptools.find_packages(PYTHON_STEM)
setuptools.setup(
name="grpcio-observability",
version=grpc_version.VERSION,
description="gRPC Python observability package",
long_description=open(README_PATH, "r").read(),
author="The gRPC Authors",
author_email="grpc-io@googlegroups.com",
url="https://grpc.io",
project_urls={
"Source Code": "https://github.com/grpc/grpc/tree/master/src/python/grpcio_observability",
"Bug Tracker": "https://github.com/grpc/grpc/issues",
},
license="Apache License 2.0",
classifiers=CLASSIFIERS,
ext_modules=extension_modules(),
packages=list(PACKAGES),
python_requires=">=3.7",
install_requires=[
"grpcio>={version}".format(version=grpc_version.VERSION),
"setuptools>=59.6.0",
],
cmdclass={
"build_ext": BuildExt,
},
)

15
src/python/grpcio_tests/tests/_loader.py
View File

@ -15,6 +15,7 @@
from __future__ import absolute_import
import importlib
import logging
import os
import pkgutil
import re
@ -23,6 +24,8 @@ import unittest
import coverage
logger = logging.getLogger(__name__)
TEST_MODULE_REGEX = r"^.*_test$"
@ -106,11 +109,15 @@ class Loader(object):
module = None
if module_name in sys.modules:
module = sys.modules[module_name]
self.visit_module(module)
else:
spec = importer.find_spec(module_name)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
self.visit_module(module)
try:
spec = importer.find_spec(module_name)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
self.visit_module(module)
except ModuleNotFoundError:
logger.debug("Skip loading %s", module_name)
def visit_module(self, module):
"""Visits the module, adding discovered tests to the test suite.

12
src/python/grpcio_tests/tests/_sanity/_sanity_test.py
View File

@ -13,7 +13,9 @@
# limitations under the License.
import json
import os
import pkgutil
import sys
import unittest
import tests
@ -41,6 +43,16 @@ class SanityTest(unittest.TestCase):
tests_json_string = pkgutil.get_data(self.TEST_PKG_PATH, "tests.json")
tests_json = json.loads(tests_json_string.decode())
# Observability is not supported in Windows and MacOS and Asyncio.
if (
os.name == "nt"
or "darwin" in sys.platform
or self.TEST_PKG_PATH == "tests_aio"
):
for test_case in tests_json:
if "_observability_test" in test_case:
tests_json.remove(test_case)
self.assertSequenceEqual(tests_json, test_suite_names)
self.assertGreater(len(test_suite_names), 0)

13
src/python/grpcio_tests/tests/observability/__init__.py Normal file
View File

@ -0,0 +1,13 @@
# 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.

120
src/python/grpcio_tests/tests/observability/_observability_test.py
View File

@ -17,6 +17,7 @@ import json
import logging
import os
import random
import sys
from typing import Any, Dict, List
import unittest
@ -60,10 +61,6 @@ _VALID_CONFIG_STATS_ONLY_STR = """
'cloud_monitoring': {}
}
"""
# Depends on grpc_core::IsTransportSuppliesClientLatencyEnabled,
# the following metrcis might not exist.
_SKIP_VEFIRY = [_cyobservability.MetricsName.CLIENT_TRANSPORT_LATENCY]
_SPAN_PREFIXS = ["Recv", "Sent", "Attempt"]
class TestExporter(_observability.Exporter):
@ -129,13 +126,13 @@ class _MethodHandler(grpc.RpcMethodHandler):
self.stream_unary = None
self.stream_stream = None
if self.request_streaming and self.response_streaming:
self.stream_stream = lambda x, y: handle_stream_stream(x, y)
self.stream_stream = handle_stream_stream
elif self.request_streaming:
self.stream_unary = lambda x, y: handle_stream_unary(x, y)
self.stream_unary = handle_stream_unary
elif self.response_streaming:
self.unary_stream = lambda x, y: handle_unary_stream(x, y)
self.unary_stream = handle_unary_stream
else:
self.unary_unary = lambda x, y: handle_unary_unary(x, y)
self.unary_unary = handle_unary_unary
class _GenericHandler(grpc.GenericRpcHandler):
@ -152,6 +149,10 @@ class _GenericHandler(grpc.GenericRpcHandler):
return None
@unittest.skipIf(
os.name == "nt" or "darwin" in sys.platform,
"Observability is not supported in Windows and MacOS",
)
class ObservabilityTest(unittest.TestCase):
def setUp(self):
self.all_metric = []
@ -175,53 +176,7 @@ class ObservabilityTest(unittest.TestCase):
unary_unary_call(port=self._port)
self.assertGreater(len(self.all_metric), 0)
self.assertGreater(len(self.all_span), 0)
self._validate_metrics(self.all_metric)
self._validate_spans(self.all_span)
def testContextPropagationToSameServer(self):
# Sends two RPCs, one from gRPC client and the other from gRPC server:
# gRPC Client -> gRPC Server 1 -> gRPC Server 1
# Verify that the trace_id was propagated to the 2nd RPC.
self._set_config_file(_VALID_CONFIG_TRACING_ONLY)
with grpc_observability.GCPOpenCensusObservability(
exporter=self.test_exporter
):
port = self._start_server()
metadata = (
TRIGGER_RPC_METADATA,
("port", str(port)),
)
unary_unary_call(port=port, metadata=metadata)
# 2 of each for ["Recv", "Sent", "Attempt"]
self.assertEqual(len(self.all_span), 6)
trace_id = self.all_span[0].trace_id
for span in self.all_span:
self.assertEqual(span.trace_id, trace_id)
def testContextPropagationToNewServer(self):
# Sends two RPCs, one from gRPC client and the other from gRPC server:
# gRPC Client -> gRPC Server 1 -> gRPC Server 2
# Verify that the trace_id was propagated to the 2nd RPC.
# This test case is to make sure that the context from one thread can
# be propagated to different thread.
self._set_config_file(_VALID_CONFIG_TRACING_ONLY)
with grpc_observability.GCPOpenCensusObservability(
exporter=self.test_exporter
):
port = self._start_server()
metadata = (
TRIGGER_RPC_METADATA,
TRIGGER_RPC_TO_NEW_SERVER_METADATA,
)
unary_unary_call(port=port, metadata=metadata)
# 2 of each for ["Recv", "Sent", "Attempt"]
self.assertEqual(len(self.all_span), 6)
trace_id = self.all_span[0].trace_id
for span in self.all_span:
self.assertEqual(span.trace_id, trace_id)
def testThrowErrorWithoutConfig(self):
with self.assertRaises(ValueError):
@ -251,7 +206,6 @@ class ObservabilityTest(unittest.TestCase):
unary_unary_call(port=self._port)
self.assertEqual(len(self.all_metric), 0)
self.assertEqual(len(self.all_span), 0)
def testThrowErrorWhenCallingMultipleInit(self):
self._set_config_file(_VALID_CONFIG_TRACING_STATS)
@ -269,7 +223,6 @@ class ObservabilityTest(unittest.TestCase):
self._start_server()
unary_unary_call(port=self._port)
self.assertEqual(len(self.all_span), 0)
self.assertGreater(len(self.all_metric), 0)
self._validate_metrics(self.all_metric)
@ -282,8 +235,6 @@ class ObservabilityTest(unittest.TestCase):
unary_unary_call(port=self._port)
self.assertEqual(len(self.all_metric), 0)
self.assertGreater(len(self.all_span), 0)
self._validate_spans(self.all_span)
def testRecordUnaryStream(self):
self._set_config_file(_VALID_CONFIG_TRACING_STATS)
@ -294,9 +245,7 @@ class ObservabilityTest(unittest.TestCase):
unary_stream_call(port=self._port)
self.assertGreater(len(self.all_metric), 0)
self.assertGreater(len(self.all_span), 0)
self._validate_metrics(self.all_metric)
self._validate_spans(self.all_span)
def testRecordStreamUnary(self):
self._set_config_file(_VALID_CONFIG_TRACING_STATS)
@ -309,7 +258,6 @@ class ObservabilityTest(unittest.TestCase):
self.assertTrue(len(self.all_metric) > 0)
self.assertTrue(len(self.all_span) > 0)
self._validate_metrics(self.all_metric)
self._validate_spans(self.all_span)
def testRecordStreamStream(self):
self._set_config_file(_VALID_CONFIG_TRACING_STATS)
@ -320,16 +268,13 @@ class ObservabilityTest(unittest.TestCase):
stream_stream_call(port=self._port)
self.assertGreater(len(self.all_metric), 0)
self.assertGreater(len(self.all_span), 0)
self._validate_metrics(self.all_metric)
self._validate_spans(self.all_span)
def testNoRecordBeforeInit(self):
self._set_config_file(_VALID_CONFIG_TRACING_STATS)
self._start_server()
unary_unary_call(port=self._port)
self.assertEqual(len(self.all_metric), 0)
self.assertEqual(len(self.all_span), 0)
self._server.stop(0)
with grpc_observability.GCPOpenCensusObservability(
@ -339,9 +284,7 @@ class ObservabilityTest(unittest.TestCase):
unary_unary_call(port=self._port)
self.assertGreater(len(self.all_metric), 0)
self.assertGreater(len(self.all_span), 0)
self._validate_metrics(self.all_metric)
self._validate_spans(self.all_span)
def testNoRecordAfterExit(self):
self._set_config_file(_VALID_CONFIG_TRACING_STATS)
@ -352,38 +295,11 @@ class ObservabilityTest(unittest.TestCase):
unary_unary_call(port=self._port)
self.assertGreater(len(self.all_metric), 0)
self.assertGreater(len(self.all_span), 0)
current_metric_len = len(self.all_metric)
current_spans_len = len(self.all_span)
self._validate_metrics(self.all_metric)
self._validate_spans(self.all_span)
unary_unary_call(port=self._port)
self.assertEqual(len(self.all_metric), current_metric_len)
self.assertEqual(len(self.all_span), current_spans_len)
def testTraceSamplingRate(self):
# Make 40 UnaryCall's
# With 50% sampling rate, we should get 10-30 traces with >99.93% probability
# Each trace will have three span (Send, Recv, Attempt)
_CALLS = 40
_LOWER_BOUND = 10 * 3
_HIGHER_BOUND = 30 * 3
_VALID_CONFIG_TRACING_ONLY_SAMPLE_HALF = {
"project_id": "test-project",
"cloud_trace": {"sampling_rate": 0.5},
}
self._set_config_file(_VALID_CONFIG_TRACING_ONLY_SAMPLE_HALF)
with grpc_observability.GCPOpenCensusObservability(
exporter=self.test_exporter
):
self._start_server()
for _ in range(_CALLS):
unary_unary_call(port=self._port)
self.assertEqual(len(self.all_metric), 0)
self.assertGreaterEqual(len(self.all_span), _LOWER_BOUND)
self.assertLessEqual(len(self.all_span), _HIGHER_BOUND)
self._validate_spans(self.all_span)
def testConfigFileOverEnvVar(self):
# env var have only stats enabled
@ -398,8 +314,6 @@ class ObservabilityTest(unittest.TestCase):
unary_unary_call(port=self._port)
self.assertEqual(len(self.all_metric), 0)
self.assertGreater(len(self.all_span), 0)
self._validate_spans(self.all_span)
def _set_config_file(self, config: Dict[str, Any]) -> None:
# Using random name here so multiple tests can run with different config files.
@ -420,8 +334,6 @@ class ObservabilityTest(unittest.TestCase):
) -> None:
metric_names = set(metric.name for metric in metrics)
for name in _cyobservability.MetricsName:
if name in _SKIP_VEFIRY:
continue
if name not in metric_names:
logger.error(
"metric %s not found in exported metrics: %s!",
@ -430,20 +342,6 @@ class ObservabilityTest(unittest.TestCase):
)
self.assertTrue(name in metric_names)
def _validate_spans(
self, tracing_data: List[_observability.TracingData]
) -> None:
span_names = set(data.name for data in tracing_data)
for prefix in _SPAN_PREFIXS:
prefix_exist = any(prefix in name for name in span_names)
if not prefix_exist:
logger.error(
"missing span with prefix %s in exported spans: %s!",
prefix,
span_names,
)
self.assertTrue(prefix_exist)
def unary_unary_call(port, metadata=None):
with grpc.insecure_channel(f"localhost:{port}") as channel:

1
src/python/grpcio_tests/tests/tests.json
View File

@ -9,6 +9,7 @@
"tests.health_check._health_servicer_test.HealthServicerTest",
"tests.interop._insecure_intraop_test.InsecureIntraopTest",
"tests.interop._secure_intraop_test.SecureIntraopTest",
"tests.observability._observability_test.ObservabilityTest",
"tests.protoc_plugin._python_plugin_test.ModuleMainTest",
"tests.protoc_plugin._python_plugin_test.PythonPluginTest",
"tests.protoc_plugin._python_plugin_test.SimpleStubsPluginTest",

19
templates/src/python/grpcio_observability/grpc_version.py.template Normal file
View File

@ -0,0 +1,19 @@
%YAML 1.2
--- |
# 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.
# AUTO-GENERATED FROM `$REPO_ROOT/templates/src/python/grpcio_observability/grpc_version.py.template`!!!
VERSION = '${settings.python_version.pep440()}'

15
tools/run_tests/helper_scripts/build_python.sh
View File

@ -138,10 +138,8 @@ pip_install() {
/usr/bin/env -i PATH="$PATH" "$VENV_PYTHON" -m pip install "$@"
}
# Pin setuptools to < 60.0.0 to restore the distutil installation, see:
# https://github.com/pypa/setuptools/pull/2896
pip_install --upgrade pip==21.3.1
pip_install --upgrade setuptools==59.6.0
pip_install --upgrade setuptools==61.0.0
pip_install --upgrade pip
# pip-installs the directory specified. Used because on MSYS the vanilla Windows
# Python gets confused when parsing paths.
@ -176,6 +174,15 @@ pip_install_dir "$ROOT"
$VENV_PYTHON "$ROOT/tools/distrib/python/make_grpcio_tools.py"
pip_install_dir_and_deps "$ROOT/tools/distrib/python/grpcio_tools"
# Build/install Observability
# Observability does not support Windows and MacOS.
if [ "$(is_mingw)" ] || [ "$(is_darwin)" ]; then
echo "Skip building grpcio_observability for Windows or MacOS"
else
$VENV_PYTHON "$ROOT/src/python/grpcio_observability/make_grpcio_observability.py"
pip_install_dir_and_deps "$ROOT/src/python/grpcio_observability"
fi
# Build/install Channelz
$VENV_PYTHON "$ROOT/src/python/grpcio_channelz/setup.py" preprocess
$VENV_PYTHON "$ROOT/src/python/grpcio_channelz/setup.py" build_package_protos