-
Notifications
You must be signed in to change notification settings - Fork 1.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #9013 from hashicorp/event-stream
Event stream
- Loading branch information
Showing
66 changed files
with
5,729 additions
and
344 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,98 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| ) | ||
|
|
||
| // Events is a set of events for a corresponding index. Events returned for the | ||
| // index depend on which topics are subscribed to when a request is made. | ||
| type Events struct { | ||
| Index uint64 | ||
| Events []Event | ||
| Err error | ||
| } | ||
|
|
||
| // Topic is an event Topic | ||
| type Topic string | ||
|
|
||
| // Event holds information related to an event that occurred in Nomad. | ||
| // The Payload is a hydrated object related to the Topic | ||
| type Event struct { | ||
| Topic Topic | ||
| Type string | ||
| Key string | ||
| FilterKeys []string | ||
| Index uint64 | ||
| Payload map[string]interface{} | ||
| } | ||
|
|
||
| // IsHeartbeat specifies if the event is an empty heartbeat used to | ||
| // keep a connection alive. | ||
| func (e *Events) IsHeartbeat() bool { | ||
| return e.Index == 0 && len(e.Events) == 0 | ||
| } | ||
|
|
||
| // EventStream is used to stream events from Nomad | ||
| type EventStream struct { | ||
| client *Client | ||
| } | ||
|
|
||
| // EventStream returns a handle to the Events endpoint | ||
| func (c *Client) EventStream() *EventStream { | ||
| return &EventStream{client: c} | ||
| } | ||
|
|
||
| // Stream establishes a new subscription to Nomad's event stream and streams | ||
| // results back to the returned channel. | ||
| func (e *EventStream) Stream(ctx context.Context, topics map[Topic][]string, index uint64, q *QueryOptions) (<-chan *Events, error) { | ||
| r, err := e.client.newRequest("GET", "/v1/event/stream") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| q = q.WithContext(ctx) | ||
| r.setQueryOptions(q) | ||
|
|
||
| // Build topic query params | ||
| for topic, keys := range topics { | ||
| for _, k := range keys { | ||
| r.params.Add("topic", fmt.Sprintf("%s:%s", topic, k)) | ||
| } | ||
| } | ||
|
|
||
| _, resp, err := requireOK(e.client.doRequest(r)) | ||
|
|
||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| eventsCh := make(chan *Events, 10) | ||
| go func() { | ||
| defer resp.Body.Close() | ||
| defer close(eventsCh) | ||
|
|
||
| dec := json.NewDecoder(resp.Body) | ||
|
|
||
| for ctx.Err() == nil { | ||
| // Decode next newline delimited json of events | ||
| var events Events | ||
| if err := dec.Decode(&events); err != nil { | ||
| // set error and fallthrough to | ||
| // select eventsCh | ||
| events = Events{Err: err} | ||
| } | ||
| if events.Err == nil && events.IsHeartbeat() { | ||
| continue | ||
| } | ||
|
|
||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case eventsCh <- &events: | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| return eventsCh, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,113 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestEvent_Stream(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| c, s := makeClient(t, nil, nil) | ||
| defer s.Stop() | ||
|
|
||
| // register job to generate events | ||
| jobs := c.Jobs() | ||
| job := testJob() | ||
| resp2, _, err := jobs.Register(job, nil) | ||
| require.Nil(t, err) | ||
| require.NotNil(t, resp2) | ||
|
|
||
| // build event stream request | ||
| events := c.EventStream() | ||
| q := &QueryOptions{} | ||
| topics := map[Topic][]string{ | ||
| "Eval": {"*"}, | ||
| } | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
|
|
||
| streamCh, err := events.Stream(ctx, topics, 0, q) | ||
| require.NoError(t, err) | ||
|
|
||
| select { | ||
| case event := <-streamCh: | ||
| if event.Err != nil { | ||
| require.Fail(t, err.Error()) | ||
| } | ||
| require.Equal(t, len(event.Events), 1) | ||
| require.Equal(t, "Eval", string(event.Events[0].Topic)) | ||
| case <-time.After(5 * time.Second): | ||
| require.Fail(t, "failed waiting for event stream event") | ||
| } | ||
| } | ||
|
|
||
| func TestEvent_Stream_Err_InvalidQueryParam(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| c, s := makeClient(t, nil, nil) | ||
| defer s.Stop() | ||
|
|
||
| // register job to generate events | ||
| jobs := c.Jobs() | ||
| job := testJob() | ||
| resp2, _, err := jobs.Register(job, nil) | ||
| require.Nil(t, err) | ||
| require.NotNil(t, resp2) | ||
|
|
||
| // build event stream request | ||
| events := c.EventStream() | ||
| q := &QueryOptions{} | ||
| topics := map[Topic][]string{ | ||
| "Eval": {"::*"}, | ||
| } | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
|
|
||
| _, err = events.Stream(ctx, topics, 0, q) | ||
| require.Error(t, err) | ||
| require.Contains(t, err.Error(), "400") | ||
| require.Contains(t, err.Error(), "Invalid key value pair") | ||
| } | ||
|
|
||
| func TestEvent_Stream_CloseCtx(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| c, s := makeClient(t, nil, nil) | ||
| defer s.Stop() | ||
|
|
||
| // register job to generate events | ||
| jobs := c.Jobs() | ||
| job := testJob() | ||
| resp2, _, err := jobs.Register(job, nil) | ||
| require.Nil(t, err) | ||
| require.NotNil(t, resp2) | ||
|
|
||
| // build event stream request | ||
| events := c.EventStream() | ||
| q := &QueryOptions{} | ||
| topics := map[Topic][]string{ | ||
| "Eval": {"*"}, | ||
| } | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
|
|
||
| streamCh, err := events.Stream(ctx, topics, 0, q) | ||
| require.NoError(t, err) | ||
|
|
||
| // cancel the request | ||
| cancel() | ||
|
|
||
| select { | ||
| case event, ok := <-streamCh: | ||
| require.False(t, ok) | ||
| require.Nil(t, event) | ||
| case <-time.After(5 * time.Second): | ||
| require.Fail(t, "failed waiting for event stream event") | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.