Cache CEL programs in order to avoid expensive heap allocations on each event match (#3667)

* use LRU cache for CEL programs

* err check boundary

* use hash expr

Author: mnafees

pkg/repository/match.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"hash/fnv"
 	"slices"
 	"time"
 
@@ -932,18 +933,29 @@ func (m *sharedRepository) processCELExpressions(ctx context.Context, events []C
 			expr = "true"
 		}
 
-		ast, issues := m.env.Compile(expr)
+		hasher := fnv.New64a()
+		hasher.Write([]byte(expr))
+		exprHash := hasher.Sum64()
 
-		if issues != nil {
-			m.l.Error().Ctx(ctx).Msgf("failed to compile CEL expression: %s", issues.String())
-			continue
-		}
+		program, ok := m.celProgramCache.Get(exprHash)
 
-		program, err := m.env.Program(ast)
+		if !ok {
+			ast, issues := m.env.Compile(expr)
 
-		if err != nil {
-			m.l.Error().Ctx(ctx).Err(err).Msgf("failed to create CEL program: %s", expr)
-			continue
+			if issues != nil && issues.Err() != nil {
+				m.l.Error().Ctx(ctx).Err(issues.Err()).Msgf("failed to compile CEL expression: %s", expr)
+				continue
+			}
+
+			compiled, err := m.env.Program(ast)
+
+			if err != nil {
+				m.l.Error().Ctx(ctx).Err(err).Msgf("failed to create CEL program: %s", expr)
+				continue
+			}
+
+			m.celProgramCache.Add(exprHash, compiled)
+			program = compiled
 		}
 
 		programs[condition.ID] = program

pkg/repository/shared.go

@@ -43,6 +43,7 @@ type sharedRepository struct {
 
 	celParser       *cel.CELParser
 	env             *celgo.Env
+	celProgramCache *lru.Cache[uint64, celgo.Program]
 	taskLookupCache *lru.Cache[taskExternalIdTenantIdTuple, *sqlcv1.FlattenExternalIdsRow]
 	payloadStore    PayloadStoreRepository
 	m               TenantLimitRepository
@@ -86,6 +87,12 @@ func newSharedRepository(
 		log.Fatalf("failed to create LRU cache: %v", err)
 	}
 
+	celProgramCache, err := lru.New[uint64, celgo.Program](50000)
+
+	if err != nil {
+		log.Fatalf("failed to create CEL program cache: %v", err)
+	}
+
 	s := &sharedRepository{
 		pool:                        pool,
 		ddlPool:                     ddlPool,
@@ -101,6 +108,7 @@ func newSharedRepository(
 		stepIdSlotRequestsCache:     stepIdSlotRequestsCache,
 		celParser:                   celParser,
 		env:                         env,
+		celProgramCache:             celProgramCache,
 		taskLookupCache:             lookupCache,
 		payloadStore:                payloadStore,
 	}