Handle downtimes based on the feed topology type

flink_jobs_v2/ApiResourceManager/pom.xml

@@ -86,6 +86,12 @@
             <version>1.58</version>
             <scope>test</scope>
         </dependency>
+
+          <dependency>
+              <groupId>com.fasterxml.jackson.core</groupId>
+              <artifactId>jackson-databind</artifactId>
+              <version>2.12.6</version>
+          </dependency>
     </dependencies>
 
 </project>

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
 }

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();

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();
+    }
+}

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;
+    }
+
+
+}