observer.go

/*
2020 © Postgres.ai
*/

// Package observer provides clone monitoring.
package observer

import (
"bytes"
"context"
"encoding/csv"
"io"
"os"
"regexp"
"strconv"
"sync"
"time"

"github.com/docker/docker/client"
"github.com/pkg/errors"

"gitlab.com/postgres-ai/database-lab/v3/internal/provision/pool"
"gitlab.com/postgres-ai/database-lab/v3/pkg/log"
"gitlab.com/postgres-ai/database-lab/v3/pkg/util/pglog"
)

const (
defaultIntervalSeconds=10
defaultMaxLockDurationSeconds=10
defaultMaxDurationSeconds=60*60// 1 hour.

statusPassed="passed"
statusFailed="failed"

observerApplicationName="observer"
)

// Observer manages observation sessions.
typeObserverstruct {
dockerClient*client.Client
sessionMu*sync.Mutex
storagemap[string]*ObservingClone
cfg*Config
replacementRules []ReplacementRule
pm*pool.Manager
}

// Config defines configuration options for observer.
typeConfigstruct {
ReplacementRulesmap[string]string`yaml:"replacementRules"`
}

// ReplacementRule describes replacement rules.
typeReplacementRulestruct {
re*regexp.Regexp
replacestring
}

// NewObserver creates an Observer instance.
funcNewObserver(dockerClient*client.Client, cfg*Config, pm*pool.Manager) *Observer {
observer:=&Observer{
dockerClient: dockerClient,
sessionMu: &sync.Mutex{},
storage: make(map[string]*ObservingClone),
cfg: cfg,
pm: pm,
replacementRules: []ReplacementRule{},
 }

forpattern, replace:=rangecfg.ReplacementRules {
rule:=ReplacementRule{
re: regexp.MustCompile(pattern),
replace: replace,
 }
observer.replacementRules=append(observer.replacementRules, rule)
 }

returnobserver
}

// GetCloneLog gets clone logs.
// TODO (akartasov): Split log to chunks.
func (o*Observer) GetCloneLog(ctx context.Context, portstring, obsClone*ObservingClone) ([]byte, error) {
clonePort, err:=strconv.Atoi(port)
iferr!=nil {
returnnil, errors.Wrap(err, "failed to parse clone port")
 }

fileSelector:=pglog.NewSelector(obsClone.pool.ClonePath(uint(clonePort)))
fileSelector.SetMinimumTime(obsClone.session.StartedAt)

iferr:=fileSelector.DiscoverLogDir(); err!=nil {
returnnil, errors.Wrap(err, "failed to init file selector")
 }

buf:=bytes.NewBuffer(make([]byte, 0, len(obsClone.CsvFields())))
buf.WriteString(obsClone.CsvFields())
buf.WriteString("\n")

for {
filename, err:=fileSelector.Next()
iferr!=nil {
iferr==pglog.ErrLastFile {
break
 }

returnnil, errors.Wrap(err, "failed to get a CSV log filename")
 }

iferr:=o.processCSVLogFile(ctx, buf, filename, obsClone); err!=nil {
iferr==pglog.ErrTimeBoundary {
break
 }

returnnil, errors.Wrap(err, "failed to process a CSV log file")
 }
 }

returnbuf.Bytes(), nil
}

func (o*Observer) processCSVLogFile(ctx context.Context, buf io.Writer, filenamestring, obsClone*ObservingClone) error {
logFile, err:=os.Open(filename)
iferr!=nil {
returnerrors.Wrap(err, "failed to open a CSV log file")
 }

deferfunc() {
iferr:=logFile.Close(); err!=nil {
log.Errf("Failed to close a CSV log file: %s", err.Error())
 }
 }()

returno.scanCSVLogFile(ctx, logFile, buf, obsClone)
}

func (o*Observer) scanCSVLogFile(ctx context.Context, reader io.Reader, writer io.Writer, obsClone*ObservingClone) error {
csvReader:=csv.NewReader(reader)
csvWriter:=csv.NewWriter(writer)

defercsvWriter.Flush()

for {
ifctx.Err() !=nil {
break
 }

entry, err:=csvReader.Read()
iferr!=nil {
iferr==io.EOF {
break
 }

returnerr
 }

// Check an application name to skip observer log entries.
ifentry[22] ==observerApplicationName {
continue
 }

logTime, err:=time.Parse("2006-01-02 15:04:05.999 MST", entry[0])
iferr!=nil {
returnerr
 }

iflogTime.Before(obsClone.session.StartedAt) {
continue
 }

iflogTime.After(obsClone.session.FinishedAt) {
returnpglog.ErrTimeBoundary
 }

iflen(o.replacementRules) >0 {
o.maskLogs(entry, obsClone.maskedIndexes)
 }

iferr:=csvWriter.Write(entry); err!=nil {
returnerr
 }
 }

returnnil
}

func (o*Observer) maskLogs(entry []string, maskedFieldIndexes []int) {
for_, maskedFieldIndex:=rangemaskedFieldIndexes {
for_, rule:=rangeo.replacementRules {
entry[maskedFieldIndex] =rule.re.ReplaceAllString(entry[maskedFieldIndex], rule.replace)
 }
 }
}

// AddObservingClone adds a new observing session to storage.
func (o*Observer) AddObservingClone(cloneIDstring, portuint, session*ObservingClone) {
o.sessionMu.Lock()
defero.sessionMu.Unlock()
session.pool=o.pm.First().Pool()
session.cloneID=cloneID
session.port=port

o.storage[cloneID] =session
}

// GetObservingClone returns an observation session from storage.
func (o*Observer) GetObservingClone(cloneIDstring) (*ObservingClone, error) {
o.sessionMu.Lock()
defero.sessionMu.Unlock()

session, ok:=o.storage[cloneID]
if!ok {
returnnil, errors.New("observer not found")
 }

returnsession, nil
}

// RemoveObservingClone removes an observing clone from storage.
func (o*Observer) RemoveObservingClone(cloneIDstring) {
o.sessionMu.Lock()
defero.sessionMu.Unlock()

delete(o.storage, cloneID)

log.Dbg("Observing clone has been removed: ", cloneID)
}