Manual compaction of a live OM.

Change-Id: I3a05cc4f41ed667d52b4ecbeca2dd9ffcdad8416

Fix compaction bugs

Change-Id: Id5beedb0f4d5519a9dcb02c94a839adbc5defd40
(cherry picked from commit 142a039dcb4838d0970aecbde10f49898df96475)

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -280,6 +280,7 @@ public static boolean isReadOnly(
     case GetObjectTagging:
     case GetQuotaRepairStatus:
     case StartQuotaRepair:
+    case Compact:
       return true;
     case CreateVolume:
     case SetVolumeProperty:

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java

@@ -1182,4 +1182,6 @@ default void deleteObjectTagging(OmKeyArgs args) throws IOException {
    * @throws IOException
    */
   void startQuotaRepair(List<String> buckets) throws IOException;
+
+  void compact(String columnFamily) throws IOException;
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -2658,6 +2658,17 @@ public void deleteObjectTagging(OmKeyArgs args) throws IOException {
     handleError(omResponse);
   }
 
+  @Override
+  public void compact(String columnFamily) throws IOException {
+    OzoneManagerProtocolProtos.CompactRequest compactRequest =
+        OzoneManagerProtocolProtos.CompactRequest.newBuilder()
+            .setColumnFamily(columnFamily)
+            .build();
+    OMRequest omRequest = createOMRequest(Type.Compact)
+        .setCompactRequest(compactRequest).build();
+    handleError(submitRequest(omRequest));
+  }
+
   private SafeMode toProtoBuf(SafeModeAction action) {
     switch (action) {
     case ENTER:

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -156,6 +156,7 @@ enum Type {
   PutObjectTagging = 140;
   GetObjectTagging = 141;
   DeleteObjectTagging = 142;
+  Compact = 143;
 }
 
 enum SafeMode {
@@ -303,6 +304,7 @@ message OMRequest {
   optional GetObjectTaggingRequest          getObjectTaggingRequest        = 140;
   optional PutObjectTaggingRequest          putObjectTaggingRequest        = 141;
   optional DeleteObjectTaggingRequest       deleteObjectTaggingRequest     = 142;
+  optional CompactRequest                   compactRequest                 = 143;
 }
 
 message OMResponse {
@@ -436,6 +438,7 @@ message OMResponse {
   optional GetObjectTaggingResponse          getObjectTaggingResponse      = 140;
   optional PutObjectTaggingResponse          putObjectTaggingResponse      = 141;
   optional DeleteObjectTaggingResponse       deleteObjectTaggingResponse   = 142;
+  optional CompactResponse                   compactResponse               = 143;
 }
 
 enum Status {
@@ -2294,6 +2297,13 @@ message DeleteObjectTaggingRequest {
 message DeleteObjectTaggingResponse {
 }
 
+message CompactRequest {
+  required string columnFamily = 1;
+}
+
+message CompactResponse {
+}
+
 /**
  The OM service that takes care of Ozone namespace.
 */

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -107,6 +107,7 @@
 import org.apache.hadoop.ozone.om.request.OMClientRequest;
 import org.apache.hadoop.ozone.om.s3.S3SecretCacheProvider;
 import org.apache.hadoop.ozone.om.s3.S3SecretStoreProvider;
+import org.apache.hadoop.ozone.om.service.CompactTask;
 import org.apache.hadoop.ozone.om.service.OMRangerBGSyncService;
 import org.apache.hadoop.ozone.om.service.QuotaRepairTask;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotUtils;
@@ -5021,4 +5022,10 @@ public void checkFeatureEnabled(OzoneManagerVersion feature) throws OMException
       throw new OMException("Feature disabled: " + feature, OMException.ResultCodes.NOT_SUPPORTED_OPERATION);
     }
   }
+
+  @Override
+  public void compact(String columnFamily) throws IOException {
+    checkAdminUserPrivilege("compact column family " + columnFamily);
+    new CompactTask(this).compact(columnFamily);
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/CompactTask.java

@@ -0,0 +1,67 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.ozone.om.service;
+
+import org.apache.hadoop.hdds.utils.db.RDBStore;
+import org.apache.hadoop.hdds.utils.db.RocksDatabase;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedCompactRangeOptions;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+
+public class CompactTask {
+  private static final Logger LOG = LoggerFactory.getLogger(
+      CompactTask.class);
+  private final OzoneManager om;
+
+  public CompactTask(OzoneManager ozoneManager) {
+    this.om = ozoneManager;
+  }
+
+  public CompletableFuture<Void> compact(String columnFamily) throws
+      IOException {
+    return CompletableFuture.supplyAsync(() -> {
+      try {
+        return compactAsync(columnFamily);
+      } catch (IOException e) {
+        LOG.warn("Failed to compact column family: {}", columnFamily, e);
+      }
+      return null;
+    });
+  }
+
+  private Void compactAsync(String columnFamilyName) throws IOException {
+    LOG.info("Compacting column family: {}", columnFamilyName);
+    long startTime = Time.monotonicNow();
+    ManagedCompactRangeOptions options =
+        new ManagedCompactRangeOptions();
+    options.setBottommostLevelCompaction(
+        ManagedCompactRangeOptions.BottommostLevelCompaction.kForce);
+    // Find CF Handler
+    RocksDatabase.ColumnFamily columnFamily =
+        ((RDBStore)om.getMetadataManager().getStore()).getDb().getColumnFamily(columnFamilyName);
+    ((RDBStore)om.getMetadataManager().getStore()).getDb().compactRange(
+        columnFamily, null, null, options);
+    LOG.info("Compaction of column family: {} completed in {} ms",
+        columnFamilyName, Time.monotonicNow() - startTime);
+    return null;
+  }
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java

@@ -402,6 +402,10 @@ public OMResponse handleReadRequest(OMRequest request) {
         OzoneManagerProtocolProtos.GetObjectTaggingResponse getObjectTaggingResponse =
             getObjectTagging(request.getGetObjectTaggingRequest());
         responseBuilder.setGetObjectTaggingResponse(getObjectTaggingResponse);
+      case Compact:
+        OzoneManagerProtocolProtos.CompactResponse compactResponse = compact(request.getCompactRequest());
+        responseBuilder
+            .setCompactResponse(compactResponse);
         break;
       default:
         responseBuilder.setSuccess(false);
@@ -419,6 +423,15 @@ public OMResponse handleReadRequest(OMRequest request) {
     return responseBuilder.build();
   }
 
+  private OzoneManagerProtocolProtos.CompactResponse compact(
+      OzoneManagerProtocolProtos.CompactRequest compactRequest)
+      throws IOException {
+    impl.compact(
+        compactRequest.getColumnFamily());
+    return OzoneManagerProtocolProtos.CompactResponse.newBuilder()
+        .build();
+  }
+
   @Override
   public OMClientResponse handleWriteRequestImpl(OMRequest omRequest, ExecutionContext context) throws IOException {
     injectPause();

hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/Compact.java

@@ -0,0 +1,83 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.ozone.admin.om;
+
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus;
+import org.apache.hadoop.util.Time;
+import picocli.CommandLine;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.stream.Collectors;
+
+import static org.apache.hadoop.ozone.OmUtils.getOmHostsFromConfig;
+import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus.PREPARE_COMPLETED;
+
+/**
+ * Handler of ozone admin om comact command.
+ */
+@CommandLine.Command(
+    name = "compact",
+    description = "Compact OM RocksDB",
+    mixinStandardHelpOptions = true,
+    versionProvider = HddsVersionProvider.class
+)
+public class Compact implements Callable<Void> {
+
+  @CommandLine.ParentCommand
+  private OMAdmin parent;
+
+  @CommandLine.Option(
+      names = {"-id", "--service-id"},
+      description = "Ozone Manager Service ID",
+      required = true
+  )
+  private String omServiceId;
+
+  @CommandLine.Option(
+      names = {"-host", "--service-host"},
+      description = "Ozone Manager Host"
+  )
+  private String omHost;
+
+  @CommandLine.Option(
+      names = {"-cf", "--column-family"},
+      description = "Column family to be compacted.",
+      defaultValue = "")
+  private String columnFamily;
+
+  @Override
+  public Void call() throws Exception {
+    OzoneManagerProtocol client = parent.createOmClient(omServiceId, omHost, false);
+    System.out.println("Compacting column family [" + columnFamily + "]...");
+
+    client.compact(columnFamily);
+    System.out.println("Compaction request issued.");
+
+    return null;
+  }
+
+}

hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/admin/om/OMAdmin.java

@@ -60,7 +60,8 @@
         UpdateRangerSubcommand.class,
         TransferOmLeaderSubCommand.class,
         FetchKeySubCommand.class,
-        LeaseSubCommand.class
+        LeaseSubCommand.class,
+        Compact.class
     })
 @MetaInfServices(AdminSubcommand.class)
 public class OMAdmin implements AdminSubcommand {