diff --git a/pkg/lambda/grpc/failure.go b/pkg/lambda/grpc/failure.go index d80cf5abd..4b8a36a4e 100644 --- a/pkg/lambda/grpc/failure.go +++ b/pkg/lambda/grpc/failure.go @@ -33,12 +33,20 @@ const ( // FailureClassHandled means the function returned an error response through // the runtime API rather than crashing. FailureClassHandled = "handled" + // FailureClassUnsupportedRPC means the connector could not resolve the + // request message's type URL against its proto registry, which is how an + // SDK older than the RPC fails. The RPC is absent, not broken. + FailureClassUnsupportedRPC = "unsupported_rpc" ) // oomErrorType is the Error Type the AWS platform stamps on the REPORT line // when it kills a sandbox for exceeding its memory limit. const oomErrorType = "Runtime.OutOfMemory" +// protoResolveErrorType is the errorType a connector runtime reports when it +// cannot resolve a message's type URL against its proto registry. +const protoResolveErrorType = "prefixError" + // LambdaInvokeFailure is a structured description of a failed Lambda invoke. // // Every field is derived from data the invoke already returns: the base64 tail @@ -105,6 +113,13 @@ func (e *LambdaInvokeFailure) Code() codes.Code { return codes.DeadlineExceeded case FailureClassOOM: return codes.ResourceExhausted + case FailureClassUnsupportedRPC: + // The connector predates the RPC, so every invoke fails identically + // until it is rebuilt. Unimplemented lets a caller skip the step the + // same way it would for any connector that does not serve the method, + // and survives error-string sanitization, which a substring match on + // the log summary does not. + return codes.Unimplemented default: return codes.Unknown } @@ -129,6 +144,8 @@ func (e *LambdaInvokeFailure) Error() string { } case FailureClassTimeout: _, _ = b.WriteString("function timed out") + case FailureClassUnsupportedRPC: + _, _ = b.WriteString("function does not support this RPC") default: _, _ = b.WriteString("function returned error") } @@ -373,6 +390,15 @@ func extractMeaningfulLogLines(raw string) string { func classifyLambdaFailure(functionError string, statusCode int32, payload []byte, rawLog string) *LambdaInvokeFailure { report, haveReport := parseLambdaReportLine(rawLog) errPayload := parseLambdaErrorPayload(payload) + + // Two different views of the same log, on purpose. + // + // LogSummary is the one that may reach a customer-visible field, so it + // keeps the truncation pre-filter that drops a partial leading line. + // Classification reads the unfiltered lines instead: dropping a line is a + // sanitization decision, and a sanitization decision must never be able to + // change which failure class an invoke lands in. + signalLogs := extractMeaningfulLogLines(rawLog) filteredLogs := extractMeaningfulLogLines(dropTruncatedFirstLine(rawLog)) failure := &LambdaInvokeFailure{ @@ -392,7 +418,15 @@ func classifyLambdaFailure(functionError string, statusCode int32, payload []byt if haveReport && report.ErrorType != "" { failure.ErrorType = report.ErrorType } - failure.FailureClass = lambdaFailureClass(functionError, payload, filteredLogs, report, failure.ErrorType) + failure.FailureClass = lambdaFailureClass( + functionError, + payload, + signalLogs, + report, + failure.ErrorType, + errPayload.ErrorType, + failure.ErrorMessage, + ) return failure } @@ -402,7 +436,21 @@ func classifyLambdaFailure(functionError string, statusCode int32, payload []byt // Timeout is checked first: a sandbox killed on its execution timeout can also // show peak memory at its ceiling, which would otherwise trip the OOM // memory-comparison fallback. -func lambdaFailureClass(functionError string, payload []byte, filteredLogs string, report lambdaReport, errorType string) string { +// +// errorType is the resolved type, where a REPORT line's platform verdict wins +// over the payload's. payloadErrorType is the function's own, always. The two +// are separate because the OOM signals want the platform's verdict while an +// absent RPC is a statement only the function can make: a REPORT line carrying +// any Error Type would otherwise overwrite it and lose the capability gap. +func lambdaFailureClass( + functionError string, + payload []byte, + signalLogs string, + report lambdaReport, + errorType string, + payloadErrorType string, + errorMessage string, +) string { // Existing signal, unchanged: the platform writes this into the error // payload on a hard timeout kill. if strings.Contains(string(payload), "Task timed out after") { @@ -410,7 +458,7 @@ func lambdaFailureClass(functionError string, payload []byte, filteredLogs strin } // Existing signal, unchanged: the function's own context deadline was // exhausted and the connector logged it. - if strings.Contains(filteredLogs, `\"error\":\"context deadline exceeded\"`) { + if strings.Contains(signalLogs, `\"error\":\"context deadline exceeded\"`) { return FailureClassTimeout } // New signal: newer runtimes stamp the outcome on the REPORT line. @@ -418,13 +466,29 @@ func lambdaFailureClass(functionError string, payload []byte, filteredLogs strin return FailureClassTimeout } + // A connector whose SDK predates the RPC cannot resolve the request + // message's type URL. Checked before the OOM signals because it is an exact + // match on the runtime's own verdict, where the memory fallback below is + // inferred. + if isUnresolvedTypeURL(payloadErrorType, errorMessage, signalLogs) { + return FailureClassUnsupportedRPC + } + // Primary OOM signal, from the REPORT line or the error payload. if errorType == oomErrorType { return FailureClassOOM } // Fallback for runtimes that report the kill without an Error Type: a // failed invoke whose peak memory reached its ceiling was an OOM. - if strings.EqualFold(report.Status, "error") && + // + // Gated on the error type not being the function's own. A process the + // platform killed reports either nothing or a Runtime.* type such as + // Runtime.ExitError ("signal: killed"), so the ceiling is the best + // explanation available. When the function surfaced its own error value it + // has already explained the failure, and peak memory sitting at the ceiling + // is a coincidence - a Go runtime routinely runs at its ceiling without + // being killed. + if !isFunctionErrorType(errorType) && strings.EqualFold(report.Status, "error") && report.MemorySizeMB > 0 && report.MaxMemoryUsedMB >= report.MemorySizeMB { return FailureClassOOM } @@ -435,6 +499,38 @@ func lambdaFailureClass(functionError string, payload []byte, filteredLogs strin return FailureClassUnhandled } +// platformErrorTypePrefix marks the error types the AWS runtime generates +// itself, such as Runtime.ExitError and Runtime.OutOfMemory. Anything without +// it came from the function's own error value. +const platformErrorTypePrefix = "Runtime." + +// isFunctionErrorType reports whether the error type came from the function +// rather than the platform. An absent type is not a function error: a hard kill +// leaves nothing behind. +func isFunctionErrorType(errorType string) bool { + return errorType != "" && !strings.HasPrefix(errorType, platformErrorTypePrefix) +} + +// isUnresolvedTypeURL reports whether the runtime failed to resolve a message's +// type URL against its proto registry. +// +// Both the error payload's message and the log lines are searched because which +// one carries the text depends on how the runtime surfaced the failure. The +// type URL itself is deliberately not matched: every RPC added after a given +// connector's SDK version fails this same way, so pinning the check to one +// method name would leave the next one unhandled. +func isUnresolvedTypeURL(errorType string, errorMessage string, signalLogs string) bool { + if errorType != protoResolveErrorType { + return false + } + for _, s := range []string{errorMessage, signalLogs} { + if strings.Contains(s, "unable to resolve") && strings.Contains(s, "type.googleapis.com/") { + return true + } + } + return false +} + // lambdaLogTailTruncationThresholdBytes is the point at which a tail log may // have been cut mid-line. AWS returns the last 4 KB of the execution log with // LogType: Tail and cuts the window at a byte offset, not a line boundary, so a @@ -493,20 +589,42 @@ func looksLikeLogLineStart(line string) bool { return true } } - // A text-format runtime line, which the platform prefixes with an - // RFC3339 timestamp such as "2006-01-02T15:04:05.000Z". + // A text-format line, identified by the timestamp its writer puts in front. return looksLikeTimestampPrefix(line) } -// looksLikeTimestampPrefix reports whether a line opens with an RFC3339-style -// date, i.e. "NNNN-NN-NNT". +// timestampLayouts are the line-leading timestamp shapes that mark a whole log +// record, written with 'N' standing for any digit. +// +// Both entries matter. The AWS platform writes RFC3339, but the connector +// runtimes writing into this log use Go's standard logger, whose default prefix +// is "2006/01/02 15:04:05" - slashes and a space, not dashes and a "T". +// Recognising only RFC3339 classifies every one of those whole lines as a +// truncated fragment. +var timestampLayouts = []string{ + "NNNN-NN-NNT", + "NNNN/NN/NN NN:NN:NN", +} + +// looksLikeTimestampPrefix reports whether a line opens with a recognised +// timestamp layout. func looksLikeTimestampPrefix(line string) bool { - const stamp = "NNNN-NN-NNT" - if len(line) < len(stamp) { + for _, layout := range timestampLayouts { + if matchesDigitLayout(line, layout) { + return true + } + } + return false +} + +// matchesDigitLayout reports whether line starts with layout, where 'N' matches +// any digit and every other byte must match exactly. +func matchesDigitLayout(line string, layout string) bool { + if len(line) < len(layout) { return false } - for i := range len(stamp) { - want := stamp[i] + for i := range len(layout) { + want := layout[i] got := line[i] switch want { case 'N': diff --git a/pkg/lambda/grpc/failure_test.go b/pkg/lambda/grpc/failure_test.go index d75d7f55a..d719f8c2d 100644 --- a/pkg/lambda/grpc/failure_test.go +++ b/pkg/lambda/grpc/failure_test.go @@ -288,6 +288,75 @@ func TestClassifyLambdaFailure(t *testing.T) { wantCode: codes.Unknown, wantLogSummary: "lambda-run: unexpected failure", }, + { + // A connector whose SDK predates the RPC cannot resolve the request + // message's type URL. The method is absent, not broken, so it maps + // to Unimplemented and the caller can skip the step. + name: "unresolved request type from an old connector sdk", + functionError: "Unhandled", + statusCode: 200, + payload: `{"errorMessage":"proto: (line 1:88): unable to resolve ` + + `\"type.googleapis.com/c1.connector.v2.ExampleServiceListExamplesRequest\": ` + + `\"not found\"","errorType":"prefixError"}`, + rawLog: "2026/05/22 21:50:47 unable to resolve type URL\n", + + wantClass: FailureClassUnsupportedRPC, + wantCode: codes.Unimplemented, + wantErrorType: "prefixError", + wantLogSummary: "2026/05/22 21:50:47 unable to resolve type URL", + wantErrorMessage: `proto: (line 1:88): unable to resolve ` + + `"type.googleapis.com/c1.connector.v2.ExampleServiceListExamplesRequest": "not found"`, + }, + { + // The memory fallback infers an OOM from peak memory reaching the + // ceiling. A function that surfaced its own error value already + // explained the failure, and a Go runtime routinely sits at its + // ceiling without being killed, so the ceiling proves nothing here. + name: "function error type at memory ceiling is not an oom", + functionError: "Unhandled", + statusCode: 200, + payload: `{"errorType":"prefixError","errorMessage":"malformed request"}`, + rawLog: "REPORT RequestId: abc-123\tDuration: 5000.00 ms\tBilled Duration: 5000 ms\t" + + "Memory Size: 128 MB\tMax Memory Used: 128 MB\tStatus: error\n", + + wantClass: FailureClassUnhandled, + wantCode: codes.Unknown, + wantRequestID: "abc-123", + wantErrorType: "prefixError", + wantMemorySize: 128, + wantMaxMemory: 128, + wantDurationMS: 5000, + wantUtilization: 100, + wantErrorMessage: "malformed request", + }, + { + // An absent RPC is a statement only the function can make, so the + // classification reads the payload's error type rather than the + // resolved one. A REPORT line carrying its own Error Type must not + // overwrite that and turn the capability gap back into a crash -- + // the ErrorType field still reports the platform's verdict, because + // that is what the field means. + name: "report error type does not mask an unresolved request type", + functionError: "Unhandled", + statusCode: 200, + payload: `{"errorMessage":"proto: (line 1:88): unable to resolve ` + + `\"type.googleapis.com/c1.connector.v2.ExampleServiceListExamplesRequest\": ` + + `\"not found\"","errorType":"prefixError"}`, + rawLog: "REPORT RequestId: def-456\tDuration: 12.00 ms\tBilled Duration: 12 ms\t" + + "Memory Size: 128 MB\tMax Memory Used: 64 MB\tStatus: error\t" + + "Error Type: Runtime.ExitError\n", + + wantClass: FailureClassUnsupportedRPC, + wantCode: codes.Unimplemented, + wantRequestID: "def-456", + wantErrorType: "Runtime.ExitError", + wantMemorySize: 128, + wantMaxMemory: 64, + wantDurationMS: 12, + wantErrorMessage: `proto: (line 1:88): unable to resolve ` + + `"type.googleapis.com/c1.connector.v2.ExampleServiceListExamplesRequest": "not found"`, + wantUtilization: 50, + }, } for _, c := range cases { @@ -449,6 +518,66 @@ func TestDropTruncatedFirstLine(t *testing.T) { raw := strings.Repeat("x", lambdaLogTailTruncationThresholdBytes+10) require.Equal(t, raw, dropTruncatedFirstLine(raw)) }) + + // Connector runtimes log through Go's standard logger, whose default prefix + // is "2006/01/02 15:04:05". Recognising only RFC3339 treated every one of + // those whole lines as a truncated fragment and dropped it. + t.Run("keeps a Go stdlib timestamped leading line in a full window", func(t *testing.T) { + var b strings.Builder + b.WriteString("2026/05/22 21:50:47 lambda-run: failed to get connector\n") + for b.Len() < lambdaLogTailTruncationThresholdBytes { + b.WriteString(`{"level":"debug","msg":"listing resources","duration_ms":12}` + "\n") + } + require.Equal(t, b.String(), dropTruncatedFirstLine(b.String())) + }) +} + +// TestClassifyLambdaFailureTruncationDoesNotChangeClass pins that sanitizing the +// log summary cannot change which class an invoke lands in. +// +// The truncation pre-filter drops a leading line it cannot recognise as a whole +// record. Classifying from that filtered text made the in-function timeout +// signal vanish whenever it landed on the first line of a full tail window, +// downgrading a retryable DeadlineExceeded into a terminal Unknown. +func TestClassifyLambdaFailureTruncationDoesNotChangeClass(t *testing.T) { + const deadlineMarker = `\"error\":\"context deadline exceeded\"` + + fullTail := func(firstLine string) string { + var b strings.Builder + b.WriteString(firstLine) + b.WriteString("\n") + for b.Len() < lambdaLogTailTruncationThresholdBytes { + b.WriteString("2026/05/22 21:50:48 still listing resources\n") + } + b.WriteString(reportHealthy) + return b.String() + } + + // A whole Go-stdlib line that the pre-filter used to misjudge as a fragment. + t.Run("whole leading line carrying the signal", func(t *testing.T) { + raw := fullTail(`2026/05/22 21:50:47 {"level":"error",` + deadlineMarker + `}`) + require.GreaterOrEqual(t, len(raw), lambdaLogTailTruncationThresholdBytes) + + failure := classifyLambdaFailure("Unhandled", 200, nil, raw) + require.Equal(t, FailureClassTimeout, failure.FailureClass) + require.Equal(t, codes.DeadlineExceeded, failure.Code(), + "a timeout must stay retryable for the sync framework") + }) + + // A genuine mid-record fragment that still carries the signal. The pre-filter + // is right to keep this out of the summary and wrong to let that decision + // reach classification, so this case is covered by reading the raw log. + t.Run("truncated leading line carrying the signal", func(t *testing.T) { + fragment := `msg":"page timed out",` + deadlineMarker + `}` + raw := fullTail(fragment) + require.GreaterOrEqual(t, len(raw), lambdaLogTailTruncationThresholdBytes) + + failure := classifyLambdaFailure("Unhandled", 200, nil, raw) + require.Equal(t, FailureClassTimeout, failure.FailureClass) + require.Equal(t, codes.DeadlineExceeded, failure.Code()) + require.NotContains(t, failure.LogSummary, fragment, + "the fragment must still be kept out of the sanitizable summary") + }) } // TestClassifyLambdaFailureTruncatedTailDoesNotLeak is the regression test for @@ -483,6 +612,11 @@ func TestLooksLikeTimestampPrefix(t *testing.T) { expected bool }{ {line: "2026-05-22T21:50:47.000Z\tabc\tINFO\thello", expected: true}, + // Go's standard logger default prefix, which connector runtimes emit. + {line: "2026/05/22 21:50:47 lambda-run: failed to get connector", expected: true}, + {line: "2026/05/22 21:50:47", expected: true}, + {line: "2026/05/22 21:50", expected: false}, + {line: "2026/05/22", expected: false}, {line: "2026-05-22 21:50:47", expected: false}, {line: "202X-05-22T21:50:47.000Z", expected: false}, {line: "2026-05-22", expected: false}, diff --git a/pkg/sync/syncer.go b/pkg/sync/syncer.go index de3c99604..f1347be8c 100644 --- a/pkg/sync/syncer.go +++ b/pkg/sync/syncer.go @@ -2108,6 +2108,12 @@ func (s *syncer) SyncStaticEntitlements(ctx context.Context, action *Action) err return s.nextPageOrFinishAction(ctx, action, "", actions...) } +// legacyUnresolvedStaticEntitlementsMarker is the raw lambda error text a +// connector too old for ListStaticEntitlements produces. It is a fallback for +// transports that do not classify the failure as codes.Unimplemented; prefer +// the status code, which survives error-string sanitization. +const legacyUnresolvedStaticEntitlementsMarker = `unable to resolve \"type.googleapis.com/c1.connector.v2.EntitlementsServiceListStaticEntitlementsRequest\": \"not found\"","errorType":"prefixError"` + func (s *syncer) syncStaticEntitlementsForResourceType(ctx context.Context, action *Action) error { ctx, span := tracer.Start(ctx, "syncer.syncStaticEntitlementsForResource") var err error @@ -2123,10 +2129,19 @@ func (s *syncer) syncStaticEntitlementsForResourceType(ctx context.Context, acti s.recordSessionUsage(resp.GetAnnotations()) s.recordConnectorWaitReport(resp.GetAnnotations(), action.ResourceTypeID) if err != nil { - // Ignore prefixError if we're calling a lambda with an old version of baton-sdk. - if strings.Contains(err.Error(), `unable to resolve \"type.googleapis.com/c1.connector.v2.EntitlementsServiceListStaticEntitlementsRequest\": \"not found\"","errorType":"prefixError"`) { + // A connector built against an SDK older than this RPC cannot resolve + // the request message's type URL, so the method is absent rather than + // broken. Skip the step and let the rest of the sync run. + // + // The status code is the signal that holds. The substring below reads + // the connector's raw log text, which a caller may sanitize out of the + // error before it reaches here; it is kept only so this still works + // against a transport that does not yet map the failure to + // Unimplemented. + if status.Code(err) == codes.Unimplemented || + strings.Contains(err.Error(), legacyUnresolvedStaticEntitlementsMarker) { l := ctxzap.Extract(ctx) - l.Info("ignoring prefixError when calling ListStaticEntitlements", zap.Error(err)) + l.Info("connector does not support ListStaticEntitlements, skipping", zap.Error(err)) s.state.FinishAction(ctx, action) return nil } diff --git a/pkg/sync/syncer_test.go b/pkg/sync/syncer_test.go index 112053fcb..2f95aeaa6 100644 --- a/pkg/sync/syncer_test.go +++ b/pkg/sync/syncer_test.go @@ -140,6 +140,74 @@ func TestExpandGrants(t *testing.T) { }) } +// TestSyncToleratesUnimplementedStaticEntitlements pins that a connector too old +// to serve ListStaticEntitlements does not fail the whole sync. +// +// A connector built against an SDK that predates the RPC cannot resolve the +// request message's type URL, and the transport reports that as +// codes.Unimplemented. The tolerance used to key off a substring of the +// connector's raw log text instead, which a caller that sanitizes transport +// errors strips out - so the tolerance silently stopped applying and every sync +// of such a connector failed. +// +// The "other errors" case is also the reachability oracle: it only fails the +// sync if the RPC is genuinely called during a full sync. +func TestSyncToleratesUnimplementedStaticEntitlements(t *testing.T) { + cases := []struct { + name string + err error + wantErr bool + }{ + { + name: "unimplemented is skipped", + // What the caller sees once the transport classifies the failure + // and sanitizes the message. + err: status.Error(codes.Unimplemented, "connector function returned an error"), + }, + { + name: "legacy unsanitized transport error is still skipped", + err: fmt.Errorf("lambda_transport: function returned error: Unhandled; logSummary: %s", + legacyUnresolvedStaticEntitlementsMarker), + }, + { + name: "unrelated errors still fail the sync", + err: status.Error(codes.Internal, "simulated connector failure"), + wantErr: true, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + runWithSyncModes(t, func(t *testing.T, extraOpts []SyncOpt) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + mc := newMockConnector() + mc.rtDB = append(mc.rtDB, groupResourceType, userResourceType) + group, _, err := mc.AddGroup(ctx, "group_0") + require.NoError(t, err) + user, err := mc.AddUser(ctx, "user_0") + require.NoError(t, err) + _ = mc.AddGroupMember(ctx, group, user) + + tempDir := t.TempDir() + c1zpath := filepath.Join(tempDir, "static-entitlements.c1z") + opts := append([]SyncOpt{WithC1ZPath(c1zpath), WithTmpDir(tempDir)}, extraOpts...) + syncer, err := NewSyncer(ctx, &staticEntitlementsErrorMockConnector{mockConnector: mc, err: c.err}, opts...) + require.NoError(t, err) + + err = syncer.Sync(ctx) + if c.wantErr { + require.Error(t, err, "an unrelated connector error must fail the sync") + } else { + require.NoError(t, err, "an absent ListStaticEntitlements must not fail the sync") + } + require.NoError(t, syncer.Close(ctx)) + }) + }) + } +} + func TestInvalidResourceTypeFilter(t *testing.T) { runWithSyncModes(t, func(t *testing.T, extraOpts []SyncOpt) { ctx := t.Context() @@ -1769,6 +1837,21 @@ func TestResumeSyncWithChildResources(t *testing.T) { } } +// staticEntitlementsErrorMockConnector fails ListStaticEntitlements with a +// caller-supplied error, standing in for a connector whose SDK predates the RPC. +type staticEntitlementsErrorMockConnector struct { + *mockConnector + err error +} + +func (mc *staticEntitlementsErrorMockConnector) ListStaticEntitlements( + ctx context.Context, + in *v2.EntitlementsServiceListStaticEntitlementsRequest, + opts ...grpc.CallOption, +) (*v2.EntitlementsServiceListStaticEntitlementsResponse, error) { + return nil, mc.err +} + // failChildResourceMockConnector wraps a mockConnector and fails ListResources calls for child resources. type failChildResourceMockConnector struct { *mockConnector