AzkabanProjectLoader { public Map<String, ValidationReport> uploadProject(final Project project, final File archive, final String fileType, final User uploader, final Props additionalProps, final String uploaderIPAddr) throws ProjectManagerException, ExecutorManagerException { final long startUploadTimeStamp = System.currentTimeMillis(); String errorMessage = null; final Map<String, Object> eventData = new HashMap<>(); eventData.put("projectId", project.getId()); eventData.put("projectName", project.getName()); eventData.put("projectVersion", project.getVersion()); final long fileSize = FileUtils.sizeOf(archive); eventData.put("projectZipSize", fileSize); eventData.put("modifiedBy", uploader.getUserId()); eventData.put("uploaderIPAddress", uploaderIPAddr); log.info("Uploading files to " + project.getName()); final Map<String, ValidationReport> reports; File folder = null; final FlowLoader loader; try { folder = unzipProject(archive, fileType); final File startupDependencies = getStartupDependenciesFile(folder); final boolean isThinProject = startupDependencies.exists(); if (isThinProject){ eventData.put("zipType", "THIN_ZIP"); } else { eventData.put("zipType", "FAT_ZIP"); } reports = isThinProject ? this.archiveUnthinner.validateThinProject(project, folder, startupDependencies, additionalProps) : this.validatorUtils.validateProject(project, folder, additionalProps); if (reports.values().stream().anyMatch(r -> !r.getModifiedFiles().isEmpty() || !r.getRemovedFiles().isEmpty())) { updateProjectZip(archive, folder); } loader = this.flowLoaderFactory.createFlowLoader(folder); reports.put(DIRECTORY_FLOW_REPORT_KEY, loader.loadProjectFlow(project, folder)); if (!isReportStatusValid(reports, project)) { FlowLoaderUtils.cleanUpDir(folder); return reports; } final File startupDependenciesOrNull = isThinProject ? startupDependencies : null; persistProject(project, loader, archive, folder, startupDependenciesOrNull, uploader, uploaderIPAddr); if (isThinProject) { commonMetrics.markUploadThinProject(); } else { commonMetrics.markUploadFatProject(); } } catch (Exception e){ errorMessage = e.toString(); throw e; } finally { final long projectUploadTime = System.currentTimeMillis() - startUploadTimeStamp; if (projectUploadTime >= 0) { eventData.put("projectUploadTime", projectUploadTime); } else{ eventData.put("projectUploadTime", 0); } eventData.putIfAbsent("zipType", "null"); if (errorMessage == null){ eventData.put("projectEventStatus", "SUCCESS"); eventData.put("errorMessage", "null"); } else { eventData.put("projectEventStatus", "ERROR"); eventData.put("errorMessage", errorMessage); } project.fireEventListeners(ProjectEvent.create(project, azkaban.spi.EventType.PROJECT_UPLOADED, eventData)); FlowLoaderUtils.cleanUpDir(folder); } cleanUpProjectOldInstallations(project); return reports; } @Inject  AzkabanProjectLoader(final Props props, final CommonMetrics commonMetrics, final ProjectLoader projectLoader,
      final ProjectStorageManager projectStorageManager, final FlowLoaderFactory flowLoaderFactory,
      final ExecutorLoader executorLoader, final DatabaseOperator databaseOperator,
      final Storage storage, final ArchiveUnthinner archiveUnthinner,
      final ValidatorUtils validatorUtils); Map<String, ValidationReport> uploadProject(final Project project,
      final File archive, final String fileType, final User uploader, final Props additionalProps,
      final String uploaderIPAddr); ProjectFileHandler getProjectFile(final Project project, int version);  }
@Test public void uploadProjectFAT() throws ExecutorManagerException { when(this.projectLoader.getLatestProjectVersion(this.project)).thenReturn(this.VERSION); final URL resource = requireNonNull( getClass().getClassLoader().getResource("sample_flow_01.zip")); final File projectZipFile = new File(resource.getPath()); final User uploader = new User("test_user"); final ExecutableFlow runningFlow = new ExecutableFlow(this.project, new Flow("x")); runningFlow.setVersion(this.VERSION); when(this.executorLoader.fetchUnfinishedFlowsMetadata()) .thenReturn(ImmutableMap.of(-1, new Pair<>(null, runningFlow))); this.project.setVersion(this.VERSION); checkValidationReport(this.azkabanProjectLoader .uploadProject(this.project, projectZipFile, "zip", uploader, null, IPv4)); verify(this.projectStorageManager) .uploadProject(this.project, this.VERSION + 1, projectZipFile, null, uploader, IPv4); verify(this.projectLoader).cleanOlderProjectVersion(this.project.getId(), this.VERSION - 3, Arrays.asList(this.VERSION)); verify(this.archiveUnthinner, never()).validateThinProject(any(), any(), any(), any()); }
@Test public void uploadProjectValidatorRemovedFileTHIN() throws Exception { final User uploader = new User("test_user"); final File projectZipFile = setupTestTHIN(); doAnswer((Answer<Map<String, ValidationReport>>) invocation -> { File libFolder = new File((File) invocation.getArguments()[1], "lib"); File randomLib = FileUtils.listFiles(libFolder, null, false).iterator().next(); randomLib.delete(); ValidationReport r = new ValidationReport(); r.addRemovedFiles(new HashSet(Arrays.asList(randomLib))); Map<String, ValidationReport> resultingReports = new HashMap(); resultingReports.put("somevalidator", r); return resultingReports; }).when(this.archiveUnthinner).validateThinProject(any(), any(), any(), any()); doAnswer(invocation -> { File zipFileBeingUploaded = (File) invocation.getArguments()[2]; File unzippedFinalProjFolder = TEMP_DIR.newFolder("finalproj"); Utils.unzip(new ZipFile(zipFileBeingUploaded), unzippedFinalProjFolder); File libFolder = new File(unzippedFinalProjFolder, "lib"); assertTrue(!libFolder.exists() || FileUtils.listFiles(libFolder, null, false).size() == 0); assertTrue(((File) invocation.getArguments()[3]).exists()); return null; }).when(this.projectStorageManager) .uploadProject(any(Project.class), anyInt(), any(File.class), any(File.class), any(User.class), anyString()); this.project.setVersion(this.VERSION); this.azkabanProjectLoader .uploadProject(this.project, projectZipFile, "zip", uploader, null, IPv6); verify(this.archiveUnthinner).validateThinProject(any(), any(), any(), any()); }
@Test public void uploadProjectValidatorModifiedFileTHIN() throws Exception { final String writtenModifiedString = "HELLOEVERYONE"; final User uploader = new User("test_user"); final File projectZipFile = setupTestTHIN(); doAnswer((Answer<Map<String, ValidationReport>>) invocation -> { File libFolder = new File((File) invocation.getArguments()[1], "lib"); File randomLib = FileUtils.listFiles(libFolder, null, false).iterator().next(); FileUtils.writeStringToFile(randomLib, writtenModifiedString); ValidationReport r = new ValidationReport(); r.addModifiedFiles(new HashSet(Arrays.asList(randomLib))); Map<String, ValidationReport> resultingReports = new HashMap(); resultingReports.put("somevalidator", r); return resultingReports; }).when(this.archiveUnthinner).validateThinProject(any(), any(), any(), any()); doAnswer(invocation -> { File zipFileBeingUploaded = (File) invocation.getArguments()[2]; File unzippedFinalProjFolder = TEMP_DIR.newFolder("finalproj"); Utils.unzip(new ZipFile(zipFileBeingUploaded), unzippedFinalProjFolder); File libFolder = new File(unzippedFinalProjFolder, "lib"); File randomLib = FileUtils.listFiles(libFolder, null, false).iterator().next(); assertEquals(writtenModifiedString, FileUtils.readFileToString(randomLib)); assertTrue(((File) invocation.getArguments()[3]).exists()); return null; }).when(this.projectStorageManager) .uploadProject(any(Project.class), anyInt(), any(File.class), any(File.class), any(User.class), anyString()); this.project.setVersion(this.VERSION); this.azkabanProjectLoader .uploadProject(this.project, projectZipFile, "zip", uploader, null, IPv6); verify(this.archiveUnthinner).validateThinProject(any(), any(), any(), any()); }
@Test public void uploadProjectTHIN() throws Exception { final User uploader = new User("test_user"); final File projectZipFile = setupTestTHIN(); doAnswer((Answer<Map<String, ValidationReport>>) invocation -> { File projectFolder = (File) invocation.getArguments()[1]; File startupDependenciesFile = (File) invocation.getArguments()[2]; assertEquals(ThinArchiveUtils.getStartupDependenciesFile(projectFolder), startupDependenciesFile); return new HashMap(); }).when(this.archiveUnthinner).validateThinProject(any(), any(), any(), any()); this.project.setVersion(this.VERSION); checkValidationReport(this.azkabanProjectLoader .uploadProject(this.project, projectZipFile, "zip", uploader, null, IPv6)); verify(this.projectStorageManager) .uploadProject(eq(this.project), eq(this.VERSION + 1), eq(projectZipFile), any(File.class), eq(uploader), anyString()); verify(this.projectLoader).cleanOlderProjectVersion(this.project.getId(), this.VERSION - 3, Arrays.asList(this.VERSION)); verify(this.archiveUnthinner).validateThinProject(any(), any(), any(), any()); }
@Test public void uploadProjectWithYamlFilesFAT() throws Exception { final File projectZipFile = ExecutionsTestUtil.getFlowFile(BASIC_FLOW_YAML_DIR, PROJECT_ZIP); final int flowVersion = 0; final User uploader = new User("test_user"); when(this.projectLoader.getLatestProjectVersion(this.project)).thenReturn(this.VERSION); when(this.projectLoader.getLatestFlowVersion(this.ID, this.VERSION, BASIC_FLOW_FILE)) .thenReturn(flowVersion); checkValidationReport(this.azkabanProjectLoader .uploadProject(this.project, projectZipFile, "zip", uploader, null, IPv6)); verify(this.projectStorageManager) .uploadProject(this.project, this.VERSION + 1, projectZipFile, null, uploader, IPv6); verify(this.projectLoader) .uploadFlowFile(eq(this.ID), eq(this.VERSION + 1), any(File.class), eq(flowVersion + 1)); }