VAULT-18934: Record individual metrics for each Auditing Event Pipeline (#22266)

* add sink wrapper to take telemetry measures

* make use of sinkwrapper

audit/event.go

@@ -97,3 +97,15 @@ func (f format) validate() error {
 func (f format) String() string {
 	return string(f)
 }
+
+// MetricTag returns a tag corresponding to this subtype to include in metrics.
+func (st subtype) MetricTag() string {
+	switch st {
+	case RequestType:
+		return "log_request"
+	case ResponseType:
+		return "log_response"
+	}
+
+	return ""
+}

audit/sink_wrapper.go

@@ -0,0 +1,42 @@
+package audit
+
+import (
+	"context"
+
+	metrics "github.com/armon/go-metrics"
+
+	"github.com/hashicorp/eventlogger"
+)
+
+// SinkWrapper is a wrapper for any kind of Sink Node that processes events
+// containing an auditEvent payload.
+type SinkWrapper struct {
+	Name string
+	Sink eventlogger.Node
+}
+
+// Process simply wraps the Process method of this SinkWrapper's sink field by
+// taking a measurement of the time elapsed since the provided Event was created
+// once this method returns.
+func (s *SinkWrapper) Process(ctx context.Context, e *eventlogger.Event) (*eventlogger.Event, error) {
+	defer func() {
+		auditEvent, ok := e.Payload.(*auditEvent)
+		if ok {
+			metrics.MeasureSince([]string{"audit", s.Name, auditEvent.Subtype.MetricTag()}, e.CreatedAt)
+		}
+	}()
+
+	return s.Sink.Process(ctx, e)
+}
+
+// Reopen simply wraps the Reopen method of this SinkWrapper's sink field
+// without doing any additional work.
+func (s *SinkWrapper) Reopen() error {
+	return s.Sink.Reopen()
+}
+
+// Type simply wraps the Type method of this SinkWrapper's sink field without
+// doing any additional work.
+func (s *SinkWrapper) Type() eventlogger.NodeType {
+	return s.Sink.Type()
+}

audit/types.go

@@ -310,6 +310,9 @@ type BackendConfig struct {
 
 	// Config is the opaque user configuration provided when mounting
 	Config map[string]string
+
+	// MountPath is the path where this Backend is mounted
+	MountPath string
 }
 
 // Factory is the factory function to create an audit backend.

builtin/audit/file/backend.go

@@ -165,18 +165,19 @@ func Factory(ctx context.Context, conf *audit.BackendConfig, useEventLogger bool
 
 		switch path {
 		case "stdout":
-			sinkNode = event.NewStdoutSinkNode(format)
+			sinkNode = &audit.SinkWrapper{Name: path, Sink: event.NewStdoutSinkNode(format)}
 		case "discard":
-			sinkNode = event.NewNoopSink()
+			sinkNode = &audit.SinkWrapper{Name: path, Sink: event.NewNoopSink()}
 		default:
 			var err error
 
 			// The NewFileSink function attempts to open the file and will
 			// return an error if it can't.
-			sinkNode, err = event.NewFileSink(b.path, format, event.WithFileMode(strconv.FormatUint(uint64(mode), 8)))
+			n, err := event.NewFileSink(b.path, format, event.WithFileMode(strconv.FormatUint(uint64(mode), 8)))
 			if err != nil {
 				return nil, fmt.Errorf("file sink creation failed for path %q: %w", path, err)
 			}
+			sinkNode = &audit.SinkWrapper{Name: conf.MountPath, Sink: n}
 		}
 
 		sinkNodeID, err := event.GenerateNodeID()

builtin/audit/socket/backend.go

@@ -138,10 +138,11 @@ func Factory(ctx context.Context, conf *audit.BackendConfig, useEventLogger bool
 		b.nodeIDList[0] = formatterNodeID
 		b.nodeMap[formatterNodeID] = f
 
-		sinkNode, err := event.NewSocketSink(format, address, event.WithSocketType(socketType), event.WithMaxDuration(writeDuration.String()))
+		n, err := event.NewSocketSink(format, address, event.WithSocketType(socketType), event.WithMaxDuration(writeDuration.String()))
 		if err != nil {
 			return nil, fmt.Errorf("error creating socket sink node: %w", err)
 		}
+		sinkNode := &audit.SinkWrapper{Name: conf.MountPath, Sink: n}
 		sinkNodeID, err := event.GenerateNodeID()
 		if err != nil {
 			return nil, fmt.Errorf("error generating random NodeID for sink node: %w", err)

builtin/audit/syslog/backend.go

@@ -133,10 +133,11 @@ func Factory(ctx context.Context, conf *audit.BackendConfig, useEventLogger bool
 		b.nodeIDList[0] = formatterNodeID
 		b.nodeMap[formatterNodeID] = f
 
-		sinkNode, err := event.NewSyslogSink(format, event.WithFacility(facility), event.WithTag(tag))
+		n, err := event.NewSyslogSink(format, event.WithFacility(facility), event.WithTag(tag))
 		if err != nil {
 			return nil, fmt.Errorf("error creating syslog sink node: %w", err)
 		}
+		sinkNode := &audit.SinkWrapper{Name: conf.MountPath, Sink: n}
 
 		sinkNodeID, err := event.GenerateNodeID()
 		if err != nil {

vault/audit.go

@@ -485,6 +485,7 @@ func (c *Core) newAuditBackend(ctx context.Context, entry *MountEntry, view logi
 		SaltView:   view,
 		SaltConfig: saltConfig,
 		Config:     conf,
+		MountPath:  entry.Path,
 	}, c.IsExperimentEnabled(experiments.VaultExperimentCoreAuditEventsAlpha1), c.auditedHeaders)
 	if err != nil {
 		return nil, err