Added event feeds to the baton SDK #92
Conversation
loganintech
changed the title
pkg/tasks/local/revoker.go
Outdated
| @@ -44,7 +44,7 @@ func (m *localRevoker) Process(ctx context.Context, task *v1.Task, cc types.Conn | |||
| return nil | |||
| } | |||
|
|
|||
| // NewGranter returns a task manager that queues a sync task. | |||
| // NewRevoker returns a task manager that queues a sync task. | |||
There was a problem hiding this comment.
Probably should fix the rest of the comment while we're at it.
There was a problem hiding this comment.
Good progress here. The main thing missing at this point is a way to persist the events, and more importantly how/where we persist the next page token.
My hunch is that we'll want to update the syncer package to have an explicit SyncEvents step that isn't called as part of a normal sync, but allows you to store the events in the c1z, as well as the next page/last token.
If not a part of the syncer, I think we'll want to still come up with a place for that logic to live.
pkg/tasks/local/event_feed.go
Outdated
| fmt.Println(string(bytes)) | ||
| } | ||
| pageToken = resp.GetNextPageToken() | ||
| if pageToken == "" { |
There was a problem hiding this comment.
I wonder if the semantics of a page token are exactly what we want. Normally we loop until the page token is empty, and then we know there isn't any data left. We may want to consider another field on the response to track the 'page token' to save for the next request.
Imagine we return 10 results per page. On the 3rd page, we only have 5 results, so we'd return an empty page token. But the connector also needs to provide a token to use for the next call to resume from after the 5th result that it returned. That way when the connector is woken up in an hour, it can start off where it was at last more easily.
There was a problem hiding this comment.
To sum that up: since it is a feed, there will always be a next page token, so we can't use the same value to know that we need to continue making more requests. Maybe it is a bool on the response message?
There was a problem hiding this comment.
Well, it gets worse. Feeds are assumed to log data from oldest to newest, but in the two connectors that we are planning on implementing this for first, they return data from newest to oldest with no way of inverting it. We got around this in another connector by looping over all events it was able to find. If we hit a pageSize cutoff, it would return the next page token that the service needs and use that as the page token. If not, we returned the most recent timestamp and put a filter to only get objects newer than that timestamp. But this doesn't really work either because events in a service may come in out of order. We either have to live with that to a certain extend while minimizing our risk for it, or sync everything, or provide some ID tracking mechanism that says "well get everything after this day - a few and only return net new IDs." Lots of ideas on how to tackle this one. This should definitely be solved by the SDK to make connector implementation easier.
pkg/tasks/c1api/event_feed.go
Outdated
| @@ -0,0 +1,43 @@ | |||
| package c1api | |||
|
|
|||
| // | |||
pkg/tasks/c1api/manager.go
Outdated
| @@ -219,6 +219,9 @@ func (c *c1ApiTaskManager) Process(ctx context.Context, task *v1.Task, cc types. | |||
| case tasks.RevokeType: | |||
| handler = newRevokeTaskHandler(task, tHelpers) | |||
|
|
|||
| // case tasks.EventFeedType: | |||
| @@ -42,6 +47,7 @@ type builderImpl struct { | |||
| resourceBuilders map[string]ResourceSyncer | |||
| resourceProvisioners map[string]ResourceProvisioner | |||
| resourceProvisionersV2 map[string]ResourceProvisionerV2 | |||
| eventFeed EventProvider | |||
There was a problem hiding this comment.
should this be map[string]EventProvider, then we could have potentially multiple streams, one per resource type?
There was a problem hiding this comment.
Hmm, maybe it is just []EventProvider? I don't know that they are tied to a resource type specifically, but the idea of supporting multiple feeds makes sense to me.
pkg/tasks/local/event_feed.go
Outdated
| var pageToken string | ||
| for { | ||
| resp, err := cc.ListEvents(ctx, &v2.ListEventsRequest{ | ||
| PageSize: 100, |
There was a problem hiding this comment.
We should maybe make this a const, even think about increasing it. Event feeds tend to have a larger volume of events than the items that give 100 like we're used to.
loganintech
force-pushed
the
anthony/logan/event-feeds
branch
from
9ea209d to
488213e
Compare
| @@ -216,6 +216,9 @@ type rotateCredentialsConfig struct { | |||
| resourceType string | |||
| } | |||
|
|
|||
| type eventStreamConfig struct { | |||
There was a problem hiding this comment.
Just placeholder for when we might eventually need a config?
There was a problem hiding this comment.
Yep, and follows the same format as the other runners.
No description provided.