Repository navigation
Expand file tree
/
Copy pathprocessor_impl.go
More file actions
127 lines (111 loc) · 4.54 KB
/
Copy pathprocessor_impl.go
File metadata and controls
127 lines (111 loc) · 4.54 KB
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
package processor
import (
"context"
"fmt"
v1 "github.com/stackrox/rox/generated/api/v1"
"github.com/stackrox/rox/generated/storage"
"github.com/stackrox/rox/pkg/integrationhealth"
"github.com/stackrox/rox/pkg/logging"
pkgNotifier "github.com/stackrox/rox/pkg/notifier"
"github.com/stackrox/rox/pkg/notifiers"
"github.com/stackrox/rox/pkg/protocompat"
"github.com/stackrox/rox/pkg/set"
)
var (
// Replacing with a background context such that outside context cancellation
// does not affect long-running go routines.
ctxBackground = context.Background()
log = logging.LoggerForModule()
)
// Processor takes in alerts and sends the notifications tied to that alert
type processorImpl struct {
ns pkgNotifier.Set
reporter integrationhealth.Reporter
}
func (p *processorImpl) HasNotifiers() bool {
return p.ns.HasNotifiers()
}
func (p *processorImpl) HasEnabledAuditNotifiers() bool {
return p.ns.HasEnabledAuditNotifiers()
}
// RemoveNotifier removes the in memory copy of the specified notifier
func (p *processorImpl) RemoveNotifier(ctx context.Context, id string) {
p.ns.RemoveNotifier(ctx, id)
}
// GetNotifier gets the in memory copy of the specified notifier id
func (p *processorImpl) GetNotifier(ctx context.Context, id string) (notifier notifiers.Notifier) {
return p.ns.GetNotifier(ctx, id)
}
// UpdateNotifier updates or adds the passed notifier into memory
func (p *processorImpl) UpdateNotifier(ctx context.Context, notifier notifiers.Notifier) {
p.ns.UpsertNotifier(ctx, notifier)
}
// ProcessAlert pushes the alert into a channel to be processed
func (p *processorImpl) ProcessAlert(ctx context.Context, alert *storage.Alert) {
if len(alert.GetPolicy().GetNotifiers()) == 0 {
return
}
alertNotifiers := set.NewStringSet(alert.GetPolicy().GetNotifiers()...)
p.ns.ForEach(ctx, func(ctx context.Context, notifier notifiers.Notifier, failures pkgNotifier.AlertSet) {
if alertNotifiers.Contains(notifier.ProtoNotifier().GetId()) {
go func() {
err := pkgNotifier.TryToAlert(ctx, notifier, alert)
if err != nil {
p.UpdateNotifierHealthStatus(notifier, storage.IntegrationHealth_UNHEALTHY, err.Error())
failures.Add(alert)
} else {
p.UpdateNotifierHealthStatus(notifier, storage.IntegrationHealth_HEALTHY, "")
}
}()
}
})
}
// ProcessAuditMessage sends the audit message with all applicable notifiers.
func (p *processorImpl) ProcessAuditMessage(ctx context.Context, msg *v1.Audit_Message) {
// TODO: Turn processorImpl into a work queue and introduce func (p *processorImpl) run(context.Context) error.
// With that, we wouldn't have to fan out n go routines (n = # notifiers in p.ns) and ensure ordering
// of audit messages.
p.ns.ForEach(ctx, func(_ context.Context, notifier notifiers.Notifier, _ pkgNotifier.AlertSet) {
go p.tryToSendAudit(ctxBackground, notifier, msg)
})
}
func (p *processorImpl) UpdateNotifierHealthStatus(notifier notifiers.Notifier, healthStatus storage.IntegrationHealth_Status, errMessage string) {
p.reporter.UpdateIntegrationHealthAsync(&storage.IntegrationHealth{
Id: notifier.ProtoNotifier().GetId(),
Name: notifier.ProtoNotifier().GetId(),
Type: storage.IntegrationHealth_NOTIFIER,
Status: healthStatus,
LastTimestamp: protocompat.TimestampNow(),
ErrorMessage: errMessage,
})
}
func (p *processorImpl) tryToSendAudit(ctx context.Context, notifier notifiers.Notifier, msg *v1.Audit_Message) {
auditNotifier, ok := notifier.(notifiers.AuditNotifier)
if ok {
if err := auditNotifier.SendAuditMessage(ctx, msg); err != nil {
protoNotifier := notifier.ProtoNotifier()
log.Errorf("Unable to send audit msg to %s (%s): %v", protoNotifier.GetName(), protoNotifier.GetType(), err)
p.UpdateNotifierHealthStatus(notifier, storage.IntegrationHealth_UNHEALTHY, fmt.Sprintf("Unable to send audit msg: %v", err))
}
p.UpdateNotifierHealthStatus(notifier, storage.IntegrationHealth_HEALTHY, "")
}
}
// Used for testing.
func (p *processorImpl) processAlertSync(ctx context.Context, alert *storage.Alert) {
alertNotifiers := set.NewStringSet(alert.GetPolicy().GetNotifiers()...)
p.ns.ForEach(ctx, func(ctx context.Context, notifier notifiers.Notifier, failures pkgNotifier.AlertSet) {
if alertNotifiers.Contains(notifier.ProtoNotifier().GetId()) {
err := pkgNotifier.TryToAlert(ctx, notifier, alert)
if err != nil {
failures.Add(alert)
}
}
})
}
// New returns a new Processor
func New(ns pkgNotifier.Set, reporter integrationhealth.Reporter) pkgNotifier.Processor {
return &processorImpl{
ns: ns,
reporter: reporter,
}
}