Time series data is a sequence of values indexed by time — sensor readings, server metrics, financial prices, log event counts. The defining characteristic: you query by time range far more often than by individual record ID, and you almost always want aggregate values (average, max, p95) over a window rather than raw individual points.
This guide covers in-memory time series structures, aggregation patterns, and integrating with purpose-built time series databases. For production metrics see Go monitoring with Prometheus. For general analytics see Go analytics and reporting.
Representing a Data Point
A time series data point minimally needs a timestamp and a value. Tags (key-value labels) let you segment and filter across multiple series:
type DataPoint struct {
Timestamp time.Time
Value float64
Tags map[string]string // e.g., {"host": "web-1", "region": "us-east"}
}
type Series struct {
Name string
Points []DataPoint // kept in ascending timestamp order
}
Sorting by timestamp is a prerequisite for efficient range queries and resampling. If you ingest data that might arrive out of order (common with distributed systems), sort after each batch insert rather than on every insert:
func (s *Series) Sort() {
sort.Slice(s.Points, func(i, j int) bool {
return s.Points[i].Timestamp.Before(s.Points[j].Timestamp)
})
}
In-Memory Store with Concurrent Access
A store manages multiple named series. Concurrent read access is common (multiple dashboards querying simultaneously), so sync.RWMutex is appropriate:
type Store struct {
mu sync.RWMutex
series map[string]*Series
}
func NewStore() *Store {
return &Store{series: make(map[string]*Series)}
}
func (s *Store) Record(name string, ts time.Time, value float64, tags map[string]string) {
s.mu.Lock()
defer s.mu.Unlock()
ser, ok := s.series[name]
if !ok {
ser = &Series{Name: name}
s.series[name] = ser
}
ser.Points = append(ser.Points, DataPoint{Timestamp: ts, Value: value, Tags: tags})
}
func (s *Store) QueryRange(name string, start, end time.Time) ([]DataPoint, error) {
s.mu.RLock()
defer s.mu.RUnlock()
ser, ok := s.series[name]
if !ok {
return nil, fmt.Errorf("series %q not found", name)
}
var result []DataPoint
for _, p := range ser.Points {
if !p.Timestamp.Before(start) && !p.Timestamp.After(end) {
result = append(result, p)
}
}
return result, nil
}
For series with millions of points, the linear scan in QueryRange becomes slow. Pre-sort and use sort.Search for O(log n) binary search on the sorted timestamp array.
Aggregation
Raw data has high resolution. Dashboards and alerts need aggregated values — one data point per minute or hour, not one per second. The aggregation pattern: bucket points by time window, apply an aggregation function to each bucket:
type AggFunc func(values []float64) float64
func Mean(values []float64) float64 {
if len(values) == 0 { return 0 }
var sum float64
for _, v := range values { sum += v }
return sum / float64(len(values))
}
func Max(values []float64) float64 {
if len(values) == 0 { return 0 }
m := values[0]
for _, v := range values[1:] {
if v > m { m = v }
}
return m
}
func Percentile(p float64) AggFunc {
return func(values []float64) float64 {
if len(values) == 0 { return 0 }
sorted := make([]float64, len(values))
copy(sorted, values)
sort.Float64s(sorted)
idx := int(math.Ceil(p/100*float64(len(sorted)))) - 1
if idx < 0 { idx = 0 }
return sorted[idx]
}
}
// Resample groups points into fixed windows and applies aggFunc to each
func Resample(points []DataPoint, window time.Duration, aggFunc AggFunc) []DataPoint {
if len(points) == 0 { return nil }
// Bucket by window start time
buckets := make(map[time.Time][]float64)
for _, p := range points {
key := p.Timestamp.Truncate(window)
buckets[key] = append(buckets[key], p.Value)
}
// Sort bucket keys for ordered output
keys := make([]time.Time, 0, len(buckets))
for k := range buckets { keys = append(keys, k) }
sort.Slice(keys, func(i, j int) bool { return keys[i].Before(keys[j]) })
result := make([]DataPoint, 0, len(keys))
for _, k := range keys {
result = append(result, DataPoint{
Timestamp: k,
Value: aggFunc(buckets[k]),
})
}
return result
}
Usage: resample raw per-second CPU measurements into per-minute averages:
raw, _ := store.QueryRange("cpu.usage", time.Now().Add(-1*time.Hour), time.Now())
minutely := Resample(raw, time.Minute, Mean)
hourlyP95 := Resample(raw, time.Hour, Percentile(95))
Downsampling for Long-Term Storage
Keeping every raw data point forever is expensive. Production systems use a tiered retention strategy: full resolution for recent data, downsampled for older data:
// DownsampleOlderThan reduces points older than cutoff to one-per-interval
func DownsampleOlderThan(points []DataPoint, cutoff time.Time, interval time.Duration, aggFunc AggFunc) []DataPoint {
var recent, old []DataPoint
for _, p := range points {
if p.Timestamp.Before(cutoff) {
old = append(old, p)
} else {
recent = append(recent, p)
}
}
downsampled := Resample(old, interval, aggFunc)
return append(downsampled, recent...)
}
// Example: keep full resolution for last 24h, hourly averages before that
points = DownsampleOlderThan(
points,
time.Now().Add(-24*time.Hour),
time.Hour,
Mean,
)
In production, this runs as a background job. InfluxDB, TimescaleDB, and VictoriaMetrics all handle retention and downsampling automatically.
Querying with Tag Filters
When multiple series share the same metric name but differ by tags (common for multi-host metrics), filter by tag values:
func (s *Store) QueryByTag(metricName, tagKey, tagValue string, start, end time.Time) []DataPoint {
s.mu.RLock()
defer s.mu.RUnlock()
ser, ok := s.series[metricName]
if !ok { return nil }
var result []DataPoint
for _, p := range ser.Points {
if p.Timestamp.Before(start) || p.Timestamp.After(end) { continue }
if p.Tags[tagKey] == tagValue {
result = append(result, p)
}
}
return result
}
// Usage: CPU usage for host web-1 only
points := store.QueryByTag("cpu.usage", "host", "web-1", start, end)
Integrating with InfluxDB
For production workloads, a purpose-built time series database is far more efficient than in-memory Go structs. InfluxDB’s line protocol is the most portable format:
import influxdb2 "github.com/influxdata/influxdb-client-go/v2"
func writeToInfluxDB(url, token, org, bucket string) {
client := influxdb2.NewClient(url, token)
defer client.Close()
writeAPI := client.WriteAPIBlocking(org, bucket)
for _, reading := range sensorReadings {
p := influxdb2.NewPoint(
"temperature", // measurement name
map[string]string{"sensor": reading.SensorID, "room": reading.Room}, // tags
map[string]interface{}{"celsius": reading.Celsius}, // fields
reading.Timestamp,
)
if err := writeAPI.WritePoint(context.Background(), p); err != nil {
log.Printf("write error: %v", err)
}
}
}
Query with InfluxQL or Flux:
queryAPI := client.QueryAPI(org)
result, err := queryAPI.Query(context.Background(), `
from(bucket: "sensors")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "temperature" and r.room == "living-room")
|> mean()
`)
Memory Management for Long-Running Stores
An in-memory store that never evicts will exhaust memory. Two strategies:
Sliding window — keep only the last N duration of data:
func (s *Store) EvictBefore(cutoff time.Time) {
s.mu.Lock()
defer s.mu.Unlock()
for name, ser := range s.series {
i := sort.Search(len(ser.Points), func(j int) bool {
return !ser.Points[j].Timestamp.Before(cutoff)
})
ser.Points = ser.Points[i:]
if len(ser.Points) == 0 {
delete(s.series, name)
}
}
}
// Run periodically
go func() {
ticker := time.NewTicker(time.Hour)
for range ticker.C {
store.EvictBefore(time.Now().Add(-7 * 24 * time.Hour)) // keep 7 days
}
}()
Max points cap — evict oldest points when the series exceeds a limit:
const maxPoints = 100_000
func (s *Series) trimIfNeeded() {
if len(s.Points) > maxPoints {
excess := len(s.Points) - maxPoints
s.Points = s.Points[excess:] // drop oldest
}
}
Summary
- Sort time series data by timestamp — binary search makes range queries O(log n) instead of O(n)
ResamplewithTruncate(window)buckets points into fixed windows; applyMean,Max, orPercentileto each bucket- Use tiered retention: full resolution for recent data, downsampled for older data
- Tag-based filtering enables per-host, per-region queries from a single metric series
- For production workloads, use InfluxDB, TimescaleDB, or VictoriaMetrics — in-memory Go stores don’t scale past single-process use
- Evict old data periodically to prevent unbounded memory growth
Comments