package org.apache.helix.monitoring;

/*
 * 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.
 */

import java.io.StringWriter;
import java.util.Date;
import java.util.List;
import java.util.TimerTask;

import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.PropertyType;
import org.apache.helix.ZNRecord;
import org.apache.helix.manager.zk.ZNRecordSerializer;
import org.apache.helix.manager.zk.ZkClient;
import org.apache.helix.util.HelixUtil;
import org.apache.log4j.Logger;
import org.apache.zookeeper.data.Stat;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.SerializationConfig;

public class ZKPathDataDumpTask extends TimerTask {
  static Logger logger = Logger.getLogger(ZKPathDataDumpTask.class);

  private final int _thresholdNoChangeInMs;
  private final HelixManager _manager;
  private final ZkClient _zkClient;

  public ZKPathDataDumpTask(HelixManager manager, ZkClient zkClient, int thresholdNoChangeInMs) {
    _manager = manager;
    _zkClient = zkClient;
    logger.info("Scannning cluster statusUpdate " + manager.getClusterName()
        + " thresholdNoChangeInMs: " + thresholdNoChangeInMs);
    _thresholdNoChangeInMs = thresholdNoChangeInMs;
  }

  @Override
  public void run() {
    // For each record in status update and error node
    // TODO: for now the status updates are dumped to cluster manager log4j log.
    // We need to think if we should create per-instance log files that contains
    // per-instance statusUpdates
    // and errors
    logger.info("Scannning status updates ...");
    try {
      HelixDataAccessor accessor = _manager.getHelixDataAccessor();
      Builder keyBuilder = accessor.keyBuilder();

      List<String> instances = accessor.getChildNames(keyBuilder.instanceConfigs());
      for (String instanceName : instances) {
        scanPath(HelixUtil.getInstancePropertyPath(_manager.getClusterName(), instanceName,
            PropertyType.STATUSUPDATES), _thresholdNoChangeInMs);
        scanPath(HelixUtil.getInstancePropertyPath(_manager.getClusterName(), instanceName,
            PropertyType.ERRORS), _thresholdNoChangeInMs * 3);
      }
      scanPath(HelixUtil.getControllerPropertyPath(_manager.getClusterName(),
          PropertyType.STATUSUPDATES_CONTROLLER), _thresholdNoChangeInMs);

      scanPath(HelixUtil.getControllerPropertyPath(_manager.getClusterName(),
          PropertyType.ERRORS_CONTROLLER), _thresholdNoChangeInMs * 3);
    } catch (Exception e) {
      logger.error(e);
    }
  }

  void scanPath(String path, int thresholdNoChangeInMs) {
    logger.info("Scannning path " + path);
    List<String> subPaths = _zkClient.getChildren(path);
    for (String subPath : subPaths) {
      try {
        String nextPath = path + "/" + subPath;
        List<String> subSubPaths = _zkClient.getChildren(nextPath);
        for (String subsubPath : subSubPaths) {
          try {
            checkAndDump(nextPath + "/" + subsubPath, thresholdNoChangeInMs);
          } catch (Exception e) {
            logger.error(e);
          }
        }
      } catch (Exception e) {
        logger.error(e);
      }
    }
  }

  void checkAndDump(String path, int thresholdNoChangeInMs) {
    List<String> subPaths = _zkClient.getChildren(path);
    if (subPaths.size() == 0) {
      subPaths.add("");
    }
    for (String subPath : subPaths) {
      String fullPath = subPath.length() > 0 ? path + "/" + subPath : path;
      Stat pathStat = _zkClient.getStat(fullPath);

      long lastModifiedTimeInMs = pathStat.getMtime();
      long nowInMs = new Date().getTime();
      // logger.info(nowInMs + " " + lastModifiedTimeInMs + " " + fullPath);

      // Check the last modified time
      if (nowInMs > lastModifiedTimeInMs) {
        long timeDiff = nowInMs - lastModifiedTimeInMs;
        if (timeDiff > thresholdNoChangeInMs) {
          logger.info("Dumping status update path " + fullPath + " " + timeDiff + "MS has passed");
          _zkClient.setZkSerializer(new ZNRecordSerializer());
          ZNRecord record = _zkClient.readData(fullPath);

          // dump the node content into log file
          ObjectMapper mapper = new ObjectMapper();
          SerializationConfig serializationConfig = mapper.getSerializationConfig();
          serializationConfig.set(SerializationConfig.Feature.INDENT_OUTPUT, true);

          StringWriter sw = new StringWriter();
          try {
            mapper.writeValue(sw, record);
            logger.info(sw.toString());
          } catch (Exception e) {
            logger
                .warn(
                    "Exception during serialization in ZKPathDataDumpTask.checkAndDump. This can mostly be ignored",
                    e);
          }
          // Delete the leaf data
          _zkClient.deleteRecursive(fullPath);
        }
      }
    }
  }
}

@@ -1861,25 +1861,24 @@
 .info(%22Scann
-n
 ing cluster 
@@ -2360,25 +2360,24 @@
 .info(%22Scann
-n
 ing status u
@@ -3417,17 +3417,16 @@
 o(%22Scann
-n
 ing path

fix typo