JAMES-3150 BlobGCService & BloomFilterGCAlgorithm

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/AttachmentBlobReferenceSource.java

@@ -19,6 +19,8 @@
 
 package org.apache.james.mailbox.cassandra.mail;
 
+import javax.inject.Inject;
+
 import org.apache.james.blob.api.BlobId;
 import org.apache.james.blob.api.BlobReferenceSource;
 
@@ -27,6 +29,7 @@
 public class AttachmentBlobReferenceSource implements BlobReferenceSource {
     private final CassandraAttachmentDAOV2 attachmentDAOV2;
 
+    @Inject
     public AttachmentBlobReferenceSource(CassandraAttachmentDAOV2 attachmentDAOV2) {
         this.attachmentDAOV2 = attachmentDAOV2;
     }

mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/MessageBlobReferenceSource.java

@@ -19,6 +19,8 @@
 
 package org.apache.james.mailbox.cassandra.mail;
 
+import javax.inject.Inject;
+
 import org.apache.james.blob.api.BlobId;
 import org.apache.james.blob.api.BlobReferenceSource;
 
@@ -27,6 +29,7 @@
 public class MessageBlobReferenceSource implements BlobReferenceSource {
     private final CassandraMessageDAOV3 daov3;
 
+    @Inject
     public MessageBlobReferenceSource(CassandraMessageDAOV3 daov3) {
         this.daov3 = daov3;
     }

mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraAttachmentDAOV2Test.java

@@ -27,7 +27,6 @@
 import org.apache.james.backends.cassandra.CassandraCluster;
 import org.apache.james.backends.cassandra.CassandraClusterExtension;
 import org.apache.james.blob.api.BlobId;
-import org.apache.james.blob.api.BlobReferenceSource;
 import org.apache.james.blob.api.HashBlobId;
 import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
 import org.apache.james.mailbox.cassandra.mail.CassandraAttachmentDAOV2.DAOAttachment;

pom.xml

@@ -1141,6 +1141,12 @@
                 <artifactId>blob-storage-strategy</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>${james.groupId}</groupId>
+                <artifactId>blob-storage-strategy</artifactId>
+                <version>${project.version}</version>
+                <type>test-jar</type>
+            </dependency>
             <dependency>
                 <groupId>${james.groupId}</groupId>
                 <artifactId>dead-letter-cassandra</artifactId>

server/blob/blob-memory/pom.xml

@@ -46,6 +46,17 @@
             <groupId>${james.groupId}</groupId>
             <artifactId>blob-storage-strategy</artifactId>
         </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>blob-storage-strategy</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>james-server-testing</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>${james.groupId}</groupId>
             <artifactId>james-server-util</artifactId>

server/blob/blob-memory/src/test/java/org/apache/james/blob/memory/MemoryBlobStoreGCAlgorithmTest.java

@@ -0,0 +1,39 @@
+/****************************************************************
+ * 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.james.blob.memory;
+
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.apache.james.server.blob.deduplication.BloomFilterGCAlgorithmContract;
+import org.junit.jupiter.api.BeforeEach;
+
+public class MemoryBlobStoreGCAlgorithmTest implements BloomFilterGCAlgorithmContract {
+
+    private BlobStoreDAO blobStoreDAO;
+
+    @BeforeEach
+    public void beforeEach() {
+        blobStoreDAO = new MemoryBlobStoreDAO();
+    }
+
+    @Override
+    public BlobStoreDAO blobStoreDAO() {
+        return blobStoreDAO;
+    }
+}

server/blob/blob-storage-strategy/pom.xml

@@ -44,6 +44,10 @@
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>james-server-task-api</artifactId>
+        </dependency>
         <dependency>
             <!-- Added because of https://issues.apache.org/jira/browse/SUREFIRE-1266 -->
             <groupId>${james.groupId}</groupId>
@@ -62,6 +66,11 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-configuration2</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.scala-lang</groupId>
             <artifactId>scala-library</artifactId>
@@ -78,6 +87,14 @@
                 <groupId>net.alchim31.maven</groupId>
                 <artifactId>scala-maven-plugin</artifactId>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <reuseForks>true</reuseForks>
+                    <forkCount>1C</forkCount>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 

server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BlobGCTask.java

@@ -0,0 +1,165 @@
+/****************************************************************
+ * 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.james.server.blob.deduplication;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.Set;
+
+import org.apache.james.blob.api.BlobReferenceSource;
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.apache.james.blob.api.BucketName;
+import org.apache.james.server.blob.deduplication.BloomFilterGCAlgorithm.Context;
+import org.apache.james.task.Task;
+import org.apache.james.task.TaskExecutionDetails;
+import org.apache.james.task.TaskType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import reactor.core.scheduler.Schedulers;
+
+public class BlobGCTask implements Task {
+    private static final Logger LOGGER = LoggerFactory.getLogger(BlobGCTask.class);
+    public static final TaskType TASK_TYPE = TaskType.of("BlobGCTask");
+
+    public static class AdditionalInformation implements TaskExecutionDetails.AdditionalInformation {
+
+        private static AdditionalInformation from(Context context) {
+            Context.Snapshot snapshot = context.snapshot();
+            return new AdditionalInformation(
+                snapshot.getReferenceSourceCount(),
+                snapshot.getBlobCount(),
+                snapshot.getGcedBlobCount(),
+                snapshot.getErrorCount(),
+                snapshot.getBloomFilterExpectedBlobCount(),
+                snapshot.getBloomFilterAssociatedProbability());
+        }
+
+        private final Instant timestamp;
+        private final long referenceSourceCount;
+        private final long blobCount;
+        private final long gcedBlobCount;
+        private final long errorCount;
+        private final long bloomFilterExpectedBlobCount;
+        private final double bloomFilterAssociatedProbability;
+
+        AdditionalInformation(long referenceSourceCount,
+                              long blobCount,
+                              long gcedBlobCount,
+                              long errorCount,
+                              long bloomFilterExpectedBlobCount,
+                              double bloomFilterAssociatedProbability) {
+            this.referenceSourceCount = referenceSourceCount;
+            this.blobCount = blobCount;
+            this.gcedBlobCount = gcedBlobCount;
+            this.errorCount = errorCount;
+            this.bloomFilterExpectedBlobCount = bloomFilterExpectedBlobCount;
+            this.bloomFilterAssociatedProbability = bloomFilterAssociatedProbability;
+            this.timestamp = Clock.systemUTC().instant();
+        }
+
+        @Override
+        public Instant timestamp() {
+            return timestamp;
+        }
+
+        public Instant getTimestamp() {
+            return timestamp;
+        }
+
+        public long getReferenceSourceCount() {
+            return referenceSourceCount;
+        }
+
+        public long getBlobCount() {
+            return blobCount;
+        }
+
+        public long getGcedBlobCount() {
+            return gcedBlobCount;
+        }
+
+        public long getErrorCount() {
+            return errorCount;
+        }
+
+        public long getBloomFilterExpectedBlobCount() {
+            return bloomFilterExpectedBlobCount;
+        }
+
+        public double getBloomFilterAssociatedProbability() {
+            return bloomFilterAssociatedProbability;
+        }
+    }
+
+    private final BlobStoreDAO blobStoreDAO;
+    private final GenerationAwareBlobId.Factory generationAwareBlobIdFactory;
+    private final GenerationAwareBlobId.Configuration generationAwareBlobIdConfiguration;
+    private final Set<BlobReferenceSource> blobReferenceSources;
+    private final Clock clock;
+    private final BucketName bucketName;
+    private final int expectedBlobCount;
+    private final double associatedProbability;
+    private final Context context;
+
+    public BlobGCTask(BlobStoreDAO blobStoreDAO,
+                      GenerationAwareBlobId.Factory generationAwareBlobIdFactory,
+                      GenerationAwareBlobId.Configuration generationAwareBlobIdConfiguration,
+                      Set<BlobReferenceSource> blobReferenceSources,
+                      BucketName bucketName,
+                      Clock clock,
+                      int expectedBlobCount,
+                      double associatedProbability) {
+        this.blobStoreDAO = blobStoreDAO;
+        this.generationAwareBlobIdFactory = generationAwareBlobIdFactory;
+        this.generationAwareBlobIdConfiguration = generationAwareBlobIdConfiguration;
+        this.blobReferenceSources = blobReferenceSources;
+        this.clock = clock;
+        this.bucketName = bucketName;
+        this.expectedBlobCount = expectedBlobCount;
+        this.associatedProbability = associatedProbability;
+        this.context = new Context(expectedBlobCount, associatedProbability);
+    }
+
+    @Override
+    public Result run() throws InterruptedException {
+        BloomFilterGCAlgorithm gcAlgorithm = new BloomFilterGCAlgorithm(
+            BlobReferenceAggregate.aggregate(blobReferenceSources),
+            blobStoreDAO,
+            generationAwareBlobIdFactory,
+            generationAwareBlobIdConfiguration,
+            clock);
+
+        return gcAlgorithm.gc(expectedBlobCount, associatedProbability, bucketName, context)
+            .subscribeOn(Schedulers.elastic())
+            .block();
+    }
+
+    @Override
+    public TaskType type() {
+        return TASK_TYPE;
+    }
+
+    @Override
+    public Optional<TaskExecutionDetails.AdditionalInformation> details() {
+        return Optional.of(AdditionalInformation.from(context));
+    }
+}