Status

Current state: [Under Discussion]

Discussion threads: 

• FLIP-75 discussion about the initial design
• FLIP-102 discussion after splitting up FLIP-75 into sub-flips

JIRA:   

Released: <Flink Version>

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

FLIP-49 has been accepted and merged in Flink 1.10, the metric in current task manager detail page could not correspond well to the design of FLIP-49

Proposed Changes

According to FLIP-49, we can sort out a table containing the correspondence between configuration and metric.

¹ These are the configuration parameters used in the Flink configuration.
² Additionally, memoryConfiguration.totalFlinkMemory  and totalProcessMemory are exposed through the REST API.
³ The metrics which are exposed through the metrics endpoint.

Frontend Design (out-dated)

Redesign the task manager metric page, this would allow users to more clearly understand the relationship between these metrics.

REST API Design

• task manager's resource contains this information, show it in 

• url: /taskmanagers/:taskmanagerid

  Code Block
  language js title JSON Schema of response collapse true
  { "type" : "object",   "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:taskmanager:TaskManagerDetailsInfo",   "properties" : {     "id" : {       "type" : "any"     },     "path" : {       "type" : "string"     },     "dataPort" : {       "type" : "integer"     },     "timeSinceLastHeartbeat" : {       "type" : "integer"     },     "slotsNumber" : {       "type" : "integer"     },     "freeSlots" : {       "type" : "integer"     },     "hardware" : {       "type" : "object",       "id" : "urn:jsonschema:org:apache:flink:runtime:instance:HardwareDescription",       "properties" : {         "cpuCores" : {           "type" : "integer"         },         "physicalMemory" : {           "type" : "integer"         },         "freeMemory" : {           "type" : "integer"         },         "managedMemory" : {           "type" : "integer"         }       }     }, "memoryConfiguration" : {       "type" : "object",       "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:taskmanager:TaskExecutorMemoryConfiguration",       "properties" : {         "frameworkHeap" : {           "type" : "long"         },         "frameworkOffHeap" : {           "type" : "long"         },         "taskHeap" : {           "type" : "long"         },         "taskOffHeap" : {           "type" : "long"         },         "networkMemory" : {           "type" : "long"         },         "managedMemory" : {           "type" : "long"         },         "jvmMetaspace" : {           "type" : "long"         },         "jvmOverhead" : {           "type" : "long"         },         "totalFlinkMemory" : {           "type" : "long"         }         "totalProcessMemory" : {           "type" : "long"         }       }     }, "metrics" : {       "type" : "object",       "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:taskmanager:TaskManagerMetricsInfo",       "properties" : {         "heapUsed" : {           "type" : "integer"         },         "heapCommitted" : {           "type" : "integer"         },         "heapMax" : {           "type" : "integer"         }, "metaspaceUsed" : {           "type" : "integer"         }, "metaspaceCommitted" : {           "type" : "integer"         }, "metaspaceMax" : {           "type" : "integer"         },       "nonHeapUsed" : {           "type" : "integer"         },         "nonHeapCommitted" : {           "type" : "integer"         },         "nonHeapMax" : {           "type" : "integer"         },         "directCount" : {           "type" : "integer"         },         "directUsed" : {           "type" : "integer"         },         "directMax" : {           "type" : "integer"         },         "mappedCount" : {           "type" : "integer"         },         "mappedUsed" : {           "type" : "integer"         },         "mappedMax" : {           "type" : "integer"         },         "memorySegmentsAvailable" : {           "type" : "integer"         },         "memorySegmentsTotal" : {           "type" : "integer"         },         "managedMemoryUsed" : {           "type" : "long"         },         "managedMemoryTotal" : {           "type" : "long"         },         "networkMemoryUsed" : {           "type" : "long"         },         "networkMemoryTotal" : {           "type" : "long"         },         "garbageCollectors" : {           "type" : "array",           "items" : {             "type" : "object",             "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:taskmanager:TaskManagerMetricsInfo:GarbageCollectorInfo",             "properties" : {               "name" : {                 "type" : "string"               },               "count" : {                 "type" : "integer"               },               "time" : {                 "type" : "integer"               }             }           }         }       }     }   } }

  
  

Implementation Proposal

Step 1: Expose configuration parameters of TaskExecutor

1. TaskManagerResourceInfo is introduced as a POJO containing the relevant values proposed in the REST response.
2. The TaskManagerResourceInfo is initialized when initializing the TaskExecutor in the same way as we do it with the HardwareDescription. It will be handed over in the same way through TaskExecutorRegistry → WorkerRegistration.
3. The TaskManagerResourceInfo will be added along with HardwareDescription in ResourceManager::requestTaskManagerInfo(ResourceId, Time).

Step 2: Introduce new metric for memory usage of NetworkBufferPool

1. add shuffle memory's size metric

   Code Block
   language java
   public long getTotalMemorySize() { return 1L * getTotalNumberOfMemorySegments() * memorySegmentSize; } public long getUsedMemorySize() { return getTotalMemorySize() - (getNumberOfAvailableMemorySegments() * memorySegmentSize); }

   
   

2. update NettyShuffleMetricFactory#registerShuffleMetrics

   Code Block
   language java
   private static final String METRIC_TOTAL_MEMORY_IN_BYTES = "TotalMemory"; private static final String METRIC_USED_MEMORY_IN_BYTES = "UsedMemory"; // ... private static void registerShuffleMetrics( String groupName, MetricGroup metricGroup, NetworkBufferPool networkBufferPool) { MetricGroup networkGroup = metricGroup.addGroup(groupName); networkGroup.<Integer, Gauge<Integer>>gauge(METRIC_TOTAL_MEMORY_SEGMENT, networkBufferPool::getTotalNumberOfMemorySegments); networkGroup.<Integer, Gauge<Integer>>gauge(METRIC_AVAILABLE_MEMORY_SEGMENT, networkBufferPool::getNumberOfAvailableMemorySegments); // === new === networkGroup.<Long, Gauge<Long>>gauge(METRIC_TOTAL_MEMORY_IN_BYTES, networkBufferPool::getTotalMemorySize); networkGroup.<Long, Gauge<Long>>gauge(METRIC_USED_MEMORY_IN_BYTES, networkBufferPool::getUsedMemorySize); }

   
   

Step 3: Introduce new metrics for Task's managed memory usage

We still have to discuss how to implement that in the right way. A brief proposal is the following one:

We would have to introduce a new metric that represents the aggregated memory usage of each TaskSlot . The aggregation can be maintained in the TaskExecutor.

1. add managed memory metric and expose it in rest API, as now managed memory is allocated in the slot, we need to sum it in TaskExecutor.

• add default memory type in MemoryManager

  Code Block
  language java
  public static final MemoryType DEFAULT_MEMORY_TYPE = MemoryType.OFF_HEAP;

  
  

• add register ManagedMemoryUsage in TaskExecutor#requestSlot:

  Code Block
  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

  Code Block
  language java
  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
• taskmanager’metrics add Status.Managed.Memory.TotalCapacity and Status.Managed.Memory.MemoryUsed
• url: /taskmanagers/:taskmanagerid
• response: metric add managedMemoryTotal and managedMemoryUsed

Step 4: Expose new metrics to REST interface

TaskManagerDetailsHandler#handleRequest  can be extended to cover also the newly added metrics:

1. TaskManagerMetricsInfo  needs to be extended adding members for the newly added fields:

2. TaskManagerDetailsHandler::createTaskManagerMetricsInfo(MetricsStore.TaskManagerMetricStore) can be used to initialize the fields.
   

Follow-Ups

• The /metrics endpoint of each TaskManager could be split up into multiple endpoints:
  • /configuration containing the static values like configuration and hardware description
  • /metrics containing the volatile values like the metrics and slot information

Test Plan

Existing tests are updated to verify feature.