Line data Source code
1 : // Copyright 2018 The LevelDB-Go and Pebble Authors. All rights reserved. Use
2 : // of this source code is governed by a BSD-style license that can be found in
3 : // the LICENSE file.
4 :
5 : package pebble
6 :
7 : import (
8 : "fmt"
9 : "strings"
10 : "time"
11 :
12 : "github.com/cockroachdb/errors"
13 : "github.com/cockroachdb/pebble/internal/base"
14 : "github.com/cockroachdb/pebble/internal/humanize"
15 : "github.com/cockroachdb/pebble/internal/invariants"
16 : "github.com/cockroachdb/pebble/internal/manifest"
17 : "github.com/cockroachdb/pebble/vfs"
18 : "github.com/cockroachdb/redact"
19 : )
20 :
21 : // TableInfo exports the manifest.TableInfo type.
22 : type TableInfo = manifest.TableInfo
23 :
24 1 : func tablesTotalSize(tables []TableInfo) uint64 {
25 1 : var size uint64
26 1 : for i := range tables {
27 1 : size += tables[i].Size
28 1 : }
29 1 : return size
30 : }
31 :
32 1 : func formatFileNums(tables []TableInfo) string {
33 1 : var buf strings.Builder
34 1 : for i := range tables {
35 1 : if i > 0 {
36 1 : buf.WriteString(" ")
37 1 : }
38 1 : buf.WriteString(tables[i].FileNum.String())
39 : }
40 1 : return buf.String()
41 : }
42 :
43 : // LevelInfo contains info pertaining to a particular level.
44 : type LevelInfo struct {
45 : Level int
46 : Tables []TableInfo
47 : Score float64
48 : }
49 :
50 0 : func (i LevelInfo) String() string {
51 0 : return redact.StringWithoutMarkers(i)
52 0 : }
53 :
54 : // SafeFormat implements redact.SafeFormatter.
55 1 : func (i LevelInfo) SafeFormat(w redact.SafePrinter, _ rune) {
56 1 : w.Printf("L%d [%s] (%s) Score=%.2f",
57 1 : redact.Safe(i.Level),
58 1 : redact.Safe(formatFileNums(i.Tables)),
59 1 : redact.Safe(humanize.Bytes.Uint64(tablesTotalSize(i.Tables))),
60 1 : redact.Safe(i.Score))
61 1 : }
62 :
63 : // CompactionInfo contains the info for a compaction event.
64 : type CompactionInfo struct {
65 : // JobID is the ID of the compaction job.
66 : JobID int
67 : // Reason is the reason for the compaction.
68 : Reason string
69 : // Input contains the input tables for the compaction organized by level.
70 : Input []LevelInfo
71 : // Output contains the output tables generated by the compaction. The output
72 : // tables are empty for the compaction begin event.
73 : Output LevelInfo
74 : // Duration is the time spent compacting, including reading and writing
75 : // sstables.
76 : Duration time.Duration
77 : // TotalDuration is the total wall-time duration of the compaction,
78 : // including applying the compaction to the database. TotalDuration is
79 : // always ≥ Duration.
80 : TotalDuration time.Duration
81 : Done bool
82 : Err error
83 :
84 : SingleLevelOverlappingRatio float64
85 : MultiLevelOverlappingRatio float64
86 :
87 : // Annotations specifies additional info to appear in a compaction's event log line
88 : Annotations compactionAnnotations
89 : }
90 :
91 : type compactionAnnotations []string
92 :
93 : // SafeFormat implements redact.SafeFormatter.
94 1 : func (ca compactionAnnotations) SafeFormat(w redact.SafePrinter, _ rune) {
95 1 : if len(ca) == 0 {
96 1 : return
97 1 : }
98 1 : for i := range ca {
99 1 : if i != 0 {
100 0 : w.Print(" ")
101 0 : }
102 1 : w.Printf("%s", redact.SafeString(ca[i]))
103 : }
104 : }
105 :
106 1 : func (i CompactionInfo) String() string {
107 1 : return redact.StringWithoutMarkers(i)
108 1 : }
109 :
110 : // SafeFormat implements redact.SafeFormatter.
111 1 : func (i CompactionInfo) SafeFormat(w redact.SafePrinter, _ rune) {
112 1 : if i.Err != nil {
113 1 : w.Printf("[JOB %d] compaction(%s) to L%d error: %s",
114 1 : redact.Safe(i.JobID), redact.SafeString(i.Reason), redact.Safe(i.Output.Level), i.Err)
115 1 : return
116 1 : }
117 :
118 1 : if !i.Done {
119 1 : w.Printf("[JOB %d] compacting(%s) ",
120 1 : redact.Safe(i.JobID),
121 1 : redact.SafeString(i.Reason))
122 1 : w.Printf("%s", i.Annotations)
123 1 : w.Printf("%s; ", levelInfos(i.Input))
124 1 : w.Printf("OverlappingRatio: Single %.2f, Multi %.2f", i.SingleLevelOverlappingRatio, i.MultiLevelOverlappingRatio)
125 1 : return
126 1 : }
127 1 : outputSize := tablesTotalSize(i.Output.Tables)
128 1 : w.Printf("[JOB %d] compacted(%s) ", redact.Safe(i.JobID), redact.SafeString(i.Reason))
129 1 : w.Printf("%s", i.Annotations)
130 1 : w.Print(levelInfos(i.Input))
131 1 : w.Printf(" -> L%d [%s] (%s), in %.1fs (%.1fs total), output rate %s/s",
132 1 : redact.Safe(i.Output.Level),
133 1 : redact.Safe(formatFileNums(i.Output.Tables)),
134 1 : redact.Safe(humanize.Bytes.Uint64(outputSize)),
135 1 : redact.Safe(i.Duration.Seconds()),
136 1 : redact.Safe(i.TotalDuration.Seconds()),
137 1 : redact.Safe(humanize.Bytes.Uint64(uint64(float64(outputSize)/i.Duration.Seconds()))))
138 : }
139 :
140 : type levelInfos []LevelInfo
141 :
142 1 : func (i levelInfos) SafeFormat(w redact.SafePrinter, _ rune) {
143 1 : for j, levelInfo := range i {
144 1 : if j > 0 {
145 1 : w.Printf(" + ")
146 1 : }
147 1 : w.Print(levelInfo)
148 : }
149 : }
150 :
151 : // DiskSlowInfo contains the info for a disk slowness event when writing to a
152 : // file.
153 : type DiskSlowInfo = vfs.DiskSlowInfo
154 :
155 : // FlushInfo contains the info for a flush event.
156 : type FlushInfo struct {
157 : // JobID is the ID of the flush job.
158 : JobID int
159 : // Reason is the reason for the flush.
160 : Reason string
161 : // Input contains the count of input memtables that were flushed.
162 : Input int
163 : // InputBytes contains the total in-memory size of the memtable(s) that were
164 : // flushed. This size includes skiplist indexing data structures.
165 : InputBytes uint64
166 : // Output contains the ouptut table generated by the flush. The output info
167 : // is empty for the flush begin event.
168 : Output []TableInfo
169 : // Duration is the time spent flushing. This duration includes writing and
170 : // syncing all of the flushed keys to sstables.
171 : Duration time.Duration
172 : // TotalDuration is the total wall-time duration of the flush, including
173 : // applying the flush to the database. TotalDuration is always ≥ Duration.
174 : TotalDuration time.Duration
175 : // Ingest is set to true if the flush is handling tables that were added to
176 : // the flushable queue via an ingestion operation.
177 : Ingest bool
178 : // IngestLevels are the output levels for each ingested table in the flush.
179 : // This field is only populated when Ingest is true.
180 : IngestLevels []int
181 : Done bool
182 : Err error
183 : }
184 :
185 1 : func (i FlushInfo) String() string {
186 1 : return redact.StringWithoutMarkers(i)
187 1 : }
188 :
189 : // SafeFormat implements redact.SafeFormatter.
190 1 : func (i FlushInfo) SafeFormat(w redact.SafePrinter, _ rune) {
191 1 : if i.Err != nil {
192 1 : w.Printf("[JOB %d] flush error: %s", redact.Safe(i.JobID), i.Err)
193 1 : return
194 1 : }
195 :
196 1 : plural := redact.SafeString("s")
197 1 : if i.Input == 1 {
198 1 : plural = ""
199 1 : }
200 1 : if !i.Done {
201 1 : w.Printf("[JOB %d] ", redact.Safe(i.JobID))
202 1 : if !i.Ingest {
203 1 : w.Printf("flushing %d memtable", redact.Safe(i.Input))
204 1 : w.SafeString(plural)
205 1 : w.Printf(" (%s) to L0", redact.Safe(humanize.Bytes.Uint64(i.InputBytes)))
206 1 : } else {
207 1 : w.Printf("flushing %d ingested table%s", redact.Safe(i.Input), plural)
208 1 : }
209 1 : return
210 : }
211 :
212 1 : outputSize := tablesTotalSize(i.Output)
213 1 : if !i.Ingest {
214 1 : if invariants.Enabled && len(i.IngestLevels) > 0 {
215 0 : panic(errors.AssertionFailedf("pebble: expected len(IngestedLevels) == 0"))
216 : }
217 1 : w.Printf("[JOB %d] flushed %d memtable%s (%s) to L0 [%s] (%s), in %.1fs (%.1fs total), output rate %s/s",
218 1 : redact.Safe(i.JobID), redact.Safe(i.Input), plural,
219 1 : redact.Safe(humanize.Bytes.Uint64(i.InputBytes)),
220 1 : redact.Safe(formatFileNums(i.Output)),
221 1 : redact.Safe(humanize.Bytes.Uint64(outputSize)),
222 1 : redact.Safe(i.Duration.Seconds()),
223 1 : redact.Safe(i.TotalDuration.Seconds()),
224 1 : redact.Safe(humanize.Bytes.Uint64(uint64(float64(outputSize)/i.Duration.Seconds()))))
225 1 : } else {
226 1 : if invariants.Enabled && len(i.IngestLevels) == 0 {
227 0 : panic(errors.AssertionFailedf("pebble: expected len(IngestedLevels) > 0"))
228 : }
229 1 : w.Printf("[JOB %d] flushed %d ingested flushable%s",
230 1 : redact.Safe(i.JobID), redact.Safe(len(i.Output)), plural)
231 1 : for j, level := range i.IngestLevels {
232 1 : file := i.Output[j]
233 1 : if j > 0 {
234 1 : w.Printf(" +")
235 1 : }
236 1 : w.Printf(" L%d:%s (%s)", level, file.FileNum, humanize.Bytes.Uint64(file.Size))
237 : }
238 1 : w.Printf(" in %.1fs (%.1fs total), output rate %s/s",
239 1 : redact.Safe(i.Duration.Seconds()),
240 1 : redact.Safe(i.TotalDuration.Seconds()),
241 1 : redact.Safe(humanize.Bytes.Uint64(uint64(float64(outputSize)/i.Duration.Seconds()))))
242 : }
243 : }
244 :
245 : // ManifestCreateInfo contains info about a manifest creation event.
246 : type ManifestCreateInfo struct {
247 : // JobID is the ID of the job the caused the manifest to be created.
248 : JobID int
249 : Path string
250 : // The file number of the new Manifest.
251 : FileNum base.DiskFileNum
252 : Err error
253 : }
254 :
255 1 : func (i ManifestCreateInfo) String() string {
256 1 : return redact.StringWithoutMarkers(i)
257 1 : }
258 :
259 : // SafeFormat implements redact.SafeFormatter.
260 1 : func (i ManifestCreateInfo) SafeFormat(w redact.SafePrinter, _ rune) {
261 1 : if i.Err != nil {
262 0 : w.Printf("[JOB %d] MANIFEST create error: %s", redact.Safe(i.JobID), i.Err)
263 0 : return
264 0 : }
265 1 : w.Printf("[JOB %d] MANIFEST created %s", redact.Safe(i.JobID), i.FileNum)
266 : }
267 :
268 : // ManifestDeleteInfo contains the info for a Manifest deletion event.
269 : type ManifestDeleteInfo struct {
270 : // JobID is the ID of the job the caused the Manifest to be deleted.
271 : JobID int
272 : Path string
273 : FileNum FileNum
274 : Err error
275 : }
276 :
277 1 : func (i ManifestDeleteInfo) String() string {
278 1 : return redact.StringWithoutMarkers(i)
279 1 : }
280 :
281 : // SafeFormat implements redact.SafeFormatter.
282 1 : func (i ManifestDeleteInfo) SafeFormat(w redact.SafePrinter, _ rune) {
283 1 : if i.Err != nil {
284 0 : w.Printf("[JOB %d] MANIFEST delete error: %s", redact.Safe(i.JobID), i.Err)
285 0 : return
286 0 : }
287 1 : w.Printf("[JOB %d] MANIFEST deleted %s", redact.Safe(i.JobID), i.FileNum)
288 : }
289 :
290 : // TableCreateInfo contains the info for a table creation event.
291 : type TableCreateInfo struct {
292 : JobID int
293 : // Reason is the reason for the table creation: "compacting", "flushing", or
294 : // "ingesting".
295 : Reason string
296 : Path string
297 : FileNum FileNum
298 : }
299 :
300 1 : func (i TableCreateInfo) String() string {
301 1 : return redact.StringWithoutMarkers(i)
302 1 : }
303 :
304 : // SafeFormat implements redact.SafeFormatter.
305 1 : func (i TableCreateInfo) SafeFormat(w redact.SafePrinter, _ rune) {
306 1 : w.Printf("[JOB %d] %s: sstable created %s",
307 1 : redact.Safe(i.JobID), redact.Safe(i.Reason), i.FileNum)
308 1 : }
309 :
310 : // TableDeleteInfo contains the info for a table deletion event.
311 : type TableDeleteInfo struct {
312 : JobID int
313 : Path string
314 : FileNum FileNum
315 : Err error
316 : }
317 :
318 1 : func (i TableDeleteInfo) String() string {
319 1 : return redact.StringWithoutMarkers(i)
320 1 : }
321 :
322 : // SafeFormat implements redact.SafeFormatter.
323 1 : func (i TableDeleteInfo) SafeFormat(w redact.SafePrinter, _ rune) {
324 1 : if i.Err != nil {
325 0 : w.Printf("[JOB %d] sstable delete error %s: %s",
326 0 : redact.Safe(i.JobID), i.FileNum, i.Err)
327 0 : return
328 0 : }
329 1 : w.Printf("[JOB %d] sstable deleted %s", redact.Safe(i.JobID), i.FileNum)
330 : }
331 :
332 : // TableIngestInfo contains the info for a table ingestion event.
333 : type TableIngestInfo struct {
334 : // JobID is the ID of the job the caused the table to be ingested.
335 : JobID int
336 : Tables []struct {
337 : TableInfo
338 : Level int
339 : }
340 : // GlobalSeqNum is the sequence number that was assigned to all entries in
341 : // the ingested table.
342 : GlobalSeqNum uint64
343 : // flushable indicates whether the ingested sstable was treated as a
344 : // flushable.
345 : flushable bool
346 : Err error
347 : }
348 :
349 1 : func (i TableIngestInfo) String() string {
350 1 : return redact.StringWithoutMarkers(i)
351 1 : }
352 :
353 : // SafeFormat implements redact.SafeFormatter.
354 1 : func (i TableIngestInfo) SafeFormat(w redact.SafePrinter, _ rune) {
355 1 : if i.Err != nil {
356 0 : w.Printf("[JOB %d] ingest error: %s", redact.Safe(i.JobID), i.Err)
357 0 : return
358 0 : }
359 :
360 1 : if i.flushable {
361 1 : w.Printf("[JOB %d] ingested as flushable", redact.Safe(i.JobID))
362 1 : } else {
363 1 : w.Printf("[JOB %d] ingested", redact.Safe(i.JobID))
364 1 : }
365 :
366 1 : for j := range i.Tables {
367 1 : t := &i.Tables[j]
368 1 : if j > 0 {
369 1 : w.Printf(",")
370 1 : }
371 1 : levelStr := ""
372 1 : if !i.flushable {
373 1 : levelStr = fmt.Sprintf("L%d:", t.Level)
374 1 : }
375 1 : w.Printf(" %s%s (%s)", redact.Safe(levelStr), t.FileNum,
376 1 : redact.Safe(humanize.Bytes.Uint64(t.Size)))
377 : }
378 : }
379 :
380 : // TableStatsInfo contains the info for a table stats loaded event.
381 : type TableStatsInfo struct {
382 : // JobID is the ID of the job that finished loading the initial tables'
383 : // stats.
384 : JobID int
385 : }
386 :
387 1 : func (i TableStatsInfo) String() string {
388 1 : return redact.StringWithoutMarkers(i)
389 1 : }
390 :
391 : // SafeFormat implements redact.SafeFormatter.
392 1 : func (i TableStatsInfo) SafeFormat(w redact.SafePrinter, _ rune) {
393 1 : w.Printf("[JOB %d] all initial table stats loaded", redact.Safe(i.JobID))
394 1 : }
395 :
396 : // TableValidatedInfo contains information on the result of a validation run
397 : // on an sstable.
398 : type TableValidatedInfo struct {
399 : JobID int
400 : Meta *fileMetadata
401 : }
402 :
403 1 : func (i TableValidatedInfo) String() string {
404 1 : return redact.StringWithoutMarkers(i)
405 1 : }
406 :
407 : // SafeFormat implements redact.SafeFormatter.
408 1 : func (i TableValidatedInfo) SafeFormat(w redact.SafePrinter, _ rune) {
409 1 : w.Printf("[JOB %d] validated table: %s", redact.Safe(i.JobID), i.Meta)
410 1 : }
411 :
412 : // WALCreateInfo contains info about a WAL creation event.
413 : type WALCreateInfo struct {
414 : // JobID is the ID of the job the caused the WAL to be created.
415 : JobID int
416 : Path string
417 : // The file number of the new WAL.
418 : FileNum base.DiskFileNum
419 : // The file number of a previous WAL which was recycled to create this
420 : // one. Zero if recycling did not take place.
421 : RecycledFileNum FileNum
422 : Err error
423 : }
424 :
425 1 : func (i WALCreateInfo) String() string {
426 1 : return redact.StringWithoutMarkers(i)
427 1 : }
428 :
429 : // SafeFormat implements redact.SafeFormatter.
430 1 : func (i WALCreateInfo) SafeFormat(w redact.SafePrinter, _ rune) {
431 1 : if i.Err != nil {
432 0 : w.Printf("[JOB %d] WAL create error: %s", redact.Safe(i.JobID), i.Err)
433 0 : return
434 0 : }
435 :
436 1 : if i.RecycledFileNum == 0 {
437 1 : w.Printf("[JOB %d] WAL created %s", redact.Safe(i.JobID), i.FileNum)
438 1 : return
439 1 : }
440 :
441 0 : w.Printf("[JOB %d] WAL created %s (recycled %s)",
442 0 : redact.Safe(i.JobID), i.FileNum, i.RecycledFileNum)
443 : }
444 :
445 : // WALDeleteInfo contains the info for a WAL deletion event.
446 : type WALDeleteInfo struct {
447 : // JobID is the ID of the job the caused the WAL to be deleted.
448 : JobID int
449 : Path string
450 : FileNum FileNum
451 : Err error
452 : }
453 :
454 1 : func (i WALDeleteInfo) String() string {
455 1 : return redact.StringWithoutMarkers(i)
456 1 : }
457 :
458 : // SafeFormat implements redact.SafeFormatter.
459 1 : func (i WALDeleteInfo) SafeFormat(w redact.SafePrinter, _ rune) {
460 1 : if i.Err != nil {
461 0 : w.Printf("[JOB %d] WAL delete error: %s", redact.Safe(i.JobID), i.Err)
462 0 : return
463 0 : }
464 1 : w.Printf("[JOB %d] WAL deleted %s", redact.Safe(i.JobID), i.FileNum)
465 : }
466 :
467 : // WriteStallBeginInfo contains the info for a write stall begin event.
468 : type WriteStallBeginInfo struct {
469 : Reason string
470 : }
471 :
472 1 : func (i WriteStallBeginInfo) String() string {
473 1 : return redact.StringWithoutMarkers(i)
474 1 : }
475 :
476 : // SafeFormat implements redact.SafeFormatter.
477 1 : func (i WriteStallBeginInfo) SafeFormat(w redact.SafePrinter, _ rune) {
478 1 : w.Printf("write stall beginning: %s", redact.Safe(i.Reason))
479 1 : }
480 :
481 : // EventListener contains a set of functions that will be invoked when various
482 : // significant DB events occur. Note that the functions should not run for an
483 : // excessive amount of time as they are invoked synchronously by the DB and may
484 : // block continued DB work. For a similar reason it is advisable to not perform
485 : // any synchronous calls back into the DB.
486 : type EventListener struct {
487 : // BackgroundError is invoked whenever an error occurs during a background
488 : // operation such as flush or compaction.
489 : BackgroundError func(error)
490 :
491 : // CompactionBegin is invoked after the inputs to a compaction have been
492 : // determined, but before the compaction has produced any output.
493 : CompactionBegin func(CompactionInfo)
494 :
495 : // CompactionEnd is invoked after a compaction has completed and the result
496 : // has been installed.
497 : CompactionEnd func(CompactionInfo)
498 :
499 : // DiskSlow is invoked after a disk write operation on a file created with a
500 : // disk health checking vfs.FS (see vfs.DefaultWithDiskHealthChecks) is
501 : // observed to exceed the specified disk slowness threshold duration. DiskSlow
502 : // is called on a goroutine that is monitoring slowness/stuckness. The callee
503 : // MUST return without doing any IO, or blocking on anything (like a mutex)
504 : // that is waiting on IO. This is imperative in order to reliably monitor for
505 : // slowness, since if this goroutine gets stuck, the monitoring will stop
506 : // working.
507 : DiskSlow func(DiskSlowInfo)
508 :
509 : // FlushBegin is invoked after the inputs to a flush have been determined,
510 : // but before the flush has produced any output.
511 : FlushBegin func(FlushInfo)
512 :
513 : // FlushEnd is invoked after a flush has complated and the result has been
514 : // installed.
515 : FlushEnd func(FlushInfo)
516 :
517 : // FormatUpgrade is invoked after the database's FormatMajorVersion
518 : // is upgraded.
519 : FormatUpgrade func(FormatMajorVersion)
520 :
521 : // ManifestCreated is invoked after a manifest has been created.
522 : ManifestCreated func(ManifestCreateInfo)
523 :
524 : // ManifestDeleted is invoked after a manifest has been deleted.
525 : ManifestDeleted func(ManifestDeleteInfo)
526 :
527 : // TableCreated is invoked when a table has been created.
528 : TableCreated func(TableCreateInfo)
529 :
530 : // TableDeleted is invoked after a table has been deleted.
531 : TableDeleted func(TableDeleteInfo)
532 :
533 : // TableIngested is invoked after an externally created table has been
534 : // ingested via a call to DB.Ingest().
535 : TableIngested func(TableIngestInfo)
536 :
537 : // TableStatsLoaded is invoked at most once, when the table stats
538 : // collector has loaded statistics for all tables that existed at Open.
539 : TableStatsLoaded func(TableStatsInfo)
540 :
541 : // TableValidated is invoked after validation runs on an sstable.
542 : TableValidated func(TableValidatedInfo)
543 :
544 : // WALCreated is invoked after a WAL has been created.
545 : WALCreated func(WALCreateInfo)
546 :
547 : // WALDeleted is invoked after a WAL has been deleted.
548 : WALDeleted func(WALDeleteInfo)
549 :
550 : // WriteStallBegin is invoked when writes are intentionally delayed.
551 : WriteStallBegin func(WriteStallBeginInfo)
552 :
553 : // WriteStallEnd is invoked when delayed writes are released.
554 : WriteStallEnd func()
555 : }
556 :
557 : // EnsureDefaults ensures that background error events are logged to the
558 : // specified logger if a handler for those events hasn't been otherwise
559 : // specified. Ensure all handlers are non-nil so that we don't have to check
560 : // for nil-ness before invoking.
561 1 : func (l *EventListener) EnsureDefaults(logger Logger) {
562 1 : if l.BackgroundError == nil {
563 1 : if logger != nil {
564 1 : l.BackgroundError = func(err error) {
565 0 : logger.Errorf("background error: %s", err)
566 0 : }
567 0 : } else {
568 0 : l.BackgroundError = func(error) {}
569 : }
570 : }
571 1 : if l.CompactionBegin == nil {
572 1 : l.CompactionBegin = func(info CompactionInfo) {}
573 : }
574 1 : if l.CompactionEnd == nil {
575 1 : l.CompactionEnd = func(info CompactionInfo) {}
576 : }
577 1 : if l.DiskSlow == nil {
578 1 : l.DiskSlow = func(info DiskSlowInfo) {}
579 : }
580 1 : if l.FlushBegin == nil {
581 1 : l.FlushBegin = func(info FlushInfo) {}
582 : }
583 1 : if l.FlushEnd == nil {
584 1 : l.FlushEnd = func(info FlushInfo) {}
585 : }
586 1 : if l.FormatUpgrade == nil {
587 1 : l.FormatUpgrade = func(v FormatMajorVersion) {}
588 : }
589 1 : if l.ManifestCreated == nil {
590 1 : l.ManifestCreated = func(info ManifestCreateInfo) {}
591 : }
592 1 : if l.ManifestDeleted == nil {
593 1 : l.ManifestDeleted = func(info ManifestDeleteInfo) {}
594 : }
595 1 : if l.TableCreated == nil {
596 1 : l.TableCreated = func(info TableCreateInfo) {}
597 : }
598 1 : if l.TableDeleted == nil {
599 1 : l.TableDeleted = func(info TableDeleteInfo) {}
600 : }
601 1 : if l.TableIngested == nil {
602 1 : l.TableIngested = func(info TableIngestInfo) {}
603 : }
604 1 : if l.TableStatsLoaded == nil {
605 1 : l.TableStatsLoaded = func(info TableStatsInfo) {}
606 : }
607 1 : if l.TableValidated == nil {
608 1 : l.TableValidated = func(validated TableValidatedInfo) {}
609 : }
610 1 : if l.WALCreated == nil {
611 1 : l.WALCreated = func(info WALCreateInfo) {}
612 : }
613 1 : if l.WALDeleted == nil {
614 1 : l.WALDeleted = func(info WALDeleteInfo) {}
615 : }
616 1 : if l.WriteStallBegin == nil {
617 1 : l.WriteStallBegin = func(info WriteStallBeginInfo) {}
618 : }
619 1 : if l.WriteStallEnd == nil {
620 1 : l.WriteStallEnd = func() {}
621 : }
622 : }
623 :
624 : // MakeLoggingEventListener creates an EventListener that logs all events to the
625 : // specified logger.
626 1 : func MakeLoggingEventListener(logger Logger) EventListener {
627 1 : if logger == nil {
628 0 : logger = DefaultLogger
629 0 : }
630 :
631 1 : return EventListener{
632 1 : BackgroundError: func(err error) {
633 0 : logger.Errorf("background error: %s", err)
634 0 : },
635 1 : CompactionBegin: func(info CompactionInfo) {
636 1 : logger.Infof("%s", info)
637 1 : },
638 0 : CompactionEnd: func(info CompactionInfo) {
639 0 : logger.Infof("%s", info)
640 0 : },
641 0 : DiskSlow: func(info DiskSlowInfo) {
642 0 : logger.Infof("%s", info)
643 0 : },
644 1 : FlushBegin: func(info FlushInfo) {
645 1 : logger.Infof("%s", info)
646 1 : },
647 0 : FlushEnd: func(info FlushInfo) {
648 0 : logger.Infof("%s", info)
649 0 : },
650 1 : FormatUpgrade: func(v FormatMajorVersion) {
651 1 : logger.Infof("upgraded to format version: %s", v)
652 1 : },
653 0 : ManifestCreated: func(info ManifestCreateInfo) {
654 0 : logger.Infof("%s", info)
655 0 : },
656 0 : ManifestDeleted: func(info ManifestDeleteInfo) {
657 0 : logger.Infof("%s", info)
658 0 : },
659 1 : TableCreated: func(info TableCreateInfo) {
660 1 : logger.Infof("%s", info)
661 1 : },
662 0 : TableDeleted: func(info TableDeleteInfo) {
663 0 : logger.Infof("%s", info)
664 0 : },
665 0 : TableIngested: func(info TableIngestInfo) {
666 0 : logger.Infof("%s", info)
667 0 : },
668 1 : TableStatsLoaded: func(info TableStatsInfo) {
669 1 : logger.Infof("%s", info)
670 1 : },
671 1 : TableValidated: func(info TableValidatedInfo) {
672 1 : logger.Infof("%s", info)
673 1 : },
674 0 : WALCreated: func(info WALCreateInfo) {
675 0 : logger.Infof("%s", info)
676 0 : },
677 0 : WALDeleted: func(info WALDeleteInfo) {
678 0 : logger.Infof("%s", info)
679 0 : },
680 1 : WriteStallBegin: func(info WriteStallBeginInfo) {
681 1 : logger.Infof("%s", info)
682 1 : },
683 1 : WriteStallEnd: func() {
684 1 : logger.Infof("write stall ending")
685 1 : },
686 : }
687 : }
688 :
689 : // TeeEventListener wraps two EventListeners, forwarding all events to both.
690 0 : func TeeEventListener(a, b EventListener) EventListener {
691 0 : a.EnsureDefaults(nil)
692 0 : b.EnsureDefaults(nil)
693 0 : return EventListener{
694 0 : BackgroundError: func(err error) {
695 0 : a.BackgroundError(err)
696 0 : b.BackgroundError(err)
697 0 : },
698 0 : CompactionBegin: func(info CompactionInfo) {
699 0 : a.CompactionBegin(info)
700 0 : b.CompactionBegin(info)
701 0 : },
702 0 : CompactionEnd: func(info CompactionInfo) {
703 0 : a.CompactionEnd(info)
704 0 : b.CompactionEnd(info)
705 0 : },
706 0 : DiskSlow: func(info DiskSlowInfo) {
707 0 : a.DiskSlow(info)
708 0 : b.DiskSlow(info)
709 0 : },
710 0 : FlushBegin: func(info FlushInfo) {
711 0 : a.FlushBegin(info)
712 0 : b.FlushBegin(info)
713 0 : },
714 0 : FlushEnd: func(info FlushInfo) {
715 0 : a.FlushEnd(info)
716 0 : b.FlushEnd(info)
717 0 : },
718 0 : FormatUpgrade: func(v FormatMajorVersion) {
719 0 : a.FormatUpgrade(v)
720 0 : b.FormatUpgrade(v)
721 0 : },
722 0 : ManifestCreated: func(info ManifestCreateInfo) {
723 0 : a.ManifestCreated(info)
724 0 : b.ManifestCreated(info)
725 0 : },
726 0 : ManifestDeleted: func(info ManifestDeleteInfo) {
727 0 : a.ManifestDeleted(info)
728 0 : b.ManifestDeleted(info)
729 0 : },
730 0 : TableCreated: func(info TableCreateInfo) {
731 0 : a.TableCreated(info)
732 0 : b.TableCreated(info)
733 0 : },
734 0 : TableDeleted: func(info TableDeleteInfo) {
735 0 : a.TableDeleted(info)
736 0 : b.TableDeleted(info)
737 0 : },
738 0 : TableIngested: func(info TableIngestInfo) {
739 0 : a.TableIngested(info)
740 0 : b.TableIngested(info)
741 0 : },
742 0 : TableStatsLoaded: func(info TableStatsInfo) {
743 0 : a.TableStatsLoaded(info)
744 0 : b.TableStatsLoaded(info)
745 0 : },
746 0 : TableValidated: func(info TableValidatedInfo) {
747 0 : a.TableValidated(info)
748 0 : b.TableValidated(info)
749 0 : },
750 0 : WALCreated: func(info WALCreateInfo) {
751 0 : a.WALCreated(info)
752 0 : b.WALCreated(info)
753 0 : },
754 0 : WALDeleted: func(info WALDeleteInfo) {
755 0 : a.WALDeleted(info)
756 0 : b.WALDeleted(info)
757 0 : },
758 0 : WriteStallBegin: func(info WriteStallBeginInfo) {
759 0 : a.WriteStallBegin(info)
760 0 : b.WriteStallBegin(info)
761 0 : },
762 0 : WriteStallEnd: func() {
763 0 : a.WriteStallEnd()
764 0 : b.WriteStallEnd()
765 0 : },
766 : }
767 : }
|
