package com.linkedin.helix.healthcheck;

import java.util.Random;
import java.util.Timer;

import org.apache.log4j.Logger;

import com.linkedin.helix.HelixManager;
import com.linkedin.helix.HelixTimerTask;
import com.linkedin.helix.controller.pipeline.Pipeline;
import com.linkedin.helix.controller.stages.ClusterEvent;
import com.linkedin.helix.controller.stages.ReadHealthDataStage;
import com.linkedin.helix.controller.stages.StatsAggregationStage;

public class HealthAggregationTask extends HelixTimerTask
{
  private static final Logger LOG = Logger.getLogger(HealthAggregationTask.class);

  public final static int DEFAULT_HEALTH_CHECK_LATENCY = 30 * 1000;

  private Timer _timer;
  private final HelixManager _manager;
  private final Pipeline _healthStatsAggregationPipeline;
  private final int _delay;
  private final int _period;

  public HealthAggregationTask(HelixManager manager, int delay, int period)
  {
    _manager = manager;
    _delay = delay;
    _period = period;

    // health stats pipeline
    _healthStatsAggregationPipeline = new Pipeline();
    _healthStatsAggregationPipeline.addStage(new ReadHealthDataStage());
    _healthStatsAggregationPipeline.addStage(new StatsAggregationStage());

  }

  public HealthAggregationTask(HelixManager manager)
  {
    this(manager, DEFAULT_HEALTH_CHECK_LATENCY, DEFAULT_HEALTH_CHECK_LATENCY);
  }

  @Override
  public void start()
  {
    LOG.info("START HealthAggregationTask");

    if (_timer == null)
    {
      _timer = new Timer();
      _timer.scheduleAtFixedRate(this, new Random().nextInt(_delay), _period);
    }
    else
    {
      LOG.warn("timer already started");
    }
  }

  @Override
  public void stop()
  {
    LOG.info("Stop HealthAggregationTask");

    if (_timer != null)
    {
      _timer.cancel();
      _timer = null;
    }
    else
    {
      LOG.warn("timer already stopped");
    }
  }

  @Override
  public void run()
  {
    LOG.info("START: HealthAggregationTask");

    if (!_manager.isLeader())
    {
      LOG.error("Cluster manager: " + _manager.getInstanceName()
          + " is not leader. Pipeline will not be invoked");
      return;
    }

    try
    {
      ClusterEvent event = new ClusterEvent("healthChange");
      event.addAttribute("helixmanager", _manager);

      _healthStatsAggregationPipeline.handle(event);
      _healthStatsAggregationPipeline.finish();
    }
    catch (Exception e)
    {
      LOG.error("Exception while executing pipeline: " + _healthStatsAggregationPipeline,
                e);
    }

    LOG.info("END: HealthAggregationTask");
  }

}

@@ -113,24 +113,151 @@
 4j.Logger;%0A%0A
+import com.linkedin.helix.ConfigAccessor;%0Aimport com.linkedin.helix.ConfigScope;%0Aimport com.linkedin.helix.ConfigScopeBuilder;%0A
 import com.l
@@ -280,24 +280,24 @@
 lixManager;%0A
-
 import com.l
@@ -1524,24 +1524,152 @@
 start()%0A  %7B%0A
+    if (!isEnabled())%0A    %7B%0A      LOG.info(%22HealthAggregationTask is disabled. Will not start the timer%22);%0A      return;%0A    %7D%0A%0A
     LOG.info
@@ -2781,32 +2781,32 @@
      e);%0A    %7D%0A%0A
-
     LOG.info(%22EN
@@ -2838,10 +2838,618 @@
 );%0A  %7D%0A%0A
+  private boolean isEnabled()%0A  %7B%0A    ConfigAccessor configAccessor = _manager.getConfigAccessor();%0A    boolean enabled = false;%0A    if (configAccessor != null)%0A    %7B%0A      // zk-based cluster manager%0A      ConfigScope scope =%0A          new ConfigScopeBuilder().forCluster(_manager.getClusterName()).build();%0A      String isEnabled = configAccessor.get(scope, %22healthChange.enabled%22);%0A      if (isEnabled != null)%0A      %7B%0A        enabled = new Boolean(isEnabled);%0A      %7D%0A    %7D%0A    else%0A    %7B%0A      LOG.debug(%22File-based cluster manager doesn't support disable healthChange%22);%0A    %7D%0A    return enabled;%0A  %7D%0A%0A
 %7D%0A

disable health check timer by default