server+plugins: Integrate server + NDBCache in decision logs.

config/config.go

@@ -32,6 +32,7 @@ type Config struct {
 	DefaultDecision              *string                    `json:"default_decision,omitempty"`
 	DefaultAuthorizationDecision *string                    `json:"default_authorization_decision,omitempty"`
 	Caching                      json.RawMessage            `json:"caching,omitempty"`
+	NDBuiltinCache               bool                       `json:"nd_builtin_cache,omitempty"`
 	PersistenceDirectory         *string                    `json:"persistence_directory,omitempty"`
 	DistributedTracing           json.RawMessage            `json:"distributed_tracing,omitempty"`
 	Storage                      *struct {
@@ -87,6 +88,11 @@ func (c Config) DefaultAuthorizationDecisionRef() ast.Ref {
 	return r
 }
 
+// NDBuiltinCacheEnabled returns if the ND builtins cache should be used.
+func (c Config) NDBuiltinCacheEnabled() bool {
+	return c.NDBuiltinCache
+}
+
 func (c *Config) validateAndInjectDefaults(id string) error {
 
 	if c.DefaultDecision == nil {

docs/content/configuration.md

@@ -740,6 +740,7 @@ For *GHCR* (Github Container Registry) you can use a developer PAT (personal acc
 | `default_authorization_decision` | `string` | No (default: `/system/authz/allow`) | Set path of default authorization decision for OPA's API. |
 | `persistence_directory` | `string` | No (default `$PWD/.opa`) | Set directory to use for persistence with options like `bundles[_].persist`. |
 | `plugins` | `object` | No (default: `{}`) | Location for custom plugin configuration. See [Plugins](../plugins) for details. |
+| `nd_builtin_cache` | `boolean` | No (default: `false`) | Enable the non-deterministic builtins caching system during policy evaluation, and include the contents of the cache in decision logs. Note that decision logs that are larger than `upload_size_limit_bytes` will drop the `nd_builtin_cache` key from the log entry before uploading. |
 
 ### Keys
 

docs/content/management-decision-logs.md

@@ -75,6 +75,7 @@ Decision log updates contain the following fields:
 | `[_].metrics` | `object` | Key-value pairs of [performance metrics](../rest-api#performance-metrics). |
 | `[_].erased` | `array[string]` | Set of JSON Pointers specifying fields in the event that were erased. |
 | `[_].masked` | `array[string]` | Set of JSON Pointers specifying fields in the event that were masked. |
+| `[_].nd_builtin_cache` | `object` | Key-value pairs of non-deterministic builtin names, paired with objects specifying the input/output mappings for each unique invocation of that builtin during policy evaluation. Intended for use in debugging and decision replay. Receivers will need to decode the JSON using Rego's JSON decoders. |
 
 If the decision log was successfully uploaded to the remote service, it should respond with an HTTP 2xx status. If the
 service responds with a non-2xx status, OPA will requeue the last chunk containing decision log events and upload it
@@ -95,6 +96,12 @@ the last chunk.
 
 `Equilibrium`: If the chunk size is between 90% and 100% of the user-configured limit, maintain soft limit value.
 
+When an event containing `nd_builtin_cache` cannot fit into a chunk smaller than `upload_size_limit_bytes`, OPA will
+drop the `nd_builtin_cache` key from the event, and will retry encoding the chunk without the non-deterministic
+builtins cache information. This best-effort approach ensures that OPA reports decision log events as much as possible,
+and bounds how large decision log events can get. This size-bounding is necessary, because some non-deterministic builtins
+(such as `http.send`) can increase the decision log event size by a potentially unbounded amount.
+
 ### Local Decision Logs
 
 Local console logging of decisions can be enabled via the `console` config option.
@@ -172,7 +179,7 @@ from the decision log event. The erased paths are recorded on the event itself:
 
 There are a few restrictions on the JSON Pointers that OPA will erase:
 
-* Pointers must be prefixed with `/input` or `/result`.
+* Pointers must be prefixed with `/input`, `/result`, or `/nd_builtin_cache`.
 * Pointers may be undefined. For example `/input/name/first` in the example
   above would be undefined. Undefined pointers are ignored.
 * Pointers must refer to object keys. Pointers to array elements will be treated

plugins/logs/mask.go

@@ -21,8 +21,9 @@ const (
 	maskOPRemove maskOP = "remove"
 	maskOPUpsert maskOP = "upsert"
 
-	partInput  = "input"
-	partResult = "result"
+	partInput    = "input"
+	partResult   = "result"
+	partNDBCache = "nd_builtin_cache"
 )
 
 var (
@@ -64,7 +65,9 @@ func newMaskRule(path string, opts ...maskRuleOption) (*maskRule, error) {
 
 	parts := strings.Split(path[1:], "/")
 
-	if parts[0] != partInput && parts[0] != partResult {
+	switch parts[0] {
+	case partInput, partResult, partNDBCache: // OK
+	default:
 		return nil, fmt.Errorf("mask prefix not allowed: %v", parts[0])
 	}
 
@@ -130,8 +133,8 @@ func withFailUndefinedPath() maskRuleOption {
 
 func (r maskRule) Mask(event *EventV1) error {
 
-	var maskObj *interface{}     // pointer to event Input|Result object
-	var maskObjPtr **interface{} // pointer to the event Input|Result pointer itself
+	var maskObj *interface{}     // pointer to event Input|Result|NDBCache object
+	var maskObjPtr **interface{} // pointer to the event Input|Result|NDBCache pointer itself
 
 	switch p := r.escapedParts[0]; p {
 	case partInput:
@@ -143,7 +146,6 @@ func (r maskRule) Mask(event *EventV1) error {
 		}
 		maskObj = event.Input
 		maskObjPtr = &event.Input
-
 	case partResult:
 		if event.Result == nil {
 			if r.failUndefinedPath {
@@ -153,6 +155,15 @@ func (r maskRule) Mask(event *EventV1) error {
 		}
 		maskObj = event.Result
 		maskObjPtr = &event.Result
+	case partNDBCache:
+		if event.NDBuiltinCache == nil {
+			if r.failUndefinedPath {
+				return errMaskInvalidObject
+			}
+			return nil
+		}
+		maskObj = event.NDBuiltinCache
+		maskObjPtr = &event.NDBuiltinCache
 	default:
 		return fmt.Errorf("illegal path value: %s", p)
 	}

plugins/logs/plugin.go

@@ -44,21 +44,22 @@ type Logger interface {
 // the struct. Any changes here MUST be reflected in the AST()
 // implementation below.
 type EventV1 struct {
-	Labels       map[string]string       `json:"labels"`
-	DecisionID   string                  `json:"decision_id"`
-	Revision     string                  `json:"revision,omitempty"` // Deprecated: Use Bundles instead
-	Bundles      map[string]BundleInfoV1 `json:"bundles,omitempty"`
-	Path         string                  `json:"path,omitempty"`
-	Query        string                  `json:"query,omitempty"`
-	Input        *interface{}            `json:"input,omitempty"`
-	Result       *interface{}            `json:"result,omitempty"`
-	MappedResult *interface{}            `json:"mapped_result,omitempty"`
-	Erased       []string                `json:"erased,omitempty"`
-	Masked       []string                `json:"masked,omitempty"`
-	Error        error                   `json:"error,omitempty"`
-	RequestedBy  string                  `json:"requested_by,omitempty"`
-	Timestamp    time.Time               `json:"timestamp"`
-	Metrics      map[string]interface{}  `json:"metrics,omitempty"`
+	Labels         map[string]string       `json:"labels"`
+	DecisionID     string                  `json:"decision_id"`
+	Revision       string                  `json:"revision,omitempty"` // Deprecated: Use Bundles instead
+	Bundles        map[string]BundleInfoV1 `json:"bundles,omitempty"`
+	Path           string                  `json:"path,omitempty"`
+	Query          string                  `json:"query,omitempty"`
+	Input          *interface{}            `json:"input,omitempty"`
+	Result         *interface{}            `json:"result,omitempty"`
+	MappedResult   *interface{}            `json:"mapped_result,omitempty"`
+	NDBuiltinCache *interface{}            `json:"nd_builtin_cache,omitempty"`
+	Erased         []string                `json:"erased,omitempty"`
+	Masked         []string                `json:"masked,omitempty"`
+	Error          error                   `json:"error,omitempty"`
+	RequestedBy    string                  `json:"requested_by,omitempty"`
+	Timestamp      time.Time               `json:"timestamp"`
+	Metrics        map[string]interface{}  `json:"metrics,omitempty"`
 
 	inputAST ast.Value
 }
@@ -87,6 +88,7 @@ var queryKey = ast.StringTerm("query")
 var inputKey = ast.StringTerm("input")
 var resultKey = ast.StringTerm("result")
 var mappedResultKey = ast.StringTerm("mapped_result")
+var ndBuiltinCacheKey = ast.StringTerm("nd_builtin_cache")
 var erasedKey = ast.StringTerm("erased")
 var maskedKey = ast.StringTerm("masked")
 var errorKey = ast.StringTerm("error")
@@ -159,6 +161,14 @@ func (e *EventV1) AST() (ast.Value, error) {
 		event.Insert(mappedResultKey, ast.NewTerm(mResults))
 	}
 
+	if e.NDBuiltinCache != nil {
+		ndbCache, err := roundtripJSONToAST(e.NDBuiltinCache)
+		if err != nil {
+			return nil, err
+		}
+		event.Insert(ndBuiltinCacheKey, ast.NewTerm(ndbCache))
+	}
+
 	if len(e.Erased) > 0 {
 		erased := make([]*ast.Term, len(e.Erased))
 		for i, v := range e.Erased {
@@ -226,6 +236,7 @@ const (
 	defaultBufferSizeLimitBytes = int64(0)     // unlimited
 	defaultMaskDecisionPath     = "/system/log/mask"
 	logDropCounterName          = "decision_logs_dropped"
+	logNDBDropCounterName       = "decision_logs_nd_builtin_cache_dropped"
 	defaultResourcePath         = "/logs"
 )
 
@@ -248,6 +259,7 @@ type Config struct {
 	MaskDecision    *string         `json:"mask_decision"`
 	ConsoleLogs     bool            `json:"console"`
 	Resource        *string         `json:"resource"`
+	NDBuiltinCache  bool            `json:"nd_builtin_cache,omitempty"`
 	maskDecisionRef ast.Ref
 }
 
@@ -557,18 +569,19 @@ func (p *Plugin) Log(ctx context.Context, decision *server.Info) error {
 	}
 
 	event := EventV1{
-		Labels:       p.manager.Labels(),
-		DecisionID:   decision.DecisionID,
-		Revision:     decision.Revision,
-		Bundles:      bundles,
-		Path:         decision.Path,
-		Query:        decision.Query,
-		Input:        decision.Input,
-		Result:       decision.Results,
-		MappedResult: decision.MappedResults,
-		RequestedBy:  decision.RemoteAddr,
-		Timestamp:    decision.Timestamp,
-		inputAST:     decision.InputAST,
+		Labels:         p.manager.Labels(),
+		DecisionID:     decision.DecisionID,
+		Revision:       decision.Revision,
+		Bundles:        bundles,
+		Path:           decision.Path,
+		Query:          decision.Query,
+		Input:          decision.Input,
+		Result:         decision.Results,
+		MappedResult:   decision.MappedResults,
+		NDBuiltinCache: decision.NDBuiltinCache,
+		RequestedBy:    decision.RemoteAddr,
+		Timestamp:      decision.Timestamp,
+		inputAST:       decision.InputAST,
 	}
 
 	if decision.Metrics != nil {
@@ -793,6 +806,9 @@ func (p *Plugin) reconfigure(config interface{}) {
 	p.config = *newConfig
 }
 
+// NOTE(philipc): Because ND builtins caching can cause unbounded growth in
+// decision log entry size, we do best-effort event encoding here, and when we
+// run out of space, we drop the ND builtins cache, and try encoding again.
 func (p *Plugin) encodeAndBufferEvent(event EventV1) {
 	if p.limiter != nil {
 		if !p.limiter.Allow() {
@@ -807,11 +823,29 @@ func (p *Plugin) encodeAndBufferEvent(event EventV1) {
 
 	result, err := p.enc.Write(event)
 	if err != nil {
-		// TODO(tsandall): revisit this now that we have an API that
-		// can return an error. Should the default behaviour be to
-		// fail-closed as we do for plugins?
-		p.logger.Error("Log encoding failed: %v.", err)
-		return
+		// If there's no ND builtins cache in the event, then we don't
+		// need to retry encoding anything.
+		if event.NDBuiltinCache == nil {
+			// TODO(tsandall): revisit this now that we have an API that
+			// can return an error. Should the default behaviour be to
+			// fail-closed as we do for plugins?
+			p.logger.Error("Log encoding failed: %v.", err)
+			return
+		}
+
+		// Attempt to encode the event again, dropping the ND builtins cache.
+		newEvent := event
+		newEvent.NDBuiltinCache = nil
+
+		result, err = p.enc.Write(newEvent)
+		if err != nil {
+			p.logger.Error("Log encoding failed: %v.", err)
+			return
+		}
+
+		// Re-encoding was successful, but we still need to alert users.
+		p.logger.Error("ND builtins cache dropped from this event to fit under maximum upload size limits. Increase upload size limit or change usage of non-deterministic builtins.")
+		p.metrics.Counter(logNDBDropCounterName).Incr()
 	}
 
 	for _, chunk := range result {