2018-05-22 21:43:36 +02:00
|
|
|
// Copyright 2016-2018, Pulumi Corporation.
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
2018-03-22 18:42:43 +01:00
|
|
|
|
2018-09-04 21:38:58 +02:00
|
|
|
package httpstate
|
2018-03-22 18:42:43 +01:00
|
|
|
|
|
|
|
import (
|
2018-05-08 03:23:03 +02:00
|
|
|
"context"
|
2018-03-22 18:42:43 +01:00
|
|
|
"fmt"
|
2018-12-21 00:34:30 +01:00
|
|
|
"sync"
|
2018-03-22 18:42:43 +01:00
|
|
|
"time"
|
|
|
|
|
2019-01-15 03:19:59 +01:00
|
|
|
"github.com/pulumi/pulumi/pkg/diag"
|
2020-03-18 22:40:07 +01:00
|
|
|
"github.com/pulumi/pulumi/sdk/go/common/util/contract"
|
2019-07-31 20:23:33 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/util/logging"
|
2018-11-09 19:01:29 +01:00
|
|
|
|
2018-05-25 22:29:59 +02:00
|
|
|
"github.com/pkg/errors"
|
2020-03-18 23:00:30 +01:00
|
|
|
"github.com/pulumi/pulumi/sdk/go/common/apitype"
|
2018-03-22 18:42:43 +01:00
|
|
|
"github.com/pulumi/pulumi/pkg/backend"
|
2018-09-05 00:40:15 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/backend/display"
|
2018-09-04 21:38:58 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/backend/httpstate/client"
|
2018-03-22 18:42:43 +01:00
|
|
|
"github.com/pulumi/pulumi/pkg/engine"
|
2019-04-19 00:57:54 +02:00
|
|
|
"github.com/pulumi/pulumi/pkg/resource/config"
|
2018-03-22 18:42:43 +01:00
|
|
|
"github.com/pulumi/pulumi/pkg/resource/deploy"
|
|
|
|
"github.com/pulumi/pulumi/pkg/resource/stack"
|
2020-03-18 22:35:53 +01:00
|
|
|
"github.com/pulumi/pulumi/sdk/go/common/workspace"
|
2018-03-22 18:42:43 +01:00
|
|
|
)
|
|
|
|
|
|
|
|
type tokenRequest chan<- tokenResponse
|
|
|
|
|
|
|
|
type tokenResponse struct {
|
|
|
|
token string
|
|
|
|
err error
|
|
|
|
}
|
|
|
|
|
|
|
|
// tokenSource is a helper type that manages the renewal of the lease token for a managed update.
|
|
|
|
type tokenSource struct {
|
|
|
|
requests chan tokenRequest
|
2018-05-09 00:28:42 +02:00
|
|
|
done chan bool
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
|
2018-05-08 03:23:03 +02:00
|
|
|
func newTokenSource(ctx context.Context, token string, backend *cloudBackend, update client.UpdateIdentifier,
|
2018-03-22 18:42:43 +01:00
|
|
|
duration time.Duration) (*tokenSource, error) {
|
|
|
|
|
|
|
|
// Perform an initial lease renewal.
|
2018-05-08 03:23:03 +02:00
|
|
|
newToken, err := backend.client.RenewUpdateLease(ctx, update, token, duration)
|
2018-03-22 18:42:43 +01:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2018-05-09 00:28:42 +02:00
|
|
|
requests, done := make(chan tokenRequest), make(chan bool)
|
2018-03-22 18:42:43 +01:00
|
|
|
go func() {
|
|
|
|
// We will renew the lease after 50% of the duration has elapsed to allow more time for retries.
|
|
|
|
ticker := time.NewTicker(duration / 2)
|
|
|
|
defer ticker.Stop()
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-ticker.C:
|
2018-05-08 03:23:03 +02:00
|
|
|
newToken, err = backend.client.RenewUpdateLease(ctx, update, token, duration)
|
2018-03-22 18:42:43 +01:00
|
|
|
if err != nil {
|
|
|
|
ticker.Stop()
|
|
|
|
} else {
|
|
|
|
token = newToken
|
|
|
|
}
|
|
|
|
|
|
|
|
case c, ok := <-requests:
|
|
|
|
if !ok {
|
2018-05-09 00:28:42 +02:00
|
|
|
close(done)
|
2018-03-22 18:42:43 +01:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
resp := tokenResponse{err: err}
|
|
|
|
if err == nil {
|
|
|
|
resp.token = token
|
|
|
|
}
|
|
|
|
c <- resp
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
2018-05-09 00:28:42 +02:00
|
|
|
return &tokenSource{requests: requests, done: done}, nil
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
func (ts *tokenSource) Close() {
|
|
|
|
close(ts.requests)
|
2018-05-09 00:28:42 +02:00
|
|
|
<-ts.done
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
func (ts *tokenSource) GetToken() (string, error) {
|
|
|
|
ch := make(chan tokenResponse)
|
|
|
|
ts.requests <- ch
|
|
|
|
resp := <-ch
|
|
|
|
return resp.token, resp.err
|
|
|
|
}
|
|
|
|
|
2019-04-30 20:48:41 +02:00
|
|
|
type cloudQuery struct {
|
2019-08-12 09:22:42 +02:00
|
|
|
root string
|
|
|
|
proj *workspace.Project
|
2019-04-30 20:48:41 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
func (q *cloudQuery) GetRoot() string {
|
|
|
|
return q.root
|
|
|
|
}
|
|
|
|
|
|
|
|
func (q *cloudQuery) GetProject() *workspace.Project {
|
|
|
|
return q.proj
|
|
|
|
}
|
|
|
|
|
2018-03-22 18:42:43 +01:00
|
|
|
// cloudUpdate is an implementation of engine.Update backed by remote state and a local program.
|
|
|
|
type cloudUpdate struct {
|
2018-05-08 03:23:03 +02:00
|
|
|
context context.Context
|
2018-03-22 18:42:43 +01:00
|
|
|
backend *cloudBackend
|
|
|
|
|
|
|
|
update client.UpdateIdentifier
|
|
|
|
tokenSource *tokenSource
|
|
|
|
|
2018-04-17 08:04:56 +02:00
|
|
|
root string
|
|
|
|
proj *workspace.Project
|
|
|
|
target *deploy.Target
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
func (u *cloudUpdate) GetRoot() string {
|
|
|
|
return u.root
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *cloudUpdate) GetProject() *workspace.Project {
|
|
|
|
return u.proj
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *cloudUpdate) GetTarget() *deploy.Target {
|
|
|
|
return u.target
|
|
|
|
}
|
|
|
|
|
|
|
|
func (u *cloudUpdate) Complete(status apitype.UpdateStatus) error {
|
|
|
|
defer u.tokenSource.Close()
|
|
|
|
|
|
|
|
token, err := u.tokenSource.GetToken()
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2018-05-08 03:23:03 +02:00
|
|
|
return u.backend.client.CompleteUpdate(u.context, u.update, status, token)
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
// recordEngineEvents will record the events with the Pulumi Service, enabling things like viewing
|
2018-12-21 00:34:30 +01:00
|
|
|
// the update logs or drilling into the timeline of an update.
|
2019-06-28 18:40:21 +02:00
|
|
|
func (u *cloudUpdate) recordEngineEvents(startingSeqNumber int, events []engine.Event) error {
|
2018-11-09 19:01:29 +01:00
|
|
|
contract.Assert(u.tokenSource != nil)
|
|
|
|
token, err := u.tokenSource.GetToken()
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2018-03-31 21:08:48 +02:00
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
var apiEvents apitype.EngineEventBatch
|
|
|
|
for idx, event := range events {
|
2019-09-07 02:07:54 +02:00
|
|
|
apiEvent, convErr := display.ConvertEngineEvent(event)
|
2019-06-28 18:40:21 +02:00
|
|
|
if convErr != nil {
|
|
|
|
return errors.Wrap(convErr, "converting engine event")
|
|
|
|
}
|
2018-12-21 00:34:30 +01:00
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
// Each event within an update must have a unique sequence number. Any request to
|
|
|
|
// emit an update with the same sequence number will fail. (Read: the caller needs
|
|
|
|
// to be accurate about this.)
|
|
|
|
apiEvent.Sequence = idx + startingSeqNumber
|
|
|
|
apiEvent.Timestamp = int(time.Now().Unix())
|
2018-12-21 00:34:30 +01:00
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
apiEvents.Events = append(apiEvents.Events, apiEvent)
|
|
|
|
}
|
2018-03-22 18:42:43 +01:00
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
return u.backend.client.RecordEngineEvents(u.context, u.update, apiEvents, token)
|
2019-01-15 03:19:59 +01:00
|
|
|
}
|
|
|
|
|
2018-11-09 19:01:29 +01:00
|
|
|
// RecordAndDisplayEvents inspects engine events from the given channel, and prints them to the CLI as well as
|
2019-05-22 21:22:40 +02:00
|
|
|
// posting them to the Pulumi service.
|
2018-10-30 23:42:33 +01:00
|
|
|
func (u *cloudUpdate) RecordAndDisplayEvents(
|
|
|
|
label string, action apitype.UpdateKind, stackRef backend.StackReference, op backend.UpdateOperation,
|
|
|
|
events <-chan engine.Event, done chan<- bool, opts display.Options, isPreview bool) {
|
2019-06-28 18:40:21 +02:00
|
|
|
// We take the channel of engine events and pass them to separate components that will display
|
|
|
|
// them to the console or persist them on the Pulumi Service. Both should terminate as soon as
|
|
|
|
// they see a CancelEvent, and when finished, close the "done" channel.
|
|
|
|
displayEvents := make(chan engine.Event) // Note: unbuffered, but we assume it won't matter in practice.
|
|
|
|
displayEventsDone := make(chan bool)
|
|
|
|
|
|
|
|
persistEvents := make(chan engine.Event, 100)
|
|
|
|
persistEventsDone := make(chan bool)
|
2018-03-22 18:42:43 +01:00
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
// We close our own done channel when both of the dependent components have finished.
|
2018-12-14 04:58:26 +01:00
|
|
|
defer func() {
|
2019-06-28 18:40:21 +02:00
|
|
|
<-displayEventsDone
|
|
|
|
<-persistEventsDone
|
2018-12-14 04:58:26 +01:00
|
|
|
close(done)
|
|
|
|
}()
|
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
// Start the Go-routines for displaying and persisting events.
|
|
|
|
go display.ShowEvents(
|
|
|
|
label, action, stackRef.Name(), op.Proj.Name,
|
|
|
|
displayEvents, displayEventsDone, opts, isPreview)
|
|
|
|
go persistEngineEvents(
|
|
|
|
u, opts.Debug, /* persist debug events */
|
|
|
|
persistEvents, persistEventsDone)
|
2018-12-21 00:34:30 +01:00
|
|
|
|
|
|
|
for e := range events {
|
2018-03-22 18:42:43 +01:00
|
|
|
displayEvents <- e
|
2019-06-28 18:40:21 +02:00
|
|
|
persistEvents <- e
|
2018-03-22 18:42:43 +01:00
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
// We stop reading from the event stream as soon as we see the CancelEvent,
|
|
|
|
// which will also signal the display/persist components to shutdown too.
|
2018-03-22 18:42:43 +01:00
|
|
|
if e.Type == engine.CancelEvent {
|
2018-12-24 19:23:32 +01:00
|
|
|
break
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
}
|
2018-12-21 00:34:30 +01:00
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
// Note that we don't return immediately, the defer'd function will block until
|
|
|
|
// the display and persistence go-routines are finished processing events.
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
|
2019-08-12 09:22:42 +02:00
|
|
|
func (b *cloudBackend) newQuery(ctx context.Context,
|
2019-10-29 23:19:44 +01:00
|
|
|
op backend.QueryOperation) (engine.QueryInfo, error) {
|
2019-04-30 20:48:41 +02:00
|
|
|
|
2019-08-12 09:22:42 +02:00
|
|
|
return &cloudQuery{root: op.Root, proj: op.Proj}, nil
|
2019-04-30 20:48:41 +02:00
|
|
|
}
|
|
|
|
|
2019-04-19 00:57:54 +02:00
|
|
|
func (b *cloudBackend) newUpdate(ctx context.Context, stackRef backend.StackReference, op backend.UpdateOperation,
|
|
|
|
update client.UpdateIdentifier, token string) (*cloudUpdate, error) {
|
2018-03-22 18:42:43 +01:00
|
|
|
|
|
|
|
// Create a token source for this update if necessary.
|
|
|
|
var tokenSource *tokenSource
|
|
|
|
if token != "" {
|
2018-05-08 03:23:03 +02:00
|
|
|
ts, err := newTokenSource(ctx, token, b, update, 5*time.Minute)
|
2018-03-22 18:42:43 +01:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
tokenSource = ts
|
|
|
|
}
|
|
|
|
|
|
|
|
// Construct the deployment target.
|
2019-04-19 00:57:54 +02:00
|
|
|
target, err := b.getTarget(ctx, stackRef, op.StackConfiguration.Config, op.StackConfiguration.Decrypter)
|
2018-03-22 18:42:43 +01:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Construct and return a new update.
|
2018-04-17 08:04:56 +02:00
|
|
|
return &cloudUpdate{
|
2018-05-08 03:23:03 +02:00
|
|
|
context: ctx,
|
2018-03-22 18:42:43 +01:00
|
|
|
backend: b,
|
|
|
|
update: update,
|
|
|
|
tokenSource: tokenSource,
|
2019-04-19 00:57:54 +02:00
|
|
|
root: op.Root,
|
|
|
|
proj: op.Proj,
|
2018-03-22 18:42:43 +01:00
|
|
|
target: target,
|
2018-04-17 08:04:56 +02:00
|
|
|
}, nil
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
|
2018-05-23 00:39:13 +02:00
|
|
|
func (b *cloudBackend) getSnapshot(ctx context.Context, stackRef backend.StackReference) (*deploy.Snapshot, error) {
|
2020-02-13 21:25:57 +01:00
|
|
|
untypedDeployment, err := b.exportDeployment(ctx, stackRef, nil /* get latest */)
|
2018-03-22 18:42:43 +01:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2018-05-25 22:29:59 +02:00
|
|
|
|
2019-08-01 19:33:52 +02:00
|
|
|
snapshot, err := stack.DeserializeUntypedDeployment(untypedDeployment, stack.DefaultSecretsProvider)
|
2018-03-22 18:42:43 +01:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2018-05-23 00:39:13 +02:00
|
|
|
return snapshot, nil
|
|
|
|
}
|
|
|
|
|
2019-04-19 00:57:54 +02:00
|
|
|
func (b *cloudBackend) getTarget(ctx context.Context, stackRef backend.StackReference,
|
|
|
|
cfg config.Map, dec config.Decrypter) (*deploy.Target, error) {
|
2019-08-12 09:22:42 +02:00
|
|
|
|
2018-05-23 00:39:13 +02:00
|
|
|
snapshot, err := b.getSnapshot(ctx, stackRef)
|
2018-03-22 18:42:43 +01:00
|
|
|
if err != nil {
|
2018-05-25 22:29:59 +02:00
|
|
|
switch err {
|
|
|
|
case stack.ErrDeploymentSchemaVersionTooOld:
|
|
|
|
return nil, fmt.Errorf("the stack '%s' is too old to be used by this version of the Pulumi CLI",
|
2018-09-05 16:20:25 +02:00
|
|
|
stackRef.Name())
|
2018-05-25 22:29:59 +02:00
|
|
|
case stack.ErrDeploymentSchemaVersionTooNew:
|
|
|
|
return nil, fmt.Errorf("the stack '%s' is newer than what this version of the Pulumi CLI understands. "+
|
2018-09-05 16:20:25 +02:00
|
|
|
"Please update your version of the Pulumi CLI", stackRef.Name())
|
2018-05-25 22:29:59 +02:00
|
|
|
default:
|
|
|
|
return nil, errors.Wrap(err, "could not deserialize deployment")
|
|
|
|
}
|
2018-03-22 18:42:43 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
return &deploy.Target{
|
2018-09-05 16:20:25 +02:00
|
|
|
Name: stackRef.Name(),
|
2019-04-19 00:57:54 +02:00
|
|
|
Config: cfg,
|
|
|
|
Decrypter: dec,
|
2018-03-22 18:42:43 +01:00
|
|
|
Snapshot: snapshot,
|
|
|
|
}, nil
|
|
|
|
}
|
2018-11-09 19:01:29 +01:00
|
|
|
|
2019-06-28 18:40:21 +02:00
|
|
|
func isDebugDiagEvent(e engine.Event) bool {
|
|
|
|
return e.Type == engine.DiagEvent && (e.Payload.(engine.DiagEventPayload)).Severity == diag.Debug
|
|
|
|
}
|
|
|
|
|
|
|
|
type engineEventBatch struct {
|
|
|
|
sequenceStart int
|
|
|
|
events []engine.Event
|
|
|
|
}
|
|
|
|
|
|
|
|
// persistEngineEvents reads from a channel of engine events and persists them on the
|
|
|
|
// Pulumi Service. This is the data that powers the logs display.
|
|
|
|
func persistEngineEvents(
|
|
|
|
update *cloudUpdate, persistDebugEvents bool,
|
|
|
|
events <-chan engine.Event, done chan<- bool) {
|
|
|
|
// A single update can emit hundreds, if not thousands, or tens of thousands of
|
|
|
|
// engine events. We transmit engine events in large batches to reduce the overhead
|
|
|
|
// associated with each HTTP request to the service. We also send multiple HTTP
|
|
|
|
// requests concurrently, as to not block processing subsequent engine events.
|
|
|
|
|
|
|
|
// Maximum number of events to batch up before transmitting.
|
|
|
|
const maxEventsToTransmit = 50
|
|
|
|
// Maximum wait time before sending all batched events.
|
|
|
|
const maxTransmissionDelay = 4 * time.Second
|
|
|
|
// Maximum number of concurrent requests to the Pulumi Service to persist
|
|
|
|
// engine events.
|
|
|
|
const maxConcurrentRequests = 3
|
|
|
|
|
|
|
|
// We don't want to indicate that we are done processing every engine event in the
|
|
|
|
// provided channel until every HTTP request has completed. We use a wait group to
|
|
|
|
// track all of those requests.
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
|
|
|
|
defer func() {
|
|
|
|
wg.Wait()
|
|
|
|
close(done)
|
|
|
|
}()
|
|
|
|
|
|
|
|
var eventBatch []engine.Event
|
|
|
|
maxDelayTicker := time.NewTicker(maxTransmissionDelay)
|
|
|
|
|
|
|
|
// We maintain a sequence counter for each event to ensure that the Pulumi Service can
|
|
|
|
// ensure events can be reconstructured in the same order they were emitted. (And not
|
|
|
|
// out of order from parallel writes and/or network delays.)
|
|
|
|
eventIdx := 0
|
|
|
|
|
|
|
|
// As we identify batches of engine events to transmit, we put them into a channel.
|
|
|
|
// This will allow us to issue HTTP requests concurrently, but also limit the maximum
|
|
|
|
// number of requests in-flight at any one time.
|
|
|
|
//
|
|
|
|
// This channel isn't buffered, so adding a new batch of events to persist will block
|
|
|
|
// until a go-routine is available to send the batch.
|
|
|
|
batchesToTransmit := make(chan engineEventBatch)
|
|
|
|
|
|
|
|
transmitBatchLoop := func() {
|
|
|
|
wg.Add(1)
|
|
|
|
defer wg.Done()
|
|
|
|
|
|
|
|
for eventBatch := range batchesToTransmit {
|
|
|
|
err := update.recordEngineEvents(eventBatch.sequenceStart, eventBatch.events)
|
|
|
|
if err != nil {
|
2019-07-31 20:23:33 +02:00
|
|
|
logging.V(3).Infof("error recording engine events: %s", err)
|
2019-06-28 18:40:21 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
// Start N different go-routines which will all pull from the batchesToTransmit channel
|
|
|
|
// and persist those engine events until the channel is closed.
|
|
|
|
for i := 0; i < maxConcurrentRequests; i++ {
|
|
|
|
go transmitBatchLoop()
|
|
|
|
}
|
|
|
|
|
|
|
|
// transmitBatch sends off the current batch of engine events (eventIdx, eventBatch) to the
|
|
|
|
// batchesToTransmit channel. Will mutate eventIdx, eventBatch as a side effect.
|
|
|
|
transmitBatch := func() {
|
|
|
|
if len(eventBatch) == 0 {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
batch := engineEventBatch{
|
|
|
|
sequenceStart: eventIdx,
|
|
|
|
events: eventBatch,
|
|
|
|
}
|
|
|
|
// This will block until one of the spawned go-routines is available to read the data.
|
|
|
|
// Effectively providing a global rate limit for how quickly we can send data to the
|
|
|
|
// Pulumi Service, if an update is particularly chatty.
|
|
|
|
batchesToTransmit <- batch
|
|
|
|
|
|
|
|
// With the values of eventIdx and eventBatch copied into engineEventBatch,
|
|
|
|
// we now modify their values for the next time transmitBatch is called.
|
|
|
|
eventIdx += len(eventBatch)
|
|
|
|
eventBatch = nil
|
|
|
|
}
|
|
|
|
|
|
|
|
var sawCancelEvent bool
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case e := <-events:
|
|
|
|
// Ignore debug events unless asked to.
|
|
|
|
if isDebugDiagEvent(e) && !persistDebugEvents {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
// Stop processing once we see the CancelEvent.
|
|
|
|
if e.Type == engine.CancelEvent {
|
|
|
|
sawCancelEvent = true
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
eventBatch = append(eventBatch, e)
|
|
|
|
if len(eventBatch) >= maxEventsToTransmit {
|
|
|
|
transmitBatch()
|
|
|
|
}
|
|
|
|
|
|
|
|
case <-maxDelayTicker.C:
|
|
|
|
// If the ticker has fired, send any batched events. This sets an upper bound for
|
|
|
|
// the delay between the event being observed and persisted.
|
|
|
|
transmitBatch()
|
|
|
|
}
|
|
|
|
|
|
|
|
if sawCancelEvent {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Transmit any lingering events.
|
|
|
|
transmitBatch()
|
|
|
|
// Closing the batchesToTransmit channel will signal the worker persistence routines to
|
|
|
|
// terminate, which will trigger the `wg` WaitGroup to be marked as complete, which will
|
|
|
|
// finally close the `done` channel so the caller knows we are finished processing the
|
|
|
|
// engine event stream.
|
|
|
|
close(batchesToTransmit)
|
|
|
|
}
|