<commit_before>/*
 * 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.zeppelin.cluster;

import com.sun.management.OperatingSystemMXBean;
import org.apache.zeppelin.cluster.meta.ClusterMeta;
import org.apache.zeppelin.cluster.meta.ClusterMetaType;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.management.ManagementFactory;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.apache.zeppelin.cluster.meta.ClusterMetaType.INTP_PROCESS_META;
import static org.apache.zeppelin.cluster.meta.ClusterMetaType.SERVER_META;

/**
 * cluster monitoring
 * 1. cluster monitoring is also used for zeppelin-Server and zeppelin Interperter,
 *    distinguish by member variable ClusterMetaType
 * 2. Report the average of the server resource CPU and MEMORY usage in the
 *    last few minutes to smooth the server's instantaneous peak
 * 3. checks the heartbeat timeout of the zeppelin-server and interperter processes
 */
public class ClusterMonitor {
  private static Logger LOGGER = LoggerFactory.getLogger(ClusterMonitor.class);

  // Whether the thread has started
  private static AtomicBoolean running = new AtomicBoolean(true);

  private ClusterManager clusterManager = null;

  // Save the CPU resource and MEmory usage of the server resources
  // in the last few minutes through the queue, and then average them through the queue.
  private Queue<UsageUtil> monitorUsageQueues = new LinkedList<>();
  private final int USAGE_QUEUE_LIMIT = 100; // queue length
  private int heartbeatInterval = 3000; // Heartbeat reporting interval（milliseconds）

  // The zeppelin-server leader checks the heartbeat timeout of
  // the zeppelin-server and zeppelin-interperter processes in the cluster metadata.
  // If this time is exceeded, the zeppelin-server and interperter processes
  // can have an exception and no heartbeat is reported.
  private int heartbeatTimeout = 9000;

  // Type of cluster monitoring object
  private ClusterMetaType clusterMetaType;

  // The key of the cluster monitoring object,
  // the name of the cluster when monitoring the zeppelin-server,
  // and the interperterGroupID when monitoring the interperter processes
  private String metaKey;

  public ClusterMonitor(ClusterManager clusterManagerServer) {
    this.clusterManager = clusterManagerServer;

    ZeppelinConfiguration zconf = new ZeppelinConfiguration();
    heartbeatInterval = zconf.getClusterHeartbeatInterval();
    heartbeatTimeout = zconf.getClusterHeartbeatTimeout();

    if (heartbeatTimeout < heartbeatInterval) {
      LOGGER.error("Heartbeat timeout must be greater than heartbeat period.");
      heartbeatTimeout = heartbeatInterval * 3;
      LOGGER.info("Heartbeat timeout is modified to 3 times the heartbeat period.");
    }

    if (heartbeatTimeout < heartbeatInterval * 3) {
      LOGGER.warn("Heartbeat timeout recommended than 3 times the heartbeat period.");
    }
  }

  //
  public void start(ClusterMetaType clusterMetaType, String metaKey) {
    this.clusterMetaType = clusterMetaType;
    this.metaKey = metaKey;

    new Thread(new Runnable() {
      @Override
      public void run() {
        while (running.get()) {
          switch (clusterMetaType) {
            case SERVER_META:
              sendMachineUsage();
              checkHealthy();
              break;
            case INTP_PROCESS_META:
              sendHeartbeat();
              break;
            default:
              LOGGER.error("unknown cluster meta type:{}", clusterMetaType);
              break;
          }

          try {
            Thread.sleep(heartbeatInterval);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
      }
    }).start();
  }

  public void shutdown() {
    running.set(false);
  }

  // Check the healthy of each service and interperter instance
  private void checkHealthy() {
    // only leader check cluster healthy
    if (!clusterManager.isClusterLeader()) {
      return;
    }

    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug("checkHealthy()");
    }

    LocalDateTime now = LocalDateTime.now();
    // check machine mate
    for (ClusterMetaType metaType : ClusterMetaType.values()) {
      Map<String, HashMap<String, Object>> clusterMeta
          = clusterManager.getClusterMeta(metaType, "");

      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("clusterMeta : {}", clusterMeta);
      }

      for (Map.Entry<String, HashMap<String, Object>> entry : clusterMeta.entrySet()) {
        String key = entry.getKey();
        Map<String, Object> meta = entry.getValue();

        // Metadata that has been offline is not processed
        String status = (String) meta.get(ClusterMeta.STATUS);
        if (status.equals(ClusterMeta.OFFLINE_STATUS)) {
          continue;
        }

        Object heartbeat = meta.get(ClusterMeta.LATEST_HEARTBEAT);
        if (heartbeat instanceof LocalDateTime) {
          LocalDateTime dHeartbeat = (LocalDateTime) heartbeat;
          Duration duration = Duration.between(dHeartbeat, now);
          long timeInterval = duration.getSeconds();<commit_msg>[ZEPPELIN-4284] Fix cluster metadata interpreter offline timeout calculation error<commit_after>/*
 * 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.zeppelin.cluster;

import com.sun.management.OperatingSystemMXBean;
import org.apache.zeppelin.cluster.meta.ClusterMeta;
import org.apache.zeppelin.cluster.meta.ClusterMetaType;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.management.ManagementFactory;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.apache.zeppelin.cluster.meta.ClusterMetaType.INTP_PROCESS_META;
import static org.apache.zeppelin.cluster.meta.ClusterMetaType.SERVER_META;

/**
 * cluster monitoring
 * 1. cluster monitoring is also used for zeppelin-Server and zeppelin Interperter,
 *    distinguish by member variable ClusterMetaType
 * 2. Report the average of the server resource CPU and MEMORY usage in the
 *    last few minutes to smooth the server's instantaneous peak
 * 3. checks the heartbeat timeout of the zeppelin-server and interperter processes
 */
public class ClusterMonitor {
  private static Logger LOGGER = LoggerFactory.getLogger(ClusterMonitor.class);

  // Whether the thread has started
  private static AtomicBoolean running = new AtomicBoolean(true);

  private ClusterManager clusterManager = null;

  // Save the CPU resource and MEmory usage of the server resources
  // in the last few minutes through the queue, and then average them through the queue.
  private Queue<UsageUtil> monitorUsageQueues = new LinkedList<>();
  private final int USAGE_QUEUE_LIMIT = 100; // queue length
  private int heartbeatInterval = 3000; // Heartbeat reporting interval（milliseconds）

  // The zeppelin-server leader checks the heartbeat timeout of
  // the zeppelin-server and zeppelin-interperter processes in the cluster metadata.
  // If this time is exceeded, the zeppelin-server and interperter processes
  // can have an exception and no heartbeat is reported.
  private int heartbeatTimeout = 9000;

  // Type of cluster monitoring object
  private ClusterMetaType clusterMetaType;

  // The key of the cluster monitoring object,
  // the name of the cluster when monitoring the zeppelin-server,
  // and the interperterGroupID when monitoring the interperter processes
  private String metaKey;

  public ClusterMonitor(ClusterManager clusterManagerServer) {
    this.clusterManager = clusterManagerServer;

    ZeppelinConfiguration zconf = new ZeppelinConfiguration();
    heartbeatInterval = zconf.getClusterHeartbeatInterval();
    heartbeatTimeout = zconf.getClusterHeartbeatTimeout();

    if (heartbeatTimeout < heartbeatInterval) {
      LOGGER.error("Heartbeat timeout must be greater than heartbeat period.");
      heartbeatTimeout = heartbeatInterval * 3;
      LOGGER.info("Heartbeat timeout is modified to 3 times the heartbeat period.");
    }

    if (heartbeatTimeout < heartbeatInterval * 3) {
      LOGGER.warn("Heartbeat timeout recommended than 3 times the heartbeat period.");
    }
  }

  //
  public void start(ClusterMetaType clusterMetaType, String metaKey) {
    this.clusterMetaType = clusterMetaType;
    this.metaKey = metaKey;

    new Thread(new Runnable() {
      @Override
      public void run() {
        while (running.get()) {
          switch (clusterMetaType) {
            case SERVER_META:
              sendMachineUsage();
              checkHealthy();
              break;
            case INTP_PROCESS_META:
              sendHeartbeat();
              break;
            default:
              LOGGER.error("unknown cluster meta type:{}", clusterMetaType);
              break;
          }

          try {
            Thread.sleep(heartbeatInterval);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
      }
    }).start();
  }

  public void shutdown() {
    running.set(false);
  }

  // Check the healthy of each service and interperter instance
  private void checkHealthy() {
    // only leader check cluster healthy
    if (!clusterManager.isClusterLeader()) {
      return;
    }

    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug("checkHealthy()");
    }

    LocalDateTime now = LocalDateTime.now();
    // check machine mate
    for (ClusterMetaType metaType : ClusterMetaType.values()) {
      Map<String, HashMap<String, Object>> clusterMeta
          = clusterManager.getClusterMeta(metaType, "");

      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("clusterMeta : {}", clusterMeta);
      }

      for (Map.Entry<String, HashMap<String, Object>> entry : clusterMeta.entrySet()) {
        String key = entry.getKey();
        Map<String, Object> meta = entry.getValue();

        // Metadata that has been offline is not processed
        String status = (String) meta.get(ClusterMeta.STATUS);
        if (status.equals(ClusterMeta.OFFLINE_STATUS)) {
          continue;
        }

        Object heartbeat = meta.get(ClusterMeta.LATEST_HEARTBEAT);
        if (heartbeat instanceof LocalDateTime) {
          LocalDateTime dHeartbeat = (LocalDateTime) heartbeat;
          Duration duration = Duration.between(dHeartbeat, now);
          long timeInterval = duration.getSeconds() * 1000; // Convert to milliseconds
{'fraction_non_alphanumeric': 0.08076322065959371, 'fraction_numerical': 0.004605641911341394, 'mean_word_length': 3.2877291960507757, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1894568', 'n_tokens_mistral': 3225, 'n_tokens_neox': 3088, 'n_words': 1059}