diff --git a/application/src/main/resources/logback.xml b/application/src/main/resources/logback.xml index f8085ec985..61caeb61fb 100644 --- a/application/src/main/resources/logback.xml +++ b/application/src/main/resources/logback.xml @@ -26,6 +26,7 @@ + diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index c9ed9d8890..21c0b9f970 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -648,6 +648,11 @@ transport: redis.enabled: "${LWM2M_REDIS_ENABLED:false}" snmp: enabled: "${SNMP_ENABLED:true}" + response_processing: + # parallelism level for executor (workStealingPool) that is responsible for handling responses from SNMP devices + parallelism_level: "${SNMP_RESPONSE_PROCESSING_PARALLELISM_LEVEL:20}" + # to configure SNMP to work over UDP or TCP + underlying_protocol: "${SNMP_UNDERLYING_PROTOCOL:udp}" swagger: api_path_regex: "${SWAGGER_API_PATH_REGEX:/api.*}" diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/SnmpDeviceTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/SnmpDeviceTransportConfiguration.java index 551bf2fcfb..a6056b4188 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/SnmpDeviceTransportConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/SnmpDeviceTransportConfiguration.java @@ -25,8 +25,10 @@ import org.thingsboard.server.common.data.transport.snmp.SnmpProtocolVersion; public class SnmpDeviceTransportConfiguration implements DeviceTransportConfiguration { private String address; private int port; - private String community; private SnmpProtocolVersion protocolVersion; + private String securityName; + private String authenticationPassphrase; // for SNMP v3 + private String privacyPassphrase; // for SNMP v3 @Override public DeviceTransportType getType() { @@ -43,6 +45,6 @@ public class SnmpDeviceTransportConfiguration implements DeviceTransportConfigur @JsonIgnore private boolean isValid() { return StringUtils.isNotBlank(address) && port > 0 && - StringUtils.isNotBlank(community) && protocolVersion != null; + StringUtils.isNotBlank(securityName) && protocolVersion != null; } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/SnmpDeviceProfileTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/SnmpDeviceProfileTransportConfiguration.java index a4f7cc57de..843a51304e 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/SnmpDeviceProfileTransportConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/SnmpDeviceProfileTransportConfiguration.java @@ -18,35 +18,21 @@ package org.thingsboard.server.common.data.device.profile; import com.fasterxml.jackson.annotation.JsonIgnore; import lombok.Data; import org.thingsboard.server.common.data.DeviceTransportType; -import org.thingsboard.server.common.data.transport.snmp.SnmpMapping; +import org.thingsboard.server.common.data.transport.snmp.configs.SnmpCommunicationConfig; -import java.util.Collections; import java.util.List; -import java.util.stream.Collectors; -import java.util.stream.Stream; @Data public class SnmpDeviceProfileTransportConfiguration implements DeviceProfileTransportConfiguration { - private int pollPeriodMs; private int timeoutMs; private int retries; - private List telemetryMappings; - private List attributesMappings; + private List communicationConfigs; @Override public DeviceTransportType getType() { return DeviceTransportType.SNMP; } - @JsonIgnore - public List getAllMappings() { - if (telemetryMappings != null && attributesMappings != null) { - return Stream.concat(telemetryMappings.stream(), attributesMappings.stream()).collect(Collectors.toList()); - } else { - return Collections.emptyList(); - } - } - @Override public void validate() { if (!isValid()) { @@ -56,9 +42,6 @@ public class SnmpDeviceProfileTransportConfiguration implements DeviceProfileTra @JsonIgnore private boolean isValid() { - List mappings = getAllMappings(); - return pollPeriodMs > 0 && timeoutMs > 0 && retries >= 0 && - !mappings.isEmpty() && mappings.stream().allMatch(SnmpMapping::isValid) && - mappings.stream().map(SnmpMapping::getOid).distinct().count() == mappings.size(); + return true; } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/JsonBasedTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpCommunicationSpec.java similarity index 51% rename from common/data/src/main/java/org/thingsboard/server/common/data/device/JsonBasedTransportConfiguration.java rename to common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpCommunicationSpec.java index c65e9dd972..f6ff7d3109 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/JsonBasedTransportConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpCommunicationSpec.java @@ -13,27 +13,28 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.common.data.device; +package org.thingsboard.server.common.data.transport.snmp; -import com.fasterxml.jackson.annotation.JsonAnyGetter; -import com.fasterxml.jackson.annotation.JsonAnySetter; -import com.fasterxml.jackson.annotation.JsonIgnore; +public enum SnmpCommunicationSpec { + TELEMETRY_QUERYING(true), + CLIENT_ATTRIBUTES_QUERYING(true), -import java.util.HashMap; -import java.util.Map; + SHARED_ATTRIBUTES_SETTING, -public abstract class JsonBasedTransportConfiguration { + TELEMETRY_TRAPS_RECEIVING, + CLIENT_ATTRIBUTES_TRAPS_RECEIVING; - @JsonIgnore - private final Map properties = new HashMap<>(); + private final boolean isRepeatingQuerying; - @JsonAnyGetter - public Map properties() { - return this.properties; + SnmpCommunicationSpec() { + this.isRepeatingQuerying = false; } - @JsonAnySetter - public void put(String name, Object value) { - this.properties.put(name, value); + SnmpCommunicationSpec(boolean isRepeatingQuerying) { + this.isRepeatingQuerying = isRepeatingQuerying; + } + + public boolean isRepeatingQuerying() { + return isRepeatingQuerying; } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpMapping.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpMapping.java index 32a3f2a25b..0269491461 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpMapping.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpMapping.java @@ -25,7 +25,6 @@ import java.util.regex.Pattern; @Data public class SnmpMapping { private String oid; - private SnmpMethod method; private String key; private DataType dataType; @@ -33,7 +32,7 @@ public class SnmpMapping { @JsonIgnore public boolean isValid() { - return StringUtils.isNotEmpty(oid) && OID_PATTERN.matcher(oid).matches() && method != null && + return StringUtils.isNotEmpty(oid) && OID_PATTERN.matcher(oid).matches() && StringUtils.isNotBlank(key) && dataType != null; } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/ClientAttributesQueryingSnmpCommunicationConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/ClientAttributesQueryingSnmpCommunicationConfig.java new file mode 100644 index 0000000000..16198f8ef6 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/ClientAttributesQueryingSnmpCommunicationConfig.java @@ -0,0 +1,25 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.transport.snmp.configs; + +import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec; + +public class ClientAttributesQueryingSnmpCommunicationConfig extends RepeatingQueryingSnmpCommunicationConfig { + @Override + public SnmpCommunicationSpec getSpec() { + return SnmpCommunicationSpec.CLIENT_ATTRIBUTES_QUERYING; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/ClientAttributesTrapsReceivingSnmpCommunicationConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/ClientAttributesTrapsReceivingSnmpCommunicationConfig.java new file mode 100644 index 0000000000..9a11b01f1b --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/ClientAttributesTrapsReceivingSnmpCommunicationConfig.java @@ -0,0 +1,25 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.transport.snmp.configs; + +import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec; + +public class ClientAttributesTrapsReceivingSnmpCommunicationConfig extends SnmpCommunicationConfig { + @Override + public SnmpCommunicationSpec getSpec() { + return SnmpCommunicationSpec.CLIENT_ATTRIBUTES_TRAPS_RECEIVING; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/RepeatingQueryingSnmpCommunicationConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/RepeatingQueryingSnmpCommunicationConfig.java new file mode 100644 index 0000000000..764bf97b15 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/RepeatingQueryingSnmpCommunicationConfig.java @@ -0,0 +1,36 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.transport.snmp.configs; + +import lombok.Data; +import lombok.EqualsAndHashCode; +import org.thingsboard.server.common.data.transport.snmp.SnmpMethod; + +@EqualsAndHashCode(callSuper = true) +@Data +public abstract class RepeatingQueryingSnmpCommunicationConfig extends SnmpCommunicationConfig { + private Long queryingFrequencyMs; + + @Override + public SnmpMethod getMethod() { + return SnmpMethod.GET; + } + + @Override + public boolean isValid() { + return true; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/SharedAttributesSettingSnmpCommunicationConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/SharedAttributesSettingSnmpCommunicationConfig.java new file mode 100644 index 0000000000..1c63dae618 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/SharedAttributesSettingSnmpCommunicationConfig.java @@ -0,0 +1,36 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.transport.snmp.configs; + +import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec; +import org.thingsboard.server.common.data.transport.snmp.SnmpMethod; + +public class SharedAttributesSettingSnmpCommunicationConfig extends SnmpCommunicationConfig { + @Override + public SnmpCommunicationSpec getSpec() { + return SnmpCommunicationSpec.SHARED_ATTRIBUTES_SETTING; + } + + @Override + public SnmpMethod getMethod() { + return SnmpMethod.SET; + } + + @Override + public boolean isValid() { + return true; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/SnmpCommunicationConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/SnmpCommunicationConfig.java new file mode 100644 index 0000000000..bfc194669b --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/SnmpCommunicationConfig.java @@ -0,0 +1,62 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.transport.snmp.configs; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonSubTypes.Type; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec; +import org.thingsboard.server.common.data.transport.snmp.SnmpMapping; +import org.thingsboard.server.common.data.transport.snmp.SnmpMethod; + +import java.util.List; + +@JsonIgnoreProperties(ignoreUnknown = true) +@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "spec") +@JsonSubTypes({ + @Type(value = TelemetryQueryingSnmpCommunicationConfig.class, name = "TELEMETRY_QUERYING"), + @Type(value = ClientAttributesQueryingSnmpCommunicationConfig.class, name = "CLIENT_ATTRIBUTES_QUERYING"), + @Type(value = SharedAttributesSettingSnmpCommunicationConfig.class, name = "SHARED_ATTRIBUTES_SETTING"), + @Type(value = TelemetryTrapsReceivingSnmpCommunicationConfig.class, name = "TELEMETRY_TRAPS_RECEIVING"), + @Type(value = ClientAttributesTrapsReceivingSnmpCommunicationConfig.class, name = "CLIENT_ATTRIBUTES_TRAPS_RECEIVING") +}) +public abstract class SnmpCommunicationConfig { + protected List mappings; + + public List getMappings() { + return mappings; + } + + public abstract SnmpCommunicationSpec getSpec(); + + @JsonIgnore + public SnmpMethod getMethod() { + return null; + } + + @JsonIgnore + public boolean isValid() { + return true; + } + + public void validate() { + if (!isValid()) { + throw new IllegalArgumentException("Communication config is not valid"); + } + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/TelemetryQueryingSnmpCommunicationConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/TelemetryQueryingSnmpCommunicationConfig.java new file mode 100644 index 0000000000..68d3c81ed2 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/TelemetryQueryingSnmpCommunicationConfig.java @@ -0,0 +1,29 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.transport.snmp.configs; + +import lombok.Data; +import lombok.EqualsAndHashCode; +import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec; + +@EqualsAndHashCode(callSuper = true) +@Data +public class TelemetryQueryingSnmpCommunicationConfig extends RepeatingQueryingSnmpCommunicationConfig { + @Override + public SnmpCommunicationSpec getSpec() { + return SnmpCommunicationSpec.TELEMETRY_QUERYING; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/TelemetryTrapsReceivingSnmpCommunicationConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/TelemetryTrapsReceivingSnmpCommunicationConfig.java new file mode 100644 index 0000000000..66be4d0cf8 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/TelemetryTrapsReceivingSnmpCommunicationConfig.java @@ -0,0 +1,30 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.transport.snmp.configs; + +import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec; + +public class TelemetryTrapsReceivingSnmpCommunicationConfig extends SnmpCommunicationConfig { + @Override + public SnmpCommunicationSpec getSpec() { + return SnmpCommunicationSpec.TELEMETRY_TRAPS_RECEIVING; + } + + @Override + public boolean isValid() { + return false; + } +} diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportServiceImpl.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportServiceImpl.java index 4ca6fb4f34..3943bc2a51 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportServiceImpl.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportServiceImpl.java @@ -717,7 +717,7 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService { * @return - value of Resource or null */ private String getResourceValueToString(LwM2mClient lwM2MClient, String path) { - LwM2mPath pathIds =new LwM2mPath(convertToObjectIdFromIdVer(path)); + LwM2mPath pathIds = new LwM2mPath(convertToObjectIdFromIdVer(path)); ResourceValue resourceValue = this.returnResourceValueFromLwM2MClient(lwM2MClient, path); return resourceValue == null ? null : this.converter.convertValue(resourceValue.getResourceValue(), this.lwM2mTransportContextServer.getLwM2MTransportConfigServer().getResourceModelType(lwM2MClient.getRegistration(), pathIds), ResourceModel.Type.STRING, pathIds).toString(); @@ -831,8 +831,10 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService { // #5.1 if (!observeOld.equals(observeNew)) { - Set observeSetOld = new Gson().fromJson(observeOld, new TypeToken>() {}.getType()); - Set observeSetNew = new Gson().fromJson(observeNew, new TypeToken>() {}.getType()); + Set observeSetOld = new Gson().fromJson(observeOld, new TypeToken>() { + }.getType()); + Set observeSetNew = new Gson().fromJson(observeNew, new TypeToken>() { + }.getType()); //#5.2 add // path Attr/Telemetry includes newObserve attributeSetOld.addAll(telemetrySetOld); @@ -968,7 +970,7 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService { LwM2mClientProfile profile = lwM2mClientContext.getProfile(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB())); Registration registration = lwM2mClientContext.getLwM2MClient(sessionInfo).getRegistration(); return profile.getPostKeyNameProfile().getAsJsonObject().entrySet().stream() - .filter(e -> e.getValue().getAsString().equals(name) && validateResourceInModelByVer(registration, e.getKey())).findFirst().map(Map.Entry::getKey) + .filter(e -> e.getValue().getAsString().equals(name) && validateResourceInModel(registration, e.getKey(), false)).findFirst().map(Map.Entry::getKey) .orElse(""); } @@ -988,7 +990,7 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService { try { LwM2mClient lwM2MClient = lwM2mClientContext.getLwM2MClient(sessionInfo); attributesResponse.getSharedAttributeListList().forEach(attr -> { - String path =this.getPathAttributeUpdate(sessionInfo, attr.getKv().getKey()); + String path = this.getPathAttributeUpdate(sessionInfo, attr.getKv().getKey()); if (path != null) { // #1.1 if (lwM2MClient.getDelayedRequests().containsKey(path) && attr.getTs() > lwM2MClient.getDelayedRequests().get(path).getTs()) { @@ -1117,7 +1119,7 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService { ConcurrentMap keyNamesIsWritable = keyNamesMap.entrySet() .stream() - .filter(e -> (attrSet.contains(e.getKey()) && resourceIsWritable(lwM2MClient.getRegistration(), e.getKey()))) + .filter(e -> (attrSet.contains(e.getKey()) && validateResourceInModel(lwM2MClient.getRegistration(), e.getKey(), true))) .collect(Collectors.toConcurrentMap(Map.Entry::getKey, Map.Entry::getValue)); Set namesIsWritable = ConcurrentHashMap.newKeySet(); @@ -1130,17 +1132,13 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService { return "LWM2M"; } - private boolean resourceIsWritable(Registration registration, String pathKey) { - return validateResourceInModelByVer(registration, pathKey) && - lwM2mTransportContextServer.getLwM2MTransportConfigServer().getResourceModel(registration, - new LwM2mPath(convertToObjectIdFromIdVer(pathKey))).operations.isWritable(); - } - - private boolean validateResourceInModelByVer(Registration registration, String pathKey) { + private boolean validateResourceInModel(Registration registration, String pathKey, boolean isWritable) { + ResourceModel resourceModel = lwM2mTransportContextServer.getLwM2MTransportConfigServer().getResourceModel(registration, + new LwM2mPath(convertToObjectIdFromIdVer(pathKey))); Integer objectId = validateObjectIdFromKey(pathKey); String objectVer = validateObjectVerFromKey(pathKey); - LwM2mPath lwM2mPath = new LwM2mPath(convertToObjectIdFromIdVer(pathKey)); - return objectId != null && objectVer != null && objectVer.equals(registration.getSupportedVersion(objectId)) && - lwM2mTransportContextServer.getLwM2MTransportConfigServer().getResourceModel(registration, lwM2mPath) != null; + return resourceModel != null && (isWritable ? + objectId != null && objectVer != null && objectVer.equals(registration.getSupportedVersion(objectId)) && resourceModel.operations.isWritable() : + objectId != null && objectVer != null && objectVer.equals(registration.getSupportedVersion(objectId))); } } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mVersionedModelProvider.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mVersionedModelProvider.java index 209f7aa68a..fc35668fef 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mVersionedModelProvider.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mVersionedModelProvider.java @@ -22,6 +22,7 @@ import org.eclipse.leshan.core.model.ObjectModel; import org.eclipse.leshan.core.model.ResourceModel; import org.eclipse.leshan.server.model.LwM2mModelProvider; import org.eclipse.leshan.server.registration.Registration; +import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClientContext; @@ -30,6 +31,7 @@ import java.util.Base64; import java.util.Collection; import java.util.Iterator; import java.util.Map; +import java.util.Optional; import static org.thingsboard.server.common.data.ResourceType.LWM2M_MODEL; import static org.thingsboard.server.common.data.lwm2m.LwM2mConstants.LWM2M_SEPARATOR_KEY; @@ -83,6 +85,7 @@ public class LwM2mVersionedModelProvider implements LwM2mModelProvider { if (objectModel != null) return objectModel.resources.get(resourceId); else + log.warn("TbResources (Object model) with id [{}/{}] not found on the server", objectId, resourceId); return null; } catch (Exception e) { log.error("", e); @@ -116,17 +119,15 @@ public class LwM2mVersionedModelProvider implements LwM2mModelProvider { private ObjectModel getObjectModelDynamic(Integer objectId, String version) { String key = getKeyIdVer(objectId, version); - String xmlB64 = lwM2mTransportContextServer.getTransportResourceCache().get( - this.tenantId, - LWM2M_MODEL, - key). - getData(); - return xmlB64 != null && !xmlB64.isEmpty() ? - lwM2mTransportContextServer.parseFromXmlToObjectModel( - Base64.getDecoder().decode(xmlB64), - key + ".xml", - new DefaultDDFFileValidator()) : - null; + + Optional tbResource = lwM2mTransportContextServer + .getTransportResourceCache() + .get(this.tenantId, LWM2M_MODEL, key); + + return tbResource.map(resource -> lwM2mTransportContextServer.parseFromXmlToObjectModel( + Base64.getDecoder().decode(resource.getData()), + key + ".xml", + new DefaultDDFFileValidator())).orElse(null); } } } diff --git a/common/transport/snmp/pom.xml b/common/transport/snmp/pom.xml index ab83ab69a7..afa35b518c 100644 --- a/common/transport/snmp/pom.xml +++ b/common/transport/snmp/pom.xml @@ -58,5 +58,11 @@ org.snmp4j snmp4j + + org.snmp4j + snmp4j-agent + 3.3.6 + test + diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulator.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulator.java deleted file mode 100644 index 7109ef4e3f..0000000000 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulator.java +++ /dev/null @@ -1,81 +0,0 @@ -/** - * Copyright © 2016-2021 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.snmp; - -import org.snmp4j.CommunityTarget; -import org.snmp4j.PDU; -import org.snmp4j.Snmp; -import org.snmp4j.Target; -import org.snmp4j.event.ResponseEvent; -import org.snmp4j.mp.SnmpConstants; -import org.snmp4j.smi.GenericAddress; -import org.snmp4j.smi.OID; -import org.snmp4j.smi.OctetString; -import org.snmp4j.smi.VariableBinding; -import org.snmp4j.transport.DefaultUdpTransportMapping; -import org.snmp4j.transport.UdpTransportMapping; - -import java.io.IOException; - -/** - * For testing purposes. Will be removed when the time comes - */ -public class SnmpDeviceSimulator { - private final Target target; - private final OID oid = new OID(".1.3.6.1.2.1.1.1.0"); - private Snmp snmp; - - public SnmpDeviceSimulator(int port) { - String address = "udp:127.0.0.1/" + port; - - CommunityTarget target = new CommunityTarget(); - target.setCommunity(new OctetString("public")); - target.setAddress(GenericAddress.parse(address)); - target.setRetries(2); - target.setTimeout(1500); - target.setVersion(SnmpConstants.version2c); - - this.target = target; - } - - public static void main(String[] args) throws IOException { - SnmpDeviceSimulator deviceSimulator = new SnmpDeviceSimulator(161); - - deviceSimulator.start(); - String response = deviceSimulator.sendRequest(PDU.GET); - - System.out.println(response); - } - - public void start() throws IOException { - UdpTransportMapping transport = new DefaultUdpTransportMapping(); - transport.addTransportListener((sourceTransport, incomingAddress, wholeMessage, tmStateReference) -> { - System.out.println(); - }); - snmp = new Snmp(transport); - - transport.listen(); - } - - public String sendRequest(int pduType) throws IOException { - PDU pdu = new PDU(); - pdu.add(new VariableBinding(oid)); - pdu.setType(pduType); - - ResponseEvent responseEvent = snmp.send(pdu, target); - return responseEvent.getResponse().get(0).getVariable().toString(); - } -} diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java index beaaef829f..877d4f1652 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java @@ -17,9 +17,7 @@ package org.thingsboard.server.transport.snmp; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.snmp4j.PDU; -import org.snmp4j.smi.OID; -import org.snmp4j.smi.VariableBinding; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.event.EventListener; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.Device; @@ -32,8 +30,6 @@ import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.data.security.DeviceCredentialsType; -import org.thingsboard.server.common.data.transport.snmp.SnmpMapping; -import org.thingsboard.server.common.data.transport.snmp.SnmpMethod; import org.thingsboard.server.common.transport.DeviceUpdatedEvent; import org.thingsboard.server.common.transport.TransportContext; import org.thingsboard.server.common.transport.TransportDeviceProfileCache; @@ -41,7 +37,6 @@ import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.auth.SessionInfoCreator; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; -import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; import org.thingsboard.server.queue.util.AfterStartUp; @@ -51,9 +46,7 @@ import org.thingsboard.server.transport.snmp.service.SnmpTransportBalancingServi import org.thingsboard.server.transport.snmp.service.SnmpTransportService; import org.thingsboard.server.transport.snmp.session.DeviceSessionContext; -import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; @@ -75,10 +68,11 @@ public class SnmpTransportContext extends TransportContext { private final SnmpTransportBalancingService balancingService; private final Map sessions = new ConcurrentHashMap<>(); - private final Map profilesTransportConfigs = new ConcurrentHashMap<>(); - private final Map> profilesPdus = new ConcurrentHashMap<>(); private Collection allSnmpDevicesIds = new ConcurrentLinkedDeque<>(); + @Value("${transport.snmp.underlying_protocol}") + private String snmpUnderlyingProtocol; + @AfterStartUp(order = 2) public void initDevicesSessions() { log.info("Initializing SNMP devices sessions"); @@ -120,22 +114,19 @@ public class SnmpTransportContext extends TransportContext { SnmpDeviceProfileTransportConfiguration profileTransportConfiguration = (SnmpDeviceProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration(); SnmpDeviceTransportConfiguration deviceTransportConfiguration = (SnmpDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration(); - profilesTransportConfigs.put(deviceProfileId, profileTransportConfiguration); - profilesPdus.computeIfAbsent(deviceProfileId, id -> createPdus(profileTransportConfiguration)); - DeviceSessionContext deviceSessionContext = new DeviceSessionContext( - device, deviceProfile, - credentials.getCredentialsId(), deviceTransportConfiguration, - this, snmpTransportService + device, deviceProfile, credentials.getCredentialsId(), + profileTransportConfiguration, deviceTransportConfiguration, + this, snmpTransportService, snmpUnderlyingProtocol ); registerSessionMsgListener(deviceSessionContext); sessions.put(device.getId(), deviceSessionContext); + snmpTransportService.createQueryingTasks(deviceSessionContext); log.info("Established SNMP device session for device {}", device.getId()); } private void updateDeviceSession(DeviceSessionContext sessionContext, Device device, DeviceProfile deviceProfile) { log.info("Updating SNMP device session for device {}", device.getId()); - DeviceProfileId deviceProfileId = deviceProfile.getId(); DeviceCredentials credentials = protoEntityService.getDeviceCredentialsByDeviceId(device.getId()); if (credentials.getCredentialsType() != DeviceCredentialsType.ACCESS_TOKEN) { @@ -148,14 +139,13 @@ public class SnmpTransportContext extends TransportContext { SnmpDeviceTransportConfiguration newDeviceTransportConfiguration = (SnmpDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration(); if (!newProfileTransportConfiguration.equals(sessionContext.getProfileTransportConfiguration())) { - profilesPdus.put(deviceProfileId, createPdus(newProfileTransportConfiguration)); - profilesTransportConfigs.put(deviceProfileId, newProfileTransportConfiguration); - sessionContext.setProfileTransportConfiguration(newProfileTransportConfiguration); - sessionContext.initTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration); + sessionContext.initializeTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration); + snmpTransportService.cancelQueryingTasks(sessionContext); + snmpTransportService.createQueryingTasks(sessionContext); } else if (!newDeviceTransportConfiguration.equals(sessionContext.getDeviceTransportConfiguration())) { sessionContext.setDeviceTransportConfiguration(newDeviceTransportConfiguration); - sessionContext.initTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration); + sessionContext.initializeTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration); } else { log.trace("Configuration of the device {} was not updated", device); } @@ -167,22 +157,14 @@ public class SnmpTransportContext extends TransportContext { sessionContext.close(); transportService.deregisterSession(sessionContext.getSessionInfo()); sessions.remove(sessionContext.getDeviceId()); - log.trace("Deregistered and removed session"); - - DeviceProfileId deviceProfileId = sessionContext.getDeviceProfile().getId(); - if (sessions.values().stream() - .map(DeviceAwareSessionContext::getDeviceProfile) - .noneMatch(deviceProfile -> deviceProfile.getId().equals(deviceProfileId))) { - log.trace("Removed values for device profile {} from configs and pdus caches", deviceProfileId); - profilesTransportConfigs.remove(deviceProfileId); - profilesPdus.remove(deviceProfileId); - } + snmpTransportService.cancelQueryingTasks(sessionContext); + log.trace("Unregistered and removed session"); } private void registerSessionMsgListener(DeviceSessionContext deviceSessionContext) { transportService.process(DeviceTransportType.SNMP, TransportProtos.ValidateDeviceTokenRequestMsg.newBuilder().setToken(deviceSessionContext.getToken()).build(), - new TransportServiceCallback() { + new TransportServiceCallback<>() { @Override public void onSuccess(ValidateDeviceCredentialsResponse msg) { if (msg.hasDeviceInfo()) { @@ -191,6 +173,7 @@ public class SnmpTransportContext extends TransportContext { ); transportService.registerAsyncSession(sessionInfo, deviceSessionContext); + transportService.process(sessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build(), null); deviceSessionContext.setSessionInfo(sessionInfo); deviceSessionContext.setDeviceInfo(msg.getDeviceInfo()); } else { @@ -205,23 +188,6 @@ public class SnmpTransportContext extends TransportContext { }); } - private List createPdus(SnmpDeviceProfileTransportConfiguration deviceProfileConfig) { - Map> bindingsPerMethod = new HashMap<>(); - - deviceProfileConfig.getAllMappings().forEach(mapping -> bindingsPerMethod - .computeIfAbsent(mapping.getMethod(), v -> new ArrayList<>()) - .add(new VariableBinding(new OID(mapping.getOid())))); - - return bindingsPerMethod.keySet().stream() - .map(method -> { - PDU request = new PDU(); - request.setType(method.getCode()); - request.addAll(bindingsPerMethod.get(method)); - return request; - }) - .collect(Collectors.toList()); - } - @EventListener(DeviceUpdatedEvent.class) public void onDeviceUpdatedOrCreated(DeviceUpdatedEvent deviceUpdatedEvent) { Device device = deviceUpdatedEvent.getDevice(); @@ -293,27 +259,4 @@ public class SnmpTransportContext extends TransportContext { return sessions.values(); } - public Map> getProfilesPdus() { - return profilesPdus; - } - - public Optional getAttributeMapping(DeviceProfileId deviceProfileId, OID responseOid) { - if (profilesTransportConfigs.containsKey(deviceProfileId)) { - return getMapping(responseOid, profilesTransportConfigs.get(deviceProfileId).getAttributesMappings()); - } - return Optional.empty(); - } - - public Optional getTelemetryMapping(DeviceProfileId deviceProfileId, OID responseOid) { - if (profilesTransportConfigs.containsKey(deviceProfileId)) { - return getMapping(responseOid, profilesTransportConfigs.get(deviceProfileId).getTelemetryMappings()); - } - return Optional.empty(); - } - - private Optional getMapping(OID oid, List mappings) { - return mappings.stream() - .filter(mapping -> new OID(mapping.getOid()).equals(oid)) - .findFirst(); - } } diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java index 86d2cead95..963b5db382 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java @@ -16,24 +16,38 @@ package org.thingsboard.server.transport.snmp.service; import com.google.gson.JsonObject; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.snmp4j.CommandResponder; +import org.snmp4j.CommandResponderEvent; import org.snmp4j.PDU; import org.snmp4j.Snmp; +import org.snmp4j.TransportMapping; import org.snmp4j.event.ResponseEvent; import org.snmp4j.smi.Null; +import org.snmp4j.smi.OID; +import org.snmp4j.smi.OctetString; +import org.snmp4j.smi.TcpAddress; +import org.snmp4j.smi.UdpAddress; import org.snmp4j.smi.VariableBinding; +import org.snmp4j.transport.DefaultTcpTransportMapping; import org.snmp4j.transport.DefaultUdpTransportMapping; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.data.TbTransportService; import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.kv.DataType; +import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec; +import org.thingsboard.server.common.data.transport.snmp.SnmpMapping; +import org.thingsboard.server.common.data.transport.snmp.configs.RepeatingQueryingSnmpCommunicationConfig; +import org.thingsboard.server.common.data.transport.snmp.configs.SnmpCommunicationConfig; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.queue.util.AfterStartUp; import org.thingsboard.server.queue.util.TbSnmpTransportComponent; import org.thingsboard.server.transport.snmp.SnmpTransportContext; import org.thingsboard.server.transport.snmp.session.DeviceSessionContext; @@ -41,21 +55,38 @@ import org.thingsboard.server.transport.snmp.session.DeviceSessionContext; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.io.IOException; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; +import java.util.stream.Collectors; +import java.util.stream.Stream; @TbSnmpTransportComponent @Service @Slf4j -public class SnmpTransportService implements TbTransportService { +public class SnmpTransportService implements TbTransportService, CommandResponder { private final SnmpTransportContext snmpTransportContext; private final TransportService transportService; + @Getter private Snmp snmp; - private ScheduledExecutorService pollingExecutor; - private ExecutorService snmpResponseProcessingExecutor; + private ScheduledExecutorService queryingExecutor; + private ExecutorService responseProcessingExecutor; + + private final Map> responseProcessors = new EnumMap<>(SnmpCommunicationSpec.class); + + @Value("${transport.snmp.response_processing.parallelism_level}") + private Integer responseProcessingParallelismLevel; + @Value("${transport.snmp.underlying_protocol}") + private String snmpUnderlyingProtocol; public SnmpTransportService(@Lazy SnmpTransportContext snmpTransportContext, TransportService transportService) { @@ -67,56 +98,144 @@ public class SnmpTransportService implements TbTransportService { private void init() throws IOException { log.info("Initializing SNMP transport service"); - pollingExecutor = Executors.newScheduledThreadPool(1, ThingsBoardThreadFactory.forName("snmp-polling")); - //TODO: Set parallelism value in the config - snmpResponseProcessingExecutor = Executors.newWorkStealingPool(20); + queryingExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), ThingsBoardThreadFactory.forName("snmp-querying")); + responseProcessingExecutor = Executors.newWorkStealingPool(responseProcessingParallelismLevel); initializeSnmp(); + initializeTrapsListener(); + configureResponseProcessors(); log.info("SNMP transport service initialized"); } private void initializeSnmp() throws IOException { - snmp = new Snmp(new DefaultUdpTransportMapping()); + TransportMapping transportMapping; + switch (snmpUnderlyingProtocol) { + case "udp": + transportMapping = new DefaultUdpTransportMapping(); + break; + case "tcp": + transportMapping = new DefaultTcpTransportMapping(); + break; + default: + throw new IllegalArgumentException("Underlying protocol " + snmpUnderlyingProtocol + " for SNMP is not supported"); + } + snmp = new Snmp(transportMapping); snmp.listen(); } - @AfterStartUp(order = 10) - public void startPolling() { - log.info("Starting SNMP polling"); - //TODO: Get poll period from configuration; - int pollPeriodSeconds = 1; + private void initializeTrapsListener() throws IOException { + int trapsListeningPort = 1062; + String bindingAddress = "0.0.0.0/" + trapsListeningPort; - pollingExecutor.scheduleWithFixedDelay(() -> { - snmpTransportContext.getSessions().forEach(this::executeSnmpRequest); - }, 0, pollPeriodSeconds, TimeUnit.SECONDS); - } + TransportMapping transportMapping; + switch (snmpUnderlyingProtocol) { + case "udp": + transportMapping = new DefaultUdpTransportMapping(new UdpAddress(bindingAddress)); + break; + case "tcp": + transportMapping = new DefaultTcpTransportMapping(new TcpAddress(bindingAddress)); + break; + default: + throw new IllegalArgumentException("Underlying protocol " + snmpUnderlyingProtocol + " for SNMP is not supported"); + } + + + Snmp trapsSnmp = new Snmp(transportMapping); + trapsSnmp.addCommandResponder(this); - private void executeSnmpRequest(DeviceSessionContext sessionContext) { - long timeNow = System.currentTimeMillis(); - long nextRequestExecutionTime = sessionContext.getPreviousRequestExecutedAt() + sessionContext.getProfileTransportConfiguration().getPollPeriodMs(); + transportMapping.listen(); + } - if (nextRequestExecutionTime < timeNow) { - sessionContext.setPreviousRequestExecutedAt(timeNow); + public void createQueryingTasks(DeviceSessionContext sessionContext) { + List> queryingTasks = sessionContext.getProfileTransportConfiguration().getCommunicationConfigs().stream() + .filter(config -> config.getSpec().isRepeatingQuerying()) + .map(config -> { + RepeatingQueryingSnmpCommunicationConfig repeatingCommunicationConfig = (RepeatingQueryingSnmpCommunicationConfig) config; + return createQueryingTaskForConfig(sessionContext, repeatingCommunicationConfig); + }) + .collect(Collectors.toList()); + sessionContext.setQueryingTasks(queryingTasks); + } - DeviceProfileId deviceProfileId = sessionContext.getDeviceProfile().getId(); - snmpTransportContext.getProfilesPdus().get(deviceProfileId).forEach(pdu -> { - try { - log.debug("[{}] Sending SNMP message for device {}", pdu.getRequestID(), sessionContext.getDeviceId()); - snmp.send(pdu, sessionContext.getTarget(), deviceProfileId, sessionContext); - } catch (Exception e) { - log.error("Failed to send SNMP request: {}", e.getMessage()); + private ScheduledFuture createQueryingTaskForConfig(DeviceSessionContext sessionContext, RepeatingQueryingSnmpCommunicationConfig communicationConfig) { + Long queryingFrequency = communicationConfig.getQueryingFrequencyMs(); + return queryingExecutor.scheduleWithFixedDelay(() -> { + try { + if (sessionContext.isActive()) { + sendRequest(sessionContext, communicationConfig); } - }); + } catch (Exception e) { + log.error("Failed to send SNMP request for device {}: {}", sessionContext.getDeviceId(), e.getMessage()); + } + }, queryingFrequency, queryingFrequency, TimeUnit.MILLISECONDS); + } + + public void cancelQueryingTasks(DeviceSessionContext sessionContext) { + sessionContext.getQueryingTasks().forEach(task -> task.cancel(true)); + sessionContext.getQueryingTasks().clear(); + } + + public void sendRequest(DeviceSessionContext sessionContext, SnmpCommunicationConfig communicationConfig) throws IOException { + PDU request = createPdu(communicationConfig); + executeRequest(sessionContext, request); + } + + public void sendRequest(DeviceSessionContext sessionContext, SnmpCommunicationConfig communicationConfig, Map values) throws IOException { + PDU request = createPduWithValues(communicationConfig, values); + executeRequest(sessionContext, request); + } + + private void executeRequest(DeviceSessionContext sessionContext, PDU request) throws IOException { + if (request.size() > 0) { + log.trace("Executing SNMP request for device {}. Variables bindings: {}", sessionContext.getDeviceId(), request.getVariableBindings()); + snmp.send(request, sessionContext.getTarget(), sessionContext.getDeviceProfile().getId(), sessionContext); } } - public void onNewDeviceResponse(ResponseEvent responseEvent, DeviceSessionContext sessionContext) { - ((Snmp) responseEvent.getSource()).cancel(responseEvent.getRequest(), sessionContext); - snmpResponseProcessingExecutor.submit(() -> processSnmpResponse(responseEvent, sessionContext)); + private PDU createPdu(SnmpCommunicationConfig communicationConfig) { + PDU pdu = new PDU(); + pdu.setType(communicationConfig.getMethod().getCode()); + pdu.addAll(communicationConfig.getMappings().stream() + .map(mapping -> new VariableBinding(new OID(mapping.getOid()))) + .collect(Collectors.toList())); + return pdu; } - private void processSnmpResponse(ResponseEvent event, DeviceSessionContext sessionContext) { + private PDU createPduWithValues(SnmpCommunicationConfig communicationConfig, Map values) { + PDU pdu = new PDU(); + pdu.setType(communicationConfig.getMethod().getCode()); + pdu.addAll(communicationConfig.getMappings().stream() + .filter(mapping -> values.containsKey(mapping.getKey())) + .map(mapping -> { + String value = values.get(mapping.getKey()); + return new VariableBinding(new OID(mapping.getOid()), new OctetString(value)); + }) + .collect(Collectors.toList())); + return pdu; + } + + + private void processTrap(CommandResponderEvent event) { + if (event.getPDU().getType() != PDU.TRAP) return; + + snmpTransportContext.getSessions().stream() + .filter(sessionContext -> { + // TODO: SNMP v3 support + return sessionContext.getTarget().getSecurityName().equals(OctetString.fromByteArray(event.getSecurityName())) && + sessionContext.getTarget().getAddress().equals(event.getPeerAddress()); + }) + .findFirst() + .ifPresentOrElse(sessionContext -> { + responseProcessingExecutor.execute(() -> processResponse(sessionContext, event.getPDU())); + }, () -> { + log.debug("SNMP event is from unknown source: {}", event); + }); + } + + public void processResponseEvent(DeviceSessionContext sessionContext, ResponseEvent event) { + ((Snmp) event.getSource()).cancel(event.getRequest(), sessionContext); + if (event.getError() != null) { log.warn("Response error: {}", event.getError().getMessage(), event.getError()); return; @@ -127,16 +246,29 @@ public class SnmpTransportService implements TbTransportService { log.warn("No response from SNMP device {}, requestId: {}", sessionContext.getDeviceId(), event.getRequest().getRequestID()); return; } - DeviceProfileId deviceProfileId = (DeviceProfileId) event.getUserObject(); log.debug("[{}] Processing SNMP response for device {} with device profile {}: {}", response.getRequestID(), sessionContext.getDeviceId(), deviceProfileId, response); - JsonObject telemetry = new JsonObject(); - JsonObject attributes = new JsonObject(); + responseProcessingExecutor.execute(() -> processResponse(sessionContext, response)); + } - for (int i = 0; i < response.size(); i++) { - VariableBinding variableBinding = response.get(i); + private void processResponse(DeviceSessionContext sessionContext, PDU responsePdu) { + Map mappings = new HashMap<>(); + Map configs = new HashMap<>(); + Map responses = new EnumMap<>(SnmpCommunicationSpec.class); + + for (SnmpCommunicationConfig config : sessionContext.getProfileTransportConfiguration().getCommunicationConfigs()) { + for (SnmpMapping mapping : config.getMappings()) { + OID oid = new OID(mapping.getOid()); + mappings.put(oid, mapping); + configs.put(oid, config); + } + responses.put(config.getSpec(), new JsonObject()); + } + + for (int i = 0; i < responsePdu.size(); i++) { + VariableBinding variableBinding = responsePdu.get(i); log.trace("Processing variable binding {}: {}", i, variableBinding); if (variableBinding.getVariable() instanceof Null) { @@ -144,37 +276,57 @@ public class SnmpTransportService implements TbTransportService { continue; } - snmpTransportContext.getTelemetryMapping(deviceProfileId, variableBinding.getOid()).ifPresent(mapping -> { - log.trace("Found telemetry mapping for oid {}: {}", variableBinding.getOid(), mapping); - processValue(mapping.getKey(), mapping.getDataType(), variableBinding.toValueString(), telemetry); - }); + OID oid = variableBinding.getOid(); + if (!mappings.containsKey(oid)) { + log.debug("No SNMP mapping for oid {}", oid); + continue; + } - snmpTransportContext.getAttributeMapping(deviceProfileId, variableBinding.getOid()).ifPresent(mapping -> { - log.trace("Found attribute mapping for oid {}: {}", variableBinding.getOid(), mapping); - processValue(mapping.getKey(), mapping.getDataType(), variableBinding.toValueString(), attributes); - }); - } + SnmpCommunicationSpec spec = configs.get(oid).getSpec(); + if (!responseProcessors.containsKey(spec)) { + log.debug("No response processor found for spec {}", spec); + continue; + } - if (telemetry.entrySet().isEmpty() && attributes.entrySet().isEmpty()) { - log.warn("No telemetry or attribute values is the SNMP response for device {}", sessionContext.getDeviceId()); - return; + SnmpMapping mapping = mappings.get(oid); + processValue(mapping.getKey(), mapping.getDataType(), variableBinding.toValueString(), responses.get(spec)); } - if (!telemetry.entrySet().isEmpty()) { - TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(telemetry); - transportService.process(sessionContext.getSessionInfo(), postTelemetryMsg, TransportServiceCallback.EMPTY); - log.debug("Posted telemetry for device {}: {}", sessionContext.getDeviceId(), telemetry); + if (responses.values().stream().allMatch(response -> response.entrySet().isEmpty())) { + log.debug("No values is the SNMP response for device {}. Request id: {}", sessionContext.getDeviceId(), responsePdu.getRequestID()); + return; } - if (!attributes.entrySet().isEmpty()) { - TransportProtos.PostAttributeMsg postAttributesMsg = JsonConverter.convertToAttributesProto(attributes); - transportService.process(sessionContext.getSessionInfo(), postAttributesMsg, TransportServiceCallback.EMPTY); - log.debug("Posted attributes for device {}: {}", sessionContext.getDeviceId(), attributes); - } + responses.forEach((spec, response) -> { + Optional.ofNullable(responseProcessors.get(spec)) + .ifPresent(responseProcessor -> { + responseProcessor.accept(response, sessionContext); + }); + }); reportActivity(sessionContext.getSessionInfo()); } + private void configureResponseProcessors() { + Stream.of(SnmpCommunicationSpec.TELEMETRY_QUERYING, SnmpCommunicationSpec.TELEMETRY_TRAPS_RECEIVING) + .forEach(telemetrySpec -> { + responseProcessors.put(telemetrySpec, (response, sessionContext) -> { + TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(response); + transportService.process(sessionContext.getSessionInfo(), postTelemetryMsg, TransportServiceCallback.EMPTY); + log.debug("Posted telemetry for device {}: {}", sessionContext.getDeviceId(), response); + }); + }); + + Stream.of(SnmpCommunicationSpec.CLIENT_ATTRIBUTES_QUERYING, SnmpCommunicationSpec.CLIENT_ATTRIBUTES_TRAPS_RECEIVING) + .forEach(clientAttributesSpec -> { + responseProcessors.put(clientAttributesSpec, (response, sessionContext) -> { + TransportProtos.PostAttributeMsg postAttributesMsg = JsonConverter.convertToAttributesProto(response); + transportService.process(sessionContext.getSessionInfo(), postAttributesMsg, TransportServiceCallback.EMPTY); + log.debug("Posted attributes for device {}: {}", sessionContext.getDeviceId(), response); + }); + }); + } + private void reportActivity(TransportProtos.SessionInfoProto sessionInfo) { transportService.process(sessionInfo, TransportProtos.SubscriptionInfoProto.newBuilder() .setAttributeSubscription(false) @@ -184,6 +336,8 @@ public class SnmpTransportService implements TbTransportService { } private void processValue(String key, DataType dataType, String value, JsonObject result) { + if (StringUtils.isEmpty(value)) return; + switch (dataType) { case LONG: result.addProperty(key, Long.parseLong(value)); @@ -199,6 +353,11 @@ public class SnmpTransportService implements TbTransportService { } } + @Override + public void processPdu(CommandResponderEvent event) { + processTrap(event); + } + @Override public String getName() { return "SNMP"; @@ -207,11 +366,11 @@ public class SnmpTransportService implements TbTransportService { @PreDestroy public void shutdown() { log.info("Stopping SNMP transport!"); - if (pollingExecutor != null) { - pollingExecutor.shutdownNow(); + if (queryingExecutor != null) { + queryingExecutor.shutdownNow(); } - if (snmpResponseProcessingExecutor != null) { - snmpResponseProcessingExecutor.shutdownNow(); + if (responseProcessingExecutor != null) { + responseProcessingExecutor.shutdownNow(); } if (snmp != null) { try { diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java index 3cfc9bc041..c0bbc843bc 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java @@ -18,18 +18,31 @@ package org.thingsboard.server.transport.snmp.session; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; +import org.snmp4j.AbstractTarget; import org.snmp4j.CommunityTarget; import org.snmp4j.Target; +import org.snmp4j.UserTarget; import org.snmp4j.event.ResponseEvent; import org.snmp4j.event.ResponseListener; +import org.snmp4j.security.AuthSHA; +import org.snmp4j.security.PrivDES; +import org.snmp4j.security.SecurityLevel; +import org.snmp4j.security.SecurityModel; +import org.snmp4j.security.SecurityModels; +import org.snmp4j.security.USM; +import org.snmp4j.security.UsmUser; import org.snmp4j.smi.GenericAddress; +import org.snmp4j.smi.OID; import org.snmp4j.smi.OctetString; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.device.data.SnmpDeviceTransportConfiguration; import org.thingsboard.server.common.data.device.profile.SnmpDeviceProfileTransportConfiguration; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec; +import org.thingsboard.server.common.data.transport.snmp.SnmpProtocolVersion; import org.thingsboard.server.common.transport.SessionMsgListener; +import org.thingsboard.server.common.transport.adaptor.JsonConverter; import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg; @@ -40,8 +53,13 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseM import org.thingsboard.server.transport.snmp.SnmpTransportContext; import org.thingsboard.server.transport.snmp.service.SnmpTransportService; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; import java.util.UUID; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; @Slf4j public class DeviceSessionContext extends DeviceAwareSessionContext implements SessionMsgListener, ResponseListener { @@ -64,11 +82,19 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S @Setter private long previousRequestExecutedAt = 0; private final AtomicInteger msgIdSeq = new AtomicInteger(0); + @Getter private boolean isActive = true; + private final String snmpUnderlyingProtocol; + + @Getter + @Setter + private List> queryingTasks = new LinkedList<>(); - public DeviceSessionContext(Device device, DeviceProfile deviceProfile, - String token, SnmpDeviceTransportConfiguration deviceTransportConfiguration, - SnmpTransportContext snmpTransportContext, SnmpTransportService snmpTransportService) { + public DeviceSessionContext(Device device, DeviceProfile deviceProfile, String token, + SnmpDeviceProfileTransportConfiguration profileTransportConfiguration, + SnmpDeviceTransportConfiguration deviceTransportConfiguration, + SnmpTransportContext snmpTransportContext, SnmpTransportService snmpTransportService, + String snmpUnderlyingProtocol) { super(UUID.randomUUID()); super.setDeviceId(device.getId()); super.setDeviceProfile(deviceProfile); @@ -78,10 +104,11 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S this.snmpTransportContext = snmpTransportContext; this.snmpTransportService = snmpTransportService; - this.profileTransportConfiguration = (SnmpDeviceProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration(); + this.profileTransportConfiguration = profileTransportConfiguration; this.deviceTransportConfiguration = deviceTransportConfiguration; - initTarget(this.profileTransportConfiguration, this.deviceTransportConfiguration); + this.snmpUnderlyingProtocol = snmpUnderlyingProtocol; + initializeTarget(profileTransportConfiguration, deviceTransportConfiguration); } @Override @@ -100,20 +127,62 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S @Override public void onResponse(ResponseEvent event) { if (isActive) { - snmpTransportService.onNewDeviceResponse(event, this); + snmpTransportService.processResponseEvent(this, event); } } - public void initTarget(SnmpDeviceProfileTransportConfiguration profileTransportConfig, SnmpDeviceTransportConfiguration deviceTransportConfig) { + public void initializeTarget(SnmpDeviceProfileTransportConfiguration profileTransportConfig, SnmpDeviceTransportConfiguration deviceTransportConfig) { log.trace("Initializing target for SNMP session of device {}", device); - CommunityTarget communityTarget = new CommunityTarget(); - communityTarget.setAddress(GenericAddress.parse(GenericAddress.TYPE_UDP + ":" + deviceTransportConfig.getAddress() + "/" + deviceTransportConfig.getPort())); - communityTarget.setVersion(deviceTransportConfig.getProtocolVersion().getCode()); - communityTarget.setCommunity(new OctetString(deviceTransportConfig.getCommunity())); - communityTarget.setTimeout(profileTransportConfig.getTimeoutMs()); - communityTarget.setRetries(profileTransportConfig.getRetries()); - this.target = communityTarget; - log.info("SNMP target initialized: {}", this.target); + + AbstractTarget target; + + SnmpProtocolVersion protocolVersion = deviceTransportConfig.getProtocolVersion(); + switch (protocolVersion) { + case V1: + CommunityTarget communityTargetV1 = new CommunityTarget(); + communityTargetV1.setSecurityModel(SecurityModel.SECURITY_MODEL_SNMPv1); + communityTargetV1.setSecurityLevel(SecurityLevel.NOAUTH_NOPRIV); + communityTargetV1.setCommunity(new OctetString(deviceTransportConfig.getSecurityName())); + target = communityTargetV1; + break; + case V2C: + CommunityTarget communityTargetV2 = new CommunityTarget(); + communityTargetV2.setSecurityModel(SecurityModel.SECURITY_MODEL_SNMPv2c); + communityTargetV2.setSecurityLevel(SecurityLevel.NOAUTH_NOPRIV); + communityTargetV2.setCommunity(new OctetString(deviceTransportConfig.getSecurityName())); + target = communityTargetV2; + break; + case V3: + USM usm = new USM(); + SecurityModels.getInstance().addSecurityModel(usm); + + OctetString securityName = new OctetString(deviceTransportConfig.getSecurityName()); + OctetString authenticationPassphrase = new OctetString(deviceTransportConfig.getAuthenticationPassphrase()); + OctetString privacyPassphrase = new OctetString(deviceTransportConfig.getPrivacyPassphrase()); + + OID authenticationProtocol = AuthSHA.ID; + OID privacyProtocol = PrivDES.ID; // FIXME: to config + + UsmUser user = new UsmUser(securityName, authenticationProtocol, authenticationPassphrase, privacyProtocol, privacyPassphrase); + snmpTransportService.getSnmp().getUSM().addUser(user); + + UserTarget userTarget = new UserTarget(); + userTarget.setSecurityName(securityName); + userTarget.setSecurityLevel(SecurityLevel.AUTH_PRIV); + + target = userTarget; + break; + default: + throw new UnsupportedOperationException("SNMP protocol version " + protocolVersion + " is not supported"); + } + + target.setAddress(GenericAddress.parse(snmpUnderlyingProtocol + ":" + deviceTransportConfig.getAddress() + "/" + deviceTransportConfig.getPort())); + target.setTimeout(profileTransportConfig.getTimeoutMs()); + target.setRetries(profileTransportConfig.getRetries()); + target.setVersion(protocolVersion.getCode()); + + this.target = target; + log.info("SNMP target initialized: {}", target); } public void close() { @@ -135,6 +204,21 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S @Override public void onAttributeUpdate(AttributeUpdateNotificationMsg attributeUpdateNotification) { + profileTransportConfiguration.getCommunicationConfigs().stream() + .filter(config -> config.getSpec() == SnmpCommunicationSpec.SHARED_ATTRIBUTES_SETTING) + .findFirst() + .ifPresent(communicationConfig -> { + Map sharedAttributes = JsonConverter.toJson(attributeUpdateNotification).entrySet().stream() + .collect(Collectors.toMap( + Map.Entry::getKey, + entry -> entry.getValue().isJsonPrimitive() ? entry.getValue().getAsString() : entry.getValue().toString() + )); + try { + snmpTransportService.sendRequest(this, communicationConfig, sharedAttributes); + } catch (Exception e) { + log.error("Failed to send request with shared attributes to SNMP device {}: {}", getDeviceId(), e.getMessage()); + } + }); } @Override diff --git a/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV2.java b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV2.java new file mode 100644 index 0000000000..d48eaadce3 --- /dev/null +++ b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV2.java @@ -0,0 +1,222 @@ +/** + * Copyright © 2016-2021 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.snmp; + +import org.snmp4j.CommandResponderEvent; +import org.snmp4j.CommunityTarget; +import org.snmp4j.PDU; +import org.snmp4j.Snmp; +import org.snmp4j.Target; +import org.snmp4j.TransportMapping; +import org.snmp4j.agent.BaseAgent; +import org.snmp4j.agent.CommandProcessor; +import org.snmp4j.agent.DuplicateRegistrationException; +import org.snmp4j.agent.MOGroup; +import org.snmp4j.agent.ManagedObject; +import org.snmp4j.agent.mo.MOAccessImpl; +import org.snmp4j.agent.mo.MOScalar; +import org.snmp4j.agent.mo.snmp.RowStatus; +import org.snmp4j.agent.mo.snmp.SnmpCommunityMIB; +import org.snmp4j.agent.mo.snmp.SnmpNotificationMIB; +import org.snmp4j.agent.mo.snmp.SnmpTargetMIB; +import org.snmp4j.agent.mo.snmp.StorageType; +import org.snmp4j.agent.mo.snmp.VacmMIB; +import org.snmp4j.agent.security.MutableVACM; +import org.snmp4j.mp.MPv3; +import org.snmp4j.mp.SnmpConstants; +import org.snmp4j.security.SecurityLevel; +import org.snmp4j.security.SecurityModel; +import org.snmp4j.security.USM; +import org.snmp4j.smi.Address; +import org.snmp4j.smi.GenericAddress; +import org.snmp4j.smi.Integer32; +import org.snmp4j.smi.OID; +import org.snmp4j.smi.OctetString; +import org.snmp4j.smi.UdpAddress; +import org.snmp4j.smi.Variable; +import org.snmp4j.smi.VariableBinding; +import org.snmp4j.transport.TransportMappings; + +import java.io.File; +import java.io.IOException; +import java.util.Map; +import java.util.Scanner; +import java.util.function.Consumer; +import java.util.stream.Collectors; + +public class SnmpDeviceSimulatorV2 extends BaseAgent { + + public static class RequestProcessor extends CommandProcessor { + private final Consumer processor; + + public RequestProcessor(Consumer processor) { + super(new OctetString(MPv3.createLocalEngineID())); + this.processor = processor; + } + + @Override + public void processPdu(CommandResponderEvent event) { + processor.accept(event); + } + } + + public static void main(String[] args) throws IOException { + SnmpDeviceSimulatorV2 device = new SnmpDeviceSimulatorV2(1610, "public"); + + device.start(); + device.setUpMappings(Map.of( + ".1.3.6.1.2.1.1.1.50", "12", + ".1.3.6.1.2.1.2.1.52", "56", + ".1.3.6.1.2.1.3.1.54", "yes", + ".1.3.6.1.2.1.7.1.58", "" + )); + + +// while (true) { +// new Scanner(System.in).nextLine(); +// device.sendTrap("127.0.0.1", 1062, Map.of(".1.3.6.1.2.87.1.56", "12")); +// System.out.println("sent"); +// } + +// Snmp snmp = new Snmp(device.transportMappings[0]); +// device.snmp.addCommandResponder(event -> { +// System.out.println(event); +// }); + + new Scanner(System.in).nextLine(); + } + + + private final Target target; + private final Address address; + private Snmp snmp; + + private final String password; + + public SnmpDeviceSimulatorV2(int port, String password) throws IOException { + super(new File("conf.agent"), new File("bootCounter.agent"), new RequestProcessor(event -> { + System.out.println("aboba"); + ((Snmp) event.getSource()).cancel(event.getPDU(), event1 -> System.out.println("canceled")); + })); + CommunityTarget target = new CommunityTarget(); + target.setCommunity(new OctetString(password)); + this.address = GenericAddress.parse("udp:0.0.0.0/" + port); + target.setAddress(address); + target.setRetries(2); + target.setTimeout(1500); + target.setVersion(SnmpConstants.version2c); + this.target = target; + this.password = password; + } + + public void start() throws IOException { + init(); + addShutdownHook(); + getServer().addContext(new OctetString("public")); + finishInit(); + run(); + sendColdStartNotification(); + snmp = new Snmp(transportMappings[0]); + } + + public void setUpMappings(Map oidToResponseMappings) { + unregisterManagedObject(getSnmpv2MIB()); + oidToResponseMappings.forEach((oid, response) -> { + registerManagedObject(new MOScalar<>(new OID(oid), MOAccessImpl.ACCESS_READ_WRITE, new OctetString(response))); + }); + } + + public void sendTrap(String host, int port, Map values) throws IOException { + PDU pdu = new PDU(); + pdu.addAll(values.entrySet().stream() + .map(entry -> new VariableBinding(new OID(entry.getKey()), new OctetString(entry.getValue()))) + .collect(Collectors.toList())); + pdu.setType(PDU.TRAP); + + CommunityTarget remoteTarget = (CommunityTarget) getTarget().clone(); + remoteTarget.setAddress(new UdpAddress(host + "/" + port)); + + snmp.send(pdu, remoteTarget); + } + + @Override + protected void registerManagedObjects() { + } + + protected void registerManagedObject(ManagedObject mo) { + try { + server.register(mo, null); + } catch (DuplicateRegistrationException ex) { + throw new RuntimeException(ex); + } + } + + protected void unregisterManagedObject(MOGroup moGroup) { + moGroup.unregisterMOs(server, getContext(moGroup)); + } + + @Override + protected void addNotificationTargets(SnmpTargetMIB targetMIB, + SnmpNotificationMIB notificationMIB) { + } + + @Override + protected void addViews(VacmMIB vacm) { + vacm.addGroup(SecurityModel.SECURITY_MODEL_SNMPv2c, new OctetString( + "cpublic"), new OctetString("v1v2group"), + StorageType.nonVolatile); + + vacm.addAccess(new OctetString("v1v2group"), new OctetString("public"), + SecurityModel.SECURITY_MODEL_ANY, SecurityLevel.NOAUTH_NOPRIV, + MutableVACM.VACM_MATCH_EXACT, new OctetString("fullReadView"), + new OctetString("fullWriteView"), new OctetString( + "fullNotifyView"), StorageType.nonVolatile); + + vacm.addViewTreeFamily(new OctetString("fullReadView"), new OID("1.3"), + new OctetString(), VacmMIB.vacmViewIncluded, + StorageType.nonVolatile); + } + + protected void addUsmUser(USM usm) { + } + + protected void initTransportMappings() { + transportMappings = new TransportMapping[]{TransportMappings.getInstance().createTransportMapping(address)}; + } + + protected void unregisterManagedObjects() { + } + + protected void addCommunities(SnmpCommunityMIB communityMIB) { + Variable[] com2sec = new Variable[]{ + new OctetString("public"), + new OctetString("cpublic"), + getAgent().getContextEngineID(), + new OctetString("public"), + new OctetString(), + new Integer32(StorageType.nonVolatile), + new Integer32(RowStatus.active) + }; + SnmpCommunityMIB.SnmpCommunityEntryRow row = communityMIB.getSnmpCommunityEntry().createRow( + new OctetString("public2public").toSubIndex(true), com2sec); + communityMIB.getSnmpCommunityEntry().addRow(row); + } + + public Target getTarget() { + return target; + } + +} diff --git a/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV3.java b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV3.java new file mode 100644 index 0000000000..63ccd7021a --- /dev/null +++ b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV3.java @@ -0,0 +1,52 @@ +/** + * Copyright © 2016-2021 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.snmp; + +import org.snmp4j.UserTarget; +import org.snmp4j.security.AuthSHA; +import org.snmp4j.security.PrivDES; +import org.snmp4j.security.SecurityLevel; +import org.snmp4j.security.SecurityModels; +import org.snmp4j.security.USM; +import org.snmp4j.security.UsmUser; +import org.snmp4j.smi.OID; +import org.snmp4j.smi.OctetString; + +import java.io.IOException; + +public class SnmpDeviceSimulatorV3 extends SnmpDeviceSimulatorV2 { + public SnmpDeviceSimulatorV3(int port, String securityName, String authenticationPassphrase, String privacyPassphrase) throws IOException { + super(12, null); +// super(new File("conf.agent"), new File("bootCounter.agent")); + + USM usm = new USM(); + SecurityModels.getInstance().addSecurityModel(usm); + + OID authenticationProtocol = AuthSHA.ID; + OID privacyProtocol = PrivDES.ID; + + UsmUser user = new UsmUser(new OctetString(securityName), authenticationProtocol, new OctetString(authenticationPassphrase), privacyProtocol, new OctetString(privacyPassphrase)); + } + + public void initV3(UsmUser user, String securityName) { +// snmp.getUSM().addUser(user); + + UserTarget userTarget = new UserTarget(); + userTarget.setSecurityName(new OctetString(securityName)); + userTarget.setSecurityLevel(SecurityLevel.AUTH_PRIV); + } + +} diff --git a/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpTest.java b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpTest.java new file mode 100644 index 0000000000..29a8c085c8 --- /dev/null +++ b/common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpTest.java @@ -0,0 +1,49 @@ +/** + * Copyright © 2016-2021 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.snmp; + +import java.io.IOException; +import java.util.Map; +import java.util.Scanner; + +public class SnmpTest { + public static void main(String[] args) throws IOException { + SnmpDeviceSimulatorV2 device = new SnmpDeviceSimulatorV2(1610, "public"); + + device.start(); + device.setUpMappings(Map.of( + ".1.3.6.1.2.1.1.1.50", "12", + ".1.3.6.1.2.1.2.1.52", "56", + ".1.3.6.1.2.1.3.1.54", "yes", + ".1.3.6.1.2.1.7.1.58", "" + )); + + +// while (true) { +// new Scanner(System.in).nextLine(); +// device.sendTrap("127.0.0.1", 1062, Map.of(".1.3.6.1.2.87.1.56", "12")); +// System.out.println("sent"); +// } + +// Snmp snmp = new Snmp(device.transportMappings[0]); +// device.snmp.addCommandResponder(event -> { +// System.out.println(event); +// }); + + new Scanner(System.in).nextLine(); + } + +} diff --git a/common/transport/snmp/src/test/resources/snmp-device-profile-transport-config.json b/common/transport/snmp/src/test/resources/snmp-device-profile-transport-config.json new file mode 100644 index 0000000000..f8291a090b --- /dev/null +++ b/common/transport/snmp/src/test/resources/snmp-device-profile-transport-config.json @@ -0,0 +1,53 @@ +{ + "timeoutMs": 500, + "retries": 0, + "communicationConfigs": [ + { + "spec": "TELEMETRY_QUERYING", + "queryingFrequencyMs": 3000, + "mappings": [ + { + "oid": ".1.3.6.1.2.1.1.1.50", + "key": "temperature", + "dataType": "LONG" + }, + { + "oid": ".1.3.6.1.2.1.2.1.52", + "key": "humidity", + "dataType": "DOUBLE" + } + ] + }, + { + "spec": "CLIENT_ATTRIBUTES_QUERYING", + "queryingFrequencyMs": 5000, + "mappings": [ + { + "oid": ".1.3.6.1.2.1.3.1.54", + "key": "isCool", + "dataType": "STRING" + } + ] + }, + { + "spec": "SHARED_ATTRIBUTES_SETTING", + "mappings": [ + { + "oid": ".1.3.6.1.2.1.7.1.58", + "key": "shared", + "dataType": "STRING" + } + ] + }, + { + "spec": "TELEMETRY_TRAPS_RECEIVING", + "mappings": [ + { + "oid": ".1.3.6.1.2.8.7.1.56", + "key": "temperature_trap", + "dataType": "LONG" + } + ] + } + ] +} diff --git a/common/transport/snmp/src/test/resources/snmp-device-transport-config.json b/common/transport/snmp/src/test/resources/snmp-device-transport-config.json new file mode 100644 index 0000000000..9a97b75e40 --- /dev/null +++ b/common/transport/snmp/src/test/resources/snmp-device-transport-config.json @@ -0,0 +1,6 @@ +{ + "address": "127.0.0.1", + "port": 1610, + "password": "public", + "protocolVersion": "V2C" +} \ No newline at end of file diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportResourceCache.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportResourceCache.java index 5f793511c6..7c72318de7 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportResourceCache.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportResourceCache.java @@ -15,13 +15,15 @@ */ package org.thingsboard.server.common.transport; -import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.ResourceType; +import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.id.TenantId; +import java.util.Optional; + public interface TransportResourceCache { - TbResource get(TenantId tenantId, ResourceType resourceType, String resourceId); + Optional get(TenantId tenantId, ResourceType resourceType, String resourceId); void update(TenantId tenantId, ResourceType resourceType, String resourceI); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java index 2c5dfafe4e..699e43e5ee 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java @@ -285,7 +285,7 @@ public class JsonConverter { return result; } - public static JsonElement toJson(AttributeUpdateNotificationMsg payload) { + public static JsonObject toJson(AttributeUpdateNotificationMsg payload) { JsonObject result = new JsonObject(); if (payload.getSharedUpdatedCount() > 0) { payload.getSharedUpdatedList().forEach(addToObjectFromProto(result)); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportResourceCache.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportResourceCache.java index 17d898508a..83439530c0 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportResourceCache.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportResourceCache.java @@ -19,8 +19,8 @@ import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.ResourceType; +import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.transport.TransportResourceCache; import org.thingsboard.server.common.transport.TransportService; @@ -53,7 +53,7 @@ public class DefaultTransportResourceCache implements TransportResourceCache { } @Override - public TbResource get(TenantId tenantId, ResourceType resourceType, String resourceKey) { + public Optional get(TenantId tenantId, ResourceType resourceType, String resourceKey) { ResourceCompositeKey compositeKey = new ResourceCompositeKey(tenantId, resourceType, resourceKey); TbResource resource; @@ -79,7 +79,7 @@ public class DefaultTransportResourceCache implements TransportResourceCache { } } - return resource; + return Optional.ofNullable(resource); } private TbResource fetchResource(ResourceCompositeKey compositeKey) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/resource/BaseTbResourceService.java b/dao/src/main/java/org/thingsboard/server/dao/resource/BaseTbResourceService.java index 91de57c849..0f1b39a459 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/resource/BaseTbResourceService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/resource/BaseTbResourceService.java @@ -87,7 +87,7 @@ public class BaseTbResourceService implements TbResourceService { String resourceKey = objectModel.id + LWM2M_SEPARATOR_KEY + objectModel.getVersion(); String name = objectModel.name; resource.setResourceKey(resourceKey); - resource.setTitle(name); + resource.setTitle(name + " id=" +objectModel.id + " v" + objectModel.getVersion()); resource.setSearchText(resourceKey + LWM2M_SEPARATOR_SEARCH_TEXT + name); } else { throw new DataValidationException(String.format("Could not parse the XML of objectModel with name %s", resource.getSearchText())); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseTbResourceServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseTbResourceServiceTest.java index d17fcc88ff..21842af176 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseTbResourceServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseTbResourceServiceTest.java @@ -123,7 +123,7 @@ public abstract class BaseTbResourceServiceTest extends AbstractServiceTest { Assert.assertNotNull(savedResource.getId()); Assert.assertTrue(savedResource.getCreatedTime() > 0); Assert.assertEquals(resource.getTenantId(), savedResource.getTenantId()); - Assert.assertEquals("My first resource", savedResource.getTitle()); + Assert.assertEquals("My first resource id=0 v1.0", savedResource.getTitle()); Assert.assertEquals("0_1.0", savedResource.getResourceKey()); Assert.assertEquals(resource.getData(), savedResource.getData()); diff --git a/transport/snmp/src/main/resources/device-data-config.json b/transport/snmp/src/main/resources/device-data-config.json deleted file mode 100644 index e425ccc3a6..0000000000 --- a/transport/snmp/src/main/resources/device-data-config.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "deviceName": "Thermostat T1", - "snmpConfig": { - "address": "192.168.1.2", - "port": 161, - "community": "U5J=$HWj6f@7", - "protocolVersion": "v2c" - } -} diff --git a/transport/snmp/src/main/resources/device-profile-config.json b/transport/snmp/src/main/resources/device-profile-config.json deleted file mode 100644 index b272913045..0000000000 --- a/transport/snmp/src/main/resources/device-profile-config.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "poolPeriodMs": 10000, - "timeoutMs": 5000, - "retries": 5, - "attributes": [ - { - "key": "snmpNodeManagerEmail", - "type": "STRING", - "method": "get", - "oid": ".1.3.6.1.2.1.1.4.0" - } - ], - "telemetry": [ - { - "key": "snmpNodeSysUpTime", - "type": "LONG", - "method": "get", - "oid": ".1.3.6.1.2.1.1.3.0" - } - ] -} \ No newline at end of file diff --git a/transport/snmp/src/main/resources/tb-snmp-transport.yml b/transport/snmp/src/main/resources/tb-snmp-transport.yml index 9f6bd131ac..db9f79b51a 100644 --- a/transport/snmp/src/main/resources/tb-snmp-transport.yml +++ b/transport/snmp/src/main/resources/tb-snmp-transport.yml @@ -189,6 +189,11 @@ transport: # Local LwM2M transport parameters snmp: enabled: "${SNMP_ENABLED:true}" + response_processing: + # parallelism level for executor (workStealingPool) that is responsible for handling responses from SNMP devices + parallelism_level: "${SNMP_RESPONSE_PROCESSING_PARALLELISM_LEVEL:20}" + # to configure SNMP to work over UDP or TCP + underlying_protocol: "${SNMP_UNDERLYING_PROTOCOL:udp}" queue: type: "${TB_QUEUE_TYPE:in-memory}" # in-memory or kafka (Apache Kafka) or aws-sqs (AWS SQS) or pubsub (PubSub) or service-bus (Azure Service Bus) or rabbitmq (RabbitMQ)