diff --git a/Dockerfile b/Dockerfile index bd74f93..0030811 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,6 @@ -# golang 1.26.5: fixes the Go stdlib pair CVE-2026-39822 (HIGH, os.Root symlink +# golang 1.26.9: fixes the Go stdlib pair CVE-2026-39822 (HIGH, os.Root symlink # traversal) / CVE-2026-42505 (crypto/tls ECH) that 1.26.4 builds carry. -FROM golang:1.26.6-bookworm@sha256:116d58cbd88c1297624acc6e967a060012422bacf9930927e23fb719189c6f36 AS build-stage +FROM golang:1.26.9-bookworm@sha256:d9c68c2c51161e12fd77e4c6320687c9cd86e1af1e3ad6e6cd63ff970641453c AS build-stage # Always compile with the base image's Go, never an auto-downloaded toolchain. ENV GOTOOLCHAIN=local diff --git a/docs/connection-lifecycle.md b/docs/connection-lifecycle.md index e91b1a6..bdf08f6 100644 --- a/docs/connection-lifecycle.md +++ b/docs/connection-lifecycle.md @@ -37,6 +37,8 @@ sequenceDiagram end ``` +**Integration errors:** Each entry in the health report carries `status`, and while `status` is `"error"` also `error` and `error_class` (`datasource_integration_failure` or `ssh_integration_failure`). Errors come from failed health checks and from proxied requests that fail with a connectivity or authentication error (kept for up to 15 minutes, cleared by the next successful request or when the datasource is re-registered or removed). Caller mistakes (bad SQL, disallowed commands, invalid input) and cancelled requests are not reported. Messages are redacted on a best-effort basis before entering the report (credentials, bearer tokens, private keys, IP addresses, `host:port`, and hosts named after `lookup`/`dial`) and truncated to 512 chars; a bare hostname or username in free-form text may not be caught. The raw error is still returned in the response to the failing request itself. Once the integration recovers, the next report simply omits the error. + **Auto-reconnect:** On disconnect, the agent reconnects with exponential backoff (3s → 6s → 12s → ... → 30s max). ## Config Sync diff --git a/go.mod b/go.mod index bfe3f69..90d5344 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module nudgebee/forager -go 1.26.6 +go 1.26.9 require ( cloud.google.com/go/auth v0.23.1 @@ -25,8 +25,8 @@ require ( github.com/spf13/viper v1.21.0 github.com/xdg-go/scram v1.2.0 go.mongodb.org/mongo-driver/v2 v2.8.0 - golang.org/x/crypto v0.56.0 - golang.org/x/sys v0.47.0 + golang.org/x/crypto v0.57.0 + golang.org/x/sys v0.48.0 golang.org/x/time v0.15.0 google.golang.org/api v0.293.0 gopkg.in/yaml.v3 v3.0.1 @@ -113,13 +113,13 @@ require ( go.uber.org/atomic v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/exp v0.0.0-20250506013437-ce4c2cf36ca6 // indirect - golang.org/x/net v0.57.0 // indirect + golang.org/x/net v0.60.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sync v0.22.0 // indirect - golang.org/x/text v0.41.0 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/text v0.42.0 // indirect google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260807164820-c8921c73eeea // indirect - google.golang.org/grpc v1.83.1 // indirect + google.golang.org/grpc v1.83.2 // indirect google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/go.sum b/go.sum index d86c542..4c79640 100644 --- a/go.sum +++ b/go.sum @@ -279,8 +279,8 @@ go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= -golang.org/x/crypto v0.56.0 h1:GUh5Ii4J5jtcseSMiRqr1jXCNHoxjeV9Fmekc2oLy6Y= -golang.org/x/crypto v0.56.0/go.mod h1:OMW5y6CY9l38uPLmxU6l6pwcXp1obtLo3e6gT7gQR2I= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= golang.org/x/exp v0.0.0-20250506013437-ce4c2cf36ca6 h1:y5zboxd6LQAqYIhHnB48p0ByQ/GnQx2BE33L8BOHQkI= golang.org/x/exp v0.0.0-20250506013437-ce4c2cf36ca6/go.mod h1:U6Lno4MTRCDY+Ba7aCcauB9T60gsv5s4ralQzP72ZoQ= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= @@ -290,14 +290,14 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= -golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= +golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU= +golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= -golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -305,20 +305,20 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= -golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= -golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= -golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= +golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= +golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= -golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= -golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= @@ -335,8 +335,8 @@ google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7 h1: google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7/go.mod h1:KqHwBx2upmfa1XSi1WuRvC+2VGCLtooKkfmyvRbUmqA= google.golang.org/genproto/googleapis/rpc v0.0.0-20260807164820-c8921c73eeea h1:kVhQEPTpKQahD5+JSBTfBB19wcgQTTjAIn45MBqnyHk= google.golang.org/genproto/googleapis/rpc v0.0.0-20260807164820-c8921c73eeea/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.83.1 h1:HIO0+BEtBP6soyqvqC8sNUjZ7bTs+0hFQuFF+RAy++Y= -google.golang.org/grpc v1.83.1/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/pkg/proxy/errors.go b/pkg/proxy/errors.go new file mode 100644 index 0000000..6b028df --- /dev/null +++ b/pkg/proxy/errors.go @@ -0,0 +1,109 @@ +package proxy + +import ( + "errors" + "io" + "net" + "regexp" + "strings" + "unicode/utf8" +) + +// Error classes surfaced on the Proxy Agent tab of the Agent Health page. +const ( + ErrorClassDatasource = "datasource_integration_failure" + ErrorClassSSH = "ssh_integration_failure" + + maxReportedErrorLen = 512 +) + +// ErrorClass maps a proxy type to the error class reported upstream. +func ErrorClass(proxyType string) string { + if proxyType == "ssh-proxy" { + return ErrorClassSSH + } + return ErrorClassDatasource +} + +// integrationErrorMarkers are substrings (lowercase) of errors that mean the +// datasource itself is unreachable or rejecting our credentials, as opposed to +// the caller sending a bad query or hitting a policy rule. +var integrationErrorMarkers = []string{ + "connection refused", "connection reset", "broken pipe", "no such host", + "i/o timeout", "network is unreachable", "no route to host", + "handshake failed", "unable to authenticate", "authentication failed", + "password authentication", "access denied for user", "permission denied (publickey", + "host key", "x509:", "tls:", "sasl", "noauth", "wrongpass", +} + +// IsIntegrationError reports whether err looks like a connectivity or +// authentication failure of the integration, rather than a caller mistake +// (invalid input, disallowed command, SQL error) which must not mark a +// healthy datasource as broken. +func IsIntegrationError(err error) bool { + if err == nil { + return false + } + var netErr net.Error + if errors.As(err, &netErr) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + return true + } + msg := strings.ToLower(err.Error()) + for _, m := range integrationErrorMarkers { + if strings.Contains(msg, m) { + return true + } + } + return false +} + +const secretKeys = `password|passwd|pwd|passphrase|secret|token|api[_-]?key|access[_-]?key|private[_-]?key|authorization|credentials?` + +var ( + // scheme://user:pass@host → scheme://[redacted]@host + urlUserinfoRe = regexp.MustCompile(`([a-zA-Z][a-zA-Z0-9+.-]*://)[^\s/@]+@`) + // password=..., "password":"...", token: ... (key may be followed by a closing quote) + secretKVRe = regexp.MustCompile(`(?i)\b(` + secretKeys + `)("?\s*[=:]\s*)("[^"]*"|'[^']*'|\S+)`) + // Authorization header schemes with an inline credential. + bearerRe = regexp.MustCompile(`(?i)\b(bearer|basic)\s+[A-Za-z0-9._~+/=-]{6,}`) + pemBlockRe = regexp.MustCompile(`(?s)-----BEGIN [A-Z ]*PRIVATE KEY-----.*?(-----END [A-Z ]*PRIVATE KEY-----|$)`) + + // [fd00::5]:5432 and bare IPv6 with at least five groups. + ipv6BracketRe = regexp.MustCompile(`\[[0-9a-fA-F:.]*:[0-9a-fA-F:.]*\]`) + ipv6BareRe = regexp.MustCompile(`\b(?:[0-9a-fA-F]{1,4}:){4,7}[0-9a-fA-F]{1,4}\b`) + ipv4Re = regexp.MustCompile(`\b\d{1,3}(?:\.\d{1,3}){3}\b`) + // host.domain:port pairs such as db.internal.example.com:5432 + hostPortRe = regexp.MustCompile(`\b[a-zA-Z0-9-]+(?:\.[a-zA-Z0-9-]+)+:\d{1,5}\b`) + // Go's resolver and dialer name the host right after these words, with or + // without dots: "lookup db.internal: no such host", "dial tcp postgres:5432". + lookupHostRe = regexp.MustCompile(`(?i)(\blookup\s+)[^\s:]+`) + dialHostRe = regexp.MustCompile(`(?i)(\bdial\s+(?:tcp|udp)\d?\s+)[^\s:]+`) +) + +// RedactError strips credentials and hostnames/IPs from an error message +// before it leaves the host, and bounds its length. +// +// It is best-effort: a bare hostname or a username in free-form error text +// that follows none of the patterns above is not detected. +func RedactError(msg string) string { + msg = strings.ToValidUTF8(msg, "") + msg = pemBlockRe.ReplaceAllString(msg, "[redacted-key]") + msg = urlUserinfoRe.ReplaceAllString(msg, "${1}[redacted]@") + msg = bearerRe.ReplaceAllString(msg, "${1} [redacted]") + msg = secretKVRe.ReplaceAllString(msg, "${1}${2}[redacted]") + msg = ipv6BracketRe.ReplaceAllString(msg, "[redacted-host]") + msg = ipv6BareRe.ReplaceAllString(msg, "[redacted-host]") + msg = ipv4Re.ReplaceAllString(msg, "[redacted-host]") + msg = hostPortRe.ReplaceAllString(msg, "[redacted-host]") + msg = lookupHostRe.ReplaceAllString(msg, "${1}[redacted-host]") + msg = dialHostRe.ReplaceAllString(msg, "${1}[redacted-host]") + msg = strings.Join(strings.Fields(msg), " ") + if len(msg) > maxReportedErrorLen { + i := maxReportedErrorLen + for i > 0 && !utf8.RuneStart(msg[i]) { + i-- + } + msg = msg[:i] + "…" + } + return msg +} diff --git a/pkg/proxy/errors_test.go b/pkg/proxy/errors_test.go new file mode 100644 index 0000000..b1fea95 --- /dev/null +++ b/pkg/proxy/errors_test.go @@ -0,0 +1,182 @@ +package proxy + +import ( + "context" + "errors" + "fmt" + "net" + "strings" + "sync" + "testing" + "unicode/utf8" +) + +func TestRedactError(t *testing.T) { + cases := []struct{ name, in, want string }{ + {"ipv4", "dial tcp 10.1.2.3:5432: connection refused", "dial tcp [redacted-host]:5432: connection refused"}, + {"dotted host", "connect to db.internal.example.com:5432 failed", "connect to [redacted-host] failed"}, + {"url userinfo", "postgres://admin:hunter2@db/x failed", "postgres://[redacted]@db/x failed"}, + {"kv", "auth failed password=hunter2 for user", "auth failed password=[redacted] for user"}, + {"json", `body {"password":"hunter2","db":"x"}`, `body {"password":[redacted],"db":"x"}`}, + {"bearer", "401 Authorization: Bearer abc123def456", "401 Authorization: [redacted] [redacted]"}, + {"basic", "sent Basic dXNlcjpwYXNz rejected", "sent Basic [redacted] rejected"}, + {"ipv6 bracket", "dial tcp [fd00::5]:5432: refused", "dial tcp [redacted-host]:5432: refused"}, + {"lookup", "dial tcp: lookup db.internal.example.com: no such host", "dial tcp: lookup [redacted-host]: no such host"}, + {"dial single label", "dial tcp postgres:5432: connect: refused", "dial tcp [redacted-host]:5432: connect: refused"}, + {"pem", "key -----BEGIN RSA PRIVATE KEY-----\nabc\n-----END RSA PRIVATE KEY-----", "key [redacted-key]"}, + {"plain", "connection refused", "connection refused"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := RedactError(tc.in); got != tc.want { + t.Errorf("RedactError(%q)\n got %q\nwant %q", tc.in, got, tc.want) + } + }) + } +} + +func TestRedactErrorTruncatesOnRuneBoundary(t *testing.T) { + // 3-byte runes so byte 512 lands mid-rune. + got := RedactError(strings.Repeat("€", 400)) + if !utf8.ValidString(got) { + t.Fatalf("invalid UTF-8 after truncation: %q", got) + } + if !strings.HasSuffix(got, "…") || len(got) > maxReportedErrorLen+len("…") { + t.Fatalf("not truncated properly: len=%d", len(got)) + } + if !utf8.ValidString(RedactError("bad \xff byte")) { + t.Fatal("invalid input bytes not scrubbed") + } +} + +func TestErrorClass(t *testing.T) { + if ErrorClass("ssh-proxy") != ErrorClassSSH || ErrorClass("db-proxy") != ErrorClassDatasource { + t.Fatal("wrong class") + } +} + +func TestIsIntegrationError(t *testing.T) { + yes := []error{ + &net.OpError{Op: "dial", Err: errors.New("boom")}, + errors.New("dial tcp 1.2.3.4:22: connect: connection refused"), + errors.New("ssh: handshake failed: ssh: unable to authenticate"), + errors.New(`pq: password authentication failed for user "x"`), + } + no := []error{ + nil, + errors.New("command \"FLUSHALL\" not allowed"), + errors.New("datasource is read-only, execute not allowed"), + errors.New("invalid base64 content"), + errors.New(`ERROR: relation "typo_table" does not exist`), + } + for _, e := range yes { + if !IsIntegrationError(e) { + t.Errorf("want integration error: %v", e) + } + } + for _, e := range no { + if IsIntegrationError(e) { + t.Errorf("want caller error: %v", e) + } + } +} + +type stubProxy struct{ Proxy } + +func (stubProxy) HealthCheck(context.Context) error { return nil } +func (stubProxy) Close() error { return nil } + +var errRefused = errors.New("dial tcp 10.0.0.5:22: connect: connection refused") + +func TestRequestErrorReportedThenCleared(t *testing.T) { + r := NewRegistry() + r.Register("ssh1", DatasourceEntry{ID: "ssh1", ProxyType: "ssh-proxy"}, stubProxy{}) + + r.RecordRequestResult(context.Background(), "ssh1", errRefused) + h := r.HealthReport(context.Background())["ssh1"] + if h.Status != "error" || h.ErrorClass != ErrorClassSSH || strings.Contains(h.Error, "10.0.0.5") || h.Error == "" { + t.Fatalf("unexpected: %+v", h) + } + + r.RecordRequestResult(context.Background(), "ssh1", nil) + if h := r.HealthReport(context.Background())["ssh1"]; h.Status != "healthy" || h.ErrorClass != "" { + t.Fatalf("not cleared: %+v", h) + } +} + +func TestCallerErrorsAndCancelledRequestsIgnored(t *testing.T) { + r := NewRegistry() + r.Register("db1", DatasourceEntry{ID: "db1", ProxyType: "db-proxy"}, stubProxy{}) + + r.RecordRequestResult(context.Background(), "db1", errors.New(`relation "typo" does not exist`)) + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + r.RecordRequestResult(cancelled, "db1", errRefused) + + if h := r.HealthReport(context.Background())["db1"]; h.Status != "healthy" { + t.Fatalf("caller error flagged datasource: %+v", h) + } +} + +func TestRequestErrorClearedOnRegisterAndRemove(t *testing.T) { + r := NewRegistry() + r.Register("ssh1", DatasourceEntry{ID: "ssh1", ProxyType: "ssh-proxy"}, stubProxy{}) + r.RecordRequestResult(context.Background(), "ssh1", errRefused) + + // Re-registering (e.g. after a corrected password) discards the old failure. + r.Register("ssh1", DatasourceEntry{ID: "ssh1", ProxyType: "ssh-proxy"}, stubProxy{}) + if h := r.HealthReport(context.Background())["ssh1"]; h.Status != "healthy" { + t.Fatalf("stale error after Register: %+v", h) + } + + r.RecordRequestResult(context.Background(), "ssh1", errRefused) + if err := r.Remove("ssh1"); err != nil { + t.Fatal(err) + } + // A late result for a removed datasource must not recreate an entry. + r.RecordRequestResult(context.Background(), "ssh1", errRefused) + r.reqErrMu.Lock() + n := len(r.reqErrors) + r.reqErrMu.Unlock() + if n != 0 { + t.Fatalf("reqErrors leaked %d entries", n) + } + + r.Register("a", DatasourceEntry{ID: "a"}, stubProxy{}) + r.RecordRequestResult(context.Background(), "a", errRefused) + r.CloseAll() + r.reqErrMu.Lock() + n = len(r.reqErrors) + r.reqErrMu.Unlock() + if n != 0 { + t.Fatalf("CloseAll leaked %d entries", n) + } +} + +func TestRecordRequestResultConcurrent(t *testing.T) { + r := NewRegistry() + for i := 0; i < 4; i++ { + r.Register(fmt.Sprint("d", i), DatasourceEntry{ID: fmt.Sprint("d", i)}, stubProxy{}) + } + var wg sync.WaitGroup + for i := 0; i < 32; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + id := fmt.Sprint("d", i%4) + for j := 0; j < 50; j++ { + switch j % 4 { + case 0: + r.RecordRequestResult(context.Background(), id, errRefused) + case 1: + r.RecordRequestResult(context.Background(), id, nil) + case 2: + r.HealthReport(context.Background()) + default: + r.Register(id, DatasourceEntry{ID: id}, stubProxy{}) + } + } + }(i) + } + wg.Wait() +} diff --git a/pkg/proxy/registry.go b/pkg/proxy/registry.go index 5abb06a..58ed19b 100644 --- a/pkg/proxy/registry.go +++ b/pkg/proxy/registry.go @@ -70,6 +70,76 @@ type Registry struct { mu sync.RWMutex proxies map[string]Proxy // datasource ID → Proxy configs map[string]DatasourceEntry + + reqErrMu sync.Mutex + reqErrors map[string]requestError // datasource ID → last unrecovered request failure +} + +type requestError struct { + msg string + at time.Time +} + +// requestErrorTTL bounds how long a request failure is reported without a +// later successful request to the same datasource. +const requestErrorTTL = 15 * time.Minute + +// RecordRequestResult notes the outcome of a proxied request. A failure that +// looks like an integration problem (see IsIntegrationError) is reported on +// the next health report (deduplicated per datasource, latest wins); a +// success clears it. Failures caused by the caller — bad input, policy +// rejections, a cancelled or timed-out request context — are ignored. +// +// Lock order is mu → reqErrMu everywhere. +func (r *Registry) RecordRequestResult(ctx context.Context, datasourceID string, err error) { + var msg string + if err != nil { + if ctx.Err() != nil || !IsIntegrationError(err) { + return + } + msg = RedactError(err.Error()) // outside the locks: several regexes + } + + r.mu.RLock() + defer r.mu.RUnlock() + // Ignore results for datasources that were removed while the request was + // in flight, so they cannot resurrect an entry nothing will clean up. + if _, ok := r.proxies[datasourceID]; !ok { + return + } + + r.reqErrMu.Lock() + defer r.reqErrMu.Unlock() + if err == nil { + delete(r.reqErrors, datasourceID) + return + } + if r.reqErrors == nil { + r.reqErrors = make(map[string]requestError) + } + r.reqErrors[datasourceID] = requestError{msg: msg, at: time.Now()} +} + +// clearRequestError drops any recorded request failure for a datasource. +// Callers hold r.mu. +func (r *Registry) clearRequestError(datasourceID string) { + r.reqErrMu.Lock() + delete(r.reqErrors, datasourceID) + r.reqErrMu.Unlock() +} + +func (r *Registry) recentRequestError(datasourceID string) (string, bool) { + r.reqErrMu.Lock() + defer r.reqErrMu.Unlock() + e, ok := r.reqErrors[datasourceID] + if !ok { + return "", false + } + if time.Since(e.at) > requestErrorTTL { + delete(r.reqErrors, datasourceID) + return "", false + } + return e.msg, true } // DatasourceEntry stores the proxy and its metadata. @@ -110,6 +180,8 @@ func (r *Registry) Register(id string, entry DatasourceEntry, p Proxy) { r.proxies[id] = p r.configs[id] = entry + // A replaced proxy is a new configuration; the old failure no longer applies. + r.clearRequestError(id) } // Remove removes and closes a proxy by datasource ID. @@ -123,6 +195,7 @@ func (r *Registry) Remove(id string) error { } delete(r.proxies, id) delete(r.configs, id) + r.clearRequestError(id) } return nil } @@ -146,7 +219,10 @@ type DatasourceHealth struct { Name string `json:"name"` Status string `json:"status"` // "healthy", "error", "unknown" Error string `json:"error,omitempty"` - LastCheck string `json:"last_check"` // RFC3339 + // ErrorClass is set only while Status is "error"; it clears on the next + // successful check, so the report always reflects current state. + ErrorClass string `json:"error_class,omitempty"` + LastCheck string `json:"last_check"` // RFC3339 } const ( @@ -181,12 +257,13 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth now := time.Now().UTC().Format(time.RFC3339) for _, t := range targets { report[t.id] = DatasourceHealth{ - Type: t.cfg.Type, - ProxyType: t.cfg.ProxyType, - Name: t.cfg.Name, - Status: "error", - Error: err.Error(), - LastCheck: now, + Type: t.cfg.Type, + ProxyType: t.cfg.ProxyType, + Name: t.cfg.Name, + Status: "error", + Error: RedactError(err.Error()), + ErrorClass: ErrorClass(t.cfg.ProxyType), + LastCheck: now, } } return report @@ -225,7 +302,8 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth if err := ctx.Err(); err != nil { health.Status = "error" - health.Error = err.Error() + health.Error = RedactError(err.Error()) + health.ErrorClass = ErrorClass(t.cfg.ProxyType) results[idx] = health continue } @@ -243,7 +321,12 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth if err != nil { health.Status = "error" - health.Error = err.Error() + health.Error = RedactError(err.Error()) + health.ErrorClass = ErrorClass(t.cfg.ProxyType) + } else if msg, ok := r.recentRequestError(t.id); ok { + health.Status = "error" + health.Error = msg + health.ErrorClass = ErrorClass(t.cfg.ProxyType) } else { health.Status = "healthy" } @@ -335,5 +418,6 @@ func (r *Registry) CloseAll() { p.Close() // nolint:errcheck delete(r.proxies, id) delete(r.configs, id) + r.clearRequestError(id) } } diff --git a/pkg/ws/handler.go b/pkg/ws/handler.go index 0512de1..5c5100c 100644 --- a/pkg/ws/handler.go +++ b/pkg/ws/handler.go @@ -174,6 +174,7 @@ func (h *Handler) handleRequest(ctx context.Context, msg []byte, requestID, data } resp, err := p.HandleRequest(ctx, &req) + h.registry.RecordRequestResult(ctx, datasourceID, err) if err != nil { h.logger.Error("proxy request failed", "action", req.Action, "datasource", datasourceID, "err", err) return h.buildErrorResponse(requestID, 500, err.Error()), nil @@ -215,6 +216,7 @@ func (h *Handler) handleLegacyRequest(ctx context.Context, msg []byte, requestID } resp, err := p.HandleRequest(ctx, req) + h.registry.RecordRequestResult(ctx, datasourceID, err) if err != nil { h.logger.Error("legacy proxy request failed", "action", req.Action, "datasource", datasourceID, "err", err) return h.buildErrorResponse(requestID, 500, err.Error()), nil