diff --git a/skywalking-collector/skywalking-collector-cluster/pom.xml b/skywalking-collector/skywalking-collector-cluster/pom.xml index c47b1d868..0f40be50a 100644 --- a/skywalking-collector/skywalking-collector-cluster/pom.xml +++ b/skywalking-collector/skywalking-collector-cluster/pom.xml @@ -13,7 +13,6 @@ jar - 4.1.6.RELEASE @@ -23,9 +22,9 @@ 1 - org.springframework - spring-context-support - ${project.spring.version} + com.lmax + disruptor + 3.3.6 com.a.eye diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractASyncMemberProvider.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractASyncMemberProvider.java new file mode 100644 index 000000000..23207cf88 --- /dev/null +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractASyncMemberProvider.java @@ -0,0 +1,61 @@ +package com.a.eye.skywalking.collector.actor; + +import akka.actor.ActorRef; +import com.a.eye.skywalking.collector.queue.DaemonThreadFactory; +import com.a.eye.skywalking.collector.queue.MessageHolder; +import com.lmax.disruptor.EventFactory; +import com.lmax.disruptor.EventHandler; +import com.lmax.disruptor.RingBuffer; +import com.lmax.disruptor.dsl.Disruptor; + +import java.lang.reflect.Constructor; + +/** + * @author pengys5 + */ +public abstract class AbstractASyncMemberProvider { + + private RingBuffer ringBuffer; + + public abstract Class memberClass(); + + public T createWorker(EventFactory eventFactory, ActorRef actorRef) throws Exception { + if (memberClass() == null) { + throw new IllegalArgumentException("cannot createInstance() with nothing obtained from memberClass()"); + } + + Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class[]{ActorRef.class}); + memberConstructor.setAccessible(true); + T member = (T) memberConstructor.newInstance(actorRef); + + // Specify the size of the ring buffer, must be power of 2. + int bufferSize = 1024; + // Construct the Disruptor + Disruptor disruptor = new Disruptor(eventFactory, bufferSize, DaemonThreadFactory.INSTANCE); + // Connect the handler + disruptor.handleEventsWith(member); + // Start the Disruptor, starts all threads running + disruptor.start(); + // Get the ring buffer from the Disruptor to be used for publishing. + ringBuffer = disruptor.getRingBuffer(); + return member; + } + + public void onData(MessageHolder message) { + long sequence = ringBuffer.next(); + try { + ringBuffer.get(sequence).setMessage(message); + } finally { + ringBuffer.publish(sequence); + } + } + + /** + * Use {@link #memberClass()} method returned class's simple name as a role name. + * + * @return is role of Worker + */ + protected String roleName() { + return memberClass().getSimpleName(); + } +} diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMember.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMember.java index 8514ebeaf..ae9d5d6bb 100644 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMember.java +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMember.java @@ -3,41 +3,43 @@ package com.a.eye.skywalking.collector.actor; import akka.actor.ActorRef; import com.a.eye.skywalking.collector.actor.selector.WorkerSelector; import com.a.eye.skywalking.collector.cluster.WorkersRefCenter; +import com.a.eye.skywalking.collector.queue.MessageHolder; +import com.lmax.disruptor.EventHandler; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import java.util.List; /** * @author pengys5 */ -public abstract class AbstractMember { +public abstract class AbstractMember implements EventHandler> { - private MemberSystem memberSystem; + private Logger logger = LogManager.getFormatterLogger(AbstractMember.class); private ActorRef actorRef; - public MemberSystem memberContext() { - return memberSystem; - } - public ActorRef getSelf() { return actorRef; } - public AbstractMember(MemberSystem memberSystem, ActorRef actorRef) { - this.memberSystem = memberSystem; + public AbstractMember(ActorRef actorRef) { this.actorRef = actorRef; } - - public abstract void preStart() throws Exception; - /** * Receive the message to analyse. * * @param message is the data send from the forward worker - * @throws Throwable is the exception thrown by that worker implementation processing + * @throws Exception is the exception thrown by that worker implementation processing */ - public abstract void receive(Object message) throws Throwable; + public abstract void receive(Object message) throws Exception; + + public void onEvent(MessageHolder event, long sequence, boolean endOfBatch) throws Exception { + T message = event.getMessage(); + event.reset(); + receive(message); + } /** * Send analysed data to next Worker. @@ -45,9 +47,10 @@ public abstract class AbstractMember { * @param targetWorkerProvider is the worker provider to create worker instance. * @param selector is the selector to select a same role worker instance form cluster. * @param message is the data used to send to next worker. - * @throws Throwable + * @throws Exception */ - public void tell(AbstractWorkerProvider targetWorkerProvider, WorkerSelector selector, T message) throws Throwable { + public void tell(AbstractWorkerProvider targetWorkerProvider, WorkerSelector selector, T message) throws Exception { + logger.debug("worker provider: %s ,role name: %s", targetWorkerProvider.getClass().getName(), targetWorkerProvider.roleName()); List availableWorks = WorkersRefCenter.INSTANCE.availableWorks(targetWorkerProvider.roleName()); selector.select(availableWorks, message).tell(message, getSelf()); } diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMemberProvider.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractSyncMemberProvider.java similarity index 65% rename from skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMemberProvider.java rename to skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractSyncMemberProvider.java index 16c5041e2..a1016e4f4 100644 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractMemberProvider.java +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractSyncMemberProvider.java @@ -7,19 +7,19 @@ import java.lang.reflect.Constructor; /** * @author pengys5 */ -public abstract class AbstractMemberProvider { +public abstract class AbstractSyncMemberProvider { + public abstract Class memberClass(); - public void createWorker(MemberSystem system, ActorRef actorRef) throws Exception { + public T createWorker(ActorRef actorRef) throws Exception { if (memberClass() == null) { throw new IllegalArgumentException("cannot createInstance() with nothing obtained from memberClass()"); } - Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class[]{MemberSystem.class, ActorRef.class}); + Constructor memberConstructor = memberClass().getDeclaredConstructor(new Class[]{ActorRef.class}); memberConstructor.setAccessible(true); - AbstractMember member = (AbstractMember) memberConstructor.newInstance(system, actorRef); - member.preStart(); - system.memberOf(member, roleName()); + T member = (T) memberConstructor.newInstance(actorRef); + return member; } /** diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorker.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorker.java index c11fb494c..d66f88473 100644 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorker.java +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorker.java @@ -1,7 +1,7 @@ package com.a.eye.skywalking.collector.actor; -import akka.actor.Terminated; import akka.actor.UntypedActor; +import akka.cluster.Cluster; import akka.cluster.ClusterEvent; import akka.cluster.Member; import akka.cluster.MemberStatus; @@ -11,7 +11,6 @@ import com.a.eye.skywalking.collector.cluster.WorkersListener; 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; @@ -43,18 +42,17 @@ public abstract class AbstractWorker extends UntypedActor { private Logger logger = LogManager.getFormatterLogger(AbstractWorker.class); - private MemberSystem memberSystem = new MemberSystem(); + private Cluster cluster = Cluster.get(getContext().system()); @Override public void preStart() throws Exception { - super.preStart(); + cluster.subscribe(getSelf(), ClusterEvent.MemberUp.class); register(); } @Override - public void preRestart(Throwable reason, Option message) throws Exception { - super.preRestart(reason, message); - register(); + public void postStop() throws Exception { + cluster.unsubscribe(getSelf()); } /** @@ -81,11 +79,11 @@ public abstract class AbstractWorker extends UntypedActor { } } } else if (message instanceof ClusterEvent.MemberUp) { - logger.info("receive ClusterEvent.MemberUp message"); ClusterEvent.MemberUp memberUp = (ClusterEvent.MemberUp) message; + logger.info("receive ClusterEvent.MemberUp message, address: %s", memberUp.member().address().toString()); register(memberUp.member()); - } else { - logger.info("message class: %s", message.getClass().getName()); + } else { + logger.debug("worker class: %s, message class: %s", this.getClass().getName(), message.getClass().getName()); receive(message); } } @@ -110,9 +108,9 @@ public abstract class AbstractWorker extends UntypedActor { * @param member is the new created or restart worker */ void register(Member member) { - System.out.println("register"); - if (member.getRoles().equals(WorkersListener.WorkName)) { + if (member.hasRole(WorkersListener.WorkName)) { WorkerListenerMessage.RegisterMessage registerMessage = new WorkerListenerMessage.RegisterMessage(getClass().getSimpleName()); + logger.info("member address: %s, worker path: %s", member.address().toString(), getSelf().path().toString()); getContext().actorSelection(member.address() + "/user/" + WorkersListener.WorkName).tell(registerMessage, getSelf()); } } @@ -121,8 +119,4 @@ public abstract class AbstractWorker extends UntypedActor { WorkerListenerMessage.RegisterMessage registerMessage = new WorkerListenerMessage.RegisterMessage(getClass().getSimpleName()); getContext().actorSelection("/user/" + WorkersListener.WorkName).tell(registerMessage, getSelf()); } - - public MemberSystem memberContext() { - return memberSystem; - } } diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorkerProvider.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorkerProvider.java index d3927754f..e4bd13727 100644 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorkerProvider.java +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/AbstractWorkerProvider.java @@ -27,7 +27,7 @@ import org.apache.logging.log4j.Logger; * }}} *

*/ -public abstract class AbstractWorkerProvider { +public abstract class AbstractWorkerProvider { private Logger logger = LogManager.getFormatterLogger(AbstractWorkerProvider.class); diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/MemberSystem.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/MemberSystem.java deleted file mode 100644 index 3c407bbaa..000000000 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/MemberSystem.java +++ /dev/null @@ -1,20 +0,0 @@ -package com.a.eye.skywalking.collector.actor; - -import java.util.HashMap; -import java.util.Map; - -/** - * @author pengys5 - */ -public class MemberSystem { - - private Map memberMap = new HashMap(); - - public void memberOf(AbstractMember member, String role) { - memberMap.put(role, member); - } - - public AbstractMember memberFor(String role) { - return memberMap.get(role); - } -} diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkerRef.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkerRef.java index d53deafca..aaf424669 100644 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkerRef.java +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/WorkerRef.java @@ -1,6 +1,8 @@ package com.a.eye.skywalking.collector.actor; import akka.actor.ActorRef; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; /** * The Worker reference @@ -8,6 +10,8 @@ import akka.actor.ActorRef; * @author pengys5 */ public class WorkerRef { + private Logger logger = LogManager.getFormatterLogger(WorkerRef.class); + final ActorRef actorRef; final String workerRole; @@ -17,8 +21,9 @@ public class WorkerRef { this.workerRole = workerRole; } - void tell(Object message, ActorRef actorRef) { - actorRef.tell(message, actorRef); + void tell(Object message, ActorRef sender) { + logger.debug("tell %s worker", actorRef.toString()); + actorRef.tell(message, sender); } public String getWorkerRole() { diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/selector/LocalSelector.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/selector/LocalSelector.java deleted file mode 100644 index 03f57ee62..000000000 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/actor/selector/LocalSelector.java +++ /dev/null @@ -1,36 +0,0 @@ -package com.a.eye.skywalking.collector.actor.selector; - -import com.a.eye.skywalking.collector.actor.AbstractWorker; -import com.a.eye.skywalking.collector.actor.WorkerRef; - -import java.util.List; - -/** - * The LocalSelector is a simple implementation of {@link WorkerSelector}. - * It choose {@link WorkerRef} nearly random, by round-robin. - * - * @author wusheng - */ -public enum LocalSelector implements WorkerSelector { - INSTANCE; - - /** - * A simple round variable. - */ - private int index = 0; - - /** - * Use round-robin to select {@link WorkerRef}. - * - * @param members given {@link WorkerRef} list, which size is greater than 0; - * @param message the {@link AbstractWorker} is going to send. - * @return the selected {@link WorkerRef} - */ - @Override - public WorkerRef select(List members, Object message) { - int size = members.size(); - index++; - int selectIndex = Math.abs(index) % size; - return members.get(selectIndex); - } -} diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/WorkersListener.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/WorkersListener.java index b7e0fbb18..c1ecbcf8e 100644 --- a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/WorkersListener.java +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/cluster/WorkersListener.java @@ -31,7 +31,7 @@ public class WorkersListener extends UntypedActor { ActorRef sender = getSender(); getContext().watch(sender); - logger.info("register worker of role %s", register.getWorkRole()); + logger.info("register worker of role: %s, path: %s", register.getWorkRole(), sender.toString()); WorkersRefCenter.INSTANCE.register(sender, register.getWorkRole()); } else if (message instanceof Terminated) { diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/queue/DaemonThreadFactory.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/queue/DaemonThreadFactory.java new file mode 100644 index 000000000..ff51aeb7d --- /dev/null +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/queue/DaemonThreadFactory.java @@ -0,0 +1,17 @@ +package com.a.eye.skywalking.collector.queue; + +import java.util.concurrent.ThreadFactory; + +/** + * @author pengys5 + */ +public enum DaemonThreadFactory implements ThreadFactory { + INSTANCE; + + @Override + public Thread newThread(Runnable r) { + Thread t = new Thread(r); + t.setDaemon(true); + return t; + } +} diff --git a/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/queue/MessageHolder.java b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/queue/MessageHolder.java new file mode 100644 index 000000000..779c2346b --- /dev/null +++ b/skywalking-collector/skywalking-collector-cluster/src/main/java/com/a/eye/skywalking/collector/queue/MessageHolder.java @@ -0,0 +1,20 @@ +package com.a.eye.skywalking.collector.queue; + +/** + * @author pengys5 + */ +public class MessageHolder { + private T message; + + public T getMessage() { + return message; + } + + public void setMessage(T message) { + this.message = message; + } + + public void reset() { + message = null; + } +} diff --git a/skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/queue/QueueTestCase.java b/skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/queue/QueueTestCase.java new file mode 100644 index 000000000..7549ef20e --- /dev/null +++ b/skywalking-collector/skywalking-collector-cluster/src/test/java/com/a/eye/skywalking/collector/queue/QueueTestCase.java @@ -0,0 +1,13 @@ +package com.a.eye.skywalking.collector.queue; + +import org.junit.Test; + +/** + * @author pengys5 + */ +public class QueueTestCase { + + @Test + public void testProducer() throws InterruptedException { + } +} diff --git a/skywalking-collector/skywalking-collector-cluster/src/test/resources/application.conf b/skywalking-collector/skywalking-collector-cluster/src/test/resources/application.conf new file mode 100644 index 000000000..79fd12aba --- /dev/null +++ b/skywalking-collector/skywalking-collector-cluster/src/test/resources/application.conf @@ -0,0 +1,19 @@ +akka { + actor { + provider = "akka.cluster.ClusterActorRefProvider" + + serializers { + java = "akka.serialization.JavaSerializer" + proto = "akka.remote.serialization.ProtobufSerializer" + } + + serialization-bindings { + "java.lang.String" = java + "com.google.protobuf.Message" = proto +// "java.io.Serializable" = none + } + +// serialize-messages = on + warn-about-java-serializer-usage = on + } +} \ No newline at end of file diff --git a/skywalking-collector/skywalking-collector-cluster/src/test/resources/log4j2.xml b/skywalking-collector/skywalking-collector-cluster/src/test/resources/log4j2.xml new file mode 100644 index 000000000..fd313a5d2 --- /dev/null +++ b/skywalking-collector/skywalking-collector-cluster/src/test/resources/log4j2.xml @@ -0,0 +1,13 @@ + + + + + + + + + + + + + diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/JsonSerializer.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/JsonSerializer.java new file mode 100644 index 000000000..985003fce --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/JsonSerializer.java @@ -0,0 +1,37 @@ +package com.a.eye.skywalking.collector.worker; + +import akka.serialization.JSerializer; +import com.a.eye.skywalking.trace.TraceSegment; +import com.a.eye.skywalking.trace.proto.SegmentMessage; +import com.google.gson.Gson; +import com.google.gson.JsonObject; +import com.google.protobuf.InvalidProtocolBufferException; + +/** + * @author pengys5 + */ +public class JsonSerializer extends JSerializer { + @Override + public boolean includeManifest() { + return false; + } + + @Override + public int identifier() { + return 123; + } + + @Override + public byte[] toBinary(Object o) { +// System.out.println("Json toBinary"); + JsonObject jsonObject = (JsonObject) o; + return jsonObject.toString().getBytes(); + } + + @Override + public Object fromBinaryJava(byte[] bytes, Class manifest) { +// System.out.println("Json fromBinaryJava"); + Gson gson = new Gson(); + return gson.fromJson(new String(bytes), JsonObject.class); + } +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceMember.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceMember.java new file mode 100644 index 000000000..bad069c99 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceMember.java @@ -0,0 +1,86 @@ +package com.a.eye.skywalking.collector.worker; + +import akka.actor.ActorRef; +import com.a.eye.skywalking.collector.actor.AbstractMember; +import com.a.eye.skywalking.collector.worker.storage.EsClient; +import com.google.gson.JsonObject; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.elasticsearch.action.bulk.BulkRequestBuilder; +import org.elasticsearch.action.bulk.BulkResponse; +import org.elasticsearch.client.transport.TransportClient; + +import java.util.HashMap; +import java.util.Map; + +/** + * @author pengys5 + */ +public abstract class PersistenceMember extends AbstractMember { + + private Logger logger = LogManager.getFormatterLogger(PersistenceMember.class); + + private long lastPersistenceTimestamp = 0; + + private Map persistenceData = new HashMap(); + + public PersistenceMember(ActorRef actorRef) { + super(actorRef); + } + + public abstract String esIndex(); + + public abstract String esType(); + + public void putData(String id, JsonObject data) { + persistenceData.put(id, data); +// if (persistenceData.size() >= 1000) { +// persistence(true); +// } + } + + public boolean containsId(String id) { + return persistenceData.containsKey(id); + } + + public JsonObject getData(String id) { + return persistenceData.get(id); + } + + public abstract void analyse(Object message) throws Exception; + + @Override + public void receive(Object message) throws Exception { + if (message instanceof PersistenceCommand) { + persistence(false); + } else { + logger.debug("receive message"); + analyse(message); + } + } + + private void persistence(boolean dataFull) { + long now = System.currentTimeMillis(); + if (now - lastPersistenceTimestamp > 5000 || dataFull) { + boolean success = saveToEs(); + if (success) { + persistenceData.clear(); + lastPersistenceTimestamp = now; + } + } + } + + private boolean saveToEs() { + TransportClient client = EsClient.client(); + BulkRequestBuilder bulkRequest = client.prepareBulk(); + + for (Map.Entry entry : persistenceData.entrySet()) { + String id = entry.getKey(); + JsonObject data = entry.getValue(); + bulkRequest.add(client.prepareIndex(esIndex(), esType(), id).setSource(data.toString())); + } + + BulkResponse bulkResponse = bulkRequest.execute().actionGet(); + return !bulkResponse.hasFailures(); + } +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceWorker.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceWorker.java index e3049af66..78f0ee48c 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceWorker.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/PersistenceWorker.java @@ -1,11 +1,13 @@ package com.a.eye.skywalking.collector.worker; import com.a.eye.skywalking.collector.actor.AbstractWorker; -import com.a.eye.skywalking.collector.worker.PersistenceCommand; -import com.a.eye.skywalking.collector.worker.tools.EsClient; +import com.a.eye.skywalking.collector.worker.storage.EsClient; import com.google.gson.JsonObject; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; +import org.elasticsearch.client.transport.TransportClient; import java.util.HashMap; import java.util.Map; @@ -15,6 +17,8 @@ import java.util.Map; */ public abstract class PersistenceWorker extends AbstractWorker { + private Logger logger = LogManager.getFormatterLogger(PersistenceWorker.class); + private long lastPersistenceTimestamp = 0; private Map persistenceData = new HashMap(); @@ -25,9 +29,9 @@ public abstract class PersistenceWorker extends AbstractWorker { public void putData(String id, JsonObject data) { persistenceData.put(id, data); - if (persistenceData.size() >= 1000) { - persistence(true); - } +// if (persistenceData.size() >= 1000) { +// persistence(true); +// } } public boolean containsId(String id) { @@ -45,6 +49,7 @@ public abstract class PersistenceWorker extends AbstractWorker { if (message instanceof PersistenceCommand) { persistence(false); } else { + logger.debug("receive message"); analyse(message); } } @@ -61,12 +66,13 @@ public abstract class PersistenceWorker extends AbstractWorker { } private boolean saveToEs() { - BulkRequestBuilder bulkRequest = EsClient.client().prepareBulk(); + TransportClient client = EsClient.client(); + BulkRequestBuilder bulkRequest = client.prepareBulk(); for (Map.Entry entry : persistenceData.entrySet()) { String id = entry.getKey(); JsonObject data = entry.getValue(); - bulkRequest.add(EsClient.client().prepareIndex(esIndex(), esType(), id).setSource(data.toString())); + bulkRequest.add(client.prepareIndex(esIndex(), esType(), id).setSource(data.toString())); } BulkResponse bulkResponse = bulkRequest.execute().actionGet(); diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/TraceSegmentSerializer.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/TraceSegmentSerializer.java new file mode 100644 index 000000000..f3aab1eb1 --- /dev/null +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/TraceSegmentSerializer.java @@ -0,0 +1,40 @@ +package com.a.eye.skywalking.collector.worker; + +import akka.serialization.JSerializer; +import com.a.eye.skywalking.trace.TraceSegment; +import com.a.eye.skywalking.trace.proto.SegmentMessage; +import com.google.protobuf.InvalidProtocolBufferException; + +/** + * @author pengys5 + */ +public class TraceSegmentSerializer extends JSerializer { + @Override + public boolean includeManifest() { + return false; + } + + @Override + public int identifier() { + return 0; + } + + @Override + public byte[] toBinary(Object o) { +// System.out.println("toBinary"); + TraceSegment traceSegment = (TraceSegment) o; + return traceSegment.serialize().toByteArray(); + } + + @Override + public Object fromBinaryJava(byte[] bytes, Class manifest) { +// System.out.println("fromBinaryJava"); + TraceSegment traceSegment = null; + try { + traceSegment = new TraceSegment(SegmentMessage.parseFrom(bytes)); + } catch (InvalidProtocolBufferException e) { + e.printStackTrace(); + } + return traceSegment; + } +} diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/WorkerConfig.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/WorkerConfig.java index 41453510f..82f6f634a 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/WorkerConfig.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/WorkerConfig.java @@ -7,15 +7,34 @@ import com.a.eye.skywalking.collector.cluster.ClusterConfig; */ public class WorkerConfig extends ClusterConfig { - public static class WorkerNum { - public static int TraceSegmentReceiver_Num = 1; + public static class Worker { + public static class TraceSegmentReceiver { + public static int Num = 5; + } - public static int DAGNodePersistence_Num = 5; - public static int DAGNodeRefPersistence_Num = 5; - public static int NodeInstancePersistence_Num = 5; - public static int ResponseCostPersistence_Num = 5; - public static int ResponseSummaryPersistence_Num = 5; - public static int TraceSegmentRecordPersistence_Num = 5; + public static class DAGNodePersistence { + public static int Num = 5; + } + + public static class NodeInstancePersistence { + public static int Num = 5; + } + + public static class ResponseCostPersistence { + public static int Num = 5; + } + + public static class ResponseSummaryPersistence { + public static int Num = 5; + } + + public static class TraceSegmentRecordPersistence { + public static int Num = 5; + } + + public static class DAGNodeRefPersistence { + public static int Num = 5; + } } } diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMember.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMember.java index 9eaac07d2..7f32b11a6 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMember.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/ApplicationMember.java @@ -2,14 +2,14 @@ package com.a.eye.skywalking.collector.worker.application; import akka.actor.ActorRef; import com.a.eye.skywalking.collector.actor.AbstractMember; -import com.a.eye.skywalking.collector.actor.AbstractMemberProvider; -import com.a.eye.skywalking.collector.actor.MemberSystem; +import com.a.eye.skywalking.collector.actor.AbstractSyncMemberProvider; import com.a.eye.skywalking.collector.actor.selector.RollingSelector; import com.a.eye.skywalking.collector.worker.application.metric.TraceSegmentRecordMember; import com.a.eye.skywalking.collector.worker.application.persistence.DAGNodePersistence; import com.a.eye.skywalking.collector.worker.application.persistence.NodeInstancePersistence; import com.a.eye.skywalking.collector.worker.application.persistence.ResponseCostPersistence; import com.a.eye.skywalking.collector.worker.application.persistence.ResponseSummaryPersistence; +import com.a.eye.skywalking.collector.worker.applicationref.presistence.DAGNodeRefPersistence; import com.a.eye.skywalking.trace.Span; import com.a.eye.skywalking.trace.TraceSegment; import com.a.eye.skywalking.trace.tag.Tags; @@ -23,24 +23,18 @@ public class ApplicationMember extends AbstractMember { private Logger logger = LogManager.getFormatterLogger(ApplicationMember.class); - public ApplicationMember(MemberSystem memberSystem, ActorRef actorRef) { - super(memberSystem, actorRef); + private TraceSegmentRecordMember traceSegmentRecordMember = TraceSegmentRecordMember.Factory.INSTANCE.createWorker(TraceSegmentRecordMember.MessageFactory.INSTANCE, getSelf()); + + public ApplicationMember(ActorRef actorRef) throws Exception { + super(actorRef); } @Override - public void preStart() throws Exception { - logger.info("create members"); - TraceSegmentRecordMember.Factory factory = new TraceSegmentRecordMember.Factory(); - factory.createWorker(memberContext(), getSelf()); - } - - @Override - public void receive(Object message) throws Throwable { + public void receive(Object message) throws Exception { if (message instanceof TraceSegment) { logger.debug("begin translate TraceSegment Object to JsonObject"); TraceSegment traceSegment = (TraceSegment) message; - AbstractMember discoverMember = memberContext().memberFor(TraceSegmentRecordMember.class.getSimpleName()); - discoverMember.receive(traceSegment); + traceSegmentRecordMember.receive(traceSegment); sendToDAGNodePersistence(traceSegment); sendToNodeInstancePersistence(traceSegment); @@ -49,14 +43,16 @@ public class ApplicationMember extends AbstractMember { } } - public static class Factory extends AbstractMemberProvider { + public static class Factory extends AbstractSyncMemberProvider { + public static Factory INSTANCE = new Factory(); + @Override public Class memberClass() { return ApplicationMember.class; } } - private void sendToDAGNodePersistence(TraceSegment traceSegment) throws Throwable { + private void sendToDAGNodePersistence(TraceSegment traceSegment) throws Exception { String code = traceSegment.getApplicationCode(); String component = null; @@ -69,10 +65,10 @@ public class ApplicationMember extends AbstractMember { } DAGNodePersistence.Metric node = new DAGNodePersistence.Metric(code, component, layer); - tell(new NodeInstancePersistence.Factory(), RollingSelector.INSTANCE, node); + tell(DAGNodeRefPersistence.Factory.INSTANCE, RollingSelector.INSTANCE, node); } - private void sendToNodeInstancePersistence(TraceSegment traceSegment) throws Throwable { + private void sendToNodeInstancePersistence(TraceSegment traceSegment) throws Exception { if (traceSegment.getPrimaryRef() != null) { String code = traceSegment.getPrimaryRef().getApplicationCode(); String address = traceSegment.getPrimaryRef().getPeerHost(); @@ -82,8 +78,10 @@ public class ApplicationMember extends AbstractMember { } } - private void sendToResponseCostPersistence(TraceSegment traceSegment) throws Throwable { + private void sendToResponseCostPersistence(TraceSegment traceSegment) throws Exception { String code = traceSegment.getApplicationCode(); + code = "test"; + long startTime = -1; long endTime = -1; Boolean isError = false; @@ -100,7 +98,7 @@ public class ApplicationMember extends AbstractMember { tell(new ResponseCostPersistence.Factory(), RollingSelector.INSTANCE, cost); } - private void sendToResponseSummaryPersistence(TraceSegment traceSegment) throws Throwable { + private void sendToResponseSummaryPersistence(TraceSegment traceSegment) throws Exception { String code = traceSegment.getApplicationCode(); boolean isError = false; diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/metric/TraceSegmentRecordMember.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/metric/TraceSegmentRecordMember.java index 17f0367a8..a8e5bbe87 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/metric/TraceSegmentRecordMember.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/metric/TraceSegmentRecordMember.java @@ -2,16 +2,18 @@ package com.a.eye.skywalking.collector.worker.application.metric; import akka.actor.ActorRef; +import com.a.eye.skywalking.collector.actor.AbstractASyncMemberProvider; import com.a.eye.skywalking.collector.actor.AbstractMember; -import com.a.eye.skywalking.collector.actor.AbstractMemberProvider; -import com.a.eye.skywalking.collector.actor.MemberSystem; -import com.a.eye.skywalking.collector.actor.selector.LocalSelector; +import com.a.eye.skywalking.collector.actor.selector.RollingSelector; +import com.a.eye.skywalking.collector.queue.MessageHolder; +import com.a.eye.skywalking.collector.worker.PersistenceMember; import com.a.eye.skywalking.collector.worker.application.persistence.TraceSegmentRecordPersistence; import com.a.eye.skywalking.trace.Span; import com.a.eye.skywalking.trace.TraceSegment; import com.a.eye.skywalking.trace.TraceSegmentRef; import com.google.gson.JsonArray; import com.google.gson.JsonObject; +import com.lmax.disruptor.EventFactory; import java.util.List; import java.util.Map; @@ -19,27 +21,43 @@ import java.util.Map; /** * @author pengys5 */ -public class TraceSegmentRecordMember extends AbstractMember { +public class TraceSegmentRecordMember extends PersistenceMember { - public TraceSegmentRecordMember(MemberSystem memberSystem, ActorRef actorRef) throws Throwable { - super(memberSystem, actorRef); + @Override + public String esIndex() { + return "application_record"; } @Override - public void preStart() throws Exception { + public String esType() { + return "trace_segment"; + } + + public TraceSegmentRecordMember(ActorRef actorRef) throws Throwable { + super(actorRef); } @Override - public void receive(Object message) throws Throwable { + public void analyse(Object message) throws Exception { if (message instanceof TraceSegment) { TraceSegment traceSegment = (TraceSegment) message; JsonObject traceSegmentJsonObj = parseTraceSegment(traceSegment); - tell(new TraceSegmentRecordPersistence.Factory(), LocalSelector.INSTANCE, traceSegmentJsonObj); + tell(TraceSegmentRecordPersistence.Factory.INSTANCE, RollingSelector.INSTANCE, traceSegmentJsonObj); } } - public static class Factory extends AbstractMemberProvider { + public static class MessageFactory implements EventFactory { + public static MessageFactory INSTANCE = new MessageFactory(); + + public MessageHolder newInstance() { + return new MessageHolder(); + } + } + + public static class Factory extends AbstractASyncMemberProvider { + public static Factory INSTANCE = new Factory(); + @Override public Class memberClass() { return TraceSegmentRecordMember.class; diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/DAGNodePersistence.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/DAGNodePersistence.java index cfcb59f82..1e4004c67 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/DAGNodePersistence.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/application/persistence/DAGNodePersistence.java @@ -7,6 +7,8 @@ import com.google.gson.JsonObject; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import java.io.Serializable; + /** * @author pengys5 */ @@ -33,12 +35,16 @@ public class DAGNodePersistence extends PersistenceWorker { + private Logger logger = LogManager.getFormatterLogger(NodeInstancePersistence.class); + @Override public String esIndex() { return "application"; @@ -29,10 +33,15 @@ public class NodeInstancePersistence extends PersistenceWorker { + private Logger logger = LogManager.getFormatterLogger(ResponseCostPersistence.class); + @Override public String esIndex() { return "application_metric"; @@ -51,10 +57,18 @@ public class ResponseCostPersistence extends PersistenceWorker { + private Logger logger = LogManager.getFormatterLogger(ResponseSummaryPersistence.class); + @Override public String esIndex() { return "application_metric"; @@ -33,7 +38,7 @@ public class ResponseSummaryPersistence extends PersistenceWorker { + public static Factory INSTANCE = new Factory(); + @Override public Class memberClass() { return ApplicationRefMember.class; diff --git a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/presistence/DAGNodeRefPersistence.java b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/presistence/DAGNodeRefPersistence.java index 36f135a76..afff42b30 100644 --- a/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/presistence/DAGNodeRefPersistence.java +++ b/skywalking-collector/skywalking-collector-worker/src/main/java/com/a/eye/skywalking/collector/worker/applicationref/presistence/DAGNodeRefPersistence.java @@ -5,6 +5,8 @@ import com.a.eye.skywalking.collector.worker.PersistenceWorker; import com.a.eye.skywalking.collector.worker.WorkerConfig; import com.google.gson.JsonObject; +import java.io.Serializable; + /** * @author pengys5 */ @@ -33,6 +35,9 @@ public class DAGNodeRefPersistence extends PersistenceWorker 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); + Thread.sleep(1000000); } }