ewilliams28/releases
1
0
Fork 0
mirror of https://github.com/open-policy-agent/opa.git synced 2026-08-12 19:32:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

status: Registering Prometheus Collectors on Status-plugin reconfigure (#5928)

Browse Source 
* status: Registering Prometheus Collectors on Status-plugin reconfigure

if `Prometheus` was enabled on active config change; and unregistering collectors if disabled.

Fixes: #5918
Signed-off-by: Johan Fylling <johan.dev@fylling.se>
This commit is contained in:
Johan Fylling
2023-05-19 10:47:32 +02:00
committed by GitHub
parent cff9ed86f4
commit 691173a347
4 changed files with 101 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/status/metrics.go
+15
View File
@@ -60,6 +60,21 @@ var (
Help: "Histogram for the bundle loading duration by stage.",
Buckets: prometheus.ExponentialBuckets(1000, 2, 20),
}, []string{"name", "stage"})
// allCollectors is a list of all collectors maintained by the status plugin.
// Note: when adding a new collector, make sure to also add it to this list,
// or it won't survive status plugin reconfigure events.
allCollectors = []prometheus.Collector{
opaInfo,
pluginStatus,
loaded,
failLoad,
lastRequest,
lastSuccessfulActivation,
lastSuccessfulDownload,
lastSuccessfulRequest,
bundleLoadDuration,
}
)
func init() {
plugins/status/plugin.go
+40 -9
View File
@@ -57,7 +57,7 @@ type Plugin struct {
lastPluginStatuses map[string]*plugins.Status
queryCh chan chan *UpdateRequestV1
stop chan chan struct{}
reconfig chan interface{}
reconfig chan reconfigure
metrics metrics.Metrics
logger logging.Logger
trigger chan trigger
@@ -73,6 +73,11 @@ type Config struct {
Trigger *plugins.TriggerMode `json:"trigger,omitempty"` // trigger mode
}
type reconfigure struct {
config interface{}
done chan struct{}
}
type trigger struct {
ctx context.Context
done chan error
@@ -198,7 +203,7 @@ func New(parsedConfig *Config, manager *plugins.Manager) *Plugin {
discoCh: make(chan bundle.Status),
decisionLogsCh: make(chan lstat.Status),
stop: make(chan chan struct{}),
reconfig: make(chan interface{}),
reconfig: make(chan reconfigure),
pluginStatusCh: make(chan map[string]*plugins.Status),
queryCh: make(chan chan *UpdateRequestV1),
logger: manager.Logger().WithFields(map[string]interface{}{"plugin": Name}),
@@ -237,10 +242,8 @@ func (p *Plugin) Start(ctx context.Context) error {
// to prevent blocking threads pushing the plugin updates.
p.manager.RegisterPluginStatusListener(Name, p.UpdatePluginStatus)
if p.config.Prometheus && p.manager.PrometheusRegister() != nil {
p.register(p.manager.PrometheusRegister(), opaInfo, pluginStatus, loaded, failLoad,
lastRequest, lastSuccessfulActivation, lastSuccessfulDownload,
lastSuccessfulRequest, bundleLoadDuration)
if p.config.Prometheus {
p.registerAll()
}
// Set the status plugin's status to OK now that everything is registered and
@@ -258,6 +261,24 @@ func (p *Plugin) register(r prom.Registerer, cs ...prom.Collector) {
}
}
func (p *Plugin) registerAll() {
if p.manager.PrometheusRegister() != nil {
p.register(p.manager.PrometheusRegister(), allCollectors...)
}
}
func (p *Plugin) unregister(r prom.Registerer, cs ...prom.Collector) {
for _, c := range cs {
r.Unregister(c)
}
}
func (p *Plugin) unregisterAll() {
if p.manager.PrometheusRegister() != nil {
p.unregister(p.manager.PrometheusRegister(), allCollectors...)
}
}
// Stop stops the plugin.
func (p *Plugin) Stop(ctx context.Context) {
p.logger.Info("Stopping status reporter.")
@@ -296,7 +317,9 @@ func (p *Plugin) UpdatePluginStatus(status map[string]*plugins.Status) {
// Reconfigure notifies the plugin with a new configuration.
func (p *Plugin) Reconfigure(_ context.Context, config interface{}) {
p.reconfig <- config
done := make(chan struct{})
p.reconfig <- reconfigure{config: config, done: done}
<-done
}
// Snapshot returns the current status.
@@ -378,8 +401,9 @@ func (p *Plugin) loop() {
p.logger.Info("Status update sent successfully in response to discovery update.")
}
}
case newConfig := <-p.reconfig:
p.reconfigure(newConfig)
case update := <-p.reconfig:
p.reconfigure(update.config)
update.done <- struct{}{}
case respCh := <-p.queryCh:
respCh <- p.snapshot()
case update := <-p.trigger:
@@ -451,6 +475,13 @@ func (p *Plugin) reconfigure(config interface{}) {
}
p.logger.Info("Status reporter configuration changed.")
if newConfig.Prometheus && !p.config.Prometheus {
p.registerAll()
} else if !newConfig.Prometheus && p.config.Prometheus {
p.unregisterAll()
}
p.config = *newConfig
}
plugins/status/plugin_go1.17_test.go
+32
View File
@@ -110,6 +110,38 @@ func TestPluginPrometheus(t *testing.T) {
if pluginsStatus != 1 {
t.Fatalf("Unexpected number of plugins (%v), got %v", 1, pluginsStatus)
}
// Assert that metrics are purged when prometheus is disabled
prometheusDisabledConfig := newConfig(fixture.manager, func(c *Config) {
c.Prometheus = false
})
fixture.plugin.Reconfigure(ctx, prometheusDisabledConfig)
eventually(t, func() bool { return fixture.plugin.config.Prometheus == false })
if len(registerMock.Collectors) != 0 {
t.Fatalf("Number of collectors expected (%v), got %v", 0, len(registerMock.Collectors))
}
// Assert that metrics are re-registered when prometheus is re-enabled
prometheusReenabledConfig := newConfig(fixture.manager, func(c *Config) {
c.Prometheus = true
})
fixture.plugin.Reconfigure(ctx, prometheusReenabledConfig)
eventually(t, func() bool { return fixture.plugin.config.Prometheus == true })
if len(registerMock.Collectors) != 9 {
t.Fatalf("Number of collectors expected (%v), got %v", 9, len(registerMock.Collectors))
}
}
func eventually(t *testing.T, predicate func() bool) {
for i := 0; i < 100; i++ {
if predicate() {
return
}
time.Sleep(time.Millisecond * 10)
}
t.Fatal("check took too long")
}
func assertOpInformationGauge(t *testing.T, registerMock *prometheusRegisterMock) {
plugins/status/plugin_test.go
+14 -8
View File
@@ -779,14 +779,7 @@ func newTestFixture(t *testing.T, m metrics.Metrics, options ...testPluginCustom
t.Fatal(err)
}
pluginConfig := []byte(`{
"service": "example",
}`)
config, _ := ParseConfig(pluginConfig, manager.Services(), nil)
for _, option := range options {
option(config)
}
config := newConfig(manager, options...)
p := New(config, manager).WithMetrics(m)
@@ -798,6 +791,19 @@ func newTestFixture(t *testing.T, m metrics.Metrics, options ...testPluginCustom
}
func newConfig(manager *plugins.Manager, options ...testPluginCustomizer) *Config {
pluginConfig := []byte(`{
"service": "example",
}`)
config, _ := ParseConfig(pluginConfig, manager.Services(), nil)
for _, option := range options {
option(config)
}
return config
}
type testServer struct {
t *testing.T
expCode int