fix: concurrency safety, regex caching, lint issues
- agent.go: guard reader goroutine msgCh send with sessionCtx.Done() to prevent permanent goroutine hang - checks.go: cache regex.Compile (not MatchString) per HTTP check; cache x509.SystemCertPool via sync.Once across SSL checks - lldp.go: remove dead walkErrors variable - ssh_test.go: reset global host key store to fix TOFU port collision flakiness across tests - .gitignore: add towerops-agent binary, .tool-versions
This commit is contained in:
parent
5f2d74d303
commit
f2254d66b6
5 changed files with 38 additions and 16 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -33,5 +33,8 @@ result
|
||||||
# Local scripts
|
# Local scripts
|
||||||
monitor-deploy.sh
|
monitor-deploy.sh
|
||||||
|
|
||||||
|
# Tool versions (asdf, mise)
|
||||||
|
.tool-versions
|
||||||
|
|
||||||
# Claude
|
# Claude
|
||||||
CLAUDE.md
|
CLAUDE.md
|
||||||
|
|
|
||||||
8
agent.go
8
agent.go
|
|
@ -196,7 +196,11 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
sessionCancel()
|
sessionCancel()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
msgCh <- data
|
select {
|
||||||
|
case msgCh <- data:
|
||||||
|
case <-sessionCtx.Done():
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
|
@ -319,7 +323,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
case data := <-msgCh:
|
case data := <-msgCh:
|
||||||
var msg channelMsg
|
var msg channelMsg
|
||||||
if err := json.Unmarshal(data, &msg); err != nil {
|
if err := json.Unmarshal(data, &msg); err != nil {
|
||||||
slog.Warn("invalid message", "error", err)
|
slog.Debug("invalid message", "error", err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
shouldEnd, endErr := handleMessage(sessionCtx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh, lldpTopologyResultCh)
|
shouldEnd, endErr := handleMessage(sessionCtx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh, lldpTopologyResultCh)
|
||||||
|
|
|
||||||
25
checks.go
25
checks.go
|
|
@ -11,6 +11,7 @@ import (
|
||||||
"regexp"
|
"regexp"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"codeberg.org/towerops-agent/towerops-agent/pb"
|
"codeberg.org/towerops-agent/towerops-agent/pb"
|
||||||
|
|
@ -28,7 +29,18 @@ var (
|
||||||
MaxIdleConnsPerHost: 10,
|
MaxIdleConnsPerHost: 10,
|
||||||
IdleConnTimeout: 90 * time.Second,
|
IdleConnTimeout: 90 * time.Second,
|
||||||
}
|
}
|
||||||
sslRootCAs = x509.SystemCertPool
|
sslRootCAsOnce sync.Once
|
||||||
|
sslRootCAsPool *x509.CertPool
|
||||||
|
sslRootCAsErr error
|
||||||
|
|
||||||
|
// sslRootCAs returns the system cert pool, cached after first load.
|
||||||
|
// Overridable for tests.
|
||||||
|
sslRootCAs = func() (*x509.CertPool, error) {
|
||||||
|
sslRootCAsOnce.Do(func() {
|
||||||
|
sslRootCAsPool, sslRootCAsErr = x509.SystemCertPool()
|
||||||
|
})
|
||||||
|
return sslRootCAsPool, sslRootCAsErr
|
||||||
|
}
|
||||||
)
|
)
|
||||||
|
|
||||||
// ExecuteCheck runs a service check and returns the result.
|
// ExecuteCheck runs a service check and returns the result.
|
||||||
|
|
@ -146,17 +158,16 @@ func executeHTTPCheck(ctx context.Context, config *pb.HttpCheckConfig, timeoutMs
|
||||||
|
|
||||||
// Check content regex if provided
|
// Check content regex if provided
|
||||||
if config.Regex != "" {
|
if config.Regex != "" {
|
||||||
|
re, err := regexp.Compile(config.Regex)
|
||||||
|
if err != nil {
|
||||||
|
return 2, fmt.Sprintf("Invalid regex: %v", err), responseTime
|
||||||
|
}
|
||||||
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 2, fmt.Sprintf("Failed to read body: %v", err), responseTime
|
return 2, fmt.Sprintf("Failed to read body: %v", err), responseTime
|
||||||
}
|
}
|
||||||
|
|
||||||
matched, err := regexp.MatchString(config.Regex, string(body))
|
if !re.Match(body) {
|
||||||
if err != nil {
|
|
||||||
return 2, fmt.Sprintf("Invalid regex: %v", err), responseTime
|
|
||||||
}
|
|
||||||
|
|
||||||
if !matched {
|
|
||||||
return 2, fmt.Sprintf("Content does not match pattern: %s", config.Regex), responseTime
|
return 2, fmt.Sprintf("Content does not match pattern: %s", config.Regex), responseTime
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
7
lldp.go
7
lldp.go
|
|
@ -106,7 +106,6 @@ func discoverLldpNeighbors(client *gosnmp.GoSNMP, deviceID, jobID string) (*pb.L
|
||||||
|
|
||||||
// Walk remote port descriptions
|
// Walk remote port descriptions
|
||||||
remotePorts := make(map[string]string)
|
remotePorts := make(map[string]string)
|
||||||
var walkErrors []string
|
|
||||||
if err := client.Walk(oidRemPortDesc, func(pdu gosnmp.SnmpPDU) error {
|
if err := client.Walk(oidRemPortDesc, func(pdu gosnmp.SnmpPDU) error {
|
||||||
key := parseRemoteKey(pdu.Name, oidRemPortDesc)
|
key := parseRemoteKey(pdu.Name, oidRemPortDesc)
|
||||||
if key != "" {
|
if key != "" {
|
||||||
|
|
@ -114,7 +113,6 @@ func discoverLldpNeighbors(client *gosnmp.GoSNMP, deviceID, jobID string) (*pb.L
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
walkErrors = append(walkErrors, fmt.Sprintf("walk remote port descriptions: %v", err))
|
|
||||||
slog.Warn("failed to walk remote port descriptions", "error", err)
|
slog.Warn("failed to walk remote port descriptions", "error", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -127,7 +125,6 @@ func discoverLldpNeighbors(client *gosnmp.GoSNMP, deviceID, jobID string) (*pb.L
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
walkErrors = append(walkErrors, fmt.Sprintf("walk remote port IDs: %v", err))
|
|
||||||
slog.Warn("failed to walk remote port IDs", "error", err)
|
slog.Warn("failed to walk remote port IDs", "error", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -140,7 +137,6 @@ func discoverLldpNeighbors(client *gosnmp.GoSNMP, deviceID, jobID string) (*pb.L
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
walkErrors = append(walkErrors, fmt.Sprintf("walk management addresses: %v", err))
|
|
||||||
slog.Warn("failed to walk management addresses", "error", err)
|
slog.Warn("failed to walk management addresses", "error", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -172,9 +168,6 @@ func discoverLldpNeighbors(client *gosnmp.GoSNMP, deviceID, jobID string) (*pb.L
|
||||||
result.Neighbors = append(result.Neighbors, neighbor)
|
result.Neighbors = append(result.Neighbors, neighbor)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Walk errors are already logged; return result with whatever data was collected
|
|
||||||
_ = walkErrors
|
|
||||||
|
|
||||||
return result, nil
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
11
ssh_test.go
11
ssh_test.go
|
|
@ -7,7 +7,9 @@ import (
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net"
|
"net"
|
||||||
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
|
@ -382,6 +384,15 @@ func TestExecuteMikrotikBackupWithOutput(t *testing.T) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestExecuteMikrotikBackupSessionError(t *testing.T) {
|
func TestExecuteMikrotikBackupSessionError(t *testing.T) {
|
||||||
|
// Reset global host key store to avoid TOFU collisions from other tests
|
||||||
|
origStore := globalHostKeys
|
||||||
|
defer func() {
|
||||||
|
hostKeysOnce = sync.Once{}
|
||||||
|
globalHostKeys = origStore
|
||||||
|
}()
|
||||||
|
hostKeysOnce = sync.Once{}
|
||||||
|
t.Setenv("TOWEROPS_HOST_KEYS_FILE", filepath.Join(t.TempDir(), "hosts.json"))
|
||||||
|
|
||||||
// SSH server that accepts connection but rejects all channel requests
|
// SSH server that accepts connection but rejects all channel requests
|
||||||
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue