metropolis/pkg/logtree/journal.go

// Copyright 2020 The Monogon Project Authors.
//
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package logtree

import (
	"errors"
	"strings"
	"sync"
)

// DN is the Distinguished Name, a dot-delimited path used to address loggers
// within a LogTree. For example, "foo.bar" designates the 'bar' logger node
// under the 'foo' logger node under the root node of the logger. An empty
// string is the root node of the tree.
type DN string

var (
	ErrInvalidDN = errors.New("invalid DN")
)

// Path return the parts of a DN, ie. all the elements of the dot-delimited DN
// path.  For the root node, an empty list will be returned. An error will be
// returned if the DN is invalid (contains empty parts, eg. `foo..bar`, `.foo`
// or `foo.`.
func (d DN) Path() ([]string, error) {
	if d == "" {
		return nil, nil
	}
	parts := strings.Split(string(d), ".")
	for _, p := range parts {
		if p == "" {
			return nil, ErrInvalidDN
		}
	}
	return parts, nil
}

// journal is the main log recording structure of logtree. It manages linked lists
// containing the actual log entries, and implements scans across them. It does not
// understand the hierarchical nature of logtree, and instead sees all entries as
// part of a global linked list and a local linked list for a given DN.
//
// The global linked list is represented by the head/tail pointers in journal and
// nextGlobal/prevGlobal pointers in entries. The local linked lists are
// represented by heads[DN]/tails[DN] pointers in journal and nextLocal/prevLocal
// pointers in entries:
//
//       .------------.        .------------.        .------------.
//       | dn: A.B    |        | dn: Z      |        | dn: A.B    |
//       | time: 1    |        | time: 2    |        | time: 3    |
//       |------------|        |------------|        |------------|
//       | nextGlobal :------->| nextGlobal :------->| nextGlobal :--> nil
// nil <-: prevGlobal |<-------: prevGlobal |<-------| prevGlobal |
//       |------------|        |------------|  n     |------------|
//       | nextLocal  :---. n  | nextLocal  :->i .-->| nextLocal  :--> nil
// nil <-: prevLocal  |<--: i<-: prevLocal  |  l :---| prevLocal  |
//       '------------'   | l  '------------'    |   '------------'
//            ^           '----------------------'         ^
//            |                      ^                     |
//            |                      |                     |
//         ( head )             ( tails[Z] )            ( tail )
//      ( heads[A.B] )          ( heads[Z] )         ( tails[A.B] )
type journal struct {
	// mu locks the rest of the structure. It must be taken during any operation on the
	// journal.
	mu sync.RWMutex

	// tail is the side of the global linked list that contains the newest log entry,
	// ie. the one that has been pushed the most recently. It can be nil when no log
	// entry has yet been pushed. The global linked list contains all log entries
	// pushed to the journal.
	tail *entry
	// head is the side of the global linked list that contains the oldest log entry.
	// It can be nil when no log entry has yet been pushed.
	head *entry

	// tails are the tail sides of a local linked list for a given DN, ie. the sides
	// that contain the newest entry. They are nil if there are no log entries for that
	// DN.
	tails map[DN]*entry
	// heads are the head sides of a local linked list for a given DN, ie. the sides
	// that contain the oldest entry. They are nil if there are no log entries for that
	// DN.
	heads map[DN]*entry

	// quota is a map from DN to quota structure, representing the quota policy of a
	// particular DN-designated logger.
	quota map[DN]*quota

	// subscribers are observer to logs. New log entries get emitted to channels
	// present in the subscriber structure, after filtering them through subscriber-
	// provided filters (eg. to limit events to subtrees that interest that particular
	// subscriber).
	subscribers []*subscriber
}

// newJournal creates a new empty journal. All journals are independent from
// eachother, and as such, all LogTrees are also independent.
func newJournal() *journal {
	return &journal{
		tails: make(map[DN]*entry),
		heads: make(map[DN]*entry),

		quota: make(map[DN]*quota),
	}
}

// filter is a predicate that returns true if a log subscriber or reader is
// interested in a given log entry.
type filter func(*entry) bool

// filterAll returns a filter that accepts all log entries.
func filterAll() filter {
	return func(*entry) bool { return true }
}

// filterExact returns a filter that accepts only log entries at a given exact
// DN.  This filter should not be used in conjunction with journal.scanEntries
// - instead, journal.getEntries should be used, as it is much faster.
func filterExact(dn DN) filter {
	return func(e *entry) bool {
		return e.origin == dn
	}
}

// filterSubtree returns a filter that accepts all log entries at a given DN and
// sub-DNs. For example, filterSubtree at "foo.bar" would allow entries at
// "foo.bar", "foo.bar.baz", but not "foo" or "foo.barr".
func filterSubtree(root DN) filter {
	if root == "" {
		return filterAll()
	}

	rootParts := strings.Split(string(root), ".")
	return func(e *entry) bool {
		parts := strings.Split(string(e.origin), ".")
		if len(parts) < len(rootParts) {
			return false
		}

		for i, p := range rootParts {
			if parts[i] != p {
				return false
			}
		}

		return true
	}
}

// filterSeverity returns a filter that accepts log entries at a given severity
// level or above. See the Severity type for more information about severity
// levels.
func filterSeverity(atLeast Severity) filter {
	return func(e *entry) bool {
		return e.leveled != nil && e.leveled.severity.AtLeast(atLeast)
	}
}

func filterOnlyRaw(e *entry) bool {
	return e.raw != nil
}

func filterOnlyLeveled(e *entry) bool {
	return e.leveled != nil
}

// scanEntries does a linear scan through the global entry list and returns all
// entries that match the given filters. If retrieving entries for an exact event,
// getEntries should be used instead, as it will leverage DN-local linked lists to
// retrieve them faster. journal.mu must be taken at R or RW level when calling
// this function.
func (j *journal) scanEntries(filters ...filter) (res []*entry) {
	cur := j.tail
	for {
		if cur == nil {
			return
		}

		passed := true
		for _, filter := range filters {
			if !filter(cur) {
				passed = false
				break
			}
		}
		if passed {
			res = append(res, cur)
		}
		cur = cur.nextGlobal
	}
}

// getEntries returns all entries at a given DN. This is faster than a
// scanEntries(filterExact), as it uses the special local linked list pointers to
// traverse the journal. Additional filters can be passed to further limit the
// entries returned, but a scan through this DN's local linked list will be
// performed regardless. journal.mu must be taken at R or RW level when calling
// this function.
func (j *journal) getEntries(exact DN, filters ...filter) (res []*entry) {
	cur := j.tails[exact]
	for {
		if cur == nil {
			return
		}

		passed := true
		for _, filter := range filters {
			if !filter(cur) {
				passed = false
				break
			}
		}
		if passed {
			res = append(res, cur)
		}
		cur = cur.nextLocal
	}

}