kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add an agent plugin to support elasticsearch7 (#6759)

This commit is contained in:
zifeihan 2021-04-16 18:06:18 +08:00 committed by GitHub
parent bc1043e9b3
commit 3a7d6c0ed4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 1111 additions and 403 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.md
View File

@ -10,6 +10,7 @@ Release Notes.
#### Java Agent
* Add `trace_segment_ref_limit_per_span` configuration mechanism to avoid OOM.
* Improve `GlobalIdGenerator` performance.
* Add an agent plugin to support elasticsearch7.
#### OAP-Backend
* BugFix: filter invalid Envoy access logs whose socket address is empty.

17
apm-sniffer/apm-sdk-plugin/elasticsearch-6.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v6/define/AdapterActionFutureInstrumentation.java
View File

@ -18,8 +18,11 @@
package org.apache.skywalking.apm.plugin.elasticsearch.v6.define;
import java.util.Collections;
import java.util.List;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.WitnessMethod;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.StaticMethodsInterceptPoint;
@ -28,6 +31,8 @@ import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import org.apache.skywalking.apm.plugin.elasticsearch.v6.interceptor.Constants;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.returns;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
public class AdapterActionFutureInstrumentation extends ClassEnhancePluginDefine {
@ -39,7 +44,7 @@ public class AdapterActionFutureInstrumentation extends ClassEnhancePluginDefine
@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
return new InstanceMethodsInterceptPoint[]{
return new InstanceMethodsInterceptPoint[] {
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
@ -71,6 +76,14 @@ public class AdapterActionFutureInstrumentation extends ClassEnhancePluginDefine
@Override
protected String[] witnessClasses() {
return new String[]{Constants.TASK_TRANSPORT_CHANNEL_WITNESS_CLASSES};
return new String[] {Constants.TASK_TRANSPORT_CHANNEL_WITNESS_CLASSES};
}
@Override
protected List<WitnessMethod> witnessMethods() {
return Collections.singletonList(new WitnessMethod(
Constants.SEARCH_HITS_WITNESS_CLASSES,
named("getTotalHits").and(takesArguments(0)).and(returns(long.class))
));
}
}

23
apm-sniffer/apm-sdk-plugin/elasticsearch-6.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v6/interceptor/AdapterActionFutureActionGetMethodsInterceptor.java
View File

@ -18,6 +18,7 @@
package org.apache.skywalking.apm.plugin.elasticsearch.v6.interceptor;
import java.lang.reflect.Method;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
@ -34,8 +35,6 @@ import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import java.lang.reflect.Method;
import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.ELASTICSEARCH_DSL_LENGTH_THRESHOLD;
import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.TRACE_DSL;
@ -45,26 +44,22 @@ public class AdapterActionFutureActionGetMethodsInterceptor implements InstanceM
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
if (!isTrace(objInst)) {
return;
if (isTrace(objInst)) {
AbstractSpan span = ContextManager.createLocalSpan(Constants.DB_TYPE + "/" + Constants.BASE_FUTURE_METHOD);
span.setComponent(ComponentsDefine.TRANSPORT_CLIENT);
Tags.DB_TYPE.set(span, Constants.DB_TYPE);
}
AbstractSpan span = ContextManager.createLocalSpan(Constants.DB_TYPE + "/" + Constants.BASE_FUTURE_METHOD);
span.setComponent(ComponentsDefine.TRANSPORT_CLIENT);
Tags.DB_TYPE.set(span, Constants.DB_TYPE);
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Object ret) throws Throwable {
if (!isTrace(objInst)) {
return ret;
if (isTrace(objInst)) {
AbstractSpan span = ContextManager.activeSpan();
parseResponseInfo((ActionResponse) ret, span);
ContextManager.stopSpan();
}
AbstractSpan span = ContextManager.activeSpan();
parseResponseInfo((ActionResponse) ret, span);
ContextManager.stopSpan();
return ret;
}

1
apm-sniffer/apm-sdk-plugin/elasticsearch-6.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v6/interceptor/Constants.java
View File

@ -43,6 +43,7 @@ public class Constants {
//witnessClasses
public static final String TASK_TRANSPORT_CHANNEL_WITNESS_CLASSES = "org.elasticsearch.transport.TaskTransportChannel";
public static final String SEARCH_HITS_WITNESS_CLASSES = "org.elasticsearch.search.SearchHits";
//es operator name
public static final String CREATE_OPERATOR_NAME = "Elasticsearch/CreateRequest";

52
apm-sniffer/apm-sdk-plugin/elasticsearch-7.x-plugin/pom.xml Normal file
View File

@ -0,0 +1,52 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You under the Apache License, Version 2.0
~ (the "License"); you may not use this file except in compliance with
~ the License. You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
~
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>apm-sdk-plugin</artifactId>
<groupId>org.apache.skywalking</groupId>
<version>8.6.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>apm-elasticsearch-7.x-plugin</artifactId>
<packaging>jar</packaging>
<name>elasticsearch-7.x-plugin</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<elasticsearch.rest.high.level.client.version>7.2.1</elasticsearch.rest.high.level.client.version>
</properties>
<dependencies>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>${elasticsearch.rest.high.level.client.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>apm-elasticsearch-6.x-plugin</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
</project>

89
apm-sniffer/apm-sdk-plugin/elasticsearch-7.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v7/define/AdapterActionFutureInstrumentation.java Normal file
View File

@ -0,0 +1,89 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.elasticsearch.v7.define;
import java.util.Collections;
import java.util.List;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.WitnessMethod;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.StaticMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassEnhancePluginDefine;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import org.apache.skywalking.apm.plugin.elasticsearch.v6.interceptor.Constants;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.returns;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
public class AdapterActionFutureInstrumentation extends ClassEnhancePluginDefine {
@Override
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
return new ConstructorInterceptPoint[0];
}
@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
return new InstanceMethodsInterceptPoint[] {
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named("actionGet");
}
@Override
public String getMethodsInterceptor() {
return "org.apache.skywalking.apm.plugin.elasticsearch.v7.interceptor.AdapterActionFutureActionGetMethodsInterceptor";
}
@Override
public boolean isOverrideArgs() {
return false;
}
}
};
}
@Override
public StaticMethodsInterceptPoint[] getStaticMethodsInterceptPoints() {
return new StaticMethodsInterceptPoint[0];
}
@Override
protected ClassMatch enhanceClass() {
return byName("org.elasticsearch.action.support.AdapterActionFuture");
}
@Override
protected String[] witnessClasses() {
return new String[] {Constants.TASK_TRANSPORT_CHANNEL_WITNESS_CLASSES};
}
@Override
protected List<WitnessMethod> witnessMethods() {
return Collections.singletonList(new WitnessMethod(
Constants.SEARCH_HITS_WITNESS_CLASSES,
named("getTotalHits").and(takesArguments(0)).and(returns(named("org.apache.lucene.search.TotalHits")))
));
}
}

161
apm-sniffer/apm-sdk-plugin/elasticsearch-7.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v7/interceptor/AdapterActionFutureActionGetMethodsInterceptor.java Normal file
View File

@ -0,0 +1,161 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.elasticsearch.v7.interceptor;
import java.lang.reflect.Method;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.plugin.elasticsearch.v6.interceptor.Constants;
import org.apache.skywalking.apm.util.StringUtil;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.ELASTICSEARCH_DSL_LENGTH_THRESHOLD;
import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.TRACE_DSL;
public class AdapterActionFutureActionGetMethodsInterceptor implements InstanceMethodsAroundInterceptor {
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
if (isTrace(objInst)) {
AbstractSpan span = ContextManager.createLocalSpan(Constants.DB_TYPE + "/" + Constants.BASE_FUTURE_METHOD);
span.setComponent(ComponentsDefine.TRANSPORT_CLIENT);
Tags.DB_TYPE.set(span, Constants.DB_TYPE);
}
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Object ret) throws Throwable {
if (isTrace(objInst)) {
AbstractSpan span = ContextManager.activeSpan();
parseResponseInfo((ActionResponse) ret, span);
ContextManager.stopSpan();
}
return ret;
}
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
ContextManager.activeSpan().log(t);
}
private boolean isTrace(EnhancedInstance objInst) {
return objInst.getSkyWalkingDynamicField() != null && (boolean) objInst.getSkyWalkingDynamicField();
}
private void parseResponseInfo(ActionResponse response, AbstractSpan span) {
// search response
if (response instanceof SearchResponse) {
parseSearchResponse((SearchResponse) response, span);
return;
}
// bulk response
if (response instanceof BulkResponse) {
parseBulkResponse((BulkResponse) response, span);
return;
}
// get response
if (response instanceof GetResponse) {
parseGetResponse((GetResponse) response, span);
return;
}
// index response
if (response instanceof IndexResponse) {
parseIndexResponse((IndexResponse) response, span);
return;
}
// update response
if (response instanceof UpdateResponse) {
parseUpdateResponse((UpdateResponse) response, span);
return;
}
// delete response
if (response instanceof DeleteResponse) {
parseDeleteResponse((DeleteResponse) response, span);
return;
}
}
private void parseSearchResponse(SearchResponse searchResponse, AbstractSpan span) {
span.tag(Constants.ES_TOOK_MILLIS, Long.toString(searchResponse.getTook().getMillis()));
span.tag(Constants.ES_TOTAL_HITS, Long.toString(searchResponse.getHits().getTotalHits().value));
if (TRACE_DSL) {
String tagValue = searchResponse.toString();
tagValue = ELASTICSEARCH_DSL_LENGTH_THRESHOLD > 0 ? StringUtil.cut(tagValue, ELASTICSEARCH_DSL_LENGTH_THRESHOLD) : tagValue;
Tags.DB_STATEMENT.set(span, tagValue);
}
}
private void parseBulkResponse(BulkResponse bulkResponse, AbstractSpan span) {
span.tag(Constants.ES_TOOK_MILLIS, Long.toString(bulkResponse.getTook().getMillis()));
span.tag(Constants.ES_INGEST_TOOK_MILLIS, Long.toString(bulkResponse.getIngestTookInMillis()));
if (TRACE_DSL) {
String tagValue = bulkResponse.toString();
tagValue = ELASTICSEARCH_DSL_LENGTH_THRESHOLD > 0 ? StringUtil.cut(tagValue, ELASTICSEARCH_DSL_LENGTH_THRESHOLD) : tagValue;
Tags.DB_STATEMENT.set(span, tagValue);
}
}
private void parseGetResponse(GetResponse getResponse, AbstractSpan span) {
if (TRACE_DSL) {
String tagValue = getResponse.toString();
tagValue = ELASTICSEARCH_DSL_LENGTH_THRESHOLD > 0 ? StringUtil.cut(tagValue, ELASTICSEARCH_DSL_LENGTH_THRESHOLD) : tagValue;
Tags.DB_STATEMENT.set(span, tagValue);
}
}
private void parseIndexResponse(IndexResponse indexResponse, AbstractSpan span) {
if (TRACE_DSL) {
String tagValue = indexResponse.toString();
tagValue = ELASTICSEARCH_DSL_LENGTH_THRESHOLD > 0 ? StringUtil.cut(tagValue, ELASTICSEARCH_DSL_LENGTH_THRESHOLD) : tagValue;
Tags.DB_STATEMENT.set(span, tagValue);
}
}
private void parseUpdateResponse(UpdateResponse updateResponse, AbstractSpan span) {
if (TRACE_DSL) {
String tagValue = updateResponse.toString();
tagValue = ELASTICSEARCH_DSL_LENGTH_THRESHOLD > 0 ? StringUtil.cut(tagValue, ELASTICSEARCH_DSL_LENGTH_THRESHOLD) : tagValue;
Tags.DB_STATEMENT.set(span, tagValue);
}
}
private void parseDeleteResponse(DeleteResponse deleteResponse, AbstractSpan span) {
if (TRACE_DSL) {
String tagValue = deleteResponse.toString();
tagValue = ELASTICSEARCH_DSL_LENGTH_THRESHOLD > 0 ? StringUtil.cut(tagValue, ELASTICSEARCH_DSL_LENGTH_THRESHOLD) : tagValue;
Tags.DB_STATEMENT.set(span, tagValue);
}
}
}

17
apm-sniffer/apm-sdk-plugin/elasticsearch-7.x-plugin/src/main/resources/skywalking-plugin.def Normal file
View File

@ -0,0 +1,17 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
elasticsearch-7.x=org.apache.skywalking.apm.plugin.elasticsearch.v7.define.AdapterActionFutureInstrumentation

1
apm-sniffer/apm-sdk-plugin/pom.xml
View File

@ -72,6 +72,7 @@
<module>activemq-5.x-plugin</module>
<module>elasticsearch-5.x-plugin</module>
<module>elasticsearch-6.x-plugin</module>
<module>elasticsearch-7.x-plugin</module>
<module>undertow-plugins</module>
<module>rabbitmq-5.x-plugin</module>
<module>dubbo-conflict-patch</module>

1
docs/en/setup/service-agent/java-agent/Plugin-list.md
View File

@ -16,6 +16,7 @@
- elastic-job-3.x
- elasticsearch-5.x
- elasticsearch-6.x
- elasticsearch-7.x
- feign-default-http-9.x
- feign-pathvar-9.x
- finagle

1
docs/en/setup/service-agent/java-agent/Supported-list.md
View File

@ -76,6 +76,7 @@ metrics based on the tracing data.
* [Elasticsearch](https://github.com/elastic/elasticsearch)
* [transport-client](https://github.com/elastic/elasticsearch/tree/v5.2.0/client/transport) 5.2.x-5.6.x
* [transport-client](https://github.com/elastic/elasticsearch/tree/v6.7.1/client/transport) 6.7.1-6.8.4
* [transport-client](https://github.com/elastic/elasticsearch/tree/7.0/client/transport) 7.0.0-7.5.2
* [rest-high-level-client](https://www.elastic.co/guide/en/elasticsearch/client/java-rest/6.7/index.html) 6.7.1-6.8.4
* [rest-high-level-client](https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.0/java-rest-high.html) 7.0.0-7.5.2
* [Solr](https://github.com/apache/solr/)

456
test/plugin/scenarios/elasticsearch-7.x-scenario/config/expectedData.yaml
View File

@ -14,161 +14,301 @@
# See the License for the specific language governing permissions and
# limitations under the License.
segmentItems:
- serviceName: elasticsearch-7.x-scenario
segmentSize: ge 1
segments:
- segmentId: not null
spans:
- operationName: Elasticsearch/Health
operationId: 0
parentSpanId: 0
spanId: 1
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
skipAnalysis: 'false'
- operationName: Elasticsearch/GetSettings
operationId: 0
parentSpanId: 0
spanId: 2
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
skipAnalysis: 'false'
- operationName: Elasticsearch/PutSettings
operationId: 0
parentSpanId: 0
spanId: 3
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
- {key: db.statement, value: not null}
skipAnalysis: 'false'
- operationName: Elasticsearch/CreateRequest
operationId: 0
parentSpanId: 0
spanId: 4
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
- {key: db.instance, value: not null}
- {key: db.statement, value: not null}
skipAnalysis: 'false'
- operationName: Elasticsearch/IndexRequest
operationId: 0
parentSpanId: 0
spanId: 5
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
- {key: db.instance, value: not null}
- {key: db.statement, value: not null}
skipAnalysis: 'false'
- operationName: Elasticsearch/GetRequest
operationId: 0
parentSpanId: 0
spanId: 6
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
- {key: db.instance, value: not null}
- {key: db.statement, value: not null}
skipAnalysis: 'false'
- operationName: Elasticsearch/SearchRequest
operationId: 0
parentSpanId: 0
spanId: 7
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
- {key: db.instance, value: not null}
- {key: db.statement, value: not null}
skipAnalysis: 'false'
- operationName: Elasticsearch/UpdateRequest
operationId: 0
parentSpanId: 0
spanId: 8
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
- {key: db.instance, value: not null}
- {key: db.statement, value: not null}
skipAnalysis: 'false'
- operationName: Elasticsearch/DeleteRequest
operationId: 0
parentSpanId: 0
spanId: 9
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- {key: db.type, value: Elasticsearch}
- {key: db.instance, value: not null}
skipAnalysis: 'false'
- operationName: /elasticsearch-case/case/elasticsearch
operationId: 0
parentSpanId: -1
spanId: 0
spanLayer: Http
startTime: nq 0
endTime: nq 0
componentId: 1
isError: false
spanType: Entry
peer: ''
tags:
- {key: url, value: 'http://localhost:8080/elasticsearch-case/case/elasticsearch'}
- {key: http.method, value: GET}
skipAnalysis: 'false'
- serviceName: elasticsearch-7.x-scenario
segmentSize: ge 1
segments:
- segmentId: not null
spans:
- operationName: Elasticsearch/Health
operationId: 0
parentSpanId: 0
spanId: 1
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
skipAnalysis: 'false'
- operationName: Elasticsearch/GetSettings
operationId: 0
parentSpanId: 0
spanId: 2
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
skipAnalysis: 'false'
- operationName: Elasticsearch/PutSettings
operationId: 0
parentSpanId: 0
spanId: 3
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/CreateRequest
operationId: 0
parentSpanId: 0
spanId: 4
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/IndexRequest
operationId: 0
parentSpanId: 0
spanId: 5
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/GetRequest
operationId: 0
parentSpanId: 0
spanId: 6
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/SearchRequest
operationId: 0
parentSpanId: 0
spanId: 7
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/UpdateRequest
operationId: 0
parentSpanId: 0
spanId: 8
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/DeleteRequest
operationId: 0
parentSpanId: 0
spanId: 9
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 77
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/IndexRequest
operationId: 0
parentSpanId: 0
spanId: 10
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 48
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: node.address, value: not null }
- { key: es.indices, value: not null }
- { key: es.types, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/actionGet
operationId: 0
parentSpanId: 0
spanId: 11
startTime: nq 0
endTime: nq 0
componentId: 48
isError: false
spanType: Local
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/GetRequest
operationId: 0
parentSpanId: 0
spanId: 12
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 48
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: node.address, value: not null }
- { key: es.indices, value: not null }
- { key: es.types, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/SearchRequest
operationId: 0
parentSpanId: 0
spanId: 13
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 48
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: node.address, value: not null }
- { key: es.indices, value: not null }
- { key: es.types, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/actionGet
operationId: 0
parentSpanId: 0
spanId: 14
startTime: nq 0
endTime: nq 0
componentId: 48
isError: false
spanType: Local
tags:
- {key: db.type, value: Elasticsearch}
- {key: es.took_millis, value: not null }
- {key: es.total_hits, value: not null }
- {key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/UpdateRequest
operationId: 0
parentSpanId: 0
spanId: 15
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 48
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: node.address, value: not null }
- { key: es.indices, value: not null }
- { key: es.types, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/DeleteRequest
operationId: 0
parentSpanId: 0
spanId: 16
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 48
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: node.address, value: not null }
- { key: es.indices, value: not null }
- { key: es.types, value: not null }
- { key: db.statement, value: not null }
skipAnalysis: 'false'
- operationName: Elasticsearch/DeleteIndexRequest
operationId: 0
parentSpanId: 0
spanId: 17
spanLayer: Database
startTime: nq 0
endTime: nq 0
componentId: 48
isError: false
spanType: Exit
peer: not null
tags:
- { key: db.type, value: Elasticsearch }
- { key: db.instance, value: not null }
- { key: node.address, value: not null }
- { key: es.indices, value: not null }
skipAnalysis: 'false'
- operationName: /elasticsearch-case/case/elasticsearch
operationId: 0
parentSpanId: -1
spanId: 0
spanLayer: Http
startTime: nq 0
endTime: nq 0
componentId: 1
isError: false
spanType: Entry
peer: ''
tags:
- { key: url, value: 'http://localhost:8080/elasticsearch-case/case/elasticsearch' }
- { key: http.method, value: GET }
skipAnalysis: 'false'

12
test/plugin/scenarios/elasticsearch-7.x-scenario/pom.xml
View File

@ -98,6 +98,18 @@
<artifactId>elasticsearch</artifactId>
<version>${test.framework.version}</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
<version>${test.framework.version}</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.plugin</groupId>
<artifactId>transport-netty4-client</artifactId>
<version>${test.framework.version}</version>
</dependency>
</dependencies>
<build>

270
test/plugin/scenarios/elasticsearch-7.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/elasticsearch/RestHighLevelClientCase.java Normal file
View File

@ -0,0 +1,270 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.testcase.elasticsearch;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.skywalking.apm.testcase.elasticsearch.controller.CaseController;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.settings.ClusterGetSettingsRequest;
import org.elasticsearch.action.admin.cluster.settings.ClusterGetSettingsResponse;
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.indices.recovery.RecoverySettings;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import static java.util.Collections.singletonMap;
@Component
public class RestHighLevelClientCase {
private static final Logger LOGGER = LogManager.getLogger(CaseController.class);
@Autowired
private RestHighLevelClient client;
public String healthCheck() throws Exception {
ClusterHealthRequest request = new ClusterHealthRequest();
request.timeout(TimeValue.timeValueSeconds(10));
request.waitForStatus(ClusterHealthStatus.GREEN);
ClusterHealthResponse response = client.cluster().health(request, RequestOptions.DEFAULT);
if (response.isTimedOut()) {
String message = "elastic search node start fail!";
LOGGER.error(message);
throw new RuntimeException(message);
}
return "Success";
}
public String elasticsearch() throws Exception {
String indexName = UUID.randomUUID().toString();
try {
// health
health();
// get settings
getSettings();
// put settings
putSettings();
// create
createIndex(indexName);
// index
index(indexName);
client.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
// get
get(indexName);
// search
search(indexName);
// update
update(indexName);
// delete
delete(indexName);
} finally {
if (null != client) {
client.close();
}
}
return "Success";
}
private void health() throws IOException {
ClusterHealthRequest request = new ClusterHealthRequest();
ClusterHealthResponse response = client.cluster().health(request, RequestOptions.DEFAULT);
if (response.isTimedOut()) {
String message = "elastic search health fail!";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void putSettings() throws IOException {
ClusterUpdateSettingsRequest request = new ClusterUpdateSettingsRequest();
String transientSettingKey =
RecoverySettings.INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING.getKey();
int transientSettingValue = 10;
Settings transientSettings = Settings.builder()
.put(transientSettingKey, transientSettingValue, ByteSizeUnit.BYTES)
.build();
request.transientSettings(transientSettings);
ClusterUpdateSettingsResponse response = client.cluster().putSettings(request, RequestOptions.DEFAULT);
if (response == null) {
String message = "elasticsearch put settings fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void getSettings() throws IOException {
ClusterGetSettingsResponse response = client.cluster()
.getSettings(
new ClusterGetSettingsRequest(),
RequestOptions.DEFAULT
);
if (response == null) {
String message = "elasticsearch get settings fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void createIndex(String indexName) throws IOException {
CreateIndexRequest request = new CreateIndexRequest(indexName);
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();
{
builder.startObject("properties");
{
builder.startObject("author");
{
builder.field("type", "keyword");
}
builder.endObject();
builder.startObject("title");
{
builder.field("type", "keyword");
}
builder.endObject();
}
builder.endObject();
}
builder.endObject();
request.mapping(builder);
request.settings(Settings.builder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0));
CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
if (!createIndexResponse.isAcknowledged()) {
String message = "elasticsearch create index fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void index(String indexName) throws IOException {
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();
{
builder.field("author", "Marker");
builder.field("title", "Java programing.");
}
builder.endObject();
IndexRequest indexRequest = new IndexRequest(indexName).id("1").source(builder);
IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
if (indexResponse.status().getStatus() >= 400) {
String message = "elasticsearch index data fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void get(String indexName) throws IOException {
GetRequest getRequest = new GetRequest(indexName, "1");
GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT);
if (!getResponse.isExists()) {
String message = "elasticsearch get data fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void update(String indexName) throws IOException {
UpdateRequest request = new UpdateRequest(indexName, "1");
Map<String, Object> parameters = singletonMap("title", "c++ programing.");
Script inline = new Script(ScriptType.INLINE, "painless", "ctx._source.title = params.title", parameters);
request.script(inline);
UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
if (updateResponse.getVersion() != 2) {
String message = "elasticsearch update data fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void delete(String indexName) throws IOException {
DeleteIndexRequest request = new DeleteIndexRequest(indexName);
AcknowledgedResponse deleteIndexResponse = client.indices().delete(request, RequestOptions.DEFAULT);
if (!deleteIndexResponse.isAcknowledged()) {
String message = "elasticsearch delete index fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void search(String indexName) throws IOException {
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.query(QueryBuilders.termQuery("author", "Marker"));
sourceBuilder.from(0);
sourceBuilder.size(10);
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices(indexName);
searchRequest.source(sourceBuilder);
SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
int length = searchResponse.getHits().getHits().length;
if (!(length > 0)) {
String message = "elasticsearch search data fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
}

100
test/plugin/scenarios/elasticsearch-7.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/elasticsearch/TransportClientCase.java Normal file
View File

@ -0,0 +1,100 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.testcase.elasticsearch;
import java.io.IOException;
import java.util.UUID;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class TransportClientCase {
private static final Logger LOGGER = LogManager.getLogger(TransportClientCase.class);
@Autowired
private TransportClient client;
public boolean elasticsearch() throws Exception {
String indexName = UUID.randomUUID().toString();
try {
// create
index(client, indexName);
// get
get(client, indexName);
// search
search(client, indexName);
// update
update(client, indexName);
// delete
delete(client, indexName);
// remove index
client.admin().indices().prepareDelete(indexName).execute();
} finally {
if (null != client) {
client.close();
}
}
return true;
}
private void index(Client client, String indexName) throws IOException {
try {
client.prepareIndex(indexName, "test", "1")
.setSource(XContentFactory.jsonBuilder()
.startObject()
.field("name", "mysql innodb")
.field("price", "0")
.field("language", "chinese")
.endObject())
.get();
} catch (IOException e) {
LOGGER.error("index document error.", e);
throw e;
}
}
private void get(Client client, String indexName) {
client.prepareGet().setIndex(indexName).setId("1").execute();
}
private void update(Client client, String indexName) throws IOException {
try {
client.prepareUpdate(indexName, "test", "1")
.setDoc(XContentFactory.jsonBuilder().startObject().field("price", "9.9").endObject())
.execute();
} catch (IOException e) {
LOGGER.error("update document error.", e);
throw e;
}
}
private void delete(Client client, String indexName) {
client.prepareDelete(indexName, "test", "1").execute();
}
private void search(Client client, String indexName) {
client.prepareSearch(indexName).setTypes("test").setSize(10).execute().actionGet();
}
}

71
test/plugin/scenarios/elasticsearch-7.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/elasticsearch/config/TransportClientConfig.java Normal file
View File

@ -0,0 +1,71 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.testcase.elasticsearch.config;
import java.net.InetAddress;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class TransportClientConfig {
@Value("${elasticsearch.server}")
private String elasticsearchHost;
public final static Integer PORT = 9300; //port
@Bean
public TransportClient getESClientConnection()
throws Exception {
TransportClient client = null;
Settings settings = Settings.builder()
.put("cluster.name", "docker-node")
.put("client.transport.sniff", false)
.build();
client = new PreBuiltTransportClient(settings);
for (TransportAddress i : parseEsHost()) {
client.addTransportAddress(i);
}
return client;
}
private TransportAddress[] parseEsHost()
throws Exception {
TransportAddress[] transportAddresses = null;
if (!elasticsearchHost.isEmpty()) {
String[] hostIp = elasticsearchHost.split(",");
transportAddresses = new TransportAddress[hostIp.length];
for (int i = 0; i < hostIp.length; ++i) {
String[] hostIpItem = hostIp[i].split(":");
String ip = hostIpItem[0].trim();
String port = hostIpItem[1].trim();
transportAddresses[i] = new TransportAddress(InetAddress.getByName(ip), PORT);
}
}
return transportAddresses;
}
}

237
test/plugin/scenarios/elasticsearch-7.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/elasticsearch/controller/CaseController.java
View File

@ -18,254 +18,39 @@
package org.apache.skywalking.apm.testcase.elasticsearch.controller;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.settings.ClusterGetSettingsRequest;
import org.elasticsearch.action.admin.cluster.settings.ClusterGetSettingsResponse;
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.indices.recovery.RecoverySettings;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.apache.skywalking.apm.testcase.elasticsearch.RestHighLevelClientCase;
import org.apache.skywalking.apm.testcase.elasticsearch.TransportClientCase;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import static java.util.Collections.singletonMap;
@RestController
@RequestMapping("/case")
@RequestMapping("/elasticsearch-case/case")
public class CaseController {
private static final Logger LOGGER = LogManager.getLogger(CaseController.class);
@Autowired
private RestHighLevelClient client;
private RestHighLevelClientCase restHighLevelClientCase;
@Autowired
private TransportClientCase transportClientCase;
@GetMapping("/healthCheck")
public String healthCheck() throws Exception {
ClusterHealthRequest request = new ClusterHealthRequest();
request.timeout(TimeValue.timeValueSeconds(10));
request.waitForStatus(ClusterHealthStatus.GREEN);
ClusterHealthResponse response = client.cluster().health(request, RequestOptions.DEFAULT);
if (response.isTimedOut()) {
String message = "elastic search node start fail!";
LOGGER.error(message);
throw new RuntimeException(message);
}
public String healthcheck() throws Exception {
restHighLevelClientCase.healthCheck();
return "Success";
}
@GetMapping("/elasticsearch")
public String elasticsearch() throws Exception {
String indexName = UUID.randomUUID().toString();
try {
// health
health();
// get settings
getSettings();
// put settings
putSettings();
// create
createIndex(indexName);
// index
index(indexName);
restHighLevelClientCase.elasticsearch();
transportClientCase.elasticsearch();
client.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
// get
get(indexName);
// search
search(indexName);
// update
update(indexName);
// delete
delete(indexName);
} finally {
if (null != client) {
client.close();
}
}
return "Success";
}
private void health() throws IOException {
ClusterHealthRequest request = new ClusterHealthRequest();
ClusterHealthResponse response = client.cluster().health(request, RequestOptions.DEFAULT);
if (response.isTimedOut()) {
String message = "elastic search health fail!";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void putSettings() throws IOException {
ClusterUpdateSettingsRequest request = new ClusterUpdateSettingsRequest();
String transientSettingKey =
RecoverySettings.INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING.getKey();
int transientSettingValue = 10;
Settings transientSettings = Settings.builder()
.put(transientSettingKey, transientSettingValue, ByteSizeUnit.BYTES)
.build();
request.transientSettings(transientSettings);
ClusterUpdateSettingsResponse response = client.cluster().putSettings(request, RequestOptions.DEFAULT);
if (response == null) {
String message = "elasticsearch put settings fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void getSettings() throws IOException {
ClusterGetSettingsResponse response = client.cluster().getSettings(new ClusterGetSettingsRequest(), RequestOptions.DEFAULT);
if (response == null) {
String message = "elasticsearch get settings fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void createIndex(String indexName) throws IOException {
CreateIndexRequest request = new CreateIndexRequest(indexName);
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();
{
builder.startObject("properties");
{
builder.startObject("author");
{
builder.field("type", "keyword");
}
builder.endObject();
builder.startObject("title");
{
builder.field("type", "keyword");
}
builder.endObject();
}
builder.endObject();
}
builder.endObject();
request.mapping(builder);
request.settings(Settings.builder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0));
CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
if (!createIndexResponse.isAcknowledged()) {
String message = "elasticsearch create index fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void index(String indexName) throws IOException {
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();
{
builder.field("author", "Marker");
builder.field("title", "Java programing.");
}
builder.endObject();
IndexRequest indexRequest = new IndexRequest(indexName).id("1").source(builder);
IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
if (indexResponse.status().getStatus() >= 400) {
String message = "elasticsearch index data fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void get(String indexName) throws IOException {
GetRequest getRequest = new GetRequest(indexName, "1");
GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT);
if (!getResponse.isExists()) {
String message = "elasticsearch get data fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void update(String indexName) throws IOException {
UpdateRequest request = new UpdateRequest(indexName, "1");
Map<String, Object> parameters = singletonMap("title", "c++ programing.");
Script inline = new Script(ScriptType.INLINE, "painless", "ctx._source.title = params.title", parameters);
request.script(inline);
UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
if (updateResponse.getVersion() != 2) {
String message = "elasticsearch update data fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void delete(String indexName) throws IOException {
DeleteIndexRequest request = new DeleteIndexRequest(indexName);
AcknowledgedResponse deleteIndexResponse = client.indices().delete(request, RequestOptions.DEFAULT);
if (!deleteIndexResponse.isAcknowledged()) {
String message = "elasticsearch delete index fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
private void search(String indexName) throws IOException {
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.query(QueryBuilders.termQuery("author", "Marker"));
sourceBuilder.from(0);
sourceBuilder.size(10);
SearchRequest searchRequest = new SearchRequest();
searchRequest.indices(indexName);
searchRequest.source(sourceBuilder);
SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
int length = searchResponse.getHits().getHits().length;
if (!(length > 0)) {
String message = "elasticsearch search data fail.";
LOGGER.error(message);
throw new RuntimeException(message);
}
}
}

4
test/plugin/scenarios/elasticsearch-7.x-scenario/src/main/resources/application.yaml
View File

@ -17,7 +17,5 @@
#
server:
port: 8080
servlet:
context-path: /elasticsearch-case
logging:
config: classpath:log4j2.xml
config: classpath:log4j2.xml