Add load data store

11 files changed, 1416 insertions, 2 deletions

diff --git a/BUILD b/BUILD
index e04d0df1a2..f899632a1b 100644
--- a/BUILD
+++ b/BUILD
@@ -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
diff --git a/Makefile b/Makefile
index c646826339..4e9b7c4dea 100644
--- a/Makefile
+++ b/Makefile
@@ -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
diff --git a/grpc.gyp b/grpc.gyp
index b36b166175..c5e10edf7f 100644
--- a/grpc.gyp
+++ b/grpc.gyp
@@ -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",