diff --git a/internal/controller/family_sink_delete_watch_envtest_test.go b/internal/controller/family_sink_delete_watch_envtest_test.go index 3274dd45..2a6525ff 100644 --- a/internal/controller/family_sink_delete_watch_envtest_test.go +++ b/internal/controller/family_sink_delete_watch_envtest_test.go @@ -9,6 +9,7 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + "github.com/prometheus/client_golang/prometheus/testutil" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" @@ -16,6 +17,7 @@ import ( kollectdevv1alpha1 "github.com/platformrelay/kollect/api/v1alpha1" "github.com/platformrelay/kollect/internal/collect" "github.com/platformrelay/kollect/internal/export" + "github.com/platformrelay/kollect/internal/metrics" "github.com/platformrelay/kollect/internal/sink" ) @@ -46,37 +48,60 @@ var _ = Describe("family-sink delete watch (envtest, BEP-1)", func() { Eventually(func() bool { return mgr.GetCache().WaitForCacheSync(watchCtx) }). WithTimeout(30 * time.Second).WithPolling(200 * time.Millisecond).Should(BeTrue()) - sinkObj := &kollectdevv1alpha1.KollectSnapshotSink{ - ObjectMeta: metav1.ObjectMeta{GenerateName: "delete-watch-", Namespace: "default"}, - Spec: kollectdevv1alpha1.KollectSnapshotSinkSpec{Type: "s3"}, - } - Expect(k8sClient.Create(watchCtx, sinkObj)).To(Succeed()) - - // The watch informer must know the sink before it is deleted, so the - // delete lands as a normal delete event with the object's UID. - Eventually(func() error { - return mgr.GetClient().Get(watchCtx, types.NamespacedName{Namespace: sinkObj.Namespace, Name: sinkObj.Name}, &kollectdevv1alpha1.KollectSnapshotSink{}) - }).WithTimeout(30 * time.Second).Should(Succeed()) - - spy := &deleteWatchSpyBackend{} - reg := sink.NewRegistry() - reg.Register("s3", func(_ kollectdevv1alpha1.KollectSinkSpec, _ sink.BuildContext) (sink.Backend, error) { - return spy, nil - }) envelope, envErr := export.MarshalEnvelope([]collect.Item{}, export.Metadata{}) Expect(envErr).NotTo(HaveOccurred()) - _, poolErr := sink.RunExportEnvelope(sink.ExportEnvelopeRequest{ - Ctx: watchCtx, - Client: k8sClient, - Registry: reg, - SinkNamespace: sinkObj.Namespace, - SinkName: sinkObj.Name, - SinkUID: sinkObj.UID, - Envelope: envelope, - SinkSpec: sinkObj.FamilySinkSpec(), - }) - Expect(poolErr).NotTo(HaveOccurred()) + // createPooledSink creates a sink, waits until the manager's cache has + // it and pools a spy backend for its UID. + createPooledSink := func() (*kollectdevv1alpha1.KollectSnapshotSink, *deleteWatchSpyBackend) { + obj := &kollectdevv1alpha1.KollectSnapshotSink{ + ObjectMeta: metav1.ObjectMeta{GenerateName: "delete-watch-", Namespace: "default"}, + Spec: kollectdevv1alpha1.KollectSnapshotSinkSpec{Type: "s3"}, + } + Expect(k8sClient.Create(watchCtx, obj)).To(Succeed()) + Eventually(func() error { + return mgr.GetClient().Get(watchCtx, types.NamespacedName{Namespace: obj.Namespace, Name: obj.Name}, &kollectdevv1alpha1.KollectSnapshotSink{}) + }).WithTimeout(30 * time.Second).Should(Succeed()) + + spy := &deleteWatchSpyBackend{} + reg := sink.NewRegistry() + reg.Register("s3", func(_ kollectdevv1alpha1.KollectSinkSpec, _ sink.BuildContext) (sink.Backend, error) { + return spy, nil + }) + _, poolErr := sink.RunExportEnvelope(sink.ExportEnvelopeRequest{ + Ctx: watchCtx, + Client: k8sClient, + Registry: reg, + SinkNamespace: obj.Namespace, + SinkName: obj.Name, + SinkUID: obj.UID, + Envelope: envelope, + SinkSpec: obj.FamilySinkSpec(), + }) + Expect(poolErr).NotTo(HaveOccurred()) + + return obj, spy + } + + const controllerLabel = "kollectsnapshotsink-delete-watch" + + reconciles := func() float64 { + return testutil.ToFloat64(metrics.ReconcileTotal.WithLabelValues(controllerLabel, metrics.ResultSuccess)) + + testutil.ToFloat64(metrics.ReconcileTotal.WithLabelValues(controllerLabel, metrics.ResultFailure)) + } + before := reconciles() + + sinkObj, spy := createPooledSink() + + // Cache sync does not mean the controller's Watches handler is attached: + // the manager starts controllers after the caches sync, asynchronously, + // and a delete issued before the handler registers is never replayed to + // it (late handlers only get synthetic Adds for objects still in the + // store). Workers start only after every source has synced, which is + // after its handler was added, so an observed reconcile of the sink + // proves the Watches handler is live before the delete. + Eventually(reconciles).WithTimeout(30*time.Second).WithPolling(50*time.Millisecond). + Should(BeNumerically(">", before), "controller never reconciled the sink: manager/controller did not start") Expect(k8sClient.Delete(watchCtx, sinkObj)).To(Succeed()) diff --git a/internal/sink/git/errors.go b/internal/sink/git/errors.go index 0e8d9707..3ba2755b 100644 --- a/internal/sink/git/errors.go +++ b/internal/sink/git/errors.go @@ -8,9 +8,13 @@ package git import ( "errors" "fmt" + "regexp" + "strconv" "strings" + "github.com/go-git/go-git/v5/plumbing" gittransport "github.com/go-git/go-git/v5/plumbing/transport" + githttp "github.com/go-git/go-git/v5/plumbing/transport/http" utilnet "k8s.io/apimachinery/pkg/util/net" kollecterrors "github.com/platformrelay/kollect/internal/errors" @@ -34,7 +38,7 @@ func ClassifyExportError(err error) error { if strings.Contains(msg, "protected branch") || strings.Contains(msg, "pre-receive hook declined") || - strings.Contains(msg, "remote rejected") && strings.Contains(msg, "403") { + strings.Contains(msg, "remote rejected") && hasAnyStatus(httpStatusCodes(err, msg), isAuthStatus) { return kollecterrors.Terminal(fmt.Errorf("git push rejected: %w", err)) } @@ -64,22 +68,77 @@ func isTransientTransportError(err error) bool { strings.Contains(msg, "timeout") || strings.Contains(msg, "temporary failure") || strings.Contains(msg, "eof") || - strings.Contains(msg, "503") || - strings.Contains(msg, "429") || + hasAnyStatus(httpStatusCodes(err, msg), isRetryableStatus) || strings.Contains(msg, "too many requests") } +var ( + // statusPhraseRE matches a status code only after a phrase that introduces + // one ("status code: 502", "returned error: 403", "HTTP 401", "code=401"). + statusPhraseRE = regexp.MustCompile(`(?:returned error:|status code:?|status:|code[=:]|http/?[0-9.]*|error:)\s*\(?([0-9]{3})(?:[^0-9]|$)`) + // statusPunctRE matches a parenthesised or colon-terminated code: "(429)", "401:". + statusPunctRE = regexp.MustCompile(`(?:^|[\s(])([0-9]{3})[):]`) + // statusTextRE matches a code followed by its reason phrase: "403 forbidden". + statusTextRE = regexp.MustCompile(`(?:^|\s)([0-9]{3}) (?:unauthorized|forbidden|too many requests|service unavailable|bad gateway|gateway timeout|internal server error)`) +) + +// httpStatusCodes extracts the HTTP status codes an export error carries. It +// prefers go-git's typed *http.Err (its Response.StatusCode is authoritative) +// and otherwise reads only status phrases, never bare digit runs: a code-like +// substring in a URL, hostname or temp/mirror directory name (team401, +// git-503.example.com, a hex mirror dir) is not a status. +func httpStatusCodes(err error, msg string) []int { + var codes []int + + var httpErr *githttp.Err + if errors.As(err, &httpErr) && httpErr.Response != nil { + codes = append(codes, httpErr.StatusCode()) + } + + // plumbing.UnexpectedError does not implement Unwrap. + var unexpected *plumbing.UnexpectedError + if errors.As(err, &unexpected) && unexpected.Err != nil { + if errors.As(unexpected.Err, &httpErr) && httpErr.Response != nil { + codes = append(codes, httpErr.StatusCode()) + } + } + + for _, re := range []*regexp.Regexp{statusPhraseRE, statusPunctRE, statusTextRE} { + for _, m := range re.FindAllStringSubmatch(msg, -1) { + if n, convErr := strconv.Atoi(m[1]); convErr == nil { + codes = append(codes, n) + } + } + } + + return codes +} + +func hasAnyStatus(codes []int, match func(int) bool) bool { + for _, c := range codes { + if match(c) { + return true + } + } + + return false +} + +func isAuthStatus(c int) bool { return c == 401 || c == 403 } + +func isRetryableStatus(c int) bool { return c == 429 || c >= 500 && c <= 599 } + func isAuthFailure(msg string, err error) bool { if strings.Contains(msg, "authentication required") || strings.Contains(msg, "invalid credentials") || strings.Contains(msg, "authorization failed") || strings.Contains(msg, "access denied") || - strings.Contains(msg, "401") || - strings.Contains(msg, "403 forbidden") { + hasAnyStatus(httpStatusCodes(err, msg), isAuthStatus) { return true } - if errors.Is(err, gittransport.ErrAuthenticationRequired) { + if errors.Is(err, gittransport.ErrAuthenticationRequired) || + errors.Is(err, gittransport.ErrAuthorizationFailed) { return true } diff --git a/internal/sink/git/errors_test.go b/internal/sink/git/errors_test.go index 1d6ca528..97fe3175 100644 --- a/internal/sink/git/errors_test.go +++ b/internal/sink/git/errors_test.go @@ -7,10 +7,15 @@ import ( "errors" "fmt" "io" + "net/http" + "net/url" + "strings" "syscall" "testing" + "github.com/go-git/go-git/v5/plumbing" gittransport "github.com/go-git/go-git/v5/plumbing/transport" + githttp "github.com/go-git/go-git/v5/plumbing/transport/http" kollecterrors "github.com/platformrelay/kollect/internal/errors" ) @@ -151,3 +156,52 @@ func TestIsNonFastForwardError(t *testing.T) { t.Fatal("expected non-fast-forward for combined rejected-refs message") } } + +// Status codes are read from typed go-git errors and status phrases only; digit +// runs inside paths, hostnames and mirror dirs are not statuses. +func TestClassifyExportError_statusCodes(t *testing.T) { + t.Parallel() + + typed := plumbing.NewUnexpectedError(&githttp.Err{ + Response: &http.Response{StatusCode: 502, Request: &http.Request{URL: &url.URL{Scheme: "https", Host: "git.example.com", Path: "/team401/infra.git/info/refs"}}}, + }) + + rows := []struct { + name string + err error + want string // terminal | transient | other + }{ + {"missing local remote, digits in temp dir", errors.New("git clone: fatal: '/tmp/x3401029481/missing.git' does not appear to be a git repository"), "other"}, + {"typed 502 with team401 in URL", typed, "transient"}, + {"text 502 with team401 in URL", errors.New(`unexpected requesting "https://git.example.com/team401/infra.git/info/refs" status code: 502`), "transient"}, + {"hex mirror dir", errors.New("mirror dir: mkdir /var/git-mirrors/3fa401bc9e0d: permission denied"), "other"}, + {"hostname with 503", errors.New("lookup git-503.example.com: no such host"), "other"}, + {"digits inside 15039", errors.New("path /tmp/a15039b"), "other"}, + {"cli 403", errors.New("fatal: unable to access 'https://h/r.git/': The requested URL returned error: 403"), "terminal"}, + {"unexpected client error 401", errors.New("unexpected client error: 401"), "terminal"}, + {"http 403 forbidden", errors.New("HTTP 403 Forbidden"), "terminal"}, + {"status code 401", errors.New("status code: 401"), "terminal"}, + {"code=401", errors.New("code=401"), "terminal"}, + {"401 colon", errors.New("push: 401: bad"), "terminal"}, + {"remote rejected (403)", errors.New("remote rejected (403)"), "terminal"}, + {"wrapped authorization failed", fmt.Errorf("push: %w", gittransport.ErrAuthorizationFailed), "terminal"}, + {"returned error 503", errors.New("The requested URL returned error: 503"), "transient"}, + {"(429)", errors.New("rate limited (429)"), "transient"}, + {"status code 429", errors.New("status code: 429"), "transient"}, + } + + for _, row := range rows { + got := "other" + + switch classified := ClassifyExportError(row.err); { + case kollecterrors.IsTerminal(classified): + got = "terminal" + case strings.Contains(classified.Error(), "git transport"): + got = "transient" + } + + if got != row.want { + t.Errorf("%s: classified %s, want %s (%v)", row.name, got, row.want, row.err) + } + } +}