heron/tmanager/src/cpp/manager/tmetrics-collector.h - incubator-heron - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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 __TMETRICS_COLLECTOR_H_
 #define __TMETRICS_COLLECTOR_H_

 #include <map>
 #include <list>
 #include <string>
 #include "basics/callback.h"
 #include "basics/sptypes.h"
 #include "network/event_loop.h"
 #include "proto/tmanager.pb.h"
 #include "proto/topology.pb.h"
 #include "metrics/tmanager-metrics.h"

 namespace heron {
 namespace tmanager {

 using std::unique_ptr;
 using std::shared_ptr;

 // Helper class to manage aggregation and and serving of metrics. Metrics are logically stored as a
 // component_name -> {instance_id ->value}n .
 // TODO(kramasamy): Store metrics persistently to prevent against crashes.
 class TMetricsCollector {
  public:
   // _max_interval is how far along we keep individual metric blobs.
   TMetricsCollector(sp_int32 _max_interval, std::shared_ptr<EventLoop> eventLoop,
                     const std::string& metrics_sinks_yaml);

   // Deletes all stored ComponentMetrics.
   virtual ~TMetricsCollector();

   // Initiated on recieving a new _metrics from metricsmanager. Will initiate appropriate calls
   // to add metrics/exception stored in '_metrics' to the respective components.
   void AddMetric(const proto::tmanager::PublishMetrics& _metrics);

   // Returns a new response to fetch metrics. The request gets propagated to Component's and
   // Instance's get metrics. Doesn't own Response.
   unique_ptr<proto::tmanager::MetricResponse> GetMetrics(
                                                 const proto::tmanager::MetricRequest& _request,
                                                 const proto::api::Topology& _topology);

   // Returns response for fetching exceptions. Doesn't own response.
   unique_ptr<proto::tmanager::ExceptionLogResponse> GetExceptions(
       const proto::tmanager::ExceptionLogRequest& request);

   // Returns exception summary response. Doesn't own response.
   unique_ptr<proto::tmanager::ExceptionLogResponse> GetExceptionsSummary(
       const proto::tmanager::ExceptionLogRequest& request);

  private:
   // Fetches exceptions for ExceptionLogRequest. Save the returned exception in
   // 'all_exceptions'.
   //  Doesn't own 'all_exceptions' pointer
   void GetExceptionsHelper(const proto::tmanager::ExceptionLogRequest& request,
                            proto::tmanager::ExceptionLogResponse& all_excepions);

   // Aggregate exceptions from 'all_exceptions' to 'aggregate_exceptions'.
   // Doesn't own 'aggregate_exceptions'.
   void AggregateExceptions(const proto::tmanager::ExceptionLogResponse& all_exceptions,
                            proto::tmanager::ExceptionLogResponse& aggregate_exceptions);

   // Add metrics for 'component_name'
   void AddMetricsForComponent(const sp_string& component_name,
                               const proto::tmanager::MetricDatum& metrics_data);
   // Add exception logs for 'component_name'
   void AddExceptionsForComponent(const sp_string& component_name,
                                  const proto::tmanager::TmanagerExceptionLog& exception_log);

   // Clean all metrics.
   void Purge(EventLoop::Status _status);

   // Timeseries of metrics.
   struct TimeBucket {
     // A list of metrics each accumulated inside the time
     // that this bucket represents
     std::list<sp_string> data_;
     // Whats the start and end time that this TimeBucket contains metrics for
     sp_int32 start_time_;
     sp_int32 end_time_;

     explicit TimeBucket(sp_int32 bucket_interval) {
       start_time_ = time(NULL);
       end_time_ = start_time_ + bucket_interval;
     }

     bool overlaps(sp_int64 start_time, sp_int64 end_time) {
       return start_time_ <= end_time && start_time <= end_time_;
     }

     sp_double64 aggregate() {
       if (data_.empty()) {
         return 0;
       } else {
         sp_double64 total = 0;
         for (std::list<sp_string>::iterator it = data_.begin(); it != data_.end(); ++it) {
           total += strtod(it->c_str(), NULL);
         }
         return total;
       }
     }

     sp_int64 count() { return data_.size(); }
   };

   // Data structure to store metrics. A metric is a Time series of data.
   // TODO(kramasamy): Use proto to store this data structure.
   class Metric {
    public:
     // TODO(kramasamy): Add ctor for default UNKNOWN type and give a set type function.
     Metric(const sp_string& name, common::TManagerMetrics::MetricAggregationType type,
            sp_int32 nbuckets, sp_int32 bucket_interval);

     // Deletes all TimeBucket.
     virtual ~Metric();

     void Purge();

     // Add a new value to the end of 'data_' extending the time series.
     void AddValueToMetric(const sp_string& value);

     // Return  past '_nbuckets' value for this metric.
     void GetMetrics(bool minutely, sp_int64 start_time, sp_int64 end_time,
                     proto::tmanager::MetricResponse::IndividualMetric* response);

    private:
     sp_string name_;
     // Time series. data_ will be ordered by their time of arrival.
     std::list<unique_ptr<TimeBucket>> data_;
     // Type of metric. This can be SUM or AVG. It specify how to aggregate these metrics for
     // display.
     common::TManagerMetrics::MetricAggregationType metric_type_;

     sp_double64 all_time_cumulative_;

     sp_int64 all_time_nitems_;

     sp_int32 bucket_interval_;
   };

   // Most granualar metrics/exception store level. This store exception and metrics
   // associated with an instance.
   class InstanceMetrics {
    public:
     // ctor. '_instance_id' is the id generated by heron. '_nbuckets' number of metrics buckets
     // stored for instances belonging to this component.
     InstanceMetrics(const sp_string& instance_id, sp_int32 nbuckets, sp_int32 bucket_interval);
     // dtor
     virtual ~InstanceMetrics();

     // Clear old metrics associated with this instance.
     void Purge();

     // Add metrics with name '_name' of type '_type' and value _value.
     void AddMetricWithName(const sp_string& name,
                            common::TManagerMetrics::MetricAggregationType type,
                            const sp_string& value);

     // Add TmanagerExceptionLog to the list of exceptions for this instance_id.
     void AddExceptions(const proto::tmanager::TmanagerExceptionLog& exception);

     // Returns the metric metrics. Doesn't own _response.
     void GetMetrics(const proto::tmanager::MetricRequest& request, sp_int64 start_time,
                     sp_int64 end_time, proto::tmanager::MetricResponse& response);

     // Fills response for fetching exceptions. Doesn't own response.
     void GetExceptionLog(proto::tmanager::ExceptionLogResponse& response);

    private:
     // Create or return existing Metric. Retains ownership of Metric object returned.
     shared_ptr<Metric> GetOrCreateMetric(const sp_string& name,
                               common::TManagerMetrics::MetricAggregationType type);

     sp_string instance_id_;
     sp_int32 nbuckets_;
     sp_int32 bucket_interval_;
     // map between metric name and its values
     std::map<sp_string, shared_ptr<Metric>> metrics_;
     // list of exceptions
     std::list<unique_ptr<proto::tmanager::TmanagerExceptionLog>> exceptions_;
   };

   // Component level metrics. A component metrics is a map storing metrics for each of its
   // instance as 'InstanceMetrics'.
   class ComponentMetrics {
    public:
     // ctor. '_component_name' is the user supplied name given to the spout/bolt. '_nbuckets' is
     // number of buckets stored for this component.
     ComponentMetrics(const sp_string& component_name, sp_int32 nbuckets, sp_int32 bucket_interval);
     // dtor
     virtual ~ComponentMetrics();

     // Remove old metrics and exception associated with this spout/bolt component.
     void Purge();

     // Add metrics for an Instance 'instance_id' of this spout/bolt component.
     void AddMetricForInstance(const sp_string& instance_id, const sp_string& name,
                               common::TManagerMetrics::MetricAggregationType type,
                               const sp_string& value);
     // Add exception for an Instance 'instance_id' of this spout/bolt component.
     void AddExceptionForInstance(const sp_string& instance_id,
                                  const proto::tmanager::TmanagerExceptionLog& exception);

     // Request aggregated metrics for this component for the '_nbucket' interval.
     // Doesn't own '_response' object.
     void GetMetrics(const proto::tmanager::MetricRequest& request, sp_int64 start_time,
                     sp_int64 end_time, proto::tmanager::MetricResponse& response);

     // Returns response for fetching exceptions. Doesn't own response.
     void GetExceptionsForInstance(const sp_string& instance_id,
                                   proto::tmanager::ExceptionLogResponse& response);

     void GetAllExceptions(proto::tmanager::ExceptionLogResponse& response);

    private:
     // Create or return existing mutable InstanceMetrics associated with 'instance_id'. This
     // method doesn't verify if the instance_id is valid fof the component.
     // Doesn't transfer ownership of returned InstanceMetrics.
     shared_ptr<InstanceMetrics> GetOrCreateInstanceMetrics(const sp_string& instance_id);

     sp_string component_name_;
     sp_int32 nbuckets_;
     sp_int32 bucket_interval_;
     // map between instance id and its set of metrics
     std::map<sp_string, shared_ptr<InstanceMetrics>> metrics_;
   };

   // Create or return existing mutable ComponentMetrics associated with 'component_name'.
   // Doesn't transfer ownership of returned ComponentMetrics
   shared_ptr<ComponentMetrics> GetOrCreateComponentMetrics(const sp_string& component_name);

   // map of component name to its metrics
   std::map<sp_string, shared_ptr<ComponentMetrics>> metrics_;
   sp_int32 max_interval_;
   sp_int32 nintervals_;
   sp_int32 interval_;
   std::shared_ptr<EventLoop> eventLoop_;
   std::string metrics_sinks_yaml_;
   std::unique_ptr<common::TManagerMetrics> tmetrics_info_;
   time_t start_time_;
 };
 }  // namespace tmanager
 }  // namespace heron

 #endif
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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 __TMETRICS_COLLECTOR_H_
	#define __TMETRICS_COLLECTOR_H_

	#include <map>
	#include <list>
	#include <string>
	#include "basics/callback.h"
	#include "basics/sptypes.h"
	#include "network/event_loop.h"
	#include "proto/tmanager.pb.h"
	#include "proto/topology.pb.h"
	#include "metrics/tmanager-metrics.h"

	namespace heron {
	namespace tmanager {

	using std::unique_ptr;
	using std::shared_ptr;

	// Helper class to manage aggregation and and serving of metrics. Metrics are logically stored as a
	// component_name -> {instance_id ->value}n .
	// TODO(kramasamy): Store metrics persistently to prevent against crashes.
	class TMetricsCollector {
	public:
	// _max_interval is how far along we keep individual metric blobs.
	TMetricsCollector(sp_int32 _max_interval, std::shared_ptr<EventLoop> eventLoop,
	const std::string& metrics_sinks_yaml);

	// Deletes all stored ComponentMetrics.
	virtual ~TMetricsCollector();

	// Initiated on recieving a new _metrics from metricsmanager. Will initiate appropriate calls
	// to add metrics/exception stored in '_metrics' to the respective components.
	void AddMetric(const proto::tmanager::PublishMetrics& _metrics);

	// Returns a new response to fetch metrics. The request gets propagated to Component's and
	// Instance's get metrics. Doesn't own Response.
	unique_ptr<proto::tmanager::MetricResponse> GetMetrics(
	const proto::tmanager::MetricRequest& _request,
	const proto::api::Topology& _topology);

	// Returns response for fetching exceptions. Doesn't own response.
	unique_ptr<proto::tmanager::ExceptionLogResponse> GetExceptions(
	const proto::tmanager::ExceptionLogRequest& request);

	// Returns exception summary response. Doesn't own response.
	unique_ptr<proto::tmanager::ExceptionLogResponse> GetExceptionsSummary(
	const proto::tmanager::ExceptionLogRequest& request);

	private:
	// Fetches exceptions for ExceptionLogRequest. Save the returned exception in
	// 'all_exceptions'.
	// Doesn't own 'all_exceptions' pointer
	void GetExceptionsHelper(const proto::tmanager::ExceptionLogRequest& request,
	proto::tmanager::ExceptionLogResponse& all_excepions);

	// Aggregate exceptions from 'all_exceptions' to 'aggregate_exceptions'.
	// Doesn't own 'aggregate_exceptions'.
	void AggregateExceptions(const proto::tmanager::ExceptionLogResponse& all_exceptions,
	proto::tmanager::ExceptionLogResponse& aggregate_exceptions);

	// Add metrics for 'component_name'
	void AddMetricsForComponent(const sp_string& component_name,
	const proto::tmanager::MetricDatum& metrics_data);
	// Add exception logs for 'component_name'
	void AddExceptionsForComponent(const sp_string& component_name,
	const proto::tmanager::TmanagerExceptionLog& exception_log);

	// Clean all metrics.
	void Purge(EventLoop::Status _status);

	// Timeseries of metrics.
	struct TimeBucket {
	// A list of metrics each accumulated inside the time
	// that this bucket represents
	std::list<sp_string> data_;
	// Whats the start and end time that this TimeBucket contains metrics for
	sp_int32 start_time_;
	sp_int32 end_time_;

	explicit TimeBucket(sp_int32 bucket_interval) {
	start_time_ = time(NULL);
	end_time_ = start_time_ + bucket_interval;
	}

	bool overlaps(sp_int64 start_time, sp_int64 end_time) {
	return start_time_ <= end_time && start_time <= end_time_;
	}

	sp_double64 aggregate() {
	if (data_.empty()) {
	return 0;
	} else {
	sp_double64 total = 0;
	for (std::list<sp_string>::iterator it = data_.begin(); it != data_.end(); ++it) {
	total += strtod(it->c_str(), NULL);
	}
	return total;
	}
	}

	sp_int64 count() { return data_.size(); }
	};

	// Data structure to store metrics. A metric is a Time series of data.
	// TODO(kramasamy): Use proto to store this data structure.
	class Metric {
	public:
	// TODO(kramasamy): Add ctor for default UNKNOWN type and give a set type function.
	Metric(const sp_string& name, common::TManagerMetrics::MetricAggregationType type,
	sp_int32 nbuckets, sp_int32 bucket_interval);

	// Deletes all TimeBucket.
	virtual ~Metric();

	void Purge();

	// Add a new value to the end of 'data_' extending the time series.
	void AddValueToMetric(const sp_string& value);

	// Return past '_nbuckets' value for this metric.
	void GetMetrics(bool minutely, sp_int64 start_time, sp_int64 end_time,
	proto::tmanager::MetricResponse::IndividualMetric* response);

	private:
	sp_string name_;
	// Time series. data_ will be ordered by their time of arrival.
	std::list<unique_ptr<TimeBucket>> data_;
	// Type of metric. This can be SUM or AVG. It specify how to aggregate these metrics for
	// display.
	common::TManagerMetrics::MetricAggregationType metric_type_;

	sp_double64 all_time_cumulative_;

	sp_int64 all_time_nitems_;

	sp_int32 bucket_interval_;
	};

	// Most granualar metrics/exception store level. This store exception and metrics
	// associated with an instance.
	class InstanceMetrics {
	public:
	// ctor. '_instance_id' is the id generated by heron. '_nbuckets' number of metrics buckets
	// stored for instances belonging to this component.
	InstanceMetrics(const sp_string& instance_id, sp_int32 nbuckets, sp_int32 bucket_interval);
	// dtor
	virtual ~InstanceMetrics();

	// Clear old metrics associated with this instance.
	void Purge();

	// Add metrics with name '_name' of type '_type' and value _value.
	void AddMetricWithName(const sp_string& name,
	common::TManagerMetrics::MetricAggregationType type,
	const sp_string& value);

	// Add TmanagerExceptionLog to the list of exceptions for this instance_id.
	void AddExceptions(const proto::tmanager::TmanagerExceptionLog& exception);

	// Returns the metric metrics. Doesn't own _response.
	void GetMetrics(const proto::tmanager::MetricRequest& request, sp_int64 start_time,
	sp_int64 end_time, proto::tmanager::MetricResponse& response);

	// Fills response for fetching exceptions. Doesn't own response.
	void GetExceptionLog(proto::tmanager::ExceptionLogResponse& response);

	private:
	// Create or return existing Metric. Retains ownership of Metric object returned.
	shared_ptr<Metric> GetOrCreateMetric(const sp_string& name,
	common::TManagerMetrics::MetricAggregationType type);

	sp_string instance_id_;
	sp_int32 nbuckets_;
	sp_int32 bucket_interval_;
	// map between metric name and its values
	std::map<sp_string, shared_ptr<Metric>> metrics_;
	// list of exceptions
	std::list<unique_ptr<proto::tmanager::TmanagerExceptionLog>> exceptions_;
	};

	// Component level metrics. A component metrics is a map storing metrics for each of its
	// instance as 'InstanceMetrics'.
	class ComponentMetrics {
	public:
	// ctor. '_component_name' is the user supplied name given to the spout/bolt. '_nbuckets' is
	// number of buckets stored for this component.
	ComponentMetrics(const sp_string& component_name, sp_int32 nbuckets, sp_int32 bucket_interval);
	// dtor
	virtual ~ComponentMetrics();

	// Remove old metrics and exception associated with this spout/bolt component.
	void Purge();

	// Add metrics for an Instance 'instance_id' of this spout/bolt component.
	void AddMetricForInstance(const sp_string& instance_id, const sp_string& name,
	common::TManagerMetrics::MetricAggregationType type,
	const sp_string& value);
	// Add exception for an Instance 'instance_id' of this spout/bolt component.
	void AddExceptionForInstance(const sp_string& instance_id,
	const proto::tmanager::TmanagerExceptionLog& exception);

	// Request aggregated metrics for this component for the '_nbucket' interval.
	// Doesn't own '_response' object.
	void GetMetrics(const proto::tmanager::MetricRequest& request, sp_int64 start_time,
	sp_int64 end_time, proto::tmanager::MetricResponse& response);

	// Returns response for fetching exceptions. Doesn't own response.
	void GetExceptionsForInstance(const sp_string& instance_id,
	proto::tmanager::ExceptionLogResponse& response);

	void GetAllExceptions(proto::tmanager::ExceptionLogResponse& response);

	private:
	// Create or return existing mutable InstanceMetrics associated with 'instance_id'. This
	// method doesn't verify if the instance_id is valid fof the component.
	// Doesn't transfer ownership of returned InstanceMetrics.
	shared_ptr<InstanceMetrics> GetOrCreateInstanceMetrics(const sp_string& instance_id);

	sp_string component_name_;
	sp_int32 nbuckets_;
	sp_int32 bucket_interval_;
	// map between instance id and its set of metrics
	std::map<sp_string, shared_ptr<InstanceMetrics>> metrics_;
	};

	// Create or return existing mutable ComponentMetrics associated with 'component_name'.
	// Doesn't transfer ownership of returned ComponentMetrics
	shared_ptr<ComponentMetrics> GetOrCreateComponentMetrics(const sp_string& component_name);

	// map of component name to its metrics
	std::map<sp_string, shared_ptr<ComponentMetrics>> metrics_;
	sp_int32 max_interval_;
	sp_int32 nintervals_;
	sp_int32 interval_;
	std::shared_ptr<EventLoop> eventLoop_;
	std::string metrics_sinks_yaml_;
	std::unique_ptr<common::TManagerMetrics> tmetrics_info_;
	time_t start_time_;
	};
	} // namespace tmanager
	} // namespace heron

	#endif