Browse Source
* Fix merge errors * Implement SNMP transports balancing * Refactor; implement transport device cache * Refactor * Finish up device lifecycle handling implementing; refactor * Refactor * Change base image to thingsboard/openjdk11 for msa snmp transport * Refactor * Change transport services names to upper-casepull/4299/head
Viacheslav Klimov
6 years ago
committed by
GitHub
GitHub
68 changed files with 1923 additions and 553 deletions
@ -0,0 +1,20 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
public interface TbTransportService { |
|||
String getName(); |
|||
} |
|||
@ -0,0 +1,35 @@ |
|||
/** |
|||
* 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.queue.discovery.event; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Getter |
|||
@ToString |
|||
public class ServiceListChangedEvent extends TbApplicationEvent { |
|||
private final List<ServiceInfo> otherServices; |
|||
private final ServiceInfo currentService; |
|||
|
|||
public ServiceListChangedEvent(List<ServiceInfo> otherServices, ServiceInfo currentService) { |
|||
super(otherServices); |
|||
this.otherServices = otherServices; |
|||
this.currentService = currentService; |
|||
} |
|||
} |
|||
@ -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.queue.util; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
|
|||
import java.lang.annotation.ElementType; |
|||
import java.lang.annotation.Retention; |
|||
import java.lang.annotation.RetentionPolicy; |
|||
import java.lang.annotation.Target; |
|||
|
|||
@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.snmp.enabled}'=='true')") |
|||
@Retention(RetentionPolicy.RUNTIME) |
|||
@Target({ElementType.TYPE, ElementType.METHOD}) |
|||
public @interface TbSnmpTransportComponent { |
|||
} |
|||
@ -0,0 +1,81 @@ |
|||
/** |
|||
* 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(); |
|||
} |
|||
} |
|||
@ -1,147 +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 lombok.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.snmp4j.Snmp; |
|||
import org.snmp4j.transport.DefaultUdpTransportMapping; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.boot.context.event.ApplicationReadyEvent; |
|||
import org.springframework.context.event.EventListener; |
|||
import org.springframework.core.annotation.Order; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.server.common.data.DeviceInfo; |
|||
import org.thingsboard.server.common.data.DeviceProfile; |
|||
import org.thingsboard.server.common.data.DeviceTransportType; |
|||
import org.thingsboard.server.common.data.Tenant; |
|||
import org.thingsboard.server.common.data.device.profile.SnmpProfileTransportConfiguration; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.PageDataIterable; |
|||
import org.thingsboard.server.dao.device.DeviceProfileService; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
import org.thingsboard.server.dao.tenant.TenantService; |
|||
import org.thingsboard.server.transport.snmp.session.DeviceSessionCtx; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.io.IOException; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Service("SnmpTransportService") |
|||
@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.snmp.enabled}'=='true')") |
|||
@Slf4j |
|||
public class SnmpTransportService { |
|||
|
|||
private static final int ENTITY_PACK_LIMIT = 1024; |
|||
|
|||
@Autowired |
|||
private SnmpTransportContext snmpTransportContext; |
|||
|
|||
@Autowired |
|||
DeviceProfileService deviceProfileService; |
|||
|
|||
@Autowired |
|||
TenantService tenantService; |
|||
|
|||
@Autowired |
|||
DeviceService deviceService; |
|||
|
|||
@Getter |
|||
private ExecutorService snmpCallbackExecutor; |
|||
private Snmp snmp; |
|||
private ScheduledExecutorService pollingExecutor; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
log.info("Starting SNMP transport..."); |
|||
pollingExecutor = Executors.newScheduledThreadPool(1, ThingsBoardThreadFactory.forName("snmp-polling")); |
|||
//TODO: Set parallelism value in the config
|
|||
snmpCallbackExecutor = Executors.newWorkStealingPool(20); |
|||
initializeSnmp(); |
|||
log.info("SNMP transport started!"); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void shutdown() { |
|||
log.info("Stopping SNMP transport!"); |
|||
if (pollingExecutor != null) { |
|||
pollingExecutor.shutdownNow(); |
|||
} |
|||
if (snmpCallbackExecutor != null) { |
|||
snmpCallbackExecutor.shutdownNow(); |
|||
} |
|||
if (snmp != null) { |
|||
try { |
|||
snmp.close(); |
|||
} catch (IOException e) { |
|||
log.error(e.getMessage(), e); |
|||
} |
|||
} |
|||
log.info("SNMP transport stopped!"); |
|||
} |
|||
|
|||
@EventListener(ApplicationReadyEvent.class) |
|||
@Order(value = 2) |
|||
public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) { |
|||
log.info("Received application ready event. Starting SNMP polling."); |
|||
initSessionCtxList(); |
|||
startPolling(); |
|||
} |
|||
|
|||
private void initializeSnmp() { |
|||
try { |
|||
this.snmp = new Snmp(new DefaultUdpTransportMapping()); |
|||
this.snmp.listen(); |
|||
} catch (IOException e) { |
|||
//TODO: what should be done if transport wasn't initialized?
|
|||
log.error(e.getMessage(), e); |
|||
} |
|||
} |
|||
|
|||
private void initSessionCtxList() { |
|||
//TODO: This approach works for monolith, in cluster the same data will be fetched by each node.
|
|||
for (Tenant tenant : new PageDataIterable<>(tenantService::findTenants, ENTITY_PACK_LIMIT)) { |
|||
TenantId tenantId = tenant.getTenantId(); |
|||
for (DeviceProfile deviceProfile : new PageDataIterable<>(pageLink -> deviceProfileService.findDeviceProfiles(tenantId, pageLink), ENTITY_PACK_LIMIT)) { |
|||
if (DeviceTransportType.SNMP.equals(deviceProfile.getTransportType())) { |
|||
snmpTransportContext.getProfileTransportConfig().put(deviceProfile.getId(), |
|||
(SnmpProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration()); |
|||
initDeviceSessions(deviceProfile); |
|||
} |
|||
} |
|||
} |
|||
snmpTransportContext.initPduListPerProfile(); |
|||
} |
|||
|
|||
private void initDeviceSessions(DeviceProfile deviceProfile) { |
|||
for (DeviceInfo deviceInfo : new PageDataIterable<>(pageLink -> deviceService.findDeviceInfosByTenantIdAndDeviceProfileId(deviceProfile.getTenantId(), deviceProfile.getId(), pageLink), ENTITY_PACK_LIMIT)) { |
|||
snmpTransportContext.updateDeviceSessionCtx(deviceInfo, deviceProfile, snmp); |
|||
} |
|||
} |
|||
|
|||
private void startPolling() { |
|||
//TODO: Get poll period from configuration;
|
|||
int poolPeriodSeconds = 1; |
|||
pollingExecutor.scheduleAtFixedRate(() -> snmpTransportContext.getDeviceSessions().values().forEach(DeviceSessionCtx::executeSnmpRequest), |
|||
0, poolPeriodSeconds, TimeUnit.SECONDS); |
|||
} |
|||
} |
|||
@ -0,0 +1,35 @@ |
|||
/** |
|||
* 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.event; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.queue.discovery.TbApplicationEventListener; |
|||
import org.thingsboard.server.queue.discovery.event.ServiceListChangedEvent; |
|||
import org.thingsboard.server.queue.util.TbSnmpTransportComponent; |
|||
import org.thingsboard.server.transport.snmp.service.SnmpTransportBalancingService; |
|||
|
|||
@TbSnmpTransportComponent |
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class ServiceListChangedEventListener extends TbApplicationEventListener<ServiceListChangedEvent> { |
|||
private final SnmpTransportBalancingService snmpTransportBalancingService; |
|||
|
|||
@Override |
|||
protected void onTbApplicationEvent(ServiceListChangedEvent event) { |
|||
snmpTransportBalancingService.onServiceListChanged(event); |
|||
} |
|||
} |
|||
@ -0,0 +1,24 @@ |
|||
/** |
|||
* 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.event; |
|||
|
|||
import org.thingsboard.server.queue.discovery.event.TbApplicationEvent; |
|||
|
|||
public class SnmpTransportListChangedEvent extends TbApplicationEvent { |
|||
public SnmpTransportListChangedEvent() { |
|||
super(new Object()); |
|||
} |
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* 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.event; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.queue.discovery.TbApplicationEventListener; |
|||
import org.thingsboard.server.queue.util.TbSnmpTransportComponent; |
|||
import org.thingsboard.server.transport.snmp.SnmpTransportContext; |
|||
|
|||
@TbSnmpTransportComponent |
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class SnmpTransportListChangedEventListener extends TbApplicationEventListener<SnmpTransportListChangedEvent> { |
|||
private final SnmpTransportContext snmpTransportContext; |
|||
|
|||
@Override |
|||
protected void onTbApplicationEvent(SnmpTransportListChangedEvent event) { |
|||
snmpTransportContext.onSnmpTransportListChanged(); |
|||
} |
|||
} |
|||
@ -0,0 +1,92 @@ |
|||
/** |
|||
* 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.service; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.device.data.DeviceData; |
|||
import org.thingsboard.server.common.data.device.data.DeviceTransportConfiguration; |
|||
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.transport.TransportService; |
|||
import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.util.TbSnmpTransportComponent; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@TbSnmpTransportComponent |
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class ProtoTransportEntityService { |
|||
private final TransportService transportService; |
|||
private final DataDecodingEncodingService dataDecodingEncodingService; |
|||
|
|||
public Device getDeviceById(DeviceId id) { |
|||
TransportProtos.GetDeviceResponseMsg deviceProto = transportService.getDevice(TransportProtos.GetDeviceRequestMsg.newBuilder() |
|||
.setDeviceIdMSB(id.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(id.getId().getLeastSignificantBits()) |
|||
.build()); |
|||
|
|||
if (deviceProto == null) { |
|||
return null; |
|||
} |
|||
|
|||
DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID( |
|||
deviceProto.getDeviceProfileIdMSB(), deviceProto.getDeviceProfileIdLSB()) |
|||
); |
|||
|
|||
Device device = new Device(); |
|||
device.setId(id); |
|||
device.setDeviceProfileId(deviceProfileId); |
|||
|
|||
DeviceTransportConfiguration deviceTransportConfiguration = (DeviceTransportConfiguration) dataDecodingEncodingService.decode( |
|||
deviceProto.getDeviceTransportConfiguration().toByteArray() |
|||
).orElseThrow(() -> new IllegalStateException("Can't find device transport configuration")); |
|||
|
|||
DeviceData deviceData = new DeviceData(); |
|||
deviceData.setTransportConfiguration(deviceTransportConfiguration); |
|||
device.setDeviceData(deviceData); |
|||
|
|||
return device; |
|||
} |
|||
|
|||
public DeviceCredentials getDeviceCredentialsByDeviceId(DeviceId deviceId) { |
|||
TransportProtos.GetDeviceCredentialsResponseMsg deviceCredentialsResponse = transportService.getDeviceCredentials( |
|||
TransportProtos.GetDeviceCredentialsRequestMsg.newBuilder() |
|||
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
|||
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
|||
.build() |
|||
); |
|||
|
|||
return (DeviceCredentials) dataDecodingEncodingService.decode(deviceCredentialsResponse.getDeviceCredentialsData().toByteArray()) |
|||
.orElseThrow(() -> new IllegalArgumentException("Device credentials not found")); |
|||
} |
|||
|
|||
public List<UUID> getAllSnmpDevicesIds() { |
|||
TransportProtos.GetSnmpDevicesResponseMsg devicesIdsResponse = transportService.getSnmpDevicesIds( |
|||
TransportProtos.GetSnmpDevicesRequestMsg.getDefaultInstance() |
|||
); |
|||
|
|||
return devicesIdsResponse.getIdsList().stream() |
|||
.map(UUID::fromString) |
|||
.collect(Collectors.toList()); |
|||
} |
|||
} |
|||
@ -0,0 +1,92 @@ |
|||
/** |
|||
* 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.service; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.context.ApplicationEventPublisher; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.discovery.event.ServiceListChangedEvent; |
|||
import org.thingsboard.server.queue.util.TbSnmpTransportComponent; |
|||
import org.thingsboard.server.transport.snmp.event.SnmpTransportListChangedEvent; |
|||
|
|||
import java.util.Comparator; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.stream.Collectors; |
|||
import java.util.stream.Stream; |
|||
|
|||
@TbSnmpTransportComponent |
|||
@Service |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class SnmpTransportBalancingService { |
|||
private final PartitionService partitionService; |
|||
private final ApplicationEventPublisher eventPublisher; |
|||
private final SnmpTransportService snmpTransportService; |
|||
|
|||
private int snmpTransportsCount = 1; |
|||
private Integer currentTransportPartitionIndex = 0; |
|||
|
|||
public void onServiceListChanged(ServiceListChangedEvent event) { |
|||
log.trace("Got service list changed event: {}", event); |
|||
recalculatePartitions(event.getOtherServices(), event.getCurrentService()); |
|||
} |
|||
|
|||
public boolean isManagedByCurrentTransport(UUID entityId) { |
|||
boolean isManaged = resolvePartitionIndexForEntity(entityId) == currentTransportPartitionIndex; |
|||
if (!isManaged) { |
|||
log.trace("Entity {} is not managed by current SNMP transport node", entityId); |
|||
} |
|||
return isManaged; |
|||
} |
|||
|
|||
private int resolvePartitionIndexForEntity(UUID entityId) { |
|||
return partitionService.resolvePartitionIndex(entityId, snmpTransportsCount); |
|||
} |
|||
|
|||
private void recalculatePartitions(List<ServiceInfo> otherServices, ServiceInfo currentService) { |
|||
log.info("Recalculating partitions for SNMP transports"); |
|||
List<ServiceInfo> snmpTransports = Stream.concat(otherServices.stream(), Stream.of(currentService)) |
|||
.filter(service -> service.getTransportsList().contains(snmpTransportService.getName())) |
|||
.sorted(Comparator.comparing(ServiceInfo::getServiceId)) |
|||
.collect(Collectors.toList()); |
|||
log.trace("Found SNMP transports: {}", snmpTransports); |
|||
|
|||
int previousCurrentTransportPartitionIndex = currentTransportPartitionIndex; |
|||
int previousSnmpTransportsCount = snmpTransportsCount; |
|||
|
|||
if (!snmpTransports.isEmpty()) { |
|||
for (int i = 0; i < snmpTransports.size(); i++) { |
|||
if (snmpTransports.get(i).equals(currentService)) { |
|||
currentTransportPartitionIndex = i; |
|||
break; |
|||
} |
|||
} |
|||
snmpTransportsCount = snmpTransports.size(); |
|||
} |
|||
|
|||
if (snmpTransportsCount != previousSnmpTransportsCount || currentTransportPartitionIndex != previousCurrentTransportPartitionIndex) { |
|||
log.info("SNMP transports partitions have changed: transports count = {}, current transport partition index = {}", snmpTransportsCount, currentTransportPartitionIndex); |
|||
eventPublisher.publishEvent(new SnmpTransportListChangedEvent()); |
|||
} else { |
|||
log.info("SNMP transports partitions have not changed"); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,166 @@ |
|||
/** |
|||
* 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.session; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.Setter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.snmp4j.CommunityTarget; |
|||
import org.snmp4j.Target; |
|||
import org.snmp4j.event.ResponseEvent; |
|||
import org.snmp4j.event.ResponseListener; |
|||
import org.snmp4j.mp.SnmpConstants; |
|||
import org.snmp4j.smi.GenericAddress; |
|||
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.SnmpProfileTransportConfiguration; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.transport.SessionMsgListener; |
|||
import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.SessionCloseNotificationProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseMsg; |
|||
import org.thingsboard.server.transport.snmp.SnmpTransportContext; |
|||
import org.thingsboard.server.transport.snmp.service.SnmpTransportService; |
|||
|
|||
import java.util.UUID; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
@Slf4j |
|||
public class DeviceSessionContext extends DeviceAwareSessionContext implements SessionMsgListener, ResponseListener { |
|||
@Getter |
|||
private Target target; |
|||
private final String token; |
|||
@Getter |
|||
@Setter |
|||
private SnmpProfileTransportConfiguration profileTransportConfiguration; |
|||
@Getter |
|||
@Setter |
|||
private SnmpDeviceTransportConfiguration deviceTransportConfiguration; |
|||
@Getter |
|||
private final Device device; |
|||
|
|||
private final SnmpTransportContext snmpTransportContext; |
|||
private final SnmpTransportService snmpTransportService; |
|||
|
|||
@Getter |
|||
@Setter |
|||
private long previousRequestExecutedAt = 0; |
|||
private final AtomicInteger msgIdSeq = new AtomicInteger(0); |
|||
private boolean isActive = true; |
|||
|
|||
public DeviceSessionContext(Device device, DeviceProfile deviceProfile, |
|||
String token, SnmpDeviceTransportConfiguration deviceTransportConfiguration, |
|||
SnmpTransportContext snmpTransportContext, SnmpTransportService snmpTransportService) { |
|||
super(UUID.randomUUID()); |
|||
super.setDeviceId(device.getId()); |
|||
super.setDeviceProfile(deviceProfile); |
|||
this.device = device; |
|||
|
|||
this.token = token; |
|||
this.snmpTransportContext = snmpTransportContext; |
|||
this.snmpTransportService = snmpTransportService; |
|||
|
|||
this.profileTransportConfiguration = (SnmpProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration(); |
|||
this.deviceTransportConfiguration = deviceTransportConfiguration; |
|||
|
|||
initTarget(this.profileTransportConfiguration, this.deviceTransportConfiguration); |
|||
} |
|||
|
|||
@Override |
|||
public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto newSessionInfo, DeviceProfile deviceProfile) { |
|||
super.onDeviceProfileUpdate(newSessionInfo, deviceProfile); |
|||
if (isActive) { |
|||
snmpTransportContext.onDeviceProfileUpdated(deviceProfile, this); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onDeviceDeleted(DeviceId deviceId) { |
|||
snmpTransportContext.onDeviceDeleted(this); |
|||
} |
|||
|
|||
@Override |
|||
public void onResponse(ResponseEvent event) { |
|||
if (isActive) { |
|||
snmpTransportService.onNewDeviceResponse(event, this); |
|||
} |
|||
} |
|||
|
|||
public void initTarget(SnmpProfileTransportConfiguration 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(getSnmpVersion(deviceTransportConfig.getProtocolVersion())); |
|||
communityTarget.setCommunity(new OctetString(deviceTransportConfig.getCommunity())); |
|||
communityTarget.setTimeout(profileTransportConfig.getTimeoutMs()); |
|||
communityTarget.setRetries(profileTransportConfig.getRetries()); |
|||
this.target = communityTarget; |
|||
log.info("SNMP target initialized: {}", this.target); |
|||
} |
|||
|
|||
public void close() { |
|||
isActive = false; |
|||
} |
|||
|
|||
public String getToken() { |
|||
return token; |
|||
} |
|||
|
|||
//TODO: replace with enum, wtih preliminary discussion of type version in config (string or integer)
|
|||
private int getSnmpVersion(String configSnmpVersion) { |
|||
switch (configSnmpVersion) { |
|||
case ("v1"): |
|||
return SnmpConstants.version1; |
|||
case ("v2c"): |
|||
return SnmpConstants.version2c; |
|||
case ("v3"): |
|||
return SnmpConstants.version3; |
|||
default: |
|||
return -1; |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public int nextMsgId() { |
|||
return msgIdSeq.incrementAndGet(); |
|||
} |
|||
|
|||
@Override |
|||
public void onGetAttributesResponse(GetAttributeResponseMsg getAttributesResponse) { |
|||
} |
|||
|
|||
@Override |
|||
public void onAttributeUpdate(AttributeUpdateNotificationMsg attributeUpdateNotification) { |
|||
} |
|||
|
|||
@Override |
|||
public void onRemoteSessionCloseCommand(SessionCloseNotificationProto sessionCloseNotification) { |
|||
} |
|||
|
|||
@Override |
|||
public void onToDeviceRpcRequest(ToDeviceRpcRequestMsg toDeviceRequest) { |
|||
} |
|||
|
|||
@Override |
|||
public void onToServerRpcResponse(ToServerRpcResponseMsg toServerResponse) { |
|||
} |
|||
} |
|||
@ -1,206 +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.session; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.Setter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.snmp4j.CommunityTarget; |
|||
import org.snmp4j.Snmp; |
|||
import org.snmp4j.Target; |
|||
import org.snmp4j.mp.SnmpConstants; |
|||
import org.snmp4j.smi.GenericAddress; |
|||
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.DeviceTransportType; |
|||
import org.thingsboard.server.common.data.device.data.SnmpDeviceTransportConfiguration; |
|||
import org.thingsboard.server.common.data.device.profile.SnmpProfileTransportConfiguration; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.transport.SessionMsgListener; |
|||
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.AttributeUpdateNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.SessionCloseNotificationProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseMsg; |
|||
import org.thingsboard.server.transport.snmp.SnmpTransportContext; |
|||
|
|||
import java.io.IOException; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.function.Consumer; |
|||
|
|||
@Slf4j |
|||
public class DeviceSessionCtx extends DeviceAwareSessionContext implements SessionMsgListener { |
|||
private final AtomicInteger msgIdSeq = new AtomicInteger(0); |
|||
|
|||
@Getter |
|||
@Setter |
|||
private SnmpDeviceTransportConfiguration deviceTransportConfig; |
|||
@Getter |
|||
@Setter |
|||
private SnmpSessionListener snmpSessionListener; |
|||
@Getter |
|||
@Setter |
|||
private Target target; |
|||
@Getter |
|||
@Setter |
|||
private volatile TransportProtos.SessionInfoProto sessionInfo; |
|||
|
|||
private Snmp snmp; |
|||
private SnmpProfileTransportConfiguration snmpProfileTransportConfiguration; |
|||
private long previousRequestExecutedAt = 0; |
|||
|
|||
public DeviceSessionCtx(SnmpTransportContext transportContext, String token, SnmpDeviceTransportConfiguration deviceTransportConfig, |
|||
Snmp snmp, DeviceId deviceId, DeviceProfile deviceProfile) { |
|||
super(UUID.randomUUID()); |
|||
this.snmpSessionListener = new SnmpSessionListener(transportContext, token); |
|||
super.setDeviceId(deviceId); |
|||
super.setDeviceProfile(deviceProfile); |
|||
//TODO: What should be done if snmp null?
|
|||
if (snmp != null) { |
|||
this.snmp = snmp; |
|||
} |
|||
this.snmpProfileTransportConfiguration = (SnmpProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration(); |
|||
initTarget(this.snmpProfileTransportConfiguration, deviceTransportConfig); |
|||
} |
|||
|
|||
@Override |
|||
public int nextMsgId() { |
|||
return msgIdSeq.incrementAndGet(); |
|||
} |
|||
|
|||
@Override |
|||
public void onGetAttributesResponse(GetAttributeResponseMsg getAttributesResponse) { |
|||
} |
|||
|
|||
@Override |
|||
public void onAttributeUpdate(AttributeUpdateNotificationMsg attributeUpdateNotification) { |
|||
} |
|||
|
|||
@Override |
|||
public void onRemoteSessionCloseCommand(SessionCloseNotificationProto sessionCloseNotification) { |
|||
} |
|||
|
|||
@Override |
|||
public void onToDeviceRpcRequest(ToDeviceRpcRequestMsg toDeviceRequest) { |
|||
} |
|||
|
|||
@Override |
|||
public void onToServerRpcResponse(ToServerRpcResponseMsg toServerResponse) { |
|||
} |
|||
|
|||
@Override |
|||
public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto newSessionInfo, DeviceProfile deviceProfile) { |
|||
super.onDeviceProfileUpdate(sessionInfo, deviceProfile); |
|||
if (DeviceTransportType.SNMP.equals(deviceProfile.getTransportType())) { |
|||
snmpProfileTransportConfiguration = (SnmpProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration(); |
|||
snmpSessionListener.getSnmpTransportContext().getProfileTransportConfig().put( |
|||
deviceProfile.getId(), |
|||
snmpProfileTransportConfiguration); |
|||
snmpSessionListener.getSnmpTransportContext().updatePduListPerProfile(deviceProfile.getId(), snmpProfileTransportConfiguration); |
|||
} else { |
|||
//TODO: should the context be removed from the map?
|
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) { |
|||
super.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); |
|||
if (super.getDeviceProfile() != null && DeviceTransportType.SNMP.equals(super.getDeviceProfile().getTransportType())) { |
|||
snmpSessionListener.getSnmpTransportContext().updateDeviceSessionCtx(device, deviceProfile, null); |
|||
SnmpProfileTransportConfiguration profileTransportConfig = (SnmpProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration(); |
|||
SnmpDeviceTransportConfiguration deviceTransportConfig = (SnmpDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration(); |
|||
initTarget(profileTransportConfig, deviceTransportConfig); |
|||
} else { |
|||
//TODO: should the context be removed from the map?
|
|||
} |
|||
} |
|||
|
|||
public void createSessionInfo(Consumer<TransportProtos.SessionInfoProto> registerSession) { |
|||
getSnmpSessionListener().getSnmpTransportContext().getTransportService().process(DeviceTransportType.SNMP, |
|||
TransportProtos.ValidateDeviceTokenRequestMsg.newBuilder().setToken(getSnmpSessionListener().getToken()).build(), |
|||
new TransportServiceCallback<ValidateDeviceCredentialsResponse>() { |
|||
@Override |
|||
public void onSuccess(ValidateDeviceCredentialsResponse msg) { |
|||
if (msg.hasDeviceInfo()) { |
|||
sessionInfo = SessionInfoCreator.create(msg, getSnmpSessionListener().getSnmpTransportContext(), UUID.randomUUID()); |
|||
registerSession.accept(sessionInfo); |
|||
setDeviceInfo(msg.getDeviceInfo()); |
|||
} else { |
|||
log.warn("[{}] Failed to process device auth", getDeviceId()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onError(Throwable e) { |
|||
log.warn("[{}] Failed to process device auth", getDeviceId(), e); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
public void executeSnmpRequest() { |
|||
long timeNow = System.currentTimeMillis(); |
|||
long nextRequestExecutionTime = previousRequestExecutedAt + snmpProfileTransportConfiguration.getPoolPeriodMs(); |
|||
if (nextRequestExecutionTime < timeNow) { |
|||
previousRequestExecutedAt = timeNow; |
|||
|
|||
snmpSessionListener.getSnmpTransportContext().getPdusPerProfile().get(deviceProfile.getId()).forEach(pdu -> { |
|||
try { |
|||
log.debug("[{}] Sending SNMP message...", pdu.getRequestID()); |
|||
snmp.send(pdu, |
|||
target, |
|||
deviceProfile.getId(), |
|||
snmpSessionListener); |
|||
} catch (IOException e) { |
|||
log.error(e.getMessage(), e); |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
|
|||
private void initTarget(SnmpProfileTransportConfiguration profileTransportConfig, SnmpDeviceTransportConfiguration deviceTransportConfig) { |
|||
this.deviceTransportConfig = deviceTransportConfig; |
|||
CommunityTarget communityTarget = new CommunityTarget(); |
|||
communityTarget.setAddress(GenericAddress.parse(GenericAddress.TYPE_UDP + ":" + this.deviceTransportConfig.getAddress() + "/" + this.deviceTransportConfig.getPort())); |
|||
communityTarget.setVersion(getSnmpVersion(this.deviceTransportConfig.getProtocolVersion())); |
|||
communityTarget.setCommunity(new OctetString(this.deviceTransportConfig.getCommunity())); |
|||
communityTarget.setTimeout(profileTransportConfig.getTimeoutMs()); |
|||
communityTarget.setRetries(profileTransportConfig.getRetries()); |
|||
this.target = communityTarget; |
|||
log.info("SNMP target initialized: {}", this.target); |
|||
} |
|||
|
|||
//TODO: replace with enum, wtih preliminary discussion of type version in config (string or integer)
|
|||
private int getSnmpVersion(String configSnmpVersion) { |
|||
switch (configSnmpVersion) { |
|||
case ("v1"): |
|||
return SnmpConstants.version1; |
|||
case ("v2c"): |
|||
return SnmpConstants.version2c; |
|||
case ("v3"): |
|||
return SnmpConstants.version3; |
|||
default: |
|||
return -1; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
/** |
|||
* 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.transport; |
|||
|
|||
import lombok.Getter; |
|||
import org.thingsboard.server.common.data.Device; |
|||
|
|||
@Getter |
|||
public class DeviceUpdatedEvent { |
|||
private final Device device; |
|||
|
|||
public DeviceUpdatedEvent(Device device) { |
|||
this.device = device; |
|||
} |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
<?xml version="1.0" encoding="UTF-8" ?> |
|||
<!-- |
|||
|
|||
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. |
|||
|
|||
--> |
|||
<!DOCTYPE configuration> |
|||
<configuration scan="true" scanPeriod="10 seconds"> |
|||
|
|||
<appender name="fileLogAppender" |
|||
class="ch.qos.logback.core.rolling.RollingFileAppender"> |
|||
<file>/var/log/tb-snmp-transport/${TB_SERVICE_ID}/tb-snmp-transport.log</file> |
|||
<rollingPolicy |
|||
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy"> |
|||
<fileNamePattern>/var/log/tb-snmp-transport/${TB_SERVICE_ID}/tb-snmp-transport.%d{yyyy-MM-dd}.%i.log</fileNamePattern> |
|||
<maxFileSize>100MB</maxFileSize> |
|||
<maxHistory>30</maxHistory> |
|||
<totalSizeCap>3GB</totalSizeCap> |
|||
</rollingPolicy> |
|||
<encoder> |
|||
<pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</pattern> |
|||
</encoder> |
|||
</appender> |
|||
|
|||
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> |
|||
<encoder> |
|||
<pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</pattern> |
|||
</encoder> |
|||
</appender> |
|||
|
|||
<logger name="org.thingsboard.server" level="INFO" /> |
|||
|
|||
<root level="INFO"> |
|||
<appender-ref ref="fileLogAppender"/> |
|||
<appender-ref ref="STDOUT"/> |
|||
</root> |
|||
|
|||
</configuration> |
|||
@ -0,0 +1,23 @@ |
|||
# |
|||
# 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. |
|||
# |
|||
|
|||
export JAVA_OPTS="$JAVA_OPTS -Xlog:gc*,heap*,age*,safepoint=debug:file=/var/log/tb-snmp-transport/${TB_SERVICE_ID}-gc.log:time,uptime,level,tags:filecount=10,filesize=10M" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/var/log/tb-snmp-transport/${TB_SERVICE_ID}-heapdump.bin" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:+UseG1GC -XX:MaxGCPauseMillis=500 -XX:+UseStringDeduplication -XX:+ParallelRefProcEnabled -XX:MaxTenuringThreshold=10" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:+ExitOnOutOfMemoryError" |
|||
export LOG_FILENAME=tb-snmp-transport.out |
|||
export LOADER_PATH=/usr/share/tb-snmp-transport/conf |
|||
@ -0,0 +1,45 @@ |
|||
<?xml version="1.0" encoding="UTF-8" ?> |
|||
<!-- |
|||
|
|||
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. |
|||
|
|||
--> |
|||
<!DOCTYPE configuration> |
|||
<configuration> |
|||
|
|||
<appender name="fileLogAppender" |
|||
class="ch.qos.logback.core.rolling.RollingFileAppender"> |
|||
<file>${pkg.logFolder}/${pkg.name}.log</file> |
|||
<rollingPolicy |
|||
class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy"> |
|||
<fileNamePattern>${pkg.logFolder}/${pkg.name}.%d{yyyy-MM-dd}.%i.log</fileNamePattern> |
|||
<maxFileSize>100MB</maxFileSize> |
|||
<maxHistory>30</maxHistory> |
|||
<totalSizeCap>3GB</totalSizeCap> |
|||
</rollingPolicy> |
|||
<encoder> |
|||
<pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</pattern> |
|||
</encoder> |
|||
</appender> |
|||
|
|||
<logger name="org.thingsboard.server" level="INFO" /> |
|||
|
|||
<logger name="com.microsoft.azure.servicebus.primitives.CoreMessageReceiver" level="OFF" /> |
|||
|
|||
<root level="INFO"> |
|||
<appender-ref ref="fileLogAppender"/> |
|||
</root> |
|||
|
|||
</configuration> |
|||
@ -0,0 +1,22 @@ |
|||
# |
|||
# 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. |
|||
# |
|||
|
|||
export JAVA_OPTS="$JAVA_OPTS -Xlog:gc*,heap*,age*,safepoint=debug:file=@pkg.logFolder@/gc.log:time,uptime,level,tags:filecount=10,filesize=10M" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:+IgnoreUnrecognizedVMOptions -XX:+HeapDumpOnOutOfMemoryError" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:-UseBiasedLocking -XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark" |
|||
export JAVA_OPTS="$JAVA_OPTS -XX:+UseG1GC -XX:MaxGCPauseMillis=500 -XX:+UseStringDeduplication -XX:+ParallelRefProcEnabled -XX:MaxTenuringThreshold=10" |
|||
export LOG_FILENAME=${pkg.name}.out |
|||
export LOADER_PATH=${pkg.installFolder}/conf |
|||
@ -0,0 +1,36 @@ |
|||
<?xml version="1.0" encoding="UTF-8" ?> |
|||
<!-- |
|||
|
|||
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. |
|||
|
|||
--> |
|||
<!DOCTYPE configuration> |
|||
<configuration scan="true" scanPeriod="10 seconds"> |
|||
|
|||
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> |
|||
<encoder> |
|||
<pattern>%d{ISO8601} [%thread] %-5level %logger{36} - %msg%n</pattern> |
|||
</encoder> |
|||
</appender> |
|||
|
|||
<logger name="org.thingsboard.server" level="TRACE" /> |
|||
|
|||
<logger name="com.microsoft.azure.servicebus.primitives.CoreMessageReceiver" level="OFF" /> |
|||
|
|||
<root level="INFO"> |
|||
<appender-ref ref="STDOUT"/> |
|||
</root> |
|||
|
|||
</configuration> |
|||
@ -0,0 +1,394 @@ |
|||
# |
|||
# 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. |
|||
# |
|||
|
|||
server: |
|||
# Server bind address |
|||
address: "${HTTP_BIND_ADDRESS:0.0.0.0}" |
|||
# Server bind port |
|||
port: "${HTTP_BIND_PORT:8085}" |
|||
# Server SSL configuration |
|||
|
|||
# Zookeeper connection parameters. Used for service discovery. |
|||
zk: |
|||
# Enable/disable zookeeper discovery service. |
|||
enabled: "${ZOOKEEPER_ENABLED:false}" |
|||
# Zookeeper connect string |
|||
url: "${ZOOKEEPER_URL:localhost:2181}" |
|||
# Zookeeper retry interval in milliseconds |
|||
retry_interval_ms: "${ZOOKEEPER_RETRY_INTERVAL_MS:3000}" |
|||
# Zookeeper connection timeout in milliseconds |
|||
connection_timeout_ms: "${ZOOKEEPER_CONNECTION_TIMEOUT_MS:3000}" |
|||
# Zookeeper session timeout in milliseconds |
|||
session_timeout_ms: "${ZOOKEEPER_SESSION_TIMEOUT_MS:3000}" |
|||
# Name of the directory in zookeeper 'filesystem' |
|||
zk_dir: "${ZOOKEEPER_NODES_DIR:/thingsboard}" |
|||
|
|||
cluster: |
|||
stats: |
|||
enabled: "${TB_CLUSTER_STATS_ENABLED:false}" |
|||
print_interval_ms: "${TB_CLUSTER_STATS_PRINT_INTERVAL_MS:10000}" |
|||
|
|||
cache: |
|||
# caffeine or redis |
|||
type: "${CACHE_TYPE:caffeine}" |
|||
attributes: |
|||
# make sure that if cache.type is 'redis' and cache.attributes.enabled is 'true' that you change 'maxmemory-policy' Redis config property to 'allkeys-lru', 'allkeys-lfu' or 'allkeys-random' |
|||
enabled: "${CACHE_ATTRIBUTES_ENABLED:true}" |
|||
|
|||
caffeine: |
|||
specs: |
|||
relations: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
deviceCredentials: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
devices: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
sessions: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
assets: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
entityViews: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
claimDevices: |
|||
timeToLiveInMinutes: 1 |
|||
maxSize: 0 |
|||
securitySettings: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
tenantProfiles: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
deviceProfiles: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 0 |
|||
attributes: |
|||
timeToLiveInMinutes: 1440 |
|||
maxSize: 100000 |
|||
|
|||
redis: |
|||
# standalone or cluster |
|||
connection: |
|||
type: "${REDIS_CONNECTION_TYPE:standalone}" |
|||
standalone: |
|||
host: "${REDIS_HOST:localhost}" |
|||
port: "${REDIS_PORT:6379}" |
|||
useDefaultClientConfig: "${REDIS_USE_DEFAULT_CLIENT_CONFIG:true}" |
|||
# this value may be used only if you used not default ClientConfig |
|||
clientName: "${REDIS_CLIENT_NAME:standalone}" |
|||
# this value may be used only if you used not default ClientConfig |
|||
connectTimeout: "${REDIS_CLIENT_CONNECT_TIMEOUT:30000}" |
|||
# this value may be used only if you used not default ClientConfig |
|||
readTimeout: "${REDIS_CLIENT_READ_TIMEOUT:60000}" |
|||
# this value may be used only if you used not default ClientConfig |
|||
usePoolConfig: "${REDIS_CLIENT_USE_POOL_CONFIG:false}" |
|||
cluster: |
|||
# Comma-separated list of "host:port" pairs to bootstrap from. |
|||
nodes: "${REDIS_NODES:}" |
|||
# Maximum number of redirects to follow when executing commands across the cluster. |
|||
max-redirects: "${REDIS_MAX_REDIRECTS:12}" |
|||
useDefaultPoolConfig: "${REDIS_USE_DEFAULT_POOL_CONFIG:true}" |
|||
# db index |
|||
db: "${REDIS_DB:0}" |
|||
# db password |
|||
password: "${REDIS_PASSWORD:}" |
|||
# pool config |
|||
pool_config: |
|||
maxTotal: "${REDIS_POOL_CONFIG_MAX_TOTAL:128}" |
|||
maxIdle: "${REDIS_POOL_CONFIG_MAX_IDLE:128}" |
|||
minIdle: "${REDIS_POOL_CONFIG_MIN_IDLE:16}" |
|||
testOnBorrow: "${REDIS_POOL_CONFIG_TEST_ON_BORROW:true}" |
|||
testOnReturn: "${REDIS_POOL_CONFIG_TEST_ON_RETURN:true}" |
|||
testWhileIdle: "${REDIS_POOL_CONFIG_TEST_WHILE_IDLE:true}" |
|||
minEvictableMs: "${REDIS_POOL_CONFIG_MIN_EVICTABLE_MS:60000}" |
|||
evictionRunsMs: "${REDIS_POOL_CONFIG_EVICTION_RUNS_MS:30000}" |
|||
maxWaitMills: "${REDIS_POOL_CONFIG_MAX_WAIT_MS:60000}" |
|||
numberTestsPerEvictionRun: "${REDIS_POOL_CONFIG_NUMBER_TESTS_PER_EVICTION_RUN:3}" |
|||
blockWhenExhausted: "${REDIS_POOL_CONFIG_BLOCK_WHEN_EXHAUSTED:true}" |
|||
|
|||
# Check new version updates parameters |
|||
updates: |
|||
# Enable/disable updates checking. |
|||
enabled: "${UPDATES_ENABLED:true}" |
|||
|
|||
# spring freemarker configuration |
|||
spring.freemarker.checkTemplateLocation: "false" |
|||
|
|||
audit-log: |
|||
# Enable/disable audit log functionality. |
|||
enabled: "${AUDIT_LOG_ENABLED:true}" |
|||
# Specify partitioning size for audit log by tenant id storage. Example MINUTES, HOURS, DAYS, MONTHS |
|||
by_tenant_partitioning: "${AUDIT_LOG_BY_TENANT_PARTITIONING:MONTHS}" |
|||
# Number of days as history period if startTime and endTime are not specified |
|||
default_query_period: "${AUDIT_LOG_DEFAULT_QUERY_PERIOD:30}" |
|||
# Logging levels per each entity type. |
|||
# Allowed values: OFF (disable), W (log write operations), RW (log read and write operations) |
|||
logging-level: |
|||
mask: |
|||
"device": "${AUDIT_LOG_MASK_DEVICE:W}" |
|||
"asset": "${AUDIT_LOG_MASK_ASSET:W}" |
|||
"dashboard": "${AUDIT_LOG_MASK_DASHBOARD:W}" |
|||
"customer": "${AUDIT_LOG_MASK_CUSTOMER:W}" |
|||
"user": "${AUDIT_LOG_MASK_USER:W}" |
|||
"rule_chain": "${AUDIT_LOG_MASK_RULE_CHAIN:W}" |
|||
"alarm": "${AUDIT_LOG_MASK_ALARM:W}" |
|||
"entity_view": "${AUDIT_LOG_MASK_ENTITY_VIEW:W}" |
|||
"device_profile": "${AUDIT_LOG_MASK_DEVICE_PROFILE:W}" |
|||
sink: |
|||
# Type of external sink. possible options: none, elasticsearch |
|||
type: "${AUDIT_LOG_SINK_TYPE:none}" |
|||
# Name of the index where audit logs stored |
|||
# Index name could contain next placeholders (not mandatory): |
|||
# @{TENANT} - substituted by tenant ID |
|||
# @{DATE} - substituted by current date in format provided in audit_log.sink.date_format |
|||
index_pattern: "${AUDIT_LOG_SINK_INDEX_PATTERN:@{TENANT}_AUDIT_LOG_@{DATE}}" |
|||
# Date format. Details of the pattern could be found here: |
|||
# https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html |
|||
date_format: "${AUDIT_LOG_SINK_DATE_FORMAT:YYYY.MM.DD}" |
|||
scheme_name: "${AUDIT_LOG_SINK_SCHEME_NAME:http}" # http or https |
|||
host: "${AUDIT_LOG_SINK_HOST:localhost}" |
|||
port: "${AUDIT_LOG_SINK_PORT:9200}" |
|||
user_name: "${AUDIT_LOG_SINK_USER_NAME:}" |
|||
password: "${AUDIT_LOG_SINK_PASSWORD:}" |
|||
|
|||
state: |
|||
# Should be greater then transport.sessions.report_timeout |
|||
defaultInactivityTimeoutInSec: "${DEFAULT_INACTIVITY_TIMEOUT:600}" |
|||
defaultStateCheckIntervalInSec: "${DEFAULT_STATE_CHECK_INTERVAL:60}" |
|||
persistToTelemetry: "${PERSIST_STATE_TO_TELEMETRY:false}" |
|||
|
|||
transport: |
|||
sessions: |
|||
inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}" |
|||
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}" |
|||
json: |
|||
# Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON |
|||
type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}" |
|||
# Maximum allowed string value length when processing Telemetry/Attributes JSON (0 value disables string value length check) |
|||
max_string_value_length: "${JSON_MAX_STRING_VALUE_LENGTH:0}" |
|||
# Enable/disable http/mqtt/coap transport protocols (has higher priority than certain protocol's 'enabled' property) |
|||
api_enabled: "${TB_TRANSPORT_API_ENABLED:true}" |
|||
# Local LwM2M transport parameters |
|||
snmp: |
|||
enabled: "${SNMP_ENABLED:true}" |
|||
|
|||
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) |
|||
in_memory: |
|||
stats: |
|||
# For debug lvl |
|||
print-interval-ms: "${TB_QUEUE_IN_MEMORY_STATS_PRINT_INTERVAL_MS:60000}" |
|||
kafka: |
|||
bootstrap.servers: "${TB_KAFKA_SERVERS:localhost:9092}" |
|||
acks: "${TB_KAFKA_ACKS:all}" |
|||
retries: "${TB_KAFKA_RETRIES:1}" |
|||
batch.size: "${TB_KAFKA_BATCH_SIZE:16384}" |
|||
linger.ms: "${TB_KAFKA_LINGER_MS:1}" |
|||
buffer.memory: "${TB_BUFFER_MEMORY:33554432}" |
|||
replication_factor: "${TB_QUEUE_KAFKA_REPLICATION_FACTOR:1}" |
|||
max_poll_interval_ms: "${TB_QUEUE_KAFKA_MAX_POLL_INTERVAL_MS:300000}" |
|||
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}" |
|||
use_confluent_cloud: "${TB_QUEUE_KAFKA_USE_CONFLUENT_CLOUD:false}" |
|||
confluent: |
|||
ssl.algorithm: "${TB_QUEUE_KAFKA_CONFLUENT_SSL_ALGORITHM:https}" |
|||
sasl.mechanism: "${TB_QUEUE_KAFKA_CONFLUENT_SASL_MECHANISM:PLAIN}" |
|||
sasl.config: "${TB_QUEUE_KAFKA_CONFLUENT_SASL_JAAS_CONFIG:org.apache.kafka.common.security.plain.PlainLoginModule required username=\"CLUSTER_API_KEY\" password=\"CLUSTER_API_SECRET\";}" |
|||
security.protocol: "${TB_QUEUE_KAFKA_CONFLUENT_SECURITY_PROTOCOL:SASL_SSL}" |
|||
other: |
|||
topic-properties: |
|||
rule-engine: "${TB_QUEUE_KAFKA_RE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}" |
|||
core: "${TB_QUEUE_KAFKA_CORE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}" |
|||
transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}" |
|||
notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}" |
|||
js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600;partitions:100}" |
|||
consumer-stats: |
|||
enabled: "${TB_QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}" |
|||
print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}" |
|||
kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}" |
|||
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}" |
|||
threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" |
|||
queue-properties: |
|||
rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" |
|||
core: "${TB_QUEUE_AWS_SQS_CORE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" |
|||
transport-api: "${TB_QUEUE_AWS_SQS_TA_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" |
|||
notifications: "${TB_QUEUE_AWS_SQS_NOTIFICATIONS_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" |
|||
js-executor: "${TB_QUEUE_AWS_SQS_JE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" |
|||
# VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds |
|||
pubsub: |
|||
project_id: "${TB_QUEUE_PUBSUB_PROJECT_ID:YOUR_PROJECT_ID}" |
|||
service_account: "${TB_QUEUE_PUBSUB_SERVICE_ACCOUNT:YOUR_SERVICE_ACCOUNT}" |
|||
max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" #in bytes |
|||
max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}" |
|||
queue-properties: |
|||
rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}" |
|||
core: "${TB_QUEUE_PUBSUB_CORE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}" |
|||
transport-api: "${TB_QUEUE_PUBSUB_TA_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}" |
|||
notifications: "${TB_QUEUE_PUBSUB_NOTIFICATIONS_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}" |
|||
js-executor: "${TB_QUEUE_PUBSUB_JE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}" |
|||
service_bus: |
|||
namespace_name: "${TB_QUEUE_SERVICE_BUS_NAMESPACE_NAME:YOUR_NAMESPACE_NAME}" |
|||
sas_key_name: "${TB_QUEUE_SERVICE_BUS_SAS_KEY_NAME:YOUR_SAS_KEY_NAME}" |
|||
sas_key: "${TB_QUEUE_SERVICE_BUS_SAS_KEY:YOUR_SAS_KEY}" |
|||
max_messages: "${TB_QUEUE_SERVICE_BUS_MAX_MESSAGES:1000}" |
|||
queue-properties: |
|||
rule-engine: "${TB_QUEUE_SERVICE_BUS_RE_QUEUE_PROPERTIES:lockDurationInSec:30;maxSizeInMb:1024;messageTimeToLiveInSec:604800}" |
|||
core: "${TB_QUEUE_SERVICE_BUS_CORE_QUEUE_PROPERTIES:lockDurationInSec:30;maxSizeInMb:1024;messageTimeToLiveInSec:604800}" |
|||
transport-api: "${TB_QUEUE_SERVICE_BUS_TA_QUEUE_PROPERTIES:lockDurationInSec:30;maxSizeInMb:1024;messageTimeToLiveInSec:604800}" |
|||
notifications: "${TB_QUEUE_SERVICE_BUS_NOTIFICATIONS_QUEUE_PROPERTIES:lockDurationInSec:30;maxSizeInMb:1024;messageTimeToLiveInSec:604800}" |
|||
js-executor: "${TB_QUEUE_SERVICE_BUS_JE_QUEUE_PROPERTIES:lockDurationInSec:30;maxSizeInMb:1024;messageTimeToLiveInSec:604800}" |
|||
rabbitmq: |
|||
exchange_name: "${TB_QUEUE_RABBIT_MQ_EXCHANGE_NAME:}" |
|||
host: "${TB_QUEUE_RABBIT_MQ_HOST:localhost}" |
|||
port: "${TB_QUEUE_RABBIT_MQ_PORT:5672}" |
|||
virtual_host: "${TB_QUEUE_RABBIT_MQ_VIRTUAL_HOST:/}" |
|||
username: "${TB_QUEUE_RABBIT_MQ_USERNAME:YOUR_USERNAME}" |
|||
password: "${TB_QUEUE_RABBIT_MQ_PASSWORD:YOUR_PASSWORD}" |
|||
automatic_recovery_enabled: "${TB_QUEUE_RABBIT_MQ_AUTOMATIC_RECOVERY_ENABLED:false}" |
|||
connection_timeout: "${TB_QUEUE_RABBIT_MQ_CONNECTION_TIMEOUT:60000}" |
|||
handshake_timeout: "${TB_QUEUE_RABBIT_MQ_HANDSHAKE_TIMEOUT:10000}" |
|||
queue-properties: |
|||
rule-engine: "${TB_QUEUE_RABBIT_MQ_RE_QUEUE_PROPERTIES:x-max-length-bytes:1048576000;x-message-ttl:604800000}" |
|||
core: "${TB_QUEUE_RABBIT_MQ_CORE_QUEUE_PROPERTIES:x-max-length-bytes:1048576000;x-message-ttl:604800000}" |
|||
transport-api: "${TB_QUEUE_RABBIT_MQ_TA_QUEUE_PROPERTIES:x-max-length-bytes:1048576000;x-message-ttl:604800000}" |
|||
notifications: "${TB_QUEUE_RABBIT_MQ_NOTIFICATIONS_QUEUE_PROPERTIES:x-max-length-bytes:1048576000;x-message-ttl:604800000}" |
|||
js-executor: "${TB_QUEUE_RABBIT_MQ_JE_QUEUE_PROPERTIES:x-max-length-bytes:1048576000;x-message-ttl:604800000}" |
|||
partitions: |
|||
hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256 |
|||
transport_api: |
|||
requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb_transport.api.requests}" |
|||
responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb_transport.api.responses}" |
|||
max_pending_requests: "${TB_QUEUE_TRANSPORT_MAX_PENDING_REQUESTS:10000}" |
|||
max_requests_timeout: "${TB_QUEUE_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}" |
|||
max_callback_threads: "${TB_QUEUE_TRANSPORT_MAX_CALLBACK_THREADS:100}" |
|||
request_poll_interval: "${TB_QUEUE_TRANSPORT_REQUEST_POLL_INTERVAL_MS:25}" |
|||
response_poll_interval: "${TB_QUEUE_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}" |
|||
core: |
|||
topic: "${TB_QUEUE_CORE_TOPIC:tb_core}" |
|||
poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" |
|||
partitions: "${TB_QUEUE_CORE_PARTITIONS:10}" |
|||
pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:2000}" |
|||
usage-stats-topic: "${TB_QUEUE_US_TOPIC:tb_usage_stats}" |
|||
stats: |
|||
enabled: "${TB_QUEUE_CORE_STATS_ENABLED:true}" |
|||
print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:60000}" |
|||
js: |
|||
# JS Eval request topic |
|||
request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js_eval.requests}" |
|||
# JS Eval responses topic prefix that is combined with node id |
|||
response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js_eval.responses}" |
|||
# JS Eval max pending requests |
|||
max_pending_requests: "${REMOTE_JS_MAX_PENDING_REQUESTS:10000}" |
|||
# JS Eval max request timeout |
|||
max_eval_requests_timeout: "${REMOTE_JS_MAX_EVAL_REQUEST_TIMEOUT:60000}" |
|||
# JS max request timeout |
|||
max_requests_timeout: "${REMOTE_JS_MAX_REQUEST_TIMEOUT:10000}" |
|||
# JS response poll interval |
|||
response_poll_interval: "${REMOTE_JS_RESPONSE_POLL_INTERVAL_MS:25}" |
|||
rule-engine: |
|||
topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}" |
|||
poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" |
|||
pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:2000}" |
|||
stats: |
|||
enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:true}" |
|||
print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:60000}" |
|||
queues: |
|||
- name: "${TB_QUEUE_RE_MAIN_QUEUE_NAME:Main}" |
|||
topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}" |
|||
poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}" |
|||
partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}" |
|||
pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:2000}" |
|||
submit-strategy: |
|||
type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL |
|||
# For BATCH only |
|||
batch-size: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_BATCH_SIZE:1000}" # Maximum number of messages in batch |
|||
processing-strategy: |
|||
type: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_TYPE:SKIP_ALL_FAILURES}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT |
|||
# For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT |
|||
retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited |
|||
failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; |
|||
pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; |
|||
max-pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:3}"# Max allowed time in seconds for pause between retries. |
|||
- name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" |
|||
topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" |
|||
poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" |
|||
partitions: "${TB_QUEUE_RE_HP_PARTITIONS:10}" |
|||
pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:2000}" |
|||
submit-strategy: |
|||
type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL |
|||
# For BATCH only |
|||
batch-size: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch |
|||
processing-strategy: |
|||
type: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT |
|||
# For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT |
|||
retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited |
|||
failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; |
|||
pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; |
|||
max-pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:5}"# Max allowed time in seconds for pause between retries. |
|||
- name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}" |
|||
topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}" |
|||
poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}" |
|||
partitions: "${TB_QUEUE_RE_SQ_PARTITIONS:10}" |
|||
pack-processing-timeout: "${TB_QUEUE_RE_SQ_PACK_PROCESSING_TIMEOUT_MS:2000}" |
|||
submit-strategy: |
|||
type: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_BY_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL |
|||
# For BATCH only |
|||
batch-size: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch |
|||
processing-strategy: |
|||
type: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT |
|||
# For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT |
|||
retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited |
|||
failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; |
|||
pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; |
|||
max-pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_MAX_RETRY_PAUSE:5}"# Max allowed time in seconds for pause between retries. |
|||
transport: |
|||
# For high priority notifications that require minimum latency and processing time |
|||
notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" |
|||
poll_interval: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_POLL_INTERVAL_MS:25}" |
|||
|
|||
event: |
|||
debug: |
|||
max-symbols: "${TB_MAX_DEBUG_EVENT_SYMBOLS:4096}" |
|||
|
|||
service: |
|||
type: "${TB_SERVICE_TYPE:tb-transport}" |
|||
# Unique id for this service (autogenerated if empty) |
|||
id: "${TB_SERVICE_ID:}" |
|||
tenant_id: "${TB_SERVICE_TENANT_ID:}" # empty or specific tenant id. |
|||
|
|||
metrics: |
|||
# Enable/disable actuator metrics. |
|||
enabled: "${METRICS_ENABLED:false}" |
|||
timer: |
|||
# Metrics percentiles returned by actuator for timer metrics. List of double values (divided by ,). |
|||
percentiles: "${METRICS_TIMER_PERCENTILES:0.5}" |
|||
|
|||
management: |
|||
endpoints: |
|||
web: |
|||
exposure: |
|||
# Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics). |
|||
include: '${METRICS_ENDPOINTS_EXPOSE:info}' |
|||
Loading…
Reference in new issue