-
Notifications
You must be signed in to change notification settings - Fork 1.3k
/
plugin.go
495 lines (421 loc) · 14.4 KB
/
plugin.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
// Copyright 2018 The OPA Authors. All rights reserved.
// Use of this source code is governed by an Apache2
// license that can be found in the LICENSE file.
// Package status implements status reporting.
package status
import (
"context"
"encoding/json"
"fmt"
"net/http"
"reflect"
prom "github.com/prometheus/client_golang/prometheus"
"github.com/open-policy-agent/opa/logging"
"github.com/open-policy-agent/opa/metrics"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/plugins/bundle"
"github.com/open-policy-agent/opa/util"
)
// Logger defines the interface for status plugins.
type Logger interface {
plugins.Plugin
Log(context.Context, *UpdateRequestV1) error
}
// UpdateRequestV1 represents the status update message that OPA sends to
// remote HTTP endpoints.
type UpdateRequestV1 struct {
Labels map[string]string `json:"labels"`
Bundle *bundle.Status `json:"bundle,omitempty"` // Deprecated: Use bulk `bundles` status updates instead
Bundles map[string]*bundle.Status `json:"bundles,omitempty"`
Discovery *bundle.Status `json:"discovery,omitempty"`
Metrics map[string]interface{} `json:"metrics,omitempty"`
Plugins map[string]*plugins.Status `json:"plugins,omitempty"`
}
// Plugin implements status reporting. Updates can be triggered by the caller.
type Plugin struct {
manager *plugins.Manager
config Config
bundleCh chan bundle.Status // Deprecated: Use bulk bundle status updates instead
lastBundleStatus *bundle.Status // Deprecated: Use bulk bundle status updates instead
bulkBundleCh chan map[string]*bundle.Status
lastBundleStatuses map[string]*bundle.Status
discoCh chan bundle.Status
lastDiscoStatus *bundle.Status
pluginStatusCh chan map[string]*plugins.Status
lastPluginStatuses map[string]*plugins.Status
queryCh chan chan *UpdateRequestV1
stop chan chan struct{}
reconfig chan interface{}
metrics metrics.Metrics
logger logging.Logger
trigger chan trigger
}
// Config contains configuration for the plugin.
type Config struct {
Plugin *string `json:"plugin"`
Service string `json:"service"`
PartitionName string `json:"partition_name,omitempty"`
ConsoleLogs bool `json:"console"`
Prometheus bool `json:"prometheus"`
Trigger *plugins.TriggerMode `json:"trigger,omitempty"` // trigger mode
}
type trigger struct {
ctx context.Context
done chan error
}
func (c *Config) validateAndInjectDefaults(services []string, pluginsList []string, trigger *plugins.TriggerMode) error {
if c.Plugin != nil {
var found bool
for _, other := range pluginsList {
if other == *c.Plugin {
found = true
break
}
}
if !found {
return fmt.Errorf("invalid plugin name %q in status", *c.Plugin)
}
} else if c.Service == "" && len(services) != 0 && !(c.ConsoleLogs || c.Prometheus) {
// For backwards compatibility allow defaulting to the first
// service listed, but only if console logging is disabled. If enabled
// we can't tell if the deployer wanted to use only console logs or
// both console logs and the default service option.
c.Service = services[0]
} else if c.Service != "" {
found := false
for _, svc := range services {
if svc == c.Service {
found = true
break
}
}
if !found {
return fmt.Errorf("invalid service name %q in status", c.Service)
}
}
t, err := plugins.ValidateAndInjectDefaultsForTriggerMode(trigger, c.Trigger)
if err != nil {
return fmt.Errorf("invalid status config: %w", err)
}
c.Trigger = t
return nil
}
// ParseConfig validates the config and injects default values.
func ParseConfig(config []byte, services []string, pluginsList []string) (*Config, error) {
t := plugins.DefaultTriggerMode
return NewConfigBuilder().WithBytes(config).WithServices(services).WithPlugins(pluginsList).WithTriggerMode(&t).Parse()
}
// ConfigBuilder assists in the construction of the plugin configuration.
type ConfigBuilder struct {
raw []byte
services []string
plugins []string
trigger *plugins.TriggerMode
}
// NewConfigBuilder returns a new ConfigBuilder to build and parse the plugin config.
func NewConfigBuilder() *ConfigBuilder {
return &ConfigBuilder{}
}
// WithBytes sets the raw plugin config.
func (b *ConfigBuilder) WithBytes(config []byte) *ConfigBuilder {
b.raw = config
return b
}
// WithServices sets the services that implement control plane APIs.
func (b *ConfigBuilder) WithServices(services []string) *ConfigBuilder {
b.services = services
return b
}
// WithPlugins sets the list of named plugins for status updates.
func (b *ConfigBuilder) WithPlugins(plugins []string) *ConfigBuilder {
b.plugins = plugins
return b
}
// WithTriggerMode sets the plugin trigger mode.
func (b *ConfigBuilder) WithTriggerMode(trigger *plugins.TriggerMode) *ConfigBuilder {
b.trigger = trigger
return b
}
// Parse validates the config and injects default values.
func (b *ConfigBuilder) Parse() (*Config, error) {
if b.raw == nil {
return nil, nil
}
var parsedConfig Config
if err := util.Unmarshal(b.raw, &parsedConfig); err != nil {
return nil, err
}
if parsedConfig.Plugin == nil && parsedConfig.Service == "" && len(b.services) == 0 && !parsedConfig.ConsoleLogs && !parsedConfig.Prometheus {
// Nothing to validate or inject
return nil, nil
}
if err := parsedConfig.validateAndInjectDefaults(b.services, b.plugins, b.trigger); err != nil {
return nil, err
}
return &parsedConfig, nil
}
// New returns a new Plugin with the given config.
func New(parsedConfig *Config, manager *plugins.Manager) *Plugin {
p := &Plugin{
manager: manager,
config: *parsedConfig,
bundleCh: make(chan bundle.Status),
bulkBundleCh: make(chan map[string]*bundle.Status),
discoCh: make(chan bundle.Status),
stop: make(chan chan struct{}),
reconfig: make(chan interface{}),
pluginStatusCh: make(chan map[string]*plugins.Status),
queryCh: make(chan chan *UpdateRequestV1),
logger: manager.Logger().WithFields(map[string]interface{}{"plugin": Name}),
trigger: make(chan trigger),
}
p.manager.UpdatePluginStatus(Name, &plugins.Status{State: plugins.StateNotReady})
return p
}
// WithMetrics sets the global metrics provider to be used by the plugin.
func (p *Plugin) WithMetrics(m metrics.Metrics) *Plugin {
p.metrics = m
return p
}
// Name identifies the plugin on manager.
const Name = "status"
// Lookup returns the status plugin registered with the manager.
func Lookup(manager *plugins.Manager) *Plugin {
if p := manager.Plugin(Name); p != nil {
return p.(*Plugin)
}
return nil
}
// Start starts the plugin.
func (p *Plugin) Start(ctx context.Context) error {
p.logger.Info("Starting status reporter.")
go p.loop()
// Setup a listener for plugin statuses, but only after starting the loop
// 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(), pluginStatus, loaded, failLoad,
lastRequest, lastSuccessfulActivation, lastSuccessfulDownload,
lastSuccessfulRequest, bundleLoadDuration)
}
// Set the status plugin's status to OK now that everything is registered and
// the loop is running. This will trigger an update on the listener with the
// current status of all the other plugins too.
p.manager.UpdatePluginStatus(Name, &plugins.Status{State: plugins.StateOK})
return nil
}
func (p *Plugin) register(r prom.Registerer, cs ...prom.Collector) {
for _, c := range cs {
if err := r.Register(c); err != nil {
p.logger.Error("Status metric failed to register on prometheus :%v.", err)
}
}
}
// Stop stops the plugin.
func (p *Plugin) Stop(ctx context.Context) {
p.logger.Info("Stopping status reporter.")
p.manager.UnregisterPluginStatusListener(Name)
done := make(chan struct{})
p.stop <- done
<-done
p.manager.UpdatePluginStatus(Name, &plugins.Status{State: plugins.StateNotReady})
}
// UpdateBundleStatus notifies the plugin that the policy bundle was updated.
// Deprecated: Use BulkUpdateBundleStatus instead.
func (p *Plugin) UpdateBundleStatus(status bundle.Status) {
p.bundleCh <- status
}
// BulkUpdateBundleStatus notifies the plugin that the policy bundle was updated.
func (p *Plugin) BulkUpdateBundleStatus(status map[string]*bundle.Status) {
p.bulkBundleCh <- status
}
// UpdateDiscoveryStatus notifies the plugin that the discovery bundle was updated.
func (p *Plugin) UpdateDiscoveryStatus(status bundle.Status) {
p.discoCh <- status
}
// UpdatePluginStatus notifies the plugin that a plugin status was updated.
func (p *Plugin) UpdatePluginStatus(status map[string]*plugins.Status) {
p.pluginStatusCh <- status
}
// Reconfigure notifies the plugin with a new configuration.
func (p *Plugin) Reconfigure(_ context.Context, config interface{}) {
p.reconfig <- config
}
// Snapshot returns the current status.
func (p *Plugin) Snapshot() *UpdateRequestV1 {
ch := make(chan *UpdateRequestV1)
p.queryCh <- ch
s := <-ch
return s
}
// Trigger can be used to control when the plugin attempts to upload
// status in manual triggering mode.
func (p *Plugin) Trigger(ctx context.Context) error {
done := make(chan error)
p.trigger <- trigger{ctx: ctx, done: done}
select {
case err := <-done:
return err
case <-ctx.Done():
return ctx.Err()
}
}
func (p *Plugin) loop() {
ctx, cancel := context.WithCancel(context.Background())
for {
select {
case statuses := <-p.pluginStatusCh:
p.lastPluginStatuses = statuses
if *p.config.Trigger == plugins.TriggerPeriodic {
err := p.oneShot(ctx)
if err != nil {
p.logger.Error("%v.", err)
} else {
p.logger.Info("Status update sent successfully in response to plugin update.")
}
}
case statuses := <-p.bulkBundleCh:
p.lastBundleStatuses = statuses
if *p.config.Trigger == plugins.TriggerPeriodic {
err := p.oneShot(ctx)
if err != nil {
p.logger.Error("%v.", err)
} else {
p.logger.Info("Status update sent successfully in response to bundle update.")
}
}
case status := <-p.bundleCh:
p.lastBundleStatus = &status
err := p.oneShot(ctx)
if err != nil {
p.logger.Error("%v.", err)
} else {
p.logger.Info("Status update sent successfully in response to bundle update.")
}
case status := <-p.discoCh:
p.lastDiscoStatus = &status
if *p.config.Trigger == plugins.TriggerPeriodic {
err := p.oneShot(ctx)
if err != nil {
p.logger.Error("%v.", err)
} else {
p.logger.Info("Status update sent successfully in response to discovery update.")
}
}
case newConfig := <-p.reconfig:
p.reconfigure(newConfig)
case respCh := <-p.queryCh:
respCh <- p.snapshot()
case update := <-p.trigger:
err := p.oneShot(update.ctx)
if err != nil {
p.logger.Error("%v.", err)
if update.ctx.Err() == nil {
update.done <- err
}
} else {
p.logger.Info("Status update sent successfully in response to manual trigger.")
}
close(update.done)
case done := <-p.stop:
cancel()
done <- struct{}{}
return
}
}
}
func (p *Plugin) oneShot(ctx context.Context) error {
req := p.snapshot()
if p.config.ConsoleLogs {
err := p.logUpdate(req)
if err != nil {
p.logger.Error("Failed to log to console: %v.", err)
}
}
if p.config.Prometheus {
updatePrometheusMetrics(req)
}
if p.config.Plugin != nil {
proxy, ok := p.manager.Plugin(*p.config.Plugin).(Logger)
if !ok {
return fmt.Errorf("plugin does not implement Logger interface")
}
return proxy.Log(ctx, req)
}
if p.config.Service != "" {
resp, err := p.manager.Client(p.config.Service).
WithJSON(req).
Do(ctx, "POST", fmt.Sprintf("/status/%v", p.config.PartitionName))
if err != nil {
return fmt.Errorf("Status update failed: %w", err)
}
defer util.Close(resp)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("status update failed, server replied with HTTP %v %v", resp.StatusCode, http.StatusText(resp.StatusCode))
}
}
return nil
}
func (p *Plugin) reconfigure(config interface{}) {
newConfig := config.(*Config)
if reflect.DeepEqual(p.config, *newConfig) {
p.logger.Debug("Status reporter configuration unchanged.")
return
}
p.logger.Info("Status reporter configuration changed.")
p.config = *newConfig
}
func (p *Plugin) snapshot() *UpdateRequestV1 {
s := &UpdateRequestV1{
Labels: p.manager.Labels(),
Discovery: p.lastDiscoStatus,
Bundle: p.lastBundleStatus,
Bundles: p.lastBundleStatuses,
Plugins: p.lastPluginStatuses,
}
if p.metrics != nil {
s.Metrics = map[string]interface{}{p.metrics.Info().Name: p.metrics.All()}
}
return s
}
func (p *Plugin) logUpdate(update *UpdateRequestV1) error {
eventBuf, err := json.Marshal(&update)
if err != nil {
return err
}
fields := map[string]interface{}{}
err = util.UnmarshalJSON(eventBuf, &fields)
if err != nil {
return err
}
p.manager.ConsoleLogger().WithFields(fields).WithFields(map[string]interface{}{
"type": "openpolicyagent.org/status",
}).Info("Status Log")
return nil
}
func updatePrometheusMetrics(u *UpdateRequestV1) {
pluginStatus.Reset()
for name, plugin := range u.Plugins {
pluginStatus.WithLabelValues(name, string(plugin.State)).Set(1)
}
lastSuccessfulActivation.Reset()
for _, bundle := range u.Bundles {
if bundle.Code == "" && !bundle.LastSuccessfulActivation.IsZero() {
loaded.WithLabelValues(bundle.Name).Inc()
} else {
failLoad.WithLabelValues(bundle.Name, bundle.Code, bundle.Message).Inc()
}
lastSuccessfulActivation.WithLabelValues(bundle.Name, bundle.ActiveRevision).Set(float64(bundle.LastSuccessfulActivation.UnixNano()))
lastSuccessfulDownload.WithLabelValues(bundle.Name).Set(float64(bundle.LastSuccessfulDownload.UnixNano()))
lastSuccessfulRequest.WithLabelValues(bundle.Name).Set(float64(bundle.LastSuccessfulRequest.UnixNano()))
lastRequest.WithLabelValues(bundle.Name).Set(float64(bundle.LastRequest.UnixNano()))
if bundle.Metrics != nil {
for stage, metric := range bundle.Metrics.All() {
switch stage {
case "timer_bundle_request_ns", "timer_rego_data_parse_ns", "timer_rego_module_parse_ns", "timer_rego_module_compile_ns", "timer_rego_load_bundles_ns":
bundleLoadDuration.WithLabelValues(bundle.Name, stage).Observe(float64(metric.(int64)))
}
}
}
}
}