Skip to content

CXP-533 Performance improvements to admin_event_feed #114

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
JavierCarnelli-ConductorOne wants to merge 5 commits into
base: main
Choose a base branch
from
Draft

CXP-533 Performance improvements to admin_event_feed #114

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5e0b005
feat: filter admin event feed server-side by event name
JavierCarnelli-ConductorOne Jun 3, 2026
eec7d0f
feat: request up to 1k entries per request on admin event feed
JavierCarnelli-ConductorOne Jun 3, 2026
9d11cc1
fix: increase page size for usage_event_feed requests
JavierCarnelli-ConductorOne Jun 4, 2026
98dd90f
fix: skip sibling events in admin event feed to prevent duplicate emi…
JavierCarnelli-ConductorOne Jun 10, 2026
868d049
refactor: hoist adminGroupEventNames lookup out of inner event loop
JavierCarnelli-ConductorOne Jun 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 91 additions & 44 deletions pkg/connector/admin_event_feed.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,13 +23,51 @@ import (
gwclient "github.com/conductorone/baton-google-workspace/pkg/client"
)

// adminEventNames lists the admin-application event names the feed subscribes to.
// The Google Reports API accepts one eventName per request, so ListEvents issues
// one ListActivities call per name and merges the results.
var adminEventNames = []string{
// GROUP_SETTINGS
"CREATE_GROUP",
"CHANGE_GROUP_DESCRIPTION",
"CHANGE_GROUP_NAME",
"CHANGE_GROUP_EMAIL",
"ADD_GROUP_MEMBER",
"UPDATE_GROUP_MEMBER",
"DELETE_GROUP",
// USER_SETTINGS

@luisina-santos luisina-santos Jun 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

DELETE_GROUP causes a wasted API call every sync

DELETE_GROUP is in adminEventNames so ListEvents issues a real ListActivities request for it on every cycle — but handleGroupEvent drops all results with a comment explaining the group can't be looked up after deletion. Consider removing it from the list (or adding it to a separate commented-out section) so we don't pay the API cost for events we can't use.

// "DELETE_GROUP" is excluded: the group is already gone by the time the event
// arrives, so the ID lookup always fails. Remove from adminEventNames.

"ACCEPT_USER_INVITATION",
"CHANGE_USER_ORGANIZATION",
"ADD_DISPLAY_NAME",
"CHANGE_DISPLAY_NAME",
"CHANGE_FIRST_NAME",
"CHANGE_LAST_NAME",
"CREATE_USER",
"RENAME_USER",
}

// adminGroupEventNames is the subset of adminEventNames that belong to GROUP_SETTINGS handling.
var adminGroupEventNames = map[string]bool{
"CREATE_GROUP": true,

@luisina-santos luisina-santos Jun 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Dual-maintenance risk between adminEventNames and adminGroupEventNames

These two declarations overlap: if a new group event is added to adminEventNames but forgotten in adminGroupEventNames, it will be silently routed to handleUserEvent with no compile-time or runtime error. Consider deriving adminEventNames from the two sets so there's a single source of truth.

var adminGroupEventNames = map[string]bool{ ... }
var adminUserEventNames  = map[string]bool{ ... }

var adminEventNames = func() []string {
    names := make([]string, 0, len(adminGroupEventNames)+len(adminUserEventNames))
    for k := range adminGroupEventNames { names = append(names, k) }
    for k := range adminUserEventNames  { names = append(names, k) }
    return names
}()

"CHANGE_GROUP_DESCRIPTION": true,
"CHANGE_GROUP_NAME": true,
"CHANGE_GROUP_EMAIL": true,
"ADD_GROUP_MEMBER": true,
"UPDATE_GROUP_MEMBER": true,
"DELETE_GROUP": true,
}

type cacheEntry struct {
Id string
DisplayName string
}

type cacheMap map[string]cacheEntry

// adminActivitiesPageSize is the number of activity items requested per ListActivities
// call. The Google Reports API maximum is 1000.
const adminActivitiesPageSize = 1000

type adminEventFeed struct {
client *gwclient.GoogleWorkspaceClient

Expand All @@ -43,68 +81,78 @@ type adminEventFeed struct {
func (f *adminEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Timestamp, pToken *pagination.StreamToken) ([]*v2.Event, *pagination.StreamState, annotations.Annotations, error) {
l := ctxzap.Extract(ctx)

var streamState *pagination.StreamState

cursor, err := unmarshalPageToken(pToken, startAt)
if err != nil {
return nil, nil, nil, fmt.Errorf("failed to unmarshal page token: %w", err)
}

r, err := f.client.ListActivities(ctx, "all", "admin", "", cursor.StartAt, cursor.NextPageToken, int64(pToken.Size))
if err != nil {
return nil, nil, nil, fmt.Errorf("google-workspace: failed to list admin activities: %w", err)
}

latestEvent, err := time.Parse(time.RFC3339, cursor.LatestEventSeen)
if err != nil {
return nil, nil, nil, fmt.Errorf("failed to parse latest event time in admin event feed: %w", err)
}

events := make([]*v2.Event, 0)
for _, activity := range r.Items {
occurredAt := convertIdTimeToTimestamp(activity.Id.Time)
if occurredAt == nil {
// Set occurred at to epoch so that it should never be after the latest event
// Unless latest event is before epoch for some reason
occurredAt = timestamppb.New(time.Unix(0, 0))
// On the first call EventPageTokens is nil — fetch every event name from StartAt.
// On continuation calls fetch only names that still have a next page token.
fetchNames := adminEventNames
if len(cursor.EventPageTokens) > 0 {
fetchNames = make([]string, 0, len(cursor.EventPageTokens))
for name := range cursor.EventPageTokens {
fetchNames = append(fetchNames, name)
}
if occurredAt.AsTime().After(latestEvent) {
cursor.LatestEventSeen = occurredAt.AsTime().Format(time.RFC3339)
latestEvent = occurredAt.AsTime()
}

events := make([]*v2.Event, 0)
nextTokens := make(map[string]string)

for _, eventName := range fetchNames {
isGroupEvent := adminGroupEventNames[eventName]
r, err := f.client.ListActivities(ctx, "all", "admin", eventName, cursor.StartAt, cursor.EventPageTokens[eventName], adminActivitiesPageSize)
if err != nil {
return nil, nil, nil, fmt.Errorf("google-workspace: failed to list admin activities for %s: %w", eventName, err)
}
// There can be multiple events, have not found an example of this yet
for _, evt := range activity.Events {
switch evt.Type {
case "GROUP_SETTINGS":
changeEvents, err := f.handleGroupEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt)
if err != nil {
l.Error("failed to handle group event", zap.Error(err))

for _, activity := range r.Items {
occurredAt := convertIdTimeToTimestamp(activity.Id.Time)
if occurredAt == nil {
occurredAt = timestamppb.New(time.Unix(0, 0))
}
if occurredAt.AsTime().After(latestEvent) {
cursor.LatestEventSeen = occurredAt.AsTime().Format(time.RFC3339)
latestEvent = occurredAt.AsTime()
}
for _, evt := range activity.Events {
Comment thread
JavierCarnelli-ConductorOne marked this conversation as resolved.
if evt.Name != eventName {
continue
}
events = append(events, changeEvents...)
case "USER_SETTINGS":
changeEvents, err := f.handleUserEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt)
if err != nil {
l.Error("failed to handle user event", zap.Error(err))
var changeEvents []*v2.Event
var evtErr error
if isGroupEvent {
changeEvents, evtErr = f.handleGroupEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt)
} else {
changeEvents, evtErr = f.handleUserEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt)
}
if evtErr != nil {
l.Error("failed to handle admin event", zap.String("event_name", evt.Name), zap.Error(evtErr))
continue
}
events = append(events, changeEvents...)
default:
l.Debug("google-workspace-event-feed: skipping event", zap.String("event", evt.Name), zap.String("type", evt.Type))
continue
}
}

if r.NextPageToken != "" {
nextTokens[eventName] = r.NextPageToken
}
}

l.Debug("google-workspace-event-feed: listed events",
zap.Int("count", len(r.Items)),
zap.String("next_page_token", r.NextPageToken),
zap.Any("start_at", startAt),
zap.Any("latest_event", cursor.LatestEventSeen),
l.Debug("google-workspace-event-feed: listed admin events",
zap.Int("event_names_fetched", len(fetchNames)),
zap.Int("events_produced", len(events)),
zap.String("latest_event", cursor.LatestEventSeen),
)

cursor.NextPageToken = r.NextPageToken
if r.NextPageToken == "" {
hasMore := len(nextTokens) > 0
cursor.EventPageTokens = nextTokens
if !hasMore {
cursor.StartAt = cursor.LatestEventSeen
cursor.LatestEventSeen = ""
}
Expand All @@ -113,12 +161,11 @@ func (f *adminEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Ti
if err != nil {
return nil, nil, nil, fmt.Errorf("failed to marshal cursor token in admin event feed: %w", err)
}
streamState = &pagination.StreamState{
Cursor: cursorToken,
HasMore: r.NextPageToken != "",
}

return events, streamState, nil, nil
return events, &pagination.StreamState{
Cursor: cursorToken,
HasMore: hasMore,
}, nil, nil
}

func (f *adminEventFeed) handleGroupEvent(ctx context.Context, uniqueQualifier int64, occurredAt *timestamppb.Timestamp, activityEvt *reports.ActivityEvents) ([]*v2.Event, error) {

@luisina-santos luisina-santos Jun 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Missing REMOVE_GROUP_MEMBER → DeleteGrantEvent

ADD_GROUP_MEMBER produces a CreateGrantEvent, but there's no corresponding DeleteGrantEvent for membership removals. The Google Reports API does emit REMOVE_GROUP_MEMBER events. Without handling it, the feed will never surface when a user is removed from a group.

This gap is in the handleGroupEvent switch body below this line.

@luisina-santos luisina-santos Jun 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(low) Silent nil drop on missing NEW_VALUE in CHANGE_GROUP_EMAIL

In the CHANGE_GROUP_EMAIL case inside handleGroupEvent (below this line), the second newGroupChangedEvent call uses "NEW_VALUE" as the parameter name. If that parameter is absent from the activity, the function returns (nil, nil) and the event is silently skipped with no log line. A l.Debug(...) on that path would make it much easier to spot malformed activities during debugging.

evt, err = f.newGroupChangedEvent(ctx, uniqueQualifier, occurredAt, "NEW_VALUE", activityEvent)
if evt == nil {
    l.Debug("google-workspace-event-feed: CHANGE_GROUP_EMAIL missing NEW_VALUE parameter, skipping")
    return events, nil
}

Expand Down
74 changes: 45 additions & 29 deletions pkg/connector/admin_event_feed_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,18 @@ type safeUserResponse struct {
}

// Minimal fake for Reports Activities.List + Directory lookups used by admin_event_feed.
func newAdminFeedTestServer(users map[string]*directoryAdmin.User, groups map[string]*directoryAdmin.Group, activities *reportsAdmin.Activities) *httptest.Server {
// activitiesByEvent maps eventName query parameter values to the Activities response
// to return. An unrecognised eventName returns an empty Activities response.
func newAdminFeedTestServer(users map[string]*directoryAdmin.User, groups map[string]*directoryAdmin.Group, activitiesByEvent map[string]*reportsAdmin.Activities) *httptest.Server {
mux := http.NewServeMux()

mux.HandleFunc("/admin/reports/v1/activity/users/all/applications/admin", func(w http.ResponseWriter, r *http.Request) {
// ignore query parsing beyond pageToken/startTime for now
_ = json.NewEncoder(w).Encode(activities)
eventName := r.URL.Query().Get("eventName")
resp, ok := activitiesByEvent[eventName]
if !ok {
resp = &reportsAdmin.Activities{}
}
_ = json.NewEncoder(w).Encode(resp)
})

mux.HandleFunc("/admin/directory/v1/users/", func(w http.ResponseWriter, r *http.Request) {
Expand Down Expand Up @@ -75,40 +81,50 @@ func TestAdminEventFeed_GroupAndUserEvents(t *testing.T) {
"group@example.com": {Id: "group-1", Name: "Group One", Email: "group@example.com"},
}

// Build Activities with admin events we handle
// Build per-event-name Activities responses. The feed now issues one
// ListActivities call per event name, so the mock routes by eventName.
now := time.Now().UTC().Format(time.RFC3339)
acts := &reportsAdmin.Activities{
Items: []*reportsAdmin.Activity{
{
actsByEvent := map[string]*reportsAdmin.Activities{
"CHANGE_GROUP_NAME": {
Items: []*reportsAdmin.Activity{{
Id: &reportsAdmin.ActivityId{Time: now, UniqueQualifier: 123},
Events: []*reportsAdmin.ActivityEvents{
{
Type: "GROUP_SETTINGS",
Name: "CHANGE_GROUP_NAME",
Parameters: []*reportsAdmin.ActivityEventsParameters{
{Name: "GROUP_EMAIL", Value: "group@example.com"},
},
Events: []*reportsAdmin.ActivityEvents{{
Type: "GROUP_SETTINGS",
Name: "CHANGE_GROUP_NAME",
Parameters: []*reportsAdmin.ActivityEventsParameters{
{Name: "GROUP_EMAIL", Value: "group@example.com"},
},
{
Type: "GROUP_SETTINGS", Name: "ADD_GROUP_MEMBER",
Parameters: []*reportsAdmin.ActivityEventsParameters{
{Name: "GROUP_EMAIL", Value: "group@example.com"},
{Name: "USER_EMAIL", Value: "user@example.com"},
},
}},
}},
},
"ADD_GROUP_MEMBER": {
Items: []*reportsAdmin.Activity{{
Id: &reportsAdmin.ActivityId{Time: now, UniqueQualifier: 124},
Events: []*reportsAdmin.ActivityEvents{{
Type: "GROUP_SETTINGS",
Name: "ADD_GROUP_MEMBER",
Parameters: []*reportsAdmin.ActivityEventsParameters{
{Name: "GROUP_EMAIL", Value: "group@example.com"},
{Name: "USER_EMAIL", Value: "user@example.com"},
},
},
},
{
}},
}},
},
"CHANGE_FIRST_NAME": {
Items: []*reportsAdmin.Activity{{
Id: &reportsAdmin.ActivityId{Time: now, UniqueQualifier: 456},
Events: []*reportsAdmin.ActivityEvents{
{Type: "USER_SETTINGS", Name: "CHANGE_FIRST_NAME", Parameters: []*reportsAdmin.ActivityEventsParameters{{Name: "USER_EMAIL", Value: "user@example.com"}}},
},
},
Events: []*reportsAdmin.ActivityEvents{{
Type: "USER_SETTINGS",
Name: "CHANGE_FIRST_NAME",
Parameters: []*reportsAdmin.ActivityEventsParameters{
{Name: "USER_EMAIL", Value: "user@example.com"},
},
}},
}},
},
NextPageToken: "",
}

server := newAdminFeedTestServer(users, groups, acts)
server := newAdminFeedTestServer(users, groups, actsByEvent)
defer server.Close()

dir := newTestDirectoryService(t, server.URL, server.Client())
Expand Down
17 changes: 12 additions & 5 deletions pkg/connector/usage_event_feed.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,10 @@ import (

var privateAppIDRegex = regexp.MustCompile("[0-9]{21}")

// usageActivitiesPageSize is the number of activity items requested per ListActivities
// call. The Google Reports API maximum is 1000.
const usageActivitiesPageSize = 1000

// maxEventFeedLookback caps how far back event feeds query the Google Reports API.
// Google page tokens expire after ~24h, so a cursor left mid-pagination (e.g. after
// a connector restart or a transient timeout) would otherwise keep requesting the
Expand Down Expand Up @@ -81,10 +85,13 @@ func hasParameter(name string, parameters []*reportsAdmin.ActivityEventsParamete
}

type pageToken struct {
LatestEventSeen string `json:"latest_event_seen,omitempty"`
NextPageToken string `json:"next_page_token,omitempty"`
StartAt string `json:"start_at,omitempty"`
PageSize int `json:"page_size,omitempty"`
LatestEventSeen string `json:"latest_event_seen,omitempty"`
NextPageToken string `json:"next_page_token,omitempty"`
StartAt string `json:"start_at,omitempty"`
PageSize int `json:"page_size,omitempty"`
// EventPageTokens holds per-event-name pagination cursors for feeds that
// issue one ListActivities request per event name (e.g. adminEventFeed).
EventPageTokens map[string]string `json:"event_page_tokens,omitempty"`

@mateoHernandez123 mateoHernandez123 Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The stale-cursor reset in unmarshalPageToken (~L127-129) clears NextPageToken and LatestEventSeen but not this new EventPageTokens map. When a cursor goes stale, its per-event-name Google page tokens are already expired, so the admin feed then fetches only the stale subset of names and re-sends expired tokens → the Reports API returns an invalid-pageToken error → C1 retries the same cursor → death spiral, defeating the reset added in #108 that sits right next to it.

Clear it alongside the others: add pt.EventPageTokens = nil in the reset branch. The adjacent pt.NextPageToken = "" / pt.LatestEventSeen = "" show the intended contract; Google page tokens expire well before the 90-day StartAt cutoff — https://developers.google.com/admin-sdk/reports/reference/rest/v1/activities/list

}

func unmarshalPageToken(token *pagination.StreamToken, defaultStart *timestamppb.Timestamp) (*pageToken, error) {
Expand Down Expand Up @@ -151,7 +158,7 @@ func (f *usageEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Ti
return nil, nil, nil, fmt.Errorf("failed to unmarshal page token in usage event feed: %w", err)
}

r, err := f.c.ListActivities(ctx, "all", "token", "authorize", cursor.StartAt, cursor.NextPageToken, int64(pToken.Size))
r, err := f.c.ListActivities(ctx, "all", "token", "authorize", cursor.StartAt, cursor.NextPageToken, usageActivitiesPageSize)
if err != nil {
return nil, nil, nil, fmt.Errorf("google-workspace: failed to list token activities: %w", err)
}
Expand Down
Loading