Rules API: Add e2e test

test/e2e/configs.go

@@ -18,6 +18,7 @@ type testType string
 
 const (
 	metrics     testType = "metrics"
+	rules 		testType = "rules"
 	logs        testType = "logs"
 	tenants     testType = "tenants"
 	interactive testType = "interactive"
@@ -30,6 +31,7 @@ const (
 	configsContainerPath = dockerLocalSharedDir + "/" + configSharedDir
 
 	envMetricsName = "e2e_metrics_read_write"
+	envRulesAPIName    = "e2e_rules_api"
 	envLogsName    = "e2e_logs_read_write_tail"
 	envTenantsName = "e2e_tenants"
 	envInteractive = "e2e_interactive"
@@ -166,3 +168,55 @@ func createDexYAML(
 	)
 	testutil.Ok(t, err)
 }
+
+const rulesYAMLTpl = `
+type: S3
+config:
+  bucket: %s
+  endpoint: %s
+  access_key: %s
+  insecure: true
+  secret_key: %s
+`
+
+func createRulesYAML(
+	t *testing.T,
+	e e2e.Environment,
+	bucket, endpoint, accessKey, secretKey string,
+) {
+	yamlContent := []byte(fmt.Sprintf(
+		rulesYAMLTpl,
+		bucket,
+		endpoint,
+		accessKey,
+		secretKey,
+	))
+
+	err := ioutil.WriteFile(
+		filepath.Join(e.SharedDir(), configSharedDir, "rules-objstore.yaml"),
+		yamlContent,
+		os.FileMode(0755),
+	)
+	testutil.Ok(t, err)
+}
+
+const recordingRuleYamlTpl = `
+groups:
+  - name: example
+    rules:
+      - record: job:http_inprogress_requests:sum
+        expr: sum by (job) (http_inprogress_requests)
+`
+
+const alertingRuleYamlTpl = `
+groups:
+- name: example
+  rules:
+  - alert: HighRequestLatency
+    expr: job:request_latency_seconds:mean5m{job="myjob"} > 0.5
+    for: 10m
+    labels:
+      severity: page
+    annotations:
+      summary: High request latency
+`

test/e2e/helpers.go

@@ -87,6 +87,8 @@ func getContainerName(t *testing.T, tt testType, serviceName string) string {
 		return envLogsName + "-" + serviceName
 	case metrics:
 		return envMetricsName + "-" + serviceName
+	case rules:
+		return envRulesAPIName + "-" + serviceName
 	case tenants:
 		return envTenantsName + "-" + serviceName
 	case interactive:
@@ -116,3 +118,13 @@ func assertResponse(t *testing.T, response string, expected string) {
 		fmt.Sprintf("failed to assert that the response '%s' contains '%s'", response, expected),
 	)
 }
+
+type tokenRoundTripper struct {
+	rt    http.RoundTripper
+	token string
+}
+
+func (rt *tokenRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
+	r.Header.Add("Authorization", "bearer "+rt.token)
+	return rt.rt.RoundTrip(r)
+}

test/e2e/interactive_test.go

@@ -20,12 +20,12 @@ func TestInteractiveSetup(t *testing.T) {
 
 	prepareConfigsAndCerts(t, interactive, e)
 	_, token, rateLimiterAddr := startBaseServices(t, e, interactive)
-	readEndpoint, writeEndpoint, readExtEndpoint := startServicesForMetrics(t, e)
+	readEndpoint, writeEndpoint, rulesEndpoint, readExtEndpoint := startServicesForMetrics(t, e)
 	logsEndpoint, logsExtEndpoint := startServicesForLogs(t, e)
 
 	api, err := newObservatoriumAPIService(
 		e,
-		withMetricsEndpoints("http://"+readEndpoint, "http://"+writeEndpoint),
+		withMetricsEndpoints("http://"+readEndpoint, "http://"+writeEndpoint, "http://"+rulesEndpoint),
 		withLogsEndpoints("http://"+logsEndpoint),
 		withRateLimiter(rateLimiterAddr),
 	)

test/e2e/metrics_test.go

@@ -26,11 +26,11 @@ func TestMetricsReadAndWrite(t *testing.T) {
 
 	prepareConfigsAndCerts(t, metrics, e)
 	_, token, rateLimiterAddr := startBaseServices(t, e, metrics)
-	readEndpoint, writeEndpoint, readExtEndpoint := startServicesForMetrics(t, e)
+	readEndpoint, writeEndpoint, rulesEndpoint, readExtEndpoint := startServicesForMetrics(t, e)
 
 	api, err := newObservatoriumAPIService(
 		e,
-		withMetricsEndpoints("http://"+readEndpoint, "http://"+writeEndpoint),
+		withMetricsEndpoints("http://"+readEndpoint, "http://"+writeEndpoint, "http://"+rulesEndpoint),
 		withRateLimiter(rateLimiterAddr),
 	)
 	testutil.Ok(t, err)
@@ -191,13 +191,3 @@ func TestMetricsReadAndWrite(t *testing.T) {
 		})
 	})
 }
-
-type tokenRoundTripper struct {
-	rt    http.RoundTripper
-	token string
-}
-
-func (rt *tokenRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
-	r.Header.Add("Authorization", "bearer "+rt.token)
-	return rt.rt.RoundTrip(r)
-}

test/e2e/rules_test.go

@@ -0,0 +1,123 @@
+// +build integration
+
+package e2e
+
+import (
+	"bytes"
+	"io/ioutil"
+	"net/http"
+	"testing"
+
+	"github.com/efficientgo/e2e"
+	"github.com/efficientgo/tools/core/pkg/testutil"
+)
+
+func TestRulesAPI(t *testing.T) {
+	t.Parallel()
+
+	e, err := e2e.NewDockerEnvironment(envRulesAPIName)
+	testutil.Ok(t, err)
+	t.Cleanup(e.Close)
+
+	prepareConfigsAndCerts(t, rules, e)
+	_, token, rateLimiterAddr := startBaseServices(t, e, rules)
+	_, _, rulesEndpoint, _ := startServicesForMetrics(t, e) // TODO: create another function just for rules?
+
+	api, err := newObservatoriumAPIService(
+		e,
+		withMetricsEndpoints("", "", "http://"+rulesEndpoint),
+		withRateLimiter(rateLimiterAddr),
+	)
+	testutil.Ok(t, err)
+	testutil.Ok(t, e2e.StartAndWaitReady(api))
+
+	t.Run("rules", func(t *testing.T) {
+		rulesEndpointURL := "https://" + api.Endpoint("https") + "/api/metrics/v1/test-oidc/api/v1/rules/raw"
+		tr := &http.Transport{
+			TLSClientConfig: getTLSClientConfig(t, e),
+		}
+
+		client := &http.Client{
+			Transport: &tokenRoundTripper{rt: tr, token: token},
+		}
+
+		// Try to list rules
+		r, err := http.NewRequest(
+			http.MethodGet,
+			rulesEndpointURL,
+			nil,
+		)
+		testutil.Ok(t, err)
+
+		res, err := client.Do(r)
+		testutil.Ok(t, err)
+		testutil.Equals(t, res.StatusCode, http.StatusNotFound)
+
+		// Set a recording rule
+		recordingRule := []byte(recordingRuleYamlTpl)
+		r, err = http.NewRequest(
+			http.MethodPut,
+			rulesEndpointURL,
+			bytes.NewReader(recordingRule),
+		)
+		testutil.Ok(t, err)
+
+		res, err = client.Do(r)
+		testutil.Ok(t, err)
+		testutil.Equals(t, res.StatusCode, http.StatusOK)
+
+		// Check if recording rule is listed
+		r, err = http.NewRequest(
+			http.MethodGet,
+			rulesEndpointURL,
+			nil,
+		)
+		testutil.Ok(t, err)
+
+		res, err = client.Do(r)
+		defer res.Body.Close()
+
+		testutil.Ok(t, err)
+		testutil.Equals(t, res.StatusCode, http.StatusOK)
+
+		body, err := ioutil.ReadAll(res.Body)
+		bodyStr := string(body)
+
+		assertResponse(t, bodyStr, "sum by (job) (http_inprogress_requests)")
+
+		// Set alerting rule
+		alertingRule := []byte(alertingRuleYamlTpl)
+		r, err = http.NewRequest(
+			http.MethodPut,
+			rulesEndpointURL,
+			bytes.NewReader(alertingRule),
+		)
+		testutil.Ok(t, err)
+
+		res, err = client.Do(r)
+		testutil.Ok(t, err)
+		testutil.Equals(t, res.StatusCode, http.StatusOK)
+
+		// Check if recording rule and alerting rule are listed
+		r, err = http.NewRequest(
+			http.MethodGet,
+			rulesEndpointURL,
+			nil,
+		)
+		testutil.Ok(t, err)
+
+		res, err = client.Do(r)
+		defer res.Body.Close()
+
+		testutil.Ok(t, err)
+		testutil.Equals(t, res.StatusCode, http.StatusOK)
+
+		body, err = ioutil.ReadAll(res.Body)
+		bodyStr = string(body)
+		//TODO: check why this fails - shouldn't it join recording+alerting rules?
+		//assertResponse(t, bodyStr, "sum by (job) (http_inprogress_requests)")
+		assertResponse(t, bodyStr, "alert: HighRequestLatency")
+
+		// TODO: add another test case for thanos-ruler-syncer flow
+	})
+}

test/e2e/services.go

@@ -25,6 +25,7 @@ const (
 	dexImage        = "dexidp/dex:v2.30.0"
 	opaImage        = "openpolicyagent/opa:0.31.0"
 	gubernatorImage = "thrawn01/gubernator:1.0.0-rc.8"
+	rulesBackendImage = "quay.io/observatorium/rules-objstore:main-2021-11-22-cc68e25"
 
 	logLevelError = "error"
 	logLevelDebug = "debug"
@@ -33,6 +34,7 @@ const (
 func startServicesForMetrics(t *testing.T, e e2e.Environment) (
 	metricsReadEndpoint string,
 	metricsWriteEndpoint string,
+	metricsRulesEndpoint string,
 	metricsExtReadEndppoint string,
 ) {
 	thanosReceive := newThanosReceiveService(e)
@@ -42,10 +44,20 @@ func startServicesForMetrics(t *testing.T, e e2e.Environment) (
 		[]string{thanosReceive.InternalEndpoint("grpc")},
 		e2edb.WithImage(thanosImage),
 	)
-	testutil.Ok(t, e2e.StartAndWaitReady(thanosReceive, thanosQuery))
+
+	// Create S3 replacement for rules backend
+	bucket := "obs_rules_test"
+	m := e2edb.NewMinio(e, "rules-minio", bucket)
+	testutil.Ok(t, e2e.StartAndWaitReady(m))
+
+	createRulesYAML(t, e, bucket, m.InternalEndpoint(e2edb.AccessPortName), e2edb.MinioAccessKey, e2edb.MinioSecretKey)
+
+	rulesBackend := newRulesBackendService(e)
+	testutil.Ok(t, e2e.StartAndWaitReady(thanosReceive, thanosQuery, rulesBackend))
 
 	return thanosQuery.InternalEndpoint("http"),
 		thanosReceive.InternalEndpoint("remote_write"),
+		rulesBackend.InternalEndpoint("http"),
 		thanosQuery.Endpoint("http")
 }
 
@@ -174,6 +186,27 @@ func newLokiService(e e2e.Environment) *e2e.InstrumentedRunnable {
 	)
 }
 
+func newRulesBackendService(e e2e.Environment) *e2e.InstrumentedRunnable {
+	ports := map[string]int{"http": 8080, "internal": 8081}
+
+	args := e2e.BuildArgs(map[string]string{
+		"-log.level":            logLevelDebug,
+		"-web.listen":           ":" + strconv.Itoa(ports["http"]),
+		"-web.internal.listen":  ":" + strconv.Itoa(ports["internal"]),
+		"-web.healthchecks.url": "http://127.0.0.1:" + strconv.Itoa(ports["http"]),
+		"-objstore.config-file": filepath.Join(configsContainerPath, "rules-objstore.yaml"),
+	})
+
+	return e2e.NewInstrumentedRunnable(e, "rules_objstore", ports, "internal").Init(
+		e2e.StartOptions{
+			Image:     rulesBackendImage,
+			Command:   e2e.NewCommand("", args...),
+			Readiness: e2e.NewHTTPReadinessProbe("internal", "/ready", 200, 200),
+			User:      strconv.Itoa(os.Getuid()),
+		},
+	)
+}
+
 func newOPAService(e e2e.Environment) *e2e.InstrumentedRunnable {
 	ports := map[string]int{"http": 8181}
 
@@ -197,6 +230,7 @@ type apiOptions struct {
 	logsEndpoint         string
 	metricsReadEndpoint  string
 	metricsWriteEndpoint string
+	metricsRulesEndpoint string
 	ratelimiterAddr      string
 }
 
@@ -208,10 +242,11 @@ func withLogsEndpoints(endpoint string) apiOption {
 	}
 }
 
-func withMetricsEndpoints(readEndpoint string, writeEndpoint string) apiOption {
+func withMetricsEndpoints(readEndpoint, writeEndpoint, rulesEndpoint string) apiOption {
 	return func(o *apiOptions) {
 		o.metricsReadEndpoint = readEndpoint
 		o.metricsWriteEndpoint = writeEndpoint
+		o.metricsRulesEndpoint = rulesEndpoint
 	}
 }
 
@@ -252,6 +287,10 @@ func newObservatoriumAPIService(
 		args = append(args, "--metrics.write.endpoint="+opts.metricsWriteEndpoint)
 	}
 
+	if opts.metricsRulesEndpoint != "" {
+		args = append(args, "--metrics.rules.endpoint="+opts.metricsRulesEndpoint)
+	}
+
 	if opts.logsEndpoint != "" {
 		args = append(args, "--logs.read.endpoint="+opts.logsEndpoint)
 		args = append(args, "--logs.tail.endpoint="+opts.logsEndpoint)

test/e2e/tenants_test.go

@@ -19,14 +19,14 @@ func TestTenantsRetryAuthenticationProviderRegistration(t *testing.T) {
 
 	prepareConfigsAndCerts(t, tenants, e)
 	dex, _, _ := startBaseServices(t, e, tenants)
-	readEndpoint, writeEndpoint, _ := startServicesForMetrics(t, e)
+	readEndpoint, writeEndpoint, rulesEndpoint, _ := startServicesForMetrics(t, e)
 
 	// Start API with stopped Dex and observe retries.
 	dex.Stop()
 
 	api, err := newObservatoriumAPIService(
 		e,
-		withMetricsEndpoints("http://"+readEndpoint, "http://"+writeEndpoint),
+		withMetricsEndpoints("http://"+readEndpoint, "http://"+writeEndpoint, "http://"+rulesEndpoint),
 	)
 	testutil.Ok(t, err)
 	testutil.Ok(t, e2e.StartAndWaitReady(api))