prometheus/promql/query_logger.go
Nathan Baulch 50cd453c8f
Some checks failed
buf.build / lint and publish (push) Waiting to run
CI / Go tests (push) Waiting to run
CI / More Go tests (push) Waiting to run
CI / Go tests with previous Go version (push) Waiting to run
CI / UI tests (push) Waiting to run
CI / Go tests on Windows (push) Waiting to run
CI / Mixins tests (push) Waiting to run
CI / Build Prometheus for common architectures (0) (push) Waiting to run
CI / Build Prometheus for common architectures (1) (push) Waiting to run
CI / Build Prometheus for common architectures (2) (push) Waiting to run
CI / Build Prometheus for all architectures (0) (push) Waiting to run
CI / Build Prometheus for all architectures (1) (push) Waiting to run
CI / Build Prometheus for all architectures (10) (push) Waiting to run
CI / Build Prometheus for all architectures (11) (push) Waiting to run
CI / Build Prometheus for all architectures (2) (push) Waiting to run
CI / Build Prometheus for all architectures (3) (push) Waiting to run
CI / Build Prometheus for all architectures (4) (push) Waiting to run
CI / Build Prometheus for all architectures (5) (push) Waiting to run
CI / Build Prometheus for all architectures (6) (push) Waiting to run
CI / Build Prometheus for all architectures (7) (push) Waiting to run
CI / Build Prometheus for all architectures (8) (push) Waiting to run
CI / Build Prometheus for all architectures (9) (push) Waiting to run
CI / Report status of build Prometheus for all architectures (push) Blocked by required conditions
CI / Check generated parser (push) Waiting to run
CI / golangci-lint (push) Waiting to run
CI / fuzzing (push) Waiting to run
CI / codeql (push) Waiting to run
CI / Publish main branch artifacts (push) Blocked by required conditions
CI / Publish release artefacts (push) Blocked by required conditions
CI / Publish UI on npm Registry (push) Blocked by required conditions
Scorecards supply-chain security / Scorecards analysis (push) Waiting to run
Push README to Docker Hub / Push README to Docker Hub (push) Has been cancelled
Push README to Docker Hub / Push README to quay.io (push) Has been cancelled
chore: Fix typos (#14868)
* Fix typos

---------

Signed-off-by: Nathan Baulch <nathan.baulch@gmail.com>
2024-09-10 22:32:03 +02:00

238 lines
6 KiB
Go

// Copyright 2019 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 (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"time"
"unicode/utf8"
"github.com/edsrzf/mmap-go"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
)
type ActiveQueryTracker struct {
mmappedFile []byte
getNextIndex chan int
logger log.Logger
closer io.Closer
maxConcurrent int
}
var _ io.Closer = &ActiveQueryTracker{}
type Entry struct {
Query string `json:"query"`
Timestamp int64 `json:"timestamp_sec"`
}
const (
entrySize int = 1000
)
func parseBrokenJSON(brokenJSON []byte) (string, bool) {
queries := strings.ReplaceAll(string(brokenJSON), "\x00", "")
if len(queries) > 0 {
queries = queries[:len(queries)-1] + "]"
}
// Conditional because of implementation detail: len() = 1 implies file consisted of a single char: '['.
if len(queries) <= 1 {
return "[]", false
}
return queries, true
}
func logUnfinishedQueries(filename string, filesize int, logger log.Logger) {
if _, err := os.Stat(filename); err == nil {
fd, err := os.Open(filename)
if err != nil {
level.Error(logger).Log("msg", "Failed to open query log file", "err", err)
return
}
defer fd.Close()
brokenJSON := make([]byte, filesize)
_, err = fd.Read(brokenJSON)
if err != nil {
level.Error(logger).Log("msg", "Failed to read query log file", "err", err)
return
}
queries, queriesExist := parseBrokenJSON(brokenJSON)
if !queriesExist {
return
}
level.Info(logger).Log("msg", "These queries didn't finish in prometheus' last run:", "queries", queries)
}
}
type mmappedFile struct {
f io.Closer
m mmap.MMap
}
func (f *mmappedFile) Close() error {
err := f.m.Unmap()
if err != nil {
err = fmt.Errorf("mmappedFile: unmapping: %w", err)
}
if fErr := f.f.Close(); fErr != nil {
return errors.Join(fmt.Errorf("close mmappedFile.f: %w", fErr), err)
}
return err
}
func getMMappedFile(filename string, filesize int, logger log.Logger) ([]byte, io.Closer, error) {
file, err := os.OpenFile(filename, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0o666)
if err != nil {
absPath, pathErr := filepath.Abs(filename)
if pathErr != nil {
absPath = filename
}
level.Error(logger).Log("msg", "Error opening query log file", "file", absPath, "err", err)
return nil, nil, err
}
err = file.Truncate(int64(filesize))
if err != nil {
file.Close()
level.Error(logger).Log("msg", "Error setting filesize.", "filesize", filesize, "err", err)
return nil, nil, err
}
fileAsBytes, err := mmap.Map(file, mmap.RDWR, 0)
if err != nil {
file.Close()
level.Error(logger).Log("msg", "Failed to mmap", "file", filename, "Attempted size", filesize, "err", err)
return nil, nil, err
}
return fileAsBytes, &mmappedFile{f: file, m: fileAsBytes}, err
}
func NewActiveQueryTracker(localStoragePath string, maxConcurrent int, logger log.Logger) *ActiveQueryTracker {
err := os.MkdirAll(localStoragePath, 0o777)
if err != nil {
level.Error(logger).Log("msg", "Failed to create directory for logging active queries")
}
filename, filesize := filepath.Join(localStoragePath, "queries.active"), 1+maxConcurrent*entrySize
logUnfinishedQueries(filename, filesize, logger)
fileAsBytes, closer, err := getMMappedFile(filename, filesize, logger)
if err != nil {
panic("Unable to create mmap-ed active query log")
}
copy(fileAsBytes, "[")
activeQueryTracker := ActiveQueryTracker{
mmappedFile: fileAsBytes,
closer: closer,
getNextIndex: make(chan int, maxConcurrent),
logger: logger,
maxConcurrent: maxConcurrent,
}
activeQueryTracker.generateIndices(maxConcurrent)
return &activeQueryTracker
}
func trimStringByBytes(str string, size int) string {
bytesStr := []byte(str)
trimIndex := len(bytesStr)
if size < len(bytesStr) {
for !utf8.RuneStart(bytesStr[size]) {
size--
}
trimIndex = size
}
return string(bytesStr[:trimIndex])
}
func _newJSONEntry(query string, timestamp int64, logger log.Logger) []byte {
entry := Entry{query, timestamp}
jsonEntry, err := json.Marshal(entry)
if err != nil {
level.Error(logger).Log("msg", "Cannot create json of query", "query", query)
return []byte{}
}
return jsonEntry
}
func newJSONEntry(query string, logger log.Logger) []byte {
timestamp := time.Now().Unix()
minEntryJSON := _newJSONEntry("", timestamp, logger)
query = trimStringByBytes(query, entrySize-(len(minEntryJSON)+1))
jsonEntry := _newJSONEntry(query, timestamp, logger)
return jsonEntry
}
func (tracker ActiveQueryTracker) generateIndices(maxConcurrent int) {
for i := 0; i < maxConcurrent; i++ {
tracker.getNextIndex <- 1 + (i * entrySize)
}
}
func (tracker ActiveQueryTracker) GetMaxConcurrent() int {
return tracker.maxConcurrent
}
func (tracker ActiveQueryTracker) Delete(insertIndex int) {
copy(tracker.mmappedFile[insertIndex:], strings.Repeat("\x00", entrySize))
tracker.getNextIndex <- insertIndex
}
func (tracker ActiveQueryTracker) Insert(ctx context.Context, query string) (int, error) {
select {
case i := <-tracker.getNextIndex:
fileBytes := tracker.mmappedFile
entry := newJSONEntry(query, tracker.logger)
start, end := i, i+entrySize
copy(fileBytes[start:], entry)
copy(fileBytes[end-1:], ",")
return i, nil
case <-ctx.Done():
return 0, ctx.Err()
}
}
// Close closes tracker.
func (tracker *ActiveQueryTracker) Close() error {
if tracker == nil || tracker.closer == nil {
return nil
}
if err := tracker.closer.Close(); err != nil {
return fmt.Errorf("close ActiveQueryTracker.closer: %w", err)
}
return nil
}