From 0d3c3d6b25c24235bbfcd106ce24d354a82ebca7 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Wed, 13 Mar 2024 14:09:05 +0200 Subject: [PATCH] Housekeeper: tests and refactoring --- .../DefaultHousekeeperService.java | 22 ++++-- .../HousekeeperReprocessingService.java | 34 +++++++--- .../src/main/resources/thingsboard.yml | 9 ++- .../housekeeper/HousekeeperServiceTest.java | 68 +++++++++++++++++-- .../resources/application-test.properties | 1 + .../server/queue/TbQueueConsumer.java | 2 + .../AbstractTbQueueConsumerTemplate.java | 5 ++ .../queue/memory/InMemoryTbQueueConsumer.java | 5 ++ .../dao/housekeeper/CleanUpService.java | 2 +- 9 files changed, 121 insertions(+), 27 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java index 63e8bb6a75..589c570144 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java +++ b/application/src/main/java/org/thingsboard/server/service/housekeeper/DefaultHousekeeperService.java @@ -17,6 +17,7 @@ package org.thingsboard.server.service.housekeeper; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.common.util.JacksonUtil; @@ -53,6 +54,7 @@ import java.util.stream.Collectors; @TbCoreComponent @Service @Slf4j +@ConditionalOnProperty(name = "queue.core.housekeeper.enabled", havingValue = "true", matchIfMissing = true) public class DefaultHousekeeperService implements HousekeeperService { private final Map> taskProcessors; @@ -136,9 +138,7 @@ public class DefaultHousekeeperService implements HousekeeperService { } if (log.isDebugEnabled()) { - log.debug("[{}] {} task {}", task.getTenantId(), - msg.getTask().getErrorsCount() == 0 ? "Processing" : "Reprocessing", - msg.getTask().getValue()); + log.debug("[{}] {} task {}", task.getTenantId(), isNew(msg.getTask()) ? "Processing" : "Reprocessing", msg.getTask().getValue()); } try { Future future = executor.submit(() -> { @@ -146,7 +146,9 @@ public class DefaultHousekeeperService implements HousekeeperService { return null; }); future.get(taskProcessingTimeout, TimeUnit.MILLISECONDS); + statsService.reportProcessed(task.getTaskType(), msg); + log.debug("[{}] Successfully {} task {}", task.getTenantId(), isNew(msg.getTask()) ? "processed" : "reprocessed", msg.getTask().getValue()); } catch (InterruptedException e) { throw e; } catch (Throwable e) { @@ -160,6 +162,7 @@ public class DefaultHousekeeperService implements HousekeeperService { task.getTenantId(), task.getEntityId().getEntityType(), task.getEntityId(), task.getTaskType(), msg.getTask().getAttempt(), task, error); reprocessingService.submitForReprocessing(msg, error); + statsService.reportFailure(task.getTaskType(), msg); notificationRuleProcessor.process(TaskProcessingFailureTrigger.builder() .task(task) @@ -185,12 +188,19 @@ public class DefaultHousekeeperService implements HousekeeperService { .build()), null); } + private boolean isNew(HousekeeperTaskProto task) { + return task.getErrorsCount() == 0; + } + @PreDestroy - private void stop() { - log.info("Stopped Housekeeper service"); + private void stop() throws Exception { stopped = true; consumer.unsubscribe(); - consumerExecutor.shutdownNow(); + consumerExecutor.shutdown(); + if (!consumerExecutor.awaitTermination(10, TimeUnit.SECONDS)) { + consumerExecutor.shutdownNow(); + } + log.info("Stopped Housekeeper service"); } } diff --git a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java index 4d361cfba4..c37496ff49 100644 --- a/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/housekeeper/HousekeeperReprocessingService.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.housekeeper; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; @@ -27,6 +28,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType; 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.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; @@ -47,16 +49,18 @@ import java.util.concurrent.TimeUnit; @TbCoreComponent @Service @Slf4j +@ConditionalOnProperty(name = "queue.core.housekeeper.enabled", havingValue = "true", matchIfMissing = true) public class HousekeeperReprocessingService { private final DefaultHousekeeperService housekeeperService; private final PartitionService partitionService; private final TbCoreQueueFactory queueFactory; - private final TbQueueProducerProvider producerProvider; + private final TbQueueProducer> producer; + private final TopicPartitionInfo submitTpi; - @Value("${queue.core.housekeeper.reprocessing-start-delay-sec:15}") // fixme: to 5 minutes + @Value("${queue.core.housekeeper.reprocessing-start-delay-sec:300}") private int startDelay; - @Value("${queue.core.housekeeper.task-reprocessing-delay-sec:30}") // fixme: to 30 minutes or 1 hour + @Value("${queue.core.housekeeper.task-reprocessing-delay-sec:3600}") private int reprocessingDelay; @Value("${queue.core.housekeeper.max-reprocessing-attempts:10}") private int maxReprocessingAttempts; @@ -74,7 +78,8 @@ public class HousekeeperReprocessingService { this.housekeeperService = housekeeperService; this.partitionService = partitionService; this.queueFactory = queueFactory; - this.producerProvider = producerProvider; + this.producer = producerProvider.getHousekeeperReprocessingMsgProducer(); + this.submitTpi = TopicPartitionInfo.builder().topic(producer.getDefaultTopic()).build(); } @PostConstruct @@ -101,7 +106,13 @@ public class HousekeeperReprocessingService { while (!stopped) { try { List> msgs = consumer.poll(pollInterval); - if (msgs.isEmpty() || msgs.stream().anyMatch(msg -> msg.getValue().getTask().getTs() >= startTs)) { // msg batch size should be 1. otherwise some tasks won't be reprocessed immediately + if (msgs.isEmpty() || msgs.stream().anyMatch(msg -> msg.getValue().getTask().getTs() >= startTs)) { + // it's not time yet to process the message + if (!consumer.isCommitSupported()) { + for (TbProtoQueueMsg msg : msgs) { + producer.send(submitTpi, new TbProtoQueueMsg<>(msg.getKey(), msg.getValue()), null); + } + } break; } @@ -114,7 +125,6 @@ public class HousekeeperReprocessingService { } catch (Throwable e) { log.error("Unexpected error during message reprocessing [{}]", msg, e); submitForReprocessing(msg.getValue(), e); - // fixme: msgs are duplicated } } consumer.commit(); @@ -150,16 +160,18 @@ public class HousekeeperReprocessingService { .build(); log.trace("Submitting for reprocessing: {}", msg); - var producer = producerProvider.getHousekeeperReprocessingMsgProducer(); - TopicPartitionInfo tpi = TopicPartitionInfo.builder().topic(producer.getDefaultTopic()).build(); - producer.send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), msg), null); // reprocessing topic has single partition, so we don't care about the msg key + producer.send(submitTpi, new TbProtoQueueMsg<>(UUID.randomUUID(), msg), null); // reprocessing topic has single partition, so we don't care about the msg key } @PreDestroy - private void stop() { + private void stop() throws Exception { stopped = true; scheduler.shutdownNow(); - consumerExecutor.shutdownNow(); + consumerExecutor.shutdown(); + if (!consumerExecutor.awaitTermination(10, TimeUnit.SECONDS)) { + consumerExecutor.shutdownNow(); + } + log.info("Stopped Housekeeper reprocessing service"); } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 4ced0655f0..a156762c71 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1427,8 +1427,10 @@ queue: ota-updates: "${TB_QUEUE_KAFKA_OTA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:10;min.insync.replicas:1}" # Kafka properties for Version Control topic version-control: "${TB_QUEUE_KAFKA_VC_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1;min.insync.replicas:1}" + # Kafka properties for Housekeeper tasks topic housekeeper: "retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:10;min.insync.replicas:1" - housekeeper-reprocessing: "retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1;min.insync.replicas:1" + # Kafka properties for Housekeeper reprocessing topic; retention.ms is set to 90 days + housekeeper-reprocessing: "retention.ms:7776000000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1;min.insync.replicas:1" consumer-stats: # Prints lag between consumer group offset and last messages offset in Kafka topics enabled: "${TB_QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}" @@ -1585,12 +1587,13 @@ queue: # Statistics printing interval for Core microservices print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:60000}" housekeeper: + enabled: "${TB_HOUSEKEEPER_ENABLED:true}" topic: "${TB_HOUSEKEEPER_TOPIC:tb_housekeeper}" reprocessing-topic: "${TB_HOUSEKEEPER_REPROCESSING_TOPIC:tb_housekeeper.reprocessing}" poll-interval-ms: "${TB_HOUSEKEEPER_POLL_INTERVAL_MS:500}" task-processing-timeout-ms: "${TB_HOUSEKEEPER_TASK_PROCESSING_TIMEOUT_MS:120000}" - reprocessing-start-delay-sec: "${TB_HOUSEKEEPER_REPROCESSING_START_DELAY_SEC:15}" # fixme: to 5 minutes - task-reprocessing-delay-sec: "${TB_HOUSEKEEPER_TASK_REPROCESSING_DELAY_SEC:30}" # fixme: to 30 minutes or 1 hour + reprocessing-start-delay-sec: "${TB_HOUSEKEEPER_REPROCESSING_START_DELAY_SEC:300}" + task-reprocessing-delay-sec: "${TB_HOUSEKEEPER_TASK_REPROCESSING_DELAY_SEC:3600}" max-reprocessing-attempts: "${TB_HOUSEKEEPER_MAX_REPROCESSING_ATTEMPTS:30}" vc: diff --git a/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java b/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java index b132cf34fd..a0564d3bcb 100644 --- a/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/housekeeper/HousekeeperServiceTest.java @@ -20,6 +20,7 @@ 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.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.mock.mockito.SpyBean; import org.springframework.test.context.TestPropertySource; @@ -33,6 +34,7 @@ import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.alarm.AlarmSeverity; 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.EntityId; import org.thingsboard.server.common.data.id.RuleChainId; @@ -56,33 +58,46 @@ import org.thingsboard.server.controller.AbstractControllerTest; import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.event.EventService; -import org.thingsboard.server.dao.housekeeper.HousekeeperService; 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.gen.transport.TransportProtos.HousekeeperTaskProto; +import org.thingsboard.server.gen.transport.TransportProtos.ToHousekeeperServiceMsg; +import org.thingsboard.server.service.housekeeper.processor.TelemetryDeletionTaskProcessor; 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.verify; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @DaoSqlTest @TestPropertySource(properties = { - "transport.http.enabled=true" + "queue.core.housekeeper.enabled=true", + "transport.http.enabled=true", + "queue.core.housekeeper.reprocessing-start-delay-sec=1", + "queue.core.housekeeper.task-reprocessing-delay-sec=2", + "queue.core.housekeeper.poll-interval-ms=1000" }) public class HousekeeperServiceTest extends AbstractControllerTest { @SpyBean - private HousekeeperService housekeeperService; + private DefaultHousekeeperService housekeeperService; + @SpyBean + private HousekeeperReprocessingService housekeeperReprocessingService; @Autowired private EventService eventService; @Autowired @@ -93,6 +108,8 @@ public class HousekeeperServiceTest extends AbstractControllerTest { private RuleChainService ruleChainService; @Autowired private AlarmService alarmService; + @SpyBean + private TelemetryDeletionTaskProcessor telemetryDeletionTaskProcessor; private TenantId tenantId; @@ -112,7 +129,7 @@ public class HousekeeperServiceTest extends AbstractControllerTest { @Test public void whenDeviceIsDeleted_thenCleanUpRelatedData() throws Exception { - Device device = createDevice("test", "test"); + Device device = createDevice("wekfwepf", "wekfwepf"); createRelatedData(device.getId()); doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); @@ -143,7 +160,7 @@ public class HousekeeperServiceTest extends AbstractControllerTest { @Test public void whenUserIsDeleted_thenCleanUpRelatedData() throws Exception { - Device device = createDevice("test", "test"); + Device device = createDevice("vneoruvhwe", "vneoruvhwe"); UserId userId = customerUserId; createRelatedData(userId); Alarm alarm = Alarm.builder() @@ -172,7 +189,7 @@ public class HousekeeperServiceTest extends AbstractControllerTest { createRelatedData(tenantId); - Device device = createDevice("test", "test"); + Device device = createDevice("oi324rujoi", "oi324rujoi"); createRelatedData(device.getId()); RuleChainMetaData ruleChainMetaData = createRuleChain(); @@ -199,6 +216,45 @@ public class HousekeeperServiceTest extends AbstractControllerTest { }); } + @Test + public void whenTaskProcessingFails_thenReprocessUntilSuccessful() throws Exception { + TimeoutException error = new TimeoutException("Test timeout"); + doThrow(error).when(telemetryDeletionTaskProcessor).process(any()); + + Device device = createDevice("vep9ruv32", "vep9ruv32"); + createRelatedData(device.getId()); + + doDelete("/api/device/" + device.getId()).andExpect(status().isOk()); + + int attempts = 3; + await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { + verify(housekeeperService).processTask(argThat(verifyTaskSubmission(device.getId(), HousekeeperTaskType.DELETE_TELEMETRY, + task -> task.getErrorsCount() == 0))); + + for (int i = 1; i <= attempts; i++) { + int attempt = i; + verify(housekeeperReprocessingService).submitForReprocessing(argThat(verifyTaskSubmission(device.getId(), HousekeeperTaskType.DELETE_TELEMETRY, + task -> task.getErrorsCount() > 0 && task.getAttempt() == attempt)), argThat(e -> e.getMessage().equals(error.getMessage()))); + verify(housekeeperService).processTask(argThat(verifyTaskSubmission(device.getId(), HousekeeperTaskType.DELETE_TELEMETRY, + task -> task.getErrorsCount() > 0 && task.getAttempt() == attempt))); + } + }); + + assertThat(getTimeseriesHistory(device.getId())).isNotEmpty(); + doCallRealMethod().when(telemetryDeletionTaskProcessor).process(any()); // fixing the code + await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { + assertThat(getTimeseriesHistory(device.getId())).isEmpty(); + }); + } + + private ArgumentMatcher verifyTaskSubmission(EntityId entityId, HousekeeperTaskType taskType, + Predicate additionalCheck) { + return msg -> { + HousekeeperTask task = JacksonUtil.fromString(msg.getTask().getValue(), HousekeeperTask.class); + return task.getEntityId().equals(entityId) && task.getTaskType() == taskType && additionalCheck.test(msg.getTask()); + }; + } + private void createRelatedData(EntityId entityId) throws Exception { createTelemetry(entityId); for (String scope : List.of(DataConstants.SERVER_SCOPE, DataConstants.SHARED_SCOPE, DataConstants.CLIENT_SCOPE)) { diff --git a/application/src/test/resources/application-test.properties b/application/src/test/resources/application-test.properties index 92959e4552..5e4900fdc1 100644 --- a/application/src/test/resources/application-test.properties +++ b/application/src/test/resources/application-test.properties @@ -54,3 +54,4 @@ sql.edge_events.partition_size=168 sql.ttl.edge_events.edge_event_ttl=2592000 server.log_controller_error_stack_trace=false +queue.core.housekeeper.enabled=false diff --git a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueConsumer.java b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueConsumer.java index 51456ab0e2..3747e6d026 100644 --- a/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueConsumer.java +++ b/common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueConsumer.java @@ -36,6 +36,8 @@ public interface TbQueueConsumer { void commit(); + boolean isCommitSupported(); + boolean isStopped(); List getFullTopicNames(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java index 531e30b343..9105c829bc 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java @@ -201,6 +201,11 @@ public abstract class AbstractTbQueueConsumerTemplate i return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList()); } + @Override + public boolean isCommitSupported() { + return true; + } + protected boolean isLongPollingSupported() { return false; } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java index de3833d124..269924087b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/memory/InMemoryTbQueueConsumer.java @@ -114,4 +114,9 @@ public class InMemoryTbQueueConsumer implements TbQueueCon return partitions.stream().map(TopicPartitionInfo::getFullTopicName).collect(Collectors.toList()); } + @Override + public boolean isCommitSupported() { + return false; + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/housekeeper/CleanUpService.java b/dao/src/main/java/org/thingsboard/server/dao/housekeeper/CleanUpService.java index 13a1e1cb09..015917e592 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/housekeeper/CleanUpService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/housekeeper/CleanUpService.java @@ -58,7 +58,7 @@ public class CleanUpService { } public void cleanUpRelatedData(TenantId tenantId, EntityId entityId) { - log.trace("[{}][{}][{}] Cleaning up related data", tenantId, entityId.getEntityType(), entityId.getId()); + log.debug("[{}][{}][{}] Cleaning up related data", tenantId, entityId.getEntityType(), entityId.getId()); // todo: skipped entities list relationService.deleteEntityRelations(tenantId, entityId); housekeeperService.ifPresent(housekeeperService -> {