Skip to content

S
SkyWalking

javalover123888 / SkyWalking
与 Fork 源项目一致

Fork自 山不在高_有仙则灵 / SkyWalking

通知 2
Star 0
Fork 0
S
SkyWalking

体验新版 GitCode，发现更多精彩内容 >>

提交 a6ca67f6 编写于 作者: P pengys5
浏览文件
操作

main test

上级 29fcbc04
隐藏空白更改
内联 并排
Showing with 249 addition and 91 deletion
skywalking-collector/pom.xml
浏览文件 @ a6ca67f6
<?xml version="1.0" encoding="UTF-8"?> <?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" <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/maven-v4_0_0.xsd"> xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<modules> <modules>
<module>skywalking-collector-cluster</module> <module>skywalking-collector-cluster</module>
<module>skywalking-collector-worker</module> <module>skywalking-collector-worker</module>
</modules> </modules>
<parent> <parent>
<artifactId>skywalking</artifactId> <artifactId>skywalking</artifactId>
<groupId>com.a.eye</groupId> <groupId>com.a.eye</groupId>
<version>3.0-2017</version> <version>3.0-2017</version>
</parent> </parent>
<artifactId>skywalking-collector</artifactId> <artifactId>skywalking-collector</artifactId>
<packaging>pom</packaging> <packaging>pom</packaging>
<properties> <properties>
<akka.version>2.4.17</akka.version> <akka.version>2.4.17</akka.version>
</properties> </properties>
<dependencies> <dependencies>
<dependency> <dependency>
<groupId>com.typesafe.akka</groupId> <groupId>com.typesafe.akka</groupId>
<artifactId>akka-cluster_2.11</artifactId> <artifactId>akka-cluster_2.11</artifactId>
<version>${akka.version}</version> <version>${akka.version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.typesafe.akka</groupId> <groupId>com.typesafe.akka</groupId>
<artifactId>akka-cluster-metrics_2.11</artifactId> <artifactId>akka-cluster-metrics_2.11</artifactId>
<version>${akka.version}</version> <version>${akka.version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.typesafe.akka</groupId> <groupId>com.typesafe.akka</groupId>
<artifactId>akka-cluster-tools_2.11</artifactId> <artifactId>akka-cluster-tools_2.11</artifactId>
<version>${akka.version}</version> <version>${akka.version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.typesafe.akka</groupId> <groupId>com.typesafe.akka</groupId>
<artifactId>akka-persistence_2.11</artifactId> <artifactId>akka-persistence_2.11</artifactId>
<version>${akka.version}</version> <version>${akka.version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.iq80.leveldb</groupId> <groupId>org.iq80.leveldb</groupId>
<artifactId>leveldb</artifactId> <artifactId>leveldb</artifactId>
<version>0.9</version> <version>0.9</version>
</dependency> </dependency>
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-logging-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-logging-impl-log4j2</artifactId>
<version>${project.version}</version>
</dependency>
<dependency> <dependency>
<groupId>com.typesafe.akka</groupId> <groupId>com.typesafe.akka</groupId>
<artifactId>akka-testkit_2.11</artifactId> <artifactId>akka-testkit_2.11</artifactId>
<version>${akka.version}</version> <version>${akka.version}</version>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.a.eye</groupId> <groupId>com.a.eye</groupId>
<artifactId>skywalking-sniffer-mock</artifactId> <artifactId>skywalking-sniffer-mock</artifactId>
<version>${project.version}</version> <version>${project.version}</version>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
</dependencies> </dependencies>
</project> </project>
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMember.java
浏览文件 @ a6ca67f6
...@@ -29,7 +29,7 @@ public abstract class AbstractMember<T> { ...@@ -29,7 +29,7 @@ public abstract class AbstractMember<T> {
} }
public abstract void preStart() throws Throwable; public abstract void preStart() throws Exception;
/** /**
* Receive the message to analyse. * Receive the message to analyse.
......
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMemberProvider.java
浏览文件 @ a6ca67f6
...@@ -18,6 +18,7 @@ public abstract class AbstractMemberProvider { ...@@ -18,6 +18,7 @@ public abstract class AbstractMemberProvider {
Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class[]{MemberSystem.class, ActorRef.class}); Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class[]{MemberSystem.class, ActorRef.class});
memberConstructor.setAccessible(true); memberConstructor.setAccessible(true);
AbstractMember member = (AbstractMember) memberConstructor.newInstance(system, actorRef); AbstractMember member = (AbstractMember) memberConstructor.newInstance(system, actorRef);
member.preStart();
system.memberOf(member, roleName()); system.memberOf(member, roleName());
} }
......
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorker.java
浏览文件 @ a6ca67f6
package com.a.eye.skywalking.collector.actor; package com.a.eye.skywalking.collector.actor;
import akka.actor.Terminated;
import akka.actor.UntypedActor; import akka.actor.UntypedActor;
import akka.cluster.ClusterEvent; import akka.cluster.ClusterEvent;
import akka.cluster.Member; import akka.cluster.Member;
...@@ -8,6 +9,9 @@ import com.a.eye.skywalking.collector.actor.selector.WorkerSelector; ...@@ -8,6 +9,9 @@ import com.a.eye.skywalking.collector.actor.selector.WorkerSelector;
import com.a.eye.skywalking.collector.cluster.WorkerListenerMessage; import com.a.eye.skywalking.collector.cluster.WorkerListenerMessage;
import com.a.eye.skywalking.collector.cluster.WorkersListener; import com.a.eye.skywalking.collector.cluster.WorkersListener;
import com.a.eye.skywalking.collector.cluster.WorkersRefCenter; import com.a.eye.skywalking.collector.cluster.WorkersRefCenter;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import scala.Option;
import java.util.List; import java.util.List;
...@@ -37,8 +41,22 @@ import java.util.List; ...@@ -37,8 +41,22 @@ import java.util.List;
*/ */
public abstract class AbstractWorker<T> extends UntypedActor { public abstract class AbstractWorker<T> extends UntypedActor {
private Logger logger = LogManager.getFormatterLogger(AbstractWorker.class);
private MemberSystem memberSystem = new MemberSystem(); private MemberSystem memberSystem = new MemberSystem();
@Override
public void preStart() throws Exception {
super.preStart();
register();
}
@Override
public void preRestart(Throwable reason, Option<Object> message) throws Exception {
super.preRestart(reason, message);
register();
}
/** /**
* Receive the message to analyse. * Receive the message to analyse.
* *
...@@ -55,6 +73,7 @@ public abstract class AbstractWorker<T> extends UntypedActor { ...@@ -55,6 +73,7 @@ public abstract class AbstractWorker<T> extends UntypedActor {
@Override @Override
public void onReceive(Object message) throws Throwable { public void onReceive(Object message) throws Throwable {
if (message instanceof ClusterEvent.CurrentClusterState) { if (message instanceof ClusterEvent.CurrentClusterState) {
logger.info("receive ClusterEvent.CurrentClusterState message");
ClusterEvent.CurrentClusterState state = (ClusterEvent.CurrentClusterState) message; ClusterEvent.CurrentClusterState state = (ClusterEvent.CurrentClusterState) message;
for (Member member : state.getMembers()) { for (Member member : state.getMembers()) {
if (member.status().equals(MemberStatus.up())) { if (member.status().equals(MemberStatus.up())) {
...@@ -62,9 +81,11 @@ public abstract class AbstractWorker<T> extends UntypedActor { ...@@ -62,9 +81,11 @@ public abstract class AbstractWorker<T> extends UntypedActor {
} }
} }
} else if (message instanceof ClusterEvent.MemberUp) { } else if (message instanceof ClusterEvent.MemberUp) {
logger.info("receive ClusterEvent.MemberUp message");
ClusterEvent.MemberUp memberUp = (ClusterEvent.MemberUp) message; ClusterEvent.MemberUp memberUp = (ClusterEvent.MemberUp) message;
register(memberUp.member()); register(memberUp.member());
} else { } else {
logger.info("message class: %s", message.getClass().getName());
receive(message); receive(message);
} }
} }
...@@ -89,12 +110,18 @@ public abstract class AbstractWorker<T> extends UntypedActor { ...@@ -89,12 +110,18 @@ public abstract class AbstractWorker<T> extends UntypedActor {
* @param member is the new created or restart worker * @param member is the new created or restart worker
*/ */
void register(Member member) { void register(Member member) {
System.out.println("register");
if (member.getRoles().equals(WorkersListener.WorkName)) { if (member.getRoles().equals(WorkersListener.WorkName)) {
WorkerListenerMessage.RegisterMessage registerMessage = new WorkerListenerMessage.RegisterMessage(getClass().getSimpleName()); WorkerListenerMessage.RegisterMessage registerMessage = new WorkerListenerMessage.RegisterMessage(getClass().getSimpleName());
getContext().actorSelection(member.address() + "/user/" + WorkersListener.WorkName).tell(registerMessage, getSelf()); getContext().actorSelection(member.address() + "/user/" + WorkersListener.WorkName).tell(registerMessage, getSelf());
} }
} }
void register() {
WorkerListenerMessage.RegisterMessage registerMessage = new WorkerListenerMessage.RegisterMessage(getClass().getSimpleName());
getContext().actorSelection("/user/" + WorkersListener.WorkName).tell(registerMessage, getSelf());
}
public MemberSystem memberContext() { public MemberSystem memberContext() {
return memberSystem; return memberSystem;
} }
......
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorkerProvider.java
浏览文件 @ a6ca67f6
...@@ -2,6 +2,8 @@ package com.a.eye.skywalking.collector.actor; ...@@ -2,6 +2,8 @@ package com.a.eye.skywalking.collector.actor;
import akka.actor.ActorSystem; import akka.actor.ActorSystem;
import akka.actor.Props; import akka.actor.Props;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
/** /**
* The <code>AbstractWorkerProvider</code> should be implemented by any class whose * The <code>AbstractWorkerProvider</code> should be implemented by any class whose
...@@ -27,6 +29,8 @@ import akka.actor.Props; ...@@ -27,6 +29,8 @@ import akka.actor.Props;
*/ */
public abstract class AbstractWorkerProvider<T> { public abstract class AbstractWorkerProvider<T> {
private Logger logger = LogManager.getFormatterLogger(AbstractWorkerProvider.class);
public abstract Class workerClass(); public abstract Class workerClass();
public abstract int workerNum(); public abstract int workerNum();
...@@ -41,6 +45,7 @@ public abstract class AbstractWorkerProvider<T> { ...@@ -41,6 +45,7 @@ public abstract class AbstractWorkerProvider<T> {
for (int i = 1; i <= workerNum(); i++) { for (int i = 1; i <= workerNum(); i++) {
system.actorOf(Props.create(workerClass()), roleName() + "_" + i); system.actorOf(Props.create(workerClass()), roleName() + "_" + i);
logger.info("create akka actor, actor id is %s", roleName() + "_" + i);
} }
} }
......
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkersCreator.java
浏览文件 @ a6ca67f6
package com.a.eye.skywalking.collector.actor; package com.a.eye.skywalking.collector.actor;
import akka.actor.ActorSystem; import akka.actor.ActorSystem;
import akka.actor.Props;
import com.a.eye.skywalking.collector.cluster.WorkersListener;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ServiceLoader; import java.util.ServiceLoader;
...@@ -13,14 +17,19 @@ import java.util.ServiceLoader; ...@@ -13,14 +17,19 @@ import java.util.ServiceLoader;
public enum WorkersCreator { public enum WorkersCreator {
INSTANCE; INSTANCE;
private Logger logger = LogManager.getFormatterLogger(WorkersCreator.class);
/** /**
* create worker to use Java Spi. * create worker to use Java Spi.
* *
* @param system is create by akka {@link ActorSystem} * @param system is create by akka {@link ActorSystem}
*/ */
public void boot(ActorSystem system) { public void boot(ActorSystem system) {
system.actorOf(Props.create(WorkersListener.class), WorkersListener.WorkName);
ServiceLoader<AbstractWorkerProvider> clusterServiceLoader = ServiceLoader.load(AbstractWorkerProvider.class); ServiceLoader<AbstractWorkerProvider> clusterServiceLoader = ServiceLoader.load(AbstractWorkerProvider.class);
for (AbstractWorkerProvider provider : clusterServiceLoader) { for (AbstractWorkerProvider provider : clusterServiceLoader) {
logger.info("create worker {%s} using java service loader", provider.workerClass().getName());
provider.createWorker(system); provider.createWorker(system);
} }
} }
......
skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/WorkersListener.java
浏览文件 @ a6ca67f6
...@@ -3,6 +3,8 @@ package com.a.eye.skywalking.collector.cluster; ...@@ -3,6 +3,8 @@ package com.a.eye.skywalking.collector.cluster;
import akka.actor.ActorRef; import akka.actor.ActorRef;
import akka.actor.Terminated; import akka.actor.Terminated;
import akka.actor.UntypedActor; import akka.actor.UntypedActor;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
/** /**
* <code>WorkersListener</code> listening the register message from workers * <code>WorkersListener</code> listening the register message from workers
...@@ -18,6 +20,8 @@ import akka.actor.UntypedActor; ...@@ -18,6 +20,8 @@ import akka.actor.UntypedActor;
*/ */
public class WorkersListener extends UntypedActor { public class WorkersListener extends UntypedActor {
private Logger logger = LogManager.getFormatterLogger(WorkersListener.class);
public static final String WorkName = "WorkersListener"; public static final String WorkName = "WorkersListener";
@Override @Override
...@@ -27,6 +31,8 @@ public class WorkersListener extends UntypedActor { ...@@ -27,6 +31,8 @@ public class WorkersListener extends UntypedActor {
ActorRef sender = getSender(); ActorRef sender = getSender();
getContext().watch(sender); getContext().watch(sender);
logger.info("register worker of role %s", register.getWorkRole());
WorkersRefCenter.INSTANCE.register(sender, register.getWorkRole()); WorkersRefCenter.INSTANCE.register(sender, register.getWorkRole());
} else if (message instanceof Terminated) { } else if (message instanceof Terminated) {
Terminated terminated = (Terminated) message; Terminated terminated = (Terminated) message;
......
skywalking-collector/skywalking-collector-worker/pom.xml
浏览文件 @ a6ca67f6
...@@ -28,5 +28,11 @@ ...@@ -28,5 +28,11 @@
<artifactId>transport</artifactId> <artifactId>transport</artifactId>
<version>5.2.2</version> <version>5.2.2</version>
</dependency> </dependency>
<dependency>
<groupId>com.a.eye</groupId>
<artifactId>skywalking-sniffer-mock</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies> </dependencies>
</project> </project>
\ No newline at end of file
skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/CollectorBootStartUp.java
浏览文件 @ a6ca67f6
package com.a.eye.skywalking.collector.worker; package com.a.eye.skywalking.collector.worker;
import akka.actor.ActorSystem;
import com.a.eye.skywalking.collector.actor.WorkersCreator;
import com.a.eye.skywalking.collector.cluster.ClusterConfig;
import com.a.eye.skywalking.collector.cluster.ClusterConfigInitializer;
import com.a.eye.skywalking.collector.cluster.NoAvailableWorkerException;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
/** /**
* @author pengys5 * @author pengys5
*/ */
public class CollectorBootStartUp { public class CollectorBootStartUp {
public static void main(String[] args) { public static void main(String[] args) throws NoAvailableWorkerException, InterruptedException {
// ActorSystem system = ActorSystem.create("ClusterSystem", config); ClusterConfigInitializer.initialize("collector.config");
// system.actorOf(Props.create(TraceConsumerActor.class), Const.Trace_Consumer_Role);
final Config config = ConfigFactory.parseString("akka.remote.netty.tcp.port=" + ClusterConfig.Cluster.Current.port).
withFallback(ConfigFactory.parseString("akka.cluster.roles = [" + ClusterConfig.Cluster.Current.roles + "]")).
withFallback(ConfigFactory.load());
ActorSystem system = ActorSystem.create(ClusterConfig.Cluster.appname, config);
WorkersCreator.INSTANCE.boot(system);
} }
} }
skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMember.java
浏览文件 @ a6ca67f6
...@@ -13,18 +13,23 @@ import com.a.eye.skywalking.collector.worker.application.persistence.ResponseSum ...@@ -13,18 +13,23 @@ import com.a.eye.skywalking.collector.worker.application.persistence.ResponseSum
import com.a.eye.skywalking.trace.Span; import com.a.eye.skywalking.trace.Span;
import com.a.eye.skywalking.trace.TraceSegment; import com.a.eye.skywalking.trace.TraceSegment;
import com.a.eye.skywalking.trace.tag.Tags; import com.a.eye.skywalking.trace.tag.Tags;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
/** /**
* @author pengys5 * @author pengys5
*/ */
public class ApplicationMember extends AbstractMember { public class ApplicationMember extends AbstractMember {
private Logger logger = LogManager.getFormatterLogger(ApplicationMember.class);
public ApplicationMember(MemberSystem memberSystem, ActorRef actorRef) { public ApplicationMember(MemberSystem memberSystem, ActorRef actorRef) {
super(memberSystem, actorRef); super(memberSystem, actorRef);
} }
@Override @Override
public void preStart() throws Throwable { public void preStart() throws Exception {
logger.info("create members");
TraceSegmentRecordMember.Factory factory = new TraceSegmentRecordMember.Factory(); TraceSegmentRecordMember.Factory factory = new TraceSegmentRecordMember.Factory();
factory.createWorker(memberContext(), getSelf()); factory.createWorker(memberContext(), getSelf());
} }
...@@ -32,6 +37,7 @@ public class ApplicationMember extends AbstractMember { ...@@ -32,6 +37,7 @@ public class ApplicationMember extends AbstractMember {
@Override @Override
public void receive(Object message) throws Throwable { public void receive(Object message) throws Throwable {
if (message instanceof TraceSegment) { if (message instanceof TraceSegment) {
logger.debug("begin translate TraceSegment Object to JsonObject");
TraceSegment traceSegment = (TraceSegment) message; TraceSegment traceSegment = (TraceSegment) message;
AbstractMember discoverMember = memberContext().memberFor(TraceSegmentRecordMember.class.getSimpleName()); AbstractMember discoverMember = memberContext().memberFor(TraceSegmentRecordMember.class.getSimpleName());
discoverMember.receive(traceSegment); discoverMember.receive(traceSegment);
...@@ -67,11 +73,13 @@ public class ApplicationMember extends AbstractMember { ...@@ -67,11 +73,13 @@ public class ApplicationMember extends AbstractMember {
} }
private void sendToNodeInstancePersistence(TraceSegment traceSegment) throws Throwable { private void sendToNodeInstancePersistence(TraceSegment traceSegment) throws Throwable {
String code = traceSegment.getPrimaryRef().getApplicationCode(); if (traceSegment.getPrimaryRef() != null) {
String address = traceSegment.getPrimaryRef().getPeerHost(); String code = traceSegment.getPrimaryRef().getApplicationCode();
String address = traceSegment.getPrimaryRef().getPeerHost();
NodeInstancePersistence.Metric property = new NodeInstancePersistence.Metric(code, address); NodeInstancePersistence.Metric property = new NodeInstancePersistence.Metric(code, address);
tell(new NodeInstancePersistence.Factory(), RollingSelector.INSTANCE, property); tell(new NodeInstancePersistence.Factory(), RollingSelector.INSTANCE, property);
}
} }
private void sendToResponseCostPersistence(TraceSegment traceSegment) throws Throwable { private void sendToResponseCostPersistence(TraceSegment traceSegment) throws Throwable {
......
skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/metric/TraceSegmentRecordMember.java
浏览文件 @ a6ca67f6
...@@ -21,12 +21,12 @@ import java.util.Map; ...@@ -21,12 +21,12 @@ import java.util.Map;
*/ */
public class TraceSegmentRecordMember extends AbstractMember { public class TraceSegmentRecordMember extends AbstractMember {
public TraceSegmentRecordMember(MemberSystem memberSystem, ActorRef actorRef) { public TraceSegmentRecordMember(MemberSystem memberSystem, ActorRef actorRef) throws Throwable {
super(memberSystem, actorRef); super(memberSystem, actorRef);
} }
@Override @Override
public void preStart() throws Throwable { public void preStart() throws Exception {
} }
@Override @Override
...@@ -53,11 +53,15 @@ public class TraceSegmentRecordMember extends AbstractMember { ...@@ -53,11 +53,15 @@ public class TraceSegmentRecordMember extends AbstractMember {
traceJsonObj.addProperty("endTime", traceSegment.getEndTime()); traceJsonObj.addProperty("endTime", traceSegment.getEndTime());
traceJsonObj.addProperty("appCode", traceSegment.getApplicationCode()); traceJsonObj.addProperty("appCode", traceSegment.getApplicationCode());
JsonObject primaryRefJsonObj = parsePrimaryRef(traceSegment.getPrimaryRef()); if (traceSegment.getPrimaryRef() != null) {
traceJsonObj.add("primaryRef", primaryRefJsonObj); JsonObject primaryRefJsonObj = parsePrimaryRef(traceSegment.getPrimaryRef());
traceJsonObj.add("primaryRef", primaryRefJsonObj);
}
JsonArray refsJsonArray = parseRefs(traceSegment.getRefs()); // if (traceSegment.getRefs() != null) {
traceJsonObj.add("refs", refsJsonArray); // JsonArray refsJsonArray = parseRefs(traceSegment.getRefs());
// traceJsonObj.add("refs", refsJsonArray);
// }
JsonArray spanJsonArray = new JsonArray(); JsonArray spanJsonArray = new JsonArray();
for (Span span : traceSegment.getSpans()) { for (Span span : traceSegment.getSpans()) {
......
skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/ApplicationRefMember.java
浏览文件 @ a6ca67f6
...@@ -13,12 +13,12 @@ import com.a.eye.skywalking.trace.TraceSegment; ...@@ -13,12 +13,12 @@ import com.a.eye.skywalking.trace.TraceSegment;
*/ */
public class ApplicationRefMember extends AbstractMember { public class ApplicationRefMember extends AbstractMember {
public ApplicationRefMember(MemberSystem memberSystem, ActorRef actorRef) { public ApplicationRefMember(MemberSystem memberSystem, ActorRef actorRef) throws Throwable {
super(memberSystem, actorRef); super(memberSystem, actorRef);
} }
@Override @Override
public void preStart() throws Throwable { public void preStart() {
} }
......
skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/metric/TraceSegmentRelationActor.java 已删除 100644 → 0
浏览文件 @ 29fcbc04
package com.a.eye.skywalking.collector.worker.metric;
import akka.actor.UntypedActor;
/**
* @author pengys5
*/
public class TraceSegmentRelationActor extends UntypedActor{
@Override
public void onReceive(Object message) throws Throwable {
}
}
skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/receiver/TraceSegmentReceiver.java
浏览文件 @ a6ca67f6
...@@ -7,22 +7,28 @@ import com.a.eye.skywalking.collector.worker.WorkerConfig; ...@@ -7,22 +7,28 @@ import com.a.eye.skywalking.collector.worker.WorkerConfig;
import com.a.eye.skywalking.collector.worker.application.ApplicationMember; import com.a.eye.skywalking.collector.worker.application.ApplicationMember;
import com.a.eye.skywalking.collector.worker.applicationref.ApplicationRefMember; import com.a.eye.skywalking.collector.worker.applicationref.ApplicationRefMember;
import com.a.eye.skywalking.trace.TraceSegment; import com.a.eye.skywalking.trace.TraceSegment;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
/** /**
* @author pengys5 * @author pengys5
*/ */
public class TraceSegmentReceiver extends AbstractWorker { public class TraceSegmentReceiver extends AbstractWorker {
private Logger logger = LogManager.getFormatterLogger(TraceSegmentReceiver.class);
@Override @Override
public void preStart() throws Exception { public void preStart() throws Exception {
ApplicationMember.Factory factory = new ApplicationMember.Factory(); ApplicationMember.Factory factory = new ApplicationMember.Factory();
factory.createWorker(memberContext(), getSelf()); factory.createWorker(memberContext(), getSelf());
super.preStart();
} }
@Override @Override
public void receive(Object message) throws Throwable { public void receive(Object message) throws Throwable {
if (message instanceof TraceSegment) { if (message instanceof TraceSegment) {
TraceSegment traceSegment = (TraceSegment) message; TraceSegment traceSegment = (TraceSegment) message;
logger.debug("receive message instanceof TraceSegment, traceSegmentId is %s", traceSegment.getTraceSegmentId());
AbstractMember applicationMember = memberContext().memberFor(ApplicationMember.class.getSimpleName()); AbstractMember applicationMember = memberContext().memberFor(ApplicationMember.class.getSimpleName());
applicationMember.receive(traceSegment); applicationMember.receive(traceSegment);
...@@ -32,7 +38,7 @@ public class TraceSegmentReceiver extends AbstractWorker { ...@@ -32,7 +38,7 @@ public class TraceSegmentReceiver extends AbstractWorker {
} }
} }
public class Factory extends AbstractWorkerProvider { public static class Factory extends AbstractWorkerProvider {
@Override @Override
public Class workerClass() { public Class workerClass() {
return TraceSegmentReceiver.class; return TraceSegmentReceiver.class;
......
skywalking-collector/skywalking-collector-worker/src/main/resources/META-INF/services/com.a.eye.skywalking.collector.actor.AbstractWorkerProvider
浏览文件 @ a6ca67f6
com.a.eye.skywalking.collector.actor.SpiTestWorkerFactory com.a.eye.skywalking.collector.worker.receiver.TraceSegmentReceiver$Factory
\ No newline at end of file com.a.eye.skywalking.collector.worker.application.persistence.TraceSegmentRecordPersistence$Factory
com.a.eye.skywalking.collector.worker.application.persistence.DAGNodePersistence$Factory
com.a.eye.skywalking.collector.worker.application.persistence.NodeInstancePersistence$Factory
com.a.eye.skywalking.collector.worker.application.persistence.ResponseCostPersistence$Factory
com.a.eye.skywalking.collector.worker.application.persistence.ResponseSummaryPersistence$Factory
com.a.eye.skywalking.collector.worker.application.persistence.TraceSegmentRecordPersistence$Factory
com.a.eye.skywalking.collector.worker.applicationref.presistence.DAGNodeRefPersistence$Factory
\ No newline at end of file
skywalking-collector/skywalking-collector-worker/src/main/resources/collector.config
浏览文件 @ a6ca67f6
cluster.current.hostname = 192.168.0.1 cluster.current.hostname = 127.0.0.1
cluster.current.port = 1000 cluster.current.port = 1000
cluster.current.roles = [Test, Test1] cluster.current.roles = [TraceSegmentReceiver]
cluster.nodes = [192.168.0.1:1000, 192.168.0.2:1000] cluster.nodes = ["akka.tcp://CollectorSystem@127.0.0.1:1000"]
\ No newline at end of file \ No newline at end of file
skywalking-collector/skywalking-collector-worker/src/main/resources/log4j2.xml 0 → 100644
浏览文件 @ a6ca67f6
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="info">
<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="info">
<AppenderRef ref="Console" />
</Root>
</Loggers>
</Configuration>
skywalking-collector/skywalking-collector-worker/src/test/java/com/a/eye/skywalking/collector/worker/StartUpTestCase.java 0 → 100644
浏览文件 @ a6ca67f6
package com.a.eye.skywalking.collector.worker;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import com.a.eye.skywalking.collector.actor.WorkerRef;
import com.a.eye.skywalking.collector.actor.WorkersCreator;
import com.a.eye.skywalking.collector.actor.selector.RollingSelector;
import com.a.eye.skywalking.collector.cluster.ClusterConfig;
import com.a.eye.skywalking.collector.cluster.ClusterConfigInitializer;
import com.a.eye.skywalking.collector.cluster.WorkersRefCenter;
import com.a.eye.skywalking.collector.worker.receiver.TraceSegmentReceiver;
import com.a.eye.skywalking.sniffer.mock.trace.TraceSegmentBuilderFactory;
import com.a.eye.skywalking.trace.TraceSegment;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.junit.Test;
import org.powermock.api.support.membermodification.MemberModifier;
import java.util.List;
/**
* @author pengys5
*/
public class StartUpTestCase {
@Test
public void test() throws Exception {
ClusterConfigInitializer.initialize("collector.config");
System.out.println(ClusterConfig.Cluster.Current.roles);
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.cluster.roles=" + ClusterConfig.Cluster.Current.roles)).
withFallback(ConfigFactory.parseString("akka.actor.provider=" + ClusterConfig.Cluster.provider)).
withFallback(ConfigFactory.parseString("akka.cluster.seed-nodes=" + ClusterConfig.Cluster.nodes)).
withFallback(ConfigFactory.load());
ActorSystem system = ActorSystem.create(ClusterConfig.Cluster.appname, config);
WorkersCreator.INSTANCE.boot(system);
Thread.sleep(2000);
for (int i = 0; i < 1; i++) {
TraceSegment traceSegment = TraceSegmentBuilderFactory.INSTANCE.singleTomcat200Trace();
List<WorkerRef> availableWorks = WorkersRefCenter.INSTANCE.availableWorks(TraceSegmentReceiver.class.getSimpleName());
WorkerRef workerRef = RollingSelector.INSTANCE.select(availableWorks, traceSegment);
ActorRef actorRef = (ActorRef) MemberModifier.field(WorkerRef.class, "actorRef").get(workerRef);
actorRef.tell(traceSegment, ActorRef.noSender());
}
Thread.sleep(10000);
}
}
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或