Add some pre-calculated metrics (#1079)

morningman · web-flow · commit afa3aa90696c · 2019-04-30T11:12:23.000+08:00
1. max io util of disks
2. max network send/receive bytes rate of all network devices
3. base/cumulative compaction request counter and failure counter
diff --git a/be/src/common/daemon.cpp b/be/src/common/daemon.cpp
@@ -51,6 +51,7 @@
 #include "exprs/hll_hash_function.h"
 #include "olap/options.h"
 #include "util/time.h"
+#include "util/system_metrics.h"
 
 namespace doris {
 
@@ -111,17 +112,28 @@ void* memory_maintenance_thread(void* dummy) {
  * this thread will calculate some metrics at a fix interval(15 sec)
  * 1. push bytes per second
  * 2. scan bytes per second
+ * 3. max io util of all disks
+ * 4. max network send bytes rate
+ * 5. max network receive bytes rate
  */
 void* calculate_metrics(void* dummy) {
     int64_t last_ts = -1L;
     int64_t lst_push_bytes = -1;
     int64_t lst_query_bytes = -1;
 
+    std::map<std::string, int64_t> lst_disks_io_time;
+    std::map<std::string, int64_t> lst_net_send_bytes;
+    std::map<std::string, int64_t> lst_net_receive_bytes;
+
     while (true) {
+        DorisMetrics::metrics()->trigger_hook();
+
         if (last_ts == -1L) {
             last_ts = GetCurrentTimeMicros() / 1000;
             lst_push_bytes = DorisMetrics::push_request_write_bytes.value();
             lst_query_bytes = DorisMetrics::query_scan_bytes.value();
+            DorisMetrics::system_metrics()->get_disks_io_time(&lst_disks_io_time);
+            DorisMetrics::system_metrics()->get_network_traffic(&lst_net_send_bytes, &lst_net_receive_bytes);
         } else {
             int64_t current_ts = GetCurrentTimeMicros() / 1000;
             long interval = (current_ts - last_ts) / 1000;
@@ -140,6 +152,22 @@ void* calculate_metrics(void* dummy) {
             DorisMetrics::query_scan_bytes_per_second.set_value(
                 qps < 0 ? 0 : qps);
             lst_query_bytes = current_query_bytes;
+
+            // 3. max disk io util
+            DorisMetrics::max_disk_io_util_percent.set_value(
+                DorisMetrics::system_metrics()->get_max_io_util(lst_disks_io_time, 15));
+            // update lst map
+            DorisMetrics::system_metrics()->get_disks_io_time(&lst_disks_io_time);
+
+            // 4. max network traffic
+            int64_t max_send = 0;
+            int64_t max_receive = 0;
+            DorisMetrics::system_metrics()->get_max_net_traffic(
+                lst_net_send_bytes, lst_net_receive_bytes, 15, &max_send, &max_receive);
+            DorisMetrics::max_network_send_bytes_rate.set_value(max_send);
+            DorisMetrics::max_network_receive_bytes_rate.set_value(max_receive);
+            // update lst map
+            DorisMetrics::system_metrics()->get_network_traffic(&lst_net_send_bytes, &lst_net_receive_bytes);
         }
 
         sleep(15); // 15 seconds
diff --git a/be/src/exec/olap_table_info.h b/be/src/exec/olap_table_info.h
@@ -116,13 +116,7 @@ class OlapTablePartKeyComparator {
 
         bool lhs_null = lhs->is_null(_slot_desc->null_indicator_offset());
         bool rhs_null = rhs->is_null(_slot_desc->null_indicator_offset());
-        if (lhs_null && rhs_null) {
-            return false;
-        } else if (lhs_null) {
-            return true;
-        } else if (rhs_null) {
-            return false;
-        }
+        if (lhs_null || rhs_null) { return !rhs_null; }
 
         auto lhs_value = lhs->get_slot(_slot_desc->tuple_offset());
         auto rhs_value = rhs->get_slot(_slot_desc->tuple_offset());
diff --git a/be/src/http/action/metrics_action.cpp b/be/src/http/action/metrics_action.cpp
@@ -57,12 +57,18 @@ class SimpleCoreMetricsVisitor : public MetricsVisitor {
     static const std::string PROCESS_THREAD_NUM;
     static const std::string PUSH_REQUEST_WRITE_BYTES_PER_SECOND;
     static const std::string QUERY_SCAN_BYTES_PER_SECOND;
+    static const std::string MAX_DISK_IO_UTIL_PERCENT;
+    static const std::string MAX_NETWORK_SEND_BYTES_RATE;
+    static const std::string MAX_NETWORK_RECEIVE_BYTES_RATE;
 };
 
 const std::string SimpleCoreMetricsVisitor::PROCESS_FD_NUM_USED = "process_fd_num_used";
 const std::string SimpleCoreMetricsVisitor::PROCESS_THREAD_NUM = "process_thread_num";
 const std::string SimpleCoreMetricsVisitor::PUSH_REQUEST_WRITE_BYTES_PER_SECOND = "push_request_write_bytes_per_second";
 const std::string SimpleCoreMetricsVisitor::QUERY_SCAN_BYTES_PER_SECOND = "query_scan_bytes_per_second";
+const std::string SimpleCoreMetricsVisitor::MAX_DISK_IO_UTIL_PERCENT = "max_disk_io_util_percent";
+const std::string SimpleCoreMetricsVisitor::MAX_NETWORK_SEND_BYTES_RATE= "max_network_send_bytes_rate";
+const std::string SimpleCoreMetricsVisitor::MAX_NETWORK_RECEIVE_BYTES_RATE= "max_network_receive_bytes_rate";
 
 void PrometheusMetricsVisitor::visit(const std::string& prefix,
                                      const std::string& name,
@@ -82,7 +88,7 @@ void PrometheusMetricsVisitor::visit(const std::string& prefix,
     case MetricType::COUNTER:
     case MetricType::GAUGE:
         for (auto& it : collector->metrics()) {
-            _visit_simple_metric(metric_name, it.first, (SimpleMetric*)it.second);
+            _visit_simple_metric(metric_name, it.first, (SimpleMetric*) it.second);
         }
         break;
     default:
@@ -117,7 +123,10 @@ void SimpleCoreMetricsVisitor::visit(const std::string& prefix,
 
     if (name != PROCESS_FD_NUM_USED && name != PROCESS_THREAD_NUM
         && name != PUSH_REQUEST_WRITE_BYTES_PER_SECOND
-        && name != QUERY_SCAN_BYTES_PER_SECOND) {
+        && name != QUERY_SCAN_BYTES_PER_SECOND
+        && name != MAX_DISK_IO_UTIL_PERCENT
+        && name != MAX_NETWORK_SEND_BYTES_RATE
+        && name != MAX_NETWORK_RECEIVE_BYTES_RATE) {
         return;
     }
 
diff --git a/be/src/olap/olap_engine.cpp b/be/src/olap/olap_engine.cpp
@@ -1828,16 +1828,19 @@ void OLAPEngine::perform_cumulative_compaction() {
     OLAPTablePtr best_table = _find_best_tablet_to_compaction(CompactionType::CUMULATIVE_COMPACTION);
     if (best_table == nullptr) { return; }
 
+    DorisMetrics::cumulative_compaction_request_total.increment(1);
     CumulativeCompaction cumulative_compaction;
     OLAPStatus res = cumulative_compaction.init(best_table);
     if (res != OLAP_SUCCESS) {
+        DorisMetrics::cumulative_compaction_request_failed.increment(1);
         LOG(WARNING) << "failed to init cumulative compaction."
                      << "table=" << best_table->full_name();
         return;
     }
 
     res = cumulative_compaction.run();
     if (res != OLAP_SUCCESS) {
+        DorisMetrics::cumulative_compaction_request_failed.increment(1);
         LOG(WARNING) << "failed to do cumulative compaction."
                      << "table=" << best_table->full_name();
         return;
@@ -1848,16 +1851,19 @@ void OLAPEngine::perform_base_compaction() {
     OLAPTablePtr best_table = _find_best_tablet_to_compaction(CompactionType::BASE_COMPACTION);
     if (best_table == nullptr) { return; }
 
+    DorisMetrics::base_compaction_request_total.increment(1);
     BaseCompaction base_compaction;
     OLAPStatus res = base_compaction.init(best_table);
     if (res != OLAP_SUCCESS) {
+        DorisMetrics::base_compaction_request_failed.increment(1);
         LOG(WARNING) << "failed to init base compaction."
                      << "table=" << best_table->full_name();
         return;
     }
 
     res = base_compaction.run();
     if (res != OLAP_SUCCESS) {
+        DorisMetrics::base_compaction_request_failed.increment(1);
         LOG(WARNING) << "failed to init base compaction."
                      << "table=" << best_table->full_name();
         return;
diff --git a/be/src/util/doris_metrics.cpp b/be/src/util/doris_metrics.cpp
@@ -98,6 +98,9 @@ IntGaugeMetricsMap DorisMetrics::disks_state;
 
 IntGauge DorisMetrics::push_request_write_bytes_per_second;
 IntGauge DorisMetrics::query_scan_bytes_per_second;
+IntGauge DorisMetrics::max_disk_io_util_percent;
+IntGauge DorisMetrics::max_network_send_bytes_rate;
+IntGauge DorisMetrics::max_network_receive_bytes_rate;
 
 DorisMetrics::DorisMetrics() : _metrics(nullptr), _system_metrics(nullptr) {
 }
@@ -220,6 +223,9 @@ void DorisMetrics::initialize(
 
     REGISTER_DORIS_METRIC(push_request_write_bytes_per_second);
     REGISTER_DORIS_METRIC(query_scan_bytes_per_second);
+    REGISTER_DORIS_METRIC(max_disk_io_util_percent);
+    REGISTER_DORIS_METRIC(max_network_send_bytes_rate);
+    REGISTER_DORIS_METRIC(max_network_receive_bytes_rate);
 
     _metrics->register_hook(_s_hook_name, std::bind(&DorisMetrics::update, this));
 
diff --git a/be/src/util/doris_metrics.h b/be/src/util/doris_metrics.h
@@ -123,6 +123,9 @@ class DorisMetrics {
     // by metric calculator
     static IntGauge push_request_write_bytes_per_second;
     static IntGauge query_scan_bytes_per_second;
+    static IntGauge max_disk_io_util_percent;
+    static IntGauge max_network_send_bytes_rate;
+    static IntGauge max_network_receive_bytes_rate;
 
     ~DorisMetrics();
     // call before calling metrics
@@ -135,6 +138,7 @@ class DorisMetrics {
 
     static DorisMetrics* instance() { return &_s_doris_metrics; }
     static MetricRegistry* metrics() { return _s_doris_metrics._metrics; }
+    static SystemMetrics* system_metrics() { return _s_doris_metrics._system_metrics; }
 private:
     // Don't allow constrctor
     DorisMetrics();
diff --git a/be/src/util/metrics.h b/be/src/util/metrics.h
@@ -302,14 +302,28 @@ class MetricRegistry {
 
     void collect(MetricsVisitor* visitor) {
         std::lock_guard<SpinLock> l(_lock);
-        // Before we collect, need to call hooks
-        for (auto& it : _hooks) {
-            it.second();
+        if (!config::enable_metric_calculator) {
+            // Before we collect, need to call hooks
+            unprotected_trigger_hook();
         }
+
         for (auto& it : _collectors) {
             it.second->collect(_name, it.first, visitor);
         }
     }
+
+    void trigger_hook() {
+        std::lock_guard<SpinLock> l(_lock);
+        unprotected_trigger_hook();
+    }
+
+private:
+    void unprotected_trigger_hook() {
+        for (auto& it : _hooks) {
+            it.second();
+        }
+    }
+
 private:
     void _deregister_locked(Metric* metric);
 
diff --git a/be/src/util/system_metrics.cpp b/be/src/util/system_metrics.cpp
@@ -435,4 +435,64 @@ void SystemMetrics::_update_fd_metrics() {
     fclose(fp);
 }
 
+int64_t SystemMetrics::get_max_io_util(
+        const std::map<std::string, int64_t>& lst_value, int64_t interval_sec) {
+    int64_t max = 0;
+    for (auto& it : _disk_metrics) {
+        int64_t cur = it.second->io_time_ms.value();
+        const auto find = lst_value.find(it.first);
+        if (find == lst_value.end()) {
+            continue;
+        }
+        int64_t incr = cur - find->second;
+        if (incr > max) max = incr;
+    }
+    return max / interval_sec / 10;
+}
+
+void SystemMetrics::get_disks_io_time(std::map<std::string, int64_t>* map) {
+    map->clear();
+    for (auto& it : _disk_metrics) {
+        map->emplace(it.first, it.second->io_time_ms.value());
+    }
+}
+
+void SystemMetrics::get_network_traffic(
+            std::map<std::string, int64_t>* send_map,
+            std::map<std::string, int64_t>* rcv_map) {
+    send_map->clear();
+    rcv_map->clear();
+    for (auto& it : _net_metrics) {
+        if (it.first == "lo") { continue; }
+        send_map->emplace(it.first, it.second->send_bytes.value());
+        rcv_map->emplace(it.first, it.second->receive_bytes.value());
+    }
+}
+
+void SystemMetrics::get_max_net_traffic(
+            const std::map<std::string, int64_t>& lst_send_map,
+            const std::map<std::string, int64_t>& lst_rcv_map,
+            int64_t interval_sec,
+            int64_t* send_rate, int64_t* rcv_rate) {
+    int64_t max_send = 0;
+    int64_t max_rcv = 0;
+    for (auto& it : _net_metrics) {
+        int64_t cur_send = it.second->send_bytes.value();
+        int64_t cur_rcv = it.second->receive_bytes.value();
+
+        const auto find_send = lst_send_map.find(it.first);
+        if (find_send != lst_send_map.end()) {
+            int64_t incr = cur_send - find_send->second;
+            if (incr > max_send) max_send = incr;
+        }
+        const auto find_rcv= lst_rcv_map.find(it.first);
+        if (find_rcv != lst_rcv_map.end()) {
+            int64_t incr = cur_rcv - find_rcv->second;
+            if (incr > max_rcv) max_rcv = incr;
+        }
+    }
+
+    *send_rate = max_send / interval_sec;
+    *rcv_rate = max_rcv / interval_sec;
 }
+} // end namespace
diff --git a/be/src/util/system_metrics.h b/be/src/util/system_metrics.h
@@ -17,6 +17,7 @@
 
 #include "util/metrics.h"
 
+#include <map>
 #include <memory>
 
 namespace doris {
@@ -40,6 +41,19 @@ class SystemMetrics {
     // update metrics
     void update();
 
+    void get_disks_io_time(std::map<std::string, int64_t>* map);
+    int64_t get_max_io_util(
+        const std::map<std::string, int64_t>& lst_value, int64_t interval_sec);
+
+    void get_network_traffic(
+            std::map<std::string, int64_t>* send_map,
+            std::map<std::string, int64_t>* rcv_map);
+    void get_max_net_traffic(
+            const std::map<std::string, int64_t>& lst_send_map,
+            const std::map<std::string, int64_t>& lst_rcv_map,
+            int64_t interval_sec,
+            int64_t* send_rate, int64_t* rcv_rate);
+
 private:
     void _install_cpu_metrics(MetricRegistry*);
     // On Intel(R) Xeon(R) CPU E5-2450 0 @ 2.10GHz;
diff --git a/fe/src/main/java/org/apache/doris/metric/MetricRepo.java b/fe/src/main/java/org/apache/doris/metric/MetricRepo.java
@@ -61,6 +61,7 @@ public final class MetricRepo {
     public static LongCounterMetric COUNTER_EDIT_LOG_SIZE_BYTES;
     public static LongCounterMetric COUNTER_IMAGE_WRITE;
     public static LongCounterMetric COUNTER_IMAGE_PUSH;
+    public static LongCounterMetric COUNTER_TXN_REJECT;
     public static LongCounterMetric COUNTER_TXN_BEGIN;
     public static LongCounterMetric COUNTER_TXN_FAILED;
     public static LongCounterMetric COUNTER_TXN_SUCCESS;
@@ -208,9 +209,10 @@ public Long getValue() {
                 "counter of image succeeded in pushing to other frontends");
         PALO_METRIC_REGISTER.addPaloMetrics(COUNTER_IMAGE_PUSH);
 
+        COUNTER_TXN_REJECT = new LongCounterMetric("txn_reject", "counter of rejected transactions");
+        PALO_METRIC_REGISTER.addPaloMetrics(COUNTER_TXN_REJECT);
         COUNTER_TXN_BEGIN = new LongCounterMetric("txn_begin", "counter of begining transactions");
         PALO_METRIC_REGISTER.addPaloMetrics(COUNTER_TXN_BEGIN);
-        COUNTER_TXN_FAILED = new LongCounterMetric("txn_failed", "counter of failed transactions");
         COUNTER_TXN_SUCCESS = new LongCounterMetric("txn_success", "counter of success transactions");
         PALO_METRIC_REGISTER.addPaloMetrics(COUNTER_TXN_SUCCESS);
         COUNTER_TXN_FAILED = new LongCounterMetric("txn_failed", "counter of failed transactions");
diff --git a/fe/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java b/fe/src/main/java/org/apache/doris/transaction/GlobalTransactionMgr.java
@@ -35,8 +35,6 @@
 import org.apache.doris.common.LabelAlreadyUsedException;
 import org.apache.doris.common.MetaNotFoundException;
 import org.apache.doris.common.UserException;
-import org.apache.doris.common.util.LogBuilder;
-import org.apache.doris.common.util.LogKey;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.common.util.Util;
 import org.apache.doris.load.Load;
@@ -168,10 +166,15 @@ public long beginTransaction(long dbId, String label, long timestamp,
             unprotectUpsertTransactionState(transactionState);
 
             if (MetricRepo.isInit.get()) {
-                MetricRepo.COUNTER_TXN_SUCCESS.increase(1L);
+                MetricRepo.COUNTER_TXN_BEGIN.increase(1L);
             }
 
             return tid;
+        } catch (Exception e) {
+            if (MetricRepo.isInit.get()) {
+                MetricRepo.COUNTER_TXN_REJECT.increase(1L);
+            }
+            throw e;
         } finally {
             writeUnlock();
         }
