STORM-3871: sweep dependency blobs that outlive their topology's cleanup

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).
      *
+     * <p>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

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<Integer> sequenceNumbers = new TreeSet<Integer>();
         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,

storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java

@@ -139,14 +139,26 @@ 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);
             }
         }
     }
 
+    /**
+     * 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);

storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java

@@ -440,6 +440,8 @@
     private final TimeCacheMap<String, WritableByteChannel> 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<String, Long> orphanedDependencyKeysDetectedMs = new HashMap<>();
     @SuppressWarnings("deprecation")
     private final TimeCacheMap<String, BufferInputStream> blobDownloaders;
     @SuppressWarnings("deprecation")
@@ -775,8 +777,18 @@
 
     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,
@@ -806,8 +818,8 @@
      */
     private static Map<String, Object> maskCredentialsForApi(Map<String, Object> conf) {
         Map<String, Object> masked = new HashMap<>(ConfigUtils.maskCredentials(conf));
         if (masked.get(BlowfishTupleSerializer.SECRET_KEY) instanceof String) {
             masked.put(BlowfishTupleSerializer.SECRET_KEY, "*****");
         }
         return masked;
     }
@@ -1126,7 +1138,7 @@
         cleanable.addAll(Utils.OR(state.heartbeatStorms(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.errorTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(store.storedTopoIds(), EMPTY_STRING_SET));
         cleanable.addAll(Utils.OR(state.backpressureTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.idsOfTopologiesWithPrivateWorkerKeys(), EMPTY_STRING_SET));
         Set<String> delayedCleanable = getExpiredTopologyIds(cleanable, conf);
         delayedCleanable.removeAll(Utils.OR(state.activeStorms(), EMPTY_STRING_LIST));
@@ -1246,8 +1258,8 @@
         ret.put(Config.TOPOLOGY_WORKER_NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
         ret.put(Config.NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
 
         if (!mergedConf.containsKey(Config.TOPOLOGY_METRICS_REPORTERS) && mergedConf.containsKey(Config.STORM_METRICS_REPORTERS)) {
             ret.put(Config.TOPOLOGY_METRICS_REPORTERS, mergedConf.get(Config.STORM_METRICS_REPORTERS));
         }
 
         // add any system metrics reporters to the topology metrics reporters
@@ -3095,6 +3107,45 @@
         Utils.forceDelete(ServerConfigUtils.masterStormDistRoot(conf, topoId));
     }
 
+    /**
+     * Remove the dependency blobs that no topology has referred to for longer than the inbox jar expiration.
+     *
+     * <p>{@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.
+     *
+     * <p>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<String> referenced) {
+        try {
+            long graceMs = TimeUnit.SECONDS.toMillis(
+                ObjectReader.getInt(conf.get(DaemonConfig.NIMBUS_INBOX_JAR_EXPIRATION_SECS), 3600));
+            long nowMs = Time.currentTimeMillis();
+            Set<String> 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.
      */
@@ -3126,7 +3177,7 @@
                 state.teardownHeartbeats(topoId);
                 state.teardownTopologyErrors(topoId);
                 state.removeAllPrivateWorkerKeys(topoId);
                 state.removeBackpressure(topoId);
                 rmDependencyBlobsInTopology(topoId, stillReferenced);
                 forceDeleteTopoDistDir(topoId);
                 rmTopologyKeys(topoId);
@@ -3134,6 +3185,12 @@
                 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={}",
@@ -3640,8 +3697,8 @@
                 waitForDesiredCodeReplication(totalConf, topoId);
                 state.setupHeatbeats(topoId, topoConf);
                 state.setupErrors(topoId, topoConf);
                 if (ObjectReader.getBoolean(totalConf.get(Config.TOPOLOGY_BACKPRESSURE_ENABLE), false)) {
                     state.setupBackpressure(topoId, topoConf);
                 }
                 notifyTopologyActionListener(topoName, "submitTopology");
                 TopologyStatus status = null;
@@ -4378,7 +4435,9 @@
             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) {

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;
+    }
+}