Skip to content

Commit

Permalink
✨ early heartbeat failure (#2388)
Browse files Browse the repository at this point in the history
When using heartbeats, currently the provider has a chance to start and
execute the heartbeat async. This means that for commands like run and
shell (and others), we gett confusing error messages:

```bash
> make cnquery/install && cnquery run aws -c asset.eol
x please update the provider plugin for aws plugin-ID=go.mondoo.com/cnquery/v9/providers/aws
→ loaded configuration from /home/zero/.config/mondoo/mondoo.yml using source default
FTL failed to run query error="rpc error: code = Canceled desc = grpc: the client connection is closing"
```

The mechanism in this PR is much more proactive at checking the
heartbeat first, before returning a running provider, thus eliminating
confusing errors:

```bash
> make cnquery/install && cnquery shell aws -c asset.eol
FTL failed to start provider aws error="please update the provider plugin for aws"
```

Signed-off-by: Dominik Richter <dominik.richter@gmail.com>
  • Loading branch information
arlimus authored Oct 27, 2023
1 parent d94b212 commit a9102f6
Showing 1 changed file with 29 additions and 20 deletions.
49 changes: 29 additions & 20 deletions providers/coordinator.go
Original file line number Diff line number Diff line change
Expand Up @@ -64,36 +64,42 @@ type RunningProvider struct {
}

// initialize the heartbeat with the provider
func (p *RunningProvider) heartbeat() {
func (p *RunningProvider) heartbeat() error {
interval := 2 * time.Second
gracePeriod := 3 * time.Second

if err := p.doOneHeartbeat(interval + gracePeriod); err != nil {
p.Shutdown()
return err
}

go func() {
for !p.isCloseOrShutdown() {
_, err := p.Plugin.Heartbeat(&pp.HeartbeatReq{
Interval: uint64(interval + gracePeriod),
})

if status, ok := status.FromError(err); ok {
if status.Code() == 12 {
log.Error().
Str("plugin-ID", p.ID).
Msg("please update the provider plugin for " + p.Name)
p.Shutdown()
break
}
}

if err != nil {
log.Error().
Str("plugin-ID", p.ID).
Msg("cannot establish heartbeat with the provider plugin for " + p.Name)
if err := p.doOneHeartbeat(interval + gracePeriod); err != nil {
p.Shutdown()
break
}

time.Sleep(interval)
}
}()

return nil
}

func (p *RunningProvider) doOneHeartbeat(t time.Duration) error {
_, err := p.Plugin.Heartbeat(&pp.HeartbeatReq{
Interval: uint64(t),
})
if err != nil {
if status, ok := status.FromError(err); ok {
if status.Code() == 12 {
return errors.New("please update the provider plugin for " + p.Name)
}
}
return errors.New("cannot establish heartbeat with the provider plugin for " + p.Name)
}
return nil
}

func (p *RunningProvider) isCloseOrShutdown() bool {
Expand Down Expand Up @@ -229,7 +235,10 @@ func (c *coordinator) Start(id string, isEphemeral bool, update UpdateProvidersC
Client: client,
Schema: provider.Schema,
}
res.heartbeat()

if err := res.heartbeat(); err != nil {
return nil, err
}

c.mutex.Lock()
if isEphemeral {
Expand Down

0 comments on commit a9102f6

Please sign in to comment.