@ -17,6 +17,8 @@ package org.thingsboard.server.service.apiusage;
import com.google.common.util.concurrent.FutureCallback ;
import com.google.common.util.concurrent.ListenableFuture ;
import jakarta.annotation.PostConstruct ;
import jakarta.annotation.PreDestroy ;
import lombok.RequiredArgsConstructor ;
import lombok.extern.slf4j.Slf4j ;
import org.checkerframework.checker.nullness.qual.Nullable ;
@ -67,8 +69,6 @@ import org.thingsboard.server.service.mail.MailExecutorService;
import org.thingsboard.server.service.partition.AbstractPartitionBasedService ;
import org.thingsboard.server.service.telemetry.InternalTelemetryService ;
import jakarta.annotation.PostConstruct ;
import jakarta.annotation.PreDestroy ;
import java.util.ArrayList ;
import java.util.Arrays ;
import java.util.Collections ;
@ -184,6 +184,9 @@ public class DefaultTbApiUsageStateService extends AbstractPartitionBasedService
if ( newHourTs ! = hourTs ) {
usageState . setHour ( newHourTs ) ;
}
if ( log . isTraceEnabled ( ) ) {
log . trace ( "[{}][{}] Processing usage stats from {} (currentCycleTs={}, currentHourTs={}): {}" , tenantId , ownerId , serviceId , ts , newHourTs , values ) ;
}
updatedEntries = new ArrayList < > ( ApiUsageRecordKey . values ( ) . length ) ;
Set < ApiFeature > apiFeatures = new HashSet < > ( ) ;
for ( UsageStatsKVProto statsItem : values ) {
@ -210,6 +213,7 @@ public class DefaultTbApiUsageStateService extends AbstractPartitionBasedService
} finally {
updateLock . unlock ( ) ;
}
log . trace ( "[{}][{}] Saving new stats: {}" , tenantId , ownerId , updatedEntries ) ;
tsWsService . saveAndNotifyInternal ( tenantId , usageState . getApiUsageState ( ) . getId ( ) , updatedEntries , VOID_CALLBACK ) ;
if ( ! result . isEmpty ( ) ) {
persistAndNotify ( usageState , result ) ;
@ -410,6 +414,9 @@ public class DefaultTbApiUsageStateService extends AbstractPartitionBasedService
myUsageStates . values ( ) . forEach ( state - > {
if ( ( state . getNextCycleTs ( ) < now ) & & ( now - state . getNextCycleTs ( ) < TimeUnit . HOURS . toMillis ( 1 ) ) ) {
state . setCycles ( state . getNextCycleTs ( ) , SchedulerUtils . getStartOfNextNextMonth ( ) ) ;
if ( log . isTraceEnabled ( ) ) {
log . trace ( "[{}][{}] Updating state cycles (currentCycleTs={},nextCycleTs={})" , state . getTenantId ( ) , state . getEntityId ( ) , state . getCurrentCycleTs ( ) , state . getNextCycleTs ( ) ) ;
}
saveNewCounts ( state , Arrays . asList ( ApiUsageRecordKey . values ( ) ) ) ;
if ( state . getEntityType ( ) = = EntityType . TENANT & & ! state . getEntityId ( ) . equals ( TenantId . SYS_TENANT_ID ) ) {
TenantId tenantId = state . getTenantId ( ) ;
@ -417,6 +424,8 @@ public class DefaultTbApiUsageStateService extends AbstractPartitionBasedService
}
}
} ) ;
} catch ( Throwable e ) {
log . error ( "Failed to check start of next cycle" , e ) ;
} finally {
updateLock . unlock ( ) ;
}
@ -483,7 +492,7 @@ public class DefaultTbApiUsageStateService extends AbstractPartitionBasedService
}
}
state . setGaugeReportInterval ( gaugeReportInterval ) ;
log . debug ( "[{}] Initialized state: {}" , ownerId , storedS tate ) ;
log . debug ( "[{}][{}] Initialized state: {}" , tenantId , ownerId , state ) ;
TopicPartitionInfo tpi = partitionService . resolve ( ServiceType . TB_CORE , tenantId , ownerId ) ;
if ( tpi . isMyPartition ( ) ) {
addEntityState ( tpi , state ) ;