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

Merge pull request #252 from wu-sheng/feature/high-performance-agent 

High performance agent, and the skeleton of modulization collector
This commit is contained in:
吴晟 Wu Sheng 2017-07-17 10:33:28 -05:00 committed by GitHub
parent 75bc0beaf4 3ef364d038
commit 04fbe180f3
900 changed files with 14062 additions and 29532 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
README.md
View File

@ -32,13 +32,15 @@ ___
# Contributors
_In chronological order_
* 吴晟 [**Project Maintainer**] [@wu-sheng](https://github.com/wu-sheng) Principle Engineer, 2012 Lab, Huawei.
* 张鑫 [**Project Maintainer**] [@ascrutae](https://github.com/ascrutae)
* 吴晟 [**PMC Member**] [@wu-sheng](https://github.com/wu-sheng) Principle Engineer, 2012 Lab, Huawei.
* 张鑫 [**PMC Member**] [@ascrutae](https://github.com/ascrutae)
* 谭真 [@mircoteam](https://github.com/mircoteam) Advanced R&D Engineers, Creative & Interactive Group.
* 徐妍 [@TastySummer](https://github.com/TastySummer)
* 彭勇升 [**Project Maintainer**] [@pengys5](https://github.com/pengys5) Technical Specialist, OneAPM.
* 彭勇升 [**PMC Member**] [@pengys5](https://github.com/pengys5) Technical Specialist, OneAPM.
* 戴文
* 柏杨 [@bai-yang](https://github.com/bai-yang) Senior Engineer, Alibaba Group.
* 陈凤 [@trey03](https://github.com/trey03)
* [More contributors](https://github.com/wu-sheng/sky-walking/graphs/contributors)
This project adheres to the Contributor Covenant [code of conduct](CODE_OF_CONDUCT.md). By participating, you are expected to uphold this code. Please report unacceptable behavior to wu.sheng@foxmail.com.

4
apm-application-toolkit/apm-toolkit-opentracing/pom.xml
View File

@ -15,12 +15,12 @@
<dependency>
<groupId>io.opentracing</groupId>
<artifactId>opentracing-api</artifactId>
<version>0.20.4</version>
<version>0.30.0</version>
</dependency>
<dependency>
<groupId>io.opentracing</groupId>
<artifactId>opentracing-noop</artifactId>
<version>0.20.4</version>
<version>0.30.0</version>
</dependency>
</dependencies>

18
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/NeedSnifferActivation.java Normal file
View File

@ -0,0 +1,18 @@
package org.skywalking.apm.toolkit.opentracing;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* The <code>NeedSnifferActivation</code> annotation is flag for reader and maintainers,
* which represents this method should be activated/intercepted in sniffer.
*
* @author wusheng
*/
@Target({ElementType.METHOD, ElementType.CONSTRUCTOR})
@Retention(RetentionPolicy.SOURCE)
public @interface NeedSnifferActivation {
String value() default "What should interceptor do?";
}

113
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkyWalkingSpan.java
View File

@ -1,113 +0,0 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import java.util.HashMap;
import java.util.Map;
/**
* Created by wusheng on 2016/12/20.
*/
public class SkyWalkingSpan implements Span, SpanContext {
private String operationName;
private long startTime;
private Map<String, String> tags;
private final Map<String, String> baggageItems;
SkyWalkingSpan(String operationName, long startTime, Map<String, String> tags) {
this.operationName = operationName;
this.startTime = startTime;
this.tags = tags;
baggageItems = new HashMap<String, String>();
}
@Override
public SpanContext context() {
return this;
}
@Override
public void finish() {
}
@Override
public void finish(long finishMicros) {
}
@Override
public void close() {
}
@Override
public Span setTag(String key, String value) {
return this;
}
@Override
public Span setTag(String key, boolean value) {
return this;
}
@Override
public Span setTag(String key, Number value) {
return this;
}
@Override
public Span log(Map<String, ?> fields) {
return this;
}
@Override
public Span log(long timestampMicroseconds, Map<String, ?> fields) {
return this;
}
@Override
public Span log(String event) {
return this;
}
@Override
public Span log(long timestampMicroseconds, String event) {
return this;
}
@Override
public Span setBaggageItem(String key, String value) {
baggageItems.put(key, value);
return this;
}
@Override
public String getBaggageItem(String key) {
return baggageItems.get(key);
}
@Override
public Span setOperationName(String operationName) {
return this;
}
@Override
public Span log(String eventName, Object payload) {
return this;
}
@Override
public Span log(long timestampMicroseconds, String eventName, Object payload) {
return this;
}
@Override
public Iterable<Map.Entry<String, String>> baggageItems() {
return baggageItems.entrySet();
}
}

110
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkyWalkingSpanBuilder.java
View File

@ -1,110 +0,0 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.References;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import io.opentracing.Tracer;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
* All source code in SkyWalkingSpanBuilder acts like an NoopSpanBuilder.
* Actually, it is NOT.
* The whole logic will be added after toolkit-activation.
* <p>
* Created by wusheng on 2016/12/20.
*/
public class SkyWalkingSpanBuilder implements Tracer.SpanBuilder {
private String operationName;
private long startTime = 0L;
private final Map<String, String> tags;
private SpanContext parentContext;
SkyWalkingSpanBuilder(String operationName) {
this.operationName = operationName;
this.tags = new HashMap<String, String>();
}
/**
* In SkyWalkingTracer, SpanContext will not be used. Tracer will build reference by itself.
*
* @param spanContext
* @return
*/
@Override
public Tracer.SpanBuilder asChildOf(SpanContext spanContext) {
this.parentContext = spanContext;
return this;
}
/**
* In SkyWalkingTracer, Parent Span will not be used. Tracer will build reference by itself.
*
* @param span
* @return
*/
@Override
public Tracer.SpanBuilder asChildOf(Span span) {
asChildOf(span.context());
return this;
}
@Override
public Tracer.SpanBuilder addReference(String referenceType, SpanContext referencedContext) {
if (referenceType.equals(References.CHILD_OF)) {
return asChildOf(referencedContext);
} else {
return this;
}
}
@Override
public Tracer.SpanBuilder withTag(String key, String value) {
if (key != null && value != null) {
tags.put(key, value);
}
return this;
}
@Override
public Tracer.SpanBuilder withTag(String key, boolean value) {
if (key != null) {
tags.put(key, Boolean.toString(value));
}
return this;
}
@Override
public Tracer.SpanBuilder withTag(String key, Number value) {
if (key != null && value != null) {
tags.put(key, value.toString());
}
return this;
}
@Override
public Tracer.SpanBuilder withStartTimestamp(long startTime) {
this.startTime = startTime;
return this;
}
@Override
public Span start() {
if (startTime == 0) {
startTime = System.currentTimeMillis();
}
return new SkyWalkingSpan(this.operationName, this.startTime, this.tags);
}
@Override
public Iterable<Map.Entry<String, String>> baggageItems() {
return parentContext == null
? Collections.<String, String>emptyMap().entrySet()
: parentContext.baggageItems();
}
}

123
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingActiveSpan.java Normal file
View File

@ -0,0 +1,123 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.ActiveSpan;
import io.opentracing.SpanContext;
import java.util.Map;
/**
* The <code>SkywalkingActiveSpan</code> is an extension of {@link SkywalkingSpan},
* but because of Java inheritance restrict, only can do with a facade mode.
*
* @author wusheng
*/
public class SkywalkingActiveSpan implements ActiveSpan {
private SkywalkingSpan span;
public SkywalkingActiveSpan(SkywalkingSpan span) {
this.span = span;
}
@Override
public void deactivate() {
span.finish();
}
@Override
public void close() {
this.deactivate();
}
@Override
public Continuation capture() {
return new SkywalkingContinuation();
}
@Override
public SpanContext context() {
return span.context();
}
@Override
public ActiveSpan setTag(String key, String value) {
span.setTag(key, value);
return this;
}
@Override
public ActiveSpan setTag(String key, boolean value) {
span.setTag(key, value);
return this;
}
@Override
public ActiveSpan setTag(String key, Number value) {
span.setTag(key, value);
return this;
}
@Override
public ActiveSpan log(Map<String, ?> fields) {
span.log(fields);
return this;
}
@Override
public ActiveSpan log(long timestampMicroseconds, Map<String, ?> fields) {
span.log(timestampMicroseconds, fields);
return this;
}
@Override
public ActiveSpan log(String event) {
span.log(event);
return this;
}
@Override
public ActiveSpan log(long timestampMicroseconds, String event) {
span.log(timestampMicroseconds, event);
return this;
}
/**
* Don't support baggage item.
*/
@Override
public ActiveSpan setBaggageItem(String key, String value) {
return this;
}
/**
* Don't support baggage item.
*
* @return null, always.
*/
@Override
public String getBaggageItem(String key) {
return null;
}
@Override
public ActiveSpan setOperationName(String operationName) {
span.setOperationName(operationName);
return this;
}
/**
* Don't support logging with payload.
*/
@Deprecated
@Override
public ActiveSpan log(String eventName, Object payload) {
return this;
}
/**
* Don't support logging with payload.
*/
@Deprecated
@Override
public ActiveSpan log(long timestampMicroseconds, String eventName, Object payload) {
return this;
}
}

21
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingContext.java Normal file
View File

@ -0,0 +1,21 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.SpanContext;
import java.util.Map;
/**
* Skywalking tracer context based on {@link ThreadLocal} auto mechanism.
*
* @author wusheng
*/
public class SkywalkingContext implements SpanContext {
public static final SkywalkingContext INSTANCE = new SkywalkingContext();
private SkywalkingContext() {
}
@Override
public Iterable<Map.Entry<String, String>> baggageItems() {
return null;
}
}

21
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingContinuation.java Normal file
View File

@ -0,0 +1,21 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.ActiveSpan;
/**
* @author wusheng
*/
public class SkywalkingContinuation implements ActiveSpan.Continuation {
@NeedSnifferActivation("1. ContextManager#capture" +
"2. set ContextSnapshot to the dynamic field")
public SkywalkingContinuation() {
}
@NeedSnifferActivation("1. get ContextSnapshot from the dynamic field" +
"2. ContextManager#continued")
@Override
public ActiveSpan activate() {
SkywalkingSpanBuilder builder = new SkywalkingSpanBuilder("Thread/" + Thread.currentThread().getName());
return builder.startActive();
}
}

128
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingSpan.java Normal file
View File

@ -0,0 +1,128 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import java.util.HashMap;
import java.util.Map;
/**
* @author wusheng
*/
public class SkywalkingSpan implements Span {
@NeedSnifferActivation(
"1.ContextManager#createSpan (Entry,Exit,Local based on builder)." +
"2.set the span reference to the dynamic field of enhanced SkywalkingSpan")
SkywalkingSpan(SkywalkingSpanBuilder builder) {
}
/**
* Create a shell span for {@link SkywalkingTracer#activeSpan()}
*
* @param tracer
*/
@NeedSnifferActivation(
"1. set the span reference to the dynamic field of enhanced SkywalkingSpan")
public SkywalkingSpan(SkywalkingTracer tracer) {
}
@NeedSnifferActivation("Override span's operationName, which has been given at ")
@Override
public Span setOperationName(String operationName) {
return this;
}
@NeedSnifferActivation("AbstractTracingSpan#log(long timestampMicroseconds, Map<String, ?> fields)")
@Override
public Span log(long timestampMicroseconds, Map<String, ?> fields) {
return this;
}
/**
* Stop the active span
*
* @param finishMicros
*/
@NeedSnifferActivation(
"1.ContextManager#stopSpan(AbstractSpan span)" +
"2. The parameter of stop methed is from the dynamic field of enhanced SkywalkingSpan")
@Override
public void finish(long finishMicros) {
}
@Override
public Span log(long timestampMicroseconds, String event) {
Map<String, String> eventMap = new HashMap<String, String>(1);
eventMap.put("event", event);
return log(timestampMicroseconds, eventMap);
}
@Override
public void finish() {
this.finish(System.currentTimeMillis());
}
@Override
public SpanContext context() {
return SkywalkingContext.INSTANCE;
}
@Override public Span setTag(String key, String value) {
return null;
}
@Override public Span setTag(String key, boolean value) {
return null;
}
@Override public Span setTag(String key, Number value) {
return null;
}
@Override
public Span log(Map<String, ?> fields) {
return log(System.currentTimeMillis(), fields);
}
@Override
public Span log(String event) {
return log(System.currentTimeMillis(), event);
}
/**
* Don't support baggage item.
*/
@Override
public Span setBaggageItem(String key, String value) {
return this;
}
/**
* Don't support baggage item.
*
* @return null, always.
*/
@Override
public String getBaggageItem(String key) {
return null;
}
/**
* Don't support logging with payload.
*/
@Deprecated
@Override
public Span log(String eventName, Object payload) {
return this;
}
/**
* Don't support logging with payload.
*/
@Deprecated
@Override
public Span log(long timestampMicroseconds, String eventName, Object payload) {
return this;
}
}

177
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingSpanBuilder.java Normal file
View File

@ -0,0 +1,177 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.ActiveSpan;
import io.opentracing.BaseSpan;
import io.opentracing.References;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import io.opentracing.Tracer;
import io.opentracing.tag.Tags;
import java.util.LinkedList;
import java.util.List;
/**
* @author wusheng
*/
public class SkywalkingSpanBuilder implements Tracer.SpanBuilder {
private List<Tag> tags = new LinkedList<Tag>();
private String operationName;
private boolean isEntry = false;
private boolean isExit = false;
private int port;
private String peer;
private String componentName;
private boolean isError = false;
private long startTime;
public SkywalkingSpanBuilder(String operationName) {
this.operationName = operationName;
}
@Override
public Tracer.SpanBuilder asChildOf(SpanContext parent) {
if (parent instanceof SkywalkingContext) {
return this;
}
throw new IllegalArgumentException("parent must be type of SpanContext");
}
@Override
public Tracer.SpanBuilder asChildOf(BaseSpan<?> parent) {
if (parent instanceof SkywalkingSpan || parent instanceof SkywalkingActiveSpan) {
return this;
}
throw new IllegalArgumentException("parent must be type of SkywalkingSpan");
}
/**
* Ignore the reference type. the span always the entry or has a parent span.
*
* @param referenceType
* @param referencedContext
* @return
*/
@Override
public Tracer.SpanBuilder addReference(String referenceType, SpanContext referencedContext) {
if (References.FOLLOWS_FROM.equals(referenceType)) {
throw new IllegalArgumentException("only support CHILD_OF reference");
}
return asChildOf(referencedContext);
}
@Override
public Tracer.SpanBuilder withTag(String key, String value) {
if (Tags.COMPONENT.getKey().equals(key)) {
componentName = value;
} else if (Tags.SPAN_KIND.getKey().equals(key)) {
if (Tags.SPAN_KIND_CLIENT.equals(value) || Tags.SPAN_KIND_PRODUCER.equals(value)) {
isEntry = false;
isExit = true;
} else if (Tags.SPAN_KIND_SERVER.equals(value) || Tags.SPAN_KIND_CONSUMER.equals(value)) {
isEntry = true;
isExit = false;
} else {
isEntry = false;
isExit = false;
}
} else if (Tags.PEER_HOST_IPV4.getKey().equals(key) || Tags.PEER_HOST_IPV6.getKey().equals(key)
|| Tags.PEER_HOSTNAME.getKey().equals(key)) {
peer = value;
} else if (Tags.PEER_SERVICE.getKey().equals(key)) {
operationName = value;
} else {
tags.add(new Tag(key, value));
}
return this;
}
@Override
public Tracer.SpanBuilder withTag(String key, boolean value) {
if (Tags.ERROR.equals(key)) {
isError = value;
} else {
tags.add(new Tag(key, value ? "true" : "false"));
}
return this;
}
@Override
public Tracer.SpanBuilder withTag(String key, Number value) {
if (Tags.PEER_PORT.getKey().equals(key)) {
port = value.intValue();
} else {
tags.add(new Tag(key, value.toString()));
}
return this;
}
@Override
public Tracer.SpanBuilder withStartTimestamp(long microseconds) {
startTime = microseconds;
return this;
}
@Override
public ActiveSpan startActive() {
return new SkywalkingActiveSpan(new SkywalkingSpan(this));
}
@Override
public Span startManual() {
return new SkywalkingSpan(this);
}
@Override
@Deprecated
public Span start() {
return startManual();
}
/**
* All the get methods are for accessing data from activation
*/
public List<Tag> getTags() {
return tags;
}
public String getOperationName() {
return operationName;
}
public boolean isEntry() {
return isEntry;
}
public boolean isExit() {
return isExit;
}
public int getPort() {
return port;
}
public String getPeer() {
return peer;
}
public String getComponentName() {
return componentName;
}
public boolean isError() {
return isError;
}
public long getStartTime() {
return startTime;
}
/**
* All the following methods are needed for activation.
*/
@Override
@NeedSnifferActivation("Stop the active span.")
public Tracer.SpanBuilder ignoreActiveSpan() {
return this;
}
}

83
apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkyWalkingTracer.java → apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/SkywalkingTracer.java
View File

@ -1,76 +1,84 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.ActiveSpan;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import io.opentracing.Tracer;
import io.opentracing.propagation.Format;
import io.opentracing.propagation.TextMap;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Map;
/**
* All source code in SkyWalkingTracer acts like an NoopTracer.
* Actually, it is NOT.
* The whole logic will be added after toolkit-activation.
* <p>
* Created by wusheng on 2016/12/20.
* @author wusheng
*/
public class SkyWalkingTracer implements Tracer {
public class SkywalkingTracer implements Tracer {
private static String TRACE_HEAD_NAME = "sw3";
public static Tracer INSTANCE = new SkyWalkingTracer();
@NeedSnifferActivation("1. ContextManager#inject" +
"2. ContextCarrier#serialize")
private String inject() {
return null;
}
@NeedSnifferActivation("1. ContextCarrier#deserialize" +
"2. ContextManager#extract")
private void extract(String carrier) {
}
@Override
public SpanBuilder buildSpan(String operationName) {
return new SkyWalkingSpanBuilder(operationName);
return new SkywalkingSpanBuilder(operationName);
}
@Override
public <C> void inject(SpanContext spanContext, Format<C> format, C carrier) {
if (Format.Builtin.TEXT_MAP.equals(format) || Format.Builtin.HTTP_HEADERS.equals(format)) {
((TextMap) carrier).put(TRACE_HEAD_NAME, formatInjectCrossProcessPropagationContextData());
((TextMap)carrier).put(TRACE_HEAD_NAME, inject());
} else if (Format.Builtin.BINARY.equals(format)) {
byte[] key = TRACE_HEAD_NAME.getBytes(ByteBufferContext.CHARSET);
byte[] value = formatInjectCrossProcessPropagationContextData().getBytes(ByteBufferContext.CHARSET);
((ByteBuffer) carrier).put(key);
((ByteBuffer) carrier).putInt(value.length);
((ByteBuffer) carrier).put(value);
byte[] value = inject().getBytes(ByteBufferContext.CHARSET);
((ByteBuffer)carrier).put(key);
((ByteBuffer)carrier).putInt(value.length);
((ByteBuffer)carrier).put(value);
} else {
throw new IllegalArgumentException("Unsupported format: " + format);
}
}
@Override
public <C> SpanContext extract(Format<C> format, C carrier) {
if (Format.Builtin.TEXT_MAP.equals(format) || Format.Builtin.HTTP_HEADERS.equals(format)) {
TextMap textMapCarrier = (TextMap)carrier;
extract(fetchContextData(textMapCarrier));
return new TextMapContext(textMapCarrier);
} else if (Format.Builtin.BINARY.equals(format)) {
ByteBuffer byteBufferCarrier = (ByteBuffer)carrier;
extract(fetchContextData(byteBufferCarrier));
return new ByteBufferContext((ByteBuffer)carrier);
} else {
throw new IllegalArgumentException("Unsupported format: " + format);
}
}
@Override
public <C> SpanContext extract(Format<C> format, C carrier) {
if (Format.Builtin.TEXT_MAP.equals(format) || Format.Builtin.HTTP_HEADERS.equals(format)) {
TextMap textMapCarrier = (TextMap) carrier;
formatExtractCrossProcessPropagationContextData(fetchContextData(textMapCarrier));
return new TextMapContext(textMapCarrier);
} else if (Format.Builtin.BINARY.equals(format)) {
ByteBuffer byteBufferCarrier = (ByteBuffer) carrier;
formatExtractCrossProcessPropagationContextData(fetchContextData(byteBufferCarrier));
return new ByteBufferContext((ByteBuffer) carrier);
public ActiveSpan activeSpan() {
return new SkywalkingActiveSpan(new SkywalkingSpan(this));
}
@Override
public ActiveSpan makeActive(Span span) {
if (span instanceof SkywalkingSpan) {
return new SkywalkingActiveSpan((SkywalkingSpan)span);
} else {
throw new IllegalArgumentException("Unsupported format: " + format);
throw new IllegalArgumentException("span must be a type of SkywalkingSpan");
}
}
/**
* set context data in toolkit-opentracing-activation
*/
private String formatInjectCrossProcessPropagationContextData() {
return "";
}
/**
* read context data in toolkit-opentracing-activation
*/
private void formatExtractCrossProcessPropagationContextData(String contextData) {
}
private String fetchContextData(TextMap textMap) {
Iterator<Map.Entry<String, String>> iterator = textMap.iterator();
while (iterator.hasNext()) {
@ -99,4 +107,5 @@ public class SkyWalkingTracer implements Tracer {
return null;
}
}
}

8
apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/tag/StringTagItem.java → apm-application-toolkit/apm-toolkit-opentracing/src/main/java/org/skywalking/apm/toolkit/opentracing/Tag.java
View File

@ -1,15 +1,13 @@
package org.skywalking.apm.agent.core.context.tag;
package org.skywalking.apm.toolkit.opentracing;
/**
* The tag item with String key and String value.
*
* @author wusheng
*/
public final class StringTagItem {
public class Tag {
private String key;
private String value;
public StringTagItem(String key, String value) {
public Tag(String key, String value) {
this.key = key;
this.value = value;
}

2
apm-application-toolkit/apm-toolkit-opentracing/src/main/resources/META-INF.services/io.opentracing.Tracer
View File

@ -1 +1 @@
SkyWalkingTracer
org.skywalking.apm.toolkit.opentracing.SkyWalkingTracer

39
apm-application-toolkit/apm-toolkit-opentracing/src/test/java/org/skywalking/apm/toolkit/opentracing/SkyWalkingTracerTest.java
View File

@ -1,39 +0,0 @@
package org.skywalking.apm.toolkit.opentracing;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import io.opentracing.Tracer;
import io.opentracing.propagation.TextMap;
import org.junit.Test;
import java.util.Iterator;
import java.util.Map;
/**
* Created by wusheng on 2016/12/21.
*/
public class SkyWalkingTracerTest {
@Test
public void testBuildSpan() {
Tracer tracer = SkyWalkingTracer.INSTANCE;
Tracer.SpanBuilder spanBuilder = tracer.buildSpan("/http/serviceName");
SpanContext context = new TextMapContext(new TextMap() {
@Override
public Iterator<Map.Entry<String, String>> iterator() {
throw new UnsupportedOperationException(
"TextMapInjectAdapter should only be used with Tracer.inject()");
}
@Override
public void put(String key, String value) {
}
});
spanBuilder.asChildOf(context).withTag("example.tag", "testtag");
Span span = spanBuilder.start();
span.finish();
}
}

20
apm-commons/apm-logging-log4j2/pom.xml → apm-collector/apm-collector-agentstream/pom.xml
View File

@ -3,29 +3,25 @@
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-commons</artifactId>
<artifactId>apm-collector</artifactId>
<groupId>org.skywalking</groupId>
<version>3.2-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>apm-logging-log4j2</artifactId>
<artifactId>apm-collector-agentstream</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.skywalking</groupId>
<artifactId>apm-logging-api</artifactId>
<artifactId>apm-collector-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>2.8</version>
<groupId>org.skywalking</groupId>
<artifactId>apm-collector-server</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>
</project>

9
apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCConfig.java Normal file
View File

@ -0,0 +1,9 @@
package org.skywalking.apm.collector.agent.stream.server.grpc;
/**
* @author pengys5
*/
public class AgentStreamGRPCConfig {
public static String HOST;
public static int PORT;
}

29
apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCConfigParser.java Normal file
View File

@ -0,0 +1,29 @@
package org.skywalking.apm.collector.agent.stream.server.grpc;
import java.util.Map;
import org.skywalking.apm.collector.core.config.ConfigParseException;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.util.StringUtils;
/**
* @author pengys5
*/
public class AgentStreamGRPCConfigParser implements ModuleConfigParser {
private static final String HOST = "host";
private static final String PORT = "port";
@Override public void parse(Map config) throws ConfigParseException {
AgentStreamGRPCConfig.HOST = (String)config.get(HOST);
if (StringUtils.isEmpty(AgentStreamGRPCConfig.HOST)) {
throw new ConfigParseException("");
}
if (StringUtils.isEmpty(config.get(PORT))) {
throw new ConfigParseException("");
} else {
AgentStreamGRPCConfig.PORT = (Integer)config.get(PORT);
}
}
}

38
apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCModuleDefine.java Normal file
View File

@ -0,0 +1,38 @@
package org.skywalking.apm.collector.agent.stream.server.grpc;
import org.skywalking.apm.collector.core.agentstream.AgentStreamModuleDefine;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.module.ModuleGroup;
import org.skywalking.apm.collector.core.module.ModuleRegistration;
import org.skywalking.apm.collector.core.server.Server;
import org.skywalking.apm.collector.server.grpc.GRPCServer;
/**
* @author pengys5
*/
public class AgentStreamGRPCModuleDefine extends AgentStreamModuleDefine {
@Override protected ModuleGroup group() {
return ModuleGroup.AgentStream;
}
@Override public String name() {
return "grpc";
}
@Override public boolean defaultModule() {
return true;
}
@Override protected ModuleConfigParser configParser() {
return new AgentStreamGRPCConfigParser();
}
@Override protected Server server() {
return new GRPCServer(AgentStreamGRPCConfig.HOST, AgentStreamGRPCConfig.PORT);
}
@Override protected ModuleRegistration registration() {
return new AgentStreamGRPCModuleRegistration();
}
}

13
apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCModuleRegistration.java Normal file
View File

@ -0,0 +1,13 @@
package org.skywalking.apm.collector.agent.stream.server.grpc;
import org.skywalking.apm.collector.core.module.ModuleRegistration;
/**
* @author pengys5
*/
public class AgentStreamGRPCModuleRegistration extends ModuleRegistration {
@Override public Value buildValue() {
return new Value(AgentStreamGRPCConfig.HOST, AgentStreamGRPCConfig.PORT, null);
}
}

10
apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyConfig.java Normal file
View File

@ -0,0 +1,10 @@
package org.skywalking.apm.collector.agent.stream.server.jetty;
/**
* @author pengys5
*/
public class AgentStreamJettyConfig {
public static String HOST;
public static int PORT;
public static String CONTEXT_PATH;
}

33
apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyConfigParser.java Normal file
View File

@ -0,0 +1,33 @@
package org.skywalking.apm.collector.agent.stream.server.jetty;
import java.util.Map;
import org.skywalking.apm.collector.core.config.ConfigParseException;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.util.StringUtils;
/**
* @author pengys5
*/
public class AgentStreamJettyConfigParser implements ModuleConfigParser {
private static final String HOST = "host";
private static final String PORT = "port";
public static final String CONTEXT_PATH = "contextPath";
@Override public void parse(Map config) throws ConfigParseException {
AgentStreamJettyConfig.HOST = (String)config.get(HOST);
AgentStreamJettyConfig.CONTEXT_PATH = "/";
if (StringUtils.isEmpty(AgentStreamJettyConfig.HOST)) {
throw new ConfigParseException("");
}
if (StringUtils.isEmpty(config.get(PORT))) {
throw new ConfigParseException("");
} else {
AgentStreamJettyConfig.PORT = (Integer)config.get(PORT);
}
if (!StringUtils.isEmpty(config.get(CONTEXT_PATH))) {
AgentStreamJettyConfig.CONTEXT_PATH = (String)config.get(CONTEXT_PATH);
}
}
}

38
apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyModuleDefine.java Normal file
View File

@ -0,0 +1,38 @@
package org.skywalking.apm.collector.agent.stream.server.jetty;
import org.skywalking.apm.collector.core.agentstream.AgentStreamModuleDefine;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.module.ModuleGroup;
import org.skywalking.apm.collector.core.module.ModuleRegistration;
import org.skywalking.apm.collector.core.server.Server;
import org.skywalking.apm.collector.server.jetty.JettyServer;
/**
* @author pengys5
*/
public class AgentStreamJettyModuleDefine extends AgentStreamModuleDefine {
@Override protected ModuleGroup group() {
return ModuleGroup.AgentStream;
}
@Override public String name() {
return "jetty";
}
@Override public boolean defaultModule() {
return false;
}
@Override protected ModuleConfigParser configParser() {
return new AgentStreamJettyConfigParser();
}
@Override protected Server server() {
return new JettyServer(AgentStreamJettyConfig.HOST, AgentStreamJettyConfig.PORT, AgentStreamJettyConfig.CONTEXT_PATH);
}
@Override protected ModuleRegistration registration() {
return new AgentStreamJettyModuleRegistration();
}
}

16
apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyModuleRegistration.java Normal file
View File

@ -0,0 +1,16 @@
package org.skywalking.apm.collector.agent.stream.server.jetty;
import com.google.gson.JsonObject;
import org.skywalking.apm.collector.core.module.ModuleRegistration;
/**
* @author pengys5
*/
public class AgentStreamJettyModuleRegistration extends ModuleRegistration {
@Override public Value buildValue() {
JsonObject data = new JsonObject();
data.addProperty(AgentStreamJettyConfigParser.CONTEXT_PATH, AgentStreamJettyConfig.CONTEXT_PATH);
return new Value(AgentStreamJettyConfig.HOST, AgentStreamJettyConfig.PORT, data);
}
}

2
apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/module.define Normal file
View File

@ -0,0 +1,2 @@
org.skywalking.apm.collector.agent.stream.server.grpc.AgentStreamGRPCModuleDefine
org.skywalking.apm.collector.agent.stream.server.jetty.AgentStreamJettyModuleDefine

37
apm-collector/apm-collector-boot/pom.xml Normal file
View File

@ -0,0 +1,37 @@
<?xml version="1.0" encoding="UTF-8"?>
<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-collector</artifactId>
<groupId>org.skywalking</groupId>
<version>3.2-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>apm-collector-boot</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.skywalking</groupId>
<artifactId>apm-collector-cluster</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.skywalking</groupId>
<artifactId>apm-collector-queue</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.skywalking</groupId>
<artifactId>apm-collector-ui</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.skywalking</groupId>
<artifactId>apm-collector-agentstream</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>

22
apm-collector/apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot/CollectorBootStartUp.java Normal file
View File

@ -0,0 +1,22 @@
package org.skywalking.apm.collector.boot;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.config.ConfigException;
import org.skywalking.apm.collector.core.framework.CollectorStarter;
import org.skywalking.apm.collector.core.framework.DefineException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class CollectorBootStartUp {
private static final Logger logger = LoggerFactory.getLogger(CollectorBootStartUp.class);
public static void main(String[] args) throws ConfigException, DefineException, ClientException {
logger.info("collector starting...");
CollectorStarter starter = new CollectorStarter();
starter.start();
}
}

26
apm-collector/apm-collector-boot/src/main/resources/application.yml Normal file
View File

@ -0,0 +1,26 @@
cluster:
# zookeeper:
# hostPort: localhost:2181
# sessionTimeout: 1000
# redis:
# host: localhost
# port: 6379
queue:
disruptor: on
data_carrier: off
agentstream:
grpc:
host: localhost
port: 1000
jetty:
host: localhost
port: 2000
context_path: /
discovery:
grpc: localhost
port: 1000
ui:
jetty:
host: localhost
port: 12800

37
apm-collector/apm-collector-client/pom.xml Normal file
View File

@ -0,0 +1,37 @@
<?xml version="1.0" encoding="UTF-8"?>
<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-collector</artifactId>
<groupId>org.skywalking</groupId>
<version>3.2-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>apm-collector-client</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.skywalking</groupId>
<artifactId>apm-collector-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<version>1.4.196</version>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.10</version>
</dependency>
</dependencies>
</project>

54
apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/h2/H2Client.java Normal file
View File

@ -0,0 +1,54 @@
package org.skywalking.apm.collector.client.h2;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import org.skywalking.apm.collector.core.client.Client;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class H2Client implements Client {
private final Logger logger = LoggerFactory.getLogger(H2Client.class);
private Connection conn;
@Override public void initialize() throws H2ClientException {
try {
Class.forName("org.h2.Driver");
conn = DriverManager.getConnection("jdbc:h2:mem:collector");
} catch (ClassNotFoundException | SQLException e) {
throw new H2ClientException(e.getMessage(), e);
}
}
public void execute(String sql) throws H2ClientException {
Statement statement = null;
try {
statement = conn.createStatement();
statement.execute(sql);
statement.closeOnCompletion();
} catch (SQLException e) {
throw new H2ClientException(e.getMessage(), e);
}
}
public void executeQuery(String sql) throws H2ClientException {
Statement statement = null;
try {
statement = conn.createStatement();
ResultSet rs = statement.executeQuery(sql);
while (rs.next()) {
logger.debug(rs.getString("ADDRESS") + "," + rs.getString("DATA"));
}
statement.closeOnCompletion();
} catch (SQLException e) {
throw new H2ClientException(e.getMessage(), e);
}
}
}

17
apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/h2/H2ClientException.java Normal file
View File

@ -0,0 +1,17 @@
package org.skywalking.apm.collector.client.h2;
import org.skywalking.apm.collector.core.client.ClientException;
/**
* @author pengys5
*/
public class H2ClientException extends ClientException {
public H2ClientException(String message) {
super(message);
}
public H2ClientException(String message, Throwable cause) {
super(message, cause);
}
}

29
apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/redis/RedisClient.java Normal file
View File

@ -0,0 +1,29 @@
package org.skywalking.apm.collector.client.redis;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import redis.clients.jedis.Jedis;
/**
* @author pengys5
*/
public class RedisClient implements Client {
private Jedis jedis;
private final String host;
private final int port;
public RedisClient(String host, int port) {
this.host = host;
this.port = port;
}
@Override public void initialize() throws ClientException {
jedis = new Jedis(host, port);
}
public void setex(String key, int seconds, String value) {
jedis.setex(key, seconds, value);
}
}

17
apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/redis/RedisClientException.java Normal file
View File

@ -0,0 +1,17 @@
package org.skywalking.apm.collector.client.redis;
import org.skywalking.apm.collector.core.client.ClientException;
/**
* @author pengys5
*/
public class RedisClientException extends ClientException {
public RedisClientException(String message) {
super(message);
}
public RedisClientException(String message, Throwable cause) {
super(message, cause);
}
}

71
apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/zookeeper/ZookeeperClient.java Normal file
View File

@ -0,0 +1,71 @@
package org.skywalking.apm.collector.client.zookeeper;
import java.io.IOException;
import java.util.List;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.skywalking.apm.collector.core.client.Client;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class ZookeeperClient implements Client {
private final Logger logger = LoggerFactory.getLogger(ZookeeperClient.class);
private ZooKeeper zk;
private final String hostPort;
private final int sessionTimeout;
public ZookeeperClient(String hostPort, int sessionTimeout) {
this.hostPort = hostPort;
this.sessionTimeout = sessionTimeout;
}
@Override public void initialize() throws ZookeeperClientException {
try {
zk = new ZooKeeper(hostPort, sessionTimeout, new ZookeeperDataListener(this));
} catch (IOException e) {
throw new ZookeeperClientException(e.getMessage(), e);
}
}
public void create(final String path, byte data[], List<ACL> acl,
CreateMode createMode) throws ZookeeperClientException {
try {
zk.create(path, data, acl, createMode);
} catch (KeeperException | InterruptedException e) {
throw new ZookeeperClientException(e.getMessage(), e);
}
}
public Stat exists(final String path, boolean watch) throws ZookeeperClientException {
try {
return zk.exists(path, watch);
} catch (KeeperException | InterruptedException e) {
throw new ZookeeperClientException(e.getMessage(), e);
}
}
public byte[] getData(String path, boolean watch, Stat stat) throws ZookeeperClientException {
try {
return zk.getData(path, watch, stat);
} catch (KeeperException | InterruptedException e) {
throw new ZookeeperClientException(e.getMessage(), e);
}
}
public Stat setData(final String path, byte data[], int version) throws ZookeeperClientException {
try {
return zk.setData(path, data, version);
} catch (KeeperException | InterruptedException e) {
throw new ZookeeperClientException(e.getMessage(), e);
}
}
}

16
apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/zookeeper/ZookeeperClientException.java Normal file
View File

@ -0,0 +1,16 @@
package org.skywalking.apm.collector.client.zookeeper;
import org.skywalking.apm.collector.core.client.ClientException;
/**
* @author pengys5
*/
public class ZookeeperClientException extends ClientException {
public ZookeeperClientException(String message) {
super(message);
}
public ZookeeperClientException(String message, Throwable cause) {
super(message, cause);
}
}

59
apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/zookeeper/ZookeeperDataListener.java Normal file
View File

@ -0,0 +1,59 @@
package org.skywalking.apm.collector.client.zookeeper;
import java.util.LinkedList;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.client.DataListener;
import org.skywalking.apm.collector.core.cluster.ClusterDataInitializer;
import org.skywalking.apm.collector.core.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class ZookeeperDataListener implements DataListener, Watcher {
private final Logger logger = LoggerFactory.getLogger(ZookeeperDataListener.class);
private ZookeeperClient client;
public ZookeeperDataListener(Client client) {
this.client = (ZookeeperClient)client;
}
@Override public void process(WatchedEvent event) {
logger.debug("path {}", event.getPath());
if (StringUtils.isEmpty(event.getPath())) {
return;
}
try {
String data = String.valueOf(client.getData(event.getPath(), false, null));
logger.debug("data {}", data);
} catch (ClientException e) {
logger.error(e.getMessage(), e);
}
}
@Override public void listen() throws ClientException {
for (String itemKey : items()) {
String[] catalogs = itemKey.split("\\.");
StringBuilder pathBuilder = new StringBuilder();
for (String catalog : catalogs) {
pathBuilder.append("/").append(catalog);
}
client.exists(pathBuilder.toString(), true);
}
}
@Override public List<String> items() {
List<String> items = new LinkedList<>();
items.add(ClusterDataInitializer.FOR_AGENT_CATALOG);
items.add(ClusterDataInitializer.FOR_UI_CATALOG);
return items;
}
}

16
apm-collector/apm-collector-client/src/main/java/org/skywalking/apm/collector/client/zookeeper/util/PathUtils.java Normal file
View File

@ -0,0 +1,16 @@
package org.skywalking.apm.collector.client.zookeeper.util;
/**
* @author pengys5
*/
public class PathUtils {
public static String convertKey2Path(String key) {
String[] keys = key.split("\\.");
StringBuilder pathBuilder = new StringBuilder();
for (String subPath : keys) {
pathBuilder.append("/").append(subPath);
}
return pathBuilder.toString();
}
}

13
apm-collector/apm-collector-cluster/pom.xml
View File

@ -12,19 +12,16 @@
<artifactId>apm-collector-cluster</artifactId>
<packaging>jar</packaging>
<properties>
</properties>
<dependencies>
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.3.6</version>
<groupId>org.skywalking</groupId>
<artifactId>apm-collector-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.skywalking</groupId>
<artifactId>apm-util</artifactId>
<artifactId>apm-collector-client</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>
</project>

50
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/AkkaSystem.java
View File

@ -1,50 +0,0 @@
package org.skywalking.apm.collector;
import akka.actor.ActorSystem;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.skywalking.apm.util.StringUtil;
import org.skywalking.apm.collector.cluster.ClusterConfig;
import org.skywalking.apm.collector.cluster.Const;
/**
* @author pengys5
*/
public enum AkkaSystem {
INSTANCE;
private Logger logger = LogManager.getFormatterLogger(AkkaSystem.class);
public ActorSystem create() {
Level logLevel = logger.getLevel();
final Config config = ConfigFactory.parseString("akka.remote.netty.tcp.HOSTNAME=" + ClusterConfig.Cluster.Current.HOSTNAME).
withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.PORT=" + ClusterConfig.Cluster.Current.PORT)).
withFallback(ConfigFactory.parseString("akka.loggers=[\"akka.event.slf4j.Slf4jLogger\"]")).
withFallback(ConfigFactory.parseString("akka.loglevel=\"" + logLevel.name() + "\"")).
withFallback(ConfigFactory.load("application.conf"));
if (!StringUtil.isEmpty(ClusterConfig.Cluster.SEED_NODES)) {
config.withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=" + generateSeedNodes()));
}
return ActorSystem.create(Const.SYSTEM_NAME, config);
}
private String generateSeedNodes() {
String[] seedNodes = ClusterConfig.Cluster.SEED_NODES.split(",");
String akkaSeedNodes = "";
for (int i = 0; i < seedNodes.length; i++) {
String akkaNodeName = "\"akka.tcp://" + Const.SYSTEM_NAME + "@" + seedNodes[i] + "\"";
if (i > 0) {
akkaSeedNodes += ",";
}
akkaSeedNodes += akkaNodeName;
}
akkaSeedNodes = "[" + akkaSeedNodes + "]";
logger.info("config seedNodes: %s, generate seedNodes: %s", ClusterConfig.Cluster.SEED_NODES, akkaSeedNodes);
return akkaSeedNodes;
}
}

62
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/CollectorSystem.java
View File

@ -1,62 +0,0 @@
package org.skywalking.apm.collector;
import akka.actor.ActorSystem;
import akka.actor.Props;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.skywalking.apm.collector.actor.*;
import org.skywalking.apm.collector.cluster.WorkersListener;
import org.skywalking.apm.collector.config.ConfigInitializer;
import java.io.IOException;
import java.util.ServiceLoader;
/**
* @author pengys5
*/
public class CollectorSystem {
private static final Logger logger = LogManager.getFormatterLogger(CollectorSystem.class);
private ClusterWorkerContext clusterContext;
public LookUp getClusterContext() {
return clusterContext;
}
public void boot() throws UsedRoleNameException, ProviderNotFoundException, IOException, IllegalAccessException {
ConfigInitializer.INSTANCE.initialize();
createAkkaSystem();
createListener();
loadLocalProviders();
createClusterWorkers();
}
private void createAkkaSystem() {
ActorSystem akkaSystem = AkkaSystem.INSTANCE.create();
clusterContext = new ClusterWorkerContext(akkaSystem);
}
private void createListener() {
clusterContext.getAkkaSystem().actorOf(Props.create(WorkersListener.class, clusterContext), WorkersListener.WORK_NAME);
}
private void createClusterWorkers() throws ProviderNotFoundException {
ServiceLoader<AbstractClusterWorkerProvider> clusterServiceLoader = ServiceLoader.load(AbstractClusterWorkerProvider.class);
for (AbstractClusterWorkerProvider provider : clusterServiceLoader) {
logger.info("create {%s} worker using java service loader", provider.workerNum());
provider.setClusterContext(clusterContext);
for (int i = 1; i <= provider.workerNum(); i++) {
provider.create(AbstractWorker.noOwner());
}
}
}
private void loadLocalProviders() throws UsedRoleNameException {
ServiceLoader<AbstractLocalWorkerProvider> clusterServiceLoader = ServiceLoader.load(AbstractLocalWorkerProvider.class);
for (AbstractLocalWorkerProvider provider : clusterServiceLoader) {
logger.info("loadLocalProviders provider name: %s", provider.getClass().getName());
provider.setClusterContext(clusterContext);
clusterContext.putProvider(provider);
}
}
}

113
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/actor/AbstractClusterWorker.java
View File

@ -1,113 +0,0 @@
package org.skywalking.apm.collector.actor;
import akka.actor.UntypedActor;
import akka.cluster.Cluster;
import akka.cluster.ClusterEvent;
import akka.cluster.Member;
import akka.cluster.MemberStatus;
import org.apache.logging.log4j.Logger;
import org.skywalking.apm.collector.cluster.WorkerListenerMessage;
import org.skywalking.apm.collector.cluster.WorkersListener;
import org.skywalking.apm.collector.log.LogManager;
/**
* The <code>AbstractClusterWorker</code> implementations represent workers,
* which receive remote messages.
* <p>
* Usually, the implementations are doing persistent, or aggregate works.
*
* @author pengys5
* @since v3.0-2017
*/
public abstract class AbstractClusterWorker extends AbstractWorker {
/**
* Construct an <code>AbstractClusterWorker</code> with the worker role and context.
*
* @param role If multi-workers are for load balance, they should be more likely called worker instance. Meaning,
* each worker have multi instances.
* @param clusterContext See {@link ClusterWorkerContext}
* @param selfContext See {@link LocalWorkerContext}
*/
protected AbstractClusterWorker(Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
super(role, clusterContext, selfContext);
}
/**
* This method use for message producer to call for send message.
*
* @param message The persistence data or metric data.
* @throws Exception The Exception happen in {@link #onWork(Object)}
*/
final public void allocateJob(Object message) throws WorkerException {
onWork(message);
}
/**
* This method use for message receiver to analyse message.
*
* @param message Cast the message object to a expect subclass.
* @throws Exception Don't handle the exception, throw it.
*/
protected abstract void onWork(Object message) throws WorkerException;
static class WorkerWithAkka extends UntypedActor {
private Logger logger = LogManager.INSTANCE.getFormatterLogger(WorkerWithAkka.class);
private Cluster cluster;
private final AbstractClusterWorker ownerWorker;
public WorkerWithAkka(AbstractClusterWorker ownerWorker) {
this.ownerWorker = ownerWorker;
cluster = Cluster.get(getContext().system());
}
@Override
public void preStart() {
cluster.subscribe(getSelf(), ClusterEvent.MemberUp.class);
}
@Override
public void postStop() {
cluster.unsubscribe(getSelf());
}
/**
* Listening {@link ClusterEvent.MemberUp} and {@link ClusterEvent.CurrentClusterState}
* cluster event, when event send from the member of {@link WorkersListener} then tell
* the sender to register self.
*/
@Override
public void onReceive(Object message) throws WorkerException {
if (message instanceof ClusterEvent.CurrentClusterState) {
ClusterEvent.CurrentClusterState state = (ClusterEvent.CurrentClusterState)message;
for (Member member : state.getMembers()) {
if (member.status().equals(MemberStatus.up())) {
register(member);
}
}
} else if (message instanceof ClusterEvent.MemberUp) {
ClusterEvent.MemberUp memberUp = (ClusterEvent.MemberUp)message;
logger.info("receive ClusterEvent.MemberUp message, address: %s", memberUp.member().address().toString());
register(memberUp.member());
} else {
logger.debug("worker class: %s, message class: %s", this.getClass().getName(), message.getClass().getName());
ownerWorker.allocateJob(message);
}
}
/**
* When member role is {@link WorkersListener#WORK_NAME} then Select actor from context
* and send register message to {@link WorkersListener}
*
* @param member is the new created or restart worker
*/
void register(Member member) {
if (member.hasRole(WorkersListener.WORK_NAME)) {
WorkerListenerMessage.RegisterMessage registerMessage = new WorkerListenerMessage.RegisterMessage(ownerWorker.getRole());
logger.info("member address: %s, worker path: %s", member.address().toString(), getSelf().path().toString());
getContext().actorSelection(member.address() + "/user/" + WorkersListener.WORK_NAME).tell(registerMessage, getSelf());
}
}
}
}

103
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/actor/AbstractLocalAsyncWorker.java
View File

@ -1,103 +0,0 @@
package org.skywalking.apm.collector.actor;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import org.skywalking.apm.collector.queue.EndOfBatchCommand;
import org.skywalking.apm.collector.queue.MessageHolder;
/**
* The <code>AbstractLocalAsyncWorker</code> implementations represent workers,
* which receive local asynchronous message.
*
* @author pengys5
* @since v3.0-2017
*/
public abstract class AbstractLocalAsyncWorker extends AbstractLocalWorker {
/**
* Construct an <code>AbstractLocalAsyncWorker</code> with the worker role and context.
*
* @param role The responsibility of worker in cluster, more than one workers can have same responsibility which use
* to provide load balancing ability.
* @param clusterContext See {@link ClusterWorkerContext}
* @param selfContext See {@link LocalWorkerContext}
*/
public AbstractLocalAsyncWorker(Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
super(role, clusterContext, selfContext);
}
/**
* The asynchronous worker always use to persistence data into db, this is the end of the streaming,
* so usually no need to create the next worker instance at the time of this worker instance create.
*
* @throws ProviderNotFoundException When worker provider not found, it will be throw this exception.
*/
@Override
public void preStart() throws ProviderNotFoundException {
}
/**
* Receive message
*
* @param message The persistence data or metric data.
* @throws WorkerException The Exception happen in {@link #onWork(Object)}
*/
final public void allocateJob(Object message) throws WorkerException {
onWork(message);
}
/**
* The data process logic in this method.
*
* @param message Cast the message object to a expect subclass.
* @throws WorkerException Don't handle the exception, throw it.
*/
protected abstract void onWork(Object message) throws WorkerException;
static class WorkerWithDisruptor implements EventHandler<MessageHolder> {
private RingBuffer<MessageHolder> ringBuffer;
private AbstractLocalAsyncWorker asyncWorker;
WorkerWithDisruptor(RingBuffer<MessageHolder> ringBuffer, AbstractLocalAsyncWorker asyncWorker) {
this.ringBuffer = ringBuffer;
this.asyncWorker = asyncWorker;
}
/**
* Receive the message from disruptor, when message in disruptor is empty, then send the cached data
* to the next workers.
*
* @param event published to the {@link RingBuffer}
* @param sequence of the event being processed
* @param endOfBatch flag to indicate if this is the last event in a batch from the {@link RingBuffer}
*/
public void onEvent(MessageHolder event, long sequence, boolean endOfBatch) {
try {
Object message = event.getMessage();
event.reset();
asyncWorker.allocateJob(message);
if (endOfBatch) {
asyncWorker.allocateJob(new EndOfBatchCommand());
}
} catch (Exception e) {
asyncWorker.saveException(e);
}
}
/**
* Push the message into disruptor ring buffer.
*
* @param message of the data to process.
*/
public void tell(Object message) {
long sequence = ringBuffer.next();
try {
ringBuffer.get(sequence).setMessage(message);
} finally {
ringBuffer.publish(sequence);
}
}
}
}

47
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/actor/AbstractLocalAsyncWorkerProvider.java
View File

@ -1,47 +0,0 @@
package org.skywalking.apm.collector.actor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.skywalking.apm.collector.queue.DaemonThreadFactory;
import org.skywalking.apm.collector.queue.MessageHolder;
import org.skywalking.apm.collector.queue.MessageHolderFactory;
/**
* @author pengys5
*/
public abstract class AbstractLocalAsyncWorkerProvider<T extends AbstractLocalAsyncWorker> extends AbstractLocalWorkerProvider<T> {
public abstract int queueSize();
@Override final public WorkerRef onCreate(
LocalWorkerContext localContext) throws ProviderNotFoundException {
T localAsyncWorker = (T)workerInstance(getClusterContext());
localAsyncWorker.preStart();
// Specify the size of the ring buffer, must be power of 2.
int bufferSize = queueSize();
if (!((((bufferSize - 1) & bufferSize) == 0) && bufferSize != 0)) {
throw new IllegalArgumentException("queue size must be power of 2");
}
// Construct the Disruptor
Disruptor<MessageHolder> disruptor = new Disruptor<MessageHolder>(MessageHolderFactory.INSTANCE, bufferSize, DaemonThreadFactory.INSTANCE);
RingBuffer<MessageHolder> ringBuffer = disruptor.getRingBuffer();
T.WorkerWithDisruptor disruptorWorker = new T.WorkerWithDisruptor(ringBuffer, localAsyncWorker);
// Connect the handler
disruptor.handleEventsWith(disruptorWorker);
// Start the Disruptor, starts all threads running
disruptor.start();
LocalAsyncWorkerRef workerRef = new LocalAsyncWorkerRef(role(), disruptorWorker);
if (localContext != null) {
localContext.put(workerRef);
}
return workerRef;
}
}

21
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/actor/ClusterWorkerRef.java
View File

@ -1,21 +0,0 @@
package org.skywalking.apm.collector.actor;
import akka.actor.ActorRef;
/**
* @author pengys5
*/
public class ClusterWorkerRef extends WorkerRef {
private ActorRef actorRef;
public ClusterWorkerRef(ActorRef actorRef, Role role) {
super(role);
this.actorRef = actorRef;
}
@Override
public void tell(Object message) {
actorRef.tell(message, ActorRef.noSender());
}
}

19
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/actor/LocalAsyncWorkerRef.java
View File

@ -1,19 +0,0 @@
package org.skywalking.apm.collector.actor;
/**
* @author pengys5
*/
public class LocalAsyncWorkerRef extends WorkerRef {
private AbstractLocalAsyncWorker.WorkerWithDisruptor workerWithDisruptor;
public LocalAsyncWorkerRef(Role role, AbstractLocalAsyncWorker.WorkerWithDisruptor workerWithDisruptor) {
super(role);
this.workerWithDisruptor = workerWithDisruptor;
}
@Override
public void tell(Object message) throws WorkerInvokeException {
workerWithDisruptor.tell(message);
}
}

13
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/actor/Role.java
View File

@ -1,13 +0,0 @@
package org.skywalking.apm.collector.actor;
import org.skywalking.apm.collector.actor.selector.WorkerSelector;
/**
* @author pengys5
*/
public interface Role {
String roleName();
WorkerSelector workerSelector();
}

24
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterConfig.java
View File

@ -1,24 +0,0 @@
package org.skywalking.apm.collector.cluster;
/**
* A static class contains some config values of cluster.
* {@link Cluster.Current#HOSTNAME} is a ip address of server which start this process.
* {@link Cluster.Current#PORT} is a PORT of server use to bind
* {@link Cluster.Current#ROLES} is a ROLES of workers that use to create workers which
* has those role in this process.
* {@link Cluster#SEED_NODES} is a SEED_NODES which cluster have, List of strings, e.g. SEED_NODES = "ip:PORT,ip:PORT"..
*
* @author pengys5
*/
public class ClusterConfig {
public static class Cluster {
public static class Current {
public static String HOSTNAME = "";
public static String PORT = "";
public static String ROLES = "";
}
public static String SEED_NODES = "";
}
}

31
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterConfigProvider.java
View File

@ -1,31 +0,0 @@
package org.skywalking.apm.collector.cluster;
import org.skywalking.apm.util.StringUtil;
import org.skywalking.apm.collector.config.ConfigProvider;
/**
* @author pengys5
*/
public class ClusterConfigProvider implements ConfigProvider {
@Override
public Class configClass() {
return ClusterConfig.class;
}
@Override
public void cliArgs() {
if (!StringUtil.isEmpty(System.getProperty("cluster.current.HOSTNAME"))) {
ClusterConfig.Cluster.Current.HOSTNAME = System.getProperty("cluster.current.HOSTNAME");
}
if (!StringUtil.isEmpty(System.getProperty("cluster.current.PORT"))) {
ClusterConfig.Cluster.Current.PORT = System.getProperty("cluster.current.PORT");
}
if (!StringUtil.isEmpty(System.getProperty("cluster.current.ROLES"))) {
ClusterConfig.Cluster.Current.ROLES = System.getProperty("cluster.current.ROLES");
}
if (!StringUtil.isEmpty(System.getProperty("cluster.SEED_NODES"))) {
ClusterConfig.Cluster.SEED_NODES = System.getProperty("cluster.SEED_NODES");
}
}
}

17
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleGroupDefine.java Normal file
View File

@ -0,0 +1,17 @@
package org.skywalking.apm.collector.cluster;
import org.skywalking.apm.collector.core.module.ModuleGroupDefine;
import org.skywalking.apm.collector.core.module.ModuleInstallMode;
/**
* @author pengys5
*/
public class ClusterModuleGroupDefine implements ModuleGroupDefine {
@Override public String name() {
return "cluster";
}
@Override public ModuleInstallMode mode() {
return ModuleInstallMode.Single;
}
}

8
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/Const.java
View File

@ -1,8 +0,0 @@
package org.skywalking.apm.collector.cluster;
/**
* @author pengys5
*/
public class Const {
public static final String SYSTEM_NAME = "ClusterSystem";
}

28
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/WorkerListenerMessage.java
View File

@ -1,28 +0,0 @@
package org.skywalking.apm.collector.cluster;
import org.skywalking.apm.collector.actor.AbstractWorker;
import org.skywalking.apm.collector.actor.Role;
import java.io.Serializable;
/**
* <code>WorkerListenerMessage</code> is a message just for the worker
* implementation of the {@link AbstractWorker}
* to register.
*
* @author pengys5
*/
public class WorkerListenerMessage {
public static class RegisterMessage implements Serializable {
private final Role role;
public RegisterMessage(Role role) {
this.role = role;
}
public Role getRole() {
return role;
}
}
}

76
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/WorkersListener.java
View File

@ -1,76 +0,0 @@
package org.skywalking.apm.collector.cluster;
import akka.actor.ActorRef;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import akka.cluster.Cluster;
import akka.cluster.ClusterEvent;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.skywalking.apm.collector.actor.AbstractWorker;
import org.skywalking.apm.collector.actor.ClusterWorkerContext;
import org.skywalking.apm.collector.actor.ClusterWorkerRef;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* <code>WorkersListener</code> listening the register message from workers
* implementation of the {@link AbstractWorker}
* and terminated message from akka cluster.
* <p>
* when listened register message then begin to watch the state for this worker
* and register to {@link ClusterWorkerContext} and {@link #relation}.
* <p>
* when listened terminate message then unregister from {@link ClusterWorkerContext} and {@link #relation} .
*
* @author pengys5
*/
public class WorkersListener extends UntypedActor {
public static final String WORK_NAME = "WorkersListener";
private static final Logger logger = LogManager.getFormatterLogger(WorkersListener.class);
private final ClusterWorkerContext clusterContext;
private Cluster cluster = Cluster.get(getContext().system());
private Map<ActorRef, ClusterWorkerRef> relation = new ConcurrentHashMap<>();
public WorkersListener(ClusterWorkerContext clusterContext) {
this.clusterContext = clusterContext;
}
@Override
public void preStart() throws Exception {
cluster.subscribe(getSelf(), ClusterEvent.UnreachableMember.class);
}
@Override
public void onReceive(Object message) throws Throwable {
if (message instanceof WorkerListenerMessage.RegisterMessage) {
WorkerListenerMessage.RegisterMessage register = (WorkerListenerMessage.RegisterMessage) message;
ActorRef sender = getSender();
logger.info("register worker of role: %s, path: %s", register.getRole().roleName(), sender.toString());
ClusterWorkerRef workerRef = new ClusterWorkerRef(sender, register.getRole());
relation.put(sender, workerRef);
clusterContext.put(new ClusterWorkerRef(sender, register.getRole()));
} else if (message instanceof Terminated) {
Terminated terminated = (Terminated) message;
clusterContext.remove(relation.get(terminated.getActor()));
relation.remove(terminated.getActor());
} else if (message instanceof ClusterEvent.UnreachableMember) {
ClusterEvent.UnreachableMember unreachableMember = (ClusterEvent.UnreachableMember) message;
Iterator<Map.Entry<ActorRef, ClusterWorkerRef>> iterator = relation.entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<ActorRef, ClusterWorkerRef> next = iterator.next();
if (next.getKey().path().address().equals(unreachableMember.member().address())) {
clusterContext.remove(next.getValue());
iterator.remove();
}
}
} else {
unhandled(message);
}
}
}

9
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisConfig.java Normal file
View File

@ -0,0 +1,9 @@
package org.skywalking.apm.collector.cluster.redis;
/**
* @author pengys5
*/
public class ClusterRedisConfig {
public static String HOST;
public static int PORT;
}

23
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisConfigParser.java Normal file
View File

@ -0,0 +1,23 @@
package org.skywalking.apm.collector.cluster.redis;
import java.util.Map;
import org.skywalking.apm.collector.core.config.ConfigParseException;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.util.StringUtils;
/**
* @author pengys5
*/
public class ClusterRedisConfigParser implements ModuleConfigParser {
private static final String HOST = "host";
private static final String PORT = "port";
@Override public void parse(Map config) throws ConfigParseException {
ClusterRedisConfig.HOST = (String)config.get(HOST);
ClusterRedisConfig.PORT = (Integer)config.get(PORT);
if (StringUtils.isEmpty(ClusterRedisConfig.HOST) || ClusterRedisConfig.PORT == 0) {
throw new ConfigParseException("");
}
}
}

24
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisDataInitializer.java Normal file
View File

@ -0,0 +1,24 @@
package org.skywalking.apm.collector.cluster.redis;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.cluster.ClusterDataInitializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class ClusterRedisDataInitializer extends ClusterDataInitializer {
private final Logger logger = LoggerFactory.getLogger(ClusterRedisDataInitializer.class);
@Override public void addItem(Client client, String itemKey) throws ClientException {
logger.info("add the redis item key \"{}\" exist", itemKey);
}
@Override public boolean existItem(Client client, String itemKey) throws ClientException {
logger.info("assess the redis item key \"{}\" exist", itemKey);
return false;
}
}

48
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisModuleDefine.java Normal file
View File

@ -0,0 +1,48 @@
package org.skywalking.apm.collector.cluster.redis;
import org.skywalking.apm.collector.client.redis.RedisClient;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.cluster.ClusterModuleDefine;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationReader;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter;
import org.skywalking.apm.collector.core.framework.DataInitializer;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.module.ModuleGroup;
/**
* @author pengys5
*/
public class ClusterRedisModuleDefine extends ClusterModuleDefine {
@Override public ModuleGroup group() {
return ModuleGroup.Cluster;
}
@Override public String name() {
return "redis";
}
@Override public boolean defaultModule() {
return false;
}
@Override protected ModuleConfigParser configParser() {
return new ClusterRedisConfigParser();
}
@Override protected Client createClient() {
return new RedisClient(ClusterRedisConfig.HOST, ClusterRedisConfig.PORT);
}
@Override protected DataInitializer dataInitializer() {
return new ClusterRedisDataInitializer();
}
@Override protected ClusterModuleRegistrationWriter registrationWriter() {
return new ClusterRedisModuleRegistrationWriter(getClient());
}
@Override protected ClusterModuleRegistrationReader registrationReader() {
return null;
}
}

28
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisModuleRegistrationWriter.java Normal file
View File

@ -0,0 +1,28 @@
package org.skywalking.apm.collector.cluster.redis;
import org.skywalking.apm.collector.client.redis.RedisClient;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter;
import org.skywalking.apm.collector.core.module.ModuleRegistration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class ClusterRedisModuleRegistrationWriter extends ClusterModuleRegistrationWriter {
private final Logger logger = LoggerFactory.getLogger(ClusterRedisModuleRegistrationWriter.class);
public ClusterRedisModuleRegistrationWriter(Client client) {
super(client);
}
@Override public void write(String key, ModuleRegistration.Value value) {
logger.debug("key {}, value {}", key, value.getHost());
key = key + "." + value.getHost() + ":" + value.getPort();
value.getData().addProperty("host", value.getHost());
value.getData().addProperty("port", value.getPort());
((RedisClient)client).setex(key, 120, value.getData().toString());
}
}

13
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneConfigParser.java Normal file
View File

@ -0,0 +1,13 @@
package org.skywalking.apm.collector.cluster.standalone;
import java.util.Map;
import org.skywalking.apm.collector.core.config.ConfigParseException;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
/**
* @author pengys5
*/
public class ClusterStandaloneConfigParser implements ModuleConfigParser {
@Override public void parse(Map config) throws ConfigParseException {
}
}

32
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneDataInitializer.java Normal file
View File

@ -0,0 +1,32 @@
package org.skywalking.apm.collector.cluster.standalone;
import org.skywalking.apm.collector.client.h2.H2Client;
import org.skywalking.apm.collector.client.h2.H2ClientException;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.cluster.ClusterDataInitializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class ClusterStandaloneDataInitializer extends ClusterDataInitializer {
private final Logger logger = LoggerFactory.getLogger(ClusterStandaloneDataInitializer.class);
@Override public void addItem(Client client, String itemKey) throws ClientException {
logger.info("add the h2 item key \"{}\" exist", itemKey);
itemKey = itemKey.replaceAll("\\.", "_");
String sql = "CREATE TABLE " + itemKey + "(ADDRESS VARCHAR(100) PRIMARY KEY,DATA VARCHAR(255));";
try {
((H2Client)client).execute(sql);
} catch (H2ClientException e) {
logger.error(e.getMessage(), e);
}
}
@Override public boolean existItem(Client client, String itemKey) throws ClientException {
return false;
}
}

48
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneModuleDefine.java Normal file
View File

@ -0,0 +1,48 @@
package org.skywalking.apm.collector.cluster.standalone;
import org.skywalking.apm.collector.client.h2.H2Client;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.cluster.ClusterModuleDefine;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationReader;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter;
import org.skywalking.apm.collector.core.framework.DataInitializer;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.module.ModuleGroup;
/**
* @author pengys5
*/
public class ClusterStandaloneModuleDefine extends ClusterModuleDefine {
@Override public ModuleGroup group() {
return ModuleGroup.Cluster;
}
@Override public String name() {
return "standalone";
}
@Override public boolean defaultModule() {
return true;
}
@Override protected ModuleConfigParser configParser() {
return new ClusterStandaloneConfigParser();
}
@Override public Client createClient() {
return new H2Client();
}
@Override protected DataInitializer dataInitializer() {
return new ClusterStandaloneDataInitializer();
}
@Override protected ClusterModuleRegistrationWriter registrationWriter() {
return new ClusterStandaloneModuleRegistrationWriter(getClient());
}
@Override protected ClusterModuleRegistrationReader registrationReader() {
return null;
}
}

34
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneModuleRegistrationWriter.java Normal file
View File

@ -0,0 +1,34 @@
package org.skywalking.apm.collector.cluster.standalone;
import org.skywalking.apm.collector.client.h2.H2Client;
import org.skywalking.apm.collector.client.h2.H2ClientException;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter;
import org.skywalking.apm.collector.core.module.ModuleRegistration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class ClusterStandaloneModuleRegistrationWriter extends ClusterModuleRegistrationWriter {
private final Logger logger = LoggerFactory.getLogger(ClusterStandaloneModuleRegistrationWriter.class);
public ClusterStandaloneModuleRegistrationWriter(Client client) {
super(client);
}
@Override public void write(String key, ModuleRegistration.Value value) {
key = key.replaceAll("\\.", "_");
String hostPort = value.getHost() + ":" + value.getPort();
String sql = "INSERT INTO " + key + " VALUES('" + hostPort + "', '" + value.getData().toString() + "');";
String sql2 = "SELECT * FROM " + key;
try {
((H2Client)client).execute(sql);
((H2Client)client).executeQuery(sql2);
} catch (H2ClientException e) {
logger.error(e.getMessage(), e);
}
}
}

9
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKConfig.java Normal file
View File

@ -0,0 +1,9 @@
package org.skywalking.apm.collector.cluster.zookeeper;
/**
* @author pengys5
*/
public class ClusterZKConfig {
public static String HOST_PORT;
public static int SESSION_TIMEOUT;
}

28
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKConfigParser.java Normal file
View File

@ -0,0 +1,28 @@
package org.skywalking.apm.collector.cluster.zookeeper;
import java.util.Map;
import org.skywalking.apm.collector.core.config.ConfigParseException;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.util.StringUtils;
/**
* @author pengys5
*/
public class ClusterZKConfigParser implements ModuleConfigParser {
private static final String HOST_PORT = "hostPort";
private static final String SESSION_TIMEOUT = "sessionTimeout";
@Override public void parse(Map config) throws ConfigParseException {
ClusterZKConfig.HOST_PORT = (String)config.get(HOST_PORT);
ClusterZKConfig.SESSION_TIMEOUT = 1000;
if (StringUtils.isEmpty(ClusterZKConfig.HOST_PORT)) {
throw new ConfigParseException("");
}
if (!StringUtils.isEmpty(config.get(SESSION_TIMEOUT))) {
ClusterZKConfig.SESSION_TIMEOUT = (Integer)config.get(SESSION_TIMEOUT);
}
}
}

49
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKDataInitializer.java Normal file
View File

@ -0,0 +1,49 @@
package org.skywalking.apm.collector.cluster.zookeeper;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.skywalking.apm.collector.client.zookeeper.ZookeeperClient;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.cluster.ClusterDataInitializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class ClusterZKDataInitializer extends ClusterDataInitializer {
private final Logger logger = LoggerFactory.getLogger(ClusterZKDataInitializer.class);
@Override public void addItem(Client client, String itemKey) throws ClientException {
logger.info("add the zookeeper item key \"{}\" exist", itemKey);
ZookeeperClient zkClient = (ZookeeperClient)client;
String[] catalogs = itemKey.split("\\.");
StringBuilder pathBuilder = new StringBuilder();
for (String catalog : catalogs) {
pathBuilder.append("/").append(catalog);
if (zkClient.exists(pathBuilder.toString(), false) == null) {
zkClient.create(pathBuilder.toString(), null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
}
}
@Override public boolean existItem(Client client, String itemKey) throws ClientException {
logger.info("assess the zookeeper item key \"{}\" exist", itemKey);
ZookeeperClient zkClient = (ZookeeperClient)client;
String[] catalogs = itemKey.split("\\.");
StringBuilder pathBuilder = new StringBuilder();
for (String catalog : catalogs) {
pathBuilder.append("/").append(catalog);
}
// if (zkClient.exists(pathBuilder.toString(), false) == null) {
// return false;
// } else {
return true;
// }
}
}

48
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKModuleDefine.java Normal file
View File

@ -0,0 +1,48 @@
package org.skywalking.apm.collector.cluster.zookeeper;
import org.skywalking.apm.collector.client.zookeeper.ZookeeperClient;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.cluster.ClusterDataInitializer;
import org.skywalking.apm.collector.core.cluster.ClusterModuleDefine;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationReader;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter;
import org.skywalking.apm.collector.core.module.ModuleConfigParser;
import org.skywalking.apm.collector.core.module.ModuleGroup;
/**
* @author pengys5
*/
public class ClusterZKModuleDefine extends ClusterModuleDefine {
@Override protected ModuleGroup group() {
return ModuleGroup.Cluster;
}
@Override public String name() {
return "zookeeper";
}
@Override public boolean defaultModule() {
return false;
}
@Override protected ModuleConfigParser configParser() {
return new ClusterZKConfigParser();
}
@Override protected Client createClient() {
return new ZookeeperClient(ClusterZKConfig.HOST_PORT, ClusterZKConfig.SESSION_TIMEOUT);
}
@Override protected ClusterDataInitializer dataInitializer() {
return new ClusterZKDataInitializer();
}
@Override protected ClusterModuleRegistrationWriter registrationWriter() {
return new ClusterZKModuleRegistrationWriter(getClient());
}
@Override protected ClusterModuleRegistrationReader registrationReader() {
return null;
}
}

13
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKModuleRegistrationReader.java Normal file
View File

@ -0,0 +1,13 @@
package org.skywalking.apm.collector.cluster.zookeeper;
import java.util.List;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationReader;
/**
* @author pengys5
*/
public class ClusterZKModuleRegistrationReader implements ClusterModuleRegistrationReader {
@Override public List<String> read(String key) {
return null;
}
}

37
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKModuleRegistrationWriter.java Normal file
View File

@ -0,0 +1,37 @@
package org.skywalking.apm.collector.cluster.zookeeper;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import org.skywalking.apm.collector.client.zookeeper.ZookeeperClient;
import org.skywalking.apm.collector.client.zookeeper.util.PathUtils;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter;
import org.skywalking.apm.collector.core.module.ModuleRegistration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author pengys5
*/
public class ClusterZKModuleRegistrationWriter extends ClusterModuleRegistrationWriter {
private final Logger logger = LoggerFactory.getLogger(ClusterZKModuleRegistrationWriter.class);
public ClusterZKModuleRegistrationWriter(Client client) {
super(client);
}
@Override public void write(String key, ModuleRegistration.Value value) throws ClientException {
logger.info("cluster zookeeper register key: {}, value: {}", key, value);
String workerUIPath = PathUtils.convertKey2Path(key) + "/" + value.getHost() + ":" + value.getPort();
Stat stat = ((ZookeeperClient)client).exists(workerUIPath, false);
if (stat == null) {
((ZookeeperClient)client).create(workerUIPath, value.getData() == null ? null : value.getData().toString().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
} else {
((ZookeeperClient)client).setData(workerUIPath, value.getData() == null ? null : value.getData().toString().getBytes(), -1);
}
}
}

40
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/config/ConfigInitializer.java
View File

@ -1,40 +0,0 @@
package org.skywalking.apm.collector.config;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.ServiceLoader;
/**
* @author pengys5
*/
public enum ConfigInitializer {
INSTANCE;
private Logger logger = LogManager.getFormatterLogger(ConfigInitializer.class);
public void initialize() throws IOException, IllegalAccessException {
InputStream configFileStream = ConfigInitializer.class.getResourceAsStream("/collector.config");
initializeConfigFile(configFileStream);
ServiceLoader<ConfigProvider> configProviders = ServiceLoader.load(ConfigProvider.class);
for (ConfigProvider provider : configProviders) {
provider.cliArgs();
}
}
private void initializeConfigFile(InputStream configFileStream) throws IOException, IllegalAccessException {
ServiceLoader<ConfigProvider> configProviders = ServiceLoader.load(ConfigProvider.class);
Properties properties = new Properties();
properties.load(configFileStream);
for (ConfigProvider provider : configProviders) {
logger.info("configProvider provider name: %s", provider.getClass().getName());
Class configClass = provider.configClass();
org.skywalking.apm.util.ConfigInitializer.initialize(properties, configClass);
}
}
}

10
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/config/ConfigProvider.java
View File

@ -1,10 +0,0 @@
package org.skywalking.apm.collector.config;
/**
* @author pengys5
*/
public interface ConfigProvider {
Class configClass();
void cliArgs();
}

14
apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/log/LogManager.java
View File

@ -1,14 +0,0 @@
package org.skywalking.apm.collector.log;
import org.apache.logging.log4j.Logger;
/**
* @author pengys5
*/
public enum LogManager {
INSTANCE;
public Logger getFormatterLogger(final Class<?> clazz) {
return org.apache.logging.log4j.LogManager.getFormatterLogger(clazz);
}
}

1
apm-collector/apm-collector-cluster/src/main/resources/META-INF/defines/group.define Normal file
View File

@ -0,0 +1 @@
org.skywalking.apm.collector.cluster.ClusterModuleGroupDefine

3
apm-collector/apm-collector-cluster/src/main/resources/META-INF/defines/module.define Normal file
View File

@ -0,0 +1,3 @@
org.skywalking.apm.collector.cluster.zookeeper.ClusterZKModuleDefine
org.skywalking.apm.collector.cluster.standalone.ClusterStandaloneModuleDefine
org.skywalking.apm.collector.cluster.redis.ClusterRedisModuleDefine

61
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/actor/AbstractClusterWorkerProviderTestCase.java
View File

@ -1,61 +0,0 @@
package org.skywalking.apm.collector.actor;
import akka.actor.ActorSystem;
import org.apache.logging.log4j.Logger;
import org.mockito.Mockito;
import org.powermock.reflect.Whitebox;
import org.skywalking.apm.collector.actor.selector.RollingSelector;
import org.skywalking.apm.collector.actor.selector.WorkerSelector;
import org.skywalking.apm.collector.log.LogManager;
/**
* @author pengys5
*/
//@RunWith(PowerMockRunner.class)
//@PrepareForTest({LogManager.class})
public class AbstractClusterWorkerProviderTestCase {
// @Test
public void testOnCreate() throws ProviderNotFoundException {
LogManager logManager = Mockito.mock(LogManager.class);
Whitebox.setInternalState(LogManager.class, "INSTANCE", logManager);
Logger logger = Mockito.mock(Logger.class);
Mockito.when(logManager.getFormatterLogger(Mockito.any())).thenReturn(logger);
ActorSystem actorSystem = Mockito.mock(ActorSystem.class);
ClusterWorkerContext clusterWorkerContext = new ClusterWorkerContext(actorSystem);
Impl impl = new Impl();
impl.onCreate(null);
}
class Impl extends AbstractClusterWorkerProvider<AbstractClusterWorkerTestCase.Impl> {
@Override
public Role role() {
return Role.INSTANCE;
}
@Override
public AbstractClusterWorkerTestCase.Impl workerInstance(ClusterWorkerContext clusterContext) {
return new AbstractClusterWorkerTestCase.Impl(role(), clusterContext, new LocalWorkerContext());
}
@Override
public int workerNum() {
return 0;
}
}
enum Role implements org.skywalking.apm.collector.actor.Role {
INSTANCE;
@Override
public String roleName() {
return AbstractClusterWorkerTestCase.Impl.class.getSimpleName();
}
@Override
public WorkerSelector workerSelector() {
return new RollingSelector();
}
}
}

100
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/actor/AbstractClusterWorkerTestCase.java
View File

@ -1,100 +0,0 @@
package org.skywalking.apm.collector.actor;
import akka.actor.Address;
import akka.cluster.ClusterEvent;
import akka.cluster.Member;
import org.apache.logging.log4j.Logger;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.reflect.Whitebox;
import org.skywalking.apm.collector.actor.selector.RollingSelector;
import org.skywalking.apm.collector.actor.selector.WorkerSelector;
import static org.mockito.Mockito.*;
/**
* @author pengys5
*/
@RunWith(PowerMockRunner.class)
@PrepareForTest( {ClusterEvent.MemberUp.class, Address.class})
@PowerMockIgnore( {"javax.management.*"})
public class AbstractClusterWorkerTestCase {
private AbstractClusterWorker.WorkerWithAkka workerWithAkka = mock(AbstractClusterWorker.WorkerWithAkka.class, CALLS_REAL_METHODS);
private AbstractClusterWorker worker = PowerMockito.spy(new Impl(WorkerRole.INSTANCE, null, null));
@Before
public void init() {
Logger logger = mock(Logger.class);
Whitebox.setInternalState(workerWithAkka, "logger", logger);
Whitebox.setInternalState(workerWithAkka, "ownerWorker", worker);
}
@Test
public void testAllocateJob() throws Exception {
String jobStr = "TestJob";
worker.allocateJob(jobStr);
verify(worker).onWork(jobStr);
}
@Test
public void testMemberUp() throws Throwable {
ClusterEvent.MemberUp memberUp = mock(ClusterEvent.MemberUp.class);
Address address = mock(Address.class);
when(address.toString()).thenReturn("address");
Member member = mock(Member.class);
when(member.address()).thenReturn(address);
when(memberUp.member()).thenReturn(member);
workerWithAkka.onReceive(memberUp);
verify(workerWithAkka).register(member);
}
@Test
public void testMessage() throws Throwable {
String message = "test";
workerWithAkka.onReceive(message);
verify(worker).allocateJob(message);
}
static class Impl extends AbstractClusterWorker {
@Override
public void preStart() throws ProviderNotFoundException {
}
public Impl(Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) {
super(role, clusterContext, selfContext);
}
@Override
protected void onWork(Object message) throws IllegalArgumentException {
}
}
public enum WorkerRole implements Role {
INSTANCE;
@Override
public String roleName() {
return Impl.class.getSimpleName();
}
@Override
public WorkerSelector workerSelector() {
return new RollingSelector();
}
}
}

59
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/actor/AbstractLocalAsyncWorkerTestCase.java
View File

@ -1,59 +0,0 @@
package org.skywalking.apm.collector.actor;
import org.junit.Test;
import org.mockito.ArgumentMatcher;
import org.skywalking.apm.collector.queue.EndOfBatchCommand;
import org.skywalking.apm.collector.queue.MessageHolder;
import static org.mockito.Mockito.*;
/**
* @author pengys5
*/
public class AbstractLocalAsyncWorkerTestCase {
@Test
public void testAllocateJob() throws Exception {
AbstractLocalAsyncWorker worker = mock(AbstractLocalAsyncWorker.class);
String message = "Test";
worker.allocateJob(message);
verify(worker).onWork(message);
}
@Test
public void testOnEventWhenNotEnd() throws Exception {
AbstractLocalAsyncWorker worker = mock(AbstractLocalAsyncWorker.class);
AbstractLocalAsyncWorker.WorkerWithDisruptor disruptor = new AbstractLocalAsyncWorker.WorkerWithDisruptor(null, worker);
MessageHolder holder = new MessageHolder();
String message = "Test";
holder.setMessage(message);
disruptor.onEvent(holder, 0, false);
verify(worker).onWork(message);
}
@Test
public void testOnEventWhenEnd() throws Exception {
AbstractLocalAsyncWorker worker = mock(AbstractLocalAsyncWorker.class);
AbstractLocalAsyncWorker.WorkerWithDisruptor disruptor = new AbstractLocalAsyncWorker.WorkerWithDisruptor(null, worker);
MessageHolder holder = new MessageHolder();
String message = "Test";
holder.setMessage(message);
disruptor.onEvent(holder, 0, true);
verify(worker, times(1)).onWork(message);
verify(worker, times(1)).onWork(argThat(new IsEndOfBatchCommandClass()));
}
class IsEndOfBatchCommandClass extends ArgumentMatcher<EndOfBatchCommand> {
public boolean matches(Object para) {
return para.getClass() == EndOfBatchCommand.class;
}
}
}

77
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/actor/AbstractWorkerProviderTestCase.java
View File

@ -1,77 +0,0 @@
package org.skywalking.apm.collector.actor;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import static org.powermock.api.mockito.PowerMockito.mock;
import static org.powermock.api.mockito.PowerMockito.when;
/**
* @author pengys5
*/
@RunWith(PowerMockRunner.class)
@PrepareForTest( {AbstractWorker.class})
@PowerMockIgnore( {"javax.management.*"})
public class AbstractWorkerProviderTestCase {
@Test(expected = IllegalArgumentException.class)
public void testNullWorkerInstanceCreate() throws ProviderNotFoundException {
AbstractWorkerProvider provider = mock(AbstractWorkerProvider.class);
when(provider.workerInstance(null)).thenReturn(null);
AbstractWorker worker = mock(AbstractWorker.class);
provider.create(worker);
}
@Test
public void testNoneWorkerOwner() throws ProviderNotFoundException {
AbstractWorkerProvider provider = mock(AbstractWorkerProvider.class);
ClusterWorkerContext context = mock(ClusterWorkerContext.class);
provider.setClusterContext(context);
AbstractWorker worker = mock(AbstractWorker.class);
when(provider.workerInstance(context)).thenReturn(worker);
provider.create(null);
Mockito.verify(provider).onCreate(null);
}
@Test
public void testHasWorkerOwner() throws ProviderNotFoundException {
AbstractWorkerProvider provider = mock(AbstractWorkerProvider.class);
ClusterWorkerContext context = mock(ClusterWorkerContext.class);
provider.setClusterContext(context);
AbstractWorker worker = mock(AbstractWorker.class);
when(provider.workerInstance(context)).thenReturn(worker);
AbstractWorker workerOwner = mock(AbstractWorker.class);
LocalWorkerContext localWorkerContext = mock(LocalWorkerContext.class);
when(workerOwner.getSelfContext()).thenReturn(localWorkerContext);
provider.create(workerOwner);
Mockito.verify(provider).onCreate(localWorkerContext);
}
@Test(expected = IllegalArgumentException.class)
public void testHasWorkerOwnerButNoneContext() throws ProviderNotFoundException {
AbstractWorkerProvider provider = mock(AbstractWorkerProvider.class);
ClusterWorkerContext context = mock(ClusterWorkerContext.class);
provider.setClusterContext(context);
AbstractWorker worker = mock(AbstractWorker.class);
when(provider.workerInstance(context)).thenReturn(worker);
AbstractWorker workerOwner = mock(AbstractWorker.class);
when(workerOwner.getSelfContext()).thenReturn(null);
provider.create(workerOwner);
}
}

24
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/actor/selector/AbstractHashMessageTestCase.java
View File

@ -1,24 +0,0 @@
package org.skywalking.apm.collector.actor.selector;
import org.junit.Assert;
import org.junit.Test;
/**
* @author pengys5
*/
public class AbstractHashMessageTestCase {
@Test
public void testGetHashCode() {
String key = "key";
Impl impl = new Impl(key);
Assert.assertEquals(key.hashCode(), impl.getHashCode());
}
class Impl extends AbstractHashMessage {
public Impl(String key) {
super(key);
}
}
}

55
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/actor/selector/HashCodeSelectorTestCase.java
View File

@ -1,55 +0,0 @@
package org.skywalking.apm.collector.actor.selector;
import org.junit.Assert;
import org.junit.Test;
import org.skywalking.apm.collector.actor.WorkerRef;
import java.util.ArrayList;
import java.util.List;
import static org.powermock.api.mockito.PowerMockito.mock;
import static org.powermock.api.mockito.PowerMockito.when;
/**
* @author pengys5
*/
public class HashCodeSelectorTestCase {
@Test
public void testSelect() {
List<WorkerRef> members = new ArrayList<>();
WorkerRef workerRef_1 = mock(WorkerRef.class);
WorkerRef workerRef_2 = mock(WorkerRef.class);
WorkerRef workerRef_3 = mock(WorkerRef.class);
members.add(workerRef_1);
members.add(workerRef_2);
members.add(workerRef_3);
AbstractHashMessage message_1 = mock(AbstractHashMessage.class);
when(message_1.getHashCode()).thenReturn(9);
AbstractHashMessage message_2 = mock(AbstractHashMessage.class);
when(message_2.getHashCode()).thenReturn(10);
AbstractHashMessage message_3 = mock(AbstractHashMessage.class);
when(message_3.getHashCode()).thenReturn(11);
HashCodeSelector selector = new HashCodeSelector();
WorkerRef select_1 = selector.select(members, message_1);
Assert.assertEquals(workerRef_1.hashCode(), select_1.hashCode());
WorkerRef select_2 = selector.select(members, message_2);
Assert.assertEquals(workerRef_2.hashCode(), select_2.hashCode());
WorkerRef select_3 = selector.select(members, message_3);
Assert.assertEquals(workerRef_3.hashCode(), select_3.hashCode());
}
@Test(expected = IllegalArgumentException.class)
public void testSelectError() {
HashCodeSelector selector = new HashCodeSelector();
selector.select(null, new Object());
}
}

41
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/actor/selector/RollingSelectorTestCase.java
View File

@ -1,41 +0,0 @@
package org.skywalking.apm.collector.actor.selector;
import org.junit.Assert;
import org.junit.Test;
import org.skywalking.apm.collector.actor.WorkerRef;
import java.util.ArrayList;
import java.util.List;
import static org.powermock.api.mockito.PowerMockito.mock;
/**
* @author pengys5
*/
public class RollingSelectorTestCase {
@Test
public void testSelect() {
List<WorkerRef> members = new ArrayList<>();
WorkerRef workerRef_1 = mock(WorkerRef.class);
WorkerRef workerRef_2 = mock(WorkerRef.class);
WorkerRef workerRef_3 = mock(WorkerRef.class);
members.add(workerRef_1);
members.add(workerRef_2);
members.add(workerRef_3);
Object message = new Object();
RollingSelector selector = new RollingSelector();
WorkerRef selected_1 = selector.select(members, message);
Assert.assertEquals(workerRef_2.hashCode(), selected_1.hashCode());
WorkerRef selected_2 = selector.select(members, message);
Assert.assertEquals(workerRef_3.hashCode(), selected_2.hashCode());
WorkerRef selected_3 = selector.select(members, message);
Assert.assertEquals(workerRef_1.hashCode(), selected_3.hashCode());
}
}

45
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/config/ConfigInitializerTestCase.java
View File

@ -1,45 +0,0 @@
package org.skywalking.apm.collector.config;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.skywalking.apm.collector.cluster.ClusterConfig;
/**
* @author pengys5
*/
public class ConfigInitializerTestCase {
@Before
public void clear() {
System.clearProperty("cluster.current.HOSTNAME");
System.clearProperty("cluster.current.PORT");
System.clearProperty("cluster.current.ROLES");
System.clearProperty("cluster.SEED_NODES");
}
@Test
public void testInitialize() throws Exception {
ConfigInitializer.INSTANCE.initialize();
Assert.assertEquals("127.0.0.1", ClusterConfig.Cluster.Current.HOSTNAME);
Assert.assertEquals("1000", ClusterConfig.Cluster.Current.PORT);
Assert.assertEquals("WorkersListener", ClusterConfig.Cluster.Current.ROLES);
Assert.assertEquals("127.0.0.1:1000", ClusterConfig.Cluster.SEED_NODES);
}
@Test
public void testInitializeWithCli() throws Exception {
System.setProperty("cluster.current.HOSTNAME", "127.0.0.2");
System.setProperty("cluster.current.PORT", "1001");
System.setProperty("cluster.current.ROLES", "Test1, Test2");
System.setProperty("cluster.SEED_NODES", "127.0.0.1:1000, 127.0.0.1:1001");
ConfigInitializer.INSTANCE.initialize();
Assert.assertEquals("127.0.0.2", ClusterConfig.Cluster.Current.HOSTNAME);
Assert.assertEquals("1001", ClusterConfig.Cluster.Current.PORT);
Assert.assertEquals("Test1, Test2", ClusterConfig.Cluster.Current.ROLES);
Assert.assertEquals("127.0.0.1:1000, 127.0.0.1:1001", ClusterConfig.Cluster.SEED_NODES);
}
}

18
apm-collector/apm-collector-cluster/src/test/java/org/skywalking/apm/collector/log/MockLog.java
View File

@ -1,18 +0,0 @@
package org.skywalking.apm.collector.log;
import org.apache.logging.log4j.Logger;
import org.mockito.Mockito;
import org.powermock.api.mockito.PowerMockito;
/**
* @author pengys5
*/
public class MockLog {
public Logger mockito() {
LogManager logManager = PowerMockito.mock(LogManager.class);
Logger logger = Mockito.mock(Logger.class);
Mockito.when(logManager.getFormatterLogger(Mockito.any())).thenReturn(logger);
return logger;
}
}

0
apm-collector/apm-collector-cluster/src/test/resources/META-INF/services/org.skywalking.apm.collector.actor.AbstractClusterWorkerProvider
View File

0
apm-collector/apm-collector-cluster/src/test/resources/META-INF/services/org.skywalking.apm.collector.actor.AbstractLocalWorkerProvider
View File

1
apm-collector/apm-collector-cluster/src/test/resources/META-INF/services/org.skywalking.apm.collector.config.ConfigProvider
View File

@ -1 +0,0 @@
org.skywalking.apm.collector.cluster.ClusterConfigProvider

11
apm-collector/apm-collector-cluster/src/test/resources/collector.config
View File

@ -1,11 +0,0 @@
cluster.current.hostname=127.0.0.1
cluster.current.port=1000
cluster.current.roles=WorkersListener
cluster.seed_nodes=127.0.0.1:1000
es.cluster.name=CollectorDBCluster
es.cluster.nodes=127.0.0.1:9300
es.cluster.transport.sniffer=true
es.index.shards.number=2
es.index.replicas.number=0

13
apm-collector/apm-collector-cluster/src/test/resources/log4j2.xml
View File

@ -1,13 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="error">
<Appenders>
<Console name="Console" target="SYSTEM_OUT">
<PatternLayout charset="UTF-8" pattern="[%d{yyyy-MM-dd HH:mm:ss:SSS}] [%p] - %l - %m%n"/>
</Console>
</Appenders>
<Loggers>
<Root level="debug">
<AppenderRef ref="Console"/>
</Root>
</Loggers>
</Configuration>

12
apm-collector/apm-collector-commons/src/main/java/org/skywalking/apm/collector/commons/config/SeedNodesFormatter.java
View File

@ -1,12 +0,0 @@
package org.skywalking.apm.collector.commons.config;
/**
* @author pengys5
*/
public enum SeedNodesFormatter {
INSTANCE;
public String formatter(String seedNodes) {
return null;
}
}

22
apm-collector/apm-collector-commons/src/main/java/org/skywalking/apm/collector/commons/role/TraceSegmentReceiverRole.java
View File

@ -1,22 +0,0 @@
package org.skywalking.apm.collector.commons.role;
import org.skywalking.apm.collector.actor.Role;
import org.skywalking.apm.collector.actor.selector.RollingSelector;
import org.skywalking.apm.collector.actor.selector.WorkerSelector;
/**
* @author pengys5
*/
public enum TraceSegmentReceiverRole implements Role {
INSTANCE;
@Override
public String roleName() {
return "TraceSegmentReceiver";
}
@Override
public WorkerSelector workerSelector() {
return new RollingSelector();
}
}

32
apm-collector/apm-collector-commons/src/main/java/org/skywalking/apm/collector/commons/serializer/JsonSerializer.java
View File

@ -1,32 +0,0 @@
package org.skywalking.apm.collector.commons.serializer;
import akka.serialization.JSerializer;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
/**
* @author pengys5
*/
public class JsonSerializer extends JSerializer {
@Override
public boolean includeManifest() {
return false;
}
@Override
public int identifier() {
return 31;
}
@Override
public byte[] toBinary(Object o) {
JsonObject jsonObject = (JsonObject) o;
return jsonObject.toString().getBytes();
}
@Override
public Object fromBinaryJava(byte[] bytes, Class<?> manifest) {
Gson gson = new Gson();
return gson.fromJson(new String(bytes), JsonObject.class);
}
}

34
apm-collector/apm-collector-commons/src/main/resources/application.conf
View File

@ -1,34 +0,0 @@
akka {
actor {
provider = "akka.cluster.ClusterActorRefProvider"
serializers {
java = "akka.serialization.JavaSerializer"
proto = "akka.remote.serialization.ProtobufSerializer"
// TraceSegment = "org.skywalking.apm.collector.worker.TraceSegmentSerializer"
json = "org.skywalking.apm.collector.commons.serializer.JsonSerializer"
}
serialization-bindings {
"java.lang.String" = java
"com.google.protobuf.Message" = proto
// "TraceSegment" = TraceSegment
"com.google.gson.JsonObject" = json
}
warn-about-java-serializer-usage = on
}
remote {
log-remote-lifecycle-events = off
netty.tcp {
}
}
cluster {
auto-down-unreachable-after = off
metrics.enabled = off
roles = ["WorkersListener"]
}
}

32
apm-collector/apm-collector-core/pom.xml Normal file
View File

@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<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-collector</artifactId>
<groupId>org.skywalking</groupId>
<version>3.2-2017</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>apm-collector-core</artifactId>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId>
<version>1.18</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.3</version>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.8.1</version>
</dependency>
</dependencies>
</project>

15
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/CollectorException.java Normal file
View File

@ -0,0 +1,15 @@
package org.skywalking.apm.collector.core;
/**
* @author pengys5
*/
public class CollectorException extends Exception {
public CollectorException(String message) {
super(message);
}
public CollectorException(String message, Throwable cause) {
super(message, cause);
}
}

40
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/agentstream/AgentStreamModuleDefine.java Normal file
View File

@ -0,0 +1,40 @@
package org.skywalking.apm.collector.core.agentstream;
import java.util.Map;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.cluster.ClusterDataInitializer;
import org.skywalking.apm.collector.core.cluster.ClusterModuleContext;
import org.skywalking.apm.collector.core.config.ConfigParseException;
import org.skywalking.apm.collector.core.framework.DataInitializer;
import org.skywalking.apm.collector.core.framework.DefineException;
import org.skywalking.apm.collector.core.module.ModuleDefine;
import org.skywalking.apm.collector.core.server.Server;
import org.skywalking.apm.collector.core.server.ServerException;
/**
* @author pengys5
*/
public abstract class AgentStreamModuleDefine extends ModuleDefine {
@Override public final void initialize(Map config) throws DefineException, ClientException {
try {
configParser().parse(config);
Server server = server();
server.initialize();
String key = ClusterDataInitializer.BASE_CATALOG + "." + name();
ClusterModuleContext.WRITER.write(key, registration().buildValue());
} catch (ConfigParseException | ServerException e) {
throw new AgentStreamModuleException(e.getMessage(), e);
}
}
@Override protected final DataInitializer dataInitializer() {
throw new UnsupportedOperationException("");
}
@Override protected final Client createClient() {
throw new UnsupportedOperationException("");
}
}

16
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/agentstream/AgentStreamModuleException.java Normal file
View File

@ -0,0 +1,16 @@
package org.skywalking.apm.collector.core.agentstream;
import org.skywalking.apm.collector.core.module.ModuleException;
/**
* @author pengys5
*/
public class AgentStreamModuleException extends ModuleException {
public AgentStreamModuleException(String message) {
super(message);
}
public AgentStreamModuleException(String message, Throwable cause) {
super(message, cause);
}
}

8
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/client/Client.java Normal file
View File

@ -0,0 +1,8 @@
package org.skywalking.apm.collector.core.client;
/**
* @author pengys5
*/
public interface Client {
void initialize() throws ClientException;
}

16
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/client/ClientException.java Normal file
View File

@ -0,0 +1,16 @@
package org.skywalking.apm.collector.core.client;
import org.skywalking.apm.collector.core.CollectorException;
/**
* @author pengys5
*/
public abstract class ClientException extends CollectorException {
public ClientException(String message) {
super(message);
}
public ClientException(String message, Throwable cause) {
super(message, cause);
}
}

12
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/client/DataListener.java Normal file
View File

@ -0,0 +1,12 @@
package org.skywalking.apm.collector.core.client;
import java.util.List;
/**
* @author pengys5
*/
public interface DataListener {
List<String> items();
void listen() throws ClientException;
}

24
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterDataInitializer.java Normal file
View File

@ -0,0 +1,24 @@
package org.skywalking.apm.collector.core.cluster;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.framework.DataInitializer;
/**
* @author pengys5
*/
public abstract class ClusterDataInitializer implements DataInitializer {
public static final String BASE_CATALOG = "skywalking";
public static final String FOR_UI_CATALOG = BASE_CATALOG + ".ui";
public static final String FOR_AGENT_CATALOG = BASE_CATALOG + ".agent";
@Override public final void initialize(Client client) throws ClientException {
if (!existItem(client, FOR_UI_CATALOG)) {
addItem(client, FOR_UI_CATALOG);
}
if (!existItem(client, FOR_AGENT_CATALOG)) {
addItem(client, FOR_AGENT_CATALOG);
}
}
}

13
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterDefinitionFile.java Normal file
View File

@ -0,0 +1,13 @@
package org.skywalking.apm.collector.core.cluster;
import org.skywalking.apm.collector.core.framework.DefinitionFile;
/**
* @author pengys5
*/
public class ClusterDefinitionFile extends DefinitionFile {
@Override protected String fileName() {
return "cluster-configuration.define";
}
}

10
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleContext.java Normal file
View File

@ -0,0 +1,10 @@
package org.skywalking.apm.collector.core.cluster;
/**
* @author pengys5
*/
public class ClusterModuleContext {
public static ClusterModuleRegistrationWriter WRITER;
public static ClusterModuleRegistrationReader READER;
}

44
apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleDefine.java Normal file
View File

@ -0,0 +1,44 @@
package org.skywalking.apm.collector.core.cluster;
import java.util.Map;
import org.skywalking.apm.collector.core.client.Client;
import org.skywalking.apm.collector.core.client.ClientException;
import org.skywalking.apm.collector.core.config.ConfigParseException;
import org.skywalking.apm.collector.core.module.ModuleDefine;
import org.skywalking.apm.collector.core.module.ModuleRegistration;
import org.skywalking.apm.collector.core.server.Server;
/**
* @author pengys5
*/
public abstract class ClusterModuleDefine extends ModuleDefine {
private Client client;
@Override public final void initialize(Map config) throws ClusterModuleException {
try {
configParser().parse(config);
client = createClient();
client.initialize();
dataInitializer().initialize(client);
} catch (ConfigParseException | ClientException e) {
throw new ClusterModuleException(e.getMessage(), e);
}
}
public final Client getClient() {
return this.client;
}
@Override public final Server server() {
throw new UnsupportedOperationException("");
}
@Override protected final ModuleRegistration registration() {
throw new UnsupportedOperationException("Cluster module do not need module registration.");
}
protected abstract ClusterModuleRegistrationWriter registrationWriter();
protected abstract ClusterModuleRegistrationReader registrationReader();
}

Some files were not shown because too many files have changed in this diff Show More