wenjin272 commented on code in PR #821: URL: https://github.com/apache/flink-agents/pull/821#discussion_r3719020563
########## plan/src/main/java/org/apache/flink/agents/plan/condition/TriggerCondition.java: ########## @@ -0,0 +1,118 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.agents.plan.condition; + +import java.util.Objects; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** One classified entry from an action's {@code trigger_conditions}. */ +public abstract class TriggerCondition { + + private static final Pattern EVENT_TYPE = + Pattern.compile( + "^(?:" + + "([A-Za-z_][A-Za-z0-9_-]*(?:\\.[A-Za-z_][A-Za-z0-9_-]*)*)" + + "|(['\"])([^\\s'\"\\\\\\p{Cntrl}]+)\\2" + + ")$"); + private static final Set<String> EXPRESSION_LITERALS = Set.of("true", "false", "null"); + + TriggerCondition() {} + + /** Classifies an entry as an event-type condition or an expression condition. */ + public static TriggerCondition classify(String source) { + if (source == null || source.trim().isEmpty()) { + throw new IllegalArgumentException("Trigger condition must be non-null and non-blank"); + } + String entry = source.trim(); + Matcher matcher = EVENT_TYPE.matcher(entry); + if (matcher.matches()) { + String bareType = matcher.group(1); + String eventType = bareType != null ? bareType : matcher.group(3); + boolean reservedExpression = + eventType.startsWith("EventType.") Review Comment: Quoted event-type selectors should bypass reserved-expression detection. Currently, `'EventType.custom'` matches the quoted branch, but this prefix check still classifies it as an `ExpressionCondition`. CEL then sees only a string literal, not an `EventType` lookup, and Runtime rejects it as non-Boolean. Apply the `EventType.` prefix check only when `bareType != null`, and add a quoted `EventType.*` regression alongside the quoted-keyword case. ########## runtime/src/main/java/org/apache/flink/agents/runtime/condition/ConditionEvaluator.java: ########## @@ -0,0 +1,186 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.agents.runtime.condition; + +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.cel.common.values.NullValue; +import dev.cel.runtime.CelEvaluationException; +import org.apache.flink.agents.api.Event; +import org.apache.flink.agents.api.EventType; +import org.apache.flink.agents.api.configuration.AgentConfigOptions.ConditionEvaluationFailureStrategy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Evaluates trigger condition expressions against event data. */ +final class ConditionEvaluator { + + private static final Logger LOG = LoggerFactory.getLogger(ConditionEvaluator.class); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private final ConditionEvaluationFailureStrategy failureStrategy; + + ConditionEvaluator(ConditionEvaluationFailureStrategy failureStrategy) { + this.failureStrategy = failureStrategy; + } + + /** Builds only the variables needed by one reached condition, then evaluates it. */ + boolean evaluate(ConditionExpressionCompiler.CompiledCondition condition, Event event) { + Map<String, Object> conditionVariables; + try { + conditionVariables = buildConditionVariables(event, condition); + } catch (RuntimeException e) { + return handleConditionFailure( + "Building trigger condition variables failed for event " + event.getId(), e); + } + + String source = condition.source(); + Object result; + try { + result = condition.program().eval(conditionVariables); + } catch (CelEvaluationException e) { + return handleConditionFailure( + "Trigger condition evaluation failed for '" + source + "'", e); + } + if (result instanceof Boolean) { + return (Boolean) result; + } + return handleConditionFailure( + String.format( + "Trigger condition '%s' returned non-boolean type %s", + source, result == null ? "null" : result.getClass().getName()), + null); + } + + /** + * Applies the failure strategy to one condition failure: {@code FAIL} throws, otherwise the + * failure is logged and the condition is treated as false. + */ + private boolean handleConditionFailure(String message, Throwable cause) { + if (failureStrategy == ConditionEvaluationFailureStrategy.FAIL) { + throw new IllegalStateException(message, cause); + } + LOG.warn("{}, treating this condition as false", message, cause); + return false; + } + + /** + * Builds the condition variables required for {@code event} from the referenced keys of {@code + * condition}. + * + * <p>The {@code attributes} entry remains the explicit root namespace for the event attribute + * map. This is needed for literal keys that cannot be represented as identifiers, such as + * {@code attributes["www.andriod.com"].ip}; simple keys are also promoted for bare-identifier + * access. + */ + Map<String, Object> buildConditionVariables( + Event event, ConditionExpressionCompiler.CompiledCondition condition) { + Set<String> referencedTopLevelAttributeKeys = condition.referencedTopLevelAttributeKeys(); + + Map<String, Object> conditionVariables = new HashMap<>(); + conditionVariables.put("type", event.getType()); + conditionVariables.put("EventType", EventType.allConstants()); + conditionVariables.put("id", event.getId().toString()); + + Map<String, Object> attrs = event.getAttributes(); + Map<String, Object> normalizedAttributes = new HashMap<>(); + for (String key : referencedTopLevelAttributeKeys) { + if (attrs.containsKey(key)) { + normalizedAttributes.put(key, normalizeValue(attrs.get(key))); + } + } + + conditionVariables.put("attributes", normalizedAttributes); + // Promote to top level for bare-identifier access; framework keys win on collision. + normalizedAttributes.forEach(conditionVariables::putIfAbsent); + return conditionVariables; + } + + /** + * Normalizes Java values for condition evaluation, converting nulls, numbers, collections, and + * Jackson-serializable objects while preserving evaluator-native values. + */ + @SuppressWarnings("unchecked") + private static Object normalizeValue(Object value) { + if (value == null) { + return NullValue.NULL_VALUE; + } + if (value instanceof Map) { + Map<String, Object> src = (Map<String, Object>) value; + Map<String, Object> dst = new HashMap<>(src.size()); + for (Map.Entry<String, Object> entry : src.entrySet()) { + dst.put(entry.getKey(), normalizeValue(entry.getValue())); + } + return dst; + } + if (value instanceof List) { + List<Object> src = (List<Object>) value; + List<Object> dst = new ArrayList<>(src.size()); + for (Object item : src) { + dst.add(normalizeValue(item)); + } + return dst; + } + if (value instanceof Byte || value instanceof Short || value instanceof Integer) { + return ((Number) value).longValue(); + } + if (value instanceof Float) { + return ((Float) value).doubleValue(); Review Comment: `Float.doubleValue()` makes condition routing depend on whether the event has crossed the JSON / cross-language boundary. For example, `0.1F` is widened to `0.10000000149011612D`, so `score > 0.1` is true for a typed Java event. Jackson serializes the same value as JSON `0.1` and deserializes it as `Double(0.1)`, making the same condition false after a round trip. Normalize a `Float` through its decimal JSON representation, for example `Double.parseDouble(Float.toString((Float) value))`, and add a typed-vs-JSON-round-trip regression using `0.1F`; the current `4.5F` test is exactly representable and misses this case. ########## dist/src/main/resources/META-INF/NOTICE: ########## @@ -14,6 +14,12 @@ This project bundles the following dependencies under the Apache Software Licens - com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.18.2 - com.fasterxml.jackson.module:jackson-module-kotlin:2.18.2 - com.fasterxml:classmate:1.7.0 +- dev.cel:cel:0.12.0 +- dev.cel:common:0.12.0 +- dev.cel:compiler:0.12.0 +- dev.cel:protobuf:0.12.0 Review Comment: After the rebase, this NOTICE no longer matches the final `dist/common` dependency tree. CEL resolves `protobuf-java:4.33.5`, `re2j:1.8`, `antlr4-runtime:4.13.2`, and `threeten-extra:1.8.0`; however, the NOTICE still declares `protobuf-java:3.25.5` and omits the other three BSD dependencies. These entries existed in the earlier PR version and should be restored. The tree also shows CEL's protobuf 4.33.5 winning over the 3.25.5 versions requested through Gemini and Milvus, so the selected protobuf version should be explicitly managed and verified rather than being chosen implicitly by Maven dependency mediation. ########## docs/yaml-schema.json: ########## @@ -35,6 +35,7 @@ }, "trigger_conditions": { "items": { + "pattern": "\\S", Review Comment: This schema update also needs to be synchronized to `dev/agent-skills/flink-agents-dev/assets/yaml-schema.json`. The bundled schema still describes condition expressions as a future feature and lacks this non-blank pattern, while `assets/yaml-contracts.yaml` still records the old `183cc7ac...` blob SHA. The skill explicitly treats that file as the matching offline schema for the repository revision, so update both the bundled schema and its manifest hash. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
