xintongsong commented on a change in pull request #8952: URL: https://github.com/apache/flink/pull/8952#discussion_r548814271
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/execution/ThresholdExceedException.java ########## @@ -0,0 +1,31 @@ +/* + * 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.runtime.execution; + +/** + * Exception thrown when a threshold exceeds. + */ +public class ThresholdExceedException extends RuntimeException { Review comment: Why is this class in the path `runtime/execution`? I think we should make it an inner class of `ThresholdMeter`, given that the two classes are closely related. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/execution/ThresholdExceedException.java ########## @@ -0,0 +1,31 @@ +/* + * 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.runtime.execution; + +/** + * Exception thrown when a threshold exceeds. + */ +public class ThresholdExceedException extends RuntimeException { + + private static final long serialVersionUID = 241871312848349L; Review comment: `serialVersionUID` should start from `1L`. https://flink.apache.org/contributing/code-style-and-quality-java.html#java-serialization ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { + private static final Double MILLISECONDS_PER_SECOND = 1000.0; + private final Supplier<Long> currentTimeMillisSupplier; + private final double maximumFailureRate; + private final Time failureInterval; + private final Queue<Long> failureTimestamps; + private long failureCounter = 0; + + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval) { + this(maximumFailureRate, failureInterval, System::currentTimeMillis); + } + + @VisibleForTesting + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval, Supplier<Long> customSupplier) { + this.maximumFailureRate = maximumFailureRate; + this.failureInterval = failureInterval; + this.failureTimestamps = new ArrayDeque<>(); + this.currentTimeMillisSupplier = customSupplier; + } + + @Override + public void markEvent() { + failureTimestamps.add(System.currentTimeMillis()); + failureCounter++; + } + + @Override + public void markEvent(long n) { + for (int i = 0; i < n; i++) { + failureTimestamps.add(System.currentTimeMillis()); + } + failureCounter = failureCounter + n; + } + + @Override + public double getRate() { + return getCurrentFailureRate() / (failureInterval.toMilliseconds() / MILLISECONDS_PER_SECOND); + } + + @Override + public long getCount() { + return failureCounter; + } + + @Override + public void checkAgainstThreshold() throws ThresholdExceedException { + if (getCurrentFailureRate() >= maximumFailureRate) { + throw new ThresholdExceedException(String.format("Maximum number of failed workers %f" + + " is detected in Resource Manager", getCurrentFailureRate())); + } + } + + private double getCurrentFailureRate() { + Long currentTimeStamp = System.currentTimeMillis(); Review comment: ```suggestion Long currentTimeStamp = currentTimeMillisSupplier.ge(); ``` ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { + private static final Double MILLISECONDS_PER_SECOND = 1000.0; + private final Supplier<Long> currentTimeMillisSupplier; + private final double maximumFailureRate; + private final Time failureInterval; + private final Queue<Long> failureTimestamps; + private long failureCounter = 0; + + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval) { + this(maximumFailureRate, failureInterval, System::currentTimeMillis); + } + + @VisibleForTesting + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval, Supplier<Long> customSupplier) { + this.maximumFailureRate = maximumFailureRate; + this.failureInterval = failureInterval; + this.failureTimestamps = new ArrayDeque<>(); + this.currentTimeMillisSupplier = customSupplier; + } + + @Override + public void markEvent() { + failureTimestamps.add(System.currentTimeMillis()); + failureCounter++; + } + + @Override + public void markEvent(long n) { + for (int i = 0; i < n; i++) { + failureTimestamps.add(System.currentTimeMillis()); + } + failureCounter = failureCounter + n; + } + + @Override + public double getRate() { + return getCurrentFailureRate() / (failureInterval.toMilliseconds() / MILLISECONDS_PER_SECOND); + } + + @Override + public long getCount() { + return failureCounter; + } + + @Override + public void checkAgainstThreshold() throws ThresholdExceedException { + if (getCurrentFailureRate() >= maximumFailureRate) { + throw new ThresholdExceedException(String.format("Maximum number of failed workers %f" + + " is detected in Resource Manager", getCurrentFailureRate())); + } + } + + private double getCurrentFailureRate() { + Long currentTimeStamp = System.currentTimeMillis(); + while (!failureTimestamps.isEmpty() && + currentTimeStamp - failureTimestamps.peek() > failureInterval.toMilliseconds()) { + failureTimestamps.remove(); + } + + return failureTimestamps.size(); Review comment: What is being returned is the count of events occurred within the interval. It's not a rate. Maybe rename the method to `getFailuresCountsRecentInterval`? ########## File path: flink-core/src/main/java/org/apache/flink/configuration/ResourceManagerOptions.java ########## @@ -67,6 +69,33 @@ "for streaming workloads, which may fail if there are not enough slots. Note that this configuration option does not take " + "effect for standalone clusters, where how many slots are allocated is not controlled by Flink."); + /** + * Defines the maximum number of worker (YARN / Mesos / Kubernetes) failures per minute before rejecting subsequent worker + * requests until the failure rate falls below the maximum. It is to quickly catch external dependency caused + * workers failure and wait for retry interval before sending new request. By default, the value is set to 10/min. + */ + public static final ConfigOption<Double> MAXIMUM_WORKERS_FAILURE_RATE = ConfigOptions + .key("resourcemanager.start-worker.max-failure-rate") + .doubleType() + .defaultValue(10.0) + .withDescription("Defines the maximum number of worker (YARN / Mesos) failures per minute before rejecting" + + " subsequent worker requests until the failure rate falls below the maximum. It is to quickly catch" + + " external dependency caused workers failure and terminate job accordingly." + + " By default, the value is set to 10/min."); + + /** + * Defines the worker creation interval in milliseconds. In case of worker creation failures, we should wait for an interval before + * trying to create new workers when the failure rate exceeds. Otherwise, ActiveResourceManager will always re-requesting + * the worker, which keeps the main thread busy. + */ + public static final ConfigOption<Duration> WORKER_CREATION_RETRY_INTERVAL = ConfigOptions + .key("resourcemanager.start-worker.retry-interval") + .durationType() + .defaultValue(Duration.ofMillis(30)) Review comment: `30ms` might be too small. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; Review comment: I think `failurerate` is a bit too detailed to be a direct child package under `runtime`. Currently, codes in this package are only used by the active resource manager. If `FailureRaterUtil` and `ThresholdMeter` are removed, we don't even need a package. We can put `TimestampBasedFailureRater` directly in package `resourcemanager.active`. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/FailureRaterUtil.java ########## @@ -0,0 +1,37 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.api.common.time.Time; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.ResourceManagerOptions; +import org.apache.flink.util.Preconditions; + +/** + * Failure rate util. + */ +public class FailureRaterUtil { Review comment: It's not necessary to introduce this util class. We can simply add a static factory method in `TimestampBasedFailureRater`. ``` public class TimestampBasedFailureRater { public static TimestampBasedFailureRater fromConfiguration(Configuration config) { ... } } ``` ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/ThresholdMeter.java ########## @@ -0,0 +1,36 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.metrics.Meter; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +/** + * Metric for measuring rate with Threshold. + */ +public interface ThresholdMeter extends Meter { + + /** + * Check whether the current rate of events exceed threshold. + * If yes, throw ThresholdExceedException. + * + * @throws ThresholdExceedException + */ + void checkAgainstThreshold() throws ThresholdExceedException; Review comment: I noticed that this interface is only implemented by `TimestampBasedFailureRater`, and only called by `ActiveResourceManager` and `TimestampBasedFailureRaterTest`. I start to wonder, do we really need this interface? Sorry for the back and forth. I think we made `ActiveResourceManager` only aware of `ThresholdMeter` (or `FailureRater` previously) because we had different implementations of this interface. Now since we no longer have `NoFailureRater`, we probably can get rid of this interface, making `ActiveResourceManager` to use `TimestampBasedFailureRater` directly. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { Review comment: Not sure about the class name `TimestampBasedFailureRater`. I believe this is named with respect to the previous interface `FailureRater` and the other implementation `NoFailureRater`. Actually, I think this implementation class should be named `ThresholdMeter`, which implements `Meter` directly. Imagine replacing all the keyword "failure" with "event", you should find this class becoming a general implementation of `Meter` with `checkAgainstThreshold` being the only additional function. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { + private static final Double MILLISECONDS_PER_SECOND = 1000.0; + private final Supplier<Long> currentTimeMillisSupplier; + private final double maximumFailureRate; + private final Time failureInterval; + private final Queue<Long> failureTimestamps; + private long failureCounter = 0; + + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval) { + this(maximumFailureRate, failureInterval, System::currentTimeMillis); + } + + @VisibleForTesting + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval, Supplier<Long> customSupplier) { + this.maximumFailureRate = maximumFailureRate; + this.failureInterval = failureInterval; + this.failureTimestamps = new ArrayDeque<>(); + this.currentTimeMillisSupplier = customSupplier; + } + + @Override + public void markEvent() { + failureTimestamps.add(System.currentTimeMillis()); Review comment: ```suggestion failureTimestamps.add(currentTimeMillisSupplier.get()); ``` ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { + private static final Double MILLISECONDS_PER_SECOND = 1000.0; + private final Supplier<Long> currentTimeMillisSupplier; + private final double maximumFailureRate; + private final Time failureInterval; + private final Queue<Long> failureTimestamps; + private long failureCounter = 0; + + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval) { + this(maximumFailureRate, failureInterval, System::currentTimeMillis); + } + + @VisibleForTesting + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval, Supplier<Long> customSupplier) { + this.maximumFailureRate = maximumFailureRate; + this.failureInterval = failureInterval; + this.failureTimestamps = new ArrayDeque<>(); + this.currentTimeMillisSupplier = customSupplier; + } + + @Override + public void markEvent() { + failureTimestamps.add(System.currentTimeMillis()); + failureCounter++; + } + + @Override + public void markEvent(long n) { + for (int i = 0; i < n; i++) { + failureTimestamps.add(System.currentTimeMillis()); Review comment: ```suggestion failureTimestamps.add(currentTimeMillisSupplier.ge()); ``` ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { + private static final Double MILLISECONDS_PER_SECOND = 1000.0; + private final Supplier<Long> currentTimeMillisSupplier; + private final double maximumFailureRate; + private final Time failureInterval; + private final Queue<Long> failureTimestamps; + private long failureCounter = 0; + + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval) { + this(maximumFailureRate, failureInterval, System::currentTimeMillis); + } + + @VisibleForTesting + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval, Supplier<Long> customSupplier) { + this.maximumFailureRate = maximumFailureRate; + this.failureInterval = failureInterval; + this.failureTimestamps = new ArrayDeque<>(); + this.currentTimeMillisSupplier = customSupplier; + } + + @Override + public void markEvent() { + failureTimestamps.add(System.currentTimeMillis()); + failureCounter++; + } + + @Override + public void markEvent(long n) { + for (int i = 0; i < n; i++) { + failureTimestamps.add(System.currentTimeMillis()); + } + failureCounter = failureCounter + n; + } + + @Override + public double getRate() { + return getCurrentFailureRate() / (failureInterval.toMilliseconds() / MILLISECONDS_PER_SECOND); + } + + @Override + public long getCount() { + return failureCounter; + } + + @Override + public void checkAgainstThreshold() throws ThresholdExceedException { + if (getCurrentFailureRate() >= maximumFailureRate) { Review comment: `getCurrentFailureRate` returns count of the event in the recent `failureInterval`, while `maximumFailureRate` is defined as "failures per minute" per the config option description. The correctness of this check strongly depends on `failureInterval` being 1min. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java ########## @@ -67,6 +67,7 @@ private MetricNames() { public static final String CHECKPOINT_ALIGNMENT_TIME = "checkpointAlignmentTime"; public static final String CHECKPOINT_START_DELAY_TIME = "checkpointStartDelayNanos"; + public static final String WORKER_FAILURE_RATE = "workFailureRate"; Review comment: ```suggestion public static final String WORKER_FAILURE_RATE = "startWorkFailure" + SUFFIX_RATE; ``` ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java ########## @@ -108,6 +111,8 @@ public static final String RESOURCE_MANAGER_NAME = "resourcemanager"; + protected final Logger log = LoggerFactory.getLogger(getClass()); Review comment: Unnecessary. The logger is already defined in `RpcEndpoint`, which is a super class of `ResourceManager. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { Review comment: And if we turn this class into a general implementation of `Meter`, we may move it to the package `org.apache.flink.metrics`. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { + private static final Double MILLISECONDS_PER_SECOND = 1000.0; + private final Supplier<Long> currentTimeMillisSupplier; + private final double maximumFailureRate; + private final Time failureInterval; + private final Queue<Long> failureTimestamps; + private long failureCounter = 0; + + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval) { + this(maximumFailureRate, failureInterval, System::currentTimeMillis); + } + + @VisibleForTesting + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval, Supplier<Long> customSupplier) { + this.maximumFailureRate = maximumFailureRate; + this.failureInterval = failureInterval; + this.failureTimestamps = new ArrayDeque<>(); + this.currentTimeMillisSupplier = customSupplier; + } + + @Override + public void markEvent() { + failureTimestamps.add(System.currentTimeMillis()); + failureCounter++; + } + + @Override + public void markEvent(long n) { + for (int i = 0; i < n; i++) { + failureTimestamps.add(System.currentTimeMillis()); Review comment: And we should avoid calling `System.currentTimeMillis()` (or `currentTimeMillisSupplier.ge()`) for `n` times. The `n` events should have the same timestamp. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/failurerate/TimestampBasedFailureRater.java ########## @@ -0,0 +1,93 @@ +/* + * 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.runtime.failurerate; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.runtime.execution.ThresholdExceedException; + +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.function.Supplier; + +/** + * A timestamp queue based failure rater implementation. + */ +public class TimestampBasedFailureRater implements ThresholdMeter { + private static final Double MILLISECONDS_PER_SECOND = 1000.0; + private final Supplier<Long> currentTimeMillisSupplier; + private final double maximumFailureRate; + private final Time failureInterval; + private final Queue<Long> failureTimestamps; + private long failureCounter = 0; + + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval) { + this(maximumFailureRate, failureInterval, System::currentTimeMillis); + } + + @VisibleForTesting + public TimestampBasedFailureRater(double maximumFailureRate, Time failureInterval, Supplier<Long> customSupplier) { + this.maximumFailureRate = maximumFailureRate; + this.failureInterval = failureInterval; + this.failureTimestamps = new ArrayDeque<>(); + this.currentTimeMillisSupplier = customSupplier; + } + + @Override + public void markEvent() { + failureTimestamps.add(System.currentTimeMillis()); + failureCounter++; + } + + @Override + public void markEvent(long n) { + for (int i = 0; i < n; i++) { + failureTimestamps.add(System.currentTimeMillis()); + } + failureCounter = failureCounter + n; + } + + @Override + public double getRate() { + return getCurrentFailureRate() / (failureInterval.toMilliseconds() / MILLISECONDS_PER_SECOND); + } + + @Override + public long getCount() { + return failureCounter; + } + + @Override + public void checkAgainstThreshold() throws ThresholdExceedException { + if (getCurrentFailureRate() >= maximumFailureRate) { Review comment: I see several different *rates* here, and I think we should make them clearly defined. * `getRate()` returns a per-second rate, which is clearly defined in the interface JavaDoc. We can also add a comment to remind people of that. * What is the definition of `maximumFailureRate`? We probably should name it `maxFailuresPerMinute`, `maxFailuresPerSecond` or `maxFailuresPerFailureInterval`. * What is the definition of `getCurrentFailureRate`? We probably should name it `getFailuresCountsRecentInterval`. ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java ########## @@ -67,6 +67,7 @@ private MetricNames() { public static final String CHECKPOINT_ALIGNMENT_TIME = "checkpointAlignmentTime"; public static final String CHECKPOINT_START_DELAY_TIME = "checkpointStartDelayNanos"; + public static final String WORKER_FAILURE_RATE = "workFailureRate"; Review comment: Not sure if per-second is a good granularity for this metric. Maybe it's not a big deal since people can always convert it to other granularity. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: [email protected]
