diff --git a/amqptrace/trace.go b/amqptrace/trace.go new file mode 100644 index 0000000..081aeca --- /dev/null +++ b/amqptrace/trace.go @@ -0,0 +1,142 @@ +// Package amqptrace provides W3C TraceContext propagation through AMQP messages +// used by the platform's event bus. It wraps platform/rabbit with OTel context +// injection/extraction so producers and consumers share the same trace. +// +// Usage (producer): +// +// import "git.k6n.net/mats/platform/amqptrace" +// +// ev := event.Event{Type: event.OrderCreated, ...} +// if err := amqptrace.Publish(ctx, conn, "order", ev); err != nil { ... } +// +// Usage (consumer): +// +// amqptrace.Listen(ch, "order", "order.created", func(ctx context.Context, d amqp.Delivery) error { +// // ctx carries the parent span from the publisher. +// _, span := otel.Tracer("my-service").Start(ctx, "handle order") +// defer span.End() +// ... +// }) +package amqptrace + +import ( + "context" + "encoding/json" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" + + "git.k6n.net/mats/platform/event" + "git.k6n.net/mats/platform/rabbit" + amqp "github.com/rabbitmq/amqp091-go" +) + +// --------------------------------------------------------------------------- +// Carrier — adapts amqp.Table to the OTel TextMapCarrier +// --------------------------------------------------------------------------- + +type amqpHeadersCarrier struct{ headers amqp.Table } + +func (c amqpHeadersCarrier) Get(key string) string { + if v, ok := c.headers[key]; ok { + if s, ok := v.(string); ok { + return s + } + } + return "" +} + +func (c amqpHeadersCarrier) Set(key, value string) { + c.headers[key] = value +} + +func (c amqpHeadersCarrier) Keys() []string { + keys := make([]string, 0, len(c.headers)) + for k := range c.headers { + keys = append(keys, k) + } + return keys +} + +// --------------------------------------------------------------------------- +// Producer +// --------------------------------------------------------------------------- + +// Publish marshals the event and publishes it to exchange with the W3C trace +// context from ctx injected into AMQP headers. Consumers using Listen or +// ListenAndHandle will automatically create child spans. +func Publish(ctx context.Context, c *amqp.Connection, exchange string, ev event.Event) error { + body, err := json.Marshal(ev) + if err != nil { + return err + } + headers := make(amqp.Table) + otel.GetTextMapPropagator().Inject(ctx, amqpHeadersCarrier{headers}) + + ch, err := c.Channel() + if err != nil { + return err + } + defer ch.Close() + return ch.Publish(exchange, string(ev.Type), true, false, amqp.Publishing{ + ContentType: "application/json", + Body: body, + DeliveryMode: amqp.Persistent, + Timestamp: time.Now(), + Headers: headers, + }) +} + +// PublishEvent is a convenience wrapper around Publish that takes a +// context.Context in its signature, matching the naming convention of +// platform/rabbit.PublishEvent. It is equivalent to calling Publish. +func PublishEvent(ctx context.Context, c *amqp.Connection, exchange string, ev event.Event) error { + return Publish(ctx, c, exchange, ev) +} + +// --------------------------------------------------------------------------- +// Consumer +// --------------------------------------------------------------------------- + +// tracedHandler wraps a delivery handler to extract the W3C trace context +// from AMQP headers and start a child span named after the routing key. +// The handler receives a context that carries the parent trace. +func tracedHandler(pattern string, handler func(context.Context, amqp.Delivery) error) func(amqp.Delivery) error { + return func(d amqp.Delivery) error { + ctx := otel.GetTextMapPropagator().Extract(context.Background(), amqpHeadersCarrier{d.Headers}) + + tracer := otel.Tracer("amqptrace.consumer") + ctx, span := tracer.Start(ctx, d.RoutingKey, + trace.WithAttributes( + attribute.String("messaging.system", "rabbitmq"), + attribute.String("messaging.destination", d.Exchange), + attribute.String("messaging.destination_kind", "topic"), + attribute.String("messaging.rabbitmq.routing_key", d.RoutingKey), + attribute.Int("messaging.rabbitmq.delivery_tag", int(d.DeliveryTag)), + ), + ) + defer span.End() + + err := handler(ctx, d) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + } + return err + } +} + +// Listen binds an exclusive queue with a routing-key pattern, wraps the +// handler to extract W3C trace context from AMQP headers, and starts a +// child span for each message. The handler type accepts a context.Context +// (with the parent span) and an amqp.Delivery. +// +// Delegates to rabbit.ListenToPattern for topology setup (exchange declare, +// queue bind, consume goroutine) — wraps only the trace-propagation layer. +func Listen(ch *amqp.Channel, exchange, pattern string, handler func(context.Context, amqp.Delivery) error) error { + wrapped := tracedHandler(pattern, handler) + return rabbit.ListenToPattern(ch, exchange, pattern, wrapped) +} diff --git a/auditlog/auditlog.go b/auditlog/auditlog.go new file mode 100644 index 0000000..a7cedd0 --- /dev/null +++ b/auditlog/auditlog.go @@ -0,0 +1,134 @@ +// Package auditlog provides a structured, append-only audit log for the +// ecommerce platform. Entries are written as JSON lines (one per line) to an +// on-disk file, and can be read back in reverse-chronological order. +// +// The package is designed for GDPR/admin auditing: every mutation of customer +// data (anonymize, merge, update, export) should be recorded with the action, +// the actor (admin user or "customer:{id}"), the target profile, and a +// human-readable detail string that MUST NOT contain PII. +// +// Thread-safe: Write uses a mutex so multiple goroutines (HTTP handlers) can +// log concurrently. Read also locks so a consistent snapshot is returned. +package auditlog + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "sync" + "time" +) + +// Entry is a single audit log line. All times are RFC3339. The Detail field +// MUST NOT contain PII (personally identifiable information) — use only +// profile IDs, order references, and action descriptions. +type Entry struct { + Timestamp string `json:"timestamp"` + Action string `json:"action"` + Actor string `json:"actor,omitempty"` + Target string `json:"target,omitempty"` + Detail string `json:"detail,omitempty"` +} + +// Log manages an append-only audit log file. +type Log struct { + mu sync.Mutex + path string + f *os.File +} + +// New opens (or creates) the audit log at the given file path. The file is +// kept open for appending until Close is called. If the parent directory does +// not exist, it is created. +func New(path string) (*Log, error) { + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, fmt.Errorf("auditlog: mkdir: %w", err) + } + f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if err != nil { + return nil, fmt.Errorf("auditlog: open: %w", err) + } + return &Log{path: path, f: f}, nil +} + +// Write appends one audit entry to the log. It is safe for concurrent use. +func (l *Log) Write(e Entry) error { + if e.Timestamp == "" { + e.Timestamp = time.Now().UTC().Format(time.RFC3339) + } + line, err := json.Marshal(e) + if err != nil { + return fmt.Errorf("auditlog: marshal: %w", err) + } + l.mu.Lock() + defer l.mu.Unlock() + if _, err := l.f.Write(append(line, '\n')); err != nil { + return fmt.Errorf("auditlog: write: %w", err) + } + return nil +} + +// Read returns up to limit entries from the log in reverse-chronological order +// (most recent first). If limit <= 0, all entries are returned. The returned +// entries are a point-in-time snapshot of the file at call time. +func (l *Log) Read(limit int) ([]Entry, error) { + l.mu.Lock() + // Sync so a concurrent Write's data is visible before we read. + _ = l.f.Sync() + l.mu.Unlock() + + f, err := os.Open(l.path) + if err != nil { + return nil, fmt.Errorf("auditlog: read open: %w", err) + } + defer f.Close() + + var entries []Entry + scanner := bufio.NewScanner(f) + buf := make([]byte, 0, 1024*64) + scanner.Buffer(buf, 1024*64) + for scanner.Scan() { + line := scanner.Bytes() + if len(line) == 0 { + continue + } + var e Entry + if err := json.Unmarshal(line, &e); err != nil { + // Skip malformed lines. + continue + } + entries = append(entries, e) + } + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("auditlog: scan: %w", err) + } + + // Reverse: newest first. + for i, j := 0, len(entries)-1; i < j; i, j = i+1, j-1 { + entries[i], entries[j] = entries[j], entries[i] + } + + // Optionally, sort by timestamp descending for robustness. + sort.Slice(entries, func(i, j int) bool { + return entries[i].Timestamp > entries[j].Timestamp + }) + + if limit > 0 && limit < len(entries) { + entries = entries[:limit] + } + return entries, nil +} + +// Close releases the audit log file handle. +func (l *Log) Close() error { + l.mu.Lock() + defer l.mu.Unlock() + if l.f != nil { + return l.f.Close() + } + return nil +} diff --git a/go.mod b/go.mod index 47b85e9..2bb993b 100644 --- a/go.mod +++ b/go.mod @@ -6,3 +6,13 @@ require ( github.com/rabbitmq/amqp091-go v1.12.0 golang.org/x/sys v0.46.0 ) + +require ( + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect +) diff --git a/go.sum b/go.sum index c26aeed..a27da12 100644 --- a/go.sum +++ b/go.sum @@ -1,4 +1,20 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/rabbitmq/amqp091-go v1.12.0 h1:V0v14Iqfs+MwHWihJt/nGS5Ulu0vw572b2Co3mwunkI= +github.com/rabbitmq/amqp091-go v1.12.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=