import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
private Table buildTable(RestRequest req, ClusterStateResponse state, NodesInfoResponse nodesInfo) {
    boolean fullId = req.paramAsBoolean("full_id", false);

    DiscoveryNodes nodes = state.getState().nodes();
    Table table = getTableWithHeader(req);

    for (DiscoveryNode node : nodes) {
        NodeInfo info = nodesInfo.getNodesMap().get(node.getId());
        for (Map.Entry<String, String> attrEntry : node.getAttributes().entrySet()) {
            table.startRow();
            table.addCell(node.getName());
            table.addCell(fullId ? node.getId() : Strings.substring(node.getId(), 0, 4));
            table.addCell(info == null ? null : info.getProcess().getId());
            table.addCell(node.getHostName());
            table.addCell(node.getHostAddress());
            table.addCell(node.getAddress().address().getPort());
            table.addCell(attrEntry.getKey());
            table.addCell(attrEntry.getValue());
            table.endRow();
        }
    }
    return table;
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
private Table buildTable(RestRequest req, ClusterStateResponse state, NodesInfoResponse nodesInfo) {
    DiscoveryNodes nodes = state.getState().nodes();
    Table table = getTableWithHeader(req);

    for (DiscoveryNode node : nodes) {
        NodeInfo info = nodesInfo.getNodesMap().get(node.getId());

        for (PluginInfo pluginInfo : info.getPlugins().getPluginInfos()) {
            table.startRow();
            table.addCell(node.getId());
            table.addCell(node.getName());
            table.addCell(pluginInfo.getName());
            table.addCell(pluginInfo.getVersion());
            table.addCell(pluginInfo.getDescription());
            table.endRow();
        }
    }

    return table;
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
public void testClusterUpdateSettingsNoAcknowledgement() {
    client().admin().indices().prepareCreate("test")
            .setSettings(Settings.builder()
                    .put("number_of_shards", between(cluster().numDataNodes(), DEFAULT_MAX_NUM_SHARDS))
                    .put("number_of_replicas", 0)).get();
    ensureGreen();

    // now that the cluster is stable, remove timeout
    removePublishTimeout();

    NodesInfoResponse nodesInfo = client().admin().cluster().prepareNodesInfo().get();
    String excludedNodeId = null;
    for (NodeInfo nodeInfo : nodesInfo.getNodes()) {
        if (nodeInfo.getNode().isDataNode()) {
            excludedNodeId = nodeInfo.getNode().getId();
            break;
        }
    }
    assertNotNull(excludedNodeId);

    ClusterUpdateSettingsResponse clusterUpdateSettingsResponse = client().admin().cluster().prepareUpdateSettings().setTimeout("0s")
            .setTransientSettings(Settings.builder().put("cluster.routing.allocation.exclude._id", excludedNodeId)).get();
    assertThat(clusterUpdateSettingsResponse.isAcknowledged(), equalTo(false));
    assertThat(clusterUpdateSettingsResponse.getTransientSettings().get("cluster.routing.allocation.exclude._id"), equalTo(excludedNodeId));
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的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.info.NodeInfo; //导入依赖的package包/类
public void testTransportClient() throws Exception {
    NodesInfoResponse nodeInfos = client().admin().cluster().prepareNodesInfo().get();
    List<NodeInfo>  nodes = nodeInfos.getNodes();
    assertTrue(nodes.size() > 0);
    TransportAddress publishAddress = randomFrom(nodes).getTransport().address().publishAddress();
    String clusterName = nodeInfos.getClusterName().value();

    Settings settings = Settings.builder()
            .put("cluster.name", clusterName)
            .put(ThreadContext.PREFIX + "." + CustomRealm.USER_HEADER, randomFrom(KNOWN_USERS))
            .put(ThreadContext.PREFIX + "." + CustomRealm.PW_HEADER, PASSWORD)
            .build();
    try (TransportClient client = new PreBuiltXPackTransportClient(settings)) {
        client.addTransportAddress(publishAddress);
        ClusterHealthResponse response = client.admin().cluster().prepareHealth().execute().actionGet();
        assertThat(response.isTimedOut(), is(false));
    }
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
@Before
public void setUpClient() throws IOException, InterruptedException, NoSuchFieldException, IllegalAccessException {
    NodeInfo[] nodes = admin().cluster().nodesInfo(Requests.nodesInfoRequest()).actionGet().getNodes();
    Assert.assertThat(nodes.length, Matchers.greaterThanOrEqualTo(1));

    TransportAddress transportAddress = nodes[0].getHttp().getAddress().publishAddress();
    Assert.assertEquals(InetSocketTransportAddress.class, transportAddress.getClass());
    InetSocketTransportAddress inetSocketTransportAddress = (InetSocketTransportAddress) transportAddress;
    InetSocketAddress socketAddress = inetSocketTransportAddress.address();

    String url = String.format("http://%s:%d", socketAddress.getHostName(), socketAddress.getPort());
    httpClient = new HttpClient(Collections.singleton(url));

    createIndex("the_index");
    ensureSearchable("the_index");
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
@Override
protected NodeRiverExecuteResponse nodeOperation(NodeRiverExecuteRequest request) throws ElasticsearchException {
    NodeInfo nodeInfo = nodeService.info(false, true, false, true, false, false, true, false, true);
    String riverType = request.getRiverType();
    String riverName = request.getRiverName();
    for (Map.Entry<RiverName, River> entry : RiverHelper.rivers(injector).entrySet()) {
        RiverName name = entry.getKey();
        if ((riverName == null || name.getName().equals(riverName))
                && (riverType == null || name.getType().equals(riverType))
                && entry.getValue() instanceof RunnableRiver) {
            RunnableRiver river = (RunnableRiver) entry.getValue();
            river.run();
            return new NodeRiverExecuteResponse(nodeInfo.getNode()).setExecuted(true);
        }
    }
    return new NodeRiverExecuteResponse(nodeInfo.getNode()).setExecuted(false);
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
public void testPluginIsLoaded() {
    NodesInfoResponse response = client().admin().cluster().prepareNodesInfo().setPlugins(true).get();
    for (NodeInfo node : response.getNodes()) {
        boolean founded = false;
        for (PluginInfo pluginInfo : node.getPlugins().getPluginInfos()) {
            if (pluginInfo.getName().equals(AnalysisOpenKoreanTextPlugin.class.getName())) {
                founded = true;
            }
        }
        Assert.assertTrue(founded);
    }
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
private boolean checkPluginInstalled(Client client) {
    NodesInfoResponse nodesInfoResponse = client.admin().cluster().prepareNodesInfo().setPlugins(true).get();
    for (NodeInfo nodeInfo : nodesInfoResponse.getNodes()) {
        for (PluginInfo pluginInfo : nodeInfo.getPlugins().getPluginInfos()) {
            if ("memgraph".equals(pluginInfo.getName())) {
                return true;
            }
        }
    }
    if (config.isErrorOnMissingMemgraphPlugin()) {
        throw new MemgraphException("Memgraph plugin cannot be found");
    }
    LOGGER.warn("Running without the server side Memgraph plugin will be deprecated in the future.");
    return false;
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
public void testDifferentPorts() throws Exception {
    if (!NetworkUtils.SUPPORTS_V6) {
        return;
    }
    logger.info("--> starting a node on ipv4 only");
    Settings ipv4Settings = Settings.builder().put("network.host", "127.0.0.1").build();
    String ipv4OnlyNode = internalCluster().startNode(ipv4Settings); // should bind 127.0.0.1:XYZ

    logger.info("--> starting a node on ipv4 and ipv6");
    Settings bothSettings = Settings.builder().put("network.host", "_local_").build();
    internalCluster().startNode(bothSettings); // should bind [::1]:XYZ and 127.0.0.1:XYZ+1

    logger.info("--> waiting for the cluster to declare itself stable");
    ensureStableCluster(2); // fails if port of publish address does not match corresponding bound address

    logger.info("--> checking if boundAddress matching publishAddress has same port");
    NodesInfoResponse nodesInfoResponse = client().admin().cluster().prepareNodesInfo().get();
    for (NodeInfo nodeInfo : nodesInfoResponse.getNodes()) {
        BoundTransportAddress boundTransportAddress = nodeInfo.getTransport().getAddress();
        if (nodeInfo.getNode().getName().equals(ipv4OnlyNode)) {
            assertThat(boundTransportAddress.boundAddresses().length, equalTo(1));
            assertThat(boundTransportAddress.boundAddresses()[0].getPort(), equalTo(boundTransportAddress.publishAddress().getPort()));
        } else {
            assertThat(boundTransportAddress.boundAddresses().length, greaterThan(1));
            for (TransportAddress boundAddress : boundTransportAddress.boundAddresses()) {
                assertThat(boundAddress, instanceOf(TransportAddress.class));
                TransportAddress inetBoundAddress = (TransportAddress) boundAddress;
                if (inetBoundAddress.address().getAddress() instanceof Inet4Address) {
                    // IPv4 address is preferred publish address for _local_
                    assertThat(inetBoundAddress.getPort(), equalTo(boundTransportAddress.publishAddress().getPort()));
                }
            }
        }
    }
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
public void testReindexFromRemote() throws Exception {
    NodeInfo nodeInfo = client().admin().cluster().prepareNodesInfo().get().getNodes().get(0);
    TransportAddress address = nodeInfo.getHttp().getAddress().publishAddress();
    RemoteInfo remote = new RemoteInfo("http", address.getAddress(), address.getPort(), new BytesArray("{\"match_all\":{}}"), null,
        null, emptyMap(), RemoteInfo.DEFAULT_SOCKET_TIMEOUT, RemoteInfo.DEFAULT_CONNECT_TIMEOUT);
    ReindexRequestBuilder request = ReindexAction.INSTANCE.newRequestBuilder(client()).source("source").destination("dest")
            .setRemoteInfo(remote);
    testCase(ReindexAction.NAME, request, matcher().created(DOC_COUNT));
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
public NodeInfo info(boolean settings, boolean os, boolean process, boolean jvm, boolean threadPool,
            boolean transport, boolean http, boolean plugin, boolean ingest, boolean indices) {
    return new NodeInfo(Version.CURRENT, Build.CURRENT, discovery.localNode(),
            settings ? settingsFilter.filter(this.settings) : null,
            os ? monitorService.osService().info() : null,
            process ? monitorService.processService().info() : null,
            jvm ? monitorService.jvmService().info() : null,
            threadPool ? this.threadPool.info() : null,
            transport ? transportService.info() : null,
            http ? (httpServerTransport == null ? null : httpServerTransport.info()) : null,
            plugin ? (pluginService == null ? null : pluginService.info()) : null,
            ingest ? (ingestService == null ? null : ingestService.info()) : null,
            indices ? indicesService.getTotalIndexingBufferBytes() : null
    );
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的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.info.NodeInfo; //导入依赖的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.info.NodeInfo; //导入依赖的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.info.NodeInfo; //导入依赖的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.info.NodeInfo; //导入依赖的package包/类
private NetworkTypes(final List<NodeInfo> nodeInfos) {
    final Map<String, AtomicInteger> transportTypes = new HashMap<>();
    final Map<String, AtomicInteger> httpTypes = new HashMap<>();
    for (final NodeInfo nodeInfo : nodeInfos) {
        final Settings settings = nodeInfo.getSettings();
        final String transportType =
            settings.get(NetworkModule.TRANSPORT_TYPE_KEY, NetworkModule.TRANSPORT_DEFAULT_TYPE_SETTING.get(settings));
        final String httpType =
            settings.get(NetworkModule.HTTP_TYPE_KEY, NetworkModule.HTTP_DEFAULT_TYPE_SETTING.get(settings));
        transportTypes.computeIfAbsent(transportType, k -> new AtomicInteger()).incrementAndGet();
        httpTypes.computeIfAbsent(httpType, k -> new AtomicInteger()).incrementAndGet();
    }
    this.transportTypes = Collections.unmodifiableMap(transportTypes);
    this.httpTypes = Collections.unmodifiableMap(httpTypes);
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的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.info.NodeInfo; //导入依赖的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);
    }
}
 

import org.elasticsearch.action.admin.cluster.node.info.NodeInfo; //导入依赖的package包/类
public void testNodeInfoStreaming() throws IOException {
    NodeInfo nodeInfo = createNodeInfo();
    try (BytesStreamOutput out = new BytesStreamOutput()) {
        nodeInfo.writeTo(out);
        try (StreamInput in = out.bytes().streamInput()) {
            NodeInfo readNodeInfo = NodeInfo.readNodeInfo(in);
            assertExpectedUnchanged(nodeInfo, readNodeInfo);
        }
    }
}