Add load reporting filter

pull/15570/head
Juanli Shen 7 years ago
parent 7a2a8ca4ba
commit f5f1d57d7a
  1. 13
      BUILD
  2. 75
      CMakeLists.txt
  3. 69
      Makefile
  4. 14
      build.yaml
  5. 3
      config.m4
  6. 3
      config.w32
  7. 2
      gRPC-C++.podspec
  8. 8
      gRPC-Core.podspec
  9. 5
      grpc.gemspec
  10. 6
      grpc.gyp
  11. 1
      include/grpc/module.modulemap
  12. 5
      package.xml
  13. 108
      src/core/ext/filters/load_reporting/registered_opencensus_objects.h
  14. 426
      src/core/ext/filters/load_reporting/server_load_reporting_filter.cc
  15. 99
      src/core/ext/filters/load_reporting/server_load_reporting_filter.h
  16. 71
      src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc
  17. 61
      src/core/ext/filters/load_reporting/server_load_reporting_plugin.h
  18. 4
      src/core/plugin_registry/grpc_cronet_plugin_registry.cc
  19. 4
      src/core/plugin_registry/grpc_plugin_registry.cc
  20. 4
      src/core/plugin_registry/grpc_unsecure_plugin_registry.cc
  21. 12
      src/cpp/common/channel_filter.h
  22. 2
      src/python/grpcio/grpc_core_dependencies.py
  23. 8
      test/core/end2end/end2end_nosec_tests.cc
  24. 8
      test/core/end2end/end2end_tests.cc
  25. 120
      test/core/end2end/fixtures/h2_load_reporting.cc
  26. 8
      test/core/end2end/gen_build_yaml.py
  27. 9
      test/core/end2end/generate_tests.bzl
  28. 1
      test/core/surface/public_headers_must_be_c89.c
  29. 11
      test/cpp/microbenchmarks/bm_call_create.cc
  30. 1
      tools/doxygen/Doxyfile.c++
  31. 1
      tools/doxygen/Doxyfile.c++.internal
  32. 1
      tools/doxygen/Doxyfile.core
  33. 5
      tools/doxygen/Doxyfile.core.internal
  34. 64
      tools/run_tests/generated/sources_and_headers.json
  35. 14808
      tools/run_tests/generated/tests.json

13
BUILD

@ -987,7 +987,6 @@ grpc_cc_library(
"grpc_client_authority_filter",
"grpc_lb_policy_pick_first",
"grpc_lb_policy_round_robin",
"grpc_server_load_reporting",
"grpc_max_age_filter",
"grpc_message_size_filter",
"grpc_resolver_dns_ares",
@ -1287,18 +1286,22 @@ grpc_cc_library(
)
grpc_cc_library(
name = "grpc_server_load_reporting",
name = "lb_server_load_reporting_filter",
srcs = [
"src/core/ext/filters/load_reporting/server_load_reporting_filter.cc",
"src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc",
],
hdrs = [
"src/core/ext/filters/load_reporting/registered_opencensus_objects.h",
"src/core/ext/filters/load_reporting/server_load_reporting_filter.h",
"src/core/ext/filters/load_reporting/server_load_reporting_plugin.h",
"src/cpp/server/load_reporter/constants.h",
],
external_deps = [
"opencensus-stats",
],
language = "c++",
deps = [
"grpc_base",
"grpc++_base",
"grpc_secure",
],
)

@ -430,7 +430,6 @@ endif()
add_dependencies(buildtests_c h2_full+trace_test)
add_dependencies(buildtests_c h2_full+workarounds_test)
add_dependencies(buildtests_c h2_http_proxy_test)
add_dependencies(buildtests_c h2_load_reporting_test)
add_dependencies(buildtests_c h2_oauth2_test)
add_dependencies(buildtests_c h2_proxy_test)
add_dependencies(buildtests_c h2_sockpair_test)
@ -454,7 +453,6 @@ endif()
add_dependencies(buildtests_c h2_full+trace_nosec_test)
add_dependencies(buildtests_c h2_full+workarounds_nosec_test)
add_dependencies(buildtests_c h2_http_proxy_nosec_test)
add_dependencies(buildtests_c h2_load_reporting_nosec_test)
add_dependencies(buildtests_c h2_proxy_nosec_test)
add_dependencies(buildtests_c h2_sockpair_nosec_test)
add_dependencies(buildtests_c h2_sockpair+trace_nosec_test)
@ -1227,8 +1225,6 @@ add_library(grpc
src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc
src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc
src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc
src/core/ext/filters/load_reporting/server_load_reporting_filter.cc
src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc
src/cpp/ext/filters/census/grpc_context.cc
src/core/ext/filters/max_age/max_age_filter.cc
src/core/ext/filters/message_size/message_size_filter.cc
@ -1303,7 +1299,6 @@ foreach(_hdr
include/grpc/grpc.h
include/grpc/grpc_posix.h
include/grpc/grpc_security_constants.h
include/grpc/load_reporting.h
include/grpc/slice.h
include/grpc/slice_buffer.h
include/grpc/status.h
@ -1604,8 +1599,6 @@ add_library(grpc_cronet
src/core/tsi/ssl/session_cache/ssl_session_openssl.cc
src/core/tsi/ssl_transport_security.cc
src/core/tsi/transport_security_grpc.cc
src/core/ext/filters/load_reporting/server_load_reporting_filter.cc
src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc
src/core/plugin_registry/grpc_cronet_plugin_registry.cc
)
@ -2523,8 +2516,6 @@ add_library(grpc_unsecure
src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc
src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc
src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc
src/core/ext/filters/load_reporting/server_load_reporting_filter.cc
src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc
src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.cc
src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel.cc
@ -2608,7 +2599,6 @@ foreach(_hdr
include/grpc/grpc.h
include/grpc/grpc_posix.h
include/grpc/grpc_security_constants.h
include/grpc/load_reporting.h
include/grpc/slice.h
include/grpc/slice_buffer.h
include/grpc/status.h
@ -2925,7 +2915,6 @@ foreach(_hdr
include/grpc/grpc.h
include/grpc/grpc_posix.h
include/grpc/grpc_security_constants.h
include/grpc/load_reporting.h
include/grpc/slice.h
include/grpc/slice_buffer.h
include/grpc/status.h
@ -3489,7 +3478,6 @@ foreach(_hdr
include/grpc/grpc.h
include/grpc/grpc_posix.h
include/grpc/grpc_security_constants.h
include/grpc/load_reporting.h
include/grpc/slice.h
include/grpc/slice_buffer.h
include/grpc/status.h
@ -4389,7 +4377,6 @@ foreach(_hdr
include/grpc/grpc.h
include/grpc/grpc_posix.h
include/grpc/grpc_security_constants.h
include/grpc/load_reporting.h
include/grpc/slice.h
include/grpc/slice_buffer.h
include/grpc/status.h
@ -5301,7 +5288,6 @@ add_library(end2end_tests
test/core/end2end/tests/invoke_large_request.cc
test/core/end2end/tests/keepalive_timeout.cc
test/core/end2end/tests/large_metadata.cc
test/core/end2end/tests/load_reporting_hook.cc
test/core/end2end/tests/max_concurrent_streams.cc
test/core/end2end/tests/max_connection_age.cc
test/core/end2end/tests/max_connection_idle.cc
@ -5422,7 +5408,6 @@ add_library(end2end_nosec_tests
test/core/end2end/tests/invoke_large_request.cc
test/core/end2end/tests/keepalive_timeout.cc
test/core/end2end/tests/large_metadata.cc
test/core/end2end/tests/load_reporting_hook.cc
test/core/end2end/tests/max_concurrent_streams.cc
test/core/end2end/tests/max_connection_age.cc
test/core/end2end/tests/max_connection_idle.cc
@ -15233,36 +15218,6 @@ target_link_libraries(h2_http_proxy_test
endif (gRPC_BUILD_TESTS)
if (gRPC_BUILD_TESTS)
add_executable(h2_load_reporting_test
test/core/end2end/fixtures/h2_load_reporting.cc
)
target_include_directories(h2_load_reporting_test
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include
PRIVATE ${_gRPC_SSL_INCLUDE_DIR}
PRIVATE ${_gRPC_PROTOBUF_INCLUDE_DIR}
PRIVATE ${_gRPC_ZLIB_INCLUDE_DIR}
PRIVATE ${_gRPC_BENCHMARK_INCLUDE_DIR}
PRIVATE ${_gRPC_CARES_INCLUDE_DIR}
PRIVATE ${_gRPC_GFLAGS_INCLUDE_DIR}
PRIVATE ${_gRPC_ADDRESS_SORTING_INCLUDE_DIR}
PRIVATE ${_gRPC_NANOPB_INCLUDE_DIR}
)
target_link_libraries(h2_load_reporting_test
${_gRPC_ALLTARGETS_LIBRARIES}
end2end_tests
grpc_test_util
grpc
gpr_test_util
gpr
)
endif (gRPC_BUILD_TESTS)
if (gRPC_BUILD_TESTS)
add_executable(h2_oauth2_test
test/core/end2end/fixtures/h2_oauth2.cc
)
@ -15779,36 +15734,6 @@ target_link_libraries(h2_http_proxy_nosec_test
endif (gRPC_BUILD_TESTS)
if (gRPC_BUILD_TESTS)
add_executable(h2_load_reporting_nosec_test
test/core/end2end/fixtures/h2_load_reporting.cc
)
target_include_directories(h2_load_reporting_nosec_test
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include
PRIVATE ${_gRPC_SSL_INCLUDE_DIR}
PRIVATE ${_gRPC_PROTOBUF_INCLUDE_DIR}
PRIVATE ${_gRPC_ZLIB_INCLUDE_DIR}
PRIVATE ${_gRPC_BENCHMARK_INCLUDE_DIR}
PRIVATE ${_gRPC_CARES_INCLUDE_DIR}
PRIVATE ${_gRPC_GFLAGS_INCLUDE_DIR}
PRIVATE ${_gRPC_ADDRESS_SORTING_INCLUDE_DIR}
PRIVATE ${_gRPC_NANOPB_INCLUDE_DIR}
)
target_link_libraries(h2_load_reporting_nosec_test
${_gRPC_ALLTARGETS_LIBRARIES}
end2end_nosec_tests
grpc_test_util_unsecure
grpc_unsecure
gpr_test_util
gpr
)
endif (gRPC_BUILD_TESTS)
if (gRPC_BUILD_TESTS)
add_executable(h2_proxy_nosec_test
test/core/end2end/fixtures/h2_proxy.cc
)

@ -1285,7 +1285,6 @@ h2_full+pipe_test: $(BINDIR)/$(CONFIG)/h2_full+pipe_test
h2_full+trace_test: $(BINDIR)/$(CONFIG)/h2_full+trace_test
h2_full+workarounds_test: $(BINDIR)/$(CONFIG)/h2_full+workarounds_test
h2_http_proxy_test: $(BINDIR)/$(CONFIG)/h2_http_proxy_test
h2_load_reporting_test: $(BINDIR)/$(CONFIG)/h2_load_reporting_test
h2_oauth2_test: $(BINDIR)/$(CONFIG)/h2_oauth2_test
h2_proxy_test: $(BINDIR)/$(CONFIG)/h2_proxy_test
h2_sockpair_test: $(BINDIR)/$(CONFIG)/h2_sockpair_test
@ -1303,7 +1302,6 @@ h2_full+pipe_nosec_test: $(BINDIR)/$(CONFIG)/h2_full+pipe_nosec_test
h2_full+trace_nosec_test: $(BINDIR)/$(CONFIG)/h2_full+trace_nosec_test
h2_full+workarounds_nosec_test: $(BINDIR)/$(CONFIG)/h2_full+workarounds_nosec_test
h2_http_proxy_nosec_test: $(BINDIR)/$(CONFIG)/h2_http_proxy_nosec_test
h2_load_reporting_nosec_test: $(BINDIR)/$(CONFIG)/h2_load_reporting_nosec_test
h2_proxy_nosec_test: $(BINDIR)/$(CONFIG)/h2_proxy_nosec_test
h2_sockpair_nosec_test: $(BINDIR)/$(CONFIG)/h2_sockpair_nosec_test
h2_sockpair+trace_nosec_test: $(BINDIR)/$(CONFIG)/h2_sockpair+trace_nosec_test
@ -1540,7 +1538,6 @@ buildtests_c: privatelibs_c \
$(BINDIR)/$(CONFIG)/h2_full+trace_test \
$(BINDIR)/$(CONFIG)/h2_full+workarounds_test \
$(BINDIR)/$(CONFIG)/h2_http_proxy_test \
$(BINDIR)/$(CONFIG)/h2_load_reporting_test \
$(BINDIR)/$(CONFIG)/h2_oauth2_test \
$(BINDIR)/$(CONFIG)/h2_proxy_test \
$(BINDIR)/$(CONFIG)/h2_sockpair_test \
@ -1558,7 +1555,6 @@ buildtests_c: privatelibs_c \
$(BINDIR)/$(CONFIG)/h2_full+trace_nosec_test \
$(BINDIR)/$(CONFIG)/h2_full+workarounds_nosec_test \
$(BINDIR)/$(CONFIG)/h2_http_proxy_nosec_test \
$(BINDIR)/$(CONFIG)/h2_load_reporting_nosec_test \
$(BINDIR)/$(CONFIG)/h2_proxy_nosec_test \
$(BINDIR)/$(CONFIG)/h2_sockpair_nosec_test \
$(BINDIR)/$(CONFIG)/h2_sockpair+trace_nosec_test \
@ -3604,8 +3600,6 @@ LIBGRPC_SRC = \
src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc \
src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc \
src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc \
src/core/ext/filters/load_reporting/server_load_reporting_filter.cc \
src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc \
src/cpp/ext/filters/census/grpc_context.cc \
src/core/ext/filters/max_age/max_age_filter.cc \
src/core/ext/filters/message_size/message_size_filter.cc \
@ -3644,7 +3638,6 @@ PUBLIC_HEADERS_C += \
include/grpc/grpc.h \
include/grpc/grpc_posix.h \
include/grpc/grpc_security_constants.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \
@ -3980,8 +3973,6 @@ LIBGRPC_CRONET_SRC = \
src/core/tsi/ssl/session_cache/ssl_session_openssl.cc \
src/core/tsi/ssl_transport_security.cc \
src/core/tsi/transport_security_grpc.cc \
src/core/ext/filters/load_reporting/server_load_reporting_filter.cc \
src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc \
src/core/plugin_registry/grpc_cronet_plugin_registry.cc \
PUBLIC_HEADERS_C += \
@ -4866,8 +4857,6 @@ LIBGRPC_UNSECURE_SRC = \
src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc \
src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc \
src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc \
src/core/ext/filters/load_reporting/server_load_reporting_filter.cc \
src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc \
src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.cc \
src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc \
src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel.cc \
@ -4916,7 +4905,6 @@ PUBLIC_HEADERS_C += \
include/grpc/grpc.h \
include/grpc/grpc_posix.h \
include/grpc/grpc_security_constants.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \
@ -5212,7 +5200,6 @@ PUBLIC_HEADERS_CXX += \
include/grpc/grpc.h \
include/grpc/grpc_posix.h \
include/grpc/grpc_security_constants.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \
@ -5783,7 +5770,6 @@ PUBLIC_HEADERS_CXX += \
include/grpc/grpc.h \
include/grpc/grpc_posix.h \
include/grpc/grpc_security_constants.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \
@ -6643,7 +6629,6 @@ PUBLIC_HEADERS_CXX += \
include/grpc/grpc.h \
include/grpc/grpc_posix.h \
include/grpc/grpc_security_constants.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \
@ -9953,7 +9938,6 @@ LIBEND2END_TESTS_SRC = \
test/core/end2end/tests/invoke_large_request.cc \
test/core/end2end/tests/keepalive_timeout.cc \
test/core/end2end/tests/large_metadata.cc \
test/core/end2end/tests/load_reporting_hook.cc \
test/core/end2end/tests/max_concurrent_streams.cc \
test/core/end2end/tests/max_connection_age.cc \
test/core/end2end/tests/max_connection_idle.cc \
@ -10071,7 +10055,6 @@ LIBEND2END_NOSEC_TESTS_SRC = \
test/core/end2end/tests/invoke_large_request.cc \
test/core/end2end/tests/keepalive_timeout.cc \
test/core/end2end/tests/large_metadata.cc \
test/core/end2end/tests/load_reporting_hook.cc \
test/core/end2end/tests/max_concurrent_streams.cc \
test/core/end2end/tests/max_connection_age.cc \
test/core/end2end/tests/max_connection_idle.cc \
@ -22861,38 +22844,6 @@ endif
endif
H2_LOAD_REPORTING_TEST_SRC = \
test/core/end2end/fixtures/h2_load_reporting.cc \
H2_LOAD_REPORTING_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(H2_LOAD_REPORTING_TEST_SRC))))
ifeq ($(NO_SECURE),true)
# You can't build secure targets if you don't have OpenSSL.
$(BINDIR)/$(CONFIG)/h2_load_reporting_test: openssl_dep_error
else
$(BINDIR)/$(CONFIG)/h2_load_reporting_test: $(H2_LOAD_REPORTING_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libend2end_tests.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
$(E) "[LD] Linking $@"
$(Q) mkdir -p `dirname $@`
$(Q) $(LD) $(LDFLAGS) $(H2_LOAD_REPORTING_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libend2end_tests.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBS) $(LDLIBS_SECURE) -o $(BINDIR)/$(CONFIG)/h2_load_reporting_test
endif
$(OBJDIR)/$(CONFIG)/test/core/end2end/fixtures/h2_load_reporting.o: $(LIBDIR)/$(CONFIG)/libend2end_tests.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
deps_h2_load_reporting_test: $(H2_LOAD_REPORTING_TEST_OBJS:.o=.dep)
ifneq ($(NO_SECURE),true)
ifneq ($(NO_DEPS),true)
-include $(H2_LOAD_REPORTING_TEST_OBJS:.o=.dep)
endif
endif
H2_OAUTH2_TEST_SRC = \
test/core/end2end/fixtures/h2_oauth2.cc \
@ -23341,26 +23292,6 @@ ifneq ($(NO_DEPS),true)
endif
H2_LOAD_REPORTING_NOSEC_TEST_SRC = \
test/core/end2end/fixtures/h2_load_reporting.cc \
H2_LOAD_REPORTING_NOSEC_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(H2_LOAD_REPORTING_NOSEC_TEST_SRC))))
$(BINDIR)/$(CONFIG)/h2_load_reporting_nosec_test: $(H2_LOAD_REPORTING_NOSEC_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libend2end_nosec_tests.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util_unsecure.a $(LIBDIR)/$(CONFIG)/libgrpc_unsecure.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
$(E) "[LD] Linking $@"
$(Q) mkdir -p `dirname $@`
$(Q) $(LD) $(LDFLAGS) $(H2_LOAD_REPORTING_NOSEC_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libend2end_nosec_tests.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util_unsecure.a $(LIBDIR)/$(CONFIG)/libgrpc_unsecure.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBS) -o $(BINDIR)/$(CONFIG)/h2_load_reporting_nosec_test
$(OBJDIR)/$(CONFIG)/test/core/end2end/fixtures/h2_load_reporting.o: $(LIBDIR)/$(CONFIG)/libend2end_nosec_tests.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util_unsecure.a $(LIBDIR)/$(CONFIG)/libgrpc_unsecure.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
deps_h2_load_reporting_nosec_test: $(H2_LOAD_REPORTING_NOSEC_TEST_OBJS:.o=.dep)
ifneq ($(NO_DEPS),true)
-include $(H2_LOAD_REPORTING_NOSEC_TEST_OBJS:.o=.dep)
endif
H2_PROXY_NOSEC_TEST_SRC = \
test/core/end2end/fixtures/h2_proxy.cc \

@ -393,7 +393,6 @@ filegroups:
- include/grpc/grpc.h
- include/grpc/grpc_posix.h
- include/grpc/grpc_security_constants.h
- include/grpc/load_reporting.h
- include/grpc/slice.h
- include/grpc/slice_buffer.h
- include/grpc/status.h
@ -823,16 +822,6 @@ filegroups:
- src/core/ext/filters/workarounds/workaround_utils.cc
uses:
- grpc_base
- name: grpc_server_load_reporting
headers:
- src/core/ext/filters/load_reporting/server_load_reporting_filter.h
- src/core/ext/filters/load_reporting/server_load_reporting_plugin.h
src:
- src/core/ext/filters/load_reporting/server_load_reporting_filter.cc
- src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc
plugin: grpc_server_load_reporting_plugin
uses:
- grpc_base
- name: grpc_test_util_base
build: test
headers:
@ -1414,7 +1403,6 @@ libs:
- grpc_resolver_dns_native
- grpc_resolver_sockaddr
- grpc_resolver_fake
- grpc_server_load_reporting
- grpc_secure
- census
- grpc_max_age_filter
@ -1441,7 +1429,6 @@ libs:
- grpc_base
- grpc_transport_cronet_client_secure
- grpc_transport_chttp2_client_secure
- grpc_server_load_reporting
generate_plugin_registry: true
platforms:
- linux
@ -1515,7 +1502,6 @@ libs:
- grpc_resolver_dns_native
- grpc_resolver_sockaddr
- grpc_resolver_fake
- grpc_server_load_reporting
- grpc_lb_policy_grpclb
- grpc_lb_policy_pick_first
- grpc_lb_policy_round_robin

@ -378,8 +378,6 @@ if test "$PHP_GRPC" != "no"; then
src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc \
src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc \
src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc \
src/core/ext/filters/load_reporting/server_load_reporting_filter.cc \
src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc \
src/cpp/ext/filters/census/grpc_context.cc \
src/core/ext/filters/max_age/max_age_filter.cc \
src/core/ext/filters/message_size/message_size_filter.cc \
@ -665,7 +663,6 @@ if test "$PHP_GRPC" != "no"; then
PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/http/client)
PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/http/message_compress)
PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/http/server)
PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/load_reporting)
PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/max_age)
PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/message_size)
PHP_ADD_BUILD_DIR($ext_builddir/src/core/ext/filters/workarounds)

@ -353,8 +353,6 @@ if (PHP_GRPC != "no") {
"src\\core\\ext\\filters\\client_channel\\resolver\\dns\\c_ares\\grpc_ares_wrapper_fallback.cc " +
"src\\core\\ext\\filters\\client_channel\\resolver\\dns\\native\\dns_resolver.cc " +
"src\\core\\ext\\filters\\client_channel\\resolver\\sockaddr\\sockaddr_resolver.cc " +
"src\\core\\ext\\filters\\load_reporting\\server_load_reporting_filter.cc " +
"src\\core\\ext\\filters\\load_reporting\\server_load_reporting_plugin.cc " +
"src\\cpp\\ext\\filters\\census\\grpc_context.cc " +
"src\\core\\ext\\filters\\max_age\\max_age_filter.cc " +
"src\\core\\ext\\filters\\message_size\\message_size_filter.cc " +
@ -676,7 +674,6 @@ if (PHP_GRPC != "no") {
FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\http\\client");
FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\http\\message_compress");
FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\http\\server");
FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\load_reporting");
FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\max_age");
FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\message_size");
FSO.CreateFolder(base_dir+"\\ext\\grpc\\src\\core\\ext\\filters\\workarounds");

@ -484,8 +484,6 @@ Pod::Spec.new do |s|
'src/core/ext/filters/client_channel/lb_policy/subchannel_list.h',
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h',
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h',
'src/core/ext/filters/load_reporting/server_load_reporting_filter.h',
'src/core/ext/filters/load_reporting/server_load_reporting_plugin.h',
'src/core/ext/filters/max_age/max_age_filter.h',
'src/core/ext/filters/message_size/message_size_filter.h',
'src/core/ext/filters/http/client_authority_filter.h',

@ -168,7 +168,6 @@ Pod::Spec.new do |s|
'include/grpc/grpc.h',
'include/grpc/grpc_posix.h',
'include/grpc/grpc_security_constants.h',
'include/grpc/load_reporting.h',
'include/grpc/slice.h',
'include/grpc/slice_buffer.h',
'include/grpc/status.h',
@ -494,8 +493,6 @@ Pod::Spec.new do |s|
'src/core/ext/filters/client_channel/lb_policy/subchannel_list.h',
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h',
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h',
'src/core/ext/filters/load_reporting/server_load_reporting_filter.h',
'src/core/ext/filters/load_reporting/server_load_reporting_plugin.h',
'src/core/ext/filters/max_age/max_age_filter.h',
'src/core/ext/filters/message_size/message_size_filter.h',
'src/core/ext/filters/http/client_authority_filter.h',
@ -793,8 +790,6 @@ Pod::Spec.new do |s|
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc',
'src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc',
'src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc',
'src/core/ext/filters/load_reporting/server_load_reporting_filter.cc',
'src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc',
'src/cpp/ext/filters/census/grpc_context.cc',
'src/core/ext/filters/max_age/max_age_filter.cc',
'src/core/ext/filters/message_size/message_size_filter.cc',
@ -1075,8 +1070,6 @@ Pod::Spec.new do |s|
'src/core/ext/filters/client_channel/lb_policy/subchannel_list.h',
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h',
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h',
'src/core/ext/filters/load_reporting/server_load_reporting_filter.h',
'src/core/ext/filters/load_reporting/server_load_reporting_plugin.h',
'src/core/ext/filters/max_age/max_age_filter.h',
'src/core/ext/filters/message_size/message_size_filter.h',
'src/core/ext/filters/http/client_authority_filter.h',
@ -1214,7 +1207,6 @@ Pod::Spec.new do |s|
'test/core/end2end/tests/invoke_large_request.cc',
'test/core/end2end/tests/keepalive_timeout.cc',
'test/core/end2end/tests/large_metadata.cc',
'test/core/end2end/tests/load_reporting_hook.cc',
'test/core/end2end/tests/max_concurrent_streams.cc',
'test/core/end2end/tests/max_connection_age.cc',
'test/core/end2end/tests/max_connection_idle.cc',

@ -173,7 +173,6 @@ Gem::Specification.new do |s|
s.files += %w( include/grpc/grpc.h )
s.files += %w( include/grpc/grpc_posix.h )
s.files += %w( include/grpc/grpc_security_constants.h )
s.files += %w( include/grpc/load_reporting.h )
s.files += %w( include/grpc/slice.h )
s.files += %w( include/grpc/slice_buffer.h )
s.files += %w( include/grpc/status.h )
@ -431,8 +430,6 @@ Gem::Specification.new do |s|
s.files += %w( src/core/ext/filters/client_channel/lb_policy/subchannel_list.h )
s.files += %w( src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h )
s.files += %w( src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h )
s.files += %w( src/core/ext/filters/load_reporting/server_load_reporting_filter.h )
s.files += %w( src/core/ext/filters/load_reporting/server_load_reporting_plugin.h )
s.files += %w( src/core/ext/filters/max_age/max_age_filter.h )
s.files += %w( src/core/ext/filters/message_size/message_size_filter.h )
s.files += %w( src/core/ext/filters/http/client_authority_filter.h )
@ -733,8 +730,6 @@ Gem::Specification.new do |s|
s.files += %w( src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc )
s.files += %w( src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc )
s.files += %w( src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc )
s.files += %w( src/core/ext/filters/load_reporting/server_load_reporting_filter.cc )
s.files += %w( src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc )
s.files += %w( src/cpp/ext/filters/census/grpc_context.cc )
s.files += %w( src/core/ext/filters/max_age/max_age_filter.cc )
s.files += %w( src/core/ext/filters/message_size/message_size_filter.cc )

@ -544,8 +544,6 @@
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc',
'src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc',
'src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc',
'src/core/ext/filters/load_reporting/server_load_reporting_filter.cc',
'src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc',
'src/cpp/ext/filters/census/grpc_context.cc',
'src/core/ext/filters/max_age/max_age_filter.cc',
'src/core/ext/filters/message_size/message_size_filter.cc',
@ -1259,8 +1257,6 @@
'src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc',
'src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc',
'src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc',
'src/core/ext/filters/load_reporting/server_load_reporting_filter.cc',
'src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc',
'src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.cc',
'src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc',
'src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel.cc',
@ -2627,7 +2623,6 @@
'test/core/end2end/tests/invoke_large_request.cc',
'test/core/end2end/tests/keepalive_timeout.cc',
'test/core/end2end/tests/large_metadata.cc',
'test/core/end2end/tests/load_reporting_hook.cc',
'test/core/end2end/tests/max_concurrent_streams.cc',
'test/core/end2end/tests/max_connection_age.cc',
'test/core/end2end/tests/max_connection_idle.cc',
@ -2719,7 +2714,6 @@
'test/core/end2end/tests/invoke_large_request.cc',
'test/core/end2end/tests/keepalive_timeout.cc',
'test/core/end2end/tests/large_metadata.cc',
'test/core/end2end/tests/load_reporting_hook.cc',
'test/core/end2end/tests/max_concurrent_streams.cc',
'test/core/end2end/tests/max_connection_age.cc',
'test/core/end2end/tests/max_connection_idle.cc',

@ -43,7 +43,6 @@ framework module grpc {
header "grpc.h"
header "grpc_posix.h"
header "grpc_security_constants.h"
header "load_reporting.h"
header "slice.h"
header "slice_buffer.h"
header "status.h"

@ -178,7 +178,6 @@
<file baseinstalldir="/" name="include/grpc/grpc.h" role="src" />
<file baseinstalldir="/" name="include/grpc/grpc_posix.h" role="src" />
<file baseinstalldir="/" name="include/grpc/grpc_security_constants.h" role="src" />
<file baseinstalldir="/" name="include/grpc/load_reporting.h" role="src" />
<file baseinstalldir="/" name="include/grpc/slice.h" role="src" />
<file baseinstalldir="/" name="include/grpc/slice_buffer.h" role="src" />
<file baseinstalldir="/" name="include/grpc/status.h" role="src" />
@ -436,8 +435,6 @@
<file baseinstalldir="/" name="src/core/ext/filters/client_channel/lb_policy/subchannel_list.h" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/load_reporting/server_load_reporting_filter.h" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/load_reporting/server_load_reporting_plugin.h" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/max_age/max_age_filter.h" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/message_size/message_size_filter.h" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/http/client_authority_filter.h" role="src" />
@ -738,8 +735,6 @@
<file baseinstalldir="/" name="src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/load_reporting/server_load_reporting_filter.cc" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc" role="src" />
<file baseinstalldir="/" name="src/cpp/ext/filters/census/grpc_context.cc" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/max_age/max_age_filter.cc" role="src" />
<file baseinstalldir="/" name="src/core/ext/filters/message_size/message_size_filter.cc" role="src" />

@ -0,0 +1,108 @@
/*
*
* Copyright 2018 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_CORE_EXT_FILTERS_LOAD_REPORTING_REGISTERED_OPENCENSUS_OBJECTS_H
#define GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_REGISTERED_OPENCENSUS_OBJECTS_H
#include <grpc/support/port_platform.h>
#include "opencensus/stats/stats.h"
#include "src/cpp/server/load_reporter/constants.h"
namespace grpc {
namespace load_reporter {
// Measures.
::opencensus::stats::MeasureInt64 MeasureStartCount() {
static const ::opencensus::stats::MeasureInt64 start_count =
::opencensus::stats::MeasureInt64::Register(
kMeasureStartCount, kMeasureStartCount, kMeasureStartCount);
return start_count;
}
::opencensus::stats::MeasureInt64 MeasureEndCount() {
static const ::opencensus::stats::MeasureInt64 end_count =
::opencensus::stats::MeasureInt64::Register(
kMeasureEndCount, kMeasureEndCount, kMeasureEndCount);
return end_count;
}
::opencensus::stats::MeasureInt64 MeasureEndBytesSent() {
static const ::opencensus::stats::MeasureInt64 end_bytes_sent =
::opencensus::stats::MeasureInt64::Register(
kMeasureEndBytesSent, kMeasureEndBytesSent, kMeasureEndBytesSent);
return end_bytes_sent;
}
::opencensus::stats::MeasureInt64 MeasureEndBytesReceived() {
static const ::opencensus::stats::MeasureInt64 end_bytes_received =
::opencensus::stats::MeasureInt64::Register(kMeasureEndBytesReceived,
kMeasureEndBytesReceived,
kMeasureEndBytesReceived);
return end_bytes_received;
}
::opencensus::stats::MeasureInt64 MeasureEndLatencyMs() {
static const ::opencensus::stats::MeasureInt64 end_latency_ms =
::opencensus::stats::MeasureInt64::Register(
kMeasureEndLatencyMs, kMeasureEndLatencyMs, kMeasureEndLatencyMs);
return end_latency_ms;
}
::opencensus::stats::MeasureDouble MeasureOtherCallMetric() {
static const ::opencensus::stats::MeasureDouble other_call_metric =
::opencensus::stats::MeasureDouble::Register(kMeasureOtherCallMetric,
kMeasureOtherCallMetric,
kMeasureOtherCallMetric);
return other_call_metric;
}
// Tags.
opencensus::stats::TagKey TagKeyToken() {
static const auto token = opencensus::stats::TagKey::Register(kTagKeyToken);
return token;
}
opencensus::stats::TagKey TagKeyHost() {
static const auto token = opencensus::stats::TagKey::Register(kTagKeyHost);
return token;
}
opencensus::stats::TagKey TagKeyUserId() {
static const auto token = opencensus::stats::TagKey::Register(kTagKeyUserId);
return token;
}
opencensus::stats::TagKey TagKeyStatus() {
static const auto token = opencensus::stats::TagKey::Register(kTagKeyStatus);
return token;
}
opencensus::stats::TagKey TagKeyMetricName() {
static const auto token =
opencensus::stats::TagKey::Register(kTagKeyMetricName);
return token;
}
} // namespace load_reporter
} // namespace grpc
#endif /* GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_REGISTERED_OPENCENSUS_OBJECTS_H \
*/

@ -18,203 +18,311 @@
#include <grpc/support/port_platform.h>
#include <string.h>
#include <grpc/grpc_security.h>
#include <grpc/load_reporting.h>
#include <grpc/slice.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include <grpc/support/sync.h>
#include "src/core/ext/filters/client_channel/parse_address.h"
#include "src/core/ext/filters/client_channel/uri_parser.h"
#include "src/core/ext/filters/load_reporting/registered_opencensus_objects.h"
#include "src/core/ext/filters/load_reporting/server_load_reporting_filter.h"
#include "src/core/ext/filters/load_reporting/server_load_reporting_plugin.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/context.h"
#include "src/core/lib/gpr/string.h"
#include "src/core/lib/iomgr/resolve_address.h"
#include "src/core/lib/iomgr/sockaddr_posix.h"
#include "src/core/lib/iomgr/socket_utils.h"
#include "src/core/lib/profiling/timers.h"
#include "src/core/lib/security/context/security_context.h"
#include "src/core/lib/slice/slice_internal.h"
#include "src/core/lib/surface/call.h"
#include "src/core/lib/transport/static_metadata.h"
namespace {
struct call_data {
intptr_t id; /**< an id unique to the call */
bool have_trailing_md_string;
grpc_slice trailing_md_string;
bool have_initial_md_string;
grpc_slice initial_md_string;
bool have_service_method;
grpc_slice service_method;
/* stores the recv_initial_metadata op's ready closure, which we wrap with our
* own (on_initial_md_ready) in order to capture the incoming initial metadata
* */
grpc_closure* ops_recv_initial_metadata_ready;
namespace grpc {
/* to get notified of the availability of the incoming initial metadata. */
grpc_closure on_initial_md_ready;
grpc_metadata_batch* recv_initial_metadata;
};
grpc_error* ServerLoadReportingChannelData::Init(
grpc_channel_element* /* elem */, grpc_channel_element_args* args) {
GPR_ASSERT(!args->is_last);
// Find and record the peer_identity.
const grpc_auth_context* auth_context =
grpc_find_auth_context_in_args(args->channel_args);
if (auth_context != nullptr &&
grpc_auth_context_peer_is_authenticated(auth_context)) {
grpc_auth_property_iterator auth_it =
grpc_auth_context_peer_identity(auth_context);
const grpc_auth_property* auth_property =
grpc_auth_property_iterator_next(&auth_it);
if (auth_property != nullptr) {
peer_identity_ = auth_property->value;
peer_identity_len_ = auth_property->value_length;
}
}
return GRPC_ERROR_NONE;
}
struct channel_data {
intptr_t id; /**< an id unique to the channel */
};
} // namespace
void ServerLoadReportingCallData::Destroy(
grpc_call_element* elem, const grpc_call_final_info* final_info,
grpc_closure* then_call_closure) {
ServerLoadReportingChannelData* chand =
reinterpret_cast<ServerLoadReportingChannelData*>(elem->channel_data);
// Only record an end if we've recorded its corresponding start, which is
// indicated by a non-null client_ip_and_lr_token_. Note that it's possible
// that we attempt to record the call end before we have recorded the call
// start, because the data needed for recording the start comes from the
// initial metadata, which may not be ready before the call finishes.
if (client_ip_and_lr_token_ != nullptr) {
opencensus::stats::Record(
{{::grpc::load_reporter::MeasureEndCount(), 1},
{::grpc::load_reporter::MeasureEndBytesSent(),
final_info->stats.transport_stream_stats.outgoing.data_bytes},
{::grpc::load_reporter::MeasureEndBytesReceived(),
final_info->stats.transport_stream_stats.incoming.data_bytes},
{::grpc::load_reporter::MeasureEndLatencyMs(),
gpr_time_to_millis(final_info->stats.latency)}},
{{::grpc::load_reporter::TagKeyToken(),
{client_ip_and_lr_token_, client_ip_and_lr_token_len_}},
{::grpc::load_reporter::TagKeyHost(),
{target_host_, target_host_len_}},
{::grpc::load_reporter::TagKeyUserId(),
{chand->peer_identity(), chand->peer_identity_len()}},
{::grpc::load_reporter::TagKeyStatus(),
GetStatusTagForStatus(final_info->final_status)}});
}
grpc_slice_unref_internal(service_method_);
}
static void on_initial_md_ready(void* user_data, grpc_error* err) {
grpc_call_element* elem = static_cast<grpc_call_element*>(user_data);
call_data* calld = static_cast<call_data*>(elem->call_data);
void ServerLoadReportingCallData::StartTransportStreamOpBatch(
grpc_call_element* elem, TransportStreamOpBatch* op) {
GPR_TIMER_SCOPE("lr_start_transport_stream_op", 0);
if (op->recv_initial_metadata() != nullptr) {
// Save some fields to use when initial metadata is ready.
peer_string_ = op->get_peer_string();
recv_initial_metadata_ = op->recv_initial_metadata();
original_recv_initial_metadata_ready_ = op->recv_initial_metadata_ready();
// Substitute the original closure for the wrapper closure.
op->set_recv_initial_metadata_ready(&recv_initial_metadata_ready_);
} else if (op->send_trailing_metadata() != nullptr) {
GRPC_LOG_IF_ERROR(
"server_load_reporting_filter",
grpc_metadata_batch_filter(op->send_trailing_metadata()->batch(),
SendTrailingMetadataFilter, elem,
"send_trailing_metadata filtering error"));
}
grpc_call_next_op(elem, op->op());
}
if (err == GRPC_ERROR_NONE) {
if (calld->recv_initial_metadata->idx.named.path != nullptr) {
calld->service_method = grpc_slice_ref_internal(
GRPC_MDVALUE(calld->recv_initial_metadata->idx.named.path->md));
calld->have_service_method = true;
} else {
err = grpc_error_add_child(
err, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Missing :path header"));
void ServerLoadReportingCallData::GetCensusSafeClientIpString(
char** client_ip_string, size_t* size) {
// Find the client URI string.
const char* client_uri_str =
reinterpret_cast<const char*>(gpr_atm_acq_load(peer_string_));
if (client_uri_str == nullptr) {
gpr_log(GPR_ERROR,
"Unable to extract client URI string (peer string) from gRPC "
"metadata.");
*client_ip_string = nullptr;
*size = 0;
return;
}
// Parse the client URI string into grpc_uri.
grpc_uri* client_uri = grpc_uri_parse(client_uri_str, true);
if (client_uri == nullptr) {
gpr_log(GPR_ERROR,
"Unable to parse the client URI string (peer string) to a client "
"URI.");
*client_ip_string = nullptr;
*size = 0;
return;
}
if (calld->recv_initial_metadata->idx.named.lb_token != nullptr) {
calld->initial_md_string = grpc_slice_ref_internal(
GRPC_MDVALUE(calld->recv_initial_metadata->idx.named.lb_token->md));
calld->have_initial_md_string = true;
grpc_metadata_batch_remove(
calld->recv_initial_metadata,
calld->recv_initial_metadata->idx.named.lb_token);
// Parse the client URI into grpc_resolved_address.
grpc_resolved_address resolved_address;
bool success = grpc_parse_uri(client_uri, &resolved_address);
grpc_uri_destroy(client_uri);
if (!success) {
gpr_log(GPR_ERROR,
"Unable to parse client URI into a grpc_resolved_address.");
*client_ip_string = nullptr;
*size = 0;
return;
}
// Convert the socket address in the grpc_resolved_address into a hex string
// according to the address family.
grpc_sockaddr* addr = reinterpret_cast<grpc_sockaddr*>(resolved_address.addr);
if (addr->sa_family == GRPC_AF_INET) {
grpc_sockaddr_in* addr4 = reinterpret_cast<grpc_sockaddr_in*>(addr);
gpr_asprintf(client_ip_string, "%08x", grpc_ntohl(addr4->sin_addr.s_addr));
*size = 8;
} else if (addr->sa_family == GRPC_AF_INET6) {
grpc_sockaddr_in6* addr6 = reinterpret_cast<grpc_sockaddr_in6*>(addr);
*client_ip_string = static_cast<char*>(gpr_malloc(32));
for (size_t i = 0; i < 16; ++i) {
sprintf(*client_ip_string + i, "%02x",
addr6->sin6_addr.__in6_u.__u6_addr8[i]);
}
*size = 32;
} else {
GRPC_ERROR_REF(err);
GPR_UNREACHABLE_CODE();
}
GRPC_CLOSURE_RUN(calld->ops_recv_initial_metadata_ready, err);
}
/* Constructor for call_data */
static grpc_error* init_call_elem(grpc_call_element* elem,
const grpc_call_element_args* args) {
call_data* calld = static_cast<call_data*>(elem->call_data);
calld->id = (intptr_t)args->call_stack;
GRPC_CLOSURE_INIT(&calld->on_initial_md_ready, on_initial_md_ready, elem,
grpc_schedule_on_exec_ctx);
/* TODO(dgq): do something with the data
channel_data *chand = elem->channel_data;
grpc_load_reporting_call_data lr_call_data = {GRPC_LR_POINT_CALL_CREATION,
(intptr_t)chand->id,
(intptr_t)calld->id,
NULL,
NULL,
NULL,
NULL};
*/
return GRPC_ERROR_NONE;
void ServerLoadReportingCallData::StoreClientIpAndLrToken(const char* lr_token,
size_t lr_token_len) {
char* client_ip;
size_t client_ip_len;
GetCensusSafeClientIpString(&client_ip, &client_ip_len);
client_ip_and_lr_token_len_ =
kLengthPrefixSize + client_ip_len + lr_token_len;
client_ip_and_lr_token_ = static_cast<char*>(
gpr_zalloc(client_ip_and_lr_token_len_ * sizeof(char)));
char* cur_pos = client_ip_and_lr_token_;
// Store the IP length prefix.
if (client_ip_len == 0) {
strncpy(cur_pos, kEmptyAddressLengthString, kLengthPrefixSize);
} else if (client_ip_len == 8) {
strncpy(cur_pos, kEncodedIpv4AddressLengthString, kLengthPrefixSize);
} else if (client_ip_len == 32) {
strncpy(cur_pos, kEncodedIpv6AddressLengthString, kLengthPrefixSize);
} else {
GPR_UNREACHABLE_CODE();
}
cur_pos += kLengthPrefixSize;
// Store the IP.
if (client_ip_len != 0) {
strncpy(cur_pos, client_ip, client_ip_len);
}
gpr_free(client_ip);
cur_pos += client_ip_len;
// Store the LR token.
if (lr_token_len != 0) {
strncpy(cur_pos, lr_token, lr_token_len);
}
GPR_ASSERT(cur_pos + lr_token_len - client_ip_and_lr_token_ ==
client_ip_and_lr_token_len_);
}
/* Destructor for call_data */
static void destroy_call_elem(grpc_call_element* elem,
const grpc_call_final_info* final_info,
grpc_closure* ignored) {
call_data* calld = static_cast<call_data*>(elem->call_data);
/* TODO(dgq): do something with the data
channel_data *chand = elem->channel_data;
grpc_load_reporting_call_data lr_call_data = {GRPC_LR_POINT_CALL_DESTRUCTION,
(intptr_t)chand->id,
(intptr_t)calld->id,
final_info,
calld->initial_md_string,
calld->trailing_md_string,
calld->service_method};
*/
if (calld->have_initial_md_string) {
grpc_slice_unref_internal(calld->initial_md_string);
grpc_filtered_mdelem ServerLoadReportingCallData::RecvInitialMetadataFilter(
void* user_data, grpc_mdelem md) {
grpc_call_element* elem = reinterpret_cast<grpc_call_element*>(user_data);
ServerLoadReportingCallData* calld =
reinterpret_cast<ServerLoadReportingCallData*>(elem->call_data);
if (grpc_slice_eq(GRPC_MDKEY(md), GRPC_MDSTR_PATH)) {
calld->service_method_ = grpc_slice_ref_internal(GRPC_MDVALUE(md));
} else if (calld->target_host_ == nullptr &&
grpc_slice_eq(GRPC_MDKEY(md), GRPC_MDSTR_AUTHORITY)) {
grpc_slice target_host_slice = GRPC_MDVALUE(md);
calld->target_host_len_ = GRPC_SLICE_LENGTH(target_host_slice);
calld->target_host_ =
reinterpret_cast<char*>(gpr_zalloc(calld->target_host_len_));
for (size_t i = 0; i < calld->target_host_len_; ++i) {
calld->target_host_[i] = static_cast<char>(
tolower(GRPC_SLICE_START_PTR(target_host_slice)[i]));
}
if (calld->have_trailing_md_string) {
grpc_slice_unref_internal(calld->trailing_md_string);
} else if (grpc_slice_eq(GRPC_MDKEY(md), GRPC_MDSTR_LB_TOKEN)) {
if (calld->client_ip_and_lr_token_ == nullptr) {
calld->StoreClientIpAndLrToken(
reinterpret_cast<const char*> GRPC_SLICE_START_PTR(GRPC_MDVALUE(md)),
GRPC_SLICE_LENGTH(GRPC_MDVALUE(md)));
}
if (calld->have_service_method) {
grpc_slice_unref_internal(calld->service_method);
return GRPC_FILTERED_REMOVE();
}
return GRPC_FILTERED_MDELEM(md);
}
/* Constructor for channel_data */
static grpc_error* init_channel_elem(grpc_channel_element* elem,
grpc_channel_element_args* args) {
GPR_ASSERT(!args->is_last);
channel_data* chand = static_cast<channel_data*>(elem->channel_data);
chand->id = (intptr_t)args->channel_stack;
/* TODO(dgq): do something with the data
grpc_load_reporting_call_data lr_call_data = {GRPC_LR_POINT_CHANNEL_CREATION,
(intptr_t)chand,
0,
NULL,
NULL,
NULL,
NULL};
*/
return GRPC_ERROR_NONE;
void ServerLoadReportingCallData::RecvInitialMetadataReady(void* arg,
grpc_error* err) {
grpc_call_element* elem = reinterpret_cast<grpc_call_element*>(arg);
ServerLoadReportingCallData* calld =
reinterpret_cast<ServerLoadReportingCallData*>(elem->call_data);
ServerLoadReportingChannelData* chand =
reinterpret_cast<ServerLoadReportingChannelData*>(elem->channel_data);
if (err == GRPC_ERROR_NONE) {
GRPC_LOG_IF_ERROR(
"server_load_reporting_filter",
grpc_metadata_batch_filter(calld->recv_initial_metadata_->batch(),
RecvInitialMetadataFilter, elem,
"recv_initial_metadata filtering error"));
// If the LB token was not found in the recv_initial_metadata, only the
// client IP part will be recorded (with an empty LB token).
if (calld->client_ip_and_lr_token_ == nullptr) {
calld->StoreClientIpAndLrToken(nullptr, 0);
}
opencensus::stats::Record(
{{::grpc::load_reporter::MeasureStartCount(), 1}},
{{::grpc::load_reporter::TagKeyToken(),
{calld->client_ip_and_lr_token_, calld->client_ip_and_lr_token_len_}},
{::grpc::load_reporter::TagKeyHost(),
{calld->target_host_, calld->target_host_len_}},
{::grpc::load_reporter::TagKeyUserId(),
{chand->peer_identity(), chand->peer_identity_len()}}});
}
GRPC_CLOSURE_RUN(calld->original_recv_initial_metadata_ready_,
GRPC_ERROR_REF(err));
}
/* Destructor for channel data */
static void destroy_channel_elem(grpc_channel_element* elem) {
/* TODO(dgq): do something with the data
channel_data *chand = elem->channel_data;
grpc_load_reporting_call_data lr_call_data = {
GRPC_LR_POINT_CHANNEL_DESTRUCTION,
(intptr_t)chand->id,
0,
NULL,
NULL,
NULL,
NULL};
*/
grpc_error* ServerLoadReportingCallData::Init(
grpc_call_element* elem, const grpc_call_element_args* args) {
service_method_ = grpc_empty_slice();
GRPC_CLOSURE_INIT(&recv_initial_metadata_ready_, RecvInitialMetadataReady,
elem, grpc_schedule_on_exec_ctx);
return GRPC_ERROR_NONE;
}
static grpc_filtered_mdelem lr_trailing_md_filter(void* user_data,
grpc_mdelem md) {
grpc_call_element* elem = static_cast<grpc_call_element*>(user_data);
call_data* calld = static_cast<call_data*>(elem->call_data);
grpc_filtered_mdelem ServerLoadReportingCallData::SendTrailingMetadataFilter(
void* user_data, grpc_mdelem md) {
grpc_call_element* elem = reinterpret_cast<grpc_call_element*>(user_data);
ServerLoadReportingCallData* calld =
reinterpret_cast<ServerLoadReportingCallData*>(elem->call_data);
ServerLoadReportingChannelData* chand =
reinterpret_cast<ServerLoadReportingChannelData*>(elem->channel_data);
// TODO(juanlishen): GRPC_MDSTR_LB_COST_BIN meaning?
if (grpc_slice_eq(GRPC_MDKEY(md), GRPC_MDSTR_LB_COST_BIN)) {
calld->trailing_md_string = GRPC_MDVALUE(md);
const grpc_slice value = GRPC_MDVALUE(md);
const size_t cost_entry_size = GRPC_SLICE_LENGTH(value);
if (cost_entry_size < sizeof(double)) {
gpr_log(GPR_ERROR,
"Cost metadata value too small (%zu bytes) to hold valid data. "
"Ignoring.",
cost_entry_size);
return GRPC_FILTERED_REMOVE();
}
const double* cost_entry_ptr =
reinterpret_cast<const double*>(GRPC_SLICE_START_PTR(value));
double cost_value = *cost_entry_ptr++;
const char* cost_name = reinterpret_cast<const char*>(cost_entry_ptr);
const size_t cost_name_len = cost_entry_size - sizeof(double);
opencensus::stats::Record(
{{::grpc::load_reporter::MeasureOtherCallMetric(), cost_value}},
{{::grpc::load_reporter::TagKeyToken(),
{calld->client_ip_and_lr_token_, calld->client_ip_and_lr_token_len_}},
{::grpc::load_reporter::TagKeyHost(),
{calld->target_host_, calld->target_host_len_}},
{::grpc::load_reporter::TagKeyUserId(),
{chand->peer_identity(), chand->peer_identity_len()}},
{::grpc::load_reporter::TagKeyMetricName(),
{cost_name, cost_name_len}}});
return GRPC_FILTERED_REMOVE();
}
return GRPC_FILTERED_MDELEM(md);
}
static void lr_start_transport_stream_op_batch(
grpc_call_element* elem, grpc_transport_stream_op_batch* op) {
GPR_TIMER_SCOPE("lr_start_transport_stream_op_batch", 0);
call_data* calld = static_cast<call_data*>(elem->call_data);
if (op->recv_initial_metadata) {
/* substitute our callback for the higher callback */
calld->recv_initial_metadata =
op->payload->recv_initial_metadata.recv_initial_metadata;
calld->ops_recv_initial_metadata_ready =
op->payload->recv_initial_metadata.recv_initial_metadata_ready;
op->payload->recv_initial_metadata.recv_initial_metadata_ready =
&calld->on_initial_md_ready;
} else if (op->send_trailing_metadata) {
GRPC_LOG_IF_ERROR(
"grpc_metadata_batch_filter",
grpc_metadata_batch_filter(
op->payload->send_trailing_metadata.send_trailing_metadata,
lr_trailing_md_filter, elem,
"LR trailing metadata filtering error"));
const char* ServerLoadReportingCallData::GetStatusTagForStatus(
grpc_status_code status) {
switch (status) {
case GRPC_STATUS_OK:
return ::grpc::load_reporter::kCallStatusOk;
case GRPC_STATUS_UNKNOWN:
case GRPC_STATUS_DEADLINE_EXCEEDED:
case GRPC_STATUS_UNIMPLEMENTED:
case GRPC_STATUS_INTERNAL:
case GRPC_STATUS_UNAVAILABLE:
case GRPC_STATUS_DATA_LOSS:
return ::grpc::load_reporter::kCallStatusServerError;
default:
return ::grpc::load_reporter::kCallStatusClientError;
}
grpc_call_next_op(elem, op);
}
const grpc_channel_filter grpc_server_load_reporting_filter = {
lr_start_transport_stream_op_batch,
grpc_channel_next_op,
sizeof(call_data),
init_call_elem,
grpc_call_stack_ignore_set_pollset_or_pollset_set,
destroy_call_elem,
sizeof(channel_data),
init_channel_elem,
destroy_channel_elem,
grpc_channel_next_get_info,
"load_reporting"};
} // namespace grpc

@ -21,10 +21,105 @@
#include <grpc/support/port_platform.h>
#include "src/core/ext/filters/load_reporting/server_load_reporting_plugin.h"
#include "src/core/lib/channel/channel_stack.h"
#include "src/cpp/common/channel_filter.h"
extern const grpc_channel_filter grpc_server_load_reporting_filter;
namespace grpc {
class ServerLoadReportingChannelData : public ChannelData {
public:
grpc_error* Init(grpc_channel_element* elem,
grpc_channel_element_args* args) override;
// Getters.
const char* peer_identity() { return peer_identity_; }
size_t peer_identity_len() { return peer_identity_len_; }
private:
// The peer's authenticated identity.
char* peer_identity_ = nullptr;
size_t peer_identity_len_ = 0;
};
class ServerLoadReportingCallData : public CallData {
public:
grpc_error* Init(grpc_call_element* elem,
const grpc_call_element_args* args) override;
void Destroy(grpc_call_element* elem, const grpc_call_final_info* final_info,
grpc_closure* then_call_closure) override;
void StartTransportStreamOpBatch(grpc_call_element* elem,
TransportStreamOpBatch* op) override;
private:
// From the peer_string_ in calld, extracts the client IP string (owned by
// caller), e.g., "01020a0b". Upon failure, set the output pointer to null and
// size to zero.
void GetCensusSafeClientIpString(char** client_ip_string, size_t* size);
// Concatenates the client IP address and the load reporting token, then
// stores the result into the call data.
void StoreClientIpAndLrToken(const char* lr_token, size_t lr_token_len);
// This matches the classification of the status codes in
// googleapis/google/rpc/code.proto.
static const char* GetStatusTagForStatus(grpc_status_code status);
// Records the call start.
static void RecvInitialMetadataReady(void* arg, grpc_error* err);
// From the initial metadata, extracts the service_method_, target_host_, and
// client_ip_and_lr_token_.
static grpc_filtered_mdelem RecvInitialMetadataFilter(void* user_data,
grpc_mdelem md);
// Records the other call metrics.
static grpc_filtered_mdelem SendTrailingMetadataFilter(void* user_data,
grpc_mdelem md);
// The peer string (a member of the recv_initial_metadata op). Note that
// gpr_atm itself is a pointer type here, making "peer_string_" effectively a
// double pointer.
const gpr_atm* peer_string_;
// The received initial metadata (a member of the recv_initial_metadata op).
// When it is ready, we will extract some data from it via
// recv_initial_metadata_ready_ closure, before the original
// recv_initial_metadata_ready closure,
MetadataBatch* recv_initial_metadata_;
// The original recv_initial_metadata closure, which is wrapped by our own
// closure (recv_initial_metadata_ready_) to capture the incoming initial
// metadata.
grpc_closure* original_recv_initial_metadata_ready_;
// The closure that wraps the original closure. Scheduled when
// recv_initial_metadata_ is ready.
grpc_closure recv_initial_metadata_ready_;
// Corresponds to the :path header.
grpc_slice service_method_;
// The backend host that the client thinks it's talking to. This may be
// different from the actual backend in the case of, for example,
// load-balanced targets. We store a copy of the metadata slice in order to
// lowercase it. */
char* target_host_;
size_t target_host_len_;
// The client IP address (including a length prefix) and the load reporting
// token.
char* client_ip_and_lr_token_;
size_t client_ip_and_lr_token_len_;
static constexpr char kEncodedIpv4AddressLengthString[] = "08";
static constexpr char kEncodedIpv6AddressLengthString[] = "32";
static constexpr char kEmptyAddressLengthString[] = "00";
static constexpr size_t kLengthPrefixSize = 2;
};
} // namespace grpc
#endif /* GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_SERVER_LOAD_REPORTING_FILTER_H \
*/

@ -1,71 +0,0 @@
/*
*
* Copyright 2016 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 <limits.h>
#include <string.h>
#include <grpc/load_reporting.h>
#include <grpc/support/alloc.h>
#include <grpc/support/sync.h>
#include "src/core/ext/filters/load_reporting/server_load_reporting_filter.h"
#include "src/core/ext/filters/load_reporting/server_load_reporting_plugin.h"
#include "src/core/lib/channel/channel_stack_builder.h"
#include "src/core/lib/slice/slice_internal.h"
#include "src/core/lib/surface/call.h"
#include "src/core/lib/surface/channel_init.h"
static bool is_load_reporting_enabled(const grpc_channel_args* a) {
return grpc_channel_arg_get_bool(
grpc_channel_args_find(a, GRPC_ARG_ENABLE_LOAD_REPORTING), false);
}
static bool maybe_add_server_load_reporting_filter(
grpc_channel_stack_builder* builder, void* arg) {
const grpc_channel_args* args =
grpc_channel_stack_builder_get_channel_arguments(builder);
const grpc_channel_filter* filter =
static_cast<const grpc_channel_filter*>(arg);
grpc_channel_stack_builder_iterator* it =
grpc_channel_stack_builder_iterator_find(builder, filter->name);
const bool already_has_load_reporting_filter =
!grpc_channel_stack_builder_iterator_is_end(it);
grpc_channel_stack_builder_iterator_destroy(it);
if (is_load_reporting_enabled(args) && !already_has_load_reporting_filter) {
return grpc_channel_stack_builder_prepend_filter(builder, filter, nullptr,
nullptr);
}
return true;
}
grpc_arg grpc_load_reporting_enable_arg() {
return grpc_channel_arg_integer_create((char*)GRPC_ARG_ENABLE_LOAD_REPORTING,
1);
}
/* Plugin registration */
void grpc_server_load_reporting_plugin_init(void) {
grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, INT_MAX,
maybe_add_server_load_reporting_filter,
(void*)&grpc_server_load_reporting_filter);
}
void grpc_server_load_reporting_plugin_shutdown() {}

@ -1,61 +0,0 @@
/*
*
* Copyright 2016 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_CORE_EXT_FILTERS_LOAD_REPORTING_SERVER_LOAD_REPORTING_PLUGIN_H
#define GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_SERVER_LOAD_REPORTING_PLUGIN_H
#include <grpc/support/port_platform.h>
#include <grpc/impl/codegen/grpc_types.h>
#include "src/core/lib/channel/channel_stack.h"
/** Identifiers for the invocation point of the users LR callback */
typedef enum grpc_load_reporting_source {
GRPC_LR_POINT_UNKNOWN = 0,
GRPC_LR_POINT_CHANNEL_CREATION,
GRPC_LR_POINT_CHANNEL_DESTRUCTION,
GRPC_LR_POINT_CALL_CREATION,
GRPC_LR_POINT_CALL_DESTRUCTION
} grpc_load_reporting_source;
/** Call information to be passed to the provided LR callback. */
typedef struct grpc_load_reporting_call_data {
const grpc_load_reporting_source source; /**< point of last data update. */
/** Unique identifier for the channel associated with the data */
intptr_t channel_id;
/** Unique identifier for the call associated with the data. If the call
* hasn't been created yet, it'll have a value of zero. */
intptr_t call_id;
/** Only valid when \a source is \a GRPC_LR_POINT_CALL_DESTRUCTION, that is,
* once the call has completed */
const grpc_call_final_info* final_info;
const char* initial_md_string; /**< value string for LR's initial md key */
const char* trailing_md_string; /**< value string for LR's trailing md key */
const char* method_name; /**< Corresponds to :path header */
} grpc_load_reporting_call_data;
/** Return a \a grpc_arg enabling load reporting */
grpc_arg grpc_load_reporting_enable_arg();
#endif /* GRPC_CORE_EXT_FILTERS_LOAD_REPORTING_SERVER_LOAD_REPORTING_PLUGIN_H \
*/

@ -30,8 +30,6 @@ void grpc_client_channel_init(void);
void grpc_client_channel_shutdown(void);
void grpc_tsi_alts_init(void);
void grpc_tsi_alts_shutdown(void);
void grpc_server_load_reporting_plugin_init(void);
void grpc_server_load_reporting_plugin_shutdown(void);
void grpc_register_built_in_plugins(void) {
grpc_register_plugin(grpc_http_filters_init,
@ -44,6 +42,4 @@ void grpc_register_built_in_plugins(void) {
grpc_client_channel_shutdown);
grpc_register_plugin(grpc_tsi_alts_init,
grpc_tsi_alts_shutdown);
grpc_register_plugin(grpc_server_load_reporting_plugin_init,
grpc_server_load_reporting_plugin_shutdown);
}

@ -46,8 +46,6 @@ void grpc_resolver_dns_native_init(void);
void grpc_resolver_dns_native_shutdown(void);
void grpc_resolver_sockaddr_init(void);
void grpc_resolver_sockaddr_shutdown(void);
void grpc_server_load_reporting_plugin_init(void);
void grpc_server_load_reporting_plugin_shutdown(void);
void grpc_max_age_filter_init(void);
void grpc_max_age_filter_shutdown(void);
void grpc_message_size_filter_init(void);
@ -84,8 +82,6 @@ void grpc_register_built_in_plugins(void) {
grpc_resolver_dns_native_shutdown);
grpc_register_plugin(grpc_resolver_sockaddr_init,
grpc_resolver_sockaddr_shutdown);
grpc_register_plugin(grpc_server_load_reporting_plugin_init,
grpc_server_load_reporting_plugin_shutdown);
grpc_register_plugin(grpc_max_age_filter_init,
grpc_max_age_filter_shutdown);
grpc_register_plugin(grpc_message_size_filter_init,

@ -38,8 +38,6 @@ void grpc_resolver_sockaddr_init(void);
void grpc_resolver_sockaddr_shutdown(void);
void grpc_resolver_fake_init(void);
void grpc_resolver_fake_shutdown(void);
void grpc_server_load_reporting_plugin_init(void);
void grpc_server_load_reporting_plugin_shutdown(void);
void grpc_lb_policy_grpclb_init(void);
void grpc_lb_policy_grpclb_shutdown(void);
void grpc_lb_policy_pick_first_init(void);
@ -74,8 +72,6 @@ void grpc_register_built_in_plugins(void) {
grpc_resolver_sockaddr_shutdown);
grpc_register_plugin(grpc_resolver_fake_init,
grpc_resolver_fake_shutdown);
grpc_register_plugin(grpc_server_load_reporting_plugin_init,
grpc_server_load_reporting_plugin_shutdown);
grpc_register_plugin(grpc_lb_policy_grpclb_init,
grpc_lb_policy_grpclb_shutdown);
grpc_register_plugin(grpc_lb_policy_pick_first_init,

@ -207,6 +207,18 @@ class TransportStreamOpBatch {
op_->payload->context[GRPC_CONTEXT_TRACING].value);
}
const gpr_atm* get_peer_string() const {
if (op_->send_initial_metadata &&
op_->payload->send_initial_metadata.peer_string != nullptr) {
return op_->payload->send_initial_metadata.peer_string;
} else if (op_->recv_initial_metadata &&
op_->payload->recv_initial_metadata.peer_string != nullptr) {
return op_->payload->recv_initial_metadata.peer_string;
} else {
return nullptr;
}
}
private:
grpc_transport_stream_op_batch* op_; // Not owned.
MetadataBatch send_initial_metadata_;

@ -352,8 +352,6 @@ CORE_SOURCE_FILES = [
'src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc',
'src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc',
'src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc',
'src/core/ext/filters/load_reporting/server_load_reporting_filter.cc',
'src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc',
'src/cpp/ext/filters/census/grpc_context.cc',
'src/core/ext/filters/max_age/max_age_filter.cc',
'src/core/ext/filters/message_size/message_size_filter.cc',

@ -88,8 +88,6 @@ extern void keepalive_timeout(grpc_end2end_test_config config);
extern void keepalive_timeout_pre_init(void);
extern void large_metadata(grpc_end2end_test_config config);
extern void large_metadata_pre_init(void);
extern void load_reporting_hook(grpc_end2end_test_config config);
extern void load_reporting_hook_pre_init(void);
extern void max_concurrent_streams(grpc_end2end_test_config config);
extern void max_concurrent_streams_pre_init(void);
extern void max_connection_age(grpc_end2end_test_config config);
@ -220,7 +218,6 @@ void grpc_end2end_tests_pre_init(void) {
invoke_large_request_pre_init();
keepalive_timeout_pre_init();
large_metadata_pre_init();
load_reporting_hook_pre_init();
max_concurrent_streams_pre_init();
max_connection_age_pre_init();
max_connection_idle_pre_init();
@ -307,7 +304,6 @@ void grpc_end2end_tests(int argc, char **argv,
invoke_large_request(config);
keepalive_timeout(config);
large_metadata(config);
load_reporting_hook(config);
max_concurrent_streams(config);
max_connection_age(config);
max_connection_idle(config);
@ -476,10 +472,6 @@ void grpc_end2end_tests(int argc, char **argv,
large_metadata(config);
continue;
}
if (0 == strcmp("load_reporting_hook", argv[i])) {
load_reporting_hook(config);
continue;
}
if (0 == strcmp("max_concurrent_streams", argv[i])) {
max_concurrent_streams(config);
continue;

@ -90,8 +90,6 @@ extern void keepalive_timeout(grpc_end2end_test_config config);
extern void keepalive_timeout_pre_init(void);
extern void large_metadata(grpc_end2end_test_config config);
extern void large_metadata_pre_init(void);
extern void load_reporting_hook(grpc_end2end_test_config config);
extern void load_reporting_hook_pre_init(void);
extern void max_concurrent_streams(grpc_end2end_test_config config);
extern void max_concurrent_streams_pre_init(void);
extern void max_connection_age(grpc_end2end_test_config config);
@ -223,7 +221,6 @@ void grpc_end2end_tests_pre_init(void) {
invoke_large_request_pre_init();
keepalive_timeout_pre_init();
large_metadata_pre_init();
load_reporting_hook_pre_init();
max_concurrent_streams_pre_init();
max_connection_age_pre_init();
max_connection_idle_pre_init();
@ -311,7 +308,6 @@ void grpc_end2end_tests(int argc, char **argv,
invoke_large_request(config);
keepalive_timeout(config);
large_metadata(config);
load_reporting_hook(config);
max_concurrent_streams(config);
max_connection_age(config);
max_connection_idle(config);
@ -484,10 +480,6 @@ void grpc_end2end_tests(int argc, char **argv,
large_metadata(config);
continue;
}
if (0 == strcmp("load_reporting_hook", argv[i])) {
load_reporting_hook(config);
continue;
}
if (0 == strcmp("max_concurrent_streams", argv[i])) {
max_concurrent_streams(config);
continue;

@ -1,120 +0,0 @@
/*
*
* Copyright 2016 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 "test/core/end2end/end2end_tests.h"
#include <string.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/sync.h>
#include "src/core/ext/filters/client_channel/client_channel.h"
#include "src/core/ext/filters/http/server/http_server_filter.h"
#include "src/core/ext/filters/load_reporting/server_load_reporting_plugin.h"
#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/connected_channel.h"
#include "src/core/lib/gpr/host_port.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/server.h"
#include "test/core/util/port.h"
#include "test/core/util/test_config.h"
typedef struct load_reporting_fixture_data {
char* localaddr;
} load_reporting_fixture_data;
static grpc_end2end_test_fixture chttp2_create_fixture_load_reporting(
grpc_channel_args* client_args, grpc_channel_args* server_args) {
grpc_end2end_test_fixture f;
int port = grpc_pick_unused_port_or_die();
load_reporting_fixture_data* ffd = static_cast<load_reporting_fixture_data*>(
gpr_malloc(sizeof(load_reporting_fixture_data)));
memset(&f, 0, sizeof(f));
gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd;
f.cq = grpc_completion_queue_create_for_next(nullptr);
f.shutdown_cq = grpc_completion_queue_create_for_pluck(nullptr);
return f;
}
void chttp2_init_client_load_reporting(grpc_end2end_test_fixture* f,
grpc_channel_args* client_args) {
load_reporting_fixture_data* ffd =
static_cast<load_reporting_fixture_data*>(f->fixture_data);
f->client =
grpc_insecure_channel_create(ffd->localaddr, client_args, nullptr);
GPR_ASSERT(f->client);
}
void chttp2_init_server_load_reporting(grpc_end2end_test_fixture* f,
grpc_channel_args* server_args) {
load_reporting_fixture_data* ffd =
static_cast<load_reporting_fixture_data*>(f->fixture_data);
grpc_arg arg = grpc_load_reporting_enable_arg();
if (f->server) {
grpc_server_destroy(f->server);
}
server_args = grpc_channel_args_copy_and_add(server_args, &arg, 1);
f->server = grpc_server_create(server_args, nullptr);
{
grpc_core::ExecCtx exec_ctx;
grpc_channel_args_destroy(server_args);
}
grpc_server_register_completion_queue(f->server, f->cq, nullptr);
GPR_ASSERT(grpc_server_add_insecure_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
void chttp2_tear_down_load_reporting(grpc_end2end_test_fixture* f) {
load_reporting_fixture_data* ffd =
static_cast<load_reporting_fixture_data*>(f->fixture_data);
gpr_free(ffd->localaddr);
gpr_free(ffd);
}
/* All test configurations */
static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack+load_reporting",
FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
nullptr, chttp2_create_fixture_load_reporting,
chttp2_init_client_load_reporting, chttp2_init_server_load_reporting,
chttp2_tear_down_load_reporting},
};
int main(int argc, char** argv) {
size_t i;
grpc_test_init(argc, argv);
grpc_end2end_tests_pre_init();
grpc_init();
for (i = 0; i < sizeof(configs) / sizeof(*configs); i++) {
grpc_end2end_tests(argc, argv, configs[i]);
}
grpc_shutdown();
return 0;
}

@ -47,7 +47,9 @@ inproc_fixture_options = default_unsecure_fixture_options._replace(
END2END_FIXTURES = {
'h2_compress': default_unsecure_fixture_options._replace(enables_compression=True),
'h2_census': default_unsecure_fixture_options,
'h2_load_reporting': default_unsecure_fixture_options,
# This cmake target is disabled for now because it depends on OpenCensus,
# which is Bazel-only.
# 'h2_load_reporting': default_unsecure_fixture_options,
'h2_fakesec': default_secure_fixture_options._replace(ci_mac=False),
'h2_fd': fd_unsecure_fixture_options,
'h2_full': default_unsecure_fixture_options,
@ -143,7 +145,9 @@ END2END_TESTS = {
'no_logging': default_test_options._replace(traceable=False),
'no_op': default_test_options,
'payload': default_test_options,
'load_reporting_hook': default_test_options,
# This cmake target is disabled for now because it depends on OpenCensus,
# which is Bazel-only.
# 'load_reporting_hook': default_test_options,
'ping_pong_streaming': default_test_options._replace(cpu_cost=LOWCPU),
'ping': connectivity_test_options._replace(proxyable=False,
cpu_cost=LOWCPU),

@ -45,7 +45,9 @@ def fixture_options(fullstack=True, includes_proxy=False, dns_resolver=True,
END2END_FIXTURES = {
'h2_compress': fixture_options(),
'h2_census': fixture_options(),
'h2_load_reporting': fixture_options(),
# TODO(juanlishen): This is disabled for now, but should be considered to re-enable once we have
# decided how the load reporting service should be enabled.
#'h2_load_reporting': fixture_options(),
'h2_fakesec': fixture_options(),
'h2_fd': fixture_options(dns_resolver=False, fullstack=False,
client_channel=False,
@ -139,7 +141,10 @@ END2END_TESTS = {
'no_logging': test_options(traceable=False),
'no_op': test_options(),
'payload': test_options(),
'load_reporting_hook': test_options(),
# TODO(juanlishen): This is disabled for now because it depends on some generated functions in
# end2end_tests.cc, which are not generated because they would depend on OpenCensus while
# OpenCensus can only be built via Bazel so far.
# 'load_reporting_hook': test_options(),
'ping_pong_streaming': test_options(),
'ping': test_options(needs_fullstack=True, proxyable=False),
'proxy_auth': test_options(needs_proxy_auth=True),

@ -40,7 +40,6 @@
#include <grpc/impl/codegen/sync.h>
#include <grpc/impl/codegen/sync_custom.h>
#include <grpc/impl/codegen/sync_generic.h>
#include <grpc/load_reporting.h>
#include <grpc/slice.h>
#include <grpc/slice_buffer.h>
#include <grpc/status.h>

@ -604,10 +604,13 @@ BENCHMARK_TEMPLATE(BM_IsolatedFilter, HttpServerFilter, SendEmptyMetadata);
typedef Fixture<&grpc_message_size_filter, CHECKS_NOT_LAST> MessageSizeFilter;
BENCHMARK_TEMPLATE(BM_IsolatedFilter, MessageSizeFilter, NoOp);
BENCHMARK_TEMPLATE(BM_IsolatedFilter, MessageSizeFilter, SendEmptyMetadata);
typedef Fixture<&grpc_server_load_reporting_filter, CHECKS_NOT_LAST>
LoadReportingFilter;
BENCHMARK_TEMPLATE(BM_IsolatedFilter, LoadReportingFilter, NoOp);
BENCHMARK_TEMPLATE(BM_IsolatedFilter, LoadReportingFilter, SendEmptyMetadata);
// This cmake target is disabled for now because it depends on OpenCensus, which
// is Bazel-only.
// typedef Fixture<&grpc_server_load_reporting_filter, CHECKS_NOT_LAST>
// LoadReportingFilter;
// BENCHMARK_TEMPLATE(BM_IsolatedFilter, LoadReportingFilter, NoOp);
// BENCHMARK_TEMPLATE(BM_IsolatedFilter, LoadReportingFilter,
// SendEmptyMetadata);
////////////////////////////////////////////////////////////////////////////////
// Benchmarks isolating grpc_call

@ -900,7 +900,6 @@ include/grpc/impl/codegen/sync_custom.h \
include/grpc/impl/codegen/sync_generic.h \
include/grpc/impl/codegen/sync_posix.h \
include/grpc/impl/codegen/sync_windows.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \

@ -901,7 +901,6 @@ include/grpc/impl/codegen/sync_custom.h \
include/grpc/impl/codegen/sync_generic.h \
include/grpc/impl/codegen/sync_posix.h \
include/grpc/impl/codegen/sync_windows.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \

@ -840,7 +840,6 @@ include/grpc/impl/codegen/sync_posix.h \
include/grpc/impl/codegen/sync_posix.h \
include/grpc/impl/codegen/sync_windows.h \
include/grpc/impl/codegen/sync_windows.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \

@ -840,7 +840,6 @@ include/grpc/impl/codegen/sync_posix.h \
include/grpc/impl/codegen/sync_posix.h \
include/grpc/impl/codegen/sync_windows.h \
include/grpc/impl/codegen/sync_windows.h \
include/grpc/load_reporting.h \
include/grpc/slice.h \
include/grpc/slice_buffer.h \
include/grpc/status.h \
@ -946,10 +945,6 @@ src/core/ext/filters/http/message_compress/message_compress_filter.cc \
src/core/ext/filters/http/message_compress/message_compress_filter.h \
src/core/ext/filters/http/server/http_server_filter.cc \
src/core/ext/filters/http/server/http_server_filter.h \
src/core/ext/filters/load_reporting/server_load_reporting_filter.cc \
src/core/ext/filters/load_reporting/server_load_reporting_filter.h \
src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc \
src/core/ext/filters/load_reporting/server_load_reporting_plugin.h \
src/core/ext/filters/max_age/max_age_filter.cc \
src/core/ext/filters/max_age/max_age_filter.h \
src/core/ext/filters/message_size/message_size_filter.cc \

@ -6011,24 +6011,6 @@
"third_party": false,
"type": "target"
},
{
"deps": [
"end2end_tests",
"gpr",
"gpr_test_util",
"grpc",
"grpc_test_util"
],
"headers": [],
"is_filegroup": false,
"language": "c",
"name": "h2_load_reporting_test",
"src": [
"test/core/end2end/fixtures/h2_load_reporting.cc"
],
"third_party": false,
"type": "target"
},
{
"deps": [
"end2end_tests",
@ -6335,24 +6317,6 @@
"third_party": false,
"type": "target"
},
{
"deps": [
"end2end_nosec_tests",
"gpr",
"gpr_test_util",
"grpc_test_util_unsecure",
"grpc_unsecure"
],
"headers": [],
"is_filegroup": false,
"language": "c",
"name": "h2_load_reporting_nosec_test",
"src": [
"test/core/end2end/fixtures/h2_load_reporting.cc"
],
"third_party": false,
"type": "target"
},
{
"deps": [
"end2end_nosec_tests",
@ -6933,7 +6897,6 @@
"grpc_resolver_sockaddr",
"grpc_secure",
"grpc_server_backward_compatibility",
"grpc_server_load_reporting",
"grpc_transport_chttp2_client_insecure",
"grpc_transport_chttp2_client_secure",
"grpc_transport_chttp2_server_insecure",
@ -6955,7 +6918,6 @@
"deps": [
"gpr",
"grpc_base",
"grpc_server_load_reporting",
"grpc_transport_chttp2_client_secure",
"grpc_transport_cronet_client_secure"
],
@ -7040,7 +7002,6 @@
"grpc_resolver_fake",
"grpc_resolver_sockaddr",
"grpc_server_backward_compatibility",
"grpc_server_load_reporting",
"grpc_transport_chttp2_client_insecure",
"grpc_transport_chttp2_server_insecure",
"grpc_transport_inproc",
@ -8783,7 +8744,6 @@
"test/core/end2end/tests/invoke_large_request.cc",
"test/core/end2end/tests/keepalive_timeout.cc",
"test/core/end2end/tests/large_metadata.cc",
"test/core/end2end/tests/load_reporting_hook.cc",
"test/core/end2end/tests/max_concurrent_streams.cc",
"test/core/end2end/tests/max_connection_age.cc",
"test/core/end2end/tests/max_connection_idle.cc",
@ -8884,7 +8844,6 @@
"test/core/end2end/tests/invoke_large_request.cc",
"test/core/end2end/tests/keepalive_timeout.cc",
"test/core/end2end/tests/large_metadata.cc",
"test/core/end2end/tests/load_reporting_hook.cc",
"test/core/end2end/tests/max_concurrent_streams.cc",
"test/core/end2end/tests/max_connection_age.cc",
"test/core/end2end/tests/max_connection_idle.cc",
@ -9501,7 +9460,6 @@
"include/grpc/grpc.h",
"include/grpc/grpc_posix.h",
"include/grpc/grpc_security_constants.h",
"include/grpc/load_reporting.h",
"include/grpc/slice.h",
"include/grpc/slice_buffer.h",
"include/grpc/status.h",
@ -9651,7 +9609,6 @@
"include/grpc/grpc.h",
"include/grpc/grpc_posix.h",
"include/grpc/grpc_security_constants.h",
"include/grpc/load_reporting.h",
"include/grpc/slice.h",
"include/grpc/slice_buffer.h",
"include/grpc/status.h",
@ -10332,27 +10289,6 @@
"third_party": false,
"type": "filegroup"
},
{
"deps": [
"gpr",
"grpc_base"
],
"headers": [
"src/core/ext/filters/load_reporting/server_load_reporting_filter.h",
"src/core/ext/filters/load_reporting/server_load_reporting_plugin.h"
],
"is_filegroup": true,
"language": "c",
"name": "grpc_server_load_reporting",
"src": [
"src/core/ext/filters/load_reporting/server_load_reporting_filter.cc",
"src/core/ext/filters/load_reporting/server_load_reporting_filter.h",
"src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc",
"src/core/ext/filters/load_reporting/server_load_reporting_plugin.h"
],
"third_party": false,
"type": "filegroup"
},
{
"deps": [
"cmdline",

File diff suppressed because it is too large Load Diff
Loading…
Cancel
Save