953 lines
23 KiB
Go
953 lines
23 KiB
Go
// Copyright 2016 Prometheus Team
|
|
// 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 silence provides a storage for silences, which can share its
|
|
// state over a mesh network and snapshot it.
|
|
package silence
|
|
|
|
import (
|
|
"bytes"
|
|
"fmt"
|
|
"io"
|
|
"math/rand"
|
|
"os"
|
|
"reflect"
|
|
"regexp"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/go-kit/kit/log"
|
|
"github.com/go-kit/kit/log/level"
|
|
"github.com/matttproud/golang_protobuf_extensions/pbutil"
|
|
"github.com/pkg/errors"
|
|
pb "github.com/prometheus/alertmanager/silence/silencepb"
|
|
"github.com/prometheus/alertmanager/types"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"github.com/prometheus/common/model"
|
|
"github.com/satori/go.uuid"
|
|
"github.com/weaveworks/mesh"
|
|
)
|
|
|
|
// ErrNotFound is returned if a silence was not found.
|
|
var ErrNotFound = fmt.Errorf("not found")
|
|
|
|
func utcNow() time.Time {
|
|
return time.Now().UTC()
|
|
}
|
|
|
|
type matcherCache map[*pb.Silence]types.Matchers
|
|
|
|
// Get retrieves the matchers for a given silence. If it is a missed cache
|
|
// access, it compiles and adds the matchers of the requested silence to the
|
|
// cache.
|
|
func (c matcherCache) Get(s *pb.Silence) (types.Matchers, error) {
|
|
if m, ok := c[s]; ok {
|
|
return m, nil
|
|
}
|
|
return c.add(s)
|
|
}
|
|
|
|
// add compiles a silences' matchers and adds them to the cache.
|
|
// It returns the compiled matchers.
|
|
func (c matcherCache) add(s *pb.Silence) (types.Matchers, error) {
|
|
var (
|
|
ms types.Matchers
|
|
mt *types.Matcher
|
|
)
|
|
|
|
for _, m := range s.Matchers {
|
|
mt = &types.Matcher{
|
|
Name: m.Name,
|
|
Value: m.Pattern,
|
|
}
|
|
switch m.Type {
|
|
case pb.Matcher_EQUAL:
|
|
mt.IsRegex = false
|
|
case pb.Matcher_REGEXP:
|
|
mt.IsRegex = true
|
|
}
|
|
err := mt.Init()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
ms = append(ms, mt)
|
|
}
|
|
|
|
c[s] = ms
|
|
|
|
return ms, nil
|
|
}
|
|
|
|
// Silences holds a silence state that can be modified, queried, and snapshot.
|
|
type Silences struct {
|
|
logger log.Logger
|
|
metrics *metrics
|
|
now func() time.Time
|
|
retention time.Duration
|
|
|
|
gossip mesh.Gossip // gossip channel for sharing silences
|
|
|
|
// We store silences in a map of IDs for now. Currently, the memory
|
|
// state is equivalent to the mesh.GossipData representation.
|
|
// In the future we'll want support for efficient queries by time
|
|
// range and affected labels.
|
|
// Mutex also guards the matcherCache, which always need write lock access.
|
|
mtx sync.Mutex
|
|
st *gossipData
|
|
mc matcherCache
|
|
}
|
|
|
|
type metrics struct {
|
|
gcDuration prometheus.Summary
|
|
snapshotDuration prometheus.Summary
|
|
snapshotSize prometheus.Gauge
|
|
queriesTotal prometheus.Counter
|
|
queryErrorsTotal prometheus.Counter
|
|
queryDuration prometheus.Histogram
|
|
silencesActive prometheus.GaugeFunc
|
|
silencesPending prometheus.GaugeFunc
|
|
silencesExpired prometheus.GaugeFunc
|
|
}
|
|
|
|
func newSilenceMetricByState(s *Silences, st types.SilenceState) prometheus.GaugeFunc {
|
|
return prometheus.NewGaugeFunc(
|
|
prometheus.GaugeOpts{
|
|
Name: "alertmanager_silences",
|
|
Help: "How many silences by state.",
|
|
ConstLabels: prometheus.Labels{"state": string(st)},
|
|
},
|
|
func() float64 {
|
|
count, err := s.CountState(st)
|
|
if err != nil {
|
|
level.Error(s.logger).Log("msg", "Counting silences failed", "err", err)
|
|
}
|
|
return float64(count)
|
|
},
|
|
)
|
|
}
|
|
|
|
func newMetrics(r prometheus.Registerer, s *Silences) *metrics {
|
|
m := &metrics{}
|
|
|
|
m.gcDuration = prometheus.NewSummary(prometheus.SummaryOpts{
|
|
Name: "alertmanager_silences_gc_duration_seconds",
|
|
Help: "Duration of the last silence garbage collection cycle.",
|
|
})
|
|
m.snapshotDuration = prometheus.NewSummary(prometheus.SummaryOpts{
|
|
Name: "alertmanager_silences_snapshot_duration_seconds",
|
|
Help: "Duration of the last silence snapshot.",
|
|
})
|
|
m.snapshotSize = prometheus.NewGauge(prometheus.GaugeOpts{
|
|
Name: "alertmanager_silences_snapshot_size_bytes",
|
|
Help: "Size of the last silence snapshot in bytes.",
|
|
})
|
|
m.queriesTotal = prometheus.NewCounter(prometheus.CounterOpts{
|
|
Name: "alertmanager_silences_queries_total",
|
|
Help: "How many silence queries were received.",
|
|
})
|
|
m.queryErrorsTotal = prometheus.NewCounter(prometheus.CounterOpts{
|
|
Name: "alertmanager_silences_query_errors_total",
|
|
Help: "How many silence received queries did not succeed.",
|
|
})
|
|
m.queryDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
|
|
Name: "alertmanager_silences_query_duration_seconds",
|
|
Help: "Duration of silence query evaluation.",
|
|
})
|
|
if s != nil {
|
|
m.silencesActive = newSilenceMetricByState(s, types.SilenceStateActive)
|
|
m.silencesPending = newSilenceMetricByState(s, types.SilenceStatePending)
|
|
m.silencesExpired = newSilenceMetricByState(s, types.SilenceStateExpired)
|
|
}
|
|
|
|
if r != nil {
|
|
r.MustRegister(
|
|
m.gcDuration,
|
|
m.snapshotDuration,
|
|
m.snapshotSize,
|
|
m.queriesTotal,
|
|
m.queryErrorsTotal,
|
|
m.queryDuration,
|
|
m.silencesActive,
|
|
m.silencesPending,
|
|
m.silencesExpired,
|
|
)
|
|
}
|
|
return m
|
|
}
|
|
|
|
// Options exposes configuration options for creating a new Silences object.
|
|
// Its zero value is a safe default.
|
|
type Options struct {
|
|
// A snapshot file or reader from which the initial state is loaded.
|
|
// None or only one of them must be set.
|
|
SnapshotFile string
|
|
SnapshotReader io.Reader
|
|
|
|
// Retention time for newly created Silences. Silences may be
|
|
// garbage collected after the given duration after they ended.
|
|
Retention time.Duration
|
|
|
|
// A function creating a mesh.Gossip on being called with a mesh.Gossiper.
|
|
Gossip func(g mesh.Gossiper) mesh.Gossip
|
|
|
|
// A logger used by background processing.
|
|
Logger log.Logger
|
|
Metrics prometheus.Registerer
|
|
}
|
|
|
|
func (o *Options) validate() error {
|
|
if o.SnapshotFile != "" && o.SnapshotReader != nil {
|
|
return fmt.Errorf("only one of SnapshotFile and SnapshotReader must be set")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// New returns a new Silences object with the given configuration.
|
|
func New(o Options) (*Silences, error) {
|
|
if err := o.validate(); err != nil {
|
|
return nil, err
|
|
}
|
|
if o.SnapshotFile != "" {
|
|
if r, err := os.Open(o.SnapshotFile); err != nil {
|
|
if !os.IsNotExist(err) {
|
|
return nil, err
|
|
}
|
|
} else {
|
|
o.SnapshotReader = r
|
|
}
|
|
}
|
|
s := &Silences{
|
|
mc: matcherCache{},
|
|
logger: log.NewNopLogger(),
|
|
retention: o.Retention,
|
|
now: utcNow,
|
|
gossip: nopGossip{},
|
|
st: newGossipData(),
|
|
}
|
|
s.metrics = newMetrics(o.Metrics, s)
|
|
|
|
if o.Logger != nil {
|
|
s.logger = o.Logger
|
|
}
|
|
if o.Gossip != nil {
|
|
s.gossip = o.Gossip(gossiper{s})
|
|
}
|
|
if o.SnapshotReader != nil {
|
|
if err := s.loadSnapshot(o.SnapshotReader); err != nil {
|
|
return s, err
|
|
}
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
type nopGossip struct{}
|
|
|
|
func (nopGossip) GossipBroadcast(d mesh.GossipData) {}
|
|
func (nopGossip) GossipUnicast(mesh.PeerName, []byte) error { return nil }
|
|
|
|
// Maintenance garbage collects the silence state at the given interval. If the snapshot
|
|
// file is set, a snapshot is written to it afterwards.
|
|
// Terminates on receiving from stopc.
|
|
func (s *Silences) Maintenance(interval time.Duration, snapf string, stopc <-chan struct{}) {
|
|
t := time.NewTicker(interval)
|
|
defer t.Stop()
|
|
|
|
f := func() error {
|
|
start := s.now()
|
|
var size int
|
|
level.Info(s.logger).Log("msg", "Running maintenance")
|
|
defer func() {
|
|
level.Info(s.logger).Log("msg", "Maintenance done", "duration", s.now().Sub(start), "size", size)
|
|
s.metrics.snapshotSize.Set(float64(size))
|
|
}()
|
|
|
|
if _, err := s.GC(); err != nil {
|
|
return err
|
|
}
|
|
if snapf == "" {
|
|
return nil
|
|
}
|
|
f, err := openReplace(snapf)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if size, err = s.Snapshot(f); err != nil {
|
|
return err
|
|
}
|
|
return f.Close()
|
|
}
|
|
|
|
Loop:
|
|
for {
|
|
select {
|
|
case <-stopc:
|
|
break Loop
|
|
case <-t.C:
|
|
if err := f(); err != nil {
|
|
level.Info(s.logger).Log("msg", "Running maintenance failed", "err", err)
|
|
}
|
|
}
|
|
}
|
|
// No need for final maintenance if we don't want to snapshot.
|
|
if snapf == "" {
|
|
return
|
|
}
|
|
if err := f(); err != nil {
|
|
level.Info(s.logger).Log("msg", "Creating shutdown snapshot failed", "err", err)
|
|
}
|
|
}
|
|
|
|
// GC runs a garbage collection that removes silences that have ended longer
|
|
// than the configured retention time ago.
|
|
func (s *Silences) GC() (int, error) {
|
|
start := time.Now()
|
|
defer func() { s.metrics.gcDuration.Observe(time.Since(start).Seconds()) }()
|
|
|
|
now := s.now()
|
|
var n int
|
|
|
|
s.mtx.Lock()
|
|
defer s.mtx.Unlock()
|
|
|
|
for id, sil := range s.st.data {
|
|
if sil.ExpiresAt.IsZero() {
|
|
return n, errors.New("unexpected zero expiration timestamp")
|
|
}
|
|
if !sil.ExpiresAt.After(now) {
|
|
delete(s.st.data, id)
|
|
delete(s.mc, sil.Silence)
|
|
n++
|
|
}
|
|
}
|
|
|
|
return n, nil
|
|
}
|
|
|
|
func validateMatcher(m *pb.Matcher) error {
|
|
if !model.LabelName(m.Name).IsValid() {
|
|
return fmt.Errorf("invalid label name %q", m.Name)
|
|
}
|
|
switch m.Type {
|
|
case pb.Matcher_EQUAL:
|
|
if !model.LabelValue(m.Pattern).IsValid() {
|
|
return fmt.Errorf("invalid label value %q", m.Pattern)
|
|
}
|
|
case pb.Matcher_REGEXP:
|
|
if _, err := regexp.Compile(m.Pattern); err != nil {
|
|
return fmt.Errorf("invalid regular expression %q: %s", m.Pattern, err)
|
|
}
|
|
default:
|
|
return fmt.Errorf("unknown matcher type %q", m.Type)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateSilence(s *pb.Silence) error {
|
|
if s.Id == "" {
|
|
return errors.New("ID missing")
|
|
}
|
|
if len(s.Matchers) == 0 {
|
|
return errors.New("at least one matcher required")
|
|
}
|
|
for i, m := range s.Matchers {
|
|
if err := validateMatcher(m); err != nil {
|
|
return fmt.Errorf("invalid label matcher %d: %s", i, err)
|
|
}
|
|
}
|
|
if s.StartsAt.IsZero() {
|
|
return errors.New("invalid zero start timestamp")
|
|
}
|
|
if s.EndsAt.IsZero() {
|
|
return errors.New("invalid zero end timestamp")
|
|
}
|
|
if s.EndsAt.Before(s.StartsAt) {
|
|
return errors.New("end time must not be before start time")
|
|
}
|
|
if s.UpdatedAt.IsZero() {
|
|
return errors.New("invalid zero update timestamp")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// cloneSilence returns a shallow copy of a silence.
|
|
func cloneSilence(sil *pb.Silence) *pb.Silence {
|
|
s := *sil
|
|
return &s
|
|
}
|
|
|
|
func (s *Silences) getSilence(id string) (*pb.Silence, bool) {
|
|
msil, ok := s.st.data[id]
|
|
if !ok {
|
|
return nil, false
|
|
}
|
|
return msil.Silence, true
|
|
}
|
|
|
|
func (s *Silences) setSilence(sil *pb.Silence) error {
|
|
sil.UpdatedAt = s.now()
|
|
|
|
if err := validateSilence(sil); err != nil {
|
|
return errors.Wrap(err, "silence invalid")
|
|
}
|
|
|
|
msil := &pb.MeshSilence{
|
|
Silence: sil,
|
|
ExpiresAt: sil.EndsAt.Add(s.retention),
|
|
}
|
|
st := &gossipData{
|
|
data: silenceMap{sil.Id: msil},
|
|
}
|
|
|
|
s.st.Merge(st)
|
|
// setSilence() is called with s.mtx locked, which can produce
|
|
// a deadlock if we call GossipBroadcast from here.
|
|
go s.gossip.GossipBroadcast(st)
|
|
|
|
return nil
|
|
}
|
|
|
|
// Set the specified silence. If a silence with the ID already exists and the modification
|
|
// modifies history, the old silence gets expired and a new one is created.
|
|
func (s *Silences) Set(sil *pb.Silence) (string, error) {
|
|
s.mtx.Lock()
|
|
defer s.mtx.Unlock()
|
|
|
|
now := s.now()
|
|
prev, ok := s.getSilence(sil.Id)
|
|
|
|
if sil.Id != "" && !ok {
|
|
return "", ErrNotFound
|
|
}
|
|
if ok {
|
|
if canUpdate(prev, sil, now) {
|
|
return sil.Id, s.setSilence(sil)
|
|
}
|
|
if getState(prev, s.now()) != types.SilenceStateExpired {
|
|
// We cannot update the silence, expire the old one.
|
|
if err := s.expire(prev.Id); err != nil {
|
|
return "", errors.Wrap(err, "expire previous silence")
|
|
}
|
|
}
|
|
}
|
|
// If we got here it's either a new silence or a replacing one.
|
|
sil.Id = uuid.NewV4().String()
|
|
|
|
if sil.StartsAt.Before(now) {
|
|
sil.StartsAt = now
|
|
}
|
|
|
|
return sil.Id, s.setSilence(sil)
|
|
}
|
|
|
|
// canUpdate returns true if silence a can be updated to b without
|
|
// affecting the historic view of silencing.
|
|
func canUpdate(a, b *pb.Silence, now time.Time) bool {
|
|
if !reflect.DeepEqual(a.Matchers, b.Matchers) {
|
|
return false
|
|
}
|
|
// Allowed timestamp modifications depend on the current time.
|
|
switch st := getState(a, now); st {
|
|
case types.SilenceStateActive:
|
|
if !b.StartsAt.Equal(a.StartsAt) {
|
|
return false
|
|
}
|
|
if b.EndsAt.Before(now) {
|
|
return false
|
|
}
|
|
case types.SilenceStatePending:
|
|
if b.StartsAt.Before(now) {
|
|
return false
|
|
}
|
|
case types.SilenceStateExpired:
|
|
return false
|
|
default:
|
|
panic("unknown silence state")
|
|
}
|
|
return true
|
|
}
|
|
|
|
// Expire the silence with the given ID immediately.
|
|
func (s *Silences) Expire(id string) error {
|
|
s.mtx.Lock()
|
|
defer s.mtx.Unlock()
|
|
return s.expire(id)
|
|
}
|
|
|
|
// Expire the silence with the given ID immediately.
|
|
func (s *Silences) expire(id string) error {
|
|
sil, ok := s.getSilence(id)
|
|
if !ok {
|
|
return ErrNotFound
|
|
}
|
|
sil = cloneSilence(sil)
|
|
now := s.now()
|
|
|
|
switch getState(sil, now) {
|
|
case types.SilenceStateExpired:
|
|
return errors.Errorf("silence %s already expired", id)
|
|
case types.SilenceStateActive:
|
|
sil.EndsAt = now
|
|
case types.SilenceStatePending:
|
|
// Set both to now to make Silence move to "expired" state
|
|
sil.StartsAt = now
|
|
sil.EndsAt = now
|
|
}
|
|
|
|
return s.setSilence(sil)
|
|
}
|
|
|
|
// QueryParam expresses parameters along which silences are queried.
|
|
type QueryParam func(*query) error
|
|
|
|
type query struct {
|
|
ids []string
|
|
filters []silenceFilter
|
|
}
|
|
|
|
// silenceFilter is a function that returns true if a silence
|
|
// should be dropped from a result set for a given time.
|
|
type silenceFilter func(*pb.Silence, *Silences, time.Time) (bool, error)
|
|
|
|
var errNotSupported = errors.New("query parameter not supported")
|
|
|
|
// QIDs configures a query to select the given silence IDs.
|
|
func QIDs(ids ...string) QueryParam {
|
|
return func(q *query) error {
|
|
q.ids = append(q.ids, ids...)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// QTimeRange configures a query to search for silences that are active
|
|
// in the given time range.
|
|
// TODO(fabxc): not supported yet.
|
|
func QTimeRange(start, end time.Time) QueryParam {
|
|
return func(q *query) error {
|
|
return errNotSupported
|
|
}
|
|
}
|
|
|
|
// QMatches returns silences that match the given label set.
|
|
func QMatches(set model.LabelSet) QueryParam {
|
|
return func(q *query) error {
|
|
f := func(sil *pb.Silence, s *Silences, _ time.Time) (bool, error) {
|
|
m, err := s.mc.Get(sil)
|
|
if err != nil {
|
|
return true, err
|
|
}
|
|
return m.Match(set), nil
|
|
}
|
|
q.filters = append(q.filters, f)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// getState returns a silence's SilenceState at the given timestamp.
|
|
func getState(sil *pb.Silence, ts time.Time) types.SilenceState {
|
|
if ts.Before(sil.StartsAt) {
|
|
return types.SilenceStatePending
|
|
}
|
|
if ts.After(sil.EndsAt) {
|
|
return types.SilenceStateExpired
|
|
}
|
|
return types.SilenceStateActive
|
|
}
|
|
|
|
// QState filters queried silences by the given states.
|
|
func QState(states ...types.SilenceState) QueryParam {
|
|
return func(q *query) error {
|
|
f := func(sil *pb.Silence, _ *Silences, now time.Time) (bool, error) {
|
|
s := getState(sil, now)
|
|
|
|
for _, ps := range states {
|
|
if s == ps {
|
|
return true, nil
|
|
}
|
|
}
|
|
return false, nil
|
|
}
|
|
q.filters = append(q.filters, f)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// QueryOne queries with the given parameters and returns the first result.
|
|
// Returns ErrNotFound if the query result is empty.
|
|
func (s *Silences) QueryOne(params ...QueryParam) (*pb.Silence, error) {
|
|
res, err := s.Query(params...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(res) == 0 {
|
|
return nil, ErrNotFound
|
|
}
|
|
return res[0], nil
|
|
}
|
|
|
|
// Query for silences based on the given query parameters.
|
|
func (s *Silences) Query(params ...QueryParam) ([]*pb.Silence, error) {
|
|
start := time.Now()
|
|
s.metrics.queriesTotal.Inc()
|
|
|
|
sils, err := func() ([]*pb.Silence, error) {
|
|
q := &query{}
|
|
for _, p := range params {
|
|
if err := p(q); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return s.query(q, s.now())
|
|
}()
|
|
if err != nil {
|
|
s.metrics.queryErrorsTotal.Inc()
|
|
}
|
|
s.metrics.queryDuration.Observe(time.Since(start).Seconds())
|
|
return sils, err
|
|
}
|
|
|
|
// Count silences by state.
|
|
func (s *Silences) CountState(states ...types.SilenceState) (int, error) {
|
|
// This could probably be optimized.
|
|
sils, err := s.Query(QState(states...))
|
|
if err != nil {
|
|
return -1, err
|
|
}
|
|
return len(sils), nil
|
|
}
|
|
|
|
func (s *Silences) query(q *query, now time.Time) ([]*pb.Silence, error) {
|
|
// If we have an ID constraint, all silences are our base set.
|
|
// This and the use of post-filter functions is the
|
|
// the trivial solution for now.
|
|
var res []*pb.Silence
|
|
|
|
s.mtx.Lock()
|
|
defer s.mtx.Unlock()
|
|
|
|
if q.ids != nil {
|
|
for _, id := range q.ids {
|
|
if s, ok := s.st.data[string(id)]; ok {
|
|
res = append(res, s.Silence)
|
|
}
|
|
}
|
|
} else {
|
|
for _, sil := range s.st.data {
|
|
res = append(res, sil.Silence)
|
|
}
|
|
}
|
|
|
|
var resf []*pb.Silence
|
|
for _, sil := range res {
|
|
remove := false
|
|
for _, f := range q.filters {
|
|
ok, err := f(sil, s, now)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !ok {
|
|
remove = true
|
|
break
|
|
}
|
|
}
|
|
if !remove {
|
|
resf = append(resf, cloneSilence(sil))
|
|
}
|
|
}
|
|
|
|
return resf, nil
|
|
}
|
|
|
|
// loadSnapshot loads a snapshot generated by Snapshot() into the state.
|
|
// Any previous state is wiped.
|
|
func (s *Silences) loadSnapshot(r io.Reader) error {
|
|
st := newGossipData()
|
|
|
|
s.mtx.Lock()
|
|
defer s.mtx.Unlock()
|
|
|
|
for {
|
|
var sil pb.MeshSilence
|
|
if _, err := pbutil.ReadDelimited(r, &sil); err != nil {
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
return err
|
|
}
|
|
// Comments list was moved to a single comment. Upgrade on loading the snapshot.
|
|
if len(sil.Silence.Comments) > 0 {
|
|
sil.Silence.Comment = sil.Silence.Comments[0].Comment
|
|
sil.Silence.CreatedBy = sil.Silence.Comments[0].Author
|
|
sil.Silence.Comments = nil
|
|
}
|
|
|
|
st.data[sil.Silence.Id] = &sil
|
|
_, err := s.mc.Get(sil.Silence)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
s.st.data = st.data
|
|
|
|
return nil
|
|
}
|
|
|
|
// Snapshot writes the full internal state into the writer and returns the number of bytes
|
|
// written.
|
|
func (s *Silences) Snapshot(w io.Writer) (int, error) {
|
|
start := time.Now()
|
|
defer func() { s.metrics.snapshotDuration.Observe(time.Since(start).Seconds()) }()
|
|
|
|
s.mtx.Lock()
|
|
defer s.mtx.Unlock()
|
|
|
|
var n int
|
|
for _, s := range s.st.data {
|
|
m, err := pbutil.WriteDelimited(w, s)
|
|
if err != nil {
|
|
return n + m, err
|
|
}
|
|
n += m
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
type gossiper struct {
|
|
*Silences
|
|
}
|
|
|
|
// Gossip implements the mesh.Gossiper interface.
|
|
func (g gossiper) Gossip() mesh.GossipData {
|
|
g.mtx.Lock()
|
|
defer g.mtx.Unlock()
|
|
|
|
return g.st.clone()
|
|
}
|
|
|
|
// OnGossip implements the mesh.Gossiper interface.
|
|
func (g gossiper) OnGossip(msg []byte) (mesh.GossipData, error) {
|
|
gd, err := decodeGossipData(msg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
g.mtx.Lock()
|
|
defer g.mtx.Unlock()
|
|
|
|
if delta := g.st.mergeDelta(gd); len(delta.data) > 0 {
|
|
return delta, nil
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
// OnGossipBroadcast implements the mesh.Gossiper interface.
|
|
func (g gossiper) OnGossipBroadcast(src mesh.PeerName, msg []byte) (mesh.GossipData, error) {
|
|
gd, err := decodeGossipData(msg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
g.mtx.Lock()
|
|
defer g.mtx.Unlock()
|
|
|
|
return g.st.mergeDelta(gd), nil
|
|
}
|
|
|
|
// OnGossipUnicast implements the mesh.Gossiper interface.
|
|
// It always panics.
|
|
func (g gossiper) OnGossipUnicast(src mesh.PeerName, msg []byte) error {
|
|
panic("not implemented")
|
|
}
|
|
|
|
type silenceMap map[string]*pb.MeshSilence
|
|
|
|
type gossipData struct {
|
|
data silenceMap
|
|
mtx sync.RWMutex
|
|
}
|
|
|
|
var _ mesh.GossipData = &gossipData{}
|
|
|
|
func newGossipData() *gossipData {
|
|
return &gossipData{
|
|
data: silenceMap{},
|
|
}
|
|
}
|
|
|
|
func decodeGossipData(msg []byte) (*gossipData, error) {
|
|
gd := newGossipData()
|
|
rd := bytes.NewReader(msg)
|
|
|
|
for {
|
|
var s pb.MeshSilence
|
|
if _, err := pbutil.ReadDelimited(rd, &s); err != nil {
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
return gd, err
|
|
}
|
|
gd.data[s.Silence.Id] = &s
|
|
}
|
|
return gd, nil
|
|
}
|
|
|
|
// Encode implements the mesh.GossipData interface.
|
|
func (gd *gossipData) Encode() [][]byte {
|
|
// Split into sub-messages of ~1MB.
|
|
const maxSize = 1024 * 1024
|
|
|
|
var (
|
|
buf bytes.Buffer
|
|
res [][]byte
|
|
n int
|
|
)
|
|
|
|
gd.mtx.RLock()
|
|
defer gd.mtx.RUnlock()
|
|
|
|
for _, s := range gd.data {
|
|
m, err := pbutil.WriteDelimited(&buf, s)
|
|
n += m
|
|
if err != nil {
|
|
// TODO(fabxc): log error and skip entry. Or can this really not happen with a bytes.Buffer?
|
|
panic(err)
|
|
}
|
|
if n > maxSize {
|
|
res = append(res, buf.Bytes())
|
|
buf = bytes.Buffer{}
|
|
}
|
|
}
|
|
if buf.Len() > 0 {
|
|
res = append(res, buf.Bytes())
|
|
}
|
|
return res
|
|
}
|
|
|
|
func (gd *gossipData) clone() *gossipData {
|
|
gd.mtx.RLock()
|
|
defer gd.mtx.RUnlock()
|
|
|
|
data := make(silenceMap, len(gd.data))
|
|
for id, s := range gd.data {
|
|
data[id] = s
|
|
}
|
|
return &gossipData{data: data}
|
|
}
|
|
|
|
// Merge the silence set with gossip data and return a new silence state.
|
|
func (gd *gossipData) Merge(other mesh.GossipData) mesh.GossipData {
|
|
ot := other.(*gossipData)
|
|
ot.mtx.RLock()
|
|
defer ot.mtx.RUnlock()
|
|
|
|
gd.mtx.Lock()
|
|
defer gd.mtx.Unlock()
|
|
|
|
for id, s := range ot.data {
|
|
// Comments list was moved to a single comment. Apply upgrade
|
|
// on silences received from peers.
|
|
if len(s.Silence.Comments) > 0 {
|
|
s.Silence.Comment = s.Silence.Comments[0].Comment
|
|
s.Silence.CreatedBy = s.Silence.Comments[0].Author
|
|
s.Silence.Comments = nil
|
|
}
|
|
|
|
prev, ok := gd.data[id]
|
|
if !ok {
|
|
gd.data[id] = s
|
|
continue
|
|
}
|
|
if prev.Silence.UpdatedAt.Before(s.Silence.UpdatedAt) {
|
|
gd.data[id] = s
|
|
}
|
|
}
|
|
return gd
|
|
}
|
|
|
|
// mergeDelta behaves like Merge but ignores expired silences, and
|
|
// returns a gossipData only containing things that have changed.
|
|
func (gd *gossipData) mergeDelta(od *gossipData) *gossipData {
|
|
delta := newGossipData()
|
|
|
|
od.mtx.RLock()
|
|
defer od.mtx.RUnlock()
|
|
|
|
gd.mtx.Lock()
|
|
defer gd.mtx.Unlock()
|
|
|
|
for id, s := range od.data {
|
|
// If a gossiped silence is expired, skip it.
|
|
// For a given silence duration exceeding a few minutes,
|
|
// active silences will have already been gossiped.
|
|
// Once the active silence is gossiped, its expiration
|
|
// should happen more or less simultaneously on the different
|
|
// alertmanager nodes. Preventing the gossiping of expired
|
|
// silences allows them to be GC'd, and doesn't affect
|
|
// consistency across the mesh.
|
|
if !s.ExpiresAt.After(utcNow()) {
|
|
continue
|
|
}
|
|
|
|
// Comments list was moved to a single comment. Apply upgrade
|
|
// on silences received from peers.
|
|
if len(s.Silence.Comments) > 0 {
|
|
s.Silence.Comment = s.Silence.Comments[0].Comment
|
|
s.Silence.CreatedBy = s.Silence.Comments[0].Author
|
|
s.Silence.Comments = nil
|
|
}
|
|
|
|
prev, ok := gd.data[id]
|
|
if !ok {
|
|
gd.data[id] = s
|
|
delta.data[id] = s
|
|
continue
|
|
}
|
|
if prev.Silence.UpdatedAt.Before(s.Silence.UpdatedAt) {
|
|
gd.data[id] = s
|
|
delta.data[id] = s
|
|
}
|
|
}
|
|
return delta
|
|
}
|
|
|
|
// replaceFile wraps a file that is moved to another filename on closing.
|
|
type replaceFile struct {
|
|
*os.File
|
|
filename string
|
|
}
|
|
|
|
func (f *replaceFile) Close() error {
|
|
if err := f.File.Sync(); err != nil {
|
|
return err
|
|
}
|
|
if err := f.File.Close(); err != nil {
|
|
return err
|
|
}
|
|
return os.Rename(f.File.Name(), f.filename)
|
|
}
|
|
|
|
// openReplace opens a new temporary file that is moved to filename on closing.
|
|
func openReplace(filename string) (*replaceFile, error) {
|
|
tmpFilename := fmt.Sprintf("%s.%x", filename, uint64(rand.Int63()))
|
|
|
|
f, err := os.Create(tmpFilename)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rf := &replaceFile{
|
|
File: f,
|
|
filename: filename,
|
|
}
|
|
return rf, nil
|
|
}
|