mirror of
https://github.com/SigNoz/signoz.git
synced 2026-04-25 13:20:24 +01:00
Compare commits
1 Commits
feat/cloud
...
ee-auditor
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a591bf4ac2 |
@@ -8,6 +8,7 @@ import (
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"github.com/SigNoz/signoz/cmd"
|
||||
"github.com/SigNoz/signoz/ee/auditor/fileauditor"
|
||||
"github.com/SigNoz/signoz/ee/auditor/otlphttpauditor"
|
||||
"github.com/SigNoz/signoz/ee/authn/callbackauthn/oidccallbackauthn"
|
||||
"github.com/SigNoz/signoz/ee/authn/callbackauthn/samlcallbackauthn"
|
||||
@@ -155,6 +156,9 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
|
||||
if err := factories.Add(otlphttpauditor.NewFactory(licensing, version.Info)); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
if err := factories.Add(fileauditor.NewFactory(licensing, version.Info)); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return factories
|
||||
},
|
||||
func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
|
||||
|
||||
37
ee/auditor/fileauditor/export.go
Normal file
37
ee/auditor/fileauditor/export.go
Normal file
@@ -0,0 +1,37 @@
|
||||
package fileauditor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/auditor"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/types/audittypes"
|
||||
)
|
||||
|
||||
// export converts a batch of audit events to OTLP-JSON log records and appends
|
||||
// the encoded payload to the configured file as one NDJSON line per batch.
|
||||
// Mirrors the wire format that otlphttpauditor sends, so downstream tools can
|
||||
// consume both transports interchangeably.
|
||||
func (provider *provider) export(ctx context.Context, events []audittypes.AuditEvent) error {
|
||||
logs := audittypes.NewPLogsFromAuditEvents(events, "signoz", provider.build.Version(), "signoz.audit")
|
||||
|
||||
payload, err := provider.marshaler.MarshalLogs(logs)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, errors.TypeInternal, auditor.ErrCodeAuditExportFailed, "failed to marshal audit logs")
|
||||
}
|
||||
|
||||
provider.mu.Lock()
|
||||
defer provider.mu.Unlock()
|
||||
|
||||
if _, err := provider.file.Write(payload); err != nil {
|
||||
provider.settings.Logger().ErrorContext(ctx, "audit export failed", errors.Attr(err), slog.Int("dropped_log_records", len(events)))
|
||||
return errors.Wrapf(err, errors.TypeInternal, auditor.ErrCodeAuditExportFailed, "failed to write audit logs")
|
||||
}
|
||||
|
||||
if _, err := provider.file.Write([]byte("\n")); err != nil {
|
||||
return errors.Wrapf(err, errors.TypeInternal, auditor.ErrCodeAuditExportFailed, "failed to write audit log newline")
|
||||
}
|
||||
|
||||
return provider.file.Sync()
|
||||
}
|
||||
99
ee/auditor/fileauditor/provider.go
Normal file
99
ee/auditor/fileauditor/provider.go
Normal file
@@ -0,0 +1,99 @@
|
||||
package fileauditor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"sync"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/auditor"
|
||||
"github.com/SigNoz/signoz/pkg/auditor/auditorserver"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/licensing"
|
||||
"github.com/SigNoz/signoz/pkg/types/audittypes"
|
||||
"github.com/SigNoz/signoz/pkg/version"
|
||||
"go.opentelemetry.io/collector/pdata/plog"
|
||||
)
|
||||
|
||||
var _ auditor.Auditor = (*provider)(nil)
|
||||
|
||||
type provider struct {
|
||||
settings factory.ScopedProviderSettings
|
||||
config auditor.Config
|
||||
licensing licensing.Licensing
|
||||
build version.Build
|
||||
server *auditorserver.Server
|
||||
marshaler plog.JSONMarshaler
|
||||
file *os.File
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func NewFactory(licensing licensing.Licensing, build version.Build) factory.ProviderFactory[auditor.Auditor, auditor.Config] {
|
||||
return factory.NewProviderFactory(factory.MustNewName("file"), func(ctx context.Context, providerSettings factory.ProviderSettings, config auditor.Config) (auditor.Auditor, error) {
|
||||
return newProvider(ctx, providerSettings, config, licensing, build)
|
||||
})
|
||||
}
|
||||
|
||||
func newProvider(_ context.Context, providerSettings factory.ProviderSettings, config auditor.Config, licensing licensing.Licensing, build version.Build) (auditor.Auditor, error) {
|
||||
settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/ee/auditor/fileauditor")
|
||||
|
||||
file, err := os.OpenFile(config.File.Path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, errors.TypeInvalidInput, auditor.ErrCodeAuditExportFailed, "failed to open audit file %q", config.File.Path)
|
||||
}
|
||||
|
||||
provider := &provider{
|
||||
settings: settings,
|
||||
config: config,
|
||||
licensing: licensing,
|
||||
build: build,
|
||||
marshaler: plog.JSONMarshaler{},
|
||||
file: file,
|
||||
}
|
||||
|
||||
server, err := auditorserver.New(settings,
|
||||
auditorserver.Config{
|
||||
BufferSize: config.BufferSize,
|
||||
BatchSize: config.BatchSize,
|
||||
FlushInterval: config.FlushInterval,
|
||||
},
|
||||
provider.export,
|
||||
)
|
||||
if err != nil {
|
||||
_ = file.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
provider.server = server
|
||||
return provider, nil
|
||||
}
|
||||
|
||||
func (provider *provider) Start(ctx context.Context) error {
|
||||
return provider.server.Start(ctx)
|
||||
}
|
||||
|
||||
func (provider *provider) Audit(ctx context.Context, event audittypes.AuditEvent) {
|
||||
if event.PrincipalAttributes.PrincipalOrgID.IsZero() {
|
||||
provider.settings.Logger().WarnContext(ctx, "audit event dropped as org_id is zero")
|
||||
return
|
||||
}
|
||||
|
||||
if _, err := provider.licensing.GetActive(ctx, event.PrincipalAttributes.PrincipalOrgID); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
provider.server.Add(ctx, event)
|
||||
}
|
||||
|
||||
func (provider *provider) Healthy() <-chan struct{} {
|
||||
return provider.server.Healthy()
|
||||
}
|
||||
|
||||
func (provider *provider) Stop(ctx context.Context) error {
|
||||
serverErr := provider.server.Stop(ctx)
|
||||
fileErr := provider.file.Close()
|
||||
if serverErr != nil {
|
||||
return serverErr
|
||||
}
|
||||
return fileErr
|
||||
}
|
||||
@@ -25,6 +25,8 @@ type Config struct {
|
||||
FlushInterval time.Duration `mapstructure:"flush_interval"`
|
||||
|
||||
OTLPHTTP OTLPHTTPConfig `mapstructure:"otlphttp"`
|
||||
|
||||
File FileConfig `mapstructure:"file"`
|
||||
}
|
||||
|
||||
// OTLPHTTPConfig holds configuration for the OTLP HTTP exporter provider.
|
||||
@@ -46,6 +48,14 @@ type OTLPHTTPConfig struct {
|
||||
Retry RetryConfig `mapstructure:"retry"`
|
||||
}
|
||||
|
||||
// FileConfig holds configuration for the file exporter provider.
|
||||
// Audit events are encoded as OTLP-JSON log records and appended to the file.
|
||||
type FileConfig struct {
|
||||
// Path is the absolute path to the audit log file. The file is opened with
|
||||
// O_APPEND|O_CREATE|O_WRONLY; existing contents are preserved across runs.
|
||||
Path string `mapstructure:"path"`
|
||||
}
|
||||
|
||||
// RetryConfig configures exponential backoff for the OTLP HTTP exporter.
|
||||
type RetryConfig struct {
|
||||
// Enabled controls whether retries are attempted on transient failures.
|
||||
@@ -111,5 +121,11 @@ func (c Config) Validate() error {
|
||||
}
|
||||
}
|
||||
|
||||
if c.Provider == "file" {
|
||||
if c.File.Path == "" {
|
||||
return errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::file::path must be set when provider is file")
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user