AuxServices extends AbstractService implements ServiceStateChangeListener, EventHandler<AuxServicesEvent> { @VisibleForTesting protected Path maybeDownloadJars(String sName, String className, String remoteFile, AuxServiceFile.TypeEnum type, Configuration conf) throws IOException { FileContext localLFS = getLocalFileContext(conf); Path nmAuxDir = dirsHandler.getLocalPathForWrite("." + Path.SEPARATOR + NM_AUX_SERVICE_DIR); if (!localLFS.util().exists(nmAuxDir)) { try { localLFS.mkdir(nmAuxDir, NM_AUX_SERVICE_DIR_PERM, true); } catch (IOException ex) { throw new YarnRuntimeException("Fail to create dir:" + nmAuxDir.toString(), ex); } } Path src = new Path(remoteFile); FileContext remoteLFS = getRemoteFileContext(src.toUri(), conf); FileStatus scFileStatus = remoteLFS.getFileStatus(src); if (!scFileStatus.getOwner().equals( this.userUGI.getShortUserName())) { throw new YarnRuntimeException("The remote jarfile owner:" + scFileStatus.getOwner() + " is not the same as the NM user:" + this.userUGI.getShortUserName() + "."); } if ((scFileStatus.getPermission().toShort() & 0022) != 0) { throw new YarnRuntimeException("The remote jarfile should not " + "be writable by group or others. " + "The current Permission is " + scFileStatus.getPermission().toShort()); } Path downloadDest = new Path(nmAuxDir, className + "_" + scFileStatus.getModificationTime()); Path targetDirPath = new Path(downloadDest, scFileStatus.getPath().getName()); FileStatus[] allSubDirs = localLFS.util().listStatus(nmAuxDir); for (FileStatus sub : allSubDirs) { if (sub.getPath().getName().equals(downloadDest.getName())) { return targetDirPath; } else { if (sub.getPath().getName().contains(className) && !sub.getPath().getName().endsWith(DEL_SUFFIX)) { Path delPath = new Path(sub.getPath().getParent(), sub.getPath().getName() + DEL_SUFFIX); localLFS.rename(sub.getPath(), delPath); LOG.info("delete old aux service jar dir:" + delPath.toString()); FileDeletionTask deletionTask = new FileDeletionTask( this.delService, null, delPath, null); this.delService.delete(deletionTask); } } } LocalResourceType srcType; if (type == AuxServiceFile.TypeEnum.STATIC) { srcType = LocalResourceType.FILE; } else if (type == AuxServiceFile.TypeEnum.ARCHIVE) { srcType = LocalResourceType.ARCHIVE; } else { throw new YarnRuntimeException( "Cannot unpack file of type " + type + " from remote-file-path:" + src + "for aux-service:" + ".\n"); } LocalResource scRsrc = LocalResource.newInstance( URL.fromURI(src.toUri()), srcType, LocalResourceVisibility.PRIVATE, scFileStatus.getLen(), scFileStatus.getModificationTime()); FSDownload download = new FSDownload(localLFS, null, conf, downloadDest, scRsrc, null); try { return download.call(); } catch (Exception ex) { throw new YarnRuntimeException( "Exception happend while downloading files " + "for aux-service:" + sName + " and remote-file-path:" + src + ".\n" + ex.getMessage()); } } AuxServices(AuxiliaryLocalPathHandler auxiliaryLocalPathHandler,
      Context nmContext, DeletionService deletionService); boolean isManifestEnabled(); Collection<AuxServiceRecord> getServiceRecords(); Map<String, ByteBuffer> getMetaData(); synchronized void reload(AuxServiceRecords services); @Override synchronized void serviceInit(Configuration conf); @Override synchronized void serviceStart(); @Override synchronized void serviceStop(); @Override void stateChanged(Service service); @Override void handle(AuxServicesEvent event); static final String NM_AUX_SERVICE_DIR; static final FsPermission NM_AUX_SERVICE_DIR_PERM; static final String CLASS_NAME; static final String SYSTEM_CLASSES; }
@Test (timeout = 15000) public void testReuseLocalizedAuxiliaryJar() throws Exception { File testJar = null; AuxServices aux = null; Configuration conf = new YarnConfiguration(); FileSystem fs = FileSystem.get(conf); String root = "target/LocalDir"; try { testJar = JarFinder.makeClassLoaderTestJar(this.getClass(), rootDir, "test-runjar.jar", 2048, ServiceB.class.getName(), LightService .class.getName()); Context mockContext = mock(Context.class); LocalDirsHandlerService mockDirsHandler = mock( LocalDirsHandlerService.class); Path rootAuxServiceDirPath = new Path(root, "nmAuxService"); when(mockDirsHandler.getLocalPathForWrite(anyString())).thenReturn( rootAuxServiceDirPath); when(mockContext.getLocalDirsHandler()).thenReturn(mockDirsHandler); aux = new AuxServices(MOCK_AUX_PATH_HANDLER, mockContext, MOCK_DEL_SERVICE); Path path = aux.maybeDownloadJars("ServiceB", ServiceB.class.getName(), testJar.getAbsolutePath(), AuxServiceFile.TypeEnum.STATIC, conf); path = aux.maybeDownloadJars("ServiceB", ServiceB.class.getName(), testJar.getAbsolutePath(), AuxServiceFile.TypeEnum.STATIC, conf); assertFalse("Failed to reuse the localized jar", path.toString().endsWith("/*")); } finally { if (testJar != null) { testJar.delete(); } if (fs.exists(new Path(root))) { fs.delete(new Path(root), true); } } }