You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by GitBox <gi...@apache.org> on 2018/10/27 11:58:17 UTC
[GitHub] wu-sheng closed pull request #1790: Add service and instance reset
function
wu-sheng closed pull request #1790: Add service and instance reset function
URL: https://github.com/apache/incubator-skywalking/pull/1790
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java
index 76ad609bf..bd9f9a65a 100644
--- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java
+++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/DataCarrier.java
@@ -152,4 +152,10 @@ public void shutdownConsumers() {
consumerPool.close();
}
}
+
+ public void clear() {
+ for (int i = 0; i < channels.getChannelSize(); i++) {
+ channels.getBuffer(i).clear();
+ }
+ }
}
diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Buffer.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Buffer.java
index 4fbb47869..d9cbb34b8 100644
--- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Buffer.java
+++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/buffer/Buffer.java
@@ -91,4 +91,10 @@ public int getBufferSize() {
return result;
}
+ public void clear() {
+ for (Object obj : buffer) {
+ obj = null;
+ }
+ }
+
}
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java
index dfdea5384..c5e305b98 100644
--- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/Config.java
@@ -16,7 +16,6 @@
*
*/
-
package org.apache.skywalking.apm.agent.core.conf;
import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
@@ -70,6 +69,21 @@
* Skywalking team may ask for these files in order to resolve compatible problem.
*/
public static boolean IS_OPEN_DEBUGGING_CLASS = false;
+
+ /**
+ * Specify register.status dir ,This is an option, the default is AGENT_HOME/option/reset.status.
+ */
+ public static String REGISTER_STATUS_DIR = "skywalking-agent/option";
+
+ /**
+ * Specify instance_uuid to ensure that the whole show is unique, for example: applicationName_ip_12
+ */
+ public static String INSTANCE_UUID = "";
+ /**
+ * enabled means that the reset function is enabled, and disabled means that the reset function is not enabled. A reset can be triggered by modifying the configuration file only if the reset feature is enabled.
+ */
+ public static String RESETER_LISTENER = "disabled";
+
}
public static class Collector {
@@ -80,7 +94,7 @@
/**
* application and service registry check interval
*/
- public static long APP_AND_SERVICE_REGISTER_CHECK_INTERVAL = 3;
+ public static long SERVICE_AND_ENDPOINT_REGISTER_CHECK_INTERVAL = 3;
/**
* Collector skywalking trace receiver service addresses.
*/
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/SnifferConfigInitializer.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/SnifferConfigInitializer.java
index d33d30471..3d8ef6363 100644
--- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/SnifferConfigInitializer.java
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/conf/SnifferConfigInitializer.java
@@ -41,24 +41,23 @@
*/
public class SnifferConfigInitializer {
private static final ILog logger = LogManager.getLogger(SnifferConfigInitializer.class);
- private static String SPECIFIED_CONFIG_PATH = "skywalking_config";
- private static String DEFAULT_CONFIG_FILE_NAME = "/config/agent.config";
- private static String ENV_KEY_PREFIX = "skywalking.";
+ private static final String SPECIFIED_CONFIG_PATH = "skywalking_config";
+ private static final String DEFAULT_CONFIG_FILE_NAME = "/config/agent.config";
+ private static final String ENV_KEY_PREFIX = "skywalking.";
+ private static final String INSTANCE_UUID_NAME = "agent.instance_uuid";
+ private static final String REGISTER_STATUS_DIR = "agent.register_status_dir";
private static boolean IS_INIT_COMPLETED = false;
/**
- * If the specified agent config path is set, the agent will try to locate the specified agent config.
- * If the specified agent config path is not set , the agent will try to locate `agent.config`, which should be in the /config dictionary of agent package.
- * <p>
- * Also try to override the config by system.env and system.properties. All the keys in these two places should
- * start with {@link #ENV_KEY_PREFIX}. e.g. in env `skywalking.agent.application_code=yourAppName` to override
- * `agent.application_code` in config file.
- * <p>
- * At the end, `agent.application_code` and `collector.servers` must be not blank.
+ * If the specified agent config path is set, the agent will try to locate the specified agent config. If the
+ * specified agent config path is not set , the agent will try to locate `agent.config`, which should be in the
+ * /config dictionary of agent package. <p> Also try to override the config by system.env and system.properties. All
+ * the keys in these two places should start with {@link #ENV_KEY_PREFIX}. e.g. in env
+ * `skywalking.agent.application_code=yourAppName` to override `agent.application_code` in config file. <p> At the
+ * end, `agent.application_code` and `collector.servers` must be not blank.
*/
public static void initialize() throws ConfigNotFoundException, AgentPackageNotFoundException {
InputStreamReader configFileStream;
-
try {
configFileStream = loadConfig();
Properties properties = new Properties();
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/dictionary/NetworkAddressDictionary.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/dictionary/NetworkAddressDictionary.java
index 208ba1725..ca4d3b519 100644
--- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/dictionary/NetworkAddressDictionary.java
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/dictionary/NetworkAddressDictionary.java
@@ -16,7 +16,6 @@
*
*/
-
package org.apache.skywalking.apm.agent.core.dictionary;
import io.netty.util.internal.ConcurrentSet;
@@ -49,6 +48,12 @@ public PossibleFound find(String networkAddress) {
}
}
+ public void clearApplicationDictionary() {
+ unRegisterApplications.clear();
+ applicationDictionary.clear();
+
+ }
+
public void syncRemoteDictionary(
NetworkAddressRegisterServiceGrpc.NetworkAddressRegisterServiceBlockingStub networkAddressRegisterServiceBlockingStub) {
if (unRegisterApplications.size() > 0) {
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/dictionary/OperationNameDictionary.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/dictionary/OperationNameDictionary.java
index 7f743b593..337bafb94 100644
--- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/dictionary/OperationNameDictionary.java
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/dictionary/OperationNameDictionary.java
@@ -61,6 +61,13 @@ private PossibleFound find0(int applicationId, String operationName,
}
}
+ public void clearOperationNameDictionary() {
+ unRegisterOperationNames.clear();
+ operationNameDictionary.clear();
+
+ }
+
+
public void syncRemoteDictionary(
ServiceNameDiscoveryServiceGrpc.ServiceNameDiscoveryServiceBlockingStub serviceNameDiscoveryServiceBlockingStub) {
if (unRegisterOperationNames.size() > 0) {
@@ -106,6 +113,10 @@ public int getApplicationId() {
return applicationId;
}
+ public void setApplicationId(int applicationId) {
+ this.applicationId = applicationId;
+ }
+
public String getOperationName() {
return operationName;
}
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/ResetConfListener.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/ResetConfListener.java
new file mode 100644
index 000000000..7360651e1
--- /dev/null
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/ResetConfListener.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.skywalking.apm.agent.core.listener;
+
+import java.io.File;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.skywalking.apm.agent.core.boot.BootService;
+import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor;
+import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory;
+import org.apache.skywalking.apm.agent.core.conf.Config;
+import org.apache.skywalking.apm.agent.core.logging.api.ILog;
+import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
+import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
+
+/**
+ * @author liu-xinyuan
+ **/
+@DefaultImplementor
+public class ResetConfListener implements BootService, Runnable {
+ private static final ILog logger = LogManager.getLogger(ResetConfListener.class);
+ private File configFile = null;
+
+ @Override public void prepare() throws Throwable {
+
+ }
+
+ @Override public void boot() {
+ if ("enabled".equals(Config.Agent.RESETER_LISTENER)) {
+ Executors.newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("ResetConfListener"))
+ .scheduleAtFixedRate(new RunnableWithExceptionProtection(this, new RunnableWithExceptionProtection.CallbackWhenException() {
+ @Override
+ public void handle(Throwable t) {
+ logger.error("unexpected exception.", t);
+ }
+ }), 0, Config.Collector.SERVICE_AND_ENDPOINT_REGISTER_CHECK_INTERVAL, TimeUnit.SECONDS);
+
+ } else {
+ logger.info("Since the agent.register_status variable is not set correctly, the reset service is not started -> Agent reset service is inactive.");
+ }
+ }
+ @Override public void onComplete() throws Throwable {
+
+ }
+
+ @Override public void shutdown() throws Throwable {
+
+ }
+
+ @Override public void run() {
+ logger.debug("ResetConfListener running.");
+
+ try {
+ if (Reseter.INSTANCE.predicateReset())
+ Reseter.INSTANCE.setStatus(ResetStatus.DONE).clearID().reportToRegisterFile();
+ } catch (SecurityException e) {
+ logger.warn(e, "Denise read access to the file {}", configFile);
+ } catch (FileNotFoundException e) {
+ logger.warn(e, "not found file {}", configFile);
+ } catch (IOException e) {
+ logger.warn(e.getMessage());
+ }
+
+ }
+
+}
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/ResetStatus.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/ResetStatus.java
new file mode 100644
index 000000000..ac9de48f4
--- /dev/null
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/ResetStatus.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.skywalking.apm.agent.core.listener;
+
+/**
+ * @author liu-xinyuan
+ **/
+public enum ResetStatus {
+ OFF("OFF"), ON("ON"), DONE("DONE");
+ private String label;
+
+ ResetStatus(String label) {
+ this.label = label;
+ }
+
+ public String value() {
+ return this.label;
+ }
+
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/Reseter.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/Reseter.java
new file mode 100644
index 000000000..cf1ecea4b
--- /dev/null
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/listener/Reseter.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.skywalking.apm.agent.core.listener;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.util.Properties;
+import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
+import org.apache.skywalking.apm.agent.core.conf.Config;
+import org.apache.skywalking.apm.agent.core.conf.RemoteDownstreamConfig;
+import org.apache.skywalking.apm.agent.core.dictionary.DictionaryUtil;
+import org.apache.skywalking.apm.agent.core.dictionary.NetworkAddressDictionary;
+import org.apache.skywalking.apm.agent.core.dictionary.OperationNameDictionary;
+import org.apache.skywalking.apm.agent.core.logging.api.ILog;
+import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
+import org.apache.skywalking.apm.agent.core.remote.TraceSegmentServiceClient;
+
+/**
+ * @author liu-xinyuan
+ **/
+public enum Reseter {
+ INSTANCE;
+ private static final ILog logger = LogManager.getLogger(Reseter.class);
+ private static final String APPLICATION_ID_NAM = "application_id";
+ private static final String INSTANCE_ID_NAME = "instance_id";
+ private static final String STATUS_NAME = "status";
+ private static final String RESET_CHILD_DIR = "/reset.status";
+ private static final String COMMENT = "#Status has three values: ON (trigger reset), DONE(reset complete), OFF(agent fist boot).\n" +
+ "Application_id: application_id of the current agent.\n" +
+ "Instance_id: the instanceid of the current agent.";
+ private volatile Properties properties = new Properties();
+ private String resetPath;
+ private ResetStatus status = ResetStatus.OFF;
+ private int isFirstRun = 0;
+ private int detectDuration = 5;
+
+ public Reseter setStatus(ResetStatus status) {
+ this.status = status;
+ return this;
+ }
+
+ public String getResetPath() throws IOException {
+ if (isFirstRun == 0) {
+ File statusDir = new File(Config.Agent.REGISTER_STATUS_DIR);
+
+ if (statusDir.exists() && statusDir.isDirectory()) {
+ resetPath = statusDir.getAbsolutePath() + RESET_CHILD_DIR;
+ } else {
+ statusDir.mkdir();
+ }
+ init();
+ isFirstRun = 1;
+ }
+ return resetPath;
+ }
+
+ public void reportToRegisterFile() throws IOException {
+ FileOutputStream outputStream = null;
+ try {
+ File configFile = new File(resetPath);
+ properties.setProperty(APPLICATION_ID_NAM, RemoteDownstreamConfig.Agent.APPLICATION_ID + "");
+ properties.setProperty(INSTANCE_ID_NAME, RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID + "");
+ properties.setProperty(STATUS_NAME, status.value());
+ outputStream = new FileOutputStream(configFile);
+ properties.store(outputStream, COMMENT);
+ } finally {
+ closeFileStream(outputStream);
+ }
+ }
+
+ public Reseter clearID() {
+ RemoteDownstreamConfig.Agent.APPLICATION_ID = DictionaryUtil.nullValue();
+ RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID = DictionaryUtil.nullValue();
+ OperationNameDictionary.INSTANCE.clearOperationNameDictionary();
+ NetworkAddressDictionary.INSTANCE.clearApplicationDictionary();
+ ServiceManager.INSTANCE.findService(TraceSegmentServiceClient.class).clearCache();
+ status = ResetStatus.DONE;
+ logger.info("clear id successfully,begin trigger reset.");
+ return this;
+ }
+
+ public Boolean predicateReset() throws IOException {
+ File resetFile = new File(getResetPath());
+ FileInputStream inputStream = null;
+ FileLock fileLock = null;
+ FileChannel fileChannel = null;
+ if (System.currentTimeMillis() - resetFile.lastModified() < detectDuration * 1000) {
+ try {
+ logger.info("The file reset.status was detected to have been modified in the last {} seconds.", detectDuration);
+ inputStream = new FileInputStream(resetFile);
+ fileChannel = inputStream.getChannel();
+ fileLock = fileChannel.tryLock(0, resetFile.length(), true);
+ if (fileLock == null) {
+ return false;
+ }
+ properties.clear();
+ properties.load(inputStream);
+ } finally {
+ fileLock.release();
+ fileChannel.close();
+ closeFileStream(inputStream);
+ }
+ if (properties.get(STATUS_NAME) != null && properties.getProperty(STATUS_NAME).equals(ResetStatus.ON.value())) {
+ return true;
+ }
+ }
+ return false;
+
+ }
+
+ public void init() throws IOException {
+ FileOutputStream outputStream = null;
+ try {
+ properties.setProperty(APPLICATION_ID_NAM, RemoteDownstreamConfig.Agent.APPLICATION_ID + "");
+ properties.setProperty(INSTANCE_ID_NAME, RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID + "");
+ properties.setProperty(STATUS_NAME, status.value());
+ File file = new File(resetPath);
+ if (!file.getParentFile().exists()) {
+ file.getParentFile().mkdir();
+ }
+ outputStream = new FileOutputStream(file);
+ properties.store(outputStream, COMMENT);
+ } finally {
+ closeFileStream(outputStream);
+ }
+ }
+
+ public void closeFileStream(Closeable stream) throws IOException {
+ if (stream != null) {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ throw new IOException("file close failed.", e);
+ }
+ } else {
+ throw new IOException("create file outputstream failed");
+ }
+ }
+}
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java
index 3ce38d970..15459c678 100644
--- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/AppAndServiceRegisterClient.java
@@ -31,16 +31,21 @@
import org.apache.skywalking.apm.agent.core.dictionary.DictionaryUtil;
import org.apache.skywalking.apm.agent.core.dictionary.NetworkAddressDictionary;
import org.apache.skywalking.apm.agent.core.dictionary.OperationNameDictionary;
+import org.apache.skywalking.apm.agent.core.listener.ResetStatus;
+import org.apache.skywalking.apm.agent.core.listener.Reseter;
import org.apache.skywalking.apm.agent.core.logging.api.ILog;
import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
import org.apache.skywalking.apm.agent.core.os.OSUtil;
import org.apache.skywalking.apm.network.language.agent.*;
+import org.apache.skywalking.apm.network.register.ServiceInstancePingGrpc;
+import org.apache.skywalking.apm.network.register.ServiceInstancePingPkg;
import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import org.apache.skywalking.apm.util.StringUtil;
/**
* @author wusheng
@@ -48,13 +53,13 @@
@DefaultImplementor
public class AppAndServiceRegisterClient implements BootService, GRPCChannelListener, Runnable, TracingContextListener {
private static final ILog logger = LogManager.getLogger(AppAndServiceRegisterClient.class);
- private static final String PROCESS_UUID = UUID.randomUUID().toString().replaceAll("-", "");
private volatile GRPCChannelStatus status = GRPCChannelStatus.DISCONNECT;
private volatile ApplicationRegisterServiceGrpc.ApplicationRegisterServiceBlockingStub applicationRegisterServiceBlockingStub;
private volatile InstanceDiscoveryServiceGrpc.InstanceDiscoveryServiceBlockingStub instanceDiscoveryServiceBlockingStub;
private volatile ServiceNameDiscoveryServiceGrpc.ServiceNameDiscoveryServiceBlockingStub serviceNameDiscoveryServiceBlockingStub;
private volatile NetworkAddressRegisterServiceGrpc.NetworkAddressRegisterServiceBlockingStub networkAddressRegisterServiceBlockingStub;
+ private volatile ServiceInstancePingGrpc.ServiceInstancePingBlockingStub serviceInstancePingBlockingStub;
private volatile ScheduledFuture<?> applicationRegisterFuture;
private volatile long lastSegmentTime = -1;
@@ -66,6 +71,7 @@ public void statusChanged(GRPCChannelStatus status) {
instanceDiscoveryServiceBlockingStub = InstanceDiscoveryServiceGrpc.newBlockingStub(channel);
serviceNameDiscoveryServiceBlockingStub = ServiceNameDiscoveryServiceGrpc.newBlockingStub(channel);
networkAddressRegisterServiceBlockingStub = NetworkAddressRegisterServiceGrpc.newBlockingStub(channel);
+ serviceInstancePingBlockingStub = ServiceInstancePingGrpc.newBlockingStub(channel);
} else {
applicationRegisterServiceBlockingStub = null;
instanceDiscoveryServiceBlockingStub = null;
@@ -82,13 +88,13 @@ public void prepare() throws Throwable {
@Override
public void boot() throws Throwable {
applicationRegisterFuture = Executors
- .newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("AppAndServiceRegisterClient"))
- .scheduleAtFixedRate(new RunnableWithExceptionProtection(this, new RunnableWithExceptionProtection.CallbackWhenException() {
- @Override
- public void handle(Throwable t) {
- logger.error("unexpected exception.", t);
- }
- }), 0, Config.Collector.APP_AND_SERVICE_REGISTER_CHECK_INTERVAL, TimeUnit.SECONDS);
+ .newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("AppAndServiceRegisterClient"))
+ .scheduleAtFixedRate(new RunnableWithExceptionProtection(this, new RunnableWithExceptionProtection.CallbackWhenException() {
+ @Override
+ public void handle(Throwable t) {
+ logger.error("unexpected exception.", t);
+ }
+ }), 0, Config.Collector.SERVICE_AND_ENDPOINT_REGISTER_CHECK_INTERVAL, TimeUnit.SECONDS);
}
@Override
@@ -105,15 +111,17 @@ public void shutdown() throws Throwable {
public void run() {
logger.debug("AppAndServiceRegisterClient running, status:{}.", status);
boolean shouldTry = true;
+ String instanceUUID = StringUtil.isEmpty(Config.Agent.INSTANCE_UUID) ? UUID.randomUUID().toString().replaceAll("-", "") : Config.Agent.INSTANCE_UUID;
while (GRPCChannelStatus.CONNECTED.equals(status) && shouldTry) {
shouldTry = false;
try {
if (RemoteDownstreamConfig.Agent.APPLICATION_ID == DictionaryUtil.nullValue()) {
if (applicationRegisterServiceBlockingStub != null) {
ApplicationMapping applicationMapping = applicationRegisterServiceBlockingStub.applicationCodeRegister(
- Application.newBuilder().setApplicationCode(Config.Agent.APPLICATION_CODE).build());
+ Application.newBuilder().setApplicationCode(Config.Agent.APPLICATION_CODE).build());
if (applicationMapping != null) {
RemoteDownstreamConfig.Agent.APPLICATION_ID = applicationMapping.getApplication().getValue();
+ Reseter.INSTANCE.reportToRegisterFile();
shouldTry = true;
}
}
@@ -122,21 +130,24 @@ public void run() {
if (RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID == DictionaryUtil.nullValue()) {
ApplicationInstanceMapping instanceMapping = instanceDiscoveryServiceBlockingStub.registerInstance(ApplicationInstance.newBuilder()
- .setApplicationId(RemoteDownstreamConfig.Agent.APPLICATION_ID)
- .setAgentUUID(PROCESS_UUID)
- .setRegisterTime(System.currentTimeMillis())
- .setOsinfo(OSUtil.buildOSInfo())
- .build());
+ .setApplicationId(RemoteDownstreamConfig.Agent.APPLICATION_ID)
+ .setAgentUUID(instanceUUID)
+ .setRegisterTime(System.currentTimeMillis())
+ .setOsinfo(OSUtil.buildOSInfo())
+ .build());
if (instanceMapping.getApplicationInstanceId() != DictionaryUtil.nullValue()) {
RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID
- = instanceMapping.getApplicationInstanceId();
+ = instanceMapping.getApplicationInstanceId();
+ Reseter.INSTANCE.setStatus(ResetStatus.OFF).reportToRegisterFile();
+
}
} else {
if (lastSegmentTime - System.currentTimeMillis() > 60 * 1000) {
- instanceDiscoveryServiceBlockingStub.heartbeat(ApplicationInstanceHeartbeat.newBuilder()
- .setApplicationInstanceId(RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID)
- .setHeartbeatTime(System.currentTimeMillis())
- .build());
+ serviceInstancePingBlockingStub.doPing(ServiceInstancePingPkg.newBuilder()
+ .setServiceInstanceId(RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID)
+ .setServiceInstanceUUID(instanceUUID)
+ .setTime(System.currentTimeMillis())
+ .build());
}
NetworkAddressDictionary.INSTANCE.syncRemoteDictionary(networkAddressRegisterServiceBlockingStub);
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
index 697d747fd..2c277fe4b 100644
--- a/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
+++ b/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/apm/agent/core/remote/TraceSegmentServiceClient.java
@@ -25,9 +25,11 @@
import org.apache.skywalking.apm.agent.core.boot.BootService;
import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor;
import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
+import org.apache.skywalking.apm.agent.core.conf.RemoteDownstreamConfig;
import org.apache.skywalking.apm.agent.core.context.TracingContext;
import org.apache.skywalking.apm.agent.core.context.TracingContextListener;
import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.dictionary.DictionaryUtil;
import org.apache.skywalking.apm.agent.core.logging.api.ILog;
import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
@@ -155,7 +157,7 @@ public void onExit() {
@Override
public void afterFinished(TraceSegment traceSegment) {
- if (traceSegment.isIgnore()) {
+ if (traceSegment.isIgnore() || RemoteDownstreamConfig.Agent.APPLICATION_INSTANCE_ID == DictionaryUtil.nullValue() || RemoteDownstreamConfig.Agent.APPLICATION_ID == DictionaryUtil.nullValue()) {
return;
}
if (!carrier.produce(traceSegment)) {
@@ -173,4 +175,9 @@ public void statusChanged(GRPCChannelStatus status) {
}
this.status = status;
}
+
+ public void clearCache() {
+ carrier.clear();
+ }
+
}
diff --git a/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.apache.skywalking.apm.agent.core.boot.BootService b/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.apache.skywalking.apm.agent.core.boot.BootService
index 9beec08f1..14bef1c0b 100644
--- a/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.apache.skywalking.apm.agent.core.boot.BootService
+++ b/apm-sniffer/apm-agent-core/src/main/resources/META-INF/services/org.apache.skywalking.apm.agent.core.boot.BootService
@@ -23,3 +23,4 @@ org.apache.skywalking.apm.agent.core.remote.GRPCChannelManager
org.apache.skywalking.apm.agent.core.jvm.JVMService
org.apache.skywalking.apm.agent.core.remote.AppAndServiceRegisterClient
org.apache.skywalking.apm.agent.core.context.ContextManagerExtendService
+org.apache.skywalking.apm.agent.core.listener.ResetConfListener
\ No newline at end of file
diff --git a/apm-sniffer/apm-agent-core/src/test/java/org/apache/skywalking/apm/agent/core/boot/ServiceManagerTest.java b/apm-sniffer/apm-agent-core/src/test/java/org/apache/skywalking/apm/agent/core/boot/ServiceManagerTest.java
index 3009f4b95..46544f37b 100644
--- a/apm-sniffer/apm-agent-core/src/test/java/org/apache/skywalking/apm/agent/core/boot/ServiceManagerTest.java
+++ b/apm-sniffer/apm-agent-core/src/test/java/org/apache/skywalking/apm/agent/core/boot/ServiceManagerTest.java
@@ -55,7 +55,7 @@ public static void afterClass() {
public void testServiceDependencies() throws Exception {
HashMap<Class, BootService> registryService = getFieldValue(ServiceManager.INSTANCE, "bootedServices");
- assertThat(registryService.size(), is(7));
+ assertThat(registryService.size(), is(8));
assertTraceSegmentServiceClient(ServiceManager.INSTANCE.findService(TraceSegmentServiceClient.class));
assertContextManager(ServiceManager.INSTANCE.findService(ContextManager.class));
diff --git a/apm-sniffer/config/agent.config b/apm-sniffer/config/agent.config
index fbf0c6baf..6c16f8fb3 100644
--- a/apm-sniffer/config/agent.config
+++ b/apm-sniffer/config/agent.config
@@ -43,3 +43,12 @@ collector.backend_service=127.0.0.1:10800
# Logging level
logging.level=DEBUG
+
+#Specify register.status dir,if dir not exists or it is a file then default AGENT_HOME/option
+#agent.register_status_dir=register_dir
+
+#Specify instance_uuid to ensure that the whole show is unique, for example: applicationName_ip_12
+#agent.instance_uuid = applicationName_ip_1
+
+#enabled means that the reset function is enabled, and disabled means that the reset function is not enabled. A reset can be triggered by modifying the configuration file only if the reset feature is enabled.
+#agent.reseter_listener = disabled
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInstanceInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInstanceInventoryRegister.java
index af0299226..1fd785183 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInstanceInventoryRegister.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInstanceInventoryRegister.java
@@ -109,4 +109,4 @@ private ServiceInstanceInventoryCache getServiceInstanceInventoryCache() {
logger.warn("Service instance {} heartbeat, but not found in storage.");
}
}
-}
+}
\ No newline at end of file
diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol
index 1122e97b5..3a83be79a 160000
--- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol
+++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol
@@ -1 +1 @@
-Subproject commit 1122e97b5604ae96447bd58ecdb248d7e02952aa
+Subproject commit 3a83be79a9c23aad6576ed2a4a04b82de6d7a829
diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/RegisterModuleProvider.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/RegisterModuleProvider.java
index 6a1014146..d6a6f2f4c 100644
--- a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/RegisterModuleProvider.java
+++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/RegisterModuleProvider.java
@@ -22,7 +22,6 @@
import org.apache.skywalking.oap.server.core.server.*;
import org.apache.skywalking.oap.server.library.module.*;
import org.apache.skywalking.oap.server.receiver.register.module.RegisterModule;
-import org.apache.skywalking.oap.server.receiver.register.provider.handler.v5.grpc.*;
import org.apache.skywalking.oap.server.receiver.register.provider.handler.v5.rest.*;
/**
@@ -46,12 +45,6 @@
}
@Override public void start() {
- GRPCHandlerRegister grpcHandlerRegister = getManager().find(CoreModule.NAME).getService(GRPCHandlerRegister.class);
- grpcHandlerRegister.addHandler(new ApplicationRegisterHandler(getManager()));
- grpcHandlerRegister.addHandler(new InstanceDiscoveryServiceHandler(getManager()));
- grpcHandlerRegister.addHandler(new ServiceNameDiscoveryHandler(getManager()));
- grpcHandlerRegister.addHandler(new NetworkAddressRegisterServiceHandler(getManager()));
-
JettyHandlerRegister jettyHandlerRegister = getManager().find(CoreModule.NAME).getService(JettyHandlerRegister.class);
jettyHandlerRegister.addHandler(new ApplicationRegisterServletHandler(getManager()));
jettyHandlerRegister.addHandler(new InstanceDiscoveryServletHandler(getManager()));
diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/ServiceInstancePingPkgHandler.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/ServiceInstancePingPkgHandler.java
new file mode 100644
index 000000000..15d50c212
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v5/ServiceInstancePingPkgHandler.java
@@ -0,0 +1,60 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. See the NOTICE file distributed with
+ * * this work for additional information regarding copyright ownership.
+ * * The ASF licenses this file to You under the Apache License, Version 2.0
+ * * (the "License"); you may not use this file except in compliance with
+ * * the License. You may obtain a copy of the License at
+ * *
+ * * http://www.apache.org/licenses/LICENSE-2.0
+ * *
+ * * Unless required by applicable law or agreed to in writing, software
+ * * distributed under the License is distributed on an "AS IS" BASIS,
+ * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * * See the License for the specific language governing permissions and
+ * * limitations under the License.
+ * *
+ *
+ */
+
+package org.apache.skywalking.oap.server.receiver.register.provider.handler.v5;
+
+import io.grpc.stub.StreamObserver;
+import org.apache.skywalking.apm.network.common.Commands;
+import org.apache.skywalking.apm.network.register.ServiceInstancePingGrpc;
+import org.apache.skywalking.apm.network.register.ServiceInstancePingPkg;
+import org.apache.skywalking.oap.server.core.Const;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory;
+import org.apache.skywalking.oap.server.core.register.service.IServiceInstanceInventoryRegister;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.cache.IServiceInstanceInventoryCacheDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.server.grpc.GRPCHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ServiceInstancePingPkgHandler extends ServiceInstancePingGrpc.ServiceInstancePingImplBase implements GRPCHandler {
+ private static final Logger logger = LoggerFactory.getLogger(ServiceInstancePingPkgHandler.class);
+
+ private final IServiceInstanceInventoryCacheDAO instanceInventoryCacheDAO;
+ private final IServiceInstanceInventoryRegister serviceInstanceInventoryRegister;
+
+ public ServiceInstancePingPkgHandler(ModuleManager moduleManager) {
+ this.instanceInventoryCacheDAO = moduleManager.find(StorageModule.NAME).getService(IServiceInstanceInventoryCacheDAO.class);
+ this.serviceInstanceInventoryRegister = moduleManager.find(CoreModule.NAME).getService(IServiceInstanceInventoryRegister.class);
+
+ }
+
+ @Override public void doPing(ServiceInstancePingPkg request, StreamObserver<Commands> responseObserver) {
+
+ ServiceInstanceInventory serviceInstanceInventory = instanceInventoryCacheDAO.get(request.getServiceInstanceId());
+ if (request.getServiceInstanceUUID().equals(serviceInstanceInventory.getName()) || serviceInstanceInventory.getServiceId() == Const.NONE) {
+ logger.error("Your metadata loss,please set the status in reset.status in the agent {} to ON to trigger a reset!", request.getServiceInstanceUUID());
+ } else {
+ serviceInstanceInventoryRegister.heartbeat(request.getServiceInstanceId(), request.getTime());
+ }
+
+ }
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services