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.action.hadoop;
020
021import java.io.IOException;
022import java.net.URISyntaxException;
023import java.util.ArrayList;
024import java.util.HashMap;
025import java.util.List;
026import java.util.Map;
027
028import org.apache.hadoop.fs.FSDataOutputStream;
029import org.apache.hadoop.fs.FileStatus;
030import org.apache.hadoop.fs.FileSystem;
031import org.apache.hadoop.fs.FileUtil;
032import org.apache.hadoop.fs.Path;
033import org.apache.hadoop.fs.permission.FsPermission;
034import org.apache.hadoop.mapred.JobConf;
035import org.apache.hadoop.security.AccessControlException;
036import org.apache.oozie.action.ActionExecutor;
037import org.apache.oozie.action.ActionExecutorException;
038import org.apache.oozie.client.WorkflowAction;
039import org.apache.oozie.service.ConfigurationService;
040import org.apache.oozie.service.HadoopAccessorException;
041import org.apache.oozie.service.HadoopAccessorService;
042import org.apache.oozie.service.Services;
043import org.apache.oozie.util.XConfiguration;
044import org.apache.oozie.util.XmlUtils;
045import org.jdom.Element;
046
047/**
048 * File system action executor. <p/> This executes the file system mkdir, move and delete commands
049 */
050public class FsActionExecutor extends ActionExecutor {
051
052    public static final String ACTION_TYPE = "fs";
053
054    private final int maxGlobCount;
055
056    public FsActionExecutor() {
057        super(ACTION_TYPE);
058        maxGlobCount = ConfigurationService.getInt(LauncherMapper.CONF_OOZIE_ACTION_FS_GLOB_MAX);
059    }
060
061    /**
062    * Initialize Action.
063    */
064    @Override
065    public void initActionType() {
066        super.initActionType();
067        registerError(AccessControlException.class.getName(), ActionExecutorException.ErrorType.ERROR, "FS014");
068    }
069
070    Path getPath(Element element, String attribute) {
071        String str = element.getAttributeValue(attribute).trim();
072        return new Path(str);
073    }
074
075    void validatePath(Path path, boolean withScheme) throws ActionExecutorException {
076        try {
077            String scheme = path.toUri().getScheme();
078            if (withScheme) {
079                if (scheme == null) {
080                    throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS001",
081                                                      "Missing scheme in path [{0}]", path);
082                }
083                else {
084                    Services.get().get(HadoopAccessorService.class).checkSupportedFilesystem(path.toUri());
085                }
086            }
087            else {
088                if (scheme != null) {
089                    throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS002",
090                                                      "Scheme [{0}] not allowed in path [{1}]", scheme, path);
091                }
092            }
093        }
094        catch (HadoopAccessorException hex) {
095            throw convertException(hex);
096        }
097    }
098
099    Path resolveToFullPath(Path nameNode, Path path, boolean withScheme) throws ActionExecutorException {
100        Path fullPath;
101
102        // If no nameNode is given, validate the path as-is and return it as-is
103        if (nameNode == null) {
104            validatePath(path, withScheme);
105            fullPath = path;
106        } else {
107            // If the path doesn't have a scheme or authority, use the nameNode which should have already been verified earlier
108            String pathScheme = path.toUri().getScheme();
109            String pathAuthority = path.toUri().getAuthority();
110            if (pathScheme == null || pathAuthority == null) {
111                if (path.isAbsolute()) {
112                    String nameNodeSchemeAuthority = nameNode.toUri().getScheme() + "://" + nameNode.toUri().getAuthority();
113                    fullPath = new Path(nameNodeSchemeAuthority + path.toString());
114                } else {
115                    throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS011",
116                            "Path [{0}] cannot be relative", path);
117                }
118            } else {
119                // If the path has a scheme and authority, but its not the nameNode then validate the path as-is and return it as-is
120                // If it is the nameNode, then it should have already been verified earlier so return it as-is
121                if (!nameNode.toUri().getScheme().equals(pathScheme) || !nameNode.toUri().getAuthority().equals(pathAuthority)) {
122                    validatePath(path, withScheme);
123                }
124                fullPath = path;
125            }
126        }
127        return fullPath;
128    }
129
130    void validateSameNN(Path source, Path dest) throws ActionExecutorException {
131        Path destPath = new Path(source, dest);
132        String t = destPath.toUri().getScheme() + destPath.toUri().getAuthority();
133        String s = source.toUri().getScheme() + source.toUri().getAuthority();
134
135        //checking whether NN prefix of source and target is same. can modify this to adjust for a set of multiple whitelisted NN
136        if(!t.equals(s)) {
137            throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS007",
138                    "move, target NN URI different from that of source", dest);
139        }
140    }
141
142    @SuppressWarnings("unchecked")
143    void doOperations(Context context, Element element) throws ActionExecutorException {
144        try {
145            FileSystem fs = context.getAppFileSystem();
146            boolean recovery = fs.exists(getRecoveryPath(context));
147            if (!recovery) {
148                fs.mkdirs(getRecoveryPath(context));
149            }
150
151            Path nameNodePath = null;
152            Element nameNodeElement = element.getChild("name-node", element.getNamespace());
153            if (nameNodeElement != null) {
154                String nameNode = nameNodeElement.getTextTrim();
155                if (nameNode != null) {
156                    nameNodePath = new Path(nameNode);
157                    // Verify the name node now
158                    validatePath(nameNodePath, true);
159                }
160            }
161
162            XConfiguration fsConf = new XConfiguration();
163            Path appPath = new Path(context.getWorkflow().getAppPath());
164            // app path could be a file
165            if (fs.isFile(appPath)) {
166                appPath = appPath.getParent();
167            }
168            JavaActionExecutor.parseJobXmlAndConfiguration(context, element, appPath, fsConf);
169
170            for (Element commandElement : (List<Element>) element.getChildren()) {
171                String command = commandElement.getName();
172                if (command.equals("mkdir")) {
173                    Path path = getPath(commandElement, "path");
174                    mkdir(context, fsConf, nameNodePath, path);
175                }
176                else {
177                    if (command.equals("delete")) {
178                        Path path = getPath(commandElement, "path");
179                        delete(context, fsConf, nameNodePath, path);
180                    }
181                    else {
182                        if (command.equals("move")) {
183                            Path source = getPath(commandElement, "source");
184                            Path target = getPath(commandElement, "target");
185                            move(context, fsConf, nameNodePath, source, target, recovery);
186                        }
187                        else {
188                            if (command.equals("chmod")) {
189                                Path path = getPath(commandElement, "path");
190                                boolean recursive = commandElement.getChild("recursive", commandElement.getNamespace()) != null;
191                                String str = commandElement.getAttributeValue("dir-files");
192                                boolean dirFiles = (str == null) || Boolean.parseBoolean(str);
193                                String permissionsMask = commandElement.getAttributeValue("permissions").trim();
194                                chmod(context, fsConf, nameNodePath, path, permissionsMask, dirFiles, recursive);
195                            }
196                            else {
197                                if (command.equals("touchz")) {
198                                    Path path = getPath(commandElement, "path");
199                                    touchz(context, fsConf, nameNodePath, path);
200                                }
201                                else {
202                                    if (command.equals("chgrp")) {
203                                        Path path = getPath(commandElement, "path");
204                                        boolean recursive = commandElement.getChild("recursive",
205                                                commandElement.getNamespace()) != null;
206                                        String group = commandElement.getAttributeValue("group");
207                                        String str = commandElement.getAttributeValue("dir-files");
208                                        boolean dirFiles = (str == null) || Boolean.parseBoolean(str);
209                                        chgrp(context, fsConf, nameNodePath, path, context.getWorkflow().getUser(),
210                                                group, dirFiles, recursive);
211                                    }
212                                }
213                            }
214                        }
215                    }
216                }
217            }
218        }
219        catch (Exception ex) {
220            throw convertException(ex);
221        }
222    }
223
224    void chgrp(Context context, XConfiguration fsConf, Path nameNodePath, Path path, String user, String group,
225            boolean dirFiles, boolean recursive) throws ActionExecutorException {
226
227        HashMap<String, String> argsMap = new HashMap<String, String>();
228        argsMap.put("user", user);
229        argsMap.put("group", group);
230        try {
231            FileSystem fs = getFileSystemFor(path, context, fsConf);
232            path = resolveToFullPath(nameNodePath, path, true);
233            Path[] pathArr = FileUtil.stat2Paths(fs.globStatus(path));
234            if (pathArr == null || pathArr.length == 0) {
235                throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS009", "chgrp"
236                        + ", path(s) that matches [{0}] does not exist", path);
237            }
238            checkGlobMax(pathArr);
239            for (Path p : pathArr) {
240                recursiveFsOperation("chgrp", fs, nameNodePath, p, argsMap, dirFiles, recursive, true);
241            }
242        }
243        catch (Exception ex) {
244            throw convertException(ex);
245        }
246    }
247
248    private void recursiveFsOperation(String op, FileSystem fs, Path nameNodePath, Path path,
249            Map<String, String> argsMap, boolean dirFiles, boolean recursive, boolean isRoot)
250            throws ActionExecutorException {
251
252        try {
253            FileStatus pathStatus = fs.getFileStatus(path);
254            List<Path> paths = new ArrayList<Path>();
255
256            if (dirFiles && pathStatus.isDir()) {
257                if (isRoot) {
258                    paths.add(path);
259                }
260                FileStatus[] filesStatus = fs.listStatus(path);
261                for (int i = 0; i < filesStatus.length; i++) {
262                    Path p = filesStatus[i].getPath();
263                    paths.add(p);
264                    if (recursive && filesStatus[i].isDir()) {
265                        recursiveFsOperation(op, fs, null, p, argsMap, dirFiles, recursive, false);
266                    }
267                }
268            }
269            else {
270                paths.add(path);
271            }
272            for (Path p : paths) {
273                doFsOperation(op, fs, p, argsMap);
274            }
275        }
276        catch (Exception ex) {
277            throw convertException(ex);
278        }
279    }
280
281    private void doFsOperation(String op, FileSystem fs, Path p, Map<String, String> argsMap)
282            throws ActionExecutorException, IOException {
283        if (op.equals("chmod")) {
284            String permissions = argsMap.get("permissions");
285            FsPermission newFsPermission = createShortPermission(permissions, p);
286            fs.setPermission(p, newFsPermission);
287        }
288        else if (op.equals("chgrp")) {
289            String user = argsMap.get("user");
290            String group = argsMap.get("group");
291            fs.setOwner(p, user, group);
292        }
293    }
294
295    /**
296     * @param path
297     * @param context
298     * @param fsConf
299     * @return FileSystem
300     * @throws HadoopAccessorException
301     */
302    private FileSystem getFileSystemFor(Path path, Context context, XConfiguration fsConf) throws HadoopAccessorException {
303        String user = context.getWorkflow().getUser();
304        HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
305        JobConf conf = has.createJobConf(path.toUri().getAuthority());
306        XConfiguration.copy(context.getProtoActionConf(), conf);
307        if (fsConf != null) {
308            XConfiguration.copy(fsConf, conf);
309        }
310        return has.createFileSystem(user, path.toUri(), conf);
311    }
312
313    /**
314     * @param path
315     * @param user
316     * @param group
317     * @return FileSystem
318     * @throws HadoopAccessorException
319     */
320    private FileSystem getFileSystemFor(Path path, String user) throws HadoopAccessorException {
321        HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
322        JobConf jobConf = has.createJobConf(path.toUri().getAuthority());
323        return has.createFileSystem(user, path.toUri(), jobConf);
324    }
325
326    void mkdir(Context context, Path path) throws ActionExecutorException {
327        mkdir(context, null, null, path);
328    }
329
330    void mkdir(Context context, XConfiguration fsConf, Path nameNodePath, Path path) throws ActionExecutorException {
331        try {
332            path = resolveToFullPath(nameNodePath, path, true);
333            FileSystem fs = getFileSystemFor(path, context, fsConf);
334
335            if (!fs.exists(path)) {
336                if (!fs.mkdirs(path)) {
337                    throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS004",
338                                                      "mkdir, path [{0}] could not create directory", path);
339                }
340            }
341        }
342        catch (Exception ex) {
343            throw convertException(ex);
344        }
345    }
346
347    /**
348     * Delete path
349     *
350     * @param context
351     * @param path
352     * @throws ActionExecutorException
353     */
354    public void delete(Context context, Path path) throws ActionExecutorException {
355        delete(context, null, null, path);
356    }
357
358    /**
359     * Delete path
360     *
361     * @param context
362     * @param fsConf
363     * @param nameNodePath
364     * @param path
365     * @throws ActionExecutorException
366     */
367    public void delete(Context context, XConfiguration fsConf, Path nameNodePath, Path path) throws ActionExecutorException {
368        try {
369            path = resolveToFullPath(nameNodePath, path, true);
370            FileSystem fs = getFileSystemFor(path, context, fsConf);
371            Path[] pathArr = FileUtil.stat2Paths(fs.globStatus(path));
372            if (pathArr != null && pathArr.length > 0) {
373                checkGlobMax(pathArr);
374                for (Path p : pathArr) {
375                    if (fs.exists(p)) {
376                        if (!fs.delete(p, true)) {
377                            throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS005",
378                                    "delete, path [{0}] could not delete path", p);
379                        }
380                    }
381                }
382            }
383        }
384        catch (Exception ex) {
385            throw convertException(ex);
386        }
387    }
388
389    /**
390     * Delete path
391     *
392     * @param user
393     * @param group
394     * @param path
395     * @throws ActionExecutorException
396     */
397    public void delete(String user, String group, Path path) throws ActionExecutorException {
398        try {
399            validatePath(path, true);
400            FileSystem fs = getFileSystemFor(path, user);
401
402            if (fs.exists(path)) {
403                if (!fs.delete(path, true)) {
404                    throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS005",
405                            "delete, path [{0}] could not delete path", path);
406                }
407            }
408        }
409        catch (Exception ex) {
410            throw convertException(ex);
411        }
412    }
413
414    /**
415     * Move source to target
416     *
417     * @param context
418     * @param source
419     * @param target
420     * @param recovery
421     * @throws ActionExecutorException
422     */
423    public void move(Context context, Path source, Path target, boolean recovery) throws ActionExecutorException {
424        move(context, null, null, source, target, recovery);
425    }
426
427    /**
428     * Move source to target
429     *
430     * @param context
431     * @param fsConf
432     * @param nameNodePath
433     * @param source
434     * @param target
435     * @param recovery
436     * @throws ActionExecutorException
437     */
438    public void move(Context context, XConfiguration fsConf, Path nameNodePath, Path source, Path target, boolean recovery)
439            throws ActionExecutorException {
440        try {
441            source = resolveToFullPath(nameNodePath, source, true);
442            validateSameNN(source, target);
443            FileSystem fs = getFileSystemFor(source, context, fsConf);
444            Path[] pathArr = FileUtil.stat2Paths(fs.globStatus(source));
445            if (( pathArr == null || pathArr.length == 0 ) ){
446                if (!recovery) {
447                    throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS006",
448                        "move, source path [{0}] does not exist", source);
449                } else {
450                    return;
451                }
452            }
453            if (pathArr.length > 1 && (!fs.exists(target) || fs.isFile(target))) {
454                if(!recovery) {
455                    throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS012",
456                            "move, could not rename multiple sources to the same target name");
457                } else {
458                    return;
459                }
460            }
461            checkGlobMax(pathArr);
462            for (Path p : pathArr) {
463                if (!fs.rename(p, target) && !recovery) {
464                    throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS008",
465                            "move, could not move [{0}] to [{1}]", p, target);
466                }
467            }
468        }
469        catch (Exception ex) {
470            throw convertException(ex);
471        }
472    }
473
474    void chmod(Context context, Path path, String permissions, boolean dirFiles, boolean recursive) throws ActionExecutorException {
475        chmod(context, null, null, path, permissions, dirFiles, recursive);
476    }
477
478    void chmod(Context context, XConfiguration fsConf, Path nameNodePath, Path path, String permissions,
479            boolean dirFiles, boolean recursive) throws ActionExecutorException {
480
481        HashMap<String, String> argsMap = new HashMap<String, String>();
482        argsMap.put("permissions", permissions);
483        try {
484            FileSystem fs = getFileSystemFor(path, context, fsConf);
485            path = resolveToFullPath(nameNodePath, path, true);
486            Path[] pathArr = FileUtil.stat2Paths(fs.globStatus(path));
487            if (pathArr == null || pathArr.length == 0) {
488                throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS009", "chmod"
489                        + ", path(s) that matches [{0}] does not exist", path);
490            }
491            checkGlobMax(pathArr);
492            for (Path p : pathArr) {
493                recursiveFsOperation("chmod", fs, nameNodePath, p, argsMap, dirFiles, recursive, true);
494            }
495
496        }
497        catch (Exception ex) {
498            throw convertException(ex);
499        }
500    }
501
502    void touchz(Context context, Path path) throws ActionExecutorException {
503        touchz(context, null, null, path);
504    }
505
506    void touchz(Context context, XConfiguration fsConf, Path nameNodePath, Path path) throws ActionExecutorException {
507        try {
508            path = resolveToFullPath(nameNodePath, path, true);
509            FileSystem fs = getFileSystemFor(path, context, fsConf);
510
511            FileStatus st;
512            if (fs.exists(path)) {
513                st = fs.getFileStatus(path);
514                if (st.isDir()) {
515                    throw new Exception(path.toString() + " is a directory");
516                } else if (st.getLen() != 0)
517                    throw new Exception(path.toString() + " must be a zero-length file");
518            }
519            FSDataOutputStream out = fs.create(path);
520            out.close();
521        }
522        catch (Exception ex) {
523            throw convertException(ex);
524        }
525    }
526
527    FsPermission createShortPermission(String permissions, Path path) throws ActionExecutorException {
528        if (permissions.length() == 3) {
529            char user = permissions.charAt(0);
530            char group = permissions.charAt(1);
531            char other = permissions.charAt(2);
532            int useri = user - '0';
533            int groupi = group - '0';
534            int otheri = other - '0';
535            int mask = useri * 100 + groupi * 10 + otheri;
536            short omask = Short.parseShort(Integer.toString(mask), 8);
537            return new FsPermission(omask);
538        }
539        else {
540            if (permissions.length() == 10) {
541                return FsPermission.valueOf(permissions);
542            }
543            else {
544                throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS010",
545                                                  "chmod, path [{0}] invalid permissions mask [{1}]", path, permissions);
546            }
547        }
548    }
549
550    @Override
551    public void check(Context context, WorkflowAction action) throws ActionExecutorException {
552    }
553
554    @Override
555    public void kill(Context context, WorkflowAction action) throws ActionExecutorException {
556    }
557
558    @Override
559    public void start(Context context, WorkflowAction action) throws ActionExecutorException {
560        try {
561            context.setStartData("-", "-", "-");
562            Element actionXml = XmlUtils.parseXml(action.getConf());
563            doOperations(context, actionXml);
564            context.setExecutionData("OK", null);
565        }
566        catch (Exception ex) {
567            throw convertException(ex);
568        }
569    }
570
571    @Override
572    public void end(Context context, WorkflowAction action) throws ActionExecutorException {
573        String externalStatus = action.getExternalStatus();
574        WorkflowAction.Status status = externalStatus.equals("OK") ? WorkflowAction.Status.OK :
575                                       WorkflowAction.Status.ERROR;
576        context.setEndData(status, getActionSignal(status));
577        if (!context.getProtoActionConf().getBoolean("oozie.action.keep.action.dir", false)) {
578            try {
579                FileSystem fs = context.getAppFileSystem();
580                fs.delete(context.getActionDir(), true);
581            }
582            catch (Exception ex) {
583                throw convertException(ex);
584            }
585        }
586    }
587
588    @Override
589    public boolean isCompleted(String externalStatus) {
590        return true;
591    }
592
593    /**
594     * @param context
595     * @return
596     * @throws HadoopAccessorException
597     * @throws IOException
598     * @throws URISyntaxException
599     */
600    public Path getRecoveryPath(Context context) throws HadoopAccessorException, IOException, URISyntaxException {
601        return new Path(context.getActionDir(), "fs-" + context.getRecoveryId());
602    }
603
604    private void checkGlobMax(Path[] pathArr) throws ActionExecutorException {
605        if(pathArr.length > maxGlobCount) {
606            throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "FS013",
607                    "too many globbed files/dirs to do FS operation");
608        }
609    }
610
611    public boolean supportsConfigurationJobXML() {
612        return true;
613    }
614}