-
Notifications
You must be signed in to change notification settings - Fork 324
/
process_manager.go
92 lines (79 loc) · 2.68 KB
/
process_manager.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
// SPDX-License-Identifier: Apache-2.0
// Copyright Authors of Tetragon
package grpc
import (
"context"
"sync"
"github.com/cilium/tetragon/api/v1/tetragon"
"github.com/cilium/tetragon/pkg/eventcache"
"github.com/cilium/tetragon/pkg/logger"
"github.com/cilium/tetragon/pkg/metrics/eventmetrics"
"github.com/cilium/tetragon/pkg/option"
"github.com/cilium/tetragon/pkg/reader/node"
"github.com/cilium/tetragon/pkg/reader/notify"
"github.com/cilium/tetragon/pkg/rthooks"
"github.com/cilium/tetragon/pkg/sensors"
"github.com/cilium/tetragon/pkg/server"
"github.com/sirupsen/logrus"
)
// ProcessManager maintains a cache of processes from tetragon exec events.
type ProcessManager struct {
nodeName string
Server *server.Server
// synchronize access to the listeners map.
mux sync.Mutex
listeners map[server.Listener]struct{}
}
// NewProcessManager returns a pointer to an initialized ProcessManager struct.
func NewProcessManager(
ctx context.Context,
wg *sync.WaitGroup,
manager *sensors.Manager,
hookRunner *rthooks.Runner,
) (*ProcessManager, error) {
pm := &ProcessManager{
nodeName: node.GetNodeNameForExport(),
listeners: make(map[server.Listener]struct{}),
}
pm.Server = server.NewServer(ctx, wg, pm, manager, hookRunner)
// Exec cache is always needed to ensure events have an associated Process{}
eventcache.New(pm.Server)
logger.GetLogger().WithField("enableCilium", option.Config.EnableCilium).WithFields(logrus.Fields{
"enableK8s": option.Config.EnableK8s,
"enableProcessCred": option.Config.EnableProcessCred,
"enableProcessNs": option.Config.EnableProcessNs,
}).Info("Starting process manager")
return pm, nil
}
// Notify implements Listener.Notify.
func (pm *ProcessManager) Notify(event notify.Message) error {
processedEvent := event.HandleMessage()
if processedEvent != nil {
pm.NotifyListener(event, processedEvent)
}
return nil
}
// Close implements Listener.Close.
func (pm *ProcessManager) Close() error {
return nil
}
func (pm *ProcessManager) AddListener(listener server.Listener) {
logger.GetLogger().WithField("getEventsListener", listener).Debug("Adding a getEventsListener")
pm.mux.Lock()
defer pm.mux.Unlock()
pm.listeners[listener] = struct{}{}
}
func (pm *ProcessManager) RemoveListener(listener server.Listener) {
logger.GetLogger().WithField("getEventsListener", listener).Debug("Removing a getEventsListener")
pm.mux.Lock()
defer pm.mux.Unlock()
delete(pm.listeners, listener)
}
func (pm *ProcessManager) NotifyListener(original interface{}, processed *tetragon.GetEventsResponse) {
pm.mux.Lock()
defer pm.mux.Unlock()
for l := range pm.listeners {
l.Notify(processed)
}
eventmetrics.ProcessEvent(original, processed)
}