podman

1
//go:build systemd
2

3
package events
4

5
import (
6
	"context"
7
	"encoding/json"
8
	"errors"
9
	"fmt"
10
	"strconv"
11
	"time"
12

13
	"github.com/containers/podman/v5/pkg/rootless"
14
	"github.com/containers/podman/v5/pkg/util"
15
	"github.com/coreos/go-systemd/v22/journal"
16
	"github.com/coreos/go-systemd/v22/sdjournal"
17
	"github.com/sirupsen/logrus"
18
)
19

20
// DefaultEventerType is journald when systemd is available
21
const DefaultEventerType = Journald
22

23
// EventJournalD is the journald implementation of an eventer
24
type EventJournalD struct {
25
	options EventerOptions
26
}
27

28
// newEventJournalD creates a new journald Eventer
29
func newEventJournalD(options EventerOptions) (Eventer, error) {
30
	return EventJournalD{options}, nil
31
}
32

33
// Write to journald
34
func (e EventJournalD) Write(ee Event) error {
35
	m := make(map[string]string)
36
	m["SYSLOG_IDENTIFIER"] = "podman"
37
	m["PODMAN_EVENT"] = ee.Status.String()
38
	m["PODMAN_TYPE"] = ee.Type.String()
39
	m["PODMAN_TIME"] = ee.Time.Format(time.RFC3339Nano)
40

41
	// Add specialized information based on the podman type
42
	switch ee.Type {
43
	case Image:
44
		m["PODMAN_NAME"] = ee.Name
45
		m["PODMAN_ID"] = ee.ID
46
		if ee.Error != "" {
47
			m["ERROR"] = ee.Error
48
		}
49
	case Container, Pod:
50
		m["PODMAN_IMAGE"] = ee.Image
51
		m["PODMAN_NAME"] = ee.Name
52
		m["PODMAN_ID"] = ee.ID
53
		if ee.ContainerExitCode != nil {
54
			m["PODMAN_EXIT_CODE"] = strconv.Itoa(*ee.ContainerExitCode)
55
		}
56
		if ee.PodID != "" {
57
			m["PODMAN_POD_ID"] = ee.PodID
58
		}
59
		// If we have container labels, we need to convert them to a string so they
60
		// can be recorded with the event
61
		if len(ee.Details.Attributes) > 0 {
62
			b, err := json.Marshal(ee.Details.Attributes)
63
			if err != nil {
64
				return err
65
			}
66
			m["PODMAN_LABELS"] = string(b)
67
		}
68
		m["PODMAN_HEALTH_STATUS"] = ee.HealthStatus
69

70
		if len(ee.Details.ContainerInspectData) > 0 {
71
			m["PODMAN_CONTAINER_INSPECT_DATA"] = ee.Details.ContainerInspectData
72
		}
73
	case Network:
74
		m["PODMAN_ID"] = ee.ID
75
		m["PODMAN_NETWORK_NAME"] = ee.Network
76
	case Volume:
77
		m["PODMAN_NAME"] = ee.Name
78
	}
79

80
	// starting with commit 7e6e267329 we set LogLevel=notice for the systemd healthcheck unit
81
	// This so it doesn't log the started/stopped unit messages al the time which spam the
82
	// journal if a small interval is used. That however broke the healthcheck event as it no
83
	// longer showed up in podman events when running as root as we only send the event on info
84
	// level. To fix this we have to send the event on notice level.
85
	// https://github.com/containers/podman/issues/20342
86
	prio := journal.PriInfo
87
	if len(ee.HealthStatus) > 0 {
88
		prio = journal.PriNotice
89
	}
90

91
	return journal.Send(ee.ToHumanReadable(false), prio, m)
92
}
93

94
// Read reads events from the journal and sends qualified events to the event channel
95
func (e EventJournalD) Read(ctx context.Context, options ReadOptions) error {
96
	defer close(options.EventChannel)
97
	filterMap, err := generateEventFilters(options.Filters, options.Since, options.Until)
98
	if err != nil {
99
		return fmt.Errorf("failed to parse event filters: %w", err)
100
	}
101

102
	var untilTime time.Time
103
	if len(options.Until) > 0 {
104
		untilTime, err = util.ParseInputTime(options.Until, false)
105
		if err != nil {
106
			return err
107
		}
108
	}
109

110
	j, err := sdjournal.NewJournal()
111
	if err != nil {
112
		return err
113
	}
114
	defer func() {
115
		if err := j.Close(); err != nil {
116
			logrus.Errorf("Unable to close journal :%v", err)
117
		}
118
	}()
119
	err = j.SetDataThreshold(0)
120
	if err != nil {
121
		logrus.Warnf("cannot set data threshold: %v", err)
122
	}
123
	// match only podman journal entries
124
	podmanJournal := sdjournal.Match{Field: "SYSLOG_IDENTIFIER", Value: "podman"}
125
	if err := j.AddMatch(podmanJournal.String()); err != nil {
126
		return fmt.Errorf("failed to add SYSLOG_IDENTIFIER journal filter for event log: %w", err)
127
	}
128

129
	// make sure we only read events for the current user
130
	uidMatch := sdjournal.Match{Field: "_UID", Value: strconv.Itoa(rootless.GetRootlessUID())}
131
	if err := j.AddMatch(uidMatch.String()); err != nil {
132
		return fmt.Errorf("failed to add _UID journal filter for event log: %w", err)
133
	}
134

135
	if len(options.Since) == 0 && len(options.Until) == 0 && options.Stream {
136
		if err := j.SeekTail(); err != nil {
137
			return fmt.Errorf("failed to seek end of journal: %w", err)
138
		}
139
		// After SeekTail calling Next moves to a random entry.
140
		// To prevent this we have to call Previous first.
141
		// see: https://bugs.freedesktop.org/show_bug.cgi?id=64614
142
		if _, err := j.Previous(); err != nil {
143
			return fmt.Errorf("failed to move journal cursor to previous entry: %w", err)
144
		}
145
	} else if len(options.Since) > 0 {
146
		since, err := util.ParseInputTime(options.Since, true)
147
		if err != nil {
148
			return err
149
		}
150
		// seek based on time which helps to reduce unnecessary event reads
151
		if err := j.SeekRealtimeUsec(uint64(since.UnixMicro())); err != nil {
152
			return err
153
		}
154
	}
155

156
	for {
157
		entry, err := GetNextEntry(ctx, j, options.Stream, untilTime)
158
		if err != nil {
159
			return err
160
		}
161
		// no entry == we hit the end
162
		if entry == nil {
163
			return nil
164
		}
165

166
		newEvent, err := newEventFromJournalEntry(entry)
167
		if err != nil {
168
			// We can't decode this event.
169
			// Don't fail hard - that would make events unusable.
170
			// Instead, log and continue.
171
			if !errors.Is(err, ErrEventTypeBlank) {
172
				logrus.Errorf("Unable to decode event: %v", err)
173
			}
174
			continue
175
		}
176
		if applyFilters(newEvent, filterMap) {
177
			options.EventChannel <- newEvent
178
		}
179
	}
180
}
181

182
func newEventFromJournalEntry(entry *sdjournal.JournalEntry) (*Event, error) {
183
	newEvent := Event{}
184
	eventType, err := StringToType(entry.Fields["PODMAN_TYPE"])
185
	if err != nil {
186
		return nil, err
187
	}
188
	eventTime, err := time.Parse(time.RFC3339Nano, entry.Fields["PODMAN_TIME"])
189
	if err != nil {
190
		return nil, err
191
	}
192
	eventStatus, err := StringToStatus(entry.Fields["PODMAN_EVENT"])
193
	if err != nil {
194
		return nil, err
195
	}
196
	newEvent.Type = eventType
197
	newEvent.Time = eventTime
198
	newEvent.Status = eventStatus
199
	newEvent.Name = entry.Fields["PODMAN_NAME"]
200

201
	switch eventType {
202
	case Container, Pod:
203
		newEvent.ID = entry.Fields["PODMAN_ID"]
204
		newEvent.Image = entry.Fields["PODMAN_IMAGE"]
205
		newEvent.PodID = entry.Fields["PODMAN_POD_ID"]
206
		if code, ok := entry.Fields["PODMAN_EXIT_CODE"]; ok {
207
			intCode, err := strconv.Atoi(code)
208
			if err != nil {
209
				logrus.Errorf("Parsing event exit code %s", code)
210
			} else {
211
				newEvent.ContainerExitCode = &intCode
212
			}
213
		}
214

215
		// we need to check for the presence of labels recorded to a container event
216
		if stringLabels, ok := entry.Fields["PODMAN_LABELS"]; ok && len(stringLabels) > 0 {
217
			labels := make(map[string]string, 0)
218
			if err := json.Unmarshal([]byte(stringLabels), &labels); err != nil {
219
				return nil, err
220
			}
221

222
			// if we have labels, add them to the event
223
			if len(labels) > 0 {
224
				newEvent.Attributes = labels
225
			}
226
		}
227
		newEvent.HealthStatus = entry.Fields["PODMAN_HEALTH_STATUS"]
228
		newEvent.Details.ContainerInspectData = entry.Fields["PODMAN_CONTAINER_INSPECT_DATA"]
229
	case Network:
230
		newEvent.ID = entry.Fields["PODMAN_ID"]
231
		newEvent.Network = entry.Fields["PODMAN_NETWORK_NAME"]
232
	case Image:
233
		newEvent.ID = entry.Fields["PODMAN_ID"]
234
		if val, ok := entry.Fields["ERROR"]; ok {
235
			newEvent.Error = val
236
		}
237
	}
238
	return &newEvent, nil
239
}
240

241
// String returns a string representation of the logger
242
func (e EventJournalD) String() string {
243
	return Journald.String()
244
}
245

246
// GetNextEntry returns the next entry in the journal. If the end of the
247
// journal is reached and stream is not set or the current time is after
248
// the until time this function returns nil,nil.
249
func GetNextEntry(ctx context.Context, j *sdjournal.Journal, stream bool, untilTime time.Time) (*sdjournal.JournalEntry, error) {
250
	for {
251
		select {
252
		case <-ctx.Done():
253
			// the consumer has cancelled
254
			return nil, nil
255
		default:
256
			// fallthrough
257
		}
258
		// the api requires a next|prev before reading the event
259
		ret, err := j.Next()
260
		if err != nil {
261
			return nil, fmt.Errorf("failed to move journal cursor to next entry: %w", err)
262
		}
263
		// ret == 0 equals EOF, see sd_journal_next(3)
264
		if ret == 0 {
265
			if !stream || (!untilTime.IsZero() && time.Now().After(untilTime)) {
266
				// we hit the end and should not keep streaming
267
				return nil, nil
268
			}
269
			// keep waiting for the next entry
270
			// j.Wait() is blocking, this would cause the goroutine to hang forever
271
			// if no more journal entries are generated and thus if the client
272
			// has closed the connection in the meantime to leak memory.
273
			// Waiting only 5 seconds makes sure we can check if the client closed in the
274
			// meantime at least every 5 seconds.
275
			t := 5 * time.Second
276
			if !untilTime.IsZero() {
277
				until := time.Until(untilTime)
278
				if until < t {
279
					t = until
280
				}
281
			}
282
			_ = j.Wait(t)
283
			continue
284
		}
285

286
		entry, err := j.GetEntry()
287
		if err != nil {
288
			return nil, fmt.Errorf("failed to read journal entry: %w", err)
289
		}
290
		return entry, nil
291
	}
292
}
293