LocalStoreMonitor implements Monitor { @Override public void monitor() throws Exception { File localStoreDir = new File(config.getLocalStoreBaseDir()); Preconditions.checkState(localStoreDir.isDirectory(), String.format("LocalStoreDir: %s is not a directory", localStoreDir.getAbsolutePath())); String localHostName = InetAddress.getLocalHost().getHostName(); for (JobInstance jobInstance : getHostAffinityEnabledJobs(localStoreDir)) { File jobDir = new File(localStoreDir, String.format("%s-%s", jobInstance.getJobName(), jobInstance.getJobId())); try { JobStatus jobStatus = jobsClient.getJobStatus(jobInstance); LOG.info("Job: {} has the status: {}.", jobInstance, jobStatus); for (Task task : jobsClient.getTasks(jobInstance)) { LOG.info("Evaluating stores for task: {}", task); for (String storeName : jobDir.list(DirectoryFileFilter.DIRECTORY)) { if (jobStatus.hasBeenStarted() && task.getStoreNames().contains(storeName) && task.getPreferredHost().equals(localHostName)) { LOG.info(String.format("Local store: %s is actively used by the task: %s.", storeName, task.getTaskName())); } else { LOG.info(String.format("Local store: %s not used by the task: %s.", storeName, task.getTaskName())); markSweepTaskStore(new StorageManagerUtil().getTaskStoreDir(jobDir, storeName, new TaskName(task.getTaskName()), TaskMode.Active)); } } } } catch (Exception ex) { localStoreMonitorMetrics.failedStoreDeletionAttempts.inc(); if (!config.getIgnoreFailures()) { throw ex; } LOG.warn("Config: {} turned on, failures will be ignored. Local store cleanup for job: {} resulted in exception: {}.", new Object[]{LocalStoreMonitorConfig.CONFIG_IGNORE_FAILURES, jobInstance, ex}); } } } LocalStoreMonitor(LocalStoreMonitorConfig config,
                           LocalStoreMonitorMetrics localStoreMonitorMetrics,
                           JobsClient jobsClient); @Override void monitor();  }
@Test public void shouldDeleteLocalTaskStoreWhenItHasNoOffsetFile() throws Exception { localStoreMonitor.monitor(); assertTrue("Task store directory should not exist.", !taskStoreDir.exists()); assertEquals(taskStoreSize, localStoreMonitorMetrics.diskSpaceFreedInBytes.getCount()); assertEquals(1, localStoreMonitorMetrics.noOfDeletedTaskPartitionStores.getCount()); }
@Test public void shouldDeleteLocalStoreWhenLastModifiedTimeOfOffsetFileIsGreaterThanOffsetTTL() throws Exception { File offsetFile = createOffsetFile(taskStoreDir); offsetFile.setLastModified(0); localStoreMonitor.monitor(); assertTrue("Offset file should not exist.", !offsetFile.exists()); assertEquals(0, localStoreMonitorMetrics.diskSpaceFreedInBytes.getCount()); }
@Test public void shouldDeleteInActiveLocalStoresOfTheJob() throws Exception { File inActiveStoreDir = new File(jobDir, "inActiveStore"); FileUtils.forceMkdir(inActiveStoreDir); File inActiveTaskDir = new File(inActiveStoreDir, "test-task"); FileUtils.forceMkdir(inActiveTaskDir); long inActiveTaskDirSize = inActiveTaskDir.getTotalSpace(); localStoreMonitor.monitor(); assertTrue("Inactive task store directory should not exist.", !inActiveTaskDir.exists()); assertEquals(taskStoreSize + inActiveTaskDirSize, localStoreMonitorMetrics.diskSpaceFreedInBytes.getCount()); assertEquals(2, localStoreMonitorMetrics.noOfDeletedTaskPartitionStores.getCount()); FileUtils.deleteDirectory(inActiveStoreDir); }
@Test public void shouldDoNothingWhenLastModifiedTimeOfOffsetFileIsLessThanOffsetTTL() throws Exception { File offsetFile = createOffsetFile(taskStoreDir); localStoreMonitor.monitor(); assertTrue("Offset file should exist.", offsetFile.exists()); assertEquals(0, localStoreMonitorMetrics.diskSpaceFreedInBytes.getCount()); }
@Test public void shouldDoNothingWhenTheJobIsRunning() throws Exception { Mockito.when(jobsClientMock.getJobStatus(Mockito.any())).thenReturn(JobStatus.STARTED); File offsetFile = createOffsetFile(taskStoreDir); localStoreMonitor.monitor(); assertTrue("Offset file should exist.", offsetFile.exists()); assertEquals(0, localStoreMonitorMetrics.diskSpaceFreedInBytes.getCount()); }
@Test public void shouldDeleteTaskStoreWhenTaskPreferredStoreIsNotLocalHost() throws Exception { Task task = new Task("notLocalHost", "test-task", "0", new ArrayList<>(), ImmutableList.of("test-store")); Mockito.when(jobsClientMock.getTasks(Mockito.any())).thenReturn(ImmutableList.of(task)); localStoreMonitor.monitor(); assertTrue("Task store directory should not exist.", !taskStoreDir.exists()); assertEquals(taskStoreSize, localStoreMonitorMetrics.diskSpaceFreedInBytes.getCount()); assertEquals(1, localStoreMonitorMetrics.noOfDeletedTaskPartitionStores.getCount()); }
@Test public void shouldContinueLocalStoreCleanUpAfterFailureToCleanUpStoreOfAJob() throws Exception { File testFailingJobDir = new File(localStoreDir, "test-jobName-jobId-1"); File testFailingTaskStoreDir = new File(new File(testFailingJobDir, "test-store"), "test-task"); FileUtils.forceMkdir(testFailingTaskStoreDir); Mockito.doThrow(new RuntimeException("Dummy exception message.")) .when(jobsClientMock) .getTasks(new JobInstance("test-jobName", "jobId-1")); Task task = new Task("notLocalHost", "test-task", "0", new ArrayList<>(), ImmutableList.of("test-store")); Mockito.when(jobsClientMock.getTasks(new JobInstance("test-jobName", "jobId"))).thenReturn(ImmutableList.of(task)); Map<String, String> configMap = new HashMap<>(config); configMap.put(LocalStoreMonitorConfig.CONFIG_IGNORE_FAILURES, "true"); LocalStoreMonitor localStoreMonitor = new LocalStoreMonitor(new LocalStoreMonitorConfig(new MapConfig(configMap)), localStoreMonitorMetrics, jobsClientMock); localStoreMonitor.monitor(); assertTrue("Task store directory should not exist.", !taskStoreDir.exists()); FileUtils.deleteDirectory(testFailingJobDir); }