diff --git a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java index bc6d575242..c36d328dd3 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -18,6 +18,7 @@ package org.thingsboard.server.controller; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -422,6 +423,25 @@ public class RuleChainController extends BaseController { } private String msgToOutput(TbMsg msg) throws Exception { + JsonNode resultNode = convertMsgToOut(msg); + return objectMapper.writeValueAsString(resultNode); + } + + private String msgToOutput(List msgs) throws Exception { + JsonNode resultNode; + if (msgs.size() > 1) { + resultNode = objectMapper.createArrayNode(); + for (TbMsg msg : msgs) { + JsonNode convertedData = convertMsgToOut(msg); + ((ArrayNode) resultNode).add(convertedData); + } + } else { + resultNode = convertMsgToOut(msgs.get(0)); + } + return objectMapper.writeValueAsString(resultNode); + } + + private JsonNode convertMsgToOut(TbMsg msg) throws Exception{ ObjectNode msgData = objectMapper.createObjectNode(); if (!StringUtils.isEmpty(msg.getData())) { msgData.set("msg", objectMapper.readTree(msg.getData())); @@ -429,7 +449,8 @@ public class RuleChainController extends BaseController { Map metadata = msg.getMetaData().getData(); msgData.set("metadata", objectMapper.valueToTree(metadata)); msgData.put("msgType", msg.getType()); - return objectMapper.writeValueAsString(msgData); + return msgData; } + } diff --git a/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java b/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java index 35f38b110e..5dffc0217c 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java +++ b/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java @@ -108,13 +108,18 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S } @Override - public TbMsg executeUpdate(TbMsg msg) throws ScriptException { + public List executeUpdate(TbMsg msg) throws ScriptException { JsonNode result = executeScript(msg); - if (!result.isObject()) { + if (result.isObject()) { + return Collections.singletonList(unbindMsg(result, msg)); + } else if (result.isArray()){ + List res = new ArrayList<>(result.size()); + result.forEach(jsonObject -> res.add(unbindMsg(jsonObject, msg))); + return res; + } else { log.warn("Wrong result type: {}", result.getNodeType()); throw new ScriptException("Wrong result type: " + result.getNodeType()); } - return unbindMsg(result, msg); } @Override diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 27b0c039b4..6ac41979d6 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -110,7 +110,7 @@ security: # Enable/disable claiming devices, if false -> the device's [claimingAllowed] SERVER_SCOPE attribute must be set to [true] to allow claiming specific device allowClaimingByDefault: "${SECURITY_CLAIM_ALLOW_CLAIMING_BY_DEFAULT:true}" # Time allowed to claim the device in milliseconds - duration: "${SECURITY_CLAIM_DURATION:60000}" # 1 minute, note this value must equal claimDevices.timeToLiveInMinutes value + duration: "${SECURITY_CLAIM_DURATION:86400000}" # 1 minute, note this value must equal claimDevices.timeToLiveInMinutes value basic: enabled: "${SECURITY_BASIC_ENABLED:false}" oauth2: @@ -348,8 +348,8 @@ caffeine: timeToLiveInMinutes: 1440 maxSize: 0 claimDevices: - timeToLiveInMinutes: 1 - maxSize: 0 + timeToLiveInMinutes: 1440 + maxSize: 1000 securitySettings: timeToLiveInMinutes: 1440 maxSize: 0 diff --git a/pom.xml b/pom.xml index beaedb9447..d1edef19fb 100755 --- a/pom.xml +++ b/pom.xml @@ -54,7 +54,8 @@ 4.10.0 4.0.5 4.3.1.0 - 3.11.9 + 3.11.10 + 3.11.0 1.2.7 28.2-jre 2.6.1 @@ -1095,6 +1096,11 @@ java-driver-query-builder ${cassandra.version} + + com.datastax.cassandra + cassandra-driver-core + ${cassandra-driver-core.version} + io.dropwizard.metrics metrics-jmx diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java index ca5f148ae3..49420e8feb 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/ScriptEngine.java @@ -25,7 +25,7 @@ import java.util.Set; public interface ScriptEngine { - TbMsg executeUpdate(TbMsg msg) throws ScriptException; + List executeUpdate(TbMsg msg) throws ScriptException; ListenableFuture> executeUpdateAsync(TbMsg msg); diff --git a/tools/pom.xml b/tools/pom.xml index bb0435d434..4877914b62 100644 --- a/tools/pom.xml +++ b/tools/pom.xml @@ -56,8 +56,8 @@ cassandra-all - com.datastax.oss - java-driver-core + com.datastax.cassandra + cassandra-driver-core commons-io @@ -70,7 +70,7 @@ maven-assembly-plugin - + org.thingsboard.client.tools.migrator.MigratorTool diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java new file mode 100644 index 0000000000..53caeaeb0d --- /dev/null +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java @@ -0,0 +1,74 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.client.tools.migrator; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.LineIterator; +import org.apache.commons.lang3.StringUtils; + +import java.io.File; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +public class DictionaryParser { + private Map dictionaryParsed = new HashMap<>(); + + public DictionaryParser(File sourceFile) throws IOException { + parseDictionaryDump(FileUtils.lineIterator(sourceFile)); + } + + public String getKeyByKeyId(String keyId) { + return dictionaryParsed.get(keyId); + } + + private boolean isBlockFinished(String line) { + return StringUtils.isBlank(line) || line.equals("\\."); + } + + private boolean isBlockStarted(String line) { + return line.startsWith("COPY public.ts_kv_dictionary ("); + } + + private void parseDictionaryDump(LineIterator iterator) { + try { + String tempLine; + while (iterator.hasNext()) { + tempLine = iterator.nextLine(); + + if (isBlockStarted(tempLine)) { + processBlock(iterator); + } + } + } finally { + iterator.close(); + } + } + + private void processBlock(LineIterator lineIterator) { + String tempLine; + String[] lineSplited; + while(lineIterator.hasNext()) { + tempLine = lineIterator.nextLine(); + if(isBlockFinished(tempLine)) { + return; + } + + lineSplited = tempLine.split("\t"); + dictionaryParsed.put(lineSplited[1], lineSplited[0]); + } + } +} diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java index e3a98e974f..526d59f936 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java @@ -30,17 +30,26 @@ public class MigratorTool { public static void main(String[] args) { CommandLine cmd = parseArgs(args); - try { - File latestSource = new File(cmd.getOptionValue("latestTelemetryFrom")); - File latestSaveDir = new File(cmd.getOptionValue("latestTelemetryOut")); - File tsSource = new File(cmd.getOptionValue("telemetryFrom")); - File tsSaveDir = new File(cmd.getOptionValue("telemetryOut")); - File partitionsSaveDir = new File(cmd.getOptionValue("partitionsOut")); boolean castEnable = Boolean.parseBoolean(cmd.getOptionValue("castEnable")); + File allTelemetrySource = new File(cmd.getOptionValue("telemetryFrom")); + File tsSaveDir = null; + File partitionsSaveDir = null; + File latestSaveDir = null; + + RelatedEntitiesParser allEntityIdsAndTypes = + new RelatedEntitiesParser(new File(cmd.getOptionValue("relatedEntities"))); + DictionaryParser dictionaryParser = new DictionaryParser(allTelemetrySource); + + if(cmd.getOptionValue("latestTelemetryOut") != null) { + latestSaveDir = new File(cmd.getOptionValue("latestTelemetryOut")); + } + if(cmd.getOptionValue("telemetryOut") != null) { + tsSaveDir = new File(cmd.getOptionValue("telemetryOut")); + partitionsSaveDir = new File(cmd.getOptionValue("partitionsOut")); + } - PgCaLatestMigrator.migrateLatest(latestSource, latestSaveDir, castEnable); - PostgresToCassandraTelemetryMigrator.migrateTs(tsSource, tsSaveDir, partitionsSaveDir, castEnable); + new PgCaMigrator(allTelemetrySource, tsSaveDir, partitionsSaveDir, latestSaveDir, allEntityIdsAndTypes, dictionaryParser, castEnable).migrate(); } catch (Throwable th) { th.printStackTrace(); @@ -52,30 +61,30 @@ public class MigratorTool { private static CommandLine parseArgs(String[] args) { Options options = new Options(); - Option latestTsOpt = new Option("latestFrom", "latestTelemetryFrom", true, "latest telemetry source file path"); - latestTsOpt.setRequired(true); - options.addOption(latestTsOpt); + Option telemetryAllFrom = new Option("telemetryFrom", "telemetryFrom", true, "telemetry source file"); + telemetryAllFrom.setRequired(true); + options.addOption(telemetryAllFrom); Option latestTsOutOpt = new Option("latestOut", "latestTelemetryOut", true, "latest telemetry save dir"); - latestTsOutOpt.setRequired(true); + latestTsOutOpt.setRequired(false); options.addOption(latestTsOutOpt); - Option tsOpt = new Option("tsFrom", "telemetryFrom", true, "telemetry source file path"); - tsOpt.setRequired(true); - options.addOption(tsOpt); - Option tsOutOpt = new Option("tsOut", "telemetryOut", true, "sstable save dir"); - tsOutOpt.setRequired(true); + tsOutOpt.setRequired(false); options.addOption(tsOutOpt); Option partitionOutOpt = new Option("partitionsOut", "partitionsOut", true, "partitions save dir"); - partitionOutOpt.setRequired(true); + partitionOutOpt.setRequired(false); options.addOption(partitionOutOpt); Option castOpt = new Option("castEnable", "castEnable", true, "cast String to Double if possible"); castOpt.setRequired(true); options.addOption(castOpt); + Option relatedOpt = new Option("relatedEntities", "relatedEntities", true, "related entities source file path"); + relatedOpt.setRequired(true); + options.addOption(relatedOpt); + HelpFormatter formatter = new HelpFormatter(); CommandLineParser parser = new BasicParser(); diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaLatestMigrator.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaLatestMigrator.java deleted file mode 100644 index 5f0a84c7f4..0000000000 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaLatestMigrator.java +++ /dev/null @@ -1,179 +0,0 @@ -/** - * Copyright © 2016-2021 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.client.tools.migrator; - -import com.google.common.collect.Lists; -import org.apache.cassandra.io.sstable.CQLSSTableWriter; -import org.apache.commons.io.FileUtils; -import org.apache.commons.io.LineIterator; -import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.math.NumberUtils; - -import java.io.File; -import java.io.IOException; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Date; -import java.util.List; -import java.util.UUID; -import java.util.stream.Collectors; - -public class PgCaLatestMigrator { - - private static final long LOG_BATCH = 1000000; - private static final long rowPerFile = 1000000; - - - private static long linesProcessed = 0; - private static long linesMigrated = 0; - private static long castErrors = 0; - private static long castedOk = 0; - - private static long currentWriterCount = 1; - private static CQLSSTableWriter currentTsWriter = null; - - public static void migrateLatest(File sourceFile, File outDir, boolean castStringsIfPossible) throws IOException { - long startTs = System.currentTimeMillis(); - long stepLineTs = System.currentTimeMillis(); - long stepOkLineTs = System.currentTimeMillis(); - LineIterator iterator = FileUtils.lineIterator(sourceFile); - currentTsWriter = WriterBuilder.getTsWriter(outDir); - - boolean isBlockStarted = false; - boolean isBlockFinished = false; - - String line; - while (iterator.hasNext()) { - if (linesProcessed++ % LOG_BATCH == 0) { - System.out.println(new Date() + " linesProcessed = " + linesProcessed + " in " + (System.currentTimeMillis() - stepLineTs) + " castOk " + castedOk + " castErr " + castErrors); - stepLineTs = System.currentTimeMillis(); - } - - line = iterator.nextLine(); - - if (isBlockFinished) { - break; - } - - if (!isBlockStarted) { - if (isBlockStarted(line)) { - System.out.println(); - System.out.println(); - System.out.println(line); - System.out.println(); - System.out.println(); - isBlockStarted = true; - } - continue; - } - - if (isBlockFinished(line)) { - isBlockFinished = true; - } else { - try { - List raw = Arrays.stream(line.trim().split("\t")) - .map(String::trim) - .filter(StringUtils::isNotEmpty) - .collect(Collectors.toList()); - List values = toValues(raw); - - if (currentWriterCount == 0) { - System.out.println(new Date() + " close writer " + new Date()); - currentTsWriter.close(); - currentTsWriter = WriterBuilder.getLatestWriter(outDir); - } - - if (castStringsIfPossible) { - currentTsWriter.addRow(castToNumericIfPossible(values)); - } else { - currentTsWriter.addRow(values); - } - currentWriterCount++; - if (currentWriterCount >= rowPerFile) { - currentWriterCount = 0; - } - - if (linesMigrated++ % LOG_BATCH == 0) { - System.out.println(new Date() + " migrated = " + linesMigrated + " in " + (System.currentTimeMillis() - stepOkLineTs)); - stepOkLineTs = System.currentTimeMillis(); - } - } catch (Exception ex) { - System.out.println(ex.getMessage() + " -> " + line); - } - - } - } - - long endTs = System.currentTimeMillis(); - System.out.println(); - System.out.println(new Date() + " Migrated rows " + linesMigrated + " in " + (endTs - startTs)); - - currentTsWriter.close(); - System.out.println(); - System.out.println("Finished migrate Latest Telemetry"); - } - - - private static List castToNumericIfPossible(List values) { - try { - if (values.get(6) != null && NumberUtils.isNumber(values.get(6).toString())) { - Double casted = NumberUtils.createDouble(values.get(6).toString()); - List numeric = Lists.newArrayList(); - numeric.addAll(values); - numeric.set(6, null); - numeric.set(8, casted); - castedOk++; - return numeric; - } - } catch (Throwable th) { - castErrors++; - } - return values; - } - - private static List toValues(List raw) { - //expected Table structure: -// COPY public.ts_kv_latest (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) FROM stdin; - - - List result = new ArrayList<>(); - result.add(raw.get(0)); - result.add(fromString(raw.get(1))); - result.add(raw.get(2)); - - long ts = Long.parseLong(raw.get(3)); - result.add(ts); - - result.add(raw.get(4).equals("\\N") ? null : raw.get(4).equals("t") ? Boolean.TRUE : Boolean.FALSE); - result.add(raw.get(5).equals("\\N") ? null : raw.get(5)); - result.add(raw.get(6).equals("\\N") ? null : Long.parseLong(raw.get(6))); - result.add(raw.get(7).equals("\\N") ? null : Double.parseDouble(raw.get(7))); - return result; - } - - public static UUID fromString(String src) { - return UUID.fromString(src.substring(7, 15) + "-" + src.substring(3, 7) + "-1" - + src.substring(0, 3) + "-" + src.substring(15, 19) + "-" + src.substring(19)); - } - - private static boolean isBlockStarted(String line) { - return line.startsWith("COPY public.ts_kv_latest"); - } - - private static boolean isBlockFinished(String line) { - return StringUtils.isBlank(line) || line.equals("\\."); - } -} diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java new file mode 100644 index 0000000000..29a0a23ad6 --- /dev/null +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java @@ -0,0 +1,282 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.client.tools.migrator; + +import com.google.common.collect.Lists; +import org.apache.cassandra.io.sstable.CQLSSTableWriter; +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.LineIterator; +import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.math.NumberUtils; + +import java.io.File; +import java.io.IOException; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import java.util.function.Function; +import java.util.stream.Collectors; + +public class PgCaMigrator { + + private final long LOG_BATCH = 1000000; + private final long rowPerFile = 1000000; + + private long linesTsMigrated = 0; + private long linesLatestMigrated = 0; + private long castErrors = 0; + private long castedOk = 0; + + private long currentWriterCount = 1; + + private final File sourceFile; + private final boolean castStringIfPossible; + + private final RelatedEntitiesParser entityIdsAndTypes; + private final DictionaryParser keyParser; + private CQLSSTableWriter currentTsWriter; + private CQLSSTableWriter currentPartitionsWriter; + private CQLSSTableWriter currentTsLatestWriter; + private final Set partitions = new HashSet<>(); + + private File outTsDir; + private File outTsLatestDir; + + public PgCaMigrator(File sourceFile, + File ourTsDir, + File outTsPartitionDir, + File outTsLatestDir, + RelatedEntitiesParser allEntityIdsAndTypes, + DictionaryParser dictionaryParser, + boolean castStringsIfPossible) { + this.sourceFile = sourceFile; + this.entityIdsAndTypes = allEntityIdsAndTypes; + this.keyParser = dictionaryParser; + this.castStringIfPossible = castStringsIfPossible; + if(outTsLatestDir != null) { + this.currentTsLatestWriter = WriterBuilder.getLatestWriter(outTsLatestDir); + this.outTsLatestDir = outTsLatestDir; + } + if(ourTsDir != null) { + this.currentTsWriter = WriterBuilder.getTsWriter(ourTsDir); + this.currentPartitionsWriter = WriterBuilder.getPartitionWriter(outTsPartitionDir); + this.outTsDir = ourTsDir; + } + } + + public void migrate() throws IOException { + boolean isTsDone = false; + boolean isLatestDone = false; + String line; + LineIterator iterator = FileUtils.lineIterator(this.sourceFile); + + try { + while(iterator.hasNext()) { + line = iterator.nextLine(); + if(!isLatestDone && isBlockLatestStarted(line)) { + System.out.println("START TO MIGRATE LATEST"); + long start = System.currentTimeMillis(); + processBlock(iterator, currentTsLatestWriter, outTsLatestDir, this::toValuesLatest); + System.out.println("TOTAL LINES MIGRATED: " + linesLatestMigrated + ", FORMING OF SSL FOR LATEST TS FINISHED WITH TIME: " + (System.currentTimeMillis() - start) + " ms."); + isLatestDone = true; + } + + if(!isTsDone && isBlockTsStarted(line)) { + System.out.println("START TO MIGRATE TS"); + long start = System.currentTimeMillis(); + processBlock(iterator, currentTsWriter, outTsDir, this::toValuesTs); + System.out.println("TOTAL LINES MIGRATED: " + linesTsMigrated + ", FORMING OF SSL FOR TS FINISHED WITH TIME: " + (System.currentTimeMillis() - start) + " ms."); + isTsDone = true; + } + } + + System.out.println("Partitions collected " + partitions.size()); + long startTs = System.currentTimeMillis(); + for (String partition : partitions) { + String[] split = partition.split("\\|"); + List values = Lists.newArrayList(); + values.add(split[0]); + values.add(UUID.fromString(split[1])); + values.add(split[2]); + values.add(Long.parseLong(split[3])); + currentPartitionsWriter.addRow(values); + } + + System.out.println(new Date() + " Migrated partitions " + partitions.size() + " in " + (System.currentTimeMillis() - startTs)); + + System.out.println(); + System.out.println("Finished migrate Telemetry"); + + } finally { + iterator.close(); + currentTsLatestWriter.close(); + currentTsWriter.close(); + currentPartitionsWriter.close(); + } + } + + private void logLinesProcessed(long lines) { + if (lines % LOG_BATCH == 0) { + System.out.println(new Date() + " lines processed = " + lines + " in, castOk " + castedOk + " castErr " + castErrors); + } + } + + private void logLinesMigrated(long lines) { + if(lines % LOG_BATCH == 0) { + System.out.println(new Date() + " lines migrated = " + lines + " in, castOk " + castedOk + " castErr " + castErrors); + } + } + + private void addTypeIdKey(List result, List raw) { + result.add(entityIdsAndTypes.getEntityType(raw.get(0))); + result.add(UUID.fromString(raw.get(0))); + result.add(keyParser.getKeyByKeyId(raw.get(1))); + } + + private void addPartitions(List result, List raw) { + long ts = Long.parseLong(raw.get(2)); + long partition = toPartitionTs(ts); + result.add(partition); + result.add(ts); + } + + private void addTimeseries(List result, List raw) { + result.add(Long.parseLong(raw.get(2))); + } + + private void addValues(List result, List raw) { + result.add(raw.get(3).equals("\\N") ? null : raw.get(3).equals("t") ? Boolean.TRUE : Boolean.FALSE); + result.add(raw.get(4).equals("\\N") ? null : raw.get(4)); + result.add(raw.get(5).equals("\\N") ? null : Long.parseLong(raw.get(5))); + result.add(raw.get(6).equals("\\N") ? null : Double.parseDouble(raw.get(6))); + result.add(raw.get(7).equals("\\N") ? null : raw.get(7)); + } + + private List toValuesTs(List raw) { + + logLinesMigrated(linesTsMigrated++); + + List result = new ArrayList<>(); + + addTypeIdKey(result, raw); + addPartitions(result, raw); + addValues(result, raw); + + processPartitions(result); + + return result; + } + + private List toValuesLatest(List raw) { + logLinesMigrated(linesLatestMigrated++); + List result = new ArrayList<>(); + + addTypeIdKey(result, raw); + addTimeseries(result, raw); + addValues(result, raw); + + return result; + } + + private long toPartitionTs(long ts) { + LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC); + return time.truncatedTo(ChronoUnit.DAYS).withDayOfMonth(1).toInstant(ZoneOffset.UTC).toEpochMilli(); + } + + private void processPartitions(List values) { + String key = values.get(0) + "|" + values.get(1) + "|" + values.get(2) + "|" + values.get(3); + partitions.add(key); + } + + private void processBlock(LineIterator iterator, CQLSSTableWriter writer, File outDir, Function, List> function) { + String currentLine; + long linesProcessed = 0; + while(iterator.hasNext()) { + logLinesProcessed(linesProcessed++); + currentLine = iterator.nextLine(); + if(isBlockFinished(currentLine)) { + return; + } + + try { + List raw = Arrays.stream(currentLine.trim().split("\t")) + .map(String::trim) + .collect(Collectors.toList()); + List values = function.apply(raw); + + if (this.currentWriterCount == 0) { + System.out.println(new Date() + " close writer " + new Date()); + writer.close(); + writer = WriterBuilder.getLatestWriter(outDir); + } + + if (this.castStringIfPossible) { + writer.addRow(castToNumericIfPossible(values)); + } else { + writer.addRow(values); + } + + currentWriterCount++; + if (currentWriterCount >= rowPerFile) { + currentWriterCount = 0; + } + } catch (Exception ex) { + System.out.println(ex.getMessage() + " -> " + currentLine); + } + } + } + + private List castToNumericIfPossible(List values) { + try { + if (values.get(6) != null && NumberUtils.isNumber(values.get(6).toString())) { + Double casted = NumberUtils.createDouble(values.get(6).toString()); + List numeric = Lists.newArrayList(); + numeric.addAll(values); + numeric.set(6, null); + numeric.set(8, casted); + castedOk++; + return numeric; + } + } catch (Throwable th) { + castErrors++; + } + + processPartitions(values); + + return values; + } + + private boolean isBlockFinished(String line) { + return StringUtils.isBlank(line) || line.equals("\\."); + } + + private boolean isBlockTsStarted(String line) { + return line.startsWith("COPY public.ts_kv ("); + } + + private boolean isBlockLatestStarted(String line) { + return line.startsWith("COPY public.ts_kv_latest ("); + } + +} diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/PostgresToCassandraTelemetryMigrator.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/PostgresToCassandraTelemetryMigrator.java deleted file mode 100644 index 3ceab85ce3..0000000000 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PostgresToCassandraTelemetryMigrator.java +++ /dev/null @@ -1,222 +0,0 @@ -/** - * Copyright © 2016-2021 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.client.tools.migrator; - -import com.google.common.collect.Lists; -import org.apache.cassandra.io.sstable.CQLSSTableWriter; -import org.apache.commons.io.FileUtils; -import org.apache.commons.io.LineIterator; -import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.math.NumberUtils; - -import java.io.File; -import java.io.IOException; -import java.time.Instant; -import java.time.LocalDateTime; -import java.time.ZoneOffset; -import java.time.temporal.ChronoUnit; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Date; -import java.util.HashSet; -import java.util.List; -import java.util.Set; -import java.util.UUID; -import java.util.stream.Collectors; - -public class PostgresToCassandraTelemetryMigrator { - - private static final long LOG_BATCH = 1000000; - private static final long rowPerFile = 1000000; - - - private static long linesProcessed = 0; - private static long linesMigrated = 0; - private static long castErrors = 0; - private static long castedOk = 0; - - private static long currentWriterCount = 1; - private static CQLSSTableWriter currentTsWriter = null; - private static CQLSSTableWriter currentPartitionWriter = null; - - private static Set partitions = new HashSet<>(); - - - public static void migrateTs(File sourceFile, File outTsDir, File outPartitionDir, boolean castStringsIfPossible) throws IOException { - long startTs = System.currentTimeMillis(); - long stepLineTs = System.currentTimeMillis(); - long stepOkLineTs = System.currentTimeMillis(); - LineIterator iterator = FileUtils.lineIterator(sourceFile); - currentTsWriter = WriterBuilder.getTsWriter(outTsDir); - currentPartitionWriter = WriterBuilder.getPartitionWriter(outPartitionDir); - - boolean isBlockStarted = false; - boolean isBlockFinished = false; - - String line; - while (iterator.hasNext()) { - if (linesProcessed++ % LOG_BATCH == 0) { - System.out.println(new Date() + " linesProcessed = " + linesProcessed + " in " + (System.currentTimeMillis() - stepLineTs) + " castOk " + castedOk + " castErr " + castErrors); - stepLineTs = System.currentTimeMillis(); - } - - line = iterator.nextLine(); - - if (isBlockFinished) { - break; - } - - if (!isBlockStarted) { - if (isBlockStarted(line)) { - System.out.println(); - System.out.println(); - System.out.println(line); - System.out.println(); - System.out.println(); - isBlockStarted = true; - } - continue; - } - - if (isBlockFinished(line)) { - isBlockFinished = true; - } else { - try { - List raw = Arrays.stream(line.trim().split("\t")) - .map(String::trim) - .filter(StringUtils::isNotEmpty) - .collect(Collectors.toList()); - List values = toValues(raw); - - if (currentWriterCount == 0) { - System.out.println(new Date() + " close writer " + new Date()); - currentTsWriter.close(); - currentTsWriter = WriterBuilder.getTsWriter(outTsDir); - } - - if (castStringsIfPossible) { - currentTsWriter.addRow(castToNumericIfPossible(values)); - } else { - currentTsWriter.addRow(values); - } - processPartitions(values); - currentWriterCount++; - if (currentWriterCount >= rowPerFile) { - currentWriterCount = 0; - } - - if (linesMigrated++ % LOG_BATCH == 0) { - System.out.println(new Date() + " migrated = " + linesMigrated + " in " + (System.currentTimeMillis() - stepOkLineTs) + " partitions = " + partitions.size()); - stepOkLineTs = System.currentTimeMillis(); - } - } catch (Exception ex) { - System.out.println(ex.getMessage() + " -> " + line); - } - - } - } - - long endTs = System.currentTimeMillis(); - System.out.println(); - System.out.println(new Date() + " Migrated rows " + linesMigrated + " in " + (endTs - startTs)); - System.out.println("Partitions collected " + partitions.size()); - - startTs = System.currentTimeMillis(); - for (String partition : partitions) { - String[] split = partition.split("\\|"); - List values = Lists.newArrayList(); - values.add(split[0]); - values.add(UUID.fromString(split[1])); - values.add(split[2]); - values.add(Long.parseLong(split[3])); - currentPartitionWriter.addRow(values); - } - currentPartitionWriter.close(); - endTs = System.currentTimeMillis(); - System.out.println(); - System.out.println(); - System.out.println(new Date() + " Migrated partitions " + partitions.size() + " in " + (endTs - startTs)); - - - currentTsWriter.close(); - System.out.println(); - System.out.println("Finished migrate Telemetry"); - } - - private static List castToNumericIfPossible(List values) { - try { - if (values.get(6) != null && NumberUtils.isNumber(values.get(6).toString())) { - Double casted = NumberUtils.createDouble(values.get(6).toString()); - List numeric = Lists.newArrayList(); - numeric.addAll(values); - numeric.set(6, null); - numeric.set(8, casted); - castedOk++; - return numeric; - } - } catch (Throwable th) { - castErrors++; - } - return values; - } - - private static void processPartitions(List values) { - String key = values.get(0) + "|" + values.get(1) + "|" + values.get(2) + "|" + values.get(3); - partitions.add(key); - } - - private static List toValues(List raw) { - //expected Table structure: -// COPY public.ts_kv (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) FROM stdin; - - - List result = new ArrayList<>(); - result.add(raw.get(0)); - result.add(fromString(raw.get(1))); - result.add(raw.get(2)); - - long ts = Long.parseLong(raw.get(3)); - long partition = toPartitionTs(ts); - result.add(partition); - result.add(ts); - - result.add(raw.get(4).equals("\\N") ? null : raw.get(4).equals("t") ? Boolean.TRUE : Boolean.FALSE); - result.add(raw.get(5).equals("\\N") ? null : raw.get(5)); - result.add(raw.get(6).equals("\\N") ? null : Long.parseLong(raw.get(6))); - result.add(raw.get(7).equals("\\N") ? null : Double.parseDouble(raw.get(7))); - return result; - } - - public static UUID fromString(String src) { - return UUID.fromString(src.substring(7, 15) + "-" + src.substring(3, 7) + "-1" - + src.substring(0, 3) + "-" + src.substring(15, 19) + "-" + src.substring(19)); - } - - private static long toPartitionTs(long ts) { - LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC); - return time.truncatedTo(ChronoUnit.DAYS).withDayOfMonth(1).toInstant(ZoneOffset.UTC).toEpochMilli(); -// return TsPartitionDate.MONTHS.truncatedTo(time).toInstant(ZoneOffset.UTC).toEpochMilli(); - } - - private static boolean isBlockStarted(String line) { - return line.startsWith("COPY public.ts_kv"); - } - - private static boolean isBlockFinished(String line) { - return StringUtils.isBlank(line) || line.equals("\\."); - } - -} diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/README.md b/tools/src/main/java/org/thingsboard/client/tools/migrator/README.md index 70c5dafcaf..a468cbd3af 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/README.md +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/README.md @@ -21,37 +21,41 @@ It will generate single jar file with all required dependencies inside `target d #### Dump data from the source Postgres Database *Do not use compression if possible because Tool can only work with uncompressed file -1. Dump table `ts_kv` table: - - `pg_dump -h localhost -U postgres -d thingsboard -t ts_kv > ts_kv.dmp` - -2. Dump table `ts_kv_latest` table: - - `pg_dump -h localhost -U postgres -d thingsboard -t ts_kv_latest > ts_kv_latest.dmp` +1. Dump related tables that need to correct save telemetry + + `pg_dump -h localhost -U postgres -d thingsboard -T admin_settings -T attribute_kv -T audit_log -T component_discriptor -T device_credentials -T event -T oauth2_client_registration -T oauth2_client_registration_info -T oauth2_client_registration_template -T relation -T rule_node_state tb_schema_settings -T user_credentials > related_entities.dmp` + +2. Dump `ts_kv` and child: + + `pg_dump -h localhost -U postgres -d thingsboard --load-via-partition-root --data-only -t ts_kv* > ts_kv_all.dmp` -3. [Optional] move table dumps to the instance where cassandra will be hosted +3. [Optional] Move table dumps to the instance where cassandra will be hosted #### Prepare directory structure for SSTables Tool use 3 different directories for saving SSTables - `ts_kv_cf`, `ts_kv_latest_cf`, `ts_kv_partitions_cf` Create 3 empty directories. For example: - /home/ubunut/migration/ts - /home/ubunut/migration/ts_latest - /home/ubunut/migration/ts_partition + /home/user/migration/ts + /home/user/migration/ts_latest + /home/user/migration/ts_partition #### Run tool -*Note: if you run this tool on remote instance - don't forget to execute this command in `screen` to avoid unexpected termination + +**If you want to migrate just `ts_kv` without `ts_kv_latest` or vice versa don't use arguments (paths) for output files* + +**Note: if you run this tool on remote instance - don't forget to execute this command in `screen` to avoid unexpected termination* ``` -java -jar ./tools-2.4.1-SNAPSHOT-jar-with-dependencies.jar - -latestFrom ./source/ts_kv_latest.dmp - -latestOut /home/ubunut/migration/ts_latest - -tsFrom ./source/ts_kv.dmp - -tsOut /home/ubunut/migration/ts - -partitionsOut /home/ubunut/migration/ts_partition - -castEnable false +java -jar ./tools-3.2.2-SNAPSHOT-jar-with-dependencies.jar + -telemetryFrom /home/user/dump/ts_kv_all.dmp + -relatedEntities /home/user/dump/related_entities.dmp + -latestOut /home/user/migration/ts_latest + -tsOut /home/user/migration/ts + -partitionsOut /home/user/migration/ts_partition + -castEnable false ``` +*Use your paths for program arguments* Tool execution time depends on DB size, CPU resources and Disk throughput @@ -59,22 +63,23 @@ Tool execution time depends on DB size, CPU resources and Disk throughput * Note that this this part works only for single node Cassandra Cluster. If you have more nodes - it is better to use `sstableloader` tool. 1. [Optional] install Cassandra on the instance -2. [Optional] Using `cqlsh` create `thingsboard` keyspace and requred tables from this file `schema-ts.cql` +2. [Optional] Using `cqlsh` create `thingsboard` keyspace and requred tables from this files `schema-ts.cql` and `schema-ts-latest.cql` using `source` command 3. Stop Cassandra -4. Copy generated SSTable files into cassandra data dir: +4. Look at `/var/lib/cassandra/data/thingsboard` and check for names of data folders +5. Copy generated SSTable files into cassandra data dir using next command: ``` - sudo find /home/ubunut/migration/ts -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_cf-0e9aaf00ee5511e9a5fa7d6f489ffd13/ \; - sudo find /home/ubunut/migration/ts_latest -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_latest_cf-161449d0ee5511e9a5fa7d6f489ffd13/ \; - sudo find /home/ubunut/migration/ts_partition -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_partitions_cf-12e8fa80ee5511e9a5fa7d6f489ffd13/ \; + sudo find /home/user/migration/ts -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_cf-0e9aaf00ee5511e9a5fa7d6f489ffd13/ \; + sudo find /home/user/migration/ts_latest -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_latest_cf-161449d0ee5511e9a5fa7d6f489ffd13/ \; + sudo find /home/user/migration/ts_partition -name '*.*' -exec mv {} /var/lib/cassandra/data/thingsboard/ts_kv_partitions_cf-12e8fa80ee5511e9a5fa7d6f489ffd13/ \; ``` - -5. Start Cassandra service and trigger compaction + *Pay attention! Data folders have similar name `ts_kv_cf-0e9aaf00ee5511e9a5fa7d6f489ffd13`, but you have to use own* +6. Start Cassandra service and trigger compaction + + Trigger compactions: `nodetool compact thingsboard` + + Check compaction status: `nodetool compactionstats` -``` - trigger compactions: nodetool compact thingsboard - check compaction status: nodetool compactionstats -``` ## Switch Thignsboard into Hybrid Mode diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java new file mode 100644 index 0000000000..7c1b29f2af --- /dev/null +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java @@ -0,0 +1,87 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.client.tools.migrator; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.LineIterator; +import org.apache.commons.lang3.StringUtils; +import org.thingsboard.server.common.data.EntityType; + +import java.io.File; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +public class RelatedEntitiesParser { + private final Map allEntityIdsAndTypes = new HashMap<>(); + + private final Map tableNameAndEntityType = Map.ofEntries( + Map.entry("COPY public.alarm ", EntityType.ALARM), + Map.entry("COPY public.asset ", EntityType.ASSET), + Map.entry("COPY public.customer ", EntityType.CUSTOMER), + Map.entry("COPY public.dashboard ", EntityType.DASHBOARD), + Map.entry("COPY public.device ", EntityType.DEVICE), + Map.entry("COPY public.rule_chain ", EntityType.RULE_CHAIN), + Map.entry("COPY public.rule_node ", EntityType.RULE_NODE), + Map.entry("COPY public.tenant ", EntityType.TENANT), + Map.entry("COPY public.tb_user ", EntityType.USER), + Map.entry("COPY public.entity_view ", EntityType.ENTITY_VIEW), + Map.entry("COPY public.widgets_bundle ", EntityType.WIDGETS_BUNDLE), + Map.entry("COPY public.widget_type ", EntityType.WIDGET_TYPE), + Map.entry("COPY public.tenant_profile ", EntityType.TENANT_PROFILE), + Map.entry("COPY public.device_profile ", EntityType.DEVICE_PROFILE), + Map.entry("COPY public.api_usage_state ", EntityType.API_USAGE_STATE) + ); + + public RelatedEntitiesParser(File source) throws IOException { + processAllTables(FileUtils.lineIterator(source)); + } + + public String getEntityType(String uuid) { + return this.allEntityIdsAndTypes.get(uuid); + } + + private boolean isBlockFinished(String line) { + return StringUtils.isBlank(line) || line.equals("\\."); + } + + private void processAllTables(LineIterator lineIterator) { + String currentLine; + try { + while (lineIterator.hasNext()) { + currentLine = lineIterator.nextLine(); + for(Map.Entry entry : tableNameAndEntityType.entrySet()) { + if(currentLine.startsWith(entry.getKey())) { + processBlock(lineIterator, entry.getValue()); + } + } + } + } finally { + lineIterator.close(); + } + } + + private void processBlock(LineIterator lineIterator, EntityType entityType) { + String currentLine; + while(lineIterator.hasNext()) { + currentLine = lineIterator.nextLine(); + if(isBlockFinished(currentLine)) { + return; + } + allEntityIdsAndTypes.put(currentLine.split("\t")[0], entityType.name()); + } + } +} diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java index cfe36aaa14..5228da98a3 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java @@ -31,6 +31,7 @@ public class WriterBuilder { " str_v text,\n" + " long_v bigint,\n" + " dbl_v double,\n" + + " json_v text,\n" + " PRIMARY KEY (( entity_type, entity_id, key, partition ), ts)\n" + ");"; @@ -43,6 +44,7 @@ public class WriterBuilder { " str_v text,\n" + " long_v bigint,\n" + " dbl_v double,\n" + + " json_v text,\n" + " PRIMARY KEY (( entity_type, entity_id ), key)\n" + ") WITH compaction = { 'class' : 'LeveledCompactionStrategy' };"; @@ -59,8 +61,8 @@ public class WriterBuilder { return CQLSSTableWriter.builder() .inDirectory(dir) .forTable(tsSchema) - .using("INSERT INTO thingsboard.ts_kv_cf (entity_type, entity_id, key, partition, ts, bool_v, str_v, long_v, dbl_v) " + - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)") + .using("INSERT INTO thingsboard.ts_kv_cf (entity_type, entity_id, key, partition, ts, bool_v, str_v, long_v, dbl_v, json_v) " + + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") .build(); } @@ -68,8 +70,8 @@ public class WriterBuilder { return CQLSSTableWriter.builder() .inDirectory(dir) .forTable(latestSchema) - .using("INSERT INTO thingsboard.ts_kv_latest_cf (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " + - "VALUES (?, ?, ?, ?, ?, ?, ?, ?)") + .using("INSERT INTO thingsboard.ts_kv_latest_cf (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v, json_v) " + + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)") .build(); }