/**
 * Copyright (C) 2012 LinkedIn Inc <opensource@linkedin.com>
 *
 * Licensed 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 com.linkedin.helix.healthcheck;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.Timer;

import org.apache.log4j.Logger;

import com.linkedin.helix.ConfigAccessor;
import com.linkedin.helix.ConfigScope;
import com.linkedin.helix.ConfigScopeBuilder;
import com.linkedin.helix.HelixManager;
import com.linkedin.helix.HelixTimerTask;
import com.linkedin.helix.controller.pipeline.Pipeline;
import com.linkedin.helix.controller.pipeline.Stage;
import com.linkedin.helix.controller.stages.ClusterEvent;
import com.linkedin.helix.controller.stages.ReadHealthDataStage;
import com.linkedin.helix.controller.stages.StatsAggregationStage;
import com.linkedin.helix.monitoring.mbeans.ClusterAlertMBeanCollection;
import com.linkedin.helix.monitoring.mbeans.HelixStageLatencyMonitor;

public class HealthStatsAggregationTask extends HelixTimerTask
{
  private static final Logger LOG = Logger.getLogger(HealthStatsAggregationTask.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;
  private final ClusterAlertMBeanCollection _alertItemCollection;
  private final Map<String, HelixStageLatencyMonitor> _stageLatencyMonitorMap =
      new HashMap<String, HelixStageLatencyMonitor>();

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

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

    registerStageLatencyMonitor(_healthStatsAggregationPipeline);
  }

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

  private void registerStageLatencyMonitor(Pipeline pipeline)
  {
    for (Stage stage : pipeline.getStages())
    {
      String stgName = stage.getStageName();
      if (!_stageLatencyMonitorMap.containsKey(stgName))
      {
        try
        {
          _stageLatencyMonitorMap.put(stage.getStageName(),
                                      new HelixStageLatencyMonitor(_manager.getClusterName(),
                                                                   stgName));
        }
        catch (Exception e)
        {
          LOG.error("Couldn't create StageLatencyMonitor mbean for stage: " + stgName, e);
        }
      }
      else
      {
        LOG.error("StageLatencyMonitor for stage: " + stgName
            + " already exists. Skip register it");
      }
    }
  }

  @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 synchronized void stop()
  {
    LOG.info("Stop HealthAggregationTask");

    if (_timer != null)
    {
      _timer.cancel();
      _timer = null;
      _alertItemCollection.reset();

      for (HelixStageLatencyMonitor stgLatencyMonitor : _stageLatencyMonitorMap.values())
      {
        stgLatencyMonitor.reset();
      }
    }
    else
    {
      LOG.warn("timer already stopped");
    }
  }

  @Override
  public synchronized void run()
  {
    if (!isEnabled())
    {
      LOG.info("HealthAggregationTask is disabled.");
      return;
    }
    
    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);
      event.addAttribute("HelixStageLatencyMonitorMap", _stageLatencyMonitorMap);

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

  private boolean isEnabled()
  {
    ConfigAccessor configAccessor = _manager.getConfigAccessor();
    boolean enabled = true;
    if (configAccessor != null)
    {
      // zk-based cluster manager
      ConfigScope scope =
          new ConfigScopeBuilder().forCluster(_manager.getClusterName()).build();
      String isEnabled = configAccessor.get(scope, "healthChange.enabled");
      if (isEnabled != null)
      {
        enabled = new Boolean(isEnabled);
      }
    }
    else
    {
      LOG.debug("File-based cluster manager doesn't support disable healthChange");
    }
    return enabled;
  }

}

@@ -687,24 +687,47 @@
 il.HashMap;%0A
+import java.util.List;%0A
 import java.
@@ -942,24 +942,69 @@
 opeBuilder;%0A
+import com.linkedin.helix.HelixDataAccessor;%0A
 import com.l
@@ -3769,32 +3769,564 @@
  == null)%0A    %7B%0A
+      // Remove all the previous health check values, if any%0A      HelixDataAccessor accessor = _manager.getHelixDataAccessor();%0A      List%3CString%3E existingHealthRecordNames = accessor.getChildNames(accessor.keyBuilder().healthReports(_manager.getInstanceName()));%0A      for(String healthReportName : existingHealthRecordNames)%0A      %7B%0A        LOG.info(%22Removing old healthrecord %22 + healthReportName);%0A        accessor.removeProperty(accessor.keyBuilder().healthReport(_manager.getInstanceName(),healthReportName));%0A      %7D%0A      %0A
       _timer = n

Remove old healthcheck records when a participant starts up