From 279ccb2916d478fae5398638a52b24cd8c5b15d5 Mon Sep 17 00:00:00 2001 From: peng-yongsheng <8082209@qq.com> Date: Tue, 2 Jan 2018 19:43:38 +0800 Subject: [PATCH] Network address register and cache. --- .../worker/instance/InstanceMappingGraph.java | 2 +- .../InstanceMappingPersistenceWorker.java | 4 +- .../register/define/graph/GraphIdDefine.java | 1 + .../register/define/graph/WorkerIdDefine.java | 2 + .../AnalysisRegisterModuleProvider.java | 4 + .../register/NetworkAddressRegisterGraph.java | 49 ++++++++++ .../NetworkAddressRegisterRemoteWorker.java | 66 ++++++++++++++ .../NetworkAddressRegisterSerialWorker.java | 91 +++++++++++++++++++ .../apm/collector/cache/CacheModule.java | 3 +- .../service/NetworkAddressCacheService.java | 30 ++++++ .../cache/guava/CacheModuleGuavaProvider.java | 12 ++- .../NetworAddressCacheGuavaService.java | 91 +++++++++++++++++++ .../apm/collector/storage/StorageModule.java | 2 + .../storage/dao/INetworkAddressCacheDAO.java | 30 ++++++ .../dao/INetworkAddressRegisterDAO.java | 33 +++++++ .../table/register/NetworkAddress.java | 71 +++++++++++++++ .../table/register/NetworkAddressTable.java | 30 ++++++ .../storage/es/StorageModuleEsProvider.java | 7 ++ .../es/dao/NetworkAddressEsCacheDAO.java | 75 +++++++++++++++ .../es/dao/NetworkAddressRegisterEsDAO.java | 62 +++++++++++++ .../define/NetworkAddressEsTableDefine.java | 42 +++++++++ .../resources/META-INF/defines/storage.define | 1 + .../storage/h2/StorageModuleH2Provider.java | 6 ++ .../h2/dao/NetworkAddressH2CacheDAO.java | 77 ++++++++++++++++ .../h2/dao/NetworkAddressRegisterH2DAO.java | 71 +++++++++++++++ .../define/NetworkAddressH2TableDefine.java | 39 ++++++++ .../resources/META-INF/defines/storage.define | 1 + 27 files changed, 893 insertions(+), 9 deletions(-) create mode 100644 apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterGraph.java create mode 100644 apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterRemoteWorker.java create mode 100644 apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterSerialWorker.java create mode 100644 apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/service/NetworkAddressCacheService.java create mode 100644 apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/service/NetworAddressCacheGuavaService.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/INetworkAddressCacheDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/INetworkAddressRegisterDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddress.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddressTable.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/NetworkAddressEsCacheDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/NetworkAddressRegisterEsDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/NetworkAddressEsTableDefine.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/NetworkAddressH2CacheDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/NetworkAddressRegisterH2DAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/NetworkAddressH2TableDefine.java diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMappingGraph.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMappingGraph.java index 0ead77baf..af3081a29 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMappingGraph.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMappingGraph.java @@ -45,6 +45,6 @@ public class InstanceMappingGraph { GraphManager.INSTANCE.createIfAbsent(MetricGraphIdDefine.INSTANCE_MAPPING_GRAPH_ID, InstanceMapping.class) .addNode(new InstanceMappingAggregationWorker.Factory(moduleManager).create(workerCreateListener)) .addNext(new InstanceMappingRemoteWorker.Factory(moduleManager, remoteSenderService, MetricGraphIdDefine.INSTANCE_MAPPING_GRAPH_ID).create(workerCreateListener)) - .addNext(new InstanceMappingAggregationWorker.Factory(moduleManager).create(workerCreateListener)); + .addNext(new InstanceMappingPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); } } diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMappingPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMappingPersistenceWorker.java index 86505a5c1..475a4c32d 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMappingPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMappingPersistenceWorker.java @@ -24,7 +24,7 @@ import org.apache.skywalking.apm.collector.analysis.worker.model.impl.Persistenc import org.apache.skywalking.apm.collector.core.module.ModuleManager; import org.apache.skywalking.apm.collector.storage.StorageModule; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; -import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.IInstanceMappingPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMapping; /** @@ -46,7 +46,7 @@ public class InstanceMappingPersistenceWorker extends PersistenceWorker persistenceDAO() { - return getModuleManager().find(StorageModule.NAME).getService(IApplicationMappingPersistenceDAO.class); + return getModuleManager().find(StorageModule.NAME).getService(IInstanceMappingPersistenceDAO.class); } public static class Factory extends PersistenceWorkerProvider { diff --git a/apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/GraphIdDefine.java b/apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/GraphIdDefine.java index 787c18931..1b99bcfe7 100644 --- a/apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/GraphIdDefine.java +++ b/apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/GraphIdDefine.java @@ -25,4 +25,5 @@ public class GraphIdDefine { public static final int APPLICATION_REGISTER_GRAPH_ID = 200; public static final int INSTANCE_REGISTER_GRAPH_ID = 201; public static final int SERVICE_NAME_REGISTER_GRAPH_ID = 202; + public static final int NETWORK_ADDRESS_NAME_REGISTER_GRAPH_ID = 203; } diff --git a/apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/WorkerIdDefine.java b/apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/WorkerIdDefine.java index b16dfc0ae..c8e13bf20 100644 --- a/apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/WorkerIdDefine.java +++ b/apm-collector/apm-collector-analysis/analysis-register/register-define/src/main/java/org/apache/skywalking/apm/collector/analysis/register/define/graph/WorkerIdDefine.java @@ -28,4 +28,6 @@ public class WorkerIdDefine { public static final int INSTANCE_REGISTER_SERIAL_WORKER = 203; public static final int SERVICE_NAME_REGISTER_REMOTE_WORKER = 204; public static final int SERVICE_NAME_REGISTER_SERIAL_WORKER = 205; + public static final int NETWORK_ADDRESS_REGISTER_REMOTE_WORKER = 206; + public static final int NETWORK_ADDRESS_REGISTER_SERIAL_WORKER = 207; } diff --git a/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/AnalysisRegisterModuleProvider.java b/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/AnalysisRegisterModuleProvider.java index 89eca44b2..7c74ba8b0 100644 --- a/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/AnalysisRegisterModuleProvider.java +++ b/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/AnalysisRegisterModuleProvider.java @@ -25,6 +25,7 @@ import org.apache.skywalking.apm.collector.analysis.register.define.service.IIns import org.apache.skywalking.apm.collector.analysis.register.define.service.IServiceNameService; import org.apache.skywalking.apm.collector.analysis.register.provider.register.ApplicationRegisterGraph; import org.apache.skywalking.apm.collector.analysis.register.provider.register.InstanceRegisterGraph; +import org.apache.skywalking.apm.collector.analysis.register.provider.register.NetworkAddressRegisterGraph; import org.apache.skywalking.apm.collector.analysis.register.provider.register.ServiceNameRegisterGraph; import org.apache.skywalking.apm.collector.analysis.register.provider.service.ApplicationIDService; import org.apache.skywalking.apm.collector.analysis.register.provider.service.InstanceIDService; @@ -85,5 +86,8 @@ public class AnalysisRegisterModuleProvider extends ModuleProvider { ServiceNameRegisterGraph serviceNameRegisterGraph = new ServiceNameRegisterGraph(getManager(), workerCreateListener); serviceNameRegisterGraph.create(); + + NetworkAddressRegisterGraph networkAddressRegisterGraph = new NetworkAddressRegisterGraph(getManager(), workerCreateListener); + networkAddressRegisterGraph.create(); } } diff --git a/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterGraph.java b/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterGraph.java new file mode 100644 index 000000000..85c868ade --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterGraph.java @@ -0,0 +1,49 @@ +/* + * 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.analysis.register.provider.register; + +import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener; +import org.apache.skywalking.apm.collector.core.graph.GraphManager; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.RemoteModule; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress; + +/** + * @author peng-yongsheng + */ +public class NetworkAddressRegisterGraph { + + private final ModuleManager moduleManager; + private final WorkerCreateListener workerCreateListener; + + public NetworkAddressRegisterGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) { + this.moduleManager = moduleManager; + this.workerCreateListener = workerCreateListener; + } + + public void create() { + RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class); + + GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.NETWORK_ADDRESS_NAME_REGISTER_GRAPH_ID, NetworkAddress.class) + .addNode(new NetworkAddressRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.NETWORK_ADDRESS_NAME_REGISTER_GRAPH_ID).create(workerCreateListener)) + .addNext(new NetworkAddressRegisterSerialWorker.Factory(moduleManager).create(workerCreateListener)); + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterRemoteWorker.java new file mode 100644 index 000000000..9dcd00661 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterRemoteWorker.java @@ -0,0 +1,66 @@ +/* + * 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.analysis.register.provider.register; + +import org.apache.skywalking.apm.collector.analysis.register.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService; +import org.apache.skywalking.apm.collector.remote.service.Selector; +import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class NetworkAddressRegisterRemoteWorker extends AbstractRemoteWorker { + + private final Logger logger = LoggerFactory.getLogger(NetworkAddressRegisterRemoteWorker.class); + + public NetworkAddressRegisterRemoteWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return WorkerIdDefine.NETWORK_ADDRESS_REGISTER_REMOTE_WORKER; + } + + @Override protected void onWork(NetworkAddress message) throws WorkerException { + logger.debug("network address: {}", message.getNetworkAddress()); + onNext(message); + } + + @Override public Selector selector() { + return Selector.ForeverFirst; + } + + public static class Factory extends AbstractRemoteWorkerProvider { + + public Factory(ModuleManager moduleManager, RemoteSenderService remoteSenderService, int graphId) { + super(moduleManager, remoteSenderService, graphId); + } + + @Override public NetworkAddressRegisterRemoteWorker workerInstance(ModuleManager moduleManager) { + return new NetworkAddressRegisterRemoteWorker(moduleManager); + } + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterSerialWorker.java b/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterSerialWorker.java new file mode 100644 index 000000000..c4e7b4d8b --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/NetworkAddressRegisterSerialWorker.java @@ -0,0 +1,91 @@ +/* + * 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.analysis.register.provider.register; + +import org.apache.skywalking.apm.collector.analysis.register.define.graph.WorkerIdDefine; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorker; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; +import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException; +import org.apache.skywalking.apm.collector.cache.CacheModule; +import org.apache.skywalking.apm.collector.cache.service.NetworkAddressCacheService; +import org.apache.skywalking.apm.collector.core.module.ModuleManager; +import org.apache.skywalking.apm.collector.storage.StorageModule; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressRegisterDAO; +import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class NetworkAddressRegisterSerialWorker extends AbstractLocalAsyncWorker { + + private final Logger logger = LoggerFactory.getLogger(NetworkAddressRegisterSerialWorker.class); + + private final INetworkAddressRegisterDAO networkAddressRegisterDAO; + private final NetworkAddressCacheService networkAddressCacheService; + + public NetworkAddressRegisterSerialWorker(ModuleManager moduleManager) { + super(moduleManager); + this.networkAddressRegisterDAO = getModuleManager().find(StorageModule.NAME).getService(INetworkAddressRegisterDAO.class); + this.networkAddressCacheService = getModuleManager().find(CacheModule.NAME).getService(NetworkAddressCacheService.class); + } + + @Override public int id() { + return WorkerIdDefine.NETWORK_ADDRESS_REGISTER_SERIAL_WORKER; + } + + @Override protected void onWork(NetworkAddress networkAddress) throws WorkerException { + logger.debug("register network address, address: {}", networkAddress.getNetworkAddress()); + int addressId = networkAddressCacheService.getAddressId(networkAddress.getNetworkAddress()); + + if (addressId == 0) { + NetworkAddress newNetworkAddress; + int min = networkAddressRegisterDAO.getMinNetworkAddressId(); + if (min == 0) { + newNetworkAddress = new NetworkAddress("-1"); + newNetworkAddress.setAddressId(-1); + newNetworkAddress.setNetworkAddress(networkAddress.getNetworkAddress()); + } else { + int max = networkAddressRegisterDAO.getMaxNetworkAddressId(); + addressId = IdAutoIncrement.INSTANCE.increment(min, max); + + newNetworkAddress = new NetworkAddress(String.valueOf(addressId)); + newNetworkAddress.setAddressId(addressId); + newNetworkAddress.setNetworkAddress(networkAddress.getNetworkAddress()); + } + networkAddressRegisterDAO.save(newNetworkAddress); + } + } + + public static class Factory extends AbstractLocalAsyncWorkerProvider { + + public Factory(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public NetworkAddressRegisterSerialWorker workerInstance(ModuleManager moduleManager) { + return new NetworkAddressRegisterSerialWorker(moduleManager); + } + + @Override public int queueSize() { + return 256; + } + } +} diff --git a/apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/CacheModule.java b/apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/CacheModule.java index c37dea195..b73f30956 100644 --- a/apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/CacheModule.java +++ b/apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/CacheModule.java @@ -20,6 +20,7 @@ package org.apache.skywalking.apm.collector.cache; import org.apache.skywalking.apm.collector.cache.service.ApplicationCacheService; import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService; +import org.apache.skywalking.apm.collector.cache.service.NetworkAddressCacheService; import org.apache.skywalking.apm.collector.cache.service.ServiceIdCacheService; import org.apache.skywalking.apm.collector.cache.service.ServiceNameCacheService; import org.apache.skywalking.apm.collector.core.module.Module; @@ -36,6 +37,6 @@ public class CacheModule extends Module { } @Override public Class[] services() { - return new Class[] {ApplicationCacheService.class, InstanceCacheService.class, ServiceIdCacheService.class, ServiceNameCacheService.class}; + return new Class[] {ApplicationCacheService.class, InstanceCacheService.class, ServiceIdCacheService.class, ServiceNameCacheService.class, NetworkAddressCacheService.class}; } } diff --git a/apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/service/NetworkAddressCacheService.java b/apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/service/NetworkAddressCacheService.java new file mode 100644 index 000000000..f0483af11 --- /dev/null +++ b/apm-collector/apm-collector-cache/collector-cache-define/src/main/java/org/apache/skywalking/apm/collector/cache/service/NetworkAddressCacheService.java @@ -0,0 +1,30 @@ +/* + * 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.cache.service; + +import org.apache.skywalking.apm.collector.core.module.Service; + +/** + * @author peng-yongsheng + */ +public interface NetworkAddressCacheService extends Service { + int getAddressId(String networkAddress); + + String getAddress(int addressId); +} diff --git a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/CacheModuleGuavaProvider.java b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/CacheModuleGuavaProvider.java index b4a9c6ae4..af2d54cb1 100644 --- a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/CacheModuleGuavaProvider.java +++ b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/CacheModuleGuavaProvider.java @@ -16,22 +16,23 @@ * */ - package org.apache.skywalking.apm.collector.cache.guava; 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.ServiceIdCacheGuavaService; import org.apache.skywalking.apm.collector.cache.guava.service.ServiceNameCacheGuavaService; -import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService; -import org.apache.skywalking.apm.collector.cache.service.ServiceIdCacheService; -import org.apache.skywalking.apm.collector.core.module.ModuleProvider; -import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException; import org.apache.skywalking.apm.collector.cache.service.ApplicationCacheService; +import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService; +import org.apache.skywalking.apm.collector.cache.service.NetworkAddressCacheService; +import org.apache.skywalking.apm.collector.cache.service.ServiceIdCacheService; import org.apache.skywalking.apm.collector.cache.service.ServiceNameCacheService; import org.apache.skywalking.apm.collector.core.module.Module; +import org.apache.skywalking.apm.collector.core.module.ModuleProvider; +import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException; import org.apache.skywalking.apm.collector.storage.StorageModule; /** @@ -52,6 +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())); } @Override public void start(Properties config) throws ServiceNotProvidedException { diff --git a/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/service/NetworAddressCacheGuavaService.java b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/service/NetworAddressCacheGuavaService.java new file mode 100644 index 000000000..847c2290c --- /dev/null +++ b/apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/apache/skywalking/apm/collector/cache/guava/service/NetworAddressCacheGuavaService.java @@ -0,0 +1,91 @@ +/* + * 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.cache.guava.service; + +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; +import org.apache.skywalking.apm.collector.cache.service.NetworkAddressCacheService; +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.core.util.StringUtils; +import org.apache.skywalking.apm.collector.storage.StorageModule; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressCacheDAO; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class NetworAddressCacheGuavaService implements NetworkAddressCacheService { + + private final Logger logger = LoggerFactory.getLogger(NetworAddressCacheGuavaService.class); + + private final Cache addressCache = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(5000).build(); + + private final ModuleManager moduleManager; + private INetworkAddressCacheDAO networkAddressCacheDAO; + + public NetworAddressCacheGuavaService(ModuleManager moduleManager) { + this.moduleManager = moduleManager; + } + + private INetworkAddressCacheDAO getNetworkAddressCacheDAO() { + if (ObjectUtils.isEmpty(networkAddressCacheDAO)) { + this.networkAddressCacheDAO = moduleManager.find(StorageModule.NAME).getService(INetworkAddressCacheDAO.class); + } + return this.networkAddressCacheDAO; + } + + public int getAddressId(String networkAddress) { + int addressId = 0; + try { + addressId = addressCache.get(networkAddress, () -> getNetworkAddressCacheDAO().getAddressId(networkAddress)); + } catch (Throwable e) { + logger.error(e.getMessage(), e); + } + + if (addressId == 0) { + addressId = getNetworkAddressCacheDAO().getAddressId(networkAddress); + if (addressId != 0) { + addressCache.put(networkAddress, addressId); + } + } + return addressId; + } + + private final Cache idCache = CacheBuilder.newBuilder().maximumSize(5000).build(); + + public String getAddress(int addressId) { + String networkAddress = Const.EMPTY_STRING; + try { + networkAddress = idCache.get(addressId, () -> getNetworkAddressCacheDAO().getAddress(addressId)); + } catch (Throwable e) { + logger.error(e.getMessage(), e); + } + + if (StringUtils.isEmpty(networkAddress)) { + networkAddress = getNetworkAddressCacheDAO().getAddress(addressId); + if (StringUtils.isNotEmpty(networkAddress)) { + addressCache.put(networkAddress, addressId); + } + } + return networkAddress; + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java index cc145db57..7b9b03231 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java @@ -57,6 +57,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceD import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricUIDAO; import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricUIDAO; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressCacheDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostUIDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentPersistenceDAO; @@ -101,6 +102,7 @@ public class StorageModule extends Module { classes.add(IApplicationCacheDAO.class); classes.add(IInstanceCacheDAO.class); classes.add(IServiceNameCacheDAO.class); + classes.add(INetworkAddressCacheDAO.class); } private void addRegisterDAO(List classes) { diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/INetworkAddressCacheDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/INetworkAddressCacheDAO.java new file mode 100644 index 000000000..a1e602339 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/INetworkAddressCacheDAO.java @@ -0,0 +1,30 @@ +/* + * 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.dao; + +import org.apache.skywalking.apm.collector.storage.base.dao.DAO; + +/** + * @author peng-yongsheng + */ +public interface INetworkAddressCacheDAO extends DAO { + int getAddressId(String networkAddress); + + String getAddress(int addressId); +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/INetworkAddressRegisterDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/INetworkAddressRegisterDAO.java new file mode 100644 index 000000000..d456b572d --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/INetworkAddressRegisterDAO.java @@ -0,0 +1,33 @@ +/* + * 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.dao; + +import org.apache.skywalking.apm.collector.storage.base.dao.DAO; +import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddress; + +/** + * @author peng-yongsheng + */ +public interface INetworkAddressRegisterDAO extends DAO { + int getMaxNetworkAddressId(); + + int getMinNetworkAddressId(); + + void save(NetworkAddress networkAddress); +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddress.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddress.java new file mode 100644 index 000000000..99aafbf95 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddress.java @@ -0,0 +1,71 @@ +/* + * 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.data.Column; +import org.apache.skywalking.apm.collector.core.data.Data; +import org.apache.skywalking.apm.collector.core.data.operator.NonOperation; + +/** + * @author peng-yongsheng + */ +public class NetworkAddress extends Data { + + private static final Column[] STRING_COLUMNS = { + new Column(NetworkAddressTable.COLUMN_ID, new NonOperation()), + new Column(NetworkAddressTable.COLUMN_NETWORK_ADDRESS, new NonOperation()), + }; + + private static final Column[] LONG_COLUMNS = { + }; + + private static final Column[] DOUBLE_COLUMNS = {}; + + private static final Column[] INTEGER_COLUMNS = { + new Column(NetworkAddressTable.COLUMN_ADDRESS_ID, new NonOperation()), + }; + + private static final Column[] BOOLEAN_COLUMNS = {}; + + private static final Column[] BYTE_COLUMNS = {}; + + public NetworkAddress(String id) { + super(id, STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS); + } + + public String getId() { + return getDataString(0); + } + + public String getNetworkAddress() { + return getDataString(1); + } + + public void setNetworkAddress(String networkAddress) { + setDataString(1, networkAddress); + } + + public Integer getAddressId() { + return getDataInteger(0); + } + + public void setAddressId(Integer addressId) { + setDataInteger(0, addressId); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddressTable.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddressTable.java new file mode 100644 index 000000000..97ffe2e76 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/register/NetworkAddressTable.java @@ -0,0 +1,30 @@ +/* + * 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.data.CommonTable; + +/** + * @author peng-yongsheng + */ +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_ADDRESS_ID = "address_id"; +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java index 40a8c1e29..984262037 100644 --- a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java @@ -66,6 +66,8 @@ import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceD import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricUIDAO; import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricUIDAO; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressCacheDAO; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressRegisterDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostUIDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentPersistenceDAO; @@ -118,6 +120,8 @@ import org.apache.skywalking.apm.collector.storage.es.dao.MemoryMetricEsPersiste import org.apache.skywalking.apm.collector.storage.es.dao.MemoryMetricEsUIDAO; import org.apache.skywalking.apm.collector.storage.es.dao.MemoryPoolMetricEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.MemoryPoolMetricEsUIDAO; +import org.apache.skywalking.apm.collector.storage.es.dao.NetworkAddressEsCacheDAO; +import org.apache.skywalking.apm.collector.storage.es.dao.NetworkAddressRegisterEsDAO; import org.apache.skywalking.apm.collector.storage.es.dao.SegmentCostEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.SegmentCostEsUIDAO; import org.apache.skywalking.apm.collector.storage.es.dao.SegmentEsPersistenceDAO; @@ -212,6 +216,7 @@ public class StorageModuleEsProvider extends ModuleProvider { this.registerServiceImplementation(IApplicationCacheDAO.class, new ApplicationEsCacheDAO(elasticSearchClient)); this.registerServiceImplementation(IInstanceCacheDAO.class, new InstanceEsCacheDAO(elasticSearchClient)); this.registerServiceImplementation(IServiceNameCacheDAO.class, new ServiceNameEsCacheDAO(elasticSearchClient)); + this.registerServiceImplementation(INetworkAddressCacheDAO.class, new NetworkAddressEsCacheDAO(elasticSearchClient)); } private void registerRegisterDAO() throws ServiceNotProvidedException { @@ -244,6 +249,8 @@ public class StorageModuleEsProvider extends ModuleProvider { } private void registerUiDAO() throws ServiceNotProvidedException { + this.registerServiceImplementation(INetworkAddressRegisterDAO.class, new NetworkAddressRegisterEsDAO(elasticSearchClient)); + this.registerServiceImplementation(IInstanceUIDAO.class, new InstanceEsUIDAO(elasticSearchClient)); this.registerServiceImplementation(ICpuMetricUIDAO.class, new CpuMetricEsUIDAO(elasticSearchClient)); diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/NetworkAddressEsCacheDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/NetworkAddressEsCacheDAO.java new file mode 100644 index 000000000..a7373514b --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/NetworkAddressEsCacheDAO.java @@ -0,0 +1,75 @@ +/* + * 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.es.dao; + +import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; +import org.apache.skywalking.apm.collector.core.util.Const; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressCacheDAO; +import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO; +import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddressTable; +import org.elasticsearch.action.get.GetRequestBuilder; +import org.elasticsearch.action.get.GetResponse; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.search.SearchType; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class NetworkAddressEsCacheDAO extends EsDAO implements INetworkAddressCacheDAO { + + private final Logger logger = LoggerFactory.getLogger(NetworkAddressEsCacheDAO.class); + + public NetworkAddressEsCacheDAO(ElasticSearchClient client) { + super(client); + } + + @Override public int getAddressId(String networkAddress) { + ElasticSearchClient client = getClient(); + + SearchRequestBuilder searchRequestBuilder = client.prepareSearch(NetworkAddressTable.TABLE); + searchRequestBuilder.setTypes("type"); + searchRequestBuilder.setSearchType(SearchType.QUERY_THEN_FETCH); + searchRequestBuilder.setQuery(QueryBuilders.termQuery(NetworkAddressTable.COLUMN_NETWORK_ADDRESS, networkAddress)); + searchRequestBuilder.setSize(1); + + 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 0; + } + + @Override public String getAddress(int addressId) { + logger.debug("get network address, address id: {}", addressId); + ElasticSearchClient client = getClient(); + GetRequestBuilder getRequestBuilder = client.prepareGet(NetworkAddressTable.TABLE, String.valueOf(addressId)); + + GetResponse getResponse = getRequestBuilder.get(); + if (getResponse.isExists()) { + return (String)getResponse.getSource().get(NetworkAddressTable.COLUMN_NETWORK_ADDRESS); + } + return Const.EMPTY_STRING; + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/NetworkAddressRegisterEsDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/NetworkAddressRegisterEsDAO.java new file mode 100644 index 000000000..f900d2c70 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/NetworkAddressRegisterEsDAO.java @@ -0,0 +1,62 @@ +/* + * 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.es.dao; + +import java.util.HashMap; +import java.util.Map; +import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressRegisterDAO; +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.index.IndexResponse; +import org.elasticsearch.action.support.WriteRequest; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class NetworkAddressRegisterEsDAO extends EsDAO implements INetworkAddressRegisterDAO { + + private final Logger logger = LoggerFactory.getLogger(NetworkAddressRegisterEsDAO.class); + + public NetworkAddressRegisterEsDAO(ElasticSearchClient client) { + super(client); + } + + @Override public int getMaxNetworkAddressId() { + return getMaxId(NetworkAddressTable.TABLE, NetworkAddressTable.COLUMN_ADDRESS_ID); + } + + @Override public int getMinNetworkAddressId() { + return getMinId(NetworkAddressTable.TABLE, NetworkAddressTable.COLUMN_ADDRESS_ID); + } + + @Override public void save(NetworkAddress networkAddress) { + logger.debug("save network address register info, address getId: {}, network address code: {}", networkAddress.getId(), networkAddress.getNetworkAddress()); + ElasticSearchClient client = getClient(); + Map source = new HashMap<>(); + source.put(NetworkAddressTable.COLUMN_NETWORK_ADDRESS, networkAddress.getNetworkAddress()); + source.put(NetworkAddressTable.COLUMN_ADDRESS_ID, networkAddress.getAddressId()); + + 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()); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/NetworkAddressEsTableDefine.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/NetworkAddressEsTableDefine.java new file mode 100644 index 000000000..3ac408e0a --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/NetworkAddressEsTableDefine.java @@ -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.es.define; + +import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchColumnDefine; +import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchTableDefine; +import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddressTable; + +/** + * @author peng-yongsheng + */ +public class NetworkAddressEsTableDefine extends ElasticSearchTableDefine { + + public NetworkAddressEsTableDefine() { + super(NetworkAddressTable.TABLE); + } + + @Override public int refreshInterval() { + return 1; + } + + @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())); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/resources/META-INF/defines/storage.define b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/resources/META-INF/defines/storage.define index 5956bf76e..c3efd65dc 100644 --- a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/resources/META-INF/defines/storage.define +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/resources/META-INF/defines/storage.define @@ -1,3 +1,4 @@ +org.apache.skywalking.apm.collector.storage.es.define.NetworkAddressEsTableDefine org.apache.skywalking.apm.collector.storage.es.define.ApplicationEsTableDefine org.apache.skywalking.apm.collector.storage.es.define.ApplicationMetricEsTableDefine org.apache.skywalking.apm.collector.storage.es.define.ApplicationAlarmEsTableDefine diff --git a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java index 0737e54d5..f3fc1c1e2 100644 --- a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java +++ b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java @@ -62,6 +62,8 @@ import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceD import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricUIDAO; import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricUIDAO; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressCacheDAO; +import org.apache.skywalking.apm.collector.storage.dao.INetworkAddressRegisterDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostUIDAO; import org.apache.skywalking.apm.collector.storage.dao.ISegmentPersistenceDAO; @@ -114,6 +116,8 @@ import org.apache.skywalking.apm.collector.storage.h2.dao.MemoryMetricH2Persiste import org.apache.skywalking.apm.collector.storage.h2.dao.MemoryMetricH2UIDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.MemoryPoolMetricH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.MemoryPoolMetricH2UIDAO; +import org.apache.skywalking.apm.collector.storage.h2.dao.NetworkAddressH2CacheDAO; +import org.apache.skywalking.apm.collector.storage.h2.dao.NetworkAddressRegisterH2DAO; import org.apache.skywalking.apm.collector.storage.h2.dao.SegmentCostH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.SegmentCostH2UIDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.SegmentH2PersistenceDAO; @@ -190,9 +194,11 @@ public class StorageModuleH2Provider extends ModuleProvider { this.registerServiceImplementation(IApplicationCacheDAO.class, new ApplicationH2CacheDAO(h2Client)); this.registerServiceImplementation(IInstanceCacheDAO.class, new InstanceH2CacheDAO(h2Client)); this.registerServiceImplementation(IServiceNameCacheDAO.class, new ServiceNameH2CacheDAO(h2Client)); + this.registerServiceImplementation(INetworkAddressCacheDAO.class, new NetworkAddressH2CacheDAO(h2Client)); } private void registerRegisterDAO() throws ServiceNotProvidedException { + this.registerServiceImplementation(INetworkAddressRegisterDAO.class, new NetworkAddressRegisterH2DAO(h2Client)); this.registerServiceImplementation(IApplicationRegisterDAO.class, new ApplicationH2RegisterDAO(h2Client)); this.registerServiceImplementation(IInstanceRegisterDAO.class, new InstanceH2RegisterDAO(h2Client)); this.registerServiceImplementation(IServiceNameRegisterDAO.class, new ServiceNameH2RegisterDAO(h2Client)); diff --git a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/NetworkAddressH2CacheDAO.java b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/NetworkAddressH2CacheDAO.java new file mode 100644 index 000000000..6c36cfadd --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/NetworkAddressH2CacheDAO.java @@ -0,0 +1,77 @@ +/* + * 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.h2.dao; + +import java.sql.ResultSet; +import java.sql.SQLException; +import org.apache.skywalking.apm.collector.client.h2.H2Client; +import org.apache.skywalking.apm.collector.client.h2.H2ClientException; +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.INetworkAddressCacheDAO; +import org.apache.skywalking.apm.collector.storage.h2.base.dao.H2DAO; +import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddressTable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng, clevertension + */ +public class NetworkAddressH2CacheDAO extends H2DAO implements INetworkAddressCacheDAO { + + private final Logger logger = LoggerFactory.getLogger(NetworkAddressH2CacheDAO.class); + + private static final String GET_ADDRESS_ID_OR_CODE_SQL = "select {0} from {1} where {2} = ?"; + + public NetworkAddressH2CacheDAO(H2Client client) { + super(client); + } + + @Override + 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}; + try (ResultSet rs = client.executeQuery(sql, params)) { + if (rs.next()) { + return rs.getInt(1); + } + } catch (SQLException | H2ClientException e) { + logger.error(e.getMessage(), e); + } + return 0; + } + + @Override public String getAddress(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); + Object[] params = new Object[] {addressId}; + try (ResultSet rs = client.executeQuery(sql, params)) { + if (rs.next()) { + return rs.getString(1); + } + } catch (SQLException | H2ClientException e) { + logger.error(e.getMessage(), e); + } + return Const.EMPTY_STRING; + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/NetworkAddressRegisterH2DAO.java b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/NetworkAddressRegisterH2DAO.java new file mode 100644 index 000000000..bc06a691b --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/NetworkAddressRegisterH2DAO.java @@ -0,0 +1,71 @@ +/* + * 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.h2.dao; + +import java.util.HashMap; +import java.util.Map; +import org.apache.skywalking.apm.collector.client.h2.H2Client; +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.INetworkAddressRegisterDAO; +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; + +/** + * @author peng-yongsheng, clevertension + */ +public class NetworkAddressRegisterH2DAO extends H2DAO implements INetworkAddressRegisterDAO { + + private final Logger logger = LoggerFactory.getLogger(NetworkAddressRegisterH2DAO.class); + + public NetworkAddressRegisterH2DAO(H2Client client) { + super(client); + } + + @Override + public int getMaxNetworkAddressId() { + return getMaxId(NetworkAddressTable.TABLE, NetworkAddressTable.COLUMN_ADDRESS_ID); + } + + @Override + public int getMinNetworkAddressId() { + return getMinId(NetworkAddressTable.TABLE, NetworkAddressTable.COLUMN_ADDRESS_ID); + } + + @Override + public void save(NetworkAddress networkAddress) { + H2Client client = getClient(); + + Map source = new HashMap<>(); + source.put(NetworkAddressTable.COLUMN_ID, networkAddress.getId()); + source.put(NetworkAddressTable.COLUMN_NETWORK_ADDRESS, networkAddress.getNetworkAddress()); + source.put(NetworkAddressTable.COLUMN_ADDRESS_ID, networkAddress.getAddressId()); + + String sql = SqlBuilder.buildBatchInsertSql(NetworkAddressTable.TABLE, source.keySet()); + Object[] params = source.values().toArray(new Object[0]); + try { + client.execute(sql, params); + } catch (H2ClientException e) { + logger.error(e.getMessage(), e); + } + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/NetworkAddressH2TableDefine.java b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/NetworkAddressH2TableDefine.java new file mode 100644 index 000000000..4d2c3f9dd --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/NetworkAddressH2TableDefine.java @@ -0,0 +1,39 @@ +/* + * 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.h2.define; + +import org.apache.skywalking.apm.collector.storage.h2.base.define.H2ColumnDefine; +import org.apache.skywalking.apm.collector.storage.h2.base.define.H2TableDefine; +import org.apache.skywalking.apm.collector.storage.table.register.NetworkAddressTable; + +/** + * @author peng-yongsheng + */ +public class NetworkAddressH2TableDefine extends H2TableDefine { + + public NetworkAddressH2TableDefine() { + super(NetworkAddressTable.TABLE); + } + + @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_ADDRESS_ID, H2ColumnDefine.Type.Int.name())); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/resources/META-INF/defines/storage.define b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/resources/META-INF/defines/storage.define index b094d673e..bb7e59145 100644 --- a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/resources/META-INF/defines/storage.define +++ b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/resources/META-INF/defines/storage.define @@ -1,3 +1,4 @@ +org.apache.skywalking.apm.collector.storage.h2.define.NetworkAddressH2TableDefine org.apache.skywalking.apm.collector.storage.h2.define.ApplicationH2TableDefine org.apache.skywalking.apm.collector.storage.h2.define.InstanceH2TableDefine org.apache.skywalking.apm.collector.storage.h2.define.ServiceNameH2TableDefine