聊聊jest的NodeChecker

序

本文主要研究一下jest的NodeChecker

NodeChecker

jest-common-6.3.1-sources.jar!/io/searchbox/client/config/discovery/NodeChecker.java

public class NodeChecker extends AbstractScheduledService {

    private final static Logger log = LoggerFactory.getLogger(NodeChecker.class);
    private final static String PUBLISH_ADDRESS_KEY = "http_address";
    private final static String PUBLISH_ADDRESS_KEY_V5 = "publish_address"; // The one that under "http" node
    private final static Pattern INETSOCKETADDRESS_PATTERN = Pattern.compile("(?:inet\\[)?(?:(?:[^:]+)?\\/)?([^:]+):(\\d+)\\]?");

    private final NodesInfo action;

    protected JestClient client;
    protected Scheduler scheduler;
    protected String defaultScheme;
    protected Set<String> bootstrapServerList;
    protected Set<String> discoveredServerList;

    public NodeChecker(JestClient jestClient, ClientConfig clientConfig) {
        action = new NodesInfo.Builder()
                .withHttp()
                .addNode(clientConfig.getDiscoveryFilter())
                .build();
        this.client = jestClient;
        this.defaultScheme = clientConfig.getDefaultSchemeForDiscoveredNodes();
        this.scheduler = Scheduler.newFixedDelaySchedule(
                0l,
                clientConfig.getDiscoveryFrequency(),
                clientConfig.getDiscoveryFrequencyTimeUnit()
        );
        this.bootstrapServerList = ImmutableSet.copyOf(clientConfig.getServerList());
        this.discoveredServerList = new LinkedHashSet<String>();
    }

    @Override
    protected void runOneIteration() throws Exception {
        JestResult result;
        try {
            result = client.execute(action);
        } catch (CouldNotConnectException cnce) {
            // Can't connect to this node, remove it from the list
            log.error("Connect exception executing NodesInfo!", cnce);
            removeNodeAndUpdateServers(cnce.getHost());
            return;
            // do not elevate the exception since that will stop the scheduled calls.
            // throw new RuntimeException("Error executing NodesInfo!", e);
        } catch (Exception e) {
            log.error("Error executing NodesInfo!", e);
            client.setServers(bootstrapServerList);
            return;
            // do not elevate the exception since that will stop the scheduled calls.
            // throw new RuntimeException("Error executing NodesInfo!", e);
        }  

        if (result.isSucceeded()) {
            LinkedHashSet<String> httpHosts = new LinkedHashSet<String>();

            JsonObject jsonMap = result.getJsonObject();
            JsonObject nodes = (JsonObject) jsonMap.get("nodes");
            if (nodes != null) {
                for (Entry<String, JsonElement> entry : nodes.entrySet()) {

                    JsonObject host = entry.getValue().getAsJsonObject();
                    JsonElement addressElement = null;
                    if (host.has("version")) {
                        int majorVersion = Integer.parseInt(Splitter.on('.').splitToList(host.get("version").getAsString()).get(0));

                        if (majorVersion >= 5) {
                            JsonObject http = host.getAsJsonObject("http");
                            if (http != null && http.has(PUBLISH_ADDRESS_KEY_V5))
                                addressElement = http.get(PUBLISH_ADDRESS_KEY_V5);
                        }
                    }

                    if (addressElement == null) {
                        // get as a JsonElement first as some nodes in the cluster may not have an http_address
                        if (host.has(PUBLISH_ADDRESS_KEY)) addressElement = host.get(PUBLISH_ADDRESS_KEY);
                    }

                    if (addressElement != null && !addressElement.isJsonNull()) {
                        String httpAddress = getHttpAddress(addressElement.getAsString());
                        if(httpAddress != null) httpHosts.add(httpAddress);
                    }
              }
            }
            if (log.isDebugEnabled()) {
                log.debug("Discovered {} HTTP hosts: {}", httpHosts.size(), Joiner.on(',').join(httpHosts));
            }
            discoveredServerList = httpHosts;
            client.setServers(discoveredServerList);
        } else {
            log.warn("NodesInfo request resulted in error: {}", result.getErrorMessage());
            client.setServers(bootstrapServerList);
        }
    }

    protected void removeNodeAndUpdateServers(final String hostToRemove) {
        log.warn("Removing host {}", hostToRemove);
        discoveredServerList.remove(hostToRemove);
        if (log.isInfoEnabled()) {
            log.info("Discovered server pool is now: {}", Joiner.on(',').join(discoveredServerList));
        }
        if (!discoveredServerList.isEmpty()) {
          client.setServers(discoveredServerList);
        } else {
          client.setServers(bootstrapServerList);
        }
    }

    @Override
    protected Scheduler scheduler() {
        return scheduler;
    }

    @Override
    protected ScheduledExecutorService executor() {
        final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
            new ThreadFactoryBuilder()
                .setDaemon(true)
                .setNameFormat(serviceName())
                .build());
        // Add a listener to shutdown the executor after the service is stopped.  This ensures that the
        // JVM shutdown will not be prevented from exiting after this service has stopped or failed.
        // Technically this listener is added after start() was called so it is a little gross, but it
        // is called within doStart() so we know that the service cannot terminate or fail concurrently
        // with adding this listener so it is impossible to miss an event that we are interested in.
        addListener(new Listener() {
            @Override public void terminated(State from) {
                executor.shutdown();
            }
            @Override public void failed(State from, Throwable failure) {
                executor.shutdown();
            }}, MoreExecutors.directExecutor());
        return executor;
    }

    /**
     * Converts the Elasticsearch reported publish address in the format "inet[<hostname>:<port>]" or
     * "inet[<hostname>/<hostaddress>:<port>]" to a normalized http address in the form "http://host:port".
     */
    protected String getHttpAddress(String httpAddress) {
        Matcher resolvedMatcher = INETSOCKETADDRESS_PATTERN.matcher(httpAddress);
        if (resolvedMatcher.matches()) {
            return defaultScheme + resolvedMatcher.group(1) + ":" + resolvedMatcher.group(2);
        }

        return null;
    }

}

• NodeChecker繼承了AbstractScheduledService，它的構造器根據clientConfig的discoveryFrequency及discoveryFrequencyTimeUnit新建了fixedDelayScheduler來執行node checker；它實現了runOneIteration方法，該方法主要是發送NodesInfo請求(GET /_nodes/_all/http)
• 若是請求拋出CouldNotConnectException則調用removeNodeAndUpdateServers方法移除該host；若是拋出其餘的Exception則將client的servers重置爲bootstrapServerList
• 若是請求成功則解析body，若是nodes下面有version且大於等於5則取http節點下面的PUBLISH_ADDRESS_KEY_V5(publish_address)屬性值添加到discoveredServerList；舊版本的則從nodes下面的PUBLISH_ADDRESS_KEY(http_address)屬性值添加到discoveredServerList

NodesInfo返回實例

{
  "_nodes" : {
    "total" : 1,
    "successful" : 1,
    "failed" : 0
  },
  "cluster_name" : "docker-cluster",
  "nodes" : {
    "RmyGhZEbTjC7JCQFVS3HWQ" : {
      "name" : "RmyGhZE",
      "transport_address" : "172.17.0.2:9300",
      "host" : "172.17.0.2",
      "ip" : "172.17.0.2",
      "version" : "6.6.2",
      "build_flavor" : "oss",
      "build_type" : "tar",
      "build_hash" : "3bd3e59",
      "roles" : [
        "master",
        "data",
        "ingest"
      ],
      "http" : {
        "bound_address" : [
          "0.0.0.0:9200"
        ],
        "publish_address" : "192.168.99.100:9200",
        "max_content_length_in_bytes" : 104857600
      }
    }
  }
}

• 若是是5版本及以上的則在nodes下面有http屬性，裏頭有publish_address屬性用於返回該node的publish address

JestHttpClient

jest-6.3.1-sources.jar!/io/searchbox/client/http/JestHttpClient.java

public class JestHttpClient extends AbstractJestClient {
    //......

    @Override
    public <T extends JestResult> T execute(Action<T> clientRequest) throws IOException {
        return execute(clientRequest, null);
    }

    public <T extends JestResult> T execute(Action<T> clientRequest, RequestConfig requestConfig) throws IOException {
        HttpUriRequest request = prepareRequest(clientRequest, requestConfig);
        CloseableHttpResponse response = null;
        try {
            response = executeRequest(request);
            return deserializeResponse(response, request, clientRequest);
        } catch (HttpHostConnectException ex) {
            throw new CouldNotConnectException(ex.getHost().toURI(), ex);
        } finally {
            if (response != null) {
                try {
                    response.close();
                } catch (IOException ex) {
                    log.error("Exception occurred while closing response stream.", ex);
                }
            }
        }
    }

    @Override
    public <T extends JestResult> void executeAsync(final Action<T> clientRequest, final JestResultHandler<? super T> resultHandler) {
        executeAsync(clientRequest, resultHandler, null);
    }

    public <T extends JestResult> void executeAsync(final Action<T> clientRequest, final JestResultHandler<? super T> resultHandler, final RequestConfig requestConfig) {
        synchronized (this) {
            if (!asyncClient.isRunning()) {
                asyncClient.start();
            }
        }

        HttpUriRequest request = prepareRequest(clientRequest, requestConfig);
        executeAsyncRequest(clientRequest, resultHandler, request);
    }

    protected <T extends JestResult> HttpUriRequest prepareRequest(final Action<T> clientRequest, final RequestConfig requestConfig) {
        String elasticSearchRestUrl = getRequestURL(getNextServer(), clientRequest.getURI(elasticsearchVersion));
        HttpUriRequest request = constructHttpMethod(clientRequest.getRestMethodName(), elasticSearchRestUrl, clientRequest.getData(gson), requestConfig);

        log.debug("Request method={} url={}", clientRequest.getRestMethodName(), elasticSearchRestUrl);

        // add headers added to action
        for (Entry<String, Object> header : clientRequest.getHeaders().entrySet()) {
            request.addHeader(header.getKey(), header.getValue().toString());
        }

        return request;
    }

    protected String getNextServer() {
        return serverPoolReference.get().getNextServer();
    }

    //......
}

• JestHttpClient繼承了AbstractJestClient，它的execute及executeAsync方法都調用了prepareRequest來構造HttpUriRequest；該方法會先調用getNextServer方法來獲取要請求的elasticSearchServer的地址；而getNextServer方法則是調用的serverPoolReference.get().getNextServer()

AbstractJestClient

jest-common-6.3.1-sources.jar!/io/searchbox/client/AbstractJestClient.java

public abstract class AbstractJestClient implements JestClient {

    private final AtomicReference<ServerPool> serverPoolReference =
            new AtomicReference<ServerPool>(new ServerPool(ImmutableSet.<String>of()));

    //......

    public void setServers(Set<String> servers) {
        if (servers.equals(serverPoolReference.get().getServers())) {
            if (log.isDebugEnabled()) {
                log.debug("Server pool already contains same list of servers: {}",
                        Joiner.on(',').join(scrubServerURIs(servers)));
            }
            return;
        }
        if (log.isInfoEnabled()) {
            log.info("Setting server pool to a list of {} servers: [{}]",
                      servers.size(), Joiner.on(',').join(scrubServerURIs(servers)));
        }
        serverPoolReference.set(new ServerPool(servers));

        if (servers.isEmpty()) {
            log.warn("No servers are currently available to connect.");
        }
    }

    //......        
}

• AbstractJestClient有一個serverPoolReference屬性，它是AtomicReference，其泛型爲ServerPool；setServers方法則是建立新的ServerPool，而後更新serverPoolReference

ServerPool

jest-common-6.3.1-sources.jar!/io/searchbox/client/AbstractJestClient.java

private static final class ServerPool {
        private final List<String> serversRing;
        private final AtomicInteger nextServerIndex = new AtomicInteger(0);

        public ServerPool(final Set<String> servers) {
            this.serversRing = ImmutableList.copyOf(servers);
        }

        public Set<String> getServers() {
            return ImmutableSet.copyOf(serversRing);
        }

        public String getNextServer() {
            if (serversRing.size() > 0) {
                try {
                    return serversRing.get(nextServerIndex.getAndIncrement() % serversRing.size());
                } catch (IndexOutOfBoundsException outOfBoundsException) {
                    // In the very rare case where nextServerIndex overflowed, this will end up with a negative number,
                    // resulting in an IndexOutOfBoundsException.
                    // We should then start back at the beginning of the server list.
                    // Note that this might happen on several threads at once, in which the reset might happen a few times
                    log.info("Resetting next server index");
                    nextServerIndex.set(0);
                    return serversRing.get(nextServerIndex.getAndIncrement() % serversRing.size());
                }
            }
            else {
                throw new NoServerConfiguredException("No Server is assigned to client to connect");
            }
        }

        public int getSize() {
            return serversRing.size();
        }
    }

• ServerPool有個AtomicInteger類型的nextServerIndex，getNextServer方法則是經過nextServerIndex.getAndIncrement() % serversRing.size()來肯定取的serversRing這個List的index，其實現的是Round Robin策略；極端狀況下出現IndexOutOfBoundsException的話，則會重置nextServerIndex爲0，而後繼續按Round Robin策略取下一個server

小結

• NodeChecker繼承了AbstractScheduledService，它的構造器根據clientConfig的discoveryFrequency及discoveryFrequencyTimeUnit新建了fixedDelayScheduler來執行node checker；它實現了runOneIteration方法，該方法主要是發送NodesInfo請求(GET /_nodes/_all/http)，而後獲取nodes的PUBLISH_ADDRESS來更新discoveredServerList
• JestHttpClient繼承了AbstractJestClient，它的execute及executeAsync方法都調用了prepareRequest來構造HttpUriRequest；該方法會先調用getNextServer方法來獲取要請求的elasticSearchServer的地址；而getNextServer方法則是調用的serverPoolReference.get().getNextServer()；AbstractJestClient有一個serverPoolReference屬性，它是AtomicReference，其泛型爲ServerPool；setServers方法則是建立新的ServerPool，而後更新serverPoolReference
• ServerPool有個AtomicInteger類型的nextServerIndex，getNextServer方法則是經過nextServerIndex.getAndIncrement() % serversRing.size()來肯定取的serversRing這個List的index，其實現的是Round Robin策略；極端狀況下出現IndexOutOfBoundsException的話，則會重置nextServerIndex爲0，而後繼續按Round Robin策略取下一個server

doc

• NodeChecker