From 6a2fd70fbbc7cb0733bf7da582f7e69aa1f1c03c Mon Sep 17 00:00:00 2001 From: cthermolia-grnet Date: Thu, 30 Jul 2026 10:37:26 +0300 Subject: [PATCH] handle downtimes based on feed topology --- flink_jobs_v2/ApiResourceManager/pom.xml | 6 + .../src/main/java/argo/amr/ApiResource.java | 2 +- .../java/argo/amr/ApiResourceManager.java | 2 +- .../main/java/argo/samr/KeycloakClient.java | 56 +++++ .../argo/samr/StatusApiRequestManager.java | 204 ++++++++++++++++++ .../argo/samr/StatusApiResourceManager.java | 163 ++++++++++++++ .../argo/samr/StatusApiResponseParser.java | 87 ++++++++ flink_jobs_v2/batch_multi/pom.xml | 1 - .../main/java/argo/batch/ArgoMultiJob.java | 163 +++++++++----- .../influxdb/connector/InfluxConnection.java | 1 + .../java/influxdb/connector/InfluxDBSink.java | 1 - flink_jobs_v2/pom.xml | 1 + 12 files changed, 626 insertions(+), 61 deletions(-) create mode 100644 flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/KeycloakClient.java create mode 100644 flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiRequestManager.java create mode 100644 flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiResourceManager.java create mode 100644 flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiResponseParser.java diff --git a/flink_jobs_v2/ApiResourceManager/pom.xml b/flink_jobs_v2/ApiResourceManager/pom.xml index 54922d11..93ea156f 100644 --- a/flink_jobs_v2/ApiResourceManager/pom.xml +++ b/flink_jobs_v2/ApiResourceManager/pom.xml @@ -86,6 +86,12 @@ 1.58 test + + + com.fasterxml.jackson.core + jackson-databind + 2.12.6 + diff --git a/flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResource.java b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResource.java index ebdf241c..71e21739 100644 --- a/flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResource.java +++ b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResource.java @@ -1,5 +1,5 @@ package argo.amr; public enum ApiResource { - CONFIG, OPS, METRIC, AGGREGATION, THRESHOLDS, TOPOENDPOINTS, TOPOGROUPS, WEIGHTS, DOWNTIMES, RECOMPUTATIONS, MTAGS, TENANTFEED + CONFIG, OPS, METRIC, AGGREGATION, THRESHOLDS, TOPOENDPOINTS, TOPOGROUPS, WEIGHTS, DOWNTIMES, RECOMPUTATIONS, MTAGS, TENANTFEED,FEEDTOPOLOGY } \ No newline at end of file diff --git a/flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResourceManager.java b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResourceManager.java index b48d9579..5cd7a587 100644 --- a/flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResourceManager.java +++ b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResourceManager.java @@ -573,7 +573,7 @@ public void getRemoteAll() throws UnknownHostException { this.getRemoteWeights(); } // get downtimes - this.getRemoteDowntimes(); + // this.getRemoteDowntimes(); // get recomptations this.getRemoteRecomputations(); this.getRemoteMetricTags(); diff --git a/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/KeycloakClient.java b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/KeycloakClient.java new file mode 100644 index 00000000..6712511c --- /dev/null +++ b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/KeycloakClient.java @@ -0,0 +1,56 @@ +package argo.samr; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.URL; +import java.net.URLEncoder; + +public class KeycloakClient { + + private final ObjectMapper objectMapper = new ObjectMapper(); + + public String retrieveAccessToken(String keycloakUrl,String clientId, String secret) throws Exception { + + URL url = new URL(keycloakUrl); + + HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + + conn.setRequestMethod("POST"); + conn.setRequestProperty("Content-Type", "application/x-www-form-urlencoded"); + conn.setDoOutput(true); + + String body = + "grant_type=client_credentials" + + "&client_id=" + URLEncoder.encode(clientId, "UTF-8") + + "&client_secret=" + URLEncoder.encode(secret, "UTF-8") + + "&scope=" + URLEncoder.encode("openid entitlements", "UTF-8"); + + OutputStream os = conn.getOutputStream(); + os.write(body.getBytes("UTF-8")); + os.flush(); + os.close(); + + int responseCode = conn.getResponseCode(); + System.out.println("Response Code: " + responseCode); + + BufferedReader in = new BufferedReader( + new InputStreamReader(conn.getInputStream(), "UTF-8")); + + StringBuilder response = new StringBuilder(); + String inputLine; + + while ((inputLine = in.readLine()) != null) { + response.append(inputLine); + } + + in.close(); + + JsonNode jsonNode = objectMapper.readTree(response.toString()); + return jsonNode.get("access_token").asText(); + } +} \ No newline at end of file diff --git a/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiRequestManager.java b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiRequestManager.java new file mode 100644 index 00000000..d4b7d07e --- /dev/null +++ b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiRequestManager.java @@ -0,0 +1,204 @@ +package argo.samr; + +/* + * To change this license header, choose License Headers in Project Properties. + * To change this template file, choose Tools | Templates + * and open the template in the editor. + */ + +import java.io.IOException; +import java.net.ConnectException; +import java.net.SocketTimeoutException; +import java.net.UnknownHostException; +import java.security.KeyManagementException; +import java.security.KeyStoreException; +import java.security.NoSuchAlgorithmException; + +import org.apache.http.client.fluent.Executor; +import org.apache.http.client.fluent.Request; +import org.apache.http.conn.ssl.NoopHostnameVerifier; +import org.apache.http.conn.ssl.SSLConnectionSocketFactory; +import org.apache.http.conn.ssl.TrustSelfSignedStrategy; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.ssl.SSLContextBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.net.ssl.SSLException; + +/** + * + * Establish a connection to the given url and request data + */ +public class StatusApiRequestManager { + + private String proxy; + private String token; + private int timeoutSec; + private boolean verify; + static Logger LOG = LoggerFactory.getLogger(StatusApiRequestManager.class); + + public StatusApiRequestManager(String proxy, String token, int timeoutSec, boolean verify) { + this.proxy = proxy; + this.token = token; + this.timeoutSec = timeoutSec; + this.verify = verify; + } + + public StatusApiRequestManager(String proxy, String token) { + this.proxy = proxy; + this.token = token; + this.timeoutSec = 30; + this.verify = true; + } + + /** + * Contacts remote argo-web-api based on the full url of a resource its + * content (expected in json format) + * + * @param fullURL String containing the full url representation of the + * argo-web-api resource + * @return A string representation of the resource json content + * @throws IOException + * @throws KeyStoreException + * @throws NoSuchAlgorithmException + * @throws KeyManagementException + */ + + public String getResource(String fullURL) throws UnknownHostException { + + Request r = Request.Get(fullURL) + .addHeader("Accept", "application/json") + .addHeader("Content-type", "application/json") + .addHeader("Authorization", "Bearer " + this.token); + + + if (!this.proxy.isEmpty()) { + r = r.viaProxy(proxy); + } + + r = r.connectTimeout(this.timeoutSec * 1000).socketTimeout(this.timeoutSec * 1000); + + String content = "{}"; + + try { + if (this.verify == false) { + CloseableHttpClient httpClient = HttpClients.custom().setSSLSocketFactory(selfSignedSSLF()).build(); + Executor executor = Executor.newInstance(httpClient); + content = executor.execute(r).returnContent().asString(); + httpClient.close(); + } else { + content = r.execute().returnContent().asString(); + } + } catch (UnknownHostException e) { + // DNS resolution failure — API domain does not exist + LOG.error("UnknownHostException: API endpoint not found: "+fullURL, e.getMessage()); + // Throw the exception again + throw new UnknownHostException("API domain not found: "+fullURL+ e.getMessage()); + } catch (ConnectException e) { + // Network error or API not reachable (e.g. timeout, refused connection) + LOG.error("ConnectException: Could not connect to API: "+ fullURL,e.getMessage()); + } catch (SocketTimeoutException e) { + // API is very slow or not responding + LOG.error("SocketTimeoutException: API did not respond in time: "+ fullURL,e.getMessage()); + } catch (SSLException e) { + // SSL errors (certs, handshake, etc.) + LOG.error("SSLException: SSL error while connecting to API: "+ fullURL,e.getMessage()); + } catch (IOException e) { + // General I/O error + LOG.error("IOException: General IO failure while accessing API: "+ fullURL,e.getMessage()); + } catch (Exception e) { + // Unexpected error + LOG.error("Unexpected exception: "+ e.getMessage()); + } + + return content; + } + + //isApiUp() checks if the api is up or down + public boolean isApiUp(String fullURL) { + try { + Request r = Request.Head(fullURL) // Use HEAD to check availability without downloading body + .connectTimeout(this.timeoutSec * 1000) + .socketTimeout(this.timeoutSec * 1000) + .addHeader("x-api-key", this.token); + + if (!this.proxy.isEmpty()) { + r = r.viaProxy(proxy); + } + + if (!this.verify) { + try (CloseableHttpClient httpClient = HttpClients.custom() + .setSSLSocketFactory(selfSignedSSLF()) + .build()) { + Executor executor = Executor.newInstance(httpClient); + int statusCode = executor.execute(r).returnResponse().getStatusLine().getStatusCode(); + return statusCode >= 200 && statusCode < 300; + } + } else { + int statusCode = r.execute().returnResponse().getStatusLine().getStatusCode(); + return statusCode >= 200 && statusCode < 300; + } + + } catch (UnknownHostException e) { + LOG.warn("API host not found: {}", fullURL, e); + } catch (ConnectException e) { + LOG.warn("Could not connect to API: {}", fullURL, e); + } catch (SocketTimeoutException e) { + LOG.warn("API timed out: {}", fullURL, e); + } catch (IOException e) { + LOG.warn("I/O error while checking API status: {}", fullURL, e); + } catch (Exception e) { + LOG.warn("Unexpected error while checking API status: {}", fullURL, e); + } + return false; + } + + + + /** + * Create an SSL Connection Socket Factory with a strategy to trust self + * signed certificates + */ + private SSLConnectionSocketFactory selfSignedSSLF() + throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException { + SSLContextBuilder sslBuild = new SSLContextBuilder(); + sslBuild.loadTrustMaterial(null, new TrustSelfSignedStrategy()); + return new SSLConnectionSocketFactory(sslBuild.build(), NoopHostnameVerifier.INSTANCE); + } + + public String getProxy() { + return proxy; + } + + public String getToken() { + return token; + } + + public int getTimeoutSec() { + return timeoutSec; + } + + public boolean isVerify() { + return verify; + } + + public void setProxy(String proxy) { + this.proxy = proxy; + } + + public void setToken(String token) { + this.token = token; + } + + public void setTimeoutSec(int timeoutSec) { + this.timeoutSec = timeoutSec; + } + + public void setVerify(boolean verify) { + this.verify = verify; + } + + +} diff --git a/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiResourceManager.java b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiResourceManager.java new file mode 100644 index 00000000..f129e083 --- /dev/null +++ b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiResourceManager.java @@ -0,0 +1,163 @@ +package argo.samr; + +import argo.amr.ApiResource; +import argo.avro.Downtime; + +import java.net.UnknownHostException; +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.List; + +/** + * APIResourceManager class fetches remote argo-web-api resources such as report + * configuration, profiles, topology, weights in JSON format + */ +public class StatusApiResourceManager { + + private EnumMap data = new EnumMap<>(ApiResource.class); + + private String endpoint; + private String token; + private String tenant; + private String date; + private String proxy; + + private StatusApiRequestManager requestManager; + private StatusApiResponseParser apiResponseParser; + + + public StatusApiResourceManager(String endpoint, String token) { + this.endpoint = endpoint; + this.token = token; + this.date = ""; + this.requestManager = new StatusApiRequestManager("", this.token); + + this.apiResponseParser = new StatusApiResponseParser( this.tenant); + } + + + public String getTenant() { + return tenant; + } + + public void setTenant(String tenant) { + this.tenant = tenant; + } + + public void setProxy(String proxy) { + + this.requestManager.setProxy(proxy); + } + + public String getProxy() { + return this.requestManager.getProxy(); + } + + public void setTimeoutSec(int timeOutSec) { + + this.requestManager.setTimeoutSec(timeOutSec); + } + + public void setVerify(boolean verify) { + this.requestManager.setVerify(verify); + } + + public boolean isVerify() { + return this.requestManager.isVerify(); + } + + public EnumMap getData() { + return data; + } + + public void setData(EnumMap data) { + this.data = data; + } + + public StatusApiResponseParser getApiResponseParser() { + return apiResponseParser; + } + + public void setApiResponseParser(StatusApiResponseParser apiResponseParser) { + this.apiResponseParser = apiResponseParser; + } + + + public String getEndpoint() { + return endpoint; + } + + public void setEndpoint(String endpoint) { + this.endpoint = endpoint; + } + + public String getToken() { + return token; + } + + public void setToken(String token) { + this.token = token; + } + + public String getDate() { + return date; + } + + public void setDate(String date) { + this.date = date; + } + + public StatusApiRequestManager getRequestManager() { + return requestManager; + } + + public void setRequestManager(StatusApiRequestManager requestManager) { + this.requestManager = requestManager; + } + /** + * Returns local resource (after has been retrieved) content based on + * resource type + * + * @param res + * @return The extracted items JSON value as string + */ + public String getResourceJSON(ApiResource res) { + return this.data.get(res); + } + + /** + * Retrieves the downtimes content and stores it to the enum map + */ + public void getRemoteDowntimes() throws UnknownHostException { + + String path = "https://%s/v1/automation/tenants/%s/downtimes/daily?date=%s"; + String fullURL = String.format(path, this.endpoint,this.tenant, this.date); + String content = this.requestManager.getResource(fullURL); + this.data.put(ApiResource.DOWNTIMES, this.apiResponseParser.getDowntimeJsonData(content, true)); + + } + + public void getRemoteFeedTopologyIsExternal() throws UnknownHostException { + + String path = "https://%s/v1/automation/tenants/%s/feeds/topology/is-external"; + String fullURL = String.format(path, this.endpoint,this.tenant); + String content = this.requestManager.getResource(fullURL); + this.data.put(ApiResource.FEEDTOPOLOGY, String.valueOf(this.apiResponseParser.getIsExternalFeedTopology(content))); + + } + public Downtime[] getListDowntimes() { + + List results = new ArrayList(); + if (!this.data.containsKey(ApiResource.DOWNTIMES)) { + Downtime[] rArr = new Downtime[results.size()]; + rArr = results.toArray(rArr); + } + + String content = this.data.get(ApiResource.DOWNTIMES); + results = this.apiResponseParser.getListDowntimes(content); + Downtime[] rArr = new Downtime[results.size()]; + rArr = results.toArray(rArr); + return rArr; + } + +} diff --git a/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiResponseParser.java b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiResponseParser.java new file mode 100644 index 00000000..2f484e6f --- /dev/null +++ b/flink_jobs_v2/ApiResourceManager/src/main/java/argo/samr/StatusApiResponseParser.java @@ -0,0 +1,87 @@ +package argo.samr; +/* + * To change this license header, choose License Headers in Project Properties. + * To change this template file, choose Tools | Templates + * and open the template in the editor. + */ + +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import argo.avro.Downtime; + +import java.util.ArrayList; +import java.util.List; + +/** + * Parses a given request's response + */ +public class StatusApiResponseParser { + + private String tenant; + + public StatusApiResponseParser() { + } + + public StatusApiResponseParser( String tenant) { + this.tenant = tenant; + + } + + + /** + * Extract first JSON item from data JSON array in api response + * + * @param content JSON content of the full repsonse (status + data) + * @return First available item in data array as JSON string representation + */ + public String getDowntimeJsonData(String content, boolean asArray) { + + JsonParser jsonParser = new JsonParser(); + // Grab the first - and only line of json from ops data + JsonElement jElement = jsonParser.parse(content); + JsonObject jRoot = jElement.getAsJsonObject(); + + // Get the data array and the first item + + return jRoot.toString(); + } + + public boolean getIsExternalFeedTopology(String content) { + JsonParser jsonParser = new JsonParser(); + JsonElement jElement = jsonParser.parse(content); + JsonObject jRoot = jElement.getAsJsonObject(); + return jRoot.get("external").getAsBoolean(); + + } + + /** + * Parses the Downtime content retrieved from argo-web-api and provides a + * list of Downtime avro objects to be used in the next steps of the + * pipeline + */ + public List getListDowntimes(String content) { + List results = new ArrayList(); + JsonParser jsonParser = new JsonParser(); + JsonElement jElement = jsonParser.parse(content); + JsonObject jRoot = jElement.getAsJsonObject(); + JsonArray jElements = jRoot.get("endpoints").getAsJsonArray(); + for (int i = 0; i < jElements.size(); i++) { + JsonObject jItem = jElements.get(i).getAsJsonObject(); + String hostname = jItem.get("hostname").getAsString(); + String service = jItem.get("service").getAsString(); + String startTime = jItem.get("start_time").getAsString(); + String endTime = jItem.get("end_time").getAsString(); + + Downtime d = new Downtime(hostname, service, startTime, endTime); + results.add(d); + } + return results; + + } + + +} + + diff --git a/flink_jobs_v2/batch_multi/pom.xml b/flink_jobs_v2/batch_multi/pom.xml index 3165fb1d..2fce2a33 100644 --- a/flink_jobs_v2/batch_multi/pom.xml +++ b/flink_jobs_v2/batch_multi/pom.xml @@ -191,7 +191,6 @@ language governing permissions and limitations under the License. --> 3.0.0 jar - diff --git a/flink_jobs_v2/batch_multi/src/main/java/argo/batch/ArgoMultiJob.java b/flink_jobs_v2/batch_multi/src/main/java/argo/batch/ArgoMultiJob.java index c8761ad5..985a50cc 100644 --- a/flink_jobs_v2/batch_multi/src/main/java/argo/batch/ArgoMultiJob.java +++ b/flink_jobs_v2/batch_multi/src/main/java/argo/batch/ArgoMultiJob.java @@ -2,40 +2,11 @@ import argo.amr.ApiResource; import argo.amr.ApiResourceManager; -import argo.ar.CalcEndpointAR; -import argo.ar.CalcGroupAR; -import argo.ar.CalcServiceAR; -import argo.ar.EndpointAR; -import argo.ar.EndpointGroupAR; -import argo.ar.ServiceAR; -import argo.avro.Downtime; -import argo.avro.GroupEndpoint; -import argo.avro.GroupGroup; -import argo.avro.MetricData; -import argo.avro.MetricProfile; -import argo.avro.Weight; -import org.apache.flink.core.fs.FileSystem; -import profilesmanager.RecomputationsManager; -import trends.calculations.ServiceTrends; -import trends.flipflops.ZeroServiceFlipFlopFilter; -import trends.status.EndpointTrendsCounter; -import trends.calculations.CalcEndpointFlipFlopTrends; -import trends.calculations.CalcGroupFlipFlopTrends; -import trends.calculations.CalcMetricFlipFlopTrends; -import trends.calculations.CalcServiceFlipFlopTrends; -import trends.calculations.EndpointTrends; -import trends.calculations.GroupTrends; -import trends.flipflops.MapEndpointTrends; -import trends.flipflops.MapGroupTrends; -import trends.flipflops.MapMetricTrends; -import trends.flipflops.MapServiceTrends; -import trends.calculations.MetricTrends; -import trends.calculations.MongoTrendsOutput; -import trends.status.StatusAndDurationFilter; -import trends.calculations.Trends; -import trends.flipflops.ZeroEndpointFlipFlopFilter; -import trends.flipflops.ZeroGroupFlipFlopFilter; -import trends.flipflops.ZeroMetricFlipFlopFilter; +import argo.ar.*; +import argo.avro.*; +import argo.samr.KeycloakClient; +import com.esotericsoftware.minlog.Log; +import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.operators.Order; import org.apache.flink.api.java.DataSet; @@ -44,22 +15,24 @@ import org.apache.flink.api.java.operators.DataSource; import org.apache.flink.api.java.tuple.Tuple8; import org.apache.flink.api.java.utils.ParameterTool; +import org.apache.flink.core.fs.FileSystem; import org.apache.flink.core.fs.Path; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.slf4j.MDC; +import profilesmanager.RecomputationsManager; import profilesmanager.ReportManager; -import trends.status.GroupTrendsCounter; -import trends.status.MetricTrendsCounter; -import trends.status.ServiceTrendsCounter; +import argo.samr.StatusApiResourceManager; +import trends.calculations.*; +import trends.flipflops.*; +import trends.status.*; import utils.Utils; +import java.net.UnknownHostException; import java.util.*; -import org.apache.flink.api.common.JobID; -import org.joda.time.DateTime; -import org.joda.time.DateTimeZone; -import org.slf4j.MDC; - /** * Implements an ARGO Status Batch Job in flink *

@@ -89,12 +62,17 @@ * If not set tags calculations will be defined from the report *

*

- * --source-data(Optinal): tenant, all, feeds-only . Defines from where the source data can be received. if not defined or is tenant the data are received from the tenant (the tenant does not combine data), if value is feeds-only the data are received from + * --source-data(Optional): tenant, all, feeds-only . Defines from where the source data can be received. if not defined or is tenant the data are received from the tenant (the tenant does not combine data), if value is feeds-only the data are received from * the tenant list defined in the feeds data(the tenant is acting as a gatherer without data on it's own), if is all the source data are received both from the tenant and the tenants list in the feeds data (the tenant already exists and has data to combine) *

*

- * --source-topo(Optinal): tenant, all . Defines from where the source topology can be received. if not defined or is tenant the data are received from the tenant (the tenant either is not combining data or acts as a gatherer without topologies on its own), + * --source-topo(Optional): tenant, all . Defines from where the source topology can be received. if not defined or is tenant the data are received from the tenant (the tenant either is not combining data or acts as a gatherer without topologies on its own), * , if is all the source data are received both from the tenant and the tenants list in the feeds data and needs a parameter to be added to the api request to receive the topology (the tenant has it's own topologies) + * --check-feed(Optional): true, false. Defines if feed topology will be checked to decide the source of downtimes. if false, argo-web-api is the source, if true there will be a check and if feed type is external argo-web-api is the source else is the monitoring status api + *--keycloak.url(Optional), the keycloak url the compute engine service accesses to obtain a token + *--status.api.endpoint(Optional), the monitoring status api url to be the source of downtimes + *--compute.engine.secret(Optional), the secret of the compute engine service + * --compute.engine.client.id(Optional), the client id of the compute engine service */ public class ArgoMultiJob { @@ -112,6 +90,7 @@ public class ArgoMultiJob { private static boolean calcStatusTrends = false; private static boolean calcFlipFlops = false; private static boolean calcTagTrends = false; + private static StatusApiResourceManager samr; enum Combined { TENANT("tenant"), @@ -190,6 +169,48 @@ public static void main(String[] args) throws Exception { amr.setDate(runDate); amr.getRemoteAll(); + DataSet downDS = env.fromElements(new Downtime()); + + boolean useStatusApi = false; + + if (params.getBoolean("check.feed")) { + + if (!hasStatusApiParams(params)) { + Log.error("Not all parameters required to connect to the Monitoring Status API are defined."); + System.exit(0); + } + + String statusApiToken = keycloakToken(params.get("keycloak.url"), params.get("compute.engine.client.id"), params.get("compute.engine.secret")); + String statusApiEndpoint = params.getRequired("status.api.endpoint"); + + connectStatusApiClient(statusApiEndpoint, statusApiToken, amr.getTenant()); + + if (params.has("status.api.proxy")) { + samr.setProxy(params.get("status.api.proxy")); + } + + samr.getRemoteFeedTopologyIsExternal(); + + boolean isExternal = Boolean.parseBoolean( + samr.getData().get(ApiResource.FEEDTOPOLOGY)); + + useStatusApi = !isExternal; + } + + Downtime[] downtimes; + if (useStatusApi) { + // downDS = fetchDowntimes(env); + samr.getRemoteDowntimes(); + downtimes = samr.getListDowntimes(); + } else { + amr.getRemoteDowntimes(); + downtimes = amr.getListDowntimes(); + } + + if (downtimes.length > 0) { + downDS = env.fromElements(downtimes); + } + DataSource confDS = env.fromElements(amr.getResourceJSON(ApiResource.CONFIG)); // Get conf data List confData = confDS.collect(); @@ -209,12 +230,12 @@ public static void main(String[] args) throws Exception { } RecomputationsManager.loadJsonString(recDS.collect()); - DataSource>> metricRecomputedDS=env.fromElements(RecomputationsManager.metricRecomputationItems); - DataSource>> endpointRecomputedDS=env.fromElements(RecomputationsManager.endpointRecomputationItems); - DataSource>> serviceRecomputedDS=env.fromElements(RecomputationsManager.serviceRecomputationItems); - DataSource>> groupRecomputedDS=env.fromElements(RecomputationsManager.groupRecomputationItems); - DataSource>>> monEngineRecDS=env.fromElements(RecomputationsManager.monEngines); - DataSource>>> groupsRecDS=env.fromElements(RecomputationsManager.groups); + DataSource>> metricRecomputedDS = env.fromElements(RecomputationsManager.metricRecomputationItems); + DataSource>> endpointRecomputedDS = env.fromElements(RecomputationsManager.endpointRecomputationItems); + DataSource>> serviceRecomputedDS = env.fromElements(RecomputationsManager.serviceRecomputationItems); + DataSource>> groupRecomputedDS = env.fromElements(RecomputationsManager.groupRecomputationItems); + DataSource>>> monEngineRecDS = env.fromElements(RecomputationsManager.monEngines); + DataSource>>> groupsRecDS = env.fromElements(RecomputationsManager.groups); DataSource mtagsDS = env.fromElements(""); if (amr.getResourceJSON(ApiResource.MTAGS) != null) { @@ -228,7 +249,6 @@ public static void main(String[] args) throws Exception { weightDS = env.fromElements(amr.getListWeights()); } - DataSet downDS = env.fromElements(new Downtime()); // begin with empty threshold datasource DataSource thrDS = env.fromElements(""); // check if report information from argo-web-api contains a threshold profile ID @@ -252,10 +272,6 @@ public static void main(String[] args) throws Exception { ggpDS = env.fromElements(amr.getListGroupGroups()); } - Downtime[] listDowntimes = amr.getListDowntimes(); - if (listDowntimes.length > 0) { - downDS = env.fromElements(amr.getListDowntimes()); - } List tenantList = new ArrayList<>(); if (sourceData.equals(Combined.TENANT) || sourceData.equals(Combined.ALL)) { @@ -285,7 +301,6 @@ public static void main(String[] args) throws Exception { } - DataSet allMetricData = null; for (Path[] path : tenantPaths) { @@ -348,7 +363,7 @@ public static void main(String[] args) throws Exception { DataSet mdataTrimDS = allMetricData.flatMap(new PickEndpoints(params)) .withBroadcastSet(mpsDS, "mps").withBroadcastSet(egpDS, "egp").withBroadcastSet(ggpDS, "ggp") .withBroadcastSet(confDS, "conf").withBroadcastSet(thrDS, "thr") - .withBroadcastSet(opsDS, "ops").withBroadcastSet(apsDS, "aps").withBroadcastSet(monEngineRecDS,"rec"); + .withBroadcastSet(opsDS, "ops").withBroadcastSet(apsDS, "aps").withBroadcastSet(monEngineRecDS, "rec"); // Combine prev and todays metric data with the generated missing metric // data @@ -560,6 +575,35 @@ public static void main(String[] args) throws Exception { } + private static String keycloakToken(String keycloakUrl, String clientId, String secret) { + KeycloakClient keycloakClient = new KeycloakClient(); + try { + return keycloakClient.retrieveAccessToken(keycloakUrl, clientId, secret); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + private static void connectStatusApiClient(String endpoint, String token, String tenant) { + samr = new StatusApiResourceManager(endpoint, token); + + // fetch + samr.setTenant(tenant); + samr.setDate(runDate); + + } + + //private static DataSet fetchDowntimes(ExecutionEnvironment env) throws UnknownHostException { + // DataSet downDS = env.fromElements(new argo.avro.Downtime()); + //samr.getRemoteDowntimes(); +// Downtime[] listDowntimes = samr.getListDowntimes(); +// if (listDowntimes.length > 0) { +// downDS = env.fromElements(listDowntimes); +// } +// return downDS; + + //} + private static void filterByStatusAndWriteMongo(MongoTrendsOutput.TrendsType mongoTrendsType, String uri, DataSet> data, String status) { DataSet> filteredData = data.filter(new StatusAndDurationFilter(status)); //filter dataset by status type and status appearances>0 @@ -648,4 +692,9 @@ private static void configJID() {//config the JID in the log4j.properties MDC.put("JID", jobId); } + + private static boolean hasStatusApiParams(ParameterTool params) { + + return params.has("keycloak.url") && params.has("status.api.endpoint") && params.has("compute.engine.secret") && params.has("compute.engine.client.id"); + } } diff --git a/flink_jobs_v2/influxdb.connector/src/main/java/influxdb/connector/InfluxConnection.java b/flink_jobs_v2/influxdb.connector/src/main/java/influxdb/connector/InfluxConnection.java index b369eecf..9e7d9eff 100644 --- a/flink_jobs_v2/influxdb.connector/src/main/java/influxdb/connector/InfluxConnection.java +++ b/flink_jobs_v2/influxdb.connector/src/main/java/influxdb/connector/InfluxConnection.java @@ -76,6 +76,7 @@ private OkHttpClient.Builder buildHttpClient() { OkHttpClient.Builder builder = new OkHttpClient.Builder(); if (verifySsl) { + LOG.info("VERIFY SSL to true-- "+verifySsl); return builder; } diff --git a/flink_jobs_v2/influxdb.connector/src/main/java/influxdb/connector/InfluxDBSink.java b/flink_jobs_v2/influxdb.connector/src/main/java/influxdb/connector/InfluxDBSink.java index 013e55d9..fb1da1f1 100644 --- a/flink_jobs_v2/influxdb.connector/src/main/java/influxdb/connector/InfluxDBSink.java +++ b/flink_jobs_v2/influxdb.connector/src/main/java/influxdb/connector/InfluxDBSink.java @@ -67,7 +67,6 @@ public void open(Configuration parameters) throws Exception { url = endpoint + ":" + port; if(params.has("influx.verify")) { influx_verify = params.getBoolean("influx.verify"); - System.out.println("VERIFY--- "+influx_verify); } LOG.info("Opening InfluxDB sink for {}", url); connection = new InfluxConnection(url, token, org, bucket, proxyURL, proxyPORT,influx_verify); diff --git a/flink_jobs_v2/pom.xml b/flink_jobs_v2/pom.xml index aae0ad21..6402a92f 100644 --- a/flink_jobs_v2/pom.xml +++ b/flink_jobs_v2/pom.xml @@ -21,6 +21,7 @@ ams_ingest_sync ams-connector influxdb.connector +