Compare commits
6 Commits
main
...
multipatte
Author | SHA1 | Date | |
---|---|---|---|
9ef2ae585e | |||
61bf6f92b9 | |||
463c5b709f | |||
4c18161c9c | |||
833ffde474 | |||
00f1647aa6 |
@ -8,6 +8,7 @@ import (
|
||||
"net"
|
||||
"os"
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
"framagit.org/ppom/reaction/logger"
|
||||
"sigs.k8s.io/yaml"
|
||||
@ -137,9 +138,15 @@ func ClientShow(format, stream, filter string, regex *regexp.Regexp) {
|
||||
if regex != nil {
|
||||
for streamName := range response.ClientStatus {
|
||||
for filterName := range response.ClientStatus[streamName] {
|
||||
for patternName := range response.ClientStatus[streamName][filterName] {
|
||||
if !regex.MatchString(patternName) {
|
||||
delete(response.ClientStatus[streamName][filterName], patternName)
|
||||
for patterns := range response.ClientStatus[streamName][filterName] {
|
||||
pmatch := false
|
||||
for _, p := range strings.Split(patterns, "\x00") {
|
||||
if regex.MatchString(p) {
|
||||
pmatch = true
|
||||
}
|
||||
}
|
||||
if !pmatch {
|
||||
delete(response.ClientStatus[streamName][filterName], patterns)
|
||||
}
|
||||
}
|
||||
if len(response.ClientStatus[streamName][filterName]) == 0 {
|
||||
@ -162,12 +169,22 @@ func ClientShow(format, stream, filter string, regex *regexp.Regexp) {
|
||||
if err != nil {
|
||||
logger.Fatalln("Failed to convert daemon binary response to text format:", err)
|
||||
}
|
||||
|
||||
// Replace \0 joined string with space joined string ("1.2.3.4\0root" -> "1.2.3.4 root")
|
||||
for streamName := range response.ClientStatus {
|
||||
for filterName := range response.ClientStatus[streamName] {
|
||||
for patterns := range response.ClientStatus[streamName][filterName] {
|
||||
text = []byte(strings.ReplaceAll(string(text), strings.Join(strings.Split(patterns, "\x00"), "\\0"), strings.Join(strings.Split(patterns, "\x00"), " ")))
|
||||
}
|
||||
}
|
||||
}
|
||||
fmt.Println(string(text))
|
||||
|
||||
os.Exit(0)
|
||||
}
|
||||
|
||||
func ClientFlush(pattern, streamfilter, format string) {
|
||||
response := SendAndRetrieve(Request{Flush, pattern})
|
||||
func ClientFlush(patterns []string, streamfilter, format string) {
|
||||
response := SendAndRetrieve(Request{Flush, strings.Join(patterns, "\x00")})
|
||||
if response.Err != nil {
|
||||
logger.Fatalln("Received error from daemon:", response.Err)
|
||||
os.Exit(1)
|
||||
|
@ -78,25 +78,38 @@ func (p *Pattern) notAnIgnore(match *string) bool {
|
||||
|
||||
// Whether one of the filter's regexes is matched on a line
|
||||
func (f *Filter) match(line *string) string {
|
||||
var result string
|
||||
for _, regex := range f.compiledRegex {
|
||||
|
||||
if matches := regex.FindStringSubmatch(*line); matches != nil {
|
||||
var pnames []string
|
||||
for _, p := range f.pattern {
|
||||
pnames = append(pnames, p.name)
|
||||
}
|
||||
|
||||
if f.pattern != nil {
|
||||
match := matches[regex.SubexpIndex(f.pattern.name)]
|
||||
|
||||
if f.pattern.notAnIgnore(&match) {
|
||||
for _, p := range f.pattern {
|
||||
match := matches[regex.SubexpIndex(p.name)]
|
||||
if p.notAnIgnore(&match) {
|
||||
logger.Printf(logger.INFO, "%s.%s: match [%v]\n", f.stream.name, f.name, match)
|
||||
return match
|
||||
if len(result) == 0 {
|
||||
result = match
|
||||
} else {
|
||||
result = strings.Join([]string{result, match}, "\x00")
|
||||
}
|
||||
}
|
||||
} else {
|
||||
logger.Printf(logger.INFO, "%s.%s: match [.]\n", f.stream.name, f.name)
|
||||
}
|
||||
if f.pattern == nil {
|
||||
// No pattern, so this match will never actually be used
|
||||
return "."
|
||||
return ""
|
||||
}
|
||||
}
|
||||
}
|
||||
return ""
|
||||
if len(strings.Split(result, "\x00")) == len(f.pattern) {
|
||||
return result
|
||||
} else {
|
||||
// Incomplete match = no match
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
func (f *Filter) sendActions(match string, at time.Time) {
|
||||
@ -112,9 +125,13 @@ func (a *Action) exec(match string) {
|
||||
|
||||
if a.filter.pattern != nil {
|
||||
computedCommand = make([]string, 0, len(a.Cmd))
|
||||
matches := strings.Split(match, "\x00")
|
||||
|
||||
for _, item := range a.Cmd {
|
||||
computedCommand = append(computedCommand, strings.ReplaceAll(item, a.filter.pattern.nameWithBraces, match))
|
||||
for i, p := range a.filter.pattern {
|
||||
item = strings.ReplaceAll(item, p.nameWithBraces, matches[i])
|
||||
}
|
||||
computedCommand = append(computedCommand, item)
|
||||
}
|
||||
} else {
|
||||
computedCommand = a.Cmd
|
||||
@ -244,7 +261,7 @@ func MatchesManager() {
|
||||
matchesManagerHandleFlush(fo)
|
||||
case pft = <-matchesC:
|
||||
|
||||
entry := LogEntry{pft.t, 0, pft.p, pft.f.stream.name, pft.f.name, 0, false}
|
||||
entry := LogEntry{pft.t, 0, strings.Join(strings.Split(pft.p, "\x00"), " / "), pft.f.stream.name, pft.f.name, 0, false}
|
||||
|
||||
entry.Exec = matchesManagerHandleMatch(pft)
|
||||
|
||||
@ -274,7 +291,7 @@ func matchesManagerHandleMatch(pft PFT) bool {
|
||||
matchesLock.Lock()
|
||||
defer matchesLock.Unlock()
|
||||
|
||||
filter, pattern, then := pft.f, pft.p, pft.t
|
||||
filter, patterns, then := pft.f, pft.p, pft.t
|
||||
pf := PF{pft.p, pft.f}
|
||||
|
||||
if filter.Retry > 1 {
|
||||
@ -299,7 +316,7 @@ func matchesManagerHandleMatch(pft PFT) bool {
|
||||
|
||||
if filter.Retry <= 1 || len(matches[pf]) >= filter.Retry {
|
||||
delete(matches, pf)
|
||||
filter.sendActions(pattern, then)
|
||||
filter.sendActions(patterns, then)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
@ -318,7 +335,7 @@ func StreamManager(s *Stream, endedSignal chan *Stream) {
|
||||
return
|
||||
}
|
||||
for _, filter := range s.Filters {
|
||||
if match := filter.match(line); match != "" {
|
||||
if match := filter.match(line); len(match) > 0 {
|
||||
matchesC <- PFT{match, filter, time.Now()}
|
||||
}
|
||||
}
|
||||
|
@ -60,7 +60,8 @@ func subCommandParse(f *flag.FlagSet, maxRemainingArgs int) {
|
||||
basicUsage()
|
||||
os.Exit(0)
|
||||
}
|
||||
if len(f.Args()) > maxRemainingArgs {
|
||||
// -1 = no limit to remaining args
|
||||
if maxRemainingArgs > -1 && len(f.Args()) > maxRemainingArgs {
|
||||
fmt.Printf("ERROR unrecognized argument(s): %v\n", f.Args()[maxRemainingArgs:])
|
||||
basicUsage()
|
||||
os.Exit(1)
|
||||
@ -102,7 +103,7 @@ func basicUsage() {
|
||||
` + bold + `reaction flush` + reset + ` TARGET
|
||||
# remove currently active matches and run currently pending actions for the specified TARGET
|
||||
# (then show flushed matches and actions)
|
||||
# e.g. reaction flush 192.168.1.1
|
||||
# e.g. reaction flush 192.168.1.1 root
|
||||
|
||||
# options:
|
||||
-s/--socket SOCKET # path to the client-daemon communication socket
|
||||
@ -194,7 +195,7 @@ func Main(version, commit string) {
|
||||
SocketPath = addSocketFlag(f)
|
||||
queryFormat := addFormatFlag(f)
|
||||
limit := addLimitFlag(f)
|
||||
subCommandParse(f, 1)
|
||||
subCommandParse(f, -1)
|
||||
if *queryFormat != "yaml" && *queryFormat != "json" {
|
||||
logger.Fatalln("only yaml and json formats are supported")
|
||||
f.PrintDefaults()
|
||||
@ -209,7 +210,7 @@ func Main(version, commit string) {
|
||||
logger.Fatalln("for now, -l/--limit is not supported")
|
||||
os.Exit(1)
|
||||
}
|
||||
ClientFlush(f.Arg(0), *limit, *queryFormat)
|
||||
ClientFlush(f.Args(), *limit, *queryFormat)
|
||||
|
||||
case "test-regex":
|
||||
// socket not needed, no interaction with the daemon
|
||||
|
@ -134,7 +134,7 @@ func rotateDB(c *Conf, logDec *gob.Decoder, flushDec *gob.Decoder, logEnc *gob.E
|
||||
}()
|
||||
|
||||
// pattern, stream, fitler → last flush
|
||||
flushes := make(map[PSF]time.Time)
|
||||
flushes := make(map[*PSF]time.Time)
|
||||
for {
|
||||
var entry LogEntry
|
||||
var filter *Filter
|
||||
@ -160,7 +160,7 @@ func rotateDB(c *Conf, logDec *gob.Decoder, flushDec *gob.Decoder, logEnc *gob.E
|
||||
}
|
||||
|
||||
// store
|
||||
flushes[PSF{entry.Pattern, entry.Stream, entry.Filter}] = entry.T
|
||||
flushes[&PSF{entry.Pattern, entry.Stream, entry.Filter}] = entry.T
|
||||
}
|
||||
|
||||
lastTimeCpt := int64(0)
|
||||
@ -201,8 +201,8 @@ func rotateDB(c *Conf, logDec *gob.Decoder, flushDec *gob.Decoder, logEnc *gob.E
|
||||
}
|
||||
|
||||
// check if it hasn't been flushed
|
||||
lastGlobalFlush := flushes[PSF{entry.Pattern, "", ""}].Unix()
|
||||
lastLocalFlush := flushes[PSF{entry.Pattern, entry.Stream, entry.Filter}].Unix()
|
||||
lastGlobalFlush := flushes[&PSF{entry.Pattern, "", ""}].Unix()
|
||||
lastLocalFlush := flushes[&PSF{entry.Pattern, entry.Stream, entry.Filter}].Unix()
|
||||
entryTime := entry.T.Unix()
|
||||
if lastLocalFlush > entryTime || lastGlobalFlush > entryTime {
|
||||
continue
|
||||
|
12
app/pipe.go
12
app/pipe.go
@ -17,14 +17,14 @@ func genClientStatus(local_actions ActionsMap, local_matches MatchesMap, local_a
|
||||
|
||||
// Painful data manipulation
|
||||
for pf, times := range local_matches {
|
||||
pattern, filter := pf.p, pf.f
|
||||
patterns, filter := pf.p, pf.f
|
||||
if cs[filter.stream.name] == nil {
|
||||
cs[filter.stream.name] = make(map[string]MapPatternStatus)
|
||||
}
|
||||
if cs[filter.stream.name][filter.name] == nil {
|
||||
cs[filter.stream.name][filter.name] = make(MapPatternStatus)
|
||||
}
|
||||
cs[filter.stream.name][filter.name][pattern] = &PatternStatus{len(times), nil}
|
||||
cs[filter.stream.name][filter.name][patterns] = &PatternStatus{len(times), nil}
|
||||
}
|
||||
|
||||
local_matchesLock.Unlock()
|
||||
@ -32,17 +32,17 @@ func genClientStatus(local_actions ActionsMap, local_matches MatchesMap, local_a
|
||||
|
||||
// Painful data manipulation
|
||||
for pa, times := range local_actions {
|
||||
pattern, action := pa.p, pa.a
|
||||
patterns, action := pa.p, pa.a
|
||||
if cs[action.filter.stream.name] == nil {
|
||||
cs[action.filter.stream.name] = make(map[string]MapPatternStatus)
|
||||
}
|
||||
if cs[action.filter.stream.name][action.filter.name] == nil {
|
||||
cs[action.filter.stream.name][action.filter.name] = make(MapPatternStatus)
|
||||
}
|
||||
if cs[action.filter.stream.name][action.filter.name][pattern] == nil {
|
||||
cs[action.filter.stream.name][action.filter.name][pattern] = new(PatternStatus)
|
||||
if cs[action.filter.stream.name][action.filter.name][patterns] == nil {
|
||||
cs[action.filter.stream.name][action.filter.name][patterns] = new(PatternStatus)
|
||||
}
|
||||
ps := cs[action.filter.stream.name][action.filter.name][pattern]
|
||||
ps := cs[action.filter.stream.name][action.filter.name][patterns]
|
||||
if ps.Actions == nil {
|
||||
ps.Actions = make(map[string][]string)
|
||||
}
|
||||
|
@ -13,6 +13,7 @@ import (
|
||||
"framagit.org/ppom/reaction/logger"
|
||||
|
||||
"github.com/google/go-jsonnet"
|
||||
"golang.org/x/exp/slices"
|
||||
)
|
||||
|
||||
func (c *Conf) setup() {
|
||||
@ -95,27 +96,17 @@ func (c *Conf) setup() {
|
||||
// Compute Regexes
|
||||
// Look for Patterns inside Regexes
|
||||
for _, regex := range filter.Regex {
|
||||
for patternName, pattern := range c.Patterns {
|
||||
for _, pattern := range c.Patterns {
|
||||
if strings.Contains(regex, pattern.nameWithBraces) {
|
||||
|
||||
if filter.pattern == nil {
|
||||
filter.pattern = pattern
|
||||
} else if filter.pattern == pattern {
|
||||
// no op
|
||||
} else {
|
||||
logger.Fatalf(
|
||||
"Bad configuration: Can't mix different patterns (%s, %s) in same filter (%s.%s)\n",
|
||||
filter.pattern.name, patternName, streamName, filterName,
|
||||
)
|
||||
if !slices.Contains(filter.pattern, pattern) {
|
||||
filter.pattern = append(filter.pattern, pattern)
|
||||
}
|
||||
|
||||
// FIXME should go in the `if filter.pattern == nil`?
|
||||
regex = strings.Replace(regex, pattern.nameWithBraces, pattern.Regex, 1)
|
||||
}
|
||||
}
|
||||
compiledRegex, err := regexp.Compile(regex)
|
||||
if err != nil {
|
||||
log.Fatalf("%vBad configuration: regex of filter %s.%s: %v", logger.FATAL, stream.name, filter.name, err)
|
||||
log.Fatal("Bad configuration: regex of filter %s.%s: %v", stream.name, filter.name, err)
|
||||
}
|
||||
filter.compiledRegex = append(filter.compiledRegex, *compiledRegex)
|
||||
}
|
||||
@ -125,7 +116,7 @@ func (c *Conf) setup() {
|
||||
}
|
||||
for actionName := range filter.Actions {
|
||||
|
||||
action := filter.Actions[actionName]
|
||||
action := filter.Actions[actionName]
|
||||
action.filter = filter
|
||||
action.name = actionName
|
||||
|
||||
|
10
app/types.go
10
app/types.go
@ -42,7 +42,7 @@ type Filter struct {
|
||||
|
||||
Regex []string `json:"regex"`
|
||||
compiledRegex []regexp.Regexp `json:"-"`
|
||||
pattern *Pattern `json:"-"`
|
||||
pattern []*Pattern `json:"-"`
|
||||
|
||||
Retry int `json:"retry"`
|
||||
RetryPeriod string `json:"retryperiod"`
|
||||
@ -87,8 +87,14 @@ type MatchesMap map[PF]map[time.Time]struct{}
|
||||
type ActionsMap map[PA]map[time.Time]struct{}
|
||||
|
||||
// Helper structs made to carry information
|
||||
// Stream, Filter
|
||||
type SF struct{ s, f string }
|
||||
type PSF struct{ p, s, f string }
|
||||
// Pattern, Stream, Filter
|
||||
type PSF struct{
|
||||
p string
|
||||
s string
|
||||
f string
|
||||
}
|
||||
type PF struct {
|
||||
p string
|
||||
f *Filter
|
||||
|
1
go.mod
1
go.mod
@ -4,6 +4,7 @@ go 1.20
|
||||
|
||||
require (
|
||||
github.com/google/go-jsonnet v0.20.0
|
||||
golang.org/x/exp v0.0.0-20240213143201-ec583247a57a
|
||||
sigs.k8s.io/yaml v1.1.0
|
||||
)
|
||||
|
||||
|
2
go.sum
2
go.sum
@ -1,6 +1,8 @@
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/google/go-jsonnet v0.20.0 h1:WG4TTSARuV7bSm4PMB4ohjxe33IHT5WVTrJSU33uT4g=
|
||||
github.com/google/go-jsonnet v0.20.0/go.mod h1:VbgWF9JX7ztlv770x/TolZNGGFfiHEVx9G6ca2eUmeA=
|
||||
golang.org/x/exp v0.0.0-20240213143201-ec583247a57a h1:HinSgX1tJRX3KsL//Gxynpw5CTOAIPhgL4W8PNiIpVE=
|
||||
golang.org/x/exp v0.0.0-20240213143201-ec583247a57a/go.mod h1:CxmFvTBINI24O/j8iY7H1xHzx2i4OsyguNBmN/uPtqc=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v2 v2.2.7 h1:VUgggvou5XRW9mHwD/yXxIYSMtY0zoKQf/v226p2nyo=
|
||||
gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
|
Loading…
Reference in New Issue
Block a user