diff --git a/pkg/connector/event_feed_common.go b/pkg/connector/event_feed_common.go index f16dc7fd..70740d4a 100644 --- a/pkg/connector/event_feed_common.go +++ b/pkg/connector/event_feed_common.go @@ -9,9 +9,8 @@ // pagination.StreamToken, and never loops internally across directory pages (see // ref-antipatterns.md, "Client-Side Pagination Loop"). // -// Ordering note: it is not documented whether activities.list returns newest-first when no -// startTime/orderBy is given, so each lookup fetches a small bounded window (not maxResults=1) -// and picks the maximum occurredAt client-side, per the plan's acceptance-criteria fallback. +// Ordering note: activities.list ordering is undocumented, so with startTime=180 days back and +// maxResults=50, each lookup still picks the maximum occurredAt client-side rather than trusting result order. package connector import ( diff --git a/pkg/connector/google_login_event_feed.go b/pkg/connector/google_login_event_feed.go index d352abcc..9cfb916f 100644 --- a/pkg/connector/google_login_event_feed.go +++ b/pkg/connector/google_login_event_feed.go @@ -2,13 +2,17 @@ package connector import ( "context" + "errors" "fmt" "strconv" + "time" v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" "github.com/conductorone/baton-sdk/pkg/annotations" "github.com/conductorone/baton-sdk/pkg/pagination" "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "go.uber.org/zap" reportsAdmin "google.golang.org/api/admin/reports/v1" "google.golang.org/protobuf/types/known/timestamppb" @@ -20,11 +24,14 @@ type bestActivity struct { occurredAt *timestamppb.Timestamp } -// googleLoginLookupMaxResults bounds the per-user Reports API lookup for Google Workspace -// sign-in events. There is exactly one target app (Google Workspace itself), so this only -// needs to cover the "does maxResults=1 return newest-first?" ordering uncertainty: fetch a -// small window and pick the maximum occurredAt client-side. -const googleLoginLookupMaxResults = 5 +// googleLoginLookupMaxResults bounds the per-user Reports API lookup; the true latest event is +// picked client-side, so this just needs to be large enough to avoid pagination. +const googleLoginLookupMaxResults = 50 + +// googleLoginLookupTimeout caps a single user's lookup, including retries. Kept above the worst +// case of a hung attempt plus backoff plus a full retry (~51s) so the hung-attempt retry in +// listActivitiesRateLimitedBounded has room to complete. +const googleLoginLookupTimeout = 60 * time.Second // googleLoginEventFeed emits UsageEvents from Google Workspace sign-in activity. // Unlike SAML/OAuth feeds, the target resource is always Google Workspace itself. @@ -48,8 +55,21 @@ func (f *googleLoginEventFeed) EventFeedMetadata(_ context.Context) *v2.EventFee } func (f *googleLoginEventFeed) lookupUser(ctx context.Context, client *gwclient.GoogleWorkspaceClient, user pendingUser) ([]*v2.Event, error) { - r, err := listActivitiesRateLimited(ctx, client, user.Email, reportsAppLogin, "login_success", "", "", googleLoginLookupMaxResults) + l := ctxzap.Extract(ctx) + startTime := time.Now().Add(-reportsLookback).UTC().Format(time.RFC3339) + + lookupCtx, cancel := context.WithTimeout(ctx, googleLoginLookupTimeout) + defer cancel() + + r, err := listActivitiesRateLimitedBounded(lookupCtx, client, user.Email, reportsAppLogin, "login_success", startTime, "", googleLoginLookupMaxResults) if err != nil { + if errors.Is(err, context.DeadlineExceeded) && ctx.Err() == nil { + // Our sub-deadline fired, not the caller's context: skip this user instead of + // failing the whole batch. + l.Debug("google-workspace-connector: timed out listing google login activities, skipping", + zap.String("user", user.Email)) + return nil, nil + } return nil, fmt.Errorf("google-workspace-connector: failed to list google login activities for %s: %w", user.Email, err) } diff --git a/pkg/connector/reports_rate_limiter.go b/pkg/connector/reports_rate_limiter.go index 672fa134..aa158921 100644 --- a/pkg/connector/reports_rate_limiter.go +++ b/pkg/connector/reports_rate_limiter.go @@ -32,6 +32,14 @@ const ( reportsMaxRetries = 5 reportsInitialBackoff = 500 * time.Millisecond reportsMaxBackoff = 30 * time.Second + + // reportsPerAttemptTimeout bounds a single ListActivities call, not the retry loop as a + // whole, so a genuinely hung request is retried like any other transient error instead of + // being confused with the caller's overall lookup deadline expiring. + reportsPerAttemptTimeout = 25 * time.Second + + // reportsLookback matches Google's Reports retention window; shared by the event feeds. + reportsLookback = 180 * 24 * time.Hour ) // reportsRateLimiter is a simple token-bucket limiter built on the standard library only @@ -92,38 +100,93 @@ func (l *reportsRateLimiter) refillLocked() { l.lastRefill = l.now() } +// listActivitiesFunc mirrors GoogleWorkspaceClient.ListActivities so retryListActivities can be +// tested with a fake implementation. +type listActivitiesFunc func(ctx context.Context, userKey, applicationName, eventName, startTime, pageToken, filters string, maxResults int64) (*reportsAdmin.Activities, error) + // listActivitiesRateLimited waits for the shared filter-query budget, then calls // client.ListActivities, retrying with exponential backoff on 429/503 — both are transient, -// SDK-retryable conditions, not connector bugs (see patterns-error-handling.md). +// SDK-retryable conditions, not connector bugs (see patterns-error-handling.md). It imposes no +// per-attempt timeout: use listActivitiesRateLimitedBounded for callers that set their own +// lookupCtx sub-deadline and want a hung attempt retried instead of just waited out. func listActivitiesRateLimited( ctx context.Context, client *gwclient.GoogleWorkspaceClient, userKey, applicationName, eventName, startTime, pageToken string, maxResults int64, ) (*reportsAdmin.Activities, error) { - return listActivitiesFilteredRateLimited(ctx, client, userKey, applicationName, eventName, startTime, pageToken, "", maxResults) + return retryListActivities( + ctx, sharedReportsRateLimiter, client.ListActivities, + 0, reportsMaxRetries, reportsInitialBackoff, reportsMaxBackoff, + userKey, applicationName, eventName, startTime, pageToken, "", maxResults, + ) +} + +// listActivitiesRateLimitedBounded is listActivitiesRateLimited plus reportsPerAttemptTimeout +// applied to every attempt. Reserved for the event feed callers (usage/google-login/saml), which +// always wrap the call in their own bounded lookupCtx, so a hung attempt can be told apart from +// the caller's own deadline expiring and retried instead of failing the lookup outright. +func listActivitiesRateLimitedBounded( + ctx context.Context, + client *gwclient.GoogleWorkspaceClient, + userKey, applicationName, eventName, startTime, pageToken string, + maxResults int64, +) (*reportsAdmin.Activities, error) { + return listActivitiesFilteredRateLimitedBounded(ctx, client, userKey, applicationName, eventName, startTime, pageToken, "", maxResults) } -// listActivitiesFilteredRateLimited is listActivitiesRateLimited plus an optional Reports API -// `filters` expression (e.g. "client_id=="), for callers that need to scope a lookup to one -// specific app rather than an entire app-type. -func listActivitiesFilteredRateLimited( +// listActivitiesFilteredRateLimitedBounded is listActivitiesRateLimitedBounded plus an optional +// Reports API `filters` expression (e.g. "client_id=="), for callers that need to scope a +// lookup to one specific app rather than an entire app-type. +func listActivitiesFilteredRateLimitedBounded( ctx context.Context, client *gwclient.GoogleWorkspaceClient, userKey, applicationName, eventName, startTime, pageToken, filters string, maxResults int64, ) (*reportsAdmin.Activities, error) { - backoff := reportsInitialBackoff + return retryListActivities( + ctx, sharedReportsRateLimiter, client.ListActivities, + reportsPerAttemptTimeout, reportsMaxRetries, reportsInitialBackoff, reportsMaxBackoff, + userKey, applicationName, eventName, startTime, pageToken, filters, maxResults, + ) +} + +// retryListActivities holds the retry/backoff/per-attempt-timeout policy, parameterized so tests +// can drive it with a fake call and short durations. perAttemptTimeout == 0 means "no per-attempt +// cap" — the caller's own ctx is used as-is and a DeadlineExceeded from it is never retried as +// a hung attempt. +func retryListActivities( + ctx context.Context, + limiter *reportsRateLimiter, + call listActivitiesFunc, + perAttemptTimeout time.Duration, + maxRetries int, + initialBackoff, maxBackoff time.Duration, + userKey, applicationName, eventName, startTime, pageToken, filters string, + maxResults int64, +) (*reportsAdmin.Activities, error) { + applyPerAttemptTimeout := perAttemptTimeout > 0 + backoff := initialBackoff for attempt := 0; ; attempt++ { - if err := sharedReportsRateLimiter.Wait(ctx); err != nil { + if err := limiter.Wait(ctx); err != nil { return nil, fmt.Errorf("google-workspace-connector: context cancelled waiting for reports api quota: %w", err) } - resp, err := client.ListActivities(ctx, userKey, applicationName, eventName, startTime, pageToken, filters, maxResults) + attemptCtx := ctx + cancel := func() {} + if applyPerAttemptTimeout { + attemptCtx, cancel = context.WithTimeout(ctx, perAttemptTimeout) + } + resp, err := call(attemptCtx, userKey, applicationName, eventName, startTime, pageToken, filters, maxResults) + cancel() if err == nil { return resp, nil } - if attempt >= reportsMaxRetries || !isRetryableReportsError(err) { + + // ctx still being live means attemptCtx's own timeout fired, not the caller's deadline — + // treat that like a retryable 429/503 rather than "out of time." + hungAttempt := applyPerAttemptTimeout && errors.Is(err, context.DeadlineExceeded) && ctx.Err() == nil + if attempt >= maxRetries || (!isRetryableReportsError(err) && !hungAttempt) { return nil, err } @@ -134,8 +197,8 @@ func listActivitiesFilteredRateLimited( return nil, ctx.Err() } backoff *= 2 - if backoff > reportsMaxBackoff { - backoff = reportsMaxBackoff + if backoff > maxBackoff { + backoff = maxBackoff } } } diff --git a/pkg/connector/reports_rate_limiter_test.go b/pkg/connector/reports_rate_limiter_test.go new file mode 100644 index 00000000..9185093d --- /dev/null +++ b/pkg/connector/reports_rate_limiter_test.go @@ -0,0 +1,260 @@ +package connector + +import ( + "context" + "errors" + "net/http" + "net/url" + "sync/atomic" + "testing" + "time" + + reportsAdmin "google.golang.org/api/admin/reports/v1" + "google.golang.org/api/googleapi" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// unlimitedRateLimiter returns a limiter whose Wait() never blocks, so tests exercise only the +// retry/timeout policy in retryListActivities, not the token bucket. +func unlimitedRateLimiter() *reportsRateLimiter { + return newReportsRateLimiter(1_000_000) +} + +// blockingCall returns a listActivitiesFunc that, on each call, either sleeps for delay and +// returns results[callIndex] (nil error means success), or returns ctx.Err() if ctx is done +// first — mirroring how the real Reports API client responds to context cancellation. +func blockingCall(delay time.Duration, results ...error) (listActivitiesFunc, *int32) { + var calls int32 + fn := func(ctx context.Context, userKey, applicationName, eventName, startTime, pageToken, filters string, maxResults int64) (*reportsAdmin.Activities, error) { + idx := atomic.AddInt32(&calls, 1) - 1 + select { + case <-time.After(delay): + var err error + if int(idx) < len(results) { + err = results[idx] + } + if err != nil { + return nil, err + } + return &reportsAdmin.Activities{}, nil + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return fn, &calls +} + +// blockingCallWithDelays is blockingCall but each call gets its own delay from delays (the last +// entry repeats if there are more calls than delays), so a test can make the first attempt hang +// and later attempts return quickly. On a ctx timeout, the returned error is ctx.Err() as-is; use +// blockingCallWithDelaysWrapped to pin a production-shaped wrapping of that error instead. +func blockingCallWithDelays(delays []time.Duration, results ...error) (listActivitiesFunc, *int32) { + return blockingCallWithDelaysWrapped(delays, func(err error) error { return err }, results...) +} + +// blockingCallWithDelaysWrapped is blockingCallWithDelays, but passes a ctx timeout's error +// through wrapCtxErr before returning it — e.g. to mirror how net/http's transport wraps a +// context error in a *url.Error, so tests can pin that errors.Is still unwraps it correctly. +func blockingCallWithDelaysWrapped(delays []time.Duration, wrapCtxErr func(error) error, results ...error) (listActivitiesFunc, *int32) { + var calls int32 + fn := func(ctx context.Context, userKey, applicationName, eventName, startTime, pageToken, filters string, maxResults int64) (*reportsAdmin.Activities, error) { + idx := int(atomic.AddInt32(&calls, 1) - 1) + delay := delays[len(delays)-1] + if idx < len(delays) { + delay = delays[idx] + } + select { + case <-time.After(delay): + var err error + if idx < len(results) { + err = results[idx] + } + if err != nil { + return nil, err + } + return &reportsAdmin.Activities{}, nil + case <-ctx.Done(): + return nil, wrapCtxErr(ctx.Err()) + } + } + return fn, &calls +} + +func rateLimitedGoogleErr() error { + return &googleapi.Error{Code: http.StatusTooManyRequests} +} + +// productionWrappedRateLimitErr mirrors what client.ListActivities actually returns for a 429: +// wrapGoogleApiErrorWithContext joins a gRPC status with the original *googleapi.Error, rather +// than returning the bare *googleapi.Error the other tests use. +func productionWrappedRateLimitErr() error { + return errors.Join(status.Error(codes.Unavailable, "rate limited"), &googleapi.Error{Code: http.StatusTooManyRequests}) +} + +func TestRetryListActivities(t *testing.T) { + const ( + perAttemptTimeout = 20 * time.Millisecond + initialBackoff = 5 * time.Millisecond + maxBackoff = 20 * time.Millisecond + ) + + t.Run("succeeds on first attempt without retry", func(t *testing.T) { + call, calls := blockingCall(0) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, perAttemptTimeout, 5, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if err != nil { + t.Fatalf("expected success, got %v", err) + } + if got := atomic.LoadInt32(calls); got != 1 { + t.Fatalf("expected 1 call, got %d", got) + } + }) + + t.Run("retries a rate-limited error and then succeeds", func(t *testing.T) { + call, calls := blockingCall(0, rateLimitedGoogleErr()) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, perAttemptTimeout, 5, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if err != nil { + t.Fatalf("expected eventual success, got %v", err) + } + if got := atomic.LoadInt32(calls); got != 2 { + t.Fatalf("expected 2 calls, got %d", got) + } + }) + + t.Run("fails immediately on a non-retryable error", func(t *testing.T) { + nonRetryable := &googleapi.Error{Code: http.StatusForbidden} + call, calls := blockingCall(0, nonRetryable, nil) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, perAttemptTimeout, 5, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if !errors.Is(err, nonRetryable) { + t.Fatalf("expected the non-retryable error back, got %v", err) + } + if got := atomic.LoadInt32(calls); got != 1 { + t.Fatalf("expected exactly 1 call (no retry), got %d", got) + } + }) + + t.Run("retries when only attemptCtx's own timeout fires (hung attempt)", func(t *testing.T) { + // First call sleeps past perAttemptTimeout so attemptCtx fires; the outer ctx has a much + // longer deadline and is still alive. Second call returns quickly and succeeds. + call, calls := blockingCallWithDelays([]time.Duration{perAttemptTimeout * 3, 0}) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, perAttemptTimeout, 5, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if err != nil { + t.Fatalf("expected the hung attempt to be retried and eventually succeed, got %v", err) + } + if got := atomic.LoadInt32(calls); got != 2 { + t.Fatalf("expected 2 calls (initial hung attempt + retry), got %d", got) + } + }) + + t.Run("returns immediately when the caller's own ctx is also done", func(t *testing.T) { + // Outer ctx deadline is shorter than perAttemptTimeout, so when attemptCtx fires it is + // because the caller's own deadline expired, not attemptCtx's independent timeout. + outerDeadline := perAttemptTimeout / 4 + call, calls := blockingCall(perAttemptTimeout * 3) + ctx, cancel := context.WithTimeout(context.Background(), outerDeadline) + defer cancel() + + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, perAttemptTimeout, 5, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected context.DeadlineExceeded, got %v", err) + } + if got := atomic.LoadInt32(calls); got != 1 { + t.Fatalf("expected exactly 1 call (no retry once caller's ctx is done), got %d", got) + } + }) + + t.Run("imposes no per-attempt timeout when perAttemptTimeout is 0, even with a deadline-bearing ctx", func(t *testing.T) { + // perAttemptTimeout == 0 must disable the sub-timeout on its own terms, regardless of + // whether ctx happens to carry a deadline (e.g. a future SDK change attaches one to the + // sync context) — this is what listActivitiesRateLimited (used by app_login.go) relies + // on to stay unbounded. + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + wantDeadline, _ := ctx.Deadline() + + var gotDeadline time.Time + var gotHasDeadline bool + call := func(ctx context.Context, userKey, applicationName, eventName, startTime, pageToken, filters string, maxResults int64) (*reportsAdmin.Activities, error) { + gotDeadline, gotHasDeadline = ctx.Deadline() + time.Sleep(perAttemptTimeout * 3) + return &reportsAdmin.Activities{}, nil + } + + start := time.Now() + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, 0, 5, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if err != nil { + t.Fatalf("expected success, got %v", err) + } + // attemptCtx must be ctx itself (same deadline as the caller's own, not a new shorter + // one), proving perAttemptTimeout==0 disabled the sub-timeout rather than the caller + // simply having no deadline to begin with. + if !gotHasDeadline || !gotDeadline.Equal(wantDeadline) { + t.Fatalf("expected attemptCtx to carry the caller's own deadline unchanged, got hasDeadline=%v deadline=%v want=%v", gotHasDeadline, gotDeadline, wantDeadline) + } + if elapsed := time.Since(start); elapsed < perAttemptTimeout*3 { + t.Fatalf("expected the call to run past perAttemptTimeout uninterrupted, only took %v", elapsed) + } + }) + + t.Run("retries a rate-limited error wrapped the way production code wraps it", func(t *testing.T) { + // Pins the contract that isRetryableReportsError's errors.As still finds the + // *googleapi.Error inside wrapGoogleApiErrorWithContext's errors.Join, not just a bare + // *googleapi.Error. + call, calls := blockingCall(0, productionWrappedRateLimitErr()) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, perAttemptTimeout, 5, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if err != nil { + t.Fatalf("expected eventual success, got %v", err) + } + if got := atomic.LoadInt32(calls); got != 2 { + t.Fatalf("expected 2 calls, got %d", got) + } + }) + + t.Run("retries a hung attempt whose timeout arrives wrapped like production's transport error", func(t *testing.T) { + // Pins the contract that hungAttempt's errors.Is still finds context.DeadlineExceeded + // inside a *url.Error, not just a bare context error. + call, calls := blockingCallWithDelaysWrapped([]time.Duration{perAttemptTimeout * 3, 0}, func(err error) error { + return &url.Error{Op: "Get", URL: "https://example.com", Err: err} + }) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, perAttemptTimeout, 5, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if err != nil { + t.Fatalf("expected the wrapped hung attempt to be retried and eventually succeed, got %v", err) + } + if got := atomic.LoadInt32(calls); got != 2 { + t.Fatalf("expected 2 calls (initial hung attempt + retry), got %d", got) + } + }) + + t.Run("stops after maxRetries and returns the last error", func(t *testing.T) { + retryable := rateLimitedGoogleErr() + call, calls := blockingCall(0, retryable, retryable, retryable) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + _, err := retryListActivities(ctx, unlimitedRateLimiter(), call, perAttemptTimeout, 2, initialBackoff, maxBackoff, "u", "app", "event", "", "", "", 10) + if err == nil { + t.Fatalf("expected an error after exhausting retries") + } + if got := atomic.LoadInt32(calls); got != 3 { + t.Fatalf("expected 3 calls (1 initial + 2 retries), got %d", got) + } + }) +} diff --git a/pkg/connector/saml_event_feed.go b/pkg/connector/saml_event_feed.go index ebae6761..85553de3 100644 --- a/pkg/connector/saml_event_feed.go +++ b/pkg/connector/saml_event_feed.go @@ -2,13 +2,17 @@ package connector import ( "context" + "errors" "fmt" "strconv" + "time" v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" "github.com/conductorone/baton-sdk/pkg/annotations" "github.com/conductorone/baton-sdk/pkg/pagination" "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "go.uber.org/zap" reportsAdmin "google.golang.org/api/admin/reports/v1" "google.golang.org/protobuf/types/known/timestamppb" @@ -21,6 +25,11 @@ import ( // by resolved app ID, keeping only the newest event per app. const samlAppLookupMaxResults = 50 +// samlAppLookupTimeout caps a single user's lookup, including retries. Kept above the worst case +// of a hung attempt plus backoff plus a full retry (~51s) so the hung-attempt retry in +// listActivitiesRateLimitedBounded has room to complete. +const samlAppLookupTimeout = 60 * time.Second + // samlEventFeed emits UsageEvents from Google Workspace SAML app login activity. type samlEventFeed struct { client *gwclient.GoogleWorkspaceClient @@ -54,8 +63,21 @@ type samlAppActivity struct { // authentication, so last login timestamps are accurate. SAML apps are identified by app name // (no numeric client_id). func (f *samlEventFeed) lookupUser(ctx context.Context, client *gwclient.GoogleWorkspaceClient, samlProfileMap map[string]string, user pendingUser) ([]*v2.Event, error) { - r, err := listActivitiesRateLimited(ctx, client, user.Email, reportsAppSAML, "login_success", "", "", samlAppLookupMaxResults) + l := ctxzap.Extract(ctx) + startTime := time.Now().Add(-reportsLookback).UTC().Format(time.RFC3339) + + lookupCtx, cancel := context.WithTimeout(ctx, samlAppLookupTimeout) + defer cancel() + + r, err := listActivitiesRateLimitedBounded(lookupCtx, client, user.Email, reportsAppSAML, "login_success", startTime, "", samlAppLookupMaxResults) if err != nil { + if errors.Is(err, context.DeadlineExceeded) && ctx.Err() == nil { + // Our sub-deadline fired, not the caller's context: skip this user instead of + // failing the whole batch. + l.Debug("google-workspace-connector: timed out listing saml login activities, skipping", + zap.String("user", user.Email)) + return nil, nil + } return nil, fmt.Errorf("google-workspace-connector: failed to list saml login activities for %s: %w", user.Email, err) } diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index 674c5a35..178275ba 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -13,6 +13,8 @@ import ( "github.com/conductorone/baton-sdk/pkg/annotations" "github.com/conductorone/baton-sdk/pkg/pagination" "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "go.uber.org/zap" reportsAdmin "google.golang.org/api/admin/reports/v1" "google.golang.org/api/googleapi" "google.golang.org/protobuf/types/known/timestamppb" @@ -22,12 +24,14 @@ import ( var privateAppIDRegex = regexp.MustCompile("[0-9]{21}") -// oauthAppLookupMaxResults bounds each per-(user, app) Reports API lookup. Since the query is -// now scoped to one specific client_id via the `filters` param, there's no cross-app crowding to -// worry about — this only needs to cover the "does maxResults=1 return newest-first?" ordering -// assumption from the acceptance criteria: with a >1 window, the true latest is picked -// client-side regardless of Google's actual ordering. -const oauthAppLookupMaxResults = 5 +// oauthAppLookupMaxResults bounds each per-(user, app) Reports API lookup; the true latest event +// is picked client-side, so this just needs to be large enough to avoid pagination. +const oauthAppLookupMaxResults = 50 + +// oauthAppLookupTimeout caps a single (user, app) lookup, including retries. Kept above the +// worst case of a hung attempt plus backoff plus a full retry (~51s) so the hung-attempt retry +// in listActivitiesFilteredRateLimitedBounded has room to complete. +const oauthAppLookupTimeout = 60 * time.Second type usageEventFeed struct { c *gwclient.GoogleWorkspaceClient @@ -96,9 +100,8 @@ func (f *usageEventFeed) lookupUser(ctx context.Context, client *gwclient.Google } return nil, fmt.Errorf("google-workspace: failed to list oauth tokens for %s: %w", user.Email, err) } - // Tokens.list can return multiple Token entries for the same client_id — e.g. a user - // granting a different scope set to the same app at different times. Dedupe here so each - // distinct app only ever triggers one Reports API lookup, never repeated ones. + // Tokens.list can return multiple Token entries for the same client_id. + // Dedupe here so each distinct app only ever triggers one Reports API lookup, never repeated ones. seenClientIDs := make(map[string]struct{}, len(tokenResp.Items)) events := make([]*v2.Event, 0, len(tokenResp.Items)) @@ -128,11 +131,24 @@ func (f *usageEventFeed) lookupUser(ctx context.Context, client *gwclient.Google } // lookupAppLogin fetches this user's most recent "authorize" activity for one specific OAuth -// app (client_id), returning nil if there is no such activity within the lookup window. +// app (client_id), returning nil if there is no such activity, or if the lookup timed out. func (f *usageEventFeed) lookupAppLogin(ctx context.Context, client *gwclient.GoogleWorkspaceClient, user pendingUser, clientID, displayName string) (*v2.Event, error) { + l := ctxzap.Extract(ctx) filters := "client_id==" + clientID - r, err := listActivitiesFilteredRateLimited(ctx, client, user.Email, "token", "authorize", "", "", filters, oauthAppLookupMaxResults) + startTime := time.Now().Add(-reportsLookback).UTC().Format(time.RFC3339) + + lookupCtx, cancel := context.WithTimeout(ctx, oauthAppLookupTimeout) + defer cancel() + + r, err := listActivitiesFilteredRateLimitedBounded(lookupCtx, client, user.Email, "token", "authorize", startTime, "", filters, oauthAppLookupMaxResults) if err != nil { + if errors.Is(err, context.DeadlineExceeded) && ctx.Err() == nil { + // Our sub-deadline fired, not the caller's context: skip this one app instead of + // failing the whole batch. + l.Debug("google-workspace: timed out listing token activities, skipping", + zap.String("user", user.Email), zap.String("client_id", clientID)) + return nil, nil + } return nil, fmt.Errorf("google-workspace: failed to list token activities for %s app %s: %w", user.Email, clientID, err) }