Merge remote-tracking branch 'origin/6.0' into 6.0
This commit is contained in:
Gao Hongtao
commit
f80de42880
|
|
@ -1 +1 @@
|
|||
Subproject commit 75c74186a1548657013a299f388e6e8b7b4b5251
|
||||
Subproject commit e259efbc20a69e23a5c8542394579f006396e6a5
|
||||
|
|
@ -82,11 +82,6 @@
|
|||
<artifactId>disruptor</artifactId>
|
||||
<version>3.3.6</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.httpcomponents</groupId>
|
||||
<artifactId>httpclient</artifactId>
|
||||
<version>4.5.3</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.eclipse.jetty</groupId>
|
||||
<artifactId>jetty-server</artifactId>
|
||||
|
|
|
|||
|
|
@ -82,32 +82,9 @@ public class Config {
|
|||
*/
|
||||
public static long APP_AND_SERVICE_REGISTER_CHECK_INTERVAL = 3;
|
||||
/**
|
||||
* discovery rest check interval
|
||||
* Collector skywalking trace receiver service addresses.
|
||||
*/
|
||||
public static long DISCOVERY_CHECK_INTERVAL = 60;
|
||||
/**
|
||||
* Collector naming/jetty service addresses.
|
||||
* Primary address setting.
|
||||
*
|
||||
* e.g.
|
||||
* SERVERS="127.0.0.1:10800" for single collector node.
|
||||
* SERVERS="10.2.45.126:10800,10.2.45.127:10800" for multi collector nodes.
|
||||
*/
|
||||
public static String SERVERS = "";
|
||||
|
||||
/**
|
||||
* Collector agent_gRPC/grpc service addresses.
|
||||
* Secondary address setting, only effect when #SERVERS is empty.
|
||||
*
|
||||
* By using this, no discovery mechanism provided. The agent only uses these addresses to uplink data.
|
||||
*
|
||||
*/
|
||||
public static String DIRECT_SERVERS = "";
|
||||
|
||||
/**
|
||||
* Collector service discovery REST service name
|
||||
*/
|
||||
public static String DISCOVERY_SERVICE_NAME = "/agent/gRPC";
|
||||
public static String BACKEND_SERVICE = "";
|
||||
}
|
||||
|
||||
public static class Jvm {
|
||||
|
|
|
|||
|
|
@ -19,8 +19,6 @@
|
|||
|
||||
package org.apache.skywalking.apm.agent.core.conf;
|
||||
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import org.apache.skywalking.apm.agent.core.dictionary.DictionaryUtil;
|
||||
|
||||
/**
|
||||
|
|
@ -35,11 +33,4 @@ public class RemoteDownstreamConfig {
|
|||
|
||||
public volatile static int APPLICATION_INSTANCE_ID = DictionaryUtil.nullValue();
|
||||
}
|
||||
|
||||
public static class Collector {
|
||||
/**
|
||||
* Collector GRPC-Service address.
|
||||
*/
|
||||
public volatile static List<String> GRPC_SERVERS = new LinkedList<String>();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -75,7 +75,7 @@ public class SnifferConfigInitializer {
|
|||
if (StringUtil.isEmpty(Config.Agent.APPLICATION_CODE)) {
|
||||
throw new ExceptionInInitializerError("`agent.application_code` is missing.");
|
||||
}
|
||||
if (StringUtil.isEmpty(Config.Collector.SERVERS) && StringUtil.isEmpty(Config.Collector.DIRECT_SERVERS)) {
|
||||
if (StringUtil.isEmpty(Config.Collector.BACKEND_SERVICE)) {
|
||||
throw new ExceptionInInitializerError("`collector.direct_servers` and `collector.servers` cannot be empty at the same time.");
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,87 +0,0 @@
|
|||
/*
|
||||
* 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.remote;
|
||||
|
||||
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.conf.RemoteDownstreamConfig;
|
||||
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;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* The <code>CollectorDiscoveryService</code> is responsible for start {@link DiscoveryRestServiceClient}.
|
||||
*
|
||||
* @author wusheng
|
||||
*/
|
||||
@DefaultImplementor
|
||||
public class CollectorDiscoveryService implements BootService {
|
||||
private static final ILog logger = LogManager.getLogger(CollectorDiscoveryService.class);
|
||||
private ScheduledFuture<?> future;
|
||||
|
||||
@Override
|
||||
public void prepare() {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void boot() {
|
||||
DiscoveryRestServiceClient discoveryRestServiceClient = new DiscoveryRestServiceClient();
|
||||
if (discoveryRestServiceClient.hasNamingServer()) {
|
||||
discoveryRestServiceClient.run();
|
||||
future = Executors.newSingleThreadScheduledExecutor(
|
||||
new DefaultNamedThreadFactory("CollectorDiscoveryService"))
|
||||
.scheduleAtFixedRate(new RunnableWithExceptionProtection(discoveryRestServiceClient, new RunnableWithExceptionProtection.CallbackWhenException() {
|
||||
@Override
|
||||
public void handle(Throwable t) {
|
||||
logger.error("unexpected exception.", t);
|
||||
}
|
||||
}),
|
||||
Config.Collector.DISCOVERY_CHECK_INTERVAL,
|
||||
Config.Collector.DISCOVERY_CHECK_INTERVAL,
|
||||
TimeUnit.SECONDS);
|
||||
} else {
|
||||
if (Config.Collector.DIRECT_SERVERS == null || Config.Collector.DIRECT_SERVERS.trim().length() == 0) {
|
||||
logger.error("Collector server and direct server addresses are both not set.");
|
||||
logger.error("Agent will not uplink any data.");
|
||||
return;
|
||||
}
|
||||
RemoteDownstreamConfig.Collector.GRPC_SERVERS = Arrays.asList(Config.Collector.DIRECT_SERVERS.split(","));
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onComplete() throws Throwable {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void shutdown() throws Throwable {
|
||||
if (future != null) {
|
||||
future.cancel(true);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,165 +0,0 @@
|
|||
/*
|
||||
* 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.remote;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
import com.google.gson.JsonArray;
|
||||
import com.google.gson.JsonElement;
|
||||
import org.apache.http.client.config.RequestConfig;
|
||||
import org.apache.http.client.methods.CloseableHttpResponse;
|
||||
import org.apache.http.client.methods.HttpGet;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
import org.apache.http.impl.client.HttpClients;
|
||||
import org.apache.http.util.EntityUtils;
|
||||
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 java.io.IOException;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Random;
|
||||
|
||||
import static org.apache.skywalking.apm.agent.core.conf.RemoteDownstreamConfig.Collector.GRPC_SERVERS;
|
||||
|
||||
/**
|
||||
* The <code>DiscoveryRestServiceClient</code> try to get the collector's grpc-server list
|
||||
* in every 60 seconds,
|
||||
* and override {@link org.apache.skywalking.apm.agent.core.conf.RemoteDownstreamConfig.Collector#GRPC_SERVERS}.
|
||||
*
|
||||
* @author wusheng
|
||||
*/
|
||||
public class DiscoveryRestServiceClient implements Runnable {
|
||||
private static final ILog logger = LogManager.getLogger(DiscoveryRestServiceClient.class);
|
||||
private static final int HTTP_CONNECT_TIMEOUT = 2000;
|
||||
private static final int HTTP_CONNECTION_REQUEST_TIMEOUT = 1000;
|
||||
private static final int HTTP_SOCKET_TIMEOUT = 2000;
|
||||
private String[] serverList;
|
||||
private volatile int selectedServer = -1;
|
||||
|
||||
public DiscoveryRestServiceClient() {
|
||||
if (Config.Collector.SERVERS == null || Config.Collector.SERVERS.trim().length() == 0) {
|
||||
logger.warn("Collector server not set.");
|
||||
return;
|
||||
}
|
||||
|
||||
serverList = Config.Collector.SERVERS.split(",");
|
||||
Random r = new Random();
|
||||
if (serverList.length > 0) {
|
||||
selectedServer = r.nextInt(serverList.length);
|
||||
}
|
||||
}
|
||||
|
||||
boolean hasNamingServer() {
|
||||
return serverList != null && serverList.length > 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
findServerList();
|
||||
} catch (Throwable t) {
|
||||
logger.error(t, "Find server list fail.");
|
||||
}
|
||||
}
|
||||
|
||||
private void findServerList() throws RESTResponseStatusError, IOException {
|
||||
CloseableHttpClient httpClient = HttpClients.custom().build();
|
||||
try {
|
||||
HttpGet httpGet = buildGet();
|
||||
if (httpGet != null) {
|
||||
CloseableHttpResponse httpResponse = httpClient.execute(httpGet);
|
||||
int statusCode = httpResponse.getStatusLine().getStatusCode();
|
||||
if (200 != statusCode) {
|
||||
findBackupServer();
|
||||
throw new RESTResponseStatusError(statusCode);
|
||||
} else {
|
||||
JsonArray serverList = new Gson().fromJson(EntityUtils.toString(httpResponse.getEntity()), JsonArray.class);
|
||||
if (serverList != null && serverList.size() > 0) {
|
||||
LinkedList<String> newServerList = new LinkedList<String>();
|
||||
for (JsonElement element : serverList) {
|
||||
newServerList.add(element.getAsString());
|
||||
}
|
||||
|
||||
if (!isListEquals(newServerList, GRPC_SERVERS)) {
|
||||
GRPC_SERVERS = newServerList;
|
||||
logger.debug("Refresh GRPC server list: {}", GRPC_SERVERS);
|
||||
} else {
|
||||
logger.debug("GRPC server list remain unchanged: {}", GRPC_SERVERS);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (IOException e) {
|
||||
findBackupServer();
|
||||
throw e;
|
||||
} finally {
|
||||
httpClient.close();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isListEquals(List<String> list1, List<String> list2) {
|
||||
if (list1.size() != list2.size()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
for (String ip1 : list1) {
|
||||
if (!list2.contains(ip1)) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Prepare the given message for HTTP Post service.
|
||||
*
|
||||
* @return {@link HttpGet}, when is ready to send. otherwise, null.
|
||||
*/
|
||||
private HttpGet buildGet() {
|
||||
if (selectedServer == -1) {
|
||||
//no available server
|
||||
return null;
|
||||
}
|
||||
HttpGet httpGet = new HttpGet("http://" + serverList[selectedServer] + Config.Collector.DISCOVERY_SERVICE_NAME);
|
||||
RequestConfig requestConfig = RequestConfig.custom()
|
||||
.setConnectTimeout(HTTP_CONNECT_TIMEOUT)
|
||||
.setConnectionRequestTimeout(HTTP_CONNECTION_REQUEST_TIMEOUT)
|
||||
.setSocketTimeout(HTTP_SOCKET_TIMEOUT).build();
|
||||
httpGet.setConfig(requestConfig);
|
||||
return httpGet;
|
||||
}
|
||||
|
||||
/**
|
||||
* Choose the next server in {@link #serverList}, by moving {@link #selectedServer}.
|
||||
*/
|
||||
private void findBackupServer() {
|
||||
selectedServer++;
|
||||
if (selectedServer >= serverList.length) {
|
||||
selectedServer = 0;
|
||||
}
|
||||
|
||||
if (serverList.length == 0) {
|
||||
selectedServer = -1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -21,6 +21,7 @@ package org.apache.skywalking.apm.agent.core.remote;
|
|||
import io.grpc.Channel;
|
||||
import io.grpc.Status;
|
||||
import io.grpc.StatusRuntimeException;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
|
|
@ -32,7 +33,6 @@ 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.conf.RemoteDownstreamConfig;
|
||||
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;
|
||||
|
|
@ -49,6 +49,7 @@ public class GRPCChannelManager implements BootService, Runnable {
|
|||
private volatile boolean reconnect = true;
|
||||
private Random random = new Random();
|
||||
private List<GRPCChannelListener> listeners = Collections.synchronizedList(new LinkedList<GRPCChannelListener>());
|
||||
private volatile List<String> grpcServers;
|
||||
|
||||
@Override
|
||||
public void prepare() throws Throwable {
|
||||
|
|
@ -57,6 +58,12 @@ public class GRPCChannelManager implements BootService, Runnable {
|
|||
|
||||
@Override
|
||||
public void boot() throws Throwable {
|
||||
if (Config.Collector.BACKEND_SERVICE.trim().length() == 0) {
|
||||
logger.error("Collector server addresses are not set.");
|
||||
logger.error("Agent will not uplink any data.");
|
||||
return;
|
||||
}
|
||||
grpcServers = Arrays.asList(Config.Collector.BACKEND_SERVICE.split(","));
|
||||
connectCheckFuture = Executors
|
||||
.newSingleThreadScheduledExecutor(new DefaultNamedThreadFactory("GRPCChannelManager"))
|
||||
.scheduleAtFixedRate(new RunnableWithExceptionProtection(this, new RunnableWithExceptionProtection.CallbackWhenException() {
|
||||
|
|
@ -85,11 +92,11 @@ public class GRPCChannelManager implements BootService, Runnable {
|
|||
public void run() {
|
||||
logger.debug("Selected collector grpc service running, reconnect:{}.", reconnect);
|
||||
if (reconnect) {
|
||||
if (RemoteDownstreamConfig.Collector.GRPC_SERVERS.size() > 0) {
|
||||
if (grpcServers.size() > 0) {
|
||||
String server = "";
|
||||
try {
|
||||
int index = Math.abs(random.nextInt()) % RemoteDownstreamConfig.Collector.GRPC_SERVERS.size();
|
||||
server = RemoteDownstreamConfig.Collector.GRPC_SERVERS.get(index);
|
||||
int index = Math.abs(random.nextInt()) % grpcServers.size();
|
||||
server = grpcServers.get(index);
|
||||
String[] ipAndPort = server.split(":");
|
||||
|
||||
managedChannel = GRPCChannel.newBuilder(ipAndPort[0], Integer.parseInt(ipAndPort[1]))
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@
|
|||
|
||||
org.apache.skywalking.apm.agent.core.remote.TraceSegmentServiceClient
|
||||
org.apache.skywalking.apm.agent.core.context.ContextManager
|
||||
org.apache.skywalking.apm.agent.core.remote.CollectorDiscoveryService
|
||||
org.apache.skywalking.apm.agent.core.sampling.SamplingService
|
||||
org.apache.skywalking.apm.agent.core.remote.GRPCChannelManager
|
||||
org.apache.skywalking.apm.agent.core.jvm.JVMService
|
||||
|
|
|
|||
|
|
@ -22,21 +22,19 @@ package org.apache.skywalking.apm.agent.core.boot;
|
|||
import java.lang.reflect.Field;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.skywalking.apm.agent.core.context.TracingContext;
|
||||
import org.apache.skywalking.apm.agent.core.test.tools.AgentServiceRule;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.apache.skywalking.apm.agent.core.context.ContextManager;
|
||||
import org.apache.skywalking.apm.agent.core.context.IgnoredTracerContext;
|
||||
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.jvm.JVMService;
|
||||
import org.apache.skywalking.apm.agent.core.remote.CollectorDiscoveryService;
|
||||
import org.apache.skywalking.apm.agent.core.remote.GRPCChannelListener;
|
||||
import org.apache.skywalking.apm.agent.core.remote.GRPCChannelManager;
|
||||
import org.apache.skywalking.apm.agent.core.remote.TraceSegmentServiceClient;
|
||||
import org.apache.skywalking.apm.agent.core.sampling.SamplingService;
|
||||
import org.apache.skywalking.apm.agent.core.test.tools.AgentServiceRule;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.hamcrest.MatcherAssert.assertThat;
|
||||
|
|
@ -57,11 +55,10 @@ public class ServiceManagerTest {
|
|||
public void testServiceDependencies() throws Exception {
|
||||
HashMap<Class, BootService> registryService = getFieldValue(ServiceManager.INSTANCE, "bootedServices");
|
||||
|
||||
assertThat(registryService.size(), is(8));
|
||||
assertThat(registryService.size(), is(7));
|
||||
|
||||
assertTraceSegmentServiceClient(ServiceManager.INSTANCE.findService(TraceSegmentServiceClient.class));
|
||||
assertContextManager(ServiceManager.INSTANCE.findService(ContextManager.class));
|
||||
assertCollectorDiscoveryService(ServiceManager.INSTANCE.findService(CollectorDiscoveryService.class));
|
||||
assertGRPCChannelManager(ServiceManager.INSTANCE.findService(GRPCChannelManager.class));
|
||||
assertSamplingService(ServiceManager.INSTANCE.findService(SamplingService.class));
|
||||
assertJVMService(ServiceManager.INSTANCE.findService(JVMService.class));
|
||||
|
|
@ -100,10 +97,6 @@ public class ServiceManagerTest {
|
|||
assertNotNull(service);
|
||||
}
|
||||
|
||||
private void assertCollectorDiscoveryService(CollectorDiscoveryService service) {
|
||||
assertNotNull(service);
|
||||
}
|
||||
|
||||
private void assertContextManager(ContextManager service) {
|
||||
assertNotNull(service);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -32,11 +32,11 @@ public class SnifferConfigInitializerTest {
|
|||
@Test
|
||||
public void testLoadConfigFromJavaAgentDir() throws AgentPackageNotFoundException, ConfigNotFoundException {
|
||||
System.setProperty("skywalking.agent.application_code", "testApp");
|
||||
System.setProperty("skywalking.collector.servers", "127.0.0.1:8090");
|
||||
System.setProperty("skywalking.collector.backend_service", "127.0.0.1:8090");
|
||||
System.setProperty("skywalking.logging.level", "info");
|
||||
SnifferConfigInitializer.initialize();
|
||||
assertThat(Config.Agent.APPLICATION_CODE, is("testApp"));
|
||||
assertThat(Config.Collector.SERVERS, is("127.0.0.1:8090"));
|
||||
assertThat(Config.Collector.BACKEND_SERVICE, is("127.0.0.1:8090"));
|
||||
assertThat(Config.Logging.LEVEL, is(LogLevel.INFO));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,133 +0,0 @@
|
|||
/*
|
||||
* 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.remote;
|
||||
|
||||
import com.github.tomakehurst.wiremock.junit.WireMockRule;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
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.test.tools.AgentServiceRule;
|
||||
import org.hamcrest.MatcherAssert;
|
||||
import org.junit.*;
|
||||
import org.apache.skywalking.apm.agent.core.conf.Config;
|
||||
|
||||
import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
|
||||
import static com.github.tomakehurst.wiremock.client.WireMock.get;
|
||||
import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
|
||||
import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.hamcrest.MatcherAssert.assertThat;
|
||||
|
||||
public class DiscoveryRestServiceClientTest {
|
||||
|
||||
@Rule
|
||||
public AgentServiceRule agentServiceRule = new AgentServiceRule();
|
||||
|
||||
private DiscoveryRestServiceClient client;
|
||||
@Rule
|
||||
public WireMockRule wireMockRule = new WireMockRule(8089);
|
||||
|
||||
@AfterClass
|
||||
public static void afterClass() {
|
||||
ServiceManager.INSTANCE.shutdown();
|
||||
}
|
||||
|
||||
@Before
|
||||
public void setUpBeforeClass() {
|
||||
Config.Collector.DISCOVERY_CHECK_INTERVAL = 1;
|
||||
stubFor(get(urlEqualTo("/withoutResult"))
|
||||
.willReturn(aResponse()
|
||||
.withStatus(200)
|
||||
.withHeader("Content-Type", "application/json")
|
||||
.withBody("[]")));
|
||||
stubFor(get(urlEqualTo("/withResult"))
|
||||
.willReturn(aResponse()
|
||||
.withStatus(200)
|
||||
.withHeader("Content-Type", "application/json")
|
||||
.withBody("['127.0.0.1:8080','127.0.0.1:8090']")));
|
||||
stubFor(get(urlEqualTo("/withSameResult"))
|
||||
.willReturn(aResponse()
|
||||
.withStatus(200)
|
||||
.withHeader("Content-Type", "application/json")
|
||||
.withBody("['127.0.0.1:8090','127.0.0.1:8080']")));
|
||||
stubFor(get(urlEqualTo("/withDifferenceResult"))
|
||||
.willReturn(aResponse()
|
||||
.withStatus(200)
|
||||
.withHeader("Content-Type", "application/json")
|
||||
.withBody("['127.0.0.1:9090','127.0.0.1:18090']")));
|
||||
stubFor(get(urlEqualTo("/with404"))
|
||||
.willReturn(aResponse()
|
||||
.withStatus(400)));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithoutCollectorServer() throws RESTResponseStatusError, IOException {
|
||||
client = new DiscoveryRestServiceClient();
|
||||
client.run();
|
||||
MatcherAssert.assertThat(RemoteDownstreamConfig.Collector.GRPC_SERVERS.size(), is(0));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithGRPCAddress() throws RESTResponseStatusError, IOException {
|
||||
Config.Collector.SERVERS = "127.0.0.1:8089";
|
||||
Config.Collector.DISCOVERY_SERVICE_NAME = "/withResult";
|
||||
client = new DiscoveryRestServiceClient();
|
||||
client.run();
|
||||
|
||||
assertThat(RemoteDownstreamConfig.Collector.GRPC_SERVERS.size(), is(2));
|
||||
assertThat(RemoteDownstreamConfig.Collector.GRPC_SERVERS.contains("127.0.0.1:8080"), is(true));
|
||||
assertThat(RemoteDownstreamConfig.Collector.GRPC_SERVERS.contains("127.0.0.1:8090"), is(true));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithoutGRPCAddress() throws RESTResponseStatusError, IOException {
|
||||
Config.Collector.SERVERS = "127.0.0.1:8089";
|
||||
Config.Collector.DISCOVERY_SERVICE_NAME = "/withoutResult";
|
||||
client = new DiscoveryRestServiceClient();
|
||||
client.run();
|
||||
|
||||
assertThat(RemoteDownstreamConfig.Collector.GRPC_SERVERS.size(), is(0));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testChangeGrpcAddress() throws RESTResponseStatusError, IOException {
|
||||
Config.Collector.SERVERS = "127.0.0.1:8089";
|
||||
Config.Collector.DISCOVERY_SERVICE_NAME = "/withResult";
|
||||
client = new DiscoveryRestServiceClient();
|
||||
client.run();
|
||||
|
||||
Config.Collector.DISCOVERY_SERVICE_NAME = "/withDifferenceResult";
|
||||
client.run();
|
||||
|
||||
assertThat(RemoteDownstreamConfig.Collector.GRPC_SERVERS.size(), is(2));
|
||||
assertThat(RemoteDownstreamConfig.Collector.GRPC_SERVERS.contains("127.0.0.1:9090"), is(true));
|
||||
assertThat(RemoteDownstreamConfig.Collector.GRPC_SERVERS.contains("127.0.0.1:18090"), is(true));
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
Config.Collector.SERVERS = "";
|
||||
Config.Collector.DISCOVERY_SERVICE_NAME = "/grpc/address";
|
||||
RemoteDownstreamConfig.Collector.GRPC_SERVERS.clear();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -1,121 +0,0 @@
|
|||
/*
|
||||
* 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.remote;
|
||||
|
||||
import io.grpc.NameResolver;
|
||||
import io.grpc.Status;
|
||||
import io.grpc.StatusRuntimeException;
|
||||
import io.grpc.netty.NettyChannelBuilder;
|
||||
import io.grpc.testing.GrpcServerRule;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import org.apache.skywalking.apm.agent.core.conf.RemoteDownstreamConfig;
|
||||
import org.junit.Before;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.Spy;
|
||||
import org.powermock.core.classloader.annotations.PrepareForTest;
|
||||
import org.powermock.modules.junit4.PowerMockRunner;
|
||||
import org.apache.skywalking.apm.agent.core.conf.Config;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.hamcrest.MatcherAssert.assertThat;
|
||||
import static org.mockito.Matchers.any;
|
||||
import static org.mockito.Matchers.anyInt;
|
||||
import static org.mockito.Matchers.anyString;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.powermock.api.mockito.PowerMockito.mockStatic;
|
||||
import static org.powermock.api.mockito.PowerMockito.when;
|
||||
|
||||
@RunWith(PowerMockRunner.class)
|
||||
@PrepareForTest({GRPCChannelManager.class, NettyChannelBuilder.class})
|
||||
public class GRPCChannelManagerTest {
|
||||
|
||||
@Rule
|
||||
private GrpcServerRule grpcServerRule = new GrpcServerRule().directExecutor();
|
||||
|
||||
@Spy
|
||||
private GRPCChannelManager grpcChannelManager = new GRPCChannelManager();
|
||||
|
||||
@Mock
|
||||
private NettyChannelBuilder mock;
|
||||
|
||||
@Spy
|
||||
private MockGRPCChannelListener listener = new MockGRPCChannelListener();
|
||||
|
||||
@Before
|
||||
public void setUp() throws Throwable {
|
||||
List<String> grpcServers = new ArrayList<String>();
|
||||
grpcServers.add("127.0.0.1:2181");
|
||||
RemoteDownstreamConfig.Collector.GRPC_SERVERS = grpcServers;
|
||||
Config.Collector.GRPC_CHANNEL_CHECK_INTERVAL = 1;
|
||||
|
||||
mockStatic(NettyChannelBuilder.class);
|
||||
when(NettyChannelBuilder.forAddress(anyString(), anyInt())).thenReturn(mock);
|
||||
when(mock.nameResolverFactory(any(NameResolver.Factory.class))).thenReturn(mock);
|
||||
when(mock.maxInboundMessageSize(anyInt())).thenReturn(mock);
|
||||
when(mock.usePlaintext(true)).thenReturn(mock);
|
||||
when(mock.build()).thenReturn(grpcServerRule.getChannel());
|
||||
|
||||
grpcChannelManager.addChannelListener(listener);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void changeStatusToConnectedWithReportError() throws Throwable {
|
||||
grpcChannelManager.reportError(new StatusRuntimeException(Status.ABORTED));
|
||||
grpcChannelManager.run();
|
||||
|
||||
verify(listener, times(1)).statusChanged(GRPCChannelStatus.CONNECTED);
|
||||
assertThat(listener.status, is(GRPCChannelStatus.CONNECTED));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void changeStatusToDisConnectedWithReportError() throws Throwable {
|
||||
doThrow(new RuntimeException()).when(mock).nameResolverFactory(any(NameResolver.Factory.class));
|
||||
grpcChannelManager.run();
|
||||
|
||||
verify(listener, times(1)).statusChanged(GRPCChannelStatus.DISCONNECT);
|
||||
assertThat(listener.status, is(GRPCChannelStatus.DISCONNECT));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void reportErrorWithoutChangeStatus() throws Throwable {
|
||||
grpcChannelManager.run();
|
||||
grpcChannelManager.reportError(new RuntimeException());
|
||||
grpcChannelManager.run();
|
||||
|
||||
verify(listener, times(1)).statusChanged(GRPCChannelStatus.CONNECTED);
|
||||
assertThat(listener.status, is(GRPCChannelStatus.CONNECTED));
|
||||
}
|
||||
|
||||
private class MockGRPCChannelListener implements GRPCChannelListener {
|
||||
private GRPCChannelStatus status;
|
||||
|
||||
@Override
|
||||
public void statusChanged(GRPCChannelStatus status) {
|
||||
this.status = status;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -15,5 +15,5 @@
|
|||
# limitations under the License.
|
||||
|
||||
agent.application_code = crmApp
|
||||
collector.servers = 127.0.0.1:8080
|
||||
collector.backend_service = 127.0.0.1:8080
|
||||
logging.level=info
|
||||
|
|
|
|||
|
|
@ -38,22 +38,8 @@ agent.application_code=Your_ApplicationName
|
|||
# Skywalking team may ask for these files in order to resolve compatible problem.
|
||||
# agent.is_open_debugging_class = true
|
||||
|
||||
# Server addresses.
|
||||
# Primary address setting.
|
||||
#
|
||||
# Mapping to `naming/jetty/ip:port` in `config/application.yml` of Collector.
|
||||
# Examples：
|
||||
# Single collector：SERVERS="127.0.0.1:8080"
|
||||
# Collector cluster：SERVERS="10.2.45.126:8080,10.2.45.127:7600"
|
||||
collector.servers=127.0.0.1:10800
|
||||
|
||||
# Collector agent_gRPC/grpc service addresses.
|
||||
# Secondary address setting, only effect when "collector.servers" is empty.
|
||||
# By using this, no discovery mechanism provided. The agent only uses these addresses to uplink data.
|
||||
# Recommend to use this only when collector cluster IPs are unreachable from agent side. Such as:
|
||||
# 1. Agent and collector cluster are in different VPC in Cloud.
|
||||
# 2. Agent uplinks data to collector cluster through Internet.
|
||||
# collector.direct_servers=www.skywalking.service.io
|
||||
# Backend service addresses.
|
||||
collector.backend_service=127.0.0.1:10800
|
||||
|
||||
# Logging level
|
||||
logging.level=DEBUG
|
||||
|
|
|
|||
|
|
@ -5,4 +5,12 @@ Also learn to build the project, even to release the official Apache version(If
|
|||
- [Compiling Guide](../How-to-build.md). Teaches developer how to build the project in local.
|
||||
- [Apache Release Guide](How-to-release.md). Apache license allows everyone to redistribute if you keep our licenses and NOTICE
|
||||
in your redistribution. This document introduces to the committer team about doing official Apache version release, to avoid
|
||||
breaking any Apache rule.
|
||||
breaking any Apache rule.
|
||||
|
||||
## Project Extensions
|
||||
SkyWalking project supports many ways to extends existing features. If you are interesting in these ways,
|
||||
read the following guides.
|
||||
|
||||
- [Java agent plugin development guide](../setup/service-agent/java-agent/Plugin-Development-Guide.md).
|
||||
This guide helps you to develop SkyWalking agent plugin to support more frameworks. Both open source plugin
|
||||
and private plugin developer should read this.
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
# JVM Metrics Service
|
||||
## Abstract
|
||||
Uplink the JVM metrics, including PermSize, HeapSize, CPU, Memory, etc., every second.
|
||||
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v2.0/JVMMetricsService.proto)
|
||||
|
|
@ -5,15 +5,25 @@ There are two types of protocols list here.
|
|||
|
||||
- [**Query Protocol**](#query-protocol). The backend provide query capability to SkyWalking own UI and others. These queries are based on GraphQL.
|
||||
|
||||
|
||||
## Probe Protocols
|
||||
They also related to the probe group, for understand that, look [Concepts and Designs](../concepts-and-designs/README.md) document.
|
||||
These groups are **Language based native agent protocol**, **Service Mesh protocol** and **3rd-party instrument protocol**.
|
||||
These groups are **Language based native agent protocol**, **Service Mesh protocol** and **3rd-party instrument protocol**.
|
||||
|
||||
## Register Protocol
|
||||
Include service, service instance, network address and endpoint meta data register.
|
||||
Purposes of register are
|
||||
1. For service, network address and endpoint, register returns the unique ID of register object, usually an integer. Probe
|
||||
can use that to represent the literal String for data compression. Further, some protocols accept IDs only.
|
||||
1. For service instance, register returns a new unique ID for every new instance. Every service instance register must contain the
|
||||
service ID.
|
||||
|
||||
|
||||
|
||||
### Language based native agent protocol
|
||||
This protocol is combined from two parts:
|
||||
* [Cross Process Propagation Headers Protocol](Skywalking-Cross-Process-Propagation-Headers-Protocol-v1.md) is for in-wire propagation.
|
||||
which is usually used to carrier the necessary info to build trace.
|
||||
By following this protocol, the trace segments in different processes could be linked.
|
||||
* [SkyWalking Trace Data Protocol](Trace-Data-Protocol.md) define the communication way and format between agent and backend.
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -5,78 +5,17 @@ Trace Data Protocol describes the data format between SkyWalking agent/sniffer a
|
|||
This protocol includes the downstream and upstream data format. Other languages agents/SDKs can use this protocol to
|
||||
uplink data to the SkyWalking backend.
|
||||
|
||||
- Discovery service provided by HTTP only
|
||||
- Other services, includes Register, Trace, etc., provided by HTTP/JSON and gRPC both.
|
||||
|
||||
### Version
|
||||
v1.1
|
||||
v2.0
|
||||
|
||||
#### gRPC proto files
|
||||
[gRPC proto files](https://github.com/apache/incubator-skywalking-data-collect-protocol/tree/v1.1.1)
|
||||
[gRPC proto files](https://github.com/apache/incubator-skywalking-data-collect-protocol/tree/v2.0)
|
||||
|
||||
## Collector discovery service
|
||||
### Abstract
|
||||
**Collector discovery service should be the first service after agent started**
|
||||
Through this service, get the gRPC service list. The agent can choose any one of them for uplink data. Recommend to
|
||||
acquire the list periodically.
|
||||
|
||||
### HTTP GET
|
||||
- Input
|
||||
GET：http://collectorIp:port/agent/gRPC
|
||||
|
||||
- Output
|
||||
JSON Array. Each element in the array ia a valid gRPC service address.
|
||||
```json
|
||||
["ip address1:port1","ip address2:port2","ip address3:port3"]
|
||||
```
|
||||
|
||||
## Application Register Service
|
||||
### Abstract
|
||||
Register Application Code to the backend, and receive an integer represents the application.
|
||||
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v1.1.1/ApplicationRegisterService.proto)
|
||||
|
||||
- applicationCode is the config in your `agent.config`.
|
||||
- The return id is **ApplicationId** as the value in `KeyWithIntegerValue`, which will be used in further data uplink.
|
||||
|
||||
## Discovery Service
|
||||
|
||||
### Register Instance Service
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v1.1.1/DiscoveryService.proto#L29)
|
||||
|
||||
- agentUUID generated by agent, should be unique. Stay same before reboot, at least.
|
||||
- **ApplicationInstanceId** will be used in further data uplink.
|
||||
|
||||
### Heart beat service
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v1.1.1/DiscoveryService.proto#L32)
|
||||
|
||||
- Recommend to report heart beat every 20-60 seconds.
|
||||
- Java agent don't use this, because JVM metrics upstream replace the capabilities of this.
|
||||
|
||||
## Service Name Discovery Service
|
||||
### Abstract
|
||||
Replace the literal String service(operation) name by an id(integer)
|
||||
|
||||
[gRPC service define](.https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v1.1.1/DiscoveryService.proto#L70)
|
||||
|
||||
- Optional service, reduce the network cost but use more memory as a buffer mapping.
|
||||
|
||||
## Network Address Register Service
|
||||
### Abstract
|
||||
Network Address includes all remove service address, includes ip, port, hostname, etc., which used in RPC framework, MQ, DB, etc.
|
||||
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v1.1.1/NetworkAddressRegisterService.proto)
|
||||
|
||||
- Optional service, reduce the network cost but use more memory as a buffer mapping.
|
||||
|
||||
## JVM Metrics Service
|
||||
### Abstract
|
||||
Uplink the JVM metrics, including PermSize, HeapSize, CPU, Memory, etc., every second.
|
||||
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v1.1.1/JVMMetricsService.proto)
|
||||
|
||||
## Trace Segment Service
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v1.1.1/TraceSegmentService.proto)
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v2.0/TraceSegmentService.proto)
|
||||
|
||||
- UniqueId represents segmentId and globalTraceId. It have 3 parts(Longs), 1) applicationInstanceId, 2) ThreadId, 3) Timestamp + 10000 + seq(seq is in [0, 100000) )
|
||||
- Span data please refs to [Plugin Development Guide](../setup/service-agent/java-agent/Plugin-Development-Guide.md)
|
||||
|
|
@ -88,68 +27,8 @@ Uplink the JVM metrics, including PermSize, HeapSize, CPU, Memory, etc., every s
|
|||
- peerId/peer
|
||||
- componentIds are defined in backend, [here](../../../apm-protocol/apm-network/src/main/java/org/apache/skywalking/apm/network/trace/component/ComponentsDefine.java)
|
||||
|
||||
# HTTP JSON Services
|
||||
All HTTP Services match the gRPC services, just adjust use short keys.
|
||||
HTTP format:
|
||||
|
||||
## Instance Register Service
|
||||
- http://ip:port/instance/register(default: localhost:12800)
|
||||
|
||||
Input:
|
||||
```
|
||||
{
|
||||
ai: x, #applicationId
|
||||
au: "", #agentUUID
|
||||
rt: x, #registerTime
|
||||
oi: "", #osinfo
|
||||
}
|
||||
```
|
||||
|
||||
Output:
|
||||
```
|
||||
{
|
||||
ai: x, #applicationId
|
||||
ii: x, #applicationInstanceId
|
||||
}
|
||||
```
|
||||
|
||||
## Heart beat service
|
||||
- http://ip:port/instance/heartbeat(default: localhost:12800)
|
||||
|
||||
Input:
|
||||
```
|
||||
{
|
||||
"ii": x, #applicationInstanceId
|
||||
"ht": x #heartbeatTime, java timestamp format
|
||||
}
|
||||
```
|
||||
|
||||
Output: empty
|
||||
|
||||
## Service Name Discovery
|
||||
- http://ip:port/servicename/discovery(default: localhost:12800)
|
||||
|
||||
Input：
|
||||
```
|
||||
{
|
||||
ai: x, #applicationId
|
||||
sn: "", #serviceName
|
||||
st: x, #srcSpanType
|
||||
}
|
||||
```
|
||||
|
||||
Output：
|
||||
```
|
||||
{
|
||||
si: x, #osinfo
|
||||
el: { #element
|
||||
ai: x, #applicationId
|
||||
sn: "", #serviceName
|
||||
st: x, #srcSpanType
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Trace Segment Service
|
||||
Input：
|
||||
```
|
||||
[
|
||||
|
|
@ -229,3 +108,106 @@ Input：
|
|||
}
|
||||
]
|
||||
```
|
||||
|
||||
|
||||
## Deprecated services
|
||||
**Deprecated service**(s) are the gRPC services SkyWalking used before. In SkyWalking v6, in order to match the common
|
||||
concepts in CloudNative world, these services are deprecated.
|
||||
Although there services are still supported at this moment, but it will be removed after 6 months later(Feb. 2019).
|
||||
|
||||
## ~~Application Register Service~~
|
||||
**Deprecated service**
|
||||
|
||||
### Abstract
|
||||
Register Application Code to the backend, and receive an integer represents the application.
|
||||
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v2.0/ApplicationRegisterService.proto)
|
||||
|
||||
- applicationCode is the config in your `agent.config`.
|
||||
- The return id is **ApplicationId** as the value in `KeyWithIntegerValue`, which will be used in further data uplink.
|
||||
|
||||
## ~~Discovery Services~~
|
||||
**Deprecated services**
|
||||
|
||||
### ~~Register Instance Service~~
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v2.0/DiscoveryService.proto#L29)
|
||||
|
||||
- agentUUID generated by agent, should be unique. Stay same before reboot, at least.
|
||||
- **ApplicationInstanceId** will be used in further data uplink.
|
||||
|
||||
HTTP format http://ip:port/instance/register(default: localhost:12800)
|
||||
|
||||
Input:
|
||||
```
|
||||
{
|
||||
ai: x, #applicationId
|
||||
au: "", #agentUUID
|
||||
rt: x, #registerTime
|
||||
oi: "", #osinfo
|
||||
}
|
||||
```
|
||||
|
||||
Output:
|
||||
```
|
||||
{
|
||||
ai: x, #applicationId
|
||||
ii: x, #applicationInstanceId
|
||||
}
|
||||
```
|
||||
|
||||
### ~~Heart beat service~~
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v2.0/DiscoveryService.proto#L32)
|
||||
|
||||
- Recommend to report heart beat every 20-60 seconds.
|
||||
- Java agent don't use this, because JVM metrics upstream replace the capabilities of this.
|
||||
|
||||
HTTP format http://ip:port/instance/heartbeat(default: localhost:12800)
|
||||
|
||||
Input:
|
||||
```
|
||||
{
|
||||
"ii": x, #applicationInstanceId
|
||||
"ht": x #heartbeatTime, java timestamp format
|
||||
}
|
||||
```
|
||||
|
||||
## ~~Service Name Discovery Service~~
|
||||
**Deprecated services**
|
||||
### Abstract
|
||||
Replace the literal String service(operation) name by an id(integer)
|
||||
|
||||
[gRPC service define](.https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v2.0/DiscoveryService.proto#L70)
|
||||
|
||||
- Optional service, reduce the network cost but use more memory as a buffer mapping.
|
||||
|
||||
HTTP format http://ip:port/servicename/discovery(default: localhost:12800)
|
||||
|
||||
Input：
|
||||
```
|
||||
{
|
||||
ai: x, #applicationId
|
||||
sn: "", #serviceName
|
||||
st: x, #srcSpanType
|
||||
}
|
||||
```
|
||||
|
||||
Output：
|
||||
```
|
||||
{
|
||||
si: x, #osinfo
|
||||
el: { #element
|
||||
ai: x, #applicationId
|
||||
sn: "", #serviceName
|
||||
st: x, #srcSpanType
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## ~~Network Address Register Service~~
|
||||
### Abstract
|
||||
Network Address includes all remove service address, includes ip, port, hostname, etc., which used in RPC framework, MQ, DB, etc.
|
||||
|
||||
[gRPC service define](https://github.com/apache/incubator-skywalking-data-collect-protocol/blob/v2.0/NetworkAddressRegisterService.proto)
|
||||
|
||||
- Optional service, reduce the network cost but use more memory as a buffer mapping.
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,61 @@
|
|||
# Component library settings
|
||||
Component library settings are about your own or 3rd part libraries used in monitored application.
|
||||
|
||||
In agent or SDK, no matter library name collected as ID or String(literally, e.g. SpringMVC), collector
|
||||
formats data in ID for better performance and less storage requirements.
|
||||
|
||||
Also, collector conjectures the remote service based on the component library, such as:
|
||||
the component library is MySQL Driver library, then the remote service should be MySQL Server.
|
||||
|
||||
For those two reasons, collector require two parts of settings in this file:
|
||||
1. Component Library id, name and languages.
|
||||
1. Remote server mapping, based on local library.
|
||||
|
||||
**All component names and IDs must be defined in this file.**
|
||||
|
||||
## Component Library id
|
||||
Define all component libraries' names and IDs, used in monitored application.
|
||||
This is a both-way mapping, agent or SDK could use the value(ID) to represent the component name in uplink data.
|
||||
|
||||
- Name: the component name used in agent and UI
|
||||
- id: Unique ID. All IDs are reserved, once it is released.
|
||||
- languages: Program languages may use this component. Multi languages should be separated by `,`
|
||||
|
||||
### ID rules
|
||||
- Java and multi languages shared: (0, 3000]
|
||||
- .NET Platform reserved: (3000, 4000]
|
||||
- Node.js Platform reserved: (4000, 5000]
|
||||
- Go reserved: (5000, 6000]
|
||||
- PHP reserved: (6000, 7000]
|
||||
|
||||
Example
|
||||
```yaml
|
||||
Tomcat:
|
||||
id: 1
|
||||
languages: Java
|
||||
HttpClient:
|
||||
id: 2
|
||||
languages: Java,C#,Node.js
|
||||
Dubbo:
|
||||
id: 3
|
||||
languages: Java
|
||||
H2:
|
||||
id: 4
|
||||
languages: Java
|
||||
```
|
||||
|
||||
## Remote server mapping
|
||||
Remote server will be conjectured by the local component. The mappings are based on Component library names.
|
||||
|
||||
- Key: client component library name
|
||||
- Value: server component name
|
||||
|
||||
```yaml
|
||||
Component-Server-Mappings:
|
||||
Jedis: Redis
|
||||
StackExchange.Redis: Redis
|
||||
SqlClient: SqlServer
|
||||
Npgsql: PostgreSQL
|
||||
MySqlConnector: Mysql
|
||||
EntityFrameworkCore.InMemory: InMemoryDatabase
|
||||
```
|
||||
|
|
@ -157,10 +157,8 @@ SpanLayer is the catalog of span. Here are 5 values:
|
|||
1. HTTP
|
||||
1. MQ
|
||||
|
||||
Component IDs are defined and protected by SkyWalking project, 0 -> 10000 IDs are reserved. If you want to contribute
|
||||
a new plugin, you can ask for an official ID, after your pull request approved and automatic test cases accepted by PMC.
|
||||
Please use > 10000 ID, if you are going to develop a private plugin or don't intend to contribute the plugin to community,
|
||||
to avoid the ID conflict.
|
||||
Component IDs are defined and reserved by SkyWalking project.
|
||||
For component name/ID extension, please follow [component library setting document](../../backend/Component-library-settings.md).
|
||||
|
||||
## Develop a plugin
|
||||
### Abstract
|
||||
|
|
|
|||
|
|
@ -67,6 +67,10 @@ Now, we have the following known plugins.
|
|||
* If you want to use annotations or SkyWalking native APIs to read context, try [SkyWalking manual APIs](Application-toolkit-trace.md)
|
||||
* If you want to continue traces across thread manually, use [across thread solution APIs](Application-toolkit-trace-cross-thread.md).
|
||||
|
||||
## Plugin Development Guide
|
||||
SkyWalking java agent supports plugin to extend [the supported list](Supported-list.md). Please follow
|
||||
our [Plugin Development Guide](Plugin-Development-Guide.md).
|
||||
|
||||
# Test
|
||||
If you are interested in plugin compatible tests or agent performance, see the following reports.
|
||||
* [Plugin Test](https://github.com/SkywalkingTest/agent-integration-test-report)
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ import java.io.IOException;
|
|||
import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.IndicatorTypeListener;
|
||||
import org.apache.skywalking.oap.server.core.annotation.AnnotationScan;
|
||||
import org.apache.skywalking.oap.server.core.cluster.*;
|
||||
import org.apache.skywalking.oap.server.core.register.annotation.InventoryTypeListener;
|
||||
import org.apache.skywalking.oap.server.core.remote.*;
|
||||
import org.apache.skywalking.oap.server.core.remote.annotation.*;
|
||||
import org.apache.skywalking.oap.server.core.remote.client.RemoteClientManager;
|
||||
|
|
@ -92,6 +93,7 @@ public class CoreModuleProvider extends ModuleProvider {
|
|||
annotationScan.registerListener(storageAnnotationListener);
|
||||
annotationScan.registerListener(streamAnnotationListener);
|
||||
annotationScan.registerListener(new IndicatorTypeListener(getManager()));
|
||||
annotationScan.registerListener(new InventoryTypeListener(getManager()));
|
||||
}
|
||||
|
||||
@Override public void start() throws ModuleStartException {
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@
|
|||
package org.apache.skywalking.oap.server.core.analysis;
|
||||
|
||||
import java.util.*;
|
||||
import org.apache.skywalking.oap.server.core.analysis.endpoint.EndpointDispatcher;
|
||||
import org.apache.skywalking.oap.server.core.analysis.generated.endpoint.EndpointDispatcher;
|
||||
import org.apache.skywalking.oap.server.core.source.Scope;
|
||||
import org.slf4j.*;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,125 +0,0 @@
|
|||
/*
|
||||
* 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.core.analysis.endpoint;
|
||||
|
||||
import java.util.*;
|
||||
import lombok.*;
|
||||
import org.apache.skywalking.oap.server.core.analysis.indicator.AvgIndicator;
|
||||
import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.IndicatorType;
|
||||
import org.apache.skywalking.oap.server.core.remote.annotation.StreamData;
|
||||
import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
|
||||
import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
|
||||
import org.apache.skywalking.oap.server.core.storage.annotation.*;
|
||||
|
||||
/**
|
||||
* @author peng-yongsheng
|
||||
*/
|
||||
@IndicatorType
|
||||
@StreamData
|
||||
@StorageEntity(name = "endpoint_latency_avg", builder = EndpointLatencyAvgIndicator.Builder.class)
|
||||
public class EndpointLatencyAvgIndicator extends AvgIndicator {
|
||||
|
||||
private static final String ID = "id";
|
||||
private static final String SERVICE_ID = "service_id";
|
||||
private static final String SERVICE_INSTANCE_ID = "service_instance_id";
|
||||
|
||||
@Setter @Getter @Column(columnName = ID) private int id;
|
||||
@Setter @Getter @Column(columnName = SERVICE_ID) private int serviceId;
|
||||
@Setter @Getter @Column(columnName = SERVICE_INSTANCE_ID) private int serviceInstanceId;
|
||||
|
||||
@Override public String id() {
|
||||
return String.valueOf(id);
|
||||
}
|
||||
|
||||
@Override public int hashCode() {
|
||||
int result = 17;
|
||||
result = 31 * result + id;
|
||||
result = 31 * result + (int)getTimeBucket();
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override public boolean equals(Object obj) {
|
||||
if (this == obj)
|
||||
return true;
|
||||
if (obj == null)
|
||||
return false;
|
||||
if (getClass() != obj.getClass())
|
||||
return false;
|
||||
|
||||
EndpointLatencyAvgIndicator indicator = (EndpointLatencyAvgIndicator)obj;
|
||||
if (id != indicator.id)
|
||||
return false;
|
||||
if (getTimeBucket() != indicator.getTimeBucket())
|
||||
return false;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override public RemoteData.Builder serialize() {
|
||||
RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
|
||||
remoteBuilder.setDataIntegers(0, getId());
|
||||
remoteBuilder.setDataIntegers(1, getServiceId());
|
||||
remoteBuilder.setDataIntegers(2, getServiceInstanceId());
|
||||
remoteBuilder.setDataIntegers(3, getCount());
|
||||
|
||||
remoteBuilder.setDataLongs(0, getTimeBucket());
|
||||
remoteBuilder.setDataLongs(1, getSummation());
|
||||
remoteBuilder.setDataLongs(2, getValue());
|
||||
|
||||
return remoteBuilder;
|
||||
}
|
||||
|
||||
@Override public void deserialize(RemoteData remoteData) {
|
||||
setId(remoteData.getDataIntegers(0));
|
||||
setServiceId(remoteData.getDataIntegers(1));
|
||||
setServiceInstanceId(remoteData.getDataIntegers(2));
|
||||
setCount(remoteData.getDataIntegers(3));
|
||||
|
||||
setTimeBucket(remoteData.getDataLongs(0));
|
||||
setSummation(remoteData.getDataLongs(1));
|
||||
setValue(remoteData.getDataLongs(2));
|
||||
}
|
||||
|
||||
public static class Builder implements StorageBuilder<EndpointLatencyAvgIndicator> {
|
||||
|
||||
@Override public EndpointLatencyAvgIndicator map2Data(Map<String, Object> dbMap) {
|
||||
EndpointLatencyAvgIndicator indicator = new EndpointLatencyAvgIndicator();
|
||||
indicator.setId((Integer)dbMap.get(ID));
|
||||
indicator.setServiceId((Integer)dbMap.get(SERVICE_ID));
|
||||
indicator.setServiceInstanceId((Integer)dbMap.get(SERVICE_INSTANCE_ID));
|
||||
indicator.setCount((Integer)dbMap.get(COUNT));
|
||||
indicator.setSummation((Long)dbMap.get(SUMMATION));
|
||||
indicator.setValue((Long)dbMap.get(VALUE));
|
||||
indicator.setTimeBucket((Long)dbMap.get(TIME_BUCKET));
|
||||
return indicator;
|
||||
}
|
||||
|
||||
@Override public Map<String, Object> data2Map(EndpointLatencyAvgIndicator storageData) {
|
||||
Map<String, Object> map = new HashMap<>();
|
||||
map.put(ID, storageData.getId());
|
||||
map.put(SERVICE_ID, storageData.getServiceId());
|
||||
map.put(SERVICE_INSTANCE_ID, storageData.getServiceInstanceId());
|
||||
map.put(COUNT, storageData.getCount());
|
||||
map.put(SUMMATION, storageData.getSummation());
|
||||
map.put(VALUE, storageData.getValue());
|
||||
map.put(TIME_BUCKET, storageData.getTimeBucket());
|
||||
return map;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -42,9 +42,9 @@ public class EndpointInventoryCacheService implements Service {
|
|||
this.moduleManager = moduleManager;
|
||||
}
|
||||
|
||||
private final Cache<String, Integer> idCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(1000000).build();
|
||||
private final Cache<String, Integer> idCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(100000).build();
|
||||
|
||||
private final Cache<Integer, EndpointInventory> sequenceCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(1000000).build();
|
||||
private final Cache<Integer, EndpointInventory> sequenceCache = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(100000).build();
|
||||
|
||||
public int get(int serviceId, String serviceName, int srcSpanType) {
|
||||
String id = serviceId + Const.ID_SPLIT + serviceName + Const.ID_SPLIT + srcSpanType;
|
||||
|
|
|
|||
|
|
@ -21,14 +21,17 @@ package org.apache.skywalking.oap.server.core.register;
|
|||
import java.util.*;
|
||||
import lombok.*;
|
||||
import org.apache.skywalking.oap.server.core.Const;
|
||||
import org.apache.skywalking.oap.server.core.register.annotation.InventoryType;
|
||||
import org.apache.skywalking.oap.server.core.remote.annotation.StreamData;
|
||||
import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
|
||||
import org.apache.skywalking.oap.server.core.source.Scope;
|
||||
import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
|
||||
import org.apache.skywalking.oap.server.core.storage.annotation.*;
|
||||
|
||||
/**
|
||||
* @author peng-yongsheng
|
||||
*/
|
||||
@InventoryType(scope = Scope.Endpoint)
|
||||
@StreamData
|
||||
@StorageEntity(name = "endpoint_inventory", builder = EndpointInventory.Builder.class)
|
||||
public class EndpointInventory extends RegisterSource {
|
||||
|
|
|
|||
|
|
@ -16,26 +16,22 @@
|
|||
*
|
||||
*/
|
||||
|
||||
package org.apache.skywalking.oap.server.core.analysis.endpoint;
|
||||
package org.apache.skywalking.oap.server.core.register.annotation;
|
||||
|
||||
import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
|
||||
import org.apache.skywalking.oap.server.core.analysis.worker.IndicatorProcess;
|
||||
import org.apache.skywalking.oap.server.core.source.Endpoint;
|
||||
import org.apache.skywalking.oap.server.core.UnexpectedException;
|
||||
import org.apache.skywalking.oap.server.core.source.Scope;
|
||||
|
||||
/**
|
||||
* @author peng-yongsheng
|
||||
*/
|
||||
public class EndpointDispatcher implements SourceDispatcher<Endpoint> {
|
||||
public class InventoryAnnotationUtils {
|
||||
|
||||
@Override public void dispatch(Endpoint source) {
|
||||
avg(source);
|
||||
}
|
||||
|
||||
private void avg(Endpoint source) {
|
||||
EndpointLatencyAvgIndicator indicator = new EndpointLatencyAvgIndicator();
|
||||
indicator.setId(source.getId());
|
||||
indicator.setTimeBucket(source.getTimeBucket());
|
||||
indicator.combine(source.getLatency(), 1);
|
||||
IndicatorProcess.INSTANCE.in(indicator);
|
||||
public static Scope getScope(Class aClass) {
|
||||
if (aClass.isAnnotationPresent(InventoryType.class)) {
|
||||
InventoryType annotation = (InventoryType)aClass.getAnnotation(InventoryType.class);
|
||||
return annotation.scope();
|
||||
} else {
|
||||
throw new UnexpectedException("");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -16,15 +16,16 @@
|
|||
*
|
||||
*/
|
||||
|
||||
package org.apache.skywalking.oap.server.receiver.istio.telemetry.provider;
|
||||
package org.apache.skywalking.oap.server.core.register.annotation;
|
||||
|
||||
import org.junit.Test;
|
||||
import java.lang.annotation.*;
|
||||
import org.apache.skywalking.oap.server.core.source.Scope;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
public class IstioTelemetryGRPCHandlerTest {
|
||||
|
||||
@Test
|
||||
public void handleMetric() {
|
||||
}
|
||||
}
|
||||
/**
|
||||
* @author peng-yongsheng
|
||||
*/
|
||||
@Target(ElementType.TYPE)
|
||||
@Retention(RetentionPolicy.RUNTIME)
|
||||
public @interface InventoryType {
|
||||
Scope scope();
|
||||
}
|
||||
|
|
@ -0,0 +1,44 @@
|
|||
/*
|
||||
* 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.core.register.annotation;
|
||||
|
||||
import java.lang.annotation.Annotation;
|
||||
import org.apache.skywalking.oap.server.core.annotation.AnnotationListener;
|
||||
import org.apache.skywalking.oap.server.core.register.worker.InventoryProcess;
|
||||
import org.apache.skywalking.oap.server.library.module.ModuleManager;
|
||||
|
||||
/**
|
||||
* @author peng-yongsheng
|
||||
*/
|
||||
public class InventoryTypeListener implements AnnotationListener {
|
||||
|
||||
private final ModuleManager moduleManager;
|
||||
|
||||
public InventoryTypeListener(ModuleManager moduleManager) {
|
||||
this.moduleManager = moduleManager;
|
||||
}
|
||||
|
||||
@Override public Class<? extends Annotation> annotation() {
|
||||
return InventoryType.class;
|
||||
}
|
||||
|
||||
@Override public void notify(Class aClass) {
|
||||
InventoryProcess.INSTANCE.create(moduleManager, aClass);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,68 @@
|
|||
/*
|
||||
* 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.core.register.worker;
|
||||
|
||||
import java.util.*;
|
||||
import org.apache.skywalking.oap.server.core.UnexpectedException;
|
||||
import org.apache.skywalking.oap.server.core.register.RegisterSource;
|
||||
import org.apache.skywalking.oap.server.core.register.annotation.InventoryAnnotationUtils;
|
||||
import org.apache.skywalking.oap.server.core.source.Scope;
|
||||
import org.apache.skywalking.oap.server.core.storage.*;
|
||||
import org.apache.skywalking.oap.server.core.storage.annotation.StorageEntityAnnotationUtils;
|
||||
import org.apache.skywalking.oap.server.core.worker.*;
|
||||
import org.apache.skywalking.oap.server.library.module.ModuleManager;
|
||||
|
||||
/**
|
||||
* @author peng-yongsheng
|
||||
*/
|
||||
public enum InventoryProcess {
|
||||
INSTANCE;
|
||||
|
||||
private Map<Class<? extends RegisterSource>, RegisterDistinctWorker> entryWorkers = new HashMap<>();
|
||||
|
||||
public void in(RegisterSource registerSource) {
|
||||
entryWorkers.get(registerSource.getClass()).in(registerSource);
|
||||
}
|
||||
|
||||
public void create(ModuleManager moduleManager, Class<? extends RegisterSource> inventoryClass) {
|
||||
String modelName = StorageEntityAnnotationUtils.getModelName(inventoryClass);
|
||||
Scope scope = InventoryAnnotationUtils.getScope(inventoryClass);
|
||||
|
||||
Class<? extends StorageBuilder> builderClass = StorageEntityAnnotationUtils.getBuilder(inventoryClass);
|
||||
|
||||
StorageDAO storageDAO = moduleManager.find(StorageModule.NAME).getService(StorageDAO.class);
|
||||
IRegisterDAO registerDAO;
|
||||
try {
|
||||
registerDAO = storageDAO.newRegisterDao(builderClass.newInstance());
|
||||
} catch (InstantiationException | IllegalAccessException e) {
|
||||
throw new UnexpectedException("");
|
||||
}
|
||||
|
||||
RegisterPersistentWorker persistentWorker = new RegisterPersistentWorker(WorkerIdGenerator.INSTANCES.generate(), modelName, moduleManager, registerDAO, scope);
|
||||
WorkerInstances.INSTANCES.put(persistentWorker.getWorkerId(), persistentWorker);
|
||||
|
||||
RegisterRemoteWorker remoteWorker = new RegisterRemoteWorker(WorkerIdGenerator.INSTANCES.generate(), moduleManager, persistentWorker);
|
||||
WorkerInstances.INSTANCES.put(remoteWorker.getWorkerId(), remoteWorker);
|
||||
|
||||
RegisterDistinctWorker distinctWorker = new RegisterDistinctWorker(WorkerIdGenerator.INSTANCES.generate(), remoteWorker);
|
||||
WorkerInstances.INSTANCES.put(distinctWorker.getWorkerId(), distinctWorker);
|
||||
|
||||
entryWorkers.put(inventoryClass, distinctWorker);
|
||||
}
|
||||
}
|
||||
|
|
@ -19,6 +19,7 @@
|
|||
package org.apache.skywalking.oap.server.core.storage;
|
||||
|
||||
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
|
||||
import org.apache.skywalking.oap.server.core.register.RegisterSource;
|
||||
import org.apache.skywalking.oap.server.library.module.Service;
|
||||
|
||||
/**
|
||||
|
|
@ -27,4 +28,6 @@ import org.apache.skywalking.oap.server.library.module.Service;
|
|||
public interface StorageDAO extends Service {
|
||||
|
||||
IIndicatorDAO newIndicatorDao(StorageBuilder<Indicator> storageBuilder);
|
||||
|
||||
IRegisterDAO newRegisterDao(StorageBuilder<RegisterSource> storageBuilder);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,33 +0,0 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!--
|
||||
~ 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.
|
||||
~
|
||||
-->
|
||||
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<parent>
|
||||
<artifactId>skywalking-istio-telemetry-receiver-plugin</artifactId>
|
||||
<groupId>org.apache.skywalking</groupId>
|
||||
<version>6.0.0-alpha-SNAPSHOT</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>istio-telemetry-receiver-module</artifactId>
|
||||
|
||||
|
||||
</project>
|
||||
|
|
@ -1,81 +0,0 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!--
|
||||
~ 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.
|
||||
~
|
||||
-->
|
||||
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<parent>
|
||||
<artifactId>skywalking-istio-telemetry-receiver-plugin</artifactId>
|
||||
<groupId>org.apache.skywalking</groupId>
|
||||
<version>6.0.0-alpha-SNAPSHOT</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>istio-telemetry-receiver-provider</artifactId>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.skywalking</groupId>
|
||||
<artifactId>istio-telemetry-receiver-module</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.skywalking</groupId>
|
||||
<artifactId>server-core</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-resources-plugin</artifactId>
|
||||
<version>2.4.3</version>
|
||||
<configuration>
|
||||
<encoding>${project.build.sourceEncoding}</encoding>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.xolstice.maven.plugins</groupId>
|
||||
<artifactId>protobuf-maven-plugin</artifactId>
|
||||
<version>0.5.0</version>
|
||||
<configuration>
|
||||
<!--
|
||||
The version of protoc must match protobuf-java. If you don't depend on
|
||||
protobuf-java directly, you will be transitively depending on the
|
||||
protobuf-java version that grpc depends on.
|
||||
-->
|
||||
<protocArtifact>com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier}
|
||||
</protocArtifact>
|
||||
<pluginId>grpc-java</pluginId>
|
||||
<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.8.0:exe:${os.detected.classifier}
|
||||
</pluginArtifact>
|
||||
</configuration>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>compile</goal>
|
||||
<goal>compile-custom</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
|
|
@ -28,11 +28,51 @@
|
|||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>skywalking-istio-telemetry-receiver-plugin</artifactId>
|
||||
<packaging>pom</packaging>
|
||||
<modules>
|
||||
<module>istio-telemetry-receiver-module</module>
|
||||
<module>istio-telemetry-receiver-provider</module>
|
||||
</modules>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.skywalking</groupId>
|
||||
<artifactId>server-core</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-resources-plugin</artifactId>
|
||||
<version>2.4.3</version>
|
||||
<configuration>
|
||||
<encoding>${project.build.sourceEncoding}</encoding>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.xolstice.maven.plugins</groupId>
|
||||
<artifactId>protobuf-maven-plugin</artifactId>
|
||||
<version>0.5.0</version>
|
||||
<configuration>
|
||||
<!--
|
||||
The version of protoc must match protobuf-java. If you don't depend on
|
||||
protobuf-java directly, you will be transitively depending on the
|
||||
protobuf-java version that grpc depends on.
|
||||
-->
|
||||
<protocArtifact>com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier}
|
||||
</protocArtifact>
|
||||
<pluginId>grpc-java</pluginId>
|
||||
<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.8.0:exe:${os.detected.classifier}
|
||||
</pluginArtifact>
|
||||
</configuration>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>compile</goal>
|
||||
<goal>compile-custom</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
|
|
@ -27,7 +27,7 @@ import io.istio.api.policy.v1beta1.TypeProto;
|
|||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.Map;
|
||||
import org.apache.skywalking.apm.network.servicemesh.DetectPoint;
|
||||
import org.apache.skywalking.apm.network.common.DetectPoint;
|
||||
import org.apache.skywalking.apm.network.servicemesh.Protocol;
|
||||
import org.apache.skywalking.apm.network.servicemesh.ServiceMeshMetric;
|
||||
import org.slf4j.Logger;
|
||||
|
|
@ -44,11 +44,13 @@ public class IstioTelemetryGRPCHandler extends HandleMetricServiceGrpc.HandleMet
|
|||
|
||||
@Override public void handleMetric(IstioMetricProto.HandleMetricRequest request,
|
||||
StreamObserver<ReportProto.ReportResult> responseObserver) {
|
||||
for (IstioMetricProto.InstanceMsg i : request.getInstancesList()) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Received msg {}", request);
|
||||
}
|
||||
for (IstioMetricProto.InstanceMsg i : request.getInstancesList()) {
|
||||
String requestMethod = string(i, "requestMethod");
|
||||
String requestPath = string(i,"requestPath");
|
||||
String requestScheme = string(i,"requestScheme");
|
||||
String requestPath = string(i, "requestPath");
|
||||
String requestScheme = string(i, "requestScheme");
|
||||
long responseCode = int64(i, "responseCode");
|
||||
String reporter = string(i, "reporter");
|
||||
String protocol = string(i, "apiProtocol");
|
||||
|
|
@ -16,23 +16,4 @@
|
|||
#
|
||||
#
|
||||
|
||||
#
|
||||
# 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.
|
||||
#
|
||||
#
|
||||
|
||||
|
||||
org.apache.skywalking.oap.server.receiver.istio.telemetry.provider.IstioTelemetryReceiverProvider
|
||||
|
|
@ -66,7 +66,7 @@
|
|||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.skywalking</groupId>
|
||||
<artifactId>istio-telemetry-receiver-provider</artifactId>
|
||||
<artifactId>skywalking-istio-telemetry-receiver-plugin</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<!-- receiver module -->
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@
|
|||
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
|
||||
|
||||
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
|
||||
import org.apache.skywalking.oap.server.core.register.RegisterSource;
|
||||
import org.apache.skywalking.oap.server.core.storage.*;
|
||||
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
|
||||
|
||||
|
|
@ -34,4 +35,8 @@ public class StorageEsDAO extends EsDAO implements StorageDAO {
|
|||
@Override public IIndicatorDAO newIndicatorDao(StorageBuilder<Indicator> storageBuilder) {
|
||||
return new IndicatorEsDAO(getClient(), storageBuilder);
|
||||
}
|
||||
|
||||
@Override public IRegisterDAO newRegisterDao(StorageBuilder<RegisterSource> storageBuilder) {
|
||||
return new RegisterEsDAO(getClient(), storageBuilder);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue