Support TCP filter to send periodical reports. (envoyproxy#971)

Automatic merge from submit-queue.

Support TCP filter to send periodical reports.

**What this PR does / why we need it**: Support TCP filter to send periodical reports for long TCP connections. The periodical reports contain delta information such as bytes received and sent during the recent period. 

**Which issue this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close that issue when PR gets merged)*: fixes envoyproxy#539 

**Special notes for your reviewer**:

**Release note**:

```release-note
NONE
```

src/envoy/mixer/mixer_control.cc

@@ -21,6 +21,12 @@ namespace Http {
 namespace Mixer {
 namespace {
 
+// Default time interval for periodical report is 10 seconds.
+const std::chrono::milliseconds kDefaultReportIntervalMs(10000);
+
+// Minimum time interval for periodical report is 1 seconds.
+const std::chrono::milliseconds kMinReportIntervalMs(1000);
+
 // A class to wrap envoy timer for mixer client timer.
 class EnvoyTimer : public ::istio::mixer_client::Timer {
  public:
@@ -74,13 +80,29 @@ HttpMixerControl::HttpMixerControl(const HttpMixerConfig& mixer_config,
 TcpMixerControl::TcpMixerControl(const TcpMixerConfig& mixer_config,
                                  Upstream::ClusterManager& cm,
                                  Event::Dispatcher& dispatcher,
-                                 Runtime::RandomGenerator& random) {
+                                 Runtime::RandomGenerator& random)
+    : dispatcher_(dispatcher) {
   ::istio::mixer_control::tcp::Controller::Options options(
       mixer_config.tcp_config);
 
   CreateEnvironment(cm, dispatcher, random, &options.env);
 
   controller_ = ::istio::mixer_control::tcp::Controller::Create(options);
+
+  if (mixer_config.tcp_config.has_report_interval() &&
+      mixer_config.tcp_config.report_interval().seconds() >= 0 &&
+      mixer_config.tcp_config.report_interval().nanos() >= 0) {
+    report_interval_ms_ = std::chrono::milliseconds(
+        mixer_config.tcp_config.report_interval().seconds() * 1000 +
+        mixer_config.tcp_config.report_interval().nanos() / 1000000);
+    // If configured time interval is less than 1 second, then set report
+    // interval to 1 second.
+    if (report_interval_ms_ < kMinReportIntervalMs) {
+      report_interval_ms_ = kMinReportIntervalMs;
+    }
+  } else {
+    report_interval_ms_ = kDefaultReportIntervalMs;
+  }
 }
 
 }  // namespace Mixer

src/envoy/mixer/mixer_control.h

@@ -62,9 +62,20 @@ class TcpMixerControl final : public ThreadLocal::ThreadLocalObject {
     return controller_.get();
   }
 
+  std::chrono::milliseconds report_interval_ms() const {
+    return report_interval_ms_;
+  }
+
+  Event::Dispatcher& dispatcher() { return dispatcher_; }
+
  private:
   // The mixer control
   std::unique_ptr<::istio::mixer_control::tcp::Controller> controller_;
+
+  // Time interval in milliseconds for sending periodical delta reports.
+  std::chrono::milliseconds report_interval_ms_;
+
+  Event::Dispatcher& dispatcher_;
 };
 
 }  // namespace Mixer

src/envoy/mixer/tcp_filter.cc

@@ -65,6 +65,13 @@ class TcpInstance : public Network::Filter,
  private:
   enum class State { NotStarted, Calling, Completed, Closed };
 
+  // This function is invoked when timer event fires. It sends periodical delta
+  // reports.
+  void OnTimer() {
+    handler_->Report(this, /* is_final_report */ false);
+    report_timer_->enableTimer(mixer_control_.report_interval_ms());
+  }
+
   istio::mixer_client::CancelFunc cancel_check_;
   TcpMixerControl& mixer_control_;
   std::unique_ptr<::istio::mixer_control::tcp::RequestHandler> handler_;
@@ -76,6 +83,9 @@ class TcpInstance : public Network::Filter,
   int check_status_code_{};
   std::chrono::time_point<std::chrono::system_clock> start_time_;
 
+  // Timer that periodically sends reports.
+  Event::TimerPtr report_timer_;
+
  public:
   TcpInstance(TcpConfigPtr config) : mixer_control_(config->mixer_control()) {
     ENVOY_LOG(debug, "Called TcpInstance: {}", __func__);
@@ -159,6 +169,9 @@ class TcpInstance : public Network::Filter,
       if (!calling_check_) {
         filter_callbacks_->continueReading();
       }
+      report_timer_ =
+          mixer_control_.dispatcher().createTimer([this]() { OnTimer(); });
+      report_timer_->enableTimer(mixer_control_.report_interval_ms());
     }
   }
 
@@ -175,7 +188,10 @@ class TcpInstance : public Network::Filter,
     if (event == Network::ConnectionEvent::RemoteClose ||
         event == Network::ConnectionEvent::LocalClose) {
       if (state_ != State::Closed && handler_) {
-        handler_->Report(this);
+        if (report_timer_) {
+          report_timer_->disableTimer();
+        }
+        handler_->Report(this, /* is_final_report */ true);
       }
       cancelCheck();
     }