Yuriy Lytvynchuk
4 years ago
206 changed files with 5610 additions and 709 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