github/karma
1
0
Fork 0
mirror of https://github.com/prymitive/karma synced 2026-08-23 11:56:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(backend): add alert history source rewrite rules

Browse Source 
Fixes #3064
This commit is contained in:
Łukasz Mierzwa
2021-05-04 13:28:35 +01:00
committed by Łukasz Mierzwa
parent 266c15cd88
commit c812c25393
16 changed files with 267 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+6
View File
@@ -5,6 +5,12 @@
### Fixed
- History queries were always failing due to wrong Prometheus API usage.
- URI handling for silence requests when proxy is used #3060.
### Added
- Ability to rewrite source URIs for alert history via `history:rewrite`
config section #3064.
## v0.84
cmd/karma/alert_history.go
+25 -5
View File
@@ -183,12 +183,13 @@ func (hp *historyPoller) knownBadLookup(key string) (*knownBadUpstream, bool) {
func (hp *historyPoller) startWorker(wid int) {
log.Debug().Int("worker", wid).Int("queue", cap(hp.queue)).Dur("timeout", hp.queryTimeout).Msg("Starting history poller")
for j := range hp.queue {
sourceURI := rewriteSource(config.Config.History.Rewrite, j.uri)
expiredAt := time.Now().Add(time.Minute * -5)
key := hashQuery(j.uri, j.labels)
key := hashQuery(sourceURI, j.labels)
if kb, found := hp.knownBadLookup(key); found && kb.timestamp.After(expiredAt) {
log.Debug().
Int("worker", wid).
Str("uri", j.uri).
Str("uri", sourceURI).
Interface("labels", j.labels).
Str("key", key).
Msg("Upstream already marked as invalid, skipping")
@@ -198,19 +199,19 @@ func (hp *historyPoller) startWorker(wid int) {
if v := hp.cacheLookup(key); v != nil && v.timestamp.After(expiredAt) {
log.Debug().
Int("worker", wid).
Str("uri", j.uri).
Str("uri", sourceURI).
Interface("labels", j.labels).
Str("key", key).
Msg("Got results from cache")
j.result <- historyQueryResult{values: v.values, err: nil}
continue
}
values, err := countAlerts(j.uri, hp.queryTimeout, j.labels)
values, err := countAlerts(sourceURI, hp.queryTimeout, j.labels)
if err != nil {
log.Error().
Err(err).
Int("worker", wid).
Str("uri", j.uri).
Str("uri", sourceURI).
Interface("labels", j.labels).
Msg("History query failed")
hp.knownBadSave(key, knownBadUpstream{timestamp: time.Now(), err: err})
@@ -233,7 +234,26 @@ func hashQuery(uri string, labels map[string]string) string {
return fmt.Sprintf("%x", hasher.Sum(nil))
}
func rewriteSource(rules []config.HistoryRewrite, uri string) string {
for _, rule := range rules {
if !rule.SourceRegex.MatchString(uri) {
continue
}
result := []byte{}
for _, submatches := range rule.SourceRegex.FindAllStringSubmatchIndex(uri, -1) {
result = rule.SourceRegex.ExpandString(result, rule.URI, uri, submatches)
}
log.Debug().Str("source", uri).Str("uri", string(result)).Msg("Alert history source rewrite")
return string(result)
}
return uri
}
func countAlerts(uri string, timeout time.Duration, labels map[string]string) (ret []OffsetSample, err error) {
if uri == "" {
return
}
client, err := api.NewClient(api.Config{
Address: uri,
RoundTripper: http.DefaultTransport,
cmd/karma/alert_history_test.go
+153 -2
View File
@@ -13,6 +13,7 @@ import (
"github.com/jarcoal/httpmock"
"github.com/prometheus/common/model"
"github.com/prymitive/karma/internal/config"
"github.com/prymitive/karma/internal/regex"
"github.com/rs/zerolog"
)
@@ -95,6 +96,7 @@ func TestAlertHistory(t *testing.T) {
enabled bool
timeout time.Duration
workers int
rewrite []config.HistoryRewrite
}
type historyQuery struct {
@@ -318,12 +320,98 @@ func TestAlertHistory(t *testing.T) {
},
},
},
{
mocks: []mock{},
config: cfg{
enabled: true,
timeout: time.Second * 5,
workers: 5,
rewrite: []config.HistoryRewrite{
{
SourceRegex: regex.MustCompileAnchored(".+"),
URI: "",
},
},
},
queries: []historyQuery{
{
payload: generateHistoryPayload(AlertHistoryPayload{
Sources: []string{"http://localhost:9092"},
Labels: map[string]string{"alertname": "Fake Alert", "cluster": "prod"},
}),
code: 200,
response: AlertHistoryResponse{
Samples: generateHistorySamples(generateIntSlice(0, 0, 24), time.Hour),
},
},
},
},
{
mocks: []mock{
{
method: "GET",
uri: regexp.MustCompile("^http://localhost:9100/api/v1/labels"),
responder: httpmock.NewJsonResponderOrPanic(200, prometheusAPIV1Labels{
Status: "success",
Data: []string{"alertname", "instance", "job"},
}),
},
{
method: "POST",
uri: regexp.MustCompile("^http://localhost:9100/api/v1/query_range"),
responder: httpmock.NewJsonResponderOrPanic(200, prometheusAPIV1QueryRange{
Status: "success",
Data: generateV1Matrix(
[]seriesValues{
{
metric: model.Metric{
"alertname": "Fake Alert",
},
values: generateIntSlice(0, 1, 24),
},
}, time.Hour),
}),
},
},
config: cfg{
enabled: true,
timeout: time.Second * 5,
workers: 5,
rewrite: []config.HistoryRewrite{
{
SourceRegex: regex.MustCompileAnchored("http://(.+):1111"),
URI: "http://$1:9100",
},
{
SourceRegex: regex.MustCompileAnchored("foo"),
URI: "",
},
{
SourceRegex: regex.MustCompileAnchored("http://(.+):909[0-9]"),
URI: "http://$1:9100",
},
},
},
queries: []historyQuery{
{
payload: generateHistoryPayload(AlertHistoryPayload{
Sources: []string{"http://localhost:9090", "http://localhost:9091", "http://localhost:1111"},
Labels: map[string]string{"alertname": "Fake Alert", "cluster": "prod"},
}),
code: 200,
response: AlertHistoryResponse{
Samples: generateHistorySamples(generateIntSlice(0, 3, 24), time.Hour),
},
},
},
},
}
defer func() {
config.Config.History.Enabled = true
config.Config.History.Timeout = time.Second * 20
config.Config.History.Workers = 30
config.Config.History.Rewrite = []config.HistoryRewrite{}
}()
httpmock.Activate()
@@ -353,6 +441,7 @@ func TestAlertHistory(t *testing.T) {
config.Config.History.Enabled = tc.config.enabled
config.Config.History.Timeout = tc.config.timeout
config.Config.History.Workers = tc.config.workers
config.Config.History.Rewrite = tc.config.rewrite
for _, q := range tc.queries {
t.Logf("Body: %s", string(q.payload))
@@ -408,9 +497,71 @@ func TestAbsTime(t *testing.T) {
t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
d := absTimeDiff(tc.a, tc.b)
if diff := cmp.Diff(tc.diff, d); diff != "" {
t.Errorf("Incorrect absTimeDiff(-want +got):\n%s", diff)
t.Errorf("Incorrect absTimeDiff result (-want +got):\n%s", diff)
}
})
}
}
func TestRewriteSource(t *testing.T) {
type testCaseT struct {
rules []config.HistoryRewrite
uri string
out string
}
testCases := []testCaseT{
{
rules: []config.HistoryRewrite{},
uri: "http://localhost",
out: "http://localhost",
},
{
rules: []config.HistoryRewrite{
{
SourceRegex: regex.MustCompileAnchored("foo"),
URI: "foo",
},
},
uri: "http://localhost",
out: "http://localhost",
},
{
rules: []config.HistoryRewrite{
{
SourceRegex: regex.MustCompileAnchored("foo"),
URI: "foo",
},
{
SourceRegex: regex.MustCompileAnchored("http://local.+"),
URI: "foo",
},
},
uri: "http://localhost",
out: "foo",
},
{
rules: []config.HistoryRewrite{
{
SourceRegex: regex.MustCompileAnchored("foo"),
URI: "foo",
},
{
SourceRegex: regex.MustCompileAnchored("http://(.+).example.com"),
URI: "https://prom-$1.example.com",
},
},
uri: "http://prod.example.com",
out: "https://prom-prod.example.com",
},
}
for i, tc := range testCases {
t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
uri := rewriteSource(tc.rules, tc.uri)
if diff := cmp.Diff(tc.out, uri); diff != "" {
t.Errorf("Incorrect rewriteSource result (-want +got):\n%s", diff)
}
})
}
}
cmd/karma/tests/testscript/039_alert_history_invalid_regex.txt
+18
View File
@@ -0,0 +1,18 @@
# Raises an error if history.rewrite rule contains invalid regex
karma.bin-should-fail --check-config
! stdout .
cmp stderr stderr.txt
-- stderr.txt --
level=error msg="Execution failed" error="history.rewrite source regex \"foo.++++++\" is invalid: error parsing regexp: invalid nested repetition operator: `++`"
-- karma.yaml --
alertmanager:
servers:
- name: default
uri: https://127.0.0.1:9093
history:
rewrite:
- source: bar.+
uri: http://bar
- source: foo.++++++
uri: http://foo
cmd/karma/tests/testscript/059_log_full_config_env.txt
+1
View File
@@ -157,6 +157,7 @@ level=info msg="history:"
level=info msg=" enabled: true"
level=info msg=" workers: 30"
level=info msg=" timeout: 20s"
level=info msg=" rewrite: []"
level=info msg="karma:"
level=info msg=" name: karma-demo"
level=info msg="labels:"
cmd/karma/tests/testscript/060_log_full_config_file.txt
+13 -3
View File
@@ -168,9 +168,12 @@ level=info msg=" order:"
level=info msg=" - severity"
level=info msg=" - cluster"
level=info msg="history:"
level=info msg=" enabled: true"
level=info msg=" workers: 30"
level=info msg=" timeout: 20s"
level=info msg=" enabled: false"
level=info msg=" workers: 123"
level=info msg=" timeout: 1h0m0s"
level=info msg=" rewrite:"
level=info msg=" - source: http://(.+).example.com"
level=info msg=" uri: https://prod-$1.example.com"
level=info msg="karma:"
level=info msg=" name: karma-demo"
level=info msg="labels:"
@@ -364,6 +367,13 @@ grid:
order:
- severity
- cluster
history:
enabled: false
workers: 123
timeout: 1h
rewrite:
- source: "http://(.+).example.com"
uri: "https://prod-$1.example.com"
karma:
name: karma-demo
labels:
cmd/karma/tests/testscript/065_proxy-with-readonly.txt
+1
View File
@@ -74,6 +74,7 @@ level=info msg="history:"
level=info msg=" enabled: true"
level=info msg=" workers: 30"
level=info msg=" timeout: 20s"
level=info msg=" rewrite: []"
level=info msg="karma:"
level=info msg=" name: karma"
level=info msg="labels:"
cmd/karma/tests/testscript/066_proxy.txt
+1
View File
@@ -74,6 +74,7 @@ level=info msg="history:"
level=info msg=" enabled: true"
level=info msg=" workers: 30"
level=info msg=" timeout: 20s"
level=info msg=" rewrite: []"
level=info msg="karma:"
level=info msg=" name: karma"
level=info msg="labels:"
cmd/karma/tests/testscript/067_readonly.txt
+1
View File
@@ -74,6 +74,7 @@ level=info msg="history:"
level=info msg=" enabled: true"
level=info msg=" workers: 30"
level=info msg=" timeout: 20s"
level=info msg=" rewrite: []"
level=info msg="karma:"
level=info msg=" name: karma"
level=info msg="labels:"
cmd/karma/tests/testscript/068_sentry.txt
+1
View File
@@ -76,6 +76,7 @@ level=info msg="history:"
level=info msg=" enabled: true"
level=info msg=" workers: 30"
level=info msg=" timeout: 20s"
level=info msg=" rewrite: []"
level=info msg="karma:"
level=info msg=" name: karma"
level=info msg="labels:"
cmd/karma/tests/testscript/070_upper_case_keys.txt
+1
View File
@@ -74,6 +74,7 @@ level=info msg="history:"
level=info msg=" enabled: true"
level=info msg=" workers: 30"
level=info msg=" timeout: 20s"
level=info msg=" rewrite: []"
level=info msg="karma:"
level=info msg=" name: karma"
level=info msg="labels:"
cmd/karma/tests/testscript/097_proxy_url_config.txt
+1
View File
@@ -74,6 +74,7 @@ level=info msg="history:"
level=info msg=" enabled: true"
level=info msg=" workers: 30"
level=info msg=" timeout: 20s"
level=info msg=" rewrite: []"
level=info msg="karma:"
level=info msg=" name: karma"
level=info msg="labels:"
docs/CONFIGURATION.md
+30
View File
@@ -725,6 +725,9 @@ history:
enabled: bool
timeout: duration
workers: integer
rewrite:
- source: regex
uri: string
```
- `enabled` - enable alert history UI and backend query support
@@ -732,6 +735,11 @@ history:
- `workers` - number of worker threads to start, each worker handles
one outgoing HTTP request, more workers allows to handle more concurrent
queries if you have a large number of Prometheus servers sending alerts
- `rewrite` - list of source rewrite rules applied before any request is send
to remote Prometheus. Rewrite rules can be used to modify URI used by karma
when connecting to Prometheus API if `source` field in alert uses addresses
not reachable from karma.
All regexes are anchored, `${N}` syntax can be used for capture groups.
Defaults:
@@ -740,6 +748,28 @@ history:
enabled: true
timeout: 20s
workers: 30
rewrite: []
```
Example with rewrite rule that will replace `https://prometheus.example.com`
with `http://localhost:9093`:
```YAML
history:
rewrite:
- source: 'https://prometheus.example.com'
uri: 'http://localhost:9093'
```
Example with rewrite rule that will replace `https://*.example.com` with
`http://prometheus-*.internal` (`https://dev.example.com` becomes
`http://prometheus-dev.example.com`):
```YAML
history:
rewrite:
- source: 'https://(.+).example.com'
uri: 'http://prometheus-$1.internal'
```
### Karma
internal/config/config.go
+7
View File
@@ -373,6 +373,13 @@ func (config *configSchema) Read(flags *pflag.FlagSet) (string, error) {
return "", fmt.Errorf("listen.tls.cert must be set when listen.tls.key is set")
}
for i := 0; i < len(config.History.Rewrite); i++ {
config.History.Rewrite[i].SourceRegex, err = regex.CompileAnchored(config.History.Rewrite[i].Source)
if err != nil {
return "", fmt.Errorf("history.rewrite source regex %q is invalid: %v", config.History.Rewrite[i].Source, err)
}
}
// accept single Alertmanager server from flag/env if nothing is set yet
if len(config.Alertmanager.Servers) == 0 && config.Alertmanager.URI != "" {
config.Alertmanager.Servers = []AlertmanagerConfig{
internal/config/config_test.go
+1
View File
@@ -92,6 +92,7 @@ history:
enabled: true
workers: 30
timeout: 20s
rewrite: []
karma:
name: another karma
labels:
internal/config/models.go
+7
View File
@@ -58,6 +58,12 @@ type AuthorizationGroup struct {
Members []string
}
type HistoryRewrite struct {
Source string `yaml:"source"`
SourceRegex *regexp.Regexp `yaml:"-"`
URI string `yaml:"uri"`
}
type configSchema struct {
Authentication struct {
Enabled bool `yaml:"-" koanf:"-"`
@@ -132,6 +138,7 @@ type configSchema struct {
Enabled bool
Workers int
Timeout time.Duration
Rewrite []HistoryRewrite
}
Karma struct {
Name string