Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

OOZIE-3719: Improve coordinator scope range checking #95

Open
wants to merge 2 commits into
base: github_actions_sandbox
Choose a base branch
from
Open

OOZIE-3719: Improve coordinator scope range checking #95

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
285008f
OOZIE-3719: Improve coordinator scope range checking
dionusos Jul 17, 2024
60c72fc
Merge branch 'apache:github_actions_sandbox' into github_actions_sandbox
dionusos Jul 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 2 additions & 43 deletions core/src/main/java/org/apache/oozie/CoordinatorEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -62,6 +61,7 @@
import org.apache.oozie.command.coord.CoordSuspendXCommand;
import org.apache.oozie.command.coord.CoordUpdateXCommand;
import org.apache.oozie.command.coord.CoordWfActionInfoXCommand;
import org.apache.oozie.coord.CoordUtils;
import org.apache.oozie.dependency.ActionDependency;
import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
Expand Down Expand Up @@ -314,48 +314,7 @@ public void streamLog(String jobId, String logRetrievalScope, String logRetrieva
// if coordinator action logs are to be retrieved based on action id range
if (logRetrievalType.equals(RestConstants.JOB_LOG_ACTION)) {
// Use set implementation that maintains order or elements to achieve reproducibility:
Set<String> actionSet = new LinkedHashSet<String>();
String[] list = logRetrievalScope.split(",");
for (String s : list) {
s = s.trim();
if (s.contains("-")) {
String[] range = s.split("-");
if (range.length != 2) {
throw new CommandException(ErrorCode.E0302, "format is wrong for action's range '" + s
+ "'");
}
int start;
int end;
try {
start = Integer.parseInt(range[0].trim());
} catch (NumberFormatException ne) {
throw new CommandException(ErrorCode.E0302, "could not parse " + range[0].trim() + "into an integer",
ne);
}
try {
end = Integer.parseInt(range[1].trim());
} catch (NumberFormatException ne) {
throw new CommandException(ErrorCode.E0302, "could not parse " + range[1].trim() + "into an integer",
ne);
}
if (start > end) {
throw new CommandException(ErrorCode.E0302, "format is wrong for action's range '" + s + "'");
}
for (int i = start; i <= end; i++) {
actionSet.add(jobId + "@" + i);
}
}
else {
try {
Integer.parseInt(s);
}
catch (NumberFormatException ne) {
throw new CommandException(ErrorCode.E0302, "format is wrong for action id'" + s
+ "'. Integer only.");
}
actionSet.add(jobId + "@" + s);
}
}
final Set<String> actionSet = CoordUtils.getActionsIds(jobId, logRetrievalScope);

if (actionSet.size() >= maxNumActionsForLog) {
throw new CommandException(ErrorCode.E0302,
Expand Down
1 change: 1 addition & 0 deletions core/src/main/java/org/apache/oozie/ErrorCode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ public enum ErrorCode {
E0306(XLog.STD, "Invalid parameter"),
E0307(XLog.STD, "Runtime error [{0}]"),
E0308(XLog.STD, "Could not parse date range parameter [{0}]"),
E0309(XLog.STD, "Invalid parameter value, [{0}] = [{1}], {2}"),

E0401(XLog.STD, "Missing configuration property [{0}]"),
E0402(XLog.STD, "Invalid callback ID [{0}]"),
Expand Down
153 changes: 110 additions & 43 deletions core/src/main/java/org/apache/oozie/coord/CoordUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,15 +22,17 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.Map;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.commons.lang3.Range;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.CoordinatorActionBean;
Expand Down Expand Up @@ -63,6 +65,7 @@


public class CoordUtils {
private static final XLog LOG = XLog.getLog(CoordUtils.class);
public static final String HADOOP_USER = "user.name";

public static String getDoneFlag(Element doneFlagElement) {
Expand Down Expand Up @@ -92,7 +95,7 @@ public static Configuration getHadoopConf(Configuration jobConf) {
* @throws CommandException thrown if failed to get coordinator actions by given date range
*/
public static List<CoordinatorActionBean> getCoordActions(String rangeType, String jobId, String scope,
boolean active) throws CommandException {
boolean active) throws CommandException {
List<CoordinatorActionBean> coordActions = null;
if (rangeType.equals(RestConstants.JOB_COORD_SCOPE_DATE)) {
coordActions = CoordUtils.getCoordActionsFromDates(jobId, scope, active);
Expand Down Expand Up @@ -196,52 +199,116 @@ public static Set<String> getActionsIds(String jobId, String scope) throws Comma
ParamChecker.notEmpty(jobId, "jobId");
ParamChecker.notEmpty(scope, "scope");

Set<String> actions = new LinkedHashSet<String>();
String[] list = scope.split(",");
for (String s : list) {
s = s.trim();
// An action range is specified with two actions separated by '-'
if (s.contains("-")) {
String[] range = s.split("-");
// Check the format for action's range
if (range.length != 2) {
throw new CommandException(ErrorCode.E0302, "format is wrong for action's range '" + s + "', an example of"
+ " correct format is 1-5");
}
int start;
int end;
//Get the starting and ending action numbers
try {
start = Integer.parseInt(range[0].trim());
} catch (NumberFormatException ne) {
throw new CommandException(ErrorCode.E0302, "could not parse " + range[0].trim() + "into an integer", ne);
}
try {
end = Integer.parseInt(range[1].trim());
} catch (NumberFormatException ne) {
throw new CommandException(ErrorCode.E0302, "could not parse " + range[1].trim() + "into an integer", ne);
}
if (start > end) {
throw new CommandException(ErrorCode.E0302, "format is wrong for action's range '" + s + "', starting action"
+ "number of the range should be less than ending action number, an example will be 1-4");
}
// Add the actionIds
for (int i = start; i <= end; i++) {
actions.add(jobId + "@" + i);
final Set<String> actions = new LinkedHashSet<>();
for (final Range<Integer> range : parseScopeToRanges(scope)) {
// Add the actionIds
for (int i = range.getMinimum(); i <= range.getMaximum(); i++) {
final String jobIdToAdd = jobId + "@" + i;
if (LOG.isTraceEnabled()) {
LOG.trace("Adding {0} to actionSet.", jobIdToAdd);
}
actions.add(jobIdToAdd);
}
else {
}
return actions;
}

/**
* Parses a string value into a {@link org.apache.commons.lang3.Range} object while does sanity checking.
*
* @param rangeToParse string representing a minimum and maximum value separated by a dash: '1-5'
* @return the parsed range class
* @throws CommandException when the provided range string is empty, invalid or starting value is greater than ending value
*/
public static Range<Integer> parseRange(final String rangeToParse) throws CommandException {
final String range = StringUtils.stripToNull(rangeToParse);
if (range == null) {
throw new CommandException(ErrorCode.E0302, String.format("Range cannot be empty: %s", rangeToParse));
}
final String[] parts = range.split("-");
if (parts.length != 2) {
throw new CommandException(
ErrorCode.E0302,
String.format("format is wrong for action's range '%s', an example of correct format is 1-5", rangeToParse)
);
}
try {
final int start = Integer.parseInt(parts[0].trim());
final int end = Integer.parseInt(parts[1].trim());

if (start > end) {
throw new CommandException(
ErrorCode.E0302,
String.format("format is wrong for action's range '%s', starting action number of the range " +
"should be less than ending action number, an example will be 1-4", rangeToParse)
);
}

return Range.between(start, end);
} catch (final NumberFormatException ne) {
throw new CommandException(
ErrorCode.E0302,
String.format("could not parse boundaries of %s into an integer", rangeToParse),
ne
);
}
}

/**
* Parses a scope definition (comma-separated list of values and ranges) into a list of
* {@link org.apache.commons.lang3.Range}s. If a single value is defined, a [1-1] range will be created.
*
* @param scope comma-separated list of values and ranges
* @return list of ranges based on the provided 'scope'
* @throws CommandException when provided scope string is empty, invalid
* or the ranges' starting value is greater than ending value
*/
public static List<Range<Integer>> parseScopeToRanges(final String scope) throws CommandException {
if (StringUtils.stripToNull(scope) == null) {
throw new CommandException(ErrorCode.E0302, "scope should not be empty");
}
final List<Range<Integer>> result = new ArrayList<>();
final String[] list = scope.split(",");
for (final String s : list) {
final String range = StringUtils.stripToNull(s);
if (range == null) {
continue;
}

if (range.contains("-")) {
// assume it is a valid range: 1-5
result.add(parseRange(range));
} else {
// assume it is a plain number
try {
Integer.parseInt(s);
}
catch (NumberFormatException ne) {
throw new CommandException(ErrorCode.E0302, "format is wrong for action id'" + s
+ "'. Integer only.");
final int elem = Integer.parseInt(range);
result.add(Range.between(elem, elem));
} catch (final NumberFormatException ne) {
throw new CommandException(ErrorCode.E0302, String.format("could not parse %s into an integer", range), ne);
}
actions.add(jobId + "@" + s);
}
}
return actions;
if (LOG.isDebugEnabled()) {
LOG.debug(
"Created the following ranges from \"{0}\": {1}",
scope,
result
.stream()
.map(r -> String.format("[%s-%s]", r.getMinimum(), r.getMaximum()))
.collect(Collectors.joining(", "))
);
}
return result;
}

/**
* Calculates the number of elements described in the list of ranges.
*
* @param ranges list of {@link org.apache.commons.lang3.Range}s
* @return the number of elements the provided ranges contain
*/
public static int getElemCountOfRanges(final List<Range<Integer>> ranges) {
return ranges.stream().mapToInt(r -> r.getMaximum() - r.getMinimum() + 1).sum();
}

/**
Expand Down
62 changes: 60 additions & 2 deletions core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,11 +28,28 @@
import javax.servlet.http.HttpServletResponse;

import com.google.common.base.Strings;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.*;
import org.apache.oozie.BaseEngineException;
import org.apache.oozie.BundleEngine;
import org.apache.oozie.BundleEngineException;
import org.apache.oozie.CoordinatorActionBean;
import org.apache.oozie.CoordinatorActionInfo;
import org.apache.oozie.CoordinatorEngine;
import org.apache.oozie.CoordinatorEngineException;
import org.apache.oozie.CoordinatorJobBean;
import org.apache.oozie.DagEngine;
import org.apache.oozie.DagEngineException;
import org.apache.oozie.ErrorCode;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
import org.apache.oozie.XException;
import org.apache.oozie.client.WorkflowAction;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.client.rest.*;
import org.apache.oozie.client.rest.JsonBean;
import org.apache.oozie.client.rest.JsonTags;
import org.apache.oozie.client.rest.JsonUtils;
import org.apache.oozie.client.rest.RestConstants;
import org.apache.oozie.command.CommandException;
import org.apache.oozie.coord.CoordUtils;
import org.apache.oozie.service.BundleEngineService;
Expand All @@ -42,6 +59,7 @@
import org.apache.oozie.service.Services;
import org.apache.oozie.service.UUIDService;
import org.apache.oozie.util.Instrumentation;
import org.apache.oozie.util.ParameterVerifierException;
import org.apache.oozie.util.graph.GraphGenerator;
import org.apache.oozie.util.XLog;
import org.apache.oozie.util.graph.GraphRenderer;
Expand Down Expand Up @@ -568,6 +586,8 @@ private JSONObject killCoordinator(HttpServletRequest request, HttpServletRespon
String rangeType = request.getParameter(RestConstants.JOB_COORD_RANGE_TYPE_PARAM);
String scope = request.getParameter(RestConstants.JOB_COORD_SCOPE_PARAM);

validateScopeSize(scope, rangeType);

try {
if (rangeType != null && scope != null) {
XLog.getLog(getClass()).info(
Expand Down Expand Up @@ -730,6 +750,8 @@ private JSONObject reRunCoordinatorActions(HttpServletRequest request, HttpServl
"Rerun coordinator for jobId=" + jobId + ", rerunType=" + rerunType + ",scope=" + scope + ",refresh="
+ refresh + ", noCleanup=" + noCleanup);

validateScopeSize(scope, rerunType);

try {
if (!(rerunType.equals(RestConstants.JOB_COORD_SCOPE_DATE) || rerunType
.equals(RestConstants.JOB_COORD_SCOPE_ACTION))) {
Expand All @@ -753,6 +775,39 @@ private JSONObject reRunCoordinatorActions(HttpServletRequest request, HttpServl
return json;
}

/**
* Validates if the number of elements defined in 'scope' (comma separated list of values and ranges)
* is less than or equal than the maximum allowed count: oozie.coord.actions.scope.max.size.
*
* @param scope comma separated list of values and ranges
* @throws XServletException if there are too many elements or ranges/values are invalid
*/
static void validateScopeSize(final String scope, final String rangeType) throws XServletException {
if (!"action".equalsIgnoreCase(StringUtils.stripToNull(rangeType))) {
return;
}

final int maxElemCount = ConfigurationService.getInt("oozie.coord.actions.scope.max.size");

try {
final int elemCountOfRanges = CoordUtils.getElemCountOfRanges(CoordUtils.parseScopeToRanges(scope));
if (elemCountOfRanges > maxElemCount) {
throw new ParameterVerifierException(
ErrorCode.E0309,
"scope",
scope,
String.format(
"too many elements are requested: %s, maximum allowed: %s",
elemCountOfRanges,
maxElemCount
)
);
}
} catch (final XException e) {
throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, e);
}
}

/**
* Get workflow job
*
Expand Down Expand Up @@ -1103,6 +1158,9 @@ protected JSONObject getJobsByParentId(HttpServletRequest request, HttpServletRe
String coordActionId;
String type = request.getParameter(RestConstants.JOB_COORD_RANGE_TYPE_PARAM);
String scope = request.getParameter(RestConstants.JOB_COORD_SCOPE_PARAM);

validateScopeSize(scope, type);

// for getting allruns for coordinator action - 2 alternate endpoints
if (type != null && type.equals(RestConstants.JOB_COORD_SCOPE_ACTION) && scope != null) {
// endpoint - oozie/v2/coord-job-id?type=action&scope=action-num&show=allruns
Expand Down
Loading
Loading