prometheus/cmd/promtool/rules.go

// Copyright 2020 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 main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/api"
	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
	"github.com/prometheus/common/model"
	"github.com/prometheus/prometheus/pkg/labels"
	"github.com/prometheus/prometheus/rules"
	"github.com/prometheus/prometheus/storage"
	"github.com/prometheus/prometheus/tsdb"
)

// ruleImporter is the importer to backfill rules.
type ruleImporter struct {
	logger log.Logger
	config ruleImporterConfig

	groups      map[string]*rules.Group
	groupLoader rules.GroupLoader

	apiClient v1.API

	writer *tsdb.BlockWriter
}

// ruleImporterConfig is the config for the rule importer.
type ruleImporterConfig struct {
	Start        time.Time
	End          time.Time
	OutputDir    string
	EvalInterval time.Duration
	URL          string
}

// newRuleImporter creates a new rule importer that can be used to backfill rules.
func newRuleImporter(logger log.Logger, config ruleImporterConfig) *ruleImporter {
	return &ruleImporter{
		logger:      logger,
		config:      config,
		groupLoader: rules.FileLoader{},
	}
}

// init initializes the rule importer which creates a new block writer
// and creates an Prometheus API client.
func (importer *ruleImporter) init() error {
	w, err := tsdb.NewBlockWriter(importer.logger,
		importer.config.OutputDir,
		tsdb.DefaultBlockDuration,
	)
	if err != nil {
		return err
	}
	importer.writer = w

	config := api.Config{
		Address: importer.config.URL,
	}
	c, err := api.NewClient(config)
	if err != nil {
		return err
	}
	importer.apiClient = v1.NewAPI(c)
	return nil
}

// close cleans up any open resources.
func (importer *ruleImporter) close() error {
	return importer.writer.Close()
}

// loadGroups reads groups from a list of rule files.
func (importer *ruleImporter) loadGroups(ctx context.Context, filenames []string) (errs []error) {
	groups := make(map[string]*rules.Group)

	for _, filename := range filenames {
		rgs, errs := importer.groupLoader.Load(filename)
		if errs != nil {
			return errs
		}

		for _, ruleGroup := range rgs.Groups {
			itv := importer.config.EvalInterval
			if ruleGroup.Interval != 0 {
				itv = time.Duration(ruleGroup.Interval)
			}

			rgRules := make([]rules.Rule, 0, len(ruleGroup.Rules))
			for _, r := range ruleGroup.Rules {
				expr, err := importer.groupLoader.Parse(r.Expr.Value)
				if err != nil {
					return []error{errors.Wrap(err, filename)}
				}
				rgRules = append(rgRules, rules.NewRecordingRule(
					r.Record.Value,
					expr,
					labels.FromMap(r.Labels),
				))
			}

			groups[rules.GroupKey(filename, ruleGroup.Name)] = rules.NewGroup(rules.GroupOptions{
				Name:     ruleGroup.Name,
				File:     filename,
				Interval: itv,
				Rules:    rgRules,
				Opts:     &rules.ManagerOptions{},
			})
		}
	}

	importer.groups = groups
	return nil
}

// importAll evaluates all the groups and rules and creates new time series
// and stores them in new blocks.
func (importer *ruleImporter) importAll(ctx context.Context) []error {
	var errs = []error{}
	var currentBlockEnd time.Time
	var appender storage.Appender

	for name, group := range importer.groups {
		level.Info(importer.logger).Log("backfiller", fmt.Sprintf("processing group, name: %s", name))
		stimeWithAlignment := group.EvalTimestamp(importer.config.Start.UnixNano())

		ts := stimeWithAlignment
		// a 2-hr block that contains all the data for each rule
		for ts.Before(importer.config.End) {
			currentBlockEnd = ts.Add(time.Duration(tsdb.DefaultBlockDuration) * time.Millisecond)
			if currentBlockEnd.After(importer.config.End) {
				currentBlockEnd = importer.config.End
			}
			// should we be creating a new appender for each block?
			appender = importer.writer.Appender(ctx)

			for i, r := range group.Rules() {
				level.Info(importer.logger).Log("backfiller", fmt.Sprintf("processing rule %d, name: %s", i+1, r.Name()))
				err := importer.importRule(ctx, r.Query().String(), r.Name(), r.Labels(), ts, currentBlockEnd, appender)
				if err != nil {
					errs = append(errs, err)
				}
			}

			ts = currentBlockEnd
			_, err := importer.writer.Flush(ctx)
			if err != nil {
				errs = append(errs, err)
			}

			err = appender.Commit()
			if err != nil {
				errs = append(errs, err)
			}
		}
	}
	return errs
}

// importRule imports the historical data for a single rule.
func (importer *ruleImporter) importRule(ctx context.Context, ruleExpr, ruleName string, ruleLabels labels.Labels, start, end time.Time, appender storage.Appender) error {
	val, warnings, err := importer.apiClient.QueryRange(ctx,
		ruleExpr,
		v1.Range{
			Start: start,
			End:   end,
			Step:  importer.config.EvalInterval,
		},
	)
	if err != nil {
		return err
	}
	if warnings != nil {
		fmt.Fprint(os.Stderr, "warning api.QueryRange:", warnings)
	}

	var matrix model.Matrix
	switch val.Type() {
	case model.ValMatrix:
		matrix = val.(model.Matrix)
		for _, sample := range matrix {

			currentLabels := make(labels.Labels, 0, len(sample.Metric))
			currentLabels = append(currentLabels, labels.Label{
				Name:  labels.MetricName,
				Value: ruleName,
			})
			for _, ruleLabel := range ruleLabels {
				currentLabels = append(currentLabels, ruleLabel)
			}
			for k, v := range sample.Metric {
				currentLabels = append(currentLabels, labels.Label{
					Name:  string(k),
					Value: string(v),
				})
			}
			for _, value := range sample.Values {
				_, err := appender.Add(currentLabels, value.Timestamp.Unix(), float64(value.Value))
				if err != nil {
					return err
				}
			}
		}
	default:
		return errors.New("rule result is wrong type")
	}
	return nil
}