From d63199d1546d1d05ca6253858c483bd714eaa652 Mon Sep 17 00:00:00 2001 From: Tim Jones Date: Tue, 17 Apr 2018 11:13:19 +0100 Subject: Add stop flag to allow VES agent to shut down Change-Id: I314b527ce624cc4bbb7df0029ecad2023ac0b5ea Issue-ID: VNFSDK-257 Signed-off-by: Tim Jones --- .../java/evel_javalibrary/att/com/AgentMain.java | 578 +++++++++++---------- .../att/com/EvelScalingMeasurement.java | 18 + .../java/evel_javalibrary/att/com/RingBuffer.java | 4 + 3 files changed, 327 insertions(+), 273 deletions(-) (limited to 'veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att') diff --git a/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/AgentMain.java b/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/AgentMain.java index 23a1b4d..5e21c6c 100644 --- a/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/AgentMain.java +++ b/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/AgentMain.java @@ -57,221 +57,238 @@ import org.apache.log4j.BasicConfigurator; public class AgentMain { - -/**************************************************************************//** - * Error codes - * - * Error codes for EVEL low level interface - *****************************************************************************/ -public enum EVEL_ERR_CODES { - EVEL_SUCCESS, /** The operation was successful. */ - EVEL_ERR_GEN_FAIL, /** Non-specific failure. */ - EVEL_CURL_LIBRARY_FAIL, /** A cURL library operation failed. */ - EVEL_PTHREAD_LIBRARY_FAIL, /** A Posix threads operation failed. */ - EVEL_OUT_OF_MEMORY, /** A memory allocation failure occurred. */ - EVEL_EVENT_BUFFER_FULL, /** Too many events in the ring-buffer. */ - EVEL_EVENT_HANDLER_INACTIVE, /** Attempt to raise event when inactive. */ - EVEL_NO_METADATA, /** Failed to retrieve OpenStack metadata. */ - EVEL_BAD_METADATA, /** OpenStack metadata invalid format. */ - EVEL_BAD_JSON_FORMAT, /** JSON failed to parse correctly. */ - EVEL_JSON_KEY_NOT_FOUND, /** Failed to find the specified JSON key. */ - EVEL_MAX_ERROR_CODES /** Maximum number of valid error codes. */ -} - - private static final Logger logger = Logger.getLogger(AgentMain.class); - - private static String url = null; - private static URL vesurl = null; - private static URL vesbatchurl = null; - private static HttpURLConnection con = null; - private static String userpass = null; - private static String keystore_pth = null; - private static String jks_passw = null; - private static String key_passw = null; - private static String version = "5"; - - /* RingBuffer to forward messages on sending AgentDispatcher thread */ - private static RingBuffer ringb = new RingBuffer(100); - - Thread thr; - - /* AgentDispatcher loops on messages in RingBuffer and POSTs them - * to external Collector - */ + + /**************************************************************************//** + * Error codes + * + * Error codes for EVEL low level interface + *****************************************************************************/ + public enum EVEL_ERR_CODES { + EVEL_SUCCESS, /** The operation was successful. */ + EVEL_ERR_GEN_FAIL, /** Non-specific failure. */ + EVEL_CURL_LIBRARY_FAIL, /** A cURL library operation failed. */ + EVEL_PTHREAD_LIBRARY_FAIL, /** A Posix threads operation failed. */ + EVEL_OUT_OF_MEMORY, /** A memory allocation failure occurred. */ + EVEL_EVENT_BUFFER_FULL, /** Too many events in the ring-buffer. */ + EVEL_EVENT_HANDLER_INACTIVE, /** Attempt to raise event when inactive. */ + EVEL_NO_METADATA, /** Failed to retrieve OpenStack metadata. */ + EVEL_BAD_METADATA, /** OpenStack metadata invalid format. */ + EVEL_BAD_JSON_FORMAT, /** JSON failed to parse correctly. */ + EVEL_JSON_KEY_NOT_FOUND, /** Failed to find the specified JSON key. */ + EVEL_MAX_ERROR_CODES /** Maximum number of valid error codes. */ + } + + private static final Logger logger = Logger.getLogger(AgentMain.class); + + private static String url = null; + private static URL vesurl = null; + private static URL vesbatchurl = null; + private static HttpURLConnection con = null; + private static String userpass = null; + private static String keystore_pth = null; + private static String jks_passw = null; + private static String key_passw = null; + private static String version = "5"; + + /* RingBuffer to forward messages on sending AgentDispatcher thread */ + private static RingBuffer ringb = new RingBuffer(100); + private static volatile boolean should_stop = false; + + private static Thread thr; + + /* AgentDispatcher loops on messages in RingBuffer and POSTs them + * to external Collector + */ private static class AgentDispatcher implements Runnable { - - private String readStream(InputStream stream) throws Exception { - StringBuilder builder = new StringBuilder(); - try (BufferedReader in = new BufferedReader(new InputStreamReader(stream))) { - String line; - while ((line = in.readLine()) != null) { - builder.append(line); // + "\r\n"(no need, json has no line breaks!) - } - in.close(); - } - logger.error("Resp: " + builder.toString()); - //System.out.println("Resp: " + builder.toString()); - return builder.toString(); - } - - // Create a trust manager that does not validate certificate chains - TrustManager[] trustAllCerts = new TrustManager[] { - new X509TrustManager() { - public java.security.cert.X509Certificate[] getAcceptedIssuers() { - return new java.security.cert.X509Certificate[0]; - } - public void checkClientTrusted( - java.security.cert.X509Certificate[] certs, String authType) { - } - public void checkServerTrusted( - java.security.cert.X509Certificate[] certs, String authType) { - } - } - }; - - public void run() { - - String datatosend=null; - for(;;){ - EvelObject tosend = ringb.take(); - if( tosend != null && ((datatosend = (String) tosend.datastr) != null)) - { - //process data - logger.trace(url + "Got an event size "+datatosend.length()); - logger.trace(datatosend); - - try { - - if( tosend.type == false) - con = (HttpURLConnection) vesurl.openConnection(); - else - con = (HttpURLConnection) vesbatchurl.openConnection(); - - if (con instanceof HttpsURLConnection) { - HttpsURLConnection httpsConnection = (HttpsURLConnection) con; - - try { - - SSLContext sc = SSLContext.getInstance("TLSv1.2"); - /* Get the JKS contents */ - if( !keystore_pth.isEmpty() && !jks_passw.isEmpty() && !key_passw.isEmpty() ) - { - final KeyStore keyStore = KeyStore.getInstance("JKS"); - try (final InputStream is = new FileInputStream(keystore_pth)) { - keyStore.load(is, jks_passw.toCharArray()); - } - final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory - .getDefaultAlgorithm()); - kmf.init(keyStore, key_passw.toCharArray()); - final TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory - .getDefaultAlgorithm()); - tmf.init(keyStore); - sc.init(kmf.getKeyManagers(), tmf.getTrustManagers(), new java.security.SecureRandom()); - } - else - { - // Init the SSLContext with a TrustManager[] and SecureRandom() - sc.init(null, trustAllCerts, new java.security.SecureRandom()); - } + + private String readStream(InputStream stream) throws Exception { + StringBuilder builder = new StringBuilder(); + try (BufferedReader in = new BufferedReader(new InputStreamReader(stream))) { + String line; + while ((line = in.readLine()) != null) { + builder.append(line); // + "\r\n"(no need, json has no line breaks!) + } + in.close(); + } + logger.error("Resp: " + builder.toString()); + //System.out.println("Resp: " + builder.toString()); + return builder.toString(); + } + + // Create a trust manager that does not validate certificate chains + TrustManager[] trustAllCerts = new TrustManager[] { + new X509TrustManager() { + public java.security.cert.X509Certificate[] getAcceptedIssuers() { + return new java.security.cert.X509Certificate[0]; + } + public void checkClientTrusted( + java.security.cert.X509Certificate[] certs, String authType) { + } + public void checkServerTrusted( + java.security.cert.X509Certificate[] certs, String authType) { + } + } + }; + + private void send_object(EvelObject tosend){ + String datatosend = (String) tosend.datastr; + //process data + logger.trace(url + "Got an event size "+datatosend.length()); + logger.trace(datatosend); + + try { + + if( tosend.type == false) + con = (HttpURLConnection) vesurl.openConnection(); + else + con = (HttpURLConnection) vesbatchurl.openConnection(); + + if (con instanceof HttpsURLConnection) { + HttpsURLConnection httpsConnection = (HttpsURLConnection) con; + + try { + + SSLContext sc = SSLContext.getInstance("TLSv1.2"); + /* Get the JKS contents */ + if( !keystore_pth.isEmpty() && !jks_passw.isEmpty() && !key_passw.isEmpty() ) + { + final KeyStore keyStore = KeyStore.getInstance("JKS"); + try (final InputStream is = new FileInputStream(keystore_pth)) { + keyStore.load(is, jks_passw.toCharArray()); + } + final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory + .getDefaultAlgorithm()); + kmf.init(keyStore, key_passw.toCharArray()); + final TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory + .getDefaultAlgorithm()); + tmf.init(keyStore); + sc.init(kmf.getKeyManagers(), tmf.getTrustManagers(), new java.security.SecureRandom()); + } + else + { + // Init the SSLContext with a TrustManager[] and SecureRandom() + sc.init(null, trustAllCerts, new java.security.SecureRandom()); + } httpsConnection.setDefaultHostnameVerifier(new HostnameVerifier() + { + public boolean verify(String hostname, SSLSession session) { - public boolean verify(String hostname, SSLSession session) - { - return true; - } + return true; + } }); - httpsConnection.setSSLSocketFactory(sc.getSocketFactory()); - con = httpsConnection; - - } - catch (final Exception exc) { - exc.printStackTrace(); - logger.error("SSL/TLS connection error"); - } - } - - //add reuqest header - con.setRequestMethod("POST"); - // No caching, we want the real thing. - con.setUseCaches (false); - // Specify the content type. - con.setRequestProperty("Content-Type", "application/json"); - con.setInstanceFollowRedirects( false ); - //Basic username password authentication - String basicAuth = "Basic " + javax.xml.bind.DatatypeConverter.printBase64Binary(userpass.getBytes("UTF-8")); - con.setRequestProperty ("Authorization", basicAuth); - - con.setReadTimeout(15000 /* milliseconds */); - con.setConnectTimeout(15000 /* milliseconds */); - // Send post request - con.setDoOutput(true); - con.setDoInput(true); - - con.setFixedLengthStreamingMode(datatosend.length()); - OutputStream os = con.getOutputStream(); - BufferedWriter writer = new BufferedWriter( - new OutputStreamWriter(os, "UTF-8")); - //Call writer POST - writer.write(datatosend); - writer.flush(); - writer.close(); - os.close(); - //Handle the response code for POST request - int respCode = con.getResponseCode(); - logger.trace(url + "Connection HTTP Response code :"+respCode); - if(respCode < HttpURLConnection.HTTP_OK ) { - logger.trace(url + " **INFO**"); - } - else if(respCode >= HttpURLConnection.HTTP_OK && respCode < HttpURLConnection.HTTP_MULT_CHOICE ) - { + httpsConnection.setSSLSocketFactory(sc.getSocketFactory()); + con = httpsConnection; + + } + catch (final Exception exc) { + exc.printStackTrace(); + logger.error("SSL/TLS connection error"); + } + } + + //add reuqest header + con.setRequestMethod("POST"); + // No caching, we want the real thing. + con.setUseCaches (false); + // Specify the content type. + con.setRequestProperty("Content-Type", "application/json"); + con.setInstanceFollowRedirects( false ); + //Basic username password authentication + String basicAuth = "Basic " + javax.xml.bind.DatatypeConverter.printBase64Binary(userpass.getBytes("UTF-8")); + con.setRequestProperty ("Authorization", basicAuth); + + con.setReadTimeout(15000 /* milliseconds */); + con.setConnectTimeout(15000 /* milliseconds */); + // Send post request + con.setDoOutput(true); + con.setDoInput(true); + + con.setFixedLengthStreamingMode(datatosend.length()); + OutputStream os = con.getOutputStream(); + BufferedWriter writer = new BufferedWriter( + new OutputStreamWriter(os, "UTF-8")); + //Call writer POST + writer.write(datatosend); + writer.flush(); + writer.close(); + os.close(); + //Handle the response code for POST request + int respCode = con.getResponseCode(); + logger.trace(url + "Connection HTTP Response code :"+respCode); + if(respCode < HttpURLConnection.HTTP_OK ) { + logger.trace(url + " **INFO**"); + } + else if(respCode >= HttpURLConnection.HTTP_OK && respCode < HttpURLConnection.HTTP_MULT_CHOICE ) + { logger.trace(url + " **OK**"); - } - else if(respCode >= HttpURLConnection.HTTP_MULT_CHOICE && respCode < HttpURLConnection.HTTP_BAD_REQUEST ) - { - logger.warn(url + " **REDIRECTION**"); - } - else if(respCode >= HttpURLConnection.HTTP_BAD_REQUEST ) - { - logger.warn(url + " **SERVER ERROR**"); + } + else if(respCode >= HttpURLConnection.HTTP_MULT_CHOICE && respCode < HttpURLConnection.HTTP_BAD_REQUEST ) + { + logger.warn(url + " **REDIRECTION**"); + } + else if(respCode >= HttpURLConnection.HTTP_BAD_REQUEST ) + { + logger.warn(url + " **SERVER ERROR**"); InputStream es = con.getErrorStream(); readStream(es); - } - - - } catch (IOException e) { - e.printStackTrace(); - } catch (Exception e) { - e.printStackTrace(); - } - - } - else - { - logger.trace(url + "Waiting for events"); - try { - Thread.sleep(5); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - logger.trace("Interrupted on "+url); - System.out.println("Interrupted on "+url); - //e.printStackTrace(); - } - } - }//end for - }//end run - }//end AgentDispatcher + } + + + } catch (IOException e) { + e.printStackTrace(); + } catch (Exception e) { + e.printStackTrace(); + } + } + + public void run() { + + String datatosend=null; + while (!should_stop){ + EvelObject tosend = ringb.take(); + if( tosend != null && ((datatosend = (String) tosend.datastr) != null)) + { + send_object(tosend); + } + else + { + logger.trace(url + "Waiting for events"); + try { + Thread.sleep(5); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.trace("Interrupted on "+url); + System.out.println("Interrupted on "+url); + //e.printStackTrace(); + } + } + }//end while + + logger.trace("Caught stop event, emptying buffer before exiting"); + + while (!ringb.isEmpty()){ + EvelObject tosend = ringb.take(); + if( tosend != null && ((datatosend = (String) tosend.datastr) != null)) + { + send_object(tosend); + } + } + logger.trace("Send buffer is empty, shutting down"); + }//end run + }//end AgentDispatcher + // Validate URL public static boolean isValidURL(String urlStr) { try { - URL url = new URL(urlStr); - return true; + URL url = new URL(urlStr); + return true; } catch (MalformedURLException e) { return false; } } - + /**************************************************************************//** * Library initialization. * @@ -293,8 +310,8 @@ public enum EVEL_ERR_CODES { * @retval ::EVEL_ERR_CODES On failure. *****************************************************************************/ public static EVEL_ERR_CODES evel_initialize( - String event_api_url, - int port, + String event_api_url, + int port, String path, String topic, String username, @@ -304,67 +321,76 @@ public enum EVEL_ERR_CODES { String key_password, Level level) throws IOException { - EVEL_ERR_CODES rc = EVEL_ERR_CODES.EVEL_SUCCESS; - - EVEL_ENTER(); - - BasicConfigurator.configure(); - - /***************************************************************************/ - /* Check assumptions. */ - /***************************************************************************/ - assert(event_api_url != null); - assert(port > 1024); - assert(username != null); - - logger.setLevel(level); - - if( !isValidURL(event_api_url) ){ - System.out.println("Invalid Event API URL"); - logger.error("Invalid Event API URL"); - rc = EVEL_ERR_CODES.EVEL_ERR_GEN_FAIL; - System.exit(1); - } - - if(path == null){ - path = ""; - } else { - version += "/example_vnf"; - } - - keystore_pth = keystore_path; - jks_passw = jks_password; - key_passw = key_password; - - url = event_api_url+":"+Integer.toString(port)+path+"/eventListener/v"+version; - vesurl = null; - try { - vesurl = new URL(url); - vesbatchurl = new URL(url+"/eventBatch"); - } catch (MalformedURLException e) { - logger.info("Error in url input"); - e.printStackTrace(); - System.exit(1); - } - userpass = username + ":" + password; + EVEL_ERR_CODES rc = EVEL_ERR_CODES.EVEL_SUCCESS; + + EVEL_ENTER(); + + BasicConfigurator.configure(); + + /***************************************************************************/ + /* Check assumptions. */ + /***************************************************************************/ + assert(event_api_url != null); + assert(port > 1024); + assert(username != null); + + logger.setLevel(level); + + if( !isValidURL(event_api_url) ){ + System.out.println("Invalid Event API URL"); + logger.error("Invalid Event API URL"); + rc = EVEL_ERR_CODES.EVEL_ERR_GEN_FAIL; + System.exit(1); + } + + if(path == null){ + path = ""; + } else { + version += "/example_vnf"; + } + + keystore_pth = keystore_path; + jks_passw = jks_password; + key_passw = key_password; + + url = event_api_url+":"+Integer.toString(port)+path+"/eventListener/v"+version; + vesurl = null; + try { + vesurl = new URL(url); + vesbatchurl = new URL(url+"/eventBatch"); + } catch (MalformedURLException e) { + logger.info("Error in url input"); + e.printStackTrace(); + System.exit(1); + } + userpass = username + ":" + password; logger.info("Starting Agent Dispatcher thread"); - Thread t = new Thread(new AgentDispatcher()); - t.start(); - + thr = new Thread(new AgentDispatcher()); + thr.start(); + EVEL_EXIT(); - return rc; - + return rc; + + } + + public static void evel_shutdown() { + if(should_stop){ + logger.warn("evel_shutdown was called whilst the agent was already stopping - this has no effect"); + } + should_stop = true; + logger.debug("Called evel_shutdown, will shutdown when the buffer is empty"); + } + + + private static void EVEL_EXIT() { + logger.trace("Exit VES Agent Main"); + } + + private static void EVEL_ENTER() { + logger.trace("Enter VES Agent Main"); } - - private static void EVEL_EXIT() { - logger.trace("Exit VES Agent Main"); - } - - private static void EVEL_ENTER() { - logger.trace("Enter VES Agent Main"); - } - + /**************************************************************************//** * Handle user formatted post message * @@ -377,16 +403,20 @@ public enum EVEL_ERR_CODES { * @retval boolean True On successful acceptance False on failure *****************************************************************************/ - public static boolean evel_post_event(EvelHeader obj) + public static boolean evel_post_event(EvelHeader obj) { - String data = obj.evel_json_encode_event().toString(); - EvelObject myobj = new EvelObject(data,false); - boolean ret = ringb.put(myobj); - logger.info("Evel Post event ret:"+ret); - return ret; + if (should_stop){ + logger.error("evel_post_event called whilst agent is shutting down - event will not be posted"); + return false; + } + String data = obj.evel_json_encode_event().toString(); + EvelObject myobj = new EvelObject(data,false); + boolean ret = ringb.put(myobj); + logger.info("Evel Post event ret:"+ret); + return ret; } - + /**************************************************************************//** * Handle user formatted post message * @@ -399,14 +429,16 @@ public enum EVEL_ERR_CODES { * @retval boolean True On successful acceptance False on failure *****************************************************************************/ - public static boolean evel_post_event(EvelBatch obj) + public static boolean evel_post_event(EvelBatch obj) { - String data = obj.evel_json_encode_event().toString(); - EvelObject myobj = new EvelObject(data,true); - boolean ret = ringb.put(myobj); - logger.info("Evel Post batch event ret:"+ret); - return ret; + if (should_stop){ + logger.error("evel_post_event called whilst agent is shutting down - event will not be posted"); + return false; + } + String data = obj.evel_json_encode_event().toString(); + EvelObject myobj = new EvelObject(data,true); + boolean ret = ringb.put(myobj); + logger.info("Evel Post batch event ret:"+ret); + return ret; } - - } diff --git a/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/EvelScalingMeasurement.java b/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/EvelScalingMeasurement.java index 1312eed..6c39d13 100644 --- a/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/EvelScalingMeasurement.java +++ b/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/EvelScalingMeasurement.java @@ -2088,6 +2088,8 @@ public class EvelScalingMeasurement extends EvelHeader { { EVEL_ENTER(); + LOGGER.warn("Called evel_measurement_vnfc_scaling_metric_set with 2 arguments. The first argument is unused and callers should use the single-parameter version instead"); + /***************************************************************************/ /* Check preconditions. */ /***************************************************************************/ @@ -2101,6 +2103,22 @@ public class EvelScalingMeasurement extends EvelHeader { EVEL_EXIT(); } + public void evel_measurement_vnfc_scaling_metric_set(int scaling_metric) + { + EVEL_ENTER(); + + /***************************************************************************/ + /* Check preconditions. */ + /***************************************************************************/ + assert(event_domain == EvelHeader.DOMAINS.EVEL_DOMAIN_MEASUREMENT); + assert(scaling_metric >= 0.0); + + vnfc_scaling_metric.SetValuePr( + scaling_metric, + "VNFC Scaling Metric"); + EVEL_EXIT(); + } + /**************************************************************************//** * Create a new Latency Bucket to be added to a Measurement event. * diff --git a/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/RingBuffer.java b/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/RingBuffer.java index 460e84a..872160f 100644 --- a/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/RingBuffer.java +++ b/veslibrary/ves_javalibrary/evel_javalib2/src/main/java/evel_javalibrary/att/com/RingBuffer.java @@ -57,6 +57,10 @@ public class RingBuffer { this.available = 0; } + public boolean isEmpty() { + return this.remainingCapacity() == this.capacity; + } + //returns available capacity public int remainingCapacity() { return this.capacity - this.available; -- cgit 1.2.3-korg