gnodet-bot commented on code in PR #29831: URL: https://github.com/apache/camel/pull/29831#discussion_r4562787329
########## components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaPolicyReloadStatus.java: ########## @@ -0,0 +1,53 @@ +/* + * 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.camel.component.opa; + +import java.time.Instant; + +/** An immutable view of the active WebAssembly revision and the last reload attempt. */ +public final class OpaPolicyReloadStatus { + + private final String activeRevision; + private final Instant lastSuccessfulReload; + private final Instant lastFailedReload; + private final String lastReloadFailureType; + + OpaPolicyReloadStatus(String activeRevision, Instant lastSuccessfulReload, Instant lastFailedReload, + String lastReloadFailureType) { + this.activeRevision = activeRevision; + this.lastSuccessfulReload = lastSuccessfulReload; + this.lastFailedReload = lastFailedReload; + this.lastReloadFailureType = lastReloadFailureType; + } + Review Comment: **Missing Javadoc on public getters.** `getActiveRevision()`, `getLastSuccessfulReload()`, and `getLastFailedReload()` are undocumented. `getLastReloadFailureType()` has a comment, the others don't. ```suggestion /** SHA-256 hex digest of the active WASM module and data bundle. */ public String getActiveRevision() { ``` ########## components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaReloadableWasmEvaluator.java: ########## @@ -0,0 +1,308 @@ +/* + * 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.camel.component.opa; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.time.Instant; +import java.util.HexFormat; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.camel.CamelContext; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Atomically replaces validated WASM policy/data snapshots while admitted evaluations drain the previous snapshot. + * Resource loading and policy construction never hold the decision-state lock. + */ +public final class OpaReloadableWasmEvaluator extends OpaPolicyEvaluator implements AutoCloseable { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final Logger LOG = LoggerFactory.getLogger(OpaReloadableWasmEvaluator.class); + + @FunctionalInterface + interface BundleLoader { + OpaWasmEvaluator.Bundle load() throws Exception; + } + + @FunctionalInterface + interface EvaluatorFactory { + OpaWasmEvaluator create(OpaWasmEvaluator.Bundle bundle) throws Exception; + } + + private final Object state = new Object(); + private final Object reloadLock = new Object(); + private final CamelContext camelContext; + private final long reloadInterval; + private final BundleLoader loader; + private final EvaluatorFactory factory; + private Snapshot active; + private boolean closed; + private boolean reloadEnabled = true; + private long watchGeneration; + private ScheduledExecutorService executor; + private ScheduledFuture<?> watch; + private volatile OpaPolicyReloadStatus status; + + OpaReloadableWasmEvaluator(CamelContext camelContext, long reloadInterval, BundleLoader loader, + EvaluatorFactory factory, String policyPath, String allowKey, String includeHeaders, + String includeProperties, boolean includeBody, boolean failOpen) throws Exception { + super(policyPath, allowKey, includeHeaders, includeProperties, includeBody, failOpen); + if (reloadInterval < 0) { + throw new IllegalArgumentException("policyReloadInterval must not be negative"); + } + this.camelContext = ObjectHelper.notNull(camelContext, "camelContext"); + this.reloadInterval = reloadInterval; + this.loader = loader; + this.factory = factory; + OpaWasmEvaluator.Bundle bundle = immutableBundle(loader.load()); + String revision = revision(bundle); + // OpaWasmEvaluator warms its pool and applies the configured entrypoint/data before returning. + this.active = new Snapshot(factory.create(bundle), revision); + this.status = new OpaPolicyReloadStatus(revision, Instant.now(), null, null); + } + + public static OpaReloadableWasmEvaluator create( + CamelContext camelContext, String policyBundle, String entrypoint, int poolSize, long borrowTimeout, + long reloadInterval, String policyPath, String allowKey, String includeHeaders, String includeProperties, + boolean includeBody, boolean failOpen) + throws Exception { + if (ObjectHelper.isEmpty(policyBundle)) { + throw new IllegalArgumentException("policyBundle is required when evaluationMode=wasm"); + } + if (poolSize < 1) { + throw new IllegalArgumentException("poolSize must be at least 1 when evaluationMode=wasm, was " + poolSize); + } + String resolvedEntrypoint = ObjectHelper.isNotEmpty(entrypoint) ? entrypoint : policyPath; + return new OpaReloadableWasmEvaluator( + camelContext, reloadInterval, + () -> OpaWasmEvaluator.loadPolicy(camelContext, policyBundle), + bundle -> new OpaWasmEvaluator( + bundle.wasm(), bundle.data(), resolvedEntrypoint, poolSize, borrowTimeout, + policyPath, allowKey, includeHeaders, includeProperties, includeBody, failOpen), + policyPath, allowKey, includeHeaders, includeProperties, includeBody, failOpen); + } + + public OpaPolicyReloadStatus getReloadStatus() { + return status; + } + + /** Re-reads the configured resource and publishes a validated replacement only when its content changed. */ + public boolean reload() throws Exception { + long generation; + synchronized (state) { + ensureOpen(-1); + if (!reloadEnabled) { + throw new IllegalStateException("The WASM policy reload watcher is stopped"); + } + generation = watchGeneration; + } + return reload(generation); + } + + private boolean reload(long generation) throws Exception { + synchronized (reloadLock) { + synchronized (state) { + ensureOpen(generation); + } + OpaWasmEvaluator replacement = null; + boolean published = false; + try { + OpaWasmEvaluator.Bundle bundle = immutableBundle(loader.load()); + String revision = revision(bundle); + synchronized (state) { + ensureOpen(generation); + if (active.revision.equals(revision)) { + status = new OpaPolicyReloadStatus(revision, Instant.now(), null, null); + return false; + } + } + replacement = factory.create(bundle); + Snapshot previous; + synchronized (state) { + ensureOpen(generation); + previous = active; + active = new Snapshot(replacement, revision); + published = true; + previous.retired = true; + status = new OpaPolicyReloadStatus(revision, Instant.now(), null, null); + if (previous.readers != 0) { + previous = null; + } + } + if (previous != null) { + previous.evaluator.close(); + } + return true; + } catch (Exception e) { + synchronized (state) { + if (!closed && (generation < 0 || generation == watchGeneration)) { + status = new OpaPolicyReloadStatus( + active.revision, status.getLastSuccessfulReload(), + Instant.now(), e.getClass().getName()); + } + } + throw e; + } finally { + if (replacement != null && !published) { + replacement.close(); + } + } + } + } + + public void startWatching() { Review Comment: **`startWatching()` and `stopWatching()` have no Javadoc.** `stopWatching()` in particular has a non-obvious side effect: after it is called, `reload()` throws `IllegalStateException` even when `reloadInterval=0` (no auto-reload was configured). That coupling should be documented. ```suggestion /** * Enables the managed polling watcher, starting the scheduler when {@code reloadInterval > 0}. * Safe to call after {@link #stopWatching()} to re-enable reloads. */ public void startWatching() { ``` ########## components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaPolicyReloadStatus.java: ########## @@ -0,0 +1,53 @@ +/* + * 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.camel.component.opa; + +import java.time.Instant; + +/** An immutable view of the active WebAssembly revision and the last reload attempt. */ +public final class OpaPolicyReloadStatus { + + private final String activeRevision; + private final Instant lastSuccessfulReload; + private final Instant lastFailedReload; + private final String lastReloadFailureType; + + OpaPolicyReloadStatus(String activeRevision, Instant lastSuccessfulReload, Instant lastFailedReload, + String lastReloadFailureType) { + this.activeRevision = activeRevision; + this.lastSuccessfulReload = lastSuccessfulReload; + this.lastFailedReload = lastFailedReload; + this.lastReloadFailureType = lastReloadFailureType; + } + + public String getActiveRevision() { + return activeRevision; + } + Review Comment: ```suggestion /** Timestamp of the last successful reload, or the construction time for the initial load. Never null. */ public Instant getLastSuccessfulReload() { ``` ########## components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaPolicyReloadStatus.java: ########## @@ -0,0 +1,53 @@ +/* + * 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.camel.component.opa; + +import java.time.Instant; + +/** An immutable view of the active WebAssembly revision and the last reload attempt. */ +public final class OpaPolicyReloadStatus { + + private final String activeRevision; + private final Instant lastSuccessfulReload; + private final Instant lastFailedReload; + private final String lastReloadFailureType; + + OpaPolicyReloadStatus(String activeRevision, Instant lastSuccessfulReload, Instant lastFailedReload, + String lastReloadFailureType) { + this.activeRevision = activeRevision; + this.lastSuccessfulReload = lastSuccessfulReload; + this.lastFailedReload = lastFailedReload; + this.lastReloadFailureType = lastReloadFailureType; + } + + public String getActiveRevision() { + return activeRevision; + } + + public Instant getLastSuccessfulReload() { + return lastSuccessfulReload; + } + Review Comment: ```suggestion /** Timestamp of the last failed reload attempt, or {@code null} if no reload has failed. */ public Instant getLastFailedReload() { ``` ########## components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaReloadableWasmEvaluator.java: ########## @@ -0,0 +1,308 @@ +/* + * 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.camel.component.opa; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.time.Instant; +import java.util.HexFormat; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.camel.CamelContext; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Atomically replaces validated WASM policy/data snapshots while admitted evaluations drain the previous snapshot. + * Resource loading and policy construction never hold the decision-state lock. + */ +public final class OpaReloadableWasmEvaluator extends OpaPolicyEvaluator implements AutoCloseable { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final Logger LOG = LoggerFactory.getLogger(OpaReloadableWasmEvaluator.class); + + @FunctionalInterface + interface BundleLoader { + OpaWasmEvaluator.Bundle load() throws Exception; + } + + @FunctionalInterface + interface EvaluatorFactory { + OpaWasmEvaluator create(OpaWasmEvaluator.Bundle bundle) throws Exception; + } + + private final Object state = new Object(); + private final Object reloadLock = new Object(); + private final CamelContext camelContext; + private final long reloadInterval; + private final BundleLoader loader; + private final EvaluatorFactory factory; + private Snapshot active; + private boolean closed; + private boolean reloadEnabled = true; + private long watchGeneration; + private ScheduledExecutorService executor; + private ScheduledFuture<?> watch; + private volatile OpaPolicyReloadStatus status; + + OpaReloadableWasmEvaluator(CamelContext camelContext, long reloadInterval, BundleLoader loader, + EvaluatorFactory factory, String policyPath, String allowKey, String includeHeaders, + String includeProperties, boolean includeBody, boolean failOpen) throws Exception { + super(policyPath, allowKey, includeHeaders, includeProperties, includeBody, failOpen); + if (reloadInterval < 0) { + throw new IllegalArgumentException("policyReloadInterval must not be negative"); + } + this.camelContext = ObjectHelper.notNull(camelContext, "camelContext"); + this.reloadInterval = reloadInterval; + this.loader = loader; + this.factory = factory; + OpaWasmEvaluator.Bundle bundle = immutableBundle(loader.load()); + String revision = revision(bundle); + // OpaWasmEvaluator warms its pool and applies the configured entrypoint/data before returning. + this.active = new Snapshot(factory.create(bundle), revision); + this.status = new OpaPolicyReloadStatus(revision, Instant.now(), null, null); + } + + public static OpaReloadableWasmEvaluator create( + CamelContext camelContext, String policyBundle, String entrypoint, int poolSize, long borrowTimeout, + long reloadInterval, String policyPath, String allowKey, String includeHeaders, String includeProperties, + boolean includeBody, boolean failOpen) + throws Exception { + if (ObjectHelper.isEmpty(policyBundle)) { + throw new IllegalArgumentException("policyBundle is required when evaluationMode=wasm"); + } + if (poolSize < 1) { + throw new IllegalArgumentException("poolSize must be at least 1 when evaluationMode=wasm, was " + poolSize); + } + String resolvedEntrypoint = ObjectHelper.isNotEmpty(entrypoint) ? entrypoint : policyPath; + return new OpaReloadableWasmEvaluator( + camelContext, reloadInterval, + () -> OpaWasmEvaluator.loadPolicy(camelContext, policyBundle), + bundle -> new OpaWasmEvaluator( + bundle.wasm(), bundle.data(), resolvedEntrypoint, poolSize, borrowTimeout, + policyPath, allowKey, includeHeaders, includeProperties, includeBody, failOpen), + policyPath, allowKey, includeHeaders, includeProperties, includeBody, failOpen); + } + + public OpaPolicyReloadStatus getReloadStatus() { + return status; + } + + /** Re-reads the configured resource and publishes a validated replacement only when its content changed. */ + public boolean reload() throws Exception { + long generation; + synchronized (state) { + ensureOpen(-1); + if (!reloadEnabled) { + throw new IllegalStateException("The WASM policy reload watcher is stopped"); + } + generation = watchGeneration; + } + return reload(generation); + } + + private boolean reload(long generation) throws Exception { + synchronized (reloadLock) { + synchronized (state) { + ensureOpen(generation); + } + OpaWasmEvaluator replacement = null; + boolean published = false; + try { + OpaWasmEvaluator.Bundle bundle = immutableBundle(loader.load()); + String revision = revision(bundle); + synchronized (state) { + ensureOpen(generation); + if (active.revision.equals(revision)) { + status = new OpaPolicyReloadStatus(revision, Instant.now(), null, null); + return false; + } + } + replacement = factory.create(bundle); + Snapshot previous; + synchronized (state) { + ensureOpen(generation); + previous = active; + active = new Snapshot(replacement, revision); + published = true; + previous.retired = true; + status = new OpaPolicyReloadStatus(revision, Instant.now(), null, null); + if (previous.readers != 0) { + previous = null; + } + } + if (previous != null) { + previous.evaluator.close(); + } + return true; + } catch (Exception e) { + synchronized (state) { + if (!closed && (generation < 0 || generation == watchGeneration)) { + status = new OpaPolicyReloadStatus( + active.revision, status.getLastSuccessfulReload(), + Instant.now(), e.getClass().getName()); + } + } + throw e; + } finally { + if (replacement != null && !published) { + replacement.close(); + } + } + } + } + + public void startWatching() { + synchronized (state) { + ensureOpen(-1); + reloadEnabled = true; + if (reloadInterval == 0 || watch != null) { + return; + } + long generation = ++watchGeneration; + executor = camelContext.getExecutorServiceManager().newSingleThreadScheduledExecutor(this, "opa-policy-reload"); + try { + watch = executor.scheduleWithFixedDelay(() -> poll(generation), reloadInterval, reloadInterval, + TimeUnit.MILLISECONDS); + } catch (RuntimeException e) { + camelContext.getExecutorServiceManager().shutdownNow(executor); + executor = null; + throw e; + } + } + } + + public void stopWatching() { Review Comment: ```suggestion /** * Disables managed polling and cancels the scheduler. After this call, {@link #reload()} throws * {@link IllegalStateException} until {@link #startWatching()} is called again. */ public void stopWatching() { ``` ########## components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaEndpoint.java: ########## @@ -165,11 +165,31 @@ private void warnAboutIgnoredServerOptions() { } private OpaPolicyEvaluator createWasmEvaluator() throws Exception { - return OpaWasmEvaluator.create( + OpaReloadableWasmEvaluator replacement = OpaReloadableWasmEvaluator.create( getCamelContext(), configuration.getPolicyBundle(), configuration.getEntrypoint(), - configuration.getPoolSize(), configuration.getBorrowTimeout(), policyPath, configuration.getAllowKey(), + configuration.getPoolSize(), configuration.getBorrowTimeout(), configuration.getPolicyReloadInterval(), + policyPath, configuration.getAllowKey(), configuration.getIncludeHeaders(), configuration.getIncludeProperties(), configuration.isIncludeBody(), configuration.isFailOpen()); + try { + replacement.startWatching(); + return replacement; + } catch (RuntimeException e) { + replacement.close(); + throw e; + } + } + + /** Explicitly reloads the configured WASM policy, retaining the active revision if validation fails. */ + public boolean reloadPolicy() throws Exception { + if (evaluator instanceof OpaReloadableWasmEvaluator wasm) { + return wasm.reload(); + } + throw new IllegalStateException("A started WASM endpoint is required to reload its policy"); + } + + public OpaPolicyReloadStatus getPolicyReloadStatus() { Review Comment: **Missing Javadoc.** ```suggestion /** Returns the active reload status, or {@code null} when this endpoint is not in WASM mode. */ public OpaPolicyReloadStatus getPolicyReloadStatus() { ``` -- 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]
