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

Jump to bottom

managed-agent will periodically call dispatcher.Dispatch #2344

Merged
merged 4 commits into from
Mar 7, 2023
Merged

managed-agent will periodically call dispatcher.Dispatch #2344

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0e82e35
Allow fleet-gateway to return empty action lists
michel-laterman Mar 2, 2023
791aa99
Change fix to periodically call the dispatcher
michel-laterman Mar 6, 2023
d3a1ebd
Fix comment
michel-laterman Mar 6, 2023
e5c2476
Fix changelog, add unit tests
michel-laterman Mar 6, 2023
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
11 changes: 6 additions & 5 deletions ...1-gateway-returns-empty-action-lists.yaml → ...s/1677797191-dispatch-periodic-flush.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,16 +11,17 @@
kind: bug-fix

# Change summary; a 80ish characters long description of the change.
summary: gateway sends empty action lists to channel
summary: Periodically run the dispatcher from a timer.
cmacknz marked this conversation as resolved.
Show resolved Hide resolved

# Long description; in case the summary is not enough to describe the change
# this field accommodate a description without length limits.
description: The fleet-gateway will send an empty action list to the
actions channel if a checkin returns no actions. This allows other
components such as the dispatcher to run at a regular interval.
description: Add a timer in the goroutine that periodically calls Dispatch with
no actions in the managed agent. This will allow the dispatcher to handle
scheduled actions if the fleet-gateway has errors or does not return any
actions.

# Affected component; a word indicating the component this changeset affects.
component: fleet-gateway
component: managed-agent

# PR number; optional; the PR number that added the changeset.
# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
Expand Down
4 changes: 3 additions & 1 deletion 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 @@ -165,7 +165,9 @@ func (f *fleetGateway) Run(ctx context.Context) error {

actions := make([]fleetapi.Action, len(resp.Actions))
copy(actions, resp.Actions)
f.actionCh <- actions
if len(actions) > 0 {
f.actionCh <- actions
}
}
}
}
Expand Down
9 changes: 4 additions & 5 deletions internal/pkg/agent/application/gateway/fleet/fleet_gateway_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@ import (
"testing"
"time"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gotest.tools/assert"

"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
"github.com/elastic/elastic-agent/internal/pkg/agent/application/gateway"
Expand Down Expand Up @@ -125,7 +125,7 @@ func TestFleetGateway(t *testing.T) {
Backoff: backoffSettings{Init: 1 * time.Second, Max: 5 * time.Second},
}

t.Run("send no event and receive empty list", withGateway(agentInfo, settings, func(
t.Run("send no event and receive no action", withGateway(agentInfo, settings, func(
t *testing.T,
gateway gateway.FleetGateway,
client *testingClient,
Expand All @@ -152,7 +152,7 @@ func TestFleetGateway(t *testing.T) {
require.NoError(t, err)
select {
case actions := <-gateway.Actions():
assert.Empty(t, actions)
t.Errorf("Expected no actions, got %v", actions)
default:
}
}))
Expand Down Expand Up @@ -244,7 +244,6 @@ func TestFleetGateway(t *testing.T) {
var count int
for {
waitFn()
<-gateway.Actions()
count++
if count == 4 {
return
Expand Down Expand Up @@ -358,7 +357,7 @@ func TestRetriesOnFailures(t *testing.T) {
require.NoError(t, err)
select {
case actions := <-gateway.Actions():
assert.Empty(t, actions)
t.Errorf("Expected no actions, got %v", actions)
default:
}
}))
Expand Down
8 changes: 8 additions & 0 deletions internal/pkg/agent/application/managed_mode.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,9 @@ import (
"github.com/elastic/elastic-agent/pkg/core/logger"
)

// dispatchFlushInterval is the min time between calls to dispatcher.Dispatch
michel-laterman marked this conversation as resolved.
Show resolved Hide resolved
const dispatchFlushInterval = time.Minute * 5

type managedConfigManager struct {
log *logger.Logger
agentInfo *info.AgentInfo
Expand Down Expand Up @@ -209,12 +212,17 @@ func (m *managedConfigManager) Run(ctx context.Context) error {

// pass actions collected from gateway to dispatcher
go func() {
t := time.NewTimer(dispatchFlushInterval)
for {
select {
case <-ctx.Done():
return
case <-t.C: // periodically call the dispatcher to handle scheduled actions.
cmacknz marked this conversation as resolved.
Show resolved Hide resolved
m.dispatcher.Dispatch(ctx, actionAcker)
t.Reset(dispatchFlushInterval)
Copy link
Contributor

@ycombinator ycombinator Mar 6, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

👍. I think this is a slightly better approach because it at least decouples the processing of the dispatcher's action queue from the Fleet Checkin call.

Copy link
Contributor

@ycombinator ycombinator Mar 6, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

I think we can have an even more efficient implementation of "when to re-dispatch" by deciding to process the action queue either:

  • when new actions are added to it (already handled by the case below this line), or
  • just-in-time for the next scheduled action (by looking at the next scheduled action's time and setting up a time.Timer accordingly) rather than on a fixed interval like we are doing here. But we can make this change in a separate PR if it's too complicated for this one.

case actions := <-gateway.Actions():
m.dispatcher.Dispatch(ctx, actionAcker, actions...)
t.Reset(dispatchFlushInterval)
}
}
}()
Expand Down