Skip to content
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

Move queue management to dispatcher #1095

Closed
wants to merge 7 commits into from
Closed

Move queue management to dispatcher #1095

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
246e23a
Move queue management to dispatcher
michel-laterman Sep 6, 2022
4b9dab1
Fix linter and unit tests
michel-laterman Sep 6, 2022
4eecbc7
minor fixes
michel-laterman Sep 7, 2022
fa49871
Port changes from attempted v2 pr
michel-laterman Sep 9, 2022
984626e
Minor cleanup
michel-laterman Sep 9, 2022
d7b34de
Merge remote-tracking branch 'origin/main' into queue-in-dispatch
michel-laterman Sep 9, 2022
6a5aed7
change to copy
michel-laterman Sep 12, 2022
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
92 changes: 4 additions & 88 deletions internal/pkg/agent/application/gateway/fleet/fleet_gateway.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@ package fleet

import (
"context"
stderr "errors"
"fmt"
"sync"
"time"
Expand Down Expand Up @@ -61,14 +60,6 @@ type stateStore interface {
AckToken() string
SetAckToken(ackToken string)
Save() error
SetQueue([]fleetapi.Action)
Actions() []fleetapi.Action
}

type actionQueue interface {
Add(fleetapi.Action, int64)
DequeueActions() []fleetapi.Action
Cancel(string) int
Actions() []fleetapi.Action
}

Expand All @@ -89,7 +80,6 @@ type fleetGateway struct {
statusController status.Controller
statusReporter status.Reporter
stateStore stateStore
queue actionQueue
}

// New creates a new fleet gateway
Expand All @@ -102,7 +92,6 @@ func New(
acker store.FleetAcker,
statusController status.Controller,
stateStore stateStore,
queue actionQueue,
) (gateway.FleetGateway, error) {

scheduler := scheduler.NewPeriodicJitter(defaultGatewaySettings.Duration, defaultGatewaySettings.Jitter)
Expand All @@ -117,7 +106,6 @@ func New(
acker,
statusController,
stateStore,
queue,
)
}

Expand All @@ -132,7 +120,6 @@ func newFleetGatewayWithScheduler(
acker store.FleetAcker,
statusController status.Controller,
stateStore stateStore,
queue actionQueue,
) (gateway.FleetGateway, error) {

// Backoff implementation doesn't support the use of a context [cancellation]
Expand All @@ -158,14 +145,13 @@ func newFleetGatewayWithScheduler(
statusReporter: statusController.RegisterComponent("gateway"),
statusController: statusController,
stateStore: stateStore,
queue: queue,
}, nil
}

func (f *fleetGateway) worker() {
for {
select {
case ts := <-f.scheduler.WaitTick():
case <-f.scheduler.WaitTick():
f.log.Debug("FleetGateway calling Checkin API")

// Execute the checkin call and for any errors returned by the fleet-server API
Expand All @@ -175,28 +161,12 @@ func (f *fleetGateway) worker() {
if err != nil {
continue
}

actions := f.queueScheduledActions(resp.Actions)
actions, err = f.dispatchCancelActions(actions)
if err != nil {
f.log.Error(err.Error())
actions := make([]fleetapi.Action, len(resp.Actions))
for idx, a := range resp.Actions {
actions[idx] = a
Copy link
Contributor

Choose a reason for hiding this comment

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

Instead of this forloop copy should work.

copy(actions, resp.Actions)

Copy link
Contributor Author

Choose a reason for hiding this comment

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

ok, loop was the original implementation. i'll change to copy

}

queued, expired := f.gatherQueuedActions(ts.UTC())
f.log.Debugf("Gathered %d actions from queue, %d actions expired", len(queued), len(expired))
f.log.Debugf("Expired actions: %v", expired)

actions = append(actions, queued...)

var errMsg string
// Persist state
f.stateStore.SetQueue(f.queue.Actions())
if err := f.stateStore.Save(); err != nil {
errMsg = fmt.Sprintf("failed to persist action_queue, error: %s", err)
f.log.Error(errMsg)
f.statusReporter.Update(state.Failed, errMsg, nil)
}

if err := f.dispatcher.Dispatch(context.Background(), f.acker, actions...); err != nil {
errMsg = fmt.Sprintf("failed to dispatch actions, error: %s", err)
f.log.Error(errMsg)
Expand All @@ -217,60 +187,6 @@ func (f *fleetGateway) worker() {
}
}

// queueScheduledActions will add any action in actions with a valid start time to the queue and return the rest.
// start time to current time comparisons are purposefully not made in case of cancel actions.
func (f *fleetGateway) queueScheduledActions(input fleetapi.Actions) []fleetapi.Action {
actions := make([]fleetapi.Action, 0, len(input))
for _, action := range input {
start, err := action.StartTime()
if err == nil {
f.log.Debugf("Adding action id: %s to queue.", action.ID())
f.queue.Add(action, start.Unix())
continue
}
if !stderr.Is(err, fleetapi.ErrNoStartTime) {
f.log.Warnf("Issue gathering start time from action id %s: %v", action.ID(), err)
}
actions = append(actions, action)
}
return actions
}

// dispatchCancelActions will separate and dispatch any cancel actions from the actions list and return the rest of the list.
// cancel actions are dispatched seperatly as they may remove items from the queue.
func (f *fleetGateway) dispatchCancelActions(actions []fleetapi.Action) ([]fleetapi.Action, error) {
// separate cancel actions from the actions list
cancelActions := make([]fleetapi.Action, 0, len(actions))
for i := len(actions) - 1; i >= 0; i-- {
action := actions[i]
if action.Type() == fleetapi.ActionTypeCancel {
cancelActions = append(cancelActions, action)
actions = append(actions[:i], actions[i+1:]...)
}
}
// Dispatch cancel actions
if len(cancelActions) > 0 {
if err := f.dispatcher.Dispatch(context.Background(), f.acker, cancelActions...); err != nil {
return actions, fmt.Errorf("failed to dispatch cancel actions: %w", err)
}
}
return actions, nil
}

// gatherQueuedActions will dequeue actions from the action queue and separate those that have already expired.
func (f *fleetGateway) gatherQueuedActions(ts time.Time) (queued, expired []fleetapi.Action) {
actions := f.queue.DequeueActions()
for _, action := range actions {
exp, _ := action.Expiration()
if ts.After(exp) {
expired = append(expired, action)
continue
}
queued = append(queued, action)
}
return queued, expired
}

func (f *fleetGateway) doExecute() (*fleetapi.CheckinResponse, error) {
f.backoff.Reset()

Expand Down
Loading