From 1554c087f408aa8bbe34d37d001be8174b7a0e69 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=88=98=E5=A8=81?= <51618159+LIU-WEI-git@users.noreply.github.com> Date: Wed, 30 Mar 2022 19:19:49 +0800 Subject: [PATCH] Refactor IoTDB storage plugin and bump up iotdb-session to 0.12.5 (#8755) --- CHANGES.md | 2 + dist-material/release-docs/LICENSE | 8 +- oap-server-bom/pom.xml | 2 +- .../storage-iotdb-plugin/pom.xml | 7 +- .../storage/plugin/iotdb/IoTDBClient.java | 110 ++-------- .../plugin/iotdb/IoTDBStorageProvider.java | 8 +- .../plugin/iotdb/IoTDBTableMetaInfo.java | 7 +- .../plugin/iotdb/base/IoTDBInsertRequest.java | 75 ++----- .../plugin/iotdb/base/IoTDBManagementDAO.java | 3 +- .../plugin/iotdb/base/IoTDBMetricsDAO.java | 7 +- .../plugin/iotdb/base/IoTDBNoneStreamDAO.java | 3 +- .../plugin/iotdb/base/IoTDBRecordDAO.java | 3 +- .../cache/IoTDBNetworkAddressAliasDAO.java | 9 +- .../IoTDBUITemplateManagementDAO.java | 52 ++--- .../profile/IoTDBProfileTaskLogQueryDAO.java | 28 +-- .../profile/IoTDBProfileTaskQueryDAO.java | 50 +++-- .../IoTDBProfileThreadSnapshotQueryDAO.java | 69 ++++--- .../iotdb/query/IoTDBAggregationQueryDAO.java | 22 +- .../iotdb/query/IoTDBAlarmQueryDAO.java | 15 +- .../iotdb/query/IoTDBBrowserLogQueryDAO.java | 34 +++- .../query/IoTDBEBPFProfilingDataDAO.java | 11 +- .../query/IoTDBEBPFProfilingScheduleDAO.java | 18 +- .../query/IoTDBEBPFProfilingTaskDAO.java | 21 +- .../iotdb/query/IoTDBEventQueryDAO.java | 43 ++-- .../plugin/iotdb/query/IoTDBLogQueryDAO.java | 36 +++- .../iotdb/query/IoTDBMetadataQueryDAO.java | 59 +++--- .../iotdb/query/IoTDBMetricsQueryDAO.java | 63 ++++-- .../iotdb/query/IoTDBTopNRecordsQueryDAO.java | 28 ++- .../iotdb/query/IoTDBTopologyQueryDAO.java | 74 ++++--- .../iotdb/query/IoTDBTraceQueryDAO.java | 43 ++-- .../iotdb/utils/IoTDBDataConverter.java | 189 ++++++++++++++++++ .../plugin/iotdb/utils/IoTDBUtils.java | 70 +++++++ .../cases/alarm/iotdb/docker-compose.yml | 2 +- .../cases/event/iotdb/docker-compose.yml | 2 +- .../e2e-v2/cases/log/iotdb/docker-compose.yml | 2 +- .../profiling/trace/iotdb/docker-compose.yml | 2 +- .../cases/storage/iotdb/docker-compose.yml | 2 +- .../e2e-v2/cases/ttl/iotdb/docker-compose.yml | 2 +- .../known-oap-backend-dependencies.txt | 8 +- 39 files changed, 755 insertions(+), 434 deletions(-) create mode 100644 oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/utils/IoTDBDataConverter.java create mode 100644 oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/utils/IoTDBUtils.java diff --git a/CHANGES.md b/CHANGES.md index 4fb9b78fed..c8f90770fb 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -148,6 +148,8 @@ NOTICE, this sharding concept is NOT just for splitting data into different data * Fix event type of export data is incorrect, it was `EventType.TOTAL` always. * Reduce redundancy ThreadLocal in MAL core. Improve MAL performance. * Trim tag's key and value in log query. +* Refactor IoTDB storage plugin, add IoTDBDataConverter and fix ModifyCollectionInEnhancedForLoop bug. +* Bump up iotdb-session to 0.12.5. #### UI diff --git a/dist-material/release-docs/LICENSE b/dist-material/release-docs/LICENSE index 48af9ec7a7..c69e7f88cb 100755 --- a/dist-material/release-docs/LICENSE +++ b/dist-material/release-docs/LICENSE @@ -329,10 +329,10 @@ The text of each license is the standard Apache 2.0 license. Armeria 1.14.1, http://github.com/line/armeria, Apache 2.0 Brotli4j 1.6.0, https://github.com/hyperxpro/Brotli4j, Apache 2.0 micrometer 1.8.2, https://github.com/micrometer-metrics/micrometer, Apache 2.0 - iotdb-session 0.12.4: https://github.com/apache/iotdb, Apache 2.0 - iotdb-thrift 0.12.4: https://github.com/apache/iotdb, Apache 2.0 - service-rpc 0.12.4: https://github.com/apache/iotdb, Apache 2.0 - tsfile 0.12.4 https://github.com/apache/iotdb Apache 2.0 + iotdb-session 0.12.5: https://github.com/apache/iotdb, Apache 2.0 + iotdb-thrift 0.12.5: https://github.com/apache/iotdb, Apache 2.0 + service-rpc 0.12.5: https://github.com/apache/iotdb, Apache 2.0 + tsfile 0.12.5 https://github.com/apache/iotdb Apache 2.0 libthrift 0.14.1: https://github.com/apache/thrift Apache 2.0 j2objc 1.3: https://github.com/google/j2objc Apache 2.0 diff --git a/oap-server-bom/pom.xml b/oap-server-bom/pom.xml index 666123f85f..fee971fea5 100644 --- a/oap-server-bom/pom.xml +++ b/oap-server-bom/pom.xml @@ -77,7 +77,7 @@ 3.0.0 4.4.13 1.21 - 0.12.4 + 0.12.5 1.6.0 diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/pom.xml b/oap-server/server-storage-plugin/storage-iotdb-plugin/pom.xml index 40b863a8f1..2fd833b57e 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/pom.xml +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/pom.xml @@ -17,7 +17,8 @@ ~ --> - + server-storage-plugin org.apache.skywalking @@ -62,8 +63,8 @@ - org.lz4 - lz4-java + org.lz4 + lz4-java diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBClient.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBClient.java index 0d7f88644f..1c2a0496fb 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBClient.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBClient.java @@ -21,8 +21,6 @@ package org.apache.skywalking.oap.server.storage.plugin.iotdb; import java.io.IOException; import java.util.ArrayList; import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import lombok.extern.slf4j.Slf4j; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; @@ -32,18 +30,16 @@ import org.apache.iotdb.session.pool.SessionPool; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.common.Field; import org.apache.iotdb.tsfile.read.common.RowRecord; -import org.apache.skywalking.oap.server.core.analysis.TimeBucket; -import org.apache.skywalking.oap.server.core.analysis.manual.log.LogRecord; -import org.apache.skywalking.oap.server.core.browser.manual.errorlog.BrowserErrorLogRecord; -import org.apache.skywalking.oap.server.core.management.ui.template.UITemplate; import org.apache.skywalking.oap.server.core.storage.StorageData; -import org.apache.skywalking.oap.server.core.storage.type.HashMapConverter; +import org.apache.skywalking.oap.server.core.storage.type.Convert2Entity; import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; import org.apache.skywalking.oap.server.library.client.Client; import org.apache.skywalking.oap.server.library.client.healthcheck.DelegatedHealthChecker; import org.apache.skywalking.oap.server.library.client.healthcheck.HealthCheckable; import org.apache.skywalking.oap.server.library.util.HealthChecker; import org.apache.skywalking.oap.server.storage.plugin.iotdb.base.IoTDBInsertRequest; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBDataConverter; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j public class IoTDBClient implements Client, HealthCheckable { @@ -69,11 +65,11 @@ public class IoTDBClient implements Client, HealthCheckable { public void connect() throws IoTDBConnectionException, StatementExecutionException { try { final int sessionPoolSize = config.getSessionPoolSize() == 0 ? - Runtime.getRuntime().availableProcessors() * 2 : config.getSessionPoolSize(); + Runtime.getRuntime().availableProcessors() * 2 : config.getSessionPoolSize(); log.info("SessionPool Size: {}", sessionPoolSize); - sessionPool = new SessionPool(config.getHost(), config.getRpcPort(), config.getUsername(), - config.getPassword(), sessionPoolSize, false, false - ); + sessionPool = new SessionPool(config.getHost(), config.getRpcPort(), + config.getUsername(), config.getPassword(), + sessionPoolSize, false, false); sessionPool.setStorageGroup(storageGroup); healthChecker.health(); @@ -119,9 +115,11 @@ public class IoTDBClient implements Client, HealthCheckable { devicePath.append(storageGroup).append(IoTDBClient.DOT).append(request.getModelName()); try { // make an index value as a layer name of the storage path + // every storage path has a fix order which has been set in IoTDBTableMetaInfo if (!request.getIndexes().isEmpty()) { - request.getIndexValues().forEach(value -> devicePath.append(IoTDBClient.DOT) - .append(indexValue2LayerName(value))); + request.getIndexValues().forEach( + value -> devicePath.append(IoTDBClient.DOT) + .append(IoTDBUtils.indexValue2LayerName(value))); } sessionPool.insertRecord(devicePath.toString(), request.getTime(), request.getMeasurements(), request.getMeasurementTypes(), @@ -158,8 +156,9 @@ public class IoTDBClient implements Client, HealthCheckable { devicePath.append(storageGroup).append(IoTDBClient.DOT).append(request.getModelName()); // make an index value as a layer name of the storage path if (!request.getIndexes().isEmpty()) { - request.getIndexValues().forEach(value -> devicePath.append(IoTDBClient.DOT) - .append(indexValue2LayerName(value))); + request.getIndexValues().forEach( + value -> devicePath.append(IoTDBClient.DOT) + .append(IoTDBUtils.indexValue2LayerName(value))); } devicePathList.add(devicePath.toString()); timeList.add(request.getTime()); @@ -188,7 +187,7 @@ public class IoTDBClient implements Client, HealthCheckable { */ public List filterQuery(String modelName, String querySQL, StorageBuilder storageBuilder) - throws IOException { + throws IOException { if (!querySQL.contains("align by device")) { throw new IOException("querySQL must contain \"align by device\""); } @@ -200,53 +199,14 @@ public class IoTDBClient implements Client, HealthCheckable { log.debug("SQL: {}, columnNames: {}", querySQL, wrapper.getColumnNames()); } - List columnNames = wrapper.getColumnNames(); IoTDBTableMetaInfo tableMetaInfo = IoTDBTableMetaInfo.get(modelName); List indexes = tableMetaInfo.getIndexes(); + List columnNames = wrapper.getColumnNames(); while (wrapper.hasNext()) { - Map map = new ConcurrentHashMap<>(); RowRecord rowRecord = wrapper.next(); - List fields = rowRecord.getFields(); - // transform timestamp to time_bucket - if (!UITemplate.INDEX_NAME.equals(modelName)) { - map.put(IoTDBClient.TIME_BUCKET, TimeBucket.getTimeBucket( - rowRecord.getTimestamp(), - tableMetaInfo.getModel().getDownsampling() - )); - } - // field.get(0) -> Device, transform layerName to indexValue - String[] layerNames = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\""); - for (int i = 0; i < indexes.size(); i++) { - map.put(indexes.get(i), layerName2IndexValue(layerNames[i + 1])); - } - for (int i = 0; i < columnNames.size() - 2; i++) { - String columnName = columnNames.get(i + 2); - Field field = fields.get(i + 1); - if (field.getDataType() == null) { - continue; - } - if (field.getDataType().equals(TSDataType.TEXT)) { - map.put(columnName, field.getStringValue()); - } else { - map.put(columnName, field.getObjectValue(field.getDataType())); - } - } - if (map.containsKey(IoTDBIndexes.LAYER_IDX)) { - String layer = (String) map.get(IoTDBIndexes.LAYER_IDX); - map.put(IoTDBIndexes.LAYER_IDX, Integer.valueOf(layer)); - } - if (modelName.equals(BrowserErrorLogRecord.INDEX_NAME) || modelName.equals(LogRecord.INDEX_NAME)) { - map.put(IoTDBClient.TIMESTAMP, map.get("\"" + IoTDBClient.TIMESTAMP + "\"")); - } - for (Map.Entry entry : map.entrySet()) { - // remove double quotes - String key = entry.getKey(); - if (key.contains(".")) { - map.put(key.substring(1, key.length() - 1), entry.getValue()); - } - } - - storageDataList.add(storageBuilder.storage2Entity(new HashMapConverter.ToEntity(map))); + Convert2Entity convert2Entity = + new IoTDBDataConverter.ToEntity(tableMetaInfo, indexes, columnNames, rowRecord); + storageDataList.add(storageBuilder.storage2Entity(convert2Entity)); } healthChecker.health(); } catch (IoTDBConnectionException | StatementExecutionException e) { @@ -316,35 +276,7 @@ public class IoTDBClient implements Client, HealthCheckable { } } - public String indexValue2LayerName(String indexValue) { - return "\"" + indexValue + "\""; - } - - public String layerName2IndexValue(String layerName) { - return layerName.substring(0, layerName.length() - 1); - } - - public StringBuilder addQueryIndexValue(String modelName, - StringBuilder query, - Map indexAndValueMap) { - List indexes = IoTDBTableMetaInfo.get(modelName).getIndexes(); - indexes.forEach(index -> { - if (indexAndValueMap.containsKey(index)) { - query.append(IoTDBClient.DOT).append(indexValue2LayerName(indexAndValueMap.get(index))); - } else { - query.append(IoTDBClient.DOT).append("*"); - } - }); - return query; - } - - public StringBuilder addQueryAsterisk(String modelName, StringBuilder query) { - List indexes = IoTDBTableMetaInfo.get(modelName).getIndexes(); - indexes.forEach(index -> query.append(IoTDBClient.DOT).append("*")); - return query; - } - - public StringBuilder addModelPath(StringBuilder query, String modelName) { - return query.append(storageGroup).append(IoTDBClient.DOT).append(modelName); + public String getStorageGroup() { + return storageGroup; } } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBStorageProvider.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBStorageProvider.java index d9deec790b..fd0e5455ea 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBStorageProvider.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBStorageProvider.java @@ -117,10 +117,10 @@ public class IoTDBStorageProvider extends ModuleProvider { this.registerServiceImplementation(UITemplateManagementDAO.class, new IoTDBUITemplateManagementDAO(client)); this.registerServiceImplementation(IProfileTaskLogQueryDAO.class, - new IoTDBProfileTaskLogQueryDAO(client, config.getFetchTaskLogMaxSize())); + new IoTDBProfileTaskLogQueryDAO(client, config.getFetchTaskLogMaxSize())); this.registerServiceImplementation(IProfileTaskQueryDAO.class, new IoTDBProfileTaskQueryDAO(client)); this.registerServiceImplementation(IProfileThreadSnapshotQueryDAO.class, - new IoTDBProfileThreadSnapshotQueryDAO(client)); + new IoTDBProfileThreadSnapshotQueryDAO(client)); this.registerServiceImplementation(IAggregationQueryDAO.class, new IoTDBAggregationQueryDAO(client)); this.registerServiceImplementation(IAlarmQueryDAO.class, new IoTDBAlarmQueryDAO(client)); @@ -141,8 +141,8 @@ public class IoTDBStorageProvider extends ModuleProvider { @Override public void start() throws ServiceNotProvidedException, ModuleStartException { MetricsCreator metricCreator = getManager().find(TelemetryModule.NAME) - .provider() - .getService(MetricsCreator.class); + .provider() + .getService(MetricsCreator.class); HealthCheckMetrics healthChecker = metricCreator.createHealthCheckerGauge( "storage_iotdb", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE); client.registerChecker(healthChecker); diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBTableMetaInfo.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBTableMetaInfo.java index 4c17e4913c..2567cde7f3 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBTableMetaInfo.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/IoTDBTableMetaInfo.java @@ -86,8 +86,11 @@ public class IoTDBTableMetaInfo { indexes.add(IoTDBIndexes.AGENT_ID_INX); } - final IoTDBTableMetaInfo tableMetaInfo = IoTDBTableMetaInfo.builder().model(model) - .columnAndTypeMap(columnAndTypeMap).indexes(indexes).build(); + final IoTDBTableMetaInfo tableMetaInfo = IoTDBTableMetaInfo.builder() + .model(model) + .columnAndTypeMap(columnAndTypeMap) + .indexes(indexes) + .build(); TABLE_META_INFOS.put(model.getName(), tableMetaInfo); } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBInsertRequest.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBInsertRequest.java index 56de4f8f7a..7f8f7e1e98 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBInsertRequest.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBInsertRequest.java @@ -19,23 +19,19 @@ package org.apache.skywalking.oap.server.storage.plugin.iotdb.base; import java.util.ArrayList; -import java.util.Iterator; import java.util.List; -import java.util.Map; import lombok.Getter; import lombok.Setter; import lombok.ToString; import lombok.extern.slf4j.Slf4j; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.skywalking.oap.server.core.Const; import org.apache.skywalking.oap.server.core.storage.StorageData; -import org.apache.skywalking.oap.server.core.storage.type.HashMapConverter; import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; -import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject; import org.apache.skywalking.oap.server.library.client.request.InsertRequest; import org.apache.skywalking.oap.server.library.client.request.UpdateRequest; -import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; -import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBTableMetaInfo; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBDataConverter; @Getter @Setter @@ -50,59 +46,24 @@ public class IoTDBInsertRequest implements InsertRequest, UpdateRequest { private List measurementTypes; private List measurementValues; - public IoTDBInsertRequest(String modelName, long time, T storageData, - StorageBuilder storageBuilder) { + public IoTDBInsertRequest(String modelName, long time) { this.modelName = modelName; this.time = time; - indexes = IoTDBTableMetaInfo.get(modelName).getIndexes(); - indexValues = new ArrayList<>(indexes.size()); - final HashMapConverter.ToStorage toStorage = new HashMapConverter.ToStorage(); + this.indexes = IoTDBTableMetaInfo.get(modelName).getIndexes(); + this.indexValues = new ArrayList<>(indexes.size()); + this.indexes.forEach(index -> this.indexValues.add(Const.EMPTY_STRING)); + + int measurementsSize = IoTDBTableMetaInfo.get(modelName).getColumnAndTypeMap().size(); + this.measurements = new ArrayList<>(measurementsSize); + this.measurementTypes = new ArrayList<>(measurementsSize); + this.measurementValues = new ArrayList<>(measurementsSize); + } + + public static IoTDBInsertRequest buildRequest(String modelName, long time, T storageData, + StorageBuilder storageBuilder) { + String id = storageData.id(); + IoTDBDataConverter.ToStorage toStorage = new IoTDBDataConverter.ToStorage(modelName, time, id); storageBuilder.entity2Storage(storageData, toStorage); - Map storageMap = toStorage.obtain(); - - indexes.forEach(index -> { - if (index.equals(IoTDBIndexes.ID_IDX)) { - indexValues.add(storageData.id()); - } else if (storageMap.containsKey(index)) { - // avoid indexValue be "null" when inserting - if (storageMap.get(index) == null) { - indexValues.add(""); - } else { - indexValues.add(String.valueOf(storageMap.get(index))); - } - storageMap.remove(index); - } - }); - - // time_bucket has changed to time before calling this method, so remove it from measurements - storageMap.remove(IoTDBClient.TIME_BUCKET); - // processing value to make it suitable for storage - Iterator> entryIterator = storageMap.entrySet().iterator(); - while (entryIterator.hasNext()) { - Map.Entry entry = entryIterator.next(); - // IoTDB doesn't allow insert null value. - if (entry.getValue() == null) { - entryIterator.remove(); - } - if (entry.getValue() instanceof StorageDataComplexObject) { - storageMap.put(entry.getKey(), ((StorageDataComplexObject) entry.getValue()).toStorageData()); - } - } - - measurements = new ArrayList<>(storageMap.keySet()); - Map columnAndTypeMap = IoTDBTableMetaInfo.get(modelName).getColumnAndTypeMap(); - measurementTypes = new ArrayList<>(measurements.size()); - for (String measurement : measurements) { - measurementTypes.add(columnAndTypeMap.get(measurement)); - } - measurementValues = new ArrayList<>(storageMap.values()); - - // IoTDB doesn't allow a measurement named `timestamp` or contains `.` - for (String key : storageMap.keySet()) { - if (key.equals(IoTDBClient.TIMESTAMP) || key.contains(".")) { - int idx = measurements.indexOf(key); - measurements.set(idx, "\"" + key + "\""); - } - } + return toStorage.obtain(); } } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBManagementDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBManagementDAO.java index 52bb6eddc9..7033e911d5 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBManagementDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBManagementDAO.java @@ -36,7 +36,8 @@ public class IoTDBManagementDAO implements IManagementDAO { @Override public void insert(Model model, ManagementData storageData) throws IOException { - IoTDBInsertRequest request = new IoTDBInsertRequest(model.getName(), 1L, storageData, storageBuilder); + IoTDBInsertRequest request = + IoTDBInsertRequest.buildRequest(model.getName(), 1L, storageData, storageBuilder); client.write(request); } } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBMetricsDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBMetricsDAO.java index 114290b68d..9dba2d7221 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBMetricsDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBMetricsDAO.java @@ -32,6 +32,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; import org.apache.skywalking.oap.server.library.client.request.InsertRequest; import org.apache.skywalking.oap.server.library.client.request.UpdateRequest; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -45,8 +46,8 @@ public class IoTDBMetricsDAO implements IMetricsDAO { query.append("select * from "); for (Metrics metric : metrics) { query.append(", "); - query = client.addModelPath(query, model.getName()); - query.append(IoTDBClient.DOT).append(client.indexValue2LayerName(metric.id())); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, model.getName()); + query.append(IoTDBClient.DOT).append(IoTDBUtils.indexValue2LayerName(metric.id())); } query.append(IoTDBClient.ALIGN_BY_DEVICE); String queryString = query.toString().replaceFirst(", ", ""); @@ -59,7 +60,7 @@ public class IoTDBMetricsDAO implements IMetricsDAO { @Override public InsertRequest prepareBatchInsert(Model model, Metrics metrics) { final long timestamp = TimeBucket.getTimestamp(metrics.getTimeBucket(), model.getDownsampling()); - return new IoTDBInsertRequest(model.getName(), timestamp, metrics, storageBuilder); + return IoTDBInsertRequest.buildRequest(model.getName(), timestamp, metrics, storageBuilder); } @Override diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBNoneStreamDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBNoneStreamDAO.java index 6a8a414b29..1885adb1ec 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBNoneStreamDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBNoneStreamDAO.java @@ -35,7 +35,8 @@ public class IoTDBNoneStreamDAO implements INoneStreamDAO { @Override public void insert(Model model, NoneStream noneStream) throws IOException { final long timestamp = TimeBucket.getTimestamp(noneStream.getTimeBucket(), model.getDownsampling()); - final IoTDBInsertRequest request = new IoTDBInsertRequest(model.getName(), timestamp, noneStream, storageBuilder); + final IoTDBInsertRequest request = + IoTDBInsertRequest.buildRequest(model.getName(), timestamp, noneStream, storageBuilder); client.write(request); } } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBRecordDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBRecordDAO.java index ff2375fbc5..b856e9ab2f 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBRecordDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/base/IoTDBRecordDAO.java @@ -40,7 +40,8 @@ public class IoTDBRecordDAO implements IRecordDAO { @Override public InsertRequest prepareBatchInsert(Model model, Record record) { final long timestamp = TimeBucket.getTimestamp(record.getTimeBucket(), model.getDownsampling()); - IoTDBInsertRequest request = new IoTDBInsertRequest(model.getName(), timestamp, record, storageBuilder); + IoTDBInsertRequest request = + IoTDBInsertRequest.buildRequest(model.getName(), timestamp, record, storageBuilder); // transform tags of SegmentRecord, LogRecord, AlarmRecord to tag1, tag2, ... List measurements = request.getMeasurements(); diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/cache/IoTDBNetworkAddressAliasDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/cache/IoTDBNetworkAddressAliasDAO.java index 47b79992b0..eecfcf5010 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/cache/IoTDBNetworkAddressAliasDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/cache/IoTDBNetworkAddressAliasDAO.java @@ -27,6 +27,7 @@ import org.apache.skywalking.oap.server.core.analysis.manual.networkalias.Networ import org.apache.skywalking.oap.server.core.storage.StorageData; import org.apache.skywalking.oap.server.core.storage.cache.INetworkAddressAliasDAO; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -38,14 +39,14 @@ public class IoTDBNetworkAddressAliasDAO implements INetworkAddressAliasDAO { public List loadLastUpdate(long timeBucket) { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, NetworkAddressAlias.INDEX_NAME); - query = client.addQueryAsterisk(NetworkAddressAlias.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, NetworkAddressAlias.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(NetworkAddressAlias.INDEX_NAME, query); query.append(" where ").append(NetworkAddressAlias.LAST_UPDATE_TIME_BUCKET).append(" >= ").append(timeBucket) - .append(IoTDBClient.ALIGN_BY_DEVICE); + .append(IoTDBClient.ALIGN_BY_DEVICE); try { List storageDataList = client.filterQuery(NetworkAddressAlias.INDEX_NAME, - query.toString(), storageBuilder); + query.toString(), storageBuilder); List networkAddressAliases = new ArrayList<>(storageDataList.size()); storageDataList.forEach(storageData -> networkAddressAliases.add((NetworkAddressAlias) storageData)); return networkAddressAliases; diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/management/IoTDBUITemplateManagementDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/management/IoTDBUITemplateManagementDAO.java index 471feee43c..9839074db1 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/management/IoTDBUITemplateManagementDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/management/IoTDBUITemplateManagementDAO.java @@ -37,6 +37,7 @@ import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; import org.apache.skywalking.oap.server.storage.plugin.iotdb.base.IoTDBInsertRequest; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -52,14 +53,14 @@ public class IoTDBUITemplateManagementDAO implements UITemplateManagementDAO { } StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, UITemplate.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, UITemplate.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.ID_IDX, id); - query = client.addQueryIndexValue(UITemplate.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(UITemplate.INDEX_NAME, query, indexAndValueMap); query.append(" limit 1").append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery( - UITemplate.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = + client.filterQuery(UITemplate.INDEX_NAME, query.toString(), storageBuilder); if (storageDataList.size() > 0) { return new DashboardConfiguration().fromEntity((UITemplate) storageDataList.get(0)); } @@ -70,20 +71,19 @@ public class IoTDBUITemplateManagementDAO implements UITemplateManagementDAO { public List getAllTemplates(Boolean includingDisabled) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, UITemplate.INDEX_NAME); - query = client.addQueryAsterisk(UITemplate.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, UITemplate.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(UITemplate.INDEX_NAME, query); if (!includingDisabled) { query.append(" where ").append(UITemplate.DISABLED).append(" = ").append(BooleanUtils.FALSE); } query.append(IoTDBClient.ALIGN_BY_DEVICE); List storageDataList = client.filterQuery(UITemplate.INDEX_NAME, query.toString(), - storageBuilder - ); + storageBuilder); List dashboardConfigurationList = new ArrayList<>(storageDataList.size()); - storageDataList.forEach(storageData -> - dashboardConfigurationList.add( - new DashboardConfiguration().fromEntity((UITemplate) storageData))); + storageDataList.forEach( + storageData -> dashboardConfigurationList.add( + new DashboardConfiguration().fromEntity((UITemplate) storageData))); return dashboardConfigurationList; } @@ -91,9 +91,9 @@ public class IoTDBUITemplateManagementDAO implements UITemplateManagementDAO { public TemplateChangeStatus addTemplate(DashboardSetting setting) throws IOException { final UITemplate uiTemplate = setting.toEntity(); - IoTDBInsertRequest request = new IoTDBInsertRequest(UITemplate.INDEX_NAME, UI_TEMPLATE_TIMESTAMP, - uiTemplate, storageBuilder - ); + IoTDBInsertRequest request = + IoTDBInsertRequest.buildRequest(UITemplate.INDEX_NAME, UI_TEMPLATE_TIMESTAMP, + uiTemplate, storageBuilder); client.write(request); return TemplateChangeStatus.builder().status(true).id(setting.getId()).build(); } @@ -104,11 +104,11 @@ public class IoTDBUITemplateManagementDAO implements UITemplateManagementDAO { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, UITemplate.INDEX_NAME); - query.append(IoTDBClient.DOT).append(client.indexValue2LayerName(uiTemplate.id())) + IoTDBUtils.addModelPath(client.getStorageGroup(), query, UITemplate.INDEX_NAME); + query.append(IoTDBClient.DOT).append(IoTDBUtils.indexValue2LayerName(uiTemplate.id())) .append(IoTDBClient.ALIGN_BY_DEVICE); List queryResult = client.filterQuery( - UITemplate.INDEX_NAME, query.toString(), storageBuilder); + UITemplate.INDEX_NAME, query.toString(), storageBuilder); if (queryResult.size() == 0) { return TemplateChangeStatus.builder() .status(false) @@ -116,9 +116,9 @@ public class IoTDBUITemplateManagementDAO implements UITemplateManagementDAO { .message("Can't find the template") .build(); } else { - IoTDBInsertRequest request = new IoTDBInsertRequest(UITemplate.INDEX_NAME, UI_TEMPLATE_TIMESTAMP, - uiTemplate, storageBuilder - ); + IoTDBInsertRequest request = + IoTDBInsertRequest.buildRequest(UITemplate.INDEX_NAME, UI_TEMPLATE_TIMESTAMP, + uiTemplate, storageBuilder); client.write(request); return TemplateChangeStatus.builder().status(true).id(setting.getId()).build(); } @@ -128,20 +128,20 @@ public class IoTDBUITemplateManagementDAO implements UITemplateManagementDAO { public TemplateChangeStatus disableTemplate(String id) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, UITemplate.INDEX_NAME); - query.append(IoTDBClient.DOT).append(client.indexValue2LayerName(id)) + IoTDBUtils.addModelPath(client.getStorageGroup(), query, UITemplate.INDEX_NAME); + query.append(IoTDBClient.DOT).append(IoTDBUtils.indexValue2LayerName(id)) .append(IoTDBClient.ALIGN_BY_DEVICE); List queryResult = client.filterQuery( - UITemplate.INDEX_NAME, query.toString(), storageBuilder); + UITemplate.INDEX_NAME, query.toString(), storageBuilder); if (queryResult.size() == 0) { return TemplateChangeStatus.builder().status(false).id(id).message("Can't find the template").build(); } else { final UITemplate uiTemplate = (UITemplate) queryResult.get(0); uiTemplate.setDisabled(BooleanUtils.TRUE); - IoTDBInsertRequest request = new IoTDBInsertRequest(UITemplate.INDEX_NAME, UI_TEMPLATE_TIMESTAMP, - uiTemplate, storageBuilder - ); + IoTDBInsertRequest request = + IoTDBInsertRequest.buildRequest(UITemplate.INDEX_NAME, UI_TEMPLATE_TIMESTAMP, + uiTemplate, storageBuilder); client.write(request); return TemplateChangeStatus.builder().status(true).id(id).build(); } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileTaskLogQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileTaskLogQueryDAO.java index 6331378825..a3dfbc1c4e 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileTaskLogQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileTaskLogQueryDAO.java @@ -30,6 +30,7 @@ import org.apache.skywalking.oap.server.core.storage.StorageData; import org.apache.skywalking.oap.server.core.storage.profiling.trace.IProfileTaskLogQueryDAO; import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -42,28 +43,31 @@ public class IoTDBProfileTaskLogQueryDAO implements IProfileTaskLogQueryDAO { public List getTaskLogList() throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ProfileTaskLogRecord.INDEX_NAME); - query = client.addQueryAsterisk(ProfileTaskLogRecord.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ProfileTaskLogRecord.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(ProfileTaskLogRecord.INDEX_NAME, query); query.append(" limit ").append(fetchTaskLogMaxSize).append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(ProfileTaskLogRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(ProfileTaskLogRecord.INDEX_NAME, + query.toString(), storageBuilder); List profileTaskLogRecordList = new ArrayList<>(storageDataList.size()); - storageDataList.forEach(storageData -> profileTaskLogRecordList.add((ProfileTaskLogRecord) storageData)); + storageDataList.forEach(storageData -> + profileTaskLogRecordList.add((ProfileTaskLogRecord) storageData)); // resort by self, because of the query result order by time. profileTaskLogRecordList.sort((ProfileTaskLogRecord r1, ProfileTaskLogRecord r2) -> - Long.compare(r2.getOperationTime(), r1.getOperationTime())); + Long.compare(r2.getOperationTime(), r1.getOperationTime())); List profileTaskLogList = new ArrayList<>(profileTaskLogRecordList.size()); - profileTaskLogRecordList.forEach(profileTaskLogRecord -> profileTaskLogList.add(parseLog(profileTaskLogRecord))); + profileTaskLogRecordList.forEach(profileTaskLogRecord -> + profileTaskLogList.add(parseLog(profileTaskLogRecord))); return profileTaskLogList; } private ProfileTaskLog parseLog(ProfileTaskLogRecord record) { return ProfileTaskLog.builder() - .id(record.id()) - .taskId(record.getTaskId()) - .instanceId(record.getInstanceId()) - .operationType(ProfileTaskLogOperationType.parse(record.getOperationType())) - .operationTime(record.getOperationTime()) - .build(); + .id(record.id()) + .taskId(record.getTaskId()) + .instanceId(record.getInstanceId()) + .operationType(ProfileTaskLogOperationType.parse(record.getOperationType())) + .operationTime(record.getOperationTime()) + .build(); } } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileTaskQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileTaskQueryDAO.java index 57df31bdff..d094d207ff 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileTaskQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileTaskQueryDAO.java @@ -34,6 +34,7 @@ import org.apache.skywalking.oap.server.core.storage.profiling.trace.IProfileTas import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -46,22 +47,28 @@ public class IoTDBProfileTaskQueryDAO implements IProfileTaskQueryDAO { Long endTimeBucket, Integer limit) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ProfileTaskRecord.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ProfileTaskRecord.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); if (StringUtil.isNotEmpty(serviceId)) { indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); } - query = client.addQueryIndexValue(ProfileTaskRecord.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(ProfileTaskRecord.INDEX_NAME, query, indexAndValueMap); StringBuilder where = new StringBuilder(" where "); if (StringUtil.isNotEmpty(endpointName)) { - where.append(ProfileTaskRecord.ENDPOINT_NAME).append(" = \"").append(endpointName).append("\"").append(" and "); + where.append(ProfileTaskRecord.ENDPOINT_NAME).append(" = \"") + .append(endpointName).append("\"") + .append(" and "); } if (Objects.nonNull(startTimeBucket)) { - where.append(IoTDBClient.TIME).append(" >= ").append(TimeBucket.getTimestamp(startTimeBucket)).append(" and "); + where.append(IoTDBClient.TIME).append(" >= ") + .append(TimeBucket.getTimestamp(startTimeBucket)) + .append(" and "); } if (Objects.nonNull(endTimeBucket)) { - where.append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endTimeBucket)).append(" and "); + where.append(IoTDBClient.TIME).append(" <= ") + .append(TimeBucket.getTimestamp(endTimeBucket)) + .append(" and "); } if (where.length() > 7) { int length = where.length(); @@ -73,9 +80,11 @@ public class IoTDBProfileTaskQueryDAO implements IProfileTaskQueryDAO { } query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(ProfileTaskRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(ProfileTaskRecord.INDEX_NAME, + query.toString(), storageBuilder); List profileTaskList = new ArrayList<>(storageDataList.size()); - storageDataList.forEach(storageData -> profileTaskList.add(record2ProfileTask((ProfileTaskRecord) storageData))); + storageDataList.forEach(storageData -> + profileTaskList.add(record2ProfileTask((ProfileTaskRecord) storageData))); return profileTaskList; } @@ -86,27 +95,28 @@ public class IoTDBProfileTaskQueryDAO implements IProfileTaskQueryDAO { } StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ProfileTaskRecord.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ProfileTaskRecord.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.ID_IDX, id); - query = client.addQueryIndexValue(ProfileTaskRecord.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(ProfileTaskRecord.INDEX_NAME, query, indexAndValueMap); query.append(" limit 1").append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(ProfileTaskRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(ProfileTaskRecord.INDEX_NAME, + query.toString(), storageBuilder); return record2ProfileTask((ProfileTaskRecord) storageDataList.get(0)); } private static ProfileTask record2ProfileTask(ProfileTaskRecord record) { return ProfileTask.builder() - .id(record.id()) - .serviceId(record.getServiceId()) - .endpointName(record.getEndpointName()) - .startTime(record.getStartTime()) - .createTime(record.getCreateTime()) - .duration(record.getDuration()) - .minDurationThreshold(record.getMinDurationThreshold()) - .dumpPeriod(record.getDumpPeriod()) - .maxSamplingCount(record.getMaxSamplingCount()) - .build(); + .id(record.id()) + .serviceId(record.getServiceId()) + .endpointName(record.getEndpointName()) + .startTime(record.getStartTime()) + .createTime(record.getCreateTime()) + .duration(record.getDuration()) + .minDurationThreshold(record.getMinDurationThreshold()) + .dumpPeriod(record.getDumpPeriod()) + .maxSamplingCount(record.getMaxSamplingCount()) + .build(); } } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileThreadSnapshotQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileThreadSnapshotQueryDAO.java index 3cb09a5a69..36bea97277 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileThreadSnapshotQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/profile/IoTDBProfileThreadSnapshotQueryDAO.java @@ -32,28 +32,32 @@ import org.apache.skywalking.oap.server.core.storage.profiling.trace.IProfileThr import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; import org.apache.skywalking.oap.server.library.util.BooleanUtils; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @RequiredArgsConstructor public class IoTDBProfileThreadSnapshotQueryDAO implements IProfileThreadSnapshotQueryDAO { private final IoTDBClient client; - private final StorageBuilder profileThreadSnapshotRecordBuilder = new ProfileThreadSnapshotRecord.Builder(); + private final StorageBuilder profileThreadSnapshotRecordBuilder = + new ProfileThreadSnapshotRecord.Builder(); private final StorageBuilder segmentRecordBuilder = new SegmentRecord.Builder(); @Override public List queryProfiledSegments(String taskId) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ProfileThreadSnapshotRecord.INDEX_NAME); - query = client.addQueryAsterisk(ProfileThreadSnapshotRecord.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ProfileThreadSnapshotRecord.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(ProfileThreadSnapshotRecord.INDEX_NAME, query); query.append(" where ").append(ProfileThreadSnapshotRecord.TASK_ID).append(" = \"").append(taskId).append("\"") - .append(" and ").append(ProfileThreadSnapshotRecord.SEQUENCE).append(" = 0") - .append(IoTDBClient.ALIGN_BY_DEVICE); + .append(" and ").append(ProfileThreadSnapshotRecord.SEQUENCE).append(" = 0") + .append(IoTDBClient.ALIGN_BY_DEVICE); List storageDataList = client.filterQuery(ProfileThreadSnapshotRecord.INDEX_NAME, - query.toString(), profileThreadSnapshotRecordBuilder); + query.toString(), + profileThreadSnapshotRecordBuilder); // We can insure the size of List, so use ArrayList to improve visit speed. (Other storage plugin use LinkedList) final List segmentIds = new ArrayList<>(storageDataList.size()); - storageDataList.forEach(storageData -> segmentIds.add(((ProfileThreadSnapshotRecord) storageData).getSegmentId())); + storageDataList.forEach( + storageData -> segmentIds.add(((ProfileThreadSnapshotRecord) storageData).getSegmentId())); if (segmentIds.isEmpty()) { return Collections.emptyList(); } @@ -62,8 +66,8 @@ public class IoTDBProfileThreadSnapshotQueryDAO implements IProfileThreadSnapsho // https://github.com/apache/iotdb/discussions/3888 query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, SegmentRecord.INDEX_NAME); - query = client.addQueryAsterisk(SegmentRecord.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, SegmentRecord.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(SegmentRecord.INDEX_NAME, query); query.append(" where ").append(SegmentRecord.SEGMENT_ID).append(" in ("); for (String segmentId : segmentIds) { query.append("\"").append(segmentId).append("\"").append(", "); @@ -74,14 +78,16 @@ public class IoTDBProfileThreadSnapshotQueryDAO implements IProfileThreadSnapsho List segmentRecordList = new ArrayList<>(storageDataList.size()); storageDataList.forEach(storageData -> segmentRecordList.add((SegmentRecord) storageData)); // resort by self, because of the select query result order by time. - segmentRecordList.sort((SegmentRecord r1, SegmentRecord r2) -> Long.compare(r2.getStartTime(), r1.getStartTime())); + segmentRecordList.sort((SegmentRecord r1, SegmentRecord r2) -> + Long.compare(r2.getStartTime(), r1.getStartTime())); List result = new ArrayList<>(segmentRecordList.size()); segmentRecordList.forEach(segmentRecord -> { BasicTrace basicTrace = new BasicTrace(); basicTrace.setSegmentId(segmentRecord.getSegmentId()); basicTrace.setStart(String.valueOf(segmentRecord.getStartTime())); - basicTrace.getEndpointNames().add(IDManager.EndpointID.analysisId(segmentRecord.getEndpointId()).getEndpointName()); + basicTrace.getEndpointNames().add(IDManager.EndpointID.analysisId(segmentRecord.getEndpointId()) + .getEndpointName()); basicTrace.setDuration(segmentRecord.getLatency()); basicTrace.setError(BooleanUtils.valueToBoolean(segmentRecord.getIsError())); basicTrace.getTraceIds().add(segmentRecord.getTraceId()); @@ -101,20 +107,23 @@ public class IoTDBProfileThreadSnapshotQueryDAO implements IProfileThreadSnapsho } @Override - public List queryRecords(String segmentId, int minSequence, int maxSequence) throws IOException { + public List queryRecords(String segmentId, int minSequence, int maxSequence) + throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ProfileThreadSnapshotRecord.INDEX_NAME); - query = client.addQueryAsterisk(ProfileThreadSnapshotRecord.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ProfileThreadSnapshotRecord.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(ProfileThreadSnapshotRecord.INDEX_NAME, query); query.append(" where ").append(ProfileThreadSnapshotRecord.SEGMENT_ID).append(" = \"").append(segmentId).append("\"") - .append(" and ").append(ProfileThreadSnapshotRecord.SEQUENCE).append(" >= ").append(minSequence) - .append(" and ").append(ProfileThreadSnapshotRecord.SEQUENCE).append(" <= ").append(maxSequence) - .append(IoTDBClient.ALIGN_BY_DEVICE); + .append(" and ").append(ProfileThreadSnapshotRecord.SEQUENCE).append(" >= ").append(minSequence) + .append(" and ").append(ProfileThreadSnapshotRecord.SEQUENCE).append(" <= ").append(maxSequence) + .append(IoTDBClient.ALIGN_BY_DEVICE); List storageDataList = client.filterQuery(ProfileThreadSnapshotRecord.INDEX_NAME, - query.toString(), profileThreadSnapshotRecordBuilder); + query.toString(), + profileThreadSnapshotRecordBuilder); List profileThreadSnapshotRecordList = new ArrayList<>(storageDataList.size()); - storageDataList.forEach(storageData -> profileThreadSnapshotRecordList.add((ProfileThreadSnapshotRecord) storageData)); + storageDataList.forEach( + storageData -> profileThreadSnapshotRecordList.add((ProfileThreadSnapshotRecord) storageData)); return profileThreadSnapshotRecordList; } @@ -122,13 +131,13 @@ public class IoTDBProfileThreadSnapshotQueryDAO implements IProfileThreadSnapsho public SegmentRecord getProfiledSegment(String segmentId) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, SegmentRecord.INDEX_NAME); - query = client.addQueryAsterisk(SegmentRecord.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, SegmentRecord.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(SegmentRecord.INDEX_NAME, query); query.append(" where ").append(SegmentRecord.SEGMENT_ID).append(" = \"").append(segmentId).append("\"") - .append(IoTDBClient.ALIGN_BY_DEVICE); + .append(IoTDBClient.ALIGN_BY_DEVICE); List storageDataList = client.filterQuery(SegmentRecord.INDEX_NAME, - query.toString(), segmentRecordBuilder); + query.toString(), segmentRecordBuilder); if (storageDataList.isEmpty()) { return null; } @@ -140,15 +149,15 @@ public class IoTDBProfileThreadSnapshotQueryDAO implements IProfileThreadSnapsho // See https://github.com/apache/iotdb/discussions/3907 StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ProfileThreadSnapshotRecord.INDEX_NAME); - query = client.addQueryAsterisk(ProfileThreadSnapshotRecord.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ProfileThreadSnapshotRecord.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(ProfileThreadSnapshotRecord.INDEX_NAME, query); query.append(" where ").append(ProfileThreadSnapshotRecord.SEGMENT_ID).append(" = \"").append(segmentId).append("\"") - .append(" and ").append(ProfileThreadSnapshotRecord.DUMP_TIME).append(" >= ").append(start) - .append(" and ").append(ProfileThreadSnapshotRecord.DUMP_TIME).append(" <= ").append(end) - .append(IoTDBClient.ALIGN_BY_DEVICE); + .append(" and ").append(ProfileThreadSnapshotRecord.DUMP_TIME).append(" >= ").append(start) + .append(" and ").append(ProfileThreadSnapshotRecord.DUMP_TIME).append(" <= ").append(end) + .append(IoTDBClient.ALIGN_BY_DEVICE); List storageDataList = client.filterQuery(ProfileThreadSnapshotRecord.INDEX_NAME, - query.toString(), profileThreadSnapshotRecordBuilder); - + query.toString(), + profileThreadSnapshotRecordBuilder); if (aggType.equals("min_value")) { int minValue = Integer.MAX_VALUE; for (Object storageData : storageDataList) { diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBAggregationQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBAggregationQueryDAO.java index 0572b46652..5cf12a16f1 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBAggregationQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBAggregationQueryDAO.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.storage.plugin.iotdb.query; +import com.google.common.base.Splitter; import java.io.IOException; import java.util.ArrayList; import java.util.Comparator; @@ -40,6 +41,7 @@ import org.apache.skywalking.oap.server.core.query.type.SelectedRecord; import org.apache.skywalking.oap.server.core.storage.query.IAggregationQueryDAO; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -49,11 +51,12 @@ public class IoTDBAggregationQueryDAO implements IAggregationQueryDAO { @Override public List sortMetrics(TopNCondition condition, String valueColumnName, Duration duration, List additionalConditions) throws IOException { - // This method maybe have poor efficiency. It queries all data which meets a condition without aggregation function. + // This method maybe have poor efficiency. + // It queries all data which meets a condition without aggregation function. // https://github.com/apache/iotdb/issues/4006 StringBuilder query = new StringBuilder(); query.append(String.format("select %s from ", valueColumnName)); - query = client.addModelPath(query, condition.getName()); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, condition.getName()); Map indexAndValueMap = new HashMap<>(); List measurementConditions = new ArrayList<>(); @@ -68,17 +71,17 @@ public class IoTDBAggregationQueryDAO implements IAggregationQueryDAO { } } if (!indexAndValueMap.isEmpty()) { - query = client.addQueryIndexValue(condition.getName(), query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(condition.getName(), query, indexAndValueMap); } else { - query = client.addQueryAsterisk(condition.getName(), query); + IoTDBUtils.addQueryAsterisk(condition.getName(), query); } query.append(" where ").append(IoTDBClient.TIME).append(" >= ").append(duration.getStartTimestamp()) - .append(" and ").append(IoTDBClient.TIME).append(" <= ").append(duration.getEndTimestamp()); + .append(" and ").append(IoTDBClient.TIME).append(" <= ").append(duration.getEndTimestamp()); if (!measurementConditions.isEmpty()) { for (KeyValue measurementCondition : measurementConditions) { query.append(" and ").append(measurementCondition.getKey()).append(" = \"") - .append(measurementCondition.getValue()).append("\""); + .append(measurementCondition.getValue()).append("\""); } } query.append(IoTDBClient.ALIGN_BY_DEVICE); @@ -97,8 +100,8 @@ public class IoTDBAggregationQueryDAO implements IAggregationQueryDAO { while (wrapper.hasNext()) { RowRecord rowRecord = wrapper.next(); List fields = rowRecord.getFields(); - String[] layerNames = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\""); - String entityId = client.layerName2IndexValue(layerNames[2]); + List layerNames = Splitter.on(IoTDBClient.DOT + "\"").splitToList(fields.get(0).getStringValue()); + String entityId = IoTDBUtils.layerName2IndexValue(layerNames.get(2)); double value = Double.parseDouble(fields.get(1).getStringValue()); entityIdAndSumMap.merge(entityId, value, Double::sum); entityIdAndCountMap.merge(entityId, 1, Integer::sum); @@ -122,7 +125,8 @@ public class IoTDBAggregationQueryDAO implements IAggregationQueryDAO { if (condition.getOrder().equals(Order.DES)) { topEntities.sort((SelectedRecord t1, SelectedRecord t2) -> - Double.compare(Double.parseDouble(t2.getValue()), Double.parseDouble(t1.getValue()))); + Double.compare(Double.parseDouble(t2.getValue()), + Double.parseDouble(t1.getValue()))); } else { topEntities.sort(Comparator.comparingDouble((SelectedRecord t) -> Double.parseDouble(t.getValue()))); } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBAlarmQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBAlarmQueryDAO.java index f646373736..084731cc15 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBAlarmQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBAlarmQueryDAO.java @@ -34,6 +34,7 @@ import org.apache.skywalking.oap.server.core.storage.query.IAlarmQueryDAO; import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @RequiredArgsConstructor public class IoTDBAlarmQueryDAO implements IAlarmQueryDAO { @@ -41,13 +42,15 @@ public class IoTDBAlarmQueryDAO implements IAlarmQueryDAO { private final StorageBuilder storageBuilder = new AlarmRecord.Builder(); @Override - public Alarms getAlarm(Integer scopeId, String keyword, int limit, int from, long startTB, long endTB, List tags) throws IOException { + public Alarms getAlarm(Integer scopeId, String keyword, int limit, int from, + long startTB, long endTB, List tags) + throws IOException { StringBuilder query = new StringBuilder(); // This method maybe have poor efficiency. It queries all data which meets a condition without select function. // https://github.com/apache/iotdb/discussions/3888 query.append("select * from "); - query = client.addModelPath(query, AlarmRecord.INDEX_NAME); - query = client.addQueryAsterisk(AlarmRecord.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, AlarmRecord.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(AlarmRecord.INDEX_NAME, query); StringBuilder where = new StringBuilder(" where "); if (Objects.nonNull(scopeId)) { @@ -73,7 +76,8 @@ public class IoTDBAlarmQueryDAO implements IAlarmQueryDAO { query.append(IoTDBClient.ALIGN_BY_DEVICE); Alarms alarms = new Alarms(); - List storageDataList = client.filterQuery(AlarmRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(AlarmRecord.INDEX_NAME, + query.toString(), storageBuilder); int limitCount = 0; for (int i = from; i < storageDataList.size(); i++) { if (limitCount < limit) { @@ -84,7 +88,8 @@ public class IoTDBAlarmQueryDAO implements IAlarmQueryDAO { } alarms.setTotal(storageDataList.size()); // resort by self, because of the select query result order by time. - alarms.getMsgs().sort((AlarmMessage m1, AlarmMessage m2) -> Long.compare(m2.getStartTime(), m1.getStartTime())); + alarms.getMsgs().sort((AlarmMessage m1, AlarmMessage m2) -> + Long.compare(m2.getStartTime(), m1.getStartTime())); return alarms; } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBBrowserLogQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBBrowserLogQueryDAO.java index dabbbb14f3..4d0cada97f 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBBrowserLogQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBBrowserLogQueryDAO.java @@ -40,6 +40,7 @@ import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -50,31 +51,42 @@ public class IoTDBBrowserLogQueryDAO implements IBrowserLogQueryDAO { @Override public BrowserErrorLogs queryBrowserErrorLogs(String serviceId, String serviceVersionId, String pagePathId, BrowserErrorCategory category, long startSecondTB, - long endSecondTB, int limit, int from) throws IOException { + long endSecondTB, int limit, int from) + throws IOException { StringBuilder query = new StringBuilder(); // This method maybe have poor efficiency. It queries all data which meets a condition without select function. // https://github.com/apache/iotdb/discussions/3888 query.append("select * from "); - query = client.addModelPath(query, BrowserErrorLogRecord.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, BrowserErrorLogRecord.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); if (StringUtil.isNotEmpty(serviceId)) { indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); } - query = client.addQueryIndexValue(BrowserErrorLogRecord.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(BrowserErrorLogRecord.INDEX_NAME, query, indexAndValueMap); StringBuilder where = new StringBuilder(" where "); if (startSecondTB != 0 && endSecondTB != 0) { - where.append(IoTDBClient.TIME).append(" >= ").append(TimeBucket.getTimestamp(startSecondTB)).append(" and "); - where.append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endSecondTB)).append(" and "); + where.append(IoTDBClient.TIME).append(" >= ") + .append(TimeBucket.getTimestamp(startSecondTB)) + .append(" and "); + where.append(IoTDBClient.TIME).append(" <= ") + .append(TimeBucket.getTimestamp(endSecondTB)) + .append(" and "); } if (StringUtil.isNotEmpty(serviceVersionId)) { - where.append(BrowserErrorLogRecord.SERVICE_VERSION_ID).append(" = \"").append(serviceVersionId).append("\"").append(" and "); + where.append(BrowserErrorLogRecord.SERVICE_VERSION_ID) + .append(" = \"").append(serviceVersionId).append("\"") + .append(" and "); } if (StringUtil.isNotEmpty(pagePathId)) { - where.append(BrowserErrorLogRecord.PAGE_PATH_ID).append(" = \"").append(pagePathId).append("\"").append(" and "); + where.append(BrowserErrorLogRecord.PAGE_PATH_ID) + .append(" = \"").append(pagePathId).append("\"") + .append(" and "); } if (Objects.nonNull(category)) { - where.append(BrowserErrorLogRecord.ERROR_CATEGORY).append(" = ").append(category.getValue()).append(" and "); + where.append(BrowserErrorLogRecord.ERROR_CATEGORY) + .append(" = ").append(category.getValue()) + .append(" and "); } if (where.length() > 7) { int length = where.length(); @@ -83,11 +95,13 @@ public class IoTDBBrowserLogQueryDAO implements IBrowserLogQueryDAO { } query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(BrowserErrorLogRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(BrowserErrorLogRecord.INDEX_NAME, + query.toString(), storageBuilder); List browserErrorLogRecordList = new ArrayList<>(storageDataList.size()); storageDataList.forEach(storageData -> browserErrorLogRecordList.add((BrowserErrorLogRecord) storageData)); // resort by self, because of the select query result order by time. - browserErrorLogRecordList.sort((BrowserErrorLogRecord b1, BrowserErrorLogRecord b2) -> Long.compare(b2.getTimestamp(), b1.getTimestamp())); + browserErrorLogRecordList.sort((BrowserErrorLogRecord b1, BrowserErrorLogRecord b2) -> + Long.compare(b2.getTimestamp(), b1.getTimestamp())); BrowserErrorLogs logs = new BrowserErrorLogs(); int limitCount = 0; for (int i = from; i < browserErrorLogRecordList.size(); i++) { diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingDataDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingDataDAO.java index b8d1a12a24..57c4bc63aa 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingDataDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingDataDAO.java @@ -31,6 +31,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -39,12 +40,13 @@ public class IoTDBEBPFProfilingDataDAO implements IEBPFProfilingDataDAO { private final StorageBuilder storageBuilder = new EBPFProfilingDataRecord.Builder(); @Override - public List queryData(String taskId, long beginTime, long endTime) throws IOException { + public List queryData(String taskId, long beginTime, long endTime) + throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, EBPFProfilingDataRecord.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, EBPFProfilingDataRecord.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); - query = client.addQueryIndexValue(EBPFProfilingDataRecord.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(EBPFProfilingDataRecord.INDEX_NAME, query, indexAndValueMap); StringBuilder where = new StringBuilder(" where "); where.append(EBPFProfilingDataRecord.TASK_ID).append(" = \"").append(taskId).append("\" and "); @@ -57,7 +59,8 @@ public class IoTDBEBPFProfilingDataDAO implements IEBPFProfilingDataDAO { } query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(EBPFProfilingDataRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(EBPFProfilingDataRecord.INDEX_NAME, + query.toString(), storageBuilder); List dataList = new ArrayList<>(storageDataList.size()); storageDataList.forEach(storageData -> dataList.add((EBPFProfilingDataRecord) storageData)); return dataList; diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingScheduleDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingScheduleDAO.java index bdc4456f25..3396f3a4d8 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingScheduleDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingScheduleDAO.java @@ -18,6 +18,9 @@ package org.apache.skywalking.oap.server.storage.plugin.iotdb.query; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.oap.server.core.analysis.TimeBucket; @@ -27,10 +30,7 @@ import org.apache.skywalking.oap.server.core.storage.StorageData; import org.apache.skywalking.oap.server.core.storage.profiling.ebpf.IEBPFProfilingScheduleDAO; import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -39,11 +39,12 @@ public class IoTDBEBPFProfilingScheduleDAO implements IEBPFProfilingScheduleDAO private final StorageBuilder storageBuilder = new EBPFProfilingScheduleRecord.Builder(); @Override - public List querySchedules(String taskId, long startTimeBucket, long endTimeBucket) throws IOException { + public List querySchedules(String taskId, long startTimeBucket, long endTimeBucket) + throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, EBPFProfilingScheduleRecord.INDEX_NAME); - query = client.addQueryAsterisk(EBPFProfilingScheduleRecord.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, EBPFProfilingScheduleRecord.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(EBPFProfilingScheduleRecord.INDEX_NAME, query); StringBuilder where = new StringBuilder(" where "); where.append(EBPFProfilingScheduleRecord.TASK_ID).append(" = \"").append(taskId).append("\" and "); @@ -56,7 +57,8 @@ public class IoTDBEBPFProfilingScheduleDAO implements IEBPFProfilingScheduleDAO } query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(EBPFProfilingScheduleRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(EBPFProfilingScheduleRecord.INDEX_NAME, + query.toString(), storageBuilder); List scheduleList = new ArrayList<>(storageDataList.size()); storageDataList.forEach(storageData -> scheduleList.add(parseSchedule((EBPFProfilingScheduleRecord) storageData))); return scheduleList; diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingTaskDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingTaskDAO.java index c095203f40..c4ec2dfebb 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingTaskDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEBPFProfilingTaskDAO.java @@ -40,6 +40,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -50,10 +51,11 @@ public class IoTDBEBPFProfilingTaskDAO implements IEBPFProfilingTaskDAO { @Override public List queryTasks(EBPFProfilingProcessFinder finder, EBPFProfilingTargetType targetType, - long taskStartTime, long latestUpdateTime) throws IOException { + long taskStartTime, long latestUpdateTime) + throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, EBPFProfilingTaskRecord.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, EBPFProfilingTaskRecord.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); if (StringUtil.isNotEmpty(finder.getServiceId())) { indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, finder.getServiceId()); @@ -69,18 +71,22 @@ public class IoTDBEBPFProfilingTaskDAO implements IEBPFProfilingTaskDAO { } final HashMap indexWithProcessId = new HashMap<>(indexAndValueMap); indexWithProcessId.put(IoTDBIndexes.PROCESS_ID_INX, processIdList.get(i)); - query = client.addQueryIndexValue(EBPFProfilingTaskRecord.INDEX_NAME, query, indexWithProcessId); + IoTDBUtils.addQueryIndexValue(EBPFProfilingTaskRecord.INDEX_NAME, query, indexWithProcessId); } } else { - query = client.addQueryIndexValue(EBPFProfilingTaskRecord.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(EBPFProfilingTaskRecord.INDEX_NAME, query, indexAndValueMap); } StringBuilder where = new StringBuilder(" where "); if (taskStartTime > 0) { - where.append(EBPFProfilingTaskRecord.START_TIME).append(" >= ").append(taskStartTime).append(" and "); + where.append(EBPFProfilingTaskRecord.START_TIME) + .append(" >= ").append(taskStartTime) + .append(" and "); } if (latestUpdateTime > 0) { - where.append(EBPFProfilingTaskRecord.LAST_UPDATE_TIME).append(" > ").append(latestUpdateTime).append(" and "); + where.append(EBPFProfilingTaskRecord.LAST_UPDATE_TIME) + .append(" > ").append(latestUpdateTime) + .append(" and "); } if (where.length() > 7) { int length = where.length(); @@ -89,7 +95,8 @@ public class IoTDBEBPFProfilingTaskDAO implements IEBPFProfilingTaskDAO { } query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(EBPFProfilingTaskRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(EBPFProfilingTaskRecord.INDEX_NAME, + query.toString(), storageBuilder); List taskList = new ArrayList<>(storageDataList.size()); storageDataList.forEach(storageData -> taskList.add(parseTask((EBPFProfilingTaskRecord) storageData))); return taskList; diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEventQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEventQueryDAO.java index 577a8f5ec8..37119fdff5 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEventQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBEventQueryDAO.java @@ -36,6 +36,7 @@ import org.apache.skywalking.oap.server.core.storage.StorageData; import org.apache.skywalking.oap.server.core.storage.query.IEventQueryDAO; import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -49,15 +50,16 @@ public class IoTDBEventQueryDAO implements IEventQueryDAO { // https://github.com/apache/iotdb/discussions/3888 StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, Event.INDEX_NAME); - query = client.addQueryAsterisk(Event.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, Event.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(Event.INDEX_NAME, query); StringBuilder where = whereSQL(condition); if (where.length() > 0) { query.append(" where ").append(where); } query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(Event.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(Event.INDEX_NAME, + query.toString(), storageBuilder); final Events events = new Events(); int limitCount = 0; PaginationUtils.Page page = PaginationUtils.INSTANCE.exchange(condition.getPaging()); @@ -89,8 +91,8 @@ public class IoTDBEventQueryDAO implements IEventQueryDAO { // https://github.com/apache/iotdb/discussions/3888 StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, Event.INDEX_NAME); - query = client.addQueryAsterisk(Event.INDEX_NAME, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, Event.INDEX_NAME); + IoTDBUtils.addQueryAsterisk(Event.INDEX_NAME, query); StringBuilder where = whereSQL(conditionList); if (where.length() > 0) { query.append(" where ").append(where); @@ -132,28 +134,42 @@ public class IoTDBEventQueryDAO implements IEventQueryDAO { final Source source = condition.getSource(); if (source != null) { if (!Strings.isNullOrEmpty(source.getService())) { - where.append(Event.SERVICE).append(" = \"").append(source.getService()).append("\"").append(" and "); + where.append(Event.SERVICE).append(" = \"") + .append(source.getService()).append("\"") + .append(" and "); } if (!Strings.isNullOrEmpty(source.getServiceInstance())) { - where.append(Event.SERVICE_INSTANCE).append(" = \"").append(source.getServiceInstance()).append("\"").append(" and "); + where.append(Event.SERVICE_INSTANCE).append(" = \"") + .append(source.getServiceInstance()).append("\"") + .append(" and "); } if (!Strings.isNullOrEmpty(source.getEndpoint())) { - where.append(Event.ENDPOINT).append(" = \"").append(source.getEndpoint()).append("\"").append(" and "); + where.append(Event.ENDPOINT).append(" = \"") + .append(source.getEndpoint()).append("\"") + .append(" and "); } } if (!Strings.isNullOrEmpty(condition.getName())) { - where.append(Event.NAME).append(" = \"").append(condition.getName()).append("\"").append(" and "); + where.append(Event.NAME).append(" = \"") + .append(condition.getName()).append("\"") + .append(" and "); } if (condition.getType() != null) { - where.append(Event.TYPE).append(" = \"").append(condition.getType().name()).append("\"").append(" and "); + where.append(Event.TYPE).append(" = \"") + .append(condition.getType().name()).append("\"") + .append(" and "); } final Duration time = condition.getTime(); if (time != null) { if (time.getStartTimestamp() > 0) { - where.append(Event.START_TIME).append(" > ").append(time.getStartTimestamp()).append(" and "); + where.append(Event.START_TIME).append(" > ") + .append(time.getStartTimestamp()) + .append(" and "); } if (time.getEndTimestamp() > 0) { - where.append(Event.END_TIME).append(" < ").append(time.getEndTimestamp()).append(" and "); + where.append(Event.END_TIME).append(" < ") + .append(time.getEndTimestamp()) + .append(" and "); } } if (where.length() > 0) { @@ -183,7 +199,8 @@ public class IoTDBEventQueryDAO implements IEventQueryDAO { } private org.apache.skywalking.oap.server.core.query.type.event.Event parseEvent(final Event event) { - final org.apache.skywalking.oap.server.core.query.type.event.Event resultEvent = new org.apache.skywalking.oap.server.core.query.type.event.Event(); + final org.apache.skywalking.oap.server.core.query.type.event.Event resultEvent = + new org.apache.skywalking.oap.server.core.query.type.event.Event(); resultEvent.setUuid(event.getUuid()); resultEvent.setSource(new Source(event.getService(), event.getServiceInstance(), event.getEndpoint())); resultEvent.setName(event.getName()); diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBLogQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBLogQueryDAO.java index 25f4f443dc..e7192b2461 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBLogQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBLogQueryDAO.java @@ -45,6 +45,7 @@ import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -61,7 +62,7 @@ public class IoTDBLogQueryDAO implements ILogQueryDAO { // This method maybe have poor efficiency. It queries all data which meets a condition without select function. // https://github.com/apache/iotdb/discussions/3888 query.append("select * from "); - query = client.addModelPath(query, LogRecord.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, LogRecord.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); if (StringUtil.isNotEmpty(serviceId)) { indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); @@ -69,30 +70,44 @@ public class IoTDBLogQueryDAO implements ILogQueryDAO { if (Objects.nonNull(relatedTrace) && StringUtil.isNotEmpty(relatedTrace.getTraceId())) { indexAndValueMap.put(IoTDBIndexes.TRACE_ID_IDX, relatedTrace.getTraceId()); } - query = client.addQueryIndexValue(LogRecord.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(LogRecord.INDEX_NAME, query, indexAndValueMap); StringBuilder where = new StringBuilder(" where "); if (startTB != 0 && endTB != 0) { - where.append(IoTDBClient.TIME).append(" >= ").append(TimeBucket.getTimestamp(startTB)).append(" and "); - where.append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endTB)).append(" and "); + where.append(IoTDBClient.TIME).append(" >= ") + .append(TimeBucket.getTimestamp(startTB)) + .append(" and "); + where.append(IoTDBClient.TIME).append(" <= ") + .append(TimeBucket.getTimestamp(endTB)) + .append(" and "); } if (StringUtil.isNotEmpty(serviceInstanceId)) { - where.append(AbstractLogRecord.SERVICE_INSTANCE_ID).append(" = \"").append(serviceInstanceId).append("\"").append(" and "); + where.append(AbstractLogRecord.SERVICE_INSTANCE_ID) + .append(" = \"").append(serviceInstanceId).append("\"") + .append(" and "); } if (StringUtil.isNotEmpty(endpointId)) { - where.append(AbstractLogRecord.ENDPOINT_ID).append(" = \"").append(endpointId).append("\"").append(" and "); + where.append(AbstractLogRecord.ENDPOINT_ID) + .append(" = \"").append(endpointId).append("\"") + .append(" and "); } if (Objects.nonNull(relatedTrace)) { if (StringUtil.isNotEmpty(relatedTrace.getSegmentId())) { - where.append(AbstractLogRecord.TRACE_SEGMENT_ID).append(" = \"").append(relatedTrace.getSegmentId()).append("\"").append(" and "); + where.append(AbstractLogRecord.TRACE_SEGMENT_ID) + .append(" = \"").append(relatedTrace.getSegmentId()).append("\"") + .append(" and "); } if (Objects.nonNull(relatedTrace.getSpanId())) { - where.append(AbstractLogRecord.SPAN_ID).append(" = ").append(relatedTrace.getSpanId()).append(" and "); + where.append(AbstractLogRecord.SPAN_ID) + .append(" = ").append(relatedTrace.getSpanId()) + .append(" and "); } } if (CollectionUtils.isNotEmpty(tags)) { for (final Tag tag : tags) { - where.append(tag.getKey()).append(" = \"").append(tag.getValue()).append("\"").append(" and "); + where.append(tag.getKey()).append(" = \"") + .append(tag.getValue()).append("\"") + .append(" and "); } } if (where.length() > 7) { @@ -103,7 +118,8 @@ public class IoTDBLogQueryDAO implements ILogQueryDAO { query.append(IoTDBClient.ALIGN_BY_DEVICE); Logs logs = new Logs(); - List storageDataList = client.filterQuery(LogRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(LogRecord.INDEX_NAME, + query.toString(), storageBuilder); int limitCount = 0; for (int i = from; i < storageDataList.size(); i++) { if (limitCount < limit) { diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBMetadataQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBMetadataQueryDAO.java index 5a8c56820d..ad3e6b11aa 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBMetadataQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBMetadataQueryDAO.java @@ -28,6 +28,7 @@ import java.util.List; import java.util.Map; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.apache.skywalking.oap.server.core.Const; import org.apache.skywalking.oap.server.core.analysis.IDManager; import org.apache.skywalking.oap.server.core.analysis.Layer; import org.apache.skywalking.oap.server.core.analysis.TimeBucket; @@ -48,6 +49,7 @@ import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder; import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -62,7 +64,7 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { public List listServices(final String layer, final String group) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ServiceTraffic.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ServiceTraffic.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); if (StringUtil.isNotEmpty(layer)) { indexAndValueMap.put(IoTDBIndexes.LAYER_IDX, String.valueOf(Layer.valueOf(layer).value())); @@ -70,10 +72,11 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { if (StringUtil.isNotEmpty(group)) { indexAndValueMap.put(IoTDBIndexes.GROUP_IDX, group); } - query = client.addQueryIndexValue(ServiceTraffic.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(ServiceTraffic.INDEX_NAME, query, indexAndValueMap); query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(ServiceTraffic.INDEX_NAME, query.toString(), serviceBuilder); + List storageDataList = client.filterQuery(ServiceTraffic.INDEX_NAME, + query.toString(), serviceBuilder); return buildServices(storageDataList); } @@ -81,29 +84,32 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { public List getServices(final String serviceId) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ServiceTraffic.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ServiceTraffic.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); - query = client.addQueryIndexValue(ServiceTraffic.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(ServiceTraffic.INDEX_NAME, query, indexAndValueMap); query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(ServiceTraffic.INDEX_NAME, query.toString(), serviceBuilder); + List storageDataList = client.filterQuery(ServiceTraffic.INDEX_NAME, + query.toString(), serviceBuilder); return buildServices(storageDataList); } @Override - public List listInstances(long startTimestamp, long endTimestamp, String serviceId) throws IOException { + public List listInstances(long startTimestamp, long endTimestamp, String serviceId) + throws IOException { final long minuteTimeBucket = TimeBucket.getMinuteTimeBucket(startTimestamp); StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, InstanceTraffic.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, InstanceTraffic.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); - query = client.addQueryIndexValue(InstanceTraffic.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(InstanceTraffic.INDEX_NAME, query, indexAndValueMap); query.append(" where ").append(InstanceTraffic.LAST_PING_TIME_BUCKET).append(" >= ").append(minuteTimeBucket) .append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(InstanceTraffic.INDEX_NAME, query.toString(), instanceBuilder); + List storageDataList = client.filterQuery(InstanceTraffic.INDEX_NAME, + query.toString(), instanceBuilder); return buildInstances(storageDataList); } @@ -111,13 +117,14 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { public ServiceInstance getInstance(final String instanceId) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ServiceTraffic.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ServiceTraffic.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.ID_IDX, instanceId); - query = client.addQueryIndexValue(ServiceTraffic.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(ServiceTraffic.INDEX_NAME, query, indexAndValueMap); query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(ServiceTraffic.INDEX_NAME, query.toString(), serviceBuilder); + List storageDataList = client.filterQuery(ServiceTraffic.INDEX_NAME, + query.toString(), serviceBuilder); final List instances = buildInstances(storageDataList); return instances.size() > 0 ? instances.get(0) : null; } @@ -126,16 +133,17 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { public List findEndpoint(String keyword, String serviceId, int limit) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, EndpointTraffic.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, EndpointTraffic.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); - query = client.addQueryIndexValue(EndpointTraffic.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(EndpointTraffic.INDEX_NAME, query, indexAndValueMap); if (!Strings.isNullOrEmpty(keyword)) { query.append(" where ").append(EndpointTraffic.NAME).append(" like '%").append(keyword).append("%'"); } query.append(" limit ").append(limit).append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(EndpointTraffic.INDEX_NAME, query.toString(), endpointBuilder); + List storageDataList = client.filterQuery(EndpointTraffic.INDEX_NAME, + query.toString(), endpointBuilder); List endpointList = new ArrayList<>(storageDataList.size()); storageDataList.forEach(storageData -> { EndpointTraffic endpointTraffic = (EndpointTraffic) storageData; @@ -148,10 +156,11 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { } @Override - public List listProcesses(String serviceId, String instanceId, String agentId) throws IOException { + public List listProcesses(String serviceId, String instanceId, String agentId) + throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ProcessTraffic.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ProcessTraffic.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); if (StringUtil.isNotEmpty(serviceId)) { indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); @@ -162,10 +171,11 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { if (StringUtil.isNotEmpty(agentId)) { indexAndValueMap.put(IoTDBIndexes.AGENT_ID_INX, agentId); } - query = client.addQueryIndexValue(ProcessTraffic.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(ProcessTraffic.INDEX_NAME, query, indexAndValueMap); query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(ProcessTraffic.INDEX_NAME, query.toString(), processBuilder); + List storageDataList = client.filterQuery(ProcessTraffic.INDEX_NAME, + query.toString(), processBuilder); return buildProcesses(storageDataList); } @@ -173,13 +183,14 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { public Process getProcess(String processId) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, ProcessTraffic.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, ProcessTraffic.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.ID_IDX, processId); - query = client.addQueryIndexValue(ProcessTraffic.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(ProcessTraffic.INDEX_NAME, query, indexAndValueMap); query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(ProcessTraffic.INDEX_NAME, query.toString(), processBuilder); + List storageDataList = client.filterQuery(ProcessTraffic.INDEX_NAME, + query.toString(), processBuilder); final List processes = buildProcesses(storageDataList); return processes.size() > 0 ? processes.get(0) : null; } @@ -205,7 +216,7 @@ public class IoTDBMetadataQueryDAO implements IMetadataQueryDAO { storageDataList.forEach(storageData -> { InstanceTraffic instanceTraffic = (InstanceTraffic) storageData; if (instanceTraffic.getName() == null) { - instanceTraffic.setName(""); + instanceTraffic.setName(Const.EMPTY_STRING); } ServiceInstance serviceInstance = new ServiceInstance(); serviceInstance.setId(instanceTraffic.id()); diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBMetricsQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBMetricsQueryDAO.java index eff62d199a..1f1cd8ab7b 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBMetricsQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBMetricsQueryDAO.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.storage.plugin.iotdb.query; +import com.google.common.base.Splitter; import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; @@ -45,6 +46,7 @@ import org.apache.skywalking.oap.server.core.storage.annotation.ValueColumnMetad import org.apache.skywalking.oap.server.core.storage.query.IMetricsQueryDAO; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -52,7 +54,10 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { private final IoTDBClient client; @Override - public long readMetricsValue(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException { + public long readMetricsValue(MetricsCondition condition, + String valueColumnName, + Duration duration) + throws IOException { final int defaultValue = ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()); final Function function = ValueColumnMetadata.INSTANCE.getValueFunction(condition.getName()); if (function == Function.Latest) { @@ -67,18 +72,20 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { op = "sum"; } query.append(String.format("select %s(%s) from ", op, valueColumnName)); - query = client.addModelPath(query, condition.getName()); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, condition.getName()); final String entityId = condition.getEntity().buildId(); if (entityId != null) { Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.ENTITY_ID_IDX, entityId); - query = client.addQueryIndexValue(condition.getName(), query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(condition.getName(), query, indexAndValueMap); } else { - query = client.addQueryAsterisk(condition.getName(), query); + IoTDBUtils.addQueryAsterisk(condition.getName(), query); } - query.append(" where ").append(String.format("%s >= %s and %s <= %s", - IoTDBClient.TIME, duration.getStartTimestamp(), IoTDBClient.TIME, duration.getEndTimestamp())) - .append(" group by level = 3"); + query.append(" where ") + .append(String.format("%s >= %s and %s <= %s", + IoTDBClient.TIME, duration.getStartTimestamp(), + IoTDBClient.TIME, duration.getEndTimestamp())) + .append(" group by level = 3"); List results = client.queryWithAgg(query.toString()); if (results.size() > 0) { @@ -100,10 +107,10 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { StringBuilder query = new StringBuilder(); query.append("select ").append(valueColumnName).append(" from "); for (String id : ids) { - query = client.addModelPath(query, condition.getName()); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, condition.getName()); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.ID_IDX, id); - query = client.addQueryIndexValue(condition.getName(), query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(condition.getName(), query, indexAndValueMap); query.append(", "); } String queryString = query.toString(); @@ -126,8 +133,9 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { while (wrapper.hasNext()) { RowRecord rowRecord = wrapper.next(); List fields = rowRecord.getFields(); - String[] layerNames = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\""); - String id = client.layerName2IndexValue(layerNames[1]); + List layerNames = Splitter.on(IoTDBClient.DOT + "\"") + .splitToList(fields.get(0).getStringValue()); + String id = IoTDBUtils.layerName2IndexValue(layerNames.get(1)); Field valueField = fields.get(1); TSDataType valueType = valueField.getDataType(); @@ -150,12 +158,18 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { sessionPool.closeResultSet(wrapper); } } - metricsValues.setValues(Util.sortValues(intValues, ids, ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()))); + metricsValues.setValues( + Util.sortValues(intValues, ids, + ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName()))); return metricsValues; } @Override - public List readLabeledMetricsValues(MetricsCondition condition, String valueColumnName, List labels, Duration duration) throws IOException { + public List readLabeledMetricsValues(MetricsCondition condition, + String valueColumnName, + List labels, + Duration duration) + throws IOException { final List pointOfTimes = duration.assembleDurationPoints(); List ids = new ArrayList<>(pointOfTimes.size()); pointOfTimes.forEach(pointOfTime -> ids.add(pointOfTime.id(condition.getEntity().buildId()))); @@ -163,10 +177,10 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { StringBuilder query = new StringBuilder(); query.append("select ").append(valueColumnName).append(" from "); for (String id : ids) { - query = client.addModelPath(query, condition.getName()); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, condition.getName()); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.ID_IDX, id); - query = client.addQueryIndexValue(condition.getName(), query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(condition.getName(), query, indexAndValueMap); query.append(", "); } String queryString = query.toString(); @@ -187,8 +201,9 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { while (wrapper.hasNext()) { RowRecord rowRecord = wrapper.next(); List fields = rowRecord.getFields(); - String[] layerNames = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\""); - String id = client.layerName2IndexValue(layerNames[1]); + List layerNames = Splitter.on(IoTDBClient.DOT + "\"") + .splitToList(fields.get(0).getStringValue()); + String id = IoTDBUtils.layerName2IndexValue(layerNames.get(1)); DataTable multipleValues = new DataTable(5); multipleValues.toObject(fields.get(1).getStringValue()); @@ -205,7 +220,10 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { } @Override - public HeatMap readHeatMap(MetricsCondition condition, String valueColumnName, Duration duration) throws IOException { + public HeatMap readHeatMap(MetricsCondition condition, + String valueColumnName, + Duration duration) + throws IOException { final List pointOfTimes = duration.assembleDurationPoints(); List ids = new ArrayList<>(pointOfTimes.size()); pointOfTimes.forEach(pointOfTime -> ids.add(pointOfTime.id(condition.getEntity().buildId()))); @@ -213,10 +231,10 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { StringBuilder query = new StringBuilder(); query.append("select ").append(valueColumnName).append(" from "); for (String id : ids) { - query = client.addModelPath(query, condition.getName()); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, condition.getName()); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.ID_IDX, id); - query = client.addQueryIndexValue(condition.getName(), query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(condition.getName(), query, indexAndValueMap); query.append(", "); } String queryString = query.toString(); @@ -238,8 +256,9 @@ public class IoTDBMetricsQueryDAO implements IMetricsQueryDAO { while (wrapper.hasNext()) { RowRecord rowRecord = wrapper.next(); List fields = rowRecord.getFields(); - String[] layerNames = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\""); - String id = client.layerName2IndexValue(layerNames[1]); + List layerNames = Splitter.on(IoTDBClient.DOT + "\"") + .splitToList(fields.get(0).getStringValue()); + String id = IoTDBUtils.layerName2IndexValue(layerNames.get(1)); heatMap.buildColumn(id, fields.get(1).getStringValue(), defaultValue); } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTopNRecordsQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTopNRecordsQueryDAO.java index da2f6b347d..b5fdd7cf99 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTopNRecordsQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTopNRecordsQueryDAO.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.storage.plugin.iotdb.query; +import com.google.common.base.Splitter; import java.io.IOException; import java.util.ArrayList; import java.util.Comparator; @@ -45,6 +46,7 @@ import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBTableMetaInfo; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -52,22 +54,27 @@ public class IoTDBTopNRecordsQueryDAO implements ITopNRecordsQueryDAO { private final IoTDBClient client; @Override - public List readSampledRecords(TopNCondition condition, String valueColumnName, Duration duration) throws IOException { + public List readSampledRecords(TopNCondition condition, + String valueColumnName, + Duration duration) + throws IOException { StringBuilder query = new StringBuilder(); query.append("select ").append(TopN.STATEMENT).append(", ").append(valueColumnName) - .append(" from "); - query = client.addModelPath(query, condition.getName()); + .append(" from "); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, condition.getName()); Map indexAndValueMap = new HashMap<>(); if (StringUtil.isNotEmpty(condition.getParentService())) { final String serviceId = IDManager.ServiceID.buildId(condition.getParentService(), condition.isNormal()); indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); } - query = client.addQueryIndexValue(condition.getName(), query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(condition.getName(), query, indexAndValueMap); StringBuilder where = new StringBuilder(" where "); if (Objects.nonNull(duration)) { - where.append(IoTDBClient.TIME).append(" >= ").append(TimeBucket.getTimestamp(duration.getStartTimeBucketInSec())).append(" and "); - where.append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(duration.getEndTimeBucketInSec())); + where.append(IoTDBClient.TIME).append(" >= ") + .append(TimeBucket.getTimestamp(duration.getStartTimeBucketInSec())).append(" and "); + where.append(IoTDBClient.TIME).append(" <= ") + .append(TimeBucket.getTimestamp(duration.getEndTimeBucketInSec())); } if (where.length() > 7) { query.append(where); @@ -92,8 +99,10 @@ public class IoTDBTopNRecordsQueryDAO implements ITopNRecordsQueryDAO { List fields = rowRecord.getFields(); record.setName(fields.get(1).getStringValue()); - String traceId = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\"")[traceIdIdx + 1]; - traceId = client.layerName2IndexValue(traceId); + String traceId = Splitter.on(IoTDBClient.DOT + "\"") + .splitToList(fields.get(0).getStringValue()) + .get(traceIdIdx + 1); + traceId = IoTDBUtils.layerName2IndexValue(traceId); record.setRefId(traceId); record.setId(record.getRefId()); @@ -111,7 +120,8 @@ public class IoTDBTopNRecordsQueryDAO implements ITopNRecordsQueryDAO { // resort by self, because of the select query result order by time. if (Order.DES.equals(condition.getOrder())) { records.sort((SelectedRecord s1, SelectedRecord s2) -> - Long.compare(Long.parseLong(s2.getValue()), Long.parseLong(s1.getValue()))); + Long.compare(Long.parseLong(s2.getValue()), + Long.parseLong(s1.getValue()))); } else { records.sort(Comparator.comparingLong((SelectedRecord s) -> Long.parseLong(s.getValue()))); } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTopologyQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTopologyQueryDAO.java index dd343b855e..8e70053865 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTopologyQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTopologyQueryDAO.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.storage.plugin.iotdb.query; +import com.google.common.base.Splitter; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -39,6 +40,7 @@ import org.apache.skywalking.oap.server.core.query.type.Call; import org.apache.skywalking.oap.server.core.source.DetectPoint; import org.apache.skywalking.oap.server.core.storage.query.ITopologyQueryDAO; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @Slf4j @RequiredArgsConstructor @@ -47,7 +49,8 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { @Override public List loadServiceRelationsDetectedAtServerSide(long startTB, long endTB, - List serviceIds) throws IOException { + List serviceIds) + throws IOException { return loadServiceCalls( ServiceRelationServerSideMetrics.INDEX_NAME, startTB, endTB, ServiceRelationServerSideMetrics.SOURCE_SERVICE_ID, @@ -57,7 +60,8 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { @Override public List loadServiceRelationDetectedAtClientSide(long startTB, long endTB, - List serviceIds) throws IOException { + List serviceIds) + throws IOException { return loadServiceCalls( ServiceRelationClientSideMetrics.INDEX_NAME, startTB, endTB, ServiceRelationClientSideMetrics.SOURCE_SERVICE_ID, @@ -66,7 +70,8 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { } @Override - public List loadServiceRelationsDetectedAtServerSide(long startTB, long endTB) throws IOException { + public List loadServiceRelationsDetectedAtServerSide(long startTB, long endTB) + throws IOException { return loadServiceCalls( ServiceRelationServerSideMetrics.INDEX_NAME, startTB, endTB, ServiceRelationServerSideMetrics.SOURCE_SERVICE_ID, @@ -75,7 +80,8 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { } @Override - public List loadServiceRelationDetectedAtClientSide(long startTB, long endTB) throws IOException { + public List loadServiceRelationDetectedAtClientSide(long startTB, long endTB) + throws IOException { return loadServiceCalls( ServiceRelationClientSideMetrics.INDEX_NAME, startTB, endTB, ServiceRelationClientSideMetrics.SOURCE_SERVICE_ID, @@ -86,7 +92,8 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { @Override public List loadInstanceRelationDetectedAtServerSide(String clientServiceId, String serverServiceId, - long startTB, long endTB) throws IOException { + long startTB, long endTB) + throws IOException { return loadServiceInstanceCalls( ServiceInstanceRelationServerSideMetrics.INDEX_NAME, startTB, endTB, ServiceInstanceRelationServerSideMetrics.SOURCE_SERVICE_ID, @@ -97,7 +104,8 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { @Override public List loadInstanceRelationDetectedAtClientSide(String clientServiceId, String serverServiceId, - long startTB, long endTB) throws IOException { + long startTB, long endTB) + throws IOException { return loadServiceInstanceCalls( ServiceInstanceRelationClientSideMetrics.INDEX_NAME, startTB, endTB, ServiceInstanceRelationClientSideMetrics.SOURCE_SERVICE_ID, @@ -106,7 +114,8 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { } @Override - public List loadEndpointRelation(long startTB, long endTB, String destEndpointId) throws IOException { + public List loadEndpointRelation(long startTB, long endTB, String destEndpointId) + throws IOException { List calls = loadEndpointFromSide( EndpointRelationServerSideMetrics.INDEX_NAME, startTB, endTB, EndpointRelationServerSideMetrics.SOURCE_ENDPOINT, @@ -122,20 +131,20 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { private List loadServiceCalls(String tableName, long startTB, long endTB, String sourceCName, String destCName, - List serviceIds, DetectPoint detectPoint) throws IOException { + List serviceIds, DetectPoint detectPoint) + throws IOException { // This method don't use "group by" like other storage plugin. StringBuilder query = new StringBuilder(); query.append("select ").append(ServiceRelationServerSideMetrics.COMPONENT_ID).append(" from "); - query = client.addModelPath(query, tableName); - query = client.addQueryAsterisk(tableName, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, tableName); + IoTDBUtils.addQueryAsterisk(tableName, query); query.append(" where ").append(IoTDBClient.TIME).append(" >= ").append(TimeBucket.getTimestamp(startTB)) - .append(" and ").append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endTB)); + .append(" and ").append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endTB)); if (serviceIds.size() > 0) { query.append(" and ("); for (int i = 0; i < serviceIds.size(); i++) { - query.append(sourceCName).append(" = \"").append(serviceIds.get(i)) - .append("\" or ") - .append(destCName).append(" = \"").append(serviceIds.get(i)).append("\""); + query.append(sourceCName).append(" = \"").append(serviceIds.get(i)).append("\" or ") + .append(destCName).append(" = \"").append(serviceIds.get(i)).append("\""); if (i != serviceIds.size() - 1) { query.append(" or "); } @@ -157,8 +166,9 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { RowRecord rowRecord = wrapper.next(); List fields = rowRecord.getFields(); Call.CallDetail call = new Call.CallDetail(); - String[] layerNames = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\""); - String entityId = client.layerName2IndexValue(layerNames[2]); + List layerNames = Splitter.on(IoTDBClient.DOT + "\"") + .splitToList(fields.get(0).getStringValue()); + String entityId = IoTDBUtils.layerName2IndexValue(layerNames.get(2)); final int componentId = fields.get(1).getIntV(); call.buildFromServiceRelation(entityId, componentId, detectPoint); calls.add(call); @@ -180,15 +190,15 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { // This method don't use "group by" like other storage plugin. StringBuilder query = new StringBuilder(); query.append("select ").append(ServiceInstanceRelationServerSideMetrics.COMPONENT_ID).append(" from "); - query = client.addModelPath(query, tableName); - query = client.addQueryAsterisk(tableName, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, tableName); + IoTDBUtils.addQueryAsterisk(tableName, query); query.append(" where ").append(IoTDBClient.TIME).append(" >= ").append(TimeBucket.getTimestamp(startTB)) - .append(" and ").append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endTB)); + .append(" and ").append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endTB)); query.append(" and ((").append(sourceCName).append(" = \"").append(sourceServiceId).append("\"") - .append(" and ").append(descCName).append(" = \"").append(destServiceId).append("\"") - .append(") or (").append(sourceCName).append(" = \"").append(destServiceId).append("\"") - .append(" and ").append(descCName).append(" = \"").append(sourceServiceId).append(" \"))") - .append(IoTDBClient.ALIGN_BY_DEVICE); + .append(" and ").append(descCName).append(" = \"").append(destServiceId).append("\"") + .append(") or (").append(sourceCName).append(" = \"").append(destServiceId).append("\"") + .append(" and ").append(descCName).append(" = \"").append(sourceServiceId).append(" \"))") + .append(IoTDBClient.ALIGN_BY_DEVICE); SessionPool sessionPool = client.getSessionPool(); SessionDataSetWrapper wrapper = null; @@ -202,8 +212,9 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { RowRecord rowRecord = wrapper.next(); List fields = rowRecord.getFields(); Call.CallDetail call = new Call.CallDetail(); - String[] layerNames = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\""); - String entityId = client.layerName2IndexValue(layerNames[2]); + List layerNames = Splitter.on(IoTDBClient.DOT + "\"") + .splitToList(fields.get(0).getStringValue()); + String entityId = IoTDBUtils.layerName2IndexValue(layerNames.get(2)); final int componentId = fields.get(1).getIntV(); call.buildFromInstanceRelation(entityId, componentId, detectPoint); calls.add(call); @@ -224,12 +235,12 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { // This method don't use "group by" like other storage plugin. StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, tableName); - query = client.addQueryAsterisk(tableName, query); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, tableName); + IoTDBUtils.addQueryAsterisk(tableName, query); query.append(" where ").append(IoTDBClient.TIME).append(" >= ").append(TimeBucket.getTimestamp(startTB)) - .append(" and ").append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endTB)); + .append(" and ").append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endTB)); query.append(" and ").append(isSourceId ? sourceCName : destCName).append(" = \"").append(id).append("\"") - .append(IoTDBClient.ALIGN_BY_DEVICE); + .append(IoTDBClient.ALIGN_BY_DEVICE); SessionPool sessionPool = client.getSessionPool(); SessionDataSetWrapper wrapper = null; @@ -243,8 +254,9 @@ public class IoTDBTopologyQueryDAO implements ITopologyQueryDAO { RowRecord rowRecord = wrapper.next(); List fields = rowRecord.getFields(); Call.CallDetail call = new Call.CallDetail(); - String[] layerNames = fields.get(0).getStringValue().split("\\" + IoTDBClient.DOT + "\""); - String entityId = client.layerName2IndexValue(layerNames[2]); + List layerNames = Splitter.on(IoTDBClient.DOT + "\"") + .splitToList(fields.get(0).getStringValue()); + String entityId = IoTDBUtils.layerName2IndexValue(layerNames.get(2)); call.buildFromEndpointRelation(entityId, DetectPoint.SERVER); calls.add(call); } diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTraceQueryDAO.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTraceQueryDAO.java index 229421d735..ec5584e8d3 100644 --- a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTraceQueryDAO.java +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/query/IoTDBTraceQueryDAO.java @@ -43,6 +43,7 @@ import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.utils.IoTDBUtils; @RequiredArgsConstructor public class IoTDBTraceQueryDAO implements ITraceQueryDAO { @@ -59,7 +60,7 @@ public class IoTDBTraceQueryDAO implements ITraceQueryDAO { // This method maybe have poor efficiency. It queries all data which meets a condition without select function. // https://github.com/apache/iotdb/discussions/3888 query.append("select * from "); - query = client.addModelPath(query, SegmentRecord.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, SegmentRecord.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); if (StringUtil.isNotEmpty(serviceId)) { indexAndValueMap.put(IoTDBIndexes.SERVICE_ID_IDX, serviceId); @@ -67,12 +68,16 @@ public class IoTDBTraceQueryDAO implements ITraceQueryDAO { if (!Strings.isNullOrEmpty(traceId)) { indexAndValueMap.put(IoTDBIndexes.TRACE_ID_IDX, traceId); } - query = client.addQueryIndexValue(SegmentRecord.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(SegmentRecord.INDEX_NAME, query, indexAndValueMap); StringBuilder where = new StringBuilder(" where "); if (startSecondTB != 0 && endSecondTB != 0) { - where.append(IoTDBClient.TIME).append(" >= ").append(TimeBucket.getTimestamp(startSecondTB)).append(" and "); - where.append(IoTDBClient.TIME).append(" <= ").append(TimeBucket.getTimestamp(endSecondTB)).append(" and "); + where.append(IoTDBClient.TIME).append(" >= ") + .append(TimeBucket.getTimestamp(startSecondTB)) + .append(" and "); + where.append(IoTDBClient.TIME).append(" <= ") + .append(TimeBucket.getTimestamp(endSecondTB)) + .append(" and "); } if (minDuration != 0) { where.append(SegmentRecord.LATENCY).append(" >= ").append(minDuration).append(" and "); @@ -81,14 +86,20 @@ public class IoTDBTraceQueryDAO implements ITraceQueryDAO { where.append(SegmentRecord.LATENCY).append(" <= ").append(maxDuration).append(" and "); } if (StringUtil.isNotEmpty(serviceInstanceId)) { - where.append(SegmentRecord.SERVICE_INSTANCE_ID).append(" = \"").append(serviceInstanceId).append("\"").append(" and "); + where.append(SegmentRecord.SERVICE_INSTANCE_ID + ).append(" = \"").append(serviceInstanceId).append("\"") + .append(" and "); } if (!Strings.isNullOrEmpty(endpointId)) { - where.append(SegmentRecord.ENDPOINT_ID).append(" = \"").append(endpointId).append("\"").append(" and "); + where.append(SegmentRecord.ENDPOINT_ID) + .append(" = \"").append(endpointId).append("\"") + .append(" and "); } if (CollectionUtils.isNotEmpty(tags)) { for (final Tag tag : tags) { - where.append(tag.getKey()).append(" = \"").append(tag.getValue()).append("\"").append(" and "); + where.append(tag.getKey()).append(" = \"") + .append(tag.getValue()).append("\"") + .append(" and "); } } switch (traceState) { @@ -107,7 +118,8 @@ public class IoTDBTraceQueryDAO implements ITraceQueryDAO { query.append(IoTDBClient.ALIGN_BY_DEVICE); TraceBrief traceBrief = new TraceBrief(); - List storageDataList = client.filterQuery(SegmentRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(SegmentRecord.INDEX_NAME, + query.toString(), storageBuilder); int limitCount = 0; for (int i = from; i < storageDataList.size(); i++) { if (limitCount < limit) { @@ -116,7 +128,8 @@ public class IoTDBTraceQueryDAO implements ITraceQueryDAO { BasicTrace basicTrace = new BasicTrace(); basicTrace.setSegmentId(segmentRecord.getSegmentId()); basicTrace.setStart(String.valueOf(segmentRecord.getStartTime())); - basicTrace.getEndpointNames().add(IDManager.EndpointID.analysisId(segmentRecord.getEndpointId()).getEndpointName()); + basicTrace.getEndpointNames().add( + IDManager.EndpointID.analysisId(segmentRecord.getEndpointId()).getEndpointName()); basicTrace.setDuration(segmentRecord.getLatency()); basicTrace.setError(BooleanUtils.valueToBoolean(segmentRecord.getIsError())); basicTrace.getTraceIds().add(segmentRecord.getTraceId()); @@ -128,11 +141,12 @@ public class IoTDBTraceQueryDAO implements ITraceQueryDAO { switch (queryOrder) { case BY_START_TIME: traceBrief.getTraces().sort((BasicTrace b1, BasicTrace b2) -> - Long.compare(Long.parseLong(b2.getStart()), Long.parseLong(b1.getStart()))); + Long.compare(Long.parseLong(b2.getStart()), + Long.parseLong(b1.getStart()))); break; case BY_DURATION: traceBrief.getTraces().sort((BasicTrace b1, BasicTrace b2) -> - Integer.compare(b2.getDuration(), b1.getDuration())); + Integer.compare(b2.getDuration(), b1.getDuration())); break; } return traceBrief; @@ -142,13 +156,14 @@ public class IoTDBTraceQueryDAO implements ITraceQueryDAO { public List queryByTraceId(String traceId) throws IOException { StringBuilder query = new StringBuilder(); query.append("select * from "); - query = client.addModelPath(query, SegmentRecord.INDEX_NAME); + IoTDBUtils.addModelPath(client.getStorageGroup(), query, SegmentRecord.INDEX_NAME); Map indexAndValueMap = new HashMap<>(); indexAndValueMap.put(IoTDBIndexes.TRACE_ID_IDX, traceId); - query = client.addQueryIndexValue(SegmentRecord.INDEX_NAME, query, indexAndValueMap); + IoTDBUtils.addQueryIndexValue(SegmentRecord.INDEX_NAME, query, indexAndValueMap); query.append(IoTDBClient.ALIGN_BY_DEVICE); - List storageDataList = client.filterQuery(SegmentRecord.INDEX_NAME, query.toString(), storageBuilder); + List storageDataList = client.filterQuery(SegmentRecord.INDEX_NAME, + query.toString(), storageBuilder); List segmentRecords = new ArrayList<>(storageDataList.size()); storageDataList.forEach(storageData -> segmentRecords.add((SegmentRecord) storageData)); return segmentRecords; diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/utils/IoTDBDataConverter.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/utils/IoTDBDataConverter.java new file mode 100644 index 0000000000..60089cd793 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/utils/IoTDBDataConverter.java @@ -0,0 +1,189 @@ +/* + * 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.iotdb.utils; + +import com.google.common.base.Splitter; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.function.Function; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.read.common.Field; +import org.apache.iotdb.tsfile.read.common.RowRecord; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.analysis.TimeBucket; +import org.apache.skywalking.oap.server.core.storage.type.Convert2Entity; +import org.apache.skywalking.oap.server.core.storage.type.Convert2Storage; +import org.apache.skywalking.oap.server.core.storage.type.StorageDataComplexObject; +import org.apache.skywalking.oap.server.library.util.CollectionUtils; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBIndexes; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBTableMetaInfo; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.base.IoTDBInsertRequest; + +public class IoTDBDataConverter { + + public static class ToEntity implements Convert2Entity { + private final IoTDBTableMetaInfo tableMetaInfo; + private final List indexValues; + private final List columnNames; + private final long timestamp; + private final List fields; + + public ToEntity(IoTDBTableMetaInfo tableMetaInfo, List indexes, + List columnNames, RowRecord rowRecord) { + this.tableMetaInfo = tableMetaInfo; + this.indexValues = new ArrayList<>(tableMetaInfo.getIndexes().size()); + this.columnNames = columnNames; + this.timestamp = rowRecord.getTimestamp(); + this.fields = rowRecord.getFields(); + // field.get(0) -> Device, transform every layerName to indexValue + List layerNames = Splitter.on(IoTDBClient.DOT + "\"") + .splitToList(fields.get(0).getStringValue()); + for (int i = 0; i < indexes.size(); i++) { + indexValues.add(IoTDBUtils.layerName2IndexValue(layerNames.get(i + 1))); + } + } + + @Override + public Object get(String fieldName) { + if (IoTDBIndexes.isIndex(fieldName)) { + if (tableMetaInfo.getIndexes().contains(fieldName)) { + String indexValue = indexValues.get(tableMetaInfo.getIndexes().indexOf(fieldName)); + // convert String to Integer for LAYER_IDX value + return IoTDBIndexes.LAYER_IDX.equals(fieldName) ? Integer.valueOf(indexValue) : indexValue; + } else { + return null; + } + } else { + // time_bucket has changed to timestamp when writing to IoTDB. + if (IoTDBClient.TIME_BUCKET.equals(fieldName)) { + return TimeBucket.getTimeBucket(timestamp, tableMetaInfo.getModel().getDownsampling()); + } + // IoTDB doesn't allow a measurement named `timestamp` or contains `.`, + // so we add double quotation mark to them. + // Also see IoTDBDataConverter.ToStorage.accept(String, Object) + if (IoTDBClient.TIMESTAMP.equals(fieldName) || fieldName.contains(".")) { + String columnName = IoTDBUtils.addQuotationMark(fieldName); + return columnNames.contains(columnName) + ? IoTDBUtils.getFieldValue(fields.get(columnNames.indexOf(columnName) - 1)) + : null; + } else { + return columnNames.contains(fieldName) + ? IoTDBUtils.getFieldValue(fields.get(columnNames.indexOf(fieldName) - 1)) + : null; + } + } + } + + @Override + public R getWith(final String fieldName, final Function typeDecoder) { + if (columnNames.contains(fieldName)) { + final T value = (T) IoTDBUtils.getFieldValue(fields.get(columnNames.indexOf(fieldName) - 1)); + return typeDecoder.apply(value); + } else { + return null; + } + } + } + + public static class ToStorage implements Convert2Storage { + private final IoTDBInsertRequest request; + private final IoTDBTableMetaInfo tableMetaInfo; + + public ToStorage(String modelName, long time, String id) { + this.request = new IoTDBInsertRequest(modelName, time); + this.tableMetaInfo = IoTDBTableMetaInfo.get(modelName); + // set id as an index + this.request.getIndexValues().set(this.request.getIndexes().indexOf(IoTDBIndexes.ID_IDX), id); + } + + @Override + public void accept(final String fieldName, final Object fieldValue) { + if (IoTDBIndexes.isIndex(fieldName)) { + List indexes = request.getIndexes(); + List indexValues = request.getIndexValues(); + // To avoid indexValue be "null" when inserting, replace null to empty string + if (indexes.contains(fieldName)) { + indexValues.set(indexes.indexOf(fieldName), + fieldValue == null ? Const.EMPTY_STRING : fieldValue.toString()); + } + } else { + // time_bucket has changed to timestamp before calling this method, + // and IoTDB v0.12 doesn't allow insert null value, + // so they don't need to be stored. + if (fieldName.equals(IoTDBClient.TIME_BUCKET) || fieldValue == null) { + return; + } + + List measurements = request.getMeasurements(); + List measurementTypes = request.getMeasurementTypes(); + List measurementValues = request.getMeasurementValues(); + // IoTDB doesn't allow a measurement named `timestamp` or contains `.` + if (fieldName.equals(IoTDBClient.TIMESTAMP) || fieldName.contains(".")) { + measurements.add(IoTDBUtils.addQuotationMark(fieldName)); + } else { + measurements.add(fieldName); + } + measurementTypes.add(tableMetaInfo.getColumnAndTypeMap().get(fieldName)); + if (fieldValue instanceof StorageDataComplexObject) { + measurementValues.add(((StorageDataComplexObject) fieldValue).toStorageData()); + } else { + measurementValues.add(fieldValue); + } + } + } + + @Override + public void accept(final String fieldName, final byte[] fieldValue) { + request.getMeasurements().add(fieldName); + request.getMeasurementTypes().add(tableMetaInfo.getColumnAndTypeMap().get(fieldName)); + if (CollectionUtils.isEmpty(fieldValue)) { + request.getMeasurementValues().add(Const.EMPTY_STRING); + } else { + request.getMeasurementValues().add(new String(Base64.getEncoder().encode(fieldValue))); + } + } + + @Override + public void accept(final String fieldName, final List fieldValue) { + request.getMeasurements().add(fieldName); + request.getMeasurementTypes().add(tableMetaInfo.getColumnAndTypeMap().get(fieldName)); + request.getMeasurementValues().add(fieldValue); + } + + @Override + public Object get(String fieldName) { + if (IoTDBIndexes.isIndex(fieldName)) { + return request.getIndexes().contains(fieldName) + ? request.getIndexValues().get(request.getIndexes().indexOf(fieldName)) + : null; + } else { + return request.getMeasurements().contains(fieldName) + ? request.getMeasurementValues().get(request.getMeasurements().indexOf(fieldName)) + : null; + } + } + + @Override + public IoTDBInsertRequest obtain() { + return request; + } + } +} diff --git a/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/utils/IoTDBUtils.java b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/utils/IoTDBUtils.java new file mode 100644 index 0000000000..91caeb0255 --- /dev/null +++ b/oap-server/server-storage-plugin/storage-iotdb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/iotdb/utils/IoTDBUtils.java @@ -0,0 +1,70 @@ +/* + * 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.iotdb.utils; + +import java.util.List; +import java.util.Map; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; +import org.apache.iotdb.tsfile.read.common.Field; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBClient; +import org.apache.skywalking.oap.server.storage.plugin.iotdb.IoTDBTableMetaInfo; + +public class IoTDBUtils { + public static String addQuotationMark(String string) { + return "\"" + string + "\""; + } + + public static String indexValue2LayerName(String indexValue) { + return addQuotationMark(indexValue); + } + + public static String layerName2IndexValue(String layerName) { + return layerName.substring(0, layerName.length() - 1); + } + + public static void addQueryIndexValue(String modelName, + StringBuilder query, + Map indexAndValueMap) { + List indexes = IoTDBTableMetaInfo.get(modelName).getIndexes(); + indexes.forEach(index -> { + if (indexAndValueMap.containsKey(index)) { + query.append(IoTDBClient.DOT).append(indexValue2LayerName(indexAndValueMap.get(index))); + } else { + query.append(IoTDBClient.DOT).append("*"); + } + }); + } + + public static void addQueryAsterisk(String modelName, StringBuilder query) { + List indexes = IoTDBTableMetaInfo.get(modelName).getIndexes(); + indexes.forEach(index -> query.append(IoTDBClient.DOT).append("*")); + } + + public static void addModelPath(String storageGroup, StringBuilder query, String modelName) { + query.append(storageGroup).append(IoTDBClient.DOT).append(modelName); + } + + public static Object getFieldValue(Field field) { + if (field.getDataType().equals(TSDataType.TEXT)) { + return field.getStringValue(); + } else { + return field.getObjectValue(field.getDataType()); + } + } +} diff --git a/test/e2e-v2/cases/alarm/iotdb/docker-compose.yml b/test/e2e-v2/cases/alarm/iotdb/docker-compose.yml index f00f91fd80..64151fda0b 100644 --- a/test/e2e-v2/cases/alarm/iotdb/docker-compose.yml +++ b/test/e2e-v2/cases/alarm/iotdb/docker-compose.yml @@ -17,7 +17,7 @@ version: '2.1' services: iotdb: - image: apache/iotdb:0.12.4-node + image: apache/iotdb:0.12.5-node expose: - 6667 networks: diff --git a/test/e2e-v2/cases/event/iotdb/docker-compose.yml b/test/e2e-v2/cases/event/iotdb/docker-compose.yml index 557238855e..f0e9c5d7d1 100644 --- a/test/e2e-v2/cases/event/iotdb/docker-compose.yml +++ b/test/e2e-v2/cases/event/iotdb/docker-compose.yml @@ -17,7 +17,7 @@ version: '3.8' services: iotdb: - image: apache/iotdb:0.12.4-node + image: apache/iotdb:0.12.5-node expose: - 6667 networks: diff --git a/test/e2e-v2/cases/log/iotdb/docker-compose.yml b/test/e2e-v2/cases/log/iotdb/docker-compose.yml index 32e40745c6..d7d04bea41 100644 --- a/test/e2e-v2/cases/log/iotdb/docker-compose.yml +++ b/test/e2e-v2/cases/log/iotdb/docker-compose.yml @@ -17,7 +17,7 @@ version: '2.1' services: iotdb: - image: apache/iotdb:0.12.4-node + image: apache/iotdb:0.12.5-node expose: - 6667 networks: diff --git a/test/e2e-v2/cases/profiling/trace/iotdb/docker-compose.yml b/test/e2e-v2/cases/profiling/trace/iotdb/docker-compose.yml index 437fc63b95..9eed5ff59a 100644 --- a/test/e2e-v2/cases/profiling/trace/iotdb/docker-compose.yml +++ b/test/e2e-v2/cases/profiling/trace/iotdb/docker-compose.yml @@ -17,7 +17,7 @@ version: '2.1' services: iotdb: - image: apache/iotdb:0.12.4-node + image: apache/iotdb:0.12.5-node expose: - 6667 networks: diff --git a/test/e2e-v2/cases/storage/iotdb/docker-compose.yml b/test/e2e-v2/cases/storage/iotdb/docker-compose.yml index f22e55db4d..80c49995d1 100644 --- a/test/e2e-v2/cases/storage/iotdb/docker-compose.yml +++ b/test/e2e-v2/cases/storage/iotdb/docker-compose.yml @@ -17,7 +17,7 @@ version: '2.1' services: iotdb: - image: apache/iotdb:0.12.4-node + image: apache/iotdb:0.12.5-node expose: - 6667 networks: diff --git a/test/e2e-v2/cases/ttl/iotdb/docker-compose.yml b/test/e2e-v2/cases/ttl/iotdb/docker-compose.yml index 39068a15f3..3aa1f03bb2 100644 --- a/test/e2e-v2/cases/ttl/iotdb/docker-compose.yml +++ b/test/e2e-v2/cases/ttl/iotdb/docker-compose.yml @@ -17,7 +17,7 @@ version: '2.1' services: iotdb: - image: apache/iotdb:0.12.4-node + image: apache/iotdb:0.12.5-node expose: - 6667 networks: diff --git a/tools/dependencies/known-oap-backend-dependencies.txt b/tools/dependencies/known-oap-backend-dependencies.txt index 0d2bdd5df9..4da9f858ba 100755 --- a/tools/dependencies/known-oap-backend-dependencies.txt +++ b/tools/dependencies/known-oap-backend-dependencies.txt @@ -66,8 +66,8 @@ httpclient-4.5.13.jar httpcore-4.4.13.jar httpcore-nio-4.4.13.jar influxdb-java-2.15.jar -iotdb-session-0.12.4.jar -iotdb-thrift-0.12.4.jar +iotdb-session-0.12.5.jar +iotdb-thrift-0.12.5.jar j2objc-annotations-1.3.jar jackson-annotations-2.12.2.jar jackson-core-2.12.2.jar @@ -140,7 +140,7 @@ protobuf-java-3.19.2.jar protobuf-java-util-3.19.2.jar reactive-streams-1.0.2.jar retrofit-2.5.0.jar -service-rpc-0.12.4.jar +service-rpc-0.12.5.jar simpleclient-0.6.0.jar simpleclient_common-0.6.0.jar simpleclient_hotspot-0.6.0.jar @@ -149,7 +149,7 @@ slf4j-api-1.7.30.jar snakeyaml-1.28.jar snappy-java-1.1.7.3.jar swagger-annotations-1.6.3.jar -tsfile-0.12.4.jar +tsfile-0.12.5.jar vavr-0.10.3.jar vavr-match-0.10.3.jar zipkin-2.9.1.jar