| /* |
| * 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.hadoop.yarn.server.resourcemanager.rmapp; |
| |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.yarn.api.records.ApplicationId; |
| import org.apache.hadoop.yarn.api.records.LogAggregationStatus; |
| import org.apache.hadoop.yarn.api.records.NodeId; |
| import org.apache.hadoop.yarn.conf.YarnConfiguration; |
| import org.apache.hadoop.yarn.server.api.protocolrecords.LogAggregationReport; |
| |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock; |
| import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock; |
| |
| /** |
| * Log aggregation logic used by RMApp. |
| * |
| */ |
| public class RMAppLogAggregation { |
| private final boolean logAggregationEnabled; |
| private final ReadLock readLock; |
| private final WriteLock writeLock; |
| private long logAggregationStartTime = 0; |
| private final long logAggregationStatusTimeout; |
| private final Map<NodeId, LogAggregationReport> logAggregationStatus = |
| new ConcurrentHashMap<>(); |
| private volatile LogAggregationStatus logAggregationStatusForAppReport; |
| private int logAggregationSucceed = 0; |
| private int logAggregationFailed = 0; |
| private Map<NodeId, List<String>> logAggregationDiagnosticsForNMs = |
| new HashMap<>(); |
| private Map<NodeId, List<String>> logAggregationFailureMessagesForNMs = |
| new HashMap<>(); |
| private final int maxLogAggregationDiagnosticsInMemory; |
| |
| RMAppLogAggregation(Configuration conf, ReadLock readLock, |
| WriteLock writeLock) { |
| this.readLock = readLock; |
| this.writeLock = writeLock; |
| this.logAggregationStatusTimeout = getLogAggregationStatusTimeout(conf); |
| this.logAggregationEnabled = getEnabledFlagFromConf(conf); |
| this.logAggregationStatusForAppReport = |
| this.logAggregationEnabled ? LogAggregationStatus.NOT_START : |
| LogAggregationStatus.DISABLED; |
| this.maxLogAggregationDiagnosticsInMemory = |
| getMaxLogAggregationDiagnostics(conf); |
| } |
| |
| private long getLogAggregationStatusTimeout(Configuration conf) { |
| long statusTimeout = |
| conf.getLong(YarnConfiguration.LOG_AGGREGATION_STATUS_TIME_OUT_MS, |
| YarnConfiguration.DEFAULT_LOG_AGGREGATION_STATUS_TIME_OUT_MS); |
| if (statusTimeout <= 0) { |
| return YarnConfiguration.DEFAULT_LOG_AGGREGATION_STATUS_TIME_OUT_MS; |
| } else { |
| return statusTimeout; |
| } |
| } |
| |
| private boolean getEnabledFlagFromConf(Configuration conf) { |
| return conf.getBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, |
| YarnConfiguration.DEFAULT_LOG_AGGREGATION_ENABLED); |
| } |
| |
| private int getMaxLogAggregationDiagnostics(Configuration conf) { |
| return conf.getInt( |
| YarnConfiguration.RM_MAX_LOG_AGGREGATION_DIAGNOSTICS_IN_MEMORY, |
| YarnConfiguration.DEFAULT_RM_MAX_LOG_AGGREGATION_DIAGNOSTICS_IN_MEMORY); |
| } |
| |
| Map<NodeId, LogAggregationReport> getLogAggregationReportsForApp( |
| RMAppImpl rmApp) { |
| this.readLock.lock(); |
| try { |
| if (!isLogAggregationFinished() && RMAppImpl.isAppInFinalState(rmApp) && |
| rmApp.getSystemClock().getTime() > this.logAggregationStartTime |
| + this.logAggregationStatusTimeout) { |
| for (Map.Entry<NodeId, LogAggregationReport> output : |
| logAggregationStatus.entrySet()) { |
| if (!output.getValue().getLogAggregationStatus() |
| .equals(LogAggregationStatus.TIME_OUT) |
| && !output.getValue().getLogAggregationStatus() |
| .equals(LogAggregationStatus.SUCCEEDED) |
| && !output.getValue().getLogAggregationStatus() |
| .equals(LogAggregationStatus.FAILED)) { |
| output.getValue().setLogAggregationStatus( |
| LogAggregationStatus.TIME_OUT); |
| } |
| } |
| } |
| return Collections.unmodifiableMap(logAggregationStatus); |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| void aggregateLogReport(NodeId nodeId, LogAggregationReport report, |
| RMAppImpl rmApp) { |
| this.writeLock.lock(); |
| try { |
| if (this.logAggregationEnabled && !isLogAggregationFinished()) { |
| LogAggregationReport curReport = this.logAggregationStatus.get(nodeId); |
| boolean stateChangedToFinal = false; |
| if (curReport == null) { |
| this.logAggregationStatus.put(nodeId, report); |
| if (isLogAggregationFinishedForNM(report)) { |
| stateChangedToFinal = true; |
| } |
| } else { |
| if (isLogAggregationFinishedForNM(report)) { |
| if (!isLogAggregationFinishedForNM(curReport)) { |
| stateChangedToFinal = true; |
| } |
| } |
| if (report.getLogAggregationStatus() != LogAggregationStatus.RUNNING |
| || curReport.getLogAggregationStatus() != |
| LogAggregationStatus.RUNNING_WITH_FAILURE) { |
| if (curReport.getLogAggregationStatus() |
| == LogAggregationStatus.TIME_OUT |
| && report.getLogAggregationStatus() |
| == LogAggregationStatus.RUNNING) { |
| // If the log aggregation status got from latest NM heartbeat |
| // is RUNNING, and current log aggregation status is TIME_OUT, |
| // based on whether there are any failure messages for this NM, |
| // we will reset the log aggregation status as RUNNING or |
| // RUNNING_WITH_FAILURE |
| if (isThereFailureMessageForNM(nodeId)) { |
| report.setLogAggregationStatus( |
| LogAggregationStatus.RUNNING_WITH_FAILURE); |
| } |
| } |
| curReport.setLogAggregationStatus(report |
| .getLogAggregationStatus()); |
| } |
| } |
| updateLogAggregationDiagnosticMessages(nodeId, report); |
| if (RMAppImpl.isAppInFinalState(rmApp) && stateChangedToFinal) { |
| updateLogAggregationStatus(nodeId); |
| } |
| } |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| public LogAggregationStatus getLogAggregationStatusForAppReport( |
| RMAppImpl rmApp) { |
| boolean appInFinalState = RMAppImpl.isAppInFinalState(rmApp); |
| this.readLock.lock(); |
| try { |
| if (!logAggregationEnabled) { |
| return LogAggregationStatus.DISABLED; |
| } |
| if (isLogAggregationFinished()) { |
| return this.logAggregationStatusForAppReport; |
| } |
| Map<NodeId, LogAggregationReport> reports = |
| getLogAggregationReportsForApp(rmApp); |
| if (reports.size() == 0) { |
| return this.logAggregationStatusForAppReport; |
| } |
| int logNotStartCount = 0; |
| int logCompletedCount = 0; |
| int logTimeOutCount = 0; |
| int logFailedCount = 0; |
| int logRunningWithFailure = 0; |
| for (Map.Entry<NodeId, LogAggregationReport> report : |
| reports.entrySet()) { |
| switch (report.getValue().getLogAggregationStatus()) { |
| case NOT_START: |
| logNotStartCount++; |
| break; |
| case RUNNING_WITH_FAILURE: |
| logRunningWithFailure ++; |
| break; |
| case SUCCEEDED: |
| logCompletedCount++; |
| break; |
| case FAILED: |
| logFailedCount++; |
| logCompletedCount++; |
| break; |
| case TIME_OUT: |
| logTimeOutCount++; |
| logCompletedCount++; |
| break; |
| default: |
| break; |
| } |
| } |
| if (logNotStartCount == reports.size()) { |
| return LogAggregationStatus.NOT_START; |
| } else if (logCompletedCount == reports.size()) { |
| // We should satisfy two condition in order to return |
| // SUCCEEDED or FAILED. |
| // 1) make sure the application is in final state |
| // 2) logs status from all NMs are SUCCEEDED/FAILED/TIMEOUT |
| // The SUCCEEDED/FAILED status is the final status which means |
| // the log aggregation is finished. And the log aggregation status will |
| // not be updated anymore. |
| if (logFailedCount > 0 && appInFinalState) { |
| this.logAggregationStatusForAppReport = |
| LogAggregationStatus.FAILED; |
| return LogAggregationStatus.FAILED; |
| } else if (logTimeOutCount > 0) { |
| this.logAggregationStatusForAppReport = |
| LogAggregationStatus.TIME_OUT; |
| return LogAggregationStatus.TIME_OUT; |
| } |
| if (appInFinalState) { |
| this.logAggregationStatusForAppReport = |
| LogAggregationStatus.SUCCEEDED; |
| return LogAggregationStatus.SUCCEEDED; |
| } |
| } else if (logRunningWithFailure > 0) { |
| return LogAggregationStatus.RUNNING_WITH_FAILURE; |
| } |
| return LogAggregationStatus.RUNNING; |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| private boolean isLogAggregationFinished() { |
| return this.logAggregationStatusForAppReport |
| .equals(LogAggregationStatus.SUCCEEDED) |
| || this.logAggregationStatusForAppReport |
| .equals(LogAggregationStatus.FAILED) |
| || this.logAggregationStatusForAppReport |
| .equals(LogAggregationStatus.TIME_OUT); |
| |
| } |
| |
| private boolean isLogAggregationFinishedForNM(LogAggregationReport report) { |
| return report.getLogAggregationStatus() == LogAggregationStatus.SUCCEEDED |
| || report.getLogAggregationStatus() == LogAggregationStatus.FAILED; |
| } |
| |
| private void updateLogAggregationDiagnosticMessages(NodeId nodeId, |
| LogAggregationReport report) { |
| if (report.getDiagnosticMessage() != null |
| && !report.getDiagnosticMessage().isEmpty()) { |
| if (report.getLogAggregationStatus() |
| == LogAggregationStatus.RUNNING ) { |
| List<String> diagnostics = logAggregationDiagnosticsForNMs.get(nodeId); |
| if (diagnostics == null) { |
| diagnostics = new ArrayList<>(); |
| logAggregationDiagnosticsForNMs.put(nodeId, diagnostics); |
| } else { |
| if (diagnostics.size() |
| == maxLogAggregationDiagnosticsInMemory) { |
| diagnostics.remove(0); |
| } |
| } |
| diagnostics.add(report.getDiagnosticMessage()); |
| this.logAggregationStatus.get(nodeId).setDiagnosticMessage( |
| StringUtils.join(diagnostics, "\n")); |
| } else if (report.getLogAggregationStatus() |
| == LogAggregationStatus.RUNNING_WITH_FAILURE) { |
| List<String> failureMessages = |
| logAggregationFailureMessagesForNMs.get(nodeId); |
| if (failureMessages == null) { |
| failureMessages = new ArrayList<>(); |
| logAggregationFailureMessagesForNMs.put(nodeId, failureMessages); |
| } else { |
| if (failureMessages.size() |
| == maxLogAggregationDiagnosticsInMemory) { |
| failureMessages.remove(0); |
| } |
| } |
| failureMessages.add(report.getDiagnosticMessage()); |
| } |
| } |
| } |
| |
| private void updateLogAggregationStatus(NodeId nodeId) { |
| LogAggregationStatus status = |
| this.logAggregationStatus.get(nodeId).getLogAggregationStatus(); |
| if (status.equals(LogAggregationStatus.SUCCEEDED)) { |
| this.logAggregationSucceed++; |
| } else if (status.equals(LogAggregationStatus.FAILED)) { |
| this.logAggregationFailed++; |
| } |
| if (this.logAggregationSucceed == this.logAggregationStatus.size()) { |
| this.logAggregationStatusForAppReport = |
| LogAggregationStatus.SUCCEEDED; |
| // Since the log aggregation status for this application for all NMs |
| // is SUCCEEDED, it means all logs are aggregated successfully. |
| // We could remove all the cached log aggregation reports |
| this.logAggregationStatus.clear(); |
| this.logAggregationDiagnosticsForNMs.clear(); |
| this.logAggregationFailureMessagesForNMs.clear(); |
| } else if (this.logAggregationSucceed + this.logAggregationFailed |
| == this.logAggregationStatus.size()) { |
| this.logAggregationStatusForAppReport = LogAggregationStatus.FAILED; |
| // We have collected the log aggregation status for all NMs. |
| // The log aggregation status is FAILED which means the log |
| // aggregation fails in some NMs. We are only interested in the |
| // nodes where the log aggregation is failed. So we could remove |
| // the log aggregation details for those succeeded NMs |
| this.logAggregationStatus.entrySet().removeIf(entry -> |
| entry.getValue().getLogAggregationStatus() |
| .equals(LogAggregationStatus.SUCCEEDED)); |
| // the log aggregation has finished/failed. |
| // and the status will not be updated anymore. |
| this.logAggregationDiagnosticsForNMs.clear(); |
| } |
| } |
| |
| String getLogAggregationFailureMessagesForNM(NodeId nodeId) { |
| this.readLock.lock(); |
| try { |
| List<String> failureMessages = |
| this.logAggregationFailureMessagesForNMs.get(nodeId); |
| if (failureMessages == null || failureMessages.isEmpty()) { |
| return StringUtils.EMPTY; |
| } |
| return StringUtils.join(failureMessages, "\n"); |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| void recordLogAggregationStartTime(long time) { |
| logAggregationStartTime = time; |
| } |
| |
| public boolean isEnabled() { |
| return logAggregationEnabled; |
| } |
| |
| private boolean hasReportForNodeManager(NodeId nodeId) { |
| return logAggregationStatus.containsKey(nodeId); |
| } |
| |
| private void addReportForNodeManager(NodeId nodeId, |
| LogAggregationReport report) { |
| logAggregationStatus.put(nodeId, report); |
| } |
| |
| public boolean isFinished() { |
| return isLogAggregationFinished(); |
| } |
| |
| private boolean isThereFailureMessageForNM(NodeId nodeId) { |
| return logAggregationFailureMessagesForNMs.get(nodeId) != null |
| && !logAggregationFailureMessagesForNMs.get(nodeId).isEmpty(); |
| } |
| |
| long getLogAggregationStartTime() { |
| return logAggregationStartTime; |
| } |
| |
| void addReportIfNecessary(NodeId nodeId, ApplicationId applicationId) { |
| if (!hasReportForNodeManager(nodeId)) { |
| LogAggregationStatus status = isEnabled() ? LogAggregationStatus.NOT_START |
| : LogAggregationStatus.DISABLED; |
| addReportForNodeManager(nodeId, |
| LogAggregationReport.newInstance(applicationId, status, "")); |
| } |
| } |
| } |
