bin/crowdsec: avoid writing errors twice when log_media=stdout (#2876)

* bin/crowdsec: avoid writing errors twice when log_media=stdout
simpler, correct hook usage
* lint
This commit is contained in:
mmetc 2024-03-07 12:29:10 +01:00 committed by GitHub
parent e611d01c90
commit 98560d0cf5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 68 additions and 69 deletions

View file

@ -1,11 +1,11 @@
package main package main
import ( import (
"errors"
"fmt" "fmt"
"runtime" "runtime"
"time" "time"
"github.com/pkg/errors"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/crowdsecurity/go-cs-lib/trace" "github.com/crowdsecurity/go-cs-lib/trace"

28
cmd/crowdsec/fatalhook.go Normal file
View file

@ -0,0 +1,28 @@
package main
import (
"io"
log "github.com/sirupsen/logrus"
)
// FatalHook is used to log fatal messages to stderr when the rest goes to a file
type FatalHook struct {
Writer io.Writer
LogLevels []log.Level
}
func (hook *FatalHook) Fire(entry *log.Entry) error {
line, err := entry.String()
if err != nil {
return err
}
_, err = hook.Writer.Write([]byte(line))
return err
}
func (hook *FatalHook) Levels() []log.Level {
return hook.LogLevels
}

View file

@ -1,43 +0,0 @@
package main
import (
"io"
"os"
log "github.com/sirupsen/logrus"
)
type ConditionalHook struct {
Writer io.Writer
LogLevels []log.Level
Enabled bool
}
func (hook *ConditionalHook) Fire(entry *log.Entry) error {
if hook.Enabled {
line, err := entry.String()
if err != nil {
return err
}
_, err = hook.Writer.Write([]byte(line))
return err
}
return nil
}
func (hook *ConditionalHook) Levels() []log.Level {
return hook.LogLevels
}
// The primal logging hook is set up before parsing config.yaml.
// Once config.yaml is parsed, the primal hook is disabled if the
// configured logger is writing to stderr. Otherwise it's used to
// report fatal errors and panics to stderr in addition to the log file.
var primalHook = &ConditionalHook{
Writer: os.Stderr,
LogLevels: []log.Level{log.FatalLevel, log.PanicLevel},
Enabled: true,
}

View file

@ -72,7 +72,7 @@ type Flags struct {
DisableCAPI bool DisableCAPI bool
Transform string Transform string
OrderEvent bool OrderEvent bool
CpuProfile string CPUProfile string
} }
type labelsMap map[string]string type labelsMap map[string]string
@ -181,7 +181,7 @@ func (f *Flags) Parse() {
} }
flag.StringVar(&dumpFolder, "dump-data", "", "dump parsers/buckets raw outputs") flag.StringVar(&dumpFolder, "dump-data", "", "dump parsers/buckets raw outputs")
flag.StringVar(&f.CpuProfile, "cpu-profile", "", "write cpu profile to file") flag.StringVar(&f.CPUProfile, "cpu-profile", "", "write cpu profile to file")
flag.Parse() flag.Parse()
} }
@ -249,7 +249,12 @@ func LoadConfig(configFile string, disableAgent bool, disableAPI bool, quiet boo
return nil, err return nil, err
} }
primalHook.Enabled = (cConfig.Common.LogMedia != "stdout") if cConfig.Common.LogMedia != "stdout" {
log.AddHook(&FatalHook{
Writer: os.Stderr,
LogLevels: []log.Level{log.FatalLevel, log.PanicLevel},
})
}
if err := csconfig.LoadFeatureFlagsFile(configFile, log.StandardLogger()); err != nil { if err := csconfig.LoadFeatureFlagsFile(configFile, log.StandardLogger()); err != nil {
return nil, err return nil, err
@ -323,7 +328,9 @@ func LoadConfig(configFile string, disableAgent bool, disableAPI bool, quiet boo
var crowdsecT0 time.Time var crowdsecT0 time.Time
func main() { func main() {
log.AddHook(primalHook) // The initial log level is INFO, even if the user provided an -error or -warning flag
// because we need feature flags before parsing cli flags
log.SetFormatter(&log.TextFormatter{TimestampFormat: time.RFC3339, FullTimestamp: true})
if err := fflag.RegisterAllFeatures(); err != nil { if err := fflag.RegisterAllFeatures(); err != nil {
log.Fatalf("failed to register features: %s", err) log.Fatalf("failed to register features: %s", err)
@ -355,13 +362,13 @@ func main() {
os.Exit(0) os.Exit(0)
} }
if flags.CpuProfile != "" { if flags.CPUProfile != "" {
f, err := os.Create(flags.CpuProfile) f, err := os.Create(flags.CPUProfile)
if err != nil { if err != nil {
log.Fatalf("could not create CPU profile: %s", err) log.Fatalf("could not create CPU profile: %s", err)
} }
log.Infof("CPU profile will be written to %s", flags.CpuProfile) log.Infof("CPU profile will be written to %s", flags.CPUProfile)
if err := pprof.StartCPUProfile(f); err != nil { if err := pprof.StartCPUProfile(f); err != nil {
f.Close() f.Close()

View file

@ -11,7 +11,6 @@ import (
) )
func runParse(input chan types.Event, output chan types.Event, parserCTX parser.UnixParserCtx, nodes []parser.Node) error { func runParse(input chan types.Event, output chan types.Event, parserCTX parser.UnixParserCtx, nodes []parser.Node) error {
LOOP: LOOP:
for { for {
select { select {
@ -56,5 +55,6 @@ LOOP:
output <- parsed output <- parsed
} }
} }
return nil return nil
} }

View file

@ -4,15 +4,17 @@ import (
"fmt" "fmt"
"time" "time"
"github.com/prometheus/client_golang/prometheus"
log "github.com/sirupsen/logrus"
"github.com/crowdsecurity/crowdsec/pkg/csconfig" "github.com/crowdsecurity/crowdsec/pkg/csconfig"
leaky "github.com/crowdsecurity/crowdsec/pkg/leakybucket" leaky "github.com/crowdsecurity/crowdsec/pkg/leakybucket"
"github.com/crowdsecurity/crowdsec/pkg/types" "github.com/crowdsecurity/crowdsec/pkg/types"
"github.com/prometheus/client_golang/prometheus"
log "github.com/sirupsen/logrus"
) )
func runPour(input chan types.Event, holders []leaky.BucketFactory, buckets *leaky.Buckets, cConfig *csconfig.Config) error { func runPour(input chan types.Event, holders []leaky.BucketFactory, buckets *leaky.Buckets, cConfig *csconfig.Config) error {
count := 0 count := 0
for { for {
// bucket is now ready // bucket is now ready
select { select {
@ -21,6 +23,7 @@ func runPour(input chan types.Event, holders []leaky.BucketFactory, buckets *lea
return nil return nil
case parsed := <-input: case parsed := <-input:
startTime := time.Now() startTime := time.Now()
count++ count++
if count%5000 == 0 { if count%5000 == 0 {
log.Infof("%d existing buckets", leaky.LeakyRoutineCount) log.Infof("%d existing buckets", leaky.LeakyRoutineCount)
@ -32,8 +35,9 @@ func runPour(input chan types.Event, holders []leaky.BucketFactory, buckets *lea
log.Warningf("Failed to unmarshal time from event '%s' : %s", parsed.MarshaledTime, err) log.Warningf("Failed to unmarshal time from event '%s' : %s", parsed.MarshaledTime, err)
} else { } else {
log.Warning("Starting buckets garbage collection ...") log.Warning("Starting buckets garbage collection ...")
if err = leaky.GarbageCollectBuckets(*z, buckets); err != nil { if err = leaky.GarbageCollectBuckets(*z, buckets); err != nil {
return fmt.Errorf("failed to start bucket GC : %s", err) return fmt.Errorf("failed to start bucket GC : %w", err)
} }
} }
} }
@ -45,13 +49,16 @@ func runPour(input chan types.Event, holders []leaky.BucketFactory, buckets *lea
log.Errorf("bucketify failed for: %v", parsed) log.Errorf("bucketify failed for: %v", parsed)
continue continue
} }
elapsed := time.Since(startTime) elapsed := time.Since(startTime)
globalPourHistogram.With(prometheus.Labels{"type": parsed.Line.Module, "source": parsed.Line.Src}).Observe(elapsed.Seconds()) globalPourHistogram.With(prometheus.Labels{"type": parsed.Line.Module, "source": parsed.Line.Src}).Observe(elapsed.Seconds())
if poured { if poured {
globalBucketPourOk.Inc() globalBucketPourOk.Inc()
} else { } else {
globalBucketPourKo.Inc() globalBucketPourKo.Inc()
} }
if len(parsed.MarshaledTime) != 0 { if len(parsed.MarshaledTime) != 0 {
if err := lastProcessedItem.UnmarshalText([]byte(parsed.MarshaledTime)); err != nil { if err := lastProcessedItem.UnmarshalText([]byte(parsed.MarshaledTime)); err != nil {
log.Warningf("failed to unmarshal time from event : %s", err) log.Warningf("failed to unmarshal time from event : %s", err)