Monitor Metrics

Overview

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

Metric

Supporting extract node

Node level metric

Table level metric

It is used for all database sync.

supporting load node

Node level metric

Table level metric

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.labels' = 'groupId=xxgroup&streamId=xxstream&nodeId=xxnode'
);
 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.labels' = 'groupId=pggroup&streamId=pgStream&nodeId=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.