Documentation
¶
Overview ¶
Package emf implements metrics.Recorder on top of the CloudWatch Embedded Metric Format.
Why EMF rather than PutMetricData: EMF metrics are extracted from structured JSON written to stdout, so a worker needs no CloudWatch API permissions, no SDK client, and no network call on the hot path. Both deployment shapes Murmur targets already ship stdout to CloudWatch Logs — Lambda natively, ECS via the awslogs driver — so wiring this up is a one-line change with no IAM edit.
rec := emf.New(emf.Config{Namespace: "Murmur"})
defer rec.Close()
handler, _ := kinesis.NewHandler(pipe, dec, kinesis.WithMetrics(rec))
Aggregation ¶
The Recorder contract asks for nanosecond-cost on the hot path, and a pipeline running at even modest throughput would be ruinous to log per-event: one EMF document per record is one CloudWatch Logs ingestion charge per record, and CloudWatch bills ingestion by the byte.
So calls only touch an in-memory aggregate under a sharded lock, and a background goroutine emits one document per flush interval carrying counters as sums and latencies as EMF StatisticSets (Max/Min/Sum/SampleCount, which CloudWatch expands into averages and percentile-capable statistics). At the default 60s interval a pipeline emits 1440 documents a day regardless of whether it processed a hundred records or a hundred million.
Sub-event names ¶
Murmur runtimes encode sub-events by suffixing the pipeline name, e.g. `RecordEvent("orders:dedup_skip")`. Emitting those verbatim would create a separate Pipeline dimension value per sub-event and fragment the dashboards. This recorder splits on the last colon instead, so `orders:dedup_skip` becomes the metric `DedupSkip` on Pipeline=orders — which is what makes dedup_skip, dedup_release, and dedup_release_failed visible at all.
Index ¶
- Constants
- type Config
- type Recorder
- func (r *Recorder) Close() error
- func (r *Recorder) Flush()
- func (r *Recorder) RecordBatch(pipeline, mode string, n int, d time.Duration)
- func (r *Recorder) RecordError(pipeline string, _ error)
- func (r *Recorder) RecordEvent(pipeline string)
- func (r *Recorder) RecordLatency(pipeline, op string, d time.Duration)
Constants ¶
const DefaultFlushInterval = 60 * time.Second
DefaultFlushInterval is how often aggregates are emitted when Config.FlushInterval is zero.
const DefaultNamespace = "Murmur"
DefaultNamespace is the CloudWatch namespace used when Config.Namespace is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// Namespace is the CloudWatch namespace. Defaults to DefaultNamespace.
Namespace string
// FlushInterval is how often aggregated metrics are written. Defaults to
// DefaultFlushInterval. Shorter intervals cost proportionally more in
// CloudWatch Logs ingestion for no extra resolution below one minute,
// which is CloudWatch's own floor for standard metrics.
FlushInterval time.Duration
// Out is where EMF documents are written. Defaults to os.Stdout, which is
// what both Lambda and the ECS awslogs driver forward to CloudWatch Logs.
Out io.Writer
// Dimensions are extra dimensions attached to every metric, e.g.
// {"Env": "soak"}. Keep this small: every distinct combination of
// dimension values is a separate CloudWatch custom metric, and they are
// billed individually.
Dimensions map[string]string
}
Config configures a Recorder.
type Recorder ¶
type Recorder struct {
// contains filtered or unexported fields
}
Recorder is a metrics.Recorder that emits CloudWatch EMF documents.
Safe for concurrent use. Call Close to flush pending aggregates and stop the background goroutine; without it, up to one flush interval of metrics is lost when the process exits.
func (*Recorder) Close ¶
Close flushes any pending aggregates and stops the background goroutine. It is safe to call more than once.
func (*Recorder) Flush ¶
func (r *Recorder) Flush()
Flush writes one EMF document per pipeline with everything aggregated since the last flush, and resets the aggregates. Called automatically on the flush interval and by Close; exported so callers can force a flush (e.g. a Lambda handler that wants metrics out before the freeze).
func (*Recorder) RecordBatch ¶
RecordBatch implements metrics.Recorder.
func (*Recorder) RecordError ¶
RecordError implements metrics.Recorder.
func (*Recorder) RecordEvent ¶
RecordEvent implements metrics.Recorder.
Source Files