From 9746df84ea7f3366c62b2e58649af32d7c53ea56 Mon Sep 17 00:00:00 2001 From: zbeacon Date: Wed, 17 Mar 2021 11:37:59 +0200 Subject: [PATCH 1/9] Fix for default values for claiming queue and duration --- application/src/main/resources/thingsboard.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 From 52d1b756fac543cb6136bdcac9c5d7b4be824192 Mon Sep 17 00:00:00 2001 From: zbeacon Date: Tue, 16 Mar 2021 14:43:22 +0200 Subject: [PATCH 2/9] Fix for test script node function --- .../controller/RuleChainController.java | 19 +++++++++++++++++++ .../script/RuleNodeJsScriptEngine.java | 11 ++++++++--- .../rule/engine/api/ScriptEngine.java | 2 +- 3 files changed, 28 insertions(+), 4 deletions(-) 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..030fa67995 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; @@ -432,4 +433,22 @@ public class RuleChainController extends BaseController { return objectMapper.writeValueAsString(msgData); } + private String msgToOutput(List msgs) throws Exception { + ArrayNode resultNode = objectMapper.createArrayNode(); + for (TbMsg msg:msgs) { + ObjectNode msgData = objectMapper.createObjectNode(); + if (!StringUtils.isEmpty(msg.getData())) { + msgData.set("msg", objectMapper.readTree(msg.getData())); + } + Map metadata = msg.getMetaData().getData(); + msgData.set("metadata", objectMapper.valueToTree(metadata)); + msgData.put("msgType", msg.getType()); + resultNode.add(msgData); + } + if (resultNode.size() == 1) { + return objectMapper.writeValueAsString(resultNode.get(0)); + } + return objectMapper.writeValueAsString(resultNode); + } + } 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/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); From bd42cfc81f4c8641ad7b41f6e24e95f6d2da80f0 Mon Sep 17 00:00:00 2001 From: zbeacon Date: Tue, 16 Mar 2021 15:23:13 +0200 Subject: [PATCH 3/9] Changes according to comment --- .../controller/RuleChainController.java | 38 ++++++++++--------- 1 file changed, 20 insertions(+), 18 deletions(-) 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 030fa67995..c36d328dd3 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -423,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())); @@ -430,25 +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; } - private String msgToOutput(List msgs) throws Exception { - ArrayNode resultNode = objectMapper.createArrayNode(); - for (TbMsg msg:msgs) { - ObjectNode msgData = objectMapper.createObjectNode(); - if (!StringUtils.isEmpty(msg.getData())) { - msgData.set("msg", objectMapper.readTree(msg.getData())); - } - Map metadata = msg.getMetaData().getData(); - msgData.set("metadata", objectMapper.valueToTree(metadata)); - msgData.put("msgType", msg.getType()); - resultNode.add(msgData); - } - if (resultNode.size() == 1) { - return objectMapper.writeValueAsString(resultNode.get(0)); - } - return objectMapper.writeValueAsString(resultNode); - } } From 92d360e14b61115f5a62ef7a43de2502fd87640c Mon Sep 17 00:00:00 2001 From: Andrew Volostnykh Date: Wed, 10 Mar 2021 19:31:01 +0200 Subject: [PATCH 4/9] Refactoring of migration tool for new Thingsboard DB structure --- tools/pom.xml | 5 +- .../tools/migrator/DictionaryParser.java | 55 ++++++++++++++ .../client/tools/migrator/MigratorTool.java | 46 ++++++----- .../tools/migrator/PgCaLatestMigrator.java | 50 ++++++------ .../PostgresToCassandraTelemetryMigrator.java | 40 +++++----- .../client/tools/migrator/README.md | 19 ++--- .../tools/migrator/RelatedEntitiesParser.java | 76 +++++++++++++++++++ .../client/tools/migrator/WriterBuilder.java | 10 ++- 8 files changed, 225 insertions(+), 76 deletions(-) create mode 100644 tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java create mode 100644 tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java diff --git a/tools/pom.xml b/tools/pom.xml index bb0435d434..3b163b44b3 100644 --- a/tools/pom.xml +++ b/tools/pom.xml @@ -56,8 +56,9 @@ cassandra-all - com.datastax.oss - java-driver-core + com.datastax.cassandra + cassandra-driver-core + 3.10.1 commons-io 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..d88b4361da --- /dev/null +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java @@ -0,0 +1,55 @@ +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) { + String tempLine; + while(iterator.hasNext()) { + tempLine = iterator.nextLine(); + + if(isBlockStarted(tempLine)) { + processBlock(iterator); + } + } + } + + 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..83ca54624a 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,25 @@ 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")); - - PgCaLatestMigrator.migrateLatest(latestSource, latestSaveDir, castEnable); - PostgresToCassandraTelemetryMigrator.migrateTs(tsSource, tsSaveDir, partitionsSaveDir, castEnable); + File allTelemetrySource = new File(cmd.getOptionValue("telemetryFrom")); + + RelatedEntitiesParser allEntityIdsAndTypes = + new RelatedEntitiesParser(new File(cmd.getOptionValue("relatedEntities"))); + DictionaryParser dictionaryParser = new DictionaryParser(allTelemetrySource); + + if(cmd.getOptionValue("latestTelemetryOut") != null) { + File latestSaveDir = new File(cmd.getOptionValue("latestTelemetryOut")); + PgCaLatestMigrator.migrateLatest(allTelemetrySource, latestSaveDir, allEntityIdsAndTypes, dictionaryParser, castEnable); + } + if(cmd.getOptionValue("telemetryOut") != null) { + File tsSaveDir = new File(cmd.getOptionValue("telemetryOut")); + File partitionsSaveDir = new File(cmd.getOptionValue("partitionsOut")); + PostgresToCassandraTelemetryMigrator.migrateTs( + allTelemetrySource, tsSaveDir, partitionsSaveDir, allEntityIdsAndTypes, dictionaryParser, castEnable + ); + } } catch (Throwable th) { th.printStackTrace(); @@ -52,30 +60,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 index 5f0a84c7f4..5d4a2a6e81 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaLatestMigrator.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaLatestMigrator.java @@ -43,14 +43,21 @@ public class PgCaLatestMigrator { 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 { + private static RelatedEntitiesParser allIdsAndTypes; + private static DictionaryParser keyPairs; + + public static void migrateLatest(File sourceFile, + File outDir, + RelatedEntitiesParser allEntityIdsAndTypes, + DictionaryParser dictionaryParser, + 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); + CQLSSTableWriter currentTsWriter = WriterBuilder.getLatestWriter(outDir); + allIdsAndTypes = allEntityIdsAndTypes; + keyPairs = dictionaryParser; boolean isBlockStarted = false; boolean isBlockFinished = false; @@ -107,7 +114,7 @@ public class PgCaLatestMigrator { } if (linesMigrated++ % LOG_BATCH == 0) { - System.out.println(new Date() + " migrated = " + linesMigrated + " in " + (System.currentTimeMillis() - stepOkLineTs)); + System.out.println(new Date() + " migrated = " + linesMigrated + " in " + (System.currentTimeMillis() - stepOkLineTs) + " ms."); stepOkLineTs = System.currentTimeMillis(); } } catch (Exception ex) { @@ -119,7 +126,7 @@ public class PgCaLatestMigrator { long endTs = System.currentTimeMillis(); System.out.println(); - System.out.println(new Date() + " Migrated rows " + linesMigrated + " in " + (endTs - startTs)); + System.out.println(new Date() + " Migrated rows " + linesMigrated + " in " + (endTs - startTs) + " ts"); currentTsWriter.close(); System.out.println(); @@ -146,34 +153,31 @@ public class PgCaLatestMigrator { 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; - + //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)); + result.add(allIdsAndTypes.getEntityType(raw.get(0))); + result.add(UUID.fromString(raw.get(0))); + result.add(keyPairs.getKeyByKeyId(raw.get(1))); - long ts = Long.parseLong(raw.get(3)); - result.add(ts); + long ts = Long.parseLong(raw.get(2)); + result.add(3, 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; - } + 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)); - 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)); + return result; } private static boolean isBlockStarted(String line) { - return line.startsWith("COPY public.ts_kv_latest"); + 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/PostgresToCassandraTelemetryMigrator.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/PostgresToCassandraTelemetryMigrator.java index 3ceab85ce3..80e617f505 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PostgresToCassandraTelemetryMigrator.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/PostgresToCassandraTelemetryMigrator.java @@ -42,7 +42,6 @@ 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; @@ -53,15 +52,23 @@ public class PostgresToCassandraTelemetryMigrator { 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 { + private static RelatedEntitiesParser entityIdsAndTypes; + private static DictionaryParser keyParser; + + public static void migrateTs(File sourceFile, + File outTsDir, + File outPartitionDir, + RelatedEntitiesParser allEntityIdsAndTypes, + DictionaryParser dictionaryParser, + 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); + entityIdsAndTypes = allEntityIdsAndTypes; + keyParser = dictionaryParser; boolean isBlockStarted = false; boolean isBlockFinished = false; @@ -182,29 +189,24 @@ public class PostgresToCassandraTelemetryMigrator { //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)); + result.add(entityIdsAndTypes.getEntityType(raw.get(0))); + result.add(UUID.fromString(raw.get(0))); + result.add(keyParser.getKeyByKeyId(raw.get(1))); - long ts = Long.parseLong(raw.get(3)); + long ts = Long.parseLong(raw.get(2)); 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))); + 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)); 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(); @@ -212,7 +214,7 @@ public class PostgresToCassandraTelemetryMigrator { } private static boolean isBlockStarted(String line) { - return line.startsWith("COPY public.ts_kv"); + return line.startsWith("COPY public.ts_kv ("); } private static boolean isBlockFinished(String line) { 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..1c5a6d54b7 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,15 +21,17 @@ 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: +*If you want to migrate just `ts_kv` without `ts_kv_latest` just don't dump an unnecessary table and when starting the tool don't use arguments (paths) for input dump and output files* - `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 tenant -t customer -t user -t dashboard -t asset -t device -t alarm -t rule_chain -t rule_node -t entity_view -t widgets_bundle -t widget_type -t tenant_profile -t device_profile -t api_usage_state -t tb_user > 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` @@ -45,9 +47,8 @@ Create 3 empty directories. For example: ``` java -jar ./tools-2.4.1-SNAPSHOT-jar-with-dependencies.jar - -latestFrom ./source/ts_kv_latest.dmp + -telemetryFrom ./source/ts_kv_all.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 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..655e623490 --- /dev/null +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java @@ -0,0 +1,76 @@ +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<>(); + + 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; + while(lineIterator.hasNext()) { + currentLine = lineIterator.nextLine(); + if(currentLine.startsWith("COPY public.alarm")) { + processBlock(lineIterator, EntityType.ALARM); + } else if (currentLine.startsWith("COPY public.asset")) { + processBlock(lineIterator, EntityType.ASSET); + } else if (currentLine.startsWith("COPY public.customer")) { + processBlock(lineIterator, EntityType.CUSTOMER); + } else if (currentLine.startsWith("COPY public.dashboard")) { + processBlock(lineIterator, EntityType.DASHBOARD); + } else if (currentLine.startsWith("COPY public.device")) { + processBlock(lineIterator, EntityType.DEVICE); + } else if (currentLine.startsWith("COPY public.rule_chain")) { + processBlock(lineIterator, EntityType.RULE_CHAIN); + } else if (currentLine.startsWith("COPY public.rule_node")) { + processBlock(lineIterator, EntityType.RULE_NODE); + } else if (currentLine.startsWith("COPY public.tenant")) { + processBlock(lineIterator, EntityType.TENANT); + } else if (currentLine.startsWith("COPY public.tb_user")) { + processBlock(lineIterator, EntityType.USER); + } else if (currentLine.startsWith("COPY public.entity_view")) { + processBlock(lineIterator, EntityType.ENTITY_VIEW); + } else if (currentLine.startsWith("COPY public.widgets_bundle")) { + processBlock(lineIterator, EntityType.WIDGETS_BUNDLE); + } else if (currentLine.startsWith("COPY public.widget_type")) { + processBlock(lineIterator, EntityType.WIDGET_TYPE); + } else if (currentLine.startsWith("COPY public.tenant_profile")) { + processBlock(lineIterator, EntityType.TENANT_PROFILE); + } else if (currentLine.startsWith("COPY public.device_profile")) { + processBlock(lineIterator, EntityType.DEVICE_PROFILE); + } else if (currentLine.startsWith("COPY public.api_usage_state")) { + processBlock(lineIterator, EntityType.API_USAGE_STATE); + } + } + } + + 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(); } From 67ee892eb84211c65cedcab5617772df11f926a3 Mon Sep 17 00:00:00 2001 From: Andrew Volostnykh Date: Thu, 11 Mar 2021 13:54:30 +0200 Subject: [PATCH 5/9] License headers added --- .../client/tools/migrator/DictionaryParser.java | 15 +++++++++++++++ .../tools/migrator/RelatedEntitiesParser.java | 15 +++++++++++++++ 2 files changed, 30 insertions(+) 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 index d88b4361da..276daf003e 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java @@ -1,3 +1,18 @@ +/** + * 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; 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 index 655e623490..ecd5164c70 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java @@ -1,3 +1,18 @@ +/** + * 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; From 1240099c815b7a673f782904b8ad64c36c253018 Mon Sep 17 00:00:00 2001 From: Andrew Volostnykh Date: Fri, 12 Mar 2021 18:39:17 +0200 Subject: [PATCH 6/9] Full refactoring and code cleaning --- tools/pom.xml | 2 + .../tools/migrator/DictionaryParser.java | 14 +- .../client/tools/migrator/MigratorTool.java | 15 +- .../tools/migrator/PgCaLatestMigrator.java | 183 ------------- .../client/tools/migrator/PgCaMigrator.java | 254 ++++++++++++++++++ .../PostgresToCassandraTelemetryMigrator.java | 224 --------------- .../tools/migrator/RelatedEntitiesParser.java | 68 ++--- 7 files changed, 309 insertions(+), 451 deletions(-) delete mode 100644 tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaLatestMigrator.java create mode 100644 tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java delete mode 100644 tools/src/main/java/org/thingsboard/client/tools/migrator/PostgresToCassandraTelemetryMigrator.java diff --git a/tools/pom.xml b/tools/pom.xml index 3b163b44b3..e3df0e08d0 100644 --- a/tools/pom.xml +++ b/tools/pom.xml @@ -54,6 +54,7 @@ org.apache.cassandra cassandra-all + 3.11.10 com.datastax.cassandra @@ -63,6 +64,7 @@ commons-io commons-io + 2.5 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 index 276daf003e..53caeaeb0d 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/DictionaryParser.java @@ -44,13 +44,17 @@ public class DictionaryParser { } private void parseDictionaryDump(LineIterator iterator) { - String tempLine; - while(iterator.hasNext()) { - tempLine = iterator.nextLine(); + try { + String tempLine; + while (iterator.hasNext()) { + tempLine = iterator.nextLine(); - if(isBlockStarted(tempLine)) { - processBlock(iterator); + if (isBlockStarted(tempLine)) { + processBlock(iterator); + } } + } finally { + iterator.close(); } } 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 83ca54624a..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 @@ -33,23 +33,24 @@ public class MigratorTool { try { 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) { - File latestSaveDir = new File(cmd.getOptionValue("latestTelemetryOut")); - PgCaLatestMigrator.migrateLatest(allTelemetrySource, latestSaveDir, allEntityIdsAndTypes, dictionaryParser, castEnable); + latestSaveDir = new File(cmd.getOptionValue("latestTelemetryOut")); } if(cmd.getOptionValue("telemetryOut") != null) { - File tsSaveDir = new File(cmd.getOptionValue("telemetryOut")); - File partitionsSaveDir = new File(cmd.getOptionValue("partitionsOut")); - PostgresToCassandraTelemetryMigrator.migrateTs( - allTelemetrySource, tsSaveDir, partitionsSaveDir, allEntityIdsAndTypes, dictionaryParser, castEnable - ); + tsSaveDir = new File(cmd.getOptionValue("telemetryOut")); + partitionsSaveDir = new File(cmd.getOptionValue("partitionsOut")); } + new PgCaMigrator(allTelemetrySource, tsSaveDir, partitionsSaveDir, latestSaveDir, allEntityIdsAndTypes, dictionaryParser, castEnable).migrate(); + } catch (Throwable th) { th.printStackTrace(); throw new IllegalStateException("failed", th); 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 5d4a2a6e81..0000000000 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaLatestMigrator.java +++ /dev/null @@ -1,183 +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 RelatedEntitiesParser allIdsAndTypes; - private static DictionaryParser keyPairs; - - public static void migrateLatest(File sourceFile, - File outDir, - RelatedEntitiesParser allEntityIdsAndTypes, - DictionaryParser dictionaryParser, - boolean castStringsIfPossible) throws IOException { - long startTs = System.currentTimeMillis(); - long stepLineTs = System.currentTimeMillis(); - long stepOkLineTs = System.currentTimeMillis(); - LineIterator iterator = FileUtils.lineIterator(sourceFile); - CQLSSTableWriter currentTsWriter = WriterBuilder.getLatestWriter(outDir); - allIdsAndTypes = allEntityIdsAndTypes; - keyPairs = dictionaryParser; - - 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) + " ms."); - 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) + " ts"); - - 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(allIdsAndTypes.getEntityType(raw.get(0))); - result.add(UUID.fromString(raw.get(0))); - result.add(keyPairs.getKeyByKeyId(raw.get(1))); - - long ts = Long.parseLong(raw.get(2)); - result.add(3, ts); - - 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)); - - return result; - } - - 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..afa341f880 --- /dev/null +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java @@ -0,0 +1,254 @@ +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 linesProcessed = 0; + 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("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("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() { + if (linesProcessed++ % LOG_BATCH == 0) { + System.out.println(new Date() + " linesProcessed = " + linesProcessed + " in, castOk " + castedOk + " castErr " + castErrors); + } + } + + private List toValuesTs(List raw) { + linesTsMigrated++; + List result = new ArrayList<>(); + result.add(entityIdsAndTypes.getEntityType(raw.get(0))); + result.add(UUID.fromString(raw.get(0))); + result.add(keyParser.getKeyByKeyId(raw.get(1))); + + long ts = Long.parseLong(raw.get(2)); + long partition = toPartitionTs(ts); + result.add(partition); + result.add(ts); + + 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)); + + processPartitions(result); + + return result; + } + + private List toValuesLatest(List raw) { + linesLatestMigrated++; + List result = new ArrayList<>(); + result.add(this.entityIdsAndTypes.getEntityType(raw.get(0))); + result.add(UUID.fromString(raw.get(0))); + result.add(this.keyParser.getKeyByKeyId(raw.get(1))); + + long ts = Long.parseLong(raw.get(2)); + result.add(3, ts); + + 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)); + + 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; + linesProcessed = 0; + while(iterator.hasNext()) { + logLinesProcessed(); + currentLine = iterator.nextLine(); + if(isBlockFinished(currentLine)) { + return; + } + + try { + List raw = Arrays.stream(currentLine.trim().split("\t")) + .map(String::trim) + .filter(StringUtils::isNotEmpty) + .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 80e617f505..0000000000 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PostgresToCassandraTelemetryMigrator.java +++ /dev/null @@ -1,224 +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<>(); - private static RelatedEntitiesParser entityIdsAndTypes; - private static DictionaryParser keyParser; - - public static void migrateTs(File sourceFile, - File outTsDir, - File outPartitionDir, - RelatedEntitiesParser allEntityIdsAndTypes, - DictionaryParser dictionaryParser, - 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); - entityIdsAndTypes = allEntityIdsAndTypes; - keyParser = dictionaryParser; - - 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(entityIdsAndTypes.getEntityType(raw.get(0))); - result.add(UUID.fromString(raw.get(0))); - result.add(keyParser.getKeyByKeyId(raw.get(1))); - - long ts = Long.parseLong(raw.get(2)); - long partition = toPartitionTs(ts); - result.add(partition); - result.add(ts); - - 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)); - return result; - } - - 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/RelatedEntitiesParser.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java index ecd5164c70..fac2609797 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java @@ -42,39 +42,43 @@ public class RelatedEntitiesParser { private void processAllTables(LineIterator lineIterator) { String currentLine; - while(lineIterator.hasNext()) { - currentLine = lineIterator.nextLine(); - if(currentLine.startsWith("COPY public.alarm")) { - processBlock(lineIterator, EntityType.ALARM); - } else if (currentLine.startsWith("COPY public.asset")) { - processBlock(lineIterator, EntityType.ASSET); - } else if (currentLine.startsWith("COPY public.customer")) { - processBlock(lineIterator, EntityType.CUSTOMER); - } else if (currentLine.startsWith("COPY public.dashboard")) { - processBlock(lineIterator, EntityType.DASHBOARD); - } else if (currentLine.startsWith("COPY public.device")) { - processBlock(lineIterator, EntityType.DEVICE); - } else if (currentLine.startsWith("COPY public.rule_chain")) { - processBlock(lineIterator, EntityType.RULE_CHAIN); - } else if (currentLine.startsWith("COPY public.rule_node")) { - processBlock(lineIterator, EntityType.RULE_NODE); - } else if (currentLine.startsWith("COPY public.tenant")) { - processBlock(lineIterator, EntityType.TENANT); - } else if (currentLine.startsWith("COPY public.tb_user")) { - processBlock(lineIterator, EntityType.USER); - } else if (currentLine.startsWith("COPY public.entity_view")) { - processBlock(lineIterator, EntityType.ENTITY_VIEW); - } else if (currentLine.startsWith("COPY public.widgets_bundle")) { - processBlock(lineIterator, EntityType.WIDGETS_BUNDLE); - } else if (currentLine.startsWith("COPY public.widget_type")) { - processBlock(lineIterator, EntityType.WIDGET_TYPE); - } else if (currentLine.startsWith("COPY public.tenant_profile")) { - processBlock(lineIterator, EntityType.TENANT_PROFILE); - } else if (currentLine.startsWith("COPY public.device_profile")) { - processBlock(lineIterator, EntityType.DEVICE_PROFILE); - } else if (currentLine.startsWith("COPY public.api_usage_state")) { - processBlock(lineIterator, EntityType.API_USAGE_STATE); + try { + while (lineIterator.hasNext()) { + currentLine = lineIterator.nextLine(); + if (currentLine.startsWith("COPY public.alarm")) { + processBlock(lineIterator, EntityType.ALARM); + } else if (currentLine.startsWith("COPY public.asset")) { + processBlock(lineIterator, EntityType.ASSET); + } else if (currentLine.startsWith("COPY public.customer")) { + processBlock(lineIterator, EntityType.CUSTOMER); + } else if (currentLine.startsWith("COPY public.dashboard")) { + processBlock(lineIterator, EntityType.DASHBOARD); + } else if (currentLine.startsWith("COPY public.device")) { + processBlock(lineIterator, EntityType.DEVICE); + } else if (currentLine.startsWith("COPY public.rule_chain")) { + processBlock(lineIterator, EntityType.RULE_CHAIN); + } else if (currentLine.startsWith("COPY public.rule_node")) { + processBlock(lineIterator, EntityType.RULE_NODE); + } else if (currentLine.startsWith("COPY public.tenant")) { + processBlock(lineIterator, EntityType.TENANT); + } else if (currentLine.startsWith("COPY public.tb_user")) { + processBlock(lineIterator, EntityType.USER); + } else if (currentLine.startsWith("COPY public.entity_view")) { + processBlock(lineIterator, EntityType.ENTITY_VIEW); + } else if (currentLine.startsWith("COPY public.widgets_bundle")) { + processBlock(lineIterator, EntityType.WIDGETS_BUNDLE); + } else if (currentLine.startsWith("COPY public.widget_type")) { + processBlock(lineIterator, EntityType.WIDGET_TYPE); + } else if (currentLine.startsWith("COPY public.tenant_profile")) { + processBlock(lineIterator, EntityType.TENANT_PROFILE); + } else if (currentLine.startsWith("COPY public.device_profile")) { + processBlock(lineIterator, EntityType.DEVICE_PROFILE); + } else if (currentLine.startsWith("COPY public.api_usage_state")) { + processBlock(lineIterator, EntityType.API_USAGE_STATE); + } } + } finally { + lineIterator.close(); } } From d5658d5b32bbfeebad90fc0b7519865d258ad280 Mon Sep 17 00:00:00 2001 From: AndrewVolosytnykhThingsboard Date: Tue, 16 Mar 2021 00:05:54 +0200 Subject: [PATCH 7/9] Code cleaning, improvement of README.md --- pom.xml | 8 ++- tools/pom.xml | 3 - .../client/tools/migrator/PgCaMigrator.java | 68 +++++++++++++------ .../client/tools/migrator/README.md | 44 ++++++------ 4 files changed, 80 insertions(+), 43 deletions(-) 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/tools/pom.xml b/tools/pom.xml index e3df0e08d0..99f966b283 100644 --- a/tools/pom.xml +++ b/tools/pom.xml @@ -54,17 +54,14 @@ org.apache.cassandra cassandra-all - 3.11.10 com.datastax.cassandra cassandra-driver-core - 3.10.1 commons-io commons-io - 2.5 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 index afa341f880..7aec7ba5f3 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java @@ -1,3 +1,18 @@ +/** + * 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; @@ -122,29 +137,52 @@ public class PgCaMigrator { } } - private void logLinesProcessed() { - if (linesProcessed++ % LOG_BATCH == 0) { - System.out.println(new Date() + " linesProcessed = " + linesProcessed + " in, castOk " + castedOk + " castErr " + castErrors); + private void logLinesProcessed(long lines) { + if (lines % LOG_BATCH == 0) { + System.out.println(new Date() + " lines processed = " + lines + " in, castOk " + castedOk + " castErr " + castErrors); } } - private List toValuesTs(List raw) { - linesTsMigrated++; - List result = new ArrayList<>(); + 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); @@ -152,20 +190,12 @@ public class PgCaMigrator { } private List toValuesLatest(List raw) { - linesLatestMigrated++; + logLinesMigrated(linesLatestMigrated++); List result = new ArrayList<>(); - result.add(this.entityIdsAndTypes.getEntityType(raw.get(0))); - result.add(UUID.fromString(raw.get(0))); - result.add(this.keyParser.getKeyByKeyId(raw.get(1))); - - long ts = Long.parseLong(raw.get(2)); - result.add(3, ts); - 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)); + addTypeIdKey(result, raw); + addTimeseries(result, raw); + addValues(result, raw); return result; } @@ -184,7 +214,7 @@ public class PgCaMigrator { String currentLine; linesProcessed = 0; while(iterator.hasNext()) { - logLinesProcessed(); + logLinesProcessed(linesProcessed++); currentLine = iterator.nextLine(); if(isBlockFinished(currentLine)) { return; 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 1c5a6d54b7..cfe1675eb2 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,8 +21,6 @@ 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 -*If you want to migrate just `ts_kv` without `ts_kv_latest` just don't dump an unnecessary table and when starting the tool don't use arguments (paths) for input dump and output files* - 1. Dump related tables that need to correct save telemetry `pg_dump -h localhost -U postgres -d thingsboard -t tenant -t customer -t user -t dashboard -t asset -t device -t alarm -t rule_chain -t rule_node -t entity_view -t widgets_bundle -t widget_type -t tenant_profile -t device_profile -t api_usage_state -t tb_user > related_entities.dmp` @@ -38,21 +36,26 @@ Tool use 3 different directories for saving SSTables - `ts_kv_cf`, `ts_kv_latest 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 - -telemetryFrom ./source/ts_kv_all.dmp + -telemetryFrom /home/user/dump/ts_kv_all.dmp + -relatedEntities /home/user/dump/relatedEntities.dmp -latestOut /home/ubunut/migration/ts_latest - -tsOut /home/ubunut/migration/ts - -partitionsOut /home/ubunut/migration/ts_partition - -castEnable false + -tsOut /home/ubunut/migration/ts + -partitionsOut /home/ubunut/migration/ts_partition + -castEnable false ``` +*Use your paths for program arguments* Tool execution time depends on DB size, CPU resources and Disk throughput @@ -62,20 +65,21 @@ Tool execution time depends on DB size, CPU resources and Disk throughput 1. [Optional] install Cassandra on the instance 2. [Optional] Using `cqlsh` create `thingsboard` keyspace and requred tables from this file `schema-ts.cql` 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 From f7f06a22978e7515c9657a3f23e93b0d13f85468 Mon Sep 17 00:00:00 2001 From: AndrewVolosytnykhThingsboard Date: Wed, 17 Mar 2021 01:04:34 +0200 Subject: [PATCH 8/9] Improvement of solution: README.md cleaned, build plugin fixed, some code cleaning --- tools/pom.xml | 2 +- .../client/tools/migrator/PgCaMigrator.java | 7 ++- .../client/tools/migrator/README.md | 14 ++--- .../tools/migrator/RelatedEntitiesParser.java | 52 ++++++++----------- 4 files changed, 33 insertions(+), 42 deletions(-) diff --git a/tools/pom.xml b/tools/pom.xml index 99f966b283..4877914b62 100644 --- a/tools/pom.xml +++ b/tools/pom.xml @@ -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/PgCaMigrator.java b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java index 7aec7ba5f3..b611fec6f0 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java @@ -43,7 +43,6 @@ public class PgCaMigrator { private final long LOG_BATCH = 1000000; private final long rowPerFile = 1000000; - private long linesProcessed = 0; private long linesTsMigrated = 0; private long linesLatestMigrated = 0; private long castErrors = 0; @@ -99,7 +98,7 @@ public class PgCaMigrator { System.out.println("START TO MIGRATE LATEST"); long start = System.currentTimeMillis(); processBlock(iterator, currentTsLatestWriter, outTsLatestDir, this::toValuesLatest); - System.out.println("FORMING OF SSL FOR LATEST TS FINISHED WITH TIME: " + (System.currentTimeMillis() - start) + " ms."); + System.out.println("TOTAL LINES MIGRATED: " + linesLatestMigrated + ", FORMING OF SSL FOR LATEST TS FINISHED WITH TIME: " + (System.currentTimeMillis() - start) + " ms."); isLatestDone = true; } @@ -107,7 +106,7 @@ public class PgCaMigrator { System.out.println("START TO MIGRATE TS"); long start = System.currentTimeMillis(); processBlock(iterator, currentTsWriter, outTsDir, this::toValuesTs); - System.out.println("FORMING OF SSL FOR TS FINISHED WITH TIME: " + (System.currentTimeMillis() - start) + " ms."); + System.out.println("TOTAL LINES MIGRATED: " + linesTsMigrated + ", FORMING OF SSL FOR TS FINISHED WITH TIME: " + (System.currentTimeMillis() - start) + " ms."); isTsDone = true; } } @@ -212,7 +211,7 @@ public class PgCaMigrator { private void processBlock(LineIterator iterator, CQLSSTableWriter writer, File outDir, Function, List> function) { String currentLine; - linesProcessed = 0; + long linesProcessed = 0; while(iterator.hasNext()) { logLinesProcessed(linesProcessed++); currentLine = iterator.nextLine(); 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 cfe1675eb2..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 @@ -23,7 +23,7 @@ It will generate single jar file with all required dependencies inside `target d 1. Dump related tables that need to correct save telemetry - `pg_dump -h localhost -U postgres -d thingsboard -t tenant -t customer -t user -t dashboard -t asset -t device -t alarm -t rule_chain -t rule_node -t entity_view -t widgets_bundle -t widget_type -t tenant_profile -t device_profile -t api_usage_state -t tb_user > related_entities.dmp` + `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: @@ -47,12 +47,12 @@ Create 3 empty directories. For example: **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 +java -jar ./tools-3.2.2-SNAPSHOT-jar-with-dependencies.jar -telemetryFrom /home/user/dump/ts_kv_all.dmp - -relatedEntities /home/user/dump/relatedEntities.dmp - -latestOut /home/ubunut/migration/ts_latest - -tsOut /home/ubunut/migration/ts - -partitionsOut /home/ubunut/migration/ts_partition + -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* @@ -63,7 +63,7 @@ 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. 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: 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 index fac2609797..7c1b29f2af 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/RelatedEntitiesParser.java @@ -27,6 +27,24 @@ 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)); @@ -45,36 +63,10 @@ public class RelatedEntitiesParser { try { while (lineIterator.hasNext()) { currentLine = lineIterator.nextLine(); - if (currentLine.startsWith("COPY public.alarm")) { - processBlock(lineIterator, EntityType.ALARM); - } else if (currentLine.startsWith("COPY public.asset")) { - processBlock(lineIterator, EntityType.ASSET); - } else if (currentLine.startsWith("COPY public.customer")) { - processBlock(lineIterator, EntityType.CUSTOMER); - } else if (currentLine.startsWith("COPY public.dashboard")) { - processBlock(lineIterator, EntityType.DASHBOARD); - } else if (currentLine.startsWith("COPY public.device")) { - processBlock(lineIterator, EntityType.DEVICE); - } else if (currentLine.startsWith("COPY public.rule_chain")) { - processBlock(lineIterator, EntityType.RULE_CHAIN); - } else if (currentLine.startsWith("COPY public.rule_node")) { - processBlock(lineIterator, EntityType.RULE_NODE); - } else if (currentLine.startsWith("COPY public.tenant")) { - processBlock(lineIterator, EntityType.TENANT); - } else if (currentLine.startsWith("COPY public.tb_user")) { - processBlock(lineIterator, EntityType.USER); - } else if (currentLine.startsWith("COPY public.entity_view")) { - processBlock(lineIterator, EntityType.ENTITY_VIEW); - } else if (currentLine.startsWith("COPY public.widgets_bundle")) { - processBlock(lineIterator, EntityType.WIDGETS_BUNDLE); - } else if (currentLine.startsWith("COPY public.widget_type")) { - processBlock(lineIterator, EntityType.WIDGET_TYPE); - } else if (currentLine.startsWith("COPY public.tenant_profile")) { - processBlock(lineIterator, EntityType.TENANT_PROFILE); - } else if (currentLine.startsWith("COPY public.device_profile")) { - processBlock(lineIterator, EntityType.DEVICE_PROFILE); - } else if (currentLine.startsWith("COPY public.api_usage_state")) { - processBlock(lineIterator, EntityType.API_USAGE_STATE); + for(Map.Entry entry : tableNameAndEntityType.entrySet()) { + if(currentLine.startsWith(entry.getKey())) { + processBlock(lineIterator, entry.getValue()); + } } } } finally { From 648dd81bfc008dc0afe99b48f5f8aa640b88ff9d Mon Sep 17 00:00:00 2001 From: Andrew Volostnykh Date: Wed, 17 Mar 2021 11:36:37 +0200 Subject: [PATCH 9/9] Redurant filter removed --- .../java/org/thingsboard/client/tools/migrator/PgCaMigrator.java | 1 - 1 file changed, 1 deletion(-) 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 index b611fec6f0..29a0a23ad6 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java @@ -222,7 +222,6 @@ public class PgCaMigrator { try { List raw = Arrays.stream(currentLine.trim().split("\t")) .map(String::trim) - .filter(StringUtils::isNotEmpty) .collect(Collectors.toList()); List values = function.apply(raw);