Heartbeat Automatic Reload #8023
Conversation
heartbeat/monitors/task.go
Outdated
| client beat.Client | ||
| } | ||
|
|
||
| type InvalidMonitorProcessorsError struct{ root error } |
There was a problem hiding this comment.
exported type InvalidMonitorProcessorsError should have comment or be unexported
heartbeat/monitors/registry.go
Outdated
| e, found := r.modules[name] | ||
| if !found { | ||
| return nil | ||
| } | ||
| return e.Create | ||
| } | ||
|
|
||
| func (r *Registrar) AddActive(name string, builder ActiveBuilder) error { | ||
| func (r *Registry) AddActive(name string, builder ActiveBuilder) error { |
There was a problem hiding this comment.
exported method Registry.AddActive should have comment or be unexported
heartbeat/monitors/registry.go
Outdated
| e, found := r.modules[name] | ||
| return e.info, found | ||
| } | ||
|
|
||
| func (r *Registrar) GetFactory(name string) Factory { | ||
| func (r *Registry) GetFactory(name string) JobFactory { |
There was a problem hiding this comment.
exported method Registry.GetFactory should have comment or be unexported
heartbeat/monitors/registry.go
Outdated
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (r *Registrar) Query(name string) (Info, bool) { | ||
| func (r *Registry) Query(name string) (BasicInfo, bool) { |
There was a problem hiding this comment.
exported method Registry.Query should have comment or be unexported
heartbeat/monitors/registry.go
Outdated
| panic(err) | ||
| } | ||
| } | ||
|
|
||
| func (r *Registrar) Register(name string, t Type, builder ActiveBuilder) error { | ||
| func (r *Registry) Register(name string, t Type, builder ActiveBuilder) error { |
There was a problem hiding this comment.
exported method Registry.Register should have comment or be unexported
heartbeat/monitors/monitor.go
Outdated
| pipeline beat.Pipeline | ||
| } | ||
|
|
||
| func NewMonitor( |
There was a problem hiding this comment.
exported function NewMonitor should have comment or be unexported
heartbeat/monitors/monitor.go
Outdated
| "github.com/elastic/beats/libbeat/logp" | ||
| ) | ||
|
|
||
| type Monitor struct { |
There was a problem hiding this comment.
exported type Monitor should have comment or be unexported
heartbeat/monitors/job.go
Outdated
|
|
||
| type JobRunner func() (beat.Event, []JobRunner, error) | ||
|
|
||
| type JobFactory func(*common.Config) ([]Job, error) |
There was a problem hiding this comment.
exported type JobFactory should have comment or be unexported
heartbeat/monitors/job.go
Outdated
| Run() (beat.Event, []JobRunner, error) | ||
| } | ||
|
|
||
| type JobRunner func() (beat.Event, []JobRunner, error) |
There was a problem hiding this comment.
exported type JobRunner should have comment or be unexported
heartbeat/monitors/job.go
Outdated
| "github.com/elastic/beats/libbeat/common" | ||
| ) | ||
|
|
||
| type Job interface { |
There was a problem hiding this comment.
exported type Job should have comment or be unexported
heartbeat/monitors/factory.go
Outdated
| return &ReloaderFactory{sched} | ||
| } | ||
|
|
||
| func (f *ReloaderFactory) Create(p beat.Pipeline, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) { |
There was a problem hiding this comment.
exported method ReloaderFactory.Create should have comment or be unexported
heartbeat/monitors/factory.go
Outdated
|
|
||
| type ReloaderFactory struct{ sched *scheduler.Scheduler } | ||
|
|
||
| func NewFactory(sched *scheduler.Scheduler) *ReloaderFactory { |
There was a problem hiding this comment.
exported function NewFactory should have comment or be unexported
heartbeat/monitors/factory.go
Outdated
| Processors processors.PluginConfig `config:"processors"` | ||
| } | ||
|
|
||
| type ReloaderFactory struct{ sched *scheduler.Scheduler } |
There was a problem hiding this comment.
exported type ReloaderFactory should have comment or be unexported
heartbeat/config/config.go
Outdated
| Processors processors.PluginConfig `config:"processors"` | ||
| } | ||
|
|
||
| type MonitorsConfig []MonitorTaskConfig |
There was a problem hiding this comment.
exported type MonitorsConfig should have comment or be unexported
heartbeat/config/config.go
Outdated
| "github.com/elastic/beats/libbeat/processors" | ||
| ) | ||
|
|
||
| type MonitorTaskConfig struct { |
There was a problem hiding this comment.
exported type MonitorTaskConfig should have comment or be unexported
heartbeat/beater/heartbeat.go
Outdated
| } | ||
| return bt, nil | ||
| } | ||
|
|
||
| func (bt *Heartbeat) Monitors() error { |
There was a problem hiding this comment.
exported method Heartbeat.Monitors should have comment or be unexported
heartbeat/monitors/factory.go
Outdated
| type factory struct{ sched *scheduler.Scheduler } | ||
|
|
||
| // NewFactory takes a scheduler and creates a factory that can create Monitor objects. | ||
| func NewFactory(sched *scheduler.Scheduler) *factory { |
There was a problem hiding this comment.
exported func NewFactory returns unexported type *monitors.factory, which can be annoying to use
heartbeat/monitors/factory.go
Outdated
| return &Factory{sched} | ||
| } | ||
|
|
||
| func (f *Factory) Create(p beat.Pipeline, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) { |
There was a problem hiding this comment.
exported method Factory.Create should have comment or be unexported
|
Is it required that we remove support for the poll file or could the file polling still happening inside the Beats reloading? An other idea to keep it more backward compatible would be to remove the poll file support from inside the reloaded configs (if this doesn't get too hacky). |
heartbeat/monitors/plugin.go
Outdated
| } | ||
| } | ||
|
|
||
| type MonitorPluginAlreadyExistsError pluginBuilder |
There was a problem hiding this comment.
exported type MonitorPluginAlreadyExistsError should have comment or be unexported
heartbeat/monitors/plugin.go
Outdated
| type Type uint8 | ||
|
|
||
| const ( | ||
| ActiveMonitor Type = iota + 1 |
There was a problem hiding this comment.
exported const ActiveMonitor should have comment (or a comment on this block) or be unexported
heartbeat/monitors/plugin.go
Outdated
| // monitors | ||
| type PluginBuilder func(string, *common.Config) ([]Job, error) | ||
|
|
||
| type Type uint8 |
There was a problem hiding this comment.
exported type Type should have comment or be unexported
| Monitors []*common.Config `config:"monitors" validate:"required"` | ||
| Scheduler Scheduler `config:"scheduler"` | ||
| Monitors []*common.Config `config:"monitors"` | ||
| ConfigMonitors *common.Config `config:"config.monitors"` |
There was a problem hiding this comment.
Could you update the heartbeat.reference.yml file? This should make it more obvious how the config will look like.
|
@ruflin I think we could bring back the Are you thinking we'd remove it in the next major or the next minor? I do think once we make this change we should delete all documentation of the |
|
@urso FYI this is ready for a preliminary review. There are no tests, but I'd appreciate your input on this approach to config file reloading. I also took the time to refactor a lot of the existing code to simplify the data structures and control flow, as well as too make the codebase more amenable to testing. I'm sure writing unit tests for all this code (which IMHO is a blocker for this), will change it more still, but before I get too far ahead of myself I think it'd be good to have feedback from you and @ruflin |
heartbeat/monitors.d/elastic.co.yml
Outdated
| name: zanzibar | ||
| enabled: true | ||
| urls: ["http://elastic.co"] | ||
| schedule: '@every 1s' |
There was a problem hiding this comment.
Is this file intended? doesn't seem related to the PR
There was a problem hiding this comment.
no and yes. It won't go in the final one, but it's useful to keep around for manual testing for now when I switch branches.
heartbeat/beater/heartbeat.go
Outdated
| } | ||
|
|
||
| func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) { | ||
| func New(b *beat.Beat, rawConfig *common.Config) (beat.Beater, error) { |
There was a problem hiding this comment.
exported function New should have comment or be unexported
heartbeat/beater/heartbeat.go
Outdated
| } | ||
|
|
||
| func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) { | ||
| func New(b *beat.Beat, rawConfig *common.Config) (beat.Beater, error) { |
There was a problem hiding this comment.
exported function New should have comment or be unexported
heartbeat/beater/heartbeat.go
Outdated
| return nil | ||
| } | ||
|
|
||
| // RunStaticMonitors runs the `heartbeat.config.monitors` portion of the yaml config if present. |
There was a problem hiding this comment.
comment on exported method Heartbeat.RunDynamicMonitors should be of the form "RunDynamicMonitors ..."
|
FYI went through the PR again and cleaned up the code a bit more, broke up some functions a bit, and removed a few dead struct members. |
There was a problem hiding this comment.
+1 on deprecating the poll file. This gives us the option to potentially remove it in 7.0 or also if we realise we can't fully replace it to either undeprecate it or find an other solution for it.
We should discuss if we remove it for 7.0, I would not remove it in a minor.
Not sure if we should remove the docs if we deprecated it, as the docs should also show that it is deprecated.
We should remove it from the config files.
heartbeat/config/config.go
Outdated
| } | ||
|
|
||
| // Scheduler defines the syntax of a heartbeat.yaml scheduler block. |
There was a problem hiding this comment.
Nit but in case I miss it later: We always use .yml (history ...)
| "github.com/elastic/beats/libbeat/common" | ||
| ) | ||
|
|
||
| func Test_newPluginsReg(t *testing.T) { |
There was a problem hiding this comment.
TestNewPluginsReg, I prefer not to have _ in function names.
There was a problem hiding this comment.
I discussed this with @urso in a previous PR. Our conclusion was that the linter doesn't complain, so this is OK. The go project itself added support for underscores since they were a requirement for doc_test.go files. In other words, they explicitly added support in the linter for underscores, but only for test files.
These names (and parts of the test bodies), were generated with https://github.com/cweill/gotests . I'm sure underscores were a consideration in the creation of that library as well.
There was a problem hiding this comment.
Good to know. Didn't know about it.
| } | ||
| } | ||
|
|
||
| func Test_pluginsReg_add(t *testing.T) { |
|
@ruflin I would love to create a more minimal PR. In general, that's my preference. However, in this case, I think a larger PR is more appropriate. I apologize for the pain of this large PR, but I do think it's the best way forward. The reason being that adding reload capability required substantive changes to the original code, and that modifying the original code with confidence was difficult due to it being hard to follow. Additionally, I think we need to add tests to this code. If we're adding test we are due for a refactor as well. The original code is hard to follow in part because of naming choices and the length of functions. This is not unexpected to find in an un-unit-tested codebase. This refactor makes understanding the code more tractable. Before, reading the code, IMHO, involved keeping in one's head many similar sounding names and some unique but confusing names. Additionally, I don't think we can merge any of this without substantive tests. The refactored code breaks up the logic more finely and makes that testing easier. |
heartbeat/beater/heartbeat.go
Outdated
| } | ||
|
|
||
| // RunDynamicMonitors runs the `heartbeat.config.monitors` portion of the yaml config if present. | ||
| func (bt *Heartbeat) RunDynamicMonitors(b *beat.Beat) (onDone func(), err error) { |
There was a problem hiding this comment.
consider to return the monitorReloader or pass the monitorReloader.
E.g. in beater Run:
if bt.config.ConfigMonitors.Enabled() {
reloader := cfgfile.NewReloader(...)
err := startReloader(reloader, ...)
if err != nil { ... }
defer reloader.Stop() // defer has function scope
}
with
func startReloader(reloader, factory) error {
if err := reloader.Check(factory); err != nil {
return err
}
go reloader.Run(factory)
return nil
}
There was a problem hiding this comment.
I'm a little unclear as to the why here? Is the goal to decouple the monitor creating logic from RunDynamicMonitors?
There was a problem hiding this comment.
It's more about symmetry. The function that instantiates some object which needs cleanup (Close/Stop/Cleanup...) should also clean it up (more akin to RAII). Returning some cleanup function is an edge-case that should be used sparily (e.g. teardown in test code is a common pattern). This way we end up with the common create-defer close pattern:
x := New...
defer x.Close()
returning a stop function to defer on from a method named 'RunX' just looks somewhat unexpected. ;)
There was a problem hiding this comment.
Ah, thanks for the heads up on that, makes sense, will change it
heartbeat/monitors/monitor.go
Outdated
| func (m *Monitor) makeTasks(config *common.Config, jobs []Job) error { | ||
| mtConf := monitorTaskConfig{} | ||
| if err := config.Unpack(&mtConf); err != nil { | ||
| return errors.Wrap(err, "unexpected error while configuring heartbeat job") |
There was a problem hiding this comment.
It's more like an 'invalid config' error. Unpacking/parsing the config into mtConf failed due to mtConf and the configuration object passed being incompatible.
heartbeat/monitors/plugin.go
Outdated
|
|
||
| // MonitorPluginAlreadyExistsError is returned when there is an attempt to register two plugins | ||
| // with the same name. | ||
| type MonitorPluginAlreadyExistsError pluginBuilder |
There was a problem hiding this comment.
Stuttering. The full type name is monitors.MonitorPluginAlreadyExistsError. Renaming it to monitors.PluginAlreadyExistsError will contain the same information.
heartbeat/monitors/plugin.go
Outdated
| ) | ||
|
|
||
| // PluginsReg maintains the canonical list of valid Heartbeat monitors at runtime. | ||
| var PluginsReg = newPluginsReg() |
heartbeat/watcher/watch_config.go
Outdated
| Poll time.Duration `config:"watch.poll_file.interval" validate:"min=1"` | ||
| } | ||
|
|
||
| var DefaultWatchConfig = watchConfig{ |
There was a problem hiding this comment.
exported var DefaultWatchConfig should have comment or be unexported
heartbeat/beater/heartbeat.go
Outdated
| @@ -56,36 +64,80 @@ func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) { | |||
| return nil, err | |||
| } | |||
|
|
|||
| sched := scheduler.NewWithLocation(limit, location) | |||
| manager, err := newMonitorManager(b.Publisher, sched, monitors.Registry, config.Monitors) | |||
| scheduler := scheduler.NewWithLocation(limit, location) | |||
| if err != nil { | |||
There was a problem hiding this comment.
Is this error check still needed now that the newMonitorManager code is gone?
heartbeat/dynamic.json
Outdated
| @@ -0,0 +1,2 @@ | |||
| {"urls": ["http://blah2.elastic.co"], "schedule": "@every 10s", "name": "foobarz", "enabled": true} | |||
There was a problem hiding this comment.
I assume thats for testing purpose
heartbeat/heartbeat.reference.yml
Outdated
|
|
||
| # Maximum amount of time to randomly delay the start of a metricset. Use 0 to | ||
| # disable startup delay. | ||
| metricbeat.max_start_delay: 10s |
There was a problem hiding this comment.
oops, I think I did some copy/paste at one point.
heartbeat/heartbeat.reference.yml
Outdated
| @@ -62,6 +62,23 @@ heartbeat.monitors: | |||
| # sub-dictionary. Default is false. | |||
| #fields_under_root: false | |||
|
|
|||
| # Config reloading allows to dynamically load monitors. Each file which is | |||
There was a problem hiding this comment.
This will probably get overwritten when you run make update. You have to modify the _meta/beat.reference.yml file
| @@ -31,7 +31,7 @@ import ( | |||
| type Config struct { | |||
| Name string `config:"name"` | |||
|
|
|||
| URLs []string `config:"urls" validate:"required"` | |||
| URLs []string `config:"urls"` | |||
There was a problem hiding this comment.
This is only not required if a poll_file is specified?
There was a problem hiding this comment.
Yes, this is a good point. I'll need to add in an error about that somewhere. I assume the annotations here aren't rich enough to encode that conditional?
There was a problem hiding this comment.
OK, I thought I had broken this with my refactor, but currently you may not use the poll file exclusively, you must declare at least one regular host.
I'll preserve that behavior and delete this line.
heartbeat/monitors/plugin.go
Outdated
| @@ -20,28 +20,120 @@ package monitors | |||
| import ( | |||
| "errors" | |||
|
|
|||
| "fmt" | |||
heartbeat/monitors/pluginconf.go
Outdated
| mpi := MonitorPluginInfo{Enabled: true} | ||
|
|
||
| if err := config.Unpack(&mpi); err != nil { | ||
| return mpi, errors.Wrap(err, "Error unpacking monitor plugin config") |
There was a problem hiding this comment.
Nit: errors we return are normally starting lower case. I wonder why hound did not complain?
There was a problem hiding this comment.
Maybe it doesn't understand errors.Wrap?
At any rate, will fix
heartbeat/monitors/task.go
Outdated
|
|
||
| type jobCanceller func() error | ||
|
|
||
| type monitorTask struct { |
There was a problem hiding this comment.
Can this just be called task. Otherwise you repeat the package name. Same below for taskConfig and all the types / methods that contain monitor in the name.
There was a problem hiding this comment.
Some comment clarifying that monitorTask is used to lift monitoring jobs into the scheduler might be helpful.
heartbeat/monitors/task.go
Outdated
| config monitorTaskConfig | ||
| monitor *Monitor | ||
| processors *processors.Processors | ||
| cancelTf jobCanceller |
There was a problem hiding this comment.
taskFunc. I think it should just be called jobCancelFn
heartbeat/beater/heartbeat.go
Outdated
| @@ -25,29 +25,37 @@ import ( | |||
| "github.com/elastic/beats/libbeat/common" | |||
| "github.com/elastic/beats/libbeat/logp" | |||
|
|
|||
| "github.com/pkg/errors" | |||
heartbeat/monitors/monitor.go
Outdated
| enabled bool | ||
| watchPollTasks []*monitorTask | ||
| watch watcher.Watch | ||
| runMtx sync.Mutex |
There was a problem hiding this comment.
I guess watchPollTasks, watch, and runMtx are all related to dynamic jobs via file config watcher. Let's either distuingish them visually by adding a newline and comment or introduce a type/struct to combine watch related fields.
heartbeat/monitors/monitor.go
Outdated
| return | ||
| } | ||
|
|
||
| m.watchPollTasks = append(m.watchPollTasks, watchTasks...) |
There was a problem hiding this comment.
this doesn't look thread-safe. The watcher is supposed to pickup changes while heartbeat is running. Maybe I'm missing something, but how do we make sure new tasks are started and old stopped?
heartbeat/beater/heartbeat.go
Outdated
| "github.com/elastic/beats/libbeat/logp" | ||
| "github.com/elastic/beats/heartbeat/monitors" | ||
|
|
||
| "github.com/pkg/errors" |
heartbeat/beater/heartbeat.go
Outdated
| manager *monitorManager | ||
| // config is used for iterating over elements of the config. | ||
| config config.Config | ||
| // rawConfig is used in places where we want to unpack the config differently. |
heartbeat/monitors/task.go
Outdated
| return t.prepareSchedulerJob(meta, t.job.Run) | ||
| } | ||
|
|
||
| func (t *task) Start() { |
There was a problem hiding this comment.
surprised that hound does not complain, same below
|
@ruflin thanks for the review I've addressed those issues in 7dabc9d |
|
The failing config reload test could be related. Looks like a race. Full log output here to make sure it's not lost. |
which forces programmer to think about implications about what's being printed.
|
@ruflin rebased, and fix for race added in 8e8a8a3 . This was a bit of a weird one, the cause was a The race wasn't 'real', beyond that. The fix, which @andrewkroh agreed with after a discussion, was to modify the interface for |
There was a problem hiding this comment.
LGTM, great work done here, it will enable heartbeat to be even more awesome in 6.5! 🎉
@andrewvc I think you can move this PR forward as @ruflin is not around, you can address any other comment he may have in future PRs.
Also, I left a comment about dynamic meta fields, but I think that can be done later, which will make reviews simpler.
| } | ||
|
|
||
| // Create makes a new Runner for a new monitor with the given Config. | ||
| func (f *RunnerFactory) Create(p beat.Pipeline, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) { |
There was a problem hiding this comment.
I would not do it on this PR, but you can pass meta to the ConnectWith call (DynamicFields field). This will be useful if you integrate autodiscover, as it will pass metadata from the provider (docker, kubernetes, etc) and enrich all events from the monitor.
|
@andrewvc can you please add a changelog entry? |
|
@exekias thanks so much for the review! I just pushed the changelog entry. I'll open a new PR once this is merged for autodiscovery, and the metadata will go with that. |
Add automatic reloading for heartbeat config files. This deprecates the `watch.poll_file` options. This patch also fixes a potential source of races in code using `cfgfile/Runner` by making that interface implement `Stringer`, the reason being that by default `cfgfile/Runner` can recursively print the backing structure, which can trigger a race. (cherry picked from commit 037a4f2)
Add automatic reloading for heartbeat config files. This deprecates the `watch.poll_file` options. This patch also fixes a potential source of races in code using `cfgfile/Runner` by making that interface implement `Stringer`, the reason being that by default `cfgfile/Runner` can recursively print the backing structure, which can trigger a race. (cherry picked from commit 037a4f2)
Docs for the changes in #8023 for heartbeat automatic reloading.
Docs for the changes in elastic#8023 for heartbeat automatic reloading. (cherry picked from commit 04597c8)
This WIP patch brings heartbeat automatic reloading in-line with other Beats, using the standard reload feature. It is not yet ready for any sort of review. I'm mostly posting this to get CI running on it. This patch:
monitors.configurationoption, that enables reloading individualmonitorYAML blocks from a directory (monitors.d/*.ymlby default).poll_filefunctionalityThis is currently a WIP, it definitely needs a good number more tests.
This will be a significant breaking change due to the removal of poll file functionality, but since heartbeat is still in beta, we can and should do this due to the non-standard nature of the
poll_file.