2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2016-2018 Ericsson. All rights reserved.
4 * ================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.apex.service.engine.runtime.impl;
23 import java.io.ByteArrayInputStream;
24 import java.util.Collection;
25 import java.util.Collections;
26 import java.util.LinkedHashMap;
28 import java.util.Map.Entry;
29 import java.util.concurrent.BlockingQueue;
30 import java.util.concurrent.LinkedBlockingQueue;
32 import org.onap.policy.apex.context.ContextException;
33 import org.onap.policy.apex.core.infrastructure.threading.ApplicationThreadFactory;
34 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
35 import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
36 import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
37 import org.onap.policy.apex.model.basicmodel.handling.ApexModelException;
38 import org.onap.policy.apex.model.basicmodel.handling.ApexModelReader;
39 import org.onap.policy.apex.model.basicmodel.service.ModelService;
40 import org.onap.policy.apex.model.enginemodel.concepts.AxEngineState;
41 import org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel;
42 import org.onap.policy.apex.service.engine.event.ApexEvent;
43 import org.onap.policy.apex.service.engine.event.ApexPeriodicEventGenerator;
44 import org.onap.policy.apex.service.engine.runtime.ApexEventListener;
45 import org.onap.policy.apex.service.engine.runtime.EngineService;
46 import org.onap.policy.apex.service.engine.runtime.EngineServiceEventInterface;
47 import org.onap.policy.apex.service.parameters.engineservice.EngineServiceParameters;
48 import org.onap.policy.common.parameters.GroupValidationResult;
49 import org.slf4j.ext.XLogger;
50 import org.slf4j.ext.XLoggerFactory;
53 * The Class EngineServiceImpl controls a thread pool that runs a set of Apex engine workers, each
54 * of which is running on an identical Apex model. This class handles the management of the engine
55 * worker instances, their threads, and event forwarding to and from the engine workers.
57 * @author Sajeevan Achuthan (sajeevan.achuthan@ericsson.com)
58 * @author Liam Fallon (liam.fallon@ericsson.com)
59 * @author John Keeney (john.keeney@ericsson.com)
61 public final class EngineServiceImpl implements EngineService, EngineServiceEventInterface {
62 // Logging static variables
63 private static final XLogger LOGGER = XLoggerFactory.getXLogger(EngineServiceImpl.class);
64 private static final boolean DEBUG_ENABLED = LOGGER.isDebugEnabled();
66 // Constants for timing
67 private static final long MAX_START_WAIT_TIME = 5000; // 5 seconds
68 private static final long MAX_STOP_WAIT_TIME = 5000; // 5 seconds
69 private static final int ENGINE_SERVICE_STOP_START_WAIT_INTERVAL = 200;
71 // The ID of this engine
72 private AxArtifactKey engineServiceKey = null;
74 // The Apex engine workers this engine service is handling
75 private final Map<AxArtifactKey, EngineService> engineWorkerMap =
76 Collections.synchronizedMap(new LinkedHashMap<AxArtifactKey, EngineService>());
78 // Event queue for events being sent into the Apex engines, it used by all engines within a
80 private final BlockingQueue<ApexEvent> queue = new LinkedBlockingQueue<>();
82 // Thread factory for thread management
83 private final ApplicationThreadFactory atFactory = new ApplicationThreadFactory("apex-engine-service", 512);
85 // Periodic event generator and its period in milliseconds
86 private ApexPeriodicEventGenerator periodicEventGenerator = null;
87 private long periodicEventPeriod;
90 * This constructor instantiates engine workers and adds them to the set of engine workers to be
91 * managed. The constructor is private to prevent subclassing.
93 * @param engineServiceKey the engine service key
94 * @param incomingThreadCount the thread count, the number of engine workers to start
95 * @param periodicEventPeriod the period in milliseconds at which periodic events are generated
96 * @throws ApexException on worker instantiation errors
98 private EngineServiceImpl(final AxArtifactKey engineServiceKey, final int incomingThreadCount,
99 final long periodicEventPeriod) throws ApexException {
100 LOGGER.entry(engineServiceKey, incomingThreadCount);
102 this.engineServiceKey = engineServiceKey;
103 this.periodicEventPeriod = periodicEventPeriod;
105 int threadCount = incomingThreadCount;
106 if (threadCount <= 0) {
107 // Just start one engine worker
111 // Start engine workers
112 for (int engineCounter = 0; engineCounter < threadCount; engineCounter++) {
113 final AxArtifactKey engineWorkerKey =
114 new AxArtifactKey(engineServiceKey.getName() + '-' + engineCounter, engineServiceKey.getVersion());
115 engineWorkerMap.put(engineWorkerKey, new EngineWorker(engineWorkerKey, queue, atFactory));
116 LOGGER.info("Created apex engine {} .", engineWorkerKey.getId());
119 LOGGER.info("APEX service created.");
124 * Create an Apex Engine Service instance. This method is deprecated and will be removed in the
127 * @param engineServiceKey the engine service key
128 * @param threadCount the thread count, the number of engine workers to start
129 * @return the Engine Service instance
130 * @throws ApexException on worker instantiation errors
131 * @deprecated Do not use this version. Use {@link #create(EngineServiceParameters)}
134 public static EngineServiceImpl create(final AxArtifactKey engineServiceKey, final int threadCount)
135 throws ApexException {
136 // Check if the Apex model specified is sane
137 if (engineServiceKey == null) {
138 LOGGER.warn("engine service key is null");
139 throw new ApexException("engine service key is null");
141 return new EngineServiceImpl(engineServiceKey, threadCount, 0);
145 * Create an Apex Engine Service instance. This method does not load the policy so
146 * {@link #updateModel(AxArtifactKey, AxPolicyModel, boolean)} or
147 * {@link #updateModel(AxArtifactKey, AxPolicyModel, boolean)} must be used to load a model.
148 * This method does not start the Engine Service so {@link #start(AxArtifactKey)} or
149 * {@link #startAll()} must be used.
151 * @param config the configuration for this Apex Engine Service.
152 * @return the Engine Service instance
153 * @throws ApexException on worker instantiation errors
155 public static EngineServiceImpl create(final EngineServiceParameters config) throws ApexException {
156 if (config == null) {
157 LOGGER.warn("Engine service configuration parameters is null");
158 throw new ApexException("engine service configuration parameters is null");
160 final GroupValidationResult validation = config.validate();
161 if (!validation.isValid()) {
162 LOGGER.warn("Invalid engine service configuration parameters: {}" + validation.getResult());
163 throw new ApexException("Invalid engine service configuration parameters: " + validation);
165 final AxArtifactKey engineServiceKey = config.getEngineKey();
166 final int threadCount = config.getInstanceCount();
168 // Check if the Apex model specified is sane
169 if (engineServiceKey == null) {
170 LOGGER.warn("engine service key is null");
171 throw new ApexException("engine service key is null");
174 return new EngineServiceImpl(engineServiceKey, threadCount, config.getPeriodicEventPeriod());
180 * @see com.ericsson.apex.service.engine.runtime.EngineService#registerActionListener(java.lang.
181 * String, com.ericsson.apex.service.engine.runtime.ApexEventListener)
184 public void registerActionListener(final String listenerName, final ApexEventListener apexEventListener) {
185 LOGGER.entry(apexEventListener);
187 // Register the Apex event listener on all engine workers, each worker will return Apex
188 // events to the listening application
189 for (final EngineService engineWorker : engineWorkerMap.values()) {
190 engineWorker.registerActionListener(listenerName, apexEventListener);
193 LOGGER.info("Added the action listener to the engine");
201 * com.ericsson.apex.service.engine.runtime.EngineService#deregisterActionListener(java.lang.
205 public void deregisterActionListener(final String listenerName) {
206 LOGGER.entry(listenerName);
208 // Register the Apex event listener on all engine workers, each worker will return Apex
209 // events to the listening application
210 for (final EngineService engineWorker : engineWorkerMap.values()) {
211 engineWorker.deregisterActionListener(listenerName);
214 LOGGER.info("Removed the action listener from the engine");
221 * @see com.ericsson.apex.service.engine.runtime.EngineService#getEngineServiceEventInterface()
224 public EngineServiceEventInterface getEngineServiceEventInterface() {
231 * @see com.ericsson.apex.service.engine.runtime.EngineService#getKey()
234 public AxArtifactKey getKey() {
235 return engineServiceKey;
241 * @see com.ericsson.apex.service.engine.runtime.EngineService#getInfo()
244 public Collection<AxArtifactKey> getEngineKeys() {
245 return engineWorkerMap.keySet();
251 * @see com.ericsson.apex.service.engine.runtime.EngineService#getApexModelKey()
254 public AxArtifactKey getApexModelKey() {
255 if (engineWorkerMap.size() == 0) {
259 return engineWorkerMap.entrySet().iterator().next().getValue().getApexModelKey();
266 * com.ericsson.apex.service.engine.runtime.EngineService#updateModel(com.ericsson.apex.model.
267 * basicmodel.concepts.AxArtifactKey, java.lang.String, boolean)
270 public void updateModel(final AxArtifactKey incomingEngineServiceKey, final String apexModelString,
271 final boolean forceFlag) throws ApexException {
272 // Check if the Apex model specified is sane
273 if (apexModelString == null || apexModelString.trim().length() == 0) {
275 "model for updating on engine service with key " + incomingEngineServiceKey.getId() + " is empty");
276 throw new ApexException(
277 "model for updating on engine service with key " + incomingEngineServiceKey.getId() + " is empty");
280 // Read the Apex model into memory using the Apex Model Reader
281 AxPolicyModel apexPolicyModel = null;
283 final ApexModelReader<AxPolicyModel> modelReader = new ApexModelReader<>(AxPolicyModel.class);
284 apexPolicyModel = modelReader.read(new ByteArrayInputStream(apexModelString.getBytes()));
285 } catch (final ApexModelException e) {
286 LOGGER.error("failed to unmarshal the apex model on engine service " + incomingEngineServiceKey.getId(), e);
287 throw new ApexException(
288 "failed to unmarshal the apex model on engine service " + incomingEngineServiceKey.getId(), e);
291 if (apexPolicyModel == null) {
292 LOGGER.error("apex model null on engine service " + incomingEngineServiceKey.getId());
293 throw new ApexException("apex model null on engine service " + incomingEngineServiceKey.getId());
297 updateModel(incomingEngineServiceKey, apexPolicyModel, forceFlag);
306 * com.ericsson.apex.service.engine.runtime.EngineService#updateModel(com.ericsson.apex.model.
307 * basicmodel.concepts.AxArtifactKey,
308 * com.ericsson.apex.model.policymodel.concepts.AxPolicyModel, boolean)
311 public void updateModel(final AxArtifactKey incomingEngineServiceKey, final AxPolicyModel apexModel,
312 final boolean forceFlag) throws ApexException {
313 LOGGER.entry(incomingEngineServiceKey);
315 // Check if the Apex model specified is sane
316 if (apexModel == null) {
318 "model for updating on engine service with key " + incomingEngineServiceKey.getId() + " is null");
319 throw new ApexException(
320 "model for updating on engine service with key " + incomingEngineServiceKey.getId() + " is null");
323 // Check if the key on the update request is correct
324 if (!this.engineServiceKey.equals(incomingEngineServiceKey)) {
325 LOGGER.warn("engine service key " + incomingEngineServiceKey.getId() + " does not match the key"
326 + engineServiceKey.getId() + " of this engine service");
327 throw new ApexException("engine service key " + incomingEngineServiceKey.getId() + " does not match the key"
328 + engineServiceKey.getId() + " of this engine service");
331 // Check model compatibility
332 if (ModelService.existsModel(AxPolicyModel.class)) {
333 // The current policy model may or may not be defined
334 final AxPolicyModel currentModel = ModelService.getModel(AxPolicyModel.class);
335 if (!currentModel.getKey().isCompatible(apexModel.getKey())) {
337 LOGGER.warn("apex model update forced, supplied model with key \"" + apexModel.getKey().getId()
338 + "\" is not a compatible model update from the existing engine model with key \""
339 + currentModel.getKey().getId() + "\"");
341 throw new ContextException(
342 "apex model update failed, supplied model with key \"" + apexModel.getKey().getId()
343 + "\" is not a compatible model update from the existing engine model with key \""
344 + currentModel.getKey().getId() + "\"");
349 final boolean wasstopped = isStopped();
352 // Stop all engines on this engine service
354 final long stoptime = System.currentTimeMillis();
355 while (!isStopped() && System.currentTimeMillis() - stoptime < MAX_STOP_WAIT_TIME) {
356 ThreadUtilities.sleep(ENGINE_SERVICE_STOP_START_WAIT_INTERVAL);
358 // Check if all engines are stopped
359 final StringBuilder notStoppedEngineIDBuilder = new StringBuilder();
360 for (final Entry<AxArtifactKey, EngineService> engineWorkerEntry : engineWorkerMap.entrySet()) {
361 if (engineWorkerEntry.getValue().getState() != AxEngineState.STOPPED) {
362 notStoppedEngineIDBuilder.append(engineWorkerEntry.getKey().getId());
363 notStoppedEngineIDBuilder.append('(');
364 notStoppedEngineIDBuilder.append(engineWorkerEntry.getValue().getState());
365 notStoppedEngineIDBuilder.append(") ");
368 if (notStoppedEngineIDBuilder.length() > 0) {
369 final String errorString = "cannot update model on engine service with key "
370 + incomingEngineServiceKey.getId() + ", engines not stopped after " + MAX_STOP_WAIT_TIME
371 + "ms are: " + notStoppedEngineIDBuilder.toString().trim();
372 LOGGER.warn(errorString);
373 throw new ApexException(errorString);
377 // Update the engines
378 for (final Entry<AxArtifactKey, EngineService> engineWorkerEntry : engineWorkerMap.entrySet()) {
379 LOGGER.info("Registering apex model on engine {}", engineWorkerEntry.getKey().getId());
380 engineWorkerEntry.getValue().updateModel(engineWorkerEntry.getKey(), apexModel, forceFlag);
384 // start all engines on this engine service if it was not stopped before the update
386 final long starttime = System.currentTimeMillis();
387 while (!isStarted() && System.currentTimeMillis() - starttime < MAX_START_WAIT_TIME) {
388 ThreadUtilities.sleep(ENGINE_SERVICE_STOP_START_WAIT_INTERVAL);
390 // Check if all engines are running
391 final StringBuilder notRunningEngineIDBuilder = new StringBuilder();
392 for (final Entry<AxArtifactKey, EngineService> engineWorkerEntry : engineWorkerMap.entrySet()) {
393 if (engineWorkerEntry.getValue().getState() != AxEngineState.READY
394 && engineWorkerEntry.getValue().getState() != AxEngineState.EXECUTING) {
395 notRunningEngineIDBuilder.append(engineWorkerEntry.getKey().getId());
396 notRunningEngineIDBuilder.append('(');
397 notRunningEngineIDBuilder.append(engineWorkerEntry.getValue().getState());
398 notRunningEngineIDBuilder.append(") ");
401 if (notRunningEngineIDBuilder.length() > 0) {
402 final String errorString = "engine start error on model update on engine service with key "
403 + incomingEngineServiceKey.getId() + ", engines not running are: "
404 + notRunningEngineIDBuilder.toString().trim();
405 LOGGER.warn(errorString);
406 throw new ApexException(errorString);
416 * @see com.ericsson.apex.service.engine.runtime.EngineService#getState()
419 public AxEngineState getState() {
420 // If one worker is running then we are running, otherwise we are stopped
421 for (final EngineService engine : engineWorkerMap.values()) {
422 if (engine.getState() != AxEngineState.STOPPED) {
423 return AxEngineState.EXECUTING;
427 return AxEngineState.STOPPED;
433 * @see com.ericsson.apex.service.engine.runtime.EngineService#startAll()
436 public void startAll() throws ApexException {
437 for (final EngineService engine : engineWorkerMap.values()) {
438 start(engine.getKey());
441 // Check if periodic events should be turned on
442 if (periodicEventPeriod > 0) {
443 startPeriodicEvents(periodicEventPeriod);
451 * com.ericsson.apex.service.engine.runtime.EngineService#start(com.ericsson.apex.core.model.
452 * concepts.AxArtifactKey)
455 public void start(final AxArtifactKey engineKey) throws ApexException {
456 LOGGER.entry(engineKey);
458 // Check if we have this key on our map
459 if (!engineWorkerMap.containsKey(engineKey)) {
460 LOGGER.warn("engine with key " + engineKey.getId() + " not found in engine service");
461 throw new ApexException("engine with key " + engineKey.getId() + " not found in engine service");
465 engineWorkerMap.get(engineKey).start(engineKey);
467 LOGGER.exit(engineKey);
473 * @see com.ericsson.apex.service.engine.runtime.EngineService#stop()
476 public void stop() throws ApexException {
480 for (final EngineService engine : engineWorkerMap.values()) {
481 if (engine.getState() != AxEngineState.STOPPED) {
493 * com.ericsson.apex.service.engine.runtime.EngineService#stop(com.ericsson.apex.core.model.
494 * concepts.AxArtifactKey)
497 public void stop(final AxArtifactKey engineKey) throws ApexException {
498 LOGGER.entry(engineKey);
500 // Check if we have this key on our map
501 if (!engineWorkerMap.containsKey(engineKey)) {
502 LOGGER.warn("engine with key " + engineKey.getId() + " not found in engine service");
503 throw new ApexException("engine with key " + engineKey.getId() + " not found in engine service");
507 engineWorkerMap.get(engineKey).stop(engineKey);
509 LOGGER.exit(engineKey);
513 * Check all engines are started.
515 * @return true if <i>all</i> engines are started
516 * @see org.onap.policy.apex.service.engine.runtime.EngineService#isStarted()
519 public boolean isStarted() {
520 for (final EngineService engine : engineWorkerMap.values()) {
521 if (!engine.isStarted()) {
532 * com.ericsson.apex.service.engine.runtime.EngineService#isStarted(com.ericsson.apex.model.
533 * basicmodel.concepts.AxArtifactKey)
536 public boolean isStarted(final AxArtifactKey engineKey) {
537 // Check if we have this key on our map
538 if (!engineWorkerMap.containsKey(engineKey)) {
539 LOGGER.warn("engine with key " + engineKey.getId() + " not found in engine service");
541 return engineWorkerMap.get(engineKey).isStarted();
545 * Check all engines are stopped.
547 * @return true if <i>all</i> engines are stopped
548 * @see org.onap.policy.apex.service.engine.runtime.EngineService#isStopped()
551 public boolean isStopped() {
552 for (final EngineService engine : engineWorkerMap.values()) {
553 if (!engine.isStopped()) {
564 * com.ericsson.apex.service.engine.runtime.EngineService#isStopped(com.ericsson.apex.model.
565 * basicmodel.concepts.AxArtifactKey)
568 public boolean isStopped(final AxArtifactKey engineKey) {
569 // Check if we have this key on our map
570 if (!engineWorkerMap.containsKey(engineKey)) {
571 LOGGER.warn("engine with key " + engineKey.getId() + " not found in engine service");
573 return engineWorkerMap.get(engineKey).isStopped();
579 * @see com.ericsson.apex.service.engine.runtime.EngineService#startPeriodicEvents(long)
582 public void startPeriodicEvents(final long period) throws ApexException {
583 // Check if periodic events are already started
584 if (periodicEventGenerator != null) {
585 LOGGER.warn("Peiodic event geneation already running on engine " + engineServiceKey.getId() + ", "
586 + periodicEventGenerator.toString());
587 throw new ApexException("Peiodic event geneation already running on engine " + engineServiceKey.getId()
588 + ", " + periodicEventGenerator.toString());
591 // Set up periodic event execution, its a Java Timer/TimerTask
592 periodicEventGenerator = new ApexPeriodicEventGenerator(this.getEngineServiceEventInterface(), period);
594 // Record the periodic event period because it may have been set over the Web Socket admin
596 this.periodicEventPeriod = period;
602 * @see com.ericsson.apex.service.engine.runtime.EngineService#stopPeriodicEvents()
605 public void stopPeriodicEvents() throws ApexException {
606 // Check if periodic events are already started
607 if (periodicEventGenerator == null) {
608 LOGGER.warn("Peiodic event geneation not running on engine " + engineServiceKey.getId());
609 throw new ApexException("Peiodic event geneation not running on engine " + engineServiceKey.getId());
612 // Stop periodic events
613 periodicEventGenerator.cancel();
614 periodicEventGenerator = null;
621 * com.ericsson.apex.service.engine.runtime.EngineService#getStatus(com.ericsson.apex.core.model
622 * .concepts.AxArtifactKey)
625 public String getStatus(final AxArtifactKey engineKey) throws ApexException {
626 // Check if we have this key on our map
627 if (!engineWorkerMap.containsKey(engineKey)) {
628 LOGGER.warn("engine with key " + engineKey.getId() + " not found in engine service");
629 throw new ApexException("engine with key " + engineKey.getId() + " not found in engine service");
632 // Return the information for this worker
633 return engineWorkerMap.get(engineKey).getStatus(engineKey);
640 * com.ericsson.apex.service.engine.runtime.EngineService#getRuntimeInfo(com.ericsson.apex.core.
641 * model.concepts.AxArtifactKey)
644 public String getRuntimeInfo(final AxArtifactKey engineKey) throws ApexException {
645 // Check if we have this key on our map
646 if (!engineWorkerMap.containsKey(engineKey)) {
647 LOGGER.warn("engine with key " + engineKey.getId() + " not found in engine service");
648 throw new ApexException("engine with key " + engineKey.getId() + " not found in engine service");
651 // Return the information for this worker
652 return engineWorkerMap.get(engineKey).getRuntimeInfo(engineKey);
659 * com.ericsson.apex.service.engine.runtime.EngineServiceEventInterface#sendEvent(com.ericsson.
660 * apex.service.engine.event.ApexEvent)
663 public void sendEvent(final ApexEvent event) {
664 // Check if we have this key on our map
665 if (getState() == AxEngineState.STOPPED) {
666 LOGGER.warn("event " + event.getName() + " not processed, no engines on engine service "
667 + engineServiceKey.getId() + " are running");
672 LOGGER.warn("Null events cannot be processed, in engine service " + engineServiceKey.getId());
677 LOGGER.debug("Forwarding Apex Event {} to the processing engine", event);
680 // Add the incoming event to the queue, the next available worker will process it