1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
|
/*-
* ============LICENSE_START=======================================================
* Copyright (C) 2016-2018 Ericsson. All rights reserved.
* ================================================================================
* 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.
*
* SPDX-License-Identifier: Apache-2.0
* ============LICENSE_END=========================================================
*/
package org.onap.policy.apex.core.deployment;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import java.util.concurrent.TimeUnit;
import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
import org.onap.policy.apex.core.protocols.Message;
import org.onap.policy.apex.core.protocols.engdep.messages.EngineServiceInfoResponse;
import org.onap.policy.apex.core.protocols.engdep.messages.GetEngineInfo;
import org.onap.policy.apex.core.protocols.engdep.messages.GetEngineServiceInfo;
import org.onap.policy.apex.core.protocols.engdep.messages.GetEngineStatus;
import org.onap.policy.apex.core.protocols.engdep.messages.Response;
import org.onap.policy.apex.core.protocols.engdep.messages.StartEngine;
import org.onap.policy.apex.core.protocols.engdep.messages.StartPeriodicEvents;
import org.onap.policy.apex.core.protocols.engdep.messages.StopEngine;
import org.onap.policy.apex.core.protocols.engdep.messages.StopPeriodicEvents;
import org.onap.policy.apex.core.protocols.engdep.messages.UpdateModel;
import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
import org.onap.policy.apex.model.basicmodel.handling.ApexModelReader;
import org.onap.policy.apex.model.basicmodel.handling.ApexModelWriter;
import org.onap.policy.apex.model.enginemodel.concepts.AxEngineModel;
import org.onap.policy.apex.model.policymodel.concepts.AxPolicyModel;
import org.onap.policy.common.utils.resources.ResourceUtils;
import org.slf4j.ext.XLogger;
import org.slf4j.ext.XLoggerFactory;
/**
* The Class Deployer deploys an Apex model held as an XML file onto an Apex engine. It uses the EngDep protocol to
* communicate with the engine, with the EngDep protocol being carried on Java web sockets.
*
* <p>his deployer is a simple command line deployer that reads the communication parameters and the location of the XML
* model file as arguments.
*
* @author Liam Fallon (liam.fallon@ericsson.com)
*/
public class EngineServiceFacade {
// Get a reference to the logger
private static final XLogger LOGGER = XLoggerFactory.getXLogger(EngineServiceFacade.class);
// Repeated string constants
private static final String RECEIVED_FROM_SERVER = " received from server";
private static final String FAILED_RESPONSE = "failed response ";
// The default message timeout and timeout increment (the amount of time between polls) in
// milliseconds
private static final int CLIENT_START_WAIT_INTERVAL = 100;
private static final int REPLY_MESSAGE_TIMEOUT_DEFAULT = 10000;
private static final int REPLY_MESSAGE_TIMEOUT_INCREMENT = 100;
// The Apex engine host and EngDep port
private final String hostName;
private final int port;
// The deployment client handles the EngDep communication session towards the Apex server
private DeploymentClient client = null;
private Thread clientThread = null;
// Information about the Engine service we are connected to
private AxArtifactKey engineServiceKey = null;
private AxArtifactKey[] engineKeyArray = null;
private AxArtifactKey apexModelKey = null;
/**
* Instantiates a new deployer.
*
* @param hostName the host name of the host running the Apex Engine
* @param port the port to use for EngDep communication with the Apex engine
*/
public EngineServiceFacade(final String hostName, final int port) {
this.hostName = hostName;
this.port = port;
}
/**
* Initializes the facade, opens an EngDep communication session with the Apex engine.
*
* @throws ApexDeploymentException thrown on deployment and communication errors
*/
public void init() throws ApexDeploymentException {
try {
LOGGER.debug("handshaking with server {}:{} . . .", hostName, port);
// Use the deployment client to handle the EngDep communication towards the Apex server.
// It runs a thread to
// monitor the session and to send
// messages
client = new DeploymentClient(hostName, port);
clientThread = new Thread(client);
clientThread.start();
// Wait for the connection to come up
while (!client.isStarted()) {
if (clientThread.isAlive()) {
ThreadUtilities.sleep(CLIENT_START_WAIT_INTERVAL);
} else {
LOGGER.error("cound not handshake with server {}:{}", hostName, port);
throw new ApexDeploymentException("cound not handshake with server " + hostName + ":" + port);
}
}
LOGGER.debug("opened connection to server {}:{} . . .", hostName, port);
// Get engine service information to see what engines we're dealing with
final GetEngineServiceInfo engineServiceInfo = new GetEngineServiceInfo(null);
LOGGER.debug("sending get engine service info message {} to server {}:{} . . .", engineServiceInfo,
hostName, port);
client.sendMessage(engineServiceInfo);
LOGGER.debug("sent get engine service info message to server {}:{} . . .", hostName, port);
final EngineServiceInfoResponse engineServiceInfoResponse = (EngineServiceInfoResponse) getResponse(
engineServiceInfo);
if (engineServiceInfoResponse.isSuccessful()) {
engineServiceKey = engineServiceInfoResponse.getEngineServiceKey();
engineKeyArray = engineServiceInfoResponse.getEngineKeyArray();
apexModelKey = engineServiceInfoResponse.getApexModelKey();
}
} catch (final Exception e) {
LOGGER.error("cound not handshake with server {}:{}", hostName, port, e);
client.stopClient();
throw new ApexDeploymentException("cound not handshake with server " + hostName + ":" + port, e);
}
}
/**
* Get the engine service key.
*
* @return the engine service key
*/
public AxArtifactKey getApexModelKey() {
return apexModelKey;
}
/**
* Get the keys of the engines on this engine service.
*
* @return the engine key array
*/
public AxArtifactKey[] getEngineKeyArray() {
return engineKeyArray;
}
/**
* Get the engine service key.
*
* @return the engine service key
*/
public AxArtifactKey getKey() {
return engineServiceKey;
}
/**
* Close the EngDep connection to the Apex server.
*/
public void close() {
LOGGER.debug("closing connection to server {}:{} . . .", hostName, port);
client.stopClient();
LOGGER.debug("closed connection to server {}:{} . . .", hostName, port);
}
/**
* Deploy an Apex model on the Apex engine service.
*
* @param modelFileName the name of the model file containing the model to deploy
* @param ignoreConflicts true if conflicts between context in polices is to be ignored
* @param force true if the model is to be applied even if it is incompatible with the existing model
* @throws ApexException on Apex errors
* @throws IOException on IO exceptions from the operating system
*/
public void deployModel(final String modelFileName, final boolean ignoreConflicts, final boolean force)
throws ApexException, IOException {
if (engineServiceKey == null || engineKeyArray == null || engineKeyArray.length == 0) {
LOGGER.error("cound not deploy apex model, deployer is not initialized");
throw new ApexDeploymentException("cound not deploy apex model, deployer is not initialized");
}
// Get the model file as a string
URL apexModelUrl = ResourceUtils.getLocalFile(modelFileName);
if (apexModelUrl == null) {
apexModelUrl = ResourceUtils.getUrlResource(modelFileName);
if (apexModelUrl == null) {
LOGGER.error("cound not create apex model, could not read from XML file {}", modelFileName);
throw new ApexDeploymentException(
"cound not create apex model, could not read XML file " + modelFileName);
}
}
deployModel(modelFileName, apexModelUrl.openStream(), ignoreConflicts, force);
}
/**
* Deploy an Apex model on the Apex engine service.
*
* @param modelFileName the name of the model file containing the model to deploy
* @param modelInputStream the stream that holds the Apex model
* @param ignoreConflicts true if conflicts between context in polices is to be ignored
* @param force true if the model is to be applied even if it is incompatible with the existing model
* @throws ApexException on model deployment errors
*/
public void deployModel(final String modelFileName, final InputStream modelInputStream,
final boolean ignoreConflicts, final boolean force) throws ApexException {
// Read the policy model from the stream
final ApexModelReader<AxPolicyModel> modelReader = new ApexModelReader<>(AxPolicyModel.class);
modelReader.setValidateFlag(!ignoreConflicts);
final AxPolicyModel apexPolicyModel = modelReader.read(modelInputStream);
if (apexPolicyModel == null) {
LOGGER.error("cound not create apex model, could not read model stream");
throw new ApexDeploymentException("cound not create apex model, could not read model stream");
}
// Deploy the model
deployModel(apexPolicyModel, ignoreConflicts, force);
}
/**
* Deploy an Apex model on the Apex engine service.
*
* @param apexPolicyModel the name of the model to deploy
* @param ignoreConflicts true if conflicts between context in polices is to be ignored
* @param force true if the model is to be applied even if it is incompatible with the existing model
* @throws ApexException on model deployment errors
*/
public void deployModel(final AxPolicyModel apexPolicyModel, final boolean ignoreConflicts, final boolean force)
throws ApexException {
// Write the model into a byte array
final ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
final ApexModelWriter<AxPolicyModel> modelWriter = new ApexModelWriter<>(AxPolicyModel.class);
modelWriter.write(apexPolicyModel, baOutputStream);
// Create and send Update message
final UpdateModel umMessage = new UpdateModel(engineServiceKey, baOutputStream.toString(), ignoreConflicts,
force);
LOGGER.debug("sending update message {} to server {}:{} . . .", umMessage, hostName, port);
client.sendMessage(umMessage);
LOGGER.debug("sent update message to server {}:{} . . .", hostName, port);
// Check if we got a response
final Response response = getResponse(umMessage);
if (!response.isSuccessful()) {
LOGGER.warn(FAILED_RESPONSE + "{} received from server {}:{}", response.getMessageData(), hostName, port);
throw new ApexException(FAILED_RESPONSE + response.getMessageData() + RECEIVED_FROM_SERVER + hostName
+ ':' + port);
}
}
/**
* Start an Apex engine on the engine service.
*
* @param engineKey the key of the engine to start
* @throws ApexDeploymentException on messaging errors
*/
public void startEngine(final AxArtifactKey engineKey) throws ApexDeploymentException {
final StartEngine startEngineMessage = new StartEngine(engineKey);
LOGGER.debug("sending start engine {} to server {}:{} . . .", startEngineMessage, hostName, port);
client.sendMessage(startEngineMessage);
LOGGER.debug("sent start engine message to server {}:{} . . .", hostName, port);
// Check if we got a response
final Response response = getResponse(startEngineMessage);
if (!response.isSuccessful()) {
String message = FAILED_RESPONSE + response.getMessageData() + RECEIVED_FROM_SERVER
+ hostName + ':' + port;
LOGGER.warn(message);
throw new ApexDeploymentException(message);
}
}
/**
* Stop an Apex engine on the engine service.
*
* @param engineKey the key of the engine to stop
* @throws ApexDeploymentException on messaging errors
*/
public void stopEngine(final AxArtifactKey engineKey) throws ApexDeploymentException {
final StopEngine stopEngineMessage = new StopEngine(engineKey);
LOGGER.debug("sending stop engine {} to server {}:{} . . .", stopEngineMessage, hostName, port);
client.sendMessage(stopEngineMessage);
LOGGER.debug("sent stop engine message to server {}:{} . . .", hostName, port);
// Check if we got a response
final Response response = getResponse(stopEngineMessage);
if (!response.isSuccessful()) {
String message = FAILED_RESPONSE + response.getMessageData() + RECEIVED_FROM_SERVER
+ hostName + ':' + port;
LOGGER.warn(message);
throw new ApexDeploymentException(message);
}
}
/**
* Start periodic events on an Apex engine on the engine service.
*
* @param engineKey the key of the engine to start periodic events on
* @param period the period in milliseconds between periodic events
* @throws ApexDeploymentException on messaging errors
*/
public void startPerioidicEvents(final AxArtifactKey engineKey, final long period) throws ApexDeploymentException {
final StartPeriodicEvents startPerioidicEventsMessage = new StartPeriodicEvents(engineKey);
startPerioidicEventsMessage.setMessageData(Long.toString(period));
LOGGER.debug("sending start perioidic events {} to server {}:{} . . .", startPerioidicEventsMessage, hostName,
port);
client.sendMessage(startPerioidicEventsMessage);
LOGGER.debug("sent start perioidic events message to server {}:{} . . .", hostName, port);
// Check if we got a response
final Response response = getResponse(startPerioidicEventsMessage);
if (!response.isSuccessful()) {
String message = FAILED_RESPONSE + response.getMessageData() + RECEIVED_FROM_SERVER
+ hostName + ':' + port;
LOGGER.warn(message);
throw new ApexDeploymentException(message);
}
}
/**
* Stop periodic events on an Apex engine on the engine service.
*
* @param engineKey the key of the engine to stop periodic events on
* @throws ApexDeploymentException on messaging errors
*/
public void stopPerioidicEvents(final AxArtifactKey engineKey) throws ApexDeploymentException {
final StopPeriodicEvents stopPerioidicEventsMessage = new StopPeriodicEvents(engineKey);
LOGGER.debug("sending stop perioidic events {} to server {}:{} . . .", stopPerioidicEventsMessage, hostName,
port);
client.sendMessage(stopPerioidicEventsMessage);
LOGGER.debug("sent stop perioidic events message to server {}:{} . . .", hostName, port);
// Check if we got a response
final Response response = getResponse(stopPerioidicEventsMessage);
if (!response.isSuccessful()) {
String message = FAILED_RESPONSE + response.getMessageData() + RECEIVED_FROM_SERVER
+ hostName + ':' + port;
LOGGER.warn(message);
throw new ApexDeploymentException(message);
}
}
/**
* Get the status of an Apex engine.
*
* @param engineKey the key of the engine to get the status of
* @return an engine model containing the status of the engine for the given key
* @throws ApexException the apex exception
*/
public AxEngineModel getEngineStatus(final AxArtifactKey engineKey) throws ApexException {
final GetEngineStatus engineStatusMessage = new GetEngineStatus(engineKey);
LOGGER.debug("sending get engine status message {} to server {}:{} . . .", engineStatusMessage, hostName, port);
client.sendMessage(engineStatusMessage);
LOGGER.debug("sent get engine status message to server {}:{} . . .", hostName, port);
// Check if we got a response
final Response response = getResponse(engineStatusMessage);
if (!response.isSuccessful()) {
String message = FAILED_RESPONSE + response.getMessageData() + RECEIVED_FROM_SERVER + hostName
+ ':' + port;
LOGGER.warn(message);
throw new ApexException(message);
}
final ByteArrayInputStream baInputStream = new ByteArrayInputStream(response.getMessageData().getBytes());
final ApexModelReader<AxEngineModel> modelReader = new ApexModelReader<>(AxEngineModel.class);
modelReader.setValidateFlag(false);
return modelReader.read(baInputStream);
}
/**
* Get the runtime information of an Apex engine.
*
* @param engineKey the key of the engine to get information for
* @return an engine model containing information on the engine for the given key
* @throws ApexException the apex exception
*/
public String getEngineInfo(final AxArtifactKey engineKey) throws ApexException {
final GetEngineInfo engineInfoMessage = new GetEngineInfo(engineKey);
LOGGER.debug("sending get engine information message {} to server {}:{} . . .", engineInfoMessage, hostName,
port);
client.sendMessage(engineInfoMessage);
LOGGER.debug("sent get engine information message to server {}:{} . . .", hostName, port);
// Check if we got a response
final Response response = getResponse(engineInfoMessage);
if (!response.isSuccessful()) {
String message = FAILED_RESPONSE + response.getMessageData() + RECEIVED_FROM_SERVER + hostName
+ ':' + port;
LOGGER.warn(message);
throw new ApexException(message);
}
return response.getMessageData();
}
/**
* Check the response to a model deployment message from the Apex server.
*
* @param sentMessage the sent message
* @return the response message
* @throws ApexDeploymentException the apex deployment exception
*/
private Response getResponse(final Message sentMessage) throws ApexDeploymentException {
// Get the amount of milliseconds we should wait for a timeout
int timeoutTime = sentMessage.getReplyTimeout();
if (timeoutTime <= 0) {
timeoutTime = REPLY_MESSAGE_TIMEOUT_DEFAULT;
}
// Wait for the required amount of milliseconds for the response from the Apex server
Message receivedMessage = null;
for (int timeWaitedSoFar = 0; receivedMessage == null
&& timeWaitedSoFar < timeoutTime; timeWaitedSoFar += REPLY_MESSAGE_TIMEOUT_INCREMENT) {
try {
receivedMessage = client.getReceiveQueue().poll(REPLY_MESSAGE_TIMEOUT_INCREMENT, TimeUnit.MILLISECONDS);
} catch (final InterruptedException e) {
// restore the interrupt status
Thread.currentThread().interrupt();
LOGGER.warn("reception of response from server interrupted {}:{}", hostName, port, e);
throw new ApexDeploymentException(
"reception of response from server interrupted " + hostName + ':' + port, e);
}
}
// Check if response to sent message
if (receivedMessage == null) {
LOGGER.warn("no response received to sent message " + sentMessage.getAction());
throw new ApexDeploymentException("no response received to sent message " + sentMessage.getAction());
}
// Check instance is a response message
if (!(receivedMessage instanceof Response)) {
LOGGER.warn("response received from server is of incorrect type {}, should be of type {}",
receivedMessage.getClass().getName(), Response.class.getName());
throw new ApexDeploymentException("response received from server is of incorrect type "
+ receivedMessage.getClass().getName() + ", should be of type " + Response.class.getName());
}
// Cast the response message
final Response responseMessage = (Response) receivedMessage;
// Check if response to sent message
if (!responseMessage.getResponseTo().equals(sentMessage)) {
LOGGER.warn("response received is not response to sent message " + sentMessage.getAction());
throw new ApexDeploymentException(
"response received is not correct response to sent message " + sentMessage.getAction());
}
// Check if successful
if (responseMessage.isSuccessful()) {
LOGGER.debug("response received: {} message was succssful: {}", sentMessage.getAction(),
responseMessage.getMessageData());
} else {
LOGGER.debug("response received: {} message failed: {}", sentMessage.getAction(),
responseMessage.getMessageData());
}
return responseMessage;
}
}
|