forked from opensearch-project/data-prepper
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Final code for opensearch for opensearch-project#1985,2264.
Signed-off-by: mallikagogoi7 <[email protected]>
- Loading branch information
1 parent
40b73ad
commit 5d5b765
Showing
41 changed files
with
2,396 additions
and
302 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
72 changes: 72 additions & 0 deletions
72
...in/java/org/opensearch/dataprepper/plugins/source/opensearch/OpenSearchClientBuilder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package org.opensearch.dataprepper.plugins.source.opensearch; | ||
|
|
||
| import java.net.URL; | ||
| import java.util.List; | ||
|
|
||
| import co.elastic.clients.json.jackson.JacksonJsonpMapper; | ||
| import co.elastic.clients.transport.ElasticsearchTransport; | ||
| import co.elastic.clients.transport.rest_client.RestClientTransport; | ||
| import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider; | ||
| import org.apache.hc.core5.http.HttpHost; | ||
| import org.apache.http.HttpHeaders; | ||
| import org.apache.http.HttpResponseInterceptor; | ||
| import org.apache.http.entity.ContentType; | ||
| import org.apache.http.message.BasicHeader; | ||
| import org.elasticsearch.client.RestClient; | ||
| import org.opensearch.client.opensearch.OpenSearchClient; | ||
| import co.elastic.clients.elasticsearch.ElasticsearchClient; | ||
| import org.opensearch.client.transport.OpenSearchTransport; | ||
| import org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportBuilder; | ||
| import org.opensearch.dataprepper.plugins.source.opensearch.client.HttpCustomClient; | ||
|
|
||
| /** | ||
| * used for creating connection | ||
| */ | ||
| public class OpenSearchClientBuilder { | ||
|
|
||
| /** | ||
| * This method create opensearch client based on host information, which will be used to call opensearch apis | ||
| * @param url | ||
| * @return | ||
| */ | ||
| public OpenSearchClient createOpenSearchClient(final URL url){ | ||
| final HttpHost host = new HttpHost(url.getProtocol(), url.getHost(), url.getPort()); | ||
| final BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider(); | ||
| final OpenSearchTransport transport = ApacheHttpClient5TransportBuilder | ||
| .builder(host) | ||
| .setMapper(new org.opensearch.client.json.jackson.JacksonJsonpMapper()) | ||
| .build(); | ||
| return new OpenSearchClient(transport); | ||
| } | ||
|
|
||
| /** | ||
| * This method create Elasticsearch client based on host information, which will be used to call opensearch apis | ||
| * @param url | ||
| * @return | ||
| */ | ||
| public ElasticsearchClient createElasticSearchClient(final URL url) { | ||
| final String HEADER_NAME = "X-Elastic-Product"; | ||
| final String HEADER_VALUE = "Elasticsearch"; | ||
|
|
||
| RestClient client = org.elasticsearch.client.RestClient.builder(new org.apache.http.HttpHost(url.getHost(), url.getPort())). | ||
| setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder | ||
| .setDefaultHeaders(List.of(new BasicHeader(HttpHeaders.CONTENT_TYPE, ContentType.APPLICATION_JSON.toString()))) | ||
| .addInterceptorLast((HttpResponseInterceptor) (response, context) -> response.addHeader(HEADER_NAME, HEADER_VALUE))).build(); | ||
| JacksonJsonpMapper jacksonJsonpMapper = new JacksonJsonpMapper(); | ||
| ElasticsearchTransport transport = new RestClientTransport(client, jacksonJsonpMapper); | ||
| return new ElasticsearchClient(transport); | ||
| } | ||
|
|
||
| /** | ||
| * This will create a custom http client. | ||
| * @param url | ||
| * @return | ||
| */ | ||
| public HttpCustomClient createCustomHttpClient(final URL url){ | ||
| return new HttpCustomClient(url); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
69 changes: 69 additions & 0 deletions
69
...in/java/org/opensearch/dataprepper/plugins/source/opensearch/OpenSearchSourceService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package org.opensearch.dataprepper.plugins.source.opensearch; | ||
|
|
||
| import com.linecorp.armeria.client.retry.Backoff; | ||
| import org.opensearch.dataprepper.model.buffer.Buffer; | ||
| import org.opensearch.dataprepper.model.event.Event; | ||
| import org.opensearch.dataprepper.model.record.Record; | ||
| import org.opensearch.dataprepper.plugins.source.opensearch.configuration.IndexParametersConfiguration; | ||
| import org.opensearch.dataprepper.plugins.source.opensearch.model.ServiceInfo; | ||
| import org.opensearch.dataprepper.plugins.source.opensearch.service.ElasticSearchService; | ||
| import org.opensearch.dataprepper.plugins.source.opensearch.service.HostsService; | ||
| import org.opensearch.dataprepper.plugins.source.opensearch.service.OpenSearchService; | ||
| import org.opensearch.dataprepper.plugins.source.opensearch.worker.OpenSearchTimerWorker; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.Timer; | ||
|
|
||
| /** | ||
| * Service that will call both worker classes | ||
| */ | ||
| public class OpenSearchSourceService { | ||
|
|
||
| private final OpenSearchService openSearchService; | ||
|
|
||
| private final ElasticSearchService elasticSearchService; | ||
|
|
||
| private final HostsService hostsService; | ||
|
|
||
| private final OpenSearchSourceConfiguration sourceConfig; | ||
|
|
||
| private final Timer timer = new Timer(); | ||
|
|
||
| private final Buffer<Record<Event>> buffer; | ||
|
|
||
| static final long INITIAL_DELAY = Duration.ofSeconds(20).toMillis(); | ||
| static final long MAXIMUM_DELAY = Duration.ofMinutes(5).toMillis(); | ||
| static final double JITTER_RATE = 0.20; | ||
|
|
||
| public OpenSearchSourceService(final OpenSearchSourceConfiguration sourceConfig, | ||
| final HostsService hostsService, | ||
| final OpenSearchService openSearchService, | ||
| final ElasticSearchService elasticSearchService, | ||
| final Buffer<Record<Event>> buffer){ | ||
| this.sourceConfig = sourceConfig; | ||
| this.hostsService = hostsService; | ||
| this.openSearchService = openSearchService; | ||
| this.elasticSearchService =elasticSearchService; | ||
| this.buffer = buffer; | ||
| } | ||
|
|
||
| public void processHosts(){ | ||
| sourceConfig.getHosts().forEach(host ->{ | ||
| final ServiceInfo serviceInfo = hostsService.findServiceDetailsByUrl(host); | ||
| IndexParametersConfiguration index = sourceConfig.getIndexParametersConfiguration(); | ||
| final Backoff backoff = Backoff.exponential(INITIAL_DELAY, MAXIMUM_DELAY).withJitter(JITTER_RATE) | ||
| .withMaxAttempts(Integer.MAX_VALUE); | ||
| timer.scheduleAtFixedRate(new OpenSearchTimerWorker(openSearchService,elasticSearchService, | ||
| sourceConfig,buffer,serviceInfo,host, backoff), | ||
| sourceConfig.getSchedulingParameterConfiguration().getStartTime().getSecond(), | ||
| sourceConfig.getSchedulingParameterConfiguration().getRate().toMillis()); | ||
| }); | ||
| } | ||
| public void stop(){ | ||
| timer.cancel(); | ||
| } | ||
| } |
56 changes: 56 additions & 0 deletions
56
...in/java/org/opensearch/dataprepper/plugins/source/opensearch/client/HttpCustomClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package org.opensearch.dataprepper.plugins.source.opensearch.client; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import org.apache.hc.client5.http.classic.methods.HttpUriRequestBase; | ||
| import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; | ||
| import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse; | ||
| import org.apache.hc.client5.http.impl.classic.HttpClients; | ||
| import org.apache.hc.core5.http.ContentType; | ||
| import org.apache.hc.core5.http.io.entity.StringEntity; | ||
|
|
||
| import java.io.BufferedReader; | ||
| import java.io.IOException; | ||
| import java.io.InputStreamReader; | ||
| import java.net.URI; | ||
| import java.net.URL; | ||
|
|
||
| public class HttpCustomClient { | ||
|
|
||
| private URL url; | ||
|
|
||
| public HttpCustomClient(URL url) { | ||
| this.url = url; | ||
| } | ||
|
|
||
| private ObjectMapper mapper = new ObjectMapper(); | ||
|
|
||
| public <T> T execute(final Class<T> responseType, final Object requestObject, final String httpMethod,final String uri) throws IOException { | ||
| StringEntity requestEntity = new StringEntity(mapper.writeValueAsString(requestObject)); | ||
| URI httpUri = URI.create(url.getProtocol()+"://"+url.getAuthority()+"/"+uri); | ||
| HttpUriRequestBase operationRequest = new HttpUriRequestBase(httpMethod, httpUri); | ||
| operationRequest.setHeader("Accept", ContentType.APPLICATION_JSON); | ||
| operationRequest.setHeader("Content-type", ContentType.APPLICATION_JSON); | ||
| operationRequest.setEntity(requestEntity); | ||
|
|
||
| CloseableHttpResponse closeableHttpResponse = getCloseableHttpResponse(operationRequest); | ||
| T response = mapper.readValue(readBuffer(closeableHttpResponse).toString(), responseType); | ||
| return response; | ||
| } | ||
|
|
||
| private StringBuffer readBuffer(CloseableHttpResponse pitCloseableResponse) throws IOException { | ||
| StringBuffer result = new StringBuffer(); | ||
| BufferedReader reader = new BufferedReader(new InputStreamReader(pitCloseableResponse.getEntity().getContent())); | ||
| String line = ""; | ||
| while ((line = reader.readLine()) != null) { | ||
| result.append(line); | ||
| } | ||
| return result; | ||
| } | ||
|
|
||
| private CloseableHttpResponse getCloseableHttpResponse(final HttpUriRequestBase operationRequest) throws IOException { | ||
| CloseableHttpClient httpClient = HttpClients.createDefault(); | ||
| CloseableHttpResponse pitResponse = httpClient.execute(operationRequest); | ||
| return pitResponse; | ||
| } | ||
|
|
||
| } |
39 changes: 39 additions & 0 deletions
39
...n/java/org/opensearch/dataprepper/plugins/source/opensearch/codec/JacksonValueParser.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.plugins.source.opensearch.codec; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import jakarta.json.stream.JsonParser; | ||
| import org.opensearch.client.json.JsonpDeserializerBase; | ||
| import org.opensearch.client.json.JsonpMapper; | ||
| import org.opensearch.client.json.jackson.JacksonJsonpParser; | ||
| import java.io.IOException; | ||
| import java.util.EnumSet; | ||
|
|
||
| public class JacksonValueParser<T> extends JsonpDeserializerBase<T> { | ||
|
|
||
| private final ObjectMapper objectMapper = new ObjectMapper(); | ||
| private final Class<T> clazz; | ||
| public JacksonValueParser(Class<T> clazz) { | ||
| super(EnumSet.allOf(JsonParser.Event.class)); | ||
| this.clazz = clazz; | ||
| } | ||
| @Override | ||
| public T deserialize(JsonParser parser, JsonpMapper mapper, JsonParser.Event event) { | ||
|
|
||
| if (!(parser instanceof JacksonJsonpParser)) { | ||
| throw new IllegalArgumentException("Jackson's ObjectMapper can only be used with the JacksonJsonpProvider"); | ||
| } | ||
| com.fasterxml.jackson.core.JsonParser jkParser = ((JacksonJsonpParser) parser).jacksonParser(); | ||
|
|
||
| try { | ||
| return objectMapper.readValue(jkParser, clazz); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
|
|
||
| } | ||
| } |
Oops, something went wrong.