You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2019/07/19 06:46:40 UTC
[skywalking] branch remove-remove-workid created (now 009e5bd)
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a change to branch remove-remove-workid
in repository https://gitbox.apache.org/repos/asf/skywalking.git.
at 009e5bd Remove the worker id, and add worker name for remote handler only.
This branch includes the following new commits:
new 009e5bd Remove the worker id, and add worker name for remote handler only.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking] 01/01: Remove the worker id,
and add worker name for remote handler only.
Posted by wu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch remove-remove-workid
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 009e5bdcd5206f9be66a0e6aa741b1bb1102f3e1
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Fri Jul 19 14:46:28 2019 +0800
Remove the worker id, and add worker name for remote handler only.
---
.../core/analysis/worker/MetricsRemoteWorker.java | 11 ++++------
.../analysis/worker/MetricsStreamProcessor.java | 12 +++++++++--
.../register/worker/InventoryStreamProcessor.java | 7 +++++-
.../core/register/worker/RegisterRemoteWorker.java | 8 +++----
.../server/core/remote/RemoteSenderService.java | 8 +++----
.../server/core/remote/RemoteServiceHandler.java | 15 +++++++++++--
.../core/remote/client/GRPCRemoteClient.java | 6 +++---
.../server/core/remote/client/RemoteClient.java | 2 +-
.../core/remote/client/SelfRemoteClient.java | 4 ++--
.../oap/server/core/worker/AbstractWorker.java | 5 -----
.../server/core/worker/IWorkerInstanceGetter.java | 2 +-
.../server/core/worker/IWorkerInstanceSetter.java | 3 +--
.../server/core/worker/WorkerInstancesService.java | 25 +++++++++++++---------
.../server-core/src/main/proto/RemoteService.proto | 2 +-
.../core/remote/RemoteServiceHandlerTestCase.java | 4 ++--
.../remote/client/GRPCRemoteClientRealClient.java | 2 +-
.../remote/client/GRPCRemoteClientTestCase.java | 2 +-
17 files changed, 69 insertions(+), 49 deletions(-)
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsRemoteWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsRemoteWorker.java
index 7ce9311..392d9b2 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsRemoteWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsRemoteWorker.java
@@ -33,21 +33,18 @@ public class MetricsRemoteWorker extends AbstractWorker<Metrics> {
private static final Logger logger = LoggerFactory.getLogger(MetricsRemoteWorker.class);
- private final AbstractWorker<Metrics> nextWorker;
private final RemoteSenderService remoteSender;
- private final String modelName;
+ private final String remoteReceiverWorkerName;
- MetricsRemoteWorker(ModuleDefineHolder moduleDefineHolder, AbstractWorker<Metrics> nextWorker,
- String modelName) {
+ MetricsRemoteWorker(ModuleDefineHolder moduleDefineHolder, String remoteReceiverWorkerName) {
super(moduleDefineHolder);
this.remoteSender = moduleDefineHolder.find(CoreModule.NAME).provider().getService(RemoteSenderService.class);
- this.nextWorker = nextWorker;
- this.modelName = modelName;
+ this.remoteReceiverWorkerName = remoteReceiverWorkerName;
}
@Override public final void in(Metrics metrics) {
try {
- remoteSender.send(nextWorker.getWorkerId(), metrics, Selector.HashCode);
+ remoteSender.send(remoteReceiverWorkerName, metrics, Selector.HashCode);
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java
index 20d09bf..33488d3 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java
@@ -28,6 +28,7 @@ import org.apache.skywalking.oap.server.core.remote.define.StreamDataMappingSett
import org.apache.skywalking.oap.server.core.storage.*;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
import org.apache.skywalking.oap.server.core.storage.model.*;
+import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
/**
@@ -90,14 +91,21 @@ public class MetricsStreamProcessor implements StreamProcessor<Metrics> {
Model model = modelSetter.putIfAbsent(metricsClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Minute));
MetricsPersistentWorker minutePersistentWorker = minutePersistentWorker(moduleDefineHolder, metricsDAO, model);
+
MetricsTransWorker transWorker = new MetricsTransWorker(moduleDefineHolder, stream.name(), minutePersistentWorker, hourPersistentWorker, dayPersistentWorker, monthPersistentWorker);
- MetricsRemoteWorker remoteWorker = new MetricsRemoteWorker(moduleDefineHolder, transWorker, stream.name());
+
+ String remoteReceiverWorkerName = stream.name() + "_rec";
+ IWorkerInstanceSetter workerInstanceSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(IWorkerInstanceSetter.class);
+ workerInstanceSetter.put(remoteReceiverWorkerName, transWorker);
+
+ MetricsRemoteWorker remoteWorker = new MetricsRemoteWorker(moduleDefineHolder, remoteReceiverWorkerName);
MetricsAggregateWorker aggregateWorker = new MetricsAggregateWorker(moduleDefineHolder, remoteWorker, stream.name());
entryWorkers.put(metricsClass, aggregateWorker);
}
- private MetricsPersistentWorker minutePersistentWorker(ModuleDefineHolder moduleDefineHolder, IMetricsDAO metricsDAO, Model model) {
+ private MetricsPersistentWorker minutePersistentWorker(ModuleDefineHolder moduleDefineHolder,
+ IMetricsDAO metricsDAO, Model model) {
AlarmNotifyWorker alarmNotifyWorker = new AlarmNotifyWorker(moduleDefineHolder);
ExportWorker exportWorker = new ExportWorker(moduleDefineHolder);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryStreamProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryStreamProcessor.java
index 596ab81..042a9bc 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryStreamProcessor.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/InventoryStreamProcessor.java
@@ -26,6 +26,7 @@ import org.apache.skywalking.oap.server.core.remote.define.StreamDataMappingSett
import org.apache.skywalking.oap.server.core.storage.*;
import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
import org.apache.skywalking.oap.server.core.storage.model.*;
+import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
/**
@@ -63,7 +64,11 @@ public class InventoryStreamProcessor implements StreamProcessor<RegisterSource>
RegisterPersistentWorker persistentWorker = new RegisterPersistentWorker(moduleDefineHolder, model.getName(), registerDAO, stream.scopeId());
- RegisterRemoteWorker remoteWorker = new RegisterRemoteWorker(moduleDefineHolder, persistentWorker);
+ String remoteReceiverWorkerName = stream.name() + "_rec";
+ IWorkerInstanceSetter workerInstanceSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(IWorkerInstanceSetter.class);
+ workerInstanceSetter.put(remoteReceiverWorkerName, persistentWorker);
+
+ RegisterRemoteWorker remoteWorker = new RegisterRemoteWorker(moduleDefineHolder, remoteReceiverWorkerName);
RegisterDistinctWorker distinctWorker = new RegisterDistinctWorker(moduleDefineHolder, remoteWorker);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterRemoteWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterRemoteWorker.java
index 513a376..8366186 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterRemoteWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterRemoteWorker.java
@@ -33,18 +33,18 @@ public class RegisterRemoteWorker extends AbstractWorker<RegisterSource> {
private static final Logger logger = LoggerFactory.getLogger(RegisterRemoteWorker.class);
- private final AbstractWorker<RegisterSource> nextWorker;
+ private final String remoteReceiverWorkerName;
private final RemoteSenderService remoteSender;
- RegisterRemoteWorker(ModuleDefineHolder moduleDefineHolder, AbstractWorker<RegisterSource> nextWorker) {
+ RegisterRemoteWorker(ModuleDefineHolder moduleDefineHolder, String remoteReceiverWorkerName) {
super(moduleDefineHolder);
this.remoteSender = moduleDefineHolder.find(CoreModule.NAME).provider().getService(RemoteSenderService.class);
- this.nextWorker = nextWorker;
+ this.remoteReceiverWorkerName = remoteReceiverWorkerName;
}
@Override public final void in(RegisterSource registerSource) {
try {
- remoteSender.send(nextWorker.getWorkerId(), registerSource, Selector.ForeverFirst);
+ remoteSender.send(remoteReceiverWorkerName, registerSource, Selector.ForeverFirst);
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java
index a028546..436ecd2 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteSenderService.java
@@ -41,22 +41,22 @@ public class RemoteSenderService implements Service {
this.rollingSelector = new RollingSelector();
}
- public void send(int nextWorkId, StreamData streamData, Selector selector) {
+ public void send(String nextWorkName, StreamData streamData, Selector selector) {
RemoteClientManager clientManager = moduleManager.find(CoreModule.NAME).provider().getService(RemoteClientManager.class);
RemoteClient remoteClient;
switch (selector) {
case HashCode:
remoteClient = hashCodeSelector.select(clientManager.getRemoteClient(), streamData);
- remoteClient.push(nextWorkId, streamData);
+ remoteClient.push(nextWorkName, streamData);
break;
case Rolling:
remoteClient = rollingSelector.select(clientManager.getRemoteClient(), streamData);
- remoteClient.push(nextWorkId, streamData);
+ remoteClient.push(nextWorkName, streamData);
break;
case ForeverFirst:
remoteClient = foreverFirstSelector.select(clientManager.getRemoteClient(), streamData);
- remoteClient.push(nextWorkId, streamData);
+ remoteClient.push(nextWorkName, streamData);
break;
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandler.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandler.java
index 2fd5887..f8fb8f7 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandler.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandler.java
@@ -24,6 +24,7 @@ import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.remote.data.StreamData;
import org.apache.skywalking.oap.server.core.remote.define.StreamDataMappingGetter;
import org.apache.skywalking.oap.server.core.remote.grpc.proto.*;
+import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceGetter;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
import org.apache.skywalking.oap.server.library.server.grpc.GRPCHandler;
@@ -47,6 +48,7 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
private IWorkerInstanceGetter workerInstanceGetter;
private CounterMetrics remoteInCounter;
private CounterMetrics remoteInErrorCounter;
+ private CounterMetrics remoteInTargetNotFoundCounter;
private HistogramMetrics remoteInHistogram;
public RemoteServiceHandler(ModuleDefineHolder moduleDefineHolder) {
@@ -58,6 +60,9 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
remoteInErrorCounter = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class)
.createCounter("remote_in_error_count", "The error number(server side) of inside remote inside aggregate rpc.",
MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
+ remoteInTargetNotFoundCounter = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class)
+ .createCounter("remote_in_target_not_found_count", "The error number(server side) of inside remote handler target worker not found. May be caused by unmatched OAL scrips.",
+ MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
remoteInHistogram = moduleDefineHolder.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class)
.createHistogramMetric("remote_in_latency", "The latency(server side) of inside remote inside aggregate rpc.",
MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
@@ -86,14 +91,20 @@ public class RemoteServiceHandler extends RemoteServiceGrpc.RemoteServiceImplBas
HistogramMetrics.Timer timer = remoteInHistogram.createTimer();
try {
int streamDataId = message.getStreamDataId();
- int nextWorkerId = message.getNextWorkerId();
+ String nextWorkerName = message.getNextWorkName();
RemoteData remoteData = message.getRemoteData();
Class<? extends StreamData> streamDataClass = streamDataMappingGetter.findClassById(streamDataId);
try {
StreamData streamData = streamDataClass.newInstance();
streamData.deserialize(remoteData);
- workerInstanceGetter.get(nextWorkerId).in(streamData);
+ AbstractWorker nextWorker = workerInstanceGetter.get(nextWorkerName);
+ if (nextWorker != null) {
+ nextWorker.in(streamData);
+ } else {
+ remoteInTargetNotFoundCounter.inc();
+ logger.warn("Work name [{}] not found. Check OAL script, make sure they are same in the whole cluster.", nextWorkerName);
+ }
} catch (Throwable t) {
remoteInErrorCounter.inc();
logger.error(t.getMessage(), t);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java
index 4927b97..c4f9e35 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClient.java
@@ -119,13 +119,13 @@ public class GRPCRemoteClient implements RemoteClient {
/**
* Push stream data which need to send to another OAP server.
*
- * @param nextWorkerId the id of a worker which will process this stream data.
+ * @param nextWorkerName the name of a worker which will process this stream data.
* @param streamData the entity contains the values.
*/
- @Override public void push(int nextWorkerId, StreamData streamData) {
+ @Override public void push(String nextWorkerName, StreamData streamData) {
int streamDataId = streamDataMappingGetter.findIdByClass(streamData.getClass());
RemoteMessage.Builder builder = RemoteMessage.newBuilder();
- builder.setNextWorkerId(nextWorkerId);
+ builder.setNextWorkName(nextWorkerName);
builder.setStreamDataId(streamDataId);
builder.setRemoteData(streamData.serialize());
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClient.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClient.java
index ccc216f..3330f3d 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClient.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClient.java
@@ -31,5 +31,5 @@ public interface RemoteClient extends Comparable<RemoteClient> {
void close();
- void push(int nextWorkerId, StreamData streamData);
+ void push(String nextWorkerName, StreamData streamData);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java
index 67827a9..c604fc7 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/SelfRemoteClient.java
@@ -53,8 +53,8 @@ public class SelfRemoteClient implements RemoteClient {
throw new UnexpectedException("Self remote client invoked to close.");
}
- @Override public void push(int nextWorkerId, StreamData streamData) {
- workerInstanceGetter.get(nextWorkerId).in(streamData);
+ @Override public void push(String nextWorkerName, StreamData streamData) {
+ workerInstanceGetter.get(nextWorkerName).in(streamData);
}
@Override public int compareTo(RemoteClient o) {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java
index c68b690..b513d38 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java
@@ -19,21 +19,16 @@
package org.apache.skywalking.oap.server.core.worker;
import lombok.Getter;
-import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
/**
* @author peng-yongsheng
*/
public abstract class AbstractWorker<INPUT> {
-
- @Getter private final int workerId;
@Getter private final ModuleDefineHolder moduleDefineHolder;
public AbstractWorker(ModuleDefineHolder moduleDefineHolder) {
this.moduleDefineHolder = moduleDefineHolder;
- IWorkerInstanceSetter workerInstanceSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(IWorkerInstanceSetter.class);
- this.workerId = workerInstanceSetter.put(this);
}
public abstract void in(INPUT input);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceGetter.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceGetter.java
index 9b54184..2c9cbb4 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceGetter.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceGetter.java
@@ -25,5 +25,5 @@ import org.apache.skywalking.oap.server.library.module.Service;
*/
public interface IWorkerInstanceGetter extends Service {
- AbstractWorker get(int workerId);
+ AbstractWorker get(String nextWorkerName);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceSetter.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceSetter.java
index eef279f..ac215e4 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceSetter.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/IWorkerInstanceSetter.java
@@ -24,6 +24,5 @@ import org.apache.skywalking.oap.server.library.module.Service;
* @author peng-yongsheng
*/
public interface IWorkerInstanceSetter extends Service {
-
- int put(AbstractWorker instance);
+ void put(String remoteReceiverWorkName, AbstractWorker instance);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/WorkerInstancesService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/WorkerInstancesService.java
index 51d671a..5a80c87 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/WorkerInstancesService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/WorkerInstancesService.java
@@ -18,28 +18,33 @@
package org.apache.skywalking.oap.server.core.worker;
-import java.util.*;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.skywalking.oap.server.core.UnexpectedException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* @author peng-yongsheng
*/
public class WorkerInstancesService implements IWorkerInstanceSetter, IWorkerInstanceGetter {
+ private static final Logger logger = LoggerFactory.getLogger(WorkerInstancesService.class);
- private final AtomicInteger generator = new AtomicInteger(1);
- private final Map<Integer, AbstractWorker> instances;
+ private final Map<String, AbstractWorker> instances;
public WorkerInstancesService() {
this.instances = new HashMap<>();
}
- @Override public AbstractWorker get(int workerId) {
- return instances.get(workerId);
+ @Override public AbstractWorker get(String nextWorkerName) {
+ return instances.get(nextWorkerName);
}
- @Override public int put(AbstractWorker instance) {
- int workerId = generator.getAndIncrement();
- instances.put(workerId, instance);
- return workerId;
+ @Override public void put(String remoteReceiverWorkName, AbstractWorker instance) {
+ if (instances.containsKey(remoteReceiverWorkName)) {
+ throw new UnexpectedException("Duplicate worker name:" + remoteReceiverWorkName);
+ }
+ instances.put(remoteReceiverWorkName, instance);
+ logger.debug("Worker {} has been registered as {}", instance.toString(), remoteReceiverWorkName);
}
}
diff --git a/oap-server/server-core/src/main/proto/RemoteService.proto b/oap-server/server-core/src/main/proto/RemoteService.proto
index 853ea82..885a2b4 100644
--- a/oap-server/server-core/src/main/proto/RemoteService.proto
+++ b/oap-server/server-core/src/main/proto/RemoteService.proto
@@ -27,7 +27,7 @@ service RemoteService {
}
message RemoteMessage {
- int32 nextWorkerId = 1;
+ string nextWorkName = 1;
int32 streamDataId = 2;
RemoteData remoteData = 3;
}
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandlerTestCase.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandlerTestCase.java
index b9081d1..7bd5dfb 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandlerTestCase.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/RemoteServiceHandlerTestCase.java
@@ -46,7 +46,7 @@ public class RemoteServiceHandlerTestCase {
@Test
public void callTest() throws DuplicateProviderException, ProviderNotFoundException, IOException {
final int streamDataClassId = 1;
- final int testWorkerId = 1;
+ final String testWorkerId = "mock-worker";
ModuleManagerTesting moduleManager = new ModuleManagerTesting();
ModuleDefineTesting moduleDefine = new ModuleDefineTesting();
@@ -102,7 +102,7 @@ public class RemoteServiceHandlerTestCase {
RemoteMessage.Builder remoteMessage = RemoteMessage.newBuilder();
remoteMessage.setStreamDataId(streamDataClassId);
- remoteMessage.setNextWorkerId(testWorkerId);
+ remoteMessage.setNextWorkName(testWorkerId);
RemoteData.Builder remoteData = RemoteData.newBuilder();
remoteData.addDataStrings("test1");
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientRealClient.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientRealClient.java
index be34824..262ec18 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientRealClient.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientRealClient.java
@@ -57,7 +57,7 @@ public class GRPCRemoteClientRealClient {
remoteClient.connect();
for (int i = 0; i < 10000; i++) {
- remoteClient.push(1, new TestStreamData());
+ remoteClient.push("mock_remote", new TestStreamData());
TimeUnit.SECONDS.sleep(1);
}
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientTestCase.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientTestCase.java
index 8a6494f..a2281f8 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientTestCase.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/remote/client/GRPCRemoteClientTestCase.java
@@ -39,7 +39,7 @@ import static org.mockito.Mockito.*;
*/
public class GRPCRemoteClientTestCase {
- private final int nextWorkerId = 1;
+ private final String nextWorkerId = "mock-worker";
private ModuleManagerTesting moduleManager;
private StreamDataMappingGetter classGetter;
@Rule public final GrpcServerRule grpcServerRule = new GrpcServerRule().directExecutor();