diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java index b62702abc5a..4805f51957b 100644 --- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java +++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java @@ -270,6 +270,10 @@ public class DaemonConfig implements Validated { * it takes to delete an inbox jar file is going to be somewhat more than NIMBUS_CLEANUP_INBOX_JAR_EXPIRATION_SECS * (depending on how often NIMBUS_CLEANUP_FREQ_SECS is set to). * + *

This is also how long a dependency blob uploaded by a client may go without any topology referring to it + * before nimbus deletes it from the blob store, which covers the time between uploading the dependencies of a + * topology and submitting it. + * * @see #NIMBUS_CLEANUP_INBOX_FREQ_SECS */ @IsInteger diff --git a/storm-server/src/main/java/org/apache/storm/blobstore/KeySequenceNumber.java b/storm-server/src/main/java/org/apache/storm/blobstore/KeySequenceNumber.java index 39e1747259c..ae0f998603b 100644 --- a/storm-server/src/main/java/org/apache/storm/blobstore/KeySequenceNumber.java +++ b/storm-server/src/main/java/org/apache/storm/blobstore/KeySequenceNumber.java @@ -124,10 +124,27 @@ public KeySequenceNumber(String key, NimbusInfo nimbusInfo) { } public synchronized int getKeySequenceNumber(CuratorFramework zkClient) throws KeyNotFoundException { + return getKeySequenceNumber(zkClient, true); + } + + /** + * Hand over the sequence number for the copy of the key held by this nimbus. + * + * @param zkClient the zookeeper client + * @param mayCreateKey whether the key may be created when zookeeper does not know it, which is only right for the + * leader storing a blob that a client uploads. Any other nimbus mirrors a key the leader created, + * so a key zookeeper does not know was deleted and must not be registered again. + * @return the sequence number + * @throws KeyNotFoundException if the key is not in zookeeper and may not be created, or it is deleted meanwhile + */ + public synchronized int getKeySequenceNumber(CuratorFramework zkClient, boolean mayCreateKey) throws KeyNotFoundException { TreeSet sequenceNumbers = new TreeSet(); try { // Key has not been created yet and it is the first time it is being created if (zkClient.checkExists().forPath(BlobStoreUtils.getBlobStoreSubtree() + "/" + key) == null) { + if (!mayCreateKey) { + throw new KeeperException.NoNodeException(BlobStoreUtils.getBlobStoreSubtree() + "/" + key); + } zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT) .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(BLOBSTORE_MAX_KEY_SEQUENCE_SUBTREE + "/" + key); zkClient.setData().forPath(BLOBSTORE_MAX_KEY_SEQUENCE_SUBTREE + "/" + key, diff --git a/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java b/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java index 5ec0b2357a4..a5d5c5bcf8e 100644 --- a/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java +++ b/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java @@ -139,7 +139,7 @@ private void setupBlobstore() throws AuthorizationException, KeyNotFoundExceptio LOG.debug("Creating list of key entries for blobstore inside zookeeper {} local {}", activeKeys, activeLocalKeys); for (String key : activeLocalKeys) { try { - state.setupBlob(key, nimbusInfo, getVersionForKey(key, nimbusInfo, zkClient)); + state.setupBlob(key, nimbusInfo, getVersionForKey(key, nimbusInfo, zkClient, false)); } catch (KeyNotFoundException e) { // invalid key, remove it from blobstore store.deleteBlob(key, NIMBUS_SUBJECT); @@ -147,6 +147,18 @@ private void setupBlobstore() throws AuthorizationException, KeyNotFoundExceptio } } + /** + * Tell whether this nimbus is the leader, which is the only one that may register a key zookeeper does not know. + * Without a leader elector there is a single nimbus, which is then the leader. + */ + private boolean isLeader() { + try { + return leaderElector == null || leaderElector.isLeader(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + private void blobSync() throws Exception { if ("distributed".equals(conf.get(Config.STORM_CLUSTER_MODE))) { @@ -216,11 +228,14 @@ public AtomicOutputStream createBlob(String key, SettableBlobMeta meta, Subject } BlobStoreFileOutputStream outputStream = null; try { + //Taken before anything is written, so that a non-leader downloading a key that was deleted meanwhile is + //refused without leaving a copy behind. + int version = getVersionForKey(key, this.nimbusInfo, zkClient, isLeader()); outputStream = new BlobStoreFileOutputStream(fbs.write(META_PREFIX + key, true)); outputStream.write(Utils.thriftSerialize(meta)); outputStream.close(); outputStream = null; - this.stormClusterState.setupBlob(key, this.nimbusInfo, getVersionForKey(key, this.nimbusInfo, zkClient)); + this.stormClusterState.setupBlob(key, this.nimbusInfo, version); return new BlobStoreFileOutputStream(fbs.write(DATA_PREFIX + key, true)); } catch (IOException e) { throw new RuntimeException(e); diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java index cc08fe2201f..75ef22c6a05 100644 --- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java +++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java @@ -440,6 +440,8 @@ public static List getNimbusAcls(Map conf) { private final TimeCacheMap uploaders; private final BlobStore blobStore; private final TopoCache topoCache; + //When a dependency blob was first seen unreferenced by any topology, only used by the cleanup pass. + private final Map orphanedDependencyKeysDetectedMs = new HashMap<>(); @SuppressWarnings("deprecation") private final TimeCacheMap blobDownloaders; @SuppressWarnings("deprecation") @@ -775,8 +777,18 @@ static List getKeyListFromId(Map conf, String id) { public static int getVersionForKey(String key, NimbusInfo nimbusInfo, CuratorFramework zkClient) throws KeyNotFoundException { + return getVersionForKey(key, nimbusInfo, zkClient, true); + } + + /** + * Get the version to register the copy of a blob held by a nimbus under. + * + * @see KeySequenceNumber#getKeySequenceNumber(CuratorFramework, boolean) + */ + public static int getVersionForKey(String key, NimbusInfo nimbusInfo, + CuratorFramework zkClient, boolean mayCreateKey) throws KeyNotFoundException { KeySequenceNumber kseq = new KeySequenceNumber(key, nimbusInfo); - return kseq.getKeySequenceNumber(zkClient); + return kseq.getKeySequenceNumber(zkClient, mayCreateKey); } private static StormTopology readStormTopology(String topoId, TopoCache tc) throws KeyNotFoundException, AuthorizationException, @@ -3095,6 +3107,45 @@ public void forceDeleteTopoDistDir(String topoId) throws IOException { Utils.forceDelete(ServerConfigUtils.masterStormDistRoot(conf, topoId)); } + /** + * Remove the dependency blobs that no topology has referred to for longer than the inbox jar expiration. + * + *

{@link #rmDependencyBlobsInTopology} only runs in the pass that cleans up the owning topology, and a dependency + * blob key carries no topology id, so a blob that outlives that pass can never be traced back to it. That happens + * when a submission fails after its dependencies were uploaded, or when another nimbus still holds a copy of the blob + * and it is downloaded back after it was removed here. This sweep reclaims those. + * + *

Only keys that are provably unique to one topology are considered: an older client that finds a shareable key + * in the store does not upload it again but refers to it, so such a key may be about to be used. A client uploads the + * dependencies before it submits the topology, so a key is only removed once it has been seen unreferenced for + * {@link DaemonConfig#NIMBUS_INBOX_JAR_EXPIRATION_SECS}, the time nimbus gives an uploaded topology jar to be + * submitted. When a key was first seen unreferenced is only kept in memory, so a new leader starts the wait over. + * + * @param referenced the dependency blob keys referenced by topologies that are not being cleaned up + */ + @VisibleForTesting + void sweepOrphanedDependencyBlobs(Set referenced) { + try { + long graceMs = TimeUnit.SECONDS.toMillis( + ObjectReader.getInt(conf.get(DaemonConfig.NIMBUS_INBOX_JAR_EXPIRATION_SECS), 3600)); + long nowMs = Time.currentTimeMillis(); + Set orphaned = blobStore.filterAndListKeys( + key -> isProvablyUniqueDependencyKey(key) && !referenced.contains(key) ? key : null); + //Forget the keys that are gone or referenced again, so that being orphaned later waits the full time again. + orphanedDependencyKeysDetectedMs.keySet().retainAll(orphaned); + for (String key : orphaned) { + long unreferencedMs = nowMs - orphanedDependencyKeysDetectedMs.computeIfAbsent(key, k -> nowMs); + if (unreferencedMs >= graceMs) { + LOG.info("Removing dependency blob {}, no topology has referred to it for {} ms", key, unreferencedMs); + rmBlobKey(blobStore, key, stormClusterState); + orphanedDependencyKeysDetectedMs.remove(key); + } + } + } catch (Exception e) { + LOG.warn("Could not sweep the dependency blobs that no topology refers to", e); + } + } + /** * Cleanup topologies and Jars. */ @@ -3134,6 +3185,12 @@ public void doCleanup() { idToExecutors.getAndUpdate(new Dissoc<>(topoId)); } + //Catches the dependency blobs that outlived the pass that cleaned up their topology. Without knowing the + //references nothing can be told to be unused, so nothing is swept then. + if (stillReferenced != null) { + sweepOrphanedDependencyBlobs(stillReferenced); + } + long cleanupDurationMs = Time.deltaMs(cleanupStartMs); if (cleanupDurationMs > 10000) { LOG.warn("doCleanup is taking too long, topoIdSelectionDurationMs={}, cleanupDurationMs={}", @@ -4378,7 +4435,9 @@ public void createStateInZookeeper(String key) throws TException { BlobStore store = blobStore; NimbusInfo ni = nimbusHostPortInfo; if (store instanceof LocalFsBlobStore) { - state.setupBlob(key, ni, getVersionForKey(key, ni, zkClient)); + //A non-leader only registers its copy of a key the leader created. If zookeeper does not know the key + //any more it was deleted while the copy was downloaded, and registering it would bring it back. + state.setupBlob(key, ni, getVersionForKey(key, ni, zkClient, isLeader())); } LOG.debug("Created state in zookeeper {} {} {}", state, store, ni); } catch (KeyNotFoundException e) { diff --git a/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java b/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java new file mode 100644 index 00000000000..8b4e8a08ba5 --- /dev/null +++ b/storm-server/src/test/java/org/apache/storm/blobstore/KeySequenceNumberTest.java @@ -0,0 +1,96 @@ +/* + * 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.storm.blobstore; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import org.apache.storm.generated.KeyNotFoundException; +import org.apache.storm.nimbus.NimbusInfo; +import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework; +import org.apache.storm.shade.org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.storm.shade.org.apache.curator.retry.ExponentialBackoffRetry; +import org.apache.storm.testing.InProcessZookeeper; +import org.junit.jupiter.api.Test; + +class KeySequenceNumberTest { + private static final String KEY = "dep-lib-11111111-1111-1111-1111-111111111111.jar"; + private static final String KEY_PATH = "/blobstore/" + KEY; + private static final String MAX_SEQUENCE_PATH = "/blobstoremaxkeysequencenumber/" + KEY; + private static final NimbusInfo LEADER = new NimbusInfo("nimbus-1", 6627, false); + private static final NimbusInfo PEER = new NimbusInfo("nimbus-2", 6627, false); + + /** + * Replays the blob store state changes behind the nimbus logs reported on STORM-3871. A non-leader that downloaded + * a blob while the leader deleted it registered the key again as new, and every other nimbus, the leader included, + * then downloaded the blob back from it. + */ + @Test + void aNonLeaderCannotRegisterAKeyAgainThatWasDeletedWhileItDownloadedIt() throws Exception { + try (InProcessZookeeper zk = new InProcessZookeeper(); + CuratorFramework zkClient = newClient(zk)) { + // the client uploads the dependency: createBlob, then createStateInZookeeper when it closes the stream + assertEquals(1, register(zkClient, LEADER, true)); + assertEquals(2, register(zkClient, LEADER, true)); + + // the topology is cleaned up and the leader deletes the blob, as LocalFsBlobStore#deleteBlob does + zkClient.delete().deletingChildrenIfNeeded().forPath(KEY_PATH); + zkClient.delete().deletingChildrenIfNeeded().forPath(MAX_SEQUENCE_PATH); + + // the non-leader finishes its download and would register the key as if it were new + assertThrows(KeyNotFoundException.class, () -> register(zkClient, PEER, false)); + assertNull(zkClient.checkExists().forPath(KEY_PATH)); + assertNull(zkClient.checkExists().forPath(MAX_SEQUENCE_PATH)); + } + } + + @Test + void aNonLeaderStillRegistersItsCopyOfAKeyTheLeaderCreated() throws Exception { + try (InProcessZookeeper zk = new InProcessZookeeper(); + CuratorFramework zkClient = newClient(zk)) { + assertEquals(1, register(zkClient, LEADER, true)); + + assertEquals(0, register(zkClient, PEER, false)); + } + } + + private static CuratorFramework newClient(InProcessZookeeper zk) { + CuratorFramework zkClient = CuratorFrameworkFactory.newClient("localhost:" + zk.getPort(), + new ExponentialBackoffRetry(1000, 3)); + zkClient.start(); + return zkClient; + } + + /** + * Do what IStormClusterState#setupBlob does with the version KeySequenceNumber hands out. + */ + private static int register(CuratorFramework zkClient, NimbusInfo nimbus, boolean mayCreateKey) throws Exception { + int version = new KeySequenceNumber(KEY, nimbus).getKeySequenceNumber(zkClient, mayCreateKey); + if (zkClient.checkExists().forPath(KEY_PATH) != null) { + for (String child : zkClient.getChildren().forPath(KEY_PATH)) { + if (child.startsWith(nimbus.toHostPortString())) { + zkClient.delete().forPath(KEY_PATH + "/" + child); + } + } + } + zkClient.create().creatingParentsIfNeeded().forPath(KEY_PATH + "/" + nimbus.toHostPortString() + "-" + version); + return version; + } +} diff --git a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java index e27b5b4ec2d..0b5215c6f14 100644 --- a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java +++ b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java @@ -39,6 +39,7 @@ import org.apache.storm.DaemonConfig; import org.apache.storm.blobstore.BlobStore; import org.apache.storm.blobstore.BlobStoreAclHandler; +import org.apache.storm.blobstore.KeyFilter; import org.apache.storm.blobstore.KeySequenceNumber; import org.apache.storm.blobstore.LocalFsBlobStore; import org.apache.storm.cluster.IStormClusterState; @@ -93,6 +94,7 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doThrow; @@ -265,14 +267,28 @@ void testCreateStateInZookeeperWhenFailToSetupBlobWithRuntimeExceptionThrowsRunt @Test void testCreateStateInZookeeperWhenKeyNotFoundHandlesException() throws Exception { try (MockedConstruction keySequenceNumber = mockConstruction(KeySequenceNumber.class, (mock, context) -> - when(mock.getKeySequenceNumber(any())).thenThrow(new KeyNotFoundException("Failed to setup blob")))) { + when(mock.getKeySequenceNumber(any(), anyBoolean())).thenThrow(new KeyNotFoundException("Failed to setup blob")))) { nimbus.createStateInZookeeper(BLOB_FILE_KEY); - verify(keySequenceNumber.constructed().get(0)).getKeySequenceNumber(any()); + verify(keySequenceNumber.constructed().get(0)).getKeySequenceNumber(any(), anyBoolean()); verify(stormClusterState, never()).setupBlob(eq(BLOB_FILE_KEY), eq(nimbusInfo), any()); } } + @Test + void testCreateStateInZookeeperOnlyLetsTheLeaderRegisterAKeyZookeeperDoesNotKnow() throws Exception { + try (MockedConstruction keySequenceNumber = mockConstruction(KeySequenceNumber.class)) { + when(leaderElector.isLeader()).thenReturn(false); + nimbus.createStateInZookeeper(BLOB_FILE_KEY); + when(leaderElector.isLeader()).thenReturn(true); + nimbus.createStateInZookeeper(BLOB_FILE_KEY); + + // a non-leader registering a key zookeeper does not know would bring back a key that was deleted + verify(keySequenceNumber.constructed().get(0)).getKeySequenceNumber(any(), eq(false)); + verify(keySequenceNumber.constructed().get(1)).getKeySequenceNumber(any(), eq(true)); + } + } + @Test void testListBlobsOnlyReturnsKeysTheCallerMayReadTheMetadataOf() throws Exception { when(localBlobStore.listKeys()).thenReturn(List.of("readable-key", "other-users-key").iterator()); @@ -575,6 +591,114 @@ void doCleanupContinuesTheReferenceScanWhenACandidateTopologyHasNoCodeBlob() thr verify(store, never()).deleteBlob(eq(LEGACY_ARTIFACT_KEY), any()); } + /** + * Make the blob store mock list exactly these keys. + */ + private static void storeKeys(BlobStore store, String... keys) { + when(store.filterAndListKeys(any())).thenAnswer(invocation -> { + KeyFilter filter = invocation.getArgument(0); + Set filtered = new HashSet<>(); + for (String key : keys) { + Object result = filter.filter(key); + if (result != null) { + filtered.add(result); + } + } + return filtered; + }); + } + + @Test + void doCleanupSweepsADependencyBlobNoTopologyRefersToOnceTheInboxJarExpirationHasPassed() throws Exception { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + BlobStore store = mock(BlobStore.class); + IStormClusterState state = mock(IStormClusterState.class); + when(store.storedTopoIds()).thenReturn(Set.of()); + when(state.activeStorms()).thenReturn(List.of()); + // outlived the pass that cleaned up its topology, e.g. because it was downloaded back from another nimbus + storeKeys(store, UNIQUE_JAR_KEY); + Nimbus nimbus = cleanupNimbus(store, state); + + nimbus.doCleanup(); + // one second short of the default nimbus.inbox.jar.expiration.secs, a submission may still be under way + Time.advanceTimeSecs(3599); + nimbus.doCleanup(); + verify(store, never()).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + + Time.advanceTimeSecs(1); + nimbus.doCleanup(); + verify(store).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + } + } + + @Test + void doCleanupNeverSweepsADependencyBlobThatIsReferencedOrCouldBeShared() throws Exception { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + BlobStore store = mock(BlobStore.class); + IStormClusterState state = mock(IStormClusterState.class); + when(store.storedTopoIds()).thenReturn(Set.of("live-topo")); + when(state.activeStorms()).thenReturn(List.of("live-topo")); + storeTopology(store, "live-topo", List.of(UNIQUE_JAR_KEY), List.of()); + // nothing refers to the legacy key, but an older client that finds it in the store refers to it without + // uploading it again, so it may be about to be used + storeKeys(store, UNIQUE_JAR_KEY, LEGACY_ARTIFACT_KEY); + Nimbus nimbus = cleanupNimbus(store, state); + + nimbus.doCleanup(); + Time.advanceTimeSecs(3600); + nimbus.doCleanup(); + + verify(store, never()).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + verify(store, never()).deleteBlob(eq(LEGACY_ARTIFACT_KEY), any()); + } + } + + @Test + void doCleanupDoesNotSweepDependencyBlobsWhenTheReferencesCannotBeRead() throws Exception { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + BlobStore store = mock(BlobStore.class); + IStormClusterState state = mock(IStormClusterState.class); + when(store.storedTopoIds()).thenReturn(Set.of("live-topo")); + when(state.activeStorms()).thenReturn(List.of("live-topo")); + // the live topology's code blob cannot be read, so it may be the one referring to the key + when(store.readBlob(eq(ConfigUtils.masterStormCodeKey("live-topo")), any())) + .thenThrow(new IOException("blob store is unhappy")); + storeKeys(store, UNIQUE_JAR_KEY); + Nimbus nimbus = cleanupNimbus(store, state); + + nimbus.doCleanup(); + Time.advanceTimeSecs(3600); + nimbus.doCleanup(); + + verify(store, never()).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + } + } + + @Test + void aDependencyBlobThatComesBackAfterItsTopologyWasCleanedUpIsRemovedAgain() throws Exception { + try (Time.SimulatedTime ignored = new Time.SimulatedTime()) { + BlobStore store = mock(BlobStore.class); + IStormClusterState state = mock(IStormClusterState.class); + when(store.storedTopoIds()).thenReturn(Set.of("dead-topo")); + when(state.activeStorms()).thenReturn(List.of()); + storeTopology(store, "dead-topo", List.of(UNIQUE_JAR_KEY), List.of()); + Nimbus nimbus = cleanupNimbus(store, state); + + nimbus.doCleanup(); + verify(store).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + + // the topology is gone for good, but another nimbus still had a copy of the blob and it was downloaded + // back, so the pass that knew which topology it belonged to is over + when(store.storedTopoIds()).thenReturn(Set.of()); + storeKeys(store, UNIQUE_JAR_KEY); + nimbus.doCleanup(); + Time.advanceTimeSecs(3600); + nimbus.doCleanup(); + + verify(store, times(2)).deleteBlob(eq(UNIQUE_JAR_KEY), any()); + } + } + @Test void everyDependencyKeyACurrentClientGeneratesIsRecognisedAsUniqueToOneTopology() { for (String fileName : List.of("commons-lang3-3.12.0.jar", "some.lib.tar.gz", "noextension")) {