[FLINK-14406] Add metric for managed memory

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Done
• Fix Version/s: 1.12.0
• Component/s: Runtime / Metrics, Runtime / Task
• Labels: pull-request-available

Release Note

New metrics are available to monitor managed memory:
`Status.Flink.Memory.Managed.[Used|Total]`

• Assignee: Matthias Pohl
• Reporter: lining
• Created: 16/Oct/19 08:35
• Updated: 22/Oct/20 07:36
• Resolved: 22/Oct/20 07:36

Description

If a user wants to get memory used in time, as there's no manage memory's metrics, it couldn't get it.

Propose

• add default memory type in MemoryManager

 

public static final MemoryType DEFAULT_MEMORY_TYPE = MemoryType.OFF_HEAP;

• add getManagedMemoryTotal in TaskExecutor:

 

public long getManagedMemoryTotal() {
    return this.taskSlotTable.getAllocatedSlots().stream().mapToLong(
        slot -> slot.getMemoryManager().getMemorySizeByType(MemoryManager.DEFAULT_MEMORY_TYPE)
    ).sum();
}

 

• add getManagedMemoryUsed in TaskExecutor:

 

public long getManagedMemoryUsed() {
    return this.taskSlotTable.getAllocatedSlots().stream().mapToLong(
        slot -> slot.getMemoryManager().getMemorySizeByType(MemoryManager.DEFAULT_MEMORY_TYPE) - slot.getMemoryManager().availableMemory(MemoryManager.DEFAULT_MEMORY_TYPE)
    ).sum();
}

 

• add instantiateMemoryManagerMetrics in MetricUtils

 

public static void instantiateMemoryManagerMetrics(MetricGroup statusMetricGroup, TaskExecutor taskExecutor) {
    checkNotNull(statusMetricGroup);
    MetricGroup memoryManagerGroup = statusMetricGroup.addGroup("Managed").addGroup("Memory");
    memoryManagerGroup.<Long, Gauge<Long>>gauge("TotalCapacity", taskExecutor::getManagedMemoryTotal);
    memoryManagerGroup.<Long, Gauge<Long>>gauge("MemoryUsed", taskExecutor::getManagedMemoryUsed);
}

• register it in TaskManagerRunner#startTaskManager

 

 

Issue Links

• links to: GitHub Pull Request #13547
• mentioned in: Page Loading...