Dmitriymush
2 years ago
547 changed files with 15008 additions and 6387 deletions
File diff suppressed because one or more lines are too long
@ -0,0 +1,29 @@ |
|||
{ |
|||
"fqn": "charts.bars", |
|||
"name": "Bars", |
|||
"deprecated": true, |
|||
"image": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAMgAAACgCAMAAAB+IdObAAAA8FBMVEUhlvNMr1Bqamp5eXl7e3t8fHx9fX1+fn5/f3+AgICCgoKDg4OEhISGhoaHh4eKioqMjIyNjY2Ojo6QkJCRkZGSkpKWlpaXl5ebm5udnZ2enp6goKChoaGkpKSnp6epqamsrKyurq6xsbGzs7O1tbW2tra3t7e4uLi7u7u9vb3BwcHCwsLDw8PGxsbKysrNzc3Ozs7R0dHS0tLT09PZ2dna2trc3Nzd3d3e3t7g4ODh4eHj4+Pk5OTm5ubn5+fo6Ojp6enu7u7w8PDz8/P0Qzb09PT29vb39/f5+fn6+vr7+/v8/Pz9/f3+/v7/wQf///+dc+aLAAAAAWJLR0RPbmZBSQAAAcFJREFUeNrt3ds2AgEYhuHsaSOZbAvZi0r2YYjCJOW7/7txZhkcDNbM6h/vdwfPmlX/ybtmEorJErGCeJLadz3rkKPpZamaLTp925DHdFvSpKelU9uQ/cLKhtcdk7YqtiHruevtojch7ZZtQ0o1dcdfRqXNqm1IbVU3OaVamm/YhvQW5zIXOknnC5JUt7qEpE5fUv/5HVePy2UHAgQIECBAgAABAgQIECBxgrwGHBAgQIAAAQIECBAgQIAAAQIECJC/QRIBN0iQ+66voDMLuRp2fQWdVUhvNun6CjqrkJ0Dx/UVdEYhzXzfcX0FnVFIrlSZ2mx/LOiMQuqHh6k972NBZ/fv13G/KeiCQkIu4358EL8UdBafSGwuOxAgQIAAAQIECJDYQB4CDggQIECAAAECBAgQIECAAAECBAgQIECA/BrSufn0DjqjkEZmLXkWaUEXEmThXMeFSAu68H4j5b1IC7rQILfZTqQFXViQ1nRTL1EWdCFBnmYuJUVZ0IUEWR1xHKcXZUEXFDLwBR2XHQgQIECAAAEC5H9ChgIOCBAgQIAAAQIECBAgQIAAAQIECBAgQIAAAQIECBAgxiBmv+L6Bl9pkxYph15gAAAAAElFTkSuQmCC", |
|||
"description": "Displays latest values of the attributes or time-series data for multiple entities as separate bars.", |
|||
"descriptor": { |
|||
"type": "latest", |
|||
"sizeX": 7, |
|||
"sizeY": 5, |
|||
"resources": [ |
|||
{ |
|||
"url": "https://cdnjs.cloudflare.com/ajax/libs/Chart.js/2.3.0/Chart.min.js" |
|||
} |
|||
], |
|||
"templateHtml": "<canvas id=\"barChart\"></canvas>\n", |
|||
"templateCss": "", |
|||
"controllerScript": "self.onInit = function() {\n $scope = self.ctx.$scope;\n utils = $scope.$injector.get(self.ctx.servicesMap.get('utils'));\n settings = utils.deepClone(self.ctx.settings) || {};\n settings.showTooltip = utils.defaultValue(settings.showTooltip, true);\n \n Chart.defaults.global.tooltips.enabled = settings.showTooltip;\n \n var barData = {\n labels: [],\n datasets: []\n };\n \n for (var i = 0; i < self.ctx.datasources.length; i++) {\n var datasource = self.ctx.datasources[i];\n for (var d = 0; d < datasource.dataKeys.length; d++) {\n var dataKey = datasource.dataKeys[d];\n var units = dataKey.units && dataKey.units.length ? dataKey.units : self.ctx.units;\n units = units ? (' (' + units + ')') : '';\n var dataset = {\n label: dataKey.label + units,\n data: [0],\n backgroundColor: [dataKey.color],\n borderColor: [dataKey.color],\n borderWidth: 1\n }\n barData.datasets.push(dataset);\n }\n }\n\n var ctx = $('#barChart', self.ctx.$container);\n self.ctx.chart = new Chart(ctx, {\n type: 'bar',\n data: barData,\n options: {\n responsive: false,\n maintainAspectRatio: false,\n scales: {\n yAxes: [{\n ticks: {\n beginAtZero:true\n }\n }]\n }\n }\n });\n \n self.onResize();\n}\n\nself.onDataUpdated = function() {\n var c = 0;\n for (var i = 0; i < self.ctx.chart.data.datasets.length; i++) {\n var dataset = self.ctx.chart.data.datasets[i];\n var cellData = self.ctx.data[i]; \n if (cellData.data.length > 0) {\n var decimals;\n if (typeof cellData.dataKey.decimals !== 'undefined' \n && cellData.dataKey.decimals !== null ) {\n decimals = cellData.dataKey.decimals; \n } else {\n decimals = self.ctx.decimals;\n }\n var tvPair = cellData.data[cellData.data.length - 1];\n var value = self.ctx.utils.formatValue(tvPair[1], decimals);\n dataset.data[0] = parseFloat(value);\n }\n }\n self.ctx.chart.update();\n}\n\nself.onResize = function() {\n self.ctx.chart.resize();\n}\n\nself.onDestroy = function() {\n self.ctx.chart.destroy();\n self.ctx.chart = null;\n}\n", |
|||
"settingsSchema": "", |
|||
"dataKeySettingsSchema": "{}\n", |
|||
"settingsDirective": "tb-chart-widget-settings", |
|||
"defaultConfig": "{\"datasources\":[{\"type\":\"function\",\"name\":\"function\",\"dataKeys\":[{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"First\",\"color\":\"#2196f3\",\"settings\":{},\"_hash\":0.15479322438769105,\"funcBody\":\"var value = (prevValue-50) + Math.random() * 2 - 1;\\nif (value < 0) {\\n\\tvalue = 0;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value+50;\"},{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Second\",\"color\":\"#4caf50\",\"settings\":{},\"_hash\":0.545701115289893,\"funcBody\":\"var value = (prevValue-20) + Math.random() * 2 - 1;\\nif (value < 0) {\\n\\tvalue = 0;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value+20;\"},{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Third\",\"color\":\"#f44336\",\"settings\":{},\"_hash\":0.2592906835158064,\"funcBody\":\"var value = (prevValue-40) + Math.random() * 2 - 1;\\nif (value < 0) {\\n\\tvalue = 0;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value+40;\"},{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Fourth\",\"color\":\"#ffc107\",\"settings\":{},\"_hash\":0.12880275585455747,\"funcBody\":\"var value = (prevValue-50) + Math.random() * 2 - 1;\\nif (value < 0) {\\n\\tvalue = 0;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value+50;\"}]}],\"timewindow\":{\"realtime\":{\"timewindowMs\":60000}},\"showTitle\":true,\"backgroundColor\":\"#fff\",\"color\":\"rgba(0, 0, 0, 0.87)\",\"padding\":\"8px\",\"settings\":{},\"title\":\"Bars\"}" |
|||
}, |
|||
"externalId": null, |
|||
"tags": [ |
|||
"bar", |
|||
"bar chart" |
|||
] |
|||
} |
|||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@ -0,0 +1,118 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.exception.ExceptionUtils; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.data.StringUtils; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.HousekeeperTaskProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.housekeeper.HousekeeperConfig; |
|||
import org.thingsboard.server.queue.provider.TbCoreQueueFactory; |
|||
import org.thingsboard.server.queue.util.AfterStartUp; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; |
|||
|
|||
import javax.annotation.PreDestroy; |
|||
import java.util.LinkedHashSet; |
|||
import java.util.List; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
@TbCoreComponent |
|||
@Service |
|||
@Slf4j |
|||
public class HousekeeperReprocessingService { |
|||
|
|||
private final HousekeeperConfig config; |
|||
private final HousekeeperService housekeeperService; |
|||
private final QueueConsumerManager<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer; |
|||
private final TbQueueProducer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> producer; |
|||
private final TopicPartitionInfo submitTpi; |
|||
|
|||
private final ExecutorService consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-reprocessing-consumer")); |
|||
|
|||
public HousekeeperReprocessingService(HousekeeperConfig config, |
|||
@Lazy HousekeeperService housekeeperService, |
|||
TbCoreQueueFactory queueFactory) { |
|||
this.config = config; |
|||
this.housekeeperService = housekeeperService; |
|||
this.consumer = QueueConsumerManager.<TbProtoQueueMsg<ToHousekeeperServiceMsg>>builder() |
|||
.name("Housekeeper reprocessing") |
|||
.msgPackProcessor(this::processMsgs) |
|||
.pollInterval(config.getPollInterval()) |
|||
.consumerCreator(queueFactory::createHousekeeperReprocessingMsgConsumer) |
|||
.consumerExecutor(consumerExecutor) |
|||
.build(); |
|||
this.producer = queueFactory.createHousekeeperReprocessingMsgProducer(); |
|||
this.submitTpi = TopicPartitionInfo.builder().topic(producer.getDefaultTopic()).build(); |
|||
} |
|||
|
|||
@AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) |
|||
public void afterStartUp() { |
|||
consumer.subscribe(); // Kafka topic for tasks reprocessing has only 1 partition, so only one TB Core will reprocess tasks
|
|||
consumer.launch(); |
|||
} |
|||
|
|||
private void processMsgs(List<TbProtoQueueMsg<ToHousekeeperServiceMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer) throws Exception { |
|||
Thread.sleep(config.getTaskReprocessingDelay()); |
|||
|
|||
for (TbProtoQueueMsg<ToHousekeeperServiceMsg> msg : msgs) { |
|||
log.trace("Reprocessing task: {}", msg); |
|||
try { |
|||
housekeeperService.processTask(msg.getValue()); |
|||
} catch (InterruptedException e) { |
|||
return; |
|||
} catch (Throwable e) { |
|||
log.error("Unexpected error during message reprocessing [{}]", msg, e); |
|||
submitForReprocessing(msg.getValue(), e); |
|||
} |
|||
} |
|||
consumer.commit(); |
|||
} |
|||
|
|||
public void submitForReprocessing(ToHousekeeperServiceMsg msg, Throwable error) { |
|||
HousekeeperTaskProto task = msg.getTask(); |
|||
Set<String> errors = new LinkedHashSet<>(task.getErrorsList()); |
|||
errors.add(StringUtils.truncate(ExceptionUtils.getStackTrace(error), 1024)); |
|||
msg = msg.toBuilder() |
|||
.setTask(task.toBuilder() |
|||
.setAttempt(task.getAttempt() + 1) |
|||
.clearErrors().addAllErrors(errors) |
|||
.build()) |
|||
.build(); |
|||
|
|||
log.trace("Submitting for reprocessing: {}", msg); |
|||
producer.send(submitTpi, new TbProtoQueueMsg<>(UUID.randomUUID(), msg), null); |
|||
} |
|||
|
|||
@PreDestroy |
|||
private void stop() { |
|||
consumer.stop(); |
|||
consumerExecutor.shutdownNow(); |
|||
log.info("Stopped Housekeeper reprocessing service"); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,166 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.TaskProcessingFailureTrigger; |
|||
import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.housekeeper.HousekeeperConfig; |
|||
import org.thingsboard.server.queue.provider.TbCoreQueueFactory; |
|||
import org.thingsboard.server.queue.util.AfterStartUp; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.housekeeper.processor.HousekeeperTaskProcessor; |
|||
import org.thingsboard.server.service.housekeeper.stats.HousekeeperStatsService; |
|||
import org.thingsboard.server.service.queue.consumer.QueueConsumerManager; |
|||
|
|||
import javax.annotation.PreDestroy; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.concurrent.ExecutionException; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.Future; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.TimeoutException; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@TbCoreComponent |
|||
@Service |
|||
@Slf4j |
|||
public class HousekeeperService { |
|||
|
|||
private final Map<HousekeeperTaskType, HousekeeperTaskProcessor<?>> taskProcessors; |
|||
|
|||
private final HousekeeperConfig config; |
|||
private final HousekeeperReprocessingService reprocessingService; |
|||
private final Optional<HousekeeperStatsService> statsService; |
|||
private final NotificationRuleProcessor notificationRuleProcessor; |
|||
private final QueueConsumerManager<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer; |
|||
|
|||
private final ExecutorService consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-consumer")); |
|||
private final ExecutorService taskExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper-task-processor")); |
|||
|
|||
public HousekeeperService(HousekeeperConfig config, |
|||
HousekeeperReprocessingService reprocessingService, |
|||
TbCoreQueueFactory queueFactory, |
|||
Optional<HousekeeperStatsService> statsService, |
|||
NotificationRuleProcessor notificationRuleProcessor, |
|||
@Lazy List<HousekeeperTaskProcessor<?>> taskProcessors) { |
|||
this.config = config; |
|||
this.reprocessingService = reprocessingService; |
|||
this.statsService = statsService; |
|||
this.notificationRuleProcessor = notificationRuleProcessor; |
|||
this.consumer = QueueConsumerManager.<TbProtoQueueMsg<ToHousekeeperServiceMsg>>builder() |
|||
.name("Housekeeper") |
|||
.msgPackProcessor(this::processMsgs) |
|||
.pollInterval(config.getPollInterval()) |
|||
.consumerCreator(queueFactory::createHousekeeperMsgConsumer) |
|||
.consumerExecutor(consumerExecutor) |
|||
.build(); |
|||
this.taskProcessors = taskProcessors.stream().collect(Collectors.toMap(HousekeeperTaskProcessor::getTaskType, p -> p)); |
|||
} |
|||
|
|||
@AfterStartUp(order = AfterStartUp.REGULAR_SERVICE) |
|||
public void afterStartUp() { |
|||
consumer.subscribe(); |
|||
consumer.launch(); |
|||
} |
|||
|
|||
private void processMsgs(List<TbProtoQueueMsg<ToHousekeeperServiceMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToHousekeeperServiceMsg>> consumer) { |
|||
for (TbProtoQueueMsg<ToHousekeeperServiceMsg> msg : msgs) { |
|||
log.trace("Processing task: {}", msg); |
|||
try { |
|||
processTask(msg.getValue()); |
|||
} catch (InterruptedException e) { |
|||
return; |
|||
} catch (Throwable e) { |
|||
log.error("Unexpected error during message processing [{}]", msg, e); |
|||
reprocessingService.submitForReprocessing(msg.getValue(), e); |
|||
} |
|||
} |
|||
consumer.commit(); |
|||
} |
|||
|
|||
@SuppressWarnings("unchecked") |
|||
protected <T extends HousekeeperTask> void processTask(ToHousekeeperServiceMsg msg) throws Exception { |
|||
HousekeeperTask task = JacksonUtil.fromString(msg.getTask().getValue(), HousekeeperTask.class); |
|||
HousekeeperTaskType taskType = task.getTaskType(); |
|||
if (config.getDisabledTaskTypes().contains(taskType)) { |
|||
log.debug("Task type {} is disabled, ignoring {}", taskType, task); |
|||
return; |
|||
} |
|||
HousekeeperTaskProcessor<T> taskProcessor = (HousekeeperTaskProcessor<T>) taskProcessors.get(taskType); |
|||
if (taskProcessor == null) { |
|||
throw new IllegalArgumentException("Unsupported task type " + taskType); |
|||
} |
|||
|
|||
try { |
|||
long startTs = System.currentTimeMillis(); |
|||
Future<Object> future = taskExecutor.submit(() -> { |
|||
taskProcessor.process((T) task); |
|||
return null; |
|||
}); |
|||
future.get(config.getTaskProcessingTimeout(), TimeUnit.MILLISECONDS); |
|||
|
|||
long timing = System.currentTimeMillis() - startTs; |
|||
if (log.isDebugEnabled()) { |
|||
log.debug("[{}] Processed {} in {} ms (attempt {})", task.getTenantId(), task.getDescription(), timing, msg.getTask().getAttempt()); |
|||
} |
|||
statsService.ifPresent(statsService -> statsService.reportProcessed(taskType, msg, timing)); |
|||
} catch (InterruptedException e) { |
|||
throw e; |
|||
} catch (Throwable e) { |
|||
Throwable error = e; |
|||
if (e instanceof ExecutionException) { |
|||
error = e.getCause(); |
|||
} else if (e instanceof TimeoutException) { |
|||
error = new TimeoutException("Timeout after " + config.getTaskProcessingTimeout() + " seconds"); |
|||
} |
|||
|
|||
if (msg.getTask().getAttempt() < config.getMaxReprocessingAttempts()) { |
|||
log.warn("[{}] Failed to process {} (attempt {}), submitting for reprocessing", |
|||
task.getTenantId(), task.getDescription(), msg.getTask().getAttempt(), error); |
|||
reprocessingService.submitForReprocessing(msg, error); |
|||
} else { |
|||
log.error("[{}] Failed to process task in {} attempts: {}", task.getTenantId(), msg.getTask().getAttempt(), msg, e); |
|||
notificationRuleProcessor.process(TaskProcessingFailureTrigger.builder() |
|||
.task(task) |
|||
.error(error) |
|||
.attempt(msg.getTask().getAttempt()) |
|||
.build()); |
|||
} |
|||
statsService.ifPresent(statsService -> statsService.reportFailure(taskType, msg)); |
|||
} |
|||
} |
|||
|
|||
@PreDestroy |
|||
private void stop() { |
|||
consumer.stop(); |
|||
consumerExecutor.shutdownNow(); |
|||
log.info("Stopped Housekeeper service"); |
|||
} |
|||
|
|||
} |
|||
@ -1,103 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.ListeningExecutorService; |
|||
import com.google.common.util.concurrent.MoreExecutors; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.springframework.transaction.event.TransactionalEventListener; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.id.AlarmId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.eventsourcing.DeleteEntityEvent; |
|||
import org.thingsboard.server.dao.housekeeper.HouseKeeperService; |
|||
import org.thingsboard.server.service.entitiy.alarm.TbAlarmService; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.List; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class InMemoryHouseKeeperServiceService implements HouseKeeperService { |
|||
|
|||
final TbAlarmService alarmService; |
|||
|
|||
ListeningExecutorService executor; |
|||
|
|||
AtomicInteger queueSize = new AtomicInteger(); |
|||
AtomicInteger totalProcessedCounter = new AtomicInteger(); |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
log.debug("Starting HouseKeeper service"); |
|||
executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("housekeeper"))); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
if (executor != null) { |
|||
log.debug("Stopping HouseKeeper service"); |
|||
executor.shutdown(); |
|||
} |
|||
} |
|||
|
|||
@TransactionalEventListener(fallbackExecution = true) |
|||
public void handleEvent(DeleteEntityEvent<?> event) { |
|||
log.trace("[{}] DeleteEntityEvent handler: {}", event.getTenantId(), event); |
|||
EntityId entityId = event.getEntityId(); |
|||
if (EntityType.USER.equals(entityId.getEntityType())) { |
|||
unassignDeletedUserAlarms(event.getTenantId(), (User) event.getEntity(), event.getTs()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<List<AlarmId>> unassignDeletedUserAlarms(TenantId tenantId, User user, long unassignTs) { |
|||
log.debug("[{}][{}] unassignDeletedUserAlarms submitting, pending queue size: {} ", tenantId, user.getId().getId(), queueSize.get()); |
|||
queueSize.incrementAndGet(); |
|||
ListenableFuture<List<AlarmId>> future = executor.submit(() -> alarmService.unassignDeletedUserAlarms(tenantId, user, unassignTs)); |
|||
Futures.addCallback(future, new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(List<AlarmId> alarmIds) { |
|||
queueSize.decrementAndGet(); |
|||
totalProcessedCounter.incrementAndGet(); |
|||
log.debug("[{}][{}] unassignDeletedUserAlarms finished, pending queue size: {}, total processed count: {} ", |
|||
tenantId, user.getId().getId(), queueSize.get(), totalProcessedCounter.get()); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable throwable) { |
|||
queueSize.decrementAndGet(); |
|||
totalProcessedCounter.incrementAndGet(); |
|||
log.error("[{}][{}] unassignDeletedUserAlarms failed, pending queue size: {}, total processed count: {}", |
|||
tenantId, user.getId().getId(), queueSize.get(), totalProcessedCounter.get(), throwable); |
|||
} |
|||
}, MoreExecutors.directExecutor()); |
|||
return future; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,80 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.housekeeper.AlarmsDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.id.AlarmId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.util.TbPair; |
|||
import org.thingsboard.server.dao.alarm.AlarmService; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class AlarmsDeletionTaskProcessor extends HousekeeperTaskProcessor<AlarmsDeletionHousekeeperTask> { |
|||
|
|||
private final AlarmService alarmService; |
|||
|
|||
@Override |
|||
public void process(AlarmsDeletionHousekeeperTask task) throws Exception { |
|||
EntityId entityId = task.getEntityId(); |
|||
EntityType entityType = entityId.getEntityType(); |
|||
TenantId tenantId = task.getTenantId(); |
|||
|
|||
if (entityType == EntityType.DEVICE || entityType == EntityType.ASSET) { |
|||
if (task.getAlarms() == null) { |
|||
AlarmId lastId = null; |
|||
long lastCreatedTime = 0; |
|||
while (true) { |
|||
List<TbPair<UUID, Long>> alarms = alarmService.findAlarmIdsByOriginatorId(tenantId, entityId, lastCreatedTime, lastId, 128); |
|||
if (alarms.isEmpty()) { |
|||
break; |
|||
} |
|||
|
|||
housekeeperClient.submitTask(new AlarmsDeletionHousekeeperTask(tenantId, entityId, alarms.stream().map(TbPair::getFirst).toList())); |
|||
|
|||
TbPair<UUID, Long> last = alarms.get(alarms.size() - 1); |
|||
lastId = new AlarmId(last.getFirst()); |
|||
lastCreatedTime = last.getSecond(); |
|||
log.debug("[{}][{}][{}] Submitted task for deleting {} alarms", tenantId, entityType, entityId, alarms.size()); |
|||
} |
|||
} else { |
|||
for (UUID alarmId : task.getAlarms()) { |
|||
alarmService.delAlarm(tenantId, new AlarmId(alarmId)); |
|||
} |
|||
log.debug("[{}][{}][{}] Deleted {} alarms", tenantId, entityType, entityId, task.getAlarms().size()); |
|||
} |
|||
} |
|||
|
|||
int count = alarmService.deleteEntityAlarmRecords(tenantId, entityId); |
|||
log.debug("[{}][{}][{}] Deleted {} entity alarms", tenantId, entityType, entityId, count); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_ALARMS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.id.AlarmId; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.AlarmsUnassignHousekeeperTask; |
|||
import org.thingsboard.server.service.entitiy.alarm.TbAlarmService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class AlarmsUnassignTaskProcessor extends HousekeeperTaskProcessor<AlarmsUnassignHousekeeperTask> { |
|||
|
|||
private final TbAlarmService alarmService; |
|||
|
|||
@Override |
|||
public void process(AlarmsUnassignHousekeeperTask task) throws Exception { |
|||
List<AlarmId> alarms = alarmService.unassignDeletedUserAlarms(task.getTenantId(), (UserId) task.getEntityId(), task.getUserTitle(), task.getTs()); |
|||
log.debug("[{}][{}] Unassigned {} alarms", task.getTenantId(), task.getEntityId(), alarms.size()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.UNASSIGN_ALARMS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.dao.attributes.AttributesService; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class AttributesDeletionTaskProcessor extends HousekeeperTaskProcessor<HousekeeperTask> { |
|||
|
|||
private final AttributesService attributesService; |
|||
|
|||
@Override |
|||
public void process(HousekeeperTask task) throws Exception { |
|||
int deletedCount = attributesService.removeAllByEntityId(task.getTenantId(), task.getEntityId()); |
|||
log.debug("[{}][{}][{}] Deleted {} attributes", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), deletedCount); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_ATTRIBUTES; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,57 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.housekeeper.EntitiesDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.EntityIdFactory; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.entity.EntityDaoService; |
|||
import org.thingsboard.server.dao.entity.EntityServiceRegistry; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class EntitiesDeletionTaskProcessor extends HousekeeperTaskProcessor<EntitiesDeletionHousekeeperTask> { |
|||
|
|||
private final EntityServiceRegistry entityServiceRegistry; |
|||
|
|||
@Override |
|||
public void process(EntitiesDeletionHousekeeperTask task) throws Exception { |
|||
EntityType entityType = task.getEntityType(); |
|||
TenantId tenantId = task.getTenantId(); |
|||
EntityDaoService entityService = entityServiceRegistry.getServiceByEntityType(entityType); |
|||
|
|||
for (UUID entityUuid : task.getEntities()) { |
|||
EntityId entityId = EntityIdFactory.getByTypeAndUuid(entityType, entityUuid); |
|||
entityService.deleteEntity(tenantId, entityId, true); |
|||
} |
|||
log.debug("[{}] Deleted {} {}s", tenantId, task.getEntities().size(), entityType.getNormalName().toLowerCase()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_ENTITIES; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.dao.event.EventService; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class EventsDeletionTaskProcessor extends HousekeeperTaskProcessor<HousekeeperTask> { |
|||
|
|||
private final EventService eventService; |
|||
|
|||
@Override |
|||
public void process(HousekeeperTask task) throws Exception { |
|||
eventService.removeEvents(task.getTenantId(), task.getEntityId(), null, 0L, System.currentTimeMillis()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_EVENTS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.msg.housekeeper.HousekeeperClient; |
|||
|
|||
public abstract class HousekeeperTaskProcessor<T extends HousekeeperTask> { |
|||
|
|||
@Autowired |
|||
protected HousekeeperClient housekeeperClient; |
|||
|
|||
public abstract void process(T task) throws Exception; |
|||
|
|||
public abstract HousekeeperTaskType getTaskType(); |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.LatestTsDeletionHousekeeperTask; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class LatestTsDeletionTaskProcessor extends HousekeeperTaskProcessor<LatestTsDeletionHousekeeperTask> { |
|||
|
|||
private final TimeseriesService timeseriesService; |
|||
|
|||
@Override |
|||
public void process(LatestTsDeletionHousekeeperTask task) throws Exception { |
|||
timeseriesService.removeLatest(task.getTenantId(), task.getEntityId(), List.of(task.getKey())).get(); |
|||
log.debug("[{}][{}][{}] Deleted latest telemetry for key '{}'", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), task.getKey()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_LATEST_TS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.LatestTsDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.TsHistoryDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class TelemetryDeletionTaskProcessor extends HousekeeperTaskProcessor<HousekeeperTask> { |
|||
|
|||
private final TimeseriesService timeseriesService; |
|||
|
|||
@Override |
|||
public void process(HousekeeperTask task) throws Exception { |
|||
TenantId tenantId = task.getTenantId(); |
|||
EntityId entityId = task.getEntityId(); |
|||
List<String> keys = timeseriesService.findAllKeysByEntityIds(tenantId, List.of(entityId)); |
|||
|
|||
for (String key : keys) { |
|||
var latestTsDeletionTask = new LatestTsDeletionHousekeeperTask(tenantId, entityId, key); |
|||
housekeeperClient.submitTask(latestTsDeletionTask); |
|||
|
|||
var tsHistoryDeletionTask = new TsHistoryDeletionHousekeeperTask(tenantId, entityId, key); |
|||
housekeeperClient.submitTask(tsHistoryDeletionTask); |
|||
} |
|||
|
|||
log.trace("[{}][{}][{}] Submitted latest and ts history deletion tasks for {} keys", tenantId, entityId.getEntityType(), entityId, keys.size()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_TELEMETRY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,63 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.housekeeper.EntitiesDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.TenantEntitiesDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.Dao; |
|||
import org.thingsboard.server.dao.entity.EntityDaoRegistry; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class TenantEntitiesDeletionTaskProcessor extends HousekeeperTaskProcessor<TenantEntitiesDeletionHousekeeperTask> { |
|||
|
|||
private final EntityDaoRegistry entityDaoRegistry; |
|||
|
|||
@Override |
|||
public void process(TenantEntitiesDeletionHousekeeperTask task) throws Exception { |
|||
EntityType entityType = task.getEntityType(); |
|||
TenantId tenantId = task.getTenantId(); |
|||
Dao<?> entityDao = entityDaoRegistry.getDao(entityType); |
|||
|
|||
UUID last = null; |
|||
while (true) { |
|||
List<UUID> entities = entityDao.findIdsByTenantIdAndIdOffset(tenantId, last, 128); |
|||
if (entities.isEmpty()) { |
|||
break; |
|||
} |
|||
|
|||
housekeeperClient.submitTask(new EntitiesDeletionHousekeeperTask(tenantId, entityType, entities)); |
|||
last = entities.get(entities.size() - 1); |
|||
log.debug("[{}] Submitted task for deleting {} {}s", tenantId, entities.size(), entityType.getNormalName().toLowerCase()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_TENANT_ENTITIES; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.processor; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.housekeeper.TsHistoryDeletionHousekeeperTask; |
|||
import org.thingsboard.server.common.data.kv.BaseDeleteTsKvQuery; |
|||
import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class TsHistoryDeletionTaskProcessor extends HousekeeperTaskProcessor<TsHistoryDeletionHousekeeperTask> { |
|||
|
|||
private final TimeseriesService timeseriesService; |
|||
|
|||
@Override |
|||
public void process(TsHistoryDeletionHousekeeperTask task) throws Exception { |
|||
DeleteTsKvQuery deleteQuery = new BaseDeleteTsKvQuery(task.getKey(), 0, System.currentTimeMillis(), false, false); |
|||
timeseriesService.remove(task.getTenantId(), task.getEntityId(), List.of(deleteQuery)).get(); |
|||
log.debug("[{}][{}][{}] Deleted timeseries history for key '{}'", task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), task.getKey()); |
|||
} |
|||
|
|||
@Override |
|||
public HousekeeperTaskType getTaskType() { |
|||
return HousekeeperTaskType.DELETE_TS_HISTORY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,125 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper.stats; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.scheduling.annotation.Scheduled; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.stats.DefaultCounter; |
|||
import org.thingsboard.server.common.stats.StatsCounter; |
|||
import org.thingsboard.server.common.stats.StatsFactory; |
|||
import org.thingsboard.server.common.stats.StatsTimer; |
|||
import org.thingsboard.server.common.stats.StatsType; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.EnumMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Objects; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@ConditionalOnProperty(name = "queue.core.housekeeper.stats.enabled", havingValue = "true", matchIfMissing = true) |
|||
public class HousekeeperStatsService { |
|||
|
|||
private final Map<HousekeeperTaskType, HousekeeperStats> stats = new EnumMap<>(HousekeeperTaskType.class); |
|||
|
|||
public HousekeeperStatsService(StatsFactory statsFactory) { |
|||
for (HousekeeperTaskType taskType : HousekeeperTaskType.values()) { |
|||
stats.put(taskType, new HousekeeperStats(taskType, statsFactory)); |
|||
} |
|||
} |
|||
|
|||
@Scheduled(initialDelayString = "${queue.core.housekeeper.stats.print-interval-ms:60000}", |
|||
fixedDelayString = "${queue.core.housekeeper.stats.print-interval-ms:60000}") |
|||
private void reportStats() { |
|||
String statsStr = stats.values().stream().map(stats -> { |
|||
String countersStr = stats.getCounters().stream() |
|||
.filter(counter -> counter.get() > 0) |
|||
.map(counter -> counter.getName() + " = [" + counter.get() + "]") |
|||
.collect(Collectors.joining(" ")); |
|||
if (countersStr.isEmpty()) { |
|||
return null; |
|||
} else { |
|||
return stats.getTaskType() + " " + countersStr + " avgProcessingTime [" + stats.getProcessingTimer().getAvg() + " ms]"; |
|||
} |
|||
}).filter(Objects::nonNull).collect(Collectors.joining(", ")); |
|||
|
|||
if (!statsStr.isEmpty()) { |
|||
stats.values().forEach(HousekeeperStats::reset); |
|||
log.info("Housekeeper stats: {}", statsStr); |
|||
} |
|||
} |
|||
|
|||
public void reportProcessed(HousekeeperTaskType taskType, ToHousekeeperServiceMsg msg, long timing) { |
|||
HousekeeperStats stats = this.stats.get(taskType); |
|||
if (msg.getTask().getErrorsCount() == 0) { |
|||
stats.getProcessedCounter().increment(); |
|||
} else { |
|||
stats.getReprocessedCounter().increment(); |
|||
} |
|||
stats.getProcessingTimer().record(timing); |
|||
} |
|||
|
|||
public void reportFailure(HousekeeperTaskType taskType, ToHousekeeperServiceMsg msg) { |
|||
HousekeeperStats stats = this.stats.get(taskType); |
|||
if (msg.getTask().getErrorsCount() == 0) { |
|||
stats.getFailedProcessingCounter().increment(); |
|||
} else { |
|||
stats.getFailedReprocessingCounter().increment(); |
|||
} |
|||
} |
|||
|
|||
@Getter |
|||
static class HousekeeperStats { |
|||
private final HousekeeperTaskType taskType; |
|||
private final List<StatsCounter> counters = new ArrayList<>(); |
|||
|
|||
private final StatsCounter processedCounter; |
|||
private final StatsCounter failedProcessingCounter; |
|||
private final StatsCounter reprocessedCounter; |
|||
private final StatsCounter failedReprocessingCounter; |
|||
|
|||
private final StatsTimer processingTimer; |
|||
|
|||
public HousekeeperStats(HousekeeperTaskType taskType, StatsFactory statsFactory) { |
|||
this.taskType = taskType; |
|||
this.processedCounter = register("processed", statsFactory); |
|||
this.failedProcessingCounter = register("failedProcessing", statsFactory); |
|||
this.reprocessedCounter = register("reprocessed", statsFactory); |
|||
this.failedReprocessingCounter = register("failedReprocessing", statsFactory); |
|||
this.processingTimer = statsFactory.createTimer(StatsType.HOUSEKEEPER, "processingTime", "taskType", taskType.name()); |
|||
} |
|||
|
|||
private StatsCounter register(String statsName, StatsFactory statsFactory) { |
|||
StatsCounter counter = statsFactory.createStatsCounter(StatsType.HOUSEKEEPER.getName(), statsName, "taskType", taskType.name()); |
|||
counters.add(counter); |
|||
return counter; |
|||
} |
|||
|
|||
public void reset() { |
|||
counters.forEach(DefaultCounter::clear); |
|||
processingTimer.reset(); |
|||
} |
|||
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,53 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.notification.rule.trigger; |
|||
|
|||
import org.apache.commons.lang3.exception.ExceptionUtils; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.StringUtils; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.notification.info.TaskProcessingFailureNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.TaskProcessingFailureTrigger; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.config.TaskProcessingFailureNotificationRuleTriggerConfig; |
|||
|
|||
@Service |
|||
public class TaskProcessingFailureTriggerProcessor implements NotificationRuleTriggerProcessor<TaskProcessingFailureTrigger, TaskProcessingFailureNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(TaskProcessingFailureTrigger trigger, TaskProcessingFailureNotificationRuleTriggerConfig triggerConfig) { |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
public TaskProcessingFailureNotificationInfo constructNotificationInfo(TaskProcessingFailureTrigger trigger) { |
|||
HousekeeperTask task = trigger.getTask(); |
|||
return TaskProcessingFailureNotificationInfo.builder() |
|||
.tenantId(task.getTenantId()) |
|||
.entityId(task.getEntityId()) |
|||
.taskType(task.getTaskType()) |
|||
.taskDescription(task.getDescription()) |
|||
.error(StringUtils.truncate(ExceptionUtils.getStackTrace(trigger.getError()), 1024)) |
|||
.attempt(trigger.getAttempt()) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.TASK_PROCESSING_FAILURE; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,109 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.queue.consumer; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.TbQueueMsg; |
|||
|
|||
import java.util.List; |
|||
import java.util.Set; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.function.Supplier; |
|||
|
|||
@Slf4j |
|||
public class QueueConsumerManager<M extends TbQueueMsg> { |
|||
|
|||
private final String name; |
|||
private final MsgPackProcessor<M> msgPackProcessor; |
|||
private final long pollInterval; |
|||
private final ExecutorService consumerExecutor; |
|||
private final String threadPrefix; |
|||
|
|||
@Getter |
|||
private final TbQueueConsumer<M> consumer; |
|||
private volatile boolean stopped; |
|||
|
|||
@Builder |
|||
public QueueConsumerManager(String name, MsgPackProcessor<M> msgPackProcessor, |
|||
long pollInterval, Supplier<TbQueueConsumer<M>> consumerCreator, |
|||
ExecutorService consumerExecutor, String threadPrefix) { |
|||
this.name = name; |
|||
this.pollInterval = pollInterval; |
|||
this.msgPackProcessor = msgPackProcessor; |
|||
this.consumerExecutor = consumerExecutor; |
|||
this.threadPrefix = threadPrefix; |
|||
this.consumer = consumerCreator.get(); |
|||
} |
|||
|
|||
public void subscribe() { |
|||
consumer.subscribe(); |
|||
} |
|||
|
|||
public void subscribe(Set<TopicPartitionInfo> partitions) { |
|||
consumer.subscribe(partitions); |
|||
} |
|||
|
|||
public void launch() { |
|||
log.info("[{}] Launching consumer", name); |
|||
consumerExecutor.submit(() -> { |
|||
if (threadPrefix != null) { |
|||
ThingsBoardThreadFactory.addThreadNamePrefix(threadPrefix); |
|||
} |
|||
try { |
|||
consumerLoop(consumer); |
|||
} catch (Throwable e) { |
|||
log.error("Failure in consumer loop", e); |
|||
} |
|||
log.info("[{}] Consumer stopped", name); |
|||
}); |
|||
} |
|||
|
|||
private void consumerLoop(TbQueueConsumer<M> consumer) { |
|||
while (!stopped && !consumer.isStopped()) { |
|||
try { |
|||
List<M> msgs = consumer.poll(pollInterval); |
|||
if (msgs.isEmpty()) { |
|||
continue; |
|||
} |
|||
msgPackProcessor.process(msgs, consumer); |
|||
} catch (Exception e) { |
|||
if (!consumer.isStopped()) { |
|||
log.warn("Failed to process messages from queue", e); |
|||
try { |
|||
Thread.sleep(pollInterval); |
|||
} catch (InterruptedException interruptedException) { |
|||
log.trace("Failed to wait until the server has capacity to handle new requests", interruptedException); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
public void stop() { |
|||
log.debug("[{}] Stopping consumer", name); |
|||
stopped = true; |
|||
consumer.unsubscribe(); |
|||
} |
|||
|
|||
public interface MsgPackProcessor<M extends TbQueueMsg> { |
|||
void process(List<M> msgs, TbQueueConsumer<M> consumer) throws Exception; |
|||
} |
|||
} |
|||
@ -0,0 +1,478 @@ |
|||
/** |
|||
* Copyright © 2016-2024 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.server.service.housekeeper; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import com.fasterxml.jackson.databind.node.TextNode; |
|||
import org.junit.After; |
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.mockito.ArgumentMatcher; |
|||
import org.mockito.Mockito; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.test.mock.mockito.SpyBean; |
|||
import org.springframework.test.context.TestPropertySource; |
|||
import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.rule.engine.metadata.TbGetAttributesNode; |
|||
import org.thingsboard.rule.engine.metadata.TbGetAttributesNodeConfiguration; |
|||
import org.thingsboard.server.common.data.ApiUsageState; |
|||
import org.thingsboard.server.common.data.AttributeScope; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.EventInfo; |
|||
import org.thingsboard.server.common.data.alarm.Alarm; |
|||
import org.thingsboard.server.common.data.alarm.AlarmSeverity; |
|||
import org.thingsboard.server.common.data.alarm.EntityAlarm; |
|||
import org.thingsboard.server.common.data.asset.Asset; |
|||
import org.thingsboard.server.common.data.event.EventType; |
|||
import org.thingsboard.server.common.data.event.LifecycleEvent; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTask; |
|||
import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType; |
|||
import org.thingsboard.server.common.data.id.AlarmId; |
|||
import org.thingsboard.server.common.data.id.AssetId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; |
|||
import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; |
|||
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
|||
import org.thingsboard.server.common.data.kv.StringDataEntry; |
|||
import org.thingsboard.server.common.data.kv.TsKvEntry; |
|||
import org.thingsboard.server.common.data.msg.TbNodeConnectionType; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.common.data.page.TimePageLink; |
|||
import org.thingsboard.server.common.data.relation.EntityRelation; |
|||
import org.thingsboard.server.common.data.relation.RelationTypeGroup; |
|||
import org.thingsboard.server.common.data.rule.RuleChain; |
|||
import org.thingsboard.server.common.data.rule.RuleChainMetaData; |
|||
import org.thingsboard.server.common.data.rule.RuleChainType; |
|||
import org.thingsboard.server.common.data.rule.RuleNode; |
|||
import org.thingsboard.server.controller.AbstractControllerTest; |
|||
import org.thingsboard.server.dao.alarm.AlarmDao; |
|||
import org.thingsboard.server.dao.alarm.AlarmService; |
|||
import org.thingsboard.server.dao.attributes.AttributesService; |
|||
import org.thingsboard.server.dao.entity.EntityServiceRegistry; |
|||
import org.thingsboard.server.dao.event.EventService; |
|||
import org.thingsboard.server.dao.relation.RelationService; |
|||
import org.thingsboard.server.dao.rule.RuleChainService; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
import org.thingsboard.server.dao.usagerecord.ApiUsageStateDao; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.HousekeeperTaskProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; |
|||
import org.thingsboard.server.service.housekeeper.processor.TsHistoryDeletionTaskProcessor; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Arrays; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.concurrent.TimeoutException; |
|||
import java.util.function.Function; |
|||
import java.util.function.Predicate; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.awaitility.Awaitility.await; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.argThat; |
|||
import static org.mockito.Mockito.doCallRealMethod; |
|||
import static org.mockito.Mockito.doThrow; |
|||
import static org.mockito.Mockito.never; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
|||
|
|||
@DaoSqlTest |
|||
@TestPropertySource(properties = { |
|||
"transport.http.enabled=true", |
|||
"queue.core.housekeeper.task-reprocessing-delay-ms=2000", |
|||
"queue.core.housekeeper.poll-interval-ms=1000", |
|||
"queue.core.housekeeper.max-reprocessing-attempts=5" |
|||
}) |
|||
public class HousekeeperServiceTest extends AbstractControllerTest { |
|||
|
|||
@SpyBean |
|||
private HousekeeperService housekeeperService; |
|||
@SpyBean |
|||
private HousekeeperReprocessingService housekeeperReprocessingService; |
|||
@Autowired |
|||
private EventService eventService; |
|||
@Autowired |
|||
private TimeseriesService timeseriesService; |
|||
@Autowired |
|||
private AttributesService attributesService; |
|||
@Autowired |
|||
private RuleChainService ruleChainService; |
|||
@Autowired |
|||
private AlarmService alarmService; |
|||
@Autowired |
|||
private AlarmDao alarmDao; |
|||
@Autowired |
|||
private RelationService relationService; |
|||
@Autowired |
|||
private ApiUsageStateDao apiUsageStateDao; |
|||
@Autowired |
|||
private EntityServiceRegistry entityServiceRegistry; |
|||
@SpyBean |
|||
private TsHistoryDeletionTaskProcessor tsHistoryDeletionTaskProcessor; |
|||
|
|||
private TenantId tenantId; |
|||
|
|||
private static final String TELEMETRY_KEY = "telemetry1"; |
|||
private static final String ATTRIBUTE_KEY = "_attribute1"; |
|||
private static final String KV_VALUE = "ewfewfwef"; |
|||
|
|||
@Before |
|||
public void setUp() throws Exception { |
|||
loginTenantAdmin(); |
|||
this.tenantId = super.tenantId; |
|||
} |
|||
|
|||
@After |
|||
public void tearDown() throws Exception { |
|||
} |
|||
|
|||
@Test |
|||
public void whenDeviceIsDeleted_thenCleanUpRelatedData() throws Exception { |
|||
Device device = createDevice("test", "test"); |
|||
createRelatedData(device.getId()); |
|||
|
|||
doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); |
|||
|
|||
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
verifyNoRelatedData(device.getId()); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenRuleChainIsDeleted_thenCleanUpRelatedData() throws Exception { |
|||
RuleChainMetaData ruleChainMetaData = createRuleChain(); |
|||
RuleChainId ruleChainId = ruleChainMetaData.getRuleChainId(); |
|||
RuleNodeId ruleNode1Id = ruleChainMetaData.getNodes().get(0).getId(); |
|||
RuleNodeId ruleNode2Id = ruleChainMetaData.getNodes().get(1).getId(); |
|||
createRelatedData(ruleChainId); |
|||
createRelatedData(ruleNode1Id); |
|||
createRelatedData(ruleNode2Id); |
|||
|
|||
doDelete("/api/ruleChain/" + ruleChainId).andExpect(status().isOk()); |
|||
|
|||
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
verifyNoRelatedData(ruleNode1Id); |
|||
verifyNoRelatedData(ruleNode2Id); |
|||
verifyNoRelatedData(ruleChainId); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenUserIsDeleted_thenCleanUpRelatedData() throws Exception { |
|||
Device device = createDevice("test", "test"); |
|||
UserId userId = customerUserId; |
|||
createRelatedData(userId); |
|||
Alarm alarm = Alarm.builder() |
|||
.type("test") |
|||
.tenantId(tenantId) |
|||
.originator(device.getId()) |
|||
.severity(AlarmSeverity.MAJOR) |
|||
.build(); |
|||
alarm = doPost("/api/alarm", alarm, Alarm.class); |
|||
AlarmId alarmId = alarm.getId(); |
|||
alarm = doPost("/api/alarm/" + alarmId + "/assign/" + userId, "", Alarm.class); |
|||
assertThat(alarm.getAssigneeId()).isEqualTo(userId); |
|||
assertThat(alarmService.findAlarmIdsByAssigneeId(tenantId, userId, new PageLink(100)).getData()).isNotEmpty(); |
|||
|
|||
doDelete("/api/user/" + userId).andExpect(status().isOk()); |
|||
|
|||
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
verifyNoRelatedData(userId); |
|||
assertThat(alarmService.findAlarmById(tenantId, alarmId).getAssigneeId()).isNull(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenDeviceIsDeleted_thenDeleteAllAlarms() throws Exception { |
|||
Device device = createDevice("test", "test"); |
|||
for (int i = 1; i <= 1000; i++) { |
|||
createAlarm(device.getId()); |
|||
} |
|||
|
|||
doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); |
|||
|
|||
await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
verifyNoAlarms(device.getId()); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenTenantIsDeleted_thenDeleteAllEntitiesAndCleanUpRelatedData() throws Exception { |
|||
loginDifferentTenant(); |
|||
tenantId = differentTenantId; |
|||
|
|||
createRelatedData(tenantId); |
|||
createDifferentTenantCustomer(); |
|||
createRelatedData(differentTenantCustomerId); |
|||
loginDifferentTenant(); |
|||
|
|||
List<DeviceId> devices = new ArrayList<>(); |
|||
for (int i = 1; i <= 300; i++) { |
|||
Device device = createDevice("test" + i, "test" + i); |
|||
devices.add(device.getId()); |
|||
} |
|||
DeviceId firstDevice = devices.get(0); |
|||
createRelatedData(firstDevice); |
|||
DeviceId lastDevice = devices.get(devices.size() - 1); |
|||
createRelatedData(lastDevice); |
|||
|
|||
Asset asset = createAsset(); |
|||
createRelatedData(asset.getId()); |
|||
createRelation(firstDevice, asset.getId()); |
|||
createAlarm(firstDevice, asset.getId()); |
|||
|
|||
RuleChainMetaData ruleChainMetaData = createRuleChain(); |
|||
RuleChainId ruleChainId = ruleChainMetaData.getRuleChainId(); |
|||
RuleNodeId ruleNode1Id = ruleChainMetaData.getNodes().get(0).getId(); |
|||
RuleNodeId ruleNode2Id = ruleChainMetaData.getNodes().get(1).getId(); |
|||
createRelatedData(ruleChainId); |
|||
createRelatedData(ruleNode1Id); |
|||
createRelatedData(ruleNode2Id); |
|||
|
|||
UserId userId = savedDifferentTenantUser.getId(); |
|||
createRelatedData(userId); |
|||
|
|||
ApiUsageState tenantApiUsageState = apiUsageStateDao.findApiUsageStateByEntityId(differentTenantId); |
|||
|
|||
loginSysAdmin(); |
|||
deleteDifferentTenant(); |
|||
|
|||
await().atMost(60, TimeUnit.SECONDS).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
for (DeviceId deviceId : devices) { |
|||
verifyNoRelatedData(deviceId); |
|||
} |
|||
verifyNoRelatedData(asset.getId()); |
|||
verifyNoRelatedData(ruleNode1Id); |
|||
verifyNoRelatedData(ruleNode2Id); |
|||
verifyNoRelatedData(ruleChainId); |
|||
verifyNoRelatedData(userId); |
|||
verifyNoRelatedData(differentTenantCustomerId); |
|||
verifyNoRelatedData(tenantApiUsageState.getId()); |
|||
verifyNoRelatedData(tenantId); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenTaskProcessingFails_thenReprocess() throws Exception { |
|||
TimeoutException error = new TimeoutException("Test timeout"); |
|||
doThrow(error).when(tsHistoryDeletionTaskProcessor).process(any()); |
|||
|
|||
Device device = createDevice("test", "test"); |
|||
createRelatedData(device.getId()); |
|||
|
|||
doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); |
|||
|
|||
int attempts = 2; |
|||
await().atMost(30, TimeUnit.SECONDS).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
for (int i = 0; i <= attempts; i++) { |
|||
int attempt = i; |
|||
verify(housekeeperReprocessingService).submitForReprocessing(argThat(getTaskMatcher(device.getId(), HousekeeperTaskType.DELETE_TS_HISTORY, |
|||
task -> task.getAttempt() == attempt)), argThat(e -> e.getMessage().equals(error.getMessage()))); |
|||
} |
|||
}); |
|||
|
|||
assertThat(getTimeseriesHistory(device.getId())).isNotEmpty(); |
|||
doCallRealMethod().when(tsHistoryDeletionTaskProcessor).process(any()); |
|||
await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
assertThat(getTimeseriesHistory(device.getId())).isEmpty(); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void whenReprocessingAttemptsExceeded_thenDropTheTask() throws Exception { |
|||
TimeoutException error = new TimeoutException("Test timeout"); |
|||
doThrow(error).when(tsHistoryDeletionTaskProcessor).process(any()); |
|||
|
|||
Device device = createDevice("test", "test"); |
|||
createRelatedData(device.getId()); |
|||
|
|||
doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); |
|||
|
|||
int maxAttempts = 5; |
|||
await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
|||
for (int i = 1; i <= maxAttempts; i++) { |
|||
verifyTaskProcessing(device.getId(), HousekeeperTaskType.DELETE_TS_HISTORY, i); |
|||
} |
|||
}); |
|||
|
|||
Mockito.clearInvocations(housekeeperService); |
|||
doCallRealMethod().when(tsHistoryDeletionTaskProcessor).process(any()); |
|||
TimeUnit.SECONDS.sleep(2); |
|||
verify(housekeeperService, never()).processTask(argThat(getTaskMatcher(device.getId(), HousekeeperTaskType.DELETE_TS_HISTORY, null))); |
|||
} |
|||
|
|||
private void verifyTaskProcessing(EntityId entityId, HousekeeperTaskType taskType, int expectedAttempt) throws Exception { |
|||
verify(housekeeperService).processTask(argThat(getTaskMatcher(entityId, taskType, task -> task.getAttempt() == expectedAttempt))); |
|||
} |
|||
|
|||
private ArgumentMatcher<ToHousekeeperServiceMsg> getTaskMatcher(EntityId entityId, HousekeeperTaskType taskType, |
|||
Predicate<HousekeeperTaskProto> additionalCheck) { |
|||
return msg -> { |
|||
HousekeeperTask task = JacksonUtil.fromString(msg.getTask().getValue(), HousekeeperTask.class); |
|||
return task.getEntityId().equals(entityId) && task.getTaskType() == taskType && (additionalCheck == null || additionalCheck.test(msg.getTask())); |
|||
}; |
|||
} |
|||
|
|||
private void createRelatedData(EntityId entityId) throws Exception { |
|||
createTelemetry(entityId); |
|||
for (AttributeScope scope : AttributeScope.values()) { |
|||
createAttribute(entityId, scope, scope + ATTRIBUTE_KEY); |
|||
} |
|||
createEvent(entityId); |
|||
} |
|||
|
|||
private void verifyNoRelatedData(EntityId entityId) throws Exception { |
|||
assertThat(entityServiceRegistry.getServiceByEntityType(entityId.getEntityType()).findEntity(tenantId, entityId)).isEmpty(); |
|||
|
|||
assertThat(getLatestTelemetry(entityId)).isNull(); |
|||
assertThat(getTimeseriesHistory(entityId)).isEmpty(); |
|||
for (AttributeScope scope : AttributeScope.values()) { |
|||
assertThat(attributesService.findAll(tenantId, entityId, scope).get()).isEmpty(); |
|||
} |
|||
assertThat(getEvents(entityId)).isEmpty(); |
|||
assertThat(alarmDao.findEntityAlarmRecordsByEntityId(tenantId, entityId)).isEmpty(); |
|||
verifyNoAlarms(entityId); |
|||
assertThat(relationService.findByTo(tenantId, entityId, RelationTypeGroup.COMMON)).isEmpty(); |
|||
assertThat(relationService.findByFrom(tenantId, entityId, RelationTypeGroup.COMMON)).isEmpty(); |
|||
} |
|||
|
|||
private void verifyNoAlarms(EntityId entityId) { |
|||
assertThat(alarmService.findAlarmIdsByOriginatorId(tenantId, entityId, 0, null, 10)).isEmpty(); |
|||
} |
|||
|
|||
private void createAttribute(EntityId entityId, AttributeScope scope, String key) throws Exception { |
|||
attributesService.save(tenantId, entityId, scope, new BaseAttributeKvEntry(System.currentTimeMillis(), new StringDataEntry(key, KV_VALUE))).get(); |
|||
} |
|||
|
|||
private void createTelemetry(EntityId entityId) throws Exception { |
|||
timeseriesService.save(tenantId, entityId, new BasicTsKvEntry(System.currentTimeMillis(), new StringDataEntry(TELEMETRY_KEY, KV_VALUE))).get(); |
|||
} |
|||
|
|||
private void createEvent(EntityId entityId) { |
|||
LifecycleEvent event = LifecycleEvent.builder() |
|||
.tenantId(tenantId) |
|||
.entityId(entityId.getId()) |
|||
.serviceId("test") |
|||
.lcEventType("test") |
|||
.success(true) |
|||
.build(); |
|||
eventService.saveAsync(event); |
|||
await().atMost(10, TimeUnit.SECONDS) |
|||
.until(() -> !getEvents(entityId).isEmpty()); |
|||
|
|||
} |
|||
|
|||
private void createRelation(DeviceId to, AssetId from) { |
|||
EntityRelation relation = new EntityRelation(from, to, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON); |
|||
relationService.saveRelation(tenantId, relation); |
|||
} |
|||
|
|||
private void createAlarm(DeviceId deviceId, EntityId propagatedEntityId) { |
|||
Alarm alarm = doPost("/api/alarm", Alarm.builder() |
|||
.tenantId(tenantId) |
|||
.originator(deviceId) |
|||
.severity(AlarmSeverity.CRITICAL) |
|||
.type("test alarm for " + deviceId) |
|||
.propagate(true) |
|||
.build(), Alarm.class); |
|||
|
|||
List<EntityAlarm> entityAlarms = alarmDao.findEntityAlarmRecords(tenantId, alarm.getId()); |
|||
assertThat(entityAlarms).anyMatch(entityAlarm -> entityAlarm.getEntityId().equals(deviceId) && entityAlarm.getAlarmType().equals(alarm.getType())); |
|||
assertThat(entityAlarms).anyMatch(entityAlarm -> entityAlarm.getEntityId().equals(propagatedEntityId) && entityAlarm.getAlarmType().equals(alarm.getType())); |
|||
assertThat(alarmService.findAlarmIdsByOriginatorId(tenantId, deviceId, 0, null, 10)).isNotEmpty(); |
|||
} |
|||
|
|||
private void createAlarm(DeviceId deviceId) { |
|||
Alarm alarm = doPost("/api/alarm", Alarm.builder() |
|||
.tenantId(tenantId) |
|||
.originator(deviceId) |
|||
.severity(AlarmSeverity.CRITICAL) |
|||
.type("test alarm for " + deviceId + " " + RandomStringUtils.randomAlphabetic(10)) |
|||
.build(), Alarm.class); |
|||
assertThat(alarmService.findAlarmIdsByOriginatorId(tenantId, deviceId, 0, null, 10)).isNotEmpty(); |
|||
} |
|||
|
|||
private TsKvEntry getLatestTelemetry(EntityId entityId) throws Exception { |
|||
return timeseriesService.findLatest(tenantId, entityId, HousekeeperServiceTest.TELEMETRY_KEY).get().orElse(null); |
|||
} |
|||
|
|||
private List<TsKvEntry> getTimeseriesHistory(EntityId entityId) throws Exception { |
|||
return timeseriesService.findAll(tenantId, entityId, List.of(new BaseReadTsKvQuery(HousekeeperServiceTest.TELEMETRY_KEY, 0, System.currentTimeMillis(), 10, "DESC"))).get(); |
|||
} |
|||
|
|||
private List<EventInfo> getEvents(EntityId entityId) { |
|||
return eventService.findEvents(tenantId, entityId, EventType.LC_EVENT, new TimePageLink(100)).getData() |
|||
.stream().filter(event -> Optional.ofNullable(event.getBody()).map(body -> body.get("event")) |
|||
.map(JsonNode::asText).orElse("").equals("test")) |
|||
.collect(Collectors.toList()); |
|||
} |
|||
|
|||
private Asset createAsset() { |
|||
Asset asset = new Asset(); |
|||
asset.setName("test"); |
|||
asset.setType("test"); |
|||
return doPost("/api/asset", asset, Asset.class); |
|||
} |
|||
|
|||
private RuleChainMetaData createRuleChain() { |
|||
RuleChain ruleChain = new RuleChain(); |
|||
ruleChain.setTenantId(tenantId); |
|||
ruleChain.setName("Test"); |
|||
ruleChain.setType(RuleChainType.CORE); |
|||
ruleChain.setDebugMode(true); |
|||
ruleChain.setConfiguration(JacksonUtil.newObjectNode().set("a", new TextNode("b"))); |
|||
ruleChain = ruleChainService.saveRuleChain(ruleChain); |
|||
RuleChainId ruleChainId = ruleChain.getId(); |
|||
|
|||
RuleChainMetaData metaData = new RuleChainMetaData(); |
|||
metaData.setRuleChainId(ruleChainId); |
|||
|
|||
RuleNode ruleNode1 = new RuleNode(); |
|||
ruleNode1.setName("Simple Rule Node 1"); |
|||
ruleNode1.setType(org.thingsboard.rule.engine.metadata.TbGetAttributesNode.class.getName()); |
|||
ruleNode1.setConfigurationVersion(TbGetAttributesNode.class.getAnnotation(org.thingsboard.rule.engine.api.RuleNode.class).version()); |
|||
ruleNode1.setDebugMode(true); |
|||
TbGetAttributesNodeConfiguration configuration1 = new TbGetAttributesNodeConfiguration(); |
|||
configuration1.setServerAttributeNames(Collections.singletonList("serverAttributeKey1")); |
|||
ruleNode1.setConfiguration(JacksonUtil.valueToTree(configuration1)); |
|||
|
|||
RuleNode ruleNode2 = new RuleNode(); |
|||
ruleNode2.setName("Simple Rule Node 2"); |
|||
ruleNode2.setType(org.thingsboard.rule.engine.metadata.TbGetAttributesNode.class.getName()); |
|||
ruleNode2.setConfigurationVersion(TbGetAttributesNode.class.getAnnotation(org.thingsboard.rule.engine.api.RuleNode.class).version()); |
|||
ruleNode2.setDebugMode(true); |
|||
TbGetAttributesNodeConfiguration configuration2 = new TbGetAttributesNodeConfiguration(); |
|||
configuration2.setServerAttributeNames(Collections.singletonList("serverAttributeKey2")); |
|||
ruleNode2.setConfiguration(JacksonUtil.valueToTree(configuration2)); |
|||
|
|||
metaData.setNodes(Arrays.asList(ruleNode1, ruleNode2)); |
|||
metaData.setFirstNodeIndex(0); |
|||
metaData.addConnectionInfo(0, 1, TbNodeConnectionType.SUCCESS); |
|||
ruleChainService.saveRuleChainMetaData(tenantId, metaData, Function.identity()); |
|||
return ruleChainService.loadRuleChainMetaData(tenantId, ruleChainId); |
|||
} |
|||
|
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue