kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #806 from peng-yongsheng/feature/servertype 

Update span layer and server type by address id.
This commit is contained in:
吴晟 Wu Sheng 2018-02-11 17:49:09 +08:00 committed by GitHub
parent 6fc4579251 eda2d8141b
commit ed82e49f49
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
23 changed files with 359 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/service/INetworkAddressIDService.java
View File

@ -24,7 +24,9 @@ import org.apache.skywalking.apm.collector.core.module.Service;
* @author peng-yongsheng
*/
public interface INetworkAddressIDService extends Service {
int create(String networkAddress, int spanLayer);
int getOrCreate(String networkAddress);
int get(String networkAddress);
void update(int addressId, int spanLayer, int serverType);
}

44
apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterSerialWorker.java
View File

@ -41,7 +41,7 @@ public class NetworkAddressRegisterSerialWorker extends AbstractLocalAsyncWorker
private final INetworkAddressRegisterDAO networkAddressRegisterDAO;
private final NetworkAddressCacheService networkAddressCacheService;
public NetworkAddressRegisterSerialWorker(ModuleManager moduleManager) {
NetworkAddressRegisterSerialWorker(ModuleManager moduleManager) {
super(moduleManager);
this.networkAddressRegisterDAO = getModuleManager().find(StorageModule.NAME).getService(INetworkAddressRegisterDAO.class);
this.networkAddressCacheService = getModuleManager().find(CacheModule.NAME).getService(NetworkAddressCacheService.class);
@ -53,28 +53,32 @@ public class NetworkAddressRegisterSerialWorker extends AbstractLocalAsyncWorker
@Override protected void onWork(NetworkAddress networkAddress) throws WorkerException {
logger.debug("register network address, address: {}", networkAddress.getNetworkAddress());
int addressId = networkAddressCacheService.getAddressId(networkAddress.getNetworkAddress());
if (networkAddress.getAddressId() == 0) {
int addressId = networkAddressCacheService.getAddressId(networkAddress.getNetworkAddress());
if (addressId == 0) {
NetworkAddress newNetworkAddress;
int min = networkAddressRegisterDAO.getMinNetworkAddressId();
if (min == 0) {
newNetworkAddress = new NetworkAddress();
newNetworkAddress.setId("-1");
newNetworkAddress.setAddressId(-1);
newNetworkAddress.setSpanLayer(networkAddress.getSpanLayer());
newNetworkAddress.setNetworkAddress(networkAddress.getNetworkAddress());
} else {
int max = networkAddressRegisterDAO.getMaxNetworkAddressId();
addressId = IdAutoIncrement.INSTANCE.increment(min, max);
if (addressId == 0) {
NetworkAddress newNetworkAddress;
int min = networkAddressRegisterDAO.getMinNetworkAddressId();
if (min == 0) {
newNetworkAddress = new NetworkAddress();
newNetworkAddress.setId("-1");
newNetworkAddress.setAddressId(-1);
newNetworkAddress.setSpanLayer(networkAddress.getSpanLayer());
newNetworkAddress.setNetworkAddress(networkAddress.getNetworkAddress());
} else {
int max = networkAddressRegisterDAO.getMaxNetworkAddressId();
addressId = IdAutoIncrement.INSTANCE.increment(min, max);
newNetworkAddress = new NetworkAddress();
newNetworkAddress.setId(String.valueOf(addressId));
newNetworkAddress.setAddressId(addressId);
newNetworkAddress.setSpanLayer(networkAddress.getSpanLayer());
newNetworkAddress.setNetworkAddress(networkAddress.getNetworkAddress());
newNetworkAddress = new NetworkAddress();
newNetworkAddress.setId(String.valueOf(addressId));
newNetworkAddress.setAddressId(addressId);
newNetworkAddress.setSpanLayer(networkAddress.getSpanLayer());
newNetworkAddress.setNetworkAddress(networkAddress.getNetworkAddress());
}
networkAddressRegisterDAO.save(newNetworkAddress);
}
networkAddressRegisterDAO.save(newNetworkAddress);
} else {
networkAddressRegisterDAO.update(networkAddress.getId(), networkAddress.getSpanLayer(), networkAddress.getServerType());
}
}

4
apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/service/InstanceIDService.java
View File

@ -72,7 +72,7 @@ public class InstanceIDService implements IInstanceIDService {
}
@Override public int getOrCreateByAgentUUID(int applicationId, String agentUUID, long registerTime, String osInfo) {
logger.debug("get or create instance id by agent UUID, application id: {}, agentUUID: {}, registerTime: {}, osInfo: {}", applicationId, agentUUID, registerTime, osInfo);
logger.debug("get or getOrCreate instance id by agent UUID, application id: {}, agentUUID: {}, registerTime: {}, osInfo: {}", applicationId, agentUUID, registerTime, osInfo);
int instanceId = getInstanceCacheService().getInstanceIdByAgentUUID(applicationId, agentUUID);
if (instanceId == 0) {
@ -93,7 +93,7 @@ public class InstanceIDService implements IInstanceIDService {
}
@Override public int getOrCreateByAddressId(int applicationId, int addressId, long registerTime) {
logger.debug("get or create instance id by address id, application id: {}, address id: {}, registerTime: {}", applicationId, addressId, registerTime);
logger.debug("get or getOrCreate instance id by address id, application id: {}, address id: {}, registerTime: {}", applicationId, addressId, registerTime);
int instanceId = getInstanceCacheService().getInstanceIdByAddressId(applicationId, addressId);
if (instanceId == 0) {

22
apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/service/NetworkAddressIDService.java
View File

@ -28,6 +28,7 @@ import org.apache.skywalking.apm.collector.cache.service.NetworkAddressCacheServ
import org.apache.skywalking.apm.collector.core.graph.Graph;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress;
@ -74,7 +75,7 @@ public class NetworkAddressIDService implements INetworkAddressIDService {
return this.networkAddressGraph;
}
@Override public int create(String networkAddress, int spanLayer) {
@Override public int getOrCreate(String networkAddress) {
int addressId = getNetworkAddressCacheService().getAddressId(networkAddress);
if (addressId != 0) {
@ -89,10 +90,11 @@ public class NetworkAddressIDService implements INetworkAddressIDService {
}
} else {
NetworkAddress newNetworkAddress = new NetworkAddress();
newNetworkAddress.setId("0");
newNetworkAddress.setId(String.valueOf(Const.NONE));
newNetworkAddress.setNetworkAddress(networkAddress);
newNetworkAddress.setSpanLayer(spanLayer);
newNetworkAddress.setAddressId(0);
newNetworkAddress.setSpanLayer(Const.NONE);
newNetworkAddress.setServerType(Const.NONE);
newNetworkAddress.setAddressId(Const.NONE);
getNetworkAddressGraph().start(newNetworkAddress);
}
@ -103,4 +105,16 @@ public class NetworkAddressIDService implements INetworkAddressIDService {
@Override public int get(String networkAddress) {
return getNetworkAddressCacheService().getAddressId(networkAddress);
}
@Override public void update(int addressId, int spanLayer, int serverType) {
if (!networkAddressCacheService.compare(addressId, spanLayer, serverType)) {
NetworkAddress newNetworkAddress = new NetworkAddress();
newNetworkAddress.setId(String.valueOf(addressId));
newNetworkAddress.setSpanLayer(spanLayer);
newNetworkAddress.setServerType(serverType);
newNetworkAddress.setAddressId(addressId);
getNetworkAddressGraph().start(newNetworkAddress);
}
}
}

2
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/buffer/SegmentBufferManager.java
View File

@ -81,7 +81,7 @@ public enum SegmentBufferManager {
}
private void newDataFile() throws IOException {
logger.debug("create new segment buffer file");
logger.debug("getOrCreate new segment buffer file");
String timeBucket = String.valueOf(TimeBucketUtils.INSTANCE.getSecondTimeBucket(System.currentTimeMillis()));
String writeFileName = DATA_FILE_PREFIX + "_" + timeBucket + "." + Const.FILE_SUFFIX;
File dataFile = new File(BufferFileConfig.BUFFER_PATH + writeFileName);

3
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/standardization/ReferenceIdExchanger.java
View File

@ -89,7 +89,8 @@ public class ReferenceIdExchanger implements IdExchanger<ReferenceDecorator> {
}
if (standardBuilder.getNetworkAddressId() == 0 && StringUtils.isNotEmpty(standardBuilder.getNetworkAddress())) {
int networkAddressId = networkAddressIDService.get(standardBuilder.getNetworkAddress());
int networkAddressId = networkAddressIDService.getOrCreate(standardBuilder.getNetworkAddress());
if (networkAddressId == 0) {
if (logger.isDebugEnabled()) {
logger.debug("network address: {} from application id: {} exchange failed", standardBuilder.getNetworkAddress(), applicationId);

7
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/standardization/SpanIdExchanger.java
View File

@ -25,6 +25,7 @@ import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decora
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.core.util.StringUtils;
import org.apache.skywalking.apm.collector.storage.table.register.ServerTypeDefine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -53,7 +54,7 @@ public class SpanIdExchanger implements IdExchanger<SpanDecorator> {
@Override public boolean exchange(SpanDecorator standardBuilder, int applicationId) {
if (standardBuilder.getPeerId() == 0 && StringUtils.isNotEmpty(standardBuilder.getPeer())) {
int peerId = networkAddressIDService.create(standardBuilder.getPeer(), standardBuilder.getSpanLayer().getNumber());
int peerId = networkAddressIDService.getOrCreate(standardBuilder.getPeer());
if (peerId == 0) {
logger.debug("peer: {} in application: {} exchange failed", standardBuilder.getPeer(), applicationId);
@ -62,6 +63,10 @@ public class SpanIdExchanger implements IdExchanger<SpanDecorator> {
standardBuilder.toBuilder();
standardBuilder.setPeerId(peerId);
standardBuilder.setPeer(Const.EMPTY_STRING);
int spanLayer = standardBuilder.getSpanLayerValue();
int serverType = ServerTypeDefine.getInstance().getServerTypeId(standardBuilder.getComponentId());
networkAddressIDService.update(peerId, spanLayer, serverType);
}
}

2
apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/service/NetworkAddressCacheService.java vendored
View File

@ -27,4 +27,6 @@ public interface NetworkAddressCacheService extends Service {
int getAddressId(String networkAddress);
String getAddress(int addressId);
boolean compare(int addressId, int spanLayer, int serverType);
}

4
apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/CacheModuleGuavaProvider.java vendored
View File

@ -22,7 +22,7 @@ import java.util.Properties;
import org.apache.skywalking.apm.collector.cache.CacheModule;
import org.apache.skywalking.apm.collector.cache.guava.service.ApplicationCacheGuavaService;
import org.apache.skywalking.apm.collector.cache.guava.service.InstanceCacheGuavaService;
import org.apache.skywalking.apm.collector.cache.guava.service.NetworAddressCacheGuavaService;
import org.apache.skywalking.apm.collector.cache.guava.service.NetworkAddressCacheGuavaService;
import org.apache.skywalking.apm.collector.cache.guava.service.ServiceIdCacheGuavaService;
import org.apache.skywalking.apm.collector.cache.guava.service.ServiceNameCacheGuavaService;
import org.apache.skywalking.apm.collector.cache.service.ApplicationCacheService;
@ -53,7 +53,7 @@ public class CacheModuleGuavaProvider extends ModuleProvider {
this.registerServiceImplementation(InstanceCacheService.class, new InstanceCacheGuavaService(getManager()));
this.registerServiceImplementation(ServiceIdCacheService.class, new ServiceIdCacheGuavaService(getManager()));
this.registerServiceImplementation(ServiceNameCacheService.class, new ServiceNameCacheGuavaService(getManager()));
this.registerServiceImplementation(NetworkAddressCacheService.class, new NetworAddressCacheGuavaService(getManager()));
this.registerServiceImplementation(NetworkAddressCacheService.class, new NetworkAddressCacheGuavaService(getManager()));
}
@Override public void start(Properties config) throws ServiceNotProvidedException {

42
apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/service/NetworAddressCacheGuavaService.java → apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/service/NetworkAddressCacheGuavaService.java vendored
View File

@ -27,22 +27,23 @@ import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
import org.apache.skywalking.apm.collector.core.util.StringUtils;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.storage.dao.cache.INetworkAddressCacheDAO;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author peng-yongsheng
*/
public class NetworAddressCacheGuavaService implements NetworkAddressCacheService {
public class NetworkAddressCacheGuavaService implements NetworkAddressCacheService {
private final Logger logger = LoggerFactory.getLogger(NetworAddressCacheGuavaService.class);
private final Logger logger = LoggerFactory.getLogger(NetworkAddressCacheGuavaService.class);
private final Cache<String, Integer> addressCache = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(5000).build();
private final ModuleManager moduleManager;
private INetworkAddressCacheDAO networkAddressCacheDAO;
public NetworAddressCacheGuavaService(ModuleManager moduleManager) {
public NetworkAddressCacheGuavaService(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
@ -57,16 +58,17 @@ public class NetworAddressCacheGuavaService implements NetworkAddressCacheServic
int addressId = 0;
try {
addressId = addressCache.get(networkAddress, () -> getNetworkAddressCacheDAO().getAddressId(networkAddress));
if (addressId == 0) {
addressId = getNetworkAddressCacheDAO().getAddressId(networkAddress);
if (addressId != 0) {
addressCache.put(networkAddress, addressId);
}
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
if (addressId == 0) {
addressId = getNetworkAddressCacheDAO().getAddressId(networkAddress);
if (addressId != 0) {
addressCache.put(networkAddress, addressId);
}
}
return addressId;
}
@ -75,17 +77,33 @@ public class NetworAddressCacheGuavaService implements NetworkAddressCacheServic
public String getAddress(int addressId) {
String networkAddress = Const.EMPTY_STRING;
try {
networkAddress = idCache.get(addressId, () -> getNetworkAddressCacheDAO().getAddress(addressId));
networkAddress = idCache.get(addressId, () -> getNetworkAddressCacheDAO().getAddressById(addressId));
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
if (StringUtils.isEmpty(networkAddress)) {
networkAddress = getNetworkAddressCacheDAO().getAddress(addressId);
networkAddress = getNetworkAddressCacheDAO().getAddressById(addressId);
if (StringUtils.isNotEmpty(networkAddress)) {
addressCache.put(networkAddress, addressId);
idCache.put(addressId, networkAddress);
}
}
return networkAddress;
}
private final Cache<Integer, NetworkAddress> addressObjCache = CacheBuilder.newBuilder().maximumSize(5000).build();
@Override public boolean compare(int addressId, int spanLayer, int serverType) {
try {
NetworkAddress address = addressObjCache.get(addressId, () -> getNetworkAddressCacheDAO().getAddress(addressId));
if (ObjectUtils.isNotEmpty(address)) {
if (spanLayer != address.getSpanLayer() || serverType != address.getServerType()) {
return false;
}
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return true;
}
}

5
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/cache/INetworkAddressCacheDAO.java vendored
View File

@ -19,6 +19,7 @@
package org.apache.skywalking.apm.collector.storage.dao.cache;
import org.apache.skywalking.apm.collector.storage.base.dao.DAO;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress;
/**
* @author peng-yongsheng
@ -26,5 +27,7 @@ import org.apache.skywalking.apm.collector.storage.base.dao.DAO;
public interface INetworkAddressCacheDAO extends DAO {
int getAddressId(String networkAddress);
String getAddress(int addressId);
String getAddressById(int addressId);
NetworkAddress getAddress(int addressId);
}

2
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/register/INetworkAddressRegisterDAO.java
View File

@ -30,4 +30,6 @@ public interface INetworkAddressRegisterDAO extends DAO {
int getMinNetworkAddressId();
void save(NetworkAddress networkAddress);
void update(String id, int spanLayer, int serverType);
}

12
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddress.java
View File

@ -20,6 +20,7 @@ package org.apache.skywalking.apm.collector.storage.table.register;
import org.apache.skywalking.apm.collector.core.data.Column;
import org.apache.skywalking.apm.collector.core.data.StreamData;
import org.apache.skywalking.apm.collector.core.data.operator.CoverOperation;
import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
/**
@ -39,7 +40,8 @@ public class NetworkAddress extends StreamData {
private static final Column[] INTEGER_COLUMNS = {
new Column(NetworkAddressTable.COLUMN_ADDRESS_ID, new NonOperation()),
new Column(NetworkAddressTable.COLUMN_SPAN_LAYER, new NonOperation()),
new Column(NetworkAddressTable.COLUMN_SPAN_LAYER, new CoverOperation()),
new Column(NetworkAddressTable.COLUMN_SERVER_TYPE, new CoverOperation()),
};
private static final Column[] BYTE_COLUMNS = {};
@ -87,4 +89,12 @@ public class NetworkAddress extends StreamData {
public void setSpanLayer(Integer spanLayer) {
setDataInteger(1, spanLayer);
}
public Integer getServerType() {
return getDataInteger(2);
}
public void setServerType(Integer serverType) {
setDataInteger(2, serverType);
}
}

1
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddressTable.java
View File

@ -27,5 +27,6 @@ public class NetworkAddressTable extends CommonTable {
public static final String TABLE = "network_address";
public static final String COLUMN_NETWORK_ADDRESS = "network_address";
public static final String COLUMN_SPAN_LAYER = "span_layer";
public static final String COLUMN_SERVER_TYPE = "server_type";
public static final String COLUMN_ADDRESS_ID = "address_id";
}

46
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/ServerType.java Normal file
View File

@ -0,0 +1,46 @@
/*
* 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.collector.storage.table.register;
/**
* @author peng-yongsheng
*/
public class ServerType {
private int componentId;
private int id;
private String name;
public ServerType(int componentId, int id, String name) {
this.componentId = componentId;
this.id = id;
this.name = name;
}
public int getId() {
return id;
}
public String getName() {
return name;
}
public int getComponentId() {
return componentId;
}
}

82
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/ServerTypeDefine.java Normal file
View File

@ -0,0 +1,82 @@
/*
* 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.collector.storage.table.register;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
/**
* @author peng-yongsheng
*/
public class ServerTypeDefine {
private static ServerTypeDefine INSTANCE = new ServerTypeDefine();
private String[] serverTypeNames;
private ServerType[] serverTypes;
private ServerTypeDefine() {
this.serverTypes = new ServerType[28];
this.serverTypeNames = new String[11];
addServerType(new ServerType(ComponentsDefine.TOMCAT.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.HTTPCLIENT.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.DUBBO.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.H2.getId(), 1, ComponentsDefine.H2.getName()));
addServerType(new ServerType(ComponentsDefine.MYSQL.getId(), 2, ComponentsDefine.MYSQL.getName()));
addServerType(new ServerType(ComponentsDefine.ORACLE.getId(), 3, ComponentsDefine.ORACLE.getName()));
addServerType(new ServerType(ComponentsDefine.REDIS.getId(), 4, ComponentsDefine.REDIS.getName()));
addServerType(new ServerType(ComponentsDefine.MOTAN.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.MONGODB.getId(), 5, ComponentsDefine.MONGODB.getName()));
addServerType(new ServerType(ComponentsDefine.RESIN.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.FEIGN.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.OKHTTP.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.SPRING_REST_TEMPLATE.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.SPRING_MVC_ANNOTATION.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.STRUTS2.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.NUTZ_MVC_ANNOTATION.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.NUTZ_HTTP.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.JETTY_CLIENT.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.JETTY_SERVER.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.MEMCACHED.getId(), 6, ComponentsDefine.MEMCACHED.getName()));
addServerType(new ServerType(ComponentsDefine.SHARDING_JDBC.getId(), 7, ComponentsDefine.SHARDING_JDBC.getName()));
addServerType(new ServerType(ComponentsDefine.POSTGRESQL.getId(), 8, ComponentsDefine.POSTGRESQL.getName()));
addServerType(new ServerType(ComponentsDefine.GRPC.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.ELASTIC_JOB.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.ROCKET_MQ.getId(), 9, ComponentsDefine.ROCKET_MQ.getName()));
addServerType(new ServerType(ComponentsDefine.HTTP_ASYNC_CLIENT.getId(), Const.NONE, Const.EMPTY_STRING));
addServerType(new ServerType(ComponentsDefine.KAFKA.getId(), 10, ComponentsDefine.KAFKA.getName()));
}
public static ServerTypeDefine getInstance() {
return INSTANCE;
}
private void addServerType(ServerType serverType) {
serverTypeNames[serverType.getId()] = serverType.getName();
serverTypes[serverType.getComponentId()] = serverType;
}
public int getServerTypeId(int componentId) {
return serverTypes[componentId].getId();
}
public String getServerType(int serverTypeId) {
return serverTypeNames[serverTypeId];
}
}

42
apm-collector/apm-collector-storage/collector-storage-define/src/test/java/org/apache/skywalking/apm/collector/storage/table/register/ServerTypeDefineTestCase.java Normal file
View File

@ -0,0 +1,42 @@
/*
* 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.collector.storage.table.register;
import java.lang.reflect.Field;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.network.trace.component.OfficialComponent;
import org.junit.Test;
/**
* @author peng-yongsheng
*/
public class ServerTypeDefineTestCase {
@Test
public void check() throws IllegalAccessException {
Field[] fields = ComponentsDefine.class.getDeclaredFields();
for (Field field : fields) {
if (field.getType().equals(OfficialComponent.class)) {
OfficialComponent component = (OfficialComponent)field.get(ComponentsDefine.getInstance());
ServerTypeDefine.getInstance().getServerTypeId(component.getId());
}
}
}
}

24
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/cache/NetworkAddressEsCacheDAO.java vendored
View File

@ -22,6 +22,7 @@ import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchCli
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.storage.dao.cache.INetworkAddressCacheDAO;
import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddressTable;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.get.GetResponse;
@ -56,12 +57,12 @@ public class NetworkAddressEsCacheDAO extends EsDAO implements INetworkAddressCa
SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
if (searchResponse.getHits().totalHits > 0) {
SearchHit searchHit = searchResponse.getHits().iterator().next();
return (int)searchHit.getSource().get(NetworkAddressTable.COLUMN_ADDRESS_ID);
return ((Number)searchHit.getSource().get(NetworkAddressTable.COLUMN_ADDRESS_ID)).intValue();
}
return 0;
return Const.NONE;
}
@Override public String getAddress(int addressId) {
@Override public String getAddressById(int addressId) {
logger.debug("get network address, address id: {}", addressId);
ElasticSearchClient client = getClient();
GetRequestBuilder getRequestBuilder = client.prepareGet(NetworkAddressTable.TABLE, String.valueOf(addressId));
@ -72,4 +73,21 @@ public class NetworkAddressEsCacheDAO extends EsDAO implements INetworkAddressCa
}
return Const.EMPTY_STRING;
}
@Override public NetworkAddress getAddress(int addressId) {
ElasticSearchClient client = getClient();
GetRequestBuilder getRequestBuilder = client.prepareGet(NetworkAddressTable.TABLE, String.valueOf(addressId));
GetResponse getResponse = getRequestBuilder.get();
if (getResponse.isExists()) {
NetworkAddress address = new NetworkAddress();
address.setId((String)getResponse.getSource().get(NetworkAddressTable.COLUMN_ID));
address.setAddressId(((Number)getResponse.getSource().get(NetworkAddressTable.COLUMN_ADDRESS_ID)).intValue());
address.setSpanLayer(((Number)getResponse.getSource().get(NetworkAddressTable.COLUMN_SPAN_LAYER)).intValue());
address.setServerType(((Number)getResponse.getSource().get(NetworkAddressTable.COLUMN_SERVER_TYPE)).intValue());
address.setNetworkAddress((String)getResponse.getSource().get(NetworkAddressTable.COLUMN_NETWORK_ADDRESS));
return address;
}
return null;
}
}

10
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/register/NetworkAddressRegisterEsDAO.java
View File

@ -56,8 +56,18 @@ public class NetworkAddressRegisterEsDAO extends EsDAO implements INetworkAddres
source.put(NetworkAddressTable.COLUMN_NETWORK_ADDRESS, networkAddress.getNetworkAddress());
source.put(NetworkAddressTable.COLUMN_ADDRESS_ID, networkAddress.getAddressId());
source.put(NetworkAddressTable.COLUMN_SPAN_LAYER, networkAddress.getSpanLayer());
source.put(NetworkAddressTable.COLUMN_SERVER_TYPE, networkAddress.getServerType());
IndexResponse response = client.prepareIndex(NetworkAddressTable.TABLE, networkAddress.getId()).setSource(source).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
logger.debug("save network address register info, address getId: {}, network address code: {}, status: {}", networkAddress.getAddressId(), networkAddress.getNetworkAddress(), response.status().name());
}
@Override public void update(String id, int spanLayer, int serverType) {
ElasticSearchClient client = getClient();
Map<String, Object> source = new HashMap<>();
source.put(NetworkAddressTable.COLUMN_SPAN_LAYER, spanLayer);
source.put(NetworkAddressTable.COLUMN_SERVER_TYPE, serverType);
client.prepareUpdate(NetworkAddressTable.TABLE, id).setDoc(source).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
}
}

2
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/register/NetworkAddressEsTableDefine.java
View File

@ -38,5 +38,7 @@ public class NetworkAddressEsTableDefine extends ElasticSearchTableDefine {
@Override public void initialize() {
addColumn(new ElasticSearchColumnDefine(NetworkAddressTable.COLUMN_NETWORK_ADDRESS, ElasticSearchColumnDefine.Type.Keyword.name()));
addColumn(new ElasticSearchColumnDefine(NetworkAddressTable.COLUMN_ADDRESS_ID, ElasticSearchColumnDefine.Type.Integer.name()));
addColumn(new ElasticSearchColumnDefine(NetworkAddressTable.COLUMN_SPAN_LAYER, ElasticSearchColumnDefine.Type.Integer.name()));
addColumn(new ElasticSearchColumnDefine(NetworkAddressTable.COLUMN_SERVER_TYPE, ElasticSearchColumnDefine.Type.Integer.name()));
}
}

29
apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/cache/NetworkAddressH2CacheDAO.java vendored
View File

@ -26,6 +26,7 @@ import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.storage.base.sql.SqlBuilder;
import org.apache.skywalking.apm.collector.storage.dao.cache.INetworkAddressCacheDAO;
import org.apache.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddressTable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -47,6 +48,7 @@ public class NetworkAddressH2CacheDAO extends H2DAO implements INetworkAddressCa
public int getAddressId(String networkAddress) {
logger.info("get the address id with network address = {}", networkAddress);
H2Client client = getClient();
String sql = SqlBuilder.buildSql(GET_ADDRESS_ID_OR_CODE_SQL, NetworkAddressTable.COLUMN_ADDRESS_ID, NetworkAddressTable.TABLE, NetworkAddressTable.COLUMN_NETWORK_ADDRESS);
Object[] params = new Object[] {networkAddress};
@ -57,10 +59,10 @@ public class NetworkAddressH2CacheDAO extends H2DAO implements INetworkAddressCa
} catch (SQLException | H2ClientException e) {
logger.error(e.getMessage(), e);
}
return 0;
return Const.NONE;
}
@Override public String getAddress(int addressId) {
@Override public String getAddressById(int addressId) {
logger.debug("get network address, address id: {}", addressId);
H2Client client = getClient();
String sql = SqlBuilder.buildSql(GET_ADDRESS_ID_OR_CODE_SQL, NetworkAddressTable.COLUMN_NETWORK_ADDRESS, NetworkAddressTable.TABLE, NetworkAddressTable.COLUMN_ADDRESS_ID);
@ -74,4 +76,27 @@ public class NetworkAddressH2CacheDAO extends H2DAO implements INetworkAddressCa
}
return Const.EMPTY_STRING;
}
@Override public NetworkAddress getAddress(int addressId) {
logger.debug("get network address, address id: {}", addressId);
H2Client client = getClient();
String dynamicSql = "select * from {0} where {1} = ?";
String sql = SqlBuilder.buildSql(dynamicSql, NetworkAddressTable.TABLE, NetworkAddressTable.COLUMN_ADDRESS_ID);
Object[] params = new Object[] {addressId};
try (ResultSet rs = client.executeQuery(sql, params)) {
if (rs.next()) {
NetworkAddress networkAddress = new NetworkAddress();
networkAddress.setId(rs.getString(NetworkAddressTable.COLUMN_ID));
networkAddress.setAddressId(rs.getInt(NetworkAddressTable.COLUMN_ADDRESS_ID));
networkAddress.setNetworkAddress(rs.getString(NetworkAddressTable.COLUMN_NETWORK_ADDRESS));
networkAddress.setSpanLayer(rs.getInt(NetworkAddressTable.COLUMN_SPAN_LAYER));
networkAddress.setServerType(rs.getInt(NetworkAddressTable.COLUMN_SERVER_TYPE));
return networkAddress;
}
} catch (SQLException | H2ClientException e) {
logger.error(e.getMessage(), e);
}
return null;
}
}

19
apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/register/NetworkAddressRegisterH2DAO.java
View File

@ -25,6 +25,7 @@ import org.apache.skywalking.apm.collector.client.h2.H2ClientException;
import org.apache.skywalking.apm.collector.storage.base.sql.SqlBuilder;
import org.apache.skywalking.apm.collector.storage.dao.register.INetworkAddressRegisterDAO;
import org.apache.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
import org.apache.skywalking.apm.collector.storage.table.register.InstanceTable;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress;
import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddressTable;
import org.slf4j.Logger;
@ -60,6 +61,7 @@ public class NetworkAddressRegisterH2DAO extends H2DAO implements INetworkAddres
source.put(NetworkAddressTable.COLUMN_NETWORK_ADDRESS, networkAddress.getNetworkAddress());
source.put(NetworkAddressTable.COLUMN_ADDRESS_ID, networkAddress.getAddressId());
source.put(NetworkAddressTable.COLUMN_SPAN_LAYER, networkAddress.getSpanLayer());
source.put(NetworkAddressTable.COLUMN_SERVER_TYPE, networkAddress.getServerType());
String sql = SqlBuilder.buildBatchInsertSql(NetworkAddressTable.TABLE, source.keySet());
Object[] params = source.values().toArray(new Object[0]);
@ -69,4 +71,21 @@ public class NetworkAddressRegisterH2DAO extends H2DAO implements INetworkAddres
logger.error(e.getMessage(), e);
}
}
@Override public void update(String id, int spanLayer, int serverType) {
H2Client client = getClient();
Map<String, Object> source = new HashMap<>();
source.put(NetworkAddressTable.COLUMN_SPAN_LAYER, spanLayer);
source.put(NetworkAddressTable.COLUMN_SERVER_TYPE, serverType);
String sql = SqlBuilder.buildBatchUpdateSql(InstanceTable.TABLE, source.keySet(), InstanceTable.COLUMN_INSTANCE_ID);
Object[] params = source.values().toArray(new Object[] {id});
try {
client.execute(sql, params);
} catch (H2ClientException e) {
logger.error(e.getMessage(), e);
}
}
}

2
apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/register/NetworkAddressH2TableDefine.java
View File

@ -34,6 +34,8 @@ public class NetworkAddressH2TableDefine extends H2TableDefine {
@Override public void initialize() {
addColumn(new H2ColumnDefine(NetworkAddressTable.COLUMN_ID, H2ColumnDefine.Type.Varchar.name()));
addColumn(new H2ColumnDefine(NetworkAddressTable.COLUMN_NETWORK_ADDRESS, H2ColumnDefine.Type.Varchar.name()));
addColumn(new H2ColumnDefine(NetworkAddressTable.COLUMN_SPAN_LAYER, H2ColumnDefine.Type.Int.name()));
addColumn(new H2ColumnDefine(NetworkAddressTable.COLUMN_SERVER_TYPE, H2ColumnDefine.Type.Int.name()));
addColumn(new H2ColumnDefine(NetworkAddressTable.COLUMN_ADDRESS_ID, H2ColumnDefine.Type.Int.name()));
}
}