Browse Source
# Conflicts: # common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRuleRecipientsConfig.javapull/15196/head
dashevchenko
7 months ago
107 changed files with 3067 additions and 427 deletions
@ -0,0 +1,105 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.edge.rpc.processor.apikey; |
||||
|
|
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.common.data.EdgeUtils; |
||||
|
import org.thingsboard.server.common.data.edge.Edge; |
||||
|
import org.thingsboard.server.common.data.edge.EdgeEvent; |
||||
|
import org.thingsboard.server.common.data.edge.EdgeEventType; |
||||
|
import org.thingsboard.server.common.data.id.ApiKeyId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.msg.TbMsgType; |
||||
|
import org.thingsboard.server.common.data.pat.ApiKey; |
||||
|
import org.thingsboard.server.exception.DataValidationException; |
||||
|
import org.thingsboard.server.gen.edge.v1.ApiKeyUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.v1.DownlinkMsg; |
||||
|
import org.thingsboard.server.gen.edge.v1.EdgeVersion; |
||||
|
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; |
||||
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
||||
|
import org.thingsboard.server.service.edge.EdgeMsgConstructorUtils; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
|
||||
|
@Slf4j |
||||
|
@Component |
||||
|
@TbCoreComponent |
||||
|
public class ApiKeyEdgeProcessor extends BaseApiKeyProcessor implements ApiKeyProcessor { |
||||
|
|
||||
|
@Override |
||||
|
public ListenableFuture<Void> processApiKeyMsgFromEdge(TenantId tenantId, Edge edge, ApiKeyUpdateMsg apiKeyUpdateMsg) { |
||||
|
ApiKeyId apiKeyId = new ApiKeyId(new UUID(apiKeyUpdateMsg.getIdMSB(), apiKeyUpdateMsg.getIdLSB())); |
||||
|
try { |
||||
|
edgeSynchronizationManager.getEdgeId().set(edge.getId()); |
||||
|
|
||||
|
return switch (apiKeyUpdateMsg.getMsgType()) { |
||||
|
case ENTITY_CREATED_RPC_MESSAGE, ENTITY_UPDATED_RPC_MESSAGE -> { |
||||
|
boolean created = saveOrUpdateApiKey(tenantId, apiKeyId, apiKeyUpdateMsg); |
||||
|
if (created) { |
||||
|
ApiKey apiKey = edgeCtx.getApiKeyService().findApiKeyById(tenantId, apiKeyId); |
||||
|
if (apiKey != null) { |
||||
|
pushEntityEventToRuleEngine(tenantId, edge, apiKey, TbMsgType.ENTITY_CREATED); |
||||
|
} |
||||
|
} |
||||
|
yield Futures.immediateFuture(null); |
||||
|
} |
||||
|
case ENTITY_DELETED_RPC_MESSAGE -> { |
||||
|
deleteApiKey(tenantId, edge, apiKeyId); |
||||
|
yield Futures.immediateFuture(null); |
||||
|
} |
||||
|
default -> handleUnsupportedMsgType(apiKeyUpdateMsg.getMsgType()); |
||||
|
}; |
||||
|
} catch (DataValidationException e) { |
||||
|
return Futures.immediateFailedFuture(e); |
||||
|
} finally { |
||||
|
edgeSynchronizationManager.getEdgeId().remove(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public DownlinkMsg convertEdgeEventToDownlink(EdgeEvent edgeEvent, EdgeVersion edgeVersion) { |
||||
|
ApiKeyId apiKeyId = new ApiKeyId(edgeEvent.getEntityId()); |
||||
|
switch (edgeEvent.getAction()) { |
||||
|
case ADDED, UPDATED -> { |
||||
|
ApiKey apiKey = edgeCtx.getApiKeyService().findApiKeyById(edgeEvent.getTenantId(), apiKeyId); |
||||
|
if (apiKey != null) { |
||||
|
UpdateMsgType msgType = getUpdateMsgType(edgeEvent.getAction()); |
||||
|
ApiKeyUpdateMsg apiKeyUpdateMsg = EdgeMsgConstructorUtils.constructApiKeyUpdatedMsg(msgType, apiKey); |
||||
|
return DownlinkMsg.newBuilder() |
||||
|
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) |
||||
|
.addApiKeyUpdateMsg(apiKeyUpdateMsg) |
||||
|
.build(); |
||||
|
} |
||||
|
} |
||||
|
case DELETED -> { |
||||
|
ApiKeyUpdateMsg apiKeyUpdateMsg = EdgeMsgConstructorUtils.constructApiKeyDeleteMsg(apiKeyId); |
||||
|
return DownlinkMsg.newBuilder() |
||||
|
.setDownlinkMsgId(EdgeUtils.nextPositiveInt()) |
||||
|
.addApiKeyUpdateMsg(apiKeyUpdateMsg) |
||||
|
.build(); |
||||
|
} |
||||
|
} |
||||
|
return null; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public EdgeEventType getEdgeEventType() { |
||||
|
return EdgeEventType.API_KEY; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,28 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.edge.rpc.processor.apikey; |
||||
|
|
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import org.thingsboard.server.common.data.edge.Edge; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.gen.edge.v1.ApiKeyUpdateMsg; |
||||
|
import org.thingsboard.server.service.edge.rpc.processor.EdgeProcessor; |
||||
|
|
||||
|
public interface ApiKeyProcessor extends EdgeProcessor { |
||||
|
|
||||
|
ListenableFuture<Void> processApiKeyMsgFromEdge(TenantId tenantId, Edge edge, ApiKeyUpdateMsg apiKeyUpdateMsg); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,63 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.edge.rpc.processor.apikey; |
||||
|
|
||||
|
import com.datastax.oss.driver.api.core.uuid.Uuids; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.server.common.data.edge.Edge; |
||||
|
import org.thingsboard.server.common.data.id.ApiKeyId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.msg.TbMsgType; |
||||
|
import org.thingsboard.server.common.data.pat.ApiKey; |
||||
|
import org.thingsboard.server.gen.edge.v1.ApiKeyUpdateMsg; |
||||
|
import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor; |
||||
|
|
||||
|
@Slf4j |
||||
|
public abstract class BaseApiKeyProcessor extends BaseEdgeProcessor { |
||||
|
|
||||
|
protected boolean saveOrUpdateApiKey(TenantId tenantId, ApiKeyId apiKeyId, ApiKeyUpdateMsg apiKeyUpdateMsg) { |
||||
|
boolean isCreated = false; |
||||
|
try { |
||||
|
ApiKey apiKey = JacksonUtil.fromString(apiKeyUpdateMsg.getEntity(), ApiKey.class, true); |
||||
|
if (apiKey == null) { |
||||
|
throw new RuntimeException("[{" + tenantId + "}] apiKeyUpdateMsg {" + apiKeyUpdateMsg + " } cannot be converted to apiKey"); |
||||
|
} |
||||
|
|
||||
|
ApiKey existingApiKey = edgeCtx.getApiKeyService().findApiKeyById(tenantId, apiKeyId); |
||||
|
if (existingApiKey == null) { |
||||
|
apiKey.setCreatedTime(Uuids.unixTimestamp(apiKeyId.getId())); |
||||
|
isCreated = true; |
||||
|
} |
||||
|
|
||||
|
apiKey.setId(apiKeyId); |
||||
|
edgeCtx.getApiKeyService().saveApiKey(tenantId, apiKey, apiKey.getValue(), false); |
||||
|
} catch (Exception e) { |
||||
|
log.error("[{}] Failed to process apiKey update msg [{}]", tenantId, apiKeyUpdateMsg, e); |
||||
|
throw e; |
||||
|
} |
||||
|
return isCreated; |
||||
|
} |
||||
|
|
||||
|
protected void deleteApiKey(TenantId tenantId, Edge edge, ApiKeyId apiKeyId) { |
||||
|
ApiKey apiKey = edgeCtx.getApiKeyService().findApiKeyById(tenantId, apiKeyId); |
||||
|
if (apiKey != null) { |
||||
|
edgeCtx.getApiKeyService().deleteApiKey(tenantId, apiKey, false); |
||||
|
pushEntityEventToRuleEngine(tenantId, edge, apiKey, TbMsgType.ENTITY_DELETED); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,71 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.controller; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.junit.After; |
||||
|
import org.junit.Before; |
||||
|
import org.junit.Test; |
||||
|
import org.thingsboard.server.common.data.pat.ApiKey; |
||||
|
import org.thingsboard.server.common.data.pat.ApiKeyInfo; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
|
||||
|
import java.net.URI; |
||||
|
import java.net.URISyntaxException; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
||||
|
|
||||
|
@Slf4j |
||||
|
@DaoSqlTest |
||||
|
public class ApiKeyWebSocketApiTest extends WebSocketApiTest { |
||||
|
|
||||
|
private ApiKey apiKey; |
||||
|
|
||||
|
@Before |
||||
|
public void setUpApiKey() { |
||||
|
ApiKeyInfo apiKeyInfo = new ApiKeyInfo(); |
||||
|
apiKeyInfo.setDescription("WS test API key"); |
||||
|
apiKeyInfo.setEnabled(true); |
||||
|
apiKeyInfo.setUserId(tenantAdminUserId); |
||||
|
apiKey = doPost("/api/apiKey", apiKeyInfo, ApiKey.class); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void tearDownApiKey() throws Exception { |
||||
|
loginTenantAdmin(); |
||||
|
doDelete("/api/apiKey/" + apiKey.getId()).andExpect(status().isOk()); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected TbTestWebSocketClient buildAndConnectWebSocketClient() throws URISyntaxException, InterruptedException { |
||||
|
return buildAndConnectWebSocketClientWithApiKey(apiKey.getValue()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testInvalidApiKeyAuthCmd_connectionClosed() throws Exception { |
||||
|
TbTestWebSocketClient client = new TbTestWebSocketClient(new URI(WS_URL + wsPort + "/api/ws")); |
||||
|
assertThat(client.connectBlocking(TIMEOUT, TimeUnit.SECONDS)).isTrue(); |
||||
|
try { |
||||
|
client.authenticateWithApiKey("invalid-key"); |
||||
|
assertThat(client.waitForClose()).isTrue(); |
||||
|
} finally { |
||||
|
client.close(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,238 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.edge; |
||||
|
|
||||
|
import com.datastax.oss.driver.api.core.uuid.Uuids; |
||||
|
import com.google.protobuf.AbstractMessage; |
||||
|
import com.google.protobuf.InvalidProtocolBufferException; |
||||
|
import org.junit.Assert; |
||||
|
import org.junit.Test; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.server.common.data.User; |
||||
|
import org.thingsboard.server.common.data.id.ApiKeyId; |
||||
|
import org.thingsboard.server.common.data.pat.ApiKey; |
||||
|
import org.thingsboard.server.common.data.pat.ApiKeyInfo; |
||||
|
import org.thingsboard.server.common.data.security.Authority; |
||||
|
import org.thingsboard.server.dao.pat.ApiKeyService; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
import org.thingsboard.server.gen.edge.v1.ApiKeyUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; |
||||
|
import org.thingsboard.server.gen.edge.v1.UplinkMsg; |
||||
|
import org.thingsboard.server.gen.edge.v1.UplinkResponseMsg; |
||||
|
import org.thingsboard.server.gen.edge.v1.UserCredentialsUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.v1.UserUpdateMsg; |
||||
|
|
||||
|
import java.util.Optional; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
||||
|
import static org.thingsboard.server.gen.edge.v1.UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
public class ApiKeyEdgeTest extends AbstractEdgeTest { |
||||
|
|
||||
|
@Autowired |
||||
|
private ApiKeyService apiKeyService; |
||||
|
|
||||
|
private static final String DEFAULT_API_KEY_DESCRIPTION = "Edge Test ApiKey"; |
||||
|
private static final String UPDATED_API_KEY_DESCRIPTION = "Updated Edge Test ApiKey"; |
||||
|
|
||||
|
@Test |
||||
|
public void testApiKey_create_update_delete_fromCloud() throws Exception { |
||||
|
// create ApiKey
|
||||
|
ApiKeyInfo apiKeyInfo = createSimpleApiKeyInfo(DEFAULT_API_KEY_DESCRIPTION); |
||||
|
|
||||
|
edgeImitator.expectMessageAmount(1); |
||||
|
ApiKey savedApiKey = doPost("/api/apiKey", apiKeyInfo, ApiKey.class); |
||||
|
Assert.assertTrue(edgeImitator.waitForMessages()); |
||||
|
|
||||
|
AbstractMessage latestMessage = edgeImitator.getLatestMessage(); |
||||
|
Assert.assertTrue(latestMessage instanceof ApiKeyUpdateMsg); |
||||
|
ApiKeyUpdateMsg apiKeyUpdateMsg = (ApiKeyUpdateMsg) latestMessage; |
||||
|
Assert.assertEquals(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, apiKeyUpdateMsg.getMsgType()); |
||||
|
Assert.assertEquals(savedApiKey.getUuidId().getMostSignificantBits(), apiKeyUpdateMsg.getIdMSB()); |
||||
|
Assert.assertEquals(savedApiKey.getUuidId().getLeastSignificantBits(), apiKeyUpdateMsg.getIdLSB()); |
||||
|
ApiKey apiKeyFromMsg = JacksonUtil.fromString(apiKeyUpdateMsg.getEntity(), ApiKey.class, true); |
||||
|
Assert.assertNotNull(apiKeyFromMsg); |
||||
|
|
||||
|
Assert.assertEquals(DEFAULT_API_KEY_DESCRIPTION, apiKeyFromMsg.getDescription()); |
||||
|
Assert.assertEquals(savedApiKey.getTenantId(), apiKeyFromMsg.getTenantId()); |
||||
|
|
||||
|
// update ApiKey
|
||||
|
edgeImitator.expectMessageAmount(1); |
||||
|
savedApiKey.setDescription(UPDATED_API_KEY_DESCRIPTION); |
||||
|
savedApiKey = doPost("/api/apiKey", new ApiKeyInfo(savedApiKey), ApiKey.class); |
||||
|
Assert.assertTrue(edgeImitator.waitForMessages()); |
||||
|
|
||||
|
latestMessage = edgeImitator.getLatestMessage(); |
||||
|
Assert.assertTrue(latestMessage instanceof ApiKeyUpdateMsg); |
||||
|
apiKeyUpdateMsg = (ApiKeyUpdateMsg) latestMessage; |
||||
|
apiKeyFromMsg = JacksonUtil.fromString(apiKeyUpdateMsg.getEntity(), ApiKey.class, true); |
||||
|
Assert.assertNotNull(apiKeyFromMsg); |
||||
|
Assert.assertEquals(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE, apiKeyUpdateMsg.getMsgType()); |
||||
|
Assert.assertEquals(UPDATED_API_KEY_DESCRIPTION, apiKeyFromMsg.getDescription()); |
||||
|
|
||||
|
// delete ApiKey
|
||||
|
edgeImitator.expectMessageAmount(1); |
||||
|
doDelete("/api/apiKey/" + savedApiKey.getUuidId()) |
||||
|
.andExpect(status().isOk()); |
||||
|
Assert.assertTrue(edgeImitator.waitForMessages()); |
||||
|
|
||||
|
latestMessage = edgeImitator.getLatestMessage(); |
||||
|
Assert.assertTrue(latestMessage instanceof ApiKeyUpdateMsg); |
||||
|
apiKeyUpdateMsg = (ApiKeyUpdateMsg) latestMessage; |
||||
|
Assert.assertEquals(UpdateMsgType.ENTITY_DELETED_RPC_MESSAGE, apiKeyUpdateMsg.getMsgType()); |
||||
|
Assert.assertEquals(savedApiKey.getUuidId().getMostSignificantBits(), apiKeyUpdateMsg.getIdMSB()); |
||||
|
Assert.assertEquals(savedApiKey.getUuidId().getLeastSignificantBits(), apiKeyUpdateMsg.getIdLSB()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testApiKey_create_update_delete_toCloud() throws Exception { |
||||
|
// create
|
||||
|
ApiKey apiKey = createSimpleApiKey(DEFAULT_API_KEY_DESCRIPTION); |
||||
|
UUID uuid = Uuids.timeBased(); |
||||
|
UplinkMsg uplinkMsg = getUplinkMsg(uuid, apiKey, UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE); |
||||
|
|
||||
|
checkApiKeyOnCloud(uplinkMsg, uuid, apiKey.getDescription()); |
||||
|
|
||||
|
// update
|
||||
|
apiKey.setDescription(UPDATED_API_KEY_DESCRIPTION); |
||||
|
UplinkMsg updatedUplinkMsg = getUplinkMsg(uuid, apiKey, UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE); |
||||
|
|
||||
|
checkApiKeyOnCloud(updatedUplinkMsg, uuid, apiKey.getDescription()); |
||||
|
|
||||
|
// delete
|
||||
|
UplinkMsg deleteUplinkMsg = getDeleteUplinkMsg(uuid); |
||||
|
edgeImitator.expectResponsesAmount(1); |
||||
|
edgeImitator.sendUplinkMsg(deleteUplinkMsg); |
||||
|
Assert.assertTrue(edgeImitator.waitForResponses()); |
||||
|
|
||||
|
ApiKeyId apiKeyId = new ApiKeyId(uuid); |
||||
|
await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> |
||||
|
Assert.assertNull(apiKeyService.findApiKeyById(tenantId, apiKeyId)) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testApiKey_pushedDuringUserSync() throws Exception { |
||||
|
// create tenant admin user - expect 3 messages: 1 UserUpdateMsg + 2 UserCredentialsUpdateMsg
|
||||
|
User user = new User(); |
||||
|
user.setAuthority(Authority.TENANT_ADMIN); |
||||
|
user.setTenantId(tenantId); |
||||
|
user.setEmail("apiKeyTestUser@thingsboard.org"); |
||||
|
user.setFirstName("ApiKey"); |
||||
|
user.setLastName("TestUser"); |
||||
|
|
||||
|
edgeImitator.expectMessageAmount(3); |
||||
|
User savedUser = createUser(user, "tenant"); |
||||
|
Assert.assertTrue(edgeImitator.waitForMessages()); |
||||
|
Assert.assertEquals(1, edgeImitator.findAllMessagesByType(UserUpdateMsg.class).size()); |
||||
|
Assert.assertEquals(2, edgeImitator.findAllMessagesByType(UserCredentialsUpdateMsg.class).size()); |
||||
|
|
||||
|
// create API key for this user - expect 1 ApiKeyUpdateMsg
|
||||
|
ApiKeyInfo apiKeyInfo = new ApiKeyInfo(); |
||||
|
apiKeyInfo.setTenantId(tenantId); |
||||
|
apiKeyInfo.setUserId(savedUser.getId()); |
||||
|
apiKeyInfo.setDescription("Test API Key for user sync"); |
||||
|
apiKeyInfo.setEnabled(true); |
||||
|
|
||||
|
edgeImitator.expectMessageAmount(1); |
||||
|
doPost("/api/apiKey", apiKeyInfo, ApiKey.class); |
||||
|
Assert.assertTrue(edgeImitator.waitForMessages()); |
||||
|
Assert.assertEquals(1, edgeImitator.findAllMessagesByType(ApiKeyUpdateMsg.class).size()); |
||||
|
|
||||
|
// update user - expect 3 messages: UserUpdateMsg + UserCredentialsUpdateMsg + ApiKeyUpdateMsg
|
||||
|
savedUser.setLastName("UpdatedLastName"); |
||||
|
edgeImitator.expectMessageAmount(3); |
||||
|
doPost("/api/user", savedUser, User.class); |
||||
|
Assert.assertTrue(edgeImitator.waitForMessages()); |
||||
|
|
||||
|
Assert.assertEquals(1, edgeImitator.findAllMessagesByType(UserUpdateMsg.class).size()); |
||||
|
Assert.assertEquals(1, edgeImitator.findAllMessagesByType(UserCredentialsUpdateMsg.class).size()); |
||||
|
Assert.assertEquals(1, edgeImitator.findAllMessagesByType(ApiKeyUpdateMsg.class).size()); |
||||
|
|
||||
|
Optional<ApiKeyUpdateMsg> apiKeyUpdateMsgOpt = edgeImitator.findMessageByType(ApiKeyUpdateMsg.class); |
||||
|
Assert.assertTrue(apiKeyUpdateMsgOpt.isPresent()); |
||||
|
ApiKeyUpdateMsg apiKeyUpdateMsg = apiKeyUpdateMsgOpt.get(); |
||||
|
Assert.assertEquals(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE, apiKeyUpdateMsg.getMsgType()); |
||||
|
} |
||||
|
|
||||
|
private ApiKeyInfo createSimpleApiKeyInfo(String description) { |
||||
|
ApiKeyInfo apiKeyInfo = new ApiKeyInfo(); |
||||
|
apiKeyInfo.setTenantId(tenantId); |
||||
|
apiKeyInfo.setUserId(tenantAdminUserId); |
||||
|
apiKeyInfo.setDescription(description); |
||||
|
apiKeyInfo.setEnabled(true); |
||||
|
return apiKeyInfo; |
||||
|
} |
||||
|
|
||||
|
private ApiKey createSimpleApiKey(String description) { |
||||
|
ApiKey apiKey = new ApiKey(); |
||||
|
apiKey.setTenantId(tenantId); |
||||
|
apiKey.setUserId(tenantAdminUserId); |
||||
|
apiKey.setDescription(description); |
||||
|
apiKey.setEnabled(true); |
||||
|
apiKey.setValue("test-api-key-value-" + UUID.randomUUID()); |
||||
|
return apiKey; |
||||
|
} |
||||
|
|
||||
|
private UplinkMsg getDeleteUplinkMsg(UUID uuid) throws InvalidProtocolBufferException { |
||||
|
UplinkMsg.Builder upLinkMsgBuilder = UplinkMsg.newBuilder(); |
||||
|
ApiKeyUpdateMsg.Builder apiKeyDeleteMsgBuilder = ApiKeyUpdateMsg.newBuilder(); |
||||
|
apiKeyDeleteMsgBuilder.setMsgType(ENTITY_DELETED_RPC_MESSAGE); |
||||
|
apiKeyDeleteMsgBuilder.setIdMSB(uuid.getMostSignificantBits()); |
||||
|
apiKeyDeleteMsgBuilder.setIdLSB(uuid.getLeastSignificantBits()); |
||||
|
testAutoGeneratedCodeByProtobuf(apiKeyDeleteMsgBuilder); |
||||
|
|
||||
|
upLinkMsgBuilder.addApiKeyUpdateMsg(apiKeyDeleteMsgBuilder.build()); |
||||
|
testAutoGeneratedCodeByProtobuf(upLinkMsgBuilder); |
||||
|
|
||||
|
return upLinkMsgBuilder.build(); |
||||
|
} |
||||
|
|
||||
|
private UplinkMsg getUplinkMsg(UUID uuid, ApiKey apiKey, UpdateMsgType updateMsgType) throws InvalidProtocolBufferException { |
||||
|
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder(); |
||||
|
ApiKeyUpdateMsg.Builder apiKeyUpdateMsgBuilder = ApiKeyUpdateMsg.newBuilder(); |
||||
|
apiKeyUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits()); |
||||
|
apiKeyUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits()); |
||||
|
apiKeyUpdateMsgBuilder.setEntity(JacksonUtil.toString(apiKey)); |
||||
|
apiKeyUpdateMsgBuilder.setMsgType(updateMsgType); |
||||
|
testAutoGeneratedCodeByProtobuf(apiKeyUpdateMsgBuilder); |
||||
|
uplinkMsgBuilder.addApiKeyUpdateMsg(apiKeyUpdateMsgBuilder.build()); |
||||
|
|
||||
|
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder); |
||||
|
|
||||
|
return uplinkMsgBuilder.build(); |
||||
|
} |
||||
|
|
||||
|
private void checkApiKeyOnCloud(UplinkMsg uplinkMsg, UUID uuid, String description) throws Exception { |
||||
|
edgeImitator.expectResponsesAmount(1); |
||||
|
edgeImitator.sendUplinkMsg(uplinkMsg); |
||||
|
|
||||
|
Assert.assertTrue(edgeImitator.waitForResponses()); |
||||
|
|
||||
|
UplinkResponseMsg latestResponseMsg = edgeImitator.getLatestResponseMsg(); |
||||
|
Assert.assertTrue(latestResponseMsg.getSuccess()); |
||||
|
|
||||
|
ApiKey apiKey = apiKeyService.findApiKeyById(tenantId, new ApiKeyId(uuid)); |
||||
|
Assert.assertNotNull(apiKey); |
||||
|
Assert.assertEquals(description, apiKey.getDescription()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,272 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.edge.rpc; |
||||
|
|
||||
|
import io.grpc.ManagedChannel; |
||||
|
import io.grpc.Server; |
||||
|
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; |
||||
|
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; |
||||
|
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; |
||||
|
import org.bouncycastle.asn1.x500.X500Name; |
||||
|
import org.bouncycastle.cert.jcajce.JcaX509CertificateConverter; |
||||
|
import org.bouncycastle.cert.jcajce.JcaX509v3CertificateBuilder; |
||||
|
import org.bouncycastle.jce.provider.BouncyCastleProvider; |
||||
|
import org.bouncycastle.openssl.jcajce.JcaPEMWriter; |
||||
|
import org.bouncycastle.openssl.jcajce.JcePEMEncryptorBuilder; |
||||
|
import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder; |
||||
|
import org.bouncycastle.util.io.pem.PemObject; |
||||
|
import org.junit.jupiter.api.AfterEach; |
||||
|
import org.junit.jupiter.params.ParameterizedTest; |
||||
|
import org.junit.jupiter.params.provider.EnumSource; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
import org.thingsboard.server.controller.AbstractWebTest; |
||||
|
import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc; |
||||
|
|
||||
|
import java.io.ByteArrayInputStream; |
||||
|
import java.math.BigInteger; |
||||
|
import java.nio.charset.StandardCharsets; |
||||
|
import java.nio.file.Files; |
||||
|
import java.nio.file.Path; |
||||
|
import java.security.KeyPair; |
||||
|
import java.security.KeyPairGenerator; |
||||
|
import java.security.PrivateKey; |
||||
|
import java.security.Security; |
||||
|
import java.security.cert.X509Certificate; |
||||
|
import java.security.spec.ECGenParameterSpec; |
||||
|
import java.util.ArrayList; |
||||
|
import java.util.Date; |
||||
|
import java.util.List; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
|
||||
|
/** |
||||
|
* Tests for Edge gRPC SSL setup using the production {@link EdgeGrpcService#setupSsl} method. |
||||
|
* <p> |
||||
|
* Covers: |
||||
|
* 1. Separate cert and key PEM inputs |
||||
|
* 2. Combined PEM (cert + key in one file) |
||||
|
* 3. Encrypted private key with password |
||||
|
* 4. Missing key in combined PEM → error |
||||
|
* <p> |
||||
|
* Each scenario is parameterized across key types: RSA-2048, RSA-4096, EC P-256, EC P-384. |
||||
|
*/ |
||||
|
class EdgeGrpcSslTest { |
||||
|
|
||||
|
static { |
||||
|
if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) { |
||||
|
Security.addProvider(new BouncyCastleProvider()); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
enum KeyType { |
||||
|
RSA_2048("RSA", 2048, null, "SHA256withRSA"), |
||||
|
RSA_4096("RSA", 4096, null, "SHA256withRSA"), |
||||
|
EC_P256("EC", 256, "secp256r1", "SHA256withECDSA"), |
||||
|
EC_P384("EC", 384, "secp384r1", "SHA384withECDSA"); |
||||
|
|
||||
|
final String algorithm; |
||||
|
final int size; |
||||
|
final String curve; |
||||
|
final String sigAlg; |
||||
|
|
||||
|
KeyType(String algorithm, int size, String curve, String sigAlg) { |
||||
|
this.algorithm = algorithm; |
||||
|
this.size = size; |
||||
|
this.curve = curve; |
||||
|
this.sigAlg = sigAlg; |
||||
|
} |
||||
|
|
||||
|
KeyPair generateKeyPair() throws Exception { |
||||
|
KeyPairGenerator kpg = KeyPairGenerator.getInstance(algorithm); |
||||
|
if (curve != null) { |
||||
|
kpg.initialize(new ECGenParameterSpec(curve)); |
||||
|
} else { |
||||
|
kpg.initialize(size); |
||||
|
} |
||||
|
return kpg.generateKeyPair(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private final List<Path> tempFiles = new ArrayList<>(); |
||||
|
private Server server; |
||||
|
private ManagedChannel channel; |
||||
|
|
||||
|
@AfterEach |
||||
|
void cleanup() throws Exception { |
||||
|
if (channel != null) { |
||||
|
channel.shutdownNow().awaitTermination(2, TimeUnit.SECONDS); |
||||
|
} |
||||
|
if (server != null) { |
||||
|
server.shutdownNow().awaitTermination(2, TimeUnit.SECONDS); |
||||
|
} |
||||
|
for (Path p : tempFiles) { |
||||
|
Files.deleteIfExists(p); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest(name = "separateCertAndKey_{0}") |
||||
|
@EnumSource(KeyType.class) |
||||
|
void separateCertAndKey(KeyType keyType) throws Exception { |
||||
|
KeyPair kp = keyType.generateKeyPair(); |
||||
|
X509Certificate cert = generateSelfSignedCert(kp, keyType.sigAlg); |
||||
|
|
||||
|
Path certFile = writeTempPem("cert", cert); |
||||
|
Path keyFile = writeTempPem("key", kp.getPrivate()); |
||||
|
|
||||
|
server = startServer(certFile.toString(), keyFile.toString(), null); |
||||
|
assertTlsConnectivity(cert); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest(name = "combinedPemWithCertAndKey_{0}") |
||||
|
@EnumSource(KeyType.class) |
||||
|
void combinedPemWithCertAndKey(KeyType keyType) throws Exception { |
||||
|
KeyPair kp = keyType.generateKeyPair(); |
||||
|
X509Certificate cert = generateSelfSignedCert(kp, keyType.sigAlg); |
||||
|
|
||||
|
Path combinedFile = writeTempPem("combined", cert, kp.getPrivate()); |
||||
|
|
||||
|
server = startServer(combinedFile.toString(), "", null); |
||||
|
assertTlsConnectivity(cert); |
||||
|
} |
||||
|
|
||||
|
// RSA-only: BouncyCastle writes encrypted EC keys in traditional PEM format (BEGIN EC PRIVATE KEY),
|
||||
|
// which after decryption produces a PEMKeyPair without public key info — causing PemSslCredentials
|
||||
|
// to fail with "Cannot invoke SubjectPublicKeyInfo.getEncoded() because getPublicKeyInfo() is null".
|
||||
|
@ParameterizedTest(name = "encryptedPrivateKey_{0}") |
||||
|
@EnumSource(value = KeyType.class, names = {"RSA_2048", "RSA_4096"}) |
||||
|
void encryptedPrivateKey(KeyType keyType) throws Exception { |
||||
|
KeyPair kp = keyType.generateKeyPair(); |
||||
|
X509Certificate cert = generateSelfSignedCert(kp, keyType.sigAlg); |
||||
|
String password = "test-password"; |
||||
|
|
||||
|
Path combinedFile = writeTempPemEncrypted("enc-combined", password, cert, kp.getPrivate()); |
||||
|
|
||||
|
server = startServer(combinedFile.toString(), "", password); |
||||
|
assertTlsConnectivity(cert); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest(name = "combinedPemWithCertOnly_throwsException_{0}") |
||||
|
@EnumSource(KeyType.class) |
||||
|
void combinedPemWithCertOnly_throwsException(KeyType keyType) throws Exception { |
||||
|
KeyPair kp = keyType.generateKeyPair(); |
||||
|
X509Certificate cert = generateSelfSignedCert(kp, keyType.sigAlg); |
||||
|
|
||||
|
Path certOnlyFile = writeTempPem("cert-only", cert); |
||||
|
|
||||
|
assertThatThrownBy(() -> startServer(certOnlyFile.toString(), "", null)) |
||||
|
.isInstanceOf(IllegalArgumentException.class); |
||||
|
} |
||||
|
|
||||
|
// --- Server startup using production EdgeGrpcService.setupSsl() ---
|
||||
|
|
||||
|
private Server startServer(String certFileResource, String privateKeyResource, String keyPassword) throws Exception { |
||||
|
EdgeGrpcService edgeGrpcService = new EdgeGrpcService(); |
||||
|
ReflectionTestUtils.setField(edgeGrpcService, "certFileResource", certFileResource); |
||||
|
ReflectionTestUtils.setField(edgeGrpcService, "privateKeyResource", privateKeyResource); |
||||
|
ReflectionTestUtils.setField(edgeGrpcService, "keyPassword", keyPassword != null ? keyPassword : ""); |
||||
|
|
||||
|
NettyServerBuilder builder = NettyServerBuilder.forPort(0) |
||||
|
.addService(new EdgeRpcServiceGrpc.EdgeRpcServiceImplBase() {}); |
||||
|
|
||||
|
edgeGrpcService.setupSsl(builder); |
||||
|
|
||||
|
return builder.build().start(); |
||||
|
} |
||||
|
|
||||
|
private void assertTlsConnectivity(X509Certificate trustedCert) throws Exception { |
||||
|
String certPem = toPem(trustedCert); |
||||
|
var clientSsl = GrpcSslContexts.forClient() |
||||
|
.trustManager(new ByteArrayInputStream(certPem.getBytes(StandardCharsets.UTF_8))) |
||||
|
.build(); |
||||
|
|
||||
|
channel = NettyChannelBuilder.forAddress("localhost", server.getPort()) |
||||
|
.sslContext(clientSsl) |
||||
|
.build(); |
||||
|
|
||||
|
channel.getState(true); // trigger connection attempt
|
||||
|
await().atMost(AbstractWebTest.TIMEOUT, TimeUnit.SECONDS) |
||||
|
.pollInterval(50, TimeUnit.MILLISECONDS) |
||||
|
.untilAsserted(() -> { |
||||
|
var state = channel.getState(false); |
||||
|
if (state == io.grpc.ConnectivityState.TRANSIENT_FAILURE) { |
||||
|
throw new AssertionError("TLS handshake failed: channel in TRANSIENT_FAILURE"); |
||||
|
} |
||||
|
assertThat(state).isEqualTo(io.grpc.ConnectivityState.READY); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
// --- Cert/key generation ---
|
||||
|
|
||||
|
private X509Certificate generateSelfSignedCert(KeyPair kp, String sigAlg) throws Exception { |
||||
|
X500Name subject = new X500Name("CN=localhost"); |
||||
|
Date now = new Date(); |
||||
|
return new JcaX509CertificateConverter().getCertificate( |
||||
|
new JcaX509v3CertificateBuilder( |
||||
|
subject, BigInteger.ONE, now, |
||||
|
new Date(now.getTime() + TimeUnit.DAYS.toMillis(1)), |
||||
|
subject, kp.getPublic()) |
||||
|
.build(new JcaContentSignerBuilder(sigAlg).build(kp.getPrivate()))); |
||||
|
} |
||||
|
|
||||
|
// --- PEM file helpers ---
|
||||
|
|
||||
|
private String toPem(Object obj) throws Exception { |
||||
|
java.io.StringWriter sw = new java.io.StringWriter(); |
||||
|
try (JcaPEMWriter w = new JcaPEMWriter(sw)) { |
||||
|
w.writeObject(obj); |
||||
|
} |
||||
|
return sw.toString(); |
||||
|
} |
||||
|
|
||||
|
private Path writeTempPem(String prefix, Object... objects) throws Exception { |
||||
|
Path p = Files.createTempFile(prefix + "-", ".pem"); |
||||
|
tempFiles.add(p); |
||||
|
try (JcaPEMWriter w = new JcaPEMWriter(Files.newBufferedWriter(p))) { |
||||
|
for (Object o : objects) { |
||||
|
w.writeObject(toPkcs8IfKey(o)); |
||||
|
} |
||||
|
} |
||||
|
return p; |
||||
|
} |
||||
|
|
||||
|
private Path writeTempPemEncrypted(String prefix, String password, Object... objects) throws Exception { |
||||
|
Path p = Files.createTempFile(prefix + "-", ".pem"); |
||||
|
tempFiles.add(p); |
||||
|
var encryptor = new JcePEMEncryptorBuilder("AES-256-CBC") |
||||
|
.setProvider(BouncyCastleProvider.PROVIDER_NAME) |
||||
|
.build(password.toCharArray()); |
||||
|
try (JcaPEMWriter w = new JcaPEMWriter(Files.newBufferedWriter(p))) { |
||||
|
for (Object o : objects) { |
||||
|
if (o instanceof PrivateKey) { |
||||
|
w.writeObject(o, encryptor); |
||||
|
} else { |
||||
|
w.writeObject(o); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
return p; |
||||
|
} |
||||
|
|
||||
|
private Object toPkcs8IfKey(Object o) { |
||||
|
if (o instanceof PrivateKey pk) { |
||||
|
return new PemObject("PRIVATE KEY", pk.getEncoded()); |
||||
|
} |
||||
|
return o; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,144 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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 org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.junit.jupiter.api.extension.ExtendWith; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.dao.notification.NotificationRequestDao; |
||||
|
import org.thingsboard.server.dao.sqlts.insert.sql.SqlPartitioningRepository; |
||||
|
import org.thingsboard.server.dao.tenant.TenantService; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.anyInt; |
||||
|
import static org.mockito.ArgumentMatchers.anyLong; |
||||
|
import static org.mockito.ArgumentMatchers.anyString; |
||||
|
import static org.mockito.ArgumentMatchers.eq; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
import static org.mockito.Mockito.times; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
public class NotificationsCleanUpServiceTest { |
||||
|
|
||||
|
@Mock |
||||
|
private PartitionService partitionService; |
||||
|
@Mock |
||||
|
private SqlPartitioningRepository partitioningRepository; |
||||
|
@Mock |
||||
|
private NotificationRequestDao notificationRequestDao; |
||||
|
@Mock |
||||
|
private TenantService tenantService; |
||||
|
|
||||
|
private NotificationsCleanUpService cleanUpService; |
||||
|
|
||||
|
private static final int BATCH_SIZE = 3; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setUp() { |
||||
|
cleanUpService = new NotificationsCleanUpService(partitionService, partitioningRepository, notificationRequestDao, tenantService); |
||||
|
ReflectionTestUtils.setField(cleanUpService, "ttlInSec", 2592000L); |
||||
|
ReflectionTestUtils.setField(cleanUpService, "partitionSizeInHours", 168); |
||||
|
ReflectionTestUtils.setField(cleanUpService, "removalBatchSize", BATCH_SIZE); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testBatchLoopCallsDaoMultipleTimes() { |
||||
|
TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build(); |
||||
|
when(partitionService.resolve(any(), any(), any())).thenReturn(myPartition); |
||||
|
when(partitioningRepository.dropPartitionsBefore(anyString(), anyLong(), anyLong())) |
||||
|
.thenReturn(System.currentTimeMillis()); |
||||
|
|
||||
|
TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
when(tenantService.findTenantsIds(any())) |
||||
|
.thenReturn(new PageData<>(List.of(tenantId), 1, 1, false)); |
||||
|
|
||||
|
// Sysadmin: returns 3 (full batch), then 1 (partial) -> 2 calls
|
||||
|
when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE))) |
||||
|
.thenReturn(BATCH_SIZE) |
||||
|
.thenReturn(1); |
||||
|
// Tenant: returns 3, 3, 0 -> 3 calls
|
||||
|
when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(tenantId), anyLong(), eq(BATCH_SIZE))) |
||||
|
.thenReturn(BATCH_SIZE) |
||||
|
.thenReturn(BATCH_SIZE) |
||||
|
.thenReturn(0); |
||||
|
|
||||
|
cleanUpService.cleanUp(); |
||||
|
|
||||
|
verify(notificationRequestDao, times(2)) |
||||
|
.removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE)); |
||||
|
verify(notificationRequestDao, times(3)) |
||||
|
.removeByTenantIdAndCreatedTimeBeforeBatch(eq(tenantId), anyLong(), eq(BATCH_SIZE)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testSkipsTenantNotOnMyPartition() { |
||||
|
TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build(); |
||||
|
TopicPartitionInfo notMyPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(false).build(); |
||||
|
when(partitionService.resolve(any(), eq(TenantId.SYS_TENANT_ID), eq(TenantId.SYS_TENANT_ID))) |
||||
|
.thenReturn(myPartition); |
||||
|
when(partitioningRepository.dropPartitionsBefore(anyString(), anyLong(), anyLong())) |
||||
|
.thenReturn(System.currentTimeMillis()); |
||||
|
|
||||
|
// Sysadmin: no records
|
||||
|
when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE))) |
||||
|
.thenReturn(0); |
||||
|
|
||||
|
TenantId myTenant = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
TenantId otherTenant = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
when(tenantService.findTenantsIds(any())) |
||||
|
.thenReturn(new PageData<>(List.of(myTenant, otherTenant), 2, 1, false)); |
||||
|
when(partitionService.resolve(any(), eq(myTenant), eq(myTenant))).thenReturn(myPartition); |
||||
|
when(partitionService.resolve(any(), eq(otherTenant), eq(otherTenant))).thenReturn(notMyPartition); |
||||
|
|
||||
|
when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(myTenant), anyLong(), eq(BATCH_SIZE))) |
||||
|
.thenReturn(0); |
||||
|
|
||||
|
cleanUpService.cleanUp(); |
||||
|
|
||||
|
verify(notificationRequestDao).removeByTenantIdAndCreatedTimeBeforeBatch(eq(myTenant), anyLong(), eq(BATCH_SIZE)); |
||||
|
verify(notificationRequestDao, never()).removeByTenantIdAndCreatedTimeBeforeBatch(eq(otherTenant), anyLong(), anyInt()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testNoPartitionsDropped_stillCleansUpRequests() { |
||||
|
TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build(); |
||||
|
when(partitionService.resolve(any(), any(), any())).thenReturn(myPartition); |
||||
|
when(partitioningRepository.dropPartitionsBefore(anyString(), anyLong(), anyLong())) |
||||
|
.thenReturn(0L); |
||||
|
|
||||
|
when(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE))) |
||||
|
.thenReturn(0); |
||||
|
when(tenantService.findTenantsIds(any())) |
||||
|
.thenReturn(new PageData<>(List.of(), 0, 0, false)); |
||||
|
|
||||
|
cleanUpService.cleanUp(); |
||||
|
|
||||
|
verify(notificationRequestDao).removeByTenantIdAndCreatedTimeBeforeBatch(eq(TenantId.SYS_TENANT_ID), anyLong(), eq(BATCH_SIZE)); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,136 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.rpc; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.junit.jupiter.api.extension.ExtendWith; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
import org.thingsboard.server.common.data.TenantProfile; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.page.PageData; |
||||
|
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; |
||||
|
import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.dao.rpc.RpcDao; |
||||
|
import org.thingsboard.server.dao.tenant.TbTenantProfileCache; |
||||
|
import org.thingsboard.server.dao.tenant.TenantService; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.anyInt; |
||||
|
import static org.mockito.ArgumentMatchers.anyLong; |
||||
|
import static org.mockito.ArgumentMatchers.eq; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
import static org.mockito.Mockito.times; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
public class RpcCleanUpServiceTest { |
||||
|
|
||||
|
@Mock |
||||
|
private PartitionService partitionService; |
||||
|
@Mock |
||||
|
private RpcDao rpcDao; |
||||
|
@Mock |
||||
|
private TenantService tenantService; |
||||
|
@Mock |
||||
|
private TbTenantProfileCache tenantProfileCache; |
||||
|
|
||||
|
private RpcCleanUpService cleanUpService; |
||||
|
|
||||
|
private static final int BATCH_SIZE = 3; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setUp() { |
||||
|
cleanUpService = new RpcCleanUpService(tenantService, partitionService, tenantProfileCache, rpcDao); |
||||
|
ReflectionTestUtils.setField(cleanUpService, "removalBatchSize", BATCH_SIZE); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testBatchLoopCallsDaoMultipleTimes() { |
||||
|
TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
setupTenant(tenantId, 7); |
||||
|
|
||||
|
// Returns 3 (full batch), 3 (full batch), 1 (partial) -> 3 calls
|
||||
|
when(rpcDao.deleteOutdatedRpcByTenantIdBatch(eq(tenantId), anyLong(), eq(BATCH_SIZE))) |
||||
|
.thenReturn(BATCH_SIZE) |
||||
|
.thenReturn(BATCH_SIZE) |
||||
|
.thenReturn(1); |
||||
|
|
||||
|
cleanUpService.cleanUp(); |
||||
|
|
||||
|
verify(rpcDao, times(3)).deleteOutdatedRpcByTenantIdBatch(eq(tenantId), anyLong(), eq(BATCH_SIZE)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testSkipsTenantNotOnMyPartition() { |
||||
|
TenantId myTenant = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
TenantId otherTenant = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
|
||||
|
TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build(); |
||||
|
TopicPartitionInfo notMyPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(false).build(); |
||||
|
|
||||
|
when(tenantService.findTenantsIds(any())) |
||||
|
.thenReturn(new PageData<>(List.of(myTenant, otherTenant), 2, 1, false)); |
||||
|
when(partitionService.resolve(any(), eq(myTenant), eq(myTenant))).thenReturn(myPartition); |
||||
|
when(partitionService.resolve(any(), eq(otherTenant), eq(otherTenant))).thenReturn(notMyPartition); |
||||
|
|
||||
|
setupTenantProfile(myTenant, 7); |
||||
|
when(rpcDao.deleteOutdatedRpcByTenantIdBatch(eq(myTenant), anyLong(), eq(BATCH_SIZE))) |
||||
|
.thenReturn(0); |
||||
|
|
||||
|
cleanUpService.cleanUp(); |
||||
|
|
||||
|
verify(rpcDao).deleteOutdatedRpcByTenantIdBatch(eq(myTenant), anyLong(), eq(BATCH_SIZE)); |
||||
|
verify(rpcDao, never()).deleteOutdatedRpcByTenantIdBatch(eq(otherTenant), anyLong(), anyInt()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testSkipsTenantWithZeroTtl() { |
||||
|
TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
setupTenant(tenantId, 0); |
||||
|
|
||||
|
cleanUpService.cleanUp(); |
||||
|
|
||||
|
verify(rpcDao, never()).deleteOutdatedRpcByTenantIdBatch(any(), anyLong(), anyInt()); |
||||
|
} |
||||
|
|
||||
|
private void setupTenant(TenantId tenantId, int rpcTtlDays) { |
||||
|
TopicPartitionInfo myPartition = TopicPartitionInfo.builder().topic("tb_core").myPartition(true).build(); |
||||
|
when(partitionService.resolve(any(), eq(tenantId), eq(tenantId))).thenReturn(myPartition); |
||||
|
when(tenantService.findTenantsIds(any())) |
||||
|
.thenReturn(new PageData<>(List.of(tenantId), 1, 1, false)); |
||||
|
setupTenantProfile(tenantId, rpcTtlDays); |
||||
|
} |
||||
|
|
||||
|
private void setupTenantProfile(TenantId tenantId, int rpcTtlDays) { |
||||
|
TenantProfile profile = new TenantProfile(); |
||||
|
TenantProfileData profileData = new TenantProfileData(); |
||||
|
DefaultTenantProfileConfiguration config = new DefaultTenantProfileConfiguration(); |
||||
|
config.setRpcTtlDays(rpcTtlDays); |
||||
|
profileData.setConfiguration(config); |
||||
|
profile.setProfileData(profileData); |
||||
|
when(tenantProfileCache.get(tenantId)).thenReturn(profile); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,137 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.transport.lwm2m.server.store; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.junit.jupiter.api.extension.ExtendWith; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
import org.springframework.data.redis.connection.RedisConnection; |
||||
|
import org.springframework.data.redis.connection.RedisConnectionFactory; |
||||
|
import org.springframework.data.redis.core.Cursor; |
||||
|
import org.springframework.data.redis.core.ScanOptions; |
||||
|
import org.thingsboard.server.transport.lwm2m.server.client.LwM2MClientState; |
||||
|
import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.Set; |
||||
|
import java.util.function.Consumer; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.Mockito.doAnswer; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
import static org.thingsboard.server.transport.lwm2m.server.store.util.LwM2MClientSerDes.serialize; |
||||
|
|
||||
|
/** |
||||
|
* Verifies that {@link TbRedisLwM2MClientStore#getAll()} uses separate connections for |
||||
|
* SCAN and GET operations to prevent Jedis 5.x response-ordering corruption that occurs |
||||
|
* when both commands share the same connection. |
||||
|
*/ |
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
class TbRedisLwM2MClientStoreTest { |
||||
|
|
||||
|
@Mock |
||||
|
RedisConnectionFactory connectionFactory; |
||||
|
|
||||
|
@Mock |
||||
|
RedisConnection scanConnection; |
||||
|
|
||||
|
@Mock |
||||
|
RedisConnection getConnection; |
||||
|
|
||||
|
TbRedisLwM2MClientStore store; |
||||
|
|
||||
|
@BeforeEach |
||||
|
void setUp() { |
||||
|
// First getConnection() call → scanConnection, second → getConnection
|
||||
|
when(connectionFactory.getConnection()) |
||||
|
.thenReturn(scanConnection) |
||||
|
.thenReturn(getConnection); |
||||
|
store = new TbRedisLwM2MClientStore(connectionFactory); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void getAll_returnsSingleClient() { |
||||
|
LwM2mClient client = new LwM2mClient("nodeId", "testEndpoint"); |
||||
|
client.setState(LwM2MClientState.REGISTERED); |
||||
|
byte[] key = "CLIENT#EP#testEndpoint".getBytes(); |
||||
|
byte[] value = serialize(client); |
||||
|
|
||||
|
// Cursor created before thenReturn to avoid Mockito unfinished-stubbing error
|
||||
|
Cursor<byte[]> cursor = cursorOf(key); |
||||
|
when(scanConnection.scan(any(ScanOptions.class))).thenReturn(cursor); |
||||
|
when(getConnection.get(key)).thenReturn(value); |
||||
|
|
||||
|
Set<LwM2mClient> result = store.getAll(); |
||||
|
|
||||
|
assertThat(result).hasSize(1); |
||||
|
assertThat(result.iterator().next().getEndpoint()).isEqualTo("testEndpoint"); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void getAll_getIsNeverCalledOnScanConnection() { |
||||
|
Cursor<byte[]> cursor = cursorOf(); |
||||
|
when(scanConnection.scan(any(ScanOptions.class))).thenReturn(cursor); |
||||
|
|
||||
|
store.getAll(); |
||||
|
|
||||
|
verify(scanConnection, never()).get(any(byte[].class)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void getAll_scanIsNeverCalledOnGetConnection() { |
||||
|
Cursor<byte[]> cursor = cursorOf(); |
||||
|
when(scanConnection.scan(any(ScanOptions.class))).thenReturn(cursor); |
||||
|
|
||||
|
store.getAll(); |
||||
|
|
||||
|
verify(getConnection, never()).scan(any(ScanOptions.class)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void getAll_skipsKeyWhenValueIsNull() { |
||||
|
byte[] key = "CLIENT#EP#gone".getBytes(); |
||||
|
Cursor<byte[]> cursor = cursorOf(key); |
||||
|
when(scanConnection.scan(any(ScanOptions.class))).thenReturn(cursor); |
||||
|
// getConnection.get(key) returns null by default — no stubbing needed
|
||||
|
|
||||
|
Set<LwM2mClient> result = store.getAll(); |
||||
|
|
||||
|
assertThat(result).isEmpty(); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Creates a mock {@link Cursor} that iterates over the given keys via {@code forEachRemaining}. |
||||
|
* The cursor is created separately (not inside a {@code thenReturn()} argument) to avoid |
||||
|
* Mockito's "unfinished stubbing" error caused by nested {@code when()} calls. |
||||
|
*/ |
||||
|
@SuppressWarnings("unchecked") |
||||
|
private static Cursor<byte[]> cursorOf(byte[]... keys) { |
||||
|
Cursor<byte[]> cursor = mock(Cursor.class); |
||||
|
List<byte[]> keyList = List.of(keys); |
||||
|
doAnswer(inv -> { |
||||
|
Consumer<byte[]> action = inv.getArgument(0); |
||||
|
keyList.forEach(action); |
||||
|
return null; |
||||
|
}).when(cursor).forEachRemaining(any(Consumer.class)); |
||||
|
return cursor; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,133 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 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.sql.notification; |
||||
|
|
||||
|
import org.junit.After; |
||||
|
import org.junit.Test; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.thingsboard.server.common.data.id.NotificationRequestId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.notification.NotificationRequest; |
||||
|
import org.thingsboard.server.common.data.notification.NotificationRequestStatus; |
||||
|
import org.thingsboard.server.dao.AbstractJpaDaoTest; |
||||
|
|
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
|
||||
|
public class JpaNotificationRequestDaoTest extends AbstractJpaDaoTest { |
||||
|
|
||||
|
@Autowired |
||||
|
JpaNotificationRequestDao notificationRequestDao; |
||||
|
|
||||
|
private final List<NotificationRequest> createdRequests = new ArrayList<>(); |
||||
|
|
||||
|
@After |
||||
|
public void tearDown() { |
||||
|
for (NotificationRequest request : createdRequests) { |
||||
|
notificationRequestDao.removeById(request.getTenantId(), request.getId().getId()); |
||||
|
} |
||||
|
createdRequests.clear(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testBatchDeletion() { |
||||
|
TenantId sysTenantId = TenantId.SYS_TENANT_ID; |
||||
|
long now = System.currentTimeMillis(); |
||||
|
long oldTimestamp = now - TimeUnit.DAYS.toMillis(30); |
||||
|
|
||||
|
NotificationRequest oldRequest1 = createNotificationRequest(sysTenantId, oldTimestamp); |
||||
|
notificationRequestDao.save(sysTenantId, oldRequest1); |
||||
|
|
||||
|
NotificationRequest oldRequest2 = createNotificationRequest(sysTenantId, oldTimestamp); |
||||
|
notificationRequestDao.save(sysTenantId, oldRequest2); |
||||
|
|
||||
|
NotificationRequest freshRequest = createNotificationRequest(sysTenantId, now); |
||||
|
notificationRequestDao.save(sysTenantId, freshRequest); |
||||
|
|
||||
|
TenantId tenant2Id = TenantId.fromUUID(UUID.fromString("3d193a7a-774b-4c05-84d5-f7fdcf7a37cf")); |
||||
|
NotificationRequest tenant2Request = createNotificationRequest(tenant2Id, oldTimestamp); |
||||
|
notificationRequestDao.save(tenant2Id, tenant2Request); |
||||
|
|
||||
|
int batchSize = 10_000; |
||||
|
|
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(sysTenantId, oldTimestamp - 1, batchSize)).isEqualTo(0); |
||||
|
|
||||
|
long expirationTime = now - TimeUnit.DAYS.toMillis(15); |
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(sysTenantId, expirationTime, batchSize)).isEqualTo(2); |
||||
|
|
||||
|
assertThat(notificationRequestDao.findById(sysTenantId, freshRequest.getId().getId())).isNotNull(); |
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenant2Id, now + 1, batchSize)).isEqualTo(1); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testBatchDeletionWithSmallBatchSize() { |
||||
|
TenantId tenantId = TenantId.SYS_TENANT_ID; |
||||
|
long oldTimestamp = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30); |
||||
|
|
||||
|
for (int i = 0; i < 10; i++) { |
||||
|
NotificationRequest request = createNotificationRequest(tenantId, oldTimestamp); |
||||
|
notificationRequestDao.save(tenantId, request); |
||||
|
} |
||||
|
|
||||
|
int batchSize = 3; |
||||
|
long expirationTime = System.currentTimeMillis(); |
||||
|
|
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(3); |
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(3); |
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(3); |
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(1); |
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenantId, expirationTime, batchSize)).isEqualTo(0); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testBatchDeletionIsolationBetweenTenants() { |
||||
|
TenantId tenant1 = TenantId.SYS_TENANT_ID; |
||||
|
TenantId tenant2 = TenantId.fromUUID(UUID.fromString("3d193a7a-774b-4c05-84d5-f7fdcf7a37cf")); |
||||
|
long oldTimestamp = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30); |
||||
|
|
||||
|
for (int i = 0; i < 5; i++) { |
||||
|
NotificationRequest request = createNotificationRequest(tenant1, oldTimestamp); |
||||
|
notificationRequestDao.save(tenant1, request); |
||||
|
} |
||||
|
|
||||
|
for (int i = 0; i < 3; i++) { |
||||
|
NotificationRequest request = createNotificationRequest(tenant2, oldTimestamp); |
||||
|
notificationRequestDao.save(tenant2, request); |
||||
|
} |
||||
|
|
||||
|
int batchSize = 10_000; |
||||
|
long expirationTime = System.currentTimeMillis(); |
||||
|
|
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenant1, expirationTime, batchSize)).isEqualTo(5); |
||||
|
assertThat(notificationRequestDao.removeByTenantIdAndCreatedTimeBeforeBatch(tenant2, expirationTime, batchSize)).isEqualTo(3); |
||||
|
} |
||||
|
|
||||
|
private NotificationRequest createNotificationRequest(TenantId tenantId, long createdTime) { |
||||
|
NotificationRequest request = new NotificationRequest(); |
||||
|
request.setId(new NotificationRequestId(UUID.randomUUID())); |
||||
|
request.setTenantId(tenantId); |
||||
|
request.setCreatedTime(createdTime); |
||||
|
request.setTargets(List.of(UUID.randomUUID())); |
||||
|
request.setStatus(NotificationRequestStatus.SENT); |
||||
|
createdRequests.add(request); |
||||
|
return request; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,458 @@ |
|||||
|
{ |
||||
|
"annotations": { |
||||
|
"list": [ |
||||
|
{ |
||||
|
"builtIn": 1, |
||||
|
"datasource": { |
||||
|
"type": "datasource", |
||||
|
"uid": "grafana" |
||||
|
}, |
||||
|
"enable": true, |
||||
|
"hide": true, |
||||
|
"iconColor": "rgba(0, 211, 255, 1)", |
||||
|
"name": "Annotations & Alerts", |
||||
|
"type": "dashboard" |
||||
|
} |
||||
|
] |
||||
|
}, |
||||
|
"editable": true, |
||||
|
"fiscalYearStartMonth": 0, |
||||
|
"graphTooltip": 0, |
||||
|
"id": null, |
||||
|
"links": [], |
||||
|
"panels": [ |
||||
|
{ |
||||
|
"datasource": { |
||||
|
"type": "prometheus", |
||||
|
"uid": "prometheus" |
||||
|
}, |
||||
|
"description": "", |
||||
|
"fieldConfig": { |
||||
|
"defaults": { |
||||
|
"color": { |
||||
|
"mode": "palette-classic" |
||||
|
}, |
||||
|
"custom": { |
||||
|
"axisBorderShow": false, |
||||
|
"axisCenteredZero": false, |
||||
|
"axisColorMode": "text", |
||||
|
"axisLabel": "", |
||||
|
"axisPlacement": "auto", |
||||
|
"barAlignment": 0, |
||||
|
"drawStyle": "line", |
||||
|
"fillOpacity": 10, |
||||
|
"gradientMode": "none", |
||||
|
"hideFrom": { |
||||
|
"legend": false, |
||||
|
"tooltip": false, |
||||
|
"viz": false |
||||
|
}, |
||||
|
"insertNulls": false, |
||||
|
"lineInterpolation": "linear", |
||||
|
"lineWidth": 1, |
||||
|
"pointSize": 5, |
||||
|
"scaleDistribution": { |
||||
|
"type": "linear" |
||||
|
}, |
||||
|
"showPoints": "never", |
||||
|
"spanNulls": true, |
||||
|
"stacking": { |
||||
|
"group": "A", |
||||
|
"mode": "none" |
||||
|
}, |
||||
|
"thresholdsStyle": { |
||||
|
"mode": "off" |
||||
|
} |
||||
|
}, |
||||
|
"mappings": [], |
||||
|
"thresholds": { |
||||
|
"mode": "absolute", |
||||
|
"steps": [ |
||||
|
{ |
||||
|
"color": "dark-orange", |
||||
|
"value": null |
||||
|
} |
||||
|
] |
||||
|
}, |
||||
|
"unit": "short" |
||||
|
}, |
||||
|
"overrides": [ |
||||
|
{ |
||||
|
"matcher": { |
||||
|
"id": "byName", |
||||
|
"options": "Total connections" |
||||
|
}, |
||||
|
"properties": [ |
||||
|
{ |
||||
|
"id": "color", |
||||
|
"value": { |
||||
|
"fixedColor": "light-orange", |
||||
|
"mode": "fixed" |
||||
|
} |
||||
|
} |
||||
|
] |
||||
|
} |
||||
|
] |
||||
|
}, |
||||
|
"gridPos": { |
||||
|
"h": 15, |
||||
|
"w": 19, |
||||
|
"x": 0, |
||||
|
"y": 0 |
||||
|
}, |
||||
|
"id": 19, |
||||
|
"options": { |
||||
|
"legend": { |
||||
|
"calcs": [], |
||||
|
"displayMode": "list", |
||||
|
"placement": "bottom", |
||||
|
"showLegend": true |
||||
|
}, |
||||
|
"tooltip": { |
||||
|
"mode": "multi", |
||||
|
"sort": "none" |
||||
|
} |
||||
|
}, |
||||
|
"targets": [ |
||||
|
{ |
||||
|
"datasource": { |
||||
|
"type": "prometheus", |
||||
|
"uid": "prometheus" |
||||
|
}, |
||||
|
"editorMode": "code", |
||||
|
"exemplar": false, |
||||
|
"expr": "sum by (protocol) (thingsboard_transport_connections_active{})", |
||||
|
"format": "time_series", |
||||
|
"instant": false, |
||||
|
"interval": "", |
||||
|
"intervalFactor": 1, |
||||
|
"legendFormat": "{{protocol}}", |
||||
|
"refId": "A" |
||||
|
} |
||||
|
], |
||||
|
"title": "Active Connections by Protocol", |
||||
|
"type": "timeseries" |
||||
|
}, |
||||
|
{ |
||||
|
"datasource": { |
||||
|
"type": "prometheus", |
||||
|
"uid": "prometheus" |
||||
|
}, |
||||
|
"fieldConfig": { |
||||
|
"defaults": { |
||||
|
"color": { |
||||
|
"mode": "thresholds" |
||||
|
}, |
||||
|
"mappings": [], |
||||
|
"thresholds": { |
||||
|
"mode": "absolute", |
||||
|
"steps": [ |
||||
|
{ |
||||
|
"color": "rgba(209, 138, 78, 1)", |
||||
|
"value": null |
||||
|
} |
||||
|
] |
||||
|
} |
||||
|
}, |
||||
|
"overrides": [] |
||||
|
}, |
||||
|
"gridPos": { |
||||
|
"h": 15, |
||||
|
"w": 5, |
||||
|
"x": 19, |
||||
|
"y": 0 |
||||
|
}, |
||||
|
"id": 21, |
||||
|
"options": { |
||||
|
"colorMode": "value", |
||||
|
"graphMode": "none", |
||||
|
"justifyMode": "auto", |
||||
|
"orientation": "auto", |
||||
|
"percentChangeColorMode": "standard", |
||||
|
"reduceOptions": { |
||||
|
"calcs": [ |
||||
|
"lastNotNull" |
||||
|
], |
||||
|
"fields": "", |
||||
|
"values": false |
||||
|
}, |
||||
|
"showPercentChange": false, |
||||
|
"text": {}, |
||||
|
"textMode": "auto", |
||||
|
"wideLayout": true |
||||
|
}, |
||||
|
"targets": [ |
||||
|
{ |
||||
|
"datasource": { |
||||
|
"type": "prometheus", |
||||
|
"uid": "prometheus" |
||||
|
}, |
||||
|
"editorMode": "code", |
||||
|
"exemplar": true, |
||||
|
"expr": "sum (thingsboard_transport_connections_active)", |
||||
|
"instant": true, |
||||
|
"interval": "", |
||||
|
"legendFormat": "", |
||||
|
"refId": "A" |
||||
|
} |
||||
|
], |
||||
|
"title": "Total Connections", |
||||
|
"type": "stat" |
||||
|
}, |
||||
|
{ |
||||
|
"datasource": { |
||||
|
"type": "prometheus", |
||||
|
"uid": "prometheus" |
||||
|
}, |
||||
|
"description": "", |
||||
|
"fieldConfig": { |
||||
|
"defaults": { |
||||
|
"color": { |
||||
|
"mode": "palette-classic" |
||||
|
}, |
||||
|
"custom": { |
||||
|
"axisBorderShow": false, |
||||
|
"axisCenteredZero": false, |
||||
|
"axisColorMode": "text", |
||||
|
"axisLabel": "", |
||||
|
"axisPlacement": "auto", |
||||
|
"barAlignment": 0, |
||||
|
"drawStyle": "line", |
||||
|
"fillOpacity": 10, |
||||
|
"gradientMode": "none", |
||||
|
"hideFrom": { |
||||
|
"legend": false, |
||||
|
"tooltip": false, |
||||
|
"viz": false |
||||
|
}, |
||||
|
"insertNulls": false, |
||||
|
"lineInterpolation": "linear", |
||||
|
"lineWidth": 1, |
||||
|
"pointSize": 5, |
||||
|
"scaleDistribution": { |
||||
|
"type": "linear" |
||||
|
}, |
||||
|
"showPoints": "never", |
||||
|
"spanNulls": true, |
||||
|
"stacking": { |
||||
|
"group": "A", |
||||
|
"mode": "none" |
||||
|
}, |
||||
|
"thresholdsStyle": { |
||||
|
"mode": "off" |
||||
|
} |
||||
|
}, |
||||
|
"mappings": [], |
||||
|
"thresholds": { |
||||
|
"mode": "absolute", |
||||
|
"steps": [ |
||||
|
{ |
||||
|
"color": "green", |
||||
|
"value": null |
||||
|
}, |
||||
|
{ |
||||
|
"color": "red", |
||||
|
"value": 80 |
||||
|
} |
||||
|
] |
||||
|
}, |
||||
|
"unit": "short" |
||||
|
}, |
||||
|
"overrides": [] |
||||
|
}, |
||||
|
"gridPos": { |
||||
|
"h": 15, |
||||
|
"w": 19, |
||||
|
"x": 0, |
||||
|
"y": 15 |
||||
|
}, |
||||
|
"id": 22, |
||||
|
"options": { |
||||
|
"legend": { |
||||
|
"calcs": [], |
||||
|
"displayMode": "list", |
||||
|
"placement": "bottom", |
||||
|
"showLegend": true |
||||
|
}, |
||||
|
"tooltip": { |
||||
|
"mode": "multi", |
||||
|
"sort": "none" |
||||
|
} |
||||
|
}, |
||||
|
"targets": [ |
||||
|
{ |
||||
|
"datasource": { |
||||
|
"type": "prometheus", |
||||
|
"uid": "prometheus" |
||||
|
}, |
||||
|
"editorMode": "code", |
||||
|
"exemplar": true, |
||||
|
"expr": "thingsboard_transport_connections_active", |
||||
|
"format": "time_series", |
||||
|
"interval": "", |
||||
|
"intervalFactor": 1, |
||||
|
"legendFormat": "{{protocol}} {{pod_name}}", |
||||
|
"range": true, |
||||
|
"refId": "A" |
||||
|
} |
||||
|
], |
||||
|
"title": "Active Connections by Pod", |
||||
|
"type": "timeseries" |
||||
|
}, |
||||
|
{ |
||||
|
"datasource": { |
||||
|
"type": "prometheus", |
||||
|
"uid": "prometheus" |
||||
|
}, |
||||
|
"fieldConfig": { |
||||
|
"defaults": { |
||||
|
"color": { |
||||
|
"mode": "thresholds" |
||||
|
}, |
||||
|
"custom": { |
||||
|
"align": "auto", |
||||
|
"cellOptions": { |
||||
|
"type": "auto" |
||||
|
}, |
||||
|
"filterable": false, |
||||
|
"inspect": false |
||||
|
}, |
||||
|
"mappings": [], |
||||
|
"min": 0, |
||||
|
"thresholds": { |
||||
|
"mode": "absolute", |
||||
|
"steps": [ |
||||
|
{ |
||||
|
"color": "rgba(209, 138, 78, 1)", |
||||
|
"value": null |
||||
|
} |
||||
|
] |
||||
|
} |
||||
|
}, |
||||
|
"overrides": [ |
||||
|
{ |
||||
|
"matcher": { |
||||
|
"id": "byName", |
||||
|
"options": "Connections" |
||||
|
}, |
||||
|
"properties": [ |
||||
|
{ |
||||
|
"id": "custom.cellOptions", |
||||
|
"value": { |
||||
|
"mode": "gradient", |
||||
|
"type": "gauge" |
||||
|
} |
||||
|
} |
||||
|
] |
||||
|
} |
||||
|
] |
||||
|
}, |
||||
|
"gridPos": { |
||||
|
"h": 15, |
||||
|
"w": 5, |
||||
|
"x": 19, |
||||
|
"y": 15 |
||||
|
}, |
||||
|
"id": 23, |
||||
|
"options": { |
||||
|
"cellHeight": "sm", |
||||
|
"footer": { |
||||
|
"countRows": false, |
||||
|
"fields": "", |
||||
|
"reducer": [ |
||||
|
"sum" |
||||
|
], |
||||
|
"show": false |
||||
|
}, |
||||
|
"showHeader": true, |
||||
|
"sortBy": [] |
||||
|
}, |
||||
|
"targets": [ |
||||
|
{ |
||||
|
"datasource": { |
||||
|
"type": "prometheus", |
||||
|
"uid": "prometheus" |
||||
|
}, |
||||
|
"editorMode": "code", |
||||
|
"exemplar": true, |
||||
|
"expr": "sum by (pod_name) (thingsboard_transport_connections_active)", |
||||
|
"format": "table", |
||||
|
"instant": true, |
||||
|
"interval": "", |
||||
|
"legendFormat": "__auto", |
||||
|
"queryType": "randomWalk", |
||||
|
"refId": "A" |
||||
|
} |
||||
|
], |
||||
|
"title": "Connection Count by Pod", |
||||
|
"transformations": [ |
||||
|
{ |
||||
|
"id": "organize", |
||||
|
"options": { |
||||
|
"excludeByName": { |
||||
|
"Time": true, |
||||
|
"__name__": true, |
||||
|
"app": true, |
||||
|
"controller_revision_hash": true, |
||||
|
"instance": true, |
||||
|
"job": true, |
||||
|
"kubernetes_name": true, |
||||
|
"kubernetes_namespace": true, |
||||
|
"node_name": true, |
||||
|
"pod_name": false, |
||||
|
"statefulset_kubernetes_io_pod_name": true |
||||
|
}, |
||||
|
"indexByName": {}, |
||||
|
"renameByName": { |
||||
|
"Time": "", |
||||
|
"Value": "Connections", |
||||
|
"pod_name": "Pod" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
{ |
||||
|
"id": "sortBy", |
||||
|
"options": { |
||||
|
"fields": {}, |
||||
|
"sort": [ |
||||
|
{ |
||||
|
"field": "Pod" |
||||
|
} |
||||
|
] |
||||
|
} |
||||
|
} |
||||
|
], |
||||
|
"type": "table" |
||||
|
} |
||||
|
], |
||||
|
"refresh": "30s", |
||||
|
"schemaVersion": 39, |
||||
|
"tags": [ |
||||
|
"thingsboard" |
||||
|
], |
||||
|
"templating": { |
||||
|
"list": [] |
||||
|
}, |
||||
|
"time": { |
||||
|
"from": "now-1h", |
||||
|
"to": "now" |
||||
|
}, |
||||
|
"timepicker": { |
||||
|
"refresh_intervals": [ |
||||
|
"10s", |
||||
|
"30s", |
||||
|
"1m", |
||||
|
"5m", |
||||
|
"15m", |
||||
|
"30m", |
||||
|
"1h", |
||||
|
"2h", |
||||
|
"1d" |
||||
|
] |
||||
|
}, |
||||
|
"timezone": "", |
||||
|
"title": "Transport Connections", |
||||
|
"uid": "lewbrlwjerwkj3", |
||||
|
"version": 1, |
||||
|
"weekStart": "" |
||||
|
} |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue