Monitor Metrics

Overview

We add metric computing for node. Sort will compute metric when user just need add with option inlong.metric that includes groupId&streamId&nodeId. Sort will export metric by flink metric group, So user can use metric reporter to get metric data.

Metric

supporting extract node

supporting load node

Usage

One example about sync mysql data to postgresql data. And We will introduce usage of metric.

• use flink sql

 create table `table_groupId_streamId_nodeId1`(
     `id` INT,
    `name` INT,
    `age` STRING,
    PRIMARY KEY(`id`) NOT ENFORCED)
    WITH (
        'connector' = 'mysql-cdc-inlong',
        'hostname' = 'xxxx',
        'username' = 'xxx',
        'password' = 'xxx',
        'database-name' = 'test',
        'scan.incremental.snapshot.enabled' = 'true',
        'server-time-zone' = 'GMT+8',
        'table-name' = 'user',
        'inlong.metric' = 'mysqlGroup&mysqlStream&mysqlNode1'
);
 CREATE TABLE `table_groupId_streamId_nodeId2`(
     PRIMARY KEY (`id`) NOT ENFORCED,
     `id` INT,
     `name` STRING,
     `age` INT)
     WITH (
         'connector' = 'jdbc-inlong',
         'url' = 'jdbc:postgresql://ip:5432/postgres',
         'username' = 'postgres',
         'password' = 'inlong',
         'table-name' = 'public.user',
         'inlong.metric' = 'pggroup&pgStream&pgNode'
         );
 INSERT INTO `table_groupId_streamId_nodeId2`
 SELECT
     `id`,
     `name`,
     `age`
 FROM `table_groupId_streamId_nodeId1`;

• We can add metric report in flink-conf.yaml

metric.reporters: promgateway
metrics.reporter.promgateway.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter
metrics.reporter.promgateway.host: ip
metrics.reporter.promgateway.port: 9091
metrics.reporter.promgateway.interval: 60 SECONDS

ip and port is your pushgateway setting.

• We can visit http://ip:port of pushgateway after execute flink sql. Metric name will add prefix flink_taskmanager_job_task_operator when metric report is org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter.
  We can see full metric name:
  flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numRecordsIn,
  flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numBytesIn,
  flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numRecordsInPerSecond,
  flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numBytesInPerSecond,
  flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numRecordsOut,
  flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numBytesOut,
  flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numRecordsOutPerSecond,
  flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numBytesOutPerSecond.