add dao

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/base/dao/IPersistenceDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.base.dao;
+
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IPersistenceDAO<I, U> {
+    Data get(String id, DataDefine dataDefine);
+
+    I prepareBatchInsert(Data data);
+
+    U prepareBatchUpdate(Data data);
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IApplicationDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.dao;
+
+import org.skywalking.apm.collector.core.data.Data;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IApplicationDAO {
+    int getMaxApplicationId();
+
+    int getMinApplicationId();
+
+    void save(Data data);
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/ICpuMetricDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.dao;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface ICpuMetricDAO {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IGCMetricDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.dao;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IGCMetricDAO {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IInstanceHeartBeatDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.dao;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IInstanceHeartBeatDAO {
+}
\ No newline at end of file

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IMemoryMetricDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.dao;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IMemoryMetricDAO {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IMemoryPoolMetricDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.dao;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IMemoryPoolMetricDAO {
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/ApplicationEsDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.es.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.support.WriteRequest;
+import org.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.dao.IApplicationDAO;
+import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
+import org.skywalking.apm.collector.storage.table.register.ApplicationTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng
+ */
+public class ApplicationEsDAO extends EsDAO implements IApplicationDAO {
+
+    private final Logger logger = LoggerFactory.getLogger(ApplicationEsDAO.class);
+
+    @Override public int getMaxApplicationId() {
+        return getMaxId(ApplicationTable.TABLE, ApplicationTable.COLUMN_APPLICATION_ID);
+    }
+
+    @Override public int getMinApplicationId() {
+        return getMinId(ApplicationTable.TABLE, ApplicationTable.COLUMN_APPLICATION_ID);
+    }
+
+    @Override public void save(Data data) {
+        logger.debug("save application register info, application id: {}, application code: {}", application.getApplicationId(), application.getApplicationCode());
+        ElasticSearchClient client = getClient();
+        Map<String, Object> source = new HashMap<>();
+        source.put(ApplicationTable.COLUMN_APPLICATION_CODE, application.getApplicationCode());
+        source.put(ApplicationTable.COLUMN_APPLICATION_ID, application.getApplicationId());
+
+        IndexResponse response = client.prepareIndex(ApplicationTable.TABLE, application.getId()).setSource(source).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
+        logger.debug("save application register info, application id: {}, application code: {}, status: {}", application.getApplicationId(), application.getApplicationCode(), response.status().name());
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/CpuMetricEsDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.es.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.ICpuMetricDAO;
+import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
+import org.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuMetricEsDAO extends EsDAO implements ICpuMetricDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder> {
+
+    private final Logger logger = LoggerFactory.getLogger(CpuMetricEsDAO.class);
+
+    @Override public Data get(String id, DataDefine dataDefine) {
+        return null;
+    }
+
+    @Override public IndexRequestBuilder prepareBatchInsert(Data data) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(CpuMetricTable.COLUMN_INSTANCE_ID, data.getDataInteger(0));
+        source.put(CpuMetricTable.COLUMN_USAGE_PERCENT, data.getDataDouble(0));
+        source.put(CpuMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(0));
+
+        logger.debug("prepare cpu metric batch insert, id: {}", data.getDataString(0));
+        return getClient().prepareIndex(CpuMetricTable.TABLE, data.getDataString(0)).setSource(source);
+    }
+
+    @Override public UpdateRequestBuilder prepareBatchUpdate(Data data) {
+        return null;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/GCMetricEsDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.es.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.IGCMetricDAO;
+import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
+import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class GCMetricEsDAO extends EsDAO implements IGCMetricDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder> {
+
+    @Override public Data get(String id, DataDefine dataDefine) {
+        return null;
+    }
+
+    @Override public IndexRequestBuilder prepareBatchInsert(Data data) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(GCMetricTable.COLUMN_INSTANCE_ID, data.getDataInteger(0));
+        source.put(GCMetricTable.COLUMN_PHRASE, data.getDataInteger(1));
+        source.put(GCMetricTable.COLUMN_COUNT, data.getDataLong(0));
+        source.put(GCMetricTable.COLUMN_TIME, data.getDataLong(1));
+        source.put(GCMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(2));
+
+        return getClient().prepareIndex(GCMetricTable.TABLE, data.getDataString(0)).setSource(source);
+    }
+
+    @Override public UpdateRequestBuilder prepareBatchUpdate(Data data) {
+        return null;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/InstanceHeartBeatEsDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.es.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+import org.skywalking.apm.collector.core.UnexpectedException;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.IInstanceHeartBeatDAO;
+import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
+import org.skywalking.apm.collector.storage.table.register.InstanceTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng
+ */
+public class InstanceHeartBeatEsDAO extends EsDAO implements IInstanceHeartBeatDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder> {
+
+    private final Logger logger = LoggerFactory.getLogger(InstanceHeartBeatEsDAO.class);
+
+    @Override public Data get(String id, DataDefine dataDefine) {
+        GetResponse getResponse = getClient().prepareGet(InstanceTable.TABLE, id).get();
+        if (getResponse.isExists()) {
+            Data data = dataDefine.build(id);
+            Map<String, Object> source = getResponse.getSource();
+            data.setDataInteger(0, (Integer)source.get(InstanceTable.COLUMN_INSTANCE_ID));
+            data.setDataLong(0, (Long)source.get(InstanceTable.COLUMN_HEARTBEAT_TIME));
+            logger.debug("id: {} is exists", id);
+            return data;
+        } else {
+            logger.debug("id: {} is not exists", id);
+            return null;
+        }
+    }
+
+    @Override public IndexRequestBuilder prepareBatchInsert(Data data) {
+        throw new UnexpectedException("There is no need to merge stream data with database data.");
+    }
+
+    @Override public UpdateRequestBuilder prepareBatchUpdate(Data data) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(InstanceTable.COLUMN_HEARTBEAT_TIME, data.getDataLong(0));
+        return getClient().prepareUpdate(InstanceTable.TABLE, data.getDataString(0)).setDoc(source);
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/MemoryMetricEsDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.es.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.IMemoryMetricDAO;
+import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
+import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMetricEsDAO extends EsDAO implements IMemoryMetricDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder> {
+
+    @Override public Data get(String id, DataDefine dataDefine) {
+        return null;
+    }
+
+    @Override public IndexRequestBuilder prepareBatchInsert(Data data) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(MemoryMetricTable.COLUMN_APPLICATION_INSTANCE_ID, data.getDataInteger(0));
+        source.put(MemoryMetricTable.COLUMN_IS_HEAP, data.getDataBoolean(0));
+        source.put(MemoryMetricTable.COLUMN_INIT, data.getDataLong(0));
+        source.put(MemoryMetricTable.COLUMN_MAX, data.getDataLong(1));
+        source.put(MemoryMetricTable.COLUMN_USED, data.getDataLong(2));
+        source.put(MemoryMetricTable.COLUMN_COMMITTED, data.getDataLong(3));
+        source.put(MemoryMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(4));
+
+        return getClient().prepareIndex(MemoryMetricTable.TABLE, data.getDataString(0)).setSource(source);
+    }
+
+    @Override public UpdateRequestBuilder prepareBatchUpdate(Data data) {
+        return null;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/MemoryPoolMetricEsDAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.es.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.IMemoryPoolMetricDAO;
+import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
+import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryPoolMetricEsDAO extends EsDAO implements IMemoryPoolMetricDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder> {
+
+    @Override public Data get(String id, DataDefine dataDefine) {
+        return null;
+    }
+
+    @Override public IndexRequestBuilder prepareBatchInsert(Data data) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(MemoryPoolMetricTable.COLUMN_INSTANCE_ID, data.getDataInteger(0));
+        source.put(MemoryPoolMetricTable.COLUMN_POOL_TYPE, data.getDataInteger(1));
+        source.put(MemoryPoolMetricTable.COLUMN_INIT, data.getDataLong(0));
+        source.put(MemoryPoolMetricTable.COLUMN_MAX, data.getDataLong(1));
+        source.put(MemoryPoolMetricTable.COLUMN_USED, data.getDataLong(2));
+        source.put(MemoryPoolMetricTable.COLUMN_COMMITTED, data.getDataLong(3));
+        source.put(MemoryPoolMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(4));
+
+        return getClient().prepareIndex(MemoryPoolMetricTable.TABLE, data.getDataString(0)).setSource(source);
+    }
+
+    @Override public UpdateRequestBuilder prepareBatchUpdate(Data data) {
+        return null;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/resources/META-INF/defines/es_dao.define

+org.skywalking.apm.collector.storage.es.dao.CpuMetricEsDAO
+org.skywalking.apm.collector.storage.es.dao.GCMetricEsDAO
+org.skywalking.apm.collector.storage.es.dao.InstanceHeartBeatEsDAO
+org.skywalking.apm.collector.storage.es.dao.MemoryMetricEsDAO
+org.skywalking.apm.collector.storage.es.dao.MemoryPoolMetricEsDAO
\ No newline at end of file

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/ApplicationH2DAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.h2.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.skywalking.apm.collector.client.h2.H2Client;
+import org.skywalking.apm.collector.client.h2.H2ClientException;
+import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
+import org.skywalking.apm.collector.storage.define.register.ApplicationTable;
+import org.skywalking.apm.collector.storage.h2.SqlBuilder;
+import org.skywalking.apm.collector.storage.h2.dao.H2DAO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng, clevertension
+ */
+public class ApplicationH2DAO extends H2DAO implements IApplicationDAO {
+    private final Logger logger = LoggerFactory.getLogger(ApplicationH2DAO.class);
+
+    @Override
+    public int getMaxApplicationId() {
+        return getMaxId(ApplicationTable.TABLE, ApplicationTable.COLUMN_APPLICATION_ID);
+    }
+
+    @Override
+    public int getMinApplicationId() {
+        return getMinId(ApplicationTable.TABLE, ApplicationTable.COLUMN_APPLICATION_ID);
+    }
+
+    @Override
+    public void save(ApplicationDataDefine.Application application) {
+        H2Client client = getClient();
+
+        Map<String, Object> source = new HashMap<>();
+        source.put(ApplicationTable.COLUMN_ID, application.getApplicationId());
+        source.put(ApplicationTable.COLUMN_APPLICATION_CODE, application.getApplicationCode());
+        source.put(ApplicationTable.COLUMN_APPLICATION_ID, application.getApplicationId());
+
+        String sql = SqlBuilder.buildBatchInsertSql(ApplicationTable.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/skywalking/apm/collector/storage/h2/dao/CpuMetricH2DAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.h2.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.ICpuMetricDAO;
+import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
+import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.skywalking.apm.collector.storage.sql.SqlBuilder;
+import org.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng, clevertension
+ */
+public class CpuMetricH2DAO extends H2DAO implements ICpuMetricDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity> {
+    private final Logger logger = LoggerFactory.getLogger(CpuMetricH2DAO.class);
+
+    @Override public Data get(String id, DataDefine dataDefine) {
+        return null;
+    }
+
+    @Override public H2SqlEntity prepareBatchInsert(Data data) {
+        H2SqlEntity entity = new H2SqlEntity();
+        Map<String, Object> source = new HashMap<>();
+        source.put(CpuMetricTable.COLUMN_ID, data.getDataString(0));
+        source.put(CpuMetricTable.COLUMN_INSTANCE_ID, data.getDataInteger(0));
+        source.put(CpuMetricTable.COLUMN_USAGE_PERCENT, data.getDataDouble(0));
+        source.put(CpuMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(0));
+
+        logger.debug("prepare cpu metric batch insert, id: {}", data.getDataString(0));
+        String sql = SqlBuilder.buildBatchInsertSql(CpuMetricTable.TABLE, source.keySet());
+        entity.setSql(sql);
+        entity.setParams(source.values().toArray(new Object[0]));
+        return entity;
+    }
+
+    @Override public H2SqlEntity prepareBatchUpdate(Data data) {
+        return null;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/GCMetricH2DAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.h2.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.IGCMetricDAO;
+import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
+import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.skywalking.apm.collector.storage.sql.SqlBuilder;
+import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable;
+
+/**
+ * @author peng-yongsheng, clevertension
+ */
+public class GCMetricH2DAO extends H2DAO implements IGCMetricDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity> {
+    @Override public Data get(String id, DataDefine dataDefine) {
+        return null;
+    }
+
+    @Override public H2SqlEntity prepareBatchInsert(Data data) {
+        H2SqlEntity entity = new H2SqlEntity();
+        Map<String, Object> source = new HashMap<>();
+        source.put(GCMetricTable.COLUMN_ID, data.getDataString(0));
+        source.put(GCMetricTable.COLUMN_INSTANCE_ID, data.getDataInteger(0));
+        source.put(GCMetricTable.COLUMN_PHRASE, data.getDataInteger(1));
+        source.put(GCMetricTable.COLUMN_COUNT, data.getDataLong(0));
+        source.put(GCMetricTable.COLUMN_TIME, data.getDataLong(1));
+        source.put(GCMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(2));
+
+        String sql = SqlBuilder.buildBatchInsertSql(GCMetricTable.TABLE, source.keySet());
+        entity.setSql(sql);
+        entity.setParams(source.values().toArray(new Object[0]));
+        return entity;
+    }
+
+    @Override public H2SqlEntity prepareBatchUpdate(Data data) {
+        return null;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/InstanceHeartBeatH2DAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.h2.dao;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.skywalking.apm.collector.client.h2.H2Client;
+import org.skywalking.apm.collector.client.h2.H2ClientException;
+import org.skywalking.apm.collector.core.UnexpectedException;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.IInstanceHeartBeatDAO;
+import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
+import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.skywalking.apm.collector.storage.sql.SqlBuilder;
+import org.skywalking.apm.collector.storage.table.register.InstanceTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng, clevertension
+ */
+public class InstanceHeartBeatH2DAO extends H2DAO implements IInstanceHeartBeatDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity> {
+
+    private final Logger logger = LoggerFactory.getLogger(InstanceHeartBeatH2DAO.class);
+
+    private static final String GET_INSTANCE_HEARTBEAT_SQL = "select * from {0} where {1} = ?";
+
+    @Override public Data get(String id, DataDefine dataDefine) {
+        H2Client client = getClient();
+        String sql = SqlBuilder.buildSql(GET_INSTANCE_HEARTBEAT_SQL, InstanceTable.TABLE, InstanceTable.COLUMN_INSTANCE_ID);
+        Object[] params = new Object[] {id};
+        try (ResultSet rs = client.executeQuery(sql, params)) {
+            if (rs.next()) {
+                Data data = dataDefine.build(id);
+                data.setDataInteger(0, rs.getInt(InstanceTable.COLUMN_INSTANCE_ID));
+                data.setDataLong(0, rs.getLong(InstanceTable.COLUMN_HEARTBEAT_TIME));
+                return data;
+            }
+        } catch (SQLException | H2ClientException e) {
+            logger.error(e.getMessage(), e);
+        }
+        return null;
+    }
+
+    @Override public H2SqlEntity prepareBatchInsert(Data data) {
+        throw new UnexpectedException("There is no need to merge stream data with database data.");
+    }
+
+    @Override public H2SqlEntity prepareBatchUpdate(Data data) {
+        H2SqlEntity entity = new H2SqlEntity();
+        Map<String, Object> source = new HashMap<>();
+        source.put(InstanceTable.COLUMN_HEARTBEAT_TIME, data.getDataLong(0));
+        String sql = SqlBuilder.buildBatchUpdateSql(InstanceTable.TABLE, source.keySet(), InstanceTable.COLUMN_INSTANCE_ID);
+        entity.setSql(sql);
+        List<Object> params = new ArrayList<>(source.values());
+        params.add(data.getDataString(0));
+        entity.setParams(params.toArray(new Object[0]));
+        return entity;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/MemoryMetricH2DAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.h2.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.IMemoryMetricDAO;
+import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
+import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.skywalking.apm.collector.storage.sql.SqlBuilder;
+import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
+
+/**
+ * @author peng-yongsheng, clevertension
+ */
+public class MemoryMetricH2DAO extends H2DAO implements IMemoryMetricDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity> {
+    @Override public Data get(String id, DataDefine dataDefine) {
+        return null;
+    }
+
+    @Override public H2SqlEntity prepareBatchInsert(Data data) {
+        H2SqlEntity entity = new H2SqlEntity();
+        Map<String, Object> source = new HashMap<>();
+        source.put(MemoryMetricTable.COLUMN_ID, data.getDataString(0));
+        source.put(MemoryMetricTable.COLUMN_APPLICATION_INSTANCE_ID, data.getDataInteger(0));
+        source.put(MemoryMetricTable.COLUMN_IS_HEAP, data.getDataBoolean(0));
+        source.put(MemoryMetricTable.COLUMN_INIT, data.getDataLong(0));
+        source.put(MemoryMetricTable.COLUMN_MAX, data.getDataLong(1));
+        source.put(MemoryMetricTable.COLUMN_USED, data.getDataLong(2));
+        source.put(MemoryMetricTable.COLUMN_COMMITTED, data.getDataLong(3));
+        source.put(MemoryMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(4));
+
+        String sql = SqlBuilder.buildBatchInsertSql(MemoryMetricTable.TABLE, source.keySet());
+        entity.setSql(sql);
+        entity.setParams(source.values().toArray(new Object[0]));
+        return entity;
+    }
+
+    @Override public H2SqlEntity prepareBatchUpdate(Data data) {
+        return null;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/MemoryPoolMetricH2DAO.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.storage.h2.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.base.define.DataDefine;
+import org.skywalking.apm.collector.storage.dao.IMemoryPoolMetricDAO;
+import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
+import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.skywalking.apm.collector.storage.sql.SqlBuilder;
+import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable;
+
+/**
+ * @author peng-yongsheng, clevertension
+ */
+public class MemoryPoolMetricH2DAO extends H2DAO implements IMemoryPoolMetricDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity> {
+    @Override public Data get(String id, DataDefine dataDefine) {
+        return null;
+    }
+
+    @Override public H2SqlEntity prepareBatchInsert(Data data) {
+        H2SqlEntity entity = new H2SqlEntity();
+        Map<String, Object> source = new HashMap<>();
+        source.put(MemoryPoolMetricTable.COLUMN_ID, data.getDataString(0));
+        source.put(MemoryPoolMetricTable.COLUMN_INSTANCE_ID, data.getDataInteger(0));
+        source.put(MemoryPoolMetricTable.COLUMN_POOL_TYPE, data.getDataInteger(1));
+        source.put(MemoryPoolMetricTable.COLUMN_INIT, data.getDataLong(0));
+        source.put(MemoryPoolMetricTable.COLUMN_MAX, data.getDataLong(1));
+        source.put(MemoryPoolMetricTable.COLUMN_USED, data.getDataLong(2));
+        source.put(MemoryPoolMetricTable.COLUMN_COMMITTED, data.getDataLong(3));
+        source.put(MemoryPoolMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(4));
+
+        String sql = SqlBuilder.buildBatchInsertSql(MemoryPoolMetricTable.TABLE, source.keySet());
+        entity.setSql(sql);
+        entity.setParams(source.values().toArray(new Object[0]));
+        return entity;
+    }
+
+    @Override public H2SqlEntity prepareBatchUpdate(Data data) {
+        return null;
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/resources/META-INF/defines/h2_dao.define

+org.skywalking.apm.collector.storage.h2.dao.CpuMetricH2DAO
+org.skywalking.apm.collector.storage.h2.dao.GCMetricH2DAO
+org.skywalking.apm.collector.storage.h2.dao.InstanceHeartBeatH2DAO
+org.skywalking.apm.collector.storage.h2.dao.MemoryMetricH2DAO
+org.skywalking.apm.collector.storage.h2.dao.MemoryPoolMetricH2DAO
\ No newline at end of file