SystemSh0cker/gocsaf
1
0
Fork 0
mirror of https://github.com/gocsaf/csaf.git synced 2025-12-22 05:40:11 +01:00
Code Issues Releases Activity Actions

Added support for structured logging in csaf_aggretator

Browse source 
This PR adds structured logging for the aggregator service. Currently, only the text handler is used, but I can extend this to use the JSON handler as well. In this case, probably some code that is shared between the aggregator and the downloader would need to be moved to a common package.

I was also wondering, whether this repo is moving to Go 1.21 at the future, since `slog` was introduced in to the standard lib in 1.21. So currently, this still relies on the `x/exp` package.

Fixes #462
This commit is contained in:
Christian Banse 2024-04-18 19:51:25 +02:00
parent d909e9de15
commit e658738b56
12 changed files with 122 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

21
cmd/csaf_aggregator/config.go
View file

@ -12,7 +12,6 @@ import (
"crypto/tls"
"errors"
"fmt"
"log"
"net/http"
"os"
"runtime"
@ -26,6 +25,7 @@ import (
"github.com/csaf-poc/csaf_distribution/v3/internal/models"
"github.com/csaf-poc/csaf_distribution/v3/internal/options"
"github.com/csaf-poc/csaf_distribution/v3/util"
"golang.org/x/exp/slog"
"golang.org/x/time/rate"
)
@ -178,9 +178,11 @@ func (p *provider) ageAccept(c *config) func(time.Time) bool {
}
if c.Verbose {
log.Printf(
"Setting up filter to accept advisories within time range %s to %s\n",
r[0].Format(time.RFC3339), r[1].Format(time.RFC3339))
slog.Debug(
"Setting up filter to accept advisories within time range",
"from", r[0].Format(time.RFC3339),
"to", r[1].Format(time.RFC3339),
)
}
return r.Contains
}
@ -393,6 +395,17 @@ func (c *config) setDefaults() {
}
}
// prepareLogging sets up the structured logging.
func (cfg *config) prepareLogging() error {
ho := slog.HandlerOptions{
Level: slog.LevelDebug,
}
handler := slog.NewTextHandler(os.Stdout, &ho)
logger := slog.New(handler)
slog.SetDefault(logger)
return nil
}
// compileIgnorePatterns compiles the configured patterns to be ignored.
func (p *provider) compileIgnorePatterns() error {
pm, err := filter.NewPatternMatcher(p.IgnorePattern)

38
cmd/csaf_aggregator/full.go
View file

@ -11,7 +11,6 @@ package main
import (
"errors"
"fmt"
"log"
"os"
"path/filepath"
"strings"
@ -20,6 +19,7 @@ import (
"github.com/csaf-poc/csaf_distribution/v3/csaf"
"github.com/csaf-poc/csaf_distribution/v3/util"
"golang.org/x/exp/slog"
)
type fullJob struct {
@ -29,11 +29,13 @@ type fullJob struct {
err error
}
// setupProviderFull fetches the provider-metadate.json for a specific provider.
// setupProviderFull fetches the provider-metadata.json for a specific provider.
func (w *worker) setupProviderFull(provider *provider) error {
log.Printf("worker #%d: %s (%s)\n",
w.num, provider.Name, provider.Domain)
w.log.Info("Setting up provider",
"provider", slog.GroupValue(
slog.String("name", provider.Name),
slog.String("domain", provider.Domain),
))
w.dir = ""
w.provider = provider
@ -55,7 +57,7 @@ func (w *worker) setupProviderFull(provider *provider) error {
"provider-metadata.json has %d validation issues", len(errors))
}
log.Printf("provider-metadata: %s\n", w.loc)
w.log.Info("Using provider-metadata", "url", w.loc)
return nil
}
@ -79,7 +81,7 @@ func (w *worker) fullWork(wg *sync.WaitGroup, jobs <-chan *fullJob) {
func (p *processor) full() error {
if p.cfg.runAsMirror() {
log.Println("Running in aggregator mode")
p.log.Info("Running in aggregator mode")
// check if we need to setup a remote validator
if p.cfg.RemoteValidatorOptions != nil {
@ -96,16 +98,18 @@ func (p *processor) full() error {
}()
}
} else {
log.Println("Running in lister mode")
p.log.Info("Running in lister mode")
}
queue := make(chan *fullJob)
var wg sync.WaitGroup
log.Printf("Starting %d workers.\n", p.cfg.Workers)
p.log.Info("Starting workers...", "num", p.cfg.Workers)
for i := 1; i <= p.cfg.Workers; i++ {
wg.Add(1)
w := newWorker(i, p)
go w.fullWork(&wg, queue)
}
@ -135,12 +139,22 @@ func (p *processor) full() error {
for i := range jobs {
j := &jobs[i]
if j.err != nil {
log.Printf("error: '%s' failed: %v\n", j.provider.Name, j.err)
p.log.Error("Job execution failed",
slog.Group("job",
slog.Group("provider"),
"name", j.provider.Name,
),
"err", j.err,
)
continue
}
if j.aggregatorProvider == nil {
log.Printf(
"error: '%s' does not produce any result.\n", j.provider.Name)
p.log.Error("Job did not produce any result",
slog.Group("job",
slog.Group("provider"),
"name", j.provider.Name,
),
)
continue
}

3
cmd/csaf_aggregator/indices.go
View file

@ -12,7 +12,6 @@ import (
"bufio"
"encoding/csv"
"fmt"
"log"
"os"
"path/filepath"
"sort"
@ -377,7 +376,7 @@ func (w *worker) writeIndices() error {
}
for label, summaries := range w.summaries {
log.Printf("%s: %d\n", label, len(summaries))
w.log.Debug("Writing indices", "label", label, "summaries.num", len(summaries))
if err := w.writeInterims(label, summaries); err != nil {
return err
}

10
cmd/csaf_aggregator/interim.go
View file

@ -17,7 +17,6 @@ import (
"errors"
"fmt"
"io"
"log"
"net/http"
"os"
"path/filepath"
@ -102,12 +101,12 @@ func (w *worker) checkInterims(
// XXX: Should we return an error here?
for _, e := range errors {
log.Printf("validation error: %s: %v\n", url, e)
w.log.Error("validation error", "url", url, "err", e)
}
// We need to write the changed content.
// This will start the transcation if not already started.
// This will start the transaction if not already started.
dst, err := tx.Dst()
if err != nil {
return nil, err
@ -159,8 +158,7 @@ func (w *worker) checkInterims(
// setupProviderInterim prepares the worker for a specific provider.
func (w *worker) setupProviderInterim(provider *provider) {
log.Printf("worker #%d: %s (%s)\n",
w.num, provider.Name, provider.Domain)
w.log.Info("Setting up worker", provider.Name, provider.Domain)
w.dir = ""
w.provider = provider
@ -262,7 +260,7 @@ func (p *processor) interim() error {
queue := make(chan *interimJob)
var wg sync.WaitGroup
log.Printf("Starting %d workers.\n", p.cfg.Workers)
p.log.Info("Starting workers...", "num", p.cfg.Workers)
for i := 1; i <= p.cfg.Workers; i++ {
wg.Add(1)
w := newWorker(i, p)

5
cmd/csaf_aggregator/lazytransaction.go
View file

@ -9,11 +9,11 @@
package main
import (
"log"
"os"
"path/filepath"
"github.com/csaf-poc/csaf_distribution/v3/util"
"golang.org/x/exp/slog"
)
type lazyTransaction struct {
@ -85,7 +85,8 @@ func (lt *lazyTransaction) commit() error {
os.RemoveAll(lt.dst)
return err
}
log.Printf("Move %q -> %q\n", symlink, lt.src)
slog.Debug("Moving directory", "from", symlink, "to", lt.src)
if err := os.Rename(symlink, lt.src); err != nil {
os.RemoveAll(lt.dst)
return err

11
cmd/csaf_aggregator/main.go
View file

@ -15,7 +15,9 @@ import (
"path/filepath"
"github.com/csaf-poc/csaf_distribution/v3/internal/options"
"github.com/gofrs/flock"
"golang.org/x/exp/slog"
)
func lock(lockFile *string, fn func() error) error {
@ -44,8 +46,9 @@ func lock(lockFile *string, fn func() error) error {
func main() {
_, cfg, err := parseArgsConfig()
options.ErrorCheck(err)
options.ErrorCheck(cfg.prepare())
p := processor{cfg: cfg}
options.ErrorCheck(lock(cfg.LockFile, p.process))
cfg.prepareLogging()
options.ErrorCheckStructured(err)
options.ErrorCheckStructured(cfg.prepare())
p := processor{cfg: cfg, log: slog.Default()}
options.ErrorCheckStructured(lock(cfg.LockFile, p.process))
}

43
cmd/csaf_aggregator/mirror.go
View file

@ -16,7 +16,7 @@ import (
"encoding/json"
"fmt"
"io"
"log"
"log/slog"
"net/http"
"net/url"
"os"
@ -47,7 +47,7 @@ func (w *worker) mirror() (*csaf.AggregatorCSAFProvider, error) {
if err != nil && w.dir != "" {
// If something goes wrong remove the debris.
if err := os.RemoveAll(w.dir); err != nil {
log.Printf("error: %v\n", err)
w.log.Error("Could not remove directory", "path", w.dir, "err", err)
}
}
return result, err
@ -166,7 +166,7 @@ func (w *worker) writeProviderMetadata() error {
{Expr: `$.public_openpgp_keys`, Action: util.ReMarshalMatcher(&pm.PGPKeys)},
}, w.metadataProvider); err != nil {
// only log the errors
log.Printf("extracting data from orignal provider failed: %v\n", err)
w.log.Error("Extracting data from original provider failed", "err", err)
}
// We are mirroring the remote public keys, too.
@ -196,11 +196,11 @@ func (w *worker) mirrorPGPKeys(pm *csaf.ProviderMetadata) error {
for i := range pm.PGPKeys {
pgpKey := &pm.PGPKeys[i]
if pgpKey.URL == nil {
log.Printf("ignoring PGP key without URL: %s\n", pgpKey.Fingerprint)
w.log.Warn("Ignoring PGP key without URL", "fingerprint", pgpKey.Fingerprint)
continue
}
if _, err := hex.DecodeString(string(pgpKey.Fingerprint)); err != nil {
log.Printf("ignoring PGP with invalid fingerprint: %s\n", *pgpKey.URL)
w.log.Warn("Ignoring PGP key with invalid fingerprint", "url", *pgpKey.URL)
continue
}
@ -344,7 +344,7 @@ func (w *worker) doMirrorTransaction() error {
// Check if there is a sysmlink already.
target := filepath.Join(w.processor.cfg.Folder, w.provider.Name)
log.Printf("target: '%s'\n", target)
w.log.Debug("Checking for path existance", "path", target)
exists, err := util.PathExists(target)
if err != nil {
@ -359,7 +359,7 @@ func (w *worker) doMirrorTransaction() error {
}
}
log.Printf("sym link: %s -> %s\n", w.dir, target)
w.log.Debug("Creating symbol", "from", w.dir, "to", target)
// Create a new symlink
if err := os.Symlink(w.dir, target); err != nil {
@ -368,7 +368,7 @@ func (w *worker) doMirrorTransaction() error {
}
// Move the symlink
log.Printf("Move: %s -> %s\n", target, webTarget)
w.log.Debug("Moving symbol", "from", target, "to", webTarget)
if err := os.Rename(target, webTarget); err != nil {
os.RemoveAll(w.dir)
return err
@ -499,14 +499,14 @@ func (w *worker) mirrorFiles(tlpLabel csaf.TLPLabel, files []csaf.AdvisoryFile)
u, err := url.Parse(file.URL())
if err != nil {
log.Printf("error: %s\n", err)
w.log.Error("Could not parse advisory file URL", "err", err)
continue
}
// Should we ignore this advisory?
if w.provider.ignoreURL(file.URL(), w.processor.cfg) {
if w.processor.cfg.Verbose {
log.Printf("Ignoring %s: %q\n", w.provider.Name, file.URL())
w.log.Info("Ignoring advisory", slog.Group("provider", "name", w.provider.Name), "file", file)
}
continue
}
@ -514,7 +514,7 @@ func (w *worker) mirrorFiles(tlpLabel csaf.TLPLabel, files []csaf.AdvisoryFile)
// Ignore not conforming filenames.
filename := filepath.Base(u.Path)
if !util.ConformingFileName(filename) {
log.Printf("Not conforming filename %q. Ignoring.\n", filename)
w.log.Warn("Ignoring advisory because of non-conforming filename", "filename", filename)
continue
}
@ -531,19 +531,18 @@ func (w *worker) mirrorFiles(tlpLabel csaf.TLPLabel, files []csaf.AdvisoryFile)
}
if err := downloadJSON(w.client, file.URL(), download); err != nil {
log.Printf("error: %v\n", err)
w.log.Error("Error while downloading JSON", "err", err)
continue
}
// Check against CSAF schema.
errors, err := csaf.ValidateCSAF(advisory)
if err != nil {
log.Printf("error: %s: %v", file, err)
w.log.Error("Error while validating CSAF schema", "err", err)
continue
}
if len(errors) > 0 {
log.Printf("CSAF file %s has %d validation errors.\n",
file, len(errors))
w.log.Error("CSAF file has validation errors", "num.errors", len(errors), "file", file)
continue
}
@ -551,29 +550,27 @@ func (w *worker) mirrorFiles(tlpLabel csaf.TLPLabel, files []csaf.AdvisoryFile)
if rmv := w.processor.remoteValidator; rmv != nil {
rvr, err := rmv.Validate(advisory)
if err != nil {
log.Printf("Calling remote validator failed: %s\n", err)
w.log.Error("Calling remote validator failed", "err", err)
continue
}
if !rvr.Valid {
log.Printf(
"CSAF file %s does not validate remotely.\n", file)
w.log.Error("CSAF file does not validate remotely", "file", file.URL())
continue
}
}
sum, err := csaf.NewAdvisorySummary(w.expr, advisory)
if err != nil {
log.Printf("error: %s: %v\n", file, err)
w.log.Error("Error while creating new advisory", "file", file, "err", err)
continue
}
if util.CleanFileName(sum.ID) != filename {
log.Printf("ID %q does not match filename %s",
sum.ID, filename)
w.log.Error("ID mismatch", "id", sum.ID, "filename", filename)
}
if err := w.extractCategories(label, advisory); err != nil {
log.Printf("error: %s: %v\n", file, err)
w.log.Error("Could not extract categories", "file", file, "err", err)
continue
}
@ -624,7 +621,7 @@ func (w *worker) downloadSignatureOrSign(url, fname string, data []byte) error {
if err != nil {
if err != errNotFound {
log.Printf("error: %s: %v\n", url, err)
w.log.Error("Could not find signature URL", "url", url, "err", err)
}
// Sign it our self.
if sig, err = w.sign(data); err != nil {

32
cmd/csaf_aggregator/processor.go
View file

@ -10,14 +10,14 @@ package main
import (
"fmt"
"log"
"os"
"path/filepath"
"github.com/ProtonMail/gopenpgp/v2/crypto"
"github.com/csaf-poc/csaf_distribution/v3/csaf"
"github.com/csaf-poc/csaf_distribution/v3/util"
"github.com/ProtonMail/gopenpgp/v2/crypto"
"golang.org/x/exp/slog"
)
type processor struct {
@ -26,6 +26,9 @@ type processor struct {
// remoteValidator is a globally configured remote validator.
remoteValidator csaf.RemoteValidator
// log is the structured logger for the whole processor.
log *slog.Logger
}
type summary struct {
@ -48,6 +51,7 @@ type worker struct {
dir string // Directory to store data to.
summaries map[string][]summary // the summaries of the advisories.
categories map[string]util.Set[string] // the categories per label.
log *slog.Logger // the structured logger, supplied with the worker number.
}
func newWorker(num int, processor *processor) *worker {
@ -55,6 +59,7 @@ func newWorker(num int, processor *processor) *worker {
num: num,
processor: processor,
expr: util.NewPathEval(),
log: processor.log.With(slog.Int("worker", num)),
}
}
@ -86,9 +91,10 @@ func (w *worker) locateProviderMetadata(domain string) error {
if w.processor.cfg.Verbose {
for i := range lpmd.Messages {
log.Printf(
"Loading provider-metadata.json of %q: %s\n",
domain, lpmd.Messages[i].Message)
w.log.Info(
"Loading provider-metadata.json",
"domain", domain,
"message", lpmd.Messages[i].Message)
}
}
@ -141,7 +147,7 @@ func (p *processor) removeOrphans() error {
fi, err := entry.Info()
if err != nil {
log.Printf("error: %v\n", err)
p.log.Error("Could not retrieve file info", "err", err)
continue
}
@ -153,13 +159,13 @@ func (p *processor) removeOrphans() error {
d := filepath.Join(path, entry.Name())
r, err := filepath.EvalSymlinks(d)
if err != nil {
log.Printf("error: %v\n", err)
p.log.Error("Could not evaluate symlink", "err", err)
continue
}
fd, err := os.Stat(r)
if err != nil {
log.Printf("error: %v\n", err)
p.log.Error("Could not retrieve file stats", "err", err)
continue
}
@ -169,18 +175,18 @@ func (p *processor) removeOrphans() error {
}
// Remove the link.
log.Printf("removing link %s -> %s\n", d, r)
p.log.Info("Removing link", "path", fmt.Sprintf("%s -> %s", d, r))
if err := os.Remove(d); err != nil {
log.Printf("error: %v\n", err)
p.log.Error("Could not remove symlink", "err", err)
continue
}
// Only remove directories which are in our folder.
if rel, err := filepath.Rel(prefix, r); err == nil &&
rel == filepath.Base(r) {
log.Printf("removing directory %s\n", r)
p.log.Info("Remove directory", "path", r)
if err := os.RemoveAll(r); err != nil {
log.Printf("error: %v\n", err)
p.log.Error("Could not remove directory", "err", err)
}
}
}

12
csaf/advisories.go
View file

@ -13,6 +13,7 @@ import (
"fmt"
"io"
"log"
"log/slog"
"net/http"
"net/url"
"strings"
@ -23,6 +24,7 @@ import (
// AdvisoryFile constructs the urls of a remote file.
type AdvisoryFile interface {
slog.LogValuer
URL() string
SHA256URL() string
SHA512URL() string
@ -46,6 +48,11 @@ func (paf PlainAdvisoryFile) SHA512URL() string { return string(paf) + ".sha512"
// SignURL returns the URL of signature file of this advisory.
func (paf PlainAdvisoryFile) SignURL() string { return string(paf) + ".asc" }
// LogValue implements [slog.LogValuer]
func (paf PlainAdvisoryFile) LogValue() slog.Value {
return slog.GroupValue(slog.String("url", paf.URL()))
}
// HashedAdvisoryFile is a more involed version of checkFile.
// Here each component can be given explicitly.
// If a component is not given it is constructed by
@ -71,6 +78,11 @@ func (haf HashedAdvisoryFile) SHA512URL() string { return haf.name(2, ".sha512")
// SignURL returns the URL of signature file of this advisory.
func (haf HashedAdvisoryFile) SignURL() string { return haf.name(3, ".asc") }
// LogValue implements [slog.LogValuer]
func (haf HashedAdvisoryFile) LogValue() slog.Value {
return slog.GroupValue(slog.String("url", haf.URL()))
}
// AdvisoryFileProcessor implements the extraction of
// advisory file names from a given provider metadata.
type AdvisoryFileProcessor struct {

2
go.mod
View file

@ -14,7 +14,7 @@ require (
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
go.etcd.io/bbolt v1.3.8
golang.org/x/crypto v0.14.0
golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa
golang.org/x/exp v0.0.0-20240416160154-fe59bbe5cc7f
golang.org/x/term v0.13.0
golang.org/x/time v0.3.0
)

2
go.sum
View file

@ -53,6 +53,8 @@ golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa h1:FRnLl4eNAQl8hwxVVC17teOw8kdjVDVAiFMtgUdTSRQ=
golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa/go.mod h1:zk2irFbV9DP96SEBUUAy67IdHUaZuSnrz1n472HUCLE=
golang.org/x/exp v0.0.0-20240416160154-fe59bbe5cc7f h1:99ci1mjWVBWwJiEKYY6jWa4d2nTQVIEhZIptnrVb1XY=
golang.org/x/exp v0.0.0-20240416160154-fe59bbe5cc7f/go.mod h1:/lliqkxwWAhPjf5oSOIJup2XcqJaw8RGS6k3TGEc7GI=
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=

10
internal/options/options.go
View file

@ -19,6 +19,7 @@ import (
"github.com/mitchellh/go-homedir"
"github.com/csaf-poc/csaf_distribution/v3/util"
"golang.org/x/exp/slog"
)
// Parser helps parsing command line arguments and loading
@ -147,3 +148,12 @@ func ErrorCheck(err error) {
log.Fatalf("error: %v\n", err)
}
}
// ErrorCheck checks if err is not nil and terminates
// the program if so.
func ErrorCheckStructured(err error) {
if err != nil {
slog.Error("Error while executing program", "err", err)
os.Exit(1)
}
}