PapaTramp/nettopo-go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: отмена скана при исчерпании соединений Postgres

Browse Source 
- Распознавание 53300/too many clients и др. в isStoreCapacityOrConnExhausted

- SaveHostResult/порты/SNMP: при фатальной ошибке БД — CancelScan; ctx отменяет дальнейший SNMP

- Воркеры при ctx.Done быстро скипают оставшиеся IP без SNMP (дренаж канала)

Made-with: Cursor
This commit is contained in:
PTah
2026-04-13 15:41:16 +10:00
parent 0086302605
commit 3292f77e5c
3 changed files with 97 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/scans/runner.go
+56 -18
View File
@@ -174,6 +174,16 @@ func (r *Runner) run(ctx context.Context, job ScanJob) {
const minProgressWriteInterval = 500 * time.Millisecond
var progWriteMu sync.Mutex
var lastProgressWrite time.Time
var storeFatalOnce sync.Once
storeAbortIfFatal := func(op string, err error) {
if err == nil || !isStoreCapacityOrConnExhausted(err) {
return
}
storeFatalOnce.Do(func() {
log.Printf("scan %s: критическая ошибка БД (%s): %v — отмена скана", job.ID, op, err)
r.CancelScan(job.ID)
})
}
pushProgress := func(force bool) {
mu.Lock()
@@ -207,21 +217,34 @@ func (r *Runner) run(ctx context.Context, job ScanJob) {
worker := func() {
defer wg.Done()
for ip := range workCh {
if ctx.Err() != nil {
continue
}
isUp := probeHost(ip, job.Options.PingTimeoutMS)
checkedAt := time.Now().UTC()
_ = r.store.SaveHostResult(job.ID, HostResult{
if err := r.store.SaveHostResult(job.ID, HostResult{
IP: ip,
IsUp: isUp,
CheckedAt: checkedAt,
})
if isUp && job.Options.PortScanEnabled {
}); err != nil {
storeAbortIfFatal("SaveHostResult", err)
if !isStoreCapacityOrConnExhausted(err) {
log.Printf("scan %s: SaveHostResult %s: %v", job.ID, ip, err)
}
} else if isUp && job.Options.PortScanEnabled {
for _, p := range job.Options.Ports {
_ = r.store.SaveOpenPortResult(job.ID, OpenPortResult{
if err := r.store.SaveOpenPortResult(job.ID, OpenPortResult{
IP: ip,
Port: p,
IsOpen: probeTCPPort(ip, p, job.Options.PingTimeoutMS),
CheckedAt: checkedAt,
})
}); err != nil {
storeAbortIfFatal("SaveOpenPortResult", err)
if !isStoreCapacityOrConnExhausted(err) {
log.Printf("scan %s: SaveOpenPortResult %s:%d: %v", job.ID, ip, p, err)
}
break
}
}
}
@@ -233,22 +256,37 @@ func (r *Runner) run(ctx context.Context, job ScanJob) {
mu.Unlock()
pushProgress(false)
if isUp && r.cfg.SNMPEnabled {
if isUp && r.cfg.SNMPEnabled && ctx.Err() == nil {
snmpRes, lldpRes, ifRes, portDevRes := probeSNMPAndLLDP(ip, r.cfg.SNMPCommunity, checkedAt, job.Options.PingTimeoutMS)
_ = r.store.SaveSNMPResult(job.ID, snmpRes)
for _, lr := range lldpRes {
if err := r.store.SaveLLDPResult(job.ID, lr); err != nil {
log.Printf("scan %s: save LLDP для %s: %v", job.ID, lr.IP, err)
if err := r.store.SaveSNMPResult(job.ID, snmpRes); err != nil {
storeAbortIfFatal("SaveSNMPResult", err)
if !isStoreCapacityOrConnExhausted(err) {
log.Printf("scan %s: SaveSNMPResult %s: %v", job.ID, ip, err)
}
}
for _, iface := range ifRes {
if err := r.store.SaveInterfaceResult(job.ID, iface); err != nil {
log.Printf("scan %s: save interface %s ifIndex=%d: %v", job.ID, iface.IP, iface.IfIndex, err)
} else {
for _, lr := range lldpRes {
if err := r.store.SaveLLDPResult(job.ID, lr); err != nil {
storeAbortIfFatal("SaveLLDPResult", err)
if !isStoreCapacityOrConnExhausted(err) {
log.Printf("scan %s: save LLDP для %s: %v", job.ID, lr.IP, err)
}
}
}
}
for _, pd := range portDevRes {
if err := r.store.SavePortDeviceResult(job.ID, pd); err != nil {
log.Printf("scan %s: save port-device %s ifIndex=%d mac=%s: %v", job.ID, pd.IP, pd.IfIndex, pd.MAC, err)
for _, iface := range ifRes {
if err := r.store.SaveInterfaceResult(job.ID, iface); err != nil {
storeAbortIfFatal("SaveInterfaceResult", err)
if !isStoreCapacityOrConnExhausted(err) {
log.Printf("scan %s: save interface %s ifIndex=%d: %v", job.ID, iface.IP, iface.IfIndex, err)
}
}
}
for _, pd := range portDevRes {
if err := r.store.SavePortDeviceResult(job.ID, pd); err != nil {
storeAbortIfFatal("SavePortDeviceResult", err)
if !isStoreCapacityOrConnExhausted(err) {
log.Printf("scan %s: save port-device %s ifIndex=%d mac=%s: %v", job.ID, pd.IP, pd.IfIndex, pd.MAC, err)
}
}
}
}
}