001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *      http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.oozie.coord;
020
021import java.text.ParseException;
022import java.util.ArrayList;
023import java.util.Date;
024import java.util.HashSet;
025import java.util.List;
026import java.util.Set;
027import java.util.Map;
028import java.util.HashMap;
029import java.util.concurrent.TimeUnit;
030
031import org.apache.hadoop.conf.Configuration;
032import org.apache.oozie.CoordinatorActionBean;
033import org.apache.oozie.CoordinatorEngine;
034import org.apache.oozie.ErrorCode;
035import org.apache.oozie.XException;
036import org.apache.oozie.client.OozieClient;
037import org.apache.oozie.client.rest.RestConstants;
038import org.apache.oozie.command.CommandException;
039import org.apache.oozie.executor.jpa.CoordActionGetJPAExecutor;
040import org.apache.oozie.executor.jpa.CoordJobGetActionForNominalTimeJPAExecutor;
041import org.apache.oozie.executor.jpa.JPAExecutorException;
042import org.apache.oozie.service.JPAService;
043import org.apache.oozie.service.Services;
044import org.apache.oozie.service.XLogService;
045import org.apache.oozie.util.CoordActionsInDateRange;
046import org.apache.oozie.util.DateUtils;
047import org.apache.oozie.util.Pair;
048import org.apache.oozie.util.ParamChecker;
049import org.apache.oozie.util.XLog;
050import org.jdom.Element;
051
052import com.google.common.annotations.VisibleForTesting;
053
054
055public class CoordUtils {
056    public static final String HADOOP_USER = "user.name";
057
058    public static String getDoneFlag(Element doneFlagElement) {
059        if (doneFlagElement != null) {
060            return doneFlagElement.getTextTrim();
061        }
062        else {
063            return CoordELConstants.DEFAULT_DONE_FLAG;
064        }
065    }
066
067    public static Configuration getHadoopConf(Configuration jobConf) {
068        Configuration conf = new Configuration();
069        ParamChecker.notNull(jobConf, "Configuration to be used for hadoop setup ");
070        String user = ParamChecker.notEmpty(jobConf.get(OozieClient.USER_NAME), OozieClient.USER_NAME);
071        conf.set(HADOOP_USER, user);
072        return conf;
073    }
074
075    /**
076     * Get the list of actions for a given coordinator job
077     * @param rangeType the rerun type (date, action)
078     * @param jobId the coordinator job id
079     * @param scope the date scope or action id scope
080     * @return the list of Coordinator actions
081     * @throws CommandException
082     */
083    public static List<CoordinatorActionBean> getCoordActions(String rangeType, String jobId, String scope,
084            boolean active) throws CommandException {
085        List<CoordinatorActionBean> coordActions = null;
086        if (rangeType.equals(RestConstants.JOB_COORD_SCOPE_DATE)) {
087            coordActions = CoordUtils.getCoordActionsFromDates(jobId, scope, active);
088        }
089        else if (rangeType.equals(RestConstants.JOB_COORD_SCOPE_ACTION)) {
090            coordActions = CoordUtils.getCoordActionsFromIds(jobId, scope);
091        }
092        return coordActions;
093    }
094
095    /**
096     * Get the list of actions for given date ranges
097     *
098     * @param jobId coordinator job id
099     * @param scope a comma-separated list of date ranges. Each date range element is specified with two dates separated by '::'
100     * @return the list of Coordinator actions for the date range
101     * @throws CommandException thrown if failed to get coordinator actions by given date range
102     */
103    static List<CoordinatorActionBean> getCoordActionsFromDates(String jobId, String scope, boolean active)
104            throws CommandException {
105        JPAService jpaService = Services.get().get(JPAService.class);
106        ParamChecker.notEmpty(jobId, "jobId");
107        ParamChecker.notEmpty(scope, "scope");
108
109        Set<CoordinatorActionBean> actionSet = new HashSet<CoordinatorActionBean>();
110        String[] list = scope.split(",");
111        for (String s : list) {
112            s = s.trim();
113            // A date range is specified with two dates separated by '::'
114            if (s.contains("::")) {
115            List<CoordinatorActionBean> listOfActions;
116            try {
117                // Get list of actions within the range of date
118                listOfActions = CoordActionsInDateRange.getCoordActionsFromDateRange(jobId, s, active);
119            }
120            catch (XException e) {
121                throw new CommandException(e);
122            }
123            actionSet.addAll(listOfActions);
124            }
125            else {
126                try {
127                    // Get action for the nominal time
128                    Date date = DateUtils.parseDateOozieTZ(s.trim());
129                    CoordinatorActionBean coordAction = jpaService
130                            .execute(new CoordJobGetActionForNominalTimeJPAExecutor(jobId, date));
131
132                    if (coordAction != null) {
133                        actionSet.add(coordAction);
134                    }
135                    else {
136                        throw new RuntimeException("This should never happen, Coordinator Action shouldn't be null");
137                    }
138                }
139                catch (ParseException e) {
140                    throw new CommandException(ErrorCode.E0302, s.trim(), e);
141                }
142                catch (JPAExecutorException e) {
143                    throw new CommandException(e);
144                }
145
146            }
147        }
148
149        List<CoordinatorActionBean> coordActions = new ArrayList<CoordinatorActionBean>();
150        for (CoordinatorActionBean coordAction : actionSet) {
151            coordActions.add(coordAction);
152        }
153        return coordActions;
154    }
155
156    /**
157     * Get the list of actions for given id ranges
158     *
159     * @param jobId coordinator job id
160     * @param scope a comma-separated list of action ranges. The action range is specified with two action numbers separated by '-'
161     * @return the list of all Coordinator actions for action range
162     * @throws CommandException thrown if failed to get coordinator actions by given id range
163     */
164     public static List<CoordinatorActionBean> getCoordActionsFromIds(String jobId, String scope) throws CommandException {
165        JPAService jpaService = Services.get().get(JPAService.class);
166        ParamChecker.notEmpty(jobId, "jobId");
167        ParamChecker.notEmpty(scope, "scope");
168
169        Set<String> actions = new HashSet<String>();
170        String[] list = scope.split(",");
171        for (String s : list) {
172            s = s.trim();
173            // An action range is specified with two actions separated by '-'
174            if (s.contains("-")) {
175                String[] range = s.split("-");
176                // Check the format for action's range
177                if (range.length != 2) {
178                    throw new CommandException(ErrorCode.E0302, "format is wrong for action's range '" + s + "', an example of correct format is 1-5");
179                }
180                int start;
181                int end;
182                //Get the starting and ending action numbers
183                try {
184                    start = Integer.parseInt(range[0].trim());
185                } catch (NumberFormatException ne) {
186                    throw new CommandException(ErrorCode.E0302, "could not parse " + range[0].trim() + "into an integer", ne);
187                }
188                try {
189                    end = Integer.parseInt(range[1].trim());
190                } catch (NumberFormatException ne) {
191                    throw new CommandException(ErrorCode.E0302, "could not parse " + range[1].trim() + "into an integer", ne);
192                }
193                if (start > end) {
194                    throw new CommandException(ErrorCode.E0302, "format is wrong for action's range '" + s + "', starting action"
195                            + "number of the range should be less than ending action number, an example will be 1-4");
196                }
197                // Add the actionIds
198                for (int i = start; i <= end; i++) {
199                    actions.add(jobId + "@" + i);
200                }
201            }
202            else {
203                try {
204                    Integer.parseInt(s);
205                }
206                catch (NumberFormatException ne) {
207                    throw new CommandException(ErrorCode.E0302, "format is wrong for action id'" + s
208                            + "'. Integer only.");
209                }
210                actions.add(jobId + "@" + s);
211            }
212        }
213        // Retrieve the actions using the corresponding actionIds
214        List<CoordinatorActionBean> coordActions = new ArrayList<CoordinatorActionBean>();
215        for (String id : actions) {
216            CoordinatorActionBean coordAction = null;
217            try {
218                coordAction = jpaService.execute(new CoordActionGetJPAExecutor(id));
219            }
220            catch (JPAExecutorException je) {
221                if (je.getErrorCode().equals(ErrorCode.E0605)) { //ignore retrieval of non-existent actions in range
222                    XLog.getLog(XLogService.class).warn(
223                            "Coord action ID num [{0}] not yet materialized. Hence skipping over it for Kill action",
224                            id.substring(id.indexOf("@") + 1));
225                    continue;
226                }
227                else {
228                    throw new CommandException(je);
229                }
230            }
231            coordActions.add(coordAction);
232        }
233        return coordActions;
234    }
235
236    // Form the where clause to filter by status values
237    public static Map<String, Object> getWhereClause(StringBuilder sb, Map<Pair<String, CoordinatorEngine.FILTER_COMPARATORS>,
238            List<Object>> filterMap) {
239        Map<String, Object> params = new HashMap<String, Object>();
240        int pcnt= 1;
241        for (Map.Entry<Pair<String, CoordinatorEngine.FILTER_COMPARATORS>, List<Object>> filter : filterMap.entrySet()) {
242            String field = filter.getKey().getFist();
243            CoordinatorEngine.FILTER_COMPARATORS comp = filter.getKey().getSecond();
244            String sqlField;
245            if (field.equals(OozieClient.FILTER_STATUS)) {
246                sqlField = "a.statusStr";
247            } else if (field.equals(OozieClient.FILTER_NOMINAL_TIME)) {
248                sqlField = "a.nominalTimestamp";
249            } else {
250                throw new IllegalArgumentException("Invalid filter key " + field);
251            }
252
253            sb.append(" and ").append(sqlField).append(" ");
254            switch (comp) {
255                case EQUALS:
256                    sb.append("IN (");
257                    params.putAll(appendParams(sb, filter.getValue(), pcnt));
258                    sb.append(")");
259                    break;
260
261                case NOT_EQUALS:
262                    sb.append("NOT IN (");
263                    params.putAll(appendParams(sb, filter.getValue(), pcnt));
264                    sb.append(")");
265                    break;
266
267                case GREATER:
268                case GREATER_EQUAL:
269                case LESSTHAN:
270                case LESSTHAN_EQUAL:
271                    if (filter.getValue().size() != 1) {
272                        throw new IllegalArgumentException(field + comp.getSign() + " can't have more than 1 values");
273                    }
274
275                    sb.append(comp.getSign()).append(" ");
276                    params.putAll(appendParams(sb, filter.getValue(), pcnt));
277                    break;
278            }
279
280            pcnt += filter.getValue().size();
281        }
282        sb.append(" ");
283        return params;
284    }
285
286    private static Map<String, Object> appendParams(StringBuilder sb, List<Object> value, int sindex) {
287        Map<String, Object> params = new HashMap<String, Object>();
288        boolean first = true;
289        for (Object val : value) {
290            String pname = "p" + sindex++;
291            params.put(pname, val);
292            if (!first) {
293                sb.append(", ");
294            }
295            sb.append(':').append(pname);
296            first = false;
297        }
298        return params;
299    }
300}