测试 promethues 数据的准确性

bearlly · Mar 21, 2019 · 24f6390 · 24f6390
1 parent 8e523c7
commit 24f6390
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 3 deletions.
diff --git a/flinkx-core/.gitignore b/flinkx-core/.gitignore
@@ -11,3 +11,4 @@ target
 .classpath
 *.eclipse.*
 *.iml
+/dependency-reduced-pom.xml
diff --git a/flinkx-core/src/main/java/com/dtstack/flinkx/metrics/SimpleAccumulatorGauge.java b/flinkx-core/src/main/java/com/dtstack/flinkx/metrics/SimpleAccumulatorGauge.java
@@ -18,21 +18,22 @@
 
 package com.dtstack.flinkx.metrics;
 
-import org.apache.flink.api.common.accumulators.SimpleAccumulator;
+import org.apache.flink.api.common.accumulators.Accumulator;
 import org.apache.flink.metrics.Gauge;
 
 import java.io.Serializable;
 
 /**
  * company: www.dtstack.com
+ *
  * @author: toutian
  * create: 2019/3/21
  */
 public class SimpleAccumulatorGauge<T extends Serializable> implements Gauge<T> {
 
-    private SimpleAccumulator<T> accumulator;
+    private Accumulator<T, T> accumulator;
 
-    public SimpleAccumulatorGauge(SimpleAccumulator<T> accumulator) {
+    public SimpleAccumulatorGauge(Accumulator<T, T> accumulator) {
         this.accumulator = accumulator;
     }
 

diff --git a/flinkx-rdb/src/main/java/com/dtstack/flinkx/rdb/inputformat/JdbcInputFormat.java b/flinkx-rdb/src/main/java/com/dtstack/flinkx/rdb/inputformat/JdbcInputFormat.java
@@ -21,6 +21,7 @@
 import com.dtstack.flinkx.common.ColumnType;
 import com.dtstack.flinkx.constants.Metrics;
 import com.dtstack.flinkx.enums.EDatabaseType;
+import com.dtstack.flinkx.metrics.SimpleAccumulatorGauge;
 import com.dtstack.flinkx.rdb.DatabaseInterface;
 import com.dtstack.flinkx.rdb.type.TypeConverterInterface;
 import com.dtstack.flinkx.rdb.util.DBUtil;
@@ -36,6 +37,7 @@
 import org.apache.flink.core.io.GenericInputSplit;
 import org.apache.flink.core.io.InputSplit;
 import org.apache.flink.core.io.InputSplitAssigner;
+import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.types.Row;
 import java.io.IOException;
 import java.sql.*;
@@ -120,16 +122,19 @@ public void configure(Configuration configuration) {
 
     private void setMetric(){
         Map<String, Accumulator<?, ?>> accumulatorMap = getRuntimeContext().getAllAccumulators();
+        final MetricGroup flinkxOutputMetricGroup = getRuntimeContext().getMetricGroup().addGroup(Metrics.METRIC_GROUP_KEY_FLINKX, Metrics.METRIC_GROUP_VALUE_OUTPUT);
 
         if(!accumulatorMap.containsKey(Metrics.TABLE_COL)){
             tableColAccumulator = new StringAccumulator();
             tableColAccumulator.add(table + "-" + increCol);
             getRuntimeContext().addAccumulator(Metrics.TABLE_COL,tableColAccumulator);
+            flinkxOutputMetricGroup.gauge(Metrics.TABLE_COL, new SimpleAccumulatorGauge<String>(tableColAccumulator));
         }
 
         if(!accumulatorMap.containsKey(Metrics.END_LOCATION)){
             endLocationAccumulator = new MaximumAccumulator();
             getRuntimeContext().addAccumulator(Metrics.END_LOCATION,endLocationAccumulator);
+            flinkxOutputMetricGroup.gauge(Metrics.END_LOCATION, new SimpleAccumulatorGauge<String>(endLocationAccumulator));
         }
 
         if (startLocation != null){
@@ -138,6 +143,7 @@ private void setMetric(){
                 startLocationAccumulator = new StringAccumulator();
                 startLocationAccumulator.add(startLocation);
                 getRuntimeContext().addAccumulator(Metrics.START_LOCATION,startLocationAccumulator);
+                flinkxOutputMetricGroup.gauge(Metrics.START_LOCATION, new SimpleAccumulatorGauge<String>(startLocationAccumulator));
             }
         }