diff --git a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java index 50a273223d..12e8e6017d 100644 --- a/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/apiusage/DefaultTbApiUsageStateService.java @@ -61,6 +61,7 @@ import org.thingsboard.server.service.telemetry.InternalTelemetryService; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -73,6 +74,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import java.util.stream.Collectors; @Slf4j @Service @@ -347,16 +349,11 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService { try { long now = System.currentTimeMillis(); myTenantStates.values().forEach(state -> { - if ((state.getNextCycleTs() > now) && (state.getNextCycleTs() - now < TimeUnit.HOURS.toMillis(1))) { + if ((state.getNextCycleTs() < now) && (now - state.getNextCycleTs() < TimeUnit.HOURS.toMillis(1))) { TenantId tenantId = state.getTenantId(); state.setCycles(state.getNextCycleTs(), SchedulerUtils.getStartOfNextNextMonth()); - ToUsageStatsServiceMsg.Builder msg = ToUsageStatsServiceMsg.newBuilder(); - msg.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); - msg.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); - for (ApiUsageRecordKey key : ApiUsageRecordKey.values()) { - msg.addValues(UsageStatsKVProto.newBuilder().setKey(key.name()).setValue(0).build()); - } - process(new TbProtoQueueMsg<>(UUID.randomUUID(), msg.build()), TbCallback.EMPTY); + saveNewCounts(state, Arrays.asList(ApiUsageRecordKey.values())); + updateTenantState(state, tenantProfileCache.get(tenantId)); } }); } finally { @@ -364,6 +361,14 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService { } } + private void saveNewCounts(TenantApiUsageState state, List keys) { + List counts = keys.stream() + .map(key -> new BasicTsKvEntry(state.getCurrentCycleTs(), new LongDataEntry(key.getApiCountKey(), 0L))) + .collect(Collectors.toList()); + + tsWsService.saveAndNotifyInternal(state.getTenantId(), state.getApiUsageState().getId(), counts, VOID_CALLBACK); + } + private TenantApiUsageState getOrFetchState(TenantId tenantId) { TenantApiUsageState tenantState = myTenantStates.get(tenantId); if (tenantState == null) { @@ -377,6 +382,7 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService { } TenantProfile tenantProfile = tenantProfileCache.get(tenantId); tenantState = new TenantApiUsageState(tenantProfile, dbStateEntity); + List newCounts = new ArrayList<>(); try { List dbValues = tsService.findAllLatest(tenantId, dbStateEntity.getId()).get(); for (ApiUsageRecordKey key : ApiUsageRecordKey.values()) { @@ -385,7 +391,13 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService { for (TsKvEntry tsKvEntry : dbValues) { if (tsKvEntry.getKey().equals(key.getApiCountKey())) { cycleEntryFound = true; - tenantState.put(key, tsKvEntry.getTs() == tenantState.getCurrentCycleTs() ? tsKvEntry.getLongValue().get() : 0L); + + boolean oldCount = tsKvEntry.getTs() == tenantState.getCurrentCycleTs(); + tenantState.put(key, oldCount ? tsKvEntry.getLongValue().get() : 0L); + + if (!oldCount) { + newCounts.add(key); + } } else if (tsKvEntry.getKey().equals(key.getApiCountKey() + HOURLY)) { hourlyEntryFound = true; tenantState.putHourly(key, tsKvEntry.getTs() == tenantState.getCurrentHourTs() ? tsKvEntry.getLongValue().get() : 0L); @@ -397,6 +409,7 @@ public class DefaultTbApiUsageStateService implements TbApiUsageStateService { } log.debug("[{}] Initialized state: {}", tenantId, dbStateEntity); myTenantStates.put(tenantId, tenantState); + saveNewCounts(tenantState, newCounts); } catch (InterruptedException | ExecutionException e) { log.warn("[{}] Failed to fetch api usage state from db.", tenantId, e); } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/tools/SchedulerUtils.java b/common/message/src/main/java/org/thingsboard/server/common/msg/tools/SchedulerUtils.java index 421066b30b..57cc578f2c 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/tools/SchedulerUtils.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/tools/SchedulerUtils.java @@ -17,22 +17,19 @@ package org.thingsboard.server.common.msg.tools; import java.time.LocalDate; import java.time.LocalDateTime; -import java.time.LocalTime; import java.time.ZoneId; -import java.time.ZoneOffset; import java.time.temporal.ChronoUnit; import java.time.temporal.TemporalAdjuster; import java.time.temporal.TemporalAdjusters; -import java.time.temporal.TemporalUnit; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import static java.time.ZoneOffset.UTC; import static java.time.temporal.ChronoField.DAY_OF_MONTH; import static java.time.temporal.ChronoUnit.MONTHS; public class SchedulerUtils { - private final static ZoneId UTC = ZoneId.of("UTC"); private static final ConcurrentMap tzMap = new ConcurrentHashMap<>(); public static ZoneId getZoneId(String tz) { @@ -44,7 +41,7 @@ public class SchedulerUtils { } public static long getStartOfCurrentHour(ZoneId zoneId) { - return LocalDateTime.now(ZoneOffset.UTC).atZone(zoneId).truncatedTo(ChronoUnit.HOURS).toInstant().toEpochMilli(); + return LocalDateTime.now(UTC).atZone(zoneId).truncatedTo(ChronoUnit.HOURS).toInstant().toEpochMilli(); } public static long getStartOfCurrentMonth() { @@ -52,7 +49,7 @@ public class SchedulerUtils { } public static long getStartOfCurrentMonth(ZoneId zoneId) { - return LocalDate.now().withDayOfMonth(1).atStartOfDay(zoneId).toInstant().toEpochMilli(); + return LocalDate.now(UTC).withDayOfMonth(1).atStartOfDay(zoneId).toInstant().toEpochMilli(); } public static long getStartOfNextMonth() { @@ -60,7 +57,7 @@ public class SchedulerUtils { } public static long getStartOfNextMonth(ZoneId zoneId) { - return LocalDate.now().with(TemporalAdjusters.firstDayOfNextMonth()).atStartOfDay(zoneId).toInstant().toEpochMilli(); + return LocalDate.now(UTC).with(TemporalAdjusters.firstDayOfNextMonth()).atStartOfDay(zoneId).toInstant().toEpochMilli(); } public static long getStartOfNextNextMonth() { @@ -68,7 +65,7 @@ public class SchedulerUtils { } public static long getStartOfNextNextMonth(ZoneId zoneId) { - return LocalDate.now().with(firstDayOfNextNextMonth()).atStartOfDay(zoneId).toInstant().toEpochMilli(); + return LocalDate.now(UTC).with(firstDayOfNextNextMonth()).atStartOfDay(zoneId).toInstant().toEpochMilli(); } public static TemporalAdjuster firstDayOfNextNextMonth() {