Browse Source
# Conflicts: # common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.javapull/7377/head
Yuriy Lytvynchuk
4 years ago
205 changed files with 5592 additions and 710 deletions
@ -0,0 +1,74 @@ |
|||||
|
-- |
||||
|
-- Copyright © 2016-2022 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. |
||||
|
-- |
||||
|
|
||||
|
DO |
||||
|
$$ |
||||
|
DECLARE table_partition RECORD; |
||||
|
BEGIN |
||||
|
-- in case of running the upgrade script a second time: |
||||
|
IF NOT (SELECT exists(SELECT FROM pg_tables WHERE tablename = 'old_audit_log')) THEN |
||||
|
ALTER TABLE audit_log RENAME TO old_audit_log; |
||||
|
ALTER INDEX IF EXISTS idx_audit_log_tenant_id_and_created_time RENAME TO idx_old_audit_log_tenant_id_and_created_time; |
||||
|
|
||||
|
FOR table_partition IN SELECT tablename AS name, split_part(tablename, '_', 3) AS partition_ts |
||||
|
FROM pg_tables WHERE tablename LIKE 'audit_log_%' |
||||
|
LOOP |
||||
|
EXECUTE format('ALTER TABLE %s RENAME TO old_audit_log_%s', table_partition.name, table_partition.partition_ts); |
||||
|
END LOOP; |
||||
|
ELSE |
||||
|
RAISE NOTICE 'Table old_audit_log already exists, leaving as is'; |
||||
|
END IF; |
||||
|
END; |
||||
|
$$; |
||||
|
|
||||
|
CREATE TABLE IF NOT EXISTS audit_log ( |
||||
|
id uuid NOT NULL, |
||||
|
created_time bigint NOT NULL, |
||||
|
tenant_id uuid, |
||||
|
customer_id uuid, |
||||
|
entity_id uuid, |
||||
|
entity_type varchar(255), |
||||
|
entity_name varchar(255), |
||||
|
user_id uuid, |
||||
|
user_name varchar(255), |
||||
|
action_type varchar(255), |
||||
|
action_data varchar(1000000), |
||||
|
action_status varchar(255), |
||||
|
action_failure_details varchar(1000000) |
||||
|
) PARTITION BY RANGE (created_time); |
||||
|
CREATE INDEX IF NOT EXISTS idx_audit_log_tenant_id_and_created_time ON audit_log(tenant_id, created_time DESC); |
||||
|
|
||||
|
CREATE OR REPLACE PROCEDURE migrate_audit_logs(IN start_time_ms BIGINT, IN end_time_ms BIGINT, IN partition_size_ms BIGINT) |
||||
|
LANGUAGE plpgsql AS |
||||
|
$$ |
||||
|
DECLARE |
||||
|
p RECORD; |
||||
|
partition_end_ts BIGINT; |
||||
|
BEGIN |
||||
|
FOR p IN SELECT DISTINCT (created_time - created_time % partition_size_ms) AS partition_ts FROM old_audit_log |
||||
|
WHERE created_time >= start_time_ms AND created_time < end_time_ms |
||||
|
LOOP |
||||
|
partition_end_ts = p.partition_ts + partition_size_ms; |
||||
|
RAISE NOTICE '[audit_log] Partition to create : [%-%]', p.partition_ts, partition_end_ts; |
||||
|
EXECUTE format('CREATE TABLE IF NOT EXISTS audit_log_%s PARTITION OF audit_log ' || |
||||
|
'FOR VALUES FROM ( %s ) TO ( %s )', p.partition_ts, p.partition_ts, partition_end_ts); |
||||
|
END LOOP; |
||||
|
|
||||
|
INSERT INTO audit_log |
||||
|
SELECT * FROM old_audit_log |
||||
|
WHERE created_time >= start_time_ms AND created_time < end_time_ms; |
||||
|
END; |
||||
|
$$; |
||||
@ -0,0 +1,61 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.ttl; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
||||
|
import org.springframework.scheduling.annotation.Scheduled; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.server.dao.audit.AuditLogDao; |
||||
|
import org.thingsboard.server.dao.sqlts.insert.sql.SqlPartitioningRepository; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
|
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.thingsboard.server.dao.model.ModelConstants.AUDIT_LOG_COLUMN_FAMILY_NAME; |
||||
|
|
||||
|
@Service |
||||
|
@ConditionalOnExpression("${sql.ttl.audit_logs.enabled:true} && ${sql.ttl.audit_logs.ttl:0} > 0") |
||||
|
@Slf4j |
||||
|
public class AuditLogsCleanUpService extends AbstractCleanUpService { |
||||
|
|
||||
|
private final AuditLogDao auditLogDao; |
||||
|
private final SqlPartitioningRepository partitioningRepository; |
||||
|
|
||||
|
@Value("${sql.ttl.audit_logs.ttl:0}") |
||||
|
private long ttlInSec; |
||||
|
@Value("${sql.audit_logs.partition_size:168}") |
||||
|
private int partitionSizeInHours; |
||||
|
|
||||
|
public AuditLogsCleanUpService(PartitionService partitionService, AuditLogDao auditLogDao, SqlPartitioningRepository partitioningRepository) { |
||||
|
super(partitionService); |
||||
|
this.auditLogDao = auditLogDao; |
||||
|
this.partitioningRepository = partitioningRepository; |
||||
|
} |
||||
|
|
||||
|
@Scheduled(initialDelayString = "#{T(org.apache.commons.lang3.RandomUtils).nextLong(0, ${sql.ttl.audit_logs.checking_interval_ms})}", |
||||
|
fixedDelayString = "${sql.ttl.audit_logs.checking_interval_ms}") |
||||
|
public void cleanUp() { |
||||
|
long auditLogsExpTime = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(ttlInSec); |
||||
|
if (isSystemTenantPartitionMine()) { |
||||
|
auditLogDao.cleanUpAuditLogs(auditLogsExpTime); |
||||
|
} else { |
||||
|
partitioningRepository.cleanupPartitionsCache(AUDIT_LOG_COLUMN_FAMILY_NAME, auditLogsExpTime, TimeUnit.HOURS.toMillis(partitionSizeInHours)); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,96 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.script; |
||||
|
|
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.test.context.TestPropertySource; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.controller.AbstractControllerTest; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.ExecutionException; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
@TestPropertySource(properties = { |
||||
|
"js.max_script_body_size=50", |
||||
|
"js.max_total_args_size=50", |
||||
|
"js.max_result_size=50", |
||||
|
"js.local.max_errors=2" |
||||
|
}) |
||||
|
class LocalJsInvokeServiceTest extends AbstractControllerTest { |
||||
|
|
||||
|
@Autowired |
||||
|
private NashornJsInvokeService jsInvokeService; |
||||
|
|
||||
|
@Value("${js.local.max_errors}") |
||||
|
private int maxJsErrors; |
||||
|
|
||||
|
@Test |
||||
|
void givenTooBigScriptForEval_thenReturnError() { |
||||
|
String hugeScript = "var a = 'qwertyqwertywertyqwabababer'; return {a: a};"; |
||||
|
|
||||
|
assertThatThrownBy(() -> { |
||||
|
evalScript(hugeScript); |
||||
|
}).hasMessageContaining("body exceeds maximum allowed size"); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenTooBigScriptInputArgs_thenReturnErrorAndReportScriptExecutionError() throws Exception { |
||||
|
String script = "return { msg: msg };"; |
||||
|
String hugeMsg = "{\"input\":\"123456781234349\"}"; |
||||
|
UUID scriptId = evalScript(script); |
||||
|
|
||||
|
for (int i = 0; i < maxJsErrors; i++) { |
||||
|
assertThatThrownBy(() -> { |
||||
|
invokeScript(scriptId, hugeMsg); |
||||
|
}).hasMessageContaining("input arguments exceed maximum"); |
||||
|
} |
||||
|
assertThatScriptIsBlocked(scriptId); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void whenScriptInvocationResultIsTooBig_thenReturnErrorAndReportScriptExecutionError() throws Exception { |
||||
|
String script = "var s = new Array(50).join('a'); return { s: s};"; |
||||
|
UUID scriptId = evalScript(script); |
||||
|
|
||||
|
for (int i = 0; i < maxJsErrors; i++) { |
||||
|
assertThatThrownBy(() -> { |
||||
|
invokeScript(scriptId, "{}"); |
||||
|
}).hasMessageContaining("result exceeds maximum allowed size"); |
||||
|
} |
||||
|
assertThatScriptIsBlocked(scriptId); |
||||
|
} |
||||
|
|
||||
|
private void assertThatScriptIsBlocked(UUID scriptId) { |
||||
|
assertThatThrownBy(() -> { |
||||
|
invokeScript(scriptId, "{}"); |
||||
|
}).hasMessageContaining("invocation is blocked due to maximum error"); |
||||
|
} |
||||
|
|
||||
|
private UUID evalScript(String script) throws ExecutionException, InterruptedException { |
||||
|
return jsInvokeService.eval(TenantId.SYS_TENANT_ID, JsScriptType.RULE_NODE_SCRIPT, script).get(); |
||||
|
} |
||||
|
|
||||
|
private String invokeScript(UUID scriptId, String msg) throws ExecutionException, InterruptedException { |
||||
|
return jsInvokeService.invokeFunction(TenantId.SYS_TENANT_ID, null, scriptId, msg, "{}", "POST_TELEMETRY_REQUEST").get(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,219 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.script; |
||||
|
|
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import org.apache.commons.lang3.StringUtils; |
||||
|
import org.junit.jupiter.api.AfterEach; |
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.mockito.ArgumentCaptor; |
||||
|
import org.thingsboard.server.common.data.ApiUsageState; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.gen.js.JsInvokeProtos; |
||||
|
import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsRequest; |
||||
|
import org.thingsboard.server.gen.js.JsInvokeProtos.RemoteJsResponse; |
||||
|
import org.thingsboard.server.queue.TbQueueRequestTemplate; |
||||
|
import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; |
||||
|
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
||||
|
import org.thingsboard.server.queue.usagestats.TbApiUsageClient; |
||||
|
import org.thingsboard.server.service.apiusage.TbApiUsageStateService; |
||||
|
|
||||
|
import java.util.HashSet; |
||||
|
import java.util.List; |
||||
|
import java.util.Set; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.argThat; |
||||
|
import static org.mockito.Mockito.doAnswer; |
||||
|
import static org.mockito.Mockito.doReturn; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
import static org.mockito.Mockito.reset; |
||||
|
import static org.mockito.Mockito.times; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.verifyNoInteractions; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
|
||||
|
class RemoteJsInvokeServiceTest { |
||||
|
|
||||
|
private RemoteJsInvokeService remoteJsInvokeService; |
||||
|
private TbQueueRequestTemplate<TbProtoJsQueueMsg<RemoteJsRequest>, TbProtoQueueMsg<RemoteJsResponse>> jsRequestTemplate; |
||||
|
|
||||
|
|
||||
|
@BeforeEach |
||||
|
public void beforeEach() { |
||||
|
TbApiUsageStateService apiUsageStateService = mock(TbApiUsageStateService.class); |
||||
|
ApiUsageState apiUsageState = mock(ApiUsageState.class); |
||||
|
when(apiUsageState.isJsExecEnabled()).thenReturn(true); |
||||
|
when(apiUsageStateService.getApiUsageState(any())).thenReturn(apiUsageState); |
||||
|
TbApiUsageClient apiUsageClient = mock(TbApiUsageClient.class); |
||||
|
|
||||
|
remoteJsInvokeService = new RemoteJsInvokeService(apiUsageStateService, apiUsageClient); |
||||
|
jsRequestTemplate = mock(TbQueueRequestTemplate.class); |
||||
|
remoteJsInvokeService.requestTemplate = jsRequestTemplate; |
||||
|
} |
||||
|
|
||||
|
@AfterEach |
||||
|
public void afterEach() { |
||||
|
reset(jsRequestTemplate); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void whenInvokingFunction_thenDoNotSendScriptBody() throws Exception { |
||||
|
mockJsEvalResponse(); |
||||
|
String scriptBody = "return { a: 'b'};"; |
||||
|
UUID scriptId = remoteJsInvokeService.eval(TenantId.SYS_TENANT_ID, JsScriptType.RULE_NODE_SCRIPT, scriptBody).get(); |
||||
|
reset(jsRequestTemplate); |
||||
|
|
||||
|
String expectedInvocationResult = "scriptInvocationResult"; |
||||
|
doReturn(Futures.immediateFuture(new TbProtoJsQueueMsg<>(UUID.randomUUID(), RemoteJsResponse.newBuilder() |
||||
|
.setInvokeResponse(JsInvokeProtos.JsInvokeResponse.newBuilder() |
||||
|
.setSuccess(true) |
||||
|
.setResult(expectedInvocationResult) |
||||
|
.build()) |
||||
|
.build()))) |
||||
|
.when(jsRequestTemplate).send(any()); |
||||
|
|
||||
|
ArgumentCaptor<TbProtoJsQueueMsg<RemoteJsRequest>> jsRequestCaptor = ArgumentCaptor.forClass(TbProtoJsQueueMsg.class); |
||||
|
Object invocationResult = remoteJsInvokeService.invokeFunction(TenantId.SYS_TENANT_ID, null, scriptId, "{}").get(); |
||||
|
verify(jsRequestTemplate).send(jsRequestCaptor.capture()); |
||||
|
|
||||
|
JsInvokeProtos.JsInvokeRequest jsInvokeRequestMade = jsRequestCaptor.getValue().getValue().getInvokeRequest(); |
||||
|
assertThat(jsInvokeRequestMade.getScriptBody()).isNullOrEmpty(); |
||||
|
assertThat(jsInvokeRequestMade.getScriptHash()).isEqualTo(getScriptHash(scriptId)); |
||||
|
assertThat(invocationResult).isEqualTo(expectedInvocationResult); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void whenInvokingFunctionAndRemoteJsExecutorRemovedScript_thenHandleNotFoundErrorAndMakeInvokeRequestWithScriptBody() throws Exception { |
||||
|
mockJsEvalResponse(); |
||||
|
String scriptBody = "return { a: 'b'};"; |
||||
|
UUID scriptId = remoteJsInvokeService.eval(TenantId.SYS_TENANT_ID, JsScriptType.RULE_NODE_SCRIPT, scriptBody).get(); |
||||
|
reset(jsRequestTemplate); |
||||
|
|
||||
|
doReturn(Futures.immediateFuture(new TbProtoJsQueueMsg<>(UUID.randomUUID(), RemoteJsResponse.newBuilder() |
||||
|
.setInvokeResponse(JsInvokeProtos.JsInvokeResponse.newBuilder() |
||||
|
.setSuccess(false) |
||||
|
.setErrorCode(JsInvokeProtos.JsInvokeErrorCode.NOT_FOUND_ERROR) |
||||
|
.build()) |
||||
|
.build()))) |
||||
|
.when(jsRequestTemplate).send(argThat(jsQueueMsg -> { |
||||
|
return StringUtils.isEmpty(jsQueueMsg.getValue().getInvokeRequest().getScriptBody()); |
||||
|
})); |
||||
|
|
||||
|
String expectedInvocationResult = "invocationResult"; |
||||
|
doReturn(Futures.immediateFuture(new TbProtoJsQueueMsg<>(UUID.randomUUID(), RemoteJsResponse.newBuilder() |
||||
|
.setInvokeResponse(JsInvokeProtos.JsInvokeResponse.newBuilder() |
||||
|
.setSuccess(true) |
||||
|
.setResult(expectedInvocationResult) |
||||
|
.build()) |
||||
|
.build()))) |
||||
|
.when(jsRequestTemplate).send(argThat(jsQueueMsg -> { |
||||
|
return StringUtils.isNotEmpty(jsQueueMsg.getValue().getInvokeRequest().getScriptBody()); |
||||
|
})); |
||||
|
|
||||
|
ArgumentCaptor<TbProtoJsQueueMsg<RemoteJsRequest>> jsRequestsCaptor = ArgumentCaptor.forClass(TbProtoJsQueueMsg.class); |
||||
|
Object invocationResult = remoteJsInvokeService.invokeFunction(TenantId.SYS_TENANT_ID, null, scriptId, "{}").get(); |
||||
|
verify(jsRequestTemplate, times(2)).send(jsRequestsCaptor.capture()); |
||||
|
|
||||
|
List<TbProtoJsQueueMsg<RemoteJsRequest>> jsInvokeRequestsMade = jsRequestsCaptor.getAllValues(); |
||||
|
|
||||
|
JsInvokeProtos.JsInvokeRequest firstRequestMade = jsInvokeRequestsMade.get(0).getValue().getInvokeRequest(); |
||||
|
assertThat(firstRequestMade.getScriptBody()).isNullOrEmpty(); |
||||
|
|
||||
|
JsInvokeProtos.JsInvokeRequest secondRequestMade = jsInvokeRequestsMade.get(1).getValue().getInvokeRequest(); |
||||
|
assertThat(secondRequestMade.getScriptBody()).contains(scriptBody); |
||||
|
|
||||
|
assertThat(jsInvokeRequestsMade.stream().map(TbProtoQueueMsg::getKey).distinct().count()).as("partition keys are same") |
||||
|
.isOne(); |
||||
|
|
||||
|
assertThat(invocationResult).isEqualTo(expectedInvocationResult); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void whenDoingEval_thenSaveScriptByHashOfTenantIdAndScriptBody() throws Exception { |
||||
|
mockJsEvalResponse(); |
||||
|
|
||||
|
TenantId tenantId1 = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
String scriptBody1 = "var msg = { temp: 42, humidity: 77 };\n" + |
||||
|
"var metadata = { data: 40 };\n" + |
||||
|
"var msgType = \"POST_TELEMETRY_REQUEST\";\n" + |
||||
|
"\n" + |
||||
|
"return { msg: msg, metadata: metadata, msgType: msgType };"; |
||||
|
|
||||
|
Set<String> scriptHashes = new HashSet<>(); |
||||
|
String tenant1Script1Hash = null; |
||||
|
for (int i = 0; i < 3; i++) { |
||||
|
UUID scriptUuid = remoteJsInvokeService.eval(tenantId1, JsScriptType.RULE_NODE_SCRIPT, scriptBody1).get(); |
||||
|
tenant1Script1Hash = getScriptHash(scriptUuid); |
||||
|
scriptHashes.add(tenant1Script1Hash); |
||||
|
} |
||||
|
assertThat(scriptHashes).as("Unique scripts ids").size().isOne(); |
||||
|
|
||||
|
TenantId tenantId2 = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
UUID scriptUuid = remoteJsInvokeService.eval(tenantId2, JsScriptType.RULE_NODE_SCRIPT, scriptBody1).get(); |
||||
|
String tenant2Script1Id = getScriptHash(scriptUuid); |
||||
|
assertThat(tenant2Script1Id).isNotEqualTo(tenant1Script1Hash); |
||||
|
|
||||
|
String scriptBody2 = scriptBody1 + ";;"; |
||||
|
scriptUuid = remoteJsInvokeService.eval(tenantId2, JsScriptType.RULE_NODE_SCRIPT, scriptBody2).get(); |
||||
|
String tenant2Script2Id = getScriptHash(scriptUuid); |
||||
|
assertThat(tenant2Script2Id).isNotEqualTo(tenant2Script1Id); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void whenReleasingScript_thenCheckForHashUsages() throws Exception { |
||||
|
mockJsEvalResponse(); |
||||
|
String scriptBody = "return { a: 'b'};"; |
||||
|
UUID scriptId1 = remoteJsInvokeService.eval(TenantId.SYS_TENANT_ID, JsScriptType.RULE_NODE_SCRIPT, scriptBody).get(); |
||||
|
UUID scriptId2 = remoteJsInvokeService.eval(TenantId.SYS_TENANT_ID, JsScriptType.RULE_NODE_SCRIPT, scriptBody).get(); |
||||
|
String scriptHash = getScriptHash(scriptId1); |
||||
|
assertThat(scriptHash).isEqualTo(getScriptHash(scriptId2)); |
||||
|
reset(jsRequestTemplate); |
||||
|
|
||||
|
doReturn(Futures.immediateFuture(new TbProtoQueueMsg<>(UUID.randomUUID(), RemoteJsResponse.newBuilder() |
||||
|
.setReleaseResponse(JsInvokeProtos.JsReleaseResponse.newBuilder() |
||||
|
.setSuccess(true) |
||||
|
.build()) |
||||
|
.build()))) |
||||
|
.when(jsRequestTemplate).send(any()); |
||||
|
|
||||
|
remoteJsInvokeService.release(scriptId1).get(); |
||||
|
verifyNoInteractions(jsRequestTemplate); |
||||
|
assertThat(remoteJsInvokeService.scriptHashToBodysMap).containsKey(scriptHash); |
||||
|
|
||||
|
remoteJsInvokeService.release(scriptId2).get(); |
||||
|
verify(jsRequestTemplate).send(any()); |
||||
|
assertThat(remoteJsInvokeService.scriptHashToBodysMap).isEmpty(); |
||||
|
} |
||||
|
|
||||
|
private String getScriptHash(UUID scriptUuid) { |
||||
|
return remoteJsInvokeService.scriptIdToNameAndHashMap.get(scriptUuid).getSecond(); |
||||
|
} |
||||
|
|
||||
|
private void mockJsEvalResponse() { |
||||
|
doAnswer(methodCall -> Futures.immediateFuture(new TbProtoJsQueueMsg<>(UUID.randomUUID(), RemoteJsResponse.newBuilder() |
||||
|
.setCompileResponse(JsInvokeProtos.JsCompileResponse.newBuilder() |
||||
|
.setSuccess(true) |
||||
|
.setScriptHash(methodCall.<TbProtoQueueMsg<RemoteJsRequest>>getArgument(0).getValue().getCompileRequest().getScriptHash()) |
||||
|
.build()) |
||||
|
.build()))) |
||||
|
.when(jsRequestTemplate).send(argThat(jsQueueMsg -> jsQueueMsg.getValue().hasCompileRequest())); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,26 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.dao.util; |
||||
|
|
||||
|
import java.lang.annotation.ElementType; |
||||
|
import java.lang.annotation.Retention; |
||||
|
import java.lang.annotation.RetentionPolicy; |
||||
|
import java.lang.annotation.Target; |
||||
|
|
||||
|
@Retention(RetentionPolicy.RUNTIME) |
||||
|
@Target(ElementType.TYPE) |
||||
|
public @interface SqlDao { |
||||
|
} |
||||
@ -0,0 +1,26 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.data; |
||||
|
|
||||
|
import org.thingsboard.server.common.data.id.RuleChainId; |
||||
|
|
||||
|
public interface HasRuleEngineProfile { |
||||
|
|
||||
|
RuleChainId getDefaultRuleChainId(); |
||||
|
|
||||
|
String getDefaultQueueName(); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,24 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.edge; |
||||
|
|
||||
|
import org.thingsboard.server.common.msg.aware.TenantAwareMsg; |
||||
|
import org.thingsboard.server.common.msg.cluster.ToAllNodesMsg; |
||||
|
|
||||
|
import java.io.Serializable; |
||||
|
|
||||
|
public interface EdgeSessionMsg extends TenantAwareMsg, ToAllNodesMsg { |
||||
|
} |
||||
@ -0,0 +1,39 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.edge; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Getter; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.MsgType; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
@AllArgsConstructor |
||||
|
@Getter |
||||
|
public class FromEdgeSyncResponse implements EdgeSessionMsg { |
||||
|
|
||||
|
private final UUID id; |
||||
|
private final TenantId tenantId; |
||||
|
private final EdgeId edgeId; |
||||
|
private final boolean success; |
||||
|
|
||||
|
@Override |
||||
|
public MsgType getMsgType() { |
||||
|
return MsgType.EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,37 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.edge; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Getter; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.MsgType; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
@AllArgsConstructor |
||||
|
@Getter |
||||
|
public class ToEdgeSyncRequest implements EdgeSessionMsg { |
||||
|
private final UUID id; |
||||
|
private final TenantId tenantId; |
||||
|
private final EdgeId edgeId; |
||||
|
|
||||
|
@Override |
||||
|
public MsgType getMsgType() { |
||||
|
return MsgType.EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,59 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.dao.aspect; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
import org.springframework.data.util.Pair; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
|
||||
|
import java.util.Map; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.ConcurrentMap; |
||||
|
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
import java.util.concurrent.atomic.AtomicLong; |
||||
|
import java.util.stream.Collectors; |
||||
|
|
||||
|
@Data |
||||
|
public class DbCallStats { |
||||
|
|
||||
|
private final TenantId tenantId; |
||||
|
private final ConcurrentMap<String, MethodCallStats> methodStats = new ConcurrentHashMap<>(); |
||||
|
private final AtomicInteger successCalls = new AtomicInteger(); |
||||
|
private final AtomicInteger failureCalls = new AtomicInteger(); |
||||
|
|
||||
|
public void onMethodCall(String methodName, boolean success, long executionTime) { |
||||
|
var methodCallStats = methodStats.computeIfAbsent(methodName, m -> new MethodCallStats()); |
||||
|
methodCallStats.getExecutions().incrementAndGet(); |
||||
|
methodCallStats.getTiming().addAndGet(executionTime); |
||||
|
if (success) { |
||||
|
successCalls.incrementAndGet(); |
||||
|
} else { |
||||
|
failureCalls.incrementAndGet(); |
||||
|
methodCallStats.getFailures().incrementAndGet(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public DbCallStatsSnapshot snapshot() { |
||||
|
return DbCallStatsSnapshot.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.totalSuccess(successCalls.get()) |
||||
|
.totalFailure(failureCalls.get()) |
||||
|
.methodStats(methodStats.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().snapshot()))) |
||||
|
.totalTiming(methodStats.values().stream().map(MethodCallStats::getTiming).map(AtomicLong::get).reduce(0L, Long::sum)) |
||||
|
.build(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,38 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.dao.aspect; |
||||
|
|
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
|
||||
|
import java.util.Map; |
||||
|
|
||||
|
@Data |
||||
|
@Builder |
||||
|
public class DbCallStatsSnapshot { |
||||
|
|
||||
|
private final TenantId tenantId; |
||||
|
private final int totalSuccess; |
||||
|
private final int totalFailure; |
||||
|
private final long totalTiming; |
||||
|
private final Map<String, MethodCallStatsSnapshot> methodStats; |
||||
|
|
||||
|
public int getTotalCalls() { |
||||
|
return totalSuccess + totalFailure; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,33 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.dao.aspect; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
import java.util.concurrent.atomic.AtomicLong; |
||||
|
|
||||
|
@Data |
||||
|
public class MethodCallStats { |
||||
|
private final AtomicInteger executions = new AtomicInteger(); |
||||
|
private final AtomicInteger failures = new AtomicInteger(); |
||||
|
private final AtomicLong timing = new AtomicLong(); |
||||
|
|
||||
|
public MethodCallStatsSnapshot snapshot() { |
||||
|
return new MethodCallStatsSnapshot(executions.get(), failures.get(), timing.get()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,25 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.dao.aspect; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
@Data |
||||
|
public class MethodCallStatsSnapshot { |
||||
|
private final int executions; |
||||
|
private final int failures; |
||||
|
private final long timing; |
||||
|
} |
||||
@ -0,0 +1,241 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2022 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.dao.aspect; |
||||
|
|
||||
|
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.extern.slf4j.Slf4j; |
||||
|
import org.apache.commons.lang3.ArrayUtils; |
||||
|
import org.apache.commons.lang3.exception.ExceptionUtils; |
||||
|
import org.aspectj.lang.ProceedingJoinPoint; |
||||
|
import org.aspectj.lang.annotation.Around; |
||||
|
import org.aspectj.lang.annotation.Aspect; |
||||
|
import org.aspectj.lang.reflect.MethodSignature; |
||||
|
import org.checkerframework.checker.nullness.qual.Nullable; |
||||
|
import org.hibernate.exception.JDBCConnectionException; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.scheduling.annotation.Scheduled; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.StringUtils; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
|
||||
|
import java.util.Arrays; |
||||
|
import java.util.Comparator; |
||||
|
import java.util.HashMap; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.Set; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.ConcurrentMap; |
||||
|
import java.util.function.Consumer; |
||||
|
import java.util.stream.Collectors; |
||||
|
|
||||
|
import static org.apache.commons.lang3.StringUtils.join; |
||||
|
|
||||
|
@Aspect |
||||
|
@ConditionalOnProperty(prefix = "sql", value = "log_tenant_stats", havingValue = "true") |
||||
|
@Component |
||||
|
@Slf4j |
||||
|
public class SqlDaoCallsAspect { |
||||
|
|
||||
|
private final Set<String> invalidTenantDbCallMethods = ConcurrentHashMap.newKeySet(); |
||||
|
private final ConcurrentMap<TenantId, DbCallStats> statsMap = new ConcurrentHashMap<>(); |
||||
|
|
||||
|
@Value("${sql.batch_sort:true}") |
||||
|
private boolean batchSortEnabled; |
||||
|
|
||||
|
private static final String DEADLOCK_DETECTED_ERROR = "deadlock detected"; |
||||
|
|
||||
|
|
||||
|
@Scheduled(initialDelayString = "${sql.log_tenant_stats_interval_ms:60000}", |
||||
|
fixedDelayString = "${sql.log_tenant_stats_interval_ms:60000}") |
||||
|
public void printStats() { |
||||
|
List<DbCallStatsSnapshot> snapshots = snapshot(); |
||||
|
if (snapshots.isEmpty()) return; |
||||
|
try { |
||||
|
if (log.isTraceEnabled()) { |
||||
|
logTopNTenants(snapshots, Comparator.comparing(DbCallStatsSnapshot::getTotalTiming).reversed(), 0, snapshot -> { |
||||
|
logSnapshot(snapshot, 0, Comparator.comparing(MethodCallStatsSnapshot::getTiming).reversed(), log::trace); |
||||
|
}); |
||||
|
|
||||
|
Map<String, Map<TenantId, MethodCallStatsSnapshot>> byMethodStats = new HashMap<>(); |
||||
|
for (DbCallStatsSnapshot snapshot : snapshots) { |
||||
|
snapshot.getMethodStats().forEach((method, stats) -> { |
||||
|
byMethodStats.computeIfAbsent(method, m -> new HashMap<>()) |
||||
|
.put(snapshot.getTenantId(), stats); |
||||
|
}); |
||||
|
} |
||||
|
byMethodStats.forEach((method, byTenantStats) -> { |
||||
|
log.trace("Top tenants for method {} by calls:", method); |
||||
|
byTenantStats.entrySet().stream() |
||||
|
.sorted(Map.Entry.comparingByValue(Comparator.comparing(MethodCallStatsSnapshot::getExecutions).reversed())) |
||||
|
.limit(10) |
||||
|
.forEach(e -> { |
||||
|
TenantId tenantId = e.getKey(); |
||||
|
MethodCallStatsSnapshot methodStats = e.getValue(); |
||||
|
log.trace("[{}] calls: {}, failures: {}, timing: {}", tenantId, |
||||
|
methodStats.getExecutions(), methodStats.getFailures(), methodStats.getTiming()); |
||||
|
}); |
||||
|
}); |
||||
|
} else if (log.isDebugEnabled()) { |
||||
|
log.debug("Total calls statistics below:"); |
||||
|
logTopNTenants(snapshots, Comparator.comparingInt(DbCallStatsSnapshot::getTotalCalls).reversed(), 10, |
||||
|
s -> logSnapshot(s, 10, Comparator.comparing(MethodCallStatsSnapshot::getExecutions).reversed(), log::debug)); |
||||
|
log.debug("Total timing statistics below:"); |
||||
|
logTopNTenants(snapshots, Comparator.comparingLong(DbCallStatsSnapshot::getTotalTiming).reversed(), |
||||
|
10, s -> logSnapshot(s, 10, Comparator.comparing(MethodCallStatsSnapshot::getTiming).reversed(), log::debug)); |
||||
|
log.debug("Total errors statistics below:"); |
||||
|
logTopNTenants(snapshots, Comparator.comparingInt(DbCallStatsSnapshot::getTotalFailure).reversed(), |
||||
|
10, s -> logSnapshot(s, 10, Comparator.comparing(MethodCallStatsSnapshot::getFailures).reversed(), log::debug)); |
||||
|
} else if (log.isInfoEnabled()) { |
||||
|
log.info("Total calls statistics below:"); |
||||
|
logTopNTenants(snapshots, Comparator.comparingInt(DbCallStatsSnapshot::getTotalFailure).reversed(), |
||||
|
3, s -> logSnapshot(s, 3, Comparator.comparing(MethodCallStatsSnapshot::getFailures).reversed(), log::info)); |
||||
|
} |
||||
|
} finally { |
||||
|
statsMap.clear(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void logSnapshot(DbCallStatsSnapshot snapshot, int limit, Comparator<MethodCallStatsSnapshot> methodStatsComparator, Consumer<String> logger) { |
||||
|
logger.accept(String.format("[%s]: calls: %s, failures: %s, exec time: %s ", |
||||
|
snapshot.getTenantId(), snapshot.getTotalCalls(), snapshot.getTotalFailure(), snapshot.getTotalTiming())); |
||||
|
var stream = snapshot.getMethodStats().entrySet().stream() |
||||
|
.sorted(Map.Entry.comparingByValue(methodStatsComparator)); |
||||
|
if (limit > 0) { |
||||
|
stream = stream.limit(limit); |
||||
|
} |
||||
|
stream.forEach(e -> { |
||||
|
MethodCallStatsSnapshot methodStats = e.getValue(); |
||||
|
logger.accept(String.format("[%s]: method: %s, calls: %s, failures: %s, exec time: %s", snapshot.getTenantId(), e.getKey(), |
||||
|
methodStats.getExecutions(), methodStats.getFailures(), methodStats.getTiming())); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private List<DbCallStatsSnapshot> snapshot() { |
||||
|
return statsMap.values().stream().map(DbCallStats::snapshot).collect(Collectors.toList()); |
||||
|
} |
||||
|
|
||||
|
private void logTopNTenants(List<DbCallStatsSnapshot> snapshots, Comparator<DbCallStatsSnapshot> comparator, |
||||
|
int n, Consumer<DbCallStatsSnapshot> logFunction) { |
||||
|
var stream = snapshots.stream().sorted(comparator); |
||||
|
if (n > 0) { |
||||
|
stream = stream.limit(n); |
||||
|
} |
||||
|
stream.forEach(logFunction); |
||||
|
} |
||||
|
|
||||
|
@SuppressWarnings({"rawtypes", "unchecked"}) |
||||
|
@Around("@within(org.thingsboard.server.dao.util.SqlDao)") |
||||
|
public Object handleSqlCall(ProceedingJoinPoint joinPoint) throws Throwable { |
||||
|
MethodSignature signature = (MethodSignature) joinPoint.getSignature(); |
||||
|
var methodName = signature.toShortString(); |
||||
|
if (invalidTenantDbCallMethods.contains(methodName)) { |
||||
|
//Simply call the method if tenant is not found
|
||||
|
return joinPoint.proceed(); |
||||
|
} |
||||
|
var tenantId = getTenantId(signature, methodName, joinPoint.getArgs()); |
||||
|
if (tenantId == null || tenantId.isNullUid()) { |
||||
|
//Simply call the method if tenant is null
|
||||
|
return joinPoint.proceed(); |
||||
|
} |
||||
|
var startTime = System.currentTimeMillis(); |
||||
|
try { |
||||
|
var result = joinPoint.proceed(); |
||||
|
if (result instanceof ListenableFuture) { |
||||
|
Futures.addCallback((ListenableFuture) result, |
||||
|
new FutureCallback<>() { |
||||
|
@Override |
||||
|
public void onSuccess(@Nullable Object result) { |
||||
|
reportSuccessfulMethodExecution(tenantId, methodName, startTime); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
reportFailedMethodExecution(tenantId, methodName, startTime, t, joinPoint); |
||||
|
} |
||||
|
}, |
||||
|
MoreExecutors.directExecutor()); |
||||
|
} else { |
||||
|
reportSuccessfulMethodExecution(tenantId, methodName, startTime); |
||||
|
} |
||||
|
return result; |
||||
|
} catch (Throwable t) { |
||||
|
reportFailedMethodExecution(tenantId, methodName, startTime, t, joinPoint); |
||||
|
throw t; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void reportFailedMethodExecution(TenantId tenantId, String method, long startTime, Throwable t, ProceedingJoinPoint joinPoint) { |
||||
|
if (t != null) { |
||||
|
if (ExceptionUtils.indexOfThrowable(t, JDBCConnectionException.class) >= 0) { |
||||
|
return; |
||||
|
} |
||||
|
if (StringUtils.containedByAny(DEADLOCK_DETECTED_ERROR, ExceptionUtils.getRootCauseMessage(t), ExceptionUtils.getMessage(t))) { |
||||
|
if (!batchSortEnabled) { |
||||
|
log.warn("Deadlock was detected for method {} (tenant: {}). You might need to enable 'sql.batch_sort' option.", method, tenantId); |
||||
|
} else { |
||||
|
log.error("Deadlock was detected for method {} (tenant: {}). Arguments passed: \n{}\n The error: ", |
||||
|
method, tenantId, join(joinPoint.getArgs(), System.lineSeparator()), t); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
reportMethodExecution(tenantId, method, false, startTime); |
||||
|
} |
||||
|
|
||||
|
private void reportSuccessfulMethodExecution(TenantId tenantId, String method, long startTime) { |
||||
|
reportMethodExecution(tenantId, method, true, startTime); |
||||
|
} |
||||
|
|
||||
|
private void reportMethodExecution(TenantId tenantId, String method, boolean success, long startTime) { |
||||
|
statsMap.computeIfAbsent(tenantId, DbCallStats::new) |
||||
|
.onMethodCall(method, success, System.currentTimeMillis() - startTime); |
||||
|
} |
||||
|
|
||||
|
TenantId getTenantId(MethodSignature signature, String methodName, Object[] args) { |
||||
|
if (args == null || args.length == 0) { |
||||
|
addAndLogInvalidMethods(methodName); |
||||
|
return null; |
||||
|
} |
||||
|
for (int i = 0; i < args.length; i++) { |
||||
|
Object arg = args[i]; |
||||
|
if (arg instanceof TenantId) { |
||||
|
return (TenantId) arg; |
||||
|
} else if (arg instanceof UUID) { |
||||
|
if (signature.getParameterNames() != null && StringUtils.equals(signature.getParameterNames()[i], "tenantId")) { |
||||
|
log.trace("Method {} uses UUID for tenantId param instead of TenantId class", methodName); |
||||
|
return TenantId.fromUUID((UUID) arg); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
if (ArrayUtils.contains(signature.getParameterTypes(), TenantId.class) || |
||||
|
ArrayUtils.contains(signature.getParameterNames(), "tenantId")) { |
||||
|
log.debug("Null was submitted as tenantId to method {}. Args: {}", methodName, Arrays.toString(args)); |
||||
|
} else { |
||||
|
addAndLogInvalidMethods(methodName); |
||||
|
} |
||||
|
return null; |
||||
|
} |
||||
|
|
||||
|
private void addAndLogInvalidMethods(String methodName) { |
||||
|
invalidTenantDbCallMethods.add(methodName); |
||||
|
} |
||||
|
|
||||
|
} |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue