From 3818d1cb68d35129e535d96f2ddb098c59812bdb Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 5 Feb 2025 12:34:10 +0200 Subject: [PATCH] debug events(wip) --- .../server/actors/ActorSystemContext.java | 95 ++++++++++++------- .../cf/ctx/state/RocksDBStateService.java | 3 +- .../permission/TenantAdminPermissions.java | 1 + .../src/main/resources/thingsboard.yml | 6 ++ .../dao/cf/BaseCalculatedFieldService.java | 2 +- .../CalculatedFieldDebugEventRepository.java | 2 +- .../main/resources/sql/schema-entities.sql | 4 +- 7 files changed, 72 insertions(+), 41 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 49e2c9c46e..652cdabf77 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -137,6 +137,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; @Slf4j @Component @@ -179,6 +180,8 @@ public class ActorSystemContext { private final ConcurrentMap debugPerTenantLimits = new ConcurrentHashMap<>(); + private final ConcurrentMap cfDebugPerTenantLimits = new ConcurrentHashMap<>(); + public ConcurrentMap getDebugPerTenantLimits() { return debugPerTenantLimits; } @@ -441,6 +444,11 @@ public class ActorSystemContext { @Getter private TbCoreToTransportService tbCoreToTransportService; + @Lazy + @Autowired(required = false) + @Getter + private ApiLimitService apiLimitService; + /** * The following Service will be null if we operate in tb-core mode */ @@ -517,11 +525,6 @@ public class ActorSystemContext { @Getter private CalculatedFieldExecutionService calculatedFieldExecutionService; - @Lazy - @Autowired(required = false) - @Getter - private ApiLimitService apiLimitService; - @Value("${actors.session.max_concurrent_sessions_per_device:1}") @Getter private long maxConcurrentSessionsPerDevice; @@ -625,6 +628,14 @@ public class ActorSystemContext { @Getter private String deviceStateNodeRateLimitConfig; + @Value("${actors.calculated_fields.debug_mode_rate_limits_per_tenant.enabled:true}") + @Getter + private boolean cfDebugPerTenantEnabled; + + @Value("${actors.calculated_fields.debug_mode_rate_limits_per_tenant.configuration:50000:3600}") + @Getter + private String cfDebugPerTenantLimitsConfiguration; + @Getter @Setter private TbActorSystem actorSystem; @@ -753,37 +764,6 @@ public class ActorSystemContext { } } - public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, Map arguments, UUID tbMsgId, TbMsgType tbMsgType, String result, Throwable error) { - try { - CalculatedFieldDebugEvent.CalculatedFieldDebugEventBuilder eventBuilder = CalculatedFieldDebugEvent.builder() - .tenantId(tenantId) - .entityId(entityId.getId()) - .serviceId(getServiceId()) - .calculatedFieldId(calculatedFieldId) - .eventEntity(entityId); - if (tbMsgId != null) { - eventBuilder.msgId(tbMsgId); - } - if (tbMsgType != null) { - eventBuilder.msgType(tbMsgType.name()); - } - if (arguments != null) { - eventBuilder.arguments(JacksonUtil.toString(arguments)); - } - if (result != null) { - eventBuilder.result(result); - } - if (error != null) { - eventBuilder.error(toString(error)); - } - - ListenableFuture future = eventService.saveAsync(eventBuilder.build()); - Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor()); - } catch (IllegalArgumentException ex) { - log.warn("Failed to persist calculated field debug message", ex); - } - } - private boolean checkLimits(TenantId tenantId, TbMsg tbMsg, Throwable error) { if (debugPerTenantEnabled) { DebugTbRateLimits debugTbRateLimits = debugPerTenantLimits.computeIfAbsent(tenantId, id -> @@ -817,6 +797,49 @@ public class ActorSystemContext { Futures.addCallback(future, RULE_CHAIN_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor()); } + public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, Map arguments, UUID tbMsgId, TbMsgType tbMsgType, String result, Throwable error) { + if (cfDebugPerTenantEnabled) { + TbRateLimits rateLimits = cfDebugPerTenantLimits.computeIfAbsent(tenantId, id -> new TbRateLimits(cfDebugPerTenantLimitsConfiguration)); + + if (rateLimits.tryConsume()) { + try { + CalculatedFieldDebugEvent.CalculatedFieldDebugEventBuilder eventBuilder = CalculatedFieldDebugEvent.builder() + .tenantId(tenantId) + .entityId(calculatedFieldId.getId()) + .serviceId(getServiceId()) + .calculatedFieldId(calculatedFieldId) + .eventEntity(entityId); + if (tbMsgId != null) { + eventBuilder.msgId(tbMsgId); + } + if (tbMsgType != null) { + eventBuilder.msgType(tbMsgType.name()); + } + if (arguments != null) { + eventBuilder.arguments(JacksonUtil.toString( + arguments.entrySet().stream() + .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue())) + )); + } + if (result != null) { + eventBuilder.result(result); + } + if (error != null) { + eventBuilder.error(toString(error)); + } + + ListenableFuture future = eventService.saveAsync(eventBuilder.build()); + Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor()); + } catch (IllegalArgumentException ex) { + log.warn("Failed to persist calculated field debug message", ex); + } + if (log.isTraceEnabled()) { + log.trace("[{}] Tenant level debug mode rate limit detected: {}", tenantId, calculatedFieldId); + } + } + } + } + public static Exception toException(Throwable error) { return Exception.class.isInstance(error) ? (Exception) error : new Exception(error); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java index 371751cc6f..8a6a5c9cb7 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java @@ -61,7 +61,8 @@ public class RocksDBStateService implements CalculatedFieldStateService { @Override public void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { CalculatedFieldStateProto stateProto = toProto(stateId, state); - if (stateProto.getSerializedSize() <= ctx.getMaxStateSizeInKBytes()) { + long maxStateSizeInKBytes = ctx.getMaxStateSizeInKBytes(); + if (maxStateSizeInKBytes <= 0 || stateProto.getSerializedSize() <= ctx.getMaxStateSizeInKBytes()) { rocksDBService.put(toProto(stateId), toProto(stateId, state)); } callback.onSuccess(); diff --git a/application/src/main/java/org/thingsboard/server/service/security/permission/TenantAdminPermissions.java b/application/src/main/java/org/thingsboard/server/service/security/permission/TenantAdminPermissions.java index 5b98a56f24..b0e35e57e4 100644 --- a/application/src/main/java/org/thingsboard/server/service/security/permission/TenantAdminPermissions.java +++ b/application/src/main/java/org/thingsboard/server/service/security/permission/TenantAdminPermissions.java @@ -55,6 +55,7 @@ public class TenantAdminPermissions extends AbstractPermissions { put(Resource.OAUTH2_CONFIGURATION_TEMPLATE, new PermissionChecker.GenericPermissionChecker(Operation.READ)); put(Resource.MOBILE_APP, tenantEntityPermissionChecker); put(Resource.MOBILE_APP_BUNDLE, tenantEntityPermissionChecker); + put(Resource.CALCULATED_FIELD, tenantEntityPermissionChecker); } public static final PermissionChecker tenantEntityPermissionChecker = new PermissionChecker() { diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 30a3e51d8c..197f05a5d5 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -510,6 +510,12 @@ actors: js_print_interval_ms: "${ACTORS_JS_STATISTICS_PRINT_INTERVAL_MS:10000}" # Actors statistic persistence frequency in milliseconds persist_frequency: "${ACTORS_STATISTICS_PERSIST_FREQUENCY:3600000}" + calculated_fields: + debug_mode_rate_limits_per_tenant: + # Enable/Disable the rate limit of persisted debug events for all calculated fields per tenant + enabled: "${ACTORS_CF_DEBUG_MODE_RATE_LIMITS_PER_TENANT_ENABLED:true}" + # The value of DEBUG mode rate limit. By default, no more than 50 thousand events per hour + configuration: "${ACTORS_CF_DEBUG_MODE_RATE_LIMITS_PER_TENANT_CONFIGURATION:50000:3600}" debug: settings: diff --git a/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java b/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java index d37a4a53c1..b062a2de5e 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/cf/BaseCalculatedFieldService.java @@ -62,9 +62,9 @@ public class BaseCalculatedFieldService extends AbstractEntityService implements try { TenantId tenantId = calculatedField.getTenantId(); log.trace("Executing save calculated field, [{}]", calculatedField); + updateDebugSettings(tenantId, calculatedField, System.currentTimeMillis()); CalculatedField savedCalculatedField = calculatedFieldDao.save(tenantId, calculatedField); createOrUpdateCalculatedFieldLink(tenantId, savedCalculatedField); - updateDebugSettings(tenantId, calculatedField, System.currentTimeMillis()); eventPublisher.publishEvent(SaveEntityEvent.builder().tenantId(savedCalculatedField.getTenantId()).entityId(savedCalculatedField.getId()) .entity(savedCalculatedField).oldEntity(oldCalculatedField).created(calculatedField.getId() == null).build()); return savedCalculatedField; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/CalculatedFieldDebugEventRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/CalculatedFieldDebugEventRepository.java index c0bd21ca74..a759babcfb 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/CalculatedFieldDebugEventRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/CalculatedFieldDebugEventRepository.java @@ -35,7 +35,7 @@ public interface CalculatedFieldDebugEventRepository extends EventRepository findLatestEvents(@Param("tenantId") UUID tenantId, @Param("entityId") UUID entityId, @Param("limit") int limit); @Override - @Query("SELECT e FROM RuleNodeDebugEventEntity e WHERE " + + @Query("SELECT e FROM CalculatedFieldDebugEventEntity e WHERE " + "e.tenantId = :tenantId " + "AND e.entityId = :entityId " + "AND (:startTime IS NULL OR e.ts >= :startTime) " + diff --git a/dao/src/main/resources/sql/schema-entities.sql b/dao/src/main/resources/sql/schema-entities.sql index a6d1e8800d..4b395a5768 100644 --- a/dao/src/main/resources/sql/schema-entities.sql +++ b/dao/src/main/resources/sql/schema-entities.sql @@ -955,10 +955,10 @@ CREATE TABLE IF NOT EXISTS cf_debug_event ( id uuid NOT NULL, tenant_id uuid NOT NULL , ts bigint NOT NULL, - entity_id uuid NOT NULL, + entity_id uuid NOT NULL, -- calculated field id service_id varchar, cf_id uuid NOT NULL, - e_entity_id uuid, + e_entity_id uuid, -- target entity id e_entity_type varchar, e_msg_id uuid, e_msg_type varchar,