diff options
-rw-r--r-- | BUILD | 14 | ||||
-rw-r--r-- | CMakeLists.txt | 84 | ||||
-rw-r--r-- | Makefile | 102 | ||||
-rw-r--r-- | build.yaml | 23 | ||||
-rw-r--r-- | grpc.gyp | 10 | ||||
-rw-r--r-- | src/cpp/server/load_reporter/load_data_store.cc | 273 | ||||
-rw-r--r-- | src/cpp/server/load_reporter/load_data_store.h | 339 | ||||
-rw-r--r-- | test/cpp/server/load_reporter/BUILD | 31 | ||||
-rw-r--r-- | test/cpp/server/load_reporter/load_data_store_test.cc | 481 | ||||
-rw-r--r-- | tools/run_tests/generated/sources_and_headers.json | 37 | ||||
-rw-r--r-- | tools/run_tests/generated/tests.json | 24 |
11 files changed, 1416 insertions, 2 deletions
@@ -1286,6 +1286,20 @@ grpc_cc_library( ) grpc_cc_library( + name = "lb_load_data_store", + srcs = [ + "src/cpp/server/load_reporter/load_data_store.cc", + ], + hdrs = [ + "src/cpp/server/load_reporter/load_data_store.h", + ], + language = "c++", + deps = [ + "grpc++", + ], +) + +grpc_cc_library( name = "grpc_resolver_dns_native", srcs = [ "src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc", diff --git a/CMakeLists.txt b/CMakeLists.txt index 3e0a93aff6..d638dbfc6e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -584,6 +584,7 @@ endif() if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX) add_dependencies(buildtests_cxx json_run_localhost) endif() +add_dependencies(buildtests_cxx lb_load_data_store_test) add_dependencies(buildtests_cxx memory_test) add_dependencies(buildtests_cxx metrics_client) add_dependencies(buildtests_cxx mock_test) @@ -4975,6 +4976,49 @@ target_link_libraries(interop_server_main endif (gRPC_BUILD_TESTS) if (gRPC_BUILD_TESTS) +add_library(lb_load_data_store + src/cpp/server/load_reporter/load_data_store.cc +) + +if(WIN32 AND MSVC) + set_target_properties(lb_load_data_store PROPERTIES COMPILE_PDB_NAME "lb_load_data_store" + COMPILE_PDB_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}" + ) + if (gRPC_INSTALL) + install(FILES ${CMAKE_CURRENT_BINARY_DIR}/lb_load_data_store.pdb + DESTINATION ${gRPC_INSTALL_LIBDIR} OPTIONAL + ) + endif() +endif() + + +target_include_directories(lb_load_data_store + PUBLIC $<INSTALL_INTERFACE:${gRPC_INSTALL_INCLUDEDIR}> $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include> + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR} + 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 third_party/googletest/googletest/include + PRIVATE third_party/googletest/googletest + PRIVATE third_party/googletest/googlemock/include + PRIVATE third_party/googletest/googlemock + PRIVATE ${_gRPC_PROTO_GENS_DIR} +) + +target_link_libraries(lb_load_data_store + ${_gRPC_PROTOBUF_LIBRARIES} + ${_gRPC_ALLTARGETS_LIBRARIES} + grpc++ +) + + +endif (gRPC_BUILD_TESTS) +if (gRPC_BUILD_TESTS) + if (gRPC_BUILD_CODEGEN) add_library(qps ${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/messages.pb.cc @@ -12271,6 +12315,46 @@ endif() endif (gRPC_BUILD_TESTS) if (gRPC_BUILD_TESTS) +add_executable(lb_load_data_store_test + test/cpp/server/load_reporter/load_data_store_test.cc + third_party/googletest/googletest/src/gtest-all.cc + third_party/googletest/googlemock/src/gmock-all.cc +) + + +target_include_directories(lb_load_data_store_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 third_party/googletest/googletest/include + PRIVATE third_party/googletest/googletest + PRIVATE third_party/googletest/googlemock/include + PRIVATE third_party/googletest/googlemock + PRIVATE ${_gRPC_PROTO_GENS_DIR} +) + +target_link_libraries(lb_load_data_store_test + ${_gRPC_PROTOBUF_LIBRARIES} + ${_gRPC_ALLTARGETS_LIBRARIES} + lb_load_data_store + grpc++_test_util + grpc_test_util + grpc++ + grpc + gpr_test_util + gpr + ${_gRPC_GFLAGS_LIBRARIES} +) + +endif (gRPC_BUILD_TESTS) +if (gRPC_BUILD_TESTS) + add_executable(memory_test test/core/gprpp/memory_test.cc third_party/googletest/googletest/src/gtest-all.cc @@ -1178,6 +1178,7 @@ interop_client: $(BINDIR)/$(CONFIG)/interop_client interop_server: $(BINDIR)/$(CONFIG)/interop_server interop_test: $(BINDIR)/$(CONFIG)/interop_test json_run_localhost: $(BINDIR)/$(CONFIG)/json_run_localhost +lb_load_data_store_test: $(BINDIR)/$(CONFIG)/lb_load_data_store_test memory_test: $(BINDIR)/$(CONFIG)/memory_test metrics_client: $(BINDIR)/$(CONFIG)/metrics_client mock_test: $(BINDIR)/$(CONFIG)/mock_test @@ -1390,9 +1391,9 @@ pc_cxx: $(LIBDIR)/$(CONFIG)/pkgconfig/grpc++.pc pc_cxx_unsecure: $(LIBDIR)/$(CONFIG)/pkgconfig/grpc++_unsecure.pc ifeq ($(EMBED_OPENSSL),true) -privatelibs_cxx: $(LIBDIR)/$(CONFIG)/libgrpc++_core_stats.a $(LIBDIR)/$(CONFIG)/libgrpc++_proto_reflection_desc_db.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_config.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util_unsecure.a $(LIBDIR)/$(CONFIG)/libgrpc_cli_libs.a $(LIBDIR)/$(CONFIG)/libhttp2_client_main.a $(LIBDIR)/$(CONFIG)/libinterop_client_helper.a $(LIBDIR)/$(CONFIG)/libinterop_client_main.a $(LIBDIR)/$(CONFIG)/libinterop_server_helper.a $(LIBDIR)/$(CONFIG)/libinterop_server_lib.a $(LIBDIR)/$(CONFIG)/libinterop_server_main.a $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libboringssl_test_util.a $(LIBDIR)/$(CONFIG)/libboringssl_crypto_test_data_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_asn1_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_base64_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_bio_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_buf_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_bytestring_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_chacha_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_aead_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_cipher_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_cmac_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_compiler_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_constant_time_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ed25519_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_spake25519_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_x25519_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_dh_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_digest_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_dsa_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ecdh_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_err_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_evp_extra_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_evp_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pbkdf_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_scrypt_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_aes_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_bn_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ec_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_p256-x86_64_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ecdsa_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_gcm_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ctrdrbg_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_hkdf_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_hmac_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_lhash_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_obj_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pkcs7_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pkcs12_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pkcs8_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_poly1305_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pool_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_refcount_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_rsa_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_file_test_gtest_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_gtest_main_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_thread_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_x509_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_tab_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_v3name_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_span_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ssl_test_lib.a $(LIBDIR)/$(CONFIG)/libbenchmark.a +privatelibs_cxx: $(LIBDIR)/$(CONFIG)/libgrpc++_core_stats.a $(LIBDIR)/$(CONFIG)/libgrpc++_proto_reflection_desc_db.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_config.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util_unsecure.a $(LIBDIR)/$(CONFIG)/libgrpc_cli_libs.a $(LIBDIR)/$(CONFIG)/libhttp2_client_main.a $(LIBDIR)/$(CONFIG)/libinterop_client_helper.a $(LIBDIR)/$(CONFIG)/libinterop_client_main.a $(LIBDIR)/$(CONFIG)/libinterop_server_helper.a $(LIBDIR)/$(CONFIG)/libinterop_server_lib.a $(LIBDIR)/$(CONFIG)/libinterop_server_main.a $(LIBDIR)/$(CONFIG)/liblb_load_data_store.a $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libboringssl_test_util.a $(LIBDIR)/$(CONFIG)/libboringssl_crypto_test_data_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_asn1_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_base64_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_bio_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_buf_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_bytestring_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_chacha_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_aead_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_cipher_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_cmac_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_compiler_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_constant_time_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ed25519_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_spake25519_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_x25519_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_dh_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_digest_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_dsa_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ecdh_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_err_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_evp_extra_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_evp_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pbkdf_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_scrypt_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_aes_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_bn_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ec_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_p256-x86_64_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ecdsa_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_gcm_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ctrdrbg_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_hkdf_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_hmac_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_lhash_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_obj_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pkcs7_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pkcs12_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pkcs8_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_poly1305_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_pool_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_refcount_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_rsa_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_file_test_gtest_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_gtest_main_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_thread_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_x509_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_tab_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_v3name_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_span_test_lib.a $(LIBDIR)/$(CONFIG)/libboringssl_ssl_test_lib.a $(LIBDIR)/$(CONFIG)/libbenchmark.a else -privatelibs_cxx: $(LIBDIR)/$(CONFIG)/libgrpc++_core_stats.a $(LIBDIR)/$(CONFIG)/libgrpc++_proto_reflection_desc_db.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_config.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util_unsecure.a $(LIBDIR)/$(CONFIG)/libgrpc_cli_libs.a $(LIBDIR)/$(CONFIG)/libhttp2_client_main.a $(LIBDIR)/$(CONFIG)/libinterop_client_helper.a $(LIBDIR)/$(CONFIG)/libinterop_client_main.a $(LIBDIR)/$(CONFIG)/libinterop_server_helper.a $(LIBDIR)/$(CONFIG)/libinterop_server_lib.a $(LIBDIR)/$(CONFIG)/libinterop_server_main.a $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libbenchmark.a +privatelibs_cxx: $(LIBDIR)/$(CONFIG)/libgrpc++_core_stats.a $(LIBDIR)/$(CONFIG)/libgrpc++_proto_reflection_desc_db.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_config.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util_unsecure.a $(LIBDIR)/$(CONFIG)/libgrpc_cli_libs.a $(LIBDIR)/$(CONFIG)/libhttp2_client_main.a $(LIBDIR)/$(CONFIG)/libinterop_client_helper.a $(LIBDIR)/$(CONFIG)/libinterop_client_main.a $(LIBDIR)/$(CONFIG)/libinterop_server_helper.a $(LIBDIR)/$(CONFIG)/libinterop_server_lib.a $(LIBDIR)/$(CONFIG)/libinterop_server_main.a $(LIBDIR)/$(CONFIG)/liblb_load_data_store.a $(LIBDIR)/$(CONFIG)/libqps.a $(LIBDIR)/$(CONFIG)/libbenchmark.a endif @@ -1660,6 +1661,7 @@ buildtests_cxx: privatelibs_cxx \ $(BINDIR)/$(CONFIG)/interop_server \ $(BINDIR)/$(CONFIG)/interop_test \ $(BINDIR)/$(CONFIG)/json_run_localhost \ + $(BINDIR)/$(CONFIG)/lb_load_data_store_test \ $(BINDIR)/$(CONFIG)/memory_test \ $(BINDIR)/$(CONFIG)/metrics_client \ $(BINDIR)/$(CONFIG)/mock_test \ @@ -1831,6 +1833,7 @@ buildtests_cxx: privatelibs_cxx \ $(BINDIR)/$(CONFIG)/interop_server \ $(BINDIR)/$(CONFIG)/interop_test \ $(BINDIR)/$(CONFIG)/json_run_localhost \ + $(BINDIR)/$(CONFIG)/lb_load_data_store_test \ $(BINDIR)/$(CONFIG)/memory_test \ $(BINDIR)/$(CONFIG)/metrics_client \ $(BINDIR)/$(CONFIG)/mock_test \ @@ -2283,6 +2286,8 @@ test_cxx: buildtests_cxx $(Q) $(BINDIR)/$(CONFIG)/inproc_sync_unary_ping_pong_test || ( echo test inproc_sync_unary_ping_pong_test failed ; exit 1 ) $(E) "[RUN] Testing interop_test" $(Q) $(BINDIR)/$(CONFIG)/interop_test || ( echo test interop_test failed ; exit 1 ) + $(E) "[RUN] Testing lb_load_data_store_test" + $(Q) $(BINDIR)/$(CONFIG)/lb_load_data_store_test || ( echo test lb_load_data_store_test failed ; exit 1 ) $(E) "[RUN] Testing memory_test" $(Q) $(BINDIR)/$(CONFIG)/memory_test || ( echo test memory_test failed ; exit 1 ) $(E) "[RUN] Testing mock_test" @@ -7207,6 +7212,55 @@ endif endif +LIBLB_LOAD_DATA_STORE_SRC = \ + src/cpp/server/load_reporter/load_data_store.cc \ + +PUBLIC_HEADERS_CXX += \ + +LIBLB_LOAD_DATA_STORE_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(LIBLB_LOAD_DATA_STORE_SRC)))) + + +ifeq ($(NO_SECURE),true) + +# You can't build secure libraries if you don't have OpenSSL. + +$(LIBDIR)/$(CONFIG)/liblb_load_data_store.a: openssl_dep_error + + +else + +ifeq ($(NO_PROTOBUF),true) + +# You can't build a C++ library if you don't have protobuf - a bit overreached, but still okay. + +$(LIBDIR)/$(CONFIG)/liblb_load_data_store.a: protobuf_dep_error + + +else + +$(LIBDIR)/$(CONFIG)/liblb_load_data_store.a: $(ZLIB_DEP) $(OPENSSL_DEP) $(CARES_DEP) $(ADDRESS_SORTING_DEP) $(PROTOBUF_DEP) $(LIBLB_LOAD_DATA_STORE_OBJS) + $(E) "[AR] Creating $@" + $(Q) mkdir -p `dirname $@` + $(Q) rm -f $(LIBDIR)/$(CONFIG)/liblb_load_data_store.a + $(Q) $(AR) $(AROPTS) $(LIBDIR)/$(CONFIG)/liblb_load_data_store.a $(LIBLB_LOAD_DATA_STORE_OBJS) +ifeq ($(SYSTEM),Darwin) + $(Q) ranlib -no_warning_for_no_symbols $(LIBDIR)/$(CONFIG)/liblb_load_data_store.a +endif + + + + +endif + +endif + +ifneq ($(NO_SECURE),true) +ifneq ($(NO_DEPS),true) +-include $(LIBLB_LOAD_DATA_STORE_OBJS:.o=.dep) +endif +endif + + LIBQPS_SRC = \ $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc \ $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc \ @@ -18027,6 +18081,49 @@ endif endif +LB_LOAD_DATA_STORE_TEST_SRC = \ + test/cpp/server/load_reporter/load_data_store_test.cc \ + +LB_LOAD_DATA_STORE_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(LB_LOAD_DATA_STORE_TEST_SRC)))) +ifeq ($(NO_SECURE),true) + +# You can't build secure targets if you don't have OpenSSL. + +$(BINDIR)/$(CONFIG)/lb_load_data_store_test: openssl_dep_error + +else + + + + +ifeq ($(NO_PROTOBUF),true) + +# You can't build the protoc plugins or protobuf-enabled targets if you don't have protobuf 3.5.0+. + +$(BINDIR)/$(CONFIG)/lb_load_data_store_test: protobuf_dep_error + +else + +$(BINDIR)/$(CONFIG)/lb_load_data_store_test: $(PROTOBUF_DEP) $(LB_LOAD_DATA_STORE_TEST_OBJS) $(LIBDIR)/$(CONFIG)/liblb_load_data_store.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a + $(E) "[LD] Linking $@" + $(Q) mkdir -p `dirname $@` + $(Q) $(LDXX) $(LDFLAGS) $(LB_LOAD_DATA_STORE_TEST_OBJS) $(LIBDIR)/$(CONFIG)/liblb_load_data_store.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBSXX) $(LDLIBS_PROTOBUF) $(LDLIBS) $(LDLIBS_SECURE) $(GTEST_LIB) -o $(BINDIR)/$(CONFIG)/lb_load_data_store_test + +endif + +endif + +$(OBJDIR)/$(CONFIG)/test/cpp/server/load_reporter/load_data_store_test.o: $(LIBDIR)/$(CONFIG)/liblb_load_data_store.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a + +deps_lb_load_data_store_test: $(LB_LOAD_DATA_STORE_TEST_OBJS:.o=.dep) + +ifneq ($(NO_SECURE),true) +ifneq ($(NO_DEPS),true) +-include $(LB_LOAD_DATA_STORE_TEST_OBJS:.o=.dep) +endif +endif + + MEMORY_TEST_SRC = \ test/core/gprpp/memory_test.cc \ @@ -23895,6 +23992,7 @@ src/cpp/common/secure_channel_arguments.cc: $(OPENSSL_DEP) src/cpp/common/secure_create_auth_context.cc: $(OPENSSL_DEP) src/cpp/ext/proto_server_reflection.cc: $(OPENSSL_DEP) src/cpp/ext/proto_server_reflection_plugin.cc: $(OPENSSL_DEP) +src/cpp/server/load_reporter/load_data_store.cc: $(OPENSSL_DEP) src/cpp/server/secure_server_credentials.cc: $(OPENSSL_DEP) src/cpp/util/core_stats.cc: $(OPENSSL_DEP) src/cpp/util/error_details.cc: $(OPENSSL_DEP) diff --git a/build.yaml b/build.yaml index 374d5a0324..5f356174ca 100644 --- a/build.yaml +++ b/build.yaml @@ -1890,6 +1890,15 @@ libs: - test/cpp/interop/interop_server_bootstrap.cc deps: - interop_server_lib +- name: lb_load_data_store + build: private + language: c++ + headers: + - src/cpp/server/load_reporter/load_data_store.h + src: + - src/cpp/server/load_reporter/load_data_store.cc + deps: + - grpc++ - name: qps build: private language: c++ @@ -4766,6 +4775,20 @@ targets: - mac - linux - posix +- name: lb_load_data_store_test + gtest: true + build: test + language: c++ + src: + - test/cpp/server/load_reporter/load_data_store_test.cc + deps: + - lb_load_data_store + - grpc++_test_util + - grpc_test_util + - grpc++ + - grpc + - gpr_test_util + - gpr - name: memory_test gtest: true build: test @@ -1638,6 +1638,16 @@ ], }, { + 'target_name': 'lb_load_data_store', + 'type': 'static_library', + 'dependencies': [ + 'grpc++', + ], + 'sources': [ + 'src/cpp/server/load_reporter/load_data_store.cc', + ], + }, + { 'target_name': 'qps', 'type': 'static_library', 'dependencies': [ diff --git a/src/cpp/server/load_reporter/load_data_store.cc b/src/cpp/server/load_reporter/load_data_store.cc new file mode 100644 index 0000000000..70f12c1102 --- /dev/null +++ b/src/cpp/server/load_reporter/load_data_store.cc @@ -0,0 +1,273 @@ +/* + * + * 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. + * + */ + +#include <cstdlib> +#include <set> +#include <unordered_map> +#include <vector> + +#include "src/cpp/server/load_reporter/load_data_store.h" + +namespace grpc { +namespace load_reporter { + +// Some helper functions. +namespace { + +// Given a map from type K to a set of value type V, finds the set associated +// with the given key and erases the value from the set. If the set becomes +// empty, also erases the key-set pair. Returns true if the value is erased +// successfully. +template <typename K, typename V> +bool UnorderedMapOfSetEraseKeyValue(std::unordered_map<K, std::set<V>>& map, + const K& key, const V& value) { + auto it = map.find(key); + if (it != map.end()) { + size_t erased = it->second.erase(value); + if (it->second.size() == 0) { + map.erase(it); + } + return erased; + } + return false; +}; + +// Given a map from type K to a set of value type V, removes the given key and +// the associated set, and returns the set. Returns an empty set if the key is +// not found. +template <typename K, typename V> +std::set<V> UnorderedMapOfSetExtract(std::unordered_map<K, std::set<V>>& map, + const K& key) { + auto it = map.find(key); + if (it != map.end()) { + auto set = std::move(it->second); + map.erase(it); + return set; + } + return {}; +}; + +// From a non-empty container, returns a pointer to a random element. +template <typename C> +const typename C::value_type* RandomElement(const C& container) { + GPR_ASSERT(!container.empty()); + auto it = container.begin(); + std::advance(it, std::rand() % container.size()); + return &(*it); +} + +} // namespace + +void PerBalancerStore::MergeRow(const LoadRecordKey& key, + const LoadRecordValue& value) { + // During suspension, the load data received will be dropped. + if (!suspended_) { + load_record_map_[key].MergeFrom(value); + gpr_log(GPR_DEBUG, + "[PerBalancerStore %p] Load data merged (Key: %s, Value: %s).", + this, key.ToString().c_str(), value.ToString().c_str()); + } else { + gpr_log(GPR_DEBUG, + "[PerBalancerStore %p] Load data dropped (Key: %s, Value: %s).", + this, key.ToString().c_str(), value.ToString().c_str()); + } + // We always keep track of num_calls_in_progress_, so that when this + // store is resumed, we still have a correct value of + // num_calls_in_progress_. + GPR_ASSERT(static_cast<int64_t>(num_calls_in_progress_) + + value.GetNumCallsInProgressDelta() >= + 0); + num_calls_in_progress_ += value.GetNumCallsInProgressDelta(); +} + +void PerBalancerStore::Suspend() { + suspended_ = true; + load_record_map_.clear(); + gpr_log(GPR_DEBUG, "[PerBalancerStore %p] Suspended.", this); +} + +void PerBalancerStore::Resume() { + suspended_ = false; + gpr_log(GPR_DEBUG, "[PerBalancerStore %p] Resumed.", this); +} + +uint64_t PerBalancerStore::GetNumCallsInProgressForReport() { + GPR_ASSERT(!suspended_); + last_reported_num_calls_in_progress_ = num_calls_in_progress_; + return num_calls_in_progress_; +} + +void PerHostStore::ReportStreamCreated(const grpc::string& lb_id, + const grpc::string& load_key) { + GPR_ASSERT(lb_id != kInvalidLbId); + SetUpForNewLbId(lb_id, load_key); + // Prior to this one, there was no load balancer receiving report, so we may + // have unassigned orphaned stores to assign to this new balancer. + // TODO(juanlishen): If the load key of this new stream is the same with + // some previously adopted orphan store, we may want to take the orphan to + // this stream. Need to discuss with LB team. + if (assigned_stores_.size() == 1) { + for (const auto& p : per_balancer_stores_) { + const grpc::string& other_lb_id = p.first; + const std::unique_ptr<PerBalancerStore>& orphaned_store = p.second; + if (other_lb_id != lb_id) { + orphaned_store->Resume(); + AssignOrphanedStore(orphaned_store.get(), lb_id); + } + } + } + // The first connected balancer will adopt the kInvalidLbId. + if (per_balancer_stores_.size() == 1) { + SetUpForNewLbId(kInvalidLbId, ""); + ReportStreamClosed(kInvalidLbId); + } +} + +void PerHostStore::ReportStreamClosed(const grpc::string& lb_id) { + auto it_store_for_gone_lb = per_balancer_stores_.find(lb_id); + GPR_ASSERT(it_store_for_gone_lb != per_balancer_stores_.end()); + // Remove this closed stream from our records. + GPR_ASSERT(UnorderedMapOfSetEraseKeyValue( + load_key_to_receiving_lb_ids_, it_store_for_gone_lb->second->load_key(), + lb_id)); + std::set<PerBalancerStore*> orphaned_stores = + UnorderedMapOfSetExtract(assigned_stores_, lb_id); + // The stores that were assigned to this balancer are orphaned now. They + // should be re-assigned to other balancers which are still receiving reports. + for (PerBalancerStore* orphaned_store : orphaned_stores) { + const grpc::string* new_receiver = nullptr; + auto it = load_key_to_receiving_lb_ids_.find(orphaned_store->load_key()); + if (it != load_key_to_receiving_lb_ids_.end()) { + // First, try to pick from the active balancers with the same load key. + new_receiver = RandomElement(it->second); + } else if (!assigned_stores_.empty()) { + // If failed, pick from all the remaining active balancers. + new_receiver = &(RandomElement(assigned_stores_)->first); + } + if (new_receiver != nullptr) { + AssignOrphanedStore(orphaned_store, *new_receiver); + } else { + // Load data for an LB ID that can't be assigned to any stream should + // be dropped. + orphaned_store->Suspend(); + } + } +} + +PerBalancerStore* PerHostStore::FindPerBalancerStore( + const grpc::string& lb_id) const { + return per_balancer_stores_.find(lb_id) != per_balancer_stores_.end() + ? per_balancer_stores_.find(lb_id)->second.get() + : nullptr; +} + +const std::set<PerBalancerStore*>* PerHostStore::GetAssignedStores( + const grpc::string& lb_id) const { + auto it = assigned_stores_.find(lb_id); + if (it == assigned_stores_.end()) return nullptr; + return &(it->second); +} + +void PerHostStore::AssignOrphanedStore(PerBalancerStore* orphaned_store, + const grpc::string& new_receiver) { + auto it = assigned_stores_.find(new_receiver); + GPR_ASSERT(it != assigned_stores_.end()); + it->second.insert(orphaned_store); + gpr_log(GPR_INFO, + "[PerHostStore %p] Re-assigned orphaned store (%p) with original LB" + " ID of %s to new receiver %s", + this, orphaned_store, orphaned_store->lb_id().c_str(), + new_receiver.c_str()); +} + +void PerHostStore::SetUpForNewLbId(const grpc::string& lb_id, + const grpc::string& load_key) { + // The top-level caller (i.e., LoadReportService) should guarantee the + // lb_id is unique for each reporting stream. + GPR_ASSERT(per_balancer_stores_.find(lb_id) == per_balancer_stores_.end()); + GPR_ASSERT(assigned_stores_.find(lb_id) == assigned_stores_.end()); + load_key_to_receiving_lb_ids_[load_key].insert(lb_id); + std::unique_ptr<PerBalancerStore> per_balancer_store( + new PerBalancerStore(lb_id, load_key)); + assigned_stores_[lb_id] = {per_balancer_store.get()}; + per_balancer_stores_[lb_id] = std::move(per_balancer_store); +} + +PerBalancerStore* LoadDataStore::FindPerBalancerStore( + const string& hostname, const string& lb_id) const { + auto it = per_host_stores_.find(hostname); + if (it != per_host_stores_.end()) { + const PerHostStore& per_host_store = it->second; + return per_host_store.FindPerBalancerStore(lb_id); + } else { + return nullptr; + } +} + +void LoadDataStore::MergeRow(const grpc::string& hostname, + const LoadRecordKey& key, + const LoadRecordValue& value) { + PerBalancerStore* per_balancer_store = + FindPerBalancerStore(hostname, key.lb_id()); + if (per_balancer_store != nullptr) { + per_balancer_store->MergeRow(key, value); + return; + } + // Unknown LB ID. Track it until its number of in-progress calls drops to + // zero. + int64_t in_progress_delta = value.GetNumCallsInProgressDelta(); + if (in_progress_delta != 0) { + auto it_tracker = unknown_balancer_id_trackers_.find(key.lb_id()); + if (it_tracker == unknown_balancer_id_trackers_.end()) { + gpr_log( + GPR_DEBUG, + "[LoadDataStore %p] Start tracking unknown balancer (lb_id_: %s).", + this, key.lb_id().c_str()); + unknown_balancer_id_trackers_.insert( + {key.lb_id(), static_cast<uint64_t>(in_progress_delta)}); + } else if ((it_tracker->second += in_progress_delta) == 0) { + unknown_balancer_id_trackers_.erase(it_tracker); + gpr_log(GPR_DEBUG, + "[LoadDataStore %p] Stop tracking unknown balancer (lb_id_: %s).", + this, key.lb_id().c_str()); + } + } +} + +const std::set<PerBalancerStore*>* LoadDataStore::GetAssignedStores( + const grpc::string& hostname, const grpc::string& lb_id) { + auto it = per_host_stores_.find(hostname); + if (it == per_host_stores_.end()) return nullptr; + return it->second.GetAssignedStores(lb_id); +} + +void LoadDataStore::ReportStreamCreated(const grpc::string& hostname, + const grpc::string& lb_id, + const grpc::string& load_key) { + per_host_stores_[hostname].ReportStreamCreated(lb_id, load_key); +} + +void LoadDataStore::ReportStreamClosed(const grpc::string& hostname, + const grpc::string& lb_id) { + auto it_per_host_store = per_host_stores_.find(hostname); + GPR_ASSERT(it_per_host_store != per_host_stores_.end()); + it_per_host_store->second.ReportStreamClosed(lb_id); +} + +} // namespace load_reporter +} // namespace grpc diff --git a/src/cpp/server/load_reporter/load_data_store.h b/src/cpp/server/load_reporter/load_data_store.h new file mode 100644 index 0000000000..feb8b2fd59 --- /dev/null +++ b/src/cpp/server/load_reporter/load_data_store.h @@ -0,0 +1,339 @@ +/* + * + * 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_SRC_CPP_SERVER_LOAD_REPORTER_LOAD_DATA_STORE_H +#define GRPC_SRC_CPP_SERVER_LOAD_REPORTER_LOAD_DATA_STORE_H + +#include <grpc/support/port_platform.h> + +#include <memory> +#include <set> +#include <unordered_map> + +#include <grpc/support/log.h> +#include <grpcpp/impl/codegen/config.h> + +namespace grpc { +namespace load_reporter { + +constexpr char kInvalidLbId[] = "<INVALID_LBID_238dsb234890rb>"; +constexpr uint8_t kLbIdLen = 8; + +// The load data storage is organized in hierarchy. The LoadDataStore is the +// top-level data store. In LoadDataStore, for each host we keep a +// PerHostStore, in which for each balancer we keep a PerBalancerStore. Each +// PerBalancerStore maintains a map of load records, mapping from LoadRecordKey +// to LoadRecordValue. The LoadRecordValue contains a map of customized call +// metrics, mapping from a call metric name to the CallMetricValue. + +// The value of a customized call metric. +class CallMetricValue { + public: + explicit CallMetricValue(uint64_t num_calls = 0, + double total_metric_value = 0) + : num_calls_(num_calls), total_metric_value_(total_metric_value) {} + + void MergeFrom(CallMetricValue other) { + num_calls_ += other.num_calls_; + total_metric_value_ += other.total_metric_value_; + } + + // Getters. + uint64_t num_calls() const { return num_calls_; } + double total_metric_value() const { return total_metric_value_; } + + private: + // The number of calls that finished with this metric. + uint64_t num_calls_ = 0; + // The sum of metric values across all the calls that finished with this + // metric. + double total_metric_value_ = 0; +}; + +// The key of a load record. +class LoadRecordKey { + public: + explicit LoadRecordKey(grpc::string lb_id, grpc::string lb_tag, + grpc::string user_id, grpc::string client_ip_hex) + : lb_id_(std::move(lb_id)), + lb_tag_(std::move(lb_tag)), + user_id_(std::move(user_id)), + client_ip_hex_(std::move(client_ip_hex)) {} + + grpc::string ToString() const { + return "[lb_id_=" + lb_id_ + ", lb_tag_=" + lb_tag_ + + ", user_id_=" + user_id_ + ", client_ip_hex_=" + client_ip_hex_ + + "]"; + } + + bool operator==(const LoadRecordKey& other) const { + return lb_id_ == other.lb_id_ && lb_tag_ == other.lb_tag_ && + user_id_ == other.user_id_ && client_ip_hex_ == other.client_ip_hex_; + } + + // Getters. + const grpc::string& lb_id() const { return lb_id_; } + const grpc::string& lb_tag() const { return lb_tag_; } + const grpc::string& user_id() const { return user_id_; } + const grpc::string& client_ip_hex() const { return client_ip_hex_; } + + struct Hasher { + void hash_combine(size_t* seed, const grpc::string& k) const { + *seed ^= std::hash<grpc::string>()(k) + 0x9e3779b9 + (*seed << 6) + + (*seed >> 2); + } + + size_t operator()(const LoadRecordKey& k) const { + size_t h = 0; + hash_combine(&h, k.lb_id_); + hash_combine(&h, k.lb_tag_); + hash_combine(&h, k.user_id_); + hash_combine(&h, k.client_ip_hex_); + return h; + } + }; + + private: + grpc::string lb_id_; + grpc::string lb_tag_; + grpc::string user_id_; + grpc::string client_ip_hex_; +}; + +// The value of a load record. +class LoadRecordValue { + public: + explicit LoadRecordValue(uint64_t start_count = 0, uint64_t ok_count = 0, + uint64_t error_count = 0, double bytes_sent = 0, + double bytes_recv = 0, double latency_ms = 0) + : start_count_(start_count), + ok_count_(ok_count), + error_count_(error_count), + bytes_sent_(bytes_sent), + bytes_recv_(bytes_recv), + latency_ms_(latency_ms) {} + + void MergeFrom(const LoadRecordValue& other) { + start_count_ += other.start_count_; + ok_count_ += other.ok_count_; + error_count_ += other.error_count_; + bytes_sent_ += other.bytes_sent_; + bytes_recv_ += other.bytes_recv_; + latency_ms_ += other.latency_ms_; + for (const auto& p : other.call_metrics_) { + const grpc::string& key = p.first; + const CallMetricValue& value = p.second; + call_metrics_[key].MergeFrom(value); + } + } + + int64_t GetNumCallsInProgressDelta() const { + return static_cast<int64_t>(start_count_ - ok_count_ - error_count_); + } + + grpc::string ToString() const { + return "[start_count_=" + grpc::to_string(start_count_) + + ", ok_count_=" + grpc::to_string(ok_count_) + + ", error_count_=" + grpc::to_string(error_count_) + + ", bytes_sent_=" + grpc::to_string(bytes_sent_) + + ", bytes_recv_=" + grpc::to_string(bytes_recv_) + + ", latency_ms_=" + grpc::to_string(latency_ms_) + "]"; + } + + bool InsertCallMetric(const grpc::string& metric_name, + const CallMetricValue& metric_value) { + return call_metrics_.insert({metric_name, metric_value}).second; + } + + // Getters. + uint64_t start_count() const { return start_count_; } + uint64_t ok_count() const { return ok_count_; } + uint64_t error_count() const { return error_count_; } + double bytes_sent() const { return bytes_sent_; } + double bytes_recv() const { return bytes_recv_; } + double latency_ms() const { return latency_ms_; } + const std::unordered_map<grpc::string, CallMetricValue>& call_metrics() + const { + return call_metrics_; + } + + private: + uint64_t start_count_ = 0; + uint64_t ok_count_ = 0; + uint64_t error_count_ = 0; + double bytes_sent_ = 0; + double bytes_recv_ = 0; + double latency_ms_ = 0; + std::unordered_map<grpc::string, CallMetricValue> call_metrics_; +}; + +// Stores the data associated with a particular LB ID. +class PerBalancerStore { + public: + using LoadRecordMap = + std::unordered_map<LoadRecordKey, LoadRecordValue, LoadRecordKey::Hasher>; + + PerBalancerStore(grpc::string lb_id, grpc::string load_key) + : lb_id_(std::move(lb_id)), load_key_(std::move(load_key)) {} + + // Merge a load record with the given key and value if the store is not + // suspended. + void MergeRow(const LoadRecordKey& key, const LoadRecordValue& value); + + // Suspend this store, so that no detailed load data will be recorded. + void Suspend(); + // Resume this store from suspension. + void Resume(); + // Is this store suspended or not? + bool IsSuspended() const { return suspended_; } + + bool IsNumCallsInProgressChangedSinceLastReport() const { + return num_calls_in_progress_ != last_reported_num_calls_in_progress_; + } + + uint64_t GetNumCallsInProgressForReport(); + + grpc::string ToString() { + return "[PerBalancerStore lb_id_=" + lb_id_ + " load_key_=" + load_key_ + + "]"; + } + + void ClearLoadRecordMap() { load_record_map_.clear(); } + + // Getters. + const grpc::string& lb_id() const { return lb_id_; } + const grpc::string& load_key() const { return load_key_; } + const LoadRecordMap& load_record_map() const { return load_record_map_; } + + private: + grpc::string lb_id_; + // TODO(juanlishen): Use bytestring protobuf type? + grpc::string load_key_; + LoadRecordMap load_record_map_; + uint64_t num_calls_in_progress_ = 0; + uint64_t last_reported_num_calls_in_progress_ = 0; + bool suspended_ = false; +}; + +// Stores the data associated with a particular host. +class PerHostStore { + public: + // When a report stream is created, a PerBalancerStore is created for the + // LB ID (guaranteed unique) associated with that stream. If it is the only + // active store, adopt all the orphaned stores. If it is the first created + // store, adopt the store of kInvalidLbId. + void ReportStreamCreated(const grpc::string& lb_id, + const grpc::string& load_key); + + // When a report stream is closed, the PerBalancerStores assigned to the + // associate LB ID need to be re-assigned to other active balancers, + // ideally with the same load key. If there is no active balancer, we have + // to suspend those stores and drop the incoming load data until they are + // resumed. + void ReportStreamClosed(const grpc::string& lb_id); + + // Returns null if not found. Caller doesn't own the returned store. + PerBalancerStore* FindPerBalancerStore(const grpc::string& lb_id) const; + + // Returns null if lb_id is not found. The returned pointer points to the + // underlying data structure, which is not owned by the caller. + const std::set<PerBalancerStore*>* GetAssignedStores( + const grpc::string& lb_id) const; + + private: + // Creates a PerBalancerStore for the given LB ID, assigns the store to + // itself, and records the LB ID to the load key. + void SetUpForNewLbId(const grpc::string& lb_id, const grpc::string& load_key); + + void AssignOrphanedStore(PerBalancerStore* orphaned_store, + const grpc::string& new_receiver); + + std::unordered_map<grpc::string, std::set<grpc::string>> + load_key_to_receiving_lb_ids_; + + // Key: LB ID. The key set includes all the LB IDs that have been + // allocated for reporting streams so far. + // Value: the unique pointer to the PerBalancerStore of the LB ID. + std::unordered_map<grpc::string, std::unique_ptr<PerBalancerStore>> + per_balancer_stores_; + + // Key: LB ID. The key set includes the LB IDs of the balancers that are + // currently receiving report. + // Value: the set of raw pointers to the PerBalancerStores assigned to the LB + // ID. Note that the sets in assigned_stores_ form a division of the value set + // of per_balancer_stores_. + std::unordered_map<grpc::string, std::set<PerBalancerStore*>> + assigned_stores_; +}; + +// Thread-unsafe two-level bookkeeper of all the load data. +// Note: We never remove any store objects from this class, as per the +// current spec. That's because premature removal of the store objects +// may lead to loss of critical information, e.g., mapping from lb_id to +// load_key, and the number of in-progress calls. Such loss will cause +// information inconsistency when the balancer is re-connected. Keeping +// all the stores should be fine for PerHostStore, since we assume there +// should only be a few hostnames. But it's a potential problem for +// PerBalancerStore. +class LoadDataStore { + public: + // Returns null if not found. Caller doesn't own the returned store. + PerBalancerStore* FindPerBalancerStore(const grpc::string& hostname, + const grpc::string& lb_id) const; + + // Returns null if hostname or lb_id is not found. The returned pointer points + // to the underlying data structure, which is not owned by the caller. + const std::set<PerBalancerStore*>* GetAssignedStores(const string& hostname, + const string& lb_id); + + // If a PerBalancerStore can be found by the hostname and LB ID in + // LoadRecordKey, the load data will be merged to that store. Otherwise, + // only track the number of the in-progress calls for this unknown LB ID. + void MergeRow(const grpc::string& hostname, const LoadRecordKey& key, + const LoadRecordValue& value); + + // Is the given lb_id a tracked unknown LB ID (i.e., the LB ID was associated + // with some received load data but unknown to this load data store)? + bool IsTrackedUnknownBalancerId(const grpc::string& lb_id) const { + return unknown_balancer_id_trackers_.find(lb_id) != + unknown_balancer_id_trackers_.end(); + } + + // Wrapper around PerHostStore::ReportStreamCreated. + void ReportStreamCreated(const grpc::string& hostname, + const grpc::string& lb_id, + const grpc::string& load_key); + + // Wrapper around PerHostStore::ReportStreamClosed. + void ReportStreamClosed(const grpc::string& hostname, + const grpc::string& lb_id); + + private: + // Buffered data that was fetched from Census but hasn't been sent to + // balancer. We need to keep this data ourselves because Census will + // delete the data once it's returned. + std::unordered_map<grpc::string, PerHostStore> per_host_stores_; + + // Tracks the number of in-progress calls for each unknown LB ID. + std::unordered_map<grpc::string, uint64_t> unknown_balancer_id_trackers_; +}; + +} // namespace load_reporter +} // namespace grpc + +#endif // GRPC_SRC_CPP_SERVER_LOAD_REPORTER_LOAD_DATA_STORE_H diff --git a/test/cpp/server/load_reporter/BUILD b/test/cpp/server/load_reporter/BUILD new file mode 100644 index 0000000000..5cb3a00f82 --- /dev/null +++ b/test/cpp/server/load_reporter/BUILD @@ -0,0 +1,31 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +licenses(["notice"]) # Apache v2 + +load("//bazel:grpc_build_system.bzl", "grpc_cc_test", "grpc_cc_library", "grpc_cc_binary", "grpc_package") + +grpc_package(name = "test/cpp/server/load_reporter") + +grpc_cc_test( + name = "lb_load_data_store_test", + srcs = ["load_data_store_test.cc"], + external_deps = [ + "gtest", + ], + deps = [ + "//:lb_load_data_store", + "//test/core/util:grpc_test_util", + ], +) diff --git a/test/cpp/server/load_reporter/load_data_store_test.cc b/test/cpp/server/load_reporter/load_data_store_test.cc new file mode 100644 index 0000000000..8280dee6a4 --- /dev/null +++ b/test/cpp/server/load_reporter/load_data_store_test.cc @@ -0,0 +1,481 @@ +/* + * + * 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. + * + */ + +#include <grpc/impl/codegen/port_platform.h> + +#include <set> +#include <vector> + +#include <grpc/grpc.h> +#include <gtest/gtest.h> + +#include "src/cpp/server/load_reporter/load_data_store.h" +#include "test/core/util/port.h" +#include "test/core/util/test_config.h" + +namespace grpc { +namespace testing { +namespace { + +using ::grpc::load_reporter::CallMetricValue; +using ::grpc::load_reporter::LoadDataStore; +using ::grpc::load_reporter::LoadRecordKey; +using ::grpc::load_reporter::LoadRecordValue; +using ::grpc::load_reporter::PerBalancerStore; +using ::grpc::load_reporter::kInvalidLbId; + +class LoadDataStoreTest : public ::testing::Test { + public: + LoadDataStoreTest() + : kKey1(kLbId1, kLbTag1, kUser1, kClientIp1), + kKey2(kLbId2, kLbTag2, kUser2, kClientIp2) {} + + // Check whether per_balancer_stores contains a store which was originally + // created for <hostname, lb_id, and load_key>. + bool PerBalancerStoresContains( + const LoadDataStore& load_data_store, + const std::set<PerBalancerStore*>* per_balancer_stores, + const grpc::string hostname, const grpc::string lb_id, + const grpc::string load_key) { + auto original_per_balancer_store = + load_data_store.FindPerBalancerStore(hostname, lb_id); + EXPECT_NE(original_per_balancer_store, nullptr); + EXPECT_EQ(original_per_balancer_store->lb_id(), lb_id); + EXPECT_EQ(original_per_balancer_store->load_key(), load_key); + for (auto per_balancer_store : *per_balancer_stores) { + if (per_balancer_store == original_per_balancer_store) { + return true; + } + } + return false; + } + + grpc::string FormatLbId(size_t index) { + return "kLbId" + std::to_string(index); + } + + const grpc::string kHostname1 = "kHostname1"; + const grpc::string kHostname2 = "kHostname2"; + const grpc::string kLbId1 = "kLbId1"; + const grpc::string kLbId2 = "kLbId2"; + const grpc::string kLbId3 = "kLbId3"; + const grpc::string kLbId4 = "kLbId4"; + const grpc::string kLoadKey1 = "kLoadKey1"; + const grpc::string kLoadKey2 = "kLoadKey2"; + const grpc::string kLbTag1 = "kLbTag1"; + const grpc::string kLbTag2 = "kLbTag2"; + const grpc::string kUser1 = "kUser1"; + const grpc::string kUser2 = "kUser2"; + const grpc::string kClientIp1 = "00"; + const grpc::string kClientIp2 = "02"; + const grpc::string kMetric1 = "kMetric1"; + const grpc::string kMetric2 = "kMetric2"; + const LoadRecordKey kKey1; + const LoadRecordKey kKey2; +}; + +using PerBalancerStoreTest = LoadDataStoreTest; + +TEST_F(LoadDataStoreTest, AssignToSelf) { + LoadDataStore load_data_store; + load_data_store.ReportStreamCreated(kHostname1, kLbId1, kLoadKey1); + auto assigned_stores = load_data_store.GetAssignedStores(kHostname1, kLbId1); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_stores, + kHostname1, kLbId1, kLoadKey1)); +} + +TEST_F(LoadDataStoreTest, ReassignOrphanStores) { + LoadDataStore load_data_store; + load_data_store.ReportStreamCreated(kHostname1, kLbId1, kLoadKey1); + load_data_store.ReportStreamCreated(kHostname1, kLbId2, kLoadKey1); + load_data_store.ReportStreamCreated(kHostname1, kLbId3, kLoadKey2); + load_data_store.ReportStreamCreated(kHostname2, kLbId4, kLoadKey1); + // 1. Close the second stream. + load_data_store.ReportStreamClosed(kHostname1, kLbId2); + auto assigned_to_lb_id_1 = + load_data_store.GetAssignedStores(kHostname1, kLbId1); + // The orphaned store is re-assigned to kLbId1 with the same load key. + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_1, + kHostname1, kLbId1, kLoadKey1)); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_1, + kHostname1, kLbId2, kLoadKey1)); + // 2. Close the first stream. + load_data_store.ReportStreamClosed(kHostname1, kLbId1); + auto assigned_to_lb_id_3 = + load_data_store.GetAssignedStores(kHostname1, kLbId3); + // The orphaned stores are re-assigned to kLbId3 with the same host, + // because there isn't any LB with the same load key. + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_3, + kHostname1, kLbId1, kLoadKey1)); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_3, + kHostname1, kLbId2, kLoadKey1)); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_3, + kHostname1, kLbId3, kLoadKey2)); + // 3. Close the third stream. + load_data_store.ReportStreamClosed(kHostname1, kLbId3); + auto assigned_to_lb_id_4 = + load_data_store.GetAssignedStores(kHostname2, kLbId4); + // There is no active LB for the first host now. kLbId4 is active but + // it's for the second host, so it wll NOT adopt the orphaned stores. + EXPECT_FALSE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_4, + kHostname1, kLbId1, kLoadKey1)); + EXPECT_FALSE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_4, + kHostname1, kLbId2, kLoadKey1)); + EXPECT_FALSE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_4, + kHostname1, kLbId3, kLoadKey2)); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_4, + kHostname2, kLbId4, kLoadKey1)); +} + +TEST_F(LoadDataStoreTest, OrphanAssignmentIsSticky) { + LoadDataStore load_data_store; + std::set<grpc::string> active_lb_ids; + size_t num_lb_ids = 1000; + for (size_t i = 0; i < num_lb_ids; ++i) { + load_data_store.ReportStreamCreated(kHostname1, FormatLbId(i), kLoadKey1); + active_lb_ids.insert(FormatLbId(i)); + } + grpc::string orphaned_lb_id = FormatLbId(std::rand() % num_lb_ids); + load_data_store.ReportStreamClosed(kHostname1, orphaned_lb_id); + active_lb_ids.erase(orphaned_lb_id); + // Find which LB is assigned the orphaned store. + grpc::string assigned_lb_id = ""; + for (auto lb_id : active_lb_ids) { + if (PerBalancerStoresContains( + load_data_store, + load_data_store.GetAssignedStores(kHostname1, lb_id), kHostname1, + orphaned_lb_id, kLoadKey1)) { + assigned_lb_id = lb_id; + break; + } + } + EXPECT_STRNE(assigned_lb_id.c_str(), ""); + // Close 10 more stream, skipping the assigned_lb_id. The assignment of + // orphaned_lb_id shouldn't change. + for (size_t _ = 0; _ < 10; ++_) { + grpc::string lb_id_to_close = ""; + for (auto lb_id : active_lb_ids) { + if (lb_id != assigned_lb_id) { + lb_id_to_close = lb_id; + break; + } + } + EXPECT_STRNE(lb_id_to_close.c_str(), ""); + load_data_store.ReportStreamClosed(kHostname1, lb_id_to_close); + active_lb_ids.erase(lb_id_to_close); + EXPECT_TRUE(PerBalancerStoresContains( + load_data_store, + load_data_store.GetAssignedStores(kHostname1, assigned_lb_id), + kHostname1, orphaned_lb_id, kLoadKey1)); + } + // Close the assigned_lb_id, orphaned_lb_id will be re-assigned again. + load_data_store.ReportStreamClosed(kHostname1, assigned_lb_id); + active_lb_ids.erase(assigned_lb_id); + size_t orphaned_lb_id_occurences = 0; + for (auto lb_id : active_lb_ids) { + if (PerBalancerStoresContains( + load_data_store, + load_data_store.GetAssignedStores(kHostname1, lb_id), kHostname1, + orphaned_lb_id, kLoadKey1)) { + orphaned_lb_id_occurences++; + } + } + EXPECT_EQ(orphaned_lb_id_occurences, 1U); +} + +TEST_F(LoadDataStoreTest, HostTemporarilyLoseAllStreams) { + LoadDataStore load_data_store; + load_data_store.ReportStreamCreated(kHostname1, kLbId1, kLoadKey1); + load_data_store.ReportStreamCreated(kHostname2, kLbId2, kLoadKey1); + auto store_lb_id_1 = load_data_store.FindPerBalancerStore(kHostname1, kLbId1); + auto store_invalid_lb_id_1 = + load_data_store.FindPerBalancerStore(kHostname1, kInvalidLbId); + EXPECT_FALSE(store_lb_id_1->IsSuspended()); + EXPECT_FALSE(store_invalid_lb_id_1->IsSuspended()); + // Disconnect all the streams of the first host. + load_data_store.ReportStreamClosed(kHostname1, kLbId1); + // All the streams of that host are suspended. + EXPECT_TRUE(store_lb_id_1->IsSuspended()); + EXPECT_TRUE(store_invalid_lb_id_1->IsSuspended()); + // Detailed load data won't be kept when the PerBalancerStore is suspended. + store_lb_id_1->MergeRow(kKey1, LoadRecordValue()); + store_invalid_lb_id_1->MergeRow(kKey1, LoadRecordValue()); + EXPECT_EQ(store_lb_id_1->load_record_map().size(), 0U); + EXPECT_EQ(store_invalid_lb_id_1->load_record_map().size(), 0U); + // The stores for different hosts won't mix, even if the load key is the same. + auto assigned_to_lb_id_2 = + load_data_store.GetAssignedStores(kHostname2, kLbId2); + EXPECT_EQ(assigned_to_lb_id_2->size(), 2U); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_2, + kHostname2, kLbId2, kLoadKey1)); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_2, + kHostname2, kInvalidLbId, "")); + // A new stream is created for the first host. + load_data_store.ReportStreamCreated(kHostname1, kLbId3, kLoadKey2); + // The stores for the first host are resumed. + EXPECT_FALSE(store_lb_id_1->IsSuspended()); + EXPECT_FALSE(store_invalid_lb_id_1->IsSuspended()); + store_lb_id_1->MergeRow(kKey1, LoadRecordValue()); + store_invalid_lb_id_1->MergeRow(kKey1, LoadRecordValue()); + EXPECT_EQ(store_lb_id_1->load_record_map().size(), 1U); + EXPECT_EQ(store_invalid_lb_id_1->load_record_map().size(), 1U); + // The resumed stores are assigned to the new LB. + auto assigned_to_lb_id_3 = + load_data_store.GetAssignedStores(kHostname1, kLbId3); + EXPECT_EQ(assigned_to_lb_id_3->size(), 3U); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_3, + kHostname1, kLbId1, kLoadKey1)); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_3, + kHostname1, kInvalidLbId, "")); + EXPECT_TRUE(PerBalancerStoresContains(load_data_store, assigned_to_lb_id_3, + kHostname1, kLbId3, kLoadKey2)); +} + +TEST_F(LoadDataStoreTest, OneStorePerLbId) { + LoadDataStore load_data_store; + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname1, kLbId1), nullptr); + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname1, kInvalidLbId), + nullptr); + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname2, kLbId2), nullptr); + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname2, kLbId3), nullptr); + // Create The first stream. + load_data_store.ReportStreamCreated(kHostname1, kLbId1, kLoadKey1); + auto store_lb_id_1 = load_data_store.FindPerBalancerStore(kHostname1, kLbId1); + auto store_invalid_lb_id_1 = + load_data_store.FindPerBalancerStore(kHostname1, kInvalidLbId); + // Two stores will be created: one is for the stream; the other one is for + // kInvalidLbId. + EXPECT_NE(store_lb_id_1, nullptr); + EXPECT_NE(store_invalid_lb_id_1, nullptr); + EXPECT_NE(store_lb_id_1, store_invalid_lb_id_1); + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname2, kLbId2), nullptr); + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname2, kLbId3), nullptr); + // Create the second stream. + load_data_store.ReportStreamCreated(kHostname2, kLbId3, kLoadKey1); + auto store_lb_id_3 = load_data_store.FindPerBalancerStore(kHostname2, kLbId3); + auto store_invalid_lb_id_2 = + load_data_store.FindPerBalancerStore(kHostname2, kInvalidLbId); + EXPECT_NE(store_lb_id_3, nullptr); + EXPECT_NE(store_invalid_lb_id_2, nullptr); + EXPECT_NE(store_lb_id_3, store_invalid_lb_id_2); + // The PerBalancerStores created for different hosts are independent. + EXPECT_NE(store_lb_id_3, store_invalid_lb_id_1); + EXPECT_NE(store_invalid_lb_id_2, store_invalid_lb_id_1); + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname2, kLbId2), nullptr); +} + +TEST_F(LoadDataStoreTest, ExactlyOnceAssignment) { + LoadDataStore load_data_store; + size_t num_create = 100; + size_t num_close = 50; + for (size_t i = 0; i < num_create; ++i) { + load_data_store.ReportStreamCreated(kHostname1, FormatLbId(i), kLoadKey1); + } + for (size_t i = 0; i < num_close; ++i) { + load_data_store.ReportStreamClosed(kHostname1, FormatLbId(i)); + } + std::set<grpc::string> reported_lb_ids; + for (size_t i = num_close; i < num_create; ++i) { + for (auto assigned_store : + *load_data_store.GetAssignedStores(kHostname1, FormatLbId(i))) { + EXPECT_TRUE(reported_lb_ids.insert(assigned_store->lb_id()).second); + } + } + // Add one for kInvalidLbId. + EXPECT_EQ(reported_lb_ids.size(), (num_create + 1)); + EXPECT_NE(reported_lb_ids.find(kInvalidLbId), reported_lb_ids.end()); +} + +TEST_F(LoadDataStoreTest, UnknownBalancerIdTracking) { + LoadDataStore load_data_store; + load_data_store.ReportStreamCreated(kHostname1, kLbId1, kLoadKey1); + // Merge data for a known LB ID. + LoadRecordValue v1(192); + load_data_store.MergeRow(kHostname1, kKey1, v1); + // Merge data for unknown LB ID. + LoadRecordValue v2(23); + EXPECT_FALSE(load_data_store.IsTrackedUnknownBalancerId(kLbId2)); + load_data_store.MergeRow( + kHostname1, LoadRecordKey(kLbId2, kLbTag1, kUser1, kClientIp1), v2); + EXPECT_TRUE(load_data_store.IsTrackedUnknownBalancerId(kLbId2)); + LoadRecordValue v3(952); + load_data_store.MergeRow( + kHostname2, LoadRecordKey(kLbId3, kLbTag1, kUser1, kClientIp1), v3); + EXPECT_TRUE(load_data_store.IsTrackedUnknownBalancerId(kLbId3)); + // The data kept for a known LB ID is correct. + auto store_lb_id_1 = load_data_store.FindPerBalancerStore(kHostname1, kLbId1); + EXPECT_EQ(store_lb_id_1->load_record_map().size(), 1U); + EXPECT_EQ(store_lb_id_1->load_record_map().find(kKey1)->second.start_count(), + v1.start_count()); + EXPECT_EQ(store_lb_id_1->GetNumCallsInProgressForReport(), v1.start_count()); + // No PerBalancerStore created for Unknown LB ID. + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname1, kLbId2), nullptr); + EXPECT_EQ(load_data_store.FindPerBalancerStore(kHostname2, kLbId3), nullptr); + // End all the started RPCs for kLbId1. + LoadRecordValue v4(0, v1.start_count()); + load_data_store.MergeRow(kHostname1, kKey1, v4); + EXPECT_EQ(store_lb_id_1->load_record_map().size(), 1U); + EXPECT_EQ(store_lb_id_1->load_record_map().find(kKey1)->second.start_count(), + v1.start_count()); + EXPECT_EQ(store_lb_id_1->load_record_map().find(kKey1)->second.ok_count(), + v4.ok_count()); + EXPECT_EQ(store_lb_id_1->GetNumCallsInProgressForReport(), 0U); + EXPECT_FALSE(load_data_store.IsTrackedUnknownBalancerId(kLbId1)); + // End all the started RPCs for kLbId2. + LoadRecordValue v5(0, v2.start_count()); + load_data_store.MergeRow( + kHostname1, LoadRecordKey(kLbId2, kLbTag1, kUser1, kClientIp1), v5); + EXPECT_FALSE(load_data_store.IsTrackedUnknownBalancerId(kLbId2)); + // End some of the started RPCs for kLbId3. + LoadRecordValue v6(0, v3.start_count() / 2); + load_data_store.MergeRow( + kHostname2, LoadRecordKey(kLbId3, kLbTag1, kUser1, kClientIp1), v6); + EXPECT_TRUE(load_data_store.IsTrackedUnknownBalancerId(kLbId3)); +} + +TEST_F(PerBalancerStoreTest, Suspend) { + PerBalancerStore per_balancer_store(kLbId1, kLoadKey1); + EXPECT_FALSE(per_balancer_store.IsSuspended()); + // Suspend the store. + per_balancer_store.Suspend(); + EXPECT_TRUE(per_balancer_store.IsSuspended()); + EXPECT_EQ(0U, per_balancer_store.load_record_map().size()); + // Data merged when the store is suspended won't be kept. + LoadRecordValue v1(139, 19); + per_balancer_store.MergeRow(kKey1, v1); + EXPECT_EQ(0U, per_balancer_store.load_record_map().size()); + // Resume the store. + per_balancer_store.Resume(); + EXPECT_FALSE(per_balancer_store.IsSuspended()); + EXPECT_EQ(0U, per_balancer_store.load_record_map().size()); + // Data merged after the store is resumed will be kept. + LoadRecordValue v2(23, 0, 51); + per_balancer_store.MergeRow(kKey1, v2); + EXPECT_EQ(1U, per_balancer_store.load_record_map().size()); + // Suspend the store. + per_balancer_store.Suspend(); + EXPECT_TRUE(per_balancer_store.IsSuspended()); + EXPECT_EQ(0U, per_balancer_store.load_record_map().size()); + // Data merged when the store is suspended won't be kept. + LoadRecordValue v3(62, 11); + per_balancer_store.MergeRow(kKey1, v3); + EXPECT_EQ(0U, per_balancer_store.load_record_map().size()); + // Resume the store. + per_balancer_store.Resume(); + EXPECT_FALSE(per_balancer_store.IsSuspended()); + EXPECT_EQ(0U, per_balancer_store.load_record_map().size()); + // Data merged after the store is resumed will be kept. + LoadRecordValue v4(225, 98); + per_balancer_store.MergeRow(kKey1, v4); + EXPECT_EQ(1U, per_balancer_store.load_record_map().size()); + // In-progress count is always kept. + EXPECT_EQ(per_balancer_store.GetNumCallsInProgressForReport(), + v1.start_count() - v1.ok_count() + v2.start_count() - + v2.error_count() + v3.start_count() - v3.ok_count() + + v4.start_count() - v4.ok_count()); +} + +TEST_F(PerBalancerStoreTest, DataAggregation) { + PerBalancerStore per_balancer_store(kLbId1, kLoadKey1); + // Construct some Values. + LoadRecordValue v1(992, 34, 13, 234.0, 164.0, 173467.38); + v1.InsertCallMetric(kMetric1, CallMetricValue(3, 2773.2)); + LoadRecordValue v2(4842, 213, 9, 393.0, 974.0, 1345.2398); + v2.InsertCallMetric(kMetric1, CallMetricValue(7, 25.234)); + v2.InsertCallMetric(kMetric2, CallMetricValue(2, 387.08)); + // v3 doesn't change the number of in-progress RPCs. + LoadRecordValue v3(293, 55, 293 - 55, 28764, 5284, 5772); + v3.InsertCallMetric(kMetric1, CallMetricValue(61, 3465.0)); + v3.InsertCallMetric(kMetric2, CallMetricValue(13, 672.0)); + // The initial state of the store. + uint64_t num_calls_in_progress = 0; + EXPECT_FALSE(per_balancer_store.IsNumCallsInProgressChangedSinceLastReport()); + EXPECT_EQ(per_balancer_store.GetNumCallsInProgressForReport(), + num_calls_in_progress); + // Merge v1 and get report of the number of in-progress calls. + per_balancer_store.MergeRow(kKey1, v1); + EXPECT_TRUE(per_balancer_store.IsNumCallsInProgressChangedSinceLastReport()); + EXPECT_EQ(per_balancer_store.GetNumCallsInProgressForReport(), + num_calls_in_progress += + (v1.start_count() - v1.ok_count() - v1.error_count())); + EXPECT_FALSE(per_balancer_store.IsNumCallsInProgressChangedSinceLastReport()); + // Merge v2 and get report of the number of in-progress calls. + per_balancer_store.MergeRow(kKey2, v2); + EXPECT_TRUE(per_balancer_store.IsNumCallsInProgressChangedSinceLastReport()); + EXPECT_EQ(per_balancer_store.GetNumCallsInProgressForReport(), + num_calls_in_progress += + (v2.start_count() - v2.ok_count() - v2.error_count())); + EXPECT_FALSE(per_balancer_store.IsNumCallsInProgressChangedSinceLastReport()); + // Merge v3 and get report of the number of in-progress calls. + per_balancer_store.MergeRow(kKey1, v3); + EXPECT_FALSE(per_balancer_store.IsNumCallsInProgressChangedSinceLastReport()); + EXPECT_EQ(per_balancer_store.GetNumCallsInProgressForReport(), + num_calls_in_progress); + // LoadRecordValue for kKey1 is aggregated correctly. + LoadRecordValue value_for_key1 = + per_balancer_store.load_record_map().find(kKey1)->second; + EXPECT_EQ(value_for_key1.start_count(), v1.start_count() + v3.start_count()); + EXPECT_EQ(value_for_key1.ok_count(), v1.ok_count() + v3.ok_count()); + EXPECT_EQ(value_for_key1.error_count(), v1.error_count() + v3.error_count()); + EXPECT_EQ(value_for_key1.bytes_sent(), v1.bytes_sent() + v3.bytes_sent()); + EXPECT_EQ(value_for_key1.bytes_recv(), v1.bytes_recv() + v3.bytes_recv()); + EXPECT_EQ(value_for_key1.latency_ms(), v1.latency_ms() + v3.latency_ms()); + EXPECT_EQ(value_for_key1.call_metrics().size(), 2U); + EXPECT_EQ(value_for_key1.call_metrics().find(kMetric1)->second.num_calls(), + v1.call_metrics().find(kMetric1)->second.num_calls() + + v3.call_metrics().find(kMetric1)->second.num_calls()); + EXPECT_EQ( + value_for_key1.call_metrics().find(kMetric1)->second.total_metric_value(), + v1.call_metrics().find(kMetric1)->second.total_metric_value() + + v3.call_metrics().find(kMetric1)->second.total_metric_value()); + EXPECT_EQ(value_for_key1.call_metrics().find(kMetric2)->second.num_calls(), + v3.call_metrics().find(kMetric2)->second.num_calls()); + EXPECT_EQ( + value_for_key1.call_metrics().find(kMetric2)->second.total_metric_value(), + v3.call_metrics().find(kMetric2)->second.total_metric_value()); + // LoadRecordValue for kKey2 is aggregated (trivially) correctly. + LoadRecordValue value_for_key2 = + per_balancer_store.load_record_map().find(kKey2)->second; + EXPECT_EQ(value_for_key2.start_count(), v2.start_count()); + EXPECT_EQ(value_for_key2.ok_count(), v2.ok_count()); + EXPECT_EQ(value_for_key2.error_count(), v2.error_count()); + EXPECT_EQ(value_for_key2.bytes_sent(), v2.bytes_sent()); + EXPECT_EQ(value_for_key2.bytes_recv(), v2.bytes_recv()); + EXPECT_EQ(value_for_key2.latency_ms(), v2.latency_ms()); + EXPECT_EQ(value_for_key2.call_metrics().size(), 2U); + EXPECT_EQ(value_for_key2.call_metrics().find(kMetric1)->second.num_calls(), + v2.call_metrics().find(kMetric1)->second.num_calls()); + EXPECT_EQ( + value_for_key2.call_metrics().find(kMetric1)->second.total_metric_value(), + v2.call_metrics().find(kMetric1)->second.total_metric_value()); + EXPECT_EQ(value_for_key2.call_metrics().find(kMetric2)->second.num_calls(), + v2.call_metrics().find(kMetric2)->second.num_calls()); + EXPECT_EQ( + value_for_key2.call_metrics().find(kMetric2)->second.total_metric_value(), + v2.call_metrics().find(kMetric2)->second.total_metric_value()); +} + +} // namespace +} // namespace testing +} // namespace grpc + +int main(int argc, char** argv) { + grpc_test_init(argc, argv); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json index 41722e063d..9092125a54 100644 --- a/tools/run_tests/generated/sources_and_headers.json +++ b/tools/run_tests/generated/sources_and_headers.json @@ -3909,6 +3909,26 @@ "gpr_test_util", "grpc", "grpc++", + "grpc++_test_util", + "grpc_test_util", + "lb_load_data_store" + ], + "headers": [], + "is_filegroup": false, + "language": "c++", + "name": "lb_load_data_store_test", + "src": [ + "test/cpp/server/load_reporter/load_data_store_test.cc" + ], + "third_party": false, + "type": "target" + }, + { + "deps": [ + "gpr", + "gpr_test_util", + "grpc", + "grpc++", "grpc++_test", "grpc_test_util" ], @@ -7469,6 +7489,23 @@ }, { "deps": [ + "grpc++" + ], + "headers": [ + "src/cpp/server/load_reporter/load_data_store.h" + ], + "is_filegroup": false, + "language": "c++", + "name": "lb_load_data_store", + "src": [ + "src/cpp/server/load_reporter/load_data_store.cc", + "src/cpp/server/load_reporter/load_data_store.h" + ], + "third_party": false, + "type": "lib" + }, + { + "deps": [ "grpc", "grpc++", "grpc++_core_stats", diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json index 067d9b300d..8eb5303e82 100644 --- a/tools/run_tests/generated/tests.json +++ b/tools/run_tests/generated/tests.json @@ -4404,6 +4404,30 @@ "flaky": false, "gtest": true, "language": "c++", + "name": "lb_load_data_store_test", + "platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "uses_polling": true + }, + { + "args": [], + "benchmark": false, + "ci_platforms": [ + "linux", + "mac", + "posix", + "windows" + ], + "cpu_cost": 1.0, + "exclude_configs": [], + "exclude_iomgrs": [], + "flaky": false, + "gtest": true, + "language": "c++", "name": "memory_test", "platforms": [ "linux", |