batch SNMP results before sending over WebSocket
Accumulate SNMP results and flush in batches of 50 or every 100ms, whichever comes first. Reduces per-message overhead when thousands of devices produce results near-simultaneously. Also flushes remaining results on disconnect/shutdown.
This commit is contained in:
parent
82f5698802
commit
da8b644f40
1 changed files with 24 additions and 2 deletions
26
agent.go
26
agent.go
|
|
@ -173,6 +173,8 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
defer heartbeatTicker.Stop()
|
defer heartbeatTicker.Stop()
|
||||||
channelHeartbeatTicker := time.NewTicker(25 * time.Second)
|
channelHeartbeatTicker := time.NewTicker(25 * time.Second)
|
||||||
defer channelHeartbeatTicker.Stop()
|
defer channelHeartbeatTicker.Stop()
|
||||||
|
flushTicker := time.NewTicker(100 * time.Millisecond)
|
||||||
|
defer flushTicker.Stop()
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
|
|
@ -183,13 +185,28 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
writerWg.Wait()
|
writerWg.Wait()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
var snmpBatch []*pb.SnmpResult
|
||||||
|
|
||||||
|
flushSnmpBatch := func() {
|
||||||
|
if len(snmpBatch) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
for _, r := range snmpBatch {
|
||||||
|
sendBinaryResult("result", r)
|
||||||
|
}
|
||||||
|
slog.Info("flushed snmp results", "count", len(snmpBatch))
|
||||||
|
snmpBatch = snmpBatch[:0]
|
||||||
|
}
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
slog.Info("shutdown signal, closing connection")
|
slog.Info("shutdown signal, closing connection")
|
||||||
|
flushSnmpBatch()
|
||||||
return nil
|
return nil
|
||||||
|
|
||||||
case err := <-errCh:
|
case err := <-errCh:
|
||||||
|
flushSnmpBatch()
|
||||||
return fmt.Errorf("read: %w", err)
|
return fmt.Errorf("read: %w", err)
|
||||||
|
|
||||||
case data := <-msgCh:
|
case data := <-msgCh:
|
||||||
|
|
@ -201,8 +218,10 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
handleMessage(msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh)
|
handleMessage(msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh)
|
||||||
|
|
||||||
case result := <-snmpResultCh:
|
case result := <-snmpResultCh:
|
||||||
sendBinaryResult("result", result)
|
snmpBatch = append(snmpBatch, result)
|
||||||
slog.Info("sent snmp result", "device", result.DeviceId, "oids", len(result.OidValues))
|
if len(snmpBatch) >= 50 {
|
||||||
|
flushSnmpBatch()
|
||||||
|
}
|
||||||
|
|
||||||
case result := <-mikrotikResultCh:
|
case result := <-mikrotikResultCh:
|
||||||
sendBinaryResult("mikrotik_result", result)
|
sendBinaryResult("mikrotik_result", result)
|
||||||
|
|
@ -216,6 +235,9 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
sendBinaryResult("monitoring_check", result)
|
sendBinaryResult("monitoring_check", result)
|
||||||
slog.Info("sent monitoring check", "device", result.DeviceId, "status", result.Status)
|
slog.Info("sent monitoring check", "device", result.DeviceId, "status", result.Status)
|
||||||
|
|
||||||
|
case <-flushTicker.C:
|
||||||
|
flushSnmpBatch()
|
||||||
|
|
||||||
case <-heartbeatTicker.C:
|
case <-heartbeatTicker.C:
|
||||||
hb := &pb.AgentHeartbeat{
|
hb := &pb.AgentHeartbeat{
|
||||||
Version: version,
|
Version: version,
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue