HDDS-12791. Make Retention Service suspendable

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

@@ -278,6 +278,7 @@ public static boolean isReadOnly(
     case GetQuotaRepairStatus:
     case StartQuotaRepair:
     case GetLifecycleConfiguration:
+    case GetLifecycleServiceStatus:
       return true;
     case CreateVolume:
     case SetVolumeProperty:
@@ -341,6 +342,7 @@ public static boolean isReadOnly(
     case DeleteObjectTagging:
     case SetLifecycleConfiguration:
     case DeleteLifecycleConfiguration:
+    case SetLifecycleServiceStatus:
     case UnknownCommand:
       return false;
     case EchoRPC:

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

@@ -66,6 +66,7 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CancelPrepareResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.EchoRPCResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetLifecycleServiceStatusResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PrepareStatusResponse.PrepareStatus;
@@ -1226,4 +1227,32 @@ default void deleteLifecycleConfiguration(String volumeName,
     throw new UnsupportedOperationException("OzoneManager does not require " +
         "this to be implemented, as write requests use a new approach.");
   }
+
+  /**
+   * Gets the lifecycle service status.
+   * @return GetLifecycleServiceStatusResponse
+   * @throws IOException
+   */
+  default GetLifecycleServiceStatusResponse getLifecycleServiceStatus() throws IOException {
+    throw new UnsupportedOperationException("OzoneManager does not require " +
+        "this to be implemented, as write requests use a new approach.");
+  }
+
+  /**
+   * Suspends the lifecycle service.
+   * @throws IOException
+   */
+  default void suspendLifecycleService() throws IOException {
+    throw new UnsupportedOperationException("OzoneManager does not require " +
+        "this to be implemented, as write requests use a new approach.");
+  }
+
+  /**
+   * Resumes the lifecycle service.
+   * @throws IOException
+   */
+  default void resumeLifecycleService() throws IOException {
+    throw new UnsupportedOperationException("OzoneManager does not require " +
+        "this to be implemented, as write requests use a new approach.");
+  }
 }

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

@@ -141,6 +141,7 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetKeyInfoResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetLifecycleConfigurationRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetLifecycleConfigurationResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetLifecycleServiceStatusResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetObjectTaggingRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetObjectTaggingResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetS3SecretRequest;
@@ -2691,6 +2692,21 @@ public OmLifecycleConfiguration getLifecycleConfiguration(String volumeName,
         resp.getLifecycleConfiguration());
   }
 
+  @Override
+  public GetLifecycleServiceStatusResponse getLifecycleServiceStatus() throws IOException {
+    OzoneManagerProtocolProtos.GetLifecycleServiceStatusRequest
+        getLifecycleServiceStatusRequest =
+        OzoneManagerProtocolProtos.GetLifecycleServiceStatusRequest
+            .newBuilder().build();
+
+    OMRequest omRequest = createOMRequest(Type.GetLifecycleServiceStatus)
+        .setGetLifecycleServiceStatusRequest(getLifecycleServiceStatusRequest)
+        .build();
+
+    return handleError(submitRequest(omRequest))
+        .getGetLifecycleServiceStatusResponse();
+  }
+
   @Override
   public void setLifecycleConfiguration(
       OmLifecycleConfiguration omLifecycleConfiguration) throws IOException {
@@ -2724,6 +2740,34 @@ public void deleteLifecycleConfiguration(String volumeName, String bucketName)
     handleError(submitRequest(omRequest));
   }
 
+  @Override
+  public void suspendLifecycleService() throws IOException {
+    OzoneManagerProtocolProtos.SetLifecycleServiceStatusRequest
+        setLifecycleServiceStatusRequest =
+        OzoneManagerProtocolProtos.SetLifecycleServiceStatusRequest
+            .newBuilder().setSuspend(true).build();
+
+    OMRequest omRequest = createOMRequest(Type.SetLifecycleServiceStatus)
+        .setSetLifecycleServiceStatusRequest(setLifecycleServiceStatusRequest)
+        .build();
+
+    handleError(submitRequest(omRequest));
+  }
+
+  @Override
+  public void resumeLifecycleService() throws IOException {
+    OzoneManagerProtocolProtos.SetLifecycleServiceStatusRequest
+        setLifecycleServiceStatusRequest =
+        OzoneManagerProtocolProtos.SetLifecycleServiceStatusRequest
+            .newBuilder().setSuspend(false).build();
+
+    OMRequest omRequest = createOMRequest(Type.SetLifecycleServiceStatus)
+        .setSetLifecycleServiceStatusRequest(setLifecycleServiceStatusRequest)
+        .build();
+
+    handleError(submitRequest(omRequest));
+  }
+
   private SafeMode toProtoBuf(SafeModeAction action) {
     switch (action) {
     case ENTER:

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java

@@ -194,6 +194,7 @@ protected static void startCluster(OzoneConfiguration conf) throws Exception {
     conf.setInt(OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL, 10);
     conf.setBoolean(OMConfigKeys.OZONE_OM_ENABLE_FILESYSTEM_PATHS, true);
     conf.setInt(ScmConfigKeys.OZONE_SCM_CONTAINER_LIST_MAX_COUNT, 1);
+    conf.setBoolean(OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_ENABLED, true);
     ozoneConfiguration = conf;
     MiniOzoneHAClusterImpl.Builder builder = MiniOzoneCluster.newHABuilder(conf);
     builder.setOMServiceId(omServiceId)
@@ -1834,6 +1835,81 @@ public void testSetEncryptionKey() throws Exception {
     assertEquals(newEncKey, volume.getBucket("bucket0").getEncryptionKeyName());
   }
 
+  @Test
+  public void testLifecycleStatus() throws UnsupportedEncodingException {
+    String[] args = new String[] {"om", "lifecycle", "status", "--service-id", omServiceId};
+    execute(ozoneAdminShell, args);
+    String output = out.toString(DEFAULT_ENCODING);
+    assertThat(output).contains("IsEnabled:");
+  }
+
+  @Test
+  public void testLifecycleSuspendAndResume() throws Exception {
+    List<OzoneManager> ozoneManagers = cluster.getOzoneManagersList();
+    for (OzoneManager om : ozoneManagers) {
+      assertNotNull(om.getKeyManager().getKeyLifecycleService());
+      assertTrue(om.getLifecycleServiceStatus().getIsEnabled());
+      assertFalse(om.getLifecycleServiceStatus().getIsSuspended());
+    }
+
+    // Execute suspend command
+    String[] args = new String[] {"om", "lifecycle", "suspend", "--service-id", omServiceId};
+    execute(ozoneAdminShell, args);
+    String output = out.toString(DEFAULT_ENCODING);
+    assertThat(output).contains("Lifecycle Service has been suspended");
+    out.reset();
+
+    // Wait for the suspend command to propagate through Ratis to all OMs
+    GenericTestUtils.waitFor(() -> {
+      for (OzoneManager om : ozoneManagers) {
+        assertNotNull(om.getKeyManager().getKeyLifecycleService());
+        if (!om.getLifecycleServiceStatus().getIsSuspended()) {
+          return false;
+        }
+      }
+      return true;
+    }, 100, 10000);
+
+    // Verify lifecycle service is suspended on all OMs
+    for (OzoneManager om : ozoneManagers) {
+      if (om.getKeyManager().getKeyLifecycleService() != null) {
+        assertTrue(om.getLifecycleServiceStatus().getIsSuspended(),
+            "Lifecycle service should be suspended on OM: " + om.getOMNodeId());
+        // isEnabled should still be true (based on configuration)
+        assertTrue(om.getLifecycleServiceStatus().getIsEnabled(),
+            "Lifecycle service isEnabled should still be true on OM: " + om.getOMNodeId());
+      }
+    }
+
+    // Execute resume command
+    args = new String[] {"om", "lifecycle", "resume", "--service-id", omServiceId};
+    execute(ozoneAdminShell, args);
+    output = out.toString(DEFAULT_ENCODING);
+    assertThat(output).contains("Lifecycle Service has been resumed");
+    out.reset();
+
+    // Wait for the resume command to propagate through Ratis to all OMs
+    GenericTestUtils.waitFor(() -> {
+      for (OzoneManager om : ozoneManagers) {
+        assertNotNull(om.getKeyManager().getKeyLifecycleService());
+        if (om.getLifecycleServiceStatus().getIsSuspended()) {
+          return false;
+        }
+      }
+      return true;
+    }, 100, 10000);
+
+    // Verify lifecycle service is resumed on all OMs
+    for (OzoneManager om : ozoneManagers) {
+      if (om.getKeyManager().getKeyLifecycleService() != null) {
+        assertFalse(om.getLifecycleServiceStatus().getIsSuspended(),
+            "Lifecycle service should be resumed on OM: " + om.getOMNodeId());
+        assertTrue(om.getLifecycleServiceStatus().getIsEnabled(),
+            "Lifecycle service isEnabled should be true on OM: " + om.getOMNodeId());
+      }
+    }
+  }
+
   @Test
   public void testCreateBucketWithECReplicationConfigWithoutReplicationParam() {
     getVolume("volume102");

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

@@ -159,6 +159,8 @@ enum Type {
   SetLifecycleConfiguration = 150;
   GetLifecycleConfiguration   = 151;
   DeleteLifecycleConfiguration = 152;
+  GetLifecycleServiceStatus    = 153;
+  SetLifecycleServiceStatus     = 154;
 }
 
 enum SafeMode {
@@ -310,6 +312,9 @@ message OMRequest {
   optional SetLifecycleConfigurationRequest setLifecycleConfigurationRequest = 150;
   optional GetLifecycleConfigurationRequest   getLifecycleConfigurationRequest   = 151;
   optional DeleteLifecycleConfigurationRequest deleteLifecycleConfigurationRequest = 152;
+
+  optional GetLifecycleServiceStatusRequest getLifecycleServiceStatusRequest = 153;
+  optional SetLifecycleServiceStatusRequest setLifecycleServiceStatusRequest = 154;
 }
 
 message OMResponse {
@@ -447,6 +452,9 @@ message OMResponse {
   optional SetLifecycleConfigurationResponse setLifecycleConfigurationResponse = 150;
   optional GetLifecycleConfigurationResponse   getLifecycleConfigurationResponse   = 151;
   optional DeleteLifecycleConfigurationResponse deleteLifecycleConfigurationResponse = 152;
+
+  optional GetLifecycleServiceStatusResponse getLifecycleServiceStatusResponse = 153;
+  optional SetLifecycleServiceStatusResponse setLifecycleServiceStatusResponse = 154;
 }
 
 enum Status {
@@ -2432,3 +2440,20 @@ service OzoneManagerService {
     rpc submitRequest(OMRequest)
           returns(OMResponse);
 }
+
+message GetLifecycleServiceStatusRequest {
+}
+
+message GetLifecycleServiceStatusResponse {
+  required bool isEnabled = 1;
+  optional bool isSuspended = 2;
+  repeated string runningBuckets = 3;
+}
+
+message SetLifecycleServiceStatusRequest {
+  required bool suspend = 1;
+}
+
+message SetLifecycleServiceStatusResponse {
+}
+

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/audit/OMAction.java

@@ -117,7 +117,9 @@ public enum OMAction implements AuditAction {
 
   GET_LIFECYCLE_CONFIGURATION,
   SET_LIFECYCLE_CONFIGURATION,
-  DELETE_LIFECYCLE_CONFIGURATION;
+  DELETE_LIFECYCLE_CONFIGURATION,
+  GET_LIFECYCLE_SERVICE_STATUS,
+  SET_LIFECYCLE_SERVICE_STATUS;
 
   @Override
   public String getAction() {

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

@@ -53,6 +53,8 @@
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_DEFAULT_BUCKET_LAYOUT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_DEFAULT_BUCKET_LAYOUT_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_DELETING_LIMIT_PER_TASK;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_ENABLED;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_ENABLED_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_EDEKCACHELOADER_INITIAL_DELAY_MS_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_EDEKCACHELOADER_INITIAL_DELAY_MS_KEY;
@@ -283,6 +285,7 @@
 import org.apache.hadoop.ozone.om.s3.S3SecretCacheProvider;
 import org.apache.hadoop.ozone.om.s3.S3SecretStoreProvider;
 import org.apache.hadoop.ozone.om.service.CompactDBService;
+import org.apache.hadoop.ozone.om.service.KeyLifecycleService;
 import org.apache.hadoop.ozone.om.service.OMRangerBGSyncService;
 import org.apache.hadoop.ozone.om.service.QuotaRepairTask;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotUtils;
@@ -295,6 +298,7 @@
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DBUpdatesRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.EchoRPCResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ExtendedUserAccessIdInfo;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetLifecycleServiceStatusResponse;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRoleInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.S3Authentication;
@@ -3159,6 +3163,18 @@ public OmLifecycleConfiguration getLifecycleConfiguration(String volumeName,
     }
   }
 
+  @Override
+  public GetLifecycleServiceStatusResponse getLifecycleServiceStatus() {
+    KeyLifecycleService keyLifecycleService = keyManager.getKeyLifecycleService();
+    if (keyLifecycleService == null) {
+      return GetLifecycleServiceStatusResponse.newBuilder()
+          .setIsEnabled(getConfiguration().getBoolean(OZONE_KEY_LIFECYCLE_SERVICE_ENABLED,
+              OZONE_KEY_LIFECYCLE_SERVICE_ENABLED_DEFAULT))
+          .build();
+    }
+    return keyLifecycleService.status();
+  }
+
   private Map<String, String> buildAuditMap(String volume) {
     Map<String, String> auditMap = new LinkedHashMap<>();
     auditMap.put(OzoneConsts.VOLUME, volume);

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

@@ -93,6 +93,7 @@ private static void init() {
     CMD_AUDIT_ACTION_MAP.put(Type.GetObjectTagging, OMAction.GET_OBJECT_TAGGING);
     CMD_AUDIT_ACTION_MAP.put(Type.PutObjectTagging, OMAction.PUT_OBJECT_TAGGING);
     CMD_AUDIT_ACTION_MAP.put(Type.DeleteObjectTagging, OMAction.DELETE_OBJECT_TAGGING);
+    CMD_AUDIT_ACTION_MAP.put(Type.SetLifecycleServiceStatus, OMAction.SET_LIFECYCLE_SERVICE_STATUS);
   }
 
   private static OMAction getAction(OzoneManagerProtocolProtos.OMRequest request) {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java

@@ -68,6 +68,7 @@
 import org.apache.hadoop.ozone.om.request.key.acl.prefix.OMPrefixSetAclRequest;
 import org.apache.hadoop.ozone.om.request.lifecycle.OMLifecycleConfigurationDeleteRequest;
 import org.apache.hadoop.ozone.om.request.lifecycle.OMLifecycleConfigurationSetRequest;
+import org.apache.hadoop.ozone.om.request.lifecycle.OMLifecycleSetServiceStatusRequest;
 import org.apache.hadoop.ozone.om.request.s3.multipart.S3ExpiredMultipartUploadsAbortRequest;
 import org.apache.hadoop.ozone.om.request.s3.security.OMSetSecretRequest;
 import org.apache.hadoop.ozone.om.request.s3.security.S3GetSecretRequest;
@@ -351,6 +352,8 @@ public static OMClientRequest createClientRequest(OMRequest omRequest,
       return new OMLifecycleConfigurationSetRequest(omRequest);
     case DeleteLifecycleConfiguration:
       return new OMLifecycleConfigurationDeleteRequest(omRequest);
+    case SetLifecycleServiceStatus:
+      return new OMLifecycleSetServiceStatusRequest(omRequest);
     default:
       throw new OMException("Unrecognized write command type request "
           + cmdType, OMException.ResultCodes.INVALID_REQUEST);