mirror of
https://github.com/prometheus/prometheus.git
synced 2024-11-09 23:24:05 -08:00
c187308366
This is based on https://github.com/prometheus/prometheus/pull/1997. This adds contexts to the relevant Storage methods and already passes PromQL's new per-query context into the storage's query methods. The immediate motivation supporting multi-tenancy in Frankenstein, but this could also be used by Prometheus's normal local storage to support cancellations and timeouts at some point.
226 lines
5.5 KiB
Go
226 lines
5.5 KiB
Go
// Copyright 2016 The Prometheus Authors
|
|
// 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 promql
|
|
|
|
import (
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
|
|
"golang.org/x/net/context"
|
|
)
|
|
|
|
func TestQueryConcurrency(t *testing.T) {
|
|
engine := NewEngine(nil, nil)
|
|
ctx, cancelCtx := context.WithCancel(context.Background())
|
|
defer cancelCtx()
|
|
|
|
block := make(chan struct{})
|
|
processing := make(chan struct{})
|
|
|
|
f := func(context.Context) error {
|
|
processing <- struct{}{}
|
|
<-block
|
|
return nil
|
|
}
|
|
|
|
for i := 0; i < DefaultEngineOptions.MaxConcurrentQueries; i++ {
|
|
q := engine.newTestQuery(f)
|
|
go q.Exec(ctx)
|
|
select {
|
|
case <-processing:
|
|
// Expected.
|
|
case <-time.After(20 * time.Millisecond):
|
|
t.Fatalf("Query within concurrency threshold not being executed")
|
|
}
|
|
}
|
|
|
|
q := engine.newTestQuery(f)
|
|
go q.Exec(ctx)
|
|
|
|
select {
|
|
case <-processing:
|
|
t.Fatalf("Query above concurrency threshold being executed")
|
|
case <-time.After(20 * time.Millisecond):
|
|
// Expected.
|
|
}
|
|
|
|
// Terminate a running query.
|
|
block <- struct{}{}
|
|
|
|
select {
|
|
case <-processing:
|
|
// Expected.
|
|
case <-time.After(20 * time.Millisecond):
|
|
t.Fatalf("Query within concurrency threshold not being executed")
|
|
}
|
|
|
|
// Terminate remaining queries.
|
|
for i := 0; i < DefaultEngineOptions.MaxConcurrentQueries; i++ {
|
|
block <- struct{}{}
|
|
}
|
|
}
|
|
|
|
func TestQueryTimeout(t *testing.T) {
|
|
engine := NewEngine(nil, &EngineOptions{
|
|
Timeout: 5 * time.Millisecond,
|
|
MaxConcurrentQueries: 20,
|
|
})
|
|
ctx, cancelCtx := context.WithCancel(context.Background())
|
|
defer cancelCtx()
|
|
|
|
query := engine.newTestQuery(func(ctx context.Context) error {
|
|
time.Sleep(50 * time.Millisecond)
|
|
return contextDone(ctx, "test statement execution")
|
|
})
|
|
|
|
res := query.Exec(ctx)
|
|
if res.Err == nil {
|
|
t.Fatalf("expected timeout error but got none")
|
|
}
|
|
if _, ok := res.Err.(ErrQueryTimeout); res.Err != nil && !ok {
|
|
t.Fatalf("expected timeout error but got: %s", res.Err)
|
|
}
|
|
}
|
|
|
|
func TestQueryCancel(t *testing.T) {
|
|
engine := NewEngine(nil, nil)
|
|
ctx, cancelCtx := context.WithCancel(context.Background())
|
|
defer cancelCtx()
|
|
|
|
// Cancel a running query before it completes.
|
|
block := make(chan struct{})
|
|
processing := make(chan struct{})
|
|
|
|
query1 := engine.newTestQuery(func(ctx context.Context) error {
|
|
processing <- struct{}{}
|
|
<-block
|
|
return contextDone(ctx, "test statement execution")
|
|
})
|
|
|
|
var res *Result
|
|
|
|
go func() {
|
|
res = query1.Exec(ctx)
|
|
processing <- struct{}{}
|
|
}()
|
|
|
|
<-processing
|
|
query1.Cancel()
|
|
block <- struct{}{}
|
|
<-processing
|
|
|
|
if res.Err == nil {
|
|
t.Fatalf("expected cancellation error for query1 but got none")
|
|
}
|
|
if ee := ErrQueryCanceled("test statement execution"); res.Err != ee {
|
|
t.Fatalf("expected error %q, got %q", ee, res.Err)
|
|
}
|
|
|
|
// Canceling a query before starting it must have no effect.
|
|
query2 := engine.newTestQuery(func(ctx context.Context) error {
|
|
return contextDone(ctx, "test statement execution")
|
|
})
|
|
|
|
query2.Cancel()
|
|
res = query2.Exec(ctx)
|
|
if res.Err != nil {
|
|
t.Fatalf("unexpected error on executing query2: %s", res.Err)
|
|
}
|
|
}
|
|
|
|
func TestEngineShutdown(t *testing.T) {
|
|
engine := NewEngine(nil, nil)
|
|
ctx, cancelCtx := context.WithCancel(context.Background())
|
|
|
|
block := make(chan struct{})
|
|
processing := make(chan struct{})
|
|
|
|
// Shutdown engine on first handler execution. Should handler execution ever become
|
|
// concurrent this test has to be adjusted accordingly.
|
|
f := func(ctx context.Context) error {
|
|
processing <- struct{}{}
|
|
<-block
|
|
return contextDone(ctx, "test statement execution")
|
|
}
|
|
query1 := engine.newTestQuery(f)
|
|
|
|
// Stopping the engine must cancel the base context. While executing queries is
|
|
// still possible, their context is canceled from the beginning and execution should
|
|
// terminate immediately.
|
|
|
|
var res *Result
|
|
go func() {
|
|
res = query1.Exec(ctx)
|
|
processing <- struct{}{}
|
|
}()
|
|
|
|
<-processing
|
|
cancelCtx()
|
|
block <- struct{}{}
|
|
<-processing
|
|
|
|
if res.Err == nil {
|
|
t.Fatalf("expected error on shutdown during query but got none")
|
|
}
|
|
if ee := ErrQueryCanceled("test statement execution"); res.Err != ee {
|
|
t.Fatalf("expected error %q, got %q", ee, res.Err)
|
|
}
|
|
|
|
query2 := engine.newTestQuery(func(context.Context) error {
|
|
t.Fatalf("reached query execution unexpectedly")
|
|
return nil
|
|
})
|
|
|
|
// The second query is started after the engine shut down. It must
|
|
// be canceled immediately.
|
|
res2 := query2.Exec(ctx)
|
|
if res2.Err == nil {
|
|
t.Fatalf("expected error on querying with canceled context but got none")
|
|
}
|
|
if _, ok := res2.Err.(ErrQueryCanceled); !ok {
|
|
t.Fatalf("expected cancelation error, got %q", res2.Err)
|
|
}
|
|
}
|
|
|
|
func TestRecoverEvaluatorRuntime(t *testing.T) {
|
|
var ev *evaluator
|
|
var err error
|
|
defer ev.recover(&err)
|
|
|
|
// Cause a runtime panic.
|
|
var a []int
|
|
a[123] = 1
|
|
|
|
if err.Error() != "unexpected error" {
|
|
t.Fatalf("wrong error message: %q, expected %q", err, "unexpected error")
|
|
}
|
|
}
|
|
|
|
func TestRecoverEvaluatorError(t *testing.T) {
|
|
var ev *evaluator
|
|
var err error
|
|
|
|
e := fmt.Errorf("custom error")
|
|
|
|
defer func() {
|
|
if err.Error() != e.Error() {
|
|
t.Fatalf("wrong error message: %q, expected %q", err, e)
|
|
}
|
|
}()
|
|
defer ev.recover(&err)
|
|
|
|
panic(e)
|
|
}
|