/*
 * 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.jackrabbit.oak.plugins.blob;

import static org.apache.jackrabbit.oak.plugins.blob.MarkSweepGarbageCollector.GarbageCollectionOperationStats.CONSISTENCY_NAME;
import static org.apache.jackrabbit.oak.plugins.blob.MarkSweepGarbageCollector.GarbageCollectionOperationStats.FINISH_FAILURE;
import static org.apache.jackrabbit.oak.plugins.blob.MarkSweepGarbageCollector.GarbageCollectionOperationStats.NAME;
import static org.apache.jackrabbit.oak.plugins.blob.MarkSweepGarbageCollector.GarbageCollectionOperationStats.NUM_BLOBS_DELETED;
import static org.apache.jackrabbit.oak.plugins.blob.MarkSweepGarbageCollector.GarbageCollectionOperationStats.NUM_CANDIDATES;
import static org.apache.jackrabbit.oak.plugins.blob.MarkSweepGarbageCollector.GarbageCollectionOperationStats.START;
import static org.apache.jackrabbit.oak.plugins.blob.MarkSweepGarbageCollector.GarbageCollectionOperationStats.TOTAL_SIZE_DELETED;
import static org.apache.jackrabbit.oak.plugins.blob.OperationsStatsMBean.TYPE;
import static org.apache.jackrabbit.oak.plugins.blob.datastore.DataStoreUtils.randomStream;
import static org.apache.jackrabbit.oak.plugins.blob.datastore.SharedDataStoreUtils.SharedStoreRecordType.REPOSITORY;
import static org.apache.jackrabbit.oak.stats.StatsOptions.METRICS_ONLY;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicReference;

import ch.qos.logback.classic.Level;
import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.io.Closer;
import org.apache.jackrabbit.oak.api.Blob;
import org.apache.jackrabbit.oak.api.CommitFailedException;
import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.api.Type;
import org.apache.jackrabbit.oak.api.blob.BlobAccessProvider;
import org.apache.jackrabbit.oak.api.blob.BlobUpload;
import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
import org.apache.jackrabbit.oak.commons.junit.LogCustomizer;
import org.apache.jackrabbit.oak.plugins.blob.datastore.DataStoreBlobStore;
import org.apache.jackrabbit.oak.plugins.blob.datastore.SharedDataStoreUtils;
import org.apache.jackrabbit.oak.spi.blob.GarbageCollectableBlobStore;
import org.apache.jackrabbit.oak.spi.cluster.ClusterRepositoryInfo;
import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeStore;
import org.apache.jackrabbit.oak.spi.whiteboard.DefaultWhiteboard;
import org.apache.jackrabbit.oak.spi.whiteboard.Registration;
import org.apache.jackrabbit.oak.spi.whiteboard.Whiteboard;
import org.apache.jackrabbit.oak.stats.Clock;
import org.apache.jackrabbit.oak.stats.DefaultStatisticsProvider;
import org.apache.jackrabbit.oak.stats.StatisticsProvider;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.internal.util.collections.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Generic class for BlobGC tests which uses custom MemoryNodeStore as well as a memory NodeStore.
 */
public class BlobGCTest {
    protected static final Logger log = LoggerFactory.getLogger(BlobGCTest.class);

    @Rule
    public TemporaryFolder folder = new TemporaryFolder(new File("target"));

    protected Whiteboard wb;

    protected Closer closer;

    protected Cluster cluster;

    protected Clock clock;

    @Before
    public void before() throws Exception {
        closer = Closer.create();
        clock = getClock();

        // add whiteboard
        final AtomicReference<Map<?, ?>> props = new AtomicReference<Map<?, ?>>();
        wb = new DefaultWhiteboard(){
            @Override
            public <T> Registration register(Class<T> type, T service, Map<?, ?> properties) {
                props.set(properties);
                return super.register(type, service, properties);
            }
        };

        TimeLapsedDataStore dataStore = new TimeLapsedDataStore(clock);
        DataStoreBlobStore blobStore = new DataStoreBlobStore(dataStore);
        MemoryBlobStoreNodeStore nodeStore = new MemoryBlobStoreNodeStore(blobStore);
        cluster = new Cluster(folder.newFolder(), blobStore, nodeStore, 0);
        closer.register(cluster);
    }

    @After
    public void after() {
        try {
            closer.close();
        } catch (IOException e) {
            log.error("Error closing cluster instances", e);
        }
    }

    protected Clock getClock() {
        return new Clock.Virtual();
    }

    class Cluster implements Closeable {
        protected final BlobStoreState blobStoreState;
        private final File root;
        private final Clock clock;
        String repoId;
        protected final TimeLapsedDataStore dataStore;
        protected final GarbageCollectableBlobStore blobStore;
        protected final NodeStore nodeStore;
        private MarkSweepGarbageCollector collector;
        protected BlobReferenceRetriever referenceRetriever;
        protected ScheduledExecutorService scheduledExecutor;
        protected ThreadPoolExecutor executor;
        protected DefaultStatisticsProvider statsProvider;
        protected long startReferenceTime;

        public Cluster(File root, GarbageCollectableBlobStore blobStore, NodeStore nodeStore, int seed) throws Exception {
            this.root = root;
            this.nodeStore = nodeStore;
            this.dataStore = (TimeLapsedDataStore) ((DataStoreBlobStore) blobStore).getDataStore();
            this.blobStore = blobStore;
            this.clock = dataStore.getClock();

            if (SharedDataStoreUtils.isShared(blobStore)) {
                repoId = ClusterRepositoryInfo.getOrCreateId(nodeStore);
                ((SharedDataStore) blobStore).setRepositoryId(repoId);
            }
            referenceRetriever = ((MemoryBlobStoreNodeStore) nodeStore).getBlobReferenceRetriever();
            startReferenceTime = clock.getTime();
            log.info("Reference time {}", startReferenceTime);
            scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
            executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);
            statsProvider = new DefaultStatisticsProvider(scheduledExecutor);

            blobStoreState = setUp(nodeStore, blobStore, 10, 5, 100, seed);
        }

        public void setRepoId(String id) {
            this.repoId = id;
        }

        public MarkSweepGarbageCollector getCollector(long blobGcMaxAgeInSecs) throws Exception {
            return getCollector(blobGcMaxAgeInSecs, false, false);
        }

        public MarkSweepGarbageCollector getCollector(long blobGcMaxAgeInSecs, boolean checkConsistency,
            boolean sweepIfRefsPastRetention) throws Exception {

            collector =
                new MarkSweepGarbageCollector(referenceRetriever, blobStore, executor, root.getAbsolutePath(), 2048,
                    blobGcMaxAgeInSecs, checkConsistency, sweepIfRefsPastRetention, repoId, wb, statsProvider);
            collector.setClock(clock);
            return collector;
        }

        @Override public void close() throws IOException {
            new ExecutorCloser(scheduledExecutor).close();
            new ExecutorCloser(executor).close();
        }
    }

    @Test
    public void sharedGC() throws Exception {
        log.info("Staring sharedGC()");

        // Setup a different cluster/repository sharing the blob store
        MemoryBlobStoreNodeStore secondClusterNodeStore = new MemoryBlobStoreNodeStore(cluster.blobStore);
        Cluster secondCluster = new Cluster(folder.newFolder(), cluster.blobStore, secondClusterNodeStore, 100);
        closer.register(secondCluster);

        Sets.SetView<String> totalPresent =
            Sets.union(cluster.blobStoreState.blobsPresent, secondCluster.blobStoreState.blobsPresent);
        Sets.SetView<String> totalAdded =
            Sets.union(cluster.blobStoreState.blobsAdded, secondCluster.blobStoreState.blobsAdded);

        // Execute mark on the default cluster
        executeGarbageCollection(cluster, cluster.getCollector(0), true);
        Set<String> existingAfterGC = executeGarbageCollection(secondCluster, secondCluster.getCollector(0), false);

        assertTrue(Sets.symmetricDifference(totalPresent, existingAfterGC).isEmpty());
        assertStats(secondCluster.statsProvider, 1, 0, totalAdded.size() - totalPresent.size(),
            totalAdded.size() - totalPresent.size(), NAME);
    }

    @Test
    public void noSharedGC() throws Exception {
        log.info("Starting noSharedGC()");

        // Setup a different cluster/repository sharing the blob store
        MemoryBlobStoreNodeStore secondClusterNodeStore = new MemoryBlobStoreNodeStore(cluster.blobStore);
        Cluster secondCluster = new Cluster(folder.newFolder(), cluster.blobStore, secondClusterNodeStore, 100);
        closer.register(secondCluster);

        Sets.SetView<String> totalAdded =
            Sets.union(cluster.blobStoreState.blobsAdded, secondCluster.blobStoreState.blobsAdded);

        Set<String> existingAfterGC = executeGarbageCollection(secondCluster, secondCluster.getCollector(0), false);

        assertEquals(totalAdded, existingAfterGC);
        assertStats(secondCluster.statsProvider, 1, 1, 0, 0, NAME);
    }

    @Test
    public void sharedGCRepositoryCloned() throws Exception {
        log.debug("Starting sharedGCRepoCloned()");

        // Setup a different cluster/repository sharing the blob store and the repository id
        MemoryBlobStoreNodeStore secondClusterNodeStore = new MemoryBlobStoreNodeStore(cluster.blobStore);
        Cluster secondCluster = new Cluster(folder.newFolder(), cluster.blobStore, secondClusterNodeStore, 100);
        closer.register(secondCluster);

        ((SharedDataStore) secondCluster.blobStore).deleteMetadataRecord(REPOSITORY.getNameFromId(secondCluster.repoId));
        secondCluster.setRepoId(cluster.repoId);

        Sets.SetView<String> totalPresent =
            Sets.union(cluster.blobStoreState.blobsPresent, secondCluster.blobStoreState.blobsPresent);

        // Execute mark on the default cluster
        executeGarbageCollection(cluster, cluster.getCollector(0), true);
        Set<String> existingAfterGC = executeGarbageCollection(secondCluster, secondCluster.getCollector(0), false);

        assertTrue(Sets.symmetricDifference(totalPresent, existingAfterGC).isEmpty());
    }

    @Test
    public void sharedGCRefsOld() throws Exception {
        log.info("Staring sharedGCRefsOld()");

        // Setup a different cluster/repository sharing the blob store
        MemoryBlobStoreNodeStore secondClusterNodeStore = new MemoryBlobStoreNodeStore(cluster.blobStore);
        Cluster secondCluster = new Cluster(folder.newFolder(), cluster.blobStore, secondClusterNodeStore, 100);
        closer.register(secondCluster);

        Sets.SetView<String> totalPresent =
            Sets.union(cluster.blobStoreState.blobsPresent, secondCluster.blobStoreState.blobsPresent);
        Sets.SetView<String> totalAdded =
            Sets.union(cluster.blobStoreState.blobsAdded, secondCluster.blobStoreState.blobsAdded);

        clock.waitUntil(clock.getTime() + 5);

        // Execute mark on the default cluster
        executeGarbageCollection(cluster, cluster.getCollector(5), true);
        executeGarbageCollection(secondCluster, secondCluster.getCollector(5), true);

        clock.waitUntil(clock.getTime() + 5);

        Set<String> existingAfterGC = executeGarbageCollection(secondCluster, secondCluster.getCollector(5, false, true), false);

        assertTrue(Sets.symmetricDifference(totalPresent, existingAfterGC).isEmpty());
        assertStats(secondCluster.statsProvider, 2, 0, totalAdded.size() - totalPresent.size(),
            totalAdded.size() - totalPresent.size(), NAME);
    }

    @Test
    public void sharedGCRefsNotOld() throws Exception {
        log.info("Staring sharedGCRefsNotOld()");

        // Setup a different cluster/repository sharing the blob store
        MemoryBlobStoreNodeStore secondClusterNodeStore = new MemoryBlobStoreNodeStore(cluster.blobStore);
        Cluster secondCluster = new Cluster(folder.newFolder(), cluster.blobStore, secondClusterNodeStore, 100);
        closer.register(secondCluster);

        Sets.SetView<String> totalPresent =
            Sets.union(cluster.blobStoreState.blobsPresent, secondCluster.blobStoreState.blobsPresent);
        Sets.SetView<String> totalAdded =
            Sets.union(cluster.blobStoreState.blobsAdded, secondCluster.blobStoreState.blobsAdded);

        // Execute mark on the default cluster
        executeGarbageCollection(cluster, cluster.getCollector(5), true);

        // Let the second cluster one not pass retention old time
        clock.waitUntil(clock.getTime() + 5);

        executeGarbageCollection(secondCluster, secondCluster.getCollector(5), true);

        Set<String> existingAfterGC = executeGarbageCollection(secondCluster, secondCluster.getCollector(6, false, true), false);

        assertTrue(Sets.symmetricDifference(totalAdded, existingAfterGC).isEmpty());
        assertStats(secondCluster.statsProvider, 2, 1, 0,0, NAME);
    }

    @Test
    public void gc() throws Exception {
        log.info("Starting gc()");

        Set<String> existingAfterGC = executeGarbageCollection(cluster, cluster.getCollector(0), false);
        assertTrue(Sets.symmetricDifference(cluster.blobStoreState.blobsPresent, existingAfterGC).isEmpty());
        assertStats(cluster.statsProvider, 1, 0,
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(),
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(), NAME);
    }

    @Test
    public void gcWithConsistencyCheck() throws Exception {
        log.info("Starting gcWithConsistencyCheck()");
        ((MemoryBlobStoreNodeStore) cluster.nodeStore).getReferencedBlobs().add("SPURIOUS");

        MarkSweepGarbageCollector collector = cluster.getCollector(0, true, false);
        Set<String> existingAfterGC = executeGarbageCollection(cluster, collector, false);
        assertFalse(Sets.symmetricDifference(cluster.blobStoreState.blobsPresent, existingAfterGC).isEmpty());
        assertStats(cluster.statsProvider, 1, 0,
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size() + 1,
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size() + 1, NAME);
        assertStatsBean(collector.getConsistencyOperationStats(), 1, 1, 1);
    }

    @Test
    public void gcWithNoDeleteDirectBinary() throws Exception {
        log.info("Starting gcWithNoDeleteDirectBinary()");

        setupDirectBinary(1, 0);
        Set<String> existingAfterGC = executeGarbageCollection(cluster, cluster.getCollector(0), false);
        assertTrue(Sets.symmetricDifference(cluster.blobStoreState.blobsPresent, existingAfterGC).isEmpty());
        assertStats(cluster.statsProvider, 1, 0,
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(),
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(), NAME);
    }

    @Test
    public void gcWithDeleteDirectBinary() throws Exception {
        log.info("Starting gcWithNoDeleteDirectBinary()");

        setupDirectBinary(5, 2);
        Set<String> existingAfterGC = executeGarbageCollection(cluster, cluster.getCollector(0), false);
        assertTrue(Sets.symmetricDifference(cluster.blobStoreState.blobsPresent, existingAfterGC).isEmpty());
        assertStats(cluster.statsProvider, 1, 0,
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(),
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(), NAME);
    }

    @Test
    public void noGc() throws Exception {
        log.info("Starting noGc()");

        long afterSetupTime = clock.getTime();
        log.info("after setup time {}", afterSetupTime);

        Set<String> existingAfterGC =
            executeGarbageCollection(cluster, cluster.getCollector(afterSetupTime - cluster.startReferenceTime + 2),
                false);
        assertTrue(Sets.symmetricDifference(cluster.blobStoreState.blobsAdded, existingAfterGC).isEmpty());
        assertStats(cluster.statsProvider, 1, 0, 0,
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(), NAME);
    }

    @Test
    public void checkConsistency() throws Exception {
        log.info("Starting checkConsistency()");

        long afterSetupTime = clock.getTime();
        log.info("after setup time {}", afterSetupTime);

        MarkSweepGarbageCollector collector = cluster.getCollector(0);
        long missing = collector.checkConsistency();

        assertEquals(0, missing);
        assertStats(cluster.statsProvider, 1, 0, 0, 0, CONSISTENCY_NAME);
        assertStatsBean(collector.getConsistencyOperationStats(), 1, 0, 0);
    }

    @Test
    public void checkConsistencyFailure() throws Exception {
        log.info("Starting checkConsistencyFailure()");

        long afterSetupTime = clock.getTime();
        log.info("after setup time {}", afterSetupTime);

        cluster.blobStore
            .countDeleteChunks(Lists.newArrayList(Iterators.getLast(cluster.blobStoreState.blobsPresent.iterator())),
                0);
        MarkSweepGarbageCollector collector = cluster.getCollector(0);
        long missing = collector.checkConsistency();

        assertEquals(1, missing);
        assertStats(cluster.statsProvider, 1, 1, 1, 0, CONSISTENCY_NAME);
        assertStatsBean(collector.getConsistencyOperationStats(), 1, 1, 1);
    }

    @Test
    public void checkConsistencyGlobal() throws Exception {
        log.info("Staring checkConsistencyGlobal()");

        // Setup a different cluster/repository sharing the blob store
        MemoryBlobStoreNodeStore secondClusterNodeStore = new MemoryBlobStoreNodeStore(cluster.blobStore, true);
        Cluster secondCluster = new Cluster(folder.newFolder(), cluster.blobStore, secondClusterNodeStore, 100);
        closer.register(secondCluster);

        secondCluster.blobStoreState.blobsPresent.add(Iterables.firstOf(cluster.blobStoreState.blobsPresent));
        // Execute mark on the default cluster
        executeGarbageCollection(cluster, cluster.getCollector(0), true);
        MarkSweepGarbageCollector globalCollector = secondCluster.getCollector(0, true, false);
        long missing = globalCollector.checkConsistency();
        assertEquals(0, missing);
        assertStats(secondCluster.statsProvider, 1, 0, 0, 0, CONSISTENCY_NAME);
        assertStatsBean(globalCollector.getConsistencyOperationStats(), 1, 0, 0);
    }

    @Test
    public void checkConsistencyGlobalFailureOther() throws Exception {
        log.info("Staring checkConsistencyGlobalFailureOther()");

        // Setup a different cluster/repository sharing the blob store
        MemoryBlobStoreNodeStore secondClusterNodeStore = new MemoryBlobStoreNodeStore(cluster.blobStore);
        Cluster secondCluster = new Cluster(folder.newFolder(), cluster.blobStore, secondClusterNodeStore, 100);
        closer.register(secondCluster);

        cluster.blobStore
            .countDeleteChunks(Lists.newArrayList(Iterators.getLast(cluster.blobStoreState.blobsPresent.iterator())),
                0);

        // Execute mark on the default cluster
        executeGarbageCollection(cluster, cluster.getCollector(0), true);
        MarkSweepGarbageCollector globalCollector = secondCluster.getCollector(0, true, false);
        long missing = globalCollector.checkConsistency();
        assertEquals(1, missing);
        assertStats(secondCluster.statsProvider, 1, 1, 1, 0, CONSISTENCY_NAME);
        assertStatsBean(globalCollector.getConsistencyOperationStats(), 1, 1, 1);
    }

    @Test
    public void checkConsistencyGlobalFailure() throws Exception {
        log.info("Staring checkConsistencyGlobalFailureOther()");

        // Setup a different cluster/repository sharing the blob store
        MemoryBlobStoreNodeStore secondClusterNodeStore = new MemoryBlobStoreNodeStore(cluster.blobStore);
        Cluster secondCluster = new Cluster(folder.newFolder(), cluster.blobStore, secondClusterNodeStore, 100);
        closer.register(secondCluster);

        secondCluster.blobStore
            .countDeleteChunks(Lists.newArrayList(Iterators.getLast(secondCluster.blobStoreState.blobsPresent.iterator())),
                0);

        // Execute mark on the default cluster
        executeGarbageCollection(cluster, cluster.getCollector(0), true);
        MarkSweepGarbageCollector globalCollector = secondCluster.getCollector(0, true, false);
        long missing = globalCollector.checkConsistency();
        assertEquals(1, missing);
        assertStats(secondCluster.statsProvider, 1, 1, 1, 0, CONSISTENCY_NAME);
        assertStatsBean(globalCollector.getConsistencyOperationStats(), 1, 1, 1);
    }

    @Test
    public void gcCheckDeletedSize() throws Exception {
        log.info("Starting gcCheckDeletedSize()");

        // Capture logs for the second round of gc
        LogCustomizer customLogs = LogCustomizer
            .forLogger(MarkSweepGarbageCollector.class.getName())
            .enable(Level.INFO)
            .filter(Level.INFO)
            .contains("Estimated size recovered for")
            .create();
        customLogs.starting();

        Set<String> existingAfterGC =
            executeGarbageCollection(cluster, cluster.getCollector(0),false);
        assertEquals(1, customLogs.getLogs().size());
        long deletedSize = (cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size()) * 100;
        assertTrue(customLogs.getLogs().get(0).contains(String.valueOf(deletedSize)));
        assertStats(cluster.statsProvider, 1, 0,
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(),
            cluster.blobStoreState.blobsAdded.size() - cluster.blobStoreState.blobsPresent.size(), NAME);
        assertEquals(deletedSize, getStatCount(cluster.statsProvider, NAME, TOTAL_SIZE_DELETED));
        customLogs.finished();
        assertTrue(Sets.symmetricDifference(cluster.blobStoreState.blobsPresent, existingAfterGC).isEmpty());
    }

    @Test
    public void gcMarkOnly() throws Exception {
        log.info("Starting gcMarkOnly()");

        Set<String> existingAfterGC =
            executeGarbageCollection(cluster, cluster.getCollector(0),true);
        assertTrue(Sets.symmetricDifference(cluster.blobStoreState.blobsAdded, existingAfterGC).isEmpty());
        assertStats(cluster.statsProvider, 1, 0, 0, 0, NAME);
    }

    protected Set<String> executeGarbageCollection(Cluster cluster, MarkSweepGarbageCollector collector, boolean markOnly)
        throws Exception {
        collector.collectGarbage(markOnly);

        assertEquals(0, cluster.executor.getTaskCount());
        Set<String> existingAfterGC = iterate(cluster.blobStore);
        log.info("{} blobs existing after gc : {}", existingAfterGC.size(), existingAfterGC);

        return existingAfterGC;
    }

    private void assertStats(StatisticsProvider statsProvider, int start, int failure, long deleted, long candidates,
        String typeName) {

        assertEquals("Start counter mismatch", start, getStatCount(statsProvider, typeName, START));
        assertEquals("Finish error mismatch", failure, getStatCount(statsProvider, typeName, FINISH_FAILURE));
        assertEquals("Num deleted mismatch", deleted, getStatCount(statsProvider, typeName, NUM_BLOBS_DELETED));
        assertEquals("Num candidates mismatch", candidates, getStatCount(statsProvider, typeName, NUM_CANDIDATES));
    }


    private void assertStatsBean(OperationsStatsMBean mbean, int start, int failure, long deleted) {
        assertEquals("Start counter mismatch", start, mbean.getStartCount());
        assertEquals("Finish error mismatch", failure, mbean.getFailureCount());
        assertEquals("Num deleted mismatch", deleted, mbean.numDeleted());
    }

    private long getStatCount(StatisticsProvider statsProvider, String typeName, String name) {
        return statsProvider.getCounterStats(
            TYPE + "." + typeName + "." + name, METRICS_ONLY).getCount();
    }

    protected Set<String> iterate(GarbageCollectableBlobStore blobStore) throws Exception {
        Iterator<String> cur = blobStore.getAllChunkIds(0);

        Set<String> existing = Sets.newHashSet();
        while (cur.hasNext()) {
            existing.add(cur.next());
        }
        return existing;
    }

    public BlobStoreState setUp (NodeStore nodeStore,
        GarbageCollectableBlobStore blobStore,
        int count,
        int deletions,
        int blobSize,
        int seed) throws Exception {

        preSetup();

        NodeBuilder a = nodeStore.getRoot().builder();
        /* Create and delete nodes with blobs stored in DS*/
        int maxDeleted  = deletions;
        int numBlobs = count;
        List<Integer> toBeDeleted = Lists.newArrayList();
        Random rand = new Random();
        for (int i = 0; i < maxDeleted; i++) {
            int n = rand.nextInt(numBlobs);
            if (!toBeDeleted.contains(n)) {
                toBeDeleted.add(n);
            }
        }

        BlobStoreState state = new BlobStoreState();
        for (int i = 0; i < numBlobs; i++) {
            Blob b = nodeStore.createBlob(
                randomStream(Integer.parseInt(String.valueOf(seed) + String.valueOf(i)), blobSize));
            Iterator<String> idIter = blobStore.resolveChunks(b.getContentIdentity());
            while (idIter.hasNext()) {
                String chunk = idIter.next();
                state.blobsAdded.add(chunk);
                if (!toBeDeleted.contains(i)) {
                    state.blobsPresent.add(chunk);
                }
            }
            a.child("c" + i).setProperty("x", b);
        }

        nodeStore.merge(a, EmptyHook.INSTANCE, CommitInfo.EMPTY);
        log.info("Created blobs : {}", state.blobsAdded.size());

        for (int id : toBeDeleted) {
            delete("c" + id, nodeStore);
        }
        log.info("Deleted nodes : {}", toBeDeleted.size());

        // Sleep a little to make eligible for cleanup
        clock.waitUntil(5);

        postSetup(nodeStore, state);

        log.info("{} blobs added : {}", state.blobsAdded.size(), state.blobsAdded);
        log.info("{} blobs remaining : {}", state.blobsPresent.size(), state.blobsPresent);

        return state;
    }

    protected void setupDirectBinary(int numCreate, int numDelete) throws CommitFailedException {
        for (int i = 0; i < numCreate; i++) {
            BlobUpload blobUpload = ((BlobAccessProvider) cluster.blobStore).initiateBlobUpload(100, 1);
            Blob blob = ((BlobAccessProvider) cluster.blobStore).completeBlobUpload(blobUpload.getUploadToken());

            cluster.blobStoreState.blobsAdded.add(blob.getContentIdentity());
            cluster.blobStoreState.blobsPresent.add(blob.getContentIdentity());
            NodeBuilder builder = cluster.nodeStore.getRoot().builder();
            builder.child("dbu" + i).setProperty("x", blob);
            cluster.nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
            PropertyState property = cluster.nodeStore.getRoot().getChildNode("dbu" + i).getProperty("x");
            Blob blobReturned = property.getValue(Type.BINARY);
            ((MemoryBlobStoreNodeStore) cluster.nodeStore).getReferencedBlobs().add(blobReturned.getContentIdentity());
        }

        for (int i = 0; i < Math.min(numCreate, numDelete); i++) {
            PropertyState property = cluster.nodeStore.getRoot().getChildNode("dbu" + i).getProperty("x");
            String blobId = property.getValue(Type.BINARY).getContentIdentity();

            delete("dbu" + i, cluster.nodeStore);
            ((MemoryBlobStoreNodeStore) cluster.nodeStore).getReferencedBlobs().remove(blobId);
            cluster.blobStoreState.blobsPresent.remove(blobId);
        }
    }

    protected Set<String> createBlobs(GarbageCollectableBlobStore blobStore, int count, int size) throws Exception {
        HashSet<String> blobSet = new HashSet<String>();
        for  (int i = 0; i < count; i++) {
            String id = blobStore.writeBlob(randomStream(10 + i, size));
            Iterator<String> idIter = blobStore.resolveChunks(id);
            while (idIter.hasNext()) {
                String chunk = idIter.next();
                blobSet.add(chunk);
            }
        }
        log.info("{} Additional created {}", blobSet.size(), blobSet);
        return blobSet;
    }

    void preSetup() {}

    protected void postSetup(NodeStore nodeStore, BlobStoreState state) {
        ((MemoryBlobStoreNodeStore) nodeStore).setReferencedBlobs(state.blobsPresent);
    }

    protected void delete(String nodeId, NodeStore nodeStore) throws CommitFailedException {
        NodeBuilder builder = nodeStore.getRoot().builder();
        builder.child(nodeId).remove();

        nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
    }

    /**
     * Represents state of the blobs after setup
     */
    class BlobStoreState {
        Set<String> blobsAdded = Sets.newHashSet();
        Set<String> blobsPresent = Sets.newHashSet();
    }

}
