Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add agent instance register and heartbeat report
- Loading branch information
Showing
60 changed files
with
1,568 additions
and
247 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
38 changes: 38 additions & 0 deletions
38
...main/java/org/skywalking/apm/collector/worker/httpserver/AbstractPostWithHttpServlet.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| package org.skywalking.apm.collector.worker.httpserver; | ||
|
|
||
| import com.google.gson.JsonObject; | ||
| import java.io.BufferedReader; | ||
| import java.io.IOException; | ||
| import javax.servlet.ServletException; | ||
| import javax.servlet.http.HttpServletRequest; | ||
| import javax.servlet.http.HttpServletResponse; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.skywalking.apm.collector.actor.WorkerRef; | ||
|
|
||
| public abstract class AbstractPostWithHttpServlet extends AbstractHttpServlet { | ||
|
|
||
| private Logger logger = LogManager.getFormatterLogger(AbstractPostWithHttpServlet.class); | ||
| protected final WorkerRef ownerWorkerRef; | ||
|
|
||
| AbstractPostWithHttpServlet(WorkerRef ownerWorkerRef) { | ||
| this.ownerWorkerRef = ownerWorkerRef; | ||
| } | ||
|
|
||
| @Override final protected void doPost(HttpServletRequest request, | ||
| HttpServletResponse response) throws ServletException, IOException { | ||
| JsonObject resJson = new JsonObject(); | ||
| try { | ||
| BufferedReader bufferedReader = request.getReader(); | ||
| doWork(bufferedReader, resJson); | ||
| reply(response, resJson, HttpServletResponse.SC_OK); | ||
| } catch (Exception e) { | ||
| logger.error(e); | ||
| resJson.addProperty("error", e.getMessage()); | ||
| reply(response, resJson, HttpServletResponse.SC_INTERNAL_SERVER_ERROR); | ||
| } | ||
| } | ||
|
|
||
| protected abstract void doWork(BufferedReader bufferedReader, JsonObject resJson) throws Exception; | ||
| } | ||
|
|
70 changes: 70 additions & 0 deletions
70
...orker/src/main/java/org/skywalking/apm/collector/worker/instance/HeartBeatReportPost.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| package org.skywalking.apm.collector.worker.instance; | ||
|
|
||
| import com.google.gson.JsonObject; | ||
| import org.skywalking.apm.collector.actor.ClusterWorkerContext; | ||
| import org.skywalking.apm.collector.actor.LocalWorkerContext; | ||
| import org.skywalking.apm.collector.actor.ProviderNotFoundException; | ||
| import org.skywalking.apm.collector.actor.Role; | ||
| import org.skywalking.apm.collector.actor.WorkerRef; | ||
| import org.skywalking.apm.collector.actor.selector.RollingSelector; | ||
| import org.skywalking.apm.collector.actor.selector.WorkerSelector; | ||
| import org.skywalking.apm.collector.worker.httpserver.AbstractPost; | ||
| import org.skywalking.apm.collector.worker.httpserver.AbstractPostProvider; | ||
| import org.skywalking.apm.collector.worker.httpserver.AbstractPostWithHttpServlet; | ||
| import org.skywalking.apm.collector.worker.instance.entity.HeartBeat; | ||
| import org.skywalking.apm.collector.worker.instance.heartbeat.HeartBeatAnalysis; | ||
|
|
||
| public class HeartBeatReportPost extends AbstractPost { | ||
|
|
||
| public HeartBeatReportPost(Role role, ClusterWorkerContext clusterContext, LocalWorkerContext selfContext) { | ||
| super(role, clusterContext, selfContext); | ||
| } | ||
|
|
||
| @Override | ||
| public void preStart() throws ProviderNotFoundException { | ||
| getClusterContext().findProvider(HeartBeatAnalysis.Role.INSTANCE).create(this); | ||
| } | ||
|
|
||
| @Override | ||
| protected void onReceive(Object message, JsonObject response) throws Exception { | ||
| if (message instanceof HeartBeat) { | ||
| getSelfContext().lookup(HeartBeatAnalysis.Role.INSTANCE).tell(message); | ||
| } | ||
| } | ||
|
|
||
| public static class Factory extends AbstractPostProvider<HeartBeatReportPost> { | ||
| @Override | ||
| public Role role() { | ||
| return HeartBeatReportPost.WorkerRole.INSTANCE; | ||
| } | ||
|
|
||
| @Override | ||
| public HeartBeatReportPost workerInstance(ClusterWorkerContext clusterContext) { | ||
| return new HeartBeatReportPost(role(), clusterContext, new LocalWorkerContext()); | ||
| } | ||
|
|
||
| @Override | ||
| public String servletPath() { | ||
| return "/heartbeat"; | ||
| } | ||
|
|
||
| @Override | ||
| public AbstractPostWithHttpServlet handleServlet(WorkerRef workerRef) { | ||
| return new HeartBeatPostWithHttpServlet(workerRef); | ||
| } | ||
| } | ||
|
|
||
| public enum WorkerRole implements Role { | ||
| INSTANCE; | ||
|
|
||
| @Override | ||
| public String roleName() { | ||
| return HeartBeatReportPost.class.getSimpleName(); | ||
| } | ||
|
|
||
| @Override | ||
| public WorkerSelector workerSelector() { | ||
| return new RollingSelector(); | ||
| } | ||
| } | ||
| } |
57 changes: 57 additions & 0 deletions
57
...orker/src/main/java/org/skywalking/apm/collector/worker/instance/IdentificationCache.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| package org.skywalking.apm.collector.worker.instance; | ||
|
|
||
| class IdentificationCache { | ||
| private static final IdentificationCache INSTANCE = new IdentificationCache(); | ||
|
|
||
| State state; | ||
| IdentificationSegment segment; | ||
|
|
||
| private IdentificationCache() { | ||
| state = State.NORMAL; | ||
| } | ||
|
|
||
| public synchronized long fetchInstanceId() { | ||
| if (state == State.ABNORMAL) { | ||
| return -1; | ||
| } | ||
|
|
||
| if (!segment.hasNext()) { | ||
| IdentificationSegmentFetcher.INSTANCE.fetchSegment(new IdentificationSegmentFetcher.Listener() { | ||
| @Override | ||
| public void failed() { | ||
| state = State.ABNORMAL; | ||
| IdentificationSegmentFetcher.INSTANCE.fetchSegmentInBackGround(INSTANCE); | ||
| } | ||
|
|
||
| @Override | ||
| public void success(IdentificationSegment idSegment) { | ||
| segment = idSegment; | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| return segment.nextInstanceId(); | ||
| } | ||
|
|
||
| public static IdentificationCache initCache() { | ||
| IdentificationSegmentFetcher.INSTANCE.fetchSegment(new IdentificationSegmentFetcher.Listener() { | ||
| @Override | ||
| public void failed() { | ||
| INSTANCE.state = State.ABNORMAL; | ||
| IdentificationSegmentFetcher.INSTANCE.fetchSegmentInBackGround(INSTANCE); | ||
| } | ||
|
|
||
| @Override | ||
| public void success(IdentificationSegment idSegment) { | ||
| INSTANCE.segment = idSegment; | ||
| } | ||
| }); | ||
|
|
||
| return INSTANCE; | ||
| } | ||
|
|
||
| enum State { | ||
| NORMAL, | ||
| ABNORMAL | ||
| } | ||
| } |
19 changes: 19 additions & 0 deletions
19
...ker/src/main/java/org/skywalking/apm/collector/worker/instance/IdentificationSegment.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| package org.skywalking.apm.collector.worker.instance; | ||
|
|
||
| public class IdentificationSegment { | ||
| private long startInstanceId; | ||
| private long endInstanceId; | ||
|
|
||
| IdentificationSegment(long start, long end) { | ||
| this.startInstanceId = start; | ||
| this.endInstanceId = end; | ||
| } | ||
|
|
||
| public long nextInstanceId() { | ||
| return startInstanceId++; | ||
| } | ||
|
|
||
| public boolean hasNext() { | ||
| return startInstanceId + 1 >= endInstanceId; | ||
| } | ||
| } |
Oops, something went wrong.