Browse Source

refactoring

pull/14327/head
Artem Barysh 10 months ago
parent
commit
7dbea4a756
  1. 4
      application/src/main/resources/thingsboard.yml
  2. 72
      common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java

4
application/src/main/resources/thingsboard.yml

@ -1295,8 +1295,8 @@ transport:
ignore_type_cast_errors: "${SNMP_RESPONSE_IGNORE_TYPE_CAST_ERRORS:false}" ignore_type_cast_errors: "${SNMP_RESPONSE_IGNORE_TYPE_CAST_ERRORS:false}"
# Thread pool size for scheduler that executes device querying tasks # Thread pool size for scheduler that executes device querying tasks
scheduler_thread_pool_size: "${SNMP_SCHEDULER_THREAD_POOL_SIZE:4}" scheduler_thread_pool_size: "${SNMP_SCHEDULER_THREAD_POOL_SIZE:4}"
# Maximum number of retry attempts for SNMP bootstrap during startup # Maximum number of retry attempts for a single SNMP devices batch during bootstrap.
bootstrap_retries: "${SNMP_BOOTSTRAP_RETRIES:8}" batch_retries: "${SNMP_BOOTSTRAP_RETRIES:8}"
stats: stats:
# Enable/Disable the collection of transport statistics # Enable/Disable the collection of transport statistics
enabled: "${TB_TRANSPORT_STATS_ENABLED:true}" enabled: "${TB_TRANSPORT_STATS_ENABLED:true}"

72
common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java

@ -15,6 +15,7 @@
*/ */
package org.thingsboard.server.transport.snmp; package org.thingsboard.server.transport.snmp;
import jakarta.annotation.PreDestroy;
import lombok.Getter; import lombok.Getter;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -60,6 +61,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@TbSnmpTransportComponent @TbSnmpTransportComponent
@Component @Component
@ -79,8 +81,8 @@ public class SnmpTransportContext extends TransportContext {
private final Set<DeviceId> allSnmpDevicesIds = ConcurrentHashMap.newKeySet(); private final Set<DeviceId> allSnmpDevicesIds = ConcurrentHashMap.newKeySet();
private final ExecutorService snmpExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("snmp-bootstrap")); private final ExecutorService snmpExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("snmp-bootstrap"));
@Value("${transport.snmp.bootstrap_retries}") @Value("${transport.snmp.batch_retries}")
private int snmpBootstrapMaxRetries; private int snmpBootstrapBatchRetries;
@AfterStartUp(order = AfterStartUp.AFTER_TRANSPORT_SERVICE) @AfterStartUp(order = AfterStartUp.AFTER_TRANSPORT_SERVICE)
public void fetchDevicesAndEstablishSessions() { public void fetchDevicesAndEstablishSessions() {
@ -88,44 +90,45 @@ public class SnmpTransportContext extends TransportContext {
} }
private void bootstrapWithRetries() { private void bootstrapWithRetries() {
for (int attempt = 1; attempt <= snmpBootstrapMaxRetries; attempt++) {
try {
doBootstrap();
return;
} catch (Exception e) {
if (attempt >= snmpBootstrapMaxRetries) {
log.error("SNMP bootstrap failed after {} attempts.", attempt, e);
return;
}
log.warn("SNMP bootstrap attempt {}/{} failed. Retrying immediately...", attempt, snmpBootstrapMaxRetries, e);
}
}
}
private void doBootstrap() {
log.info("Initializing SNMP devices sessions"); log.info("Initializing SNMP devices sessions");
int batchIndex = 0; int batchIndex = 0;
int batchSize = 512; int batchSize = 512;
boolean nextBatchExists = true; boolean nextBatchExists = true;
while (nextBatchExists) { while (nextBatchExists) {
TransportProtos.GetSnmpDevicesResponseMsg snmpDevicesResponse = protoEntityService.getSnmpDevicesIds(batchIndex, batchSize); for (int attempt = 1; attempt <= snmpBootstrapBatchRetries; attempt++) {
snmpDevicesResponse.getIdsList().stream() try {
.map(id -> new DeviceId(UUID.fromString(id))) TransportProtos.GetSnmpDevicesResponseMsg snmpDevicesResponse = protoEntityService.getSnmpDevicesIds(batchIndex, batchSize);
.peek(allSnmpDevicesIds::add) snmpDevicesResponse.getIdsList().stream()
.filter(deviceId -> balancingService.isManagedByCurrentTransport(deviceId.getId())) .map(id -> new DeviceId(UUID.fromString(id)))
.map(protoEntityService::getDeviceById) .peek(allSnmpDevicesIds::add)
.forEach(device -> { .filter(deviceId -> balancingService.isManagedByCurrentTransport(deviceId.getId()))
if (!sessions.containsKey(device.getId())) { .map(protoEntityService::getDeviceById)
getExecutor().execute(() -> establishDeviceSession(device)); .forEach(device -> getExecutor().execute(() -> establishDeviceSession(device)));
} nextBatchExists = snmpDevicesResponse.getHasNextPage();
}); batchIndex++;
break;
} catch (Exception e) {
if (e instanceof InterruptedException) {
log.warn("SNMP bootstrap interrupted. Stopping bootstrap task.", e);
return;
}
nextBatchExists = snmpDevicesResponse.getHasNextPage(); if (attempt >= snmpBootstrapBatchRetries) {
batchIndex++; log.error("SNMP bootstrap: batch {} failed after {} attempts.", batchIndex, attempt, e);
return;
}
log.warn("SNMP bootstrap: batch {} attempt {}/{} failed.", batchIndex, attempt, snmpBootstrapBatchRetries, e);
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException ex) {
log.warn("SNMP bootstrap interrupted. Stopping bootstrap task.");
return;
}
}
}
} }
log.debug("Found SNMP devices ids: {}", allSnmpDevicesIds);
log.debug("Found all SNMP devices ids: {}", allSnmpDevicesIds);
} }
private void establishDeviceSession(Device device) { private void establishDeviceSession(Device device) {
@ -330,4 +333,9 @@ public class SnmpTransportContext extends TransportContext {
return sessions.values(); return sessions.values();
} }
@PreDestroy
public void destroy() {
snmpExecutor.shutdown();
}
} }

Loading…
Cancel
Save