Network address register and cache.

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));
     }
 }

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<Instance
 
     @SuppressWarnings("unchecked")
     @Override protected IPersistenceDAO<?, ?, InstanceMapping> persistenceDAO() {
-        return getModuleManager().find(StorageModule.NAME).getService(IApplicationMappingPersistenceDAO.class);
+        return getModuleManager().find(StorageModule.NAME).getService(IInstanceMappingPersistenceDAO.class);
     }
 
     public static class Factory extends PersistenceWorkerProvider<InstanceMapping, InstanceMappingPersistenceWorker> {

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;
 }

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;
 }

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();
     }
 }

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));
+    }
+}

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<NetworkAddress, NetworkAddress> {
+
+    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<NetworkAddress, NetworkAddress, NetworkAddressRegisterRemoteWorker> {
+
+        public Factory(ModuleManager moduleManager, RemoteSenderService remoteSenderService, int graphId) {
+            super(moduleManager, remoteSenderService, graphId);
+        }
+
+        @Override public NetworkAddressRegisterRemoteWorker workerInstance(ModuleManager moduleManager) {
+            return new NetworkAddressRegisterRemoteWorker(moduleManager);
+        }
+    }
+}

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<NetworkAddress, NetworkAddress> {
+
+    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<NetworkAddress, NetworkAddress, NetworkAddressRegisterSerialWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public NetworkAddressRegisterSerialWorker workerInstance(ModuleManager moduleManager) {
+            return new NetworkAddressRegisterSerialWorker(moduleManager);
+        }
+
+        @Override public int queueSize() {
+            return 256;
+        }
+    }
+}

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};
     }
 }

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);
+}

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 {

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<String, Integer> 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<Integer, String> 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;
+    }
+}

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<Class> classes) {

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);
+}

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);
+}

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);
+    }
+}

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";
+}

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));

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;
+    }
+}

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<String, Object> 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());
+    }
+}

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()));
+    }
+}

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

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));

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;
+    }
+}

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<String, Object> 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);
+        }
+    }
+}

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()));
+    }
+}

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