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"
@@ -29,10 +30,11 @@ const (
 	certsContainerPath   = dockerLocalSharedDir + "/" + certsSharedDir
 	configsContainerPath = dockerLocalSharedDir + "/" + configSharedDir
 
-	envMetricsName = "e2e_metrics_read_write"
-	envLogsName    = "e2e_logs_read_write_tail"
-	envTenantsName = "e2e_tenants"
-	envInteractive = "e2e_interactive"
+	envMetricsName  = "e2e_metrics_read_write"
+	envRulesAPIName = "e2e_rules_api"
+	envLogsName     = "e2e_logs_read_write_tail"
+	envTenantsName  = "e2e_tenants"
+	envInteractive  = "e2e_interactive"
 
 	defaultTenantID = "1610b0c3-c509-4592-a256-a1871353dbfa"
 	mtlsTenantID    = "845cdfd9-f936-443c-979c-2ee7dc91f646"
@@ -166,3 +168,74 @@ 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
+`
+const recordAndAlertingRulesYamlTpl = `
+groups:
+- name: node_rules
+  rules:
+  - record: job:up:avg
+    expr: avg without(instance)(up{job="node"})
+  - alert: ManyInstancesDown
+    expr: job:up:avg{job="node"} < 0.5
+`
+
+const invalidRulesYamlTpl = `
+invalid:
+- name: testing
+ invalid_rules:
+ - rule1: job:up:avg
+   expr: avg without(instance)(up{job="node"})
+ - rule2: ManyInstancesDown
+   expr: job:up:avg{job="node"} < 0.5
+`

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/metrics_test.go

@@ -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,170 @@
+// +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 := startServicesForRules(t, e)
+
+	api, err := newObservatoriumAPIService(
+		e,
+		withRulesEndpoint("http://"+rulesEndpoint),
+		withRateLimiter(rateLimiterAddr),
+	)
+	testutil.Ok(t, err)
+	testutil.Ok(t, e2e.StartAndWaitReady(api))
+
+	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},
+	}
+
+	t.Run("read-write-recording-rules", func(t *testing.T) {
+		// 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, http.StatusNotFound, res.StatusCode)
+
+		// Set a file containing 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, http.StatusOK, res.StatusCode)
+
+		// 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, http.StatusOK, res.StatusCode)
+
+		body, err := ioutil.ReadAll(res.Body)
+		bodyStr := string(body)
+
+		assertResponse(t, bodyStr, "sum by (job) (http_inprogress_requests)")
+	})
+
+	t.Run("read-write-alerting-rules", func(t *testing.T) {
+		// Set a file containing an 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, http.StatusOK, res.StatusCode)
+
+		// Check if the alerting 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, http.StatusOK, res.StatusCode)
+
+		body, err := ioutil.ReadAll(res.Body)
+		bodyStr := string(body)
+		assertResponse(t, bodyStr, "alert: HighRequestLatency")
+	})
+
+	t.Run("read-write-recording-and-alerting-rules", func(t *testing.T) {
+		// Set a file containing both recording and alerting rules
+		recordAndAlertingRules := []byte(recordAndAlertingRulesYamlTpl)
+		r, err := http.NewRequest(
+			http.MethodPut,
+			rulesEndpointURL,
+			bytes.NewReader(recordAndAlertingRules),
+		)
+		testutil.Ok(t, err)
+
+		res, err := client.Do(r)
+		testutil.Ok(t, err)
+		testutil.Equals(t, http.StatusOK, res.StatusCode)
+
+		// Check if both recording and alerting rules 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, http.StatusOK, res.StatusCode)
+
+		body, err := ioutil.ReadAll(res.Body)
+		bodyStr := string(body)
+		assertResponse(t, bodyStr, "record: job:up:avg")
+		assertResponse(t, bodyStr, "alert: ManyInstancesDown")
+	})
+
+	t.Run("write-invalid-rules", func(t *testing.T) {
+		// Set an invalid rules file
+		invalidRules := []byte(invalidRulesYamlTpl)
+		r, err := http.NewRequest(
+			http.MethodPut,
+			rulesEndpointURL,
+			bytes.NewReader(invalidRules),
+		)
+		testutil.Ok(t, err)
+		res, err := client.Do(r)
+		//TODO: an error/http status code is not being returned to the API
+		//testutil.NotOk(t, err)
+		testutil.Equals(t, http.StatusOK, res.StatusCode) // should this be http.StatusBadRequest instead? (from: https://github.com/observatorium/rules-objstore/blob/main/pkg/server/server.go#L80)
+	})
+}

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"
+	rulesObjectStoreImage = "quay.io/observatorium/rules-objstore:main-2022-01-19-8650540"
 
 	logLevelError = "error"
 	logLevelDebug = "debug"
@@ -49,6 +50,19 @@ func startServicesForMetrics(t *testing.T, e e2e.Environment) (
 		thanosQuery.Endpoint("http")
 }
 
+func startServicesForRules(t *testing.T, e e2e.Environment) (metricsRulesEndpoint string) {
+	// 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(rulesBackend))
+
+	return rulesBackend.InternalEndpoint("http")
+}
+
 func startServicesForLogs(t *testing.T, e e2e.Environment) (
 	logsEndpoint string,
 	logsExtEndpoint string,
@@ -174,6 +188,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:     rulesObjectStoreImage,
+			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 +232,7 @@ type apiOptions struct {
 	logsEndpoint         string
 	metricsReadEndpoint  string
 	metricsWriteEndpoint string
+	metricsRulesEndpoint string
 	ratelimiterAddr      string
 }
 
@@ -215,6 +251,12 @@ func withMetricsEndpoints(readEndpoint string, writeEndpoint string) apiOption {
 	}
 }
 
+func withRulesEndpoint(rulesEndpoint string) apiOption {
+	return func(o *apiOptions) {
+		o.metricsRulesEndpoint = rulesEndpoint
+	}
+}
+
 func withRateLimiter(addr string) apiOption {
 	return func(o *apiOptions) {
 		o.ratelimiterAddr = addr
@@ -252,6 +294,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)