diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index a230f95cae..a8443709f4 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -197,15 +197,15 @@ public class DefaultDeviceStateService implements DeviceStateService { if (lastReportedActivity > 0 && lastReportedActivity > lastSavedActivity) { DeviceStateData stateData = getOrFetchDeviceStateData(deviceId); if (stateData != null) { - DeviceState state = stateData.getState(); - stateData.getState().setLastActivityTime(lastReportedActivity); - stateData.getMetaData().putValue("scope", SERVER_SCOPE); - pushRuleEngineMessage(stateData, ACTIVITY_EVENT); save(deviceId, LAST_ACTIVITY_TIME, lastReportedActivity); deviceLastSavedActivity.put(deviceId, lastReportedActivity); + DeviceState state = stateData.getState(); + state.setLastActivityTime(lastReportedActivity); if (!state.isActive()) { state.setActive(true); save(deviceId, ACTIVITY_STATE, state.isActive()); + stateData.getMetaData().putValue("scope", SERVER_SCOPE); + pushRuleEngineMessage(stateData, ACTIVITY_EVENT); } } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 71af1d7c86..cafed53f90 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -617,6 +617,16 @@ queue: max_poll_records: "${TB_QUEUE_KAFKA_MAX_POLL_RECORDS:8192}" max_partition_fetch_bytes: "${TB_QUEUE_KAFKA_MAX_PARTITION_FETCH_BYTES:16777216}" fetch_max_bytes: "${TB_QUEUE_KAFKA_FETCH_MAX_BYTES:134217728}" + other: +# Properties for Confluent cloud +# - key: "ssl.endpoint.identification.algorithm" +# value: "https" +# - key: "sasl.mechanism" +# value: "PLAIN" +# - key: "sasl.jaas.config" +# value: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"CLUSTER_API_KEY\" password=\"CLUSTER_API_SECRET\";" +# - key: "security.protocol" +# value: "SASL_SSL" topic-properties: rule-engine: "${TB_QUEUE_KAFKA_RE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" core: "${TB_QUEUE_KAFKA_CORE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" @@ -624,6 +634,7 @@ queue: notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000}" js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600}" aws_sqs: + use_default_credential_provider_chain: "${TB_QUEUE_AWS_SQS_USE_DEFAULT_CREDENTIAL_PROVIDER_CHAIN:false}" access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}" secret_access_key: "${TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY:YOUR_SECRET}" region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java index 659dd19bda..d9da969eb9 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java @@ -16,10 +16,12 @@ package org.thingsboard.server.queue.kafka; import lombok.Getter; +import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.producer.ProducerConfig; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; import java.util.List; @@ -30,6 +32,7 @@ import java.util.Properties; */ @Slf4j @ConditionalOnExpression("'${queue.type:null}'=='kafka'") +@ConfigurationProperties(prefix = "queue.kafka") @Component public class TbKafkaSettings { @@ -67,7 +70,7 @@ public class TbKafkaSettings { @Getter private int fetchMaxBytes; - @Value("${kafka.other:#{null}}") + @Setter private List other; public Properties toProps() { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java index 8e293e6dff..f99755a0af 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java @@ -16,8 +16,10 @@ package org.thingsboard.server.queue.sqs; import com.amazonaws.auth.AWSCredentials; +import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.auth.AWSStaticCredentialsProvider; import com.amazonaws.auth.BasicAWSCredentials; +import com.amazonaws.auth.DefaultAWSCredentialsProviderChain; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClientBuilder; import com.amazonaws.services.sqs.model.CreateQueueRequest; @@ -37,9 +39,16 @@ public class TbAwsSqsAdmin implements TbQueueAdmin { public TbAwsSqsAdmin(TbAwsSqsSettings sqsSettings, Map attributes) { this.attributes = attributes; - AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey()); + AWSCredentialsProvider credentialsProvider; + if (sqsSettings.getUseDefaultCredentialProviderChain()) { + credentialsProvider = new DefaultAWSCredentialsProviderChain(); + } else { + AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey()); + credentialsProvider = new AWSStaticCredentialsProvider(awsCredentials); + } + sqsClient = AmazonSQSClientBuilder.standard() - .withCredentials(new AWSStaticCredentialsProvider(awsCredentials)) + .withCredentials(credentialsProvider) .withRegion(sqsSettings.getRegion()) .build(); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsConsumerTemplate.java index f4f279a02b..3e30123980 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsConsumerTemplate.java @@ -16,8 +16,10 @@ package org.thingsboard.server.queue.sqs; import com.amazonaws.auth.AWSCredentials; +import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.auth.AWSStaticCredentialsProvider; import com.amazonaws.auth.BasicAWSCredentials; +import com.amazonaws.auth.DefaultAWSCredentialsProviderChain; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClientBuilder; import com.amazonaws.services.sqs.model.DeleteMessageBatchRequestEntry; @@ -67,13 +69,19 @@ public class TbAwsSqsConsumerTemplate extends AbstractPara this.decoder = decoder; this.sqsSettings = sqsSettings; - AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey()); - AWSStaticCredentialsProvider credProvider = new AWSStaticCredentialsProvider(awsCredentials); + AWSCredentialsProvider credentialsProvider; + if (sqsSettings.getUseDefaultCredentialProviderChain()) { + credentialsProvider = new DefaultAWSCredentialsProviderChain(); + } else { + AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey()); + credentialsProvider = new AWSStaticCredentialsProvider(awsCredentials); + } - this.sqsClient = AmazonSQSClientBuilder.standard() - .withCredentials(credProvider) + sqsClient = AmazonSQSClientBuilder.standard() + .withCredentials(credentialsProvider) .withRegion(sqsSettings.getRegion()) .build(); + } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java index 6110d08c5e..3e508b09e6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java @@ -16,8 +16,10 @@ package org.thingsboard.server.queue.sqs; import com.amazonaws.auth.AWSCredentials; +import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.auth.AWSStaticCredentialsProvider; import com.amazonaws.auth.BasicAWSCredentials; +import com.amazonaws.auth.DefaultAWSCredentialsProviderChain; import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClientBuilder; import com.amazonaws.services.sqs.model.SendMessageRequest; @@ -54,14 +56,18 @@ public class TbAwsSqsProducerTemplate implements TbQueuePr this.admin = admin; this.defaultTopic = defaultTopic; - AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey()); - AWSStaticCredentialsProvider credProvider = new AWSStaticCredentialsProvider(awsCredentials); + AWSCredentialsProvider credentialsProvider; + if (sqsSettings.getUseDefaultCredentialProviderChain()) { + credentialsProvider = new DefaultAWSCredentialsProviderChain(); + } else { + AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey()); + credentialsProvider = new AWSStaticCredentialsProvider(awsCredentials); + } - this.sqsClient = AmazonSQSClientBuilder.standard() - .withCredentials(credProvider) + sqsClient = AmazonSQSClientBuilder.standard() + .withCredentials(credentialsProvider) .withRegion(sqsSettings.getRegion()) .build(); - producerExecutor = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool()); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java index 922a2b1062..7a5c3332ad 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java @@ -27,6 +27,9 @@ import org.springframework.stereotype.Component; @Data public class TbAwsSqsSettings { + @Value("${queue.aws_sqs.use_default_credential_provider_chain}") + private Boolean useDefaultCredentialProviderChain; + @Value("${queue.aws_sqs.access_key_id}") private String accessKeyId; diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttSslHandlerProvider.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttSslHandlerProvider.java index 4e3a39699a..0c872c351f 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttSslHandlerProvider.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttSslHandlerProvider.java @@ -53,7 +53,7 @@ import java.util.concurrent.TimeUnit; */ @Slf4j @Component("MqttSslHandlerProvider") -@ConditionalOnExpression("'${transport.type:null}'=='null' || ('${transport.type}'=='local' && '${transport.http.enabled}'=='true')") +@ConditionalOnExpression("'${transport.type:null}'=='null' || ('${transport.type}'=='local' && '${transport.mqtt.enabled}'=='true')") @ConditionalOnProperty(prefix = "transport.mqtt.ssl", value = "enabled", havingValue = "true", matchIfMissing = false) public class MqttSslHandlerProvider { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java index 45370a2c89..caa627451f 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java @@ -207,6 +207,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { try { return new JsonParser().parse(payload); } catch (JsonSyntaxException ex) { + log.warn("Payload is in incorrect format: {}", payload); throw new AdaptorException(ex); } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java index 83b7e752e8..7de0339800 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java @@ -553,7 +553,13 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository { String lowerSearchText = "%" + searchText.toLowerCase() + "%"; ctx.addStringParameter("lowerSearchTextParam", lowerSearchText); List searchAliases = selectionMapping.stream().map(EntityKeyMapping::getValueAlias).collect(Collectors.toList()); - return String.format(" WHERE LOWER(CONCAT(%s)) LIKE :%s", String.join(" , ", searchAliases), "lowerSearchTextParam"); + String searchAliasesExpression; + if (searchAliases.size() > 1) { + searchAliasesExpression = "CONCAT(" + String.join(" , ", searchAliases) + ")"; + } else { + searchAliasesExpression = searchAliases.get(0); + } + return String.format(" WHERE LOWER(%s) LIKE :%s", searchAliasesExpression, "lowerSearchTextParam"); } else { return ""; } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java index 071ed9ca08..30bf2d4d6e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java @@ -18,7 +18,6 @@ package org.thingsboard.rule.engine.action; import com.fasterxml.jackson.databind.JsonNode; 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.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -59,7 +58,7 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode processAlarm(TbContext ctx, TbMsg msg) { String alarmType = TbNodeUtils.processPattern(this.config.getAlarmType(), msg.getMetaData()); ListenableFuture alarmFuture; - if(msg.getOriginator().getEntityType().equals(EntityType.ALARM)){ + if (msg.getOriginator().getEntityType().equals(EntityType.ALARM)) { alarmFuture = ctx.getAlarmService().findAlarmByIdAsync(ctx.getTenantId(), new AlarmId(msg.getOriginator().getId())); } else { alarmFuture = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), alarmType); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java index 6996a1117a..ae4fb3df1e 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java @@ -36,6 +36,7 @@ import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleChainId; @@ -97,6 +98,7 @@ public class TbAlarmNodeTest { private ListeningExecutor dbExecutor; private EntityId originator = new DeviceId(Uuids.timeBased()); + private EntityId alarmOriginator = new AlarmId(Uuids.timeBased()); private TenantId tenantId = new TenantId(Uuids.timeBased()); private TbMsgMetaData metaData = new TbMsgMetaData(); private String rawJson = "{\"name\": \"Vit\", \"passed\": 5}"; @@ -328,6 +330,55 @@ public class TbAlarmNodeTest { assertEquals(expectedAlarm, actualAlarm); } + @Test + public void alarmCanBeClearedWithAlarmOriginator() throws ScriptException, IOException { + initWithClearAlarmScript(); + metaData.putValue("key", "value"); + TbMsg msg = TbMsg.newMsg( "USER", alarmOriginator, metaData, TbMsgDataType.JSON, rawJson, ruleChainId, ruleNodeId); + + long oldEndDate = System.currentTimeMillis(); + AlarmId id = new AlarmId(alarmOriginator.getId()); + Alarm activeAlarm = Alarm.builder().type("SomeType").tenantId(tenantId).originator(originator).status(ACTIVE_UNACK).severity(WARNING).endTs(oldEndDate).build(); + activeAlarm.setId(id); + + when(detailsJs.executeJsonAsync(msg)).thenReturn(Futures.immediateFuture(null)); + when(alarmService.findAlarmByIdAsync(tenantId, id)).thenReturn(Futures.immediateFuture(activeAlarm)); + when(alarmService.clearAlarm(eq(activeAlarm.getTenantId()), eq(activeAlarm.getId()), org.mockito.Mockito.any(JsonNode.class), anyLong())).thenReturn(Futures.immediateFuture(true)); +// doAnswer((Answer) invocationOnMock -> (Alarm) (invocationOnMock.getArguments())[0]).when(alarmService).createOrUpdateAlarm(activeAlarm); + + node.onMsg(ctx, msg); + + verify(ctx).tellNext(any(), eq("Cleared")); + + ArgumentCaptor msgCaptor = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor typeCaptor = ArgumentCaptor.forClass(String.class); + ArgumentCaptor originatorCaptor = ArgumentCaptor.forClass(EntityId.class); + ArgumentCaptor metadataCaptor = ArgumentCaptor.forClass(TbMsgMetaData.class); + ArgumentCaptor dataCaptor = ArgumentCaptor.forClass(String.class); + verify(ctx).transformMsg(msgCaptor.capture(), typeCaptor.capture(), originatorCaptor.capture(), metadataCaptor.capture(), dataCaptor.capture()); + + assertEquals("ALARM", typeCaptor.getValue()); + assertEquals(alarmOriginator, originatorCaptor.getValue()); + assertEquals("value", metadataCaptor.getValue().getValue("key")); + assertEquals(Boolean.TRUE.toString(), metadataCaptor.getValue().getValue(IS_CLEARED_ALARM)); + assertNotSame(metaData, metadataCaptor.getValue()); + + Alarm actualAlarm = new ObjectMapper().readValue(dataCaptor.getValue().getBytes(), Alarm.class); + Alarm expectedAlarm = Alarm.builder() + .tenantId(tenantId) + .originator(originator) + .status(CLEARED_UNACK) + .severity(WARNING) + .propagate(false) + .type("SomeType") + .details(null) + .endTs(oldEndDate) + .build(); + expectedAlarm.setId(id); + + assertEquals(expectedAlarm, actualAlarm); + } + private void initWithCreateAlarmScript() { try { TbCreateAlarmNodeConfiguration config = new TbCreateAlarmNodeConfiguration();