From 215eab74288328e13f2c0fecf222168632685ff9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E6=99=9F=20Wu=20Sheng?= Date: Wed, 28 Nov 2018 22:59:02 +0800 Subject: [PATCH] Mysql style storage implementation (#1964) Also fixed several bugs in this merge. --- docs/en/guides/storage-extention.md | 38 +++- docs/en/setup/backend/backend-storage.md | 16 ++ .../oap/server/core/CoreModule.java | 2 + .../oap/server/core/CoreModuleProvider.java | 2 + .../oap/server/core/alarm/AlarmRecord.java | 8 +- .../core/alarm/AlarmStandardPersistence.java | 2 +- .../manual/segment/SegmentRecord.java | 2 +- .../oap/server/core/query/DurationUtils.java | 132 ++++++------ .../annotation/StorageAnnotationListener.java | 18 +- .../server/core/storage/model/ColumnName.java | 21 +- .../core/storage/model/IModelOverride.java | 30 +++ .../core/storage/model/ModelInstaller.java | 17 +- .../core/storage/ttl/DataTTLKeeperTimer.java | 42 ++-- .../jdbc/hikaricp/JDBCHikariCPClient.java | 69 +++++-- .../query/graphql/resolver/AlarmQuery.java | 4 +- .../query/graphql/resolver/MetadataQuery.java | 16 +- .../query/graphql/resolver/TopologyQuery.java | 4 +- .../trace/provider/TraceModuleProvider.java | 2 +- .../listener/segment/SegmentSpanListener.java | 2 +- .../receiver/trace/mock/AgentDataMock.java | 6 +- .../receiver/trace/mock/ServiceBMock.java | 3 +- .../receiver/trace/mock/ServiceCMock.java | 2 +- .../assembly/datasource-settings.properties | 31 +++ .../src/main/resources/application.yml | 1 + .../resources/datasource-settings.properties | 31 +++ .../base/HistoryDeleteEsDAO.java | 6 +- .../storage-jdbc-hikaricp-plugin/pom.xml | 5 + .../storage/plugin/jdbc/SQLExecutor.java | 6 + .../plugin/jdbc/h2/H2StorageProvider.java | 55 +++-- .../jdbc/h2/dao/H2AggregationQueryDAO.java | 35 ++-- .../plugin/jdbc/h2/dao/H2AlarmQueryDAO.java | 67 +++++- .../jdbc/h2/dao/H2HistoryDeleteDAO.java | 17 ++ .../jdbc/h2/dao/H2MetadataQueryDAO.java | 156 ++++++-------- .../plugin/jdbc/h2/dao/H2MetricQueryDAO.java | 18 +- .../plugin/jdbc/h2/dao/H2RegisterDAO.java | 17 +- .../plugin/jdbc/h2/dao/H2SQLExecutor.java | 22 +- .../h2/dao/H2ServiceInventoryCacheDAO.java | 8 +- .../plugin/jdbc/h2/dao/H2TableInstaller.java | 16 +- .../jdbc/h2/dao/H2TopologyQueryDAO.java | 28 +-- .../plugin/jdbc/h2/dao/H2TraceQueryDAO.java | 25 +-- .../jdbc/mysql/MySQLAggregationQueryDAO.java | 79 +++++++ .../plugin/jdbc/mysql/MySQLAlarmQueryDAO.java | 98 +++++++++ .../mysql/MySQLRegisterLockInstaller.java | 97 +++++++++ .../jdbc/mysql/MySQLRegisterTableLockDAO.java | 85 ++++++++ .../jdbc/mysql/MySQLStorageProvider.java | 143 +++++++++++++ .../jdbc/mysql/MySQLTableInstaller.java | 193 ++++++++++++++++++ .../plugin/jdbc/mysql/MySQLTraceQueryDAO.java | 138 +++++++++++++ ...g.oap.server.library.module.ModuleProvider | 3 +- ...PreventRedistributionMySQLDriverTest.java} | 30 +-- 49 files changed, 1487 insertions(+), 361 deletions(-) create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/IModelOverride.java create mode 100644 oap-server/server-starter/src/main/assembly/datasource-settings.properties create mode 100644 oap-server/server-starter/src/main/resources/datasource-settings.properties create mode 100644 oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLAggregationQueryDAO.java create mode 100644 oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLAlarmQueryDAO.java create mode 100644 oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLRegisterLockInstaller.java create mode 100644 oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLRegisterTableLockDAO.java create mode 100644 oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java create mode 100644 oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java create mode 100644 oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTraceQueryDAO.java rename oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/{main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2RegisterLockInstaller.java => test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/PreventRedistributionMySQLDriverTest.java} (51%) diff --git a/docs/en/guides/storage-extention.md b/docs/en/guides/storage-extention.md index 66c6fbf3d..a88e34b3d 100644 --- a/docs/en/guides/storage-extention.md +++ b/docs/en/guides/storage-extention.md @@ -1,3 +1,37 @@ # Extend storage -SkyWalking has already provided H2 and ElasticSearch as storage solutions. In this document, you could -learn how to implement a new storage easily. \ No newline at end of file +SkyWalking has already provided several storage solutions. In this document, you could +learn how to implement a new storage easily. + +## Define your storage provider +1. Define a class extends `org.apache.skywalking.oap.server.library.module.ModuleProvider`. +2. Set this provider targeting to Storage module. +```java +@Override +public Class module() { + return StorageModule.class; +} +``` + +## Implement all DAOs +Here is the list of all DAO interfaces in storage +1. IServiceInventoryCacheDAO +1. IServiceInstanceInventoryCacheDAO +1. H2EndpointInventoryCacheDAO +1. H2NetworkAddressInventoryCacheDAO +1. IBatchDAO +1. StorageDAO +1. IRegisterLockDAO +1. H2TopologyQueryDAO +1. IMetricQueryDAO +1. ITraceQueryDAO +1. IMetadataQueryDAO +1. IAggregationQueryDAO +1. IAlarmQueryDAO +1. IHistoryDeleteDAO + +## Register all service implementations +In `public void prepare()`, use `this#registerServiceImplementation` method to do register binding your implementation with the above interfaces. + +## Example +Take `org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchProvider` +or `org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql.MySQLStorageProvider` as a good example. \ No newline at end of file diff --git a/docs/en/setup/backend/backend-storage.md b/docs/en/setup/backend/backend-storage.md index e7de556c2..01321daf4 100644 --- a/docs/en/setup/backend/backend-storage.md +++ b/docs/en/setup/backend/backend-storage.md @@ -4,6 +4,7 @@ use is by changing the `application.yml` - [**H2**](#h2) - [**ElasticSearch 6**](#elasticsearch-6) +- [**MySQL**](#mysql) ## H2 Active H2 as storage, set storage provider to **H2** In-Memory Databases. Default in distribution package. @@ -44,6 +45,21 @@ storage: monthMetricDataTTL: 18 # Unit is month ``` +## MySQL +Active MySQL as storage, set storage provider to **mysql**. + +**NOTICE:** MySQL driver is NOT allowed in Apache official distribution and source codes. +Please download MySQL driver by yourself. Copy the connection driver jar to `oap-libs`. + +```yaml +storage: + mysql: +``` + +All connection related settings including link url, username and password +are in `databsource-settings.properties`. +This setting file follow [HikariCP](https://github.com/brettwooldridge/HikariCP) connection pool document. + ## More storage solution extension Follow [Storage extension development guide](../../guides/storage-extention.md) diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java index dade6d0a7..1baa2507d 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModule.java @@ -29,6 +29,7 @@ import org.apache.skywalking.oap.server.core.remote.client.RemoteClientManager; import org.apache.skywalking.oap.server.core.server.*; import org.apache.skywalking.oap.server.core.source.SourceReceiver; import org.apache.skywalking.oap.server.core.storage.model.IModelGetter; +import org.apache.skywalking.oap.server.core.storage.model.IModelOverride; import org.apache.skywalking.oap.server.library.module.*; /** @@ -73,6 +74,7 @@ public class CoreModule extends ModuleDefine { private void addInsideService(List classes) { classes.add(IModelGetter.class); + classes.add(IModelOverride.class); classes.add(StreamDataClassGetter.class); classes.add(RemoteClientManager.class); classes.add(RemoteSenderService.class); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java index f0e908499..20ea0bcbf 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java @@ -36,6 +36,7 @@ import org.apache.skywalking.oap.server.core.source.*; import org.apache.skywalking.oap.server.core.storage.PersistenceTimer; import org.apache.skywalking.oap.server.core.storage.annotation.StorageAnnotationListener; import org.apache.skywalking.oap.server.core.storage.model.IModelGetter; +import org.apache.skywalking.oap.server.core.storage.model.IModelOverride; import org.apache.skywalking.oap.server.core.storage.ttl.DataTTLKeeperTimer; import org.apache.skywalking.oap.server.library.module.*; import org.apache.skywalking.oap.server.library.server.ServerException; @@ -106,6 +107,7 @@ public class CoreModuleProvider extends ModuleProvider { this.registerServiceImplementation(RemoteSenderService.class, new RemoteSenderService(getManager())); this.registerServiceImplementation(IModelGetter.class, storageAnnotationListener); + this.registerServiceImplementation(IModelOverride.class, storageAnnotationListener); this.registerServiceImplementation(ServiceInventoryCache.class, new ServiceInventoryCache(getManager())); this.registerServiceImplementation(IServiceInventoryRegister.class, new ServiceInventoryRegister(getManager())); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmRecord.java index 90dc8ddc8..9257829a2 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmRecord.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmRecord.java @@ -38,10 +38,10 @@ public class AlarmRecord extends Record { public static final String INDEX_NAME = "alarm_record"; public static final String SCOPE = "scope"; - private static final String NAME = "name"; - private static final String ID0 = "id0"; - private static final String ID1 = "id1"; - private static final String START_TIME = "start_time"; + public static final String NAME = "name"; + public static final String ID0 = "id0"; + public static final String ID1 = "id1"; + public static final String START_TIME = "start_time"; public static final String ALARM_MESSAGE = "alarm_message"; @Override public String id() { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmStandardPersistence.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmStandardPersistence.java index 556dfc507..2fec9ae23 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmStandardPersistence.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmStandardPersistence.java @@ -45,7 +45,7 @@ public class AlarmStandardPersistence implements AlarmCallback { record.setName(message.getName()); record.setAlarmMessage(message.getAlarmMessage()); record.setStartTime(message.getStartTime()); - record.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(message.getStartTime())); + record.setTimeBucket(TimeBucketUtils.INSTANCE.getSecondTimeBucket(message.getStartTime())); RecordProcess.INSTANCE.in(record); }); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java index b85de76fd..40a02cf81 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/segment/SegmentRecord.java @@ -38,7 +38,7 @@ import org.apache.skywalking.oap.server.library.util.CollectionUtils; * @author peng-yongsheng */ @RecordType -@StorageEntity(name = SegmentRecord.INDEX_NAME, builder = SegmentRecord.Builder.class, deleteHistory = false, source = Scope.Segment) +@StorageEntity(name = SegmentRecord.INDEX_NAME, builder = SegmentRecord.Builder.class, source = Scope.Segment) public class SegmentRecord extends Record { public static final String INDEX_NAME = "segment"; diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/DurationUtils.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/DurationUtils.java index 999bbc1b0..196d5ddd6 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/DurationUtils.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/DurationUtils.java @@ -80,6 +80,38 @@ public enum DurationUtils { return secondTimeBucket; } + public long startTimeToTimestamp(Step step, String dateStr) throws ParseException { + switch (step) { + case MONTH: + return new SimpleDateFormat("yyyy-MM").parse(dateStr).getTime(); + case DAY: + return new SimpleDateFormat("yyyy-MM-dd").parse(dateStr).getTime(); + case HOUR: + return new SimpleDateFormat("yyyy-MM-dd HH").parse(dateStr).getTime(); + case MINUTE: + return new SimpleDateFormat("yyyy-MM-dd HHmm").parse(dateStr).getTime(); + case SECOND: + return new SimpleDateFormat("yyyy-MM-dd HHmmss").parse(dateStr).getTime(); + } + throw new UnexpectedException("Unsupported step " + step.name()); + } + + public long endTimeToTimestamp(Step step, String dateStr) throws ParseException { + switch (step) { + case MONTH: + return new DateTime(new SimpleDateFormat("yyyy-MM").parse(dateStr)).plusMonths(1).getMillis(); + case DAY: + return new DateTime(new SimpleDateFormat("yyyy-MM-dd").parse(dateStr)).plusDays(1).getMillis(); + case HOUR: + return new DateTime(new SimpleDateFormat("yyyy-MM-dd HH").parse(dateStr)).plusHours(1).getMillis(); + case MINUTE: + return new DateTime(new SimpleDateFormat("yyyy-MM-dd HHmm").parse(dateStr)).plusMinutes(1).getMillis(); + case SECOND: + return new DateTime(new SimpleDateFormat("yyyy-MM-dd HHmmss").parse(dateStr)).plusSeconds(1).getMillis(); + } + throw new UnexpectedException("Unsupported step " + step.name()); + } + public int minutesBetween(Step step, long startTimeBucket, long endTimeBucket) throws ParseException { Date startDate = formatDate(step, startTimeBucket); Date endDate = formatDate(step, endTimeBucket); @@ -128,57 +160,6 @@ public enum DurationUtils { } } - private Date formatDate(Step step, long timeBucket) throws ParseException { - Date date = null; - switch (step) { - case MONTH: - date = new SimpleDateFormat("yyyyMM").parse(String.valueOf(timeBucket)); - break; - case DAY: - date = new SimpleDateFormat("yyyyMMdd").parse(String.valueOf(timeBucket)); - break; - case HOUR: - date = new SimpleDateFormat("yyyyMMddHH").parse(String.valueOf(timeBucket)); - break; - case MINUTE: - date = new SimpleDateFormat("yyyyMMddHHmm").parse(String.valueOf(timeBucket)); - break; - case SECOND: - date = new SimpleDateFormat("yyyyMMddHHmmss").parse(String.valueOf(timeBucket)); - break; - } - return date; - } - - public DateTime parseToDateTime(Step step, long time) throws ParseException { - DateTime dateTime = null; - - switch (step) { - case MONTH: - Date date = new SimpleDateFormat("yyyyMM").parse(String.valueOf(time)); - dateTime = new DateTime(date); - break; - case DAY: - date = new SimpleDateFormat("yyyyMMdd").parse(String.valueOf(time)); - dateTime = new DateTime(date); - break; - case HOUR: - date = new SimpleDateFormat("yyyyMMddHH").parse(String.valueOf(time)); - dateTime = new DateTime(date); - break; - case MINUTE: - date = new SimpleDateFormat("yyyyMMddHHmm").parse(String.valueOf(time)); - dateTime = new DateTime(date); - break; - case SECOND: - date = new SimpleDateFormat("yyyyMMddHHmmss").parse(String.valueOf(time)); - dateTime = new DateTime(date); - break; - } - - return dateTime; - } - public List getDurationPoints(Step step, long startTimeBucket, long endTimeBucket) throws ParseException { DateTime dateTime = parseToDateTime(step, startTimeBucket); @@ -225,19 +206,54 @@ public enum DurationUtils { return durations; } - public long toTimestamp(Step step, String dateStr) throws ParseException { + private Date formatDate(Step step, long timeBucket) throws ParseException { + Date date = null; switch (step) { case MONTH: - return new SimpleDateFormat("yyyy-MM").parse(dateStr).getTime(); + date = new SimpleDateFormat("yyyyMM").parse(String.valueOf(timeBucket)); + break; case DAY: - return new SimpleDateFormat("yyyy-MM-dd").parse(dateStr).getTime(); + date = new SimpleDateFormat("yyyyMMdd").parse(String.valueOf(timeBucket)); + break; case HOUR: - return new SimpleDateFormat("yyyy-MM-dd HH").parse(dateStr).getTime(); + date = new SimpleDateFormat("yyyyMMddHH").parse(String.valueOf(timeBucket)); + break; case MINUTE: - return new SimpleDateFormat("yyyy-MM-dd HHmm").parse(dateStr).getTime(); + date = new SimpleDateFormat("yyyyMMddHHmm").parse(String.valueOf(timeBucket)); + break; case SECOND: - return new SimpleDateFormat("yyyy-MM-dd HHmmss").parse(dateStr).getTime(); + date = new SimpleDateFormat("yyyyMMddHHmmss").parse(String.valueOf(timeBucket)); + break; } - throw new UnexpectedException("Unsupported step " + step.name()); + return date; + } + + private DateTime parseToDateTime(Step step, long time) throws ParseException { + DateTime dateTime = null; + + switch (step) { + case MONTH: + Date date = new SimpleDateFormat("yyyyMM").parse(String.valueOf(time)); + dateTime = new DateTime(date); + break; + case DAY: + date = new SimpleDateFormat("yyyyMMdd").parse(String.valueOf(time)); + dateTime = new DateTime(date); + break; + case HOUR: + date = new SimpleDateFormat("yyyyMMddHH").parse(String.valueOf(time)); + dateTime = new DateTime(date); + break; + case MINUTE: + date = new SimpleDateFormat("yyyyMMddHHmm").parse(String.valueOf(time)); + dateTime = new DateTime(date); + break; + case SECOND: + date = new SimpleDateFormat("yyyyMMddHHmmss").parse(String.valueOf(time)); + dateTime = new DateTime(date); + break; + } + + return dateTime; } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/StorageAnnotationListener.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/StorageAnnotationListener.java index e9e09e98f..e31f34721 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/StorageAnnotationListener.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/annotation/StorageAnnotationListener.java @@ -29,6 +29,7 @@ import org.apache.skywalking.oap.server.core.annotation.AnnotationListener; import org.apache.skywalking.oap.server.core.source.Scope; import org.apache.skywalking.oap.server.core.storage.model.ColumnName; import org.apache.skywalking.oap.server.core.storage.model.IModelGetter; +import org.apache.skywalking.oap.server.core.storage.model.IModelOverride; import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.core.storage.model.ModelColumn; import org.slf4j.Logger; @@ -37,7 +38,7 @@ import org.slf4j.LoggerFactory; /** * @author peng-yongsheng */ -public class StorageAnnotationListener implements AnnotationListener, IModelGetter { +public class StorageAnnotationListener implements AnnotationListener, IModelGetter, IModelOverride { private static final Logger logger = LoggerFactory.getLogger(StorageAnnotationListener.class); @@ -70,7 +71,7 @@ public class StorageAnnotationListener implements AnnotationListener, IModelGett for (Field field : fields) { if (field.isAnnotationPresent(Column.class)) { Column column = field.getAnnotation(Column.class); - modelColumns.add(new ModelColumn(new ColumnName(column.columnName(), column.columnName()), field.getType(), column.matchQuery())); + modelColumns.add(new ModelColumn(new ColumnName(column.columnName()), field.getType(), column.matchQuery())); if (logger.isDebugEnabled()) { logger.debug("The field named {} with the {} type", column.columnName(), field.getType()); } @@ -84,4 +85,17 @@ public class StorageAnnotationListener implements AnnotationListener, IModelGett retrieval(clazz.getSuperclass(), modelName, modelColumns); } } + + @Override public void overrideColumnName(String columnName, String newName) { + models.forEach(model -> { + model.getColumns().forEach(column -> { + ColumnName existColumnName = column.getColumnName(); + String name = existColumnName.getName(); + if (name.equals(columnName)) { + existColumnName.setStorageName(newName); + logger.debug("Model {} column {} has been override. The new column name is {}.", model.getName(), name, newName); + } + }); + }); + } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ColumnName.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ColumnName.java index be7e77564..d2469e68d 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ColumnName.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ColumnName.java @@ -19,23 +19,28 @@ package org.apache.skywalking.oap.server.core.storage.model; /** + * Short column name unsupported for now. No define in @Column annotation. The storage implementation need to use name + * to do match. + * * @author peng-yongsheng */ public class ColumnName { - private final String fullName; - private final String shortName; - private boolean useShortName = false; + private String fullName; + private String storageName = null; - public ColumnName(String fullName, String shortName) { + public ColumnName(String fullName) { this.fullName = fullName; - this.shortName = shortName; } public String getName() { - return useShortName ? shortName : fullName; + return fullName; } - public void useShortName() { - this.useShortName = true; + public String getStorageName() { + return storageName != null ? storageName : fullName; + } + + public void setStorageName(String storageName) { + this.storageName = storageName; } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/IModelOverride.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/IModelOverride.java new file mode 100644 index 000000000..3344fb81a --- /dev/null +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/IModelOverride.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.oap.server.core.storage.model; + +import org.apache.skywalking.oap.server.library.module.Service; + +/** + * Override service provides ways to rename the existing column or table name. + * + * @author wusheng + */ +public interface IModelOverride extends Service { + void overrideColumnName(String columnName, String newName); +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java index 1c3b463ff..4eef35812 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/ModelInstaller.java @@ -18,13 +18,17 @@ package org.apache.skywalking.oap.server.core.storage.model; -import java.util.*; -import org.apache.skywalking.oap.server.core.*; +import java.util.ArrayList; +import java.util.List; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.config.DownsamplingConfigService; -import org.apache.skywalking.oap.server.core.storage.*; +import org.apache.skywalking.oap.server.core.storage.Downsampling; +import org.apache.skywalking.oap.server.core.storage.StorageException; import org.apache.skywalking.oap.server.library.client.Client; import org.apache.skywalking.oap.server.library.module.ModuleManager; -import org.slf4j.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @author peng-yongsheng @@ -75,6 +79,11 @@ public abstract class ModelInstaller { } } + public final void overrideColumnName(String columnName, String newName) { + IModelOverride modelOverride = moduleManager.find(CoreModule.NAME).provider().getService(IModelOverride.class); + modelOverride.overrideColumnName(columnName, newName); + } + protected abstract boolean isExists(Client client, Model model) throws StorageException; protected abstract void columnCheck(Client client, Model model) throws StorageException; diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java index ba3de7a85..575953a77 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java @@ -20,20 +20,29 @@ package org.apache.skywalking.oap.server.core.storage.ttl; import java.io.IOException; import java.util.List; -import java.util.concurrent.*; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import lombok.Setter; import org.apache.skywalking.apm.util.RunnableWithExceptionProtection; -import org.apache.skywalking.oap.server.core.*; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.DataTTL; import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; import org.apache.skywalking.oap.server.core.analysis.record.Record; -import org.apache.skywalking.oap.server.core.cluster.*; +import org.apache.skywalking.oap.server.core.cluster.ClusterModule; +import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery; +import org.apache.skywalking.oap.server.core.cluster.RemoteInstance; import org.apache.skywalking.oap.server.core.config.DownsamplingConfigService; -import org.apache.skywalking.oap.server.core.storage.*; -import org.apache.skywalking.oap.server.core.storage.model.*; +import org.apache.skywalking.oap.server.core.storage.Downsampling; +import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO; +import org.apache.skywalking.oap.server.core.storage.StorageModule; +import org.apache.skywalking.oap.server.core.storage.model.IModelGetter; +import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.joda.time.DateTime; -import org.slf4j.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @author peng-yongsheng @@ -75,19 +84,19 @@ public enum DataTTLKeeperTimer { List models = modelGetter.getModels(); models.forEach(model -> { if (model.isIndicator()) { - execute(model.getName(), timeBuckets.minuteTimeBucketBefore, Indicator.TIME_BUCKET); + execute(model, model.getName(), timeBuckets.minuteTimeBucketBefore, Indicator.TIME_BUCKET); if (downsamplingConfigService.shouldToHour()) { - execute(model.getName() + Const.ID_SPLIT + Downsampling.Hour.getName(), timeBuckets.hourTimeBucketBefore, Indicator.TIME_BUCKET); + execute(model, model.getName() + Const.ID_SPLIT + Downsampling.Hour.getName(), timeBuckets.hourTimeBucketBefore, Indicator.TIME_BUCKET); } if (downsamplingConfigService.shouldToDay()) { - execute(model.getName() + Const.ID_SPLIT + Downsampling.Day.getName(), timeBuckets.dayTimeBucketBefore, Indicator.TIME_BUCKET); + execute(model, model.getName() + Const.ID_SPLIT + Downsampling.Day.getName(), timeBuckets.dayTimeBucketBefore, Indicator.TIME_BUCKET); } if (downsamplingConfigService.shouldToMonth()) { - execute(model.getName() + Const.ID_SPLIT + Downsampling.Month.getName(), timeBuckets.monthTimeBucketBefore, Indicator.TIME_BUCKET); + execute(model, model.getName() + Const.ID_SPLIT + Downsampling.Month.getName(), timeBuckets.monthTimeBucketBefore, Indicator.TIME_BUCKET); } } else { - execute(model.getName(), timeBuckets.recordDataTTL, Record.TIME_BUCKET); + execute(model, model.getName(), timeBuckets.recordDataTTL, Record.TIME_BUCKET); } }); } @@ -95,7 +104,7 @@ public enum DataTTLKeeperTimer { TimeBuckets convertTimeBucket(DateTime currentTime) { TimeBuckets timeBuckets = new TimeBuckets(); - timeBuckets.recordDataTTL = Long.valueOf(currentTime.plusMinutes(0 - dataTTL.getRecordDataTTL()).toString("yyyyMMddHHmm")); + timeBuckets.recordDataTTL = Long.valueOf(currentTime.plusMinutes(0 - dataTTL.getRecordDataTTL()).toString("yyyyMMddHHmmss")); timeBuckets.minuteTimeBucketBefore = Long.valueOf(currentTime.plusMinutes(0 - dataTTL.getMinuteMetricsDataTTL()).toString("yyyyMMddHHmm")); timeBuckets.hourTimeBucketBefore = Long.valueOf(currentTime.plusHours(0 - dataTTL.getHourMetricsDataTTL()).toString("yyyyMMddHH")); timeBuckets.dayTimeBucketBefore = Long.valueOf(currentTime.plusDays(0 - dataTTL.getDayMetricsDataTTL()).toString("yyyyMMdd")); @@ -104,11 +113,14 @@ public enum DataTTLKeeperTimer { return timeBuckets; } - private void execute(String modelName, long timeBucketBefore, String timeBucketColumnName) { + private void execute(Model model, String modelName, long timeBucketBefore, String timeBucketColumnName) { try { - moduleManager.find(StorageModule.NAME).provider().getService(IHistoryDeleteDAO.class).deleteHistory(modelName, timeBucketColumnName, timeBucketBefore); + if (model.isDeleteHistory()) { + moduleManager.find(StorageModule.NAME).provider().getService(IHistoryDeleteDAO.class).deleteHistory(modelName, timeBucketColumnName, timeBucketBefore); + } } catch (IOException e) { - logger.warn("History delete failure, error message: {}", e.getMessage()); + logger.warn("History of {} delete failure, time bucket {}", modelName, timeBucketBefore); + logger.error(e.getMessage(), e); } } diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java index 65a6b18f6..fcdeb119b 100644 --- a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java +++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java @@ -47,41 +47,76 @@ public class JDBCHikariCPClient implements Client { @Override public void shutdown() { } + /** + * Default getConnection is not set in auto-commit. + * + * @return + * @throws JDBCClientException + */ public Connection getConnection() throws JDBCClientException { + return getConnection(true); + } + + public Connection getTransactionConnection() throws JDBCClientException { + return getConnection(false); + } + + public Connection getConnection(boolean autoCommit) throws JDBCClientException { try { Connection connection = dataSource.getConnection(); - connection.setAutoCommit(true); + connection.setAutoCommit(autoCommit); return connection; } catch (SQLException e) { throw new JDBCClientException(e.getMessage(), e); } } - public void close(Connection connection) { - if (connection != null) { - try { - connection.commit(); - connection.close(); - } catch (SQLException e) { - } - } - } - public void execute(Connection connection, String sql) throws JDBCClientException { - try { - connection.setReadOnly(true); - } catch (SQLException e) { - - } logger.debug("execute aql: {}", sql); try (Statement statement = connection.createStatement()) { statement.execute(sql); - statement.closeOnCompletion(); } catch (SQLException e) { throw new JDBCClientException(e.getMessage(), e); } } + public boolean execute(Connection connection, String sql, Object... params) throws JDBCClientException { + logger.debug("execute query with result: {}", sql); + boolean result; + PreparedStatement statement = null; + try { + statement = connection.prepareStatement(sql); + if (params != null) { + for (int i = 0; i < params.length; i++) { + Object param = params[i]; + if (param instanceof String) { + statement.setString(i + 1, (String)param); + } else if (param instanceof Integer) { + statement.setInt(i + 1, (int)param); + } else if (param instanceof Double) { + statement.setDouble(i + 1, (double)param); + } else if (param instanceof Long) { + statement.setLong(i + 1, (long)param); + } else { + throw new JDBCClientException("Unsupported data type, type=" + param.getClass().getName()); + } + } + } + result = statement.execute(); + statement.closeOnCompletion(); + } catch (SQLException e) { + if (statement != null) { + try { + statement.close(); + } catch (SQLException e1) { + } + } + throw new JDBCClientException(e.getMessage(), e); + } + + return result; + } + public ResultSet executeQuery(Connection connection, String sql, Object... params) throws JDBCClientException { logger.debug("execute query with result: {}", sql); ResultSet rs; diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java index 2cfac2366..c69b162c6 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/AlarmQuery.java @@ -52,8 +52,8 @@ public class AlarmQuery implements GraphQLQueryResolver { public Alarms getAlarm(final Duration duration, final Scope scope, final String keyword, final Pagination paging) throws IOException { - long startTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getStart()); - long endTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getEnd()); + long startTimeBucket = DurationUtils.INSTANCE.startTimeDurationToSecondTimeBucket(duration.getStep(), duration.getStart()); + long endTimeBucket = DurationUtils.INSTANCE.endTimeDurationToSecondTimeBucket(duration.getStep(), duration.getEnd()); return getQueryService().getAlarm(scope, keyword, paging, startTimeBucket, endTimeBucket); } diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/MetadataQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/MetadataQuery.java index c7ae17eda..157a639a7 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/MetadataQuery.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/MetadataQuery.java @@ -48,23 +48,23 @@ public class MetadataQuery implements GraphQLQueryResolver { } public ClusterBrief getGlobalBrief(final Duration duration) throws IOException, ParseException { - long startTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getStart()); - long endTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getEnd()); + long startTimestamp = DurationUtils.INSTANCE.startTimeToTimestamp(duration.getStep(), duration.getStart()); + long endTimestamp = DurationUtils.INSTANCE.endTimeToTimestamp(duration.getStep(), duration.getEnd()); return getMetadataQueryService().getGlobalBrief(startTimestamp, endTimestamp); } public List getAllServices(final Duration duration) throws IOException, ParseException { - long startTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getStart()); - long endTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getEnd()); + long startTimestamp = DurationUtils.INSTANCE.startTimeToTimestamp(duration.getStep(), duration.getStart()); + long endTimestamp = DurationUtils.INSTANCE.endTimeToTimestamp(duration.getStep(), duration.getEnd()); return getMetadataQueryService().getAllServices(startTimestamp, endTimestamp); } public List searchServices(final Duration duration, final String keyword) throws IOException, ParseException { - long startTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getStart()); - long endTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getEnd()); + long startTimestamp = DurationUtils.INSTANCE.startTimeToTimestamp(duration.getStep(), duration.getStart()); + long endTimestamp = DurationUtils.INSTANCE.endTimeToTimestamp(duration.getStep(), duration.getEnd()); return getMetadataQueryService().searchServices(startTimestamp, endTimestamp, keyword); } @@ -75,8 +75,8 @@ public class MetadataQuery implements GraphQLQueryResolver { public List getServiceInstances(final Duration duration, final String serviceId) throws IOException, ParseException { - long startTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getStart()); - long endTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getEnd()); + long startTimestamp = DurationUtils.INSTANCE.startTimeToTimestamp(duration.getStep(), duration.getStart()); + long endTimestamp = DurationUtils.INSTANCE.endTimeToTimestamp(duration.getStep(), duration.getEnd()); return getMetadataQueryService().getServiceInstances(startTimestamp, endTimestamp, serviceId); } diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/TopologyQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/TopologyQuery.java index de9f50439..26a8890fc 100644 --- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/TopologyQuery.java +++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/TopologyQuery.java @@ -50,8 +50,8 @@ public class TopologyQuery implements GraphQLQueryResolver { long startTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getStart()); long endTimeBucket = DurationUtils.INSTANCE.exchangeToTimeBucket(duration.getEnd()); - long startTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getStart()); - long endTimestamp = DurationUtils.INSTANCE.toTimestamp(duration.getStep(), duration.getEnd()); + long startTimestamp = DurationUtils.INSTANCE.startTimeToTimestamp(duration.getStep(), duration.getStart()); + long endTimestamp = DurationUtils.INSTANCE.endTimeToTimestamp(duration.getStep(), duration.getEnd()); return getQueryService().getGlobalTopology(duration.getStep(), startTimeBucket, endTimeBucket, startTimestamp, endTimestamp); } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java index 10147a940..480998b6b 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java @@ -84,7 +84,7 @@ public class TraceModuleProvider extends ModuleProvider { grpcHandlerRegister.addHandler(new TraceSegmentReportServiceHandler(segmentProducerV2)); jettyHandlerRegister.addHandler(new TraceSegmentServletHandler(segmentProducer)); - SegmentStandardizationWorker standardizationWorker = new SegmentStandardizationWorker(segmentProducer, moduleConfig.getBufferPath() + "-v5", moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(), moduleConfig.isBufferFileCleanWhenRestart()); + SegmentStandardizationWorker standardizationWorker = new SegmentStandardizationWorker(segmentProducer, moduleConfig.getBufferPath() + "v5", moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(), moduleConfig.isBufferFileCleanWhenRestart()); segmentProducer.setStandardizationWorker(standardizationWorker); SegmentStandardizationWorker standardizationWorker2 = new SegmentStandardizationWorker(segmentProducer, moduleConfig.getBufferPath(), moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(), moduleConfig.isBufferFileCleanWhenRestart()); diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/segment/SegmentSpanListener.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/segment/SegmentSpanListener.java index 9b43c5337..f7e81117a 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/segment/SegmentSpanListener.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/segment/SegmentSpanListener.java @@ -52,7 +52,7 @@ public class SegmentSpanListener implements FirstSpanListener, EntrySpanListener @Override public void parseFirst(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) { - long timeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(segmentCoreInfo.getStartTime()); + long timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(segmentCoreInfo.getStartTime()); segment.setSegmentId(segmentCoreInfo.getSegmentId()); segment.setServiceId(segmentCoreInfo.getServiceId()); diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java index 924d1e205..0883f4705 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java @@ -68,9 +68,9 @@ public class AgentDataMock { serviceASegmentId = UniqueIdBuilder.INSTANCE.create(); serviceBSegmentId = UniqueIdBuilder.INSTANCE.create(); serviceCSegmentId = UniqueIdBuilder.INSTANCE.create(); - serviceAMock.mock(streamObserver, globalTraceId, serviceASegmentId, startTimestamp, false); - serviceBMock.mock(streamObserver, globalTraceId, serviceBSegmentId, serviceASegmentId, startTimestamp, false); - serviceCMock.mock(streamObserver, globalTraceId, serviceCSegmentId, serviceBSegmentId, startTimestamp, false); + serviceAMock.mock(streamObserver, globalTraceId, serviceASegmentId, startTimestamp, true); + serviceBMock.mock(streamObserver, globalTraceId, serviceBSegmentId, serviceASegmentId, startTimestamp, true); + serviceCMock.mock(streamObserver, globalTraceId, serviceCSegmentId, serviceBSegmentId, startTimestamp, true); } streamObserver.onCompleted(); diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceBMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceBMock.java index 66023b4c7..344e0490f 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceBMock.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceBMock.java @@ -31,6 +31,7 @@ class ServiceBMock { private final RegisterMock registerMock; private static int SERVICE_ID; static int SERVICE_INSTANCE_ID; + static String DUBBO_PROVIDER_ENDPOINT = "org.skywaking.apm.testcase.dubbo.services.GreetServiceImpl.doBusiness()"; static String ROCKET_MQ_ENDPOINT = "org.apache.skywalking.RocketMQ"; static String ROCKET_MQ_ADDRESS = "RocketMQAddress:2000"; @@ -98,7 +99,7 @@ class ServiceBMock { span.addRefs(createReference(uniqueId, isPrepare)); if (isPrepare) { - span.setOperationName(ServiceAMock.DUBBO_ENDPOINT); + span.setOperationName(ServiceBMock.DUBBO_PROVIDER_ENDPOINT); } else { span.setOperationNameId(4); } diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceCMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceCMock.java index 2126b4d3f..b1f537e2c 100644 --- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceCMock.java +++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceCMock.java @@ -90,7 +90,7 @@ class ServiceCMock { reference.setRefType(RefType.CrossProcess); if (isPrepare) { - reference.setParentServiceName(ServiceBMock.ROCKET_MQ_ENDPOINT); + reference.setParentServiceName(ServiceBMock.DUBBO_PROVIDER_ENDPOINT); reference.setNetworkAddress(ServiceBMock.ROCKET_MQ_ADDRESS); reference.setEntryServiceName(ServiceAMock.REST_ENDPOINT); } else { diff --git a/oap-server/server-starter/src/main/assembly/datasource-settings.properties b/oap-server/server-starter/src/main/assembly/datasource-settings.properties new file mode 100644 index 000000000..70b009915 --- /dev/null +++ b/oap-server/server-starter/src/main/assembly/datasource-settings.properties @@ -0,0 +1,31 @@ +# +# 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. +# +# + +jdbcUrl=jdbc:mysql://localhost:3306/swtest +dataSource.user=root +dataSource.password=root@1234 +dataSource.cachePrepStmts=true +dataSource.prepStmtCacheSize=250 +dataSource.prepStmtCacheSqlLimit=2048 +dataSource.useServerPrepStmts=true +dataSource.useLocalSessionState=true +dataSource.rewriteBatchedStatements=true +dataSource.cacheResultSetMetadata=true +dataSource.cacheServerConfiguration=true +dataSource.elideSetAutoCommits=true +dataSource.maintainTimeStats=false \ No newline at end of file diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml index d94a82ae4..663f833e3 100644 --- a/oap-server/server-starter/src/main/resources/application.yml +++ b/oap-server/server-starter/src/main/resources/application.yml @@ -59,6 +59,7 @@ storage: # driver: ${SW_STORAGE_H2_DRIVER:org.h2.jdbcx.JdbcDataSource} # url: ${SW_STORAGE_H2_URL:jdbc:h2:mem:skywalking-oap-db} # user: ${SW_STORAGE_H2_USER:sa} +# mysql: receiver-register: default: receiver-trace: diff --git a/oap-server/server-starter/src/main/resources/datasource-settings.properties b/oap-server/server-starter/src/main/resources/datasource-settings.properties new file mode 100644 index 000000000..70b009915 --- /dev/null +++ b/oap-server/server-starter/src/main/resources/datasource-settings.properties @@ -0,0 +1,31 @@ +# +# 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. +# +# + +jdbcUrl=jdbc:mysql://localhost:3306/swtest +dataSource.user=root +dataSource.password=root@1234 +dataSource.cachePrepStmts=true +dataSource.prepStmtCacheSize=250 +dataSource.prepStmtCacheSqlLimit=2048 +dataSource.useServerPrepStmts=true +dataSource.useLocalSessionState=true +dataSource.rewriteBatchedStatements=true +dataSource.cacheResultSetMetadata=true +dataSource.cacheServerConfiguration=true +dataSource.elideSetAutoCommits=true +dataSource.maintainTimeStats=false \ No newline at end of file diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/HistoryDeleteEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/HistoryDeleteEsDAO.java index d105fddd2..1ad43713b 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/HistoryDeleteEsDAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/HistoryDeleteEsDAO.java @@ -19,10 +19,10 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base; import java.io.IOException; -import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO; import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient; -import org.slf4j.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @author peng-yongsheng @@ -37,7 +37,7 @@ public class HistoryDeleteEsDAO extends EsDAO implements IHistoryDeleteDAO { @Override public void deleteHistory(String modelName, String timeBucketColumnName, Long timeBucketBefore) throws IOException { - int statusCode = getClient().delete(modelName, Indicator.TIME_BUCKET, timeBucketBefore); + int statusCode = getClient().delete(modelName, timeBucketColumnName, timeBucketBefore); if (logger.isDebugEnabled()) { logger.debug("Delete history from {} index, status code {}", modelName, statusCode); } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/pom.xml b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/pom.xml index 6a7e571a6..1fd36e005 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/pom.xml +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/pom.xml @@ -43,6 +43,11 @@ com.h2database h2 + + + + + \ No newline at end of file diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/SQLExecutor.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/SQLExecutor.java index 52ab0d4fd..54b11928e 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/SQLExecutor.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/SQLExecutor.java @@ -22,6 +22,8 @@ import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.SQLException; import java.util.List; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * A SQL executor. @@ -29,6 +31,8 @@ import java.util.List; * @author wusheng */ public class SQLExecutor { + private final Logger logger = LoggerFactory.getLogger(SQLExecutor.class); + private String sql; private List param; @@ -43,6 +47,8 @@ public class SQLExecutor { for (int i = 0; i < param.size(); i++) { preparedStatement.setObject(i + 1, param.get(i)); } + + logger.debug("execute aql in batch: {}", sql); preparedStatement.execute(); } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java index ac8cb361d..169b81f44 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java @@ -19,18 +19,49 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2; import java.util.Properties; -import org.apache.skywalking.oap.server.core.storage.*; -import org.apache.skywalking.oap.server.core.storage.cache.*; -import org.apache.skywalking.oap.server.core.storage.query.*; +import org.apache.skywalking.oap.server.core.storage.IBatchDAO; +import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO; +import org.apache.skywalking.oap.server.core.storage.IRegisterLockDAO; +import org.apache.skywalking.oap.server.core.storage.StorageDAO; +import org.apache.skywalking.oap.server.core.storage.StorageException; +import org.apache.skywalking.oap.server.core.storage.StorageModule; +import org.apache.skywalking.oap.server.core.storage.cache.IEndpointInventoryCacheDAO; +import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressInventoryCacheDAO; +import org.apache.skywalking.oap.server.core.storage.cache.IServiceInstanceInventoryCacheDAO; +import org.apache.skywalking.oap.server.core.storage.cache.IServiceInventoryCacheDAO; +import org.apache.skywalking.oap.server.core.storage.query.IAggregationQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.IAlarmQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.IMetadataQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.IMetricQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.ITopologyQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.ITraceQueryDAO; import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; -import org.apache.skywalking.oap.server.library.module.*; -import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.*; -import org.slf4j.*; +import org.apache.skywalking.oap.server.library.module.ModuleConfig; +import org.apache.skywalking.oap.server.library.module.ModuleDefine; +import org.apache.skywalking.oap.server.library.module.ModuleProvider; +import org.apache.skywalking.oap.server.library.module.ModuleStartException; +import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2AggregationQueryDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2AlarmQueryDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2BatchDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2EndpointInventoryCacheDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2HistoryDeleteDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2MetadataQueryDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2MetricQueryDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2NetworkAddressInventoryCacheDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2RegisterLockDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2ServiceInstanceInventoryCacheDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2ServiceInventoryCacheDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2StorageDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TableInstaller; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TopologyQueryDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TraceQueryDAO; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** - * H2 Storage provider is for demonstration and preview only. - * I will find that haven't implemented several interfaces, because not necessary, - * and don't consider about performance very much. + * H2 Storage provider is for demonstration and preview only. I will find that haven't implemented several interfaces, + * because not necessary, and don't consider about performance very much. * * If someone wants to implement SQL-style database as storage, please just refer the logic. * @@ -81,8 +112,8 @@ public class H2StorageProvider extends ModuleProvider { this.registerServiceImplementation(ITraceQueryDAO.class, new H2TraceQueryDAO(h2Client)); this.registerServiceImplementation(IMetadataQueryDAO.class, new H2MetadataQueryDAO(h2Client)); this.registerServiceImplementation(IAggregationQueryDAO.class, new H2AggregationQueryDAO(h2Client)); - this.registerServiceImplementation(IAlarmQueryDAO.class, new H2AlarmQueryDAO()); - this.registerServiceImplementation(IHistoryDeleteDAO.class, new H2HistoryDeleteDAO()); + this.registerServiceImplementation(IAlarmQueryDAO.class, new H2AlarmQueryDAO(h2Client)); + this.registerServiceImplementation(IHistoryDeleteDAO.class, new H2HistoryDeleteDAO(h2Client)); } @Override public void start() throws ServiceNotProvidedException, ModuleStartException { @@ -91,8 +122,6 @@ public class H2StorageProvider extends ModuleProvider { H2TableInstaller installer = new H2TableInstaller(getManager()); installer.install(h2Client); - - new H2RegisterLockInstaller().install(h2Client); } catch (StorageException e) { throw new ModuleStartException(e.getMessage(), e); } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2AggregationQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2AggregationQueryDAO.java index 71376434b..d87412b0f 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2AggregationQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2AggregationQueryDAO.java @@ -92,36 +92,39 @@ public class H2AggregationQueryDAO implements IAggregationQueryDAO { sql.append(" group by ").append(Indicator.ENTITY_ID); sql.append(") order by value ").append(order.equals(Order.ASC) ? "asc" : "desc").append(" limit ").append(topN); - Connection connection = null; List topNEntities = new ArrayList<>(); - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), conditions.toArray(new Object[0])); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), conditions.toArray(new Object[0]))) { - try { - while (resultSet.next()) { - TopNEntity topNEntity = new TopNEntity(); - topNEntity.setId(resultSet.getString(Indicator.ENTITY_ID)); - topNEntity.setValue(resultSet.getLong("value")); - topNEntities.add(topNEntity); + try { + while (resultSet.next()) { + TopNEntity topNEntity = new TopNEntity(); + topNEntity.setId(resultSet.getString(Indicator.ENTITY_ID)); + topNEntity.setValue(resultSet.getLong("value")); + topNEntities.add(topNEntity); + } + } catch (SQLException e) { + throw new IOException(e); } - } catch (SQLException e) { - throw new IOException(e); } - } finally { - h2Client.close(connection); + } catch (SQLException e) { + throw new IOException(e); } return topNEntities; } - private void setTimeRangeCondition(StringBuilder sql, List conditions, long startTimestamp, + public JDBCHikariCPClient getClient() { + return h2Client; + } + + protected void setTimeRangeCondition(StringBuilder sql, List conditions, long startTimestamp, long endTimestamp) { sql.append(Indicator.TIME_BUCKET).append(" >= ? and ").append(Indicator.TIME_BUCKET).append(" <= ?"); conditions.add(startTimestamp); conditions.add(endTimestamp); } - private interface AppendCondition { + protected interface AppendCondition { void append(StringBuilder sql, List conditions); } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2AlarmQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2AlarmQueryDAO.java index 4982d43b4..daa2b24b9 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2AlarmQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2AlarmQueryDAO.java @@ -19,19 +19,80 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao; import java.io.IOException; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import org.apache.skywalking.oap.server.core.alarm.AlarmRecord; +import org.apache.skywalking.oap.server.core.query.entity.AlarmMessage; import org.apache.skywalking.oap.server.core.query.entity.Alarms; import org.apache.skywalking.oap.server.core.source.Scope; import org.apache.skywalking.oap.server.core.storage.query.IAlarmQueryDAO; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.library.util.StringUtils; /** - * As a demo show env, not necessary to support alarm. - * * @author wusheng */ public class H2AlarmQueryDAO implements IAlarmQueryDAO { + private JDBCHikariCPClient client; + + public H2AlarmQueryDAO(JDBCHikariCPClient client) { + this.client = client; + } + @Override public Alarms getAlarm(Scope scope, String keyword, int limit, int from, long startTB, long endTB) throws IOException { - return new Alarms(); + + StringBuilder sql = new StringBuilder(); + List parameters = new ArrayList<>(10); + sql.append("from ").append(AlarmRecord.INDEX_NAME).append(" where "); + sql.append(" 1=1 "); + if (startTB != 0 && endTB != 0) { + sql.append(" and ").append(AlarmRecord.TIME_BUCKET).append(" >= ?"); + parameters.add(startTB); + sql.append(" and ").append(AlarmRecord.TIME_BUCKET).append(" <= ?"); + parameters.add(endTB); + } + + if (StringUtils.isNotEmpty(keyword)) { + sql.append(" and ").append(AlarmRecord.ALARM_MESSAGE).append(" like '%").append(keyword).append("%' "); + } + sql.append(" order by ").append(AlarmRecord.START_TIME).append(" desc "); + + Alarms alarms = new Alarms(); + try (Connection connection = client.getConnection()) { + + try (ResultSet resultSet = client.executeQuery(connection, "select count(1) total from (select 1 " + sql.toString() + " )", parameters.toArray(new Object[0]))) { + while (resultSet.next()) { + alarms.setTotal(resultSet.getInt("total")); + } + } + + this.buildLimit(sql, from, limit); + + try (ResultSet resultSet = client.executeQuery(connection, "select * " + sql.toString(), parameters.toArray(new Object[0]))) { + while (resultSet.next()) { + AlarmMessage message = new AlarmMessage(); + message.setId(resultSet.getString(AlarmRecord.ID0)); + message.setMessage(resultSet.getString(AlarmRecord.ALARM_MESSAGE)); + message.setStartTime(resultSet.getLong(AlarmRecord.START_TIME)); + message.setScope(Scope.valueOf(resultSet.getInt(AlarmRecord.SCOPE))); + + alarms.getMsgs().add(message); + } + } + } catch (SQLException e) { + throw new IOException(e); + } + + return alarms; + } + + protected void buildLimit(StringBuilder sql, int from, int limit) { + sql.append(" LIMIT ").append(limit); + sql.append(" OFFSET ").append(from); } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2HistoryDeleteDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2HistoryDeleteDAO.java index 60d5eedd1..5c0d48e1c 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2HistoryDeleteDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2HistoryDeleteDAO.java @@ -19,14 +19,31 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao; import java.io.IOException; +import java.sql.Connection; +import java.sql.SQLException; import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO; +import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLBuilder; /** * @author wusheng */ public class H2HistoryDeleteDAO implements IHistoryDeleteDAO { + private JDBCHikariCPClient client; + + public H2HistoryDeleteDAO(JDBCHikariCPClient client) { + this.client = client; + } + @Override public void deleteHistory(String modelName, String timeBucketColumnName, Long timeBucketBefore) throws IOException { + SQLBuilder dataDeleteSQL = new SQLBuilder("delete from " + modelName + " where ").append(timeBucketColumnName).append("<= ?"); + try (Connection connection = client.getConnection()) { + client.execute(connection, dataDeleteSQL.toString(), timeBucketBefore); + } catch (JDBCClientException | SQLException e) { + throw new IOException(e.getMessage(), e); + } } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetadataQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetadataQueryDAO.java index 4f82a826f..b6c66815a 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetadataQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetadataQueryDAO.java @@ -57,17 +57,14 @@ public class H2MetadataQueryDAO implements IMetadataQueryDAO { setTimeRangeCondition(sql, condition, startTimestamp, endTimestamp); sql.append(" and ").append(ServiceInventory.IS_ADDRESS).append("=0"); - Connection connection = null; - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0])); - while (resultSet.next()) { - return resultSet.getInt("num"); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]))) { + while (resultSet.next()) { + return resultSet.getInt("num"); + } } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return 0; } @@ -76,21 +73,17 @@ public class H2MetadataQueryDAO implements IMetadataQueryDAO { StringBuilder sql = new StringBuilder(); List condition = new ArrayList<>(5); sql.append("select count(*) num from ").append(EndpointInventory.MODEL_NAME).append(" where "); - setTimeRangeCondition(sql, condition, startTimestamp, endTimestamp); - sql.append(" and ").append(EndpointInventory.DETECT_POINT).append("=").append(DetectPoint.SERVER.ordinal()); + sql.append(EndpointInventory.DETECT_POINT).append("=").append(DetectPoint.SERVER.ordinal()); - Connection connection = null; - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0])); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]))) { - while (resultSet.next()) { - return resultSet.getInt("num"); + while (resultSet.next()) { + return resultSet.getInt("num"); + } } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return 0; } @@ -100,21 +93,17 @@ public class H2MetadataQueryDAO implements IMetadataQueryDAO { StringBuilder sql = new StringBuilder(); List condition = new ArrayList<>(5); sql.append("select count(*) num from ").append(NetworkAddressInventory.MODEL_NAME).append(" where "); - setTimeRangeCondition(sql, condition, startTimestamp, endTimestamp); - sql.append(" and ").append(NetworkAddressInventory.SRC_LAYER).append("=?"); + sql.append(NetworkAddressInventory.SRC_LAYER).append("=?"); condition.add(srcLayer); - Connection connection = null; - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0])); - while (resultSet.next()) { - return resultSet.getInt("num"); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]))) { + while (resultSet.next()) { + return resultSet.getInt("num"); + } } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return 0; } @@ -128,15 +117,12 @@ public class H2MetadataQueryDAO implements IMetadataQueryDAO { sql.append(" and ").append(ServiceInventory.IS_ADDRESS).append("=? limit 100"); condition.add(BooleanUtils.FALSE); - Connection connection = null; - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0])); - return buildServices(resultSet); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]))) { + return buildServices(resultSet); + } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } } @@ -153,15 +139,12 @@ public class H2MetadataQueryDAO implements IMetadataQueryDAO { } sql.append(" limit 100"); - Connection connection = null; - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0])); - return buildServices(resultSet); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]))) { + return buildServices(resultSet); + } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } } @@ -174,21 +157,18 @@ public class H2MetadataQueryDAO implements IMetadataQueryDAO { sql.append(" and ").append(ServiceInventory.NAME).append(" = ?"); condition.add(serviceCode); - Connection connection = null; - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0])); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]))) { - while (resultSet.next()) { - Service service = new Service(); - service.setId(resultSet.getInt(ServiceInventory.SEQUENCE)); - service.setName(resultSet.getString(ServiceInventory.NAME)); - return service; + while (resultSet.next()) { + Service service = new Service(); + service.setId(resultSet.getInt(ServiceInventory.SEQUENCE)); + service.setName(resultSet.getString(ServiceInventory.NAME)); + return service; + } } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return null; @@ -209,21 +189,18 @@ public class H2MetadataQueryDAO implements IMetadataQueryDAO { sql.append(" limit ").append(limit); List endpoints = new ArrayList<>(); - Connection connection = null; - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0])); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]))) { - while (resultSet.next()) { - Endpoint endpoint = new Endpoint(); - endpoint.setId(resultSet.getInt(EndpointInventory.SEQUENCE)); - endpoint.setName(resultSet.getString(EndpointInventory.NAME)); - endpoints.add(endpoint); + while (resultSet.next()) { + Endpoint endpoint = new Endpoint(); + endpoint.setId(resultSet.getInt(EndpointInventory.SEQUENCE)); + endpoint.setName(resultSet.getString(EndpointInventory.NAME)); + endpoints.add(endpoint); + } } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return endpoints; } @@ -237,40 +214,37 @@ public class H2MetadataQueryDAO implements IMetadataQueryDAO { sql.append(" and ").append(ServiceInstanceInventory.SERVICE_ID).append("=?"); condition.add(serviceId); - Connection connection = null; List serviceInstances = new ArrayList<>(); - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0])); + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), condition.toArray(new Object[0]))) { - while (resultSet.next()) { - ServiceInstance serviceInstance = new ServiceInstance(); - serviceInstance.setId(resultSet.getString(ServiceInstanceInventory.SEQUENCE)); - serviceInstance.setName(resultSet.getString(ServiceInstanceInventory.NAME)); - int languageId = resultSet.getInt(ServiceInstanceInventory.LANGUAGE); - serviceInstance.setLanguage(LanguageTrans.INSTANCE.value(languageId)); + while (resultSet.next()) { + ServiceInstance serviceInstance = new ServiceInstance(); + serviceInstance.setId(resultSet.getString(ServiceInstanceInventory.SEQUENCE)); + serviceInstance.setName(resultSet.getString(ServiceInstanceInventory.NAME)); + int languageId = resultSet.getInt(ServiceInstanceInventory.LANGUAGE); + serviceInstance.setLanguage(LanguageTrans.INSTANCE.value(languageId)); - String osName = resultSet.getString(ServiceInstanceInventory.OS_NAME); - if (StringUtils.isNotEmpty(osName)) { - serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.OS_NAME, osName)); + String osName = resultSet.getString(ServiceInstanceInventory.OS_NAME); + if (StringUtils.isNotEmpty(osName)) { + serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.OS_NAME, osName)); + } + String hostName = resultSet.getString(ServiceInstanceInventory.HOST_NAME); + if (StringUtils.isNotEmpty(hostName)) { + serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.HOST_NAME, hostName)); + } + serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.PROCESS_NO, resultSet.getString(ServiceInstanceInventory.PROCESS_NO))); + + List ipv4s = ServiceInstanceInventory.AgentOsInfo.ipv4sDeserialize(resultSet.getString(ServiceInstanceInventory.IPV4S)); + for (String ipv4 : ipv4s) { + serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.IPV4S, ipv4)); + } + + serviceInstances.add(serviceInstance); } - String hostName = resultSet.getString(ServiceInstanceInventory.HOST_NAME); - if (StringUtils.isNotEmpty(hostName)) { - serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.HOST_NAME, hostName)); - } - serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.PROCESS_NO, resultSet.getString(ServiceInstanceInventory.PROCESS_NO))); - - List ipv4s = ServiceInstanceInventory.AgentOsInfo.ipv4sDeserialize(resultSet.getString(ServiceInstanceInventory.IPV4S)); - for (String ipv4 : ipv4s) { - serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.IPV4S, ipv4)); - } - - serviceInstances.add(serviceInstance); } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return serviceInstances; } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricQueryDAO.java index a8093a6ba..a61c6db97 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2MetricQueryDAO.java @@ -91,9 +91,7 @@ public class H2MetricQueryDAO extends H2SQLExecutor implements IMetricQueryDAO { } IntValues intValues = new IntValues(); - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet resultSet = h2Client.executeQuery(connection, "select " + Indicator.ENTITY_ID + " id, " + op + "(" + valueCName + ") value from " + tableName + " where " + whereSql + Indicator.TIME_BUCKET + ">= ? and " + Indicator.TIME_BUCKET + "<=?" @@ -109,8 +107,6 @@ public class H2MetricQueryDAO extends H2SQLExecutor implements IMetricQueryDAO { } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return orderWithDefault0(intValues, ids); } @@ -129,9 +125,7 @@ public class H2MetricQueryDAO extends H2SQLExecutor implements IMetricQueryDAO { IntValues intValues = new IntValues(); - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet resultSet = h2Client.executeQuery(connection, "select id, " + valueCName + " from " + tableName + " where id in (" + idValues.toString() + ")")) { while (resultSet.next()) { KVInt kv = new KVInt(); @@ -142,8 +136,6 @@ public class H2MetricQueryDAO extends H2SQLExecutor implements IMetricQueryDAO { } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return orderWithDefault0(intValues, ids); } @@ -183,9 +175,7 @@ public class H2MetricQueryDAO extends H2SQLExecutor implements IMetricQueryDAO { List> thermodynamicValueCollection = new ArrayList<>(); Map> thermodynamicValueMatrix = new HashMap<>(); - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { Thermodynamic thermodynamic = new Thermodynamic(); int numOfSteps = 0; int axisYStep = 0; @@ -232,8 +222,6 @@ public class H2MetricQueryDAO extends H2SQLExecutor implements IMetricQueryDAO { return thermodynamic; } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2RegisterDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2RegisterDAO.java index 6cdf4ec3e..c6b2e6bdd 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2RegisterDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2RegisterDAO.java @@ -19,17 +19,24 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao; import java.io.IOException; -import java.sql.*; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; import org.apache.skywalking.oap.server.core.Const; import org.apache.skywalking.oap.server.core.register.RegisterSource; -import org.apache.skywalking.oap.server.core.storage.*; +import org.apache.skywalking.oap.server.core.storage.IRegisterDAO; +import org.apache.skywalking.oap.server.core.storage.StorageBuilder; import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @author wusheng */ public class H2RegisterDAO extends H2SQLExecutor implements IRegisterDAO { + private static final Logger logger = LoggerFactory.getLogger(H2RegisterDAO.class); + private final JDBCHikariCPClient h2Client; private final StorageBuilder storageBuilder; @@ -40,9 +47,7 @@ public class H2RegisterDAO extends H2SQLExecutor implements IRegisterDAO { } @Override public int max(String modelName) throws IOException { - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet rs = h2Client.executeQuery(connection, "SELECT max(sequence) max_id FROM " + modelName)) { while (rs.next()) { int maxId = rs.getInt("max_id"); @@ -57,8 +62,6 @@ public class H2RegisterDAO extends H2SQLExecutor implements IRegisterDAO { throw new IOException(e.getMessage(), e); } catch (JDBCClientException e) { throw new IOException(e.getMessage(), e); - } finally { - h2Client.close(connection); } return Const.NONE; } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2SQLExecutor.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2SQLExecutor.java index 53729d6a4..8cd217532 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2SQLExecutor.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2SQLExecutor.java @@ -48,9 +48,7 @@ public class H2SQLExecutor { protected StorageData getByID(JDBCHikariCPClient h2Client, String modelName, String id, StorageBuilder storageBuilder) throws IOException { - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet rs = h2Client.executeQuery(connection, "SELECT * FROM " + modelName + " WHERE id = ?", id)) { return toStorageData(rs, modelName, storageBuilder); } @@ -58,16 +56,12 @@ public class H2SQLExecutor { throw new IOException(e.getMessage(), e); } catch (JDBCClientException e) { throw new IOException(e.getMessage(), e); - } finally { - h2Client.close(connection); } } protected StorageData getByColumn(JDBCHikariCPClient h2Client, String modelName, String columnName, Object value, StorageBuilder storageBuilder) throws IOException { - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet rs = h2Client.executeQuery(connection, "SELECT * FROM " + modelName + " WHERE " + columnName + " = ?", value)) { return toStorageData(rs, modelName, storageBuilder); } @@ -75,8 +69,6 @@ public class H2SQLExecutor { throw new IOException(e.getMessage(), e); } catch (JDBCClientException e) { throw new IOException(e.getMessage(), e); - } finally { - h2Client.close(connection); } } @@ -86,7 +78,7 @@ public class H2SQLExecutor { Map data = new HashMap(); List columns = TableMetaInfo.get(modelName).getColumns(); for (ModelColumn column : columns) { - data.put(column.getColumnName().getName(), rs.getObject(column.getColumnName().getName())); + data.put(column.getColumnName().getName(), rs.getObject(column.getColumnName().getStorageName())); } return storageBuilder.map2Data(data); } @@ -94,9 +86,7 @@ public class H2SQLExecutor { } protected int getEntityIDByID(JDBCHikariCPClient h2Client, String entityColumnName, String modelName, String id) { - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet rs = h2Client.executeQuery(connection, "SELECT " + entityColumnName + " FROM " + modelName + " WHERE ID=?", id)) { while (rs.next()) { return rs.getInt(ServiceInstanceInventory.SEQUENCE); @@ -106,8 +96,6 @@ public class H2SQLExecutor { logger.error(e.getMessage(), e); } catch (JDBCClientException e) { logger.error(e.getMessage(), e); - } finally { - h2Client.close(connection); } return Const.NONE; } @@ -149,7 +137,7 @@ public class H2SQLExecutor { List param = new ArrayList<>(); for (int i = 0; i < columns.size(); i++) { ModelColumn column = columns.get(i); - sqlBuilder.append(column.getColumnName().getName() + "= ?"); + sqlBuilder.append(column.getColumnName().getStorageName() + "= ?"); if (i != columns.size() - 1) { sqlBuilder.append(","); } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2ServiceInventoryCacheDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2ServiceInventoryCacheDAO.java index f28a1c6ac..20da50056 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2ServiceInventoryCacheDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2ServiceInventoryCacheDAO.java @@ -70,11 +70,7 @@ public class H2ServiceInventoryCacheDAO extends H2SQLExecutor implements IServic sql.append(" where ").append(ServiceInventory.IS_ADDRESS).append("=? "); sql.append(" and ").append(ServiceInventory.MAPPING_LAST_UPDATE_TIME).append(">?"); - sql.append(" LIMIT 50 "); - - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet resultSet = h2Client.executeQuery(connection, sql.toString(), BooleanUtils.TRUE, System.currentTimeMillis() - 10000)) { while (resultSet.next()) { ServiceInventory serviceInventory = (ServiceInventory)toStorageData(resultSet, ServiceInventory.MODEL_NAME, new ServiceInventory.Builder()); @@ -85,8 +81,6 @@ public class H2ServiceInventoryCacheDAO extends H2SQLExecutor implements IServic } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } } catch (Throwable e) { logger.error(e.getMessage()); diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java index e7bfbaae0..946ae0387 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TableInstaller.java @@ -44,6 +44,7 @@ public class H2TableInstaller extends ModelInstaller { } @Override protected boolean isExists(Client client, Model model) throws StorageException { + TableMetaInfo.addModel(model); JDBCHikariCPClient h2Client = (JDBCHikariCPClient)client; try (Connection conn = h2Client.getConnection()) { try (ResultSet rset = conn.getMetaData().getTables(null, null, model.getName(), null)) { @@ -68,14 +69,13 @@ public class H2TableInstaller extends ModelInstaller { } @Override protected void createTable(Client client, Model model) throws StorageException { - TableMetaInfo.addModel(model); JDBCHikariCPClient h2Client = (JDBCHikariCPClient)client; SQLBuilder tableCreateSQL = new SQLBuilder("CREATE TABLE IF NOT EXISTS " + model.getName() + " ("); - tableCreateSQL.appendLine("id VARCHAR2(300), "); + tableCreateSQL.appendLine("id VARCHAR(300) PRIMARY KEY, "); for (int i = 0; i < model.getColumns().size(); i++) { ModelColumn column = model.getColumns().get(i); ColumnName name = column.getColumnName(); - tableCreateSQL.appendLine(name.getName() + " " + getColumnType(column.getType()) + (i != model.getColumns().size() - 1 ? "," : "")); + tableCreateSQL.appendLine(name.getStorageName() + " " + getColumnType(model, name, column.getType()) + (i != model.getColumns().size() - 1 ? "," : "")); } tableCreateSQL.appendLine(")"); @@ -83,19 +83,17 @@ public class H2TableInstaller extends ModelInstaller { logger.debug("creating table: " + tableCreateSQL.toStringInNewLine()); } - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { h2Client.execute(connection, tableCreateSQL.toString()); } catch (JDBCClientException e) { throw new StorageException(e.getMessage(), e); - } finally { - h2Client.close(connection); + } catch (SQLException e) { + throw new StorageException(e.getMessage(), e); } } - private String getColumnType(Class type) { + protected String getColumnType(Model model, ColumnName name, Class type) { if (Integer.class.equals(type) || int.class.equals(type)) { return "INT"; } else if (Long.class.equals(type) || long.class.equals(type)) { diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TopologyQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TopologyQueryDAO.java index f24e9b65e..f1d73cc97 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TopologyQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TopologyQueryDAO.java @@ -94,21 +94,18 @@ public class H2TopologyQueryDAO implements ITopologyQueryDAO { serviceIdMatchSql.append(")"); } List calls = new ArrayList<>(); - Connection connection = null; - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, "select " + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, "select " + Indicator.ENTITY_ID - + " component_id from " + tableName + " where " + + " from " + tableName + " where " + Indicator.TIME_BUCKET + ">= ? and " + Indicator.TIME_BUCKET + "<=? " + serviceIdMatchSql.toString() + " group by " + Indicator.ENTITY_ID, - conditions); - buildCalls(resultSet, calls, isClientSide); + conditions)) { + buildCalls(resultSet, calls, isClientSide); + } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return calls; } @@ -119,22 +116,19 @@ public class H2TopologyQueryDAO implements ITopologyQueryDAO { conditions[0] = startTB; conditions[1] = endTB; conditions[2] = id; - Connection connection = null; List calls = new ArrayList<>(); - try { - connection = h2Client.getConnection(); - ResultSet resultSet = h2Client.executeQuery(connection, "select " + try (Connection connection = h2Client.getConnection()) { + try (ResultSet resultSet = h2Client.executeQuery(connection, "select " + Indicator.ENTITY_ID + " from " + tableName + " where " + Indicator.TIME_BUCKET + ">= ? and " + Indicator.TIME_BUCKET + "<=? and " + (isSourceId ? sourceCName : destCName) + "=?" + " group by " + Indicator.ENTITY_ID, - conditions); - buildCalls(resultSet, calls, isSourceId); + conditions)) { + buildCalls(resultSet, calls, isSourceId); + } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return calls; } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TraceQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TraceQueryDAO.java index af7d03dbb..08e352e42 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TraceQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2TraceQueryDAO.java @@ -102,13 +102,9 @@ public class H2TraceQueryDAO implements ITraceQueryDAO { sql.append(" order by ").append(SegmentRecord.LATENCY).append(" ").append(SortOrder.DESC); break; } - sql.append(" LIMIT ").append(limit); - sql.append(" OFFSET ").append(from); TraceBrief traceBrief = new TraceBrief(); - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet resultSet = h2Client.executeQuery(connection, "select count(1) total from (select 1 " + sql.toString() + " )", parameters.toArray(new Object[0]))) { while (resultSet.next()) { @@ -116,6 +112,8 @@ public class H2TraceQueryDAO implements ITraceQueryDAO { } } + buildLimit(sql, from, limit); + try (ResultSet resultSet = h2Client.executeQuery(connection, "select * " + sql.toString(), parameters.toArray(new Object[0]))) { while (resultSet.next()) { BasicTrace basicTrace = new BasicTrace(); @@ -132,18 +130,19 @@ public class H2TraceQueryDAO implements ITraceQueryDAO { } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return traceBrief; } + protected void buildLimit(StringBuilder sql, int from, int limit) { + sql.append(" LIMIT ").append(limit); + sql.append(" OFFSET ").append(from); + } + @Override public List queryByTraceId(String traceId) throws IOException { List segmentRecords = new ArrayList<>(); - Connection connection = null; - try { - connection = h2Client.getConnection(); + try (Connection connection = h2Client.getConnection()) { try (ResultSet resultSet = h2Client.executeQuery(connection, "select * from " + SegmentRecord.INDEX_NAME + " where " + SegmentRecord.TRACE_ID + " = ?", traceId)) { while (resultSet.next()) { @@ -166,9 +165,11 @@ public class H2TraceQueryDAO implements ITraceQueryDAO { } } catch (SQLException e) { throw new IOException(e); - } finally { - h2Client.close(connection); } return segmentRecords; } + + protected JDBCHikariCPClient getClient() { + return h2Client; + } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLAggregationQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLAggregationQueryDAO.java new file mode 100644 index 000000000..f6e86fc8c --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLAggregationQueryDAO.java @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator; +import org.apache.skywalking.oap.server.core.query.entity.Order; +import org.apache.skywalking.oap.server.core.query.entity.Step; +import org.apache.skywalking.oap.server.core.query.entity.TopNEntity; +import org.apache.skywalking.oap.server.core.storage.DownSamplingModelNameBuilder; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2AggregationQueryDAO; + +/** + * @author wusheng + */ +public class MySQLAggregationQueryDAO extends H2AggregationQueryDAO { + public MySQLAggregationQueryDAO( + JDBCHikariCPClient client) { + super(client); + } + + @Override + public List topNQuery(String indName, String valueCName, int topN, Step step, + long startTB, long endTB, Order order, AppendCondition appender) throws IOException { + String tableName = DownSamplingModelNameBuilder.build(step, indName); + StringBuilder sql = new StringBuilder(); + List conditions = new ArrayList<>(10); + sql.append("select * from (select avg(").append(valueCName).append(") value,").append(Indicator.ENTITY_ID).append(" from ") + .append(tableName).append(" where "); + this.setTimeRangeCondition(sql, conditions, startTB, endTB); + if (appender != null) { + appender.append(sql, conditions); + } + sql.append(" group by ").append(Indicator.ENTITY_ID); + sql.append(") AS INDICATOR order by value ").append(order.equals(Order.ASC) ? "asc" : "desc").append(" limit ").append(topN); + + List topNEntities = new ArrayList<>(); + try (Connection connection = getClient().getConnection()) { + try (ResultSet resultSet = getClient().executeQuery(connection, sql.toString(), conditions.toArray(new Object[0]))) { + + try { + while (resultSet.next()) { + TopNEntity topNEntity = new TopNEntity(); + topNEntity.setId(resultSet.getString(Indicator.ENTITY_ID)); + topNEntity.setValue(resultSet.getLong("value")); + topNEntities.add(topNEntity); + } + } catch (SQLException e) { + throw new IOException(e); + } + } + } catch (SQLException e) { + throw new IOException(e); + } + return topNEntities; + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLAlarmQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLAlarmQueryDAO.java new file mode 100644 index 000000000..38ce85c00 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLAlarmQueryDAO.java @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import org.apache.skywalking.oap.server.core.alarm.AlarmRecord; +import org.apache.skywalking.oap.server.core.query.entity.AlarmMessage; +import org.apache.skywalking.oap.server.core.query.entity.Alarms; +import org.apache.skywalking.oap.server.core.source.Scope; +import org.apache.skywalking.oap.server.core.storage.query.IAlarmQueryDAO; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.library.util.StringUtils; + +/** + * @author wusheng + */ +public class MySQLAlarmQueryDAO implements IAlarmQueryDAO { + private JDBCHikariCPClient client; + + public MySQLAlarmQueryDAO(JDBCHikariCPClient client) { + this.client = client; + } + + @Override + public Alarms getAlarm(Scope scope, String keyword, int limit, int from, long startTB, + long endTB) throws IOException { + + StringBuilder sql = new StringBuilder(); + List parameters = new ArrayList<>(10); + sql.append("from ").append(AlarmRecord.INDEX_NAME).append(" where "); + sql.append(" scope = ?"); + parameters.add(scope.ordinal()); + if (startTB != 0 && endTB != 0) { + sql.append(" and ").append(AlarmRecord.TIME_BUCKET).append(" >= ?"); + parameters.add(startTB); + sql.append(" and ").append(AlarmRecord.TIME_BUCKET).append(" <= ?"); + parameters.add(endTB); + } + + if (StringUtils.isNotEmpty(keyword)) { + sql.append(" and ").append(AlarmRecord.ALARM_MESSAGE).append(" like '%").append(keyword).append("%' "); + } + sql.append(" order by ").append(AlarmRecord.START_TIME).append(" desc "); + + Alarms alarms = new Alarms(); + try (Connection connection = client.getConnection()) { + + try (ResultSet resultSet = client.executeQuery(connection, "select count(1) total from (select 1 " + sql.toString() + " ) AS alarm", parameters.toArray(new Object[0]))) { + while (resultSet.next()) { + alarms.setTotal(resultSet.getInt("total")); + } + } + + this.buildLimit(sql, from, limit); + + try (ResultSet resultSet = client.executeQuery(connection, "select * " + sql.toString(), parameters.toArray(new Object[0]))) { + while (resultSet.next()) { + AlarmMessage message = new AlarmMessage(); + message.setId(resultSet.getString(AlarmRecord.ID0)); + message.setMessage(resultSet.getString(AlarmRecord.ALARM_MESSAGE)); + message.setStartTime(resultSet.getLong(AlarmRecord.START_TIME)); + message.setScope(Scope.valueOf(resultSet.getInt(AlarmRecord.SCOPE))); + + alarms.getMsgs().add(message); + } + } + } catch (SQLException e) { + throw new IOException(e); + } + + return alarms; + } + + protected void buildLimit(StringBuilder sql, int from, int limit) { + sql.append(" LIMIT ").append(from).append(", ").append(limit); + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLRegisterLockInstaller.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLRegisterLockInstaller.java new file mode 100644 index 000000000..8f96c94ac --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLRegisterLockInstaller.java @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import org.apache.skywalking.oap.server.core.register.worker.InventoryProcess; +import org.apache.skywalking.oap.server.core.source.Scope; +import org.apache.skywalking.oap.server.core.storage.StorageException; +import org.apache.skywalking.oap.server.core.storage.annotation.StorageEntityAnnotationUtils; +import org.apache.skywalking.oap.server.library.client.Client; +import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author wusheng + */ +public class MySQLRegisterLockInstaller { + public static final String LOCK_TABLE_NAME = "register_lock"; + + private static final Logger logger = LoggerFactory.getLogger(MySQLRegisterLockInstaller.class); + + /** + * In MySQL lock storage, lock table created. The row lock is used in {@link MySQLRegisterTableLockDAO} + * + * @param client + * @throws StorageException + */ + public void install(Client client, MySQLRegisterTableLockDAO dao) throws StorageException { + JDBCHikariCPClient h2Client = (JDBCHikariCPClient)client; + SQLBuilder tableCreateSQL = new SQLBuilder("CREATE TABLE IF NOT EXISTS " + LOCK_TABLE_NAME + " ("); + tableCreateSQL.appendLine("id int PRIMARY KEY, "); + tableCreateSQL.appendLine("name VARCHAR(100)"); + tableCreateSQL.appendLine(")"); + + if (logger.isDebugEnabled()) { + logger.debug("creating table: " + tableCreateSQL.toStringInNewLine()); + } + + try (Connection connection = h2Client.getConnection()) { + h2Client.execute(connection, tableCreateSQL.toString()); + + for (Class registerSource : InventoryProcess.INSTANCE.getAllRegisterSources()) { + Scope sourceScope = StorageEntityAnnotationUtils.getSourceScope(registerSource); + dao.init(sourceScope); + putIfAbsent(h2Client, connection, sourceScope.ordinal(), sourceScope.name()); + } + } catch (JDBCClientException e) { + throw new StorageException(e.getMessage(), e); + } catch (SQLException e) { + throw new StorageException(e.getMessage(), e); + } + } + + private void putIfAbsent(JDBCHikariCPClient h2Client, Connection connection, int scopeId, + String scopeName) throws StorageException { + boolean existed = false; + try (ResultSet resultSet = h2Client.executeQuery(connection, "select 1 from " + LOCK_TABLE_NAME + " where id = " + scopeId)) { + if (resultSet.next()) { + existed = true; + } + } catch (SQLException | JDBCClientException e) { + throw new StorageException(e.getMessage(), e); + } + if (!existed) { + try (PreparedStatement statement = connection.prepareStatement("insert into " + LOCK_TABLE_NAME + "(id, name) values (?, ?)")) { + statement.setInt(1, scopeId); + statement.setString(2, scopeName); + + statement.execute(); + } catch (SQLException e) { + throw new StorageException(e.getMessage(), e); + } + } + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLRegisterTableLockDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLRegisterTableLockDAO.java new file mode 100644 index 000000000..3382bc160 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLRegisterTableLockDAO.java @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; + +import java.sql.Connection; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.Map; +import org.apache.skywalking.oap.server.core.source.Scope; +import org.apache.skywalking.oap.server.core.storage.IRegisterLockDAO; +import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * In MySQL, use a row lock of LOCK table. + * + * @author wusheng + */ +public class MySQLRegisterTableLockDAO implements IRegisterLockDAO { + private static final Logger logger = LoggerFactory.getLogger(MySQLRegisterTableLockDAO.class); + + private JDBCHikariCPClient h2Client; + private Map onLockingConnection; + + public MySQLRegisterTableLockDAO(JDBCHikariCPClient h2Client) { + this.h2Client = h2Client; + onLockingConnection = new HashMap<>(); + } + + void init(Scope scope) { + if (!onLockingConnection.containsKey(scope)) { + onLockingConnection.put(scope, null); + } + } + + @Override public boolean tryLock(Scope scope) { + if (onLockingConnection.containsKey(scope)) { + try { + Connection connection = h2Client.getTransactionConnection(); + onLockingConnection.put(scope, connection); + connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED); + h2Client.execute(connection, "select * from " + MySQLRegisterLockInstaller.LOCK_TABLE_NAME + " where id = " + scope.ordinal() + " for update"); + return true; + } catch (JDBCClientException | SQLException e) { + logger.error("try inventory register lock for scope id={} name={} failure.", scope.ordinal(), scope.name()); + logger.error("tryLock error", e); + return false; + } + } + return false; + } + + @Override public void releaseLock(Scope scope) { + Connection connection = onLockingConnection.get(scope); + if (connection != null) { + try { + connection.commit(); + connection.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ); + connection.close(); + } catch (SQLException e) { + logger.error("release lock failure.", e); + } finally { + onLockingConnection.put(scope, null); + } + } + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java new file mode 100644 index 000000000..3fb67fd59 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLStorageProvider.java @@ -0,0 +1,143 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; + +import java.io.IOException; +import java.util.Properties; +import org.apache.skywalking.oap.server.core.storage.IBatchDAO; +import org.apache.skywalking.oap.server.core.storage.IHistoryDeleteDAO; +import org.apache.skywalking.oap.server.core.storage.IRegisterLockDAO; +import org.apache.skywalking.oap.server.core.storage.StorageDAO; +import org.apache.skywalking.oap.server.core.storage.StorageException; +import org.apache.skywalking.oap.server.core.storage.StorageModule; +import org.apache.skywalking.oap.server.core.storage.cache.IEndpointInventoryCacheDAO; +import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressInventoryCacheDAO; +import org.apache.skywalking.oap.server.core.storage.cache.IServiceInstanceInventoryCacheDAO; +import org.apache.skywalking.oap.server.core.storage.cache.IServiceInventoryCacheDAO; +import org.apache.skywalking.oap.server.core.storage.query.IAggregationQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.IAlarmQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.IMetadataQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.IMetricQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.ITopologyQueryDAO; +import org.apache.skywalking.oap.server.core.storage.query.ITraceQueryDAO; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.library.module.ModuleConfig; +import org.apache.skywalking.oap.server.library.module.ModuleDefine; +import org.apache.skywalking.oap.server.library.module.ModuleProvider; +import org.apache.skywalking.oap.server.library.module.ModuleStartException; +import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException; +import org.apache.skywalking.oap.server.library.util.ResourceUtils; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.H2StorageConfig; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.H2StorageProvider; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2BatchDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2EndpointInventoryCacheDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2HistoryDeleteDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2MetadataQueryDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2MetricQueryDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2NetworkAddressInventoryCacheDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2ServiceInstanceInventoryCacheDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2ServiceInventoryCacheDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2StorageDAO; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TopologyQueryDAO; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * MySQL storage provider should be secondary choice for production usage as SkyWalking storage solution. It enhanced + * and came from H2StorageProvider, but consider more in using in production. + * + * Because this module is not really related to MySQL, instead, it is based on MySQL SQL style with JDBC, so, by having + * this storage implementation, we could also use this in MySQL-compatible projects, such as, Apache ShardingSphere, + * TiDB + * + * @author wusheng + */ +public class MySQLStorageProvider extends ModuleProvider { + private static final Logger logger = LoggerFactory.getLogger(H2StorageProvider.class); + + private H2StorageConfig config; + private JDBCHikariCPClient mysqlClient; + private MySQLRegisterTableLockDAO lockDAO; + + public MySQLStorageProvider() { + config = new H2StorageConfig(); + } + + @Override public String name() { + return "mysql"; + } + + @Override public Class module() { + return StorageModule.class; + } + + @Override public ModuleConfig createConfigBeanIfAbsent() { + return config; + } + + @Override public void prepare() throws ServiceNotProvidedException, ModuleStartException { + Properties settings = new Properties(); + try { + settings.load(ResourceUtils.read("datasource-settings.properties")); + } catch (IOException e) { + throw new ModuleStartException("load datasource setting file failure.", e); + } + + mysqlClient = new JDBCHikariCPClient(settings); + + this.registerServiceImplementation(IBatchDAO.class, new H2BatchDAO(mysqlClient)); + this.registerServiceImplementation(StorageDAO.class, new H2StorageDAO(mysqlClient)); + lockDAO = new MySQLRegisterTableLockDAO(mysqlClient); + this.registerServiceImplementation(IRegisterLockDAO.class, lockDAO); + + this.registerServiceImplementation(IServiceInventoryCacheDAO.class, new H2ServiceInventoryCacheDAO(mysqlClient)); + this.registerServiceImplementation(IServiceInstanceInventoryCacheDAO.class, new H2ServiceInstanceInventoryCacheDAO(mysqlClient)); + this.registerServiceImplementation(IEndpointInventoryCacheDAO.class, new H2EndpointInventoryCacheDAO(mysqlClient)); + this.registerServiceImplementation(INetworkAddressInventoryCacheDAO.class, new H2NetworkAddressInventoryCacheDAO(mysqlClient)); + + this.registerServiceImplementation(ITopologyQueryDAO.class, new H2TopologyQueryDAO(mysqlClient)); + this.registerServiceImplementation(IMetricQueryDAO.class, new H2MetricQueryDAO(mysqlClient)); + this.registerServiceImplementation(ITraceQueryDAO.class, new MySQLTraceQueryDAO(mysqlClient)); + this.registerServiceImplementation(IMetadataQueryDAO.class, new H2MetadataQueryDAO(mysqlClient)); + this.registerServiceImplementation(IAggregationQueryDAO.class, new MySQLAggregationQueryDAO(mysqlClient)); + this.registerServiceImplementation(IAlarmQueryDAO.class, new MySQLAlarmQueryDAO(mysqlClient)); + this.registerServiceImplementation(IHistoryDeleteDAO.class, new H2HistoryDeleteDAO(mysqlClient)); + } + + @Override public void start() throws ServiceNotProvidedException, ModuleStartException { + try { + mysqlClient.connect(); + + MySQLTableInstaller installer = new MySQLTableInstaller(getManager()); + installer.install(mysqlClient); + + new MySQLRegisterLockInstaller().install(mysqlClient, lockDAO); + } catch (StorageException e) { + throw new ModuleStartException(e.getMessage(), e); + } + } + + @Override public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException { + + } + + @Override public String[] requiredModules() { + return new String[0]; + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java new file mode 100644 index 000000000..eeced149f --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTableInstaller.java @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; + +import java.sql.Connection; +import java.sql.SQLException; +import org.apache.skywalking.oap.server.core.analysis.indicator.IntKeyLongValueArray; +import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord; +import org.apache.skywalking.oap.server.core.register.RegisterSource; +import org.apache.skywalking.oap.server.core.source.Scope; +import org.apache.skywalking.oap.server.core.storage.StorageException; +import org.apache.skywalking.oap.server.core.storage.model.ColumnName; +import org.apache.skywalking.oap.server.core.storage.model.Model; +import org.apache.skywalking.oap.server.library.client.Client; +import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLBuilder; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TableInstaller; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Extend H2TableInstaller but match MySQL SQL syntax. + * + * @author wusheng + */ +public class MySQLTableInstaller extends H2TableInstaller { + private static final Logger logger = LoggerFactory.getLogger(MySQLTableInstaller.class); + + public MySQLTableInstaller(ModuleManager moduleManager) { + super(moduleManager); + /** + * Override column because the default column names in core have syntax conflict with MySQL. + */ + this.overrideColumnName("precision", "cal_precision"); + this.overrideColumnName("match", "match_num"); + } + + @Override protected void createTable(Client client, Model model) throws StorageException { + super.createTable(client, model); + JDBCHikariCPClient jdbcHikariCPClient = (JDBCHikariCPClient)client; + this.createIndexes(jdbcHikariCPClient, model); + } + + @Override protected void deleteTable(Client client, Model model) throws StorageException { + JDBCHikariCPClient jdbcClient = (JDBCHikariCPClient)client; + try (Connection connection = jdbcClient.getConnection()) { + jdbcClient.execute(connection, "drop table " + model.getName()); + } catch (SQLException | JDBCClientException e) { + throw new StorageException(e.getMessage(), e); + } + } + + @Override + protected String getColumnType(Model model, ColumnName name, Class type) { + if (Integer.class.equals(type) || int.class.equals(type)) { + return "INT"; + } else if (Long.class.equals(type) || long.class.equals(type)) { + return "BIGINT"; + } else if (Double.class.equals(type) || double.class.equals(type)) { + return "DOUBLE"; + } else if (String.class.equals(type)) { + if (Scope.Segment.equals(model.getSource())) { + if (name.getName().equals(SegmentRecord.TRACE_ID) || name.getName().equals(SegmentRecord.SEGMENT_ID)) + return "VARCHAR(300)"; + } + return "VARCHAR(2000)"; + } else if (IntKeyLongValueArray.class.equals(type)) { + return "MEDIUMTEXT"; + } else if (byte[].class.equals(type)) { + return "MEDIUMTEXT"; + } else { + throw new IllegalArgumentException("Unsupported data type: " + type.getName()); + } + } + + protected void createIndexes(JDBCHikariCPClient client, Model model) throws StorageException { + switch (model.getSource()) { + case ServiceInventory: + case ServiceInstanceInventory: + case NetworkAddress: + case EndpointInventory: + createInventoryIndexes(client, model); + return; + case Segment: + createSegmentIndexes(client, model); + return; + case Alarm: + createAlarmIndexes(client, model); + return; + default: + createIndexesForAllIndicators(client, model); + + } + } + + private void createIndexesForAllIndicators(JDBCHikariCPClient client, Model model) throws StorageException { + try (Connection connection = client.getConnection()) { + SQLBuilder tableIndexSQL = new SQLBuilder("CREATE INDEX "); + tableIndexSQL.append(model.getName().toUpperCase()).append("_TIME_BUCKET "); + tableIndexSQL.append("ON ").append(model.getName()).append("(").append(SegmentRecord.TIME_BUCKET).append(")"); + createIndex(client, connection, model, tableIndexSQL); + } catch (JDBCClientException e) { + throw new StorageException(e.getMessage(), e); + } catch (SQLException e) { + throw new StorageException(e.getMessage(), e); + } + } + + private void createAlarmIndexes(JDBCHikariCPClient client, Model model) throws StorageException { + try (Connection connection = client.getConnection()) { + SQLBuilder tableIndexSQL = new SQLBuilder("CREATE INDEX "); + tableIndexSQL.append(model.getName().toUpperCase()).append("_TIME_BUCKET "); + tableIndexSQL.append("ON ").append(model.getName()).append("(").append(SegmentRecord.TIME_BUCKET).append(")"); + createIndex(client, connection, model, tableIndexSQL); + } catch (JDBCClientException e) { + throw new StorageException(e.getMessage(), e); + } catch (SQLException e) { + throw new StorageException(e.getMessage(), e); + } + } + + private void createSegmentIndexes(JDBCHikariCPClient client, Model model) throws StorageException { + try (Connection connection = client.getConnection()) { + SQLBuilder tableIndexSQL = new SQLBuilder("CREATE INDEX "); + tableIndexSQL.append(model.getName().toUpperCase()).append("_TRACE_ID "); + tableIndexSQL.append("ON ").append(model.getName()).append("(").append(SegmentRecord.TRACE_ID).append(")"); + createIndex(client, connection, model, tableIndexSQL); + + tableIndexSQL = new SQLBuilder("CREATE INDEX "); + tableIndexSQL.append(model.getName().toUpperCase()).append("_ENDPOINT_ID "); + tableIndexSQL.append("ON ").append(model.getName()).append("(").append(SegmentRecord.ENDPOINT_ID).append(")"); + createIndex(client, connection, model, tableIndexSQL); + + tableIndexSQL = new SQLBuilder("CREATE INDEX "); + tableIndexSQL.append(model.getName().toUpperCase()).append("_LATENCY "); + tableIndexSQL.append("ON ").append(model.getName()).append("(").append(SegmentRecord.LATENCY).append(")"); + createIndex(client, connection, model, tableIndexSQL); + + tableIndexSQL = new SQLBuilder("CREATE INDEX "); + tableIndexSQL.append(model.getName().toUpperCase()).append("_TIME_BUCKET "); + tableIndexSQL.append("ON ").append(model.getName()).append("(").append(SegmentRecord.TIME_BUCKET).append(")"); + createIndex(client, connection, model, tableIndexSQL); + } catch (JDBCClientException e) { + throw new StorageException(e.getMessage(), e); + } catch (SQLException e) { + throw new StorageException(e.getMessage(), e); + } + } + + private void createInventoryIndexes(JDBCHikariCPClient client, Model model) throws StorageException { + try (Connection connection = client.getConnection()) { + SQLBuilder tableIndexSQL = new SQLBuilder("CREATE UNIQUE INDEX "); + tableIndexSQL.append(model.getName().toUpperCase()).append("_SEQ "); + tableIndexSQL.append("ON ").append(model.getName()).append("(").append(RegisterSource.SEQUENCE).append(")"); + createIndex(client, connection, model, tableIndexSQL); + + tableIndexSQL = new SQLBuilder("CREATE INDEX "); + tableIndexSQL.append(model.getName().toUpperCase()).append("_TIME "); + tableIndexSQL.append("ON ").append(model.getName()).append("(").append(RegisterSource.HEARTBEAT_TIME).append(", ").append(RegisterSource.REGISTER_TIME).append(")"); + createIndex(client, connection, model, tableIndexSQL); + } catch (JDBCClientException e) { + throw new StorageException(e.getMessage(), e); + } catch (SQLException e) { + throw new StorageException(e.getMessage(), e); + } + } + + private void createIndex(JDBCHikariCPClient client, Connection connection, Model model, + SQLBuilder indexSQL) throws JDBCClientException { + if (logger.isDebugEnabled()) { + logger.debug("create index for table {}, sql: {} ", model.getName(), indexSQL.toStringInNewLine()); + } + client.execute(connection, indexSQL.toString()); + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTraceQueryDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTraceQueryDAO.java new file mode 100644 index 000000000..8a9417666 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/MySQLTraceQueryDAO.java @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentRecord; +import org.apache.skywalking.oap.server.core.query.entity.BasicTrace; +import org.apache.skywalking.oap.server.core.query.entity.QueryOrder; +import org.apache.skywalking.oap.server.core.query.entity.TraceBrief; +import org.apache.skywalking.oap.server.core.query.entity.TraceState; +import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.library.util.BooleanUtils; +import org.apache.skywalking.oap.server.library.util.StringUtils; +import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TraceQueryDAO; +import org.elasticsearch.search.sort.SortOrder; + +/** + * @author wusheng + */ +public class MySQLTraceQueryDAO extends H2TraceQueryDAO { + public MySQLTraceQueryDAO(JDBCHikariCPClient mysqlClient) { + super(mysqlClient); + } + + @Override + public TraceBrief queryBasicTraces(long startSecondTB, long endSecondTB, long minDuration, long maxDuration, + String endpointName, int serviceId, int endpointId, String traceId, int limit, int from, TraceState traceState, + QueryOrder queryOrder) throws IOException { + StringBuilder sql = new StringBuilder(); + List parameters = new ArrayList<>(10); + + sql.append("from ").append(SegmentRecord.INDEX_NAME).append(" where "); + sql.append(" 1=1 "); + if (startSecondTB != 0 && endSecondTB != 0) { + sql.append(" and ").append(SegmentRecord.TIME_BUCKET).append(" >= ?"); + parameters.add(startSecondTB); + sql.append(" and ").append(SegmentRecord.TIME_BUCKET).append(" <= ?"); + parameters.add(endSecondTB); + } + if (minDuration != 0 || maxDuration != 0) { + if (minDuration != 0) { + sql.append(" and ").append(SegmentRecord.LATENCY).append(" >= ?"); + parameters.add(minDuration); + } + if (maxDuration != 0) { + sql.append(" and ").append(SegmentRecord.LATENCY).append(" <= ?"); + parameters.add(maxDuration); + } + } + if (StringUtils.isNotEmpty(endpointName)) { + sql.append(" and ").append(SegmentRecord.ENDPOINT_NAME).append(" like '%" + endpointName + "%'"); + } + if (serviceId != 0) { + sql.append(" and ").append(SegmentRecord.SERVICE_ID).append(" = ?"); + parameters.add(serviceId); + } + if (endpointId != 0) { + sql.append(" and ").append(SegmentRecord.ENDPOINT_ID).append(" = ?"); + parameters.add(endpointId); + } + if (StringUtils.isNotEmpty(traceId)) { + sql.append(" and ").append(SegmentRecord.TRACE_ID).append(" = ?"); + parameters.add(traceId); + } + switch (traceState) { + case ERROR: + sql.append(" and ").append(SegmentRecord.IS_ERROR).append(" = ").append(BooleanUtils.TRUE); + break; + case SUCCESS: + sql.append(" and ").append(SegmentRecord.IS_ERROR).append(" = ").append(BooleanUtils.FALSE); + break; + } + switch (queryOrder) { + case BY_START_TIME: + sql.append(" order by ").append(SegmentRecord.START_TIME).append(" ").append(SortOrder.DESC); + break; + case BY_DURATION: + sql.append(" order by ").append(SegmentRecord.LATENCY).append(" ").append(SortOrder.DESC); + break; + } + + TraceBrief traceBrief = new TraceBrief(); + try (Connection connection = getClient().getConnection()) { + + try (ResultSet resultSet = getClient().executeQuery(connection, "select count(1) total from (select 1 " + sql.toString() + " ) AS TRACE", parameters.toArray(new Object[0]))) { + while (resultSet.next()) { + traceBrief.setTotal(resultSet.getInt("total")); + } + } + + buildLimit(sql, from, limit); + + try (ResultSet resultSet = getClient().executeQuery(connection, "select * " + sql.toString(), parameters.toArray(new Object[0]))) { + while (resultSet.next()) { + BasicTrace basicTrace = new BasicTrace(); + + basicTrace.setSegmentId(resultSet.getString(SegmentRecord.SEGMENT_ID)); + basicTrace.setStart(resultSet.getString(SegmentRecord.START_TIME)); + basicTrace.getEndpointNames().add(resultSet.getString(SegmentRecord.ENDPOINT_NAME)); + basicTrace.setDuration(resultSet.getInt(SegmentRecord.LATENCY)); + basicTrace.setError(BooleanUtils.valueToBoolean(resultSet.getInt(SegmentRecord.IS_ERROR))); + String traceIds = resultSet.getString(SegmentRecord.TRACE_ID); + basicTrace.getTraceIds().add(traceIds); + traceBrief.getTraces().add(basicTrace); + } + } + } catch (SQLException e) { + throw new IOException(e); + } + + return traceBrief; + } + + @Override protected void buildLimit(StringBuilder sql, int from, int limit) { + sql.append(" LIMIT ").append(from).append(", ").append(limit); + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider index 86c2c5661..c8fa0b9f8 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider @@ -16,4 +16,5 @@ # # -org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.H2StorageProvider \ No newline at end of file +org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.H2StorageProvider +org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql.MySQLStorageProvider \ No newline at end of file diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2RegisterLockInstaller.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/PreventRedistributionMySQLDriverTest.java similarity index 51% rename from oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2RegisterLockInstaller.java rename to oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/PreventRedistributionMySQLDriverTest.java index 1d1896967..77db0fe6b 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2RegisterLockInstaller.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/mysql/PreventRedistributionMySQLDriverTest.java @@ -16,29 +16,21 @@ * */ -package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2; +package org.apache.skywalking.oap.server.storage.plugin.jdbc.mysql; -import org.apache.skywalking.oap.server.core.storage.StorageException; -import org.apache.skywalking.oap.server.library.client.Client; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.junit.Test; /** + * This is a very special test case. It isn't for feature testing. + * + * In Apache, we can't redistribute MySQL Driver, because of GPL license, but we deliver MySQL solution source codes and + * distribution by using JDBC. + * * @author wusheng */ -public class H2RegisterLockInstaller { - public static final String LOCK_TABLE_NAME = "register_lock"; - - private static final Logger logger = LoggerFactory.getLogger(H2RegisterLockInstaller.class); - - /** - * For H2 storage, no concurrency situation, so, on lock table required. If someone wants to implement a storage by - * referring H2, please consider to create a LOCK table. - * - * @param client - * @throws StorageException - */ - public void install(Client client) throws StorageException { - +public class PreventRedistributionMySQLDriverTest { + @Test(expected = ClassNotFoundException.class) + public void TestMySQLDriverNotExist() throws ClassNotFoundException { + Class.forName("com.mysql.cj.jdbc.Driver"); } }