| /** |
| * 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.nodemanager.recovery; |
| |
| import static org.assertj.core.api.Assertions.assertThat; |
| import static org.fusesource.leveldbjni.JniDBFactory.bytes; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertNotEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.mockito.ArgumentMatchers.any; |
| import static org.mockito.Mockito.isNull; |
| import static org.mockito.Mockito.mock; |
| import static org.mockito.Mockito.spy; |
| import static org.mockito.Mockito.timeout; |
| import static org.mockito.Mockito.times; |
| import static org.mockito.Mockito.verify; |
| import static org.mockito.Mockito.when; |
| |
| import java.io.File; |
| import java.io.IOException; |
| |
| import java.io.Serializable; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileUtil; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.service.ServiceStateException; |
| import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest; |
| import org.apache.hadoop.yarn.api.records.ApplicationAccessType; |
| import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; |
| import org.apache.hadoop.yarn.api.records.ApplicationId; |
| import org.apache.hadoop.yarn.api.records.ContainerExitStatus; |
| import org.apache.hadoop.yarn.api.records.ContainerId; |
| import org.apache.hadoop.yarn.api.records.ContainerLaunchContext; |
| import org.apache.hadoop.yarn.api.records.LocalResource; |
| import org.apache.hadoop.yarn.api.records.LocalResourceType; |
| import org.apache.hadoop.yarn.api.records.LocalResourceVisibility; |
| import org.apache.hadoop.yarn.api.records.Priority; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.Token; |
| import org.apache.hadoop.yarn.api.records.URL; |
| import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationIdPBImpl; |
| import org.apache.hadoop.yarn.api.records.impl.pb.LocalResourcePBImpl; |
| import org.apache.hadoop.yarn.conf.YarnConfiguration; |
| import org.apache.hadoop.yarn.proto.YarnProtos.LocalResourceProto; |
| import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.ContainerManagerApplicationProto; |
| import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.DeletionServiceDeleteTaskProto; |
| import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.LocalizedResourceProto; |
| import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.LogDeleterProto; |
| import org.apache.hadoop.yarn.security.ContainerTokenIdentifier; |
| import org.apache.hadoop.yarn.server.api.records.MasterKey; |
| import org.apache.hadoop.yarn.server.nodemanager.amrmproxy.AMRMProxyTokenSecretManager; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ResourceMappings; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.resourceplugin.fpga.FpgaDevice; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceAllocation; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.resourceplugin.gpu.GpuDevice; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.LocalResourceTrackerState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredAMRMProxyState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredApplicationsState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerStatus; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerTokensState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerType; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredDeletionServiceState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredLocalizationState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredLogDeleterState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredNMTokensState; |
| import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredUserResources; |
| import org.apache.hadoop.yarn.server.records.Version; |
| import org.apache.hadoop.yarn.server.security.BaseContainerTokenSecretManager; |
| import org.apache.hadoop.yarn.server.security.BaseNMTokenSecretManager; |
| import org.apache.hadoop.yarn.server.utils.BuilderUtils; |
| import org.iq80.leveldb.DB; |
| import org.iq80.leveldb.DBException; |
| import org.junit.After; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.mockito.Mockito; |
| |
| public class TestNMLeveldbStateStoreService { |
| private static final File TMP_DIR = new File( |
| System.getProperty("test.build.data", |
| System.getProperty("java.io.tmpdir")), |
| TestNMLeveldbStateStoreService.class.getName()); |
| |
| YarnConfiguration conf; |
| NMLeveldbStateStoreService stateStore; |
| |
| @Before |
| public void setup() throws IOException { |
| FileUtil.fullyDelete(TMP_DIR); |
| conf = new YarnConfiguration(); |
| conf.setBoolean(YarnConfiguration.NM_RECOVERY_ENABLED, true); |
| conf.set(YarnConfiguration.NM_RECOVERY_DIR, TMP_DIR.toString()); |
| restartStateStore(); |
| } |
| |
| @After |
| public void cleanup() throws IOException { |
| if (stateStore != null) { |
| stateStore.close(); |
| } |
| FileUtil.fullyDelete(TMP_DIR); |
| } |
| |
| private List<RecoveredContainerState> loadContainersState( |
| RecoveryIterator<RecoveredContainerState> it) throws IOException { |
| List<RecoveredContainerState> containers = |
| new ArrayList<RecoveredContainerState>(); |
| while (it.hasNext()) { |
| RecoveredContainerState rcs = it.next(); |
| containers.add(rcs); |
| } |
| return containers; |
| } |
| |
| private List<ContainerManagerApplicationProto> loadApplicationProtos( |
| RecoveryIterator<ContainerManagerApplicationProto> it) |
| throws IOException { |
| List<ContainerManagerApplicationProto> applicationProtos = |
| new ArrayList<ContainerManagerApplicationProto>(); |
| while (it.hasNext()) { |
| applicationProtos.add(it.next()); |
| } |
| return applicationProtos; |
| } |
| |
| private List<DeletionServiceDeleteTaskProto> loadDeletionTaskProtos( |
| RecoveryIterator<DeletionServiceDeleteTaskProto> it) throws IOException { |
| List<DeletionServiceDeleteTaskProto> deleteTaskProtos = |
| new ArrayList<DeletionServiceDeleteTaskProto>(); |
| while (it.hasNext()) { |
| deleteTaskProtos.add(it.next()); |
| } |
| return deleteTaskProtos; |
| } |
| |
| private Map<String, RecoveredUserResources> loadUserResources( |
| RecoveryIterator<Map.Entry<String, RecoveredUserResources>> it) |
| throws IOException { |
| Map<String, RecoveredUserResources> userResources = |
| new HashMap<String, RecoveredUserResources>(); |
| while (it.hasNext()) { |
| Map.Entry<String, RecoveredUserResources> entry = it.next(); |
| userResources.put(entry.getKey(), entry.getValue()); |
| } |
| return userResources; |
| } |
| |
| private Map<ApplicationAttemptId, MasterKey> loadNMTokens( |
| RecoveryIterator<Map.Entry<ApplicationAttemptId, MasterKey>> it) |
| throws IOException { |
| Map<ApplicationAttemptId, MasterKey> nmTokens = |
| new HashMap<ApplicationAttemptId, MasterKey>(); |
| while (it.hasNext()) { |
| Map.Entry<ApplicationAttemptId, MasterKey> entry = it.next(); |
| nmTokens.put(entry.getKey(), entry.getValue()); |
| } |
| return nmTokens; |
| } |
| |
| private Map<ContainerId, Long> loadContainerTokens( |
| RecoveryIterator<Map.Entry<ContainerId, Long>> it) throws IOException { |
| Map<ContainerId, Long> containerTokens = |
| new HashMap<ContainerId, Long>(); |
| while (it.hasNext()) { |
| Map.Entry<ContainerId, Long> entry = it.next(); |
| containerTokens.put(entry.getKey(), entry.getValue()); |
| } |
| return containerTokens; |
| } |
| |
| private List<LocalizedResourceProto> loadCompletedResources( |
| RecoveryIterator<LocalizedResourceProto> it) throws IOException { |
| List<LocalizedResourceProto> completedResources = |
| new ArrayList<LocalizedResourceProto>(); |
| while (it != null && it.hasNext()) { |
| completedResources.add(it.next()); |
| } |
| return completedResources; |
| } |
| |
| private Map<LocalResourceProto, Path> loadStartedResources( |
| RecoveryIterator <Map.Entry<LocalResourceProto, Path>> it) |
| throws IOException { |
| Map<LocalResourceProto, Path> startedResources = |
| new HashMap<LocalResourceProto, Path>(); |
| while (it != null &&it.hasNext()) { |
| Map.Entry<LocalResourceProto, Path> entry = it.next(); |
| startedResources.put(entry.getKey(), entry.getValue()); |
| } |
| return startedResources; |
| } |
| |
| private void restartStateStore() throws IOException { |
| // need to close so leveldb releases database lock |
| if (stateStore != null) { |
| stateStore.close(); |
| } |
| stateStore = new NMLeveldbStateStoreService(); |
| stateStore.init(conf); |
| stateStore.start(); |
| } |
| |
| private void verifyEmptyState() throws IOException { |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| assertNotNull(state); |
| LocalResourceTrackerState pubts = state.getPublicTrackerState(); |
| assertNotNull(pubts); |
| assertTrue(loadCompletedResources(pubts.getCompletedResourcesIterator()) |
| .isEmpty()); |
| assertTrue(loadStartedResources(pubts.getStartedResourcesIterator()) |
| .isEmpty()); |
| assertTrue(loadUserResources(state.getIterator()).isEmpty()); |
| } |
| |
| @Test |
| public void testIsNewlyCreated() throws IOException { |
| assertTrue(stateStore.isNewlyCreated()); |
| restartStateStore(); |
| assertFalse(stateStore.isNewlyCreated()); |
| } |
| |
| @Test |
| public void testEmptyState() throws IOException { |
| assertTrue(stateStore.canRecover()); |
| verifyEmptyState(); |
| } |
| |
| @Test |
| public void testCheckVersion() throws IOException { |
| // default version |
| Version defaultVersion = stateStore.getCurrentVersion(); |
| Assert.assertEquals(defaultVersion, stateStore.loadVersion()); |
| |
| // compatible version |
| Version compatibleVersion = |
| Version.newInstance(defaultVersion.getMajorVersion(), |
| defaultVersion.getMinorVersion() + 2); |
| stateStore.storeVersion(compatibleVersion); |
| Assert.assertEquals(compatibleVersion, stateStore.loadVersion()); |
| restartStateStore(); |
| // overwrite the compatible version |
| Assert.assertEquals(defaultVersion, stateStore.loadVersion()); |
| |
| // incompatible version |
| Version incompatibleVersion = |
| Version.newInstance(defaultVersion.getMajorVersion() + 1, |
| defaultVersion.getMinorVersion()); |
| stateStore.storeVersion(incompatibleVersion); |
| try { |
| restartStateStore(); |
| Assert.fail("Incompatible version, should expect fail here."); |
| } catch (ServiceStateException e) { |
| Assert.assertTrue("Exception message mismatch", |
| e.getMessage().contains("Incompatible version for NM state:")); |
| } |
| } |
| |
| @Test |
| public void testApplicationStorage() throws IOException { |
| // test empty when no state |
| RecoveredApplicationsState state = stateStore.loadApplicationsState(); |
| List<ContainerManagerApplicationProto> apps = |
| loadApplicationProtos(state.getIterator()); |
| assertTrue(apps.isEmpty()); |
| |
| // store an application and verify recovered |
| final ApplicationId appId1 = ApplicationId.newInstance(1234, 1); |
| ContainerManagerApplicationProto.Builder builder = |
| ContainerManagerApplicationProto.newBuilder(); |
| builder.setId(((ApplicationIdPBImpl) appId1).getProto()); |
| builder.setUser("user1"); |
| ContainerManagerApplicationProto appProto1 = builder.build(); |
| stateStore.storeApplication(appId1, appProto1); |
| restartStateStore(); |
| state = stateStore.loadApplicationsState(); |
| apps = loadApplicationProtos(state.getIterator()); |
| assertEquals(1, apps.size()); |
| assertEquals(appProto1, apps.get(0)); |
| |
| // add a new app |
| final ApplicationId appId2 = ApplicationId.newInstance(1234, 2); |
| builder = ContainerManagerApplicationProto.newBuilder(); |
| builder.setId(((ApplicationIdPBImpl) appId2).getProto()); |
| builder.setUser("user2"); |
| ContainerManagerApplicationProto appProto2 = builder.build(); |
| stateStore.storeApplication(appId2, appProto2); |
| restartStateStore(); |
| state = stateStore.loadApplicationsState(); |
| apps = loadApplicationProtos(state.getIterator()); |
| assertEquals(2, apps.size()); |
| assertTrue(apps.contains(appProto1)); |
| assertTrue(apps.contains(appProto2)); |
| |
| // test removing an application |
| stateStore.removeApplication(appId2); |
| restartStateStore(); |
| state = stateStore.loadApplicationsState(); |
| apps = loadApplicationProtos(state.getIterator()); |
| assertEquals(1, apps.size()); |
| assertEquals(appProto1, apps.get(0)); |
| } |
| |
| |
| @Test |
| public void testContainerStorageWhenContainerIsRequested() |
| throws IOException { |
| final ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| restartStateStore(); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| final RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(0, rcs.getVersion()); |
| assertEquals(containerParams.getContainerStartTime().longValue(), |
| rcs.getStartTime()); |
| assertEquals(RecoveredContainerStatus.REQUESTED, rcs.getStatus()); |
| assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode()); |
| assertEquals(false, rcs.getKilled()); |
| assertEquals(containerParams.getContainerRequest(), rcs.getStartRequest()); |
| assertTrue(rcs.getDiagnostics().isEmpty()); |
| assertEquals(containerParams.getContainerResource(), rcs.getCapability()); |
| } |
| |
| |
| |
| @Test |
| public void testContainerStorageWhenContainerIsQueued() |
| throws IOException { |
| ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| ContainerId containerId = containerParams.getContainerId(); |
| StartContainerRequest containerReq = containerParams.getContainerRequest(); |
| Resource containerResource = containerParams.getContainerResource(); |
| ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId(); |
| |
| storeNewContainerRecordWithoutStartContainerRequest(appAttemptId); |
| |
| stateStore.storeContainerQueued(containerId); |
| restartStateStore(); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(RecoveredContainerStatus.QUEUED, rcs.getStatus()); |
| assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode()); |
| assertEquals(false, rcs.getKilled()); |
| assertEquals(containerReq, rcs.getStartRequest()); |
| assertTrue(rcs.getDiagnostics().isEmpty()); |
| assertEquals(containerResource, rcs.getCapability()); |
| } |
| |
| @Test |
| public void testContainerStorageWhenContainerIsLaunched() |
| throws IOException { |
| ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| ContainerId containerId = containerParams.getContainerId(); |
| StartContainerRequest containerReq = containerParams.getContainerRequest(); |
| Resource containerResource = containerParams.getContainerResource(); |
| ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId(); |
| |
| storeNewContainerRecordWithoutStartContainerRequest(appAttemptId); |
| stateStore.storeContainerQueued(containerId); |
| |
| StringBuilder diags = launchContainerWithDiagnostics(containerId); |
| restartStateStore(); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(RecoveredContainerStatus.LAUNCHED, rcs.getStatus()); |
| assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode()); |
| assertEquals(false, rcs.getKilled()); |
| assertEquals(containerReq, rcs.getStartRequest()); |
| assertEquals(diags.toString(), rcs.getDiagnostics()); |
| assertEquals(containerResource, rcs.getCapability()); |
| } |
| |
| @Test |
| public void testContainerStorageWhenContainerIsPaused() |
| throws IOException { |
| ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| ContainerId containerId = containerParams.getContainerId(); |
| StartContainerRequest containerReq = containerParams.getContainerRequest(); |
| ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId(); |
| |
| storeNewContainerRecordWithoutStartContainerRequest(appAttemptId); |
| stateStore.storeContainerQueued(containerId); |
| |
| stateStore.storeContainerPaused(containerId); |
| restartStateStore(); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(RecoveredContainerStatus.PAUSED, rcs.getStatus()); |
| assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode()); |
| assertEquals(false, rcs.getKilled()); |
| assertEquals(containerReq, rcs.getStartRequest()); |
| |
| // Resume the container |
| stateStore.removeContainerPaused(containerId); |
| restartStateStore(); |
| recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| } |
| |
| @Test |
| public void testContainerStorageWhenContainerSizeIncreased() |
| throws IOException { |
| ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| ContainerId containerId = containerParams.getContainerId(); |
| ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId(); |
| |
| storeNewContainerRecordWithoutStartContainerRequest(appAttemptId); |
| stateStore.storeContainerQueued(containerId); |
| launchContainerWithDiagnostics(containerId); |
| |
| increaseContainerSize(containerId); |
| restartStateStore(); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(0, rcs.getVersion()); |
| assertEquals(RecoveredContainerStatus.LAUNCHED, rcs.getStatus()); |
| assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode()); |
| assertEquals(false, rcs.getKilled()); |
| assertEquals(Resource.newInstance(2468, 4), rcs.getCapability()); |
| } |
| |
| @Test |
| public void testContainerStorageWhenContainerMarkedAsKilled() |
| throws IOException { |
| ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| ContainerId containerId = containerParams.getContainerId(); |
| ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId(); |
| |
| storeNewContainerRecordWithoutStartContainerRequest(appAttemptId); |
| stateStore.storeContainerQueued(containerId); |
| StringBuilder diags = launchContainerWithDiagnostics(containerId); |
| ContainerTokenIdentifier updateTokenIdentifier = |
| increaseContainerSize(containerId); |
| |
| markContainerAsKilled(containerId, diags); |
| restartStateStore(); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(RecoveredContainerStatus.LAUNCHED, rcs.getStatus()); |
| assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode()); |
| assertTrue(rcs.getKilled()); |
| ContainerTokenIdentifier tokenReadFromRequest = BuilderUtils |
| .newContainerTokenIdentifier(rcs.getStartRequest() |
| .getContainerToken()); |
| assertEquals(updateTokenIdentifier, tokenReadFromRequest); |
| assertEquals(diags.toString(), rcs.getDiagnostics()); |
| } |
| |
| @Test |
| public void testContainerStorageWhenContainerCompleted() |
| throws IOException { |
| ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| ContainerId containerId = containerParams.getContainerId(); |
| ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId(); |
| |
| storeNewContainerRecordWithoutStartContainerRequest(appAttemptId); |
| stateStore.storeContainerQueued(containerId); |
| StringBuilder diags = launchContainerWithDiagnostics(containerId); |
| markContainerAsKilled(containerId, diags); |
| |
| // add yet more diags, mark container completed |
| diags.append("some final diags"); |
| stateStore.storeContainerDiagnostics(containerId, diags); |
| stateStore.storeContainerCompleted(containerId, 21); |
| restartStateStore(); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(RecoveredContainerStatus.COMPLETED, rcs.getStatus()); |
| assertEquals(21, rcs.getExitCode()); |
| assertTrue(rcs.getKilled()); |
| assertEquals(diags.toString(), rcs.getDiagnostics()); |
| } |
| |
| @Test |
| public void testContainerStorage() throws IOException { |
| ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| ContainerId containerId = containerParams.getContainerId(); |
| |
| // remaining retry attempts, work dir and log dir are stored |
| stateStore.storeContainerRemainingRetryAttempts(containerId, 6); |
| stateStore.storeContainerWorkDir(containerId, "/test/workdir"); |
| stateStore.storeContainerLogDir(containerId, "/test/logdir"); |
| restartStateStore(); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(6, rcs.getRemainingRetryAttempts()); |
| assertEquals("/test/workdir", rcs.getWorkDir()); |
| assertEquals("/test/logdir", rcs.getLogDir()); |
| validateRetryAttempts(containerId); |
| } |
| |
| @Test |
| public void testContainerStorageWhenContainerRemoved() |
| throws IOException { |
| ContainerStateConstructParams containerParams = |
| storeContainerInStateStore(); |
| ContainerId containerId = containerParams.getContainerId(); |
| |
| // remove the container and verify not recovered |
| stateStore.removeContainer(containerId); |
| restartStateStore(); |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertTrue(recoveredContainers.isEmpty()); |
| // recover again to check remove clears all containers |
| restartStateStore(); |
| NMStateStoreService nmStoreSpy = spy(stateStore); |
| loadContainersState(nmStoreSpy.getContainerStateIterator()); |
| verify(nmStoreSpy, times(0)).removeContainer(any(ContainerId.class)); |
| } |
| |
| private ContainerStateConstructParams storeContainerInStateStore() |
| throws IOException { |
| // test empty when no state |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertTrue(recoveredContainers.isEmpty()); |
| |
| // create a container request |
| ApplicationId appId = ApplicationId.newInstance(1234, 3); |
| ApplicationAttemptId appAttemptId = |
| ApplicationAttemptId.newInstance(appId, 4); |
| ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5); |
| Resource containerResource = Resource.newInstance(1024, 2); |
| StartContainerRequest containerReq = |
| createContainerRequest(containerId, containerResource); |
| |
| long anyContainerStartTime = 1573155078494L; |
| stateStore.storeContainer(containerId, 0, anyContainerStartTime, |
| containerReq); |
| |
| // verify the container version key is not stored for new containers |
| DB db = stateStore.getDB(); |
| assertNull("version key present for new container", db.get(bytes( |
| stateStore.getContainerVersionKey(containerId.toString())))); |
| |
| return new ContainerStateConstructParams() |
| .setContainerRequest(containerReq) |
| .setContainerResource(containerResource) |
| .setContainerStartTime(anyContainerStartTime) |
| .setAppAttemptId(appAttemptId) |
| .setContainerId(containerId); |
| } |
| |
| private static class ContainerStateConstructParams { |
| private StartContainerRequest containerRequest; |
| private Resource containerResource; |
| private Long containerStartTime; |
| private ApplicationAttemptId appAttemptId; |
| private ContainerId containerId; |
| |
| public ApplicationAttemptId getAppAttemptId() { |
| return appAttemptId; |
| } |
| public ContainerStateConstructParams setAppAttemptId(ApplicationAttemptId |
| theAppAttemptId) { |
| this.appAttemptId = theAppAttemptId; |
| return this; |
| } |
| public ContainerId getContainerId() { |
| return containerId; |
| } |
| public ContainerStateConstructParams setContainerId(ContainerId |
| theContainerId) { |
| this.containerId = theContainerId; |
| return this; |
| } |
| |
| public StartContainerRequest getContainerRequest() { |
| return containerRequest; |
| } |
| public ContainerStateConstructParams setContainerRequest( |
| StartContainerRequest theContainerRequest) { |
| this.containerRequest = theContainerRequest; |
| return this; |
| } |
| |
| public Resource getContainerResource() { |
| return containerResource; |
| } |
| |
| public ContainerStateConstructParams setContainerResource( |
| Resource theContainerResource) { |
| this.containerResource = theContainerResource; |
| return this; |
| } |
| |
| public Long getContainerStartTime() { |
| return containerStartTime; |
| } |
| |
| public ContainerStateConstructParams setContainerStartTime( |
| Long theContainerStartTime) { |
| this.containerStartTime = theContainerStartTime; |
| return this; |
| } |
| } |
| |
| private void markContainerAsKilled(ContainerId containerId, |
| StringBuilder diags) throws IOException { |
| // mark the container killed, add some more diags |
| diags.append("some more diags for container"); |
| stateStore.storeContainerDiagnostics(containerId, diags); |
| stateStore.storeContainerKilled(containerId); |
| } |
| |
| private ContainerTokenIdentifier increaseContainerSize( |
| ContainerId containerId) throws IOException { |
| ContainerTokenIdentifier updateTokenIdentifier = |
| new ContainerTokenIdentifier(containerId, "host", "user", |
| Resource.newInstance(2468, 4), 9876543210L, 42, 2468, |
| Priority.newInstance(7), 13579); |
| stateStore |
| .storeContainerUpdateToken(containerId, updateTokenIdentifier); |
| return updateTokenIdentifier; |
| } |
| |
| private StringBuilder launchContainerWithDiagnostics(ContainerId containerId) |
| throws IOException { |
| StringBuilder diags = new StringBuilder(); |
| stateStore.storeContainerLaunched(containerId); |
| diags.append("some diags for container"); |
| stateStore.storeContainerDiagnostics(containerId, diags); |
| return diags; |
| } |
| |
| private void storeNewContainerRecordWithoutStartContainerRequest( |
| ApplicationAttemptId appAttemptId) throws IOException { |
| // store a new container record without StartContainerRequest |
| ContainerId containerId1 = ContainerId.newContainerId(appAttemptId, 6); |
| stateStore.storeContainerLaunched(containerId1); |
| |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| // check whether the new container record is discarded |
| assertEquals(1, recoveredContainers.size()); |
| } |
| |
| private void validateRetryAttempts(ContainerId containerId) |
| throws IOException { |
| // store finishTimeForRetryAttempts |
| List<Long> finishTimeForRetryAttempts = Arrays.asList(1462700529039L, |
| 1462700529050L, 1462700529120L); |
| stateStore.storeContainerRestartTimes(containerId, |
| finishTimeForRetryAttempts); |
| restartStateStore(); |
| RecoveredContainerState rcs = |
| loadContainersState(stateStore.getContainerStateIterator()).get(0); |
| List<Long> recoveredRestartTimes = rcs.getRestartTimes(); |
| assertEquals(1462700529039L, (long)recoveredRestartTimes.get(0)); |
| assertEquals(1462700529050L, (long)recoveredRestartTimes.get(1)); |
| assertEquals(1462700529120L, (long)recoveredRestartTimes.get(2)); |
| } |
| |
| private StartContainerRequest createContainerRequest( |
| ContainerId containerId, Resource res) { |
| return createContainerRequestInternal(containerId, res); |
| } |
| |
| private StartContainerRequest createContainerRequestInternal(ContainerId |
| containerId, Resource res) { |
| LocalResource lrsrc = LocalResource.newInstance( |
| URL.newInstance("hdfs", "somehost", 12345, "/some/path/to/rsrc"), |
| LocalResourceType.FILE, LocalResourceVisibility.APPLICATION, 123L, |
| 1234567890L); |
| Map<String, LocalResource> localResources = |
| new HashMap<String, LocalResource>(); |
| localResources.put("rsrc", lrsrc); |
| Map<String, String> env = new HashMap<String, String>(); |
| env.put("somevar", "someval"); |
| List<String> containerCmds = new ArrayList<String>(); |
| containerCmds.add("somecmd"); |
| containerCmds.add("somearg"); |
| Map<String, ByteBuffer> serviceData = new HashMap<String, ByteBuffer>(); |
| serviceData.put("someservice", |
| ByteBuffer.wrap(new byte[] {0x1, 0x2, 0x3})); |
| ByteBuffer containerTokens = |
| ByteBuffer.wrap(new byte[] {0x7, 0x8, 0x9, 0xa}); |
| Map<ApplicationAccessType, String> acls = |
| new HashMap<ApplicationAccessType, String>(); |
| acls.put(ApplicationAccessType.VIEW_APP, "viewuser"); |
| acls.put(ApplicationAccessType.MODIFY_APP, "moduser"); |
| ContainerLaunchContext clc = ContainerLaunchContext.newInstance( |
| localResources, env, containerCmds, serviceData, containerTokens, |
| acls); |
| Resource containerRsrc = Resource.newInstance(1357, 3); |
| |
| if (res != null) { |
| containerRsrc = res; |
| } |
| ContainerTokenIdentifier containerTokenId = |
| new ContainerTokenIdentifier(containerId, "host", "user", |
| containerRsrc, 9876543210L, 42, 2468, Priority.newInstance(7), |
| 13579); |
| Token containerToken = Token.newInstance(containerTokenId.getBytes(), |
| ContainerTokenIdentifier.KIND.toString(), "password".getBytes(), |
| "tokenservice"); |
| return StartContainerRequest.newInstance(clc, containerToken); |
| } |
| |
| @Test |
| public void testLocalTrackerStateIterator() throws IOException { |
| String user1 = "somebody"; |
| ApplicationId appId1 = ApplicationId.newInstance(1, 1); |
| ApplicationId appId2 = ApplicationId.newInstance(2, 2); |
| |
| String user2 = "someone"; |
| ApplicationId appId3 = ApplicationId.newInstance(3, 3); |
| |
| // start and finish local resource for applications |
| Path appRsrcPath1 = new Path("hdfs://some/app/resource1"); |
| LocalResourcePBImpl rsrcPb1 = (LocalResourcePBImpl) |
| LocalResource.newInstance( |
| URL.fromPath(appRsrcPath1), |
| LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION, |
| 123L, 456L); |
| LocalResourceProto appRsrcProto1 = rsrcPb1.getProto(); |
| Path appRsrcLocalPath1 = new Path("/some/local/dir/for/apprsrc1"); |
| Path appRsrcPath2 = new Path("hdfs://some/app/resource2"); |
| LocalResourcePBImpl rsrcPb2 = (LocalResourcePBImpl) |
| LocalResource.newInstance( |
| URL.fromPath(appRsrcPath2), |
| LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION, |
| 123L, 456L); |
| LocalResourceProto appRsrcProto2 = rsrcPb2.getProto(); |
| Path appRsrcLocalPath2 = new Path("/some/local/dir/for/apprsrc2"); |
| Path appRsrcPath3 = new Path("hdfs://some/app/resource3"); |
| LocalResourcePBImpl rsrcPb3 = (LocalResourcePBImpl) |
| LocalResource.newInstance( |
| URL.fromPath(appRsrcPath3), |
| LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION, |
| 123L, 456L); |
| LocalResourceProto appRsrcProto3 = rsrcPb3.getProto(); |
| Path appRsrcLocalPath3 = new Path("/some/local/dir/for/apprsrc2"); |
| |
| stateStore.startResourceLocalization(user1, appId1, appRsrcProto1, |
| appRsrcLocalPath1); |
| stateStore.startResourceLocalization(user1, appId2, appRsrcProto2, |
| appRsrcLocalPath2); |
| stateStore.startResourceLocalization(user2, appId3, appRsrcProto3, |
| appRsrcLocalPath3); |
| |
| LocalizedResourceProto appLocalizedProto1 = |
| LocalizedResourceProto.newBuilder() |
| .setResource(appRsrcProto1) |
| .setLocalPath(appRsrcLocalPath1.toString()) |
| .setSize(1234567L) |
| .build(); |
| LocalizedResourceProto appLocalizedProto2 = |
| LocalizedResourceProto.newBuilder() |
| .setResource(appRsrcProto2) |
| .setLocalPath(appRsrcLocalPath2.toString()) |
| .setSize(1234567L) |
| .build(); |
| LocalizedResourceProto appLocalizedProto3 = |
| LocalizedResourceProto.newBuilder() |
| .setResource(appRsrcProto3) |
| .setLocalPath(appRsrcLocalPath3.toString()) |
| .setSize(1234567L) |
| .build(); |
| |
| |
| stateStore.finishResourceLocalization(user1, appId1, appLocalizedProto1); |
| stateStore.finishResourceLocalization(user1, appId2, appLocalizedProto2); |
| stateStore.finishResourceLocalization(user2, appId3, appLocalizedProto3); |
| |
| |
| List<LocalizedResourceProto> completedResources = |
| new ArrayList<LocalizedResourceProto>(); |
| Map<LocalResourceProto, Path> startedResources = |
| new HashMap<LocalResourceProto, Path>(); |
| |
| // restart and verify two users exist and two apps completed for user1. |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertEquals(2, userResources.size()); |
| |
| RecoveredUserResources uResource = userResources.get(user1); |
| assertEquals(2, uResource.getAppTrackerStates().size()); |
| LocalResourceTrackerState app1ts = |
| uResource.getAppTrackerStates().get(appId1); |
| assertNotNull(app1ts); |
| completedResources = loadCompletedResources( |
| app1ts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| app1ts.getStartedResourcesIterator()); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(1, completedResources.size()); |
| assertEquals(appLocalizedProto1, |
| completedResources.iterator().next()); |
| LocalResourceTrackerState app2ts = |
| uResource.getAppTrackerStates().get(appId2); |
| assertNotNull(app2ts); |
| completedResources = loadCompletedResources( |
| app2ts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| app2ts.getStartedResourcesIterator()); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(1, completedResources.size()); |
| assertEquals(appLocalizedProto2, |
| completedResources.iterator().next()); |
| } |
| |
| @Test |
| public void testStartResourceLocalizationForApplicationResource() |
| throws IOException { |
| String user = "somebody"; |
| ApplicationId appId = ApplicationId.newInstance(1, 1); |
| |
| // start a local resource for an application |
| Path appRsrcPath = new Path("hdfs://some/app/resource"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) |
| LocalResource.newInstance( |
| URL.fromPath(appRsrcPath), |
| LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION, |
| 123L, 456L); |
| LocalResourceProto appRsrcProto = rsrcPb.getProto(); |
| Path appRsrcLocalPath = new Path("/some/local/dir/for/apprsrc"); |
| stateStore.startResourceLocalization(user, appId, appRsrcProto, |
| appRsrcLocalPath); |
| |
| List<LocalizedResourceProto> completedResources = |
| new ArrayList<LocalizedResourceProto>(); |
| Map<LocalResourceProto, Path> startedResources = |
| new HashMap<LocalResourceProto, Path>(); |
| |
| // restart and verify only app resource is marked in-progress |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| LocalResourceTrackerState pubts = state.getPublicTrackerState(); |
| completedResources = loadCompletedResources( |
| pubts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| pubts.getStartedResourcesIterator()); |
| assertTrue(completedResources.isEmpty()); |
| assertTrue(startedResources.isEmpty()); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertEquals(1, userResources.size()); |
| RecoveredUserResources rur = userResources.get(user); |
| LocalResourceTrackerState privts = rur.getPrivateTrackerState(); |
| assertNotNull(privts); |
| completedResources = loadCompletedResources( |
| privts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| privts.getStartedResourcesIterator()); |
| assertTrue(completedResources.isEmpty()); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(1, rur.getAppTrackerStates().size()); |
| LocalResourceTrackerState appts = rur.getAppTrackerStates().get(appId); |
| assertNotNull(appts); |
| completedResources = loadCompletedResources( |
| appts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| appts.getStartedResourcesIterator()); |
| assertTrue(completedResources.isEmpty()); |
| assertEquals(1, startedResources.size()); |
| assertEquals(appRsrcLocalPath, |
| startedResources.get(appRsrcProto)); |
| } |
| |
| @Test |
| public void testStartResourceLocalizationForPublicResources() |
| throws IOException { |
| Path pubRsrcPath1 = new Path("hdfs://some/public/resource1"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource |
| .newInstance( |
| URL.fromPath(pubRsrcPath1), |
| LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, |
| 789L, 135L); |
| LocalResourceProto pubRsrcProto1 = rsrcPb.getProto(); |
| Path pubRsrcLocalPath1 = new Path("/some/local/dir/for/pubrsrc1"); |
| stateStore.startResourceLocalization(null, null, pubRsrcProto1, |
| pubRsrcLocalPath1); |
| Path pubRsrcPath2 = new Path("hdfs://some/public/resource2"); |
| rsrcPb = (LocalResourcePBImpl) LocalResource.newInstance( |
| URL.fromPath(pubRsrcPath2), |
| LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, |
| 789L, 135L); |
| LocalResourceProto pubRsrcProto2 = rsrcPb.getProto(); |
| Path pubRsrcLocalPath2 = new Path("/some/local/dir/for/pubrsrc2"); |
| stateStore.startResourceLocalization(null, null, pubRsrcProto2, |
| pubRsrcLocalPath2); |
| |
| // restart and verify resources are marked in-progress |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| LocalResourceTrackerState pubts = state.getPublicTrackerState(); |
| List<LocalizedResourceProto> completedResources = loadCompletedResources( |
| pubts.getCompletedResourcesIterator()); |
| Map<LocalResourceProto, Path> startedResources = loadStartedResources( |
| pubts.getStartedResourcesIterator()); |
| assertTrue(completedResources.isEmpty()); |
| assertEquals(2, startedResources.size()); |
| assertEquals(pubRsrcLocalPath1, |
| startedResources.get(pubRsrcProto1)); |
| assertEquals(pubRsrcLocalPath2, |
| startedResources.get(pubRsrcProto2)); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertEquals(0, userResources.size()); |
| } |
| |
| @Test |
| public void testStartResourceLocalizationForPrivateResource() |
| throws IOException { |
| Path privRsrcPath = new Path("hdfs://some/private/resource"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource |
| .newInstance( |
| URL.fromPath(privRsrcPath), |
| LocalResourceType.PATTERN, LocalResourceVisibility.PRIVATE, |
| 789L, 680L, "*pattern*"); |
| LocalResourceProto privRsrcProto = rsrcPb.getProto(); |
| Path privRsrcLocalPath = new Path("/some/local/dir/for/privrsrc"); |
| String user = "somebody"; |
| stateStore.startResourceLocalization(user, null, privRsrcProto, |
| privRsrcLocalPath); |
| |
| // restart and verify resources are marked in-progress |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertEquals(1, userResources.size()); |
| RecoveredUserResources rur = userResources.get(user); |
| LocalResourceTrackerState privts = rur.getPrivateTrackerState(); |
| assertNotNull(privts); |
| List<LocalizedResourceProto> completedResources = loadCompletedResources( |
| privts.getCompletedResourcesIterator()); |
| Map<LocalResourceProto, Path> startedResources = loadStartedResources( |
| privts.getStartedResourcesIterator()); |
| assertTrue(completedResources.isEmpty()); |
| assertEquals(1, startedResources.size()); |
| assertEquals(privRsrcLocalPath, |
| startedResources.get(privRsrcProto)); |
| assertEquals(0, rur.getAppTrackerStates().size()); |
| } |
| |
| @Test |
| public void testFinishResourceLocalizationForApplicationResource() |
| throws IOException { |
| String user = "somebody"; |
| ApplicationId appId = ApplicationId.newInstance(1, 1); |
| |
| // start and finish a local resource for an application |
| Path appRsrcPath = new Path("hdfs://some/app/resource"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) |
| LocalResource.newInstance( |
| URL.fromPath(appRsrcPath), |
| LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION, |
| 123L, 456L); |
| LocalResourceProto appRsrcProto = rsrcPb.getProto(); |
| Path appRsrcLocalPath = new Path("/some/local/dir/for/apprsrc"); |
| stateStore.startResourceLocalization(user, appId, appRsrcProto, |
| appRsrcLocalPath); |
| LocalizedResourceProto appLocalizedProto = |
| LocalizedResourceProto.newBuilder() |
| .setResource(appRsrcProto) |
| .setLocalPath(appRsrcLocalPath.toString()) |
| .setSize(1234567L) |
| .build(); |
| stateStore.finishResourceLocalization(user, appId, appLocalizedProto); |
| |
| List<LocalizedResourceProto> completedResources = |
| new ArrayList<LocalizedResourceProto>(); |
| Map<LocalResourceProto, Path> startedResources = |
| new HashMap<LocalResourceProto, Path>(); |
| |
| // restart and verify only app resource is completed |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| LocalResourceTrackerState pubts = state.getPublicTrackerState(); |
| completedResources = loadCompletedResources( |
| pubts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| pubts.getStartedResourcesIterator()); |
| assertTrue(completedResources.isEmpty()); |
| assertTrue(startedResources.isEmpty()); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertEquals(1, userResources.size()); |
| RecoveredUserResources rur = userResources.get(user); |
| LocalResourceTrackerState privts = rur.getPrivateTrackerState(); |
| assertNotNull(privts); |
| completedResources = loadCompletedResources( |
| privts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| privts.getStartedResourcesIterator()); |
| assertTrue(completedResources.isEmpty()); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(1, rur.getAppTrackerStates().size()); |
| LocalResourceTrackerState appts = rur.getAppTrackerStates().get(appId); |
| assertNotNull(appts); |
| completedResources = loadCompletedResources( |
| appts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| appts.getStartedResourcesIterator()); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(1, completedResources.size()); |
| assertEquals(appLocalizedProto, |
| completedResources.iterator().next()); |
| } |
| |
| @Test |
| public void testFinishResourceLocalizationForPublicResources() |
| throws IOException { |
| Path pubRsrcPath1 = new Path("hdfs://some/public/resource1"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource |
| .newInstance( |
| URL.fromPath(pubRsrcPath1), |
| LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, |
| 789L, 135L); |
| LocalResourceProto pubRsrcProto1 = rsrcPb.getProto(); |
| Path pubRsrcLocalPath1 = new Path("/some/local/dir/for/pubrsrc1"); |
| stateStore.startResourceLocalization(null, null, pubRsrcProto1, |
| pubRsrcLocalPath1); |
| Path pubRsrcPath2 = new Path("hdfs://some/public/resource2"); |
| rsrcPb = (LocalResourcePBImpl) LocalResource.newInstance( |
| URL.fromPath(pubRsrcPath2), |
| LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, |
| 789L, 135L); |
| LocalResourceProto pubRsrcProto2 = rsrcPb.getProto(); |
| Path pubRsrcLocalPath2 = new Path("/some/local/dir/for/pubrsrc2"); |
| stateStore.startResourceLocalization(null, null, pubRsrcProto2, |
| pubRsrcLocalPath2); |
| |
| // finish some of the resources |
| LocalizedResourceProto pubLocalizedProto1 = |
| LocalizedResourceProto.newBuilder() |
| .setResource(pubRsrcProto1) |
| .setLocalPath(pubRsrcLocalPath1.toString()) |
| .setSize(pubRsrcProto1.getSize()) |
| .build(); |
| stateStore.finishResourceLocalization(null, null, pubLocalizedProto1); |
| |
| // restart and verify state |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| LocalResourceTrackerState pubts = state.getPublicTrackerState(); |
| List<LocalizedResourceProto> completedResources = loadCompletedResources( |
| pubts.getCompletedResourcesIterator()); |
| Map<LocalResourceProto, Path> startedResources = loadStartedResources( |
| pubts.getStartedResourcesIterator()); |
| assertEquals(1, completedResources.size()); |
| assertEquals(pubLocalizedProto1, |
| completedResources.iterator().next()); |
| assertEquals(1, startedResources.size()); |
| assertEquals(pubRsrcLocalPath2, |
| startedResources.get(pubRsrcProto2)); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertEquals(0, userResources.size()); |
| } |
| |
| @Test |
| public void testFinishResourceLocalizationForPrivateResource() |
| throws IOException { |
| String user = "somebody"; |
| ApplicationId appId = ApplicationId.newInstance(1, 1); |
| |
| Path privRsrcPath = new Path("hdfs://some/private/resource"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource |
| .newInstance( |
| URL.fromPath(privRsrcPath), |
| LocalResourceType.PATTERN, LocalResourceVisibility.PRIVATE, |
| 789L, 680L, "*pattern*"); |
| LocalResourceProto privRsrcProto = rsrcPb.getProto(); |
| Path privRsrcLocalPath = new Path("/some/local/dir/for/privrsrc"); |
| stateStore.startResourceLocalization(user, null, privRsrcProto, |
| privRsrcLocalPath); |
| |
| LocalizedResourceProto privLocalizedProto = |
| LocalizedResourceProto.newBuilder() |
| .setResource(privRsrcProto) |
| .setLocalPath(privRsrcLocalPath.toString()) |
| .setSize(privRsrcProto.getSize()) |
| .build(); |
| stateStore.finishResourceLocalization(user, null, privLocalizedProto); |
| |
| // restart and verify state |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| LocalResourceTrackerState pubts = state.getPublicTrackerState(); |
| List<LocalizedResourceProto> completedResources = loadCompletedResources( |
| pubts.getCompletedResourcesIterator()); |
| Map<LocalResourceProto, Path> startedResources = loadStartedResources( |
| pubts.getStartedResourcesIterator()); |
| assertEquals(0, completedResources.size()); |
| assertEquals(0, startedResources.size()); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertEquals(1, userResources.size()); |
| RecoveredUserResources rur = userResources.get(user); |
| LocalResourceTrackerState privts = rur.getPrivateTrackerState(); |
| assertNotNull(privts); |
| completedResources = loadCompletedResources( |
| privts.getCompletedResourcesIterator()); |
| startedResources = loadStartedResources( |
| privts.getStartedResourcesIterator()); |
| assertEquals(1, completedResources.size()); |
| assertEquals(privLocalizedProto, |
| completedResources.iterator().next()); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(0, rur.getAppTrackerStates().size()); |
| LocalResourceTrackerState appts = rur.getAppTrackerStates().get(appId); |
| assertNull(appts); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(1, completedResources.size()); |
| } |
| |
| @Test |
| public void testRemoveLocalizedResourceForApplicationResource() |
| throws IOException { |
| String user = "somebody"; |
| ApplicationId appId = ApplicationId.newInstance(1, 1); |
| |
| // go through the complete lifecycle for an application local resource |
| Path appRsrcPath = new Path("hdfs://some/app/resource"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) |
| LocalResource.newInstance( |
| URL.fromPath(appRsrcPath), |
| LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION, |
| 123L, 456L); |
| LocalResourceProto appRsrcProto = rsrcPb.getProto(); |
| Path appRsrcLocalPath = new Path("/some/local/dir/for/apprsrc"); |
| stateStore.startResourceLocalization(user, appId, appRsrcProto, |
| appRsrcLocalPath); |
| LocalizedResourceProto appLocalizedProto = |
| LocalizedResourceProto.newBuilder() |
| .setResource(appRsrcProto) |
| .setLocalPath(appRsrcLocalPath.toString()) |
| .setSize(1234567L) |
| .build(); |
| stateStore.finishResourceLocalization(user, appId, appLocalizedProto); |
| stateStore.removeLocalizedResource(user, appId, appRsrcLocalPath); |
| |
| restartStateStore(); |
| verifyEmptyState(); |
| |
| // remove an app resource that didn't finish |
| stateStore.startResourceLocalization(user, appId, appRsrcProto, |
| appRsrcLocalPath); |
| stateStore.removeLocalizedResource(user, appId, appRsrcLocalPath); |
| |
| restartStateStore(); |
| verifyEmptyState(); |
| } |
| |
| @Test |
| public void testRemoveLocalizedResourceForPublicResources() |
| throws IOException { |
| // add public resources and remove some |
| Path pubRsrcPath1 = new Path("hdfs://some/public/resource1"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource |
| .newInstance( |
| URL.fromPath(pubRsrcPath1), |
| LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, |
| 789L, 135L); |
| LocalResourceProto pubRsrcProto1 = rsrcPb.getProto(); |
| Path pubRsrcLocalPath1 = new Path("/some/local/dir/for/pubrsrc1"); |
| stateStore.startResourceLocalization(null, null, pubRsrcProto1, |
| pubRsrcLocalPath1); |
| LocalizedResourceProto pubLocalizedProto1 = |
| LocalizedResourceProto.newBuilder() |
| .setResource(pubRsrcProto1) |
| .setLocalPath(pubRsrcLocalPath1.toString()) |
| .setSize(789L) |
| .build(); |
| stateStore.finishResourceLocalization(null, null, pubLocalizedProto1); |
| Path pubRsrcPath2 = new Path("hdfs://some/public/resource2"); |
| rsrcPb = (LocalResourcePBImpl) LocalResource.newInstance( |
| URL.fromPath(pubRsrcPath2), |
| LocalResourceType.FILE, LocalResourceVisibility.PUBLIC, |
| 789L, 135L); |
| LocalResourceProto pubRsrcProto2 = rsrcPb.getProto(); |
| Path pubRsrcLocalPath2 = new Path("/some/local/dir/for/pubrsrc2"); |
| stateStore.startResourceLocalization(null, null, pubRsrcProto2, |
| pubRsrcLocalPath2); |
| LocalizedResourceProto pubLocalizedProto2 = |
| LocalizedResourceProto.newBuilder() |
| .setResource(pubRsrcProto2) |
| .setLocalPath(pubRsrcLocalPath2.toString()) |
| .setSize(7654321L) |
| .build(); |
| stateStore.finishResourceLocalization(null, null, pubLocalizedProto2); |
| stateStore.removeLocalizedResource(null, null, pubRsrcLocalPath2); |
| |
| // restart and verify state |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| LocalResourceTrackerState pubts = state.getPublicTrackerState(); |
| List<LocalizedResourceProto> completedResources = |
| loadCompletedResources(pubts.getCompletedResourcesIterator()); |
| Map<LocalResourceProto, Path> startedResources = |
| loadStartedResources(pubts.getStartedResourcesIterator()); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(1, completedResources.size()); |
| assertEquals(pubLocalizedProto1, |
| completedResources.iterator().next()); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertTrue(userResources.isEmpty()); |
| } |
| |
| @Test |
| public void testRemoveLocalizedResourceForPrivateResource() |
| throws IOException { |
| String user = "somebody"; |
| |
| Path privRsrcPath = new Path("hdfs://some/private/resource"); |
| LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource |
| .newInstance( |
| URL.fromPath(privRsrcPath), |
| LocalResourceType.PATTERN, LocalResourceVisibility.PRIVATE, |
| 789L, 680L, "*pattern*"); |
| LocalResourceProto privRsrcProto = rsrcPb.getProto(); |
| Path privRsrcLocalPath = new Path("/some/local/dir/for/privrsrc"); |
| stateStore.startResourceLocalization(user, null, privRsrcProto, |
| privRsrcLocalPath); |
| stateStore.removeLocalizedResource(user, null, privRsrcLocalPath); |
| |
| // restart and verify state |
| restartStateStore(); |
| RecoveredLocalizationState state = stateStore.loadLocalizationState(); |
| LocalResourceTrackerState pubts = state.getPublicTrackerState(); |
| List<LocalizedResourceProto> completedResources = |
| loadCompletedResources(pubts.getCompletedResourcesIterator()); |
| Map<LocalResourceProto, Path> startedResources = |
| loadStartedResources(pubts.getStartedResourcesIterator()); |
| assertTrue(startedResources.isEmpty()); |
| assertEquals(0, completedResources.size()); |
| Map<String, RecoveredUserResources> userResources = |
| loadUserResources(state.getIterator()); |
| assertTrue(userResources.isEmpty()); |
| } |
| |
| @Test |
| public void testDeletionTaskStorage() throws IOException { |
| // test empty when no state |
| RecoveredDeletionServiceState state = |
| stateStore.loadDeletionServiceState(); |
| List<DeletionServiceDeleteTaskProto> deleteTaskProtos = |
| loadDeletionTaskProtos(state.getIterator()); |
| assertTrue(deleteTaskProtos.isEmpty()); |
| |
| // store a deletion task and verify recovered |
| DeletionServiceDeleteTaskProto proto = |
| DeletionServiceDeleteTaskProto.newBuilder() |
| .setId(7) |
| .setUser("someuser") |
| .setSubdir("some/subdir") |
| .addBasedirs("some/dir/path") |
| .addBasedirs("some/other/dir/path") |
| .setDeletionTime(123456L) |
| .addSuccessorIds(8) |
| .addSuccessorIds(9) |
| .build(); |
| stateStore.storeDeletionTask(proto.getId(), proto); |
| restartStateStore(); |
| state = stateStore.loadDeletionServiceState(); |
| deleteTaskProtos = loadDeletionTaskProtos(state.getIterator()); |
| assertEquals(1, deleteTaskProtos.size()); |
| assertEquals(proto, deleteTaskProtos.get(0)); |
| |
| // store another deletion task |
| DeletionServiceDeleteTaskProto proto2 = |
| DeletionServiceDeleteTaskProto.newBuilder() |
| .setId(8) |
| .setUser("user2") |
| .setSubdir("subdir2") |
| .setDeletionTime(789L) |
| .build(); |
| stateStore.storeDeletionTask(proto2.getId(), proto2); |
| restartStateStore(); |
| state = stateStore.loadDeletionServiceState(); |
| deleteTaskProtos = loadDeletionTaskProtos(state.getIterator()); |
| assertEquals(2, deleteTaskProtos.size()); |
| assertTrue(deleteTaskProtos.contains(proto)); |
| assertTrue(deleteTaskProtos.contains(proto2)); |
| |
| |
| // delete a task and verify gone after recovery |
| stateStore.removeDeletionTask(proto2.getId()); |
| restartStateStore(); |
| state = stateStore.loadDeletionServiceState(); |
| deleteTaskProtos = loadDeletionTaskProtos(state.getIterator()); |
| assertEquals(1, deleteTaskProtos.size()); |
| assertEquals(proto, deleteTaskProtos.get(0)); |
| |
| // delete the last task and verify none left |
| stateStore.removeDeletionTask(proto.getId()); |
| restartStateStore(); |
| state = stateStore.loadDeletionServiceState(); |
| deleteTaskProtos = loadDeletionTaskProtos(state.getIterator()); |
| assertTrue(deleteTaskProtos.isEmpty()); } |
| |
| @Test |
| public void testNMTokenStorage() throws IOException { |
| // test empty when no state |
| RecoveredNMTokensState state = stateStore.loadNMTokensState(); |
| Map<ApplicationAttemptId, MasterKey> loadedAppKeys = |
| loadNMTokens(state.getIterator()); |
| assertNull(state.getCurrentMasterKey()); |
| assertNull(state.getPreviousMasterKey()); |
| assertTrue(loadedAppKeys.isEmpty()); |
| |
| // store a master key and verify recovered |
| NMTokenSecretManagerForTest secretMgr = new NMTokenSecretManagerForTest(); |
| MasterKey currentKey = secretMgr.generateKey(); |
| stateStore.storeNMTokenCurrentMasterKey(currentKey); |
| restartStateStore(); |
| state = stateStore.loadNMTokensState(); |
| loadedAppKeys = loadNMTokens(state.getIterator()); |
| assertEquals(currentKey, state.getCurrentMasterKey()); |
| assertNull(state.getPreviousMasterKey()); |
| assertTrue(loadedAppKeys.isEmpty()); |
| |
| // store a previous key and verify recovered |
| MasterKey prevKey = secretMgr.generateKey(); |
| stateStore.storeNMTokenPreviousMasterKey(prevKey); |
| restartStateStore(); |
| state = stateStore.loadNMTokensState(); |
| loadedAppKeys = loadNMTokens(state.getIterator()); |
| assertEquals(currentKey, state.getCurrentMasterKey()); |
| assertEquals(prevKey, state.getPreviousMasterKey()); |
| assertTrue(loadedAppKeys.isEmpty()); |
| |
| // store a few application keys and verify recovered |
| ApplicationAttemptId attempt1 = ApplicationAttemptId.newInstance( |
| ApplicationId.newInstance(1, 1), 1); |
| MasterKey attemptKey1 = secretMgr.generateKey(); |
| stateStore.storeNMTokenApplicationMasterKey(attempt1, attemptKey1); |
| ApplicationAttemptId attempt2 = ApplicationAttemptId.newInstance( |
| ApplicationId.newInstance(2, 3), 4); |
| MasterKey attemptKey2 = secretMgr.generateKey(); |
| stateStore.storeNMTokenApplicationMasterKey(attempt2, attemptKey2); |
| restartStateStore(); |
| state = stateStore.loadNMTokensState(); |
| loadedAppKeys = loadNMTokens(state.getIterator()); |
| assertEquals(currentKey, state.getCurrentMasterKey()); |
| assertEquals(prevKey, state.getPreviousMasterKey()); |
| assertEquals(2, loadedAppKeys.size()); |
| assertEquals(attemptKey1, loadedAppKeys.get(attempt1)); |
| assertEquals(attemptKey2, loadedAppKeys.get(attempt2)); |
| |
| // add/update/remove keys and verify recovered |
| ApplicationAttemptId attempt3 = ApplicationAttemptId.newInstance( |
| ApplicationId.newInstance(5, 6), 7); |
| MasterKey attemptKey3 = secretMgr.generateKey(); |
| stateStore.storeNMTokenApplicationMasterKey(attempt3, attemptKey3); |
| stateStore.removeNMTokenApplicationMasterKey(attempt1); |
| attemptKey2 = prevKey; |
| stateStore.storeNMTokenApplicationMasterKey(attempt2, attemptKey2); |
| prevKey = currentKey; |
| stateStore.storeNMTokenPreviousMasterKey(prevKey); |
| currentKey = secretMgr.generateKey(); |
| stateStore.storeNMTokenCurrentMasterKey(currentKey); |
| restartStateStore(); |
| state = stateStore.loadNMTokensState(); |
| loadedAppKeys = loadNMTokens(state.getIterator()); |
| assertEquals(currentKey, state.getCurrentMasterKey()); |
| assertEquals(prevKey, state.getPreviousMasterKey()); |
| assertEquals(2, loadedAppKeys.size()); |
| assertNull(loadedAppKeys.get(attempt1)); |
| assertEquals(attemptKey2, loadedAppKeys.get(attempt2)); |
| assertEquals(attemptKey3, loadedAppKeys.get(attempt3)); |
| } |
| |
| @Test |
| public void testContainerTokenStorage() throws IOException { |
| // test empty when no state |
| RecoveredContainerTokensState state = |
| stateStore.loadContainerTokensState(); |
| Map<ContainerId, Long> loadedActiveTokens = loadContainerTokens(state.it); |
| assertNull(state.getCurrentMasterKey()); |
| assertNull(state.getPreviousMasterKey()); |
| assertTrue(loadedActiveTokens.isEmpty()); |
| |
| // store a master key and verify recovered |
| ContainerTokenKeyGeneratorForTest keygen = |
| new ContainerTokenKeyGeneratorForTest(new YarnConfiguration()); |
| MasterKey currentKey = keygen.generateKey(); |
| stateStore.storeContainerTokenCurrentMasterKey(currentKey); |
| restartStateStore(); |
| state = stateStore.loadContainerTokensState(); |
| loadedActiveTokens = loadContainerTokens(state.it); |
| assertEquals(currentKey, state.getCurrentMasterKey()); |
| assertNull(state.getPreviousMasterKey()); |
| assertTrue(loadedActiveTokens.isEmpty()); |
| |
| // store a previous key and verify recovered |
| MasterKey prevKey = keygen.generateKey(); |
| stateStore.storeContainerTokenPreviousMasterKey(prevKey); |
| restartStateStore(); |
| state = stateStore.loadContainerTokensState(); |
| loadedActiveTokens = loadContainerTokens(state.it); |
| assertEquals(currentKey, state.getCurrentMasterKey()); |
| assertEquals(prevKey, state.getPreviousMasterKey()); |
| assertTrue(loadedActiveTokens.isEmpty()); |
| |
| // store a few container tokens and verify recovered |
| ContainerId cid1 = BuilderUtils.newContainerId(1, 1, 1, 1); |
| Long expTime1 = 1234567890L; |
| ContainerId cid2 = BuilderUtils.newContainerId(2, 2, 2, 2); |
| Long expTime2 = 9876543210L; |
| stateStore.storeContainerToken(cid1, expTime1); |
| stateStore.storeContainerToken(cid2, expTime2); |
| restartStateStore(); |
| state = stateStore.loadContainerTokensState(); |
| loadedActiveTokens = loadContainerTokens(state.it); |
| assertEquals(currentKey, state.getCurrentMasterKey()); |
| assertEquals(prevKey, state.getPreviousMasterKey()); |
| assertEquals(2, loadedActiveTokens.size()); |
| assertEquals(expTime1, loadedActiveTokens.get(cid1)); |
| assertEquals(expTime2, loadedActiveTokens.get(cid2)); |
| |
| // add/update/remove tokens and verify recovered |
| ContainerId cid3 = BuilderUtils.newContainerId(3, 3, 3, 3); |
| Long expTime3 = 135798642L; |
| stateStore.storeContainerToken(cid3, expTime3); |
| stateStore.removeContainerToken(cid1); |
| expTime2 += 246897531L; |
| stateStore.storeContainerToken(cid2, expTime2); |
| prevKey = currentKey; |
| stateStore.storeContainerTokenPreviousMasterKey(prevKey); |
| currentKey = keygen.generateKey(); |
| stateStore.storeContainerTokenCurrentMasterKey(currentKey); |
| restartStateStore(); |
| state = stateStore.loadContainerTokensState(); |
| loadedActiveTokens = loadContainerTokens(state.it); |
| assertEquals(currentKey, state.getCurrentMasterKey()); |
| assertEquals(prevKey, state.getPreviousMasterKey()); |
| assertEquals(2, loadedActiveTokens.size()); |
| assertNull(loadedActiveTokens.get(cid1)); |
| assertEquals(expTime2, loadedActiveTokens.get(cid2)); |
| assertEquals(expTime3, loadedActiveTokens.get(cid3)); |
| } |
| |
| @Test |
| public void testLogDeleterStorage() throws IOException { |
| // test empty when no state |
| RecoveredLogDeleterState state = stateStore.loadLogDeleterState(); |
| assertTrue(state.getLogDeleterMap().isEmpty()); |
| |
| // store log deleter state |
| final ApplicationId appId1 = ApplicationId.newInstance(1, 1); |
| LogDeleterProto proto1 = LogDeleterProto.newBuilder() |
| .setUser("user1") |
| .setDeletionTime(1234) |
| .build(); |
| stateStore.storeLogDeleter(appId1, proto1); |
| |
| // restart state store and verify recovered |
| restartStateStore(); |
| state = stateStore.loadLogDeleterState(); |
| assertEquals(1, state.getLogDeleterMap().size()); |
| assertEquals(proto1, state.getLogDeleterMap().get(appId1)); |
| |
| // store another log deleter |
| final ApplicationId appId2 = ApplicationId.newInstance(2, 2); |
| LogDeleterProto proto2 = LogDeleterProto.newBuilder() |
| .setUser("user2") |
| .setDeletionTime(5678) |
| .build(); |
| stateStore.storeLogDeleter(appId2, proto2); |
| |
| // restart state store and verify recovered |
| restartStateStore(); |
| state = stateStore.loadLogDeleterState(); |
| assertEquals(2, state.getLogDeleterMap().size()); |
| assertEquals(proto1, state.getLogDeleterMap().get(appId1)); |
| assertEquals(proto2, state.getLogDeleterMap().get(appId2)); |
| |
| // remove a deleter and verify removed after restart and recovery |
| stateStore.removeLogDeleter(appId1); |
| restartStateStore(); |
| state = stateStore.loadLogDeleterState(); |
| assertEquals(1, state.getLogDeleterMap().size()); |
| assertEquals(proto2, state.getLogDeleterMap().get(appId2)); |
| |
| // remove last deleter and verify empty after restart and recovery |
| stateStore.removeLogDeleter(appId2); |
| restartStateStore(); |
| state = stateStore.loadLogDeleterState(); |
| assertTrue(state.getLogDeleterMap().isEmpty()); |
| } |
| |
| @Test |
| public void testCompactionCycle() throws IOException { |
| final DB mockdb = mock(DB.class); |
| conf.setInt(YarnConfiguration.NM_RECOVERY_COMPACTION_INTERVAL_SECS, 1); |
| NMLeveldbStateStoreService store = new NMLeveldbStateStoreService() { |
| @Override |
| protected void checkVersion() {} |
| |
| @Override |
| protected DB openDatabase(Configuration conf) { |
| return mockdb; |
| } |
| }; |
| store.init(conf); |
| store.start(); |
| verify(mockdb, timeout(10000).atLeastOnce()).compactRange( |
| (byte[]) isNull(), (byte[]) isNull()); |
| store.close(); |
| } |
| |
| @Test |
| public void testUnexpectedKeyDoesntThrowException() throws IOException { |
| // test empty when no state |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertTrue(recoveredContainers.isEmpty()); |
| |
| ApplicationId appId = ApplicationId.newInstance(1234, 3); |
| ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId, |
| 4); |
| ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5); |
| StartContainerRequest startContainerRequest = storeMockContainer( |
| containerId); |
| |
| // add a invalid key |
| byte[] invalidKey = ("ContainerManager/containers/" |
| + containerId.toString() + "/invalidKey1234").getBytes(); |
| stateStore.getDB().put(invalidKey, new byte[1]); |
| restartStateStore(); |
| recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| assertEquals(RecoveredContainerStatus.REQUESTED, rcs.getStatus()); |
| assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode()); |
| assertEquals(false, rcs.getKilled()); |
| assertEquals(startContainerRequest, rcs.getStartRequest()); |
| assertTrue(rcs.getDiagnostics().isEmpty()); |
| assertEquals(RecoveredContainerType.KILL, rcs.getRecoveryType()); |
| // assert unknown keys are cleaned up finally |
| assertNotNull(stateStore.getDB().get(invalidKey)); |
| stateStore.removeContainer(containerId); |
| assertNull(stateStore.getDB().get(invalidKey)); |
| } |
| |
| @Test |
| public void testAMRMProxyStorage() throws IOException { |
| RecoveredAMRMProxyState state = stateStore.loadAMRMProxyState(); |
| assertThat(state.getCurrentMasterKey()).isNull(); |
| assertThat(state.getNextMasterKey()).isNull(); |
| assertThat(state.getAppContexts()).isEmpty(); |
| |
| ApplicationId appId1 = ApplicationId.newInstance(1, 1); |
| ApplicationId appId2 = ApplicationId.newInstance(1, 2); |
| ApplicationAttemptId attemptId1 = |
| ApplicationAttemptId.newInstance(appId1, 1); |
| ApplicationAttemptId attemptId2 = |
| ApplicationAttemptId.newInstance(appId2, 2); |
| String key1 = "key1"; |
| String key2 = "key2"; |
| byte[] data1 = "data1".getBytes(); |
| byte[] data2 = "data2".getBytes(); |
| |
| AMRMProxyTokenSecretManager secretManager = |
| new AMRMProxyTokenSecretManager(stateStore); |
| secretManager.init(conf); |
| // Generate currentMasterKey |
| secretManager.start(); |
| |
| try { |
| // Add two applications, each with two data entries |
| stateStore.storeAMRMProxyAppContextEntry(attemptId1, key1, data1); |
| stateStore.storeAMRMProxyAppContextEntry(attemptId2, key1, data1); |
| stateStore.storeAMRMProxyAppContextEntry(attemptId1, key2, data2); |
| stateStore.storeAMRMProxyAppContextEntry(attemptId2, key2, data2); |
| |
| // restart state store and verify recovered |
| restartStateStore(); |
| secretManager.setNMStateStoreService(stateStore); |
| state = stateStore.loadAMRMProxyState(); |
| assertEquals(state.getCurrentMasterKey(), |
| secretManager.getCurrentMasterKeyData().getMasterKey()); |
| assertThat(state.getNextMasterKey()).isNull(); |
| assertThat(state.getAppContexts()).hasSize(2); |
| // app1 |
| Map<String, byte[]> map = state.getAppContexts().get(attemptId1); |
| assertNotEquals(map, null); |
| assertThat(map).hasSize(2); |
| assertTrue(Arrays.equals(map.get(key1), data1)); |
| assertTrue(Arrays.equals(map.get(key2), data2)); |
| // app2 |
| map = state.getAppContexts().get(attemptId2); |
| assertNotEquals(map, null); |
| assertThat(map).hasSize(2); |
| assertTrue(Arrays.equals(map.get(key1), data1)); |
| assertTrue(Arrays.equals(map.get(key2), data2)); |
| |
| // Generate next master key and remove one entry of app2 |
| secretManager.rollMasterKey(); |
| stateStore.removeAMRMProxyAppContextEntry(attemptId2, key1); |
| |
| // restart state store and verify recovered |
| restartStateStore(); |
| secretManager.setNMStateStoreService(stateStore); |
| state = stateStore.loadAMRMProxyState(); |
| assertEquals(state.getCurrentMasterKey(), |
| secretManager.getCurrentMasterKeyData().getMasterKey()); |
| assertEquals(state.getNextMasterKey(), |
| secretManager.getNextMasterKeyData().getMasterKey()); |
| assertEquals(state.getAppContexts().size(), 2); |
| // app1 |
| map = state.getAppContexts().get(attemptId1); |
| assertThat(map).isNotNull(); |
| assertThat(map).hasSize(2); |
| assertTrue(Arrays.equals(map.get(key1), data1)); |
| assertTrue(Arrays.equals(map.get(key2), data2)); |
| // app2 |
| map = state.getAppContexts().get(attemptId2); |
| assertThat(map).isNotNull(); |
| assertThat(map).hasSize(1); |
| assertTrue(Arrays.equals(map.get(key2), data2)); |
| |
| // Activate next master key and remove all entries of app1 |
| secretManager.activateNextMasterKey(); |
| stateStore.removeAMRMProxyAppContext(attemptId1); |
| |
| // restart state store and verify recovered |
| restartStateStore(); |
| secretManager.setNMStateStoreService(stateStore); |
| state = stateStore.loadAMRMProxyState(); |
| assertEquals(state.getCurrentMasterKey(), |
| secretManager.getCurrentMasterKeyData().getMasterKey()); |
| assertThat(state.getNextMasterKey()).isNull(); |
| assertThat(state.getAppContexts()).hasSize(1); |
| // app2 only |
| map = state.getAppContexts().get(attemptId2); |
| assertThat(map).isNotNull(); |
| assertThat(map).hasSize(1); |
| assertTrue(Arrays.equals(map.get(key2), data2)); |
| } finally { |
| secretManager.stop(); |
| } |
| } |
| |
| @Test |
| public void testStateStoreForResourceMapping() throws IOException { |
| // test that stateStore is initially empty |
| List<RecoveredContainerState> recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertTrue(recoveredContainers.isEmpty()); |
| |
| ApplicationId appId = ApplicationId.newInstance(1234, 3); |
| ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId, |
| 4); |
| ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5); |
| storeMockContainer(containerId); |
| |
| Container container = mock(Container.class); |
| when(container.getContainerId()).thenReturn(containerId); |
| ResourceMappings resourceMappings = new ResourceMappings(); |
| when(container.getResourceMappings()).thenReturn(resourceMappings); |
| |
| stateStore.storeAssignedResources(container, "gpu", |
| Arrays.asList(new GpuDevice(1, 1), new GpuDevice(2, 2), |
| new GpuDevice(3, 3))); |
| |
| // This will overwrite the above |
| List<Serializable> gpuRes1 = Arrays.asList( |
| new GpuDevice(1, 1), new GpuDevice(2, 2), new GpuDevice(4, 4)); |
| stateStore.storeAssignedResources(container, "gpu", gpuRes1); |
| |
| List<Serializable> fpgaRes = Arrays.asList( |
| new FpgaDevice("testType", 3, 3, "testIPID"), |
| new FpgaDevice("testType", 4, 4, "testIPID"), |
| new FpgaDevice("testType", 5, 5, "testIPID"), |
| new FpgaDevice("testType", 6, 6, "testIPID")); |
| stateStore.storeAssignedResources(container, "fpga", fpgaRes); |
| |
| List<Serializable> numaRes = Arrays.asList( |
| new NumaResourceAllocation("testmemNodeId", 2048, "testCpuNodeId", 10)); |
| stateStore.storeAssignedResources(container, "numa", numaRes); |
| |
| restartStateStore(); |
| recoveredContainers = |
| loadContainersState(stateStore.getContainerStateIterator()); |
| assertEquals(1, recoveredContainers.size()); |
| RecoveredContainerState rcs = recoveredContainers.get(0); |
| List<Serializable> resources = rcs.getResourceMappings() |
| .getAssignedResources("gpu"); |
| Assert.assertEquals(gpuRes1, resources); |
| Assert.assertEquals(gpuRes1, resourceMappings.getAssignedResources("gpu")); |
| |
| resources = rcs.getResourceMappings().getAssignedResources("fpga"); |
| Assert.assertEquals(fpgaRes, resources); |
| Assert.assertEquals(fpgaRes, resourceMappings.getAssignedResources("fpga")); |
| |
| resources = rcs.getResourceMappings().getAssignedResources("numa"); |
| Assert.assertEquals(numaRes, resources); |
| Assert.assertEquals(numaRes, resourceMappings.getAssignedResources("numa")); |
| // test removing numa resources from state store |
| stateStore.releaseAssignedResources(containerId, "numa"); |
| recoveredContainers = loadContainersState(stateStore.getContainerStateIterator()); |
| resourceMappings = recoveredContainers.get(0).getResourceMappings(); |
| assertTrue(resourceMappings.getAssignedResources("numa").isEmpty()); |
| |
| // testing calling deletion of non-existing key doesn't break anything |
| try { |
| stateStore.releaseAssignedResources(containerId, "numa"); |
| }catch (RuntimeException e){ |
| Assert.fail("Should not throw exception while deleting non existing key from statestore"); |
| } |
| } |
| |
| @Test |
| public void testStateStoreNodeHealth() throws IOException { |
| // keep the working DB clean, break a temp DB |
| DB keepDB = stateStore.getDB(); |
| DB myMocked = mock(DB.class); |
| stateStore.setDB(myMocked); |
| |
| ApplicationId appId = ApplicationId.newInstance(1234, 1); |
| ApplicationAttemptId appAttemptId = |
| ApplicationAttemptId.newInstance(appId, 1); |
| DBException toThrow = new DBException(); |
| Mockito.doThrow(toThrow).when(myMocked). |
| put(any(byte[].class), any(byte[].class)); |
| // write some data |
| try { |
| // chosen a simple method could be any of the "void" methods |
| ContainerId containerId = ContainerId.newContainerId(appAttemptId, 1); |
| stateStore.storeContainerKilled(containerId); |
| } catch (IOException ioErr) { |
| // Cause should be wrapped DBException |
| assertTrue(ioErr.getCause() instanceof DBException); |
| // check the store is marked unhealthy |
| assertFalse("Statestore should have been unhealthy", |
| stateStore.isHealthy()); |
| return; |
| } finally { |
| // restore the working DB |
| stateStore.setDB(keepDB); |
| } |
| Assert.fail("Expected exception not thrown"); |
| } |
| |
| @Test |
| public void testEmptyRestartTimes() throws IOException { |
| List<Long> restartTimes = new ArrayList<>(); |
| ApplicationId appId = ApplicationId.newInstance(1234, 3); |
| ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId, |
| 4); |
| ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5); |
| storeMockContainer(containerId); |
| stateStore.storeContainerRestartTimes(containerId, |
| restartTimes); |
| restartStateStore(); |
| RecoveredContainerState rcs = |
| loadContainersState(stateStore.getContainerStateIterator()).get(0); |
| List<Long> recoveredRestartTimes = rcs.getRestartTimes(); |
| assertTrue(recoveredRestartTimes.isEmpty()); |
| } |
| |
| private StartContainerRequest storeMockContainer(ContainerId containerId) |
| throws IOException { |
| // create a container request |
| LocalResource lrsrc = LocalResource.newInstance( |
| URL.newInstance("hdfs", "somehost", 12345, "/some/path/to/rsrc"), |
| LocalResourceType.FILE, LocalResourceVisibility.APPLICATION, 123L, |
| 1234567890L); |
| Map<String, LocalResource> localResources = |
| new HashMap<String, LocalResource>(); |
| localResources.put("rsrc", lrsrc); |
| Map<String, String> env = new HashMap<String, String>(); |
| env.put("somevar", "someval"); |
| List<String> containerCmds = new ArrayList<String>(); |
| containerCmds.add("somecmd"); |
| containerCmds.add("somearg"); |
| Map<String, ByteBuffer> serviceData = new HashMap<String, ByteBuffer>(); |
| serviceData.put("someservice", |
| ByteBuffer.wrap(new byte[] {0x1, 0x2, 0x3})); |
| ByteBuffer containerTokens = ByteBuffer |
| .wrap(new byte[] {0x7, 0x8, 0x9, 0xa}); |
| Map<ApplicationAccessType, String> acls = |
| new HashMap<ApplicationAccessType, String>(); |
| acls.put(ApplicationAccessType.VIEW_APP, "viewuser"); |
| acls.put(ApplicationAccessType.MODIFY_APP, "moduser"); |
| ContainerLaunchContext clc = ContainerLaunchContext.newInstance( |
| localResources, env, containerCmds, |
| serviceData, containerTokens, acls); |
| Resource containerRsrc = Resource.newInstance(1357, 3); |
| ContainerTokenIdentifier containerTokenId = new ContainerTokenIdentifier( |
| containerId, "host", "user", containerRsrc, 9876543210L, 42, 2468, |
| Priority.newInstance(7), 13579); |
| Token containerToken = Token.newInstance(containerTokenId.getBytes(), |
| ContainerTokenIdentifier.KIND.toString(), "password".getBytes(), |
| "tokenservice"); |
| StartContainerRequest containerReq = StartContainerRequest.newInstance(clc, |
| containerToken); |
| stateStore.storeContainer(containerId, 0, 0, containerReq); |
| return containerReq; |
| } |
| |
| private static class NMTokenSecretManagerForTest extends |
| BaseNMTokenSecretManager { |
| public MasterKey generateKey() { |
| return createNewMasterKey().getMasterKey(); |
| } |
| } |
| |
| private static class ContainerTokenKeyGeneratorForTest extends |
| BaseContainerTokenSecretManager { |
| public ContainerTokenKeyGeneratorForTest(Configuration conf) { |
| super(conf); |
| } |
| |
| public MasterKey generateKey() { |
| return createNewMasterKey().getMasterKey(); |
| } |
| } |
| } |
