<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.accumulo.monitor;

import static java.nio.charset.StandardCharsets.UTF_8;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.MasterClient;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.gc.thrift.GCMonitorService;
import org.apache.accumulo.core.gc.thrift.GCStatus;
import org.apache.accumulo.core.master.thrift.MasterClientService;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.master.thrift.TableInfo;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.ActiveScan;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService.Client;
import org.apache.accumulo.core.trace.DistributedTrace;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.util.Daemon;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.ServerServices;
import org.apache.accumulo.core.util.ServerServices.Service;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.util.LoggingRunnable;
import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy;
import org.apache.accumulo.monitor.servlets.DefaultServlet;
import org.apache.accumulo.monitor.servlets.GcStatusServlet;
import org.apache.accumulo.monitor.servlets.JSONServlet;
import org.apache.accumulo.monitor.servlets.LogServlet;
import org.apache.accumulo.monitor.servlets.MasterServlet;
import org.apache.accumulo.monitor.servlets.OperationServlet;
import org.apache.accumulo.monitor.servlets.ProblemServlet;
import org.apache.accumulo.monitor.servlets.ReplicationServlet;
import org.apache.accumulo.monitor.servlets.ScanServlet;
import org.apache.accumulo.monitor.servlets.ShellServlet;
import org.apache.accumulo.monitor.servlets.TServersServlet;
import org.apache.accumulo.monitor.servlets.TablesServlet;
import org.apache.accumulo.monitor.servlets.VisServlet;
import org.apache.accumulo.monitor.servlets.XMLServlet;
import org.apache.accumulo.monitor.servlets.trace.ListType;
import org.apache.accumulo.monitor.servlets.trace.ShowTrace;
import org.apache.accumulo.monitor.servlets.trace.Summary;
import org.apache.accumulo.server.Accumulo;
import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.accumulo.server.ServerOpts;
import org.apache.accumulo.server.client.HdfsZooInstance;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.monitor.LogService;
import org.apache.accumulo.server.problems.ProblemReports;
import org.apache.accumulo.server.problems.ProblemType;
import org.apache.accumulo.server.security.SecurityUtil;
import org.apache.accumulo.server.util.Halt;
import org.apache.accumulo.server.util.TableInfoUtil;
import org.apache.accumulo.server.util.time.SimpleTimer;
import org.apache.accumulo.server.zookeeper.ZooLock;
import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.net.HostAndPort;

/**
 * Serve master statistics with an embedded web server.
 */
public class Monitor {
  private static final Logger log = LoggerFactory.getLogger(Monitor.class);

  private static final int REFRESH_TIME = 5;
  private static long lastRecalc = 0L;
  private static double totalIngestRate = 0.0;
  private static double totalQueryRate = 0.0;
  private static double totalScanRate = 0.0;
  private static long totalEntries = 0L;
  private static int totalTabletCount = 0;
  private static long totalHoldTime = 0;
  private static long totalLookups = 0;
  private static int totalTables = 0;

  private static class MaxList<T> extends LinkedList<Pair<Long,T>> {
    private static final long serialVersionUID = 1L;

    private long maxDelta;

    public MaxList(long maxDelta) {
      this.maxDelta = maxDelta;
    }

    @Override
    public boolean add(Pair<Long,T> obj) {
      boolean result = super.add(obj);

      if (obj.getFirst() - get(0).getFirst() > maxDelta)
        remove(0);

      return result;
    }

  }

  private static final int MAX_TIME_PERIOD = 60 * 60 * 1000;
  private static final List<Pair<Long,Double>> loadOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> ingestRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> ingestByteRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Integer>> minorCompactionsOverTime = Collections.synchronizedList(new MaxList<Integer>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Integer>> majorCompactionsOverTime = Collections.synchronizedList(new MaxList<Integer>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> lookupsOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Integer>> queryRateOverTime = Collections.synchronizedList(new MaxList<Integer>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Integer>> scanRateOverTime = Collections.synchronizedList(new MaxList<Integer>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> queryByteRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> indexCacheHitRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> dataCacheHitRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static EventCounter lookupRateTracker = new EventCounter();
  private static EventCounter indexCacheHitTracker = new EventCounter();
  private static EventCounter indexCacheRequestTracker = new EventCounter();
  private static EventCounter dataCacheHitTracker = new EventCounter();
  private static EventCounter dataCacheRequestTracker = new EventCounter();

  private static volatile boolean fetching = false;
  private static MasterMonitorInfo mmi;
  private static Map<String,Map<ProblemType,Integer>> problemSummary = Collections.emptyMap();
  private static Exception problemException;
  private static GCStatus gcStatus;

  private static Instance instance;

  private static ServerConfigurationFactory config;
  private static AccumuloServerContext context;

  private static EmbeddedWebServer server;

  private ZooLock monitorLock;

  private static final String DEFAULT_INSTANCE_NAME = "(Unavailable)";
  public static final AtomicReference<String> cachedInstanceName = new AtomicReference<String>(DEFAULT_INSTANCE_NAME);

  private static class EventCounter {

    Map<String,Pair<Long,Long>> prevSamples = new HashMap<String,Pair<Long,Long>>();
    Map<String,Pair<Long,Long>> samples = new HashMap<String,Pair<Long,Long>>();
    Set<String> serversUpdated = new HashSet<String>();

    void startingUpdates() {
      serversUpdated.clear();
    }

    void updateTabletServer(String name, long sampleTime, long numEvents) {
      Pair<Long,Long> newSample = new Pair<Long,Long>(sampleTime, numEvents);
      Pair<Long,Long> lastSample = samples.get(name);

      if (lastSample == null || !lastSample.equals(newSample)) {
        samples.put(name, newSample);
        if (lastSample != null) {
          prevSamples.put(name, lastSample);
        }
      }
      serversUpdated.add(name);
    }

    void finishedUpdating() {
      // remove any tablet servers not updated
      samples.keySet().retainAll(serversUpdated);
      prevSamples.keySet().retainAll(serversUpdated);
    }

    double calculateRate() {
      double totalRate = 0;

      for (Entry<String,Pair<Long,Long>> entry : prevSamples.entrySet()) {
        Pair<Long,Long> prevSample = entry.getValue();
        Pair<Long,Long> sample = samples.get(entry.getKey());

        totalRate += (sample.getSecond() - prevSample.getSecond()) / ((sample.getFirst() - prevSample.getFirst()) / (double) 1000);
      }

      return totalRate;
    }

    long calculateCount() {
      long count = 0;

      for (Entry<String,Pair<Long,Long>> entry : prevSamples.entrySet()) {
        Pair<Long,Long> prevSample = entry.getValue();
        Pair<Long,Long> sample = samples.get(entry.getKey());

        count += sample.getSecond() - prevSample.getSecond();
      }

      return count;
    }
  }

  public static void fetchData() {
    double totalIngestRate = 0.;
    double totalIngestByteRate = 0.;
    double totalQueryRate = 0.;
    double totalQueryByteRate = 0.;
    double totalScanRate = 0.;
    long totalEntries = 0;
    int totalTabletCount = 0;
    long totalHoldTime = 0;
    long totalLookups = 0;
    boolean retry = true;

    // only recalc every so often
    long currentTime = System.currentTimeMillis();
    if (currentTime - lastRecalc < REFRESH_TIME * 1000)
      return;

    synchronized (Monitor.class) {
      // Learn our instance name asynchronously so we don't hang up if zookeeper is down
      if (cachedInstanceName.get().equals(DEFAULT_INSTANCE_NAME)) {
        SimpleTimer.getInstance(config.getConfiguration()).schedule(new TimerTask() {
          @Override
          public void run() {
            synchronized (Monitor.class) {
              if (cachedInstanceName.get().equals(DEFAULT_INSTANCE_NAME)) {
                cachedInstanceName.set(HdfsZooInstance.getInstance().getInstanceName());
              }
            }
          }
        }, 0);
      }
    }

    synchronized (Monitor.class) {
      if (fetching)
        return;
      fetching = true;
    }

    try {
      while (retry) {
        MasterClientService.Iface client = null;
        try {
          client = MasterClient.getConnection(context);
          if (client != null) {
            mmi = client.getMasterStats(Tracer.traceInfo(), context.rpcCreds());
            retry = false;
          } else {
            mmi = null;
          }
          Monitor.gcStatus = fetchGcStatus();
        } catch (Exception e) {
          mmi = null;
          log.info("Error fetching stats: " + e);
        } finally {
          if (client != null) {
            MasterClient.close(client);
          }
        }
        if (mmi == null)
          UtilWaitThread.sleep(1000);
      }
      if (mmi != null) {
        int majorCompactions = 0;
        int minorCompactions = 0;

        lookupRateTracker.startingUpdates();
        indexCacheHitTracker.startingUpdates();
        indexCacheRequestTracker.startingUpdates();
        dataCacheHitTracker.startingUpdates();
        dataCacheRequestTracker.startingUpdates();

        for (TabletServerStatus server : mmi.tServerInfo) {
          TableInfo summary = TableInfoUtil.summarizeTableStats(server);
          totalIngestRate += summary.ingestRate;
          totalIngestByteRate += summary.ingestByteRate;
          totalQueryRate += summary.queryRate;
          totalScanRate += summary.scanRate;
          totalQueryByteRate += summary.queryByteRate;
          totalEntries += summary.recs;
          totalHoldTime += server.holdTime;
          totalLookups += server.lookups;
          majorCompactions += summary.majors.running;
          minorCompactions += summary.minors.running;
          lookupRateTracker.updateTabletServer(server.name, server.lastContact, server.lookups);
          indexCacheHitTracker.updateTabletServer(server.name, server.lastContact, server.indexCacheHits);
          indexCacheRequestTracker.updateTabletServer(server.name, server.lastContact, server.indexCacheRequest);
          dataCacheHitTracker.updateTabletServer(server.name, server.lastContact, server.dataCacheHits);
          dataCacheRequestTracker.updateTabletServer(server.name, server.lastContact, server.dataCacheRequest);
        }

        lookupRateTracker.finishedUpdating();
        indexCacheHitTracker.finishedUpdating();
        indexCacheRequestTracker.finishedUpdating();
        dataCacheHitTracker.finishedUpdating();
        dataCacheRequestTracker.finishedUpdating();

        int totalTables = 0;
        for (TableInfo tInfo : mmi.tableMap.values()) {
          totalTabletCount += tInfo.tablets;
          totalTables++;
        }
        Monitor.totalIngestRate = totalIngestRate;
        Monitor.totalTables = totalTables;
        totalIngestByteRate = totalIngestByteRate / 1000000.0;
        Monitor.totalQueryRate = totalQueryRate;
        Monitor.totalScanRate = totalScanRate;
        totalQueryByteRate = totalQueryByteRate / 1000000.0;
        Monitor.totalEntries = totalEntries;
        Monitor.totalTabletCount = totalTabletCount;
        Monitor.totalHoldTime = totalHoldTime;
        Monitor.totalLookups = totalLookups;

        ingestRateOverTime.add(new Pair<Long,Double>(currentTime, totalIngestRate));
        ingestByteRateOverTime.add(new Pair<Long,Double>(currentTime, totalIngestByteRate));

        double totalLoad = 0.;
        for (TabletServerStatus status : mmi.tServerInfo) {
          if (status != null)
            totalLoad += status.osLoad;
        }
        loadOverTime.add(new Pair<Long,Double>(currentTime, totalLoad));

        minorCompactionsOverTime.add(new Pair<Long,Integer>(currentTime, minorCompactions));
        majorCompactionsOverTime.add(new Pair<Long,Integer>(currentTime, majorCompactions));

        lookupsOverTime.add(new Pair<Long,Double>(currentTime, lookupRateTracker.calculateRate()));

        queryRateOverTime.add(new Pair<Long,Integer>(currentTime, (int) totalQueryRate));
        queryByteRateOverTime.add(new Pair<Long,Double>(currentTime, totalQueryByteRate));

        scanRateOverTime.add(new Pair<Long,Integer>(currentTime, (int) totalScanRate));

        calcCacheHitRate(indexCacheHitRateOverTime, currentTime, indexCacheHitTracker, indexCacheRequestTracker);
        calcCacheHitRate(dataCacheHitRateOverTime, currentTime, dataCacheHitTracker, dataCacheRequestTracker);
      }
      try {
        Monitor.problemSummary = ProblemReports.getInstance(getContext()).summarize();
        Monitor.problemException = null;
      } catch (Exception e) {
        log.info("Failed to obtain problem reports ", e);
        Monitor.problemSummary = Collections.emptyMap();
        Monitor.problemException = e;
      }

    } finally {
      synchronized (Monitor.class) {
        fetching = false;
        lastRecalc = currentTime;<commit_msg>ACCUMULO-3867 Switch concurrently-accessed long to AtomicLong<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.accumulo.monitor;

import static java.nio.charset.StandardCharsets.UTF_8;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.MasterClient;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.gc.thrift.GCMonitorService;
import org.apache.accumulo.core.gc.thrift.GCStatus;
import org.apache.accumulo.core.master.thrift.MasterClientService;
import org.apache.accumulo.core.master.thrift.MasterMonitorInfo;
import org.apache.accumulo.core.master.thrift.TableInfo;
import org.apache.accumulo.core.master.thrift.TabletServerStatus;
import org.apache.accumulo.core.rpc.ThriftUtil;
import org.apache.accumulo.core.tabletserver.thrift.ActiveScan;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService.Client;
import org.apache.accumulo.core.trace.DistributedTrace;
import org.apache.accumulo.core.trace.Tracer;
import org.apache.accumulo.core.util.Daemon;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.ServerServices;
import org.apache.accumulo.core.util.ServerServices.Service;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.zookeeper.ZooUtil;
import org.apache.accumulo.fate.util.LoggingRunnable;
import org.apache.accumulo.fate.zookeeper.ZooLock.LockLossReason;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeExistsPolicy;
import org.apache.accumulo.fate.zookeeper.ZooUtil.NodeMissingPolicy;
import org.apache.accumulo.monitor.servlets.DefaultServlet;
import org.apache.accumulo.monitor.servlets.GcStatusServlet;
import org.apache.accumulo.monitor.servlets.JSONServlet;
import org.apache.accumulo.monitor.servlets.LogServlet;
import org.apache.accumulo.monitor.servlets.MasterServlet;
import org.apache.accumulo.monitor.servlets.OperationServlet;
import org.apache.accumulo.monitor.servlets.ProblemServlet;
import org.apache.accumulo.monitor.servlets.ReplicationServlet;
import org.apache.accumulo.monitor.servlets.ScanServlet;
import org.apache.accumulo.monitor.servlets.ShellServlet;
import org.apache.accumulo.monitor.servlets.TServersServlet;
import org.apache.accumulo.monitor.servlets.TablesServlet;
import org.apache.accumulo.monitor.servlets.VisServlet;
import org.apache.accumulo.monitor.servlets.XMLServlet;
import org.apache.accumulo.monitor.servlets.trace.ListType;
import org.apache.accumulo.monitor.servlets.trace.ShowTrace;
import org.apache.accumulo.monitor.servlets.trace.Summary;
import org.apache.accumulo.server.Accumulo;
import org.apache.accumulo.server.AccumuloServerContext;
import org.apache.accumulo.server.ServerOpts;
import org.apache.accumulo.server.client.HdfsZooInstance;
import org.apache.accumulo.server.conf.ServerConfigurationFactory;
import org.apache.accumulo.server.fs.VolumeManager;
import org.apache.accumulo.server.fs.VolumeManagerImpl;
import org.apache.accumulo.server.monitor.LogService;
import org.apache.accumulo.server.problems.ProblemReports;
import org.apache.accumulo.server.problems.ProblemType;
import org.apache.accumulo.server.security.SecurityUtil;
import org.apache.accumulo.server.util.Halt;
import org.apache.accumulo.server.util.TableInfoUtil;
import org.apache.accumulo.server.util.time.SimpleTimer;
import org.apache.accumulo.server.zookeeper.ZooLock;
import org.apache.accumulo.server.zookeeper.ZooReaderWriter;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.net.HostAndPort;

/**
 * Serve master statistics with an embedded web server.
 */
public class Monitor {
  private static final Logger log = LoggerFactory.getLogger(Monitor.class);

  private static final int REFRESH_TIME = 5;
  private static AtomicLong lastRecalc = new AtomicLong(0L);
  private static double totalIngestRate = 0.0;
  private static double totalQueryRate = 0.0;
  private static double totalScanRate = 0.0;
  private static long totalEntries = 0L;
  private static int totalTabletCount = 0;
  private static long totalHoldTime = 0;
  private static long totalLookups = 0;
  private static int totalTables = 0;

  private static class MaxList<T> extends LinkedList<Pair<Long,T>> {
    private static final long serialVersionUID = 1L;

    private long maxDelta;

    public MaxList(long maxDelta) {
      this.maxDelta = maxDelta;
    }

    @Override
    public boolean add(Pair<Long,T> obj) {
      boolean result = super.add(obj);

      if (obj.getFirst() - get(0).getFirst() > maxDelta)
        remove(0);

      return result;
    }

  }

  private static final int MAX_TIME_PERIOD = 60 * 60 * 1000;
  private static final List<Pair<Long,Double>> loadOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> ingestRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> ingestByteRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Integer>> minorCompactionsOverTime = Collections.synchronizedList(new MaxList<Integer>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Integer>> majorCompactionsOverTime = Collections.synchronizedList(new MaxList<Integer>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> lookupsOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Integer>> queryRateOverTime = Collections.synchronizedList(new MaxList<Integer>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Integer>> scanRateOverTime = Collections.synchronizedList(new MaxList<Integer>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> queryByteRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> indexCacheHitRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static final List<Pair<Long,Double>> dataCacheHitRateOverTime = Collections.synchronizedList(new MaxList<Double>(MAX_TIME_PERIOD));
  private static EventCounter lookupRateTracker = new EventCounter();
  private static EventCounter indexCacheHitTracker = new EventCounter();
  private static EventCounter indexCacheRequestTracker = new EventCounter();
  private static EventCounter dataCacheHitTracker = new EventCounter();
  private static EventCounter dataCacheRequestTracker = new EventCounter();

  private static volatile boolean fetching = false;
  private static MasterMonitorInfo mmi;
  private static Map<String,Map<ProblemType,Integer>> problemSummary = Collections.emptyMap();
  private static Exception problemException;
  private static GCStatus gcStatus;

  private static Instance instance;

  private static ServerConfigurationFactory config;
  private static AccumuloServerContext context;

  private static EmbeddedWebServer server;

  private ZooLock monitorLock;

  private static final String DEFAULT_INSTANCE_NAME = "(Unavailable)";
  public static final AtomicReference<String> cachedInstanceName = new AtomicReference<String>(DEFAULT_INSTANCE_NAME);

  private static class EventCounter {

    Map<String,Pair<Long,Long>> prevSamples = new HashMap<String,Pair<Long,Long>>();
    Map<String,Pair<Long,Long>> samples = new HashMap<String,Pair<Long,Long>>();
    Set<String> serversUpdated = new HashSet<String>();

    void startingUpdates() {
      serversUpdated.clear();
    }

    void updateTabletServer(String name, long sampleTime, long numEvents) {
      Pair<Long,Long> newSample = new Pair<Long,Long>(sampleTime, numEvents);
      Pair<Long,Long> lastSample = samples.get(name);

      if (lastSample == null || !lastSample.equals(newSample)) {
        samples.put(name, newSample);
        if (lastSample != null) {
          prevSamples.put(name, lastSample);
        }
      }
      serversUpdated.add(name);
    }

    void finishedUpdating() {
      // remove any tablet servers not updated
      samples.keySet().retainAll(serversUpdated);
      prevSamples.keySet().retainAll(serversUpdated);
    }

    double calculateRate() {
      double totalRate = 0;

      for (Entry<String,Pair<Long,Long>> entry : prevSamples.entrySet()) {
        Pair<Long,Long> prevSample = entry.getValue();
        Pair<Long,Long> sample = samples.get(entry.getKey());

        totalRate += (sample.getSecond() - prevSample.getSecond()) / ((sample.getFirst() - prevSample.getFirst()) / (double) 1000);
      }

      return totalRate;
    }

    long calculateCount() {
      long count = 0;

      for (Entry<String,Pair<Long,Long>> entry : prevSamples.entrySet()) {
        Pair<Long,Long> prevSample = entry.getValue();
        Pair<Long,Long> sample = samples.get(entry.getKey());

        count += sample.getSecond() - prevSample.getSecond();
      }

      return count;
    }
  }

  public static void fetchData() {
    double totalIngestRate = 0.;
    double totalIngestByteRate = 0.;
    double totalQueryRate = 0.;
    double totalQueryByteRate = 0.;
    double totalScanRate = 0.;
    long totalEntries = 0;
    int totalTabletCount = 0;
    long totalHoldTime = 0;
    long totalLookups = 0;
    boolean retry = true;

    // only recalc every so often
    long currentTime = System.currentTimeMillis();
    if (currentTime - lastRecalc.get() < REFRESH_TIME * 1000)
      return;

    synchronized (Monitor.class) {
      // Learn our instance name asynchronously so we don't hang up if zookeeper is down
      if (cachedInstanceName.get().equals(DEFAULT_INSTANCE_NAME)) {
        SimpleTimer.getInstance(config.getConfiguration()).schedule(new TimerTask() {
          @Override
          public void run() {
            synchronized (Monitor.class) {
              if (cachedInstanceName.get().equals(DEFAULT_INSTANCE_NAME)) {
                cachedInstanceName.set(HdfsZooInstance.getInstance().getInstanceName());
              }
            }
          }
        }, 0);
      }
    }

    synchronized (Monitor.class) {
      if (fetching)
        return;
      fetching = true;
    }

    try {
      while (retry) {
        MasterClientService.Iface client = null;
        try {
          client = MasterClient.getConnection(context);
          if (client != null) {
            mmi = client.getMasterStats(Tracer.traceInfo(), context.rpcCreds());
            retry = false;
          } else {
            mmi = null;
          }
          Monitor.gcStatus = fetchGcStatus();
        } catch (Exception e) {
          mmi = null;
          log.info("Error fetching stats: " + e);
        } finally {
          if (client != null) {
            MasterClient.close(client);
          }
        }
        if (mmi == null)
          UtilWaitThread.sleep(1000);
      }
      if (mmi != null) {
        int majorCompactions = 0;
        int minorCompactions = 0;

        lookupRateTracker.startingUpdates();
        indexCacheHitTracker.startingUpdates();
        indexCacheRequestTracker.startingUpdates();
        dataCacheHitTracker.startingUpdates();
        dataCacheRequestTracker.startingUpdates();

        for (TabletServerStatus server : mmi.tServerInfo) {
          TableInfo summary = TableInfoUtil.summarizeTableStats(server);
          totalIngestRate += summary.ingestRate;
          totalIngestByteRate += summary.ingestByteRate;
          totalQueryRate += summary.queryRate;
          totalScanRate += summary.scanRate;
          totalQueryByteRate += summary.queryByteRate;
          totalEntries += summary.recs;
          totalHoldTime += server.holdTime;
          totalLookups += server.lookups;
          majorCompactions += summary.majors.running;
          minorCompactions += summary.minors.running;
          lookupRateTracker.updateTabletServer(server.name, server.lastContact, server.lookups);
          indexCacheHitTracker.updateTabletServer(server.name, server.lastContact, server.indexCacheHits);
          indexCacheRequestTracker.updateTabletServer(server.name, server.lastContact, server.indexCacheRequest);
          dataCacheHitTracker.updateTabletServer(server.name, server.lastContact, server.dataCacheHits);
          dataCacheRequestTracker.updateTabletServer(server.name, server.lastContact, server.dataCacheRequest);
        }

        lookupRateTracker.finishedUpdating();
        indexCacheHitTracker.finishedUpdating();
        indexCacheRequestTracker.finishedUpdating();
        dataCacheHitTracker.finishedUpdating();
        dataCacheRequestTracker.finishedUpdating();

        int totalTables = 0;
        for (TableInfo tInfo : mmi.tableMap.values()) {
          totalTabletCount += tInfo.tablets;
          totalTables++;
        }
        Monitor.totalIngestRate = totalIngestRate;
        Monitor.totalTables = totalTables;
        totalIngestByteRate = totalIngestByteRate / 1000000.0;
        Monitor.totalQueryRate = totalQueryRate;
        Monitor.totalScanRate = totalScanRate;
        totalQueryByteRate = totalQueryByteRate / 1000000.0;
        Monitor.totalEntries = totalEntries;
        Monitor.totalTabletCount = totalTabletCount;
        Monitor.totalHoldTime = totalHoldTime;
        Monitor.totalLookups = totalLookups;

        ingestRateOverTime.add(new Pair<Long,Double>(currentTime, totalIngestRate));
        ingestByteRateOverTime.add(new Pair<Long,Double>(currentTime, totalIngestByteRate));

        double totalLoad = 0.;
        for (TabletServerStatus status : mmi.tServerInfo) {
          if (status != null)
            totalLoad += status.osLoad;
        }
        loadOverTime.add(new Pair<Long,Double>(currentTime, totalLoad));

        minorCompactionsOverTime.add(new Pair<Long,Integer>(currentTime, minorCompactions));
        majorCompactionsOverTime.add(new Pair<Long,Integer>(currentTime, majorCompactions));

        lookupsOverTime.add(new Pair<Long,Double>(currentTime, lookupRateTracker.calculateRate()));

        queryRateOverTime.add(new Pair<Long,Integer>(currentTime, (int) totalQueryRate));
        queryByteRateOverTime.add(new Pair<Long,Double>(currentTime, totalQueryByteRate));

        scanRateOverTime.add(new Pair<Long,Integer>(currentTime, (int) totalScanRate));

        calcCacheHitRate(indexCacheHitRateOverTime, currentTime, indexCacheHitTracker, indexCacheRequestTracker);
        calcCacheHitRate(dataCacheHitRateOverTime, currentTime, dataCacheHitTracker, dataCacheRequestTracker);
      }
      try {
        Monitor.problemSummary = ProblemReports.getInstance(getContext()).summarize();
        Monitor.problemException = null;
      } catch (Exception e) {
        log.info("Failed to obtain problem reports ", e);
        Monitor.problemSummary = Collections.emptyMap();
        Monitor.problemException = e;
      }

    } finally {
      synchronized (Monitor.class) {
        fetching = false;
        lastRecalc.set(currentTime);
{'fraction_non_alphanumeric': 0.09835259404966806, 'fraction_numerical': 0.004733218588640275, 'mean_word_length': 4.025019305019305, 'pattern_counts': {'":': 0, '<': 147, '<?xml version=': 0, '>': 147, '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': '1089575', 'n_tokens_mistral': 9474, 'n_tokens_neox': 9064, 'n_words': 1880}