AMBARI-15133. Functionality to purge operational logs from the ambari database. (Laszlo Puskas via stoader)

ambari-server/pom.xml

@@ -1262,6 +1262,11 @@
       <artifactId>jetty-util-ajax</artifactId>
       <version>${jetty.version}</version>
     </dependency>
+    <dependency>
+      <groupId>commons-cli</groupId>
+      <artifactId>commons-cli</artifactId>
+      <version>1.3.1</version>
+    </dependency>
   </dependencies>
 
   <pluginRepositories>

ambari-server/sbin/ambari-server

@@ -155,11 +155,15 @@ case "$1" in
         ;;
   setup-sso)
         echo -e "Setting up SSO authentication properties..."
-                $PYTHON /usr/sbin/ambari-server.py $@
-                ;;
+        $PYTHON "$AMBARI_PYTHON_EXECUTABLE" $@
+        ;;
+  db-cleanup)
+        echo -e "Cleanup database..."
+        $PYTHON "$AMBARI_PYTHON_EXECUTABLE" $@
+        ;;
   *)
         echo "Usage: $AMBARI_PYTHON_EXECUTABLE
-        {start|stop|restart|setup|setup-jce|upgrade|status|upgradestack|setup-ldap|sync-ldap|set-current|setup-security|setup-sso|refresh-stack-hash|backup|restore|update-host-names|enable-stack|check-database} [options]
+        {start|stop|restart|setup|setup-jce|upgrade|status|upgradestack|setup-ldap|sync-ldap|set-current|setup-security|setup-sso|refresh-stack-hash|backup|restore|update-host-names|enable-stack|check-database|db-cleanup} [options]
         Use $AMBARI_PYTHON_EXECUTABLE <action> --help to get details on options available.
         Or, simply invoke ambari-server.py --help to print the options."
         exit 1

ambari-server/src/main/java/org/apache/ambari/server/cleanup/ClasspathScannerUtils.java

@@ -0,0 +1,129 @@
+/**
+ * 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.ambari.server.cleanup;
+
+import java.io.IOException;
+import java.lang.annotation.Annotation;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.reflect.ClassPath;
+
+/**
+ * Utility for looking up classes on the classpath that are potentially subject to be bound by a multibinder.
+ */
+public class ClasspathScannerUtils {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ClasspathScannerUtils.class);
+
+  /**
+   * Scans the classpath for classes based on the provided arguments
+   *
+   * @param packageName the package to be scanned
+   * @param exclusions  a list with classes excluded from the result
+   * @param selectors   a list with annotation and interface classes that identify classes to be found (lookup criteria)
+   * @return a list of classes from the classpath that match the lookup criteria
+   */
+  public static Set<Class> findOnClassPath(String packageName, List<Class> exclusions, List<Class> selectors) {
+
+    Set<Class> bindingSet = new HashSet<>();
+    try {
+      ClassPath classpath = ClassPath.from(ClasspathScannerUtils.class.getClassLoader());
+      LOGGER.info("Checking package [{}] for binding candidates.", packageName);
+
+      for (ClassPath.ClassInfo classInfo : classpath.getTopLevelClassesRecursive(packageName)) {
+        Class candidate = classInfo.load();
+
+        if (exclusions.contains(candidate)) {
+          LOGGER.debug("Candidate [{}] is excluded excluded.", candidate);
+          continue;
+        }
+
+        if (isEligible(candidate, selectors)) {
+          LOGGER.info("Found class [{}]", candidate);
+          bindingSet.add(candidate);
+        } else {
+          LOGGER.debug("Candidate [{}] doesn't match.", candidate);
+        }
+      }
+
+    } catch (IOException e) {
+      LOGGER.error("Failure during configuring JUICE bindings.", e);
+      throw new IllegalArgumentException(e);
+    }
+    return bindingSet;
+  }
+
+
+  /**
+   * Checks whether the candidate class matches lookup conditions.
+   *
+   * @param candidate the type to be checked
+   * @return true if the class matches, false otherwise
+   */
+  private static boolean isEligible(Class candidate, List<Class> selectors) {
+    return checkSubClasses(candidate, selectors) || checkAnnotations(candidate, selectors);
+  }
+
+  /**
+   * Checks if the candidate has annotations listed in the selection criteria
+   *
+   * @param candidate the type to be checked
+   * @return true if the candidate has annotations listed in the selection criteria, false otherwise
+   */
+  private static boolean checkAnnotations(Class candidate, List<Class> selectors) {
+    LOGGER.debug("Checking annotations for: [{}]", candidate);
+    boolean ret = false;
+    for (Annotation candidateAnn : candidate.getDeclaredAnnotations()) {
+      if (selectors.contains(candidateAnn)) {
+        ret = true;
+        break;
+      }
+    }
+    return ret;
+  }
+
+  /**
+   * Checks if the candidate implements interfaces listed in the selection criteria
+   *
+   * @param candidate the type to be checked
+   * @return true if the candidate implements interfaces listed in the selection criteria, false otherwise
+   */
+  private static boolean checkSubClasses(Class candidate, List<Class> selectors) {
+    boolean ret = false;
+    LOGGER.debug("Checking interfaces for: [{}]", candidate);
+    List<Class> interfaces = Arrays.asList(candidate.getInterfaces());
+
+    for (Class selectorItf : selectors) {
+      if (interfaces.contains(selectorItf)) {
+        LOGGER.debug("Checking candidate for subclassing interface: ", selectorItf);
+        if (selectorItf.getClass().isAssignableFrom(candidate.getClass())) {
+          ret = true;
+          break;
+        }
+      }
+    }
+    return ret;
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/cleanup/CleanupDriver.java

@@ -0,0 +1,117 @@
+/*
+ * 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.ambari.server.cleanup;
+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+import org.apache.ambari.server.controller.ControllerModule;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.persist.jpa.AmbariJpaPersistService;
+
+/**
+ * Class in charge for driving the cleanup process.
+ */
+public class CleanupDriver {
+  private static final Logger LOGGER = LoggerFactory.getLogger(CleanupDriver.class);
+
+  private static final String DATE_PATTERN = "yyyy-MM-dd";
+  private static final String CLUSTER_NAME_ARG = "cluster-name";
+  private static final String FROM_DATE_ARG = "from-date";
+
+  private static Options getOptions() {
+    Options options = new Options();
+    options.addOption(Option.builder().longOpt(CLUSTER_NAME_ARG).desc("The cluster name").required().type(String.class).hasArg().valueSeparator(' ').build());
+    options.addOption(Option.builder().longOpt(FROM_DATE_ARG).desc("The day from which the cleanup runs").required().type(String.class).hasArg().valueSeparator(' ').build());
+    return options;
+  }
+
+  private static CleanupContext processArguments(String... args) {
+    CommandLineParser cmdLineParser = new DefaultParser();
+    HelpFormatter formatter = new HelpFormatter();
+    DateFormat df = new SimpleDateFormat(DATE_PATTERN);
+    CleanupContext ctx = null;
+
+    try {
+      CommandLine line = cmdLineParser.parse(getOptions(), args);
+      String clusterName = (String) line.getParsedOptionValue(CLUSTER_NAME_ARG);
+      Date fromDate = df.parse(line.getOptionValue(FROM_DATE_ARG));
+      ctx = new CleanupContext(clusterName, fromDate.getTime());
+    } catch (Exception exp) {
+      System.err.println("Parsing failed.  Reason: " + exp.getMessage());
+      LOGGER.error("Parsing failed.  Reason: ", exp);
+      formatter.printHelp("cleanup", getOptions());
+      System.exit(1);
+    }
+    return ctx;
+  }
+
+
+  public static void main(String... args) throws Exception {
+    LOGGER.info("DB-CLEANUP - Starting the cleanup process ...");
+
+    CleanupContext cleanupContext = processArguments(args);
+
+    // set up the guice context
+    Injector injector = Guice.createInjector(new ControllerModule(), new CleanupModule());
+
+    // explicitly starting the persist service
+    injector.getInstance(AmbariJpaPersistService.class).start();
+
+    CleanupServiceImpl cleanupService = injector.getInstance(CleanupServiceImpl.class);
+    long affected = cleanupService.cleanup(new TimeBasedCleanupPolicy(cleanupContext.getClusterName(), cleanupContext.getFromDayTimestamp()));
+
+    // explicitly stopping the persist service
+    injector.getInstance(AmbariJpaPersistService.class).stop();
+
+    LOGGER.info("DB-CLEANUP - completed. Number of affected records [{}]", affected);
+  }
+
+  /**
+   * Context object that encapsulates values passed in as arguments to the driver class.
+   * Represents the input for the cleanup process.
+   */
+  private static class CleanupContext {
+    private String clusterName;
+    private Long fromDayTimestamp;
+
+    public CleanupContext(String clusterName, Long fromDayTimestamp) {
+      this.clusterName = clusterName;
+      this.fromDayTimestamp = fromDayTimestamp;
+    }
+
+    public String getClusterName() {
+      return clusterName;
+    }
+
+    public Long getFromDayTimestamp() {
+      return fromDayTimestamp;
+    }
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/cleanup/CleanupModule.java

@@ -0,0 +1,76 @@
+/*
+ * 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.ambari.server.cleanup;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.ambari.server.orm.dao.Cleanable;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Scopes;
+import com.google.inject.multibindings.Multibinder;
+
+/**
+ * Configuration module for the cleanup framework.
+ */
+public class CleanupModule extends AbstractModule {
+
+  /**
+   * Selectors identifying objects to be bound.
+   *
+   * @return a list with interface and annotation types
+   */
+  protected List<Class> getSelectors() {
+    List<Class> selectorList = new ArrayList<>();
+    selectorList.add(Cleanable.class);
+    return selectorList;
+  }
+
+  /**
+   * Gets the list of types to be excluded from bindings.
+   *
+   * @return a list with types to be left out from dynamic bindings
+   */
+  protected List<Class> getExclusions() {
+    return Collections.emptyList();
+  }
+
+  /**
+   * Returns the package to be scanned for bindings of this module.
+   *
+   * @return the name of the package to be scanned
+   */
+  protected String getPackageToScan() {
+    return Cleanable.class.getPackage().getName();
+  }
+
+
+  @Override
+  protected void configure() {
+
+    Multibinder<Cleanable> multiBinder = Multibinder.newSetBinder(binder(), Cleanable.class);
+    Set<Class> bindingSet = ClasspathScannerUtils.findOnClassPath(getPackageToScan(), getExclusions(), getSelectors());
+    for (Class clazz : bindingSet) {
+      multiBinder.addBinding().to(clazz).in(Scopes.SINGLETON);
+    }
+    bind(CleanupServiceImpl.class).in(Scopes.SINGLETON);
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/cleanup/CleanupService.java

@@ -0,0 +1,33 @@
+/*
+ * 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.ambari.server.cleanup;
+
+/**
+ * Contract for services in charge for cleaning operational data.
+ * @param <T> the type based on which the cleanup is done
+ */
+public interface CleanupService<T> {
+
+  /**
+   * Triggers the cleanup for the given cleanup policy.
+   *
+   * @param cleanupPolicy the cleanup policy based on which the cleanup is executed.
+   * @return the affected "rows"
+   */
+  long cleanup(T cleanupPolicy);
+}

ambari-server/src/main/java/org/apache/ambari/server/cleanup/CleanupServiceImpl.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
+ *
+ *     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.ambari.server.cleanup;
+
+import java.util.Set;
+
+import javax.inject.Inject;
+
+import org.apache.ambari.server.orm.dao.Cleanable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Singleton;
+
+/**
+ * Service in charge to perform the cleanup/purge on the entities that support this functionality.
+ */
+@Singleton
+public class CleanupServiceImpl implements CleanupService<TimeBasedCleanupPolicy> {
+  private static final Logger LOGGER = LoggerFactory.getLogger(CleanupServiceImpl.class);
+
+
+  // this Set is automatically populated by the guice framework (based on the cleanup interface)
+  private Set<Cleanable> cleanables;
+
+  /**
+   * Constructor for testing purposes.
+   *
+   * @param cleanables
+   */
+  @Inject
+  protected CleanupServiceImpl(Set<Cleanable> cleanables) {
+    this.cleanables = cleanables;
+  }
+
+  /**
+   * Triggers the cleanup process on the registered DAOs.
+   *
+   * @param cleanupPolicy the policy based on which the cleanup is done
+   * @return the number of affected rows
+   */
+  public long cleanup(TimeBasedCleanupPolicy cleanupPolicy) {
+    long affectedRows = 0;
+    for (Cleanable cleanable : cleanables) {
+      LOGGER.info("Running the purge process for DAO: [{}] with cleanup policy: [{}]", cleanable, cleanupPolicy);
+      affectedRows += cleanable.cleanup(cleanupPolicy);
+    }
+    return affectedRows;
+  }
+
+}
+

ambari-server/src/main/java/org/apache/ambari/server/cleanup/PurgePolicy.java

@@ -0,0 +1,33 @@
+/*
+ * 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.ambari.server.cleanup;
+
+/**
+ * Supported purge policies.
+ */
+public enum PurgePolicy {
+  /**
+   * Delete from the database
+   */
+  DELETE,
+
+  /**
+   * Archive records, but do not delete them
+   */
+  ARCHIVE
+}

ambari-server/src/main/java/org/apache/ambari/server/cleanup/TimeBasedCleanupPolicy.java

@@ -0,0 +1,62 @@
+package org.apache.ambari.server.cleanup;
+
+/**
+ * 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.
+ */
+
+/**
+ * Time and cluster id based cleanup policy
+ */
+public class TimeBasedCleanupPolicy {
+
+  private String clusterName;
+  private Long toDateInMillis;
+
+  /**
+   * Constructs an instance based on the given arguments.
+   *
+   * @param clusterName      the cluster name
+   * @param toDateInMillis timestamp before that entities are purged.
+   */
+  public TimeBasedCleanupPolicy(String clusterName, Long toDateInMillis) {
+    this.clusterName = clusterName;
+    this.toDateInMillis = toDateInMillis;
+  }
+
+  /**
+   * @return the cluster name
+   */
+  public String getClusterName() {
+    return clusterName;
+  }
+
+  /**
+   *
+   * @return the timestamp before that entities are purged
+   */
+  public Long getToDateInMillis() {
+    return toDateInMillis;
+  }
+
+  /**
+   *
+   * @return The used purge policy
+   */
+  public PurgePolicy getPurgePolicy() {
+    return PurgePolicy.DELETE;
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/AlertsDAO.java

@@ -37,8 +37,10 @@ import javax.persistence.metamodel.SingularAttribute;
 
 import org.apache.ambari.annotations.Experimental;
 import org.apache.ambari.annotations.ExperimentalFeature;
+import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.api.query.JpaPredicateVisitor;
 import org.apache.ambari.server.api.query.JpaSortBuilder;
+import org.apache.ambari.server.cleanup.TimeBasedCleanupPolicy;
 import org.apache.ambari.server.configuration.Configuration;
 import org.apache.ambari.server.controller.AlertCurrentRequest;
 import org.apache.ambari.server.controller.AlertHistoryRequest;
@@ -51,6 +53,7 @@ import org.apache.ambari.server.orm.entities.AlertCurrentEntity;
 import org.apache.ambari.server.orm.entities.AlertCurrentEntity_;
 import org.apache.ambari.server.orm.entities.AlertHistoryEntity;
 import org.apache.ambari.server.orm.entities.AlertHistoryEntity_;
+import org.apache.ambari.server.orm.entities.AlertNoticeEntity;
 import org.apache.ambari.server.state.AlertState;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
@@ -81,7 +84,7 @@ import com.google.inject.persist.Transactional;
  */
 @Singleton
 @Experimental(feature = ExperimentalFeature.ALERT_CACHING)
-public class AlertsDAO {
+public class AlertsDAO implements Cleanable {
   /**
    * Logger.
    */
@@ -1185,6 +1188,25 @@ public class AlertsDAO {
     return cachedAlerts;
   }
 
+  @Transactional
+  @Override
+  public long cleanup(TimeBasedCleanupPolicy policy) {
+    long affectedRows = 0;
+    Long clusterId = null;
+    try {
+      clusterId = m_clusters.get().getCluster(policy.getClusterName()).getClusterId();
+      affectedRows += cleanAlertNoticesForClusterBeforeDate(clusterId, policy.getToDateInMillis());
+      affectedRows += cleanAlertCurrentsForClusterBeforeDate(clusterId, policy.getToDateInMillis());
+      affectedRows += cleanAlertHistoriesForClusterBeforeDate(clusterId, policy.getToDateInMillis());
+    } catch (AmbariException e) {
+      LOG.error("Error while looking up cluster with name: {}", policy.getClusterName(), e);
+      throw new IllegalStateException(e);
+    }
+
+    return affectedRows;
+  }
+
+
   /**
    * The {@link HistoryPredicateVisitor} is used to convert an Ambari
    * {@link Predicate} into a JPA {@link javax.persistence.criteria.Predicate}.
@@ -1410,4 +1432,71 @@ public class AlertsDAO {
   @SuppressWarnings("serial")
   private static final class AlertNotYetCreatedException extends Exception {
   }
+
+
+  /**
+   * Deletes AlertNotice records in relation with AlertHistory entries older than the given date.
+   *
+   * @param clusterId        the identifier of the cluster the AlertNotices belong to
+   * @param beforeDateMillis the date in milliseconds the
+   * @return a long representing the number of affected (deleted) records
+   */
+  @Transactional
+  private int cleanAlertNoticesForClusterBeforeDate(Long clusterId, long beforeDateMillis) {
+    return executeQuery("AlertNoticeEntity.removeByAlertHistoryBeforeDate", AlertNoticeEntity.class, clusterId, beforeDateMillis);
+  }
+
+
+  /**
+   * Deletes AlertCurrent records in relation with AlertHistory entries older than the given date.
+   *
+   * @param clusterId        the identifier of the cluster the AlertCurrents belong to
+   * @param beforeDateMillis the date in milliseconds the
+   * @return a long representing the number of affected (deleted) records
+   */
+  @Transactional
+  private int cleanAlertCurrentsForClusterBeforeDate(long clusterId, long beforeDateMillis) {
+    return executeQuery("AlertCurrentEntity.removeByAlertHistoryBeforeDate", AlertCurrentEntity.class, clusterId, beforeDateMillis);
+  }
+
+  /**
+   * Deletes AlertHistory entries in a cluster older than the given date.
+   *
+   * @param clusterId        the identifier of the cluster the AlertHistory entries belong to
+   * @param beforeDateMillis the date in milliseconds the
+   * @return a long representing the number of affected (deleted) records
+   */
+
+  @Transactional
+  private int cleanAlertHistoriesForClusterBeforeDate(Long clusterId, long beforeDateMillis) {
+    return executeQuery("AlertHistoryEntity.removeInClusterBeforeDate", AlertHistoryEntity.class, clusterId, beforeDateMillis);
+  }
+
+  /**
+   * Utility method for executing update or delete named queries having as input parameters the cluster id and a timestamp.
+   *
+   * @param namedQuery the named query to be executed
+   * @param entityType the type of the entity
+   * @param clusterId  the cluster identifier
+   * @param timestamp  timestamp
+   * @return the number of rows affected by the query execution.
+   */
+  private int executeQuery(String namedQuery, Class entityType, long clusterId, long timestamp) {
+    LOG.info("Starting: Delete/update entries older than [ {} ] for entity [{}]", timestamp, entityType);
+    TypedQuery query = m_entityManagerProvider.get().createNamedQuery(namedQuery, entityType);
+
+    query.setParameter("clusterId", clusterId);
+    query.setParameter("beforeDate", timestamp);
+
+    int affectedRows = query.executeUpdate();
+
+    m_entityManagerProvider.get().flush();
+    m_entityManagerProvider.get().clear();
+
+    LOG.info("Completed: Delete/update entries older than [ {} ] for entity: [{}]. Number of entities deleted: [{}]",
+        timestamp, entityType, affectedRows);
+
+    return affectedRows;
+  }
+
 }

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/Cleanable.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.ambari.server.orm.dao;
+
+import org.apache.ambari.server.cleanup.TimeBasedCleanupPolicy;
+
+/**
+ * Interface to be implemented by all DAOs that support the cleanup functionality.
+ * All implementing DAO are automatically configured in the cleanup process.
+ *
+ */
+public interface Cleanable {
+
+  /**
+   * Performs the cleanup for the entiries the implementing DAO is responsible for.
+   *
+   * @param policy the policy with the parameters of the cleanup
+   * @return the number of affected records if available
+   */
+  long cleanup(TimeBasedCleanupPolicy policy);
+}

ambari-server/src/main/java/org/apache/ambari/server/orm/entities/AlertCurrentEntity.java

@@ -58,7 +58,9 @@ import org.apache.ambari.server.state.MaintenanceState;
     @NamedQuery(name = "AlertCurrentEntity.removeDisabled", query = "DELETE FROM AlertCurrentEntity alert WHERE alert.alertDefinition.enabled = 0"),
     @NamedQuery(name = "AlertCurrentEntity.removeByService", query = "DELETE FROM AlertCurrentEntity alert WHERE alert.alertHistory.serviceName = :serviceName"),
     @NamedQuery(name = "AlertCurrentEntity.removeByHost", query = "DELETE FROM AlertCurrentEntity alert WHERE alert.alertHistory.hostName = :hostName"),
-    @NamedQuery(name = "AlertCurrentEntity.removeByHostComponent", query = "DELETE FROM AlertCurrentEntity alert WHERE alert.alertHistory.serviceName = :serviceName AND alert.alertHistory.componentName = :componentName AND alert.alertHistory.hostName = :hostName") })
+    @NamedQuery(name = "AlertCurrentEntity.removeByHostComponent", query = "DELETE FROM AlertCurrentEntity alert WHERE alert.alertHistory.serviceName = :serviceName AND alert.alertHistory.componentName = :componentName AND alert.alertHistory.hostName = :hostName"),
+    @NamedQuery(name = "AlertCurrentEntity.removeByAlertHistoryBeforeDate", query = "DELETE FROM AlertCurrentEntity alert WHERE alert.alertHistory.clusterId = :clusterId AND alert.alertHistory.alertTimestamp <= :beforeDate")
+})
 public class AlertCurrentEntity {
 
   @Id

ambari-server/src/main/java/org/apache/ambari/server/orm/entities/AlertHistoryEntity.java

@@ -51,7 +51,9 @@ import org.apache.ambari.server.state.AlertState;
     @NamedQuery(name = "AlertHistoryEntity.findAllInClusterBetweenDates", query = "SELECT alertHistory FROM AlertHistoryEntity alertHistory WHERE alertHistory.clusterId = :clusterId AND alertHistory.alertTimestamp BETWEEN :startDate AND :endDate"),
     @NamedQuery(name = "AlertHistoryEntity.findAllInClusterBeforeDate", query = "SELECT alertHistory FROM AlertHistoryEntity alertHistory WHERE alertHistory.clusterId = :clusterId AND alertHistory.alertTimestamp <= :beforeDate"),
     @NamedQuery(name = "AlertHistoryEntity.findAllInClusterAfterDate", query = "SELECT alertHistory FROM AlertHistoryEntity alertHistory WHERE alertHistory.clusterId = :clusterId AND alertHistory.alertTimestamp >= :afterDate"),
-    @NamedQuery(name = "AlertHistoryEntity.removeByDefinitionId", query = "DELETE FROM AlertHistoryEntity alertHistory WHERE alertHistory.alertDefinition.definitionId = :definitionId") })
+    @NamedQuery(name = "AlertHistoryEntity.removeByDefinitionId", query = "DELETE FROM AlertHistoryEntity alertHistory WHERE alertHistory.alertDefinition.definitionId = :definitionId"),
+    @NamedQuery(name = "AlertHistoryEntity.removeInClusterBeforeDate", query = "DELETE FROM AlertHistoryEntity alertHistory WHERE alertHistory.clusterId= :clusterId AND alertHistory.alertTimestamp <= :beforeDate")
+})
 public class AlertHistoryEntity {
 
   @Id

ambari-server/src/main/java/org/apache/ambari/server/orm/entities/AlertNoticeEntity.java

@@ -49,7 +49,9 @@ import org.apache.ambari.server.state.NotificationState;
     @NamedQuery(name = "AlertNoticeEntity.findAll", query = "SELECT notice FROM AlertNoticeEntity notice"),
     @NamedQuery(name = "AlertNoticeEntity.findByState", query = "SELECT notice FROM AlertNoticeEntity notice WHERE notice.notifyState = :notifyState"),
     @NamedQuery(name = "AlertNoticeEntity.findByUuid", query = "SELECT notice FROM AlertNoticeEntity notice WHERE notice.uuid = :uuid"),
-    @NamedQuery(name = "AlertNoticeEntity.removeByDefinitionId", query = "DELETE FROM AlertNoticeEntity notice WHERE notice.alertHistory.alertDefinition.definitionId = :definitionId") })
+    @NamedQuery(name = "AlertNoticeEntity.removeByDefinitionId", query = "DELETE FROM AlertNoticeEntity notice WHERE notice.alertHistory.alertDefinition.definitionId = :definitionId"),
+    @NamedQuery(name = "AlertNoticeEntity.removeByAlertHistoryBeforeDate", query = "DELETE FROM AlertNoticeEntity notice WHERE notice.alertHistory.clusterId = :clusterId AND notice.alertHistory.alertTimestamp <= :beforeDate")
+})
 public class AlertNoticeEntity {
 
   @Id

ambari-server/src/main/python/ambari-server.py

@@ -37,6 +37,7 @@ from ambari_server.serverSetup import reset, setup, setup_jce_policy
 from ambari_server.serverUpgrade import upgrade, upgrade_stack, set_current
 from ambari_server.setupHttps import setup_https, setup_truststore
 from ambari_server.setupSso import setup_sso
+from ambari_server.dbCleanup import db_cleanup
 from ambari_server.hostUpdate import update_host_names
 from ambari_server.checkDatabase import check_database
 from ambari_server.enableStack import enable_stack_version
@@ -45,7 +46,7 @@ from ambari_server.setupActions import BACKUP_ACTION, LDAP_SETUP_ACTION, LDAP_SY
   REFRESH_STACK_HASH_ACTION, RESET_ACTION, RESTORE_ACTION, UPDATE_HOST_NAMES_ACTION, CHECK_DATABASE_ACTION, \
   SETUP_ACTION, SETUP_SECURITY_ACTION,START_ACTION, STATUS_ACTION, STOP_ACTION, UPGRADE_ACTION, UPGRADE_STACK_ACTION, \
   SETUP_JCE_ACTION, SET_CURRENT_ACTION, START_ACTION, STATUS_ACTION, STOP_ACTION, UPGRADE_ACTION, UPGRADE_STACK_ACTION, SETUP_JCE_ACTION, \
-  SET_CURRENT_ACTION, ENABLE_STACK_ACTION, SETUP_SSO_ACTION
+  SET_CURRENT_ACTION, ENABLE_STACK_ACTION, SETUP_SSO_ACTION, DB_CLEANUP_ACTION
 from ambari_server.setupSecurity import setup_ldap, sync_ldap, setup_master_key, setup_ambari_krb5_jaas
 from ambari_server.userInput import get_validated_string_input
 
@@ -392,6 +393,8 @@ def init_parser_options(parser):
                     help="Specify stack version that needs to be enabled. All other stacks versions will be disabled")
   parser.add_option('--stack', dest="stack_name", default=None, type="string",
                     help="Specify stack name for the stack versions that needs to be enabled")
+  parser.add_option("-d", "--from-date", dest="cleanup_from_date", default=None, type="string", help="Specify date for the cleanup process in 'yyyy-MM-dd' format")
+
 
 @OsFamilyFuncImpl(OSConst.WINSRV_FAMILY)
 def are_cmd_line_db_args_blank(options):
@@ -546,7 +549,8 @@ def create_user_action_map(args, options):
         UPDATE_HOST_NAMES_ACTION: UserActionPossibleArgs(update_host_names, [2], args, options),
         CHECK_DATABASE_ACTION: UserAction(check_database, options),
         ENABLE_STACK_ACTION: UserAction(enable_stack, options, args),
-        SETUP_SSO_ACTION: UserActionRestart(setup_sso, options)
+        SETUP_SSO_ACTION: UserActionRestart(setup_sso, options),
+        DB_CLEANUP_ACTION: UserAction(db_cleanup, options)
       }
   return action_map
 

ambari-server/src/main/python/ambari_server/dbCleanup.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+from ambari_commons.logging_utils import print_info_msg, print_error_msg
+from ambari_commons.os_utils import run_os_command
+from ambari_server.dbConfiguration import ensure_jdbc_driver_is_installed
+from ambari_server.serverConfiguration import configDefaults, \
+    get_ambari_properties, get_java_exe_path, read_ambari_user, get_db_type
+from ambari_server.setupSecurity import generate_env, ensure_can_start_under_current_user
+from ambari_server.userInput import get_YN_input
+from ambari_server.serverClassPath import ServerClassPath
+from ambari_server.serverUtils import is_server_runing
+import datetime
+
+DB_CLEANUP_CMD = "{0} -cp {1} org.apache.ambari.server.cleanup.CleanupDriver --cluster-name {2} --from-date {3}> " + configDefaults.SERVER_OUT_FILE + " 2>&1"
+
+#
+# Run the db cleanup process
+#
+def run_db_cleanup(options):
+
+    if validate_args(options):
+        return 1
+
+    db_title = get_db_type(get_ambari_properties()).title
+
+    confirmBackup = get_YN_input("Ambari Server configured for {0}. Confirm you have made a backup of the Ambari Server database [y/n]".format(
+            db_title), True)
+    if not confirmBackup:
+        print_info_msg("Ambari Server Database cleanup aborted")
+        return 0
+
+    status, stateDesc = is_server_runing()
+    if status:
+        print_error_msg("The database cleanup cannot proceed while Ambari Server is running. Please shut down Ambari first.")
+        return 1
+
+    confirm = get_YN_input(
+        "Ambari server is using db type {0}. Cleanable database entries older than {1} will be cleaned up. Proceed [y/n]".format(
+            db_title, options.cleanup_from_date), True)
+    if not confirm:
+        print_info_msg("Ambari Server Database cleanup aborted")
+        return 0
+
+    jdk_path = get_java_exe_path()
+    if jdk_path is None:
+        print_error_msg("No JDK found, please run the \"setup\" command to install a JDK automatically or install any "
+                        "JDK manually to {0}".format(configDefaults.JDK_INSTALL_DIR));
+        return 1
+
+    ensure_jdbc_driver_is_installed(options, get_ambari_properties())
+
+    serverClassPath = ServerClassPath(get_ambari_properties(), options)
+    class_path = serverClassPath.get_full_ambari_classpath_escaped_for_shell()
+
+    ambari_user = read_ambari_user()
+    current_user = ensure_can_start_under_current_user(ambari_user)
+    environ = generate_env(options, ambari_user, current_user)
+
+    print "Cleaning up the database ..."
+    command = DB_CLEANUP_CMD.format(jdk_path, class_path, options.cluster_name, options.cleanup_from_date)
+    (retcode, stdout, stderr) = run_os_command(command, env=environ)
+
+    print_info_msg("Return code from database cleanup command, retcode = " + str(retcode))
+
+    if stdout:
+        print "Console output from database cleanup command:"
+        print stdout
+        print
+    if stderr:
+        print "Error output from database cleanup command:"
+        print stderr
+        print
+    if retcode > 0:
+        print_error_msg("Error wncountered while cleaning up the Ambari Server Database. Check the ambari-server.log for details.")
+    else:
+        print "Cleanup completed. Check the ambari-server.log for details."
+    return retcode
+
+#
+# Database cleanup
+#
+def db_cleanup(options):
+    return run_db_cleanup(options)
+
+
+def validate_args(options):
+    if not options.cluster_name:
+        print_error_msg("Please provide the --cluster-name argument.")
+        return 1
+
+    if not options.cleanup_from_date:
+        print_error_msg("Please provide the --from-date argument.")
+        return 1
+
+    try:
+        datetime.datetime.strptime(options.cleanup_from_date, "%Y-%m-%d")
+    except ValueError as e:
+        print_error_msg("The --from-date argument has an invalid format. {0}".format(e.args[0]))
+        return 1;

ambari-server/src/main/python/ambari_server/setupActions.py

@@ -41,4 +41,5 @@ CHECK_DATABASE_ACTION = "check-database"
 BACKUP_ACTION = "backup"
 RESTORE_ACTION = "restore"
 SETUP_JCE_ACTION = "setup-jce"
-ENABLE_STACK_ACTION = "enable-stack"
+ENABLE_STACK_ACTION = "enable-stack"
+DB_CLEANUP_ACTION = "db-cleanup"

ambari-server/src/test/java/org/apache/ambari/server/cleanup/CleanupServiceFunctionalTest.java

@@ -0,0 +1,161 @@
+/*
+ *
+ *  * 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.ambari.server.cleanup;
+
+
+import java.util.Properties;
+
+import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.controller.ControllerModule;
+import org.eclipse.persistence.config.PersistenceUnitProperties;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Module;
+import com.google.inject.persist.jpa.AmbariJpaPersistModule;
+import com.google.inject.persist.jpa.AmbariJpaPersistService;
+
+import junit.framework.Assert;
+
+/**
+ * Functional test for the Cleanup process.
+ */
+@Ignore("Ignored in order not to run with the unit tests as it's time consuming. Should be part of a functional test suit.")
+public class CleanupServiceFunctionalTest {
+
+  private static Injector injector;
+
+  @BeforeClass
+  public static void beforeClass() throws Exception {
+
+    injector = Guice.createInjector(
+        new CleanupModule(),
+        new ControllerModule(getTestProperties())
+    );
+
+    // start the persistService
+    injector.getInstance(AmbariJpaPersistService.class).start();
+
+  }
+
+  private static Module getTestPersistModule() {
+    AmbariJpaPersistModule persistModule = new AmbariJpaPersistModule("ambari-server");
+    Configuration testConfiguration = new Configuration(getTestProperties());
+    Properties persistenceProperties = ControllerModule.getPersistenceProperties(testConfiguration);
+
+    // overriding JPA properties with test specific values
+    persistenceProperties.setProperty(PersistenceUnitProperties.SCHEMA_GENERATION_DATABASE_ACTION, PersistenceUnitProperties.NONE);
+
+    // this doesn't work for something similar to what's described here: http://stackoverflow.com/questions/3606825/problems-with-generating-sql-via-eclipselink-missing-separator
+    // line breaks are not supported; SQL statements need to be in one line! - appears to be fixed in eclipselink 2.6
+    //persistenceProperties.setProperty(PersistenceUnitProperties.SCHEMA_GENERATION_CREATE_SCRIPT_SOURCE, "Ambari-DDL-Postgres-CREATE.sql");
+
+    // Let the initialization be performed by the jPA provider!
+    //persistenceProperties.setProperty(PersistenceUnitProperties.SCHEMA_GENERATION_SQL_LOAD_SCRIPT_SOURCE, getTestDataDDL());
+
+    //persistenceProperties.setProperty(PersistenceUnitProperties.SCHEMA_GENERATION_DROP_SOURCE, PersistenceUnitProperties.SCHEMA_GENERATION_METADATA_SOURCE);
+    persistenceProperties.setProperty(PersistenceUnitProperties.THROW_EXCEPTIONS, "true");
+
+    // todo remove these when switching to derby!
+    persistenceProperties.setProperty(PersistenceUnitProperties.JDBC_USER, "ambari");
+    persistenceProperties.setProperty(PersistenceUnitProperties.JDBC_PASSWORD, "bigdata");
+
+    return persistModule.properties(persistenceProperties);
+  }
+
+  private static String getTestDataDDL() {
+    return "ddl-func-test/ddl-cleanup-test-data.sql";
+  }
+
+  @AfterClass
+  public static void afterClass() {
+    injector.getInstance(AmbariJpaPersistService.class).stop();
+  }
+
+  /**
+   * Override JPA config values read from the ambari.properties
+   *
+   * @return
+   */
+  private static Properties getTestProperties() {
+    Properties properties = new Properties();
+    properties.put("server.jdbc.connection-pool", "internal");
+    properties.put("server.persistence.type", "remote");
+
+    properties.put("server.jdbc.driver", "org.postgresql.Driver");
+    properties.put("server.jdbc.user.name", "ambari");
+    //properties.put("server.jdbc.user.passwd", "bigdata");
+    properties.put("server.jdbc.url", "jdbc:postgresql://192.168.59.103:5432/ambari");
+    properties.put(Configuration.SHARED_RESOURCES_DIR_KEY, "/Users/lpuskas/prj/ambari/ambari-server/src/test/resources");
+
+    return properties;
+  }
+
+  @Before
+  public void setUp() throws Exception {
+  }
+
+  @Test
+  public void testIOCContext() throws Exception {
+
+    // WHEN
+    CleanupServiceImpl cleanupService = injector.getInstance(CleanupServiceImpl.class);
+
+    // THEN
+    Assert.assertNotNull("The cleanupService instance should be present in the IoC context", cleanupService);
+    //Assert.assertFalse("The cleanup registry shouldn't be empty", cleanupService.showCleanupRegistry().isEmpty());
+  }
+
+  @Test
+  public void testRunCleanup() throws Exception {
+    // GIVEN
+    CleanupService cleanupService = injector.getInstance(CleanupServiceImpl.class);
+    TimeBasedCleanupPolicy cleanupPolicy = new TimeBasedCleanupPolicy("cluster-1", 1455891250758L);
+
+    // WHEN
+    cleanupService.cleanup(cleanupPolicy);
+
+    // THEN
+    // todo assert eg.:on the amount of deleted rows
+
+  }
+
+  @Test
+  public void testServicesShouldBeInSingletonScope() throws Exception {
+    // GIVEN
+    // the cleanup guice context is build
+
+    // WHEN
+    CleanupService cleanupService1 = injector.getInstance(CleanupServiceImpl.class);
+    CleanupService cleanupService2 = injector.getInstance(CleanupServiceImpl.class);
+
+    // THEN
+    Assert.assertEquals("The ChainedCleanupService is not in Singleton scope!", cleanupService1, cleanupService2);
+    //Assert.assertEquals("Registered services are not is not in Singleton scope!", cleanupService1.showCleanupRegistry(), cleanupService2.showCleanupRegistry());
+
+  }
+
+}

ambari-server/src/test/java/org/apache/ambari/server/cleanup/CleanupServiceImplTest.java

@@ -0,0 +1,82 @@
+/**
+ * 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.ambari.server.cleanup;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.ambari.server.orm.dao.Cleanable;
+import org.easymock.Capture;
+import org.easymock.EasyMockRule;
+import org.easymock.Mock;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import junit.framework.Assert;
+
+import static org.easymock.EasyMock.capture;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.newCapture;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.reset;
+
+
+public class CleanupServiceImplTest {
+
+  private static final String CLUSTER_NAME = "cluster-1";
+  private static final Long FROM_DATE_TIMESTAMP = 10L;
+
+  @Rule
+  public EasyMockRule mocks = new EasyMockRule(this);
+
+  @Mock
+  private Cleanable cleanableDao;
+
+  private CleanupServiceImpl cleanupServiceImpl;
+  private TimeBasedCleanupPolicy cleanupPolicy;
+  private Capture<TimeBasedCleanupPolicy> timeBasedCleanupPolicyCapture;
+  private Set cleanables;
+
+  @Before
+  public void setUp() throws Exception {
+    reset(cleanableDao);
+    timeBasedCleanupPolicyCapture = newCapture();
+    cleanupPolicy = new TimeBasedCleanupPolicy(CLUSTER_NAME, FROM_DATE_TIMESTAMP);
+  }
+
+  @Test
+  public void testShouldDaosBeCalledWithTheCleanupPolicy() throws Exception {
+    // GIVEN
+    cleanables = new HashSet<>();
+    cleanables.add(cleanableDao);
+    expect(cleanableDao.cleanup(capture(timeBasedCleanupPolicyCapture))).andReturn(2L);
+
+    replay(cleanableDao);
+    cleanupServiceImpl = new CleanupServiceImpl(cleanables);
+
+    // WHEN
+    long rows = cleanupServiceImpl.cleanup(cleanupPolicy);
+
+    // THEN
+    Assert.assertNotNull("The argument is null", timeBasedCleanupPolicyCapture.getValue());
+    Assert.assertEquals("The cluster name is wrong!", timeBasedCleanupPolicyCapture.getValue().getClusterName(), CLUSTER_NAME);
+    Assert.assertEquals("The to date is wrong!", timeBasedCleanupPolicyCapture.getValue().getToDateInMillis(), FROM_DATE_TIMESTAMP);
+  }
+
+}

ambari-server/src/test/resources/ddl-func-test/ddl-cleanup-test-data.sql

@@ -0,0 +1,6 @@
+INSERT INTO stack (stack_id, stack_name, stack_version) VALUES (1, 'HDP', '2.4');
+INSERT INTO adminresourcetype (resource_type_id, resource_type_name) VALUES (1, 'AMBARI');
+INSERT INTO adminresource (resource_id, resource_type_id) VALUES (1, 1);
+INSERT INTO clusters (cluster_id, resource_id, cluster_info, cluster_name, provisioning_state, security_type, desired_cluster_state, desired_stack_id) VALUES (1, 1, '', 'downscaletest', 'INSTALLED', 'NONE', '', 1);
+INSERT INTO alert_definition (definition_id, cluster_id, definition_name, service_name, component_name, scope, label, description, enabled, schedule_interval, source_type, alert_source, hash, ignore_host) VALUES (1, 1, 'yarn_resourcemanager_rpc_latency', 'YARN', 'RESOURCEMANAGER', 'ANY', 'ResourceManager RPC Latency', 'This host-level alert is triggered if the ResourceManager operations RPC latency exceeds the configured critical threshold. Typically an increase in the RPC processing time increases the RPC queue length, causing the average queue wait time to increase for ResourceManager operations. The threshold values are in milliseconds.', 1, 5, 'METRIC', '{"uri":{"http":"{{yarn-site/yarn.resourcemanager.webapp.address}}","https":"{{yarn-site/yarn.resourcemanager.webapp.https.address}}","https_property":"{{yarn-site/yarn.http.policy}}","https_property_value":"HTTPS_ONLY","kerberos_keytab":"{{yarn-site/yarn.resourcemanager.webapp.spnego-keytab-file}}","kerberos_principal":"{{yarn-site/yarn.resourcemanager.webapp.spnego-principal}}","default_port":0,"connection_timeout":5.0,"high_availability":{"alias_key":"{{yarn-site/yarn.resourcemanager.ha.rm-ids}}","http_pattern":"{{yarn-site/yarn.resourcemanager.webapp.address.{{alias}}}}","https_pattern":"{{yarn-site/yarn.resourcemanager.webapp.https.address.{{alias}}}}"}},"jmx":{"property_list":["Hadoop:service\u003dResourceManager,name\u003dRpcActivityForPort*/RpcQueueTimeAvgTime","Hadoop:service\u003dResourceManager,name\u003dRpcActivityForPort*/RpcProcessingTimeAvgTime"],"value":"{0}"},"type":"METRIC","reporting":{"ok":{"text":"Average Queue Time:[{0}], Average Processing Time:[{1}]"},"warning":{"text":"Average Queue Time:[{0}], Average Processing Time:[{1}]","value":3000.0},"critical":{"text":"Average Queue Time:[{0}], Average Processing Time:[{1}]","value":5000.0},"units":"ms"}}', '1babf240-2131-40f6-81f5-65067bc82c14', 0);
+INSERT INTO alert_history (alert_id, cluster_id, alert_definition_id, service_name, component_name, host_name, alert_instance, alert_timestamp, alert_label, alert_state, alert_text) VALUES (1, 1, 1, 'AMBARI', 'AMBARI_AGENT', 'host-10-0-1-13.node.dc1.consul', NULL, 1455885271767, 'Host Disk Usage', 'OK', 'Capacity Used: [10.87%, 4.7 GB], Capacity Total: [42.9 GB], path=/usr/hdp');