SimpleStatistics: implement serialization roundtrip (envoyproxy#659)

Part of the horizontal scaling effort: `SimpleStatistic` and `StreamingStatistic ` get used
for tracking header byte counts, hence we should be able to wire
transfer statistics for those to be able to properly aggregate.

Signed-off-by: Otto van der Schaaf <[email protected]>

Author: Otto van der Schaaf

internal_proto/README.md

@@ -0,0 +1,3 @@
+# Internal proto definitions
+
+Warning: this directory contains proto defininitions which are used internally by Nighthawk, and are subject to change.

internal_proto/statistic/BUILD

@@ -0,0 +1,11 @@
+load("@envoy_api//bazel:api_build_system.bzl", "api_cc_py_proto_library")
+
+licenses(["notice"])  # Apache 2
+
+api_cc_py_proto_library(
+    name = "statistic",
+    srcs = [
+        "statistic.proto",
+    ],
+    visibility = ["//visibility:public"],
+)

internal_proto/statistic/statistic.proto

@@ -0,0 +1,26 @@
+syntax = "proto3";
+
+package nighthawk.internal;
+
+// This package contains proto definitions which allow serialization / deserialization
+// of statistics implementations. Naming maps 1:1 with the statistics implementations
+// over at source/common/statistics_impl.h. See the code & doc comments there for further
+// information about the corresponding statistics implementations.
+
+message SimpleStatistic {
+  uint64 count = 1;
+  string id = 2;
+  uint64 min = 5;
+  uint64 max = 6;
+  double sum_x = 7;
+  double sum_x_2 = 8;
+}
+
+message StreamingStatistic {
+  uint64 count = 1;
+  string id = 2;
+  uint64 min = 5;
+  uint64 max = 6;
+  double mean = 7;
+  double accumulated_variance = 8;
+}

source/common/BUILD

@@ -146,6 +146,7 @@ envoy_cc_library(
         "//api/client:grpc_service_lib",
         "//include/nighthawk/client:client_includes",
         "//include/nighthawk/common:base_includes",
+        "//internal_proto/statistic:statistic_cc_proto",
         "@com_google_absl//absl/random",
         "@com_google_absl//absl/strings",
         "@dep_hdrhistogram_c//:hdrhistogram_c",

source/common/statistic_impl.cc

@@ -9,6 +9,8 @@
 #include "external/envoy/source/common/common/assert.h"
 #include "external/envoy/source/common/protobuf/utility.h"
 
+#include "internal_proto/statistic/statistic.pb.h"
+
 namespace Nighthawk {
 
 namespace {
@@ -104,6 +106,38 @@ StatisticPtr SimpleStatistic::combine(const Statistic& statistic) const {
   return combined;
 }
 
+absl::StatusOr<std::unique_ptr<std::istream>> SimpleStatistic::serializeNative() const {
+  nighthawk::internal::SimpleStatistic proto;
+  proto.set_id(id());
+  proto.set_count(count());
+  proto.set_min(min());
+  proto.set_max(max());
+  proto.set_sum_x(sum_x_);
+  proto.set_sum_x_2(sum_x2_);
+
+  std::string tmp;
+  proto.SerializeToString(&tmp);
+  auto write_stream = std::make_unique<std::stringstream>();
+  *write_stream << tmp;
+  return write_stream;
+}
+
+absl::Status SimpleStatistic::deserializeNative(std::istream& stream) {
+  nighthawk::internal::SimpleStatistic proto;
+  std::string tmp(std::istreambuf_iterator<char>(stream), {});
+  if (!proto.ParseFromString(tmp)) {
+    ENVOY_LOG(error, "Failed to read back SimpleStatistic data.");
+    return absl::Status(absl::StatusCode::kInternal, "Failed to read back SimpleStatistic data");
+  }
+  id_ = proto.id();
+  count_ = proto.count();
+  min_ = proto.min();
+  max_ = proto.max();
+  sum_x_ = proto.sum_x();
+  sum_x2_ = proto.sum_x_2();
+  return absl::OkStatus();
+}
+
 void StreamingStatistic::addValue(uint64_t value) {
   double delta, delta_n;
   StatisticImpl::addValue(value);
@@ -142,6 +176,38 @@ StatisticPtr StreamingStatistic::combine(const Statistic& statistic) const {
   return combined;
 }
 
+absl::StatusOr<std::unique_ptr<std::istream>> StreamingStatistic::serializeNative() const {
+  nighthawk::internal::StreamingStatistic proto;
+  proto.set_id(id());
+  proto.set_count(count());
+  proto.set_min(min());
+  proto.set_max(max());
+  proto.set_mean(mean_);
+  proto.set_accumulated_variance(accumulated_variance_);
+
+  std::string tmp;
+  proto.SerializeToString(&tmp);
+  auto write_stream = std::make_unique<std::stringstream>();
+  *write_stream << tmp;
+  return write_stream;
+}
+
+absl::Status StreamingStatistic::deserializeNative(std::istream& stream) {
+  nighthawk::internal::StreamingStatistic proto;
+  std::string tmp(std::istreambuf_iterator<char>(stream), {});
+  if (!proto.ParseFromString(tmp)) {
+    ENVOY_LOG(error, "Failed to read back StreamingStatistic data.");
+    return absl::Status(absl::StatusCode::kInternal, "Failed to read back StreamingStatistic data");
+  }
+  id_ = proto.id();
+  count_ = proto.count();
+  min_ = proto.min();
+  max_ = proto.max();
+  mean_ = proto.mean();
+  accumulated_variance_ = proto.accumulated_variance();
+  return absl::OkStatus();
+}
+
 InMemoryStatistic::InMemoryStatistic() : streaming_stats_(std::make_unique<StreamingStatistic>()) {}
 
 void InMemoryStatistic::addValue(uint64_t sample_value) {

source/common/statistic_impl.h

@@ -69,6 +69,8 @@ class SimpleStatistic : public StatisticImpl {
   StatisticPtr createNewInstanceOfSameType() const override {
     return std::make_unique<SimpleStatistic>();
   };
+  absl::StatusOr<std::unique_ptr<std::istream>> serializeNative() const override;
+  absl::Status deserializeNative(std::istream&) override;
 
 private:
   double sum_x_{0};
@@ -94,6 +96,8 @@ class StreamingStatistic : public StatisticImpl {
   StatisticPtr createNewInstanceOfSameType() const override {
     return std::make_unique<StreamingStatistic>();
   };
+  absl::StatusOr<std::unique_ptr<std::istream>> serializeNative() const override;
+  absl::Status deserializeNative(std::istream&) override;
 
 private:
   double mean_{0};