TestEntityGroupFSTimelineStore.java
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.yarn.server.timeline;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileContextTestHelper;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
import org.apache.hadoop.metrics2.lib.MutableStat;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.ApplicationClassLoader;
import org.apache.hadoop.util.JarFinder;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.timeline.TimelineEntities;
import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.timeline.EntityGroupFSTimelineStore.AppState;
import org.apache.hadoop.yarn.server.timeline.TimelineReader.Field;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.util.MinimalPrettyPrinter;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.type.TypeFactory;
import com.fasterxml.jackson.module.jaxb.JaxbAnnotationIntrospector;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class TestEntityGroupFSTimelineStore extends TimelineStoreTestUtils {
private static final String SAMPLE_APP_PREFIX_CACHE_TEST = "1234_000";
private static final int CACHE_TEST_CACHE_SIZE = 5;
private static final String TEST_SUMMARY_LOG_FILE_NAME
= EntityGroupFSTimelineStore.SUMMARY_LOG_PREFIX + "test";
private static final String TEST_DOMAIN_LOG_FILE_NAME
= EntityGroupFSTimelineStore.DOMAIN_LOG_PREFIX + "test";
private static final Path TEST_ROOT_DIR
= new Path(System.getProperty("test.build.data",
System.getProperty("java.io.tmpdir")),
TestEntityGroupFSTimelineStore.class.getSimpleName());
private static Configuration config = new YarnConfiguration();
private static MiniDFSCluster hdfsCluster;
private static FileSystem fs;
private static FileContext fc;
private static FileContextTestHelper fileContextTestHelper =
new FileContextTestHelper("/tmp/TestEntityGroupFSTimelineStore");
private static List<ApplicationId> sampleAppIds;
private static ApplicationId mainTestAppId;
private static Path mainTestAppDirPath;
private static Path testDoneDirPath;
private static Path testActiveDirPath;
private static String mainEntityLogFileName;
private EntityGroupFSTimelineStore store;
private TimelineEntity entityNew;
private File rootDir;
private File testJar;
@BeforeAll
public static void setupClass() throws Exception {
config.setBoolean(YarnConfiguration.TIMELINE_SERVICE_TTL_ENABLE, false);
config.set(
YarnConfiguration
.TIMELINE_SERVICE_ENTITYGROUP_FS_STORE_SUMMARY_ENTITY_TYPES,
"YARN_APPLICATION,YARN_APPLICATION_ATTEMPT,YARN_CONTAINER");
config.setInt(
YarnConfiguration.TIMELINE_SERVICE_ENTITYGROUP_FS_STORE_APP_CACHE_SIZE,
CACHE_TEST_CACHE_SIZE);
config.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, TEST_ROOT_DIR.toString());
HdfsConfiguration hdfsConfig = new HdfsConfiguration();
hdfsCluster
= new MiniDFSCluster.Builder(hdfsConfig).numDataNodes(1).build();
fs = hdfsCluster.getFileSystem();
fc = FileContext.getFileContext(hdfsCluster.getURI(0), config);
sampleAppIds = new ArrayList<>(CACHE_TEST_CACHE_SIZE + 1);
for (int i = 0; i < CACHE_TEST_CACHE_SIZE + 1; i++) {
ApplicationId appId = ApplicationId.fromString(
ConverterUtils.APPLICATION_PREFIX + "_" + SAMPLE_APP_PREFIX_CACHE_TEST
+ i);
sampleAppIds.add(appId);
}
testActiveDirPath = getTestRootPath("active");
// Among all sample applicationIds, choose the first one for most of the
// tests.
mainTestAppId = sampleAppIds.get(0);
mainTestAppDirPath = new Path(testActiveDirPath, mainTestAppId.toString());
mainEntityLogFileName = EntityGroupFSTimelineStore.ENTITY_LOG_PREFIX
+ EntityGroupPlugInForTest.getStandardTimelineGroupId(mainTestAppId);
testDoneDirPath = getTestRootPath("done");
config.set(YarnConfiguration.TIMELINE_SERVICE_ENTITYGROUP_FS_STORE_DONE_DIR,
testDoneDirPath.toString());
config.set(
YarnConfiguration.TIMELINE_SERVICE_ENTITYGROUP_FS_STORE_ACTIVE_DIR,
testActiveDirPath.toString());
}
@BeforeEach
public void setup(TestInfo testInfo) throws Exception {
for (ApplicationId appId : sampleAppIds) {
Path attemotDirPath =
new Path(new Path(testActiveDirPath, appId.toString()),
getAttemptDirName(appId));
createTestFiles(appId, attemotDirPath);
}
store = new EntityGroupFSTimelineStore();
if (testInfo.getTestMethod().get().getName().contains("Plugin")) {
rootDir = GenericTestUtils.getTestDir(getClass()
.getSimpleName());
if (!rootDir.exists()) {
rootDir.mkdirs();
}
testJar = null;
testJar = JarFinder.makeClassLoaderTestJar(this.getClass(), rootDir,
"test-runjar.jar", 2048,
EntityGroupPlugInForTest.class.getName());
config.set(
YarnConfiguration.TIMELINE_SERVICE_ENTITY_GROUP_PLUGIN_CLASSPATH,
testJar.getAbsolutePath());
// add "-org.apache.hadoop." as system classes
String systemClasses = "-org.apache.hadoop." + "," +
ApplicationClassLoader.SYSTEM_CLASSES_DEFAULT;
config.set(
YarnConfiguration.TIMELINE_SERVICE_ENTITY_GROUP_PLUGIN_SYSTEM_CLASSES,
systemClasses);
config.set(YarnConfiguration.TIMELINE_SERVICE_ENTITY_GROUP_PLUGIN_CLASSES,
EntityGroupPlugInForTest.class.getName());
}
store.init(config);
store.setFs(fs);
store.start();
}
@AfterEach
public void tearDown() throws Exception {
store.stop();
for (ApplicationId appId : sampleAppIds) {
fs.delete(new Path(testActiveDirPath,appId.toString()), true);
}
if (testJar != null) {
testJar.delete();
rootDir.delete();
}
}
@AfterAll
public static void tearDownClass() throws Exception {
hdfsCluster.shutdown();
FileContext fileContext = FileContext.getLocalFSFileContext();
fileContext.delete(new Path(
config.get(YarnConfiguration.TIMELINE_SERVICE_LEVELDB_PATH)), true);
}
@Test
void testAppLogsScanLogs() throws Exception {
EntityGroupFSTimelineStore.AppLogs appLogs =
store.new AppLogs(mainTestAppId, mainTestAppDirPath,
AppState.COMPLETED);
appLogs.scanForLogs();
List<LogInfo> summaryLogs = appLogs.getSummaryLogs();
List<LogInfo> detailLogs = appLogs.getDetailLogs();
assertEquals(2, summaryLogs.size());
assertEquals(1, detailLogs.size());
for (LogInfo log : summaryLogs) {
String fileName = log.getFilename();
assertTrue(fileName.equals(TEST_SUMMARY_LOG_FILE_NAME)
|| fileName.equals(TEST_DOMAIN_LOG_FILE_NAME));
}
for (LogInfo log : detailLogs) {
String fileName = log.getFilename();
assertEquals(fileName, mainEntityLogFileName);
}
}
@Test
void testAppLogsDomainLogLastlyScanned() throws Exception {
EntityGroupFSTimelineStore.AppLogs appLogs =
store.new AppLogs(mainTestAppId, mainTestAppDirPath,
AppState.COMPLETED);
Path attemptDirPath = new Path(new Path(testActiveDirPath,
mainTestAppId.toString()),
getAttemptDirName(mainTestAppId));
//Delete the domain log from AppDirPath so first scan won't find it
fs.delete(new Path(attemptDirPath, TEST_DOMAIN_LOG_FILE_NAME), false);
appLogs.scanForLogs();
List<LogInfo> summaryLogs = appLogs.getSummaryLogs();
assertEquals(1, summaryLogs.size());
assertEquals(TEST_SUMMARY_LOG_FILE_NAME, summaryLogs.get(0).getFilename());
//Generate the domain log
FSDataOutputStream out = fs.create(
new Path(attemptDirPath, TEST_DOMAIN_LOG_FILE_NAME));
out.close();
appLogs.scanForLogs();
assertEquals(2, summaryLogs.size());
assertEquals(TEST_DOMAIN_LOG_FILE_NAME, summaryLogs.get(0).getFilename());
}
@Test
void testMoveToDone() throws Exception {
EntityGroupFSTimelineStore.AppLogs appLogs =
store.new AppLogs(mainTestAppId, mainTestAppDirPath,
AppState.COMPLETED);
Path pathBefore = appLogs.getAppDirPath();
appLogs.moveToDone();
Path pathAfter = appLogs.getAppDirPath();
assertNotEquals(pathBefore, pathAfter);
assertTrue(pathAfter.toString().contains(testDoneDirPath.toString()));
fs.delete(pathAfter, true);
}
@Test
void testParseSummaryLogs() throws Exception {
TimelineDataManager tdm = PluginStoreTestUtils.getTdmWithMemStore(config);
MutableCounterLong scanned = store.metrics.getEntitiesReadToSummary();
long beforeScan = scanned.value();
EntityGroupFSTimelineStore.AppLogs appLogs =
store.new AppLogs(mainTestAppId, mainTestAppDirPath,
AppState.COMPLETED);
appLogs.scanForLogs();
appLogs.parseSummaryLogs(tdm);
PluginStoreTestUtils.verifyTestEntities(tdm);
assertEquals(beforeScan + 2L, scanned.value());
}
@Test
void testWithAnonymousUser() throws Exception {
try {
TimelineDataManager tdm = PluginStoreTestUtils.getTdmWithMemStore(config);
EntityGroupFSTimelineStore.AppLogs appLogs =
store.new AppLogs(mainTestAppId, mainTestAppDirPath,
AppState.COMPLETED);
FileStatus fileStatus = mock(FileStatus.class);
when(fileStatus.getOwner()).thenReturn(null);
appLogs.scanForLogs();
appLogs.parseSummaryLogs(tdm);
PluginStoreTestUtils.verifyTestEntities(tdm);
} catch (IllegalArgumentException ie) {
fail("No exception needs to be thrown as anonymous user is configured");
}
}
@Test
void testCleanLogs() throws Exception {
// Create test dirs and files
// Irrelevant file, should not be reclaimed
String appDirName = mainTestAppId.toString();
String attemptDirName = ApplicationAttemptId.appAttemptIdStrPrefix
+ appDirName + "_1";
Path irrelevantFilePath = new Path(
testDoneDirPath, "irrelevant.log");
FSDataOutputStream stream = fs.create(irrelevantFilePath);
stream.close();
// Irrelevant directory, should not be reclaimed
Path irrelevantDirPath = new Path(testDoneDirPath, "irrelevant");
fs.mkdirs(irrelevantDirPath);
Path doneAppHomeDir = new Path(new Path(new Path(testDoneDirPath,
Long.toString(mainTestAppId.getClusterTimestamp())), "0000"), "001");
// First application, untouched after creation
Path appDirClean = new Path(doneAppHomeDir, appDirName);
Path attemptDirClean = new Path(appDirClean, attemptDirName);
fs.mkdirs(attemptDirClean);
Path filePath = new Path(attemptDirClean, "test.log");
stream = fs.create(filePath);
stream.close();
// Second application, one file touched after creation
Path appDirHoldByFile = new Path(doneAppHomeDir, appDirName + "1");
Path attemptDirHoldByFile
= new Path(appDirHoldByFile, attemptDirName);
fs.mkdirs(attemptDirHoldByFile);
Path filePathHold = new Path(attemptDirHoldByFile, "test1.log");
stream = fs.create(filePathHold);
stream.close();
// Third application, one dir touched after creation
Path appDirHoldByDir = new Path(doneAppHomeDir, appDirName + "2");
Path attemptDirHoldByDir = new Path(appDirHoldByDir, attemptDirName);
fs.mkdirs(attemptDirHoldByDir);
Path dirPathHold = new Path(attemptDirHoldByDir, "hold");
fs.mkdirs(dirPathHold);
// Fourth application, empty dirs
Path appDirEmpty = new Path(doneAppHomeDir, appDirName + "3");
Path attemptDirEmpty = new Path(appDirEmpty, attemptDirName);
fs.mkdirs(attemptDirEmpty);
Path dirPathEmpty = new Path(attemptDirEmpty, "empty");
fs.mkdirs(dirPathEmpty);
// Should retain all logs after this run
MutableCounterLong dirsCleaned = store.metrics.getLogsDirsCleaned();
long before = dirsCleaned.value();
store.cleanLogs(testDoneDirPath, 10000);
assertTrue(fs.exists(irrelevantDirPath));
assertTrue(fs.exists(irrelevantFilePath));
assertTrue(fs.exists(filePath));
assertTrue(fs.exists(filePathHold));
assertTrue(fs.exists(dirPathHold));
assertTrue(fs.exists(dirPathEmpty));
// Make sure the created dir is old enough
Thread.sleep(2000);
// Touch the second application
stream = fs.append(filePathHold);
stream.writeBytes("append");
stream.close();
// Touch the third application by creating a new dir
fs.mkdirs(new Path(dirPathHold, "holdByMe"));
store.cleanLogs(testDoneDirPath, 1000);
// Verification after the second cleaner call
assertTrue(fs.exists(irrelevantDirPath));
assertTrue(fs.exists(irrelevantFilePath));
assertTrue(fs.exists(filePathHold));
assertTrue(fs.exists(dirPathHold));
assertTrue(fs.exists(doneAppHomeDir));
// appDirClean and appDirEmpty should be cleaned up
assertFalse(fs.exists(appDirClean));
assertFalse(fs.exists(appDirEmpty));
assertEquals(before + 2L, dirsCleaned.value());
}
@Test
void testCleanBuckets() throws Exception {
// ClusterTimeStampDir with App Log Dirs
Path clusterTimeStampDir1 = new Path(testDoneDirPath,
Long.toString(sampleAppIds.get(0).getClusterTimestamp()));
Path appDir1 = new Path(new Path(new Path(
clusterTimeStampDir1, "0000"), "000"), sampleAppIds.get(0).toString());
Path appDir2 = new Path(new Path(new Path(
clusterTimeStampDir1, "0000"), "001"), sampleAppIds.get(1).toString());
Path appDir3 = new Path(new Path(new Path(
clusterTimeStampDir1, "0000"), "002"), sampleAppIds.get(2).toString());
Path appDir4 = new Path(new Path(new Path(
clusterTimeStampDir1, "0001"), "000"), sampleAppIds.get(3).toString());
// ClusterTimeStampDir with no App Log Dirs
Path clusterTimeStampDir2 = new Path(testDoneDirPath, "1235");
// Irrevelant ClusterTimeStampDir
Path clusterTimeStampDir3 = new Path(testDoneDirPath, "irrevelant");
Path appDir5 = new Path(new Path(new Path(
clusterTimeStampDir3, "0000"), "000"), sampleAppIds.get(4).toString());
fs.mkdirs(appDir1);
fs.mkdirs(appDir2);
fs.mkdirs(appDir3);
fs.mkdirs(appDir4);
fs.mkdirs(clusterTimeStampDir2);
fs.mkdirs(appDir5);
Thread.sleep(2000);
store.cleanLogs(testDoneDirPath, 1000);
// ClusterTimeStampDir will be removed only if no App Log Dir Present
assertTrue(fs.exists(clusterTimeStampDir1));
assertFalse(fs.exists(appDir1));
assertFalse(fs.exists(appDir2));
assertFalse(fs.exists(appDir3));
assertFalse(fs.exists(appDir4));
assertFalse(fs.exists(clusterTimeStampDir2));
assertTrue(fs.exists(appDir5));
store.cleanLogs(testDoneDirPath, 1000);
assertFalse(fs.exists(clusterTimeStampDir1));
}
@Test
void testNullCheckGetEntityTimelines() throws Exception {
try {
store.getEntityTimelines("YARN_APPLICATION", null, null, null, null,
null);
} catch (NullPointerException e) {
fail("NPE when getEntityTimelines called with Null EntityIds");
}
}
@Test
void testPluginRead() throws Exception {
// Verify precondition
assertEquals(EntityGroupPlugInForTest.class.getName(),
store.getConfig().get(
YarnConfiguration.TIMELINE_SERVICE_ENTITY_GROUP_PLUGIN_CLASSES));
List<TimelineEntityGroupPlugin> currPlugins = store.getPlugins();
for (TimelineEntityGroupPlugin plugin : currPlugins) {
ClassLoader pluginClassLoader = plugin.getClass().getClassLoader();
assertTrue(pluginClassLoader instanceof ApplicationClassLoader,
"Should set up ApplicationClassLoader");
URL[] paths = ((URLClassLoader) pluginClassLoader).getURLs();
boolean foundJAR = false;
for (URL path : paths) {
if (path.toString().contains(testJar.getAbsolutePath())) {
foundJAR = true;
}
}
assertTrue(foundJAR, "Not found path " + testJar.getAbsolutePath()
+ " for plugin " + plugin.getClass().getName());
}
// Load data and cache item, prepare timeline store by making a cache item
EntityGroupFSTimelineStore.AppLogs appLogs =
store.new AppLogs(mainTestAppId, mainTestAppDirPath,
AppState.COMPLETED);
EntityCacheItem cacheItem = new EntityCacheItem(
EntityGroupPlugInForTest.getStandardTimelineGroupId(mainTestAppId),
config);
cacheItem.setAppLogs(appLogs);
store.setCachedLogs(
EntityGroupPlugInForTest.getStandardTimelineGroupId(mainTestAppId),
cacheItem);
MutableCounterLong detailLogEntityRead =
store.metrics.getGetEntityToDetailOps();
MutableStat cacheRefresh = store.metrics.getCacheRefresh();
long numEntityReadBefore = detailLogEntityRead.value();
long cacheRefreshBefore = cacheRefresh.lastStat().numSamples();
// Generate TDM
TimelineDataManager tdm
= PluginStoreTestUtils.getTdmWithStore(config, store);
// Verify single entity read
TimelineEntity entity3 = tdm.getEntity("type_3", mainTestAppId.toString(),
EnumSet.allOf(TimelineReader.Field.class),
UserGroupInformation.getLoginUser());
assertNotNull(entity3);
assertEquals(entityNew.getStartTime(), entity3.getStartTime());
// Verify multiple entities read
NameValuePair primaryFilter = new NameValuePair(
EntityGroupPlugInForTest.APP_ID_FILTER_NAME, mainTestAppId.toString());
TimelineEntities entities = tdm.getEntities("type_3", primaryFilter, null,
null, null, null, null, null, EnumSet.allOf(TimelineReader.Field.class),
UserGroupInformation.getLoginUser());
assertEquals(1, entities.getEntities().size());
for (TimelineEntity entity : entities.getEntities()) {
assertEquals(entityNew.getStartTime(), entity.getStartTime());
}
// Verify metrics
assertEquals(numEntityReadBefore + 2L, detailLogEntityRead.value());
assertEquals(cacheRefreshBefore + 1L, cacheRefresh.lastStat().numSamples());
}
@Test
void testSummaryRead() throws Exception {
// Load data
EntityGroupFSTimelineStore.AppLogs appLogs =
store.new AppLogs(mainTestAppId, mainTestAppDirPath,
AppState.COMPLETED);
MutableCounterLong summaryLogEntityRead
= store.metrics.getGetEntityToSummaryOps();
long numEntityReadBefore = summaryLogEntityRead.value();
TimelineDataManager tdm
= PluginStoreTestUtils.getTdmWithStore(config, store);
appLogs.scanForLogs();
appLogs.parseSummaryLogs(tdm);
// Verify single entity read
PluginStoreTestUtils.verifyTestEntities(tdm);
// Verify multiple entities read
TimelineEntities entities = tdm.getEntities("type_1", null, null, null,
null, null, null, null, EnumSet.allOf(TimelineReader.Field.class),
UserGroupInformation.getLoginUser());
assertThat(entities.getEntities()).hasSize(1);
for (TimelineEntity entity : entities.getEntities()) {
assertEquals((Long) 123L, entity.getStartTime());
}
// Verify metrics
assertEquals(numEntityReadBefore + 5L, summaryLogEntityRead.value());
}
@Test
void testGetEntityPluginRead() throws Exception {
EntityGroupFSTimelineStore store = null;
ApplicationId appId =
ApplicationId.fromString("application_1501509265053_0001");
String user = UserGroupInformation.getCurrentUser().getShortUserName();
Path userBase = new Path(testActiveDirPath, user);
Path userAppRoot = new Path(userBase, appId.toString());
Path attemotDirPath = new Path(userAppRoot, getAttemptDirName(appId));
try {
store = createAndStartTimelineStore(AppState.ACTIVE);
String logFileName = EntityGroupFSTimelineStore.ENTITY_LOG_PREFIX
+ EntityGroupPlugInForTest.getStandardTimelineGroupId(appId);
createTestFiles(appId, attemotDirPath, logFileName);
TimelineEntity entity = store.getEntity(entityNew.getEntityId(),
entityNew.getEntityType(), EnumSet.allOf(Field.class));
assertNotNull(entity);
assertEquals(entityNew.getEntityId(), entity.getEntityId());
assertEquals(entityNew.getEntityType(), entity.getEntityType());
} finally {
if (store != null) {
store.stop();
}
fs.delete(userBase, true);
}
}
@Test
void testScanActiveLogsWithInvalidFile() throws Exception {
Path invalidFile = new Path(testActiveDirPath, "invalidfile");
try {
if (!fs.exists(invalidFile)) {
fs.createNewFile(invalidFile);
}
store.scanActiveLogs();
} catch (StackOverflowError error) {
fail("EntityLogScanner crashed with StackOverflowError");
} finally {
if (fs.exists(invalidFile)) {
fs.delete(invalidFile, false);
}
}
}
@Test
void testScanActiveLogsAndMoveToDonePluginRead() throws Exception {
EntityGroupFSTimelineStore store = null;
ApplicationId appId =
ApplicationId.fromString("application_1501509265053_0002");
String user = UserGroupInformation.getCurrentUser().getShortUserName();
Path userBase = new Path(testActiveDirPath, user);
Path userAppRoot = new Path(userBase, appId.toString());
Path attemotDirPath = new Path(userAppRoot, getAttemptDirName(appId));
try {
store = createAndStartTimelineStore(AppState.COMPLETED);
String logFileName = EntityGroupFSTimelineStore.ENTITY_LOG_PREFIX
+ EntityGroupPlugInForTest.getStandardTimelineGroupId(appId);
createTestFiles(appId, attemotDirPath, logFileName);
store.scanActiveLogs();
TimelineEntity entity = store.getEntity(entityNew.getEntityId(),
entityNew.getEntityType(), EnumSet.allOf(Field.class));
assertNotNull(entity);
assertEquals(entityNew.getEntityId(), entity.getEntityId());
assertEquals(entityNew.getEntityType(), entity.getEntityType());
} finally {
if (store != null) {
store.stop();
}
fs.delete(userBase, true);
}
}
// TestTimelineStore to validate the put entities call
static class TestTimelineStore extends LeveldbTimelineStore {
static final AtomicInteger ENTITIES_COUNT = new AtomicInteger(0);
TestTimelineStore() {
super();
}
@Override
public TimelinePutResponse put(TimelineEntities entities) {
ENTITIES_COUNT.getAndAdd(entities.getEntities().size());
return new TimelinePutResponse();
}
public static int getEntitiesCount() {
return ENTITIES_COUNT.get();
}
}
@Test
void testIfAnyDuplicateEntities() throws Exception {
// Create an application with some entities
ApplicationId appId =
ApplicationId.fromString("application_1501509265053_0002");
String user = UserGroupInformation.getCurrentUser().getShortUserName();
Path activeDirPath = getTestRootPath("active1");
Path doneDirPath = getTestRootPath("done1");
Path userBase = new Path(activeDirPath, user);
Path userAppRoot = new Path(userBase, appId.toString());
Path attemptDirPath = new Path(userAppRoot, getAttemptDirName(appId));
String logFileName = EntityGroupFSTimelineStore.ENTITY_LOG_PREFIX
+ EntityGroupPlugInForTest.getStandardTimelineGroupId(appId);
createTestFiles(appId, attemptDirPath, logFileName);
// stop the default store before creating new store to get the lock
store.stop();
EntityGroupFSTimelineStore newStore = new EntityGroupFSTimelineStore() {
@Override
protected AppState getAppState(ApplicationId appId) throws IOException {
return AppState.ACTIVE;
}
};
try {
// Start ATS with TestTimelineStore
Configuration newConfig = new YarnConfiguration(config);
newConfig.set(YarnConfiguration.TIMELINE_SERVICE_ENTITYGROUP_FS_STORE_SUMMARY_STORE,
TestTimelineStore.class.getName());
newConfig.set(YarnConfiguration.TIMELINE_SERVICE_ENTITYGROUP_FS_STORE_DONE_DIR,
doneDirPath.toString());
newConfig.set(
YarnConfiguration.TIMELINE_SERVICE_ENTITYGROUP_FS_STORE_ACTIVE_DIR,
activeDirPath.toString());
newStore.init(newConfig);
newStore.setFs(fs);
newStore.start();
// Validate if the initial entities count are correct
newStore.scanActiveLogs();
GenericTestUtils.waitFor(new Supplier<Boolean>() {
@Override
public Boolean get() {
return TestTimelineStore.getEntitiesCount() == 2;
}
}, 100, 10000);
assertEquals(2, TestTimelineStore.getEntitiesCount(), "Wrong Initial Entities Count");
// Append the Summary log file with few more entities
TimelineEntities entities = PluginStoreTestUtils.generateTestEntities();
FSDataOutputStream outStream = fs.append(
new Path(attemptDirPath, TEST_SUMMARY_LOG_FILE_NAME));
JsonGenerator jsonGenerator
= new JsonFactory().createGenerator((OutputStream) outStream);
jsonGenerator.setPrettyPrinter(new MinimalPrettyPrinter("\n"));
ObjectMapper objMapper = new ObjectMapper();
objMapper.setAnnotationIntrospector(
new JaxbAnnotationIntrospector(TypeFactory.defaultInstance()));
objMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
for (TimelineEntity entity : entities.getEntities()) {
objMapper.writeValue(jsonGenerator, entity);
}
outStream.close();
// Validate if there are any duplicates
newStore.scanActiveLogs();
GenericTestUtils.waitFor(new Supplier<Boolean>() {
@Override
public Boolean get() {
return TestTimelineStore.getEntitiesCount() == 4;
}
}, 100, 10000);
assertEquals(4, TestTimelineStore.getEntitiesCount(), "Duplicate Entities present");
} finally {
if (newStore != null) {
newStore.stop();
}
fs.delete(userAppRoot, true);
}
}
@Test
void testStateStoreAndRecovery() throws Exception {
// Prepare the AppLogs Data
EntityGroupFSTimelineStore.AppLogs appLogs =
store.new AppLogs(mainTestAppId, mainTestAppDirPath, AppState.COMPLETED);
appLogs.scanForLogs();
List<LogInfo> summaryLogs = appLogs.getSummaryLogs();
List<EntityGroupFSTimelineStore.AppLogs> logsList = new ArrayList<>();
logsList.add(appLogs);
// Store the Log files
Path checkpointFile = new Path(fs.getHomeDirectory(), "atscheckpoint");
try (DataOutputStream dataOutputStream = fs.create(checkpointFile)) {
store.storeLogFiles(logsList, dataOutputStream);
} catch (IOException e) {
fail("Failed to store the log files");
}
// Recover the Log files and validate the contents
try (DataInputStream dataInputStream = fs.open(checkpointFile)) {
HashMap<String, Pair<Long, Long>> logFiles =
store.recoverLogFiles(dataInputStream);
assertEquals(summaryLogs.size(), logFiles.size());
for (LogInfo logInfo : summaryLogs) {
String logFileName = logInfo.getAttemptDirName() +
Path.SEPARATOR + logInfo.getFilename();
Pair<Long, Long> pair = logFiles.get(logFileName);
assertNotNull(pair, "Failed to recover " + logFileName);
assertTrue(logInfo.getLastProcessedTime() == pair.getLeft(),
"LastProcessedTime is not same");
assertTrue(logInfo.getOffset() == pair.getRight(),
"Offset is not same");
}
} catch (IOException e) {
fail("Failed to recover the log files");
}
}
private EntityGroupFSTimelineStore createAndStartTimelineStore(
AppState appstate) {
// stop before creating new store to get the lock
store.stop();
EntityGroupFSTimelineStore newStore = new EntityGroupFSTimelineStore() {
@Override
protected AppState getAppState(ApplicationId appId) throws IOException {
return appstate;
}
};
newStore.init(config);
newStore.setFs(fs);
newStore.start();
return newStore;
}
private void createTestFiles(ApplicationId appId, Path attemptDirPath)
throws IOException {
createTestFiles(appId, attemptDirPath, mainEntityLogFileName);
}
private void createTestFiles(ApplicationId appId, Path attemptDirPath,
String logPath) throws IOException {
TimelineEntities entities = PluginStoreTestUtils.generateTestEntities();
PluginStoreTestUtils.writeEntities(entities,
new Path(attemptDirPath, TEST_SUMMARY_LOG_FILE_NAME), fs);
Map<String, Set<Object>> primaryFilters = new HashMap<>();
Set<Object> appSet = new HashSet<Object>();
appSet.add(appId.toString());
primaryFilters.put(EntityGroupPlugInForTest.APP_ID_FILTER_NAME, appSet);
entityNew = PluginStoreTestUtils
.createEntity(appId.toString(), "type_3", 789L, null, null,
primaryFilters, null, "domain_id_1");
TimelineEntities entityList = new TimelineEntities();
entityList.addEntity(entityNew);
PluginStoreTestUtils.writeEntities(entityList,
new Path(attemptDirPath, logPath), fs);
FSDataOutputStream out = fs.create(
new Path(attemptDirPath, TEST_DOMAIN_LOG_FILE_NAME));
out.close();
}
private static Path getTestRootPath(String pathString) {
return fileContextTestHelper.getTestRootPath(fc, pathString);
}
private static String getAttemptDirName(ApplicationId appId) {
return ApplicationAttemptId.appAttemptIdStrPrefix + appId.toString() + "_1";
}
}