@Substitute
public synchronized void setNodes(Collection<Node> nodes) {
    if (nodes == null || nodes.isEmpty()) {
        throw new IllegalArgumentException("nodes must not be null or empty");
    }
    AuthCache authCache = new NoSerializationBasicAuthCache();

    Map<HttpHost, Node> nodesByHost = new LinkedHashMap<>();
    for (Node node : nodes) {
        Objects.requireNonNull(node, "node cannot be null");
        // TODO should we throw an IAE if we have two nodes with the same host?
        nodesByHost.put(node.getHost(), node);
        authCache.put(node.getHost(), new BasicScheme());
    }
    this.nodeTuple = new NodeTuple<>(Collections.unmodifiableList(new ArrayList<>(nodesByHost.values())),
            authCache);
    this.blacklist.clear();
}
 

@Before
public void setUp() throws Exception {
    Configurator.initialize(null, "config/log4j2.xml");
    SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");

    String SSL_TRUSTSTORE_PATH = "certs/truststore.jks";
    String SSL_TRUSTSTORE_PW = "changeit";
    String SSL_KEYSTORE_PATH = "certs/jmeter-keystore.jks";
    String SSL_KEYSTORE_PW = "changeit";

    System.setProperty("javax.net.ssl.keyStore", SSL_KEYSTORE_PATH);
    System.setProperty("javax.net.ssl.keyStorePassword", SSL_KEYSTORE_PW);
    System.setProperty("javax.net.ssl.keyStoreType",
            FilenameUtils.getExtension(SSL_KEYSTORE_PATH).toLowerCase().equals("jks") ? "JKS" : "PKCS12");

    System.setProperty("javax.net.ssl.trustStore", SSL_TRUSTSTORE_PATH);
    System.setProperty("javax.net.ssl.trustStorePassword", SSL_TRUSTSTORE_PW);
    System.setProperty("javax.net.ssl.trustStoreType",
            FilenameUtils.getExtension(SSL_TRUSTSTORE_PATH).toLowerCase().equals("jks") ? "JKS" : "PKCS12");

    client = RestClient.builder(new HttpHost("localhost", Integer.parseInt("9200"), "https"))
            .setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder.setConnectTimeout(5000)
                    .setSocketTimeout((int) 200L))
            .setFailureListener(new RestClient.FailureListener() {
                @Override
                public void onFailure(Node node) {
                    System.err.println("Error with node: " + node.toString());
                }
            }).setMaxRetryTimeoutMillis(60000).build();
    sender = new ElasticSearchMetricSender(client, "test_" + sdf.format(new Date()), "logstashTest",
            "logstashTest", "");
}
 

@Override
public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
    RestClientBuilder restClientBuilder = (RestClientBuilder) (allArguments[0]);
    RestClient restClient = restClientBuilder.build();

    RestClientEnhanceInfo restClientEnhanceInfo = new RestClientEnhanceInfo();
    List<Node> nodeList = restClient.getNodes();
    for (Node node : nodeList) {
        restClientEnhanceInfo.addHttpHost(node.getHost());
    }

    objInst.setSkyWalkingDynamicField(restClientEnhanceInfo);
}
 

@Before
public void setUp() throws Exception {
    List<Node> nodeList = new ArrayList<Node>();
    nodeList.add(new Node(new HttpHost("127.0.0.1", 9200)));
    nodeList.add(new Node(new HttpHost("127.0.0.1", 9300)));
    restHighLevelClientConInterceptor = new RestHighLevelClientConInterceptor();
    when(restClientBuilder.build()).thenReturn(restClient);
    when(restClient.getNodes()).thenReturn(nodeList);
    allArguments = new Object[] {restClientBuilder};
}
 

private HttpClient(final KeyStore trustStore, final String basicCredentials, final KeyStore keystore,
        final char[] keyPassword, final String keystoreAlias, final boolean verifyHostnames, final boolean ssl, String[] supportedProtocols, String[] supportedCipherSuites, final String... servers)
        throws UnrecoverableKeyException, KeyManagementException, NoSuchAlgorithmException, KeyStoreException, CertificateException,
        IOException {
    super();
    this.trustStore = trustStore;
    this.basicCredentials = basicCredentials;
    this.keystore = keystore;
    this.keyPassword = keyPassword;
    this.verifyHostnames = verifyHostnames;
    this.ssl = ssl;
    this.supportedProtocols = supportedProtocols;
    this.supportedCipherSuites = supportedCipherSuites;
    this.keystoreAlias = keystoreAlias;

    HttpHost[] hosts = Arrays.stream(servers)
            .map(s->s.split(":"))
            .map(s->new HttpHost(s[0], Integer.parseInt(s[1]),ssl?"https":"http"))
            .collect(Collectors.toList()).toArray(new HttpHost[0]);


    RestClientBuilder builder = RestClient.builder(hosts);
    //builder.setMaxRetryTimeoutMillis(10000);

    builder.setFailureListener(new RestClient.FailureListener() {
        @Override
        public void onFailure(Node node) {

        }

    });

    builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
        @Override
        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
            try {
                return asyncClientBuilder(httpClientBuilder);
            } catch (Exception e) {
                log.error("Unable to build http client",e);
                throw new RuntimeException(e);
            }
        }
    });

    rclient = new RestHighLevelClient(builder);
}
 

@Override
public void setupTest(BackendListenerContext context) throws Exception {
    try {
        this.filters = new HashSet<>();
        this.fields = new HashSet<>();
        this.modes = new HashSet<>(Arrays.asList("info", "debug", "error", "quiet"));
        this.bulkSize = Integer.parseInt(context.getParameter(ES_BULK_SIZE));
        this.timeoutMs = Integer.parseInt((context.getParameter(ES_TIMEOUT_MS)));
        this.buildNumber = (JMeterUtils.getProperty(ElasticsearchBackendClient.BUILD_NUMBER) != null
                && !JMeterUtils.getProperty(ElasticsearchBackendClient.BUILD_NUMBER).trim().equals(""))
                        ? Integer.parseInt(JMeterUtils.getProperty(ElasticsearchBackendClient.BUILD_NUMBER)) : 0;

        setSSLConfiguration(context);

        if (context.getParameter(ES_AWS_ENDPOINT).equalsIgnoreCase("")) {
            client = RestClient
                    .builder(new HttpHost(context.getParameter(ES_HOST),
                            Integer.parseInt(context.getParameter(ES_PORT)), context.getParameter(ES_SCHEME)))
                    .setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder.setConnectTimeout(5000)
                            .setSocketTimeout((int) timeoutMs))
                    .setFailureListener(new RestClient.FailureListener() {
                        @Override
                        public void onFailure(Node node) {
                            logger.error("Error with node: " + node.toString());
                        }
                    }).build();
        } else {
            AWS4Signer signer = new AWS4Signer();
            signer.setServiceName(SERVICE_NAME);
            signer.setRegionName(context.getParameter(ES_AWS_REGION));
            HttpRequestInterceptor interceptor = new AWSRequestSigningApacheInterceptor(SERVICE_NAME, signer,
                    credentialsProvider);
            client = RestClient.builder(HttpHost.create(context.getParameter(ES_AWS_ENDPOINT)))
                    .setHttpClientConfigCallback(hacb -> hacb.addInterceptorLast(interceptor)).build();
        }

        convertParameterToSet(context, ES_SAMPLE_FILTER, this.filters);
        convertParameterToSet(context, ES_FIELDS, this.fields);

        this.sender = new ElasticSearchMetricSender(client, context.getParameter(ES_INDEX).toLowerCase(),
                context.getParameter(ES_AUTH_USER), context.getParameter(ES_AUTH_PWD),
                context.getParameter(ES_AWS_ENDPOINT));
        this.sender.createIndex();
        this.esVersion = sender.getElasticSearchVersion();

        checkTestMode(context.getParameter(ES_TEST_MODE));
        super.setupTest(context);
    } catch (Exception e) {
        throw new IllegalStateException("Unable to connect to the ElasticSearch engine", e);
    }
}
 

private List<HttpHost> getHosts(String hosts) throws IOException {
    Map<String,Serializable> params = createConnectionParams();
    params.put(ElasticDataStoreFactory.HOSTNAME.key, hosts);
    ElasticDataStoreFactory factory = new ElasticDataStoreFactory();
    return factory.createRestClient(params).getNodes().stream().map(Node::getHost).collect(Collectors.toList());
}