vmalert: support object storage for rules (#519)

* vmalert: support object storage for rules

Support loading of alerting and recording rules from object
storages `gcs://`, `gs://`, `s3://`.

* review fixes

app/vmalert/config/config.go

@@ -5,8 +5,6 @@ import (
 	"fmt"
 	"hash/fnv"
 	"net/url"
-	"os"
-	"path/filepath"
 	"sort"
 	"strings"
 
@@ -203,19 +201,15 @@ type ValidateTplFn func(annotations map[string]string) error
 
 // Parse parses rule configs from given file patterns
 func Parse(pathPatterns []string, validateTplFn ValidateTplFn, validateExpressions bool) ([]Group, error) {
-	var fp []string
-	for _, pattern := range pathPatterns {
-		matches, err := filepath.Glob(pattern)
-		if err != nil {
-			return nil, fmt.Errorf("error reading file pattern %s: %w", pattern, err)
-		}
-		fp = append(fp, matches...)
+	files, err := readFromFS(pathPatterns)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read from the config: %s", err)
 	}
 	errGroup := new(utils.ErrGroup)
 	var groups []Group
-	for _, file := range fp {
+	for file, data := range files {
 		uniqueGroups := map[string]struct{}{}
-		gr, err := parseFile(file)
+		gr, err := parseConfig(data)
 		if err != nil {
 			errGroup.Add(fmt.Errorf("failed to parse file %q: %w", file, err))
 			continue
@@ -243,14 +237,10 @@ func Parse(pathPatterns []string, validateTplFn ValidateTplFn, validateExpressio
 	return groups, nil
 }
 
-func parseFile(path string) ([]Group, error) {
-	data, err := os.ReadFile(path)
+func parseConfig(data []byte) ([]Group, error) {
+	data, err := envtemplate.ReplaceBytes(data)
 	if err != nil {
-		return nil, fmt.Errorf("error reading alert rule file %q: %w", path, err)
-	}
-	data, err = envtemplate.ReplaceBytes(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot expand environment vars in %q: %w", path, err)
+		return nil, fmt.Errorf("cannot expand environment vars: %w", err)
 	}
 	g := struct {
 		Groups []Group `yaml:"groups"`

app/vmalert/config/fs.go

@@ -0,0 +1,89 @@
+package config
+
+import (
+	"fmt"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config/fslocal"
+	"strings"
+	"sync"
+)
+
+// FS represent a file system abstract for reading files.
+type FS interface {
+	// Init initializes FS.
+	Init() error
+
+	// String must return human-readable representation of FS.
+	String() string
+
+	// Read returns a list of read files in form of a map
+	// where key is a file name and value is a content of read file.
+	// Read must be called only after the successful Init call.
+	Read() (map[string][]byte, error)
+}
+
+var (
+	fsRegistryMu sync.Mutex
+	fsRegistry   = make(map[string]FS)
+)
+
+// readFromFS parses the given path list and inits FS for each item.
+// Once inited, readFromFS will try to read and return files from each FS.
+// readFromFS returns an error if at least one FS failed to init.
+// The function can be called multiple times but each unique path
+// will be inited only once.
+//
+// It is allowed to mix different FS types in path list.
+func readFromFS(paths []string) (map[string][]byte, error) {
+	var err error
+	result := make(map[string][]byte)
+	for _, path := range paths {
+
+		fsRegistryMu.Lock()
+		fs, ok := fsRegistry[path]
+		if !ok {
+			fs, err = newFS(path)
+			if err != nil {
+				fsRegistryMu.Unlock()
+				return nil, fmt.Errorf("error while parsing path %q: %w", path, err)
+			}
+			if err := fs.Init(); err != nil {
+				fsRegistryMu.Unlock()
+				return nil, fmt.Errorf("error while initializing path %q: %w", path, err)
+			}
+			fsRegistry[path] = fs
+		}
+		fsRegistryMu.Unlock()
+
+		files, err := fs.Read()
+		if err != nil {
+			return nil, fmt.Errorf("error while reading files from %q: %w", fs, err)
+		}
+		for k, v := range files {
+			if _, ok := result[k]; ok {
+				return nil, fmt.Errorf("duplicate found for file name %q: file names must be unique", k)
+			}
+			result[k] = v
+		}
+	}
+	return result, nil
+}
+
+// newFS creates FS based on the give path.
+// Supported file systems are: fs
+func newFS(path string) (FS, error) {
+	scheme := "fs"
+	n := strings.Index(path, "://")
+	if n >= 0 {
+		scheme = path[:n]
+		path = path[n+len("://"):]
+	}
+	if len(path) == 0 {
+		return nil, fmt.Errorf("path cannot be empty")
+	}
+	switch scheme {
+	case "fs":
+		return &fslocal.FS{Pattern: path}, nil
+	default:
+		return nil, fmt.Errorf("unsupported scheme %q", scheme)
+	}
+}

app/vmalert/config/fs_test.go

@@ -0,0 +1,39 @@
+package config
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestNewFS(t *testing.T) {
+	f := func(path, expStr string) {
+		t.Helper()
+		fs, err := newFS(path)
+		if err != nil {
+			t.Fatalf("unexpected err: %s", err)
+		}
+		if fs.String() != expStr {
+			t.Fatalf("expected FS %q; got %q", expStr, fs.String())
+		}
+	}
+
+	f("/foo/bar", "Local FS{MatchPattern: \"/foo/bar\"}")
+	f("fs:///foo/bar", "Local FS{MatchPattern: \"/foo/bar\"}")
+}
+
+func TestNewFSNegative(t *testing.T) {
+	f := func(path, expErr string) {
+		t.Helper()
+		_, err := newFS(path)
+		if err == nil {
+			t.Fatalf("expected to have err: %s", expErr)
+		}
+		if !strings.Contains(err.Error(), expErr) {
+			t.Fatalf("expected to have err %q; got %q instead", expErr, err)
+		}
+	}
+
+	f("", "path cannot be empty")
+	f("fs://", "path cannot be empty")
+	f("foobar://baz", `unsupported scheme "foobar"`)
+}

app/vmalert/config/fslocal/fslocal.go

@@ -0,0 +1,44 @@
+package fslocal
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+)
+
+// FS represents a local file system
+type FS struct {
+	// Pattern is used for matching one or multiple files.
+	// The pattern may describe hierarchical names such as
+	// /usr/*/bin/ed (assuming the Separator is '/').
+	Pattern string
+}
+
+// Init verifies that configured Pattern is correct
+func (fs *FS) Init() error {
+	_, err := filepath.Glob(fs.Pattern)
+	return err
+}
+
+// String implements Stringer interface
+func (fs *FS) String() string {
+	return fmt.Sprintf("Local FS{MatchPattern: %q}", fs.Pattern)
+}
+
+// Read returns a map of read files where
+// key is the file name and value is file's content.
+func (fs *FS) Read() (map[string][]byte, error) {
+	matches, err := filepath.Glob(fs.Pattern)
+	if err != nil {
+		return nil, fmt.Errorf("error while matching files via pattern %s: %w", fs.Pattern, err)
+	}
+	result := make(map[string][]byte)
+	for _, path := range matches {
+		data, err := os.ReadFile(path)
+		if err != nil {
+			return nil, fmt.Errorf("error while reading file %q: %w", path, err)
+		}
+		result[path] = data
+	}
+	return result, nil
+}

app/vmalert/main.go

@@ -28,13 +28,18 @@ import (
 )
 
 var (
-	rulePath = flagutil.NewArrayString("rule", `Path to the file with alert rules.
-Supports patterns. Flag can be specified multiple times.
+	rulePath = flagutil.NewArrayString("rule", `Path to the files with alert rules.
+Example: gs://bucket/path/to/rules, s3://bucket/path/to/rules, or fs:///path/to/local/rules/dir
+If scheme remote storage scheme is omitted, local file system is used.
+Local file system supports hierarchical patterns and regexes. 
+Remote file system supports only matching by prefix, e.g. s3://bucket/dir/rule_ will match all files with prefix
+rule_ in folder dir.
+This flag can be specified multiple times.
 Examples:
  -rule="/path/to/file". Path to a single file with alerting rules
- -rule="dir/*.yaml" -rule="/*.yaml". Relative path to all .yaml files in "dir" folder,
-absolute path to all .yaml files in root.
-Rule files may contain %{ENV_VAR} placeholders, which are substituted by the corresponding env vars.`)
+ -rule="dir/*.yaml" -rule="/*.yaml" -rule="gcs://vmalert-rules/tenant_%{TENANT_ID}/prod". 
+Rule files may contain %{ENV_VAR} placeholders, which are substituted by the corresponding env vars
+`)
 
 	ruleTemplatesPath = flagutil.NewArrayString("rule.templates", `Path or glob pattern to location with go template definitions
 	for rules annotations templating. Flag can be specified multiple times.