feat: add service check execution (HTTP/TCP/DNS)
Extends agent to execute HTTP, TCP, and DNS service checks. Agent Changes: - Extended protobuf schema with Check, CheckResult, CheckList messages - Added checks.go with HTTP/TCP/DNS executors (completely stateless) - Added checks worker pool (50 workers) - Handles check_jobs WebSocket event from Phoenix - Reports check_result back to Phoenix Check Executors: - HTTP: Full request support with regex, SSL, headers - TCP: Port connectivity with send/expect patterns - DNS: Resolution for A, AAAA, CNAME, MX, TXT records Architecture: Agent is stateless - receives jobs, executes, reports back. Phoenix handles all scheduling and state tracking. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
d141633586
commit
3ec645a90a
4 changed files with 1044 additions and 96 deletions
51
agent.go
51
agent.go
|
|
@ -95,6 +95,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
|||
snmp: newWorkerPool(100),
|
||||
mikrotik: newWorkerPool(20),
|
||||
ping: newWorkerPool(50),
|
||||
checks: newWorkerPool(50),
|
||||
}
|
||||
|
||||
// Result channels
|
||||
|
|
@ -102,6 +103,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
|||
mikrotikResultCh := make(chan *pb.MikrotikResult, 5000)
|
||||
credTestResultCh := make(chan *pb.CredentialTestResult, 5000)
|
||||
monitoringCheckCh := make(chan *pb.MonitoringCheck, 10000)
|
||||
checkResultCh := make(chan *pb.CheckResult, 10000)
|
||||
|
||||
// Ref counter for outbound messages
|
||||
var refCounter atomic.Uint64
|
||||
|
|
@ -234,6 +236,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
|||
pools.snmp.stop()
|
||||
pools.mikrotik.stop()
|
||||
pools.ping.stop()
|
||||
pools.checks.stop()
|
||||
close(writeCh)
|
||||
writerWg.Wait()
|
||||
}()
|
||||
|
|
@ -272,7 +275,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
|||
slog.Warn("invalid message", "error", err)
|
||||
continue
|
||||
}
|
||||
if handleMessage(ctx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh) {
|
||||
if handleMessage(ctx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh) {
|
||||
flushSnmpBatch()
|
||||
return errRestartRequested
|
||||
}
|
||||
|
|
@ -295,6 +298,10 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
|||
sendBinaryResult("monitoring_check", result)
|
||||
slog.Info("sent monitoring check", "device", result.DeviceId, "status", result.Status)
|
||||
|
||||
case result := <-checkResultCh:
|
||||
sendBinaryResult("check_result", result)
|
||||
slog.Info("sent check result", "check", result.CheckId, "status", result.Status)
|
||||
|
||||
case <-flushTicker.C:
|
||||
flushSnmpBatch()
|
||||
|
||||
|
|
@ -335,6 +342,7 @@ func handleMessage(
|
|||
mikrotikResultCh chan<- *pb.MikrotikResult,
|
||||
credTestResultCh chan<- *pb.CredentialTestResult,
|
||||
monitoringCheckCh chan<- *pb.MonitoringCheck,
|
||||
checkResultCh chan<- *pb.CheckResult,
|
||||
) bool {
|
||||
switch msg.Event {
|
||||
case "phx_reply":
|
||||
|
|
@ -364,7 +372,30 @@ func handleMessage(
|
|||
}
|
||||
slog.Info("received jobs", "count", len(jobList.Jobs))
|
||||
for _, job := range jobList.Jobs {
|
||||
dispatchJob(ctx, job, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh)
|
||||
dispatchJob(ctx, job, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh)
|
||||
}
|
||||
|
||||
case "check_jobs":
|
||||
var payload struct {
|
||||
Binary string `json:"binary"`
|
||||
}
|
||||
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
||||
slog.Error("decode check_jobs payload", "error", err)
|
||||
return false
|
||||
}
|
||||
bin, err := base64.StdEncoding.DecodeString(payload.Binary)
|
||||
if err != nil {
|
||||
slog.Error("decode check_jobs base64", "error", err)
|
||||
return false
|
||||
}
|
||||
var checkList pb.CheckList
|
||||
if err := proto.Unmarshal(bin, &checkList); err != nil {
|
||||
slog.Error("unmarshal check list", "error", err)
|
||||
return false
|
||||
}
|
||||
slog.Info("received checks", "count", len(checkList.Checks))
|
||||
for _, check := range checkList.Checks {
|
||||
executeCheck(ctx, check, pools, checkResultCh)
|
||||
}
|
||||
|
||||
case "restart":
|
||||
|
|
@ -396,6 +427,7 @@ type jobPools struct {
|
|||
snmp *workerPool
|
||||
mikrotik *workerPool
|
||||
ping *workerPool
|
||||
checks *workerPool
|
||||
}
|
||||
|
||||
// dispatchJob routes a job to the appropriate worker pool.
|
||||
|
|
@ -407,6 +439,7 @@ func dispatchJob(
|
|||
mikrotikResultCh chan<- *pb.MikrotikResult,
|
||||
credTestResultCh chan<- *pb.CredentialTestResult,
|
||||
monitoringCheckCh chan<- *pb.MonitoringCheck,
|
||||
checkResultCh chan<- *pb.CheckResult,
|
||||
) {
|
||||
slog.Info("starting job", "job_id", job.JobId, "type", job.JobType)
|
||||
|
||||
|
|
@ -448,3 +481,17 @@ func zeroBytes(b []byte) {
|
|||
}
|
||||
|
||||
func strPtr(s string) *string { return &s }
|
||||
|
||||
// executeCheck dispatches a check to the worker pool.
|
||||
func executeCheck(ctx context.Context, check *pb.Check, pools *jobPools, checkResultCh chan<- *pb.CheckResult) {
|
||||
ok := pools.checks.submit(ctx, func() {
|
||||
result := ExecuteCheck(ctx, check)
|
||||
select {
|
||||
case checkResultCh <- result:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
})
|
||||
if !ok {
|
||||
slog.Warn("check rejected (pool full)", "check_id", check.Id, "type", check.CheckType)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
269
checks.go
Normal file
269
checks.go
Normal file
|
|
@ -0,0 +1,269 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/towerops-app/towerops-agent/pb"
|
||||
)
|
||||
|
||||
// ExecuteCheck runs a service check and returns the result.
|
||||
// Agent is stateless - just executes what it's told and reports back.
|
||||
func ExecuteCheck(ctx context.Context, check *pb.Check) *pb.CheckResult {
|
||||
startTime := time.Now()
|
||||
|
||||
var status uint32
|
||||
var output string
|
||||
var responseTimeMs float64
|
||||
|
||||
switch check.CheckType {
|
||||
case "http":
|
||||
if httpConfig := check.GetHttp(); httpConfig != nil {
|
||||
status, output, responseTimeMs = executeHTTPCheck(ctx, httpConfig, check.TimeoutMs)
|
||||
} else {
|
||||
status, output = 3, "Missing HTTP config"
|
||||
}
|
||||
|
||||
case "tcp":
|
||||
if tcpConfig := check.GetTcp(); tcpConfig != nil {
|
||||
status, output, responseTimeMs = executeTCPCheck(ctx, tcpConfig, check.TimeoutMs)
|
||||
} else {
|
||||
status, output = 3, "Missing TCP config"
|
||||
}
|
||||
|
||||
case "dns":
|
||||
if dnsConfig := check.GetDns(); dnsConfig != nil {
|
||||
status, output, responseTimeMs = executeDNSCheck(ctx, dnsConfig, check.TimeoutMs)
|
||||
} else {
|
||||
status, output = 3, "Missing DNS config"
|
||||
}
|
||||
|
||||
default:
|
||||
status, output = 3, fmt.Sprintf("Unknown check type: %s", check.CheckType)
|
||||
}
|
||||
|
||||
// If responseTimeMs wasn't set by executor, calculate from start time
|
||||
if responseTimeMs == 0 {
|
||||
responseTimeMs = float64(time.Since(startTime).Milliseconds())
|
||||
}
|
||||
|
||||
return &pb.CheckResult{
|
||||
CheckId: check.Id,
|
||||
Status: status,
|
||||
Output: output,
|
||||
ResponseTimeMs: responseTimeMs,
|
||||
Timestamp: time.Now().Unix(),
|
||||
}
|
||||
}
|
||||
|
||||
// executeHTTPCheck performs an HTTP/HTTPS check
|
||||
func executeHTTPCheck(ctx context.Context, config *pb.HttpCheckConfig, timeoutMs uint32) (uint32, string, float64) {
|
||||
timeout := time.Duration(timeoutMs) * time.Millisecond
|
||||
|
||||
// Create HTTP client with timeout and TLS config
|
||||
client := &http.Client{
|
||||
Timeout: timeout,
|
||||
Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{
|
||||
InsecureSkipVerify: !config.VerifySsl,
|
||||
},
|
||||
},
|
||||
CheckRedirect: func(req *http.Request, via []*http.Request) error {
|
||||
if !config.FollowRedirects {
|
||||
return http.ErrUseLastResponse
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
method := strings.ToUpper(config.Method)
|
||||
if method == "" {
|
||||
method = "GET"
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, method, config.Url, strings.NewReader(config.Body))
|
||||
if err != nil {
|
||||
return 2, fmt.Sprintf("Failed to create request: %v", err), 0
|
||||
}
|
||||
|
||||
// Add headers
|
||||
for key, value := range config.Headers {
|
||||
req.Header.Set(key, value)
|
||||
}
|
||||
|
||||
startTime := time.Now()
|
||||
resp, err := client.Do(req)
|
||||
responseTime := float64(time.Since(startTime).Milliseconds())
|
||||
|
||||
if err != nil {
|
||||
return 2, fmt.Sprintf("Request failed: %v", err), responseTime
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
// Check status code
|
||||
expectedStatus := int(config.ExpectedStatus)
|
||||
if expectedStatus == 0 {
|
||||
expectedStatus = 200
|
||||
}
|
||||
|
||||
if resp.StatusCode != expectedStatus {
|
||||
return 2, fmt.Sprintf("HTTP %d, expected %d", resp.StatusCode, expectedStatus), responseTime
|
||||
}
|
||||
|
||||
// Check content regex if provided
|
||||
if config.Regex != "" {
|
||||
body := make([]byte, 1024*1024) // Read up to 1MB
|
||||
n, _ := resp.Body.Read(body)
|
||||
|
||||
matched, err := regexp.MatchString(config.Regex, string(body[:n]))
|
||||
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 0, fmt.Sprintf("HTTP %d OK", resp.StatusCode), responseTime
|
||||
}
|
||||
|
||||
// executeTCPCheck performs a TCP port connectivity check
|
||||
func executeTCPCheck(ctx context.Context, config *pb.TcpCheckConfig, timeoutMs uint32) (uint32, string, float64) {
|
||||
timeout := time.Duration(timeoutMs) * time.Millisecond
|
||||
address := fmt.Sprintf("%s:%d", config.Host, config.Port)
|
||||
|
||||
startTime := time.Now()
|
||||
conn, err := net.DialTimeout("tcp", address, timeout)
|
||||
responseTime := float64(time.Since(startTime).Milliseconds())
|
||||
|
||||
if err != nil {
|
||||
return 2, fmt.Sprintf("Connection failed: %v", err), responseTime
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
// If send/expect strings provided, test them
|
||||
if config.Send != "" {
|
||||
conn.SetDeadline(time.Now().Add(timeout))
|
||||
|
||||
_, err = conn.Write([]byte(config.Send))
|
||||
if err != nil {
|
||||
return 2, fmt.Sprintf("Send failed: %v", err), responseTime
|
||||
}
|
||||
|
||||
if config.Expect != "" {
|
||||
buffer := make([]byte, 4096)
|
||||
n, err := conn.Read(buffer)
|
||||
if err != nil {
|
||||
return 2, fmt.Sprintf("Receive failed: %v", err), responseTime
|
||||
}
|
||||
|
||||
received := string(buffer[:n])
|
||||
if !strings.Contains(received, config.Expect) {
|
||||
return 2, fmt.Sprintf("Unexpected response: %s", received), responseTime
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return 0, fmt.Sprintf("TCP port %d open", config.Port), responseTime
|
||||
}
|
||||
|
||||
// executeDNSCheck performs a DNS resolution check
|
||||
func executeDNSCheck(ctx context.Context, config *pb.DnsCheckConfig, timeoutMs uint32) (uint32, string, float64) {
|
||||
timeout := time.Duration(timeoutMs) * time.Millisecond
|
||||
|
||||
resolver := &net.Resolver{}
|
||||
if config.Server != "" {
|
||||
resolver = &net.Resolver{
|
||||
PreferGo: true,
|
||||
Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
|
||||
d := net.Dialer{Timeout: timeout}
|
||||
return d.DialContext(ctx, "udp", config.Server+":53")
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
recordType := strings.ToUpper(config.RecordType)
|
||||
if recordType == "" {
|
||||
recordType = "A"
|
||||
}
|
||||
|
||||
startTime := time.Now()
|
||||
|
||||
var results []string
|
||||
var err error
|
||||
|
||||
switch recordType {
|
||||
case "A":
|
||||
ips, lookupErr := resolver.LookupIP(ctx, "ip4", config.Hostname)
|
||||
err = lookupErr
|
||||
for _, ip := range ips {
|
||||
results = append(results, ip.String())
|
||||
}
|
||||
|
||||
case "AAAA":
|
||||
ips, lookupErr := resolver.LookupIP(ctx, "ip6", config.Hostname)
|
||||
err = lookupErr
|
||||
for _, ip := range ips {
|
||||
results = append(results, ip.String())
|
||||
}
|
||||
|
||||
case "CNAME":
|
||||
cname, lookupErr := resolver.LookupCNAME(ctx, config.Hostname)
|
||||
err = lookupErr
|
||||
if cname != "" {
|
||||
results = append(results, cname)
|
||||
}
|
||||
|
||||
case "MX":
|
||||
mxs, lookupErr := resolver.LookupMX(ctx, config.Hostname)
|
||||
err = lookupErr
|
||||
for _, mx := range mxs {
|
||||
results = append(results, fmt.Sprintf("%d %s", mx.Pref, mx.Host))
|
||||
}
|
||||
|
||||
case "TXT":
|
||||
txts, lookupErr := resolver.LookupTXT(ctx, config.Hostname)
|
||||
err = lookupErr
|
||||
results = txts
|
||||
|
||||
default:
|
||||
return 3, fmt.Sprintf("Unsupported record type: %s", recordType), 0
|
||||
}
|
||||
|
||||
responseTime := float64(time.Since(startTime).Milliseconds())
|
||||
|
||||
if err != nil {
|
||||
return 2, fmt.Sprintf("DNS query failed: %v", err), responseTime
|
||||
}
|
||||
|
||||
if len(results) == 0 {
|
||||
return 2, fmt.Sprintf("No %s records found", recordType), responseTime
|
||||
}
|
||||
|
||||
// Check expected result if provided
|
||||
if config.Expected != "" {
|
||||
found := false
|
||||
for _, result := range results {
|
||||
if result == config.Expected {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if !found {
|
||||
return 2, fmt.Sprintf("Expected '%s', got: %s", config.Expected, strings.Join(results, ", ")), responseTime
|
||||
}
|
||||
}
|
||||
|
||||
return 0, fmt.Sprintf("Resolved to: %s", strings.Join(results, ", ")), responseTime
|
||||
}
|
||||
765
pb/agent.pb.go
765
pb/agent.pb.go
File diff suppressed because it is too large
Load diff
|
|
@ -9,6 +9,7 @@ message AgentConfig {
|
|||
string version = 1;
|
||||
uint32 poll_interval_seconds = 2;
|
||||
repeated Device devices = 3;
|
||||
repeated Check checks = 4; // Service checks (HTTP/TCP/DNS)
|
||||
}
|
||||
|
||||
message Device {
|
||||
|
|
@ -57,6 +58,7 @@ message Metric {
|
|||
InterfaceStat interface_stat = 2;
|
||||
NeighborDiscovery neighbor_discovery = 3;
|
||||
MonitoringCheck monitoring_check = 4;
|
||||
CheckResult check_result = 5; // Service check results (HTTP/TCP/DNS)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -99,6 +101,59 @@ message MonitoringCheck {
|
|||
int64 timestamp = 4; // Unix timestamp in seconds
|
||||
}
|
||||
|
||||
// Service Checks (HTTP/TCP/DNS)
|
||||
|
||||
message Check {
|
||||
string id = 1;
|
||||
string check_type = 2; // "http", "tcp", "dns"
|
||||
uint32 interval_seconds = 3;
|
||||
uint32 timeout_ms = 4;
|
||||
|
||||
// Type-specific config (JSON from Phoenix config field)
|
||||
oneof config {
|
||||
HttpCheckConfig http = 5;
|
||||
TcpCheckConfig tcp = 6;
|
||||
DnsCheckConfig dns = 7;
|
||||
}
|
||||
}
|
||||
|
||||
message HttpCheckConfig {
|
||||
string url = 1;
|
||||
string method = 2; // GET, POST, etc.
|
||||
uint32 expected_status = 3;
|
||||
bool verify_ssl = 4;
|
||||
map<string, string> headers = 5;
|
||||
string body = 6;
|
||||
string regex = 7; // Content match regex
|
||||
bool follow_redirects = 8;
|
||||
}
|
||||
|
||||
message TcpCheckConfig {
|
||||
string host = 1;
|
||||
uint32 port = 2;
|
||||
string send = 3; // Data to send after connection
|
||||
string expect = 4; // Expected response
|
||||
}
|
||||
|
||||
message DnsCheckConfig {
|
||||
string hostname = 1;
|
||||
string server = 2; // DNS server (optional)
|
||||
string record_type = 3; // A, AAAA, CNAME, MX, TXT
|
||||
string expected = 4; // Expected result (optional)
|
||||
}
|
||||
|
||||
message CheckResult {
|
||||
string check_id = 1;
|
||||
uint32 status = 2; // 0=OK, 1=WARNING, 2=CRITICAL, 3=UNKNOWN
|
||||
string output = 3;
|
||||
double response_time_ms = 4;
|
||||
int64 timestamp = 5; // Unix timestamp in seconds
|
||||
}
|
||||
|
||||
message CheckList {
|
||||
repeated Check checks = 1;
|
||||
}
|
||||
|
||||
// Heartbeat metadata
|
||||
message HeartbeatMetadata {
|
||||
string version = 1;
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue