add metrics to mysql cdc connector

Author: baisui1981

tis-incr/tis-flink-cdc-mysql-plugin/pom.xml

@@ -58,7 +58,7 @@
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-mysql-cdc</artifactId>
+            <artifactId>flink-connector-mysql-cdc-tis</artifactId>
             <version>${flink.cdc.version}</version>
             <exclusions>
                 <exclusion>

tis-incr/tis-flink-extends/pom.xml

@@ -123,6 +123,12 @@
             <groupId>com.qlangtech.tis</groupId>
             <artifactId>tis-base-test</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>com.qlangtech.tis</groupId>
+            <artifactId>tis-hadoop-rpc</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <!--        <dependency>-->
         <!--            <groupId>com.alibaba.datax</groupId>-->
         <!--            <artifactId>datax-config</artifactId>-->

tis-incr/tis-flink-extends/src/main/java/com/qlangtech/plugins/incr/flink/metrics/TISPBReporter.java

@@ -18,14 +18,21 @@
 
 package com.qlangtech.plugins.incr.flink.metrics;
 
+import com.qlangtech.tis.cloud.ITISCoordinator;
+import com.qlangtech.tis.realtime.yarn.rpc.MasterJob;
+import com.qlangtech.tis.realtime.yarn.rpc.UpdateCounterMap;
+import com.tis.hadoop.rpc.RpcServiceReference;
+import com.tis.hadoop.rpc.StatusRpcClientFactory;
 import org.apache.flink.metrics.CharacterFilter;
 import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Metric;
 import org.apache.flink.metrics.MetricConfig;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.metrics.reporter.AbstractReporter;
 import org.apache.flink.metrics.reporter.Scheduled;
 
 import java.util.Map;
+import java.util.Optional;
 
 /**
  * https://github.com/datavane/tis/issues/397
@@ -39,9 +46,30 @@ public String filterCharacters(String s) {
         return CharacterFilter.NO_OP_FILTER.filterCharacters(s);
     }
 
+    private RpcServiceReference rpcService;
+
     @Override
     public void open(MetricConfig metricConfig) {
 
+        try {
+            ITISCoordinator coordinator = ITISCoordinator.create(true, Optional.empty());
+            this.rpcService = StatusRpcClientFactory.getService(coordinator);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
+//<jobmanager>
+//  <job.JobName.jobID>
+//    <task.TaskName>
+//      <operator.OperatorName>
+//        <subtask_index>
+
+        String[] scope = group.getScopeComponents();
+// 输出示例：["jobmanager", "job_MyJob", "task_Source", "operator_Map", "0"]
+        super.notifyOfAddedMetric(metric, metricName, group);
     }
 
     @Override
@@ -51,7 +79,36 @@ public void close() {
 
     @Override
     public void report() {
+        UpdateCounterMap upateCounter = null;
+
+        UpdateCounterMap updateCounterMap = new UpdateCounterMap();
+//        updateCounterMap.setGcCounter(BasicONSListener.getGarbageCollectionCount());
+//        updateCounterMap.setFrom(hostName);
+//        long currentTimeInSec = ConsumeDataKeeper.getCurrentTimeInSec();
+//        updateCounterMap.setUpdateTime(currentTimeInSec);
+        // 汇总一个节点中所有索引的增量信息
+//        for (IOnsListenerStatus l : incrChannels) {
+//            TableSingleDataIndexStatus tableUpdateCounter = new TableSingleDataIndexStatus();
+//            tableUpdateCounter.setBufferQueueRemainingCapacity(l.getBufferQueueRemainingCapacity());
+//            tableUpdateCounter.setBufferQueueUsedSize(l.getBufferQueueUsedSize());
+//            tableUpdateCounter.setConsumeErrorCount((int) l.getConsumeErrorCount());
+//            tableUpdateCounter.setIgnoreRowsCount((int) l.getIgnoreRowsCount());
+//            tableUpdateCounter.setUUID(this.indexUUID.get(l.getCollectionName()));
+//            tableUpdateCounter.setTis30sAvgRT(((BasicONSListener) l).getTis30sAvgRT());
+//            // 汇总一个索引中所有focus table的增量信息
+//            for (Map.Entry<String, IIncreaseCounter> entry : l.getUpdateStatic()) {
+//                // IncrCounter tableIncrCounter = new
+//                // IncrCounter((int)entry.getValue().getIncreasePastLast());
+//                // tableIncrCounter.setAccumulationCount(entry.getValue().getAccumulation());
+//                // tableUpdateCounter.put(entry.getKey(), tableIncrCounter);
+//                // 只记录一个消费总量和当前时间
+//                tableUpdateCounter.put(entry.getKey(), entry.getValue().getAccumulation());
+//            }
+//            tableUpdateCounter.put(TABLE_CONSUME_COUNT, ((BasicONSListener) l).getTableConsumeCount());
+//            updateCounterMap.addTableCounter(l.getCollectionName(), tableUpdateCounter);
+//        }
 
+       // MasterJob masterJob = this.rpcService.reportStatus(upateCounter);
         for (Map.Entry<Counter, String> entry : counters.entrySet()) {
             Counter counter = entry.getKey();
             String metricName = entry.getValue();

tis-incr/tis-flink-extends/src/main/java/com/qlangtech/tis/realtime/SourceProcessFunction.java

@@ -18,6 +18,7 @@
 
 package com.qlangtech.tis.realtime;
 
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.ProcessFunction;
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.OutputTag;
@@ -36,7 +37,13 @@ public abstract class SourceProcessFunction<RECORD_TYPE> extends ProcessFunction
     public SourceProcessFunction(Map<String, OutputTag<RECORD_TYPE>> tab2OutputTag) {
         this.tab2OutputTag = tab2OutputTag;
     }
-
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        // 注册或获取名为 "filteredRecords" 的计数器
+//        this.filteredRecordsCounter = getRuntimeContext()
+//                .getMetricGroup()
+//                .counter(KEY_SKIP_UPDATE_BEFORE_EVENT + "Count");
+    }
     /**
      * 在主流中为每个表打标签
      *

tis-incr/tis-incr-test/pom.xml

@@ -44,7 +44,7 @@
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-mysql-cdc</artifactId>
+            <artifactId>flink-connector-mysql-cdc-tis</artifactId>
             <version>${flink.cdc.version}</version>
         </dependency>