import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
@Override
public void ensureEstimatedStats() {
    if (size() > 0) {
        NodesStatsResponse nodeStats = client().admin().cluster().prepareNodesStats()
                .clear().setBreaker(true).setIndices(true).execute().actionGet();
        for (NodeStats stats : nodeStats.getNodes()) {
            assertThat("Fielddata breaker not reset to 0 on node: " + stats.getNode(),
                    stats.getBreaker().getStats(CircuitBreaker.FIELDDATA).getEstimated(), equalTo(0L));
            // ExternalTestCluster does not check the request breaker,
            // because checking it requires a network request, which in
            // turn increments the breaker, making it non-0

            assertThat("Fielddata size must be 0 on node: " +
                stats.getNode(), stats.getIndices().getFieldData().getMemorySizeInBytes(), equalTo(0L));
            assertThat("Query cache size must be 0 on node: " +
                stats.getNode(), stats.getIndices().getQueryCache().getMemorySizeInBytes(), equalTo(0L));
            assertThat("FixedBitSet cache size must be 0 on node: " +
                stats.getNode(), stats.getIndices().getSegments().getBitsetMemoryInBytes(), equalTo(0L));
        }
    }
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
private Table buildTable(final RestRequest request, final NodesStatsResponse nodeStatses) {
    Table table = getTableWithHeader(request);

    for (NodeStats nodeStats: nodeStatses.getNodes()) {
        if (nodeStats.getIndices().getFieldData().getFields() != null) {
            for (ObjectLongCursor<String> cursor : nodeStats.getIndices().getFieldData().getFields()) {
                table.startRow();
                table.addCell(nodeStats.getNode().getId());
                table.addCell(nodeStats.getNode().getHostName());
                table.addCell(nodeStats.getNode().getHostAddress());
                table.addCell(nodeStats.getNode().getName());
                table.addCell(cursor.key);
                table.addCell(new ByteSizeValue(cursor.value));
                table.endRow();
            }
        }
    }

    return table;
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
public NodeStats stats(CommonStatsFlags indices, boolean os, boolean process, boolean jvm, boolean threadPool,
                       boolean fs, boolean transport, boolean http, boolean circuitBreaker,
                       boolean script, boolean discoveryStats, boolean ingest) {
    // for indices stats we want to include previous allocated shards stats as well (it will
    // only be applied to the sensible ones to use, like refresh/merge/flush/indexing stats)
    return new NodeStats(discovery.localNode(), System.currentTimeMillis(),
            indices.anySet() ? indicesService.stats(true, indices) : null,
            os ? monitorService.osService().stats() : null,
            process ? monitorService.processService().stats() : null,
            jvm ? monitorService.jvmService().stats() : null,
            threadPool ? this.threadPool.stats() : null,
            fs ? monitorService.fsService().stats() : null,
            transport ? transportService.stats() : null,
            http ? (httpServerTransport == null ? null : httpServerTransport.stats()) : null,
            circuitBreaker ? circuitBreakerService.stats() : null,
            script ? scriptService.stats() : null,
            discoveryStats ? discovery.stats() : null,
            ingest ? ingestService.getPipelineExecutionService().stats() : null
    );
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
/** Returns true if any of the nodes used a noop breaker */
private boolean noopBreakerUsed() {
    NodesStatsResponse stats = client().admin().cluster().prepareNodesStats().setBreaker(true).get();
    for (NodeStats nodeStats : stats.getNodes()) {
        if (nodeStats.getBreaker().getStats(CircuitBreaker.REQUEST).getLimit() == NoopCircuitBreaker.LIMIT) {
            return true;
        }
        if (nodeStats.getBreaker().getStats(CircuitBreaker.IN_FLIGHT_REQUESTS).getLimit() == NoopCircuitBreaker.LIMIT) {
            return true;
        }
        if (nodeStats.getBreaker().getStats(CircuitBreaker.FIELDDATA).getLimit() == NoopCircuitBreaker.LIMIT) {
            return true;
        }
    }
    return false;
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
/** Issues a cache clear and waits 30 seconds for the field data breaker to be cleared */
public void clearFieldData() throws Exception {
    client().admin().indices().prepareClearCache().setFieldDataCache(true).execute().actionGet();
    assertBusy(new Runnable() {
        @Override
        public void run() {
            NodesStatsResponse resp = client().admin().cluster().prepareNodesStats()
                    .clear().setBreaker(true).get(new TimeValue(15, TimeUnit.SECONDS));
            for (NodeStats nStats : resp.getNodes()) {
                assertThat("fielddata breaker never reset back to 0",
                        nStats.getBreaker().getStats(CircuitBreaker.FIELDDATA).getEstimated(),
                        equalTo(0L));
            }
        }
    }, 30, TimeUnit.SECONDS);
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
private static CompletableFuture<NodeStats> getNodeStats(final Client client) {
    CompletableFuture<NodeStats> result = new CompletableFuture<>();

    NodesStatsRequest nodesStatsRequest = new NodesStatsRequest("_local").all();
    client.admin().cluster().nodesStats(nodesStatsRequest, new ActionListener<NodesStatsResponse>() {
        @Override
        public void onResponse(NodesStatsResponse nodesStatsResponse) {
            NodeStats stats = nodesStatsResponse.getNodes().get(0);
            result.complete(stats);
        }

        @Override
        public void onFailure(Exception e) {
            result.completeExceptionally(e);
        }
    });
    return result;
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
private void logNodeStats(Map<String, NodeStats> statsMap) {
  Map<String, String> tags = new HashMap<>();
  for (NodeStats stat : statsMap.values()) {
    tags.put("esnode", stat.getHostname());
    Stats.setGauge(StatsUtil.getStatsName("eshealth", "heapUsedPercent", tags),
        stat.getJvm().getMem().getHeapUsedPrecent());
    Stats.setGauge(StatsUtil.getStatsName("eshealth", "heapMaxMB", tags),
        stat.getJvm().getMem().getHeapMax().getMbFrac());
    Stats.setGauge(StatsUtil.getStatsName("eshealth", "heapUsedMB", tags),
        stat.getJvm().getMem().getHeapUsed().getMbFrac());
    Stats.setGauge(StatsUtil.getStatsName("eshealth", "upMinutes", tags),
        stat.getJvm().getUptime().getMinutesFrac());
    Stats.setGauge(StatsUtil.getStatsName("eshealth", "docCount", tags),
        stat.getIndices().getDocs().getCount());
  }
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
public NodeStats stats() throws IOException {
    // for indices stats we want to include previous allocated shards stats as well (it will
    // only be applied to the sensible ones to use, like refresh/merge/flush/indexing stats)
    return new NodeStats(discovery.localNode(), System.currentTimeMillis(),
            indicesService.stats(true),
            monitorService.osService().stats(),
            monitorService.processService().stats(),
            monitorService.jvmService().stats(),
            threadPool.stats(),
            monitorService.fsService().stats(),
            transportService.stats(),
            httpServer == null ? null : httpServer.stats(),
            circuitBreakerService.stats(),
            scriptService.stats()
    );
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
@Override
protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeRequest) {
    NodeInfo nodeInfo = nodeService.info(false, true, false, true, false, true, false, true);
    NodeStats nodeStats = nodeService.stats(CommonStatsFlags.NONE, true, true, true, false, true, false, false, false, false);
    List<ShardStats> shardsStats = new ArrayList<>();
    for (IndexService indexService : indicesService) {
        for (IndexShard indexShard : indexService) {
            if (indexShard.routingEntry() != null && indexShard.routingEntry().active()) {
                // only report on fully started shards
                shardsStats.add(new ShardStats(indexShard.routingEntry(), indexShard.shardPath(), new CommonStats(indexShard, SHARD_STATS_FLAGS), indexShard.commitStats()));
            }
        }
    }

    ClusterHealthStatus clusterStatus = null;
    if (clusterService.state().nodes().localNodeMaster()) {
        clusterStatus = new ClusterStateHealth(clusterService.state()).getStatus();
    }

    return new ClusterStatsNodeResponse(nodeInfo.getNode(), clusterStatus, nodeInfo, nodeStats, shardsStats.toArray(new ShardStats[shardsStats.size()]));

}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
public void addNodeStats(NodeStats nodeStats) {
    if (nodeStats.getProcess() == null) {
        return;
    }
    count++;
    if (nodeStats.getProcess().getCpu() != null) {
        cpuPercent += nodeStats.getProcess().getCpu().getPercent();
    }
    long fd = nodeStats.getProcess().getOpenFileDescriptors();
    if (fd > 0) {
        // fd can be -1 if not supported on platform
        totalOpenFileDescriptors += fd;
    }
    // we still do min max calc on -1, so we'll have an indication of it not being supported on one of the nodes.
    minOpenFileDescriptors = Math.min(minOpenFileDescriptors, fd);
    maxOpenFileDescriptors = Math.max(maxOpenFileDescriptors, fd);
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
@Test
public void testCircuitBreakerOnShard() throws Exception {
  // Update circuit breaker settings
  Settings settings = settingsBuilder()
          .put(HierarchyCircuitBreakerService.REQUEST_CIRCUIT_BREAKER_LIMIT_SETTING, "8b")
          .build();
  assertAcked(client().admin().cluster().prepareUpdateSettings().setTransientSettings(settings));

  SearchRequestBuilder searchRequest = new CoordinateSearchRequestBuilder(client()).setIndices("index1").setQuery(
          QueryBuilders.filterJoin("foreign_key").indices("index2").types("type").path("id").query(
                  boolQuery().filter(termQuery("tag", "aaa"))
          ).termsEncoding(TermsByQueryRequest.TermsEncoding.LONG)
  );
  assertFailures(searchRequest, RestStatus.INTERNAL_SERVER_ERROR,
          containsString("Data too large, data for [<terms_set>] would be larger than limit of [8/8b]"));

  NodesStatsResponse stats = client().admin().cluster().prepareNodesStats().setBreaker(true).get();
  int breaks = 0;
  for (NodeStats stat : stats.getNodes()) {
    CircuitBreakerStats breakerStats = stat.getBreaker().getStats(CircuitBreaker.REQUEST);
    breaks += breakerStats.getTrippedCount();
  }
  assertThat(breaks, greaterThanOrEqualTo(1));
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
@Test
public void testCircuitBreakerOnCoordinator() throws Exception {
  // Update circuit breaker settings
  Settings settings = settingsBuilder()
          .put(HierarchyCircuitBreakerService.REQUEST_CIRCUIT_BREAKER_LIMIT_SETTING, "60b")
          .build();
  assertAcked(client().admin().cluster().prepareUpdateSettings().setTransientSettings(settings));

  SearchRequestBuilder searchRequest = new CoordinateSearchRequestBuilder(client()).setIndices("index1").setQuery(
          QueryBuilders.filterJoin("foreign_key").indices("index2").types("type").path("id").query(
                  boolQuery().filter(termQuery("tag", "aaa"))
          ).termsEncoding(TermsByQueryRequest.TermsEncoding.LONG)
  );
  assertFailures(searchRequest, RestStatus.INTERNAL_SERVER_ERROR,
          containsString("Data too large, data for [<terms_set>] would be larger than limit of [60/60b]"));

  NodesStatsResponse stats = client().admin().cluster().prepareNodesStats().setBreaker(true).get();
  int breaks = 0;
  for (NodeStats stat : stats.getNodes()) {
    CircuitBreakerStats breakerStats = stat.getBreaker().getStats(CircuitBreaker.REQUEST);
    breaks += breakerStats.getTrippedCount();
  }
  assertThat(breaks, greaterThanOrEqualTo(1));
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
/** Create a fake NodeStats for the given node and usage */
public static NodeStats makeStats(String nodeName, DiskUsage usage) {
    FsInfo.Path[] paths = new FsInfo.Path[1];
    FsInfo.Path path = new FsInfo.Path("/dev/null", null,
        usage.getTotalBytes(), usage.getFreeBytes(), usage.getFreeBytes());
    paths[0] = path;
    FsInfo fsInfo = new FsInfo(System.currentTimeMillis(), null, paths);
    return new NodeStats(new DiscoveryNode(nodeName, ESTestCase.buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT),
        System.currentTimeMillis(),
        null, null, null, null, null,
        fsInfo,
        null, null, null,
        null, null, null);
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
@Override
protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeRequest) {
    NodeInfo nodeInfo = nodeService.info(true, true, false, true, false, true, false, true, false, false);
    NodeStats nodeStats = nodeService.stats(CommonStatsFlags.NONE, true, true, true, false, true, false, false, false, false, false, false);
    List<ShardStats> shardsStats = new ArrayList<>();
    for (IndexService indexService : indicesService) {
        for (IndexShard indexShard : indexService) {
            if (indexShard.routingEntry() != null && indexShard.routingEntry().active()) {
                // only report on fully started shards
                shardsStats.add(
                    new ShardStats(
                        indexShard.routingEntry(),
                        indexShard.shardPath(),
                        new CommonStats(indicesService.getIndicesQueryCache(), indexShard, SHARD_STATS_FLAGS),
                        indexShard.commitStats(),
                        indexShard.seqNoStats()));
            }
        }
    }

    ClusterHealthStatus clusterStatus = null;
    if (clusterService.state().nodes().isLocalNodeElectedMaster()) {
        clusterStatus = new ClusterStateHealth(clusterService.state()).getStatus();
    }

    return new ClusterStatsNodeResponse(nodeInfo.getNode(), clusterStatus, nodeInfo, nodeStats, shardsStats.toArray(new ShardStats[shardsStats.size()]));

}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
ClusterStatsNodes(List<ClusterStatsNodeResponse> nodeResponses) {
    this.versions = new HashSet<>();
    this.fs = new FsInfo.Path();
    this.plugins = new HashSet<>();

    Set<InetAddress> seenAddresses = new HashSet<>(nodeResponses.size());
    List<NodeInfo> nodeInfos = new ArrayList<>();
    List<NodeStats> nodeStats = new ArrayList<>();
    for (ClusterStatsNodeResponse nodeResponse : nodeResponses) {
        nodeInfos.add(nodeResponse.nodeInfo());
        nodeStats.add(nodeResponse.nodeStats());
        this.versions.add(nodeResponse.nodeInfo().getVersion());
        this.plugins.addAll(nodeResponse.nodeInfo().getPlugins().getPluginInfos());

        // now do the stats that should be deduped by hardware (implemented by ip deduping)
        TransportAddress publishAddress = nodeResponse.nodeInfo().getTransport().address().publishAddress();
        final InetAddress inetAddress = publishAddress.address().getAddress();
        if (!seenAddresses.add(inetAddress)) {
            continue;
        }
        if (nodeResponse.nodeStats().getFs() != null) {
            this.fs.add(nodeResponse.nodeStats().getFs().getTotal());
        }
    }
    this.counts = new Counts(nodeInfos);
    this.os = new OsStats(nodeInfos, nodeStats);
    this.process = new ProcessStats(nodeStats);
    this.jvm = new JvmStats(nodeInfos, nodeStats);
    this.networkTypes = new NetworkTypes(nodeInfos);
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
/**
 * Build the stats from information about each node.
 */
private OsStats(List<NodeInfo> nodeInfos, List<NodeStats> nodeStatsList) {
    this.names = new ObjectIntHashMap<>();
    int availableProcessors = 0;
    int allocatedProcessors = 0;
    for (NodeInfo nodeInfo : nodeInfos) {
        availableProcessors += nodeInfo.getOs().getAvailableProcessors();
        allocatedProcessors += nodeInfo.getOs().getAllocatedProcessors();

        if (nodeInfo.getOs().getName() != null) {
            names.addTo(nodeInfo.getOs().getName(), 1);
        }
    }
    this.availableProcessors = availableProcessors;
    this.allocatedProcessors = allocatedProcessors;

    long totalMemory = 0;
    long freeMemory = 0;
    for (NodeStats nodeStats : nodeStatsList) {
        if (nodeStats.getOs() != null) {
            long total = nodeStats.getOs().getMem().getTotal().getBytes();
            if (total > 0) {
                totalMemory += total;
            }
            long free = nodeStats.getOs().getMem().getFree().getBytes();
            if (free > 0) {
                freeMemory += free;
            }
        }
    }
    this.mem = new org.elasticsearch.monitor.os.OsStats.Mem(totalMemory, freeMemory);
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
/**
 * Build from looking at a list of node statistics.
 */
private ProcessStats(List<NodeStats> nodeStatsList) {
    int count = 0;
    int cpuPercent = 0;
    long totalOpenFileDescriptors = 0;
    long minOpenFileDescriptors = Long.MAX_VALUE;
    long maxOpenFileDescriptors = Long.MIN_VALUE;
    for (NodeStats nodeStats : nodeStatsList) {
        if (nodeStats.getProcess() == null) {
            continue;
        }
        count++;
        if (nodeStats.getProcess().getCpu() != null) {
            cpuPercent += nodeStats.getProcess().getCpu().getPercent();
        }
        long fd = nodeStats.getProcess().getOpenFileDescriptors();
        if (fd > 0) {
            // fd can be -1 if not supported on platform
            totalOpenFileDescriptors += fd;
        }
        // we still do min max calc on -1, so we'll have an indication of it not being supported on one of the nodes.
        minOpenFileDescriptors = Math.min(minOpenFileDescriptors, fd);
        maxOpenFileDescriptors = Math.max(maxOpenFileDescriptors, fd);
    }
    this.count = count;
    this.cpuPercent = cpuPercent;
    this.totalOpenFileDescriptors = totalOpenFileDescriptors;
    this.minOpenFileDescriptors = minOpenFileDescriptors;
    this.maxOpenFileDescriptors = maxOpenFileDescriptors;
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
/**
 * Build from lists of information about each node.
 */
private JvmStats(List<NodeInfo> nodeInfos, List<NodeStats> nodeStatsList) {
    this.versions = new ObjectIntHashMap<>();
    long threads = 0;
    long maxUptime = 0;
    long heapMax = 0;
    long heapUsed = 0;
    for (NodeInfo nodeInfo : nodeInfos) {
        versions.addTo(new JvmVersion(nodeInfo.getJvm()), 1);
    }

    for (NodeStats nodeStats : nodeStatsList) {
        org.elasticsearch.monitor.jvm.JvmStats js = nodeStats.getJvm();
        if (js == null) {
            continue;
        }
        if (js.getThreads() != null) {
            threads += js.getThreads().getCount();
        }
        maxUptime = Math.max(maxUptime, js.getUptime().millis());
        if (js.getMem() != null) {
            heapUsed += js.getMem().getHeapUsed().getBytes();
            heapMax += js.getMem().getHeapMax().getBytes();
        }
    }
    this.threads = threads;
    this.maxUptime = maxUptime;
    this.heapUsed = heapUsed;
    this.heapMax = heapMax;
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
public ClusterStatsNodeResponse(DiscoveryNode node, @Nullable ClusterHealthStatus clusterStatus, NodeInfo nodeInfo, NodeStats nodeStats, ShardStats[] shardsStats) {
    super(node);
    this.nodeInfo = nodeInfo;
    this.nodeStats = nodeStats;
    this.shardsStats = shardsStats;
    this.clusterStatus = clusterStatus;
}
 

import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    clusterStatus = null;
    if (in.readBoolean()) {
        clusterStatus = ClusterHealthStatus.fromValue(in.readByte());
    }
    this.nodeInfo = NodeInfo.readNodeInfo(in);
    this.nodeStats = NodeStats.readNodeStats(in);
    int size = in.readVInt();
    shardsStats = new ShardStats[size];
    for (int i = 0; i < size; i++) {
        shardsStats[i] = ShardStats.readShardStats(in);
    }
}