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: Unable to render Jira issues macro, execution error.

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.

JVM Metrics

These JVM metrics are exposed and can be used through the TaskManager's metrics REST API.

JVM	Metric	Used key	Total key
Heap	Status.JVM.Memory.Heap	Used	Max
Direct	Status.JVM.Memory.Direct	Used	Max
Metaspace	Status.JVM.Memory.Metaspace Unable to render Jira issues macro, execution error.	Used	Max
Mapped	Status.JVM.Memory.Mapped	MemoryUsed	TotalCapacity
NonHeap	Status.JVM.Memory.NonHeap	MemoryUsed	TotalCapacity

Memory Configuration

Flink's memory model (as described in org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec) can be mapped to the following Flink configuration parameters. There are a few that have a correlating Flink metric.

Flink Memory Model	Flink configuration¹	REST API²	Metric³	Used key	Total key
Framework Heap	`taskmanager.memory.framework.heap.size`	memoryConfiguration.frameworkHeap	Status.JVM.Memory.Heap	Used	Max
Task Heap	`taskmanager.memory.task.heap.size`	memoryConfiguration.taskHeap	Status.JVM.Memory.Heap	Used	Max
Framework OffHeap	`taskmanager.memory.framework.off-heap.size`	memoryConfiguration.frameworkOffHeap	-	-	-
Task OffHeap	`taskmanager.memory.task.off-heap.size`	memoryConfiguration.taskOffHeap	-	-	-
Network Memory	`taskmanager.memory.network.min` `taskmanager.memory.network.max`	memoryConfiguration.networkMemory	Status.Shuffle.Netty Unable to render Jira issues macro, execution error.	UsedMemory	TotalMemory
Managed Memory	`taskmanager.memory.managed.size`	memoryConfiguration.managedMemory	Status.ManagedMemory Unable to render Jira issues macro, execution error.	(is up for discussion, still)	(is up for discussion, still)
JVM Metaspace	`taskmanager.memory.jvm-metaspace.size`	memoryConfiguration.jvmMetaspace	Status.JVM.Memory.Metaspace	Used	Max
JVM Overhead	`taskmanager.memory.jvm-overhead.min` `taskmanager.memory.jvm-overhead.max`	memoryConfiguration.jvmOverhead	-	-	-

¹ These are the configuration parameters used in the Flink configuration.
² These are the Json paths to address the properties in the HTTP REST API response. Additionally, memoryConfiguration.totalFlinkMemory and totalProcessMemory are exposed through the REST API.
³ The metrics which are exposed through the TaskManager's metrics REST API.

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

JSON Schema of response

{
  "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

TaskManagerResourceInfo is introduced as a POJO containing the relevant values proposed in the REST response.
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.
The TaskManagerResourceInfo will be added along with HardwareDescription in ResourceManager::requestTaskManagerInfo(ResourceId, Time).

Step 2: Introduce new metric for memory usage of NetworkBufferPool

add shuffle memory's size metric

public long getTotalMemorySize() {
	return 1L * getTotalNumberOfMemorySegments() * memorySegmentSize;
}

public long getUsedMemorySize() {
	return getTotalMemorySize() - (getNumberOfAvailableMemorySegments() * memorySegmentSize);
}

update NettyShuffleMetricFactory#registerShuffleMetrics

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.

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

public static final MemoryType DEFAULT_MEMORY_TYPE = MemoryType.OFF_HEAP;

add register ManagedMemoryUsage in TaskExecutor#requestSlot:

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
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:

TaskManagerMetricsInfo needs to be extended adding members for the newly added fields:
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.

Page tree

FLIP-102: Add More Metrics to TaskManager