Documentation
¶
Overview ¶
Package events delivers individual occurrences inside the engine to programmatic subscribers, without ever blocking the code that reports them.
Why this is not metrics and not logging ¶
The three observability seams in GatewayDNS answer different questions and are deliberately not interchangeable:
- Metrics are always-on aggregates. "How many NXDOMAIN responses in the last minute" is a counter: fixed cost, no cardinality per query, safe to leave enabled forever. A metric cannot tell you *which* query.
- Logs are prose for humans and for log aggregators. They are formatted, sampled, levelled and usually thrown away. A log line is a bad API: a consumer that has to parse one is coupled to its wording.
- Events are individual occurrences, typed, delivered to code. The REST API streaming a live query feed to a dashboard, an analytics sink writing to a database, GatewayDNS Desktop's device manager reacting the moment an unrecognised device appears — none of those can be built on a counter, and none of them should be built by scraping log output.
Different consumers, different costs, different lifetimes. Metrics live for the process, logs live for their retention window, an event lives exactly as long as it takes a subscriber to handle it, and usually there are no subscribers at all.
Delivery guarantees ¶
Stated plainly, because the whole design follows from them:
- At-most-once. An event is delivered to a subscriber zero times or once. There is no acknowledgement, no retry and no redelivery.
- Per-subscriber bounded queue. Every Subscription owns its own buffered channel. One subscriber's backlog is invisible to every other subscriber and to the publisher.
- Drops are counted and reported, never silent. When a queue is full the bus discards an event, increments Subscription.Dropped, and calls the bus's OnDrop hook. A subscriber can always tell that it missed something, and how much.
- Ordering is per-subscriber and only as good as the publishers. Events enqueued to one subscription by one goroutine arrive in the order they were published; events published concurrently from different goroutines have no defined order relative to each other, because the engine has no global sequencer and adding one would be the contention it is designed to avoid.
- A subscriber added while a publish is in flight may miss that event, and one removed mid-flight may still receive it. The subscriber set is read without a lock; see Bus.Publish.
This bus is NOT durable. Nothing is written to disk, nothing survives a restart, and a subscriber that falls behind loses data by design. Do not use it where loss is unacceptable — billing, audit trails, security forensics or anything a regulator will ask about. For those, have the subscriber write to storage that *is* durable and treat Subscription.Dropped as a fault condition rather than a statistic.
The hot path must never block ¶
A publisher is on the DNS query path. At the design target of 10,000 queries per second, a publisher that waits for even one slow subscriber has converted a dashboard's stalled WebSocket into a resolver outage. So Bus.Publish never blocks, never waits on a subscriber and never panics; a stalled subscriber degrades only itself, and its damage is bounded by its own queue size.
The zero-subscriber case is the one that matters most, because it is the normal one: a bus with no subscribers costs a single atomic load per publish. Callers on the query path that would have to build an event to publish it should guard with Bus.Subscribed or Bus.SubscribedTo first — converting a concrete event to the Event interface costs one allocation, and that allocation happens at the call site whether or not anybody is listening:
if bus.SubscribedTo(events.KindQuery) {
bus.Publish(events.QueryEvent{Time: now, Question: q, RCode: rc})
}
Serialisation ¶
Every event field carries a JSON tag, because these cross the REST API in a later milestone. Three consequences are worth knowing before the UI is written:
- Errors are carried as strings, never as error values. A live error is not serialisable, its Error() text can change under a reader, and it can retain arbitrary memory — a wrapped error holding a *http.Response holds a connection — for as long as the event sits in a queue.
- Names render as presentation format, since dnsmsg.Name implements encoding.TextMarshaler. Types, classes and response codes render as their wire integers. That is deliberate: the numbers are stable forever, while mnemonics are presentation, and a UI that wants "NXDOMAIN" can call dnsmsg.RCode.String.
- No event carries a redundant kind field; Event.Kind already reports it. A transport that needs one on the wire should wrap the event in its own envelope, so that the kind and the payload cannot drift apart.
Construction ¶
Nothing in this package is global. A Bus is constructed with NewBus and injected; there is no default bus, no package-level registry and no init-time side effect. A nil *Bus is a valid no-op bus, so a component whose events nobody wants can hold nil rather than a special case.
Index ¶
- Constants
- Variables
- type BlockedEvent
- type Bus
- type BusOptions
- type CacheEvent
- type ConfigReloadedEvent
- type DeviceSeenEvent
- type DropPolicy
- type Event
- type Kind
- type PolicyMatchedEvent
- type QueryEvent
- type ServerEvent
- type SubOptions
- type Subscription
- type UpstreamFailureEvent
- type UpstreamRecoveredEvent
Constants ¶
const DefaultQueueSize = 256
DefaultQueueSize is the per-subscriber queue depth used when neither BusOptions.QueueSize nor SubOptions.QueueSize says otherwise.
It is a compromise, and worth understanding before overriding. A deeper queue absorbs a longer stall — at the design target of 10,000 queries per second, 256 events is roughly 25 milliseconds of full-rate traffic — but it also means a subscriber that falls behind is working on data up to that far in the past, and it multiplies by the number of subscribers in memory. Deep queues do not prevent loss under sustained overload; they only delay it, while making what eventually arrives staler. Size for the burst you expect to absorb, not for the backlog you hope never to have.
Variables ¶
var ErrUnknownKind = errors.New("events: unknown event kind")
ErrUnknownKind is returned by Kind.UnmarshalText for a name this build does not recognise, which is what a client written against a newer server looks like.
Functions ¶
This section is empty.
Types ¶
type BlockedEvent ¶
type BlockedEvent struct {
// Time is when the block decision was made. It is set by the publisher;
// the bus never stamps an event, so an unset Time reaches every subscriber
// as the zero instant and is silently outside every time filter there is.
Time time.Time `json:"time"`
// Device identifies the client whose query was blocked.
Device string `json:"device"`
// Question is the question that was blocked.
Question dnsmsg.Question `json:"question"`
// Rule identifies the rule that matched — a list entry, a regexp, a
// category — precisely enough that an operator can find and change it.
Rule string `json:"rule"`
// Reason is the human-facing explanation shown to whoever asks why. It
// is deliberately distinct from Rule: the rule is the identifier an
// operator edits, the reason is the sentence a user reads.
Reason string `json:"reason"`
// RCode is the response code the client actually received. A block is not
// always an NXDOMAIN: see [policy.BlockMode], where REFUSED and an empty
// NOERROR are both supported answers, and a subscriber that assumed
// NXDOMAIN would misreport every deployment that chose either.
RCode dnsmsg.RCode `json:"rcode"`
}
BlockedEvent reports a query answered from policy rather than from DNS data.
It is separate from QueryEvent rather than a flag on it because the two have different audiences and different volumes: a blocked query is what a user looks at when they wonder why a site will not load, and it must carry enough to answer that question without a second lookup.
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus is a typed, non-blocking, fan-out event bus.
A Bus is safe for concurrent use by any number of publishers and subscribers, and a nil *Bus is a valid bus on which every operation is a no-op — so a component that was given no bus needs no nil check of its own.
The subscriber set is kept as an immutable snapshot behind an atomic pointer and replaced wholesale on subscribe and unsubscribe. Publishing therefore takes no lock at all: it loads a count, loads a pointer, and walks a slice that nobody will ever mutate. Subscribing and unsubscribing are O(n) copies under a mutex, which is the right trade when publishes outnumber subscription changes by six orders of magnitude. A Bus must be created by NewBus. The zero value is not usable and will panic: it has no clock and no subscriber snapshot. A nil *Bus, by contrast, IS usable — every method tolerates it and does nothing — so that a component with no bus configured needs no nil check on its publish path.
func NewBus ¶
func NewBus(opts BusOptions) *Bus
NewBus returns a bus configured by opts. It starts no goroutines and owns no resources beyond its subscribers' queues, so a bus that is never closed leaks nothing; Bus.Close exists for the subscribers' sake, not the bus's.
func (*Bus) Close ¶
func (b *Bus) Close()
Close closes every subscription on the bus and makes all further publishing a no-op. It is idempotent.
Readers blocked on Subscription.C are released, and readers with events still queued drain them before seeing the channel close: shutting down is not a reason to discard work already handed over. Close does not wait for anyone to finish reading, because a bus that waited for its subscribers would hand every subscriber a veto over shutdown.
func (*Bus) Now ¶
Now returns the current time from the bus's clock.
It is here so that a component holding a bus can stamp its events from the same seam the bus was built with, rather than reaching for time.Now and quietly making itself untestable. The bus never rewrites an event's Time: an Event is an interface over a value type, and a bus that wanted to stamp one would have to copy and re-box it on the query path.
func (*Bus) Publish ¶
Publish delivers e to every subscriber that wants its kind.
Publish never blocks, never panics and is safe from any number of goroutines. A subscriber whose queue is full loses an event and nothing else happens: no waiting, no backpressure, no effect on any other subscriber and none at all on the caller. That asymmetry is the point of the package — the caller is answering a DNS query and a stalled dashboard is not permitted to be its problem.
With no subscribers Publish is a single atomic load. With subscribers it is a pointer load, one filter test per subscriber, and one non-blocking channel send per interested subscriber; it allocates nothing itself, though converting a concrete event to Event at the call site does.
A nil bus, a nil event and a closed bus are all no-ops. The last matters: shutdown ordering in a real server is never as tidy as intended, and a listener draining its last few queries must not panic because the bus it reports to was closed first.
func (*Bus) Subscribe ¶
func (b *Bus) Subscribe(opts SubOptions) *Subscription
Subscribe returns a new subscription to this bus.
The caller owns the returned subscription and must call Subscription.Close when finished; until it does, every matching event costs a queue slot and the subscription keeps whatever those events reference alive. Subscribing to a closed bus is not an error — it returns a subscription whose channel is already closed, so a consumer loop written as a range over Subscription.C terminates immediately instead of blocking forever on a bus that will never publish again.
func (*Bus) Subscribed ¶
Subscribed reports whether the bus has any subscriber at all.
Call it before building an event that is expensive to assemble. Converting a concrete event to the Event interface allocates, and that allocation happens at the call site whether or not Bus.Publish does anything with it, so the only way to make an unobserved event truly free is not to build it.
func (*Bus) SubscribedTo ¶
SubscribedTo reports whether any subscriber wants kind k.
It is the precise form of Bus.Subscribed and answers from the union of every subscriber's filter, so a resolver can skip assembling a PolicyMatchedEvent while an operator streams only blocks. The answer is conservative: a true may be stale by the time the event is published, in which case Publish simply finds nobody and returns.
type BusOptions ¶
type BusOptions struct {
// Clock is the time source exposed by [Bus.Now]. Nil means the system
// clock. The bus never stamps an event itself — an event's Time is set
// by whoever knows when the thing happened — but it carries the clock so
// that a component holding only a bus still has one seam for time, and so
// that a test can drive both from one [clock.Fake].
Clock clock.Clock
// QueueSize is the default queue depth for subscriptions that do not set
// their own. Zero or negative means [DefaultQueueSize].
QueueSize int
// OnDrop is called once per enqueue attempt that lost events, with the
// kind of the newest event lost. It exists so that loss is never silent
// even for a subscriber nobody is watching, and its usual implementation
// is a metrics counter increment.
//
// It runs on the publishing goroutine, which is the DNS query path, while
// no bus lock is held. It must not block, must not panic and must not
// publish — a hook that publishes an event on a bus whose queues are
// already full recurses until the stack gives out.
OnDrop func(Kind)
}
BusOptions configures a Bus. The zero value is valid and yields a bus with DefaultQueueSize queues, the system clock and no drop hook.
type CacheEvent ¶
type CacheEvent struct {
// Time is when the cache operation happened.
Time time.Time `json:"time"`
// Op is which operation this was: [KindCacheHit], [KindCacheMiss] or
// [KindCacheEvict]. It is the value returned by Kind.
Op Kind `json:"op"`
// Question is the key the cache was consulted for.
Question dnsmsg.Question `json:"question"`
// TTL is the remaining lifetime for a hit, and the lifetime the entry
// had for an eviction. It can be negative on a hit when a stale answer
// was served under RFC 8767, which is precisely when a subscriber wants
// to know. It marshals as an integer count of nanoseconds.
TTL time.Duration `json:"ttl_ns"`
// Reason explains an eviction — expiry, capacity, an explicit flush —
// and is empty for a hit or a miss. Without it, a cache that is thrashing
// under memory pressure looks exactly like one whose entries are simply
// short-lived.
Reason string `json:"reason,omitempty"`
}
CacheEvent reports a cache hit, miss or eviction.
One type serves three kinds because the payload is identical and a subscriber building a hit-rate graph wants all three through one channel in one order. CacheEvent.Op carries which, and CacheEvent.Kind reports it — an unset Op is KindUnknown and will match no subscription, which is the intended way for a caller that forgot to set it to find out.
func (CacheEvent) Kind ¶
func (e CacheEvent) Kind() Kind
Kind implements Event, reporting CacheEvent.Op.
type ConfigReloadedEvent ¶
type ConfigReloadedEvent struct {
// Time is when the new configuration took effect.
Time time.Time `json:"time"`
// Version increases monotonically with each accepted configuration. It
// is not a hash and not a file modification time: a reload that produces
// identical configuration still gets a new version, and a version never
// goes backwards within a process.
Version uint64 `json:"version"`
}
ConfigReloadedEvent reports configuration replaced at runtime.
Version is what makes the event actionable: a subscriber holding state derived from configuration can compare it to what it built against and rebuild only when it actually changed, which matters because reloads arrive in bursts when a file is saved by an editor that writes it in pieces.
type DeviceSeenEvent ¶
type DeviceSeenEvent struct {
// Time is when the traffic was seen.
Time time.Time `json:"time"`
// Device is the stable identifier assigned to the client, empty if the
// device manager has not assigned one yet.
Device string `json:"device"`
// ClientAddr is the address the traffic came from.
ClientAddr netip.Addr `json:"client_addr"`
// Name is the friendly label for the device — a DHCP hostname, a
// configured name — and is empty when nothing is known.
Name string `json:"name,omitempty"`
// First reports whether this is the first time the device has been seen
// in this process's lifetime. A subscriber that only cares about new
// devices filters on this rather than on the kind, because the kind is
// emitted for every device however familiar.
First bool `json:"first"`
}
DeviceSeenEvent reports traffic from a client.
It exists for GatewayDNS Desktop's device manager and anything like it: the moment an unrecognised device sends its first query is the moment a policy decision has to be made about it, and polling a table to discover that is both slower and racier than being told.
type DropPolicy ¶
type DropPolicy uint8
DropPolicy decides which event a full subscriber queue gives up.
There is no third option. "Block until there is room" is the one thing this bus may never do, because the caller with no room to spare is the DNS query path, and "grow the queue" is unbounded memory with extra steps.
const ( // DropNewest discards the arriving event and keeps the queue as it is. // It is the default and the cheapest: a failed non-blocking send and // nothing else, with no second operation on the channel and no exclusive // lock. Choose it for anything that aggregates — an analytics sink, a // counter, a log writer — where events are interchangeable and losing the // most recent one is no worse than losing any other. DropNewest DropPolicy = iota // DropOldest evicts the oldest queued event to make room for the // arriving one, so the queue always holds the freshest events. Choose it // for anything that displays state — a live UI, a status panel, a // tail-style feed — where a viewer would rather see the last hundred // things that happened than the first hundred things that happened after // they stopped keeping up. It costs an exclusive lock on the overflow // path, so it is measurably more expensive precisely when the subscriber // is already in trouble. DropOldest )
Drop policies.
type Event ¶
type Event interface {
// Kind reports what happened. It must be constant for a given value and
// must never be [KindUnknown] for an event that is actually published.
Kind() Kind
// At reports when it happened, per the publisher's clock. It is not the
// time the event was delivered, which may be much later on a backed-up
// subscription.
At() time.Time
}
Event is one thing that happened.
The interface is two methods wide on purpose. Everything the bus needs to do — filter by kind, and let a subscriber order or age what it received — is here, and everything else is the concrete type's business. A wider interface would have to be implemented by every future event type, and would push the bus towards understanding payloads it has no reason to understand.
Implementations are value types with value receivers, so both QueryEvent and *QueryEvent satisfy Event. Publish a value unless the event is unusually large and you are publishing it to many buses.
type Kind ¶
type Kind uint16
Kind identifies what happened, and is the only thing a subscriber may filter on.
Filtering is a Kind rather than a Go type assertion because the filter has to be evaluated on the publishing goroutine, once per subscriber, on the query path: a small integer compared against a bitmask costs nothing, while a type switch over a growing set of event types costs more with every type added. It is uint16 rather than a string for the same reason, and because it makes the set closed — a subscriber written today cannot be surprised by a kind invented tomorrow, it simply never matches it.
const ( // KindUnknown is the zero Kind. It exists so that a struct with an // unset kind field — [CacheEvent.Op], say — matches no filter and is // visibly wrong in a log, rather than silently masquerading as whatever // kind happened to be numbered zero. KindUnknown Kind = iota // KindQuery reports a query that was answered, whatever the answer was. // It is the highest-volume kind by orders of magnitude. KindQuery // KindBlocked reports a query answered from policy rather than from DNS // data, and carries the rule that decided it. KindBlocked // KindCacheHit reports an answer served from cache. KindCacheHit // KindCacheMiss reports a lookup that had to go upstream. KindCacheMiss // KindCacheEvict reports an entry leaving the cache, whether it expired // per RFC 2308 or was displaced under memory pressure. KindCacheEvict // KindUpstreamFailure reports one upstream resolver failing a query. // Consecutive failures are counted so a subscriber can distinguish a // blip from an outage without keeping state. KindUpstreamFailure // KindUpstreamRecovered reports an upstream answering again after a run // of failures. It is emitted once per recovery, not once per success. KindUpstreamRecovered // KindDeviceSeen reports traffic from a client, and is what a device // manager subscribes to in order to notice a new device on the network. KindDeviceSeen // KindConfigReloaded reports configuration replaced at runtime, so a // subscriber holding derived state knows to rebuild it. KindConfigReloaded // KindServerStarted reports a listener accepting traffic. KindServerStarted // KindServerStopped reports a listener no longer accepting traffic, // including when it stopped because of an error. KindServerStopped // KindPolicyMatched reports a policy rule matching, whether or not the // match changed the answer. It is separate from [KindBlocked] because // an allow rule matching is just as interesting to an auditor as a deny // rule, and far less interesting to a dashboard. KindPolicyMatched )
Event kinds. The zero value is reserved so that an uninitialised Kind can never be mistaken for a real one; see KindUnknown.
func (Kind) MarshalText ¶
MarshalText implements encoding.TextMarshaler so that a Kind crosses the REST API as "cache_hit" rather than as 3.
Kinds are the one place where the name beats the number, because unlike a response code a Kind has no wire format to be stable against: its numeric value is an implementation detail of this package and may be renumbered when kinds are added, while the name may not.
func (Kind) String ¶
String returns the kind's stable snake_case name, or "kind(n)" for a value this build does not know.
The names are lowercase and underscore-separated because they are consumed by machines — JSON fields, metric labels, dashboard filters — not by prose. They are part of the API: renaming one breaks every stored query that mentions it.
func (*Kind) UnmarshalText ¶
UnmarshalText implements encoding.TextUnmarshaler, so a subscription filter can be expressed in configuration or in a query string.
type PolicyMatchedEvent ¶
type PolicyMatchedEvent struct {
// Time is when the rule matched.
Time time.Time `json:"time"`
// Device identifies the client the policy was evaluated for.
Device string `json:"device"`
// Question is the question the policy was evaluated against.
Question dnsmsg.Question `json:"question"`
// Policy names the policy the rule belongs to.
Policy string `json:"policy"`
// Rule identifies the specific rule that matched.
Rule string `json:"rule"`
// Action is what the rule asked for — allow, block, redirect, log — as
// decided by the rule, which is not necessarily what finally happened
// once every policy had its say.
Action string `json:"action"`
}
PolicyMatchedEvent reports a policy rule matching, whether or not the match changed the answer.
It is the audit record that BlockedEvent is not: a block is a user-facing outcome, while a match is evidence about the rule set itself, including allow rules and rules that matched but were overridden. Subscribing to it is expensive on a busy resolver, which is exactly why it is a separate kind that costs nothing when nobody asks for it.
type QueryEvent ¶
type QueryEvent struct {
// Time is when the query was answered.
Time time.Time `json:"time"`
// Device identifies the client, as resolved by the device manager. It is
// empty when the client is unknown, which is not the same as unnamed.
Device string `json:"device"`
// ClientAddr is the address the query arrived from. It marshals as text
// because [netip.Addr] implements [encoding.TextMarshaler], so an IPv6
// client is readable in the UI without help.
ClientAddr netip.Addr `json:"client_addr"`
// Question is the question that was asked (RFC 1035 section 4.1.2).
// Only the first question is carried: no deployed resolver honours
// QDCOUNT above one, and modelling a slice here would cost an allocation
// per event to represent something that never happens.
Question dnsmsg.Question `json:"question"`
// RCode is the effective response code, including any extended bits
// carried in an OPT record per RFC 6891 section 6.1.3.
RCode dnsmsg.RCode `json:"rcode"`
// Duration is the wall time spent answering, measured from receipt to
// response. It marshals as an integer count of nanoseconds, which is what
// the field name says.
Duration time.Duration `json:"duration_ns"`
// Source names where the answer came from: the upstream provider, the
// cache, the policy engine, a local zone. It is free-form because the
// set of answer sources grows with every resolver feature, and a closed
// enum here would have to be extended in lockstep with them.
Source string `json:"source"`
// Cached reports whether the answer came from cache. It duplicates what
// Source usually implies, because "what fraction of queries were cached"
// is the single most asked question of this event and a subscriber should
// not have to string-match to answer it.
Cached bool `json:"cached"`
// Blocked reports that policy answered this query rather than DNS data,
// and Rule names the rule that did it.
//
// They are here as well as on [BlockedEvent] because the two events serve
// different readers and a query log must not choose between them. A log
// driven by KindQuery alone sees every query exactly once; without these
// fields it would have to correlate a second stream to learn which of
// those rows were blocks, and any writer that simply recorded both streams
// would enter a blocked query twice — once truthfully and once as an
// ordinary NXDOMAIN. RCode above stays the code the client actually
// received, which is not always NXDOMAIN.
Blocked bool `json:"blocked"`
Rule string `json:"rule,omitempty"`
// Answers is how many records the answer section carried. Zero on a
// negative answer, on a block, and on a failure — which is most of what
// anyone reads a query log to find.
Answers int `json:"answers"`
}
QueryEvent reports one answered query and is the highest-volume event in the system.
It is a value type with no slices, maps or pointers, which is what makes it safe to hand the same event to every subscriber concurrently: there is nothing in it a subscriber can mutate and nothing that keeps a buffer alive. Anything added to it later must preserve that property; a []dnsmsg.RR answer section here would let one subscriber corrupt another's view and would pin decoder memory for the lifetime of the slowest queue.
type ServerEvent ¶
type ServerEvent struct {
// Time is when the transition happened.
Time time.Time `json:"time"`
// Op is which transition: [KindServerStarted] or [KindServerStopped].
// It is the value returned by Kind.
Op Kind `json:"op"`
// Addr is the address the listener is bound to, in the form the listener
// reports it.
Addr string `json:"addr"`
// Proto names the transport: udp, tcp, tls, https, quic.
Proto string `json:"proto"`
// Err is why a listener stopped, as text, and is empty for a clean stop
// or a start. A stop with a non-empty Err is a fault; a stop without one
// is a shutdown.
Err string `json:"err,omitempty"`
}
ServerEvent reports a listener starting or stopping.
One type serves both because a supervisor subscribing to lifecycle wants them in one ordered stream; two types would let a stop overtake its own start in a subscriber that merged two channels.
func (ServerEvent) Kind ¶
func (e ServerEvent) Kind() Kind
Kind implements Event, reporting ServerEvent.Op.
type SubOptions ¶
type SubOptions struct {
// Kinds restricts what this subscriber receives; empty means every kind,
// including kinds added in later versions. The slice is converted to a
// bitmask at subscribe time and is not retained, so the caller may reuse
// or modify it afterwards.
//
// Filter as narrowly as the subscriber can stand. The filter is evaluated
// before the queue is touched, so a kind a subscriber did not ask for
// costs it nothing at all — not a channel operation, not a lock, not a
// slot in its queue that a wanted event would otherwise have used.
Kinds []Kind
// QueueSize is this subscription's queue depth. Zero or negative means
// the bus default.
QueueSize int
// Policy decides which event is lost when the queue is full. See
// [DropNewest] and [DropOldest].
Policy DropPolicy
}
SubOptions configures a Subscription. The zero value is valid and yields a subscription to every kind, with the bus's default queue size and DropNewest.
type Subscription ¶
type Subscription struct {
// contains filtered or unexported fields
}
Subscription is one consumer's view of the bus: a filter, a bounded queue and the counters that say how well the consumer is keeping up.
A Subscription is safe for concurrent use, though its channel is best consumed by a single goroutine — two readers on one subscription get an arbitrary split of the events and neither sees the order the other saw.
func (*Subscription) C ¶
func (s *Subscription) C() <-chan Event
C returns the channel events are delivered on.
The channel is closed when the subscription or its bus is closed, so the idiomatic consumer is a range loop that ends by itself:
for e := range sub.C() {
handle(e)
}
Read it promptly. Every moment spent handling an event is a moment the queue is filling, and a handler that does I/O should hand off to its own worker rather than do it here — this queue is a shock absorber, not a work queue.
func (*Subscription) Close ¶
func (s *Subscription) Close()
Close ends the subscription and closes its channel.
It is idempotent, safe to call from any goroutine, and safe to call while another goroutine is publishing — the close is serialised against in-flight sends, so it can never turn a publish into a send on a closed channel. Already queued events survive: a reader may drain them and will then see the channel closed.
func (*Subscription) Delivered ¶
func (s *Subscription) Delivered() uint64
Delivered returns the number of events this subscription's consumer has received or will receive.
It counts enqueues net of evictions, not handled events: an event counted here may still be sitting in the queue, or may be discarded unread when the subscription closes.
Netting off evictions is what makes the number mean the same thing under both drop policies. Under DropNewest an event that does not fit is never enqueued, so it is only ever counted in Dropped. Under DropOldest it IS enqueued, and an older one is evicted to make room — so without the decrement, Delivered would count an event the consumer provably never sees, and two subscriptions with different policies over identical traffic would report different totals. Received is Delivered under either policy, and Delivered+Dropped is the number published to this subscription.
func (*Subscription) Dropped ¶
func (s *Subscription) Dropped() uint64
Dropped returns the number of events this subscription lost because its queue was full.
This is the number that makes loss visible, and a non-zero value means exactly one thing: this consumer is slower than the publisher, and it must either get faster, filter harder, or ask for a deeper queue. Events published after the subscription is closed are not counted — the consumer asked to stop listening, which is not the same as failing to keep up, and conflating them would make every clean shutdown look like an overload.
type UpstreamFailureEvent ¶
type UpstreamFailureEvent struct {
// Time is when the failure was observed.
Time time.Time `json:"time"`
// Provider names the upstream that failed.
Provider string `json:"provider"`
// Err is the failure rendered as text. It is a string and not an error
// for the reason given in the package documentation: an event may be
// serialised, may outlive the operation that produced it, and must not
// retain whatever an error value happens to wrap.
Err string `json:"err"`
// Consecutive is the number of failures in a row, including this one. It
// is 1 for the first failure after a success.
Consecutive int `json:"consecutive"`
}
UpstreamFailureEvent reports one upstream resolver failing.
Consecutive is carried so that a subscriber can distinguish a packet loss blip from a dead resolver without keeping per-provider state of its own, which is exactly the state the health checker already keeps.
type UpstreamRecoveredEvent ¶
type UpstreamRecoveredEvent struct {
// Time is when the first successful response after the outage arrived.
Time time.Time `json:"time"`
// Provider names the upstream that recovered.
Provider string `json:"provider"`
// Failures is how many consecutive failures preceded the recovery, so
// that a subscriber can report the severity of what just ended.
Failures int `json:"failures"`
// Downtime is how long the upstream was considered failed. It marshals
// as an integer count of nanoseconds.
Downtime time.Duration `json:"downtime_ns"`
}
UpstreamRecoveredEvent reports an upstream answering again after a run of failures.
It is emitted once, on the transition, not on every subsequent success — otherwise recovery would be indistinguishable from normal operation and every subscriber would have to deduplicate.
func (UpstreamRecoveredEvent) At ¶
func (e UpstreamRecoveredEvent) At() time.Time
At implements Event.
func (UpstreamRecoveredEvent) Kind ¶
func (e UpstreamRecoveredEvent) Kind() Kind
Kind implements Event.
Source Files