diff options
Diffstat (limited to 'feature-pooling-messages/src/main/java/org')
25 files changed, 3918 insertions, 0 deletions
diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/CancellableScheduledTask.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/CancellableScheduledTask.java new file mode 100644 index 00000000..4e9112e6 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/CancellableScheduledTask.java @@ -0,0 +1,34 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +/** + * A scheduled task that can be cancelled. + */ +@FunctionalInterface +public interface CancellableScheduledTask { + + /** + * Cancels the scheduled task. + */ + void cancel(); +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java new file mode 100644 index 00000000..6411dd81 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java @@ -0,0 +1,397 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +import java.util.List; +import java.util.Properties; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import lombok.AccessLevel; +import lombok.Getter; +import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure; +import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager; +import org.onap.policy.common.endpoints.event.comm.TopicSink; +import org.onap.policy.common.endpoints.event.comm.TopicSource; +import org.onap.policy.common.utils.properties.SpecProperties; +import org.onap.policy.common.utils.properties.exception.PropertyException; +import org.onap.policy.drools.controller.DroolsController; +import org.onap.policy.drools.features.DroolsControllerFeatureApi; +import org.onap.policy.drools.features.PolicyControllerFeatureApi; +import org.onap.policy.drools.features.PolicyEngineFeatureApi; +import org.onap.policy.drools.persistence.SystemPersistenceConstants; +import org.onap.policy.drools.system.PolicyController; +import org.onap.policy.drools.system.PolicyControllerConstants; +import org.onap.policy.drools.system.PolicyEngine; +import org.onap.policy.drools.util.FeatureEnabledChecker; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Controller/session pooling. Multiple hosts may be launched, all servicing the same + * controllers/sessions. When this feature is enabled, the requests are divided across the different + * hosts, instead of all running on a single, active host. + * + * <p>With each controller, there is an + * associated DMaaP topic that is used for internal communication between the different hosts + * serving the controller. + */ +public class PoolingFeature implements PolicyEngineFeatureApi, PolicyControllerFeatureApi, DroolsControllerFeatureApi { + + private static final Logger logger = LoggerFactory.getLogger(PoolingFeature.class); + + /** + * ID of this host. + */ + @Getter + private final String host; + + /** + * Entire set of feature properties, including those specific to various controllers. + */ + private Properties featProps = null; + + /** + * Maps a controller name to its associated manager. + */ + private final ConcurrentHashMap<String, PoolingManagerImpl> ctlr2pool = new ConcurrentHashMap<>(107); + + /** + * Decremented each time a manager enters the Active state. Used by junit tests. + */ + @Getter(AccessLevel.PROTECTED) + private final CountDownLatch activeLatch = new CountDownLatch(1); + + /** + * Topic names passed to beforeOffer(), which are saved for when the beforeInsert() is + * called later. As multiple threads can be active within the methods at the same + * time, we must keep this in thread local storage. + */ + private ThreadLocal<String> offerTopics = new ThreadLocal<>(); + + /** + * Constructor. + */ + public PoolingFeature() { + super(); + + this.host = UUID.randomUUID().toString(); + } + + @Override + public int getSequenceNumber() { + return 0; + } + + @Override + public boolean beforeStart(PolicyEngine engine) { + logger.info("initializing {}", PoolingProperties.FEATURE_NAME); + featProps = getProperties(PoolingProperties.FEATURE_NAME); + + // remove any generic pooling topic - always use controller-specific property + featProps.remove(PoolingProperties.POOLING_TOPIC); + + initTopicSources(featProps); + initTopicSinks(featProps); + + return false; + } + + @Override + public boolean beforeStart(PolicyController controller) { + return doManager(controller, mgr -> { + mgr.beforeStart(); + return false; + }); + } + + /** + * Adds the controller and a new pooling manager to {@link #ctlr2pool}. + * + * @throws PoolingFeatureRtException if an error occurs + */ + @Override + public boolean afterCreate(PolicyController controller) { + + if (featProps == null) { + logger.error("pooling feature properties have not been loaded"); + throw new PoolingFeatureRtException(new IllegalStateException("missing pooling feature properties")); + } + + String name = controller.getName(); + + var specProps = new SpecProperties(PoolingProperties.PREFIX, name, featProps); + + if (FeatureEnabledChecker.isFeatureEnabled(specProps, PoolingProperties.FEATURE_ENABLED)) { + try { + // get & validate the properties + var props = new PoolingProperties(name, featProps); + + logger.info("pooling enabled for {}", name); + ctlr2pool.computeIfAbsent(name, xxx -> makeManager(host, controller, props, activeLatch)); + + } catch (PropertyException e) { + logger.error("pooling disabled due to exception for {}", name); + throw new PoolingFeatureRtException(e); + } + + } else { + logger.info("pooling disabled for {}", name); + } + + + return false; + } + + @Override + public boolean afterStart(PolicyController controller) { + return doManager(controller, mgr -> { + mgr.afterStart(); + return false; + }); + } + + @Override + public boolean beforeStop(PolicyController controller) { + return doManager(controller, mgr -> { + mgr.beforeStop(); + return false; + }); + } + + @Override + public boolean afterStop(PolicyController controller) { + return doManager(controller, mgr -> { + mgr.afterStop(); + return false; + }); + } + + @Override + public boolean afterShutdown(PolicyController controller) { + return commonShutdown(controller); + } + + @Override + public boolean afterHalt(PolicyController controller) { + return commonShutdown(controller); + } + + private boolean commonShutdown(PolicyController controller) { + deleteManager(controller); + return false; + } + + @Override + public boolean beforeLock(PolicyController controller) { + return doManager(controller, mgr -> { + mgr.beforeLock(); + return false; + }); + } + + @Override + public boolean afterUnlock(PolicyController controller) { + return doManager(controller, mgr -> { + mgr.afterUnlock(); + return false; + }); + } + + @Override + public boolean beforeOffer(PolicyController controller, CommInfrastructure protocol, String topic2, String event) { + /* + * As this is invoked a lot, we'll directly call the manager's method instead of using the + * functional interface via doManager(). + */ + PoolingManagerImpl mgr = ctlr2pool.get(controller.getName()); + if (mgr == null) { + return false; + } + + if (mgr.beforeOffer(topic2, event)) { + return true; + } + + offerTopics.set(topic2); + return false; + } + + @Override + public boolean beforeInsert(DroolsController droolsController, Object fact) { + + String topic = offerTopics.get(); + if (topic == null) { + logger.warn("missing arguments for feature-pooling-messages in beforeInsert"); + return false; + } + + PolicyController controller; + try { + controller = getController(droolsController); + + } catch (IllegalArgumentException | IllegalStateException e) { + logger.warn("cannot get controller for {} {}", droolsController.getGroupId(), + droolsController.getArtifactId(), e); + return false; + } + + + if (controller == null) { + logger.warn("cannot determine controller for {} {}", droolsController.getGroupId(), + droolsController.getArtifactId()); + return false; + } + + /* + * As this is invoked a lot, we'll directly call the manager's method instead of using the + * functional interface via doManager(). + */ + PoolingManagerImpl mgr = ctlr2pool.get(controller.getName()); + if (mgr == null) { + return false; + } + + return mgr.beforeInsert(topic, fact); + } + + @Override + public boolean afterOffer(PolicyController controller, CommInfrastructure protocol, String topic, String event, + boolean success) { + + // clear any stored arguments + offerTopics.remove(); + + return false; + } + + /** + * Executes a function using the manager associated with the controller. Catches any exceptions + * from the function and re-throws it as a runtime exception. + * + * @param controller controller + * @param func function to be executed + * @return {@code true} if the function handled the request, {@code false} otherwise + * @throws PoolingFeatureRtException if an error occurs + */ + private boolean doManager(PolicyController controller, MgrFunc func) { + PoolingManagerImpl mgr = ctlr2pool.get(controller.getName()); + if (mgr == null) { + return false; + } + + try { + return func.apply(mgr); + + } catch (PoolingFeatureException e) { + throw new PoolingFeatureRtException(e); + } + } + + /** + * Deletes the manager associated with a controller. + * + * @param controller controller + * @throws PoolingFeatureRtException if an error occurs + */ + private void deleteManager(PolicyController controller) { + + String name = controller.getName(); + logger.info("remove feature-pooling-messages manager for {}", name); + + ctlr2pool.remove(name); + } + + /** + * Function that operates on a manager. + */ + @FunctionalInterface + private static interface MgrFunc { + + /** + * Apply. + * + * @param mgr manager + * @return {@code true} if the request was handled by the manager, {@code false} otherwise + * @throws PoolingFeatureException feature exception + */ + boolean apply(PoolingManagerImpl mgr) throws PoolingFeatureException; + } + + /* + * The remaining methods may be overridden by junit tests. + */ + + /** + * Get properties. + * + * @param featName feature name + * @return the properties for the specified feature + */ + protected Properties getProperties(String featName) { + return SystemPersistenceConstants.getManager().getProperties(featName); + } + + /** + * Makes a pooling manager for a controller. + * + * @param host name/uuid of this host + * @param controller controller + * @param props properties to use to configure the manager + * @param activeLatch decremented when the manager goes Active + * @return a new pooling manager + */ + protected PoolingManagerImpl makeManager(String host, PolicyController controller, PoolingProperties props, + CountDownLatch activeLatch) { + return new PoolingManagerImpl(host, controller, props, activeLatch); + } + + /** + * Gets the policy controller associated with a drools controller. + * + * @param droolsController drools controller + * @return the policy controller associated with a drools controller + */ + protected PolicyController getController(DroolsController droolsController) { + return PolicyControllerConstants.getFactory().get(droolsController); + } + + /** + * Initializes the topic sources. + * + * @param props properties used to configure the topics + * @return the topic sources + */ + protected List<TopicSource> initTopicSources(Properties props) { + return TopicEndpointManager.getManager().addTopicSources(props); + } + + /** + * Initializes the topic sinks. + * + * @param props properties used to configure the topics + * @return the topic sinks + */ + protected List<TopicSink> initTopicSinks(Properties props) { + return TopicEndpointManager.getManager().addTopicSinks(props); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeatureException.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeatureException.java new file mode 100644 index 00000000..5d7b9f76 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeatureException.java @@ -0,0 +1,54 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +import java.io.Serial; + +/** + * Exception thrown by the pooling feature. + */ +public class PoolingFeatureException extends Exception { + @Serial + private static final long serialVersionUID = 1L; + + public PoolingFeatureException() { + super(); + } + + public PoolingFeatureException(String message) { + super(message); + } + + public PoolingFeatureException(Throwable cause) { + super(cause); + } + + public PoolingFeatureException(String message, Throwable cause) { + super(message, cause); + } + + public PoolingFeatureException(String message, Throwable cause, boolean enableSuppression, + boolean writableStackTrace) { + super(message, cause, enableSuppression, writableStackTrace); + } + +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeatureRtException.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeatureRtException.java new file mode 100644 index 00000000..5d0a2755 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingFeatureRtException.java @@ -0,0 +1,54 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +import java.io.Serial; + +/** + * A runtime exception thrown by the pooling feature. + */ +public class PoolingFeatureRtException extends RuntimeException { + @Serial + private static final long serialVersionUID = 1L; + + public PoolingFeatureRtException() { + super(); + } + + public PoolingFeatureRtException(String message) { + super(message); + } + + public PoolingFeatureRtException(Throwable cause) { + super(cause); + } + + public PoolingFeatureRtException(String message, Throwable cause) { + super(message, cause); + } + + public PoolingFeatureRtException(String message, Throwable cause, boolean enableSuppression, + boolean writableStackTrace) { + super(message, cause, enableSuppression, writableStackTrace); + } + +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingManager.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingManager.java new file mode 100644 index 00000000..5e358e61 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingManager.java @@ -0,0 +1,133 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2020 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +import org.onap.policy.drools.pooling.message.BucketAssignments; +import org.onap.policy.drools.pooling.message.Message; +import org.onap.policy.drools.pooling.state.State; +import org.onap.policy.drools.pooling.state.StateTimerTask; + +/** + * Pooling manager for a single PolicyController. + */ +public interface PoolingManager { + + /** + * Gets the properties used to configure the manager. + * + * @return pooling properties + */ + PoolingProperties getProperties(); + + /** + * Gets the host id. + * + * @return the host id + */ + String getHost(); + + /** + * Gets the name of the internal DMaaP topic used by this manager to communicate with + * its other hosts. + * + * @return the name of the internal DMaaP topic + */ + String getTopic(); + + /** + * Starts distributing requests according to the given bucket assignments. + * + * @param assignments must <i>not</i> be {@code null} + */ + void startDistributing(BucketAssignments assignments); + + /** + * Gets the current bucket assignments. + * + * @return the current bucket assignments, or {@code null} if no assignments have been + * made + */ + BucketAssignments getAssignments(); + + /** + * Publishes a message to the internal topic on the administrative channel. + * + * @param msg message to be published + */ + void publishAdmin(Message msg); + + /** + * Publishes a message to the internal topic on a particular channel. + * + * @param channel channel on which the message should be published + * @param msg message to be published + */ + void publish(String channel, Message msg); + + /** + * Schedules a timer to fire after a delay. + * + * @param delayMs delay, in milliseconds + * @param task task + * @return a new scheduled task + */ + CancellableScheduledTask schedule(long delayMs, StateTimerTask task); + + /** + * Schedules a timer to fire repeatedly. + * + * @param initialDelayMs initial delay, in milliseconds + * @param delayMs delay, in milliseconds + * @param task task + * @return a new scheduled task + */ + CancellableScheduledTask scheduleWithFixedDelay(long initialDelayMs, long delayMs, StateTimerTask task); + + /** + * Transitions to the "start" state. + * + * @return the new state + */ + State goStart(); + + /** + * Transitions to the "query" state. + * + * @return the new state + */ + State goQuery(); + + /** + * Transitions to the "active" state. + * + * @return the new state + */ + State goActive(); + + /** + * Transitions to the "inactive" state. + * + * @return the new state + */ + State goInactive(); + +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java new file mode 100644 index 00000000..7c0436eb --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java @@ -0,0 +1,647 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +import com.google.gson.JsonParseException; +import java.lang.reflect.InvocationTargetException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import lombok.Getter; +import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure; +import org.onap.policy.common.endpoints.event.comm.TopicListener; +import org.onap.policy.drools.controller.DroolsController; +import org.onap.policy.drools.pooling.message.BucketAssignments; +import org.onap.policy.drools.pooling.message.Leader; +import org.onap.policy.drools.pooling.message.Message; +import org.onap.policy.drools.pooling.message.Offline; +import org.onap.policy.drools.pooling.state.ActiveState; +import org.onap.policy.drools.pooling.state.IdleState; +import org.onap.policy.drools.pooling.state.InactiveState; +import org.onap.policy.drools.pooling.state.QueryState; +import org.onap.policy.drools.pooling.state.StartState; +import org.onap.policy.drools.pooling.state.State; +import org.onap.policy.drools.pooling.state.StateTimerTask; +import org.onap.policy.drools.protocol.coders.EventProtocolCoderConstants; +import org.onap.policy.drools.system.PolicyController; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Implementation of a {@link PoolingManager}. Until bucket assignments have been made, + * events coming from external topics are saved in a queue for later processing. Once + * assignments are made, the saved events are processed. In addition, while the controller + * is locked, events are still forwarded to other hosts and bucket assignments are still + * updated, based on any {@link Leader} messages that it receives. + */ +public class PoolingManagerImpl implements PoolingManager, TopicListener { + + private static final Logger logger = LoggerFactory.getLogger(PoolingManagerImpl.class); + + /** + * Maximum number of times a message can be forwarded. + */ + public static final int MAX_HOPS = 5; + + /** + * ID of this host. + */ + @Getter + private final String host; + + /** + * Properties with which this was configured. + */ + @Getter + private final PoolingProperties properties; + + /** + * Associated controller. + */ + private final PolicyController controller; + + /** + * Decremented each time the manager enters the Active state. Used by junit tests. + */ + private final CountDownLatch activeLatch; + + /** + * Used to encode & decode request objects received from & sent to a rule engine. + */ + private final Serializer serializer; + + /** + * Internal DMaaP topic used by this controller. + */ + @Getter + private final String topic; + + /** + * Manager for the internal DMaaP topic. + */ + private final TopicMessageManager topicMessageManager; + + /** + * Lock used while updating {@link #current}. In general, public methods must use + * this, while private methods assume the lock is already held. + */ + private final Object curLocker = new Object(); + + /** + * Current state. + * + * <p>This uses a finite state machine, wherein the state object contains all of the data + * relevant to that state. Each state object has a process() method, specific to each + * type of {@link Message} subclass. The method returns the next state object, or + * {@code null} if the state is to remain the same. + */ + private State current; + + /** + * Current bucket assignments or {@code null}. + */ + @Getter + private BucketAssignments assignments = null; + + /** + * Pool used to execute timers. + */ + private ScheduledThreadPoolExecutor scheduler = null; + + /** + * Constructs the manager, initializing all the data structures. + * + * @param host name/uuid of this host + * @param controller controller with which this is associated + * @param props feature properties specific to the controller + * @param activeLatch latch to be decremented each time the manager enters the Active + * state + */ + public PoolingManagerImpl(String host, PolicyController controller, PoolingProperties props, + CountDownLatch activeLatch) { + this.host = host; + this.controller = controller; + this.properties = props; + this.activeLatch = activeLatch; + + try { + this.serializer = new Serializer(); + this.topic = props.getPoolingTopic(); + this.topicMessageManager = makeTopicMessagesManager(props.getPoolingTopic()); + this.current = new IdleState(this); + + logger.info("allocating host {} to controller {} for topic {}", host, controller.getName(), topic); + + } catch (ClassCastException e) { + logger.error("not a topic listener, controller {}", controller.getName()); + throw new PoolingFeatureRtException(e); + + } catch (PoolingFeatureException e) { + logger.error("failed to attach internal DMaaP topic to controller {}", controller.getName()); + throw new PoolingFeatureRtException(e); + } + } + + /** + * Should only be used by junit tests. + * + * @return the current state + */ + protected State getCurrent() { + synchronized (curLocker) { + return current; + } + } + + /** + * Indicates that the controller is about to start. Starts the publisher for the + * internal topic, and creates a thread pool for the timers. + */ + public void beforeStart() { + synchronized (curLocker) { + if (scheduler == null) { + topicMessageManager.startPublisher(); + + logger.debug("make scheduler thread for topic {}", getTopic()); + scheduler = makeScheduler(); + + /* + * Only a handful of timers at any moment, thus we can afford to take the + * time to remove them when they're cancelled. + */ + scheduler.setRemoveOnCancelPolicy(true); + scheduler.setMaximumPoolSize(1); + scheduler.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); + scheduler.setContinueExistingPeriodicTasksAfterShutdownPolicy(false); + } + } + } + + /** + * Indicates that the controller has successfully started. Starts the consumer for the + * internal topic, enters the {@link StartState}, and sets the filter for the initial + * state. + */ + public void afterStart() { + synchronized (curLocker) { + if (current instanceof IdleState) { + topicMessageManager.startConsumer(this); + changeState(new StartState(this)); + } + } + } + + /** + * Indicates that the controller is about to stop. Stops the consumer, the scheduler, + * and the current state. + */ + public void beforeStop() { + ScheduledThreadPoolExecutor sched; + + synchronized (curLocker) { + sched = scheduler; + scheduler = null; + + if (!(current instanceof IdleState)) { + changeState(new IdleState(this)); + topicMessageManager.stopConsumer(this); + publishAdmin(new Offline(getHost())); + } + + assignments = null; + } + + if (sched != null) { + logger.debug("stop scheduler for topic {}", getTopic()); + sched.shutdownNow(); + } + } + + /** + * Indicates that the controller has stopped. Stops the publisher and logs a warning + * if any events are still in the queue. + */ + public void afterStop() { + synchronized (curLocker) { + /* + * stop the publisher, but allow time for any Offline message to be + * transmitted + */ + topicMessageManager.stopPublisher(properties.getOfflinePubWaitMs()); + } + } + + /** + * Indicates that the controller is about to be locked. Enters the idle state, as all + * it will be doing is forwarding messages. + */ + public void beforeLock() { + logger.info("locking manager for topic {}", getTopic()); + + synchronized (curLocker) { + changeState(new IdleState(this)); + } + } + + /** + * Indicates that the controller has been unlocked. Enters the start state, if the + * controller is running. + */ + public void afterUnlock() { + logger.info("unlocking manager for topic {}", getTopic()); + + synchronized (curLocker) { + if (controller.isAlive() && current instanceof IdleState && scheduler != null) { + changeState(new StartState(this)); + } + } + } + + /** + * Changes the finite state machine to a new state, provided the new state is not + * {@code null}. + * + * @param newState new state, or {@code null} if to remain unchanged + */ + private void changeState(State newState) { + if (newState != null) { + current.cancelTimers(); + current = newState; + + newState.start(); + } + } + + @Override + public CancellableScheduledTask schedule(long delayMs, StateTimerTask task) { + // wrap the task in a TimerAction and schedule it + ScheduledFuture<?> fut = scheduler.schedule(new TimerAction(task), delayMs, TimeUnit.MILLISECONDS); + + // wrap the future in a "CancellableScheduledTask" + return () -> fut.cancel(false); + } + + @Override + public CancellableScheduledTask scheduleWithFixedDelay(long initialDelayMs, long delayMs, StateTimerTask task) { + // wrap the task in a TimerAction and schedule it + ScheduledFuture<?> fut = scheduler.scheduleWithFixedDelay(new TimerAction(task), initialDelayMs, delayMs, + TimeUnit.MILLISECONDS); + + // wrap the future in a "CancellableScheduledTask" + return () -> fut.cancel(false); + } + + @Override + public void publishAdmin(Message msg) { + publish(Message.ADMIN, msg); + } + + @Override + public void publish(String channel, Message msg) { + logger.info("publish {} to {} on topic {}", msg.getClass().getSimpleName(), channel, getTopic()); + + msg.setChannel(channel); + + try { + // ensure it's valid before we send it + msg.checkValidity(); + + String txt = serializer.encodeMsg(msg); + topicMessageManager.publish(txt); + + } catch (JsonParseException e) { + logger.error("failed to serialize message for topic {} channel {}", topic, channel, e); + + } catch (PoolingFeatureException e) { + logger.error("failed to publish message for topic {} channel {}", topic, channel, e); + } + } + + /** + * Handles an event from the internal topic. + * + * @param commType comm infrastructure + * @param topic2 topic + * @param event event + */ + @Override + public void onTopicEvent(CommInfrastructure commType, String topic2, String event) { + + if (event == null) { + logger.error("null event on topic {}", topic); + return; + } + + synchronized (curLocker) { + // it's on the internal topic + handleInternal(event); + } + } + + /** + * Called by the PolicyController before it offers the event to the DroolsController. + * If the controller is locked, then it isn't processing events. However, they still + * need to be forwarded, thus in that case, they are decoded and forwarded. + * + * <p>On the other hand, if the controller is not locked, then we just return immediately + * and let {@link #beforeInsert(String, Object) beforeInsert()} handle + * it instead, as it already has the decoded message. + * + * @param topic2 topic + * @param event event + * @return {@code true} if the event was handled by the manager, {@code false} if it + * must still be handled by the invoker + */ + public boolean beforeOffer(String topic2, String event) { + + if (!controller.isLocked()) { + // we should NOT intercept this message - let the invoker handle it + return false; + } + + return handleExternal(topic2, decodeEvent(topic2, event)); + } + + /** + * Called by the DroolsController before it inserts the event into the rule engine. + * + * @param topic2 topic + * @param event event, as an object + * @return {@code true} if the event was handled by the manager, {@code false} if it + * must still be handled by the invoker + */ + public boolean beforeInsert(String topic2, Object event) { + return handleExternal(topic2, event); + } + + /** + * Handles an event from an external topic. + * + * @param topic2 topic + * @param event event, as an object, or {@code null} if it cannot be decoded + * @return {@code true} if the event was handled by the manager, {@code false} if it + * must still be handled by the invoker + */ + private boolean handleExternal(String topic2, Object event) { + if (event == null) { + // no event - let the invoker handle it + return false; + } + + synchronized (curLocker) { + return handleExternal(topic2, event, event.hashCode()); + } + } + + /** + * Handles an event from an external topic. + * + * @param topic2 topic + * @param event event, as an object + * @param eventHashCode event's hash code + * @return {@code true} if the event was handled, {@code false} if the invoker should + * handle it + */ + private boolean handleExternal(String topic2, Object event, int eventHashCode) { + if (assignments == null) { + // no bucket assignments yet - handle locally + logger.info("handle event locally for request {}", event); + + // we did NOT consume the event + return false; + + } else { + return handleEvent(topic2, event, eventHashCode); + } + } + + /** + * Handles a {@link Forward} event, possibly forwarding it again. + * + * @param topic2 topic + * @param event event, as an object + * @param eventHashCode event's hash code + * @return {@code true} if the event was handled, {@code false} if the invoker should + * handle it + */ + private boolean handleEvent(String topic2, Object event, int eventHashCode) { + String target = assignments.getAssignedHost(eventHashCode); + + if (target == null) { + /* + * This bucket has no assignment - just discard the event + */ + logger.warn("discarded event for unassigned bucket from topic {}", topic2); + return true; + } + + if (target.equals(host)) { + /* + * Message belongs to this host - allow the controller to handle it. + */ + logger.info("handle local event for request {} from topic {}", event, topic2); + return false; + } + + // not our message, consume the event + logger.warn("discarded event for host {} from topic {}", target, topic2); + return true; + } + + /** + * Decodes an event from a String into an event Object. + * + * @param topic2 topic + * @param event event + * @return the decoded event object, or {@code null} if it can't be decoded + */ + private Object decodeEvent(String topic2, String event) { + DroolsController drools = controller.getDrools(); + + // check if this topic has a decoder + + if (!canDecodeEvent(drools, topic2)) { + + logger.warn("{}: DECODING-UNSUPPORTED {}:{}:{}", drools, topic2, drools.getGroupId(), + drools.getArtifactId()); + return null; + } + + // decode + + try { + return decodeEventWrapper(drools, topic2, event); + + } catch (UnsupportedOperationException | IllegalStateException | IllegalArgumentException e) { + logger.debug("{}: DECODE FAILED: {} <- {} because of {}", drools, topic2, event, e.getMessage(), e); + return null; + } + } + + /** + * Handles an event from the internal topic. This uses reflection to identify the + * appropriate process() method to invoke, based on the type of Message that was + * decoded. + * + * @param event the serialized {@link Message} read from the internal topic + */ + private void handleInternal(String event) { + Class<?> clazz = null; + + try { + Message msg = serializer.decodeMsg(event); + + // get the class BEFORE checking the validity + clazz = msg.getClass(); + + msg.checkValidity(); + + var meth = current.getClass().getMethod("process", msg.getClass()); + changeState((State) meth.invoke(current, msg)); + + } catch (JsonParseException e) { + logger.warn("failed to decode message for topic {}", topic, e); + + } catch (NoSuchMethodException | SecurityException e) { + logger.error("no processor for message {} for topic {}", clazz, topic, e); + + } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException + | PoolingFeatureException e) { + logger.error("failed to process message {} for topic {}", clazz, topic, e); + } + } + + @Override + public void startDistributing(BucketAssignments asgn) { + synchronized (curLocker) { + int sz = (asgn == null ? 0 : asgn.getAllHosts().size()); + logger.info("new assignments for {} hosts on topic {}", sz, getTopic()); + assignments = asgn; + } + } + + @Override + public State goStart() { + return new StartState(this); + } + + @Override + public State goQuery() { + return new QueryState(this); + } + + @Override + public State goActive() { + activeLatch.countDown(); + return new ActiveState(this); + } + + @Override + public State goInactive() { + return new InactiveState(this); + } + + /** + * Action to run a timer task. Only runs the task if the machine is still in the state + * that it was in when the timer was created. + */ + private class TimerAction implements Runnable { + + /** + * State of the machine when the timer was created. + */ + private State origState; + + /** + * Task to be executed. + */ + private StateTimerTask task; + + /** + * Constructor. + * + * @param task task to execute when this timer runs + */ + public TimerAction(StateTimerTask task) { + this.origState = current; + this.task = task; + } + + @Override + public void run() { + synchronized (curLocker) { + if (current == origState) { + changeState(task.fire()); + } + } + } + } + + /** + * Creates a DMaaP manager. + * + * @param topic name of the internal DMaaP topic + * @return a new topic messages manager + * @throws PoolingFeatureException if an error occurs + */ + protected TopicMessageManager makeTopicMessagesManager(String topic) throws PoolingFeatureException { + return new TopicMessageManager(topic); + } + + /** + * Creates a scheduled thread pool. + * + * @return a new scheduled thread pool + */ + protected ScheduledThreadPoolExecutor makeScheduler() { + return new ScheduledThreadPoolExecutor(1); + } + + /** + * Determines if the event can be decoded. + * + * @param drools drools controller + * @param topic topic on which the event was received + * @return {@code true} if the event can be decoded, {@code false} otherwise + */ + protected boolean canDecodeEvent(DroolsController drools, String topic) { + return EventProtocolCoderConstants.getManager().isDecodingSupported(drools.getGroupId(), drools.getArtifactId(), + topic); + } + + /** + * Decodes the event. + * + * @param drools drools controller + * @param topic topic on which the event was received + * @param event event text to be decoded + * @return the decoded event + * @throws IllegalArgumentException illegal argument + * @throws UnsupportedOperationException unsupported operation + * @throws IllegalStateException illegal state + */ + protected Object decodeEventWrapper(DroolsController drools, String topic, String event) { + return EventProtocolCoderConstants.getManager().decode(drools.getGroupId(), drools.getArtifactId(), topic, + event); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingProperties.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingProperties.java new file mode 100644 index 00000000..fd1c3d3d --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/PoolingProperties.java @@ -0,0 +1,150 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +import java.util.Properties; +import lombok.Getter; +import lombok.Setter; +import org.onap.policy.common.utils.properties.BeanConfigurator; +import org.onap.policy.common.utils.properties.Property; +import org.onap.policy.common.utils.properties.SpecProperties; +import org.onap.policy.common.utils.properties.exception.PropertyException; + +/** + * Properties used by the pooling feature, specific to a controller. + */ +@Getter +@Setter +public class PoolingProperties { + + /** + * The feature name, used to retrieve properties. + */ + public static final String FEATURE_NAME = "feature-pooling-messages"; + + /** + * Feature properties all begin with this prefix. + */ + public static final String PREFIX = "pooling."; + + public static final String FEATURE_ENABLED = PREFIX + "enabled"; + public static final String POOLING_TOPIC = PREFIX + "topic"; + public static final String OFFLINE_LIMIT = PREFIX + "offline.queue.limit"; + public static final String OFFLINE_AGE_MS = PREFIX + "offline.queue.age.milliseconds"; + public static final String OFFLINE_PUB_WAIT_MS = PREFIX + "offline.publish.wait.milliseconds"; + public static final String START_HEARTBEAT_MS = PREFIX + "start.heartbeat.milliseconds"; + public static final String REACTIVATE_MS = PREFIX + "reactivate.milliseconds"; + public static final String IDENTIFICATION_MS = PREFIX + "identification.milliseconds"; + public static final String ACTIVE_HEARTBEAT_MS = PREFIX + "active.heartbeat.milliseconds"; + public static final String INTER_HEARTBEAT_MS = PREFIX + "inter.heartbeat.milliseconds"; + + /** + * Type of item that the extractors will be extracting. + */ + public static final String EXTRACTOR_TYPE = "requestId"; + + /** + * Prefix for extractor properties. + */ + public static final String PROP_EXTRACTOR_PREFIX = "extractor." + EXTRACTOR_TYPE; + + /** + * Properties from which this was constructed. + */ + private Properties source; + + /** + * Topic used for inter-host communication. + */ + @Property(name = POOLING_TOPIC) + private String poolingTopic; + + /** + * Maximum number of events to retain in the queue while waiting for + * buckets to be assigned. + */ + @Property(name = OFFLINE_LIMIT, defaultValue = "1000") + private int offlineLimit; + + /** + * Maximum age, in milliseconds, of events to be retained in the queue. + * Events older than this are discarded. + */ + @Property(name = OFFLINE_AGE_MS, defaultValue = "60000") + private long offlineAgeMs; + + /** + * Time, in milliseconds, to wait for an "Offline" message to be published + * to topic. + */ + @Property(name = OFFLINE_PUB_WAIT_MS, defaultValue = "3000") + private long offlinePubWaitMs; + + /** + * Time, in milliseconds, to wait for this host's heart beat during the + * start-up state. + */ + @Property(name = START_HEARTBEAT_MS, defaultValue = "100000") + private long startHeartbeatMs; + + /** + * Time, in milliseconds, to wait before attempting to reactivate this + * host when it has no bucket assignments. + */ + @Property(name = REACTIVATE_MS, defaultValue = "50000") + private long reactivateMs; + + /** + * Time, in milliseconds, to wait for all Identification messages to + * arrive during the query state. + */ + @Property(name = IDENTIFICATION_MS, defaultValue = "50000") + private long identificationMs; + + /** + * Time, in milliseconds, to wait for heart beats from this host, or its + * predecessor, during the active state. + */ + @Property(name = ACTIVE_HEARTBEAT_MS, defaultValue = "50000") + private long activeHeartbeatMs; + + /** + * Time, in milliseconds, to wait between heart beat generations during + * the active and start-up states. + */ + @Property(name = INTER_HEARTBEAT_MS, defaultValue = "15000") + private long interHeartbeatMs; + + /** + * Constructor. + * + * @param controllerName the name of the controller + * @param props set of properties used to configure this + * @throws PropertyException if an error occurs + * + */ + public PoolingProperties(String controllerName, Properties props) throws PropertyException { + source = props; + + new BeanConfigurator().configureFromProperties(this, new SpecProperties(PREFIX, controllerName, props)); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/Serializer.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/Serializer.java new file mode 100644 index 00000000..2894b1da --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/Serializer.java @@ -0,0 +1,124 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +import com.google.gson.Gson; +import com.google.gson.JsonElement; +import com.google.gson.JsonParseException; +import java.util.HashMap; +import java.util.Map; +import org.onap.policy.drools.pooling.message.Heartbeat; +import org.onap.policy.drools.pooling.message.Identification; +import org.onap.policy.drools.pooling.message.Leader; +import org.onap.policy.drools.pooling.message.Message; +import org.onap.policy.drools.pooling.message.Offline; +import org.onap.policy.drools.pooling.message.Query; + +/** + * Serialization helper functions. + */ +public class Serializer { + + /** + * The message type is stored in fields of this name within the JSON. + */ + private static final String TYPE_FIELD = "type"; + + /** + * Used to encode & decode JSON messages sent & received, respectively, on the + * internal topic. + */ + private final Gson gson = new Gson(); + + /** + * Maps a message subclass to its type. + */ + private static final Map<Class<? extends Message>, String> class2type = new HashMap<>(); + + /** + * Maps a message type to the appropriate subclass. + */ + private static final Map<String, Class<? extends Message>> type2class = new HashMap<>(); + + static { + class2type.put(Heartbeat.class, "heartbeat"); + class2type.put(Identification.class, "identification"); + class2type.put(Leader.class, "leader"); + class2type.put(Offline.class, "offline"); + class2type.put(Query.class, "query"); + + class2type.forEach((clazz, type) -> type2class.put(type, clazz)); + } + + /** + * Encodes a filter. + * + * @param filter filter to be encoded + * @return the filter, serialized as a JSON string + */ + public String encodeFilter(Map<String, Object> filter) { + return gson.toJson(filter); + } + + /** + * Encodes a message. + * + * @param msg message to be encoded + * @return the message, serialized as a JSON string + */ + public String encodeMsg(Message msg) { + JsonElement jsonEl = gson.toJsonTree(msg); + + String type = class2type.get(msg.getClass()); + if (type == null) { + throw new JsonParseException("cannot serialize " + msg.getClass()); + } + + jsonEl.getAsJsonObject().addProperty(TYPE_FIELD, type); + + return gson.toJson(jsonEl); + } + + /** + * Decodes a JSON string into a Message. + * + * @param msg JSON string representing the message + * @return the message + */ + public Message decodeMsg(String msg) { + JsonElement jsonEl = gson.fromJson(msg, JsonElement.class); + + JsonElement typeEl = jsonEl.getAsJsonObject().get(TYPE_FIELD); + if (typeEl == null) { + throw new JsonParseException("cannot deserialize " + Message.class + + " because it does not contain a field named " + TYPE_FIELD); + + } + + Class<? extends Message> clazz = type2class.get(typeEl.getAsString()); + if (clazz == null) { + throw new JsonParseException("cannot deserialize " + typeEl); + } + + return gson.fromJson(jsonEl, clazz); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/TopicMessageManager.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/TopicMessageManager.java new file mode 100644 index 00000000..4c9b3f34 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/TopicMessageManager.java @@ -0,0 +1,233 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling; + +import java.util.List; +import lombok.Getter; +import org.onap.policy.common.endpoints.event.comm.TopicEndpoint; +import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager; +import org.onap.policy.common.endpoints.event.comm.TopicListener; +import org.onap.policy.common.endpoints.event.comm.TopicSink; +import org.onap.policy.common.endpoints.event.comm.TopicSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Manages the internal topic. Assumes all topics are managed by + * {@link TopicEndpoint}. + */ +public class TopicMessageManager { + + private static final Logger logger = LoggerFactory.getLogger(TopicMessageManager.class); + + /** + * Name of the topic. + */ + @Getter + private final String topic; + + /** + * Topic source whose filter is to be manipulated. + */ + private final TopicSource topicSource; + + /** + * Where to publish messages. + */ + private final TopicSink topicSink; + + /** + * {@code True} if the consumer is running, {@code false} otherwise. + */ + private boolean consuming = false; + + /** + * {@code True} if the publisher is running, {@code false} otherwise. + */ + private boolean publishing = false; + + /** + * Constructs the manager, but does not start the source or sink. + * + * @param topic name of the internal topic + * @throws PoolingFeatureException if an error occurs + */ + public TopicMessageManager(String topic) throws PoolingFeatureException { + + logger.info("initializing bus for topic {}", topic); + + try { + this.topic = topic; + + this.topicSource = findTopicSource(); + this.topicSink = findTopicSink(); + + } catch (IllegalArgumentException e) { + logger.error("failed to attach to topic {}", topic); + throw new PoolingFeatureException(e); + } + } + + /** + * Finds the topic source associated with the internal topic. + * + * @return the topic source + * @throws PoolingFeatureException if the source doesn't exist or is not filterable + */ + private TopicSource findTopicSource() throws PoolingFeatureException { + for (TopicSource src : getTopicSources()) { + if (topic.equals(src.getTopic())) { + return src; + } + } + + throw new PoolingFeatureException("missing topic source " + topic); + } + + /** + * Finds the topic sink associated with the internal topic. + * + * @return the topic sink + * @throws PoolingFeatureException if the sink doesn't exist + */ + private TopicSink findTopicSink() throws PoolingFeatureException { + for (TopicSink sink : getTopicSinks()) { + if (topic.equals(sink.getTopic())) { + return sink; + } + } + + throw new PoolingFeatureException("missing topic sink " + topic); + } + + /** + * Starts the publisher, if it isn't already running. + */ + public void startPublisher() { + if (publishing) { + return; + } + + logger.info("start publishing to topic {}", topic); + publishing = true; + } + + /** + * Stops the publisher. + * + * @param waitMs time, in milliseconds, to wait for the sink to transmit any queued messages and + * close + */ + public void stopPublisher(long waitMs) { + if (!publishing) { + return; + } + + /* + * Give the sink a chance to transmit messages in the queue. It would be better if "waitMs" + * could be passed to sink.stop(), but that isn't an option at this time. + */ + try { + Thread.sleep(waitMs); + + } catch (InterruptedException e) { + logger.warn("message transmission stopped due to {}", e.getMessage()); + Thread.currentThread().interrupt(); + } + + logger.info("stop publishing to topic {}", topic); + publishing = false; + } + + /** + * Starts the consumer, if it isn't already running. + * + * @param listener listener to register with the source + */ + public void startConsumer(TopicListener listener) { + if (consuming) { + return; + } + + logger.info("start consuming from topic {}", topic); + topicSource.register(listener); + consuming = true; + } + + /** + * Stops the consumer. + * + * @param listener listener to unregister with the source + */ + public void stopConsumer(TopicListener listener) { + if (!consuming) { + return; + } + + logger.info("stop consuming from topic {}", topic); + consuming = false; + topicSource.unregister(listener); + } + + /** + * Publishes a message to the sink. + * + * @param msg message to be published + * @throws PoolingFeatureException if an error occurs or the publisher isn't running + */ + public void publish(String msg) throws PoolingFeatureException { + if (!publishing) { + throw new PoolingFeatureException(new IllegalStateException("no topic sink " + topic)); + } + + try { + if (!topicSink.send(msg)) { + throw new PoolingFeatureException("failed to send to topic sink " + topic); + } + + } catch (IllegalStateException e) { + throw new PoolingFeatureException("cannot send to topic sink " + topic, e); + } + } + + /* + * The remaining methods may be overridden by junit tests. + */ + + /** + * Get topic source. + * + * @return the topic sources + */ + protected List<TopicSource> getTopicSources() { + return TopicEndpointManager.getManager().getTopicSources(); + } + + /** + * Get topic sinks. + * + * @return the topic sinks + */ + protected List<TopicSink> getTopicSinks() { + return TopicEndpointManager.getManager().getTopicSinks(); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/BucketAssignments.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/BucketAssignments.java new file mode 100644 index 00000000..485a9d2e --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/BucketAssignments.java @@ -0,0 +1,205 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.message; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import org.onap.policy.drools.pooling.PoolingFeatureException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Bucket assignments, which is simply an array of host names. + */ +@Getter +@Setter +@NoArgsConstructor +public class BucketAssignments { + + private static final Logger logger = LoggerFactory.getLogger(BucketAssignments.class); + + /** + * The number of bits in the maximum number of buckets. + */ + private static final int MAX_BUCKET_BITS = 10; + + /** + * Maximum number of buckets. Must be a power of two. + */ + public static final int MAX_BUCKETS = 1 << MAX_BUCKET_BITS; + + /** + * Used to ensure that a hash code is not negative. + */ + private static final int MAX_BUCKETS_MASK = MAX_BUCKETS - 1; + + /** + * Identifies the host serving a particular bucket. + */ + private String[] hostArray = null; + + + /** + * Constructor. + * + * @param hostArray maps a bucket number (i.e., array index) to a host. All values + * must be non-null + */ + public BucketAssignments(String[] hostArray) { + this.hostArray = hostArray; + } + + /** + * Gets the leader, which is the host with the minimum UUID. + * + * @return the assignment leader + */ + public String getLeader() { + if (hostArray == null) { + return null; + } + + String leader = null; + + for (String host : hostArray) { + if (host != null && (leader == null || host.compareTo(leader) < 0)) { + leader = host; + } + } + + return leader; + + } + + /** + * Determines if a host has an assignment. + * + * @param host host to be checked + * @return {@code true} if the host has an assignment, {@code false} otherwise + */ + public boolean hasAssignment(String host) { + if (hostArray == null) { + return false; + } + + for (String host2 : hostArray) { + if (host.equals(host2)) { + return true; + } + } + + return false; + } + + /** + * Gets all the hosts that have an assignment. + * + * @return all the hosts that have an assignment + */ + public Set<String> getAllHosts() { + Set<String> set = new HashSet<>(); + if (hostArray == null) { + return set; + } + + for (String host : hostArray) { + if (host != null) { + set.add(host); + } + } + + return set; + } + + /** + * Gets the host assigned to a given bucket. + * + * @param hashCode hash code of the item whose assignment is desired + * @return the assigned host, or {@code null} if the item has no assigned host + */ + public String getAssignedHost(int hashCode) { + if (hostArray == null || hostArray.length == 0) { + logger.error("no buckets have been assigned"); + return null; + } + + return hostArray[(Math.abs(hashCode) & MAX_BUCKETS_MASK) % hostArray.length]; + } + + /** + * Gets the number of buckets. + * + * @return the number of buckets + */ + public int size() { + return (hostArray != null ? hostArray.length : 0); + } + + /** + * Checks the validity of the assignments, verifying that all buckets have been + * assigned to a host. + * + * @throws PoolingFeatureException if the assignments are invalid + */ + public void checkValidity() throws PoolingFeatureException { + if (hostArray == null || hostArray.length == 0) { + throw new PoolingFeatureException("missing hosts in message bucket assignments"); + } + + if (hostArray.length > MAX_BUCKETS) { + throw new PoolingFeatureException("too many hosts in message bucket assignments"); + } + + for (var x = 0; x < hostArray.length; ++x) { + if (hostArray[x] == null) { + throw new PoolingFeatureException("bucket " + x + " has no assignment"); + } + } + } + + @Override + public int hashCode() { + final var prime = 31; + var result = 1; + result = prime * result + Arrays.hashCode(hostArray); + return result; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + BucketAssignments other = (BucketAssignments) obj; + return Arrays.equals(hostArray, other.hostArray); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Heartbeat.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Heartbeat.java new file mode 100644 index 00000000..4a8bdc3b --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Heartbeat.java @@ -0,0 +1,52 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.message; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +/** + * Heart beat message sent to self, or to the succeeding host. + */ +@Getter +@Setter +@NoArgsConstructor +public class Heartbeat extends Message { + + /** + * Time, in milliseconds, when this was created. + */ + private long timestampMs; + + /** + * Constructor. + * + * @param source host on which the message originated + * @param timestampMs time, in milliseconds, associated with the message + */ + public Heartbeat(String source, long timestampMs) { + super(source); + + this.timestampMs = timestampMs; + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Identification.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Identification.java new file mode 100644 index 00000000..b8fd8414 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Identification.java @@ -0,0 +1,41 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.message; + +import lombok.NoArgsConstructor; + +/** + * Identifies the source host and the bucket assignments which it knows about. + */ +@NoArgsConstructor +public class Identification extends MessageWithAssignments { + + /** + * Constructor. + * + * @param source host on which the message originated + * @param assignments assignments + */ + public Identification(String source, BucketAssignments assignments) { + super(source, assignments); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Leader.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Leader.java new file mode 100644 index 00000000..10c33382 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Leader.java @@ -0,0 +1,70 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2019, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.message; + +import lombok.NoArgsConstructor; +import org.onap.policy.drools.pooling.PoolingFeatureException; + +/** + * Indicates that the "source" of this message is now the "lead" host. + */ +@NoArgsConstructor +public class Leader extends MessageWithAssignments { + + /** + * Constructor. + * + * @param source host on which the message originated + * @param assignments assignments + */ + public Leader(String source, BucketAssignments assignments) { + super(source, assignments); + } + + /** + * Also verifies that buckets have been assigned and that the source is + * indeed the leader. + */ + @Override + public void checkValidity() throws PoolingFeatureException { + + super.checkValidity(); + + BucketAssignments assignments = getAssignments(); + if (assignments == null) { + throw new PoolingFeatureException("missing message bucket assignments"); + } + + String leader = getSource(); + + if (!assignments.hasAssignment(leader)) { + throw new PoolingFeatureException("leader " + leader + " has no bucket assignments"); + } + + for (String host : assignments.getHostArray()) { + if (host.compareTo(leader) < 0) { + throw new PoolingFeatureException("invalid leader " + leader + ", should be " + host); + } + } + } + +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Message.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Message.java new file mode 100644 index 00000000..bb973142 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Message.java @@ -0,0 +1,79 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2019, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.message; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import org.onap.policy.drools.pooling.PoolingFeatureException; + +/** + * Messages sent on the internal topic. + */ +@Getter +@Setter +@NoArgsConstructor +public class Message { + + /** + * Name of the administrative channel. + */ + public static final String ADMIN = "_admin"; + + /** + * Host that originated the message. + */ + private String source; + + /** + * Channel on which the message is routed, which is either the target host + * or {@link #ADMIN}. + */ + private String channel; + + + /** + * Constructor. + * + * @param source host on which the message originated + */ + public Message(String source) { + this.source = source; + } + + /** + * Checks the validity of the message, including verifying that required + * fields are not missing. + * + * @throws PoolingFeatureException if the message is invalid + */ + public void checkValidity() throws PoolingFeatureException { + if (source == null || source.isEmpty()) { + throw new PoolingFeatureException("missing message source"); + } + + if (channel == null || channel.isEmpty()) { + throw new PoolingFeatureException("missing message channel"); + } + } + +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/MessageWithAssignments.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/MessageWithAssignments.java new file mode 100644 index 00000000..b8521dd8 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/MessageWithAssignments.java @@ -0,0 +1,68 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2019, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.message; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import org.onap.policy.drools.pooling.PoolingFeatureException; + +/** + * A Message that includes bucket assignments. + */ +@Setter +@Getter +@NoArgsConstructor +public class MessageWithAssignments extends Message { + + /** + * Bucket assignments, as known by the source host. + */ + private BucketAssignments assignments; + + + /** + * Constructor. + * + * @param source host on which the message originated + * @param assignments assignments + */ + public MessageWithAssignments(String source, BucketAssignments assignments) { + super(source); + + this.assignments = assignments; + } + + /** + * If there are any assignments, it verifies there validity. + */ + @Override + public void checkValidity() throws PoolingFeatureException { + + super.checkValidity(); + + if (assignments != null) { + assignments.checkValidity(); + } + } + +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Offline.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Offline.java new file mode 100644 index 00000000..3ff7bf1e --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Offline.java @@ -0,0 +1,41 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.message; + +import lombok.NoArgsConstructor; + +/** + * Indicates that the source host is going offline and will be unable to process + * any further requests. + */ +@NoArgsConstructor +public class Offline extends Message { + + /** + * Constructor. + * + * @param source host on which the message originated + */ + public Offline(String source) { + super(source); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Query.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Query.java new file mode 100644 index 00000000..4c856bb7 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/message/Query.java @@ -0,0 +1,40 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.message; + +import lombok.NoArgsConstructor; + +/** + * Query the other hosts for their identification. + */ +@NoArgsConstructor +public class Query extends Message { + + /** + * Constructor. + * + * @param source host on which the message originated + */ + public Query(String source) { + super(source); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/ActiveState.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/ActiveState.java new file mode 100644 index 00000000..cafcb45b --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/ActiveState.java @@ -0,0 +1,270 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2020-2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.state; + +import java.util.Arrays; +import java.util.TreeSet; +import lombok.AccessLevel; +import lombok.Getter; +import org.onap.policy.drools.pooling.PoolingManager; +import org.onap.policy.drools.pooling.message.Heartbeat; +import org.onap.policy.drools.pooling.message.Leader; +import org.onap.policy.drools.pooling.message.Offline; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The active state. In this state, this host has one more bucket assignments and + * processes any events associated with one of its buckets. Other events are forwarded to + * appropriate target hosts. + */ +@Getter(AccessLevel.PROTECTED) +public class ActiveState extends ProcessingState { + + private static final Logger logger = LoggerFactory.getLogger(ActiveState.class); + + /** + * Set of hosts that have been assigned a bucket. + */ + @Getter(AccessLevel.NONE) + private final TreeSet<String> assigned = new TreeSet<>(); + + /** + * Host that comes after this host, or {@code null} if it has no successor. + */ + private String succHost = null; + + /** + * Host that comes before this host, or "" if it has no predecessor. + */ + private String predHost = ""; + + /** + * {@code True} if we saw this host's heart beat since the last check, {@code false} + * otherwise. + */ + private boolean myHeartbeatSeen = false; + + /** + * {@code True} if we saw the predecessor's heart beat since the last check, + * {@code false} otherwise. + */ + private boolean predHeartbeatSeen = false; + + + /** + * Constructor. + * + * @param mgr pooling manager + */ + public ActiveState(PoolingManager mgr) { + super(mgr, mgr.getAssignments().getLeader()); + + assigned.addAll(Arrays.asList(mgr.getAssignments().getHostArray())); + + detmNeighbors(); + } + + /** + * Determine this host's neighbors based on the order of the host UUIDs. Updates + * {@link #succHost} and {@link #predHost}. + */ + private void detmNeighbors() { + if (assigned.size() < 2) { + logger.info("this host has no neighbors on topic {}", getTopic()); + /* + * this host is the only one with any assignments - it has no neighbors + */ + succHost = null; + predHost = ""; + return; + } + + if ((succHost = assigned.higher(getHost())) == null) { + // wrapped around - successor is the first host in the set + succHost = assigned.first(); + } + logger.info("this host's successor is {} on topic {}", succHost, getTopic()); + + if ((predHost = assigned.lower(getHost())) == null) { + // wrapped around - predecessor is the last host in the set + predHost = assigned.last(); + } + logger.info("this host's predecessor is {} on topic {}", predHost, getTopic()); + } + + @Override + public void start() { + super.start(); + addTimers(); + genHeartbeat(); + } + + /** + * Adds the timers. + */ + private void addTimers() { + logger.info("add timers"); + + /* + * heart beat generator + */ + long genMs = getProperties().getInterHeartbeatMs(); + + scheduleWithFixedDelay(genMs, genMs, () -> { + genHeartbeat(); + return null; + }); + + /* + * my heart beat checker + */ + long waitMs = getProperties().getActiveHeartbeatMs(); + + scheduleWithFixedDelay(waitMs, waitMs, () -> { + if (myHeartbeatSeen) { + myHeartbeatSeen = false; + return null; + } + + // missed my heart beat + logger.error("missed my heartbeat on topic {}", getTopic()); + + return missedHeartbeat(); + }); + + /* + * predecessor heart beat checker + */ + if (!predHost.isEmpty()) { + + scheduleWithFixedDelay(waitMs, waitMs, () -> { + if (predHeartbeatSeen) { + predHeartbeatSeen = false; + return null; + } + + // missed the predecessor's heart beat + logger.warn("missed predecessor's heartbeat on topic {}", getTopic()); + + publish(makeQuery()); + + return goQuery(); + }); + } + } + + /** + * Generates a heart beat for this host and its successor. + */ + private void genHeartbeat() { + var msg = makeHeartbeat(System.currentTimeMillis()); + publish(getHost(), msg); + + if (succHost != null) { + publish(succHost, msg); + } + } + + @Override + public State process(Heartbeat msg) { + String src = msg.getSource(); + + if (src == null) { + logger.warn("Heartbeat message has no source on topic {}", getTopic()); + + } else if (src.equals(getHost())) { + logger.info("saw my heartbeat on topic {}", getTopic()); + myHeartbeatSeen = true; + + } else if (src.equals(predHost)) { + logger.info("saw heartbeat from {} on topic {}", src, getTopic()); + predHeartbeatSeen = true; + + } else { + logger.info("ignored heartbeat message from {} on topic {}", src, getTopic()); + } + + return null; + } + + @Override + public State process(Leader msg) { + if (!isValid(msg)) { + return null; + } + + String src = msg.getSource(); + + if (getHost().compareTo(src) < 0) { + // our host would be a better leader - find out what's up + logger.warn("unexpected Leader message from {} on topic {}", src, getTopic()); + return goQuery(); + } + + logger.info("have a new leader {} on topic {}", src, getTopic()); + + return goActive(msg.getAssignments()); + } + + @Override + public State process(Offline msg) { + String src = msg.getSource(); + + if (src == null) { + logger.warn("Offline message has no source on topic {}", getTopic()); + return null; + + } else if (!assigned.contains(src)) { + /* + * the offline host wasn't assigned any buckets, so just ignore the message + */ + logger.info("ignore Offline message from unassigned source {} on topic {}", src, getTopic()); + return null; + + } else if (isLeader() || (predHost.equals(src) && predHost.equals(assigned.first()))) { + /* + * Case 1: We are the leader. + * + * Case 2: Our predecessor was the leader, and it has gone offline - we should + * become the leader. + * + * In either case, we are now the leader, and we must re-balance the buckets + * since one of the hosts has gone offline. + */ + + logger.info("Offline message from source {} on topic {}", src, getTopic()); + + assigned.remove(src); + + return becomeLeader(assigned); + + } else { + /* + * Otherwise, we don't care right now - we'll wait for the leader to tell us + * it's been removed. + */ + logger.info("ignore Offline message from source {} on topic {}", src, getTopic()); + return null; + } + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/IdleState.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/IdleState.java new file mode 100644 index 00000000..06418280 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/IdleState.java @@ -0,0 +1,34 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.state; + +import org.onap.policy.drools.pooling.PoolingManager; + +/** + * Idle state, used when offline. + */ +public class IdleState extends State { + + public IdleState(PoolingManager mgr) { + super(mgr); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/InactiveState.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/InactiveState.java new file mode 100644 index 00000000..7fc220a0 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/InactiveState.java @@ -0,0 +1,81 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.state; + +import org.onap.policy.drools.pooling.PoolingManager; +import org.onap.policy.drools.pooling.message.Leader; +import org.onap.policy.drools.pooling.message.Query; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The inactive state. In this state, we just wait a bit and then try to re-activate. In + * the meantime, all messages are ignored. + */ +public class InactiveState extends State { + + private static final Logger logger = LoggerFactory.getLogger(InactiveState.class); + + /** + * Constructor. + * + * @param mgr pooling manager + */ + public InactiveState(PoolingManager mgr) { + super(mgr); + } + + @Override + public void start() { + super.start(); + schedule(getProperties().getReactivateMs(), this::goStart); + } + + @Override + public State process(Leader msg) { + if (isValid(msg)) { + logger.info("received Leader message from {} on topic {}", msg.getSource(), getTopic()); + return goActive(msg.getAssignments()); + } + + return null; + } + + /** + * Generates an Identification message and goes to the query state. + */ + @Override + public State process(Query msg) { + logger.info("received Query message on topic {}", getTopic()); + publish(makeIdentification()); + return goQuery(); + } + + /** + * Remains in this state, without resetting any timers. + */ + @Override + protected State goInactive() { + return null; + } + +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/ProcessingState.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/ProcessingState.java new file mode 100644 index 00000000..76914b75 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/ProcessingState.java @@ -0,0 +1,398 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.state; + +import java.util.Collection; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; +import lombok.Getter; +import lombok.NonNull; +import lombok.Setter; +import org.onap.policy.drools.pooling.PoolingManager; +import org.onap.policy.drools.pooling.message.BucketAssignments; +import org.onap.policy.drools.pooling.message.Leader; +import org.onap.policy.drools.pooling.message.Query; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Any state in which events are being processed locally and forwarded, as appropriate. + */ +@Setter +@Getter +public class ProcessingState extends State { + + private static final Logger logger = LoggerFactory.getLogger(ProcessingState.class); + + /** + * Current known leader, never {@code null}. + */ + @NonNull + private String leader; + + /** + * Constructor. + * + * @param mgr pooling manager + * @param leader current known leader, which need not be the same as the assignment + * leader. Never {@code null} + * @throws IllegalArgumentException if an argument is invalid + */ + public ProcessingState(PoolingManager mgr, @NonNull String leader) { + super(mgr); + + BucketAssignments assignments = mgr.getAssignments(); + + if (assignments != null) { + String[] arr = assignments.getHostArray(); + if (arr != null && arr.length == 0) { + throw new IllegalArgumentException("zero-length bucket assignments"); + } + } + + this.leader = leader; + } + + /** + * Generates an Identification message and goes to the query state. + */ + @Override + public State process(Query msg) { + logger.info("received Query message on topic {}", getTopic()); + publish(makeIdentification()); + return goQuery(); + } + + /** + * Sets the assignments. + * + * @param assignments new assignments, or {@code null} + */ + protected final void setAssignments(BucketAssignments assignments) { + if (assignments != null) { + startDistributing(assignments); + } + } + + /** + * Determines if this host is the leader, based on the current assignments. + * + * @return {@code true} if this host is the leader, {@code false} otherwise + */ + public boolean isLeader() { + return getHost().equals(leader); + } + + /** + * Becomes the leader. Publishes a Leader message and enters the {@link ActiveState}. + * + * @param alive hosts that are known to be alive + * + * @return the new state + */ + protected State becomeLeader(SortedSet<String> alive) { + String newLeader = getHost(); + + if (!newLeader.equals(alive.first())) { + throw new IllegalArgumentException(newLeader + " cannot replace " + alive.first()); + } + + var msg = makeLeader(alive); + logger.info("{}/{} hosts have an assignment", msg.getAssignments().getAllHosts().size(), alive.size()); + + publish(msg); + + return goActive(msg.getAssignments()); + } + + /** + * Makes a leader message. Assumes "this" host is the leader, and thus appears as the + * first host in the set of hosts that are still alive. + * + * @param alive hosts that are known to be alive + * + * @return a new message + */ + private Leader makeLeader(Set<String> alive) { + return new Leader(getHost(), makeAssignments(alive)); + } + + /** + * Makes a set of bucket assignments. Assumes "this" host is the leader. + * + * @param alive hosts that are known to be alive + * + * @return a new set of bucket assignments + */ + private BucketAssignments makeAssignments(Set<String> alive) { + + // make a working array from the CURRENT assignments + String[] bucket2host = makeBucketArray(); + + TreeSet<String> avail = new TreeSet<>(alive); + + // if we have more hosts than buckets, then remove the extra hosts + removeExcessHosts(bucket2host.length, avail); + + // create a host bucket for each available host + Map<String, HostBucket> host2hb = new HashMap<>(); + avail.forEach(host -> host2hb.put(host, new HostBucket(host))); + + // add bucket indices to the appropriate host bucket + addIndicesToHostBuckets(bucket2host, host2hb); + + // convert the collection back to an array + fillArray(host2hb.values(), bucket2host); + + // update bucket2host with new assignments + rebalanceBuckets(host2hb.values(), bucket2host); + + return new BucketAssignments(bucket2host); + } + + /** + * Makes a bucket array, copying the current assignments, if available. + * + * @return a new bucket array + */ + private String[] makeBucketArray() { + BucketAssignments asgn = getAssignments(); + if (asgn == null) { + return new String[BucketAssignments.MAX_BUCKETS]; + } + + String[] oldArray = asgn.getHostArray(); + if (oldArray.length == 0) { + return new String[BucketAssignments.MAX_BUCKETS]; + } + + var newArray = new String[oldArray.length]; + System.arraycopy(oldArray, 0, newArray, 0, oldArray.length); + + return newArray; + } + + /** + * Removes excess hosts from the set of available hosts. Assumes "this" host is the + * leader, and thus appears as the first host in the set. + * + * @param maxHosts maximum number of hosts to be retained + * @param avail available hosts + */ + private void removeExcessHosts(int maxHosts, SortedSet<String> avail) { + while (avail.size() > maxHosts) { + /* + * Don't remove this host, as it's the leader. Since the leader is always at + * the front of the sorted set, we'll just pick off hosts from the back of the + * set. + */ + String host = avail.last(); + avail.remove(host); + + logger.warn("not using extra host {} for topic {}", host, getTopic()); + } + } + + /** + * Adds bucket indices to {@link HostBucket} objects. Buckets that are unassigned or + * assigned to a host that does not appear within the map are re-assigned to a host + * that appears within the map. + * + * @param bucket2host bucket assignments + * @param host2data maps a host name to its {@link HostBucket} + */ + private void addIndicesToHostBuckets(String[] bucket2host, Map<String, HostBucket> host2data) { + LinkedList<Integer> nullBuckets = new LinkedList<>(); + + for (var x = 0; x < bucket2host.length; ++x) { + String host = bucket2host[x]; + if (host == null) { + nullBuckets.add(x); + + } else { + HostBucket hb = host2data.get(host); + if (hb == null) { + nullBuckets.add(x); + + } else { + hb.add(x); + } + } + } + + // assign the null buckets to other hosts + assignNullBuckets(nullBuckets, host2data.values()); + } + + /** + * Assigns null buckets (i.e., those having no assignment) to available hosts. + * + * @param buckets buckets that still need to be assigned to hosts + * @param coll collection of current host-bucket assignments + */ + private void assignNullBuckets(Queue<Integer> buckets, Collection<HostBucket> coll) { + // assign null buckets to the hosts with the fewest buckets + TreeSet<HostBucket> assignments = new TreeSet<>(coll); + + for (Integer index : buckets) { + // add it to the host with the shortest bucket list + HostBucket newhb = assignments.pollFirst(); + assert newhb != null; + newhb.add(index); + + // put the item back into the queue, with its new count + assignments.add(newhb); + } + } + + /** + * Re-balances the buckets, taking from those that have a larger count and giving to + * those that have a smaller count. Populates an output array with the new + * assignments. + * + * @param coll current bucket assignment + * @param bucket2host array to be populated with the new assignments + */ + private void rebalanceBuckets(Collection<HostBucket> coll, String[] bucket2host) { + if (coll.size() <= 1) { + // only one hosts - nothing to rebalance + return; + } + + TreeSet<HostBucket> assignments = new TreeSet<>(coll); + + for (;;) { + HostBucket smaller = assignments.pollFirst(); + HostBucket larger = assignments.pollLast(); + + assert larger != null && smaller != null; + if (larger.size() - smaller.size() <= 1) { + // it's as balanced as it will get + break; + } + + // move the bucket from the larger to the smaller + Integer bucket = larger.remove(); + smaller.add(bucket); + + bucket2host[bucket] = smaller.host; + + // put the items back, with their new counts + assignments.add(larger); + assignments.add(smaller); + } + + } + + /** + * Fills the array with the host assignments. + * + * @param coll the host assignments + * @param bucket2host array to be filled + */ + private void fillArray(Collection<HostBucket> coll, String[] bucket2host) { + for (HostBucket hb : coll) { + for (Integer index : hb.buckets) { + bucket2host[index] = hb.host; + } + } + } + + /** + * Tracks buckets that have been assigned to a host. + */ + protected static class HostBucket implements Comparable<HostBucket> { + /** + * Host to which the buckets have been assigned. + */ + private String host; + + /** + * Buckets that have been assigned to this host. + */ + private Queue<Integer> buckets = new LinkedList<>(); + + /** + * Constructor. + * + * @param host host + */ + public HostBucket(String host) { + this.host = host; + } + + /** + * Removes the next bucket from the list. + * + * @return the next bucket + */ + public final Integer remove() { + return buckets.remove(); + } + + /** + * Adds a bucket to the list. + * + * @param index index of the bucket to add + */ + public final void add(Integer index) { + buckets.add(index); + } + + /** + * Size. + * + * @return the number of buckets assigned to this host + */ + public final int size() { + return buckets.size(); + } + + /** + * Compares host buckets, first by the number of buckets, and then by the host + * name. + */ + @Override + public final int compareTo(HostBucket other) { + int diff = buckets.size() - other.buckets.size(); + if (diff == 0) { + diff = host.compareTo(other.host); + } + return diff; + } + + @Override + public final int hashCode() { + throw new UnsupportedOperationException("HostBucket cannot be hashed"); + } + + @Override + public final boolean equals(Object obj) { + throw new UnsupportedOperationException("cannot compare HostBuckets"); + } + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/QueryState.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/QueryState.java new file mode 100644 index 00000000..ef401dcb --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/QueryState.java @@ -0,0 +1,204 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.state; + +import java.util.TreeSet; +import org.onap.policy.drools.pooling.PoolingManager; +import org.onap.policy.drools.pooling.message.BucketAssignments; +import org.onap.policy.drools.pooling.message.Identification; +import org.onap.policy.drools.pooling.message.Leader; +import org.onap.policy.drools.pooling.message.Offline; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The Query state. In this state, the host waits for the other hosts to identify + * themselves. Eventually, a leader should come forth. If not, it will transition to the + * active or inactive state, depending on whether it has an assignment in the + * current bucket assignments. The other possibility is that it may <i>become</i> the + * leader, in which case it will also transition to the active state. + */ +public class QueryState extends ProcessingState { + + private static final Logger logger = LoggerFactory.getLogger(QueryState.class); + + /** + * Hosts that have sent an "Identification" message. Always includes this host. + */ + private final TreeSet<String> alive = new TreeSet<>(); + + /** + * {@code True} if we saw our own Identification method, {@code false} otherwise. + */ + private boolean sawSelfIdent = false; + + /** + * Constructor. + * + * @param mgr manager + */ + public QueryState(PoolingManager mgr) { + // this host is the leader, until a better candidate identifies itself + super(mgr, mgr.getHost()); + + alive.add(getHost()); + } + + @Override + public void start() { + super.start(); + + // start identification timer + awaitIdentification(); + } + + /** + * Starts a timer to wait for all Identification messages to arrive. + */ + private void awaitIdentification() { + + /* + * Once we've waited long enough for all Identification messages to arrive, become + * the leader, assuming we should. + */ + + schedule(getProperties().getIdentificationMs(), () -> { + + if (!sawSelfIdent) { + // didn't see our identification + logger.error("missed our own Ident message on topic {}", getTopic()); + return missedHeartbeat(); + + } else if (isLeader()) { + // "this" host is the new leader + logger.info("this host is the new leader for topic {}", getTopic()); + return becomeLeader(alive); + + } else { + // not the leader - return to previous state + logger.info("no new leader on topic {}", getTopic()); + return goActive(getAssignments()); + } + }); + } + + @Override + public State goQuery() { + return null; + } + + @Override + public State process(Identification msg) { + + if (getHost().equals(msg.getSource())) { + logger.info("saw our own Ident message on topic {}", getTopic()); + sawSelfIdent = true; + + } else { + logger.info("received Ident message from {} on topic {}", msg.getSource(), getTopic()); + recordInfo(msg.getSource(), msg.getAssignments()); + } + + return null; + } + + /** + * If the message leader is better than the leader we have, then go active with it. + * Otherwise, simply treat it like an {@link Identification} message. + */ + @Override + public State process(Leader msg) { + if (!isValid(msg)) { + return null; + } + + String source = msg.getSource(); + BucketAssignments asgn = msg.getAssignments(); + + // go active, if this has a leader that's the same or better than the one we have + if (source.compareTo(getLeader()) <= 0) { + logger.warn("leader with {} on topic {}", source, getTopic()); + return goActive(asgn); + } + + /* + * The message does not have an acceptable leader, but we'll still record its + * info. + */ + logger.info("record leader info from {} on topic {}", source, getTopic()); + recordInfo(source, asgn); + + return null; + } + + @Override + public State process(Offline msg) { + String host = msg.getSource(); + + if (host != null && !host.equals(getHost())) { + logger.warn("host {} offline on topic {}", host, getTopic()); + alive.remove(host); + setLeader(alive.first()); + + } else { + logger.info("ignored offline message from {} on topic {}", msg.getSource(), getTopic()); + } + + return null; + } + + /** + * Records info from a message, adding the source host name to {@link #alive}, and + * updating the bucket assignments. + * + * @param source the message's source host + * @param assignments assignments, or {@code null} + */ + private void recordInfo(String source, BucketAssignments assignments) { + // add this message's source host to "alive" + if (source != null) { + alive.add(source); + setLeader(alive.first()); + } + + if (assignments == null || assignments.getLeader() == null) { + return; + } + + // record assignments, if we don't have any yet + BucketAssignments current = getAssignments(); + if (current == null) { + logger.info("received initial assignments on topic {}", getTopic()); + setAssignments(assignments); + return; + } + + /* + * Record assignments, if the new assignments have a better (i.e., lesser) leader. + */ + String curldr = current.getLeader(); + if (curldr == null || assignments.getLeader().compareTo(curldr) < 0) { + logger.info("use new assignments from {} on topic {}", source, getTopic()); + setAssignments(assignments); + } + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/StartState.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/StartState.java new file mode 100644 index 00000000..73717d7c --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/StartState.java @@ -0,0 +1,99 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2020-2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.state; + +import lombok.Getter; +import org.onap.policy.drools.pooling.PoolingManager; +import org.onap.policy.drools.pooling.message.Heartbeat; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The start state. Upon entry, a heart beat is generated and the event filter is changed + * to look for just that particular message. Once the message is seen, it goes into the + * {@link QueryState}. + */ +@Getter +public class StartState extends State { + + private static final Logger logger = LoggerFactory.getLogger(StartState.class); + + /** + * Time stamp inserted into the heart beat message. + */ + private long hbTimestampMs = System.currentTimeMillis(); + + /** + * Constructor. + * + * @param mgr pooling manager + */ + public StartState(PoolingManager mgr) { + super(mgr); + } + + @Override + public void start() { + + super.start(); + + var hb = makeHeartbeat(hbTimestampMs); + publish(getHost(), hb); + + /* + * heart beat generator + */ + long genMs = getProperties().getInterHeartbeatMs(); + + scheduleWithFixedDelay(genMs, genMs, () -> { + publish(getHost(), hb); + return null; + }); + + /* + * my heart beat checker + */ + schedule(getProperties().getStartHeartbeatMs(), () -> { + logger.error("missed heartbeat on topic {}", getTopic()); + return internalTopicFailed(); + }); + } + + /** + * Transitions to the query state if the heart beat originated from this host and its + * time stamp matches. + */ + @Override + public State process(Heartbeat msg) { + if (msg.getTimestampMs() == hbTimestampMs && getHost().equals(msg.getSource())) { + // saw our own heart beat - transition to query state + logger.info("saw our own heartbeat on topic {}", getTopic()); + publish(makeQuery()); + return goQuery(); + + } else { + logger.info("ignored old heartbeat message from {} on topic {}", msg.getSource(), getTopic()); + } + + return null; + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/State.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/State.java new file mode 100644 index 00000000..e2cf9586 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/State.java @@ -0,0 +1,373 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018, 2020-2021 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.state; + +import java.util.LinkedList; +import java.util.List; +import org.onap.policy.drools.pooling.CancellableScheduledTask; +import org.onap.policy.drools.pooling.PoolingManager; +import org.onap.policy.drools.pooling.PoolingProperties; +import org.onap.policy.drools.pooling.message.BucketAssignments; +import org.onap.policy.drools.pooling.message.Heartbeat; +import org.onap.policy.drools.pooling.message.Identification; +import org.onap.policy.drools.pooling.message.Leader; +import org.onap.policy.drools.pooling.message.Offline; +import org.onap.policy.drools.pooling.message.Query; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A state in the finite state machine. + * + * <p>A state may have several timers associated with it, which must be cancelled whenever + * the state is changed. Assumes that timers are not continuously added to the same state. + */ +public abstract class State { + + private static final Logger logger = LoggerFactory.getLogger(State.class); + + /** + * Host pool manager. + */ + private final PoolingManager mgr; + + /** + * Timers added by this state. + */ + private final List<CancellableScheduledTask> timers = new LinkedList<>(); + + /** + * Constructor. + * + * @param mgr pooling manager + */ + protected State(PoolingManager mgr) { + this.mgr = mgr; + } + + /** + * Cancels the timers added by this state. + */ + public final void cancelTimers() { + timers.forEach(CancellableScheduledTask::cancel); + } + + /** + * Starts the state. The default method simply logs a message and returns. + */ + public void start() { + logger.info("entered {} for topic {}", getClass().getSimpleName(), getTopic()); + } + + /** + * Transitions to the "start" state. + * + * @return the new state + */ + public final State goStart() { + return mgr.goStart(); + } + + /** + * Transitions to the "query" state. + * + * @return the new state + */ + public State goQuery() { + return mgr.goQuery(); + } + + /** + * Goes active with a new set of assignments. + * + * @param asgn new assignments + * @return the new state, either Active or Inactive, depending on whether or not this + * host has an assignment + */ + protected State goActive(BucketAssignments asgn) { + startDistributing(asgn); + + if (asgn != null && asgn.hasAssignment(getHost())) { + return mgr.goActive(); + + } else { + return goInactive(); + } + } + + /** + * Transitions to the "inactive" state. + * + * @return the new state + */ + protected State goInactive() { + return mgr.goInactive(); + } + + /** + * Processes a message. The default method just returns {@code null}. + * + * @param msg message to be processed + * @return the new state, or {@code null} if the state is unchanged + */ + public State process(Heartbeat msg) { + logger.info("ignored heartbeat message from {} on topic {}", msg.getSource(), getTopic()); + return null; + } + + /** + * Processes a message. The default method just returns {@code null}. + * + * @param msg message to be processed + * @return the new state, or {@code null} if the state is unchanged + */ + public State process(Identification msg) { + logger.info("ignored ident message from {} on topic {}", msg.getSource(), getTopic()); + return null; + } + + /** + * Processes a message. The default method copies the assignments and then returns + * {@code null}. + * + * @param msg message to be processed + * @return the new state, or {@code null} if the state is unchanged + */ + public State process(Leader msg) { + if (isValid(msg)) { + logger.info("extract assignments from Leader message from {} on topic {}", msg.getSource(), getTopic()); + startDistributing(msg.getAssignments()); + } + + return null; + } + + /** + * Processes a message. The default method just returns {@code null}. + * + * @param msg message to be processed + * @return the new state, or {@code null} if the state is unchanged + */ + public State process(Offline msg) { + logger.info("ignored offline message from {} on topic {}", msg.getSource(), getTopic()); + return null; + } + + /** + * Processes a message. The default method just returns {@code null}. + * + * @param msg message to be processed + * @return the new state, or {@code null} if the state is unchanged + */ + public State process(Query msg) { + logger.info("ignored Query message from {} on topic {}", msg.getSource(), getTopic()); + return null; + } + + /** + * Determines if a message is valid and did not originate from this host. + * + * @param msg message to be validated + * @return {@code true} if the message is valid, {@code false} otherwise + */ + protected boolean isValid(Leader msg) { + BucketAssignments asgn = msg.getAssignments(); + if (asgn == null) { + logger.warn("Leader message from {} has no assignments for topic {}", msg.getSource(), getTopic()); + return false; + } + + // ignore Leader messages from ourself + String source = msg.getSource(); + if (source == null || source.equals(getHost())) { + logger.debug("ignore Leader message from {} for topic {}", msg.getSource(), getTopic()); + return false; + } + + // the new leader must equal the source + boolean result = source.equals(asgn.getLeader()); + + if (!result) { + logger.warn("Leader message from {} has an invalid assignment for topic {}", msg.getSource(), getTopic()); + } + + return result; + } + + /** + * Publishes a message. + * + * @param msg message to be published + */ + protected final void publish(Identification msg) { + mgr.publishAdmin(msg); + } + + /** + * Publishes a message. + * + * @param msg message to be published + */ + protected final void publish(Leader msg) { + mgr.publishAdmin(msg); + } + + /** + * Publishes a message. + * + * @param msg message to be published + */ + protected final void publish(Offline msg) { + mgr.publishAdmin(msg); + } + + /** + * Publishes a message. + * + * @param msg message to be published + */ + protected final void publish(Query msg) { + mgr.publishAdmin(msg); + } + + /** + * Publishes a message on the specified channel. + * + * @param channel channel + * @param msg message to be published + */ + protected final void publish(String channel, Heartbeat msg) { + mgr.publish(channel, msg); + } + + /** + * Starts distributing messages using the specified bucket assignments. + * + * @param assignments assignments + */ + protected final void startDistributing(BucketAssignments assignments) { + if (assignments != null) { + mgr.startDistributing(assignments); + } + } + + /** + * Schedules a timer to fire after a delay. + * + * @param delayMs delay in ms + * @param task task + */ + protected final void schedule(long delayMs, StateTimerTask task) { + timers.add(mgr.schedule(delayMs, task)); + } + + /** + * Schedules a timer to fire repeatedly. + * + * @param initialDelayMs initial delay ms + * @param delayMs delay ms + * @param task task + */ + protected final void scheduleWithFixedDelay(long initialDelayMs, long delayMs, StateTimerTask task) { + timers.add(mgr.scheduleWithFixedDelay(initialDelayMs, delayMs, task)); + } + + /** + * Indicates that we failed to see our own heartbeat; must be a problem with the + * internal topic. Assumes the problem is temporary and continues to use the current + * bucket assignments. + * + * @return a new {@link StartState} + */ + protected final State missedHeartbeat() { + publish(makeOffline()); + + return mgr.goStart(); + } + + /** + * Indicates that the internal topic failed; this should only be invoked from the + * StartState. Discards bucket assignments and begins processing everything locally. + * + * @return a new {@link InactiveState} + */ + protected final State internalTopicFailed() { + publish(makeOffline()); + mgr.startDistributing(null); + + return mgr.goInactive(); + } + + /** + * Makes a heart beat message. + * + * @param timestampMs time, in milliseconds, associated with the message + * + * @return a new message + */ + protected final Heartbeat makeHeartbeat(long timestampMs) { + return new Heartbeat(getHost(), timestampMs); + } + + /** + * Makes an Identification message. + * + * @return a new message + */ + protected Identification makeIdentification() { + return new Identification(getHost(), getAssignments()); + } + + /** + * Makes an "offline" message. + * + * @return a new message + */ + protected final Offline makeOffline() { + return new Offline(getHost()); + } + + /** + * Makes a query message. + * + * @return a new message + */ + protected final Query makeQuery() { + return new Query(getHost()); + } + + public final BucketAssignments getAssignments() { + return mgr.getAssignments(); + } + + public final String getHost() { + return mgr.getHost(); + } + + public final String getTopic() { + return mgr.getTopic(); + } + + public final PoolingProperties getProperties() { + return mgr.getProperties(); + } +} diff --git a/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/StateTimerTask.java b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/StateTimerTask.java new file mode 100644 index 00000000..892a1767 --- /dev/null +++ b/feature-pooling-messages/src/main/java/org/onap/policy/drools/pooling/state/StateTimerTask.java @@ -0,0 +1,37 @@ +/* + * ============LICENSE_START======================================================= + * ONAP + * ================================================================================ + * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved. + * Modifications Copyright (C) 2024 Nordix Foundation. + * ================================================================================ + * Licensed 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.drools.pooling.state; + +/** + * Task to be executed when a timer fires within a {@link State}. + */ +@FunctionalInterface +public interface StateTimerTask { + + /** + * Fires the timer. + * + * @return the new state, or {@code null} if the state is unchanged + */ + State fire(); + +} |