Browse Source

add Close for ActiveQueryTracker to close the file.

Signed-off-by: ianwoolf <btw515wolf2@gmail.com>
pull/10367/head
ianwoolf 3 years ago committed by zhangzizhao
parent
commit
79e4bdee8e
  1. 1
      promql/engine_test.go
  2. 22
      promql/query_logger.go
  3. 5
      promql/query_logger_test.go

1
promql/engine_test.go

@ -49,6 +49,7 @@ func TestQueryConcurrency(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer os.RemoveAll(dir) defer os.RemoveAll(dir)
queryTracker := NewActiveQueryTracker(dir, maxConcurrency, nil) queryTracker := NewActiveQueryTracker(dir, maxConcurrency, nil)
t.Cleanup(queryTracker.Close)
opts := EngineOpts{ opts := EngineOpts{
Logger: nil, Logger: nil,

22
promql/query_logger.go

@ -16,6 +16,7 @@ package promql
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"io"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
@ -31,6 +32,7 @@ type ActiveQueryTracker struct {
mmapedFile []byte mmapedFile []byte
getNextIndex chan int getNextIndex chan int
logger log.Logger logger log.Logger
closer io.Closer
maxConcurrent int maxConcurrent int
} }
@ -81,7 +83,7 @@ func logUnfinishedQueries(filename string, filesize int, logger log.Logger) {
} }
} }
func getMMapedFile(filename string, filesize int, logger log.Logger) ([]byte, error) { func getMMapedFile(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) file, err := os.OpenFile(filename, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0o666)
if err != nil { if err != nil {
absPath, pathErr := filepath.Abs(filename) absPath, pathErr := filepath.Abs(filename)
@ -89,22 +91,22 @@ func getMMapedFile(filename string, filesize int, logger log.Logger) ([]byte, er
absPath = filename absPath = filename
} }
level.Error(logger).Log("msg", "Error opening query log file", "file", absPath, "err", err) level.Error(logger).Log("msg", "Error opening query log file", "file", absPath, "err", err)
return nil, err return nil, nil, err
} }
err = file.Truncate(int64(filesize)) err = file.Truncate(int64(filesize))
if err != nil { if err != nil {
level.Error(logger).Log("msg", "Error setting filesize.", "filesize", filesize, "err", err) level.Error(logger).Log("msg", "Error setting filesize.", "filesize", filesize, "err", err)
return nil, err return nil, nil, err
} }
fileAsBytes, err := mmap.Map(file, mmap.RDWR, 0) fileAsBytes, err := mmap.Map(file, mmap.RDWR, 0)
if err != nil { if err != nil {
level.Error(logger).Log("msg", "Failed to mmap", "file", filename, "Attempted size", filesize, "err", err) level.Error(logger).Log("msg", "Failed to mmap", "file", filename, "Attempted size", filesize, "err", err)
return nil, err return nil, nil, err
} }
return fileAsBytes, err return fileAsBytes, file, err
} }
func NewActiveQueryTracker(localStoragePath string, maxConcurrent int, logger log.Logger) *ActiveQueryTracker { func NewActiveQueryTracker(localStoragePath string, maxConcurrent int, logger log.Logger) *ActiveQueryTracker {
@ -116,7 +118,7 @@ func NewActiveQueryTracker(localStoragePath string, maxConcurrent int, logger lo
filename, filesize := filepath.Join(localStoragePath, "queries.active"), 1+maxConcurrent*entrySize filename, filesize := filepath.Join(localStoragePath, "queries.active"), 1+maxConcurrent*entrySize
logUnfinishedQueries(filename, filesize, logger) logUnfinishedQueries(filename, filesize, logger)
fileAsBytes, err := getMMapedFile(filename, filesize, logger) fileAsBytes, closer, err := getMMapedFile(filename, filesize, logger)
if err != nil { if err != nil {
panic("Unable to create mmap-ed active query log") panic("Unable to create mmap-ed active query log")
} }
@ -124,6 +126,7 @@ func NewActiveQueryTracker(localStoragePath string, maxConcurrent int, logger lo
copy(fileAsBytes, "[") copy(fileAsBytes, "[")
activeQueryTracker := ActiveQueryTracker{ activeQueryTracker := ActiveQueryTracker{
mmapedFile: fileAsBytes, mmapedFile: fileAsBytes,
closer: closer,
getNextIndex: make(chan int, maxConcurrent), getNextIndex: make(chan int, maxConcurrent),
logger: logger, logger: logger,
maxConcurrent: maxConcurrent, maxConcurrent: maxConcurrent,
@ -198,3 +201,10 @@ func (tracker ActiveQueryTracker) Insert(ctx context.Context, query string) (int
return 0, ctx.Err() return 0, ctx.Err()
} }
} }
func (tracker *ActiveQueryTracker) Close() {
if tracker == nil || tracker.closer == nil {
return
}
tracker.closer.Close()
}

5
promql/query_logger_test.go

@ -110,7 +110,10 @@ func TestMMapFile(t *testing.T) {
filename := file.Name() filename := file.Name()
defer os.Remove(filename) defer os.Remove(filename)
fileAsBytes, err := getMMapedFile(filename, 2, nil) fileAsBytes, closer, err := getMMapedFile(filename, 2, nil)
if err != nil {
t.Cleanup(func() { closer.Close() })
}
require.NoError(t, err) require.NoError(t, err)
copy(fileAsBytes, "ab") copy(fileAsBytes, "ab")

Loading…
Cancel
Save