Vladyslav_Prykhodko
6 years ago
72 changed files with 1034 additions and 179 deletions
@ -0,0 +1,26 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.actors; |
||||
|
|
||||
|
public class TbRuleNodeUpdateException extends RuntimeException { |
||||
|
|
||||
|
private static final long serialVersionUID = 8209771144711980882L; |
||||
|
|
||||
|
public TbRuleNodeUpdateException(String message, Throwable cause) { |
||||
|
super(message, cause); |
||||
|
} |
||||
|
} |
||||
|
|
||||
@ -0,0 +1,40 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.common.msg.plugin; |
||||
|
|
||||
|
import lombok.ToString; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
||||
|
import org.thingsboard.server.common.msg.MsgType; |
||||
|
|
||||
|
import java.util.Optional; |
||||
|
|
||||
|
/** |
||||
|
* @author Andrew Shvayka |
||||
|
*/ |
||||
|
@ToString |
||||
|
public class RuleNodeUpdatedMsg extends ComponentLifecycleMsg { |
||||
|
|
||||
|
public RuleNodeUpdatedMsg(TenantId tenantId, EntityId entityId) { |
||||
|
super(tenantId, entityId, ComponentLifecycleEvent.UPDATED); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public MsgType getMsgType() { |
||||
|
return MsgType.RULE_NODE_UPDATED_MSG; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,37 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.queue.discovery; |
||||
|
|
||||
|
import lombok.Getter; |
||||
|
import org.springframework.context.ApplicationEvent; |
||||
|
|
||||
|
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
|
||||
|
public class TbApplicationEvent extends ApplicationEvent { |
||||
|
|
||||
|
private static final long serialVersionUID = 3884264064887765146L; |
||||
|
|
||||
|
private static final AtomicInteger sequence = new AtomicInteger(); |
||||
|
|
||||
|
@Getter |
||||
|
private final int sequenceNumber; |
||||
|
|
||||
|
public TbApplicationEvent(Object source) { |
||||
|
super(source); |
||||
|
sequenceNumber = sequence.incrementAndGet(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,52 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.queue.discovery; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.context.ApplicationListener; |
||||
|
|
||||
|
import java.util.concurrent.locks.Lock; |
||||
|
import java.util.concurrent.locks.ReentrantLock; |
||||
|
|
||||
|
@Slf4j |
||||
|
public abstract class TbApplicationEventListener<T extends TbApplicationEvent> implements ApplicationListener<T> { |
||||
|
|
||||
|
private int lastProcessedSequenceNumber = Integer.MIN_VALUE; |
||||
|
private final Lock seqNumberLock = new ReentrantLock(); |
||||
|
|
||||
|
@Override |
||||
|
public void onApplicationEvent(T event) { |
||||
|
boolean validUpdate = false; |
||||
|
seqNumberLock.lock(); |
||||
|
try { |
||||
|
if (event.getSequenceNumber() > lastProcessedSequenceNumber) { |
||||
|
validUpdate = true; |
||||
|
lastProcessedSequenceNumber = event.getSequenceNumber(); |
||||
|
} |
||||
|
} finally { |
||||
|
seqNumberLock.unlock(); |
||||
|
} |
||||
|
if (validUpdate) { |
||||
|
onTbApplicationEvent(event); |
||||
|
} else { |
||||
|
log.info("Application event ignored due to invalid sequence number ({} > {}). Event: {}", lastProcessedSequenceNumber, event.getSequenceNumber(), event); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
protected abstract void onTbApplicationEvent(T event); |
||||
|
|
||||
|
|
||||
|
} |
||||
@ -0,0 +1,37 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.queue.kafka; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Getter; |
||||
|
import lombok.NoArgsConstructor; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
|
||||
|
@Component |
||||
|
@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka") |
||||
|
@Getter |
||||
|
@AllArgsConstructor |
||||
|
@NoArgsConstructor |
||||
|
public class TbKafkaConsumerStatisticConfig { |
||||
|
@Value("${queue.kafka.consumer-stats.enabled:true}") |
||||
|
private Boolean enabled; |
||||
|
@Value("${queue.kafka.consumer-stats.print-interval-ms:60000}") |
||||
|
private Long printIntervalMs; |
||||
|
@Value("${queue.kafka.consumer-stats.kafka-response-timeout-ms:1000}") |
||||
|
private Long kafkaResponseTimeoutMs; |
||||
|
} |
||||
@ -0,0 +1,184 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.server.queue.kafka; |
||||
|
|
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.kafka.clients.admin.AdminClient; |
||||
|
import org.apache.kafka.clients.consumer.Consumer; |
||||
|
import org.apache.kafka.clients.consumer.ConsumerConfig; |
||||
|
import org.apache.kafka.clients.consumer.KafkaConsumer; |
||||
|
import org.apache.kafka.clients.consumer.OffsetAndMetadata; |
||||
|
import org.apache.kafka.common.TopicPartition; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.springframework.util.StringUtils; |
||||
|
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
|
||||
|
import javax.annotation.PostConstruct; |
||||
|
import javax.annotation.PreDestroy; |
||||
|
import java.time.Duration; |
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.Properties; |
||||
|
import java.util.Set; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.Executors; |
||||
|
import java.util.concurrent.ScheduledExecutorService; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
@Slf4j |
||||
|
@Component |
||||
|
@RequiredArgsConstructor |
||||
|
@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka") |
||||
|
public class TbKafkaConsumerStatsService { |
||||
|
private final Set<String> monitoredGroups = ConcurrentHashMap.newKeySet(); |
||||
|
|
||||
|
private final TbKafkaSettings kafkaSettings; |
||||
|
private final TbKafkaConsumerStatisticConfig statsConfig; |
||||
|
private final PartitionService partitionService; |
||||
|
|
||||
|
private AdminClient adminClient; |
||||
|
private Consumer<String, byte[]> consumer; |
||||
|
private ScheduledExecutorService statsPrintScheduler; |
||||
|
|
||||
|
@PostConstruct |
||||
|
public void init() { |
||||
|
if (!statsConfig.getEnabled()) { |
||||
|
return; |
||||
|
} |
||||
|
this.adminClient = AdminClient.create(kafkaSettings.toAdminProps()); |
||||
|
this.statsPrintScheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("kafka-consumer-stats")); |
||||
|
|
||||
|
Properties consumerProps = kafkaSettings.toConsumerProps(); |
||||
|
consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-stats-loader-client"); |
||||
|
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-stats-loader-client-group"); |
||||
|
this.consumer = new KafkaConsumer<>(consumerProps); |
||||
|
|
||||
|
startLogScheduling(); |
||||
|
} |
||||
|
|
||||
|
private void startLogScheduling() { |
||||
|
Duration timeoutDuration = Duration.ofMillis(statsConfig.getKafkaResponseTimeoutMs()); |
||||
|
statsPrintScheduler.scheduleWithFixedDelay(() -> { |
||||
|
if (!isStatsPrintRequired()) { |
||||
|
return; |
||||
|
} |
||||
|
for (String groupId : monitoredGroups) { |
||||
|
try { |
||||
|
Map<TopicPartition, OffsetAndMetadata> groupOffsets = adminClient.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata() |
||||
|
.get(statsConfig.getKafkaResponseTimeoutMs(), TimeUnit.MILLISECONDS); |
||||
|
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(groupOffsets.keySet(), timeoutDuration); |
||||
|
|
||||
|
List<GroupTopicStats> lagTopicsStats = getTopicsStatsWithLag(groupOffsets, endOffsets); |
||||
|
if (!lagTopicsStats.isEmpty()) { |
||||
|
StringBuilder builder = new StringBuilder(); |
||||
|
for (int i = 0; i < lagTopicsStats.size(); i++) { |
||||
|
builder.append(lagTopicsStats.get(i).toString()); |
||||
|
if (i != lagTopicsStats.size() - 1) { |
||||
|
builder.append(", "); |
||||
|
} |
||||
|
} |
||||
|
log.info("[{}] Topic partitions with lag: [{}].", groupId, builder.toString()); |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.warn("[{}] Failed to get consumer group stats. Reason - {}.", groupId, e.getMessage()); |
||||
|
log.trace("Detailed error: ", e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
}, statsConfig.getPrintIntervalMs(), statsConfig.getPrintIntervalMs(), TimeUnit.MILLISECONDS); |
||||
|
} |
||||
|
|
||||
|
private boolean isStatsPrintRequired() { |
||||
|
boolean isMyRuleEnginePartition = partitionService.resolve(ServiceType.TB_RULE_ENGINE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition(); |
||||
|
boolean isMyCorePartition = partitionService.resolve(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition(); |
||||
|
return log.isInfoEnabled() && (isMyRuleEnginePartition || isMyCorePartition); |
||||
|
} |
||||
|
|
||||
|
private List<GroupTopicStats> getTopicsStatsWithLag(Map<TopicPartition, OffsetAndMetadata> groupOffsets, Map<TopicPartition, Long> endOffsets) { |
||||
|
List<GroupTopicStats> consumerGroupStats = new ArrayList<>(); |
||||
|
for (TopicPartition topicPartition : groupOffsets.keySet()) { |
||||
|
long endOffset = endOffsets.get(topicPartition); |
||||
|
long committedOffset = groupOffsets.get(topicPartition).offset(); |
||||
|
long lag = endOffset - committedOffset; |
||||
|
if (lag != 0) { |
||||
|
GroupTopicStats groupTopicStats = GroupTopicStats.builder() |
||||
|
.topic(topicPartition.topic()) |
||||
|
.partition(topicPartition.partition()) |
||||
|
.committedOffset(committedOffset) |
||||
|
.endOffset(endOffset) |
||||
|
.lag(lag) |
||||
|
.build(); |
||||
|
consumerGroupStats.add(groupTopicStats); |
||||
|
} |
||||
|
} |
||||
|
return consumerGroupStats; |
||||
|
} |
||||
|
|
||||
|
public void registerClientGroup(String groupId) { |
||||
|
if (statsConfig.getEnabled() && !StringUtils.isEmpty(groupId)) { |
||||
|
monitoredGroups.add(groupId); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void unregisterClientGroup(String groupId) { |
||||
|
if (statsConfig.getEnabled() && !StringUtils.isEmpty(groupId)) { |
||||
|
monitoredGroups.remove(groupId); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@PreDestroy |
||||
|
public void destroy() { |
||||
|
if (statsPrintScheduler != null) { |
||||
|
statsPrintScheduler.shutdownNow(); |
||||
|
} |
||||
|
if (adminClient != null) { |
||||
|
adminClient.close(); |
||||
|
} |
||||
|
if (consumer != null) { |
||||
|
consumer.close(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
|
||||
|
@Builder |
||||
|
@Data |
||||
|
private static class GroupTopicStats { |
||||
|
private String topic; |
||||
|
private int partition; |
||||
|
private long committedOffset; |
||||
|
private long endOffset; |
||||
|
private long lag; |
||||
|
|
||||
|
@Override |
||||
|
public String toString() { |
||||
|
return "[" + |
||||
|
"topic=[" + topic + ']' + |
||||
|
", partition=[" + partition + "]" + |
||||
|
", committedOffset=[" + committedOffset + "]" + |
||||
|
", endOffset=[" + endOffset + "]" + |
||||
|
", lag=[" + lag + "]" + |
||||
|
"]"; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,53 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
|
||||
|
import com.google.gson.JsonParser; |
||||
|
import org.junit.Assert; |
||||
|
import org.junit.Test; |
||||
|
import org.junit.runner.RunWith; |
||||
|
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
import org.thingsboard.server.common.transport.adaptor.JsonConverter; |
||||
|
|
||||
|
@RunWith(MockitoJUnitRunner.class) |
||||
|
public class JsonConverterTest { |
||||
|
|
||||
|
private static final JsonParser JSON_PARSER = new JsonParser(); |
||||
|
|
||||
|
@Test |
||||
|
public void testParseBigDecimalAsLong() { |
||||
|
var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1E+1}"), 0L); |
||||
|
Assert.assertEquals(10L, result.get(0L).get(0).getLongValue().get().longValue()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testParseBigDecimalAsDouble() { |
||||
|
var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 101E-1}"), 0L); |
||||
|
Assert.assertEquals(10.1, result.get(0L).get(0).getDoubleValue().get(), 0.0); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testParseAsDouble() { |
||||
|
var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1.1}"), 0L); |
||||
|
Assert.assertEquals(1.1, result.get(0L).get(0).getDoubleValue().get(), 0.0); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testParseAsLong() { |
||||
|
var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 11}"), 0L); |
||||
|
Assert.assertEquals(11L, result.get(0L).get(0).getLongValue().get().longValue()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue