mxm commented on code in PR #778: URL: https://github.com/apache/flink-kubernetes-operator/pull/778#discussion_r1495910099
########## flink-autoscaler/src/main/java/org/apache/flink/autoscaler/tuning/ConfigChanges.java: ########## @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.autoscaler.tuning; + +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.FallbackKey; + +import lombok.Getter; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** Holds the configuration overrides and removals for a Flink Configuration. */ +public class ConfigChanges { + + /** Overrides which will be applied on top of the existing config. */ + @Getter private final Map<String, String> overrides = new HashMap<>(); + + /** Removals which will be removed from the existing config. */ + @Getter private final Set<String> removals = new HashSet<>(); + + private final transient Configuration confObject = new Configuration(); + + public <T> ConfigChanges addOverride(ConfigOption<T> configOption, T value) { + // Slightly awkward but safe way to work with the limited API of the Configuration class. + // We want to avoid using the serialization code of Configuration and use Strings instead. + confObject.set(configOption, value); + overrides.putAll(confObject.toMap()); + confObject.removeKey(configOption.key()); Review Comment: Thanks, that works better! ########## flink-autoscaler/src/main/java/org/apache/flink/autoscaler/tuning/MemoryTuning.java: ########## @@ -80,57 +85,93 @@ public static Configuration tuneTaskManagerHeapMemory( return EMPTY_CONFIG; } - var maxHeapSize = memSpecs.getFlinkMemory().getJvmHeapMemorySize(); - LOG.debug("Current configured heap size: {}", maxHeapSize); + MemorySize specHeapSize = memSpecs.getFlinkMemory().getJvmHeapMemorySize(); + MemorySize specManagedSize = memSpecs.getFlinkMemory().getManaged(); + MemorySize specNetworkSize = memSpecs.getFlinkMemory().getNetwork(); + MemorySize specMetaspaceSize = memSpecs.getJvmMetaspaceSize(); + LOG.info( + "Spec memory - heap: {}, managed: {}, network: {}, meta: {}", + specHeapSize.toHumanReadableString(), + specManagedSize.toHumanReadableString(), + specNetworkSize.toHumanReadableString(), + specMetaspaceSize.toHumanReadableString()); + + MemorySize maxMemoryBySpec = context.getTaskManagerMemory().orElse(MemorySize.ZERO); + if (maxMemoryBySpec.compareTo(MemorySize.ZERO) <= 0) { + LOG.warn("Spec TaskManager memory size could not be determined."); + return EMPTY_CONFIG; + } + MemoryBudget memBudget = new MemoryBudget(maxMemoryBySpec.getBytes()); + // Add these current settings from the budget + memBudget.budget(memSpecs.getFlinkMemory().getFrameworkOffHeap().getBytes()); + memBudget.budget(memSpecs.getFlinkMemory().getTaskOffHeap().getBytes()); + memBudget.budget(memSpecs.getJvmOverheadSize().getBytes()); - MemorySize newHeapSize = determineNewHeapSize(evaluatedMetrics, config, maxHeapSize); - LOG.info("New TM heap memory {}", newHeapSize.toHumanReadableString()); + var globalMetrics = evaluatedMetrics.getGlobalMetrics(); + // The order matters in case the memory usage is higher than the maximum available memory. + // Managed memory comes last because it can grow arbitrary for RocksDB jobs. + MemorySize newHeapSize = + determineNewSize(HEAP_MEMORY_USED, globalMetrics, config, memBudget); + MemorySize newMetaspaceSize = + determineNewSize(METASPACE_MEMORY_USED, globalMetrics, config, memBudget); + MemorySize newNetworkSize = + determineNewSize(NETWORK_MEMORY_USED, globalMetrics, config, memBudget); + MemorySize newManagedSize = + determineNewSize(MANAGED_MEMORY_USED, globalMetrics, config, memBudget); + LOG.info( + "Optimized memory sizes: heap: {} managed: {}, network: {}, meta: {}", + newHeapSize.toHumanReadableString(), + newManagedSize.toHumanReadableString(), + newNetworkSize.toHumanReadableString(), + newMetaspaceSize.toHumanReadableString()); // Diff can be negative (memory shrinks) or positive (memory grows) - final long heapDiffBytes = newHeapSize.getBytes() - maxHeapSize.getBytes(); - - final MemorySize totalMemory = adjustTotalTmMemory(context, heapDiffBytes); + final long heapDiffBytes = newHeapSize.getBytes() - specHeapSize.getBytes(); + final long managedDiffBytes = newManagedSize.getBytes() - specManagedSize.getBytes(); + final long networkDiffBytes = newNetworkSize.getBytes() - specNetworkSize.getBytes(); + final long flinkMemoryDiffBytes = heapDiffBytes + managedDiffBytes + networkDiffBytes; + + // Update total memory according to memory diffs + final MemorySize totalMemory = + new MemorySize(maxMemoryBySpec.getBytes() - memBudget.getRemaining()); if (totalMemory.equals(MemorySize.ZERO)) { return EMPTY_CONFIG; } // Prepare the tuning config for new configuration values - var tuningConfig = new Configuration(); - // Update total memory according to new heap size - // Adjust the total container memory and the JVM heap size accordingly. - tuningConfig.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, totalMemory); - // Framework and Task heap memory configs add up together yield the max heap memory. - // To simplify the calculation, set the framework heap memory to zero. - tuningConfig.set(TaskManagerOptions.FRAMEWORK_HEAP_MEMORY, MemorySize.ZERO); - tuningConfig.set(TaskManagerOptions.TASK_HEAP_MEMORY, newHeapSize); - - // All memory options which can be configured via fractions need to be set to their - // absolute values or, if there is no absolute setting, the fractions need to be - // re-calculated. - MemorySize managedMemory = memSpecs.getFlinkMemory().getManaged(); - if (shouldTransferHeapToManagedMemory(config, heapDiffBytes)) { - // If RocksDB is configured, give back the heap memory as managed memory to RocksDB - MemorySize newManagedMemory = - new MemorySize(managedMemory.getBytes() + Math.abs(heapDiffBytes)); - LOG.info( - "Increasing managed memory size from {} to {}", - managedMemory, - newManagedMemory); - tuningConfig.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, newManagedMemory); - } else { - tuningConfig.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, managedMemory); - } - - tuningConfig.set( + var tuningConfig = new ConfigChanges(); + // Adjust the total container memory + tuningConfig.addOverride(TaskManagerOptions.TOTAL_PROCESS_MEMORY, totalMemory); + // We do not set the framework/task heap memory because those are automatically derived from + // setting the other mandatory memory options for managed memory, network, metaspace and jvm + // overhead. However, we do precise accounting for heap memory above. In contrast to other + // memory pools, there are no fractional variants for heap memory. Setting the absolute heap + // memory options could cause invalid configuration states when users adapt the total amount + // of memory. We also need to take care to remove any user-provided overrides for those. + tuningConfig.addRemoval(TaskManagerOptions.TASK_HEAP_MEMORY); + // Set default to zero because we already account for heap via task heap. + tuningConfig.addOverride(TaskManagerOptions.FRAMEWORK_HEAP_MEMORY, MemorySize.ZERO); + + MemorySize flinkMemorySize = + new MemorySize( + memSpecs.getTotalFlinkMemorySize().getBytes() + flinkMemoryDiffBytes); + + // All memory options which can be configured via fractions need to be re-calculated. + tuningConfig.addOverride( + TaskManagerOptions.MANAGED_MEMORY_FRACTION, + getFraction(newManagedSize, flinkMemorySize)); + tuningConfig.addRemoval(TaskManagerOptions.MANAGED_MEMORY_SIZE); Review Comment: We remove them because if the user has them set, it will corrupt the memory configuration. If memory tuning is enabled, those configurations need to be removed. Once memory tuning is disabled again, the user-set values will be restored automatically. Generally, I've moved away from settings any absolute values. Fractions support the user updating the total memory without things breaking instantly. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
