Andrii Landiak
2 years ago
committed by
GitHub
GitHub
106 changed files with 2025 additions and 773 deletions
@ -1,222 +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.edge; |
|||
|
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import jakarta.annotation.PostConstruct; |
|||
import jakarta.annotation.PreDestroy; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.context.ApplicationEventPublisher; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.common.util.ThingsBoardExecutors; |
|||
import org.thingsboard.server.common.data.audit.ActionType; |
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|||
import org.thingsboard.server.dao.edge.EdgeService; |
|||
import org.thingsboard.server.dao.eventsourcing.ActionEntityEvent; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.edge.rpc.processor.alarm.AlarmEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.asset.AssetEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.asset.profile.AssetProfileEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.customer.CustomerEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.dashboard.DashboardEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.device.DeviceEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.device.profile.DeviceProfileEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.edge.EdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.entityview.EntityViewEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.notification.NotificationEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.oauth2.OAuth2EdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.ota.OtaPackageEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.queue.QueueEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.relation.RelationEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.resource.ResourceEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.rule.RuleChainEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.tenant.TenantEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.tenant.TenantProfileEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.user.UserEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.widget.WidgetBundleEdgeProcessor; |
|||
import org.thingsboard.server.service.edge.rpc.processor.widget.WidgetTypeEdgeProcessor; |
|||
|
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Service |
|||
@TbCoreComponent |
|||
@Slf4j |
|||
public class DefaultEdgeNotificationService implements EdgeNotificationService { |
|||
|
|||
public static final String EDGE_IS_ROOT_BODY_KEY = "isRoot"; |
|||
|
|||
@Autowired |
|||
private EdgeService edgeService; |
|||
|
|||
@Autowired |
|||
private EdgeProcessor edgeProcessor; |
|||
|
|||
@Autowired |
|||
private AssetEdgeProcessor assetProcessor; |
|||
|
|||
@Autowired |
|||
private AssetProfileEdgeProcessor assetProfileEdgeProcessor; |
|||
|
|||
@Autowired |
|||
private DeviceEdgeProcessor deviceProcessor; |
|||
|
|||
@Autowired |
|||
private DeviceProfileEdgeProcessor deviceProfileEdgeProcessor; |
|||
|
|||
@Autowired |
|||
private EntityViewEdgeProcessor entityViewProcessor; |
|||
|
|||
@Autowired |
|||
private DashboardEdgeProcessor dashboardProcessor; |
|||
|
|||
@Autowired |
|||
private RuleChainEdgeProcessor ruleChainProcessor; |
|||
|
|||
@Autowired |
|||
private UserEdgeProcessor userProcessor; |
|||
|
|||
@Autowired |
|||
private CustomerEdgeProcessor customerProcessor; |
|||
|
|||
@Autowired |
|||
private OtaPackageEdgeProcessor otaPackageProcessor; |
|||
|
|||
@Autowired |
|||
private WidgetBundleEdgeProcessor widgetBundleProcessor; |
|||
|
|||
@Autowired |
|||
private WidgetTypeEdgeProcessor widgetTypeProcessor; |
|||
|
|||
@Autowired |
|||
private QueueEdgeProcessor queueProcessor; |
|||
|
|||
@Autowired |
|||
private TenantEdgeProcessor tenantEdgeProcessor; |
|||
|
|||
@Autowired |
|||
private TenantProfileEdgeProcessor tenantProfileEdgeProcessor; |
|||
|
|||
@Autowired |
|||
private AlarmEdgeProcessor alarmProcessor; |
|||
|
|||
@Autowired |
|||
private RelationEdgeProcessor relationProcessor; |
|||
|
|||
@Autowired |
|||
private ResourceEdgeProcessor resourceEdgeProcessor; |
|||
|
|||
@Autowired |
|||
private NotificationEdgeProcessor notificationEdgeProcessor; |
|||
|
|||
@Autowired |
|||
private OAuth2EdgeProcessor oAuth2EdgeProcessor; |
|||
|
|||
@Autowired |
|||
protected ApplicationEventPublisher eventPublisher; |
|||
|
|||
@Value("${actors.system.edge_dispatcher_pool_size:4}") |
|||
private int edgeDispatcherSize; |
|||
|
|||
private ExecutorService executor; |
|||
|
|||
@PostConstruct |
|||
public void initExecutor() { |
|||
executor = ThingsBoardExecutors.newWorkStealingPool(edgeDispatcherSize, "edge-notifications"); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void shutdownExecutor() { |
|||
if (executor != null) { |
|||
executor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public Edge setEdgeRootRuleChain(TenantId tenantId, Edge edge, RuleChainId ruleChainId) { |
|||
edge.setRootRuleChainId(ruleChainId); |
|||
Edge savedEdge = edgeService.saveEdge(edge); |
|||
ObjectNode isRootBody = JacksonUtil.newObjectNode(); |
|||
isRootBody.put(EDGE_IS_ROOT_BODY_KEY, Boolean.TRUE); |
|||
eventPublisher.publishEvent(ActionEntityEvent.builder().tenantId(tenantId).edgeId(edge.getId()).entityId(ruleChainId) |
|||
.body(JacksonUtil.toString(isRootBody)).actionType(ActionType.UPDATED).build()); |
|||
return savedEdge; |
|||
} |
|||
|
|||
@Override |
|||
public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) { |
|||
TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); |
|||
log.debug("[{}] Pushing notification to edge {}", tenantId, edgeNotificationMsg); |
|||
final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60); |
|||
try { |
|||
executor.submit(() -> { |
|||
try { |
|||
if (deadline < System.nanoTime()) { |
|||
log.warn("[{}] Skipping notification message because deadline reached {}", tenantId, edgeNotificationMsg); |
|||
return; |
|||
} |
|||
EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); |
|||
switch (type) { |
|||
case EDGE -> edgeProcessor.processEdgeNotification(tenantId, edgeNotificationMsg); |
|||
case ASSET -> assetProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case ASSET_PROFILE -> assetProfileEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case DEVICE -> deviceProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case DEVICE_PROFILE -> deviceProfileEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case ENTITY_VIEW -> entityViewProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case DASHBOARD -> dashboardProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case RULE_CHAIN -> ruleChainProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case USER -> userProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case CUSTOMER -> customerProcessor.processCustomerNotification(tenantId, edgeNotificationMsg); |
|||
case OTA_PACKAGE -> otaPackageProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case WIDGETS_BUNDLE -> widgetBundleProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case WIDGET_TYPE -> widgetTypeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case QUEUE -> queueProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case ALARM -> alarmProcessor.processAlarmNotification(tenantId, edgeNotificationMsg); |
|||
case ALARM_COMMENT -> alarmProcessor.processAlarmCommentNotification(tenantId, edgeNotificationMsg); |
|||
case RELATION -> relationProcessor.processRelationNotification(tenantId, edgeNotificationMsg); |
|||
case TENANT -> tenantEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case TENANT_PROFILE -> tenantProfileEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case NOTIFICATION_RULE, NOTIFICATION_TARGET, NOTIFICATION_TEMPLATE -> |
|||
notificationEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case TB_RESOURCE -> resourceEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case DOMAIN, OAUTH2_CLIENT -> oAuth2EdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); |
|||
default -> log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type); |
|||
} |
|||
} catch (Exception e) { |
|||
callBackFailure(tenantId, edgeNotificationMsg, callback, e); |
|||
} |
|||
}); |
|||
callback.onSuccess(); |
|||
} catch (Exception e) { |
|||
callBackFailure(tenantId, edgeNotificationMsg, callback, e); |
|||
} |
|||
} |
|||
|
|||
private void callBackFailure(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback, Throwable throwable) { |
|||
log.error("[{}] Can't push to edge updates, edgeNotificationMsg [{}]", tenantId, edgeNotificationMsg, throwable); |
|||
callback.onFailure(throwable); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,334 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
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.MoreExecutors; |
|||
import lombok.Data; |
|||
import lombok.Getter; |
|||
import lombok.Setter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.checkerframework.checker.nullness.qual.Nullable; |
|||
import org.jetbrains.annotations.NotNull; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.scheduling.annotation.Scheduled; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.edge.Edge; |
|||
import org.thingsboard.server.common.data.edge.EdgeEventType; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
import org.thingsboard.server.common.data.queue.QueueConfig; |
|||
import org.thingsboard.server.common.msg.edge.EdgeSessionMsg; |
|||
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|||
import org.thingsboard.server.common.stats.StatsFactory; |
|||
import org.thingsboard.server.common.util.ProtoUtils; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.EdgeNotificationMsgProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; |
|||
import org.thingsboard.server.queue.TbQueueConsumer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.discovery.QueueKey; |
|||
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; |
|||
import org.thingsboard.server.queue.provider.TbCoreQueueFactory; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.edge.EdgeContextComponent; |
|||
import org.thingsboard.server.service.queue.consumer.MainQueueConsumerManager; |
|||
import org.thingsboard.server.service.queue.processing.AbstractConsumerService; |
|||
import org.thingsboard.server.service.queue.processing.IdMsgPair; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.CountDownLatch; |
|||
import java.util.concurrent.Future; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Slf4j |
|||
@Service |
|||
@TbCoreComponent |
|||
public class DefaultTbEdgeConsumerService extends AbstractConsumerService<ToEdgeNotificationMsg> implements TbEdgeConsumerService { |
|||
|
|||
@Value("${queue.edge.pool-interval:25}") |
|||
private int pollInterval; |
|||
@Value("${queue.edge.pack-processing-timeout:10000}") |
|||
private int packProcessingTimeout; |
|||
@Value("${queue.core.consumer-per-partition:true}") |
|||
private boolean consumerPerPartition; |
|||
@Value("${queue.edge.pack-processing-retries:3}") |
|||
private int packProcessingRetries; |
|||
@Value("${queue.edge.stats.enabled:false}") |
|||
private boolean statsEnabled; |
|||
|
|||
private final TbCoreQueueFactory queueFactory; |
|||
private final EdgeContextComponent edgeCtx; |
|||
private final EdgeConsumerStats stats; |
|||
|
|||
private MainQueueConsumerManager<TbProtoQueueMsg<ToEdgeMsg>, EdgeQueueConfig> mainConsumer; |
|||
|
|||
public DefaultTbEdgeConsumerService(TbCoreQueueFactory tbCoreQueueFactory, ActorSystemContext actorContext, |
|||
StatsFactory statsFactory, EdgeContextComponent edgeCtx) { |
|||
super(actorContext, null, null, null, null, null, |
|||
null, null); |
|||
this.edgeCtx = edgeCtx; |
|||
this.stats = new EdgeConsumerStats(statsFactory); |
|||
this.queueFactory = tbCoreQueueFactory; |
|||
} |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
super.init("tb-edge"); |
|||
|
|||
this.mainConsumer = MainQueueConsumerManager.<TbProtoQueueMsg<ToEdgeMsg>, EdgeQueueConfig>builder() |
|||
.queueKey(new QueueKey(ServiceType.TB_CORE).withQueueName(DataConstants.EDGE_QUEUE_NAME)) |
|||
.config(EdgeQueueConfig.of(consumerPerPartition, (int) pollInterval)) |
|||
.msgPackProcessor(this::processMsgs) |
|||
.consumerCreator((config, partitionId) -> queueFactory.createEdgeMsgConsumer()) |
|||
.consumerExecutor(consumersExecutor) |
|||
.scheduler(scheduler) |
|||
.taskExecutor(mgmtExecutor) |
|||
.build(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void destroy() { |
|||
super.destroy(); |
|||
} |
|||
|
|||
@Override |
|||
protected void startConsumers() { |
|||
super.startConsumers(); |
|||
} |
|||
|
|||
@Override |
|||
protected void onTbApplicationEvent(PartitionChangeEvent event) { |
|||
if (ServiceType.TB_CORE.equals(event.getServiceType())) { |
|||
var partitions = event.getEdgePartitions(); |
|||
log.info("Subscribing to partitions: {}", partitions); |
|||
mainConsumer.update(partitions); |
|||
} |
|||
} |
|||
|
|||
private void processMsgs(List<TbProtoQueueMsg<ToEdgeMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToEdgeMsg>> consumer, EdgeQueueConfig edgeQueueConfig) throws InterruptedException { |
|||
List<IdMsgPair<ToEdgeMsg>> orderedMsgList = msgs.stream().map(msg -> new IdMsgPair<>(UUID.randomUUID(), msg)).toList(); |
|||
ConcurrentMap<UUID, TbProtoQueueMsg<ToEdgeMsg>> pendingMap = orderedMsgList.stream().collect( |
|||
Collectors.toConcurrentMap(IdMsgPair::getUuid, IdMsgPair::getMsg)); |
|||
CountDownLatch processingTimeoutLatch = new CountDownLatch(1); |
|||
TbPackProcessingContext<TbProtoQueueMsg<ToEdgeMsg>> ctx = new TbPackProcessingContext<>( |
|||
processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); |
|||
PendingMsgHolder pendingMsgHolder = new PendingMsgHolder(); |
|||
Future<?> submitFuture = consumersExecutor.submit(() -> { |
|||
orderedMsgList.forEach((element) -> { |
|||
UUID id = element.getUuid(); |
|||
TbProtoQueueMsg<ToEdgeMsg> msg = element.getMsg(); |
|||
TbCallback callback = new TbPackCallback<>(id, ctx); |
|||
try { |
|||
ToEdgeMsg toEdgeMsg = msg.getValue(); |
|||
pendingMsgHolder.setToEdgeMsg(toEdgeMsg); |
|||
if (toEdgeMsg.hasEdgeNotificationMsg()) { |
|||
pushNotificationToEdge(toEdgeMsg.getEdgeNotificationMsg(), 0, packProcessingRetries, callback); |
|||
} |
|||
if (statsEnabled) { |
|||
stats.log(toEdgeMsg); |
|||
} |
|||
} catch (Throwable e) { |
|||
log.warn("[{}] Failed to process message: {}", id, msg, e); |
|||
callback.onFailure(e); |
|||
} |
|||
}); |
|||
}); |
|||
if (!processingTimeoutLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) { |
|||
if (!submitFuture.isDone()) { |
|||
submitFuture.cancel(true); |
|||
ToEdgeMsg lastSubmitMsg = pendingMsgHolder.getToEdgeMsg(); |
|||
log.info("Timeout to process message: {}", lastSubmitMsg); |
|||
} |
|||
ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); |
|||
} |
|||
consumer.commit(); |
|||
} |
|||
|
|||
private static class PendingMsgHolder { |
|||
@Getter |
|||
@Setter |
|||
private volatile ToEdgeMsg toEdgeMsg; |
|||
} |
|||
|
|||
@Override |
|||
protected ServiceType getServiceType() { |
|||
return ServiceType.TB_CORE; |
|||
} |
|||
|
|||
@Override |
|||
protected long getNotificationPollDuration() { |
|||
return pollInterval; |
|||
} |
|||
|
|||
@Override |
|||
protected long getNotificationPackProcessingTimeout() { |
|||
return packProcessingTimeout; |
|||
} |
|||
|
|||
@Override |
|||
protected int getMgmtThreadPoolSize() { |
|||
return Math.max(Runtime.getRuntime().availableProcessors(), 4); |
|||
} |
|||
|
|||
@Override |
|||
protected TbQueueConsumer<TbProtoQueueMsg<ToEdgeNotificationMsg>> createNotificationsConsumer() { |
|||
return queueFactory.createToEdgeNotificationsMsgConsumer(); |
|||
} |
|||
|
|||
@Override |
|||
protected void handleNotification(UUID id, TbProtoQueueMsg<ToEdgeNotificationMsg> msg, TbCallback callback) { |
|||
ToEdgeNotificationMsg toEdgeNotificationMsg = msg.getValue(); |
|||
try { |
|||
if (toEdgeNotificationMsg.hasEdgeHighPriority()) { |
|||
EdgeSessionMsg edgeSessionMsg = ProtoUtils.fromProto(toEdgeNotificationMsg.getEdgeHighPriority()); |
|||
edgeCtx.getEdgeRpcService().onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg); |
|||
callback.onSuccess(); |
|||
} else if (toEdgeNotificationMsg.hasEdgeEventUpdate()) { |
|||
EdgeSessionMsg edgeSessionMsg = ProtoUtils.fromProto(toEdgeNotificationMsg.getEdgeEventUpdate()); |
|||
edgeCtx.getEdgeRpcService().onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg); |
|||
callback.onSuccess(); |
|||
} else if (toEdgeNotificationMsg.hasToEdgeSyncRequest()) { |
|||
EdgeSessionMsg edgeSessionMsg = ProtoUtils.fromProto(toEdgeNotificationMsg.getToEdgeSyncRequest()); |
|||
edgeCtx.getEdgeRpcService().onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg); |
|||
callback.onSuccess(); |
|||
} else if (toEdgeNotificationMsg.hasFromEdgeSyncResponse()) { |
|||
EdgeSessionMsg edgeSessionMsg = ProtoUtils.fromProto(toEdgeNotificationMsg.getFromEdgeSyncResponse()); |
|||
edgeCtx.getEdgeRpcService().onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg); |
|||
callback.onSuccess(); |
|||
} else if (toEdgeNotificationMsg.hasComponentLifecycle()) { |
|||
ComponentLifecycleMsg componentLifecycle = ProtoUtils.fromProto(toEdgeNotificationMsg.getComponentLifecycle()); |
|||
TenantId tenantId = componentLifecycle.getTenantId(); |
|||
EdgeId edgeId = new EdgeId(componentLifecycle.getEntityId().getId()); |
|||
if (ComponentLifecycleEvent.DELETED.equals(componentLifecycle.getEvent())) { |
|||
edgeCtx.getEdgeRpcService().deleteEdge(tenantId, edgeId); |
|||
} else if (ComponentLifecycleEvent.UPDATED.equals(componentLifecycle.getEvent())) { |
|||
Edge edge = edgeCtx.getEdgeService().findEdgeById(tenantId, edgeId); |
|||
edgeCtx.getEdgeRpcService().updateEdge(tenantId, edge); |
|||
} |
|||
callback.onSuccess(); |
|||
} |
|||
} catch (Exception e) { |
|||
log.error("Error processing edge notification message: {}", e.getMessage(), e); |
|||
callback.onFailure(e); |
|||
} |
|||
|
|||
if (statsEnabled) { |
|||
stats.log(msg.getValue()); |
|||
} |
|||
} |
|||
|
|||
private void pushNotificationToEdge(EdgeNotificationMsgProto edgeNotificationMsg, int retryCount, int retryLimit, TbCallback callback) { |
|||
TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); |
|||
log.debug("[{}] Pushing notification to edge {}", tenantId, edgeNotificationMsg); |
|||
try { |
|||
EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); |
|||
ListenableFuture<Void> future; |
|||
switch (type) { |
|||
case EDGE -> future = edgeCtx.getEdgeProcessor().processEdgeNotification(tenantId, edgeNotificationMsg); |
|||
case ASSET -> future = edgeCtx.getAssetProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case ASSET_PROFILE -> future = edgeCtx.getAssetProfileProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case DEVICE -> future = edgeCtx.getDeviceProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case DEVICE_PROFILE -> future = edgeCtx.getDeviceProfileProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case ENTITY_VIEW -> future = edgeCtx.getEntityViewProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case DASHBOARD -> future = edgeCtx.getDashboardProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case RULE_CHAIN -> future = edgeCtx.getRuleChainProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case USER -> future = edgeCtx.getUserProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case CUSTOMER -> future = edgeCtx.getCustomerProcessor().processCustomerNotification(tenantId, edgeNotificationMsg); |
|||
case OTA_PACKAGE -> future = edgeCtx.getOtaPackageProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case WIDGETS_BUNDLE -> future = edgeCtx.getWidgetBundleProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case WIDGET_TYPE -> future = edgeCtx.getWidgetTypeProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case QUEUE -> future = edgeCtx.getQueueProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case ALARM -> future = edgeCtx.getAlarmProcessor().processAlarmNotification(tenantId, edgeNotificationMsg); |
|||
case ALARM_COMMENT -> future = edgeCtx.getAlarmProcessor().processAlarmCommentNotification(tenantId, edgeNotificationMsg); |
|||
case RELATION -> future = edgeCtx.getRelationProcessor().processRelationNotification(tenantId, edgeNotificationMsg); |
|||
case TENANT -> future = edgeCtx.getTenantProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case TENANT_PROFILE -> future = edgeCtx.getTenantProfileProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case NOTIFICATION_RULE, NOTIFICATION_TARGET, NOTIFICATION_TEMPLATE -> |
|||
future = edgeCtx.getNotificationEdgeProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case TB_RESOURCE -> future = edgeCtx.getResourceProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
case DOMAIN, OAUTH2_CLIENT -> future = edgeCtx.getOAuth2EdgeProcessor().processEntityNotification(tenantId, edgeNotificationMsg); |
|||
default -> { |
|||
future = Futures.immediateFuture(null); |
|||
log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type); |
|||
} |
|||
} |
|||
Futures.addCallback(future, new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(@Nullable Void unused) { |
|||
callback.onSuccess(); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(@NotNull Throwable throwable) { |
|||
if (retryCount < retryLimit) { |
|||
log.warn("[{}] Retry {} for message due to failure: {}", tenantId, retryCount + 1, throwable.getMessage()); |
|||
pushNotificationToEdge(edgeNotificationMsg, retryCount + 1, retryLimit, callback); |
|||
} else { |
|||
callBackFailure(tenantId, edgeNotificationMsg, callback, throwable); |
|||
} |
|||
} |
|||
}, MoreExecutors.directExecutor()); |
|||
} catch (Exception e) { |
|||
if (retryCount < retryLimit) { |
|||
log.warn("[{}] Retry {} for message due to exception: {}", tenantId, retryCount + 1, e.getMessage()); |
|||
pushNotificationToEdge(edgeNotificationMsg, retryCount + 1, retryLimit, callback); |
|||
} else { |
|||
callBackFailure(tenantId, edgeNotificationMsg, callback, e); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void callBackFailure(TenantId tenantId, EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback, Throwable throwable) { |
|||
log.error("[{}] Can't push to edge updates, edgeNotificationMsg [{}]", tenantId, edgeNotificationMsg, throwable); |
|||
callback.onFailure(throwable); |
|||
} |
|||
|
|||
@Scheduled(fixedDelayString = "${queue.edge.stats.print-interval-ms}") |
|||
public void printStats() { |
|||
if (statsEnabled) { |
|||
stats.printStats(); |
|||
stats.reset(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected void stopConsumers() { |
|||
super.stopConsumers(); |
|||
mainConsumer.stop(); |
|||
mainConsumer.awaitStop(); |
|||
} |
|||
|
|||
@Data(staticConstructor = "of") |
|||
public static class EdgeQueueConfig implements QueueConfig { |
|||
private final boolean consumerPerPartition; |
|||
private final int pollInterval; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,99 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.common.stats.StatsCounter; |
|||
import org.thingsboard.server.common.stats.StatsFactory; |
|||
import org.thingsboard.server.common.stats.StatsType; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeNotificationMsg; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
|
|||
@Slf4j |
|||
public class EdgeConsumerStats { |
|||
|
|||
public static final String TOTAL_MSGS = "totalMsgs"; |
|||
public static final String EDGE_NOTIFICATIONS = "edgeNfs"; |
|||
public static final String TO_CORE_NF_EDGE_EVENT = "coreNfEdgeHPUpd"; |
|||
public static final String TO_CORE_NF_EDGE_EVENT_UPDATE = "coreNfEdgeUpd"; |
|||
public static final String TO_CORE_NF_EDGE_SYNC_REQUEST = "coreNfEdgeSyncReq"; |
|||
public static final String TO_CORE_NF_EDGE_SYNC_RESPONSE = "coreNfEdgeSyncResp"; |
|||
public static final String TO_CORE_NF_EDGE_COMPONENT_LIFECYCLE = "coreNfEdgeCompLfcl"; |
|||
|
|||
private final StatsCounter totalCounter; |
|||
private final StatsCounter edgeNotificationsCounter; |
|||
private final StatsCounter edgeHighPriorityCounter; |
|||
private final StatsCounter edgeEventUpdateCounter; |
|||
private final StatsCounter edgeSyncRequestCounter; |
|||
private final StatsCounter edgeSyncResponseCounter; |
|||
private final StatsCounter edgeComponentLifecycle; |
|||
|
|||
private final List<StatsCounter> counters = new ArrayList<>(7); |
|||
|
|||
public EdgeConsumerStats(StatsFactory statsFactory) { |
|||
String statsKey = StatsType.EDGE.getName(); |
|||
|
|||
this.totalCounter = register(statsFactory.createStatsCounter(statsKey, TOTAL_MSGS)); |
|||
this.edgeNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, EDGE_NOTIFICATIONS)); |
|||
this.edgeHighPriorityCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_EVENT)); |
|||
this.edgeEventUpdateCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_EVENT_UPDATE)); |
|||
this.edgeSyncRequestCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_SYNC_REQUEST)); |
|||
this.edgeSyncResponseCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_SYNC_RESPONSE)); |
|||
this.edgeComponentLifecycle = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_COMPONENT_LIFECYCLE)); |
|||
} |
|||
|
|||
private StatsCounter register(StatsCounter counter) { |
|||
counters.add(counter); |
|||
return counter; |
|||
} |
|||
|
|||
public void log(ToEdgeNotificationMsg msg) { |
|||
totalCounter.increment(); |
|||
if (msg.hasEdgeHighPriority()) { |
|||
edgeHighPriorityCounter.increment(); |
|||
} else if (msg.hasEdgeEventUpdate()) { |
|||
edgeEventUpdateCounter.increment(); |
|||
} else if (msg.hasToEdgeSyncRequest()) { |
|||
edgeSyncRequestCounter.increment(); |
|||
} else if (msg.hasFromEdgeSyncResponse()) { |
|||
edgeSyncResponseCounter.increment(); |
|||
} else if (msg.hasComponentLifecycle()) { |
|||
edgeComponentLifecycle.increment(); |
|||
} |
|||
} |
|||
|
|||
public void log(ToEdgeMsg msg) { |
|||
totalCounter.increment(); |
|||
edgeNotificationsCounter.increment(); |
|||
} |
|||
|
|||
public void printStats() { |
|||
int total = totalCounter.get(); |
|||
if (total > 0) { |
|||
StringBuilder stats = new StringBuilder(); |
|||
counters.forEach(counter -> stats.append(counter.getName()).append(" = [").append(counter.get()).append("] ")); |
|||
log.info("Edge Stats: {}", stats); |
|||
} |
|||
} |
|||
|
|||
public void reset() { |
|||
counters.forEach(StatsCounter::clear); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,23 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
import org.springframework.context.ApplicationListener; |
|||
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; |
|||
|
|||
public interface TbEdgeConsumerService extends ApplicationListener<PartitionChangeEvent> { |
|||
|
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
/** |
|||
* 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.queue.settings; |
|||
|
|||
import lombok.Data; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
@Lazy |
|||
@Data |
|||
@Component |
|||
public class TbQueueEdgeSettings { |
|||
|
|||
@Value("${queue.edge.topic}") |
|||
private String topic; |
|||
|
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue