Discussion thread	https://lists.apache.org/thread/h2lgqgrvcptzj3c8q4cjzv5jopgtwx9o
Vote thread	TBDhttps://lists.apache.org/thread/09lfq4j7w23h6o1lmywl6k9691sqykhm
ISSUE	https://github.com/apache/incubator-paimon/issues/742
Release	TBD0.6

Motivation

Currently Paimon runs without monitoring and metrics out of box. In the production environment, users need to know how their Paimon Table behaves like what is the commit duration, how many files each commit added or deleted, the status of compaction operation, the duration of scan query, lag of streaming reading and so on. So we need to introduce a metrics system and more metrics for Paimon.

...

Class Metrics is the core of metrics system, there are MetricRegistryMetricGroup and MetricsReporter container in it. When the Metrics instance is initiating, the MetricRegistry MetricGroup is instantiated and metrics reporters are started.

...

Code Block

language	java

public class Metrics {

    /** The registrymetrics that holds the metricsreporters container. */
  	private final MetricRegistry registry;

    List<MetricsReporter> reporters = new ArrayList<>();

	/** The metrics reporters containergroup. */  
  	private final List<MetricsReporter> reporters; List<MetricGroup> metricGroups = new ArrayList<>();   

	/** Add Registera metrics to MetricRegistryreporter. */
 		@parampublic namevoid The name of metric.
		@param metric The metric to register.	
	addReporter(MetricsReporter reporter) {};

	/** Add a metric group. */
	public void registerMetrics(String name, Metric metricaddGroup(MetricGroup group) {};

	/** Get metrics reporters. */
	public List<MetricsReporter> getReporters() {}
}

MetricRegistry

;

	/** Get metric groups. */
	public List<MetricGroup> getMetricGroups() {};
}

MetricGroup

MetricGroup is a class responsible for metrics registering and taggingMetricRegistry is a class responsible for metrics registering, there is a metrics container in it. It provides register method for each type of measurable metric, registering metrics will put metrics to the metrics container.

Code Block

language	java

public classinterface MetricRegistryMetricGroup {

	   /** Map ofRegister gauge metricsmetric. */
 	public void  private final Map<String, Gauge<?>> gauges = new HashMap<>();

    gauge(String name, Gauge gauge) {}

	/** MapRegister of counter metricsmetric. */
	public void   private final Map<Stringcounter(String name, Counter>Counter counters = new HashMap<>();counter) {}

	/** Register gaugehistogram metric. */
 	public void gaugehistogram(String name, Gauge gauge) {}

	/** Register counter metric. */
	public void counter(String name, Counter counterHistogram counter) {}
}

MetricsReporter

MetricsReporter is used to report metrics to external backend, Paimon will implement an out of box reporter as JMX `MetricsReporter`.

Code Block

language	java

public interface MetricsReporter {
	/** Configure reporter after instantiating it.*/
     void open();

    /** Closes this reporter. */
    void close();

	/** Report the current measurements. This method is called periodically by the Metrics. */
	void report();
}

Proposed Changes

Architecture

Paimon support to add tags for metric groups, tags is likely to scopes in Flink's metric group. The tagged metrics group will report metric value with tags as a part of prefix of the metric name.

Table partition and bucket can be added as tag for some metric groups. Like the number of files monitoring in committing / scan / compaction, users can monitor the files of different buckets to check the data skew, and by checking the compaction duration of different buckets to see which unexpected bucket caused the long compaction duration.

MetricsReporter

MetricsReporter is used to report metrics to external backend, Paimon will implement an out of box reporter as JMX MetricsReporter .

Code Block

language	java

public interface MetricsReporter {
	/** Configure reporter after instantiating it.*/
     void open();

    /** Closes this reporter. */
    void close();

	/** Report the current measurements. This method is called periodically by the Metrics. */
	void report();
}

Proposed Changes

Architecture

draw.io Diagram

diagramName
border	true

draw.io Diagram

border	true
diagramName	MetricsSystem
simpleViewer	false
width
links	auto
tbstyle	top
lbox	true
diagramWidth	1141
revision	46

Metrics Registering

Take CommitMetrics as example, the CommitMetrics will be instantiated by FileStoreCommitImpl, then commit related metrics will be registered by MetricRegistry the corresponding MetricGroup in singleton the Metrics instance.

The Metrics has instance of MetricRegistry MetricGroups and MetricsReporters set. MetricRegistry MetricGroup maintains metrics map containers. Metrics registering is a process of putting metrics instances into the metric (gauge, counter) map container.

...

We introduce CommitMetrics , CompactionMetrics , ScanMetrics as metrics set to measure the stats of Paimon table committing, compaction and scanning.

Common metrics

CommitMetrics

, ScanMetrics as metrics set to measure the stats of Paimon table committing, compaction and scanning.

Common metrics

CommitMetrics

Code Block

language	java

public class CommitMetrics {
	private final String LAST_COMMIT_DURATION_METRIC = "lastCommitDuration";
	...

    private final Map<Integer, BucketMetricGroup> bucketMetricGroups = new HashMap<>();
    private final Map<BinaryRow, PartitionMetricGroup> partitionMetricGroups = new HashMap<>();
    private final MetricGroup genericMetricGroup;

	private void registerGenericCommitMetrics() {   
		group.gauge(LAST_COMMIT_DURATION_METRIC, new CommitDurationGauge());
		...
	}

 	private void registerTaggedCommitMetrics() {
		if(isNewBucket()) {
			registerTaggedMetrics();
		}

	 	if(isNewBucket()) {
			registerTaggedMetrics();
		}
	}

Code Block

language	java

public class CommitMetrics {
	private Metrics metrics;
	private final String COMMIT_DURATION_METRIC = "lastCommitDuration";
	...
	private void registerCommitMetrics(Metrics metrics) {
		metrics.gauge(COMMIT_DURATION_METRIC, new CommitDurationTimer());
		...
	}
	...
}

CommitMetrics list includes commit duration, counter of files / records etc.

Metric Name	Description	Type	Tagged	Unit	Update at
lastCommitDuration	The time it took to complete the last commit.	Gauge	none	Ms	Timer starts before commit starting, update commit duration after the last commit finished.
commitDuration	Distributions of the time taken by the last few commits.	Histogram	none	Ms	Timer starts before commit starting, update commit duration after each commit finished.
lastCommitAttempts	The number of attempts the last commit made.	Gauge	Counternone	Number	Increment by 1 when trying to commit once, clear the counter after the last commit finished.
numTableFilesAdded	Number of added table files in last commit, including newly created data files and compacted after.	Gauge	Number	Collecting changes from committables
numTableFilesDeleted	Number of deleted table files in last commit, which comes from compacted before.	Gauge	Number	Collecting changes from committables
.
lastTableFilesAddednumTableFilesAppended	Number of appended added table files in last commit, which means the including newly created data files and compacted after.	Gauge	none	Number	Collecting changes from committables
numTableFilesCompatedlastTableFilesDeleted	Number of compacted deleted table files in last commit, including which comes from compacted before and after.	Gauge	none	Number	Collecting changes from committables
numChangelogFilesAppendedlastTableFilesAppended	Number of appended changelog table files in last commit	Gauge	Number	Collecting changes from committables	numChangelogFileCompacted	Number of compacted changelog files in last commit	, which means the newly created data files.	Gauge	noneGauge	Number	Collecting changes from committables
totalTablesFileslastTableFilesCommitCompated	Number of total data files currently maintained on storage.compacted table files in last commit, including compacted before and after.	Gauge	noneCounter	Number	Collecting changes from committables
totalChangelogFileslastChangelogFilesAppended	Number of total appended changelog files currently maintained on storage.in last commit	Gauge	noneCounter	Number	Collecting changes from committables
numGeneratedSnapshotslastChangelogFileCommitCompacted	Number of snapshot compacted changelog files generated in last commit, maybe 1 snapshot or 2 snapshots.	Gauge	none	Number	After collecting Collecting changes from committables
totalSnapshotslastGeneratedSnapshots	Number of currently retained total snapshots.	Counter	Number	When trying to commit, the counter will increment by the number of snapshots generated When expiring snapshots, the counter will decrement by the number of expiring snapshots.
numTotalRecordsAppended	Total records count in last commit with APPEND commit kind	Gauge	Number	Preparing snapshot file with APPEND commit kind
snapshot files generated in last commit, maybe 1 snapshot or 2 snapshots.	Gauge	none	Number	After collecting changes from committables
lastDeltaRecordsAppendednumDeltaRecordsAppended	Delta records count in last commit with APPEND commit kind	Gauge	none	Number	Preparing snapshot file with APPEND commit kind
numChangelogRecordsAppendedlastChangelogRecordsAppended	Changelog records count in last commit with APPEND commit kind	Gauge	none	Number	Preparing snapshot file with APPEND commit kind
numTotalRecordsCompatedlastDeltaRecordsCommitCompated	Total Delta records count in last commit with COMPACT commit kind	Gauge	Number	Preparing snapshot file with COMPACT commit kind	numDeltaRecordsCompated	Delta records count in last commit with COMPACT commit kind	Gauge	none	Number	Preparing snapshot file with COMPACT commit kind
numChangelogRecordsCompatedlastChangelogRecordsCommitCompated	Changelog records count in last commit with COMPACT commit kind	Gauge	none	Number	Preparing snapshot file with COMPACT commit kind
numPartitionsWrittenlastPartitionsWritten	Number of partitions written in last commit	Gauge	none	Number	After collecting changes from committables
numBucketsWrittenlastBucketsWritten	Number of buckets written in last commit	Gauge	none	Number	After collecting changes from committables

...

Code Block

public class ScanMetrics {
	private Metrics metrics;
	private final String SCAN_FILES_METRIC = "lastScanDuration";
	...
 final String LAST_SCAN_SKIPPED_MANIFESTS_METRIC = "lastScanSkippedManifests";
	...
 	
	private final Map<Integer, BucketMetricGroup> bucketMetricGroups = new HashMap<>();
    private final Map<BinaryRow, PartitionMetricGroup> partitionMetricGroups = new HashMap<>();

 	private void registerScanMetricsregisterTaggedScanMetrics(Metrics metrics) {
		if(isNewBucket()) {
			metrics.counter(SCAN_FILES_METRIC, new ScanFilesCounterregisterTaggedMetrics();
		}

	 	if(isNewPartition()); {
			...
registerTaggedMetrics();
		}
	...}       
}

ScanMetrics list includes duration, data files and manifest files counter.

Metric Name	Description	Type	Tagged	Unit	Update at
lastScanDuration	The time it took to complete the last scan planning.	Gauge	none	Ms	Timer starts before the scan planning starts, update after the last scan planning operation finished
scanDuration	Distributions of the time taken by the last few scan planning operations.	Histogram	none	Ms	Timer starts before the scan planning starts, update after each planning finished
numScannedManifestslastScannedManifests	Number of scanned manifests files in the last scan planning.	Gauge	none	Number	Scan planning
numSkippedManifestslastScanSkippedManifests	Number of skipped manifests files in the last scan planning.	Gauge	none	Number	Scan planning
numResultTableFileslastScanResultTableFiles	Number of result table files in the last scan planning.	Gauge	Number	Scan planning	numGenerateSplits	Number of splits generated by the last scan planning.	Gaugenone	Number	Scan planning

CompactionMetrics

Code Block

public class CompactionMetrics {

	private Metrics metrics;
	 final String LAST_TABLE_FILES_COMPACTED_BEFORE_METRIC = "lastTableFilesCompactedBefore";
 	...
 	
	private final Map<Integer, BucketMetricGroup> bucketMetricGroups = new HashMap<>();
    private final String COMPACTED_FILES_METRICMap<BinaryRow, PartitionMetricGroup> partitionMetricGroups = "lastCompactionDuration" new HashMap<>();
	...
 	private void registerCompactionMetricsregisterTaggedCompactionMetrics(Metrics metrics) {
		if(isNewBucket()) {
			metrics.counter(COMPACTED_FILES_METRIC, new CompactedFilesCounterregisterTaggedMetrics();
		}

	 	if(isNewPartition()); {
			...
registerTaggedMetrics();
		}
	...}
}

CompactionMetrics list includes duration, and counter of files, sorted runs etc.

Metric Name	Description	Type	Tagged	Unit	Update at
lastCompactionDuration	The time it took to complete the last compaction.	Gauge	none	Ms	Timer starts before compaction, update after the last compaction finished
compactionDuration	Distributions of the time taken by the last few compactions.	Histogram	none	Ms	Timer starts before compaction, update after each compaction finished
numTableFilesCompactedBeforelastTableFilesCompactedBefore	Number of deleted files in last compaction	Gauge	none	Number	After getting compaction result.
numTableFilesCompactedAfterlastTableFilesCompactedAfter	Number of added files in last compaction	Gauge	none	Number	After getting compaction result.
numChangelogFilesCompactedlastChangelogFilesCompacted	Number of changelog files compacted in last compaction	Gauge	none	Number	After getting compaction result.
numSortedRunslastCompactOutputLevel	Current number Number of total sorted runs of all levels at last compaction	Gauge	none	Number	Updating levels after getting compaction result.
numLevel0FileslastCompactFileSize	Current number of files at level 0	Sum of file size compacted in last compaction	Gauge	noneGauge	Number	Updating levels after getting compaction result.

...

Page tree

Versions Compared

Old Version 20

New Version Current

Key

Motivation

MetricRegistry

MetricGroup

MetricsReporter

Proposed Changes

Architecture

MetricsReporter

Proposed Changes

Architecture

Metrics Registering

Common metrics

CommitMetrics

Common metrics

CommitMetrics

CompactionMetrics

Page tree

Page History

Versions Compared

Old Version 20

New Version Current

Key

Motivation

MetricRegistry

MetricGroup

MetricsReporter

Proposed Changes

Architecture

MetricsReporter

Proposed Changes

Architecture

Metrics Registering

Common metrics

CommitMetrics

Common metrics

CommitMetrics

CompactionMetrics