ewilliams28/releases
1
0
Fork 0
mirror of https://github.com/open-policy-agent/opa.git synced 2026-08-12 19:32:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

inmem: allow passing triggers (AST) data without conversion (#7959)

Browse Source 
Fixes #7958

Signed-off-by: Anders Eknert <anders@eknert.com>
This commit is contained in:
Anders Eknert
2025-11-19 14:04:00 +01:00
committed by GitHub
parent 3188e04a57
commit f2f0ef080c
4 changed files with 146 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
v1/storage/inmem/inmem.go
+24 -13
View File
@@ -349,10 +349,25 @@ func (h *handle) Unregister(_ context.Context, txn storage.Transaction) {
}
func (db *store) runOnCommitTriggers(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) {
if db.returnASTValuesOnRead && len(db.triggers) > 0 {
// FIXME: Not very performant for large data.
// While it's unlikely, the API allows one trigger to be configured to want
// data conversion, and another that doesn't. So let's handle that properly.
var wantsDataConversion bool
if db.returnASTValuesOnRead && len(event.Data) > 0 {
for _, t := range db.triggers {
if !t.SkipDataConversion {
wantsDataConversion = true
break
}
}
}
dataEvents := make([]storage.DataEvent, 0, len(event.Data))
var converted storage.TriggerEvent
if wantsDataConversion {
converted = storage.TriggerEvent{
Policy: event.Policy,
Data: make([]storage.DataEvent, 0, len(event.Data)),
Context: event.Context,
}
for _, dataEvent := range event.Data {
if astData, ok := dataEvent.Data.(ast.Value); ok {
@@ -360,25 +375,21 @@ func (db *store) runOnCommitTriggers(ctx context.Context, txn storage.Transactio
if err != nil {
panic(err)
}
dataEvents = append(dataEvents, storage.DataEvent{
converted.Data = append(converted.Data, storage.DataEvent{
Path: dataEvent.Path,
Data: jsn,
Removed: dataEvent.Removed,
})
} else {
dataEvents = append(dataEvents, dataEvent)
}
}
event = storage.TriggerEvent{
Policy: event.Policy,
Data: dataEvents,
Context: event.Context,
}
}
for _, t := range db.triggers {
t.OnCommit(ctx, txn, event)
if wantsDataConversion && !t.SkipDataConversion {
t.OnCommit(ctx, txn, converted)
} else {
t.OnCommit(ctx, txn, event)
}
}
}
v1/storage/inmem/inmem_bench_test.go
+58 -2
View File
@@ -211,8 +211,8 @@ func BenchmarkWriteAndCommit(b *testing.B) {
AllStores(map[string]any{}).Bench(b, operation)
}
// Go 48750 ns/op 27040 B/op 311 allocs/op (no additional cost of triggers)
// AST 191501 ns/op 37585 B/op 616 allocs/op (extra cost du to converting back to Go values.. why?)
// Go 48399 ns/op 16209 B/op 304 allocs/op (no additional cost of triggers)
// AST 191501 ns/op 26673 B/op 605 allocs/op (extra cost due to converting back to Go values)
func BenchmarkWriteAndCommitWithTriggers(b *testing.B) {
paths := make([]storage.Path, 100)
for i := range 100 {
@@ -239,6 +239,62 @@ func BenchmarkWriteAndCommitWithTriggers(b *testing.B) {
Bench(b, operation)
}
// AST 189485 ns/op 17011 B/op 304 allocs/op
func BenchmarkWriteAndCommitWithTriggersSkipConversion(b *testing.B) {
paths := make([]storage.Path, 100)
for i := range 100 {
paths[i] = storage.Path{strconv.Itoa(i)}
}
values := make([]ast.Value, 100)
for i := range 100 {
values[i] = ast.String(paths[i][0])
}
operation := func(ctx context.Context, target *target) error {
txn, _ := target.store.NewTransaction(b.Context(), storage.WriteParams)
for i := range 100 {
if err := target.store.Write(b.Context(), txn, storage.AddOp, paths[i], values[i]); err != nil {
return err
}
}
return target.store.Commit(b.Context(), txn)
}
triggerCount := 0
trigger := storage.TriggerConfig{
SkipDataConversion: true,
OnCommit: func(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) {
if event.DataChanged() {
if len(event.Data) != 100 {
b.Fatalf("Expected 100 data changes but got: %d", len(event.Data))
}
if _, ok := event.Data[0].Data.(ast.Value); !ok {
b.Fatalf("Expected ast.Value data but got: %T", event.Data[0].Data)
}
triggerCount++
}
},
}
onlyAstStores := targets{{
name: "AST",
store: inmem.NewFromObjectWithOpts(map[string]any{}, inmem.OptReturnASTValuesOnRead(true)),
isAST: true,
}}
onlyAstStores.
SetupWithTxn(b, writeTxn, func(ctx context.Context, target *target) error {
return onlyError(target.store.Register(b.Context(), target.txn, trigger))
}).
Bench(b, operation)
if triggerCount == 0 {
b.Fatalf("Expected trigger to be called at least once")
}
}
func (t targets) VerifyRead(b *testing.B, path storage.Path, expected any) targets {
b.Helper()
for _, target := range t {
v1/storage/inmem/inmem_test.go
+60
View File
@@ -1150,6 +1150,66 @@ func TestInMemoryTriggers(t *testing.T) {
}
}
func TestASTInMemoryTriggersDataConversion(t *testing.T) {
ctx := t.Context()
store := NewFromObjectWithOpts(loadSmallTestData(), OptReturnASTValuesOnRead(true))
noConversionTriggerCount := 0
conversionTriggerCount := 0
// Register a trigger that doesn't want data conversion
err := storage.Txn(t.Context(), store, storage.WriteParams, func(txn storage.Transaction) error {
_, err := store.Register(ctx, txn, storage.TriggerConfig{
SkipDataConversion: true,
OnCommit: func(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) {
if event.DataChanged() {
if _, ok := event.Data[0].Data.(ast.Value); !ok {
t.Fatalf("Expected ast.Value data but got: %T", event.Data[0].Data)
}
noConversionTriggerCount++
}
},
})
return err
})
if err != nil {
t.Fatalf("Failed to register trigger: %v", err)
}
// Register a trigger that wants data conversion (skip data conversion not set, as is the default)
err = storage.Txn(t.Context(), store, storage.WriteParams, func(txn storage.Transaction) error {
_, err := store.Register(ctx, txn, storage.TriggerConfig{
OnCommit: func(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) {
if event.DataChanged() {
if _, ok := event.Data[0].Data.(ast.Value); ok {
t.Fatalf("Expected non-ast.Value data but got: %T", event.Data[0].Data)
}
conversionTriggerCount++
}
},
})
return err
})
if err != nil {
t.Fatalf("Failed to register trigger: %v", err)
}
err = storage.Txn(t.Context(), store, storage.WriteParams, func(txn storage.Transaction) error {
return store.Write(ctx, txn, storage.ReplaceOp, storage.MustParsePath("/a"), "hello")
})
if err != nil {
t.Fatalf("Failed to write data: %v", err)
}
if noConversionTriggerCount != 1 {
t.Fatalf("Expected no conversion trigger to be called once but got: %d", noConversionTriggerCount)
}
if conversionTriggerCount != 1 {
t.Fatalf("Expected conversion trigger to be called once but got: %d", conversionTriggerCount)
}
}
func TestInMemoryTriggersUnregister(t *testing.T) {
ctx := t.Context()
store := NewFromObject(loadSmallTestData())
v1/storage/interface.go
+4
View File
@@ -210,6 +210,10 @@ func (e TriggerEvent) DataChanged() bool {
// TriggerConfig contains the trigger registration configuration.
type TriggerConfig struct {
// SkipDataConversion when set to true, avoids converting data passed to
// trigger functions from the store to Go types, and instead passes the
// original representation (e.g., ast.Value).
SkipDataConversion bool
// OnCommit is invoked when a transaction is successfully committed. The
// callback is invoked with a handle to the write transaction that