Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ARGO-4879 Streaming job fails due to null pointer exception when argo web-api is unreachable #430

Open
wants to merge 1 commit into
base: devel
Choose a base branch
from
Open

ARGO-4879 Streaming job fails due to null pointer exception when argo web-api is unreachable #430

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 86 additions & 23 deletions flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/ApiResourceManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ public class ApiResourceManager {
private boolean isCombined;
//private boolean verify;
//private int timeoutSec;
private boolean shouldRepeat;

public ApiResourceManager(String endpoint, String token) {
this.endpoint = endpoint;
Expand All @@ -64,8 +65,10 @@ public ApiResourceManager(String endpoint, String token) {
this.egroup = "";
this.requestManager = new RequestManager("", this.token);
this.apiResponseParser = new ApiResponseParser(this.reportName, this.metricID, this.aggregationID, this.opsID, this.threshID, this.tenant, this.egroup);

}


public void setProxy(String proxy) {

this.requestManager.setProxy(proxy);
Expand Down Expand Up @@ -196,6 +199,15 @@ public void setEgroup(String egroup) {
this.egroup = egroup;
}

public boolean isShouldRepeat() {
return shouldRepeat;
}

public void setShouldRepeat(boolean shouldRepeat) {
this.shouldRepeat = shouldRepeat;
}


/**
* Retrieves the remote report configuration based on reportID main class
* attribute and stores the content in the enum map
Expand All @@ -204,6 +216,11 @@ public void getRemoteConfig() {
String path = "https://%s/api/v2/reports/%s";
String fullURL = String.format(path, this.endpoint, this.reportID);
String content = this.requestManager.getResource(fullURL);
if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.CONFIG, this.apiResponseParser.getJsonData(content, false));

}
Expand All @@ -217,7 +234,10 @@ public void getRemoteMetric() {
String path = "https://%s/api/v2/metric_profiles/%s?date=%s";
String fullURL = String.format(path, this.endpoint, this.metricID, this.date);
String content = this.requestManager.getResource(fullURL);

if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}
this.data.put(ApiResource.METRIC, this.apiResponseParser.getJsonData(content, false));

}
Expand All @@ -244,8 +264,10 @@ public MetricProfile[] getNewEntriesMetrics() throws ParseException {

String path = "https://%s/api/v2/metric_profiles/%s?date=%s";
String fullURL = String.format(path, this.endpoint, this.metricID, yesterdaystr);
yesterdayContent = this.apiResponseParser.getJsonData(this.requestManager.getResource(fullURL), false);

String jsoncontent = this.requestManager.getResource(fullURL);
if (!jsoncontent.equals("{}")) {
yesterdayContent = this.apiResponseParser.getJsonData(this.requestManager.getResource(fullURL), false);
}
}
List<MetricProfile> newentries = this.apiResponseParser.getListNewMetrics(content, yesterdayContent);

Expand All @@ -263,7 +285,10 @@ public void getRemoteAggregation() {
String path = "https://%s/api/v2/aggregation_profiles/%s?date=%s";
String fullURL = String.format(path, this.endpoint, this.aggregationID, this.date);
String content = this.requestManager.getResource(fullURL);

if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}
this.data.put(ApiResource.AGGREGATION, this.apiResponseParser.getJsonData(content, false));

}
Expand All @@ -278,6 +303,10 @@ public void getRemoteOps() {
String fullURL = String.format(path, this.endpoint, this.opsID, this.date);

String content = this.requestManager.getResource(fullURL);
if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.OPS, this.apiResponseParser.getJsonData(content, false));

Expand All @@ -292,6 +321,10 @@ public void getRemoteThresholds() {
String path = "https://%s/api/v2/thresholds_profiles/%s?date=%s";
String fullURL = String.format(path, this.endpoint, this.threshID, this.date);
String content = this.requestManager.getResource(fullURL);
if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.THRESHOLDS, this.apiResponseParser.getJsonData(content, false));

Expand All @@ -301,13 +334,17 @@ public void getRemoteThresholds() {
* Retrieves the topology endpoint content and stores it to the enum map
*/
public void getRemoteTopoEndpoints() {
String combinedparam="";
if(isSourceTopoAll){
combinedparam="&mode=combined";
String combinedparam = "";
if (isSourceTopoAll) {
combinedparam = "&mode=combined";
}
String path = "https://%s/api/v2/topology/endpoints/by_report/%s?date=%s";
String fullURL = String.format(path, this.endpoint, this.reportName, this.date+combinedparam);
String fullURL = String.format(path, this.endpoint, this.reportName, this.date + combinedparam);
String content = this.requestManager.getResource(fullURL);
if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.TOPOENDPOINTS, this.apiResponseParser.getJsonData(content, true));

Expand All @@ -317,14 +354,18 @@ public void getRemoteTopoEndpoints() {
* Retrieves the topology groups content and stores it to the enum map
*/
public void getRemoteTopoGroups() {
String combinedparam="";
if(isSourceTopoAll){
combinedparam="&mode=combined";
String combinedparam = "";
if (isSourceTopoAll) {
combinedparam = "&mode=combined";
}

String path = "https://%s/api/v2/topology/groups/by_report/%s?date=%s";
String fullURL = String.format(path, this.endpoint, this.reportName, this.date+combinedparam);
String fullURL = String.format(path, this.endpoint, this.reportName, this.date + combinedparam);
String content = this.requestManager.getResource(fullURL);
if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.TOPOGROUPS, this.apiResponseParser.getJsonData(content, true));

Expand All @@ -337,6 +378,10 @@ public void getRemoteWeights() {
String path = "https://%s/api/v2/weights/%s?date=%s";
String fullURL = String.format(path, this.endpoint, this.weightsID, this.date);
String content = this.requestManager.getResource(fullURL);
if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.WEIGHTS, this.apiResponseParser.getJsonData(content, false));

Expand All @@ -349,6 +394,11 @@ public void getRemoteDowntimes() {
String path = "https://%s/api/v2/downtimes?date=%s";
String fullURL = String.format(path, this.endpoint, this.date);
String content = this.requestManager.getResource(fullURL);
if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.DOWNTIMES, this.apiResponseParser.getJsonData(content, false));

}
Expand All @@ -357,6 +407,10 @@ public void getRemoteRecomputations() {
String path = "https://%s/api/v2/recomputations?date=%s";
String fullURL = String.format(path, this.endpoint, this.date);
String content = this.requestManager.getResource(fullURL);
if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.RECOMPUTATIONS, this.apiResponseParser.getJsonData(content, true));

Expand All @@ -370,9 +424,14 @@ public void getRemoteMetricTags() {
String path = "https://%s/api/v2/metrics/by_report/%s";
String fullURL = String.format(path, this.endpoint, this.reportName);
String content = this.requestManager.getResource(fullURL);
if (!content.equals("{}")) {
this.data.put(ApiResource.MTAGS, this.apiResponseParser.getJsonData(content, true));
if (content.equals("{}")) {

this.shouldRepeat = true;
return;
}

this.data.put(ApiResource.MTAGS, this.apiResponseParser.getJsonData(content, true));

}

/**
Expand Down Expand Up @@ -558,7 +617,7 @@ public void getRemoteAll() {
this.getRemoteThresholds();
}
// Go to topology

this.getRemoteTopoEndpoints();
this.getRemoteTopoGroups();
// get weights
Expand All @@ -576,18 +635,21 @@ public void getRemoteAll() {
* Retrieves the topology endpoint content and stores it to the enum map
*/
public void getAllRemoteTopoEndpoints() {
String combinedparam="";
if(isSourceTopoAll){
combinedparam="&mode=combined";
String combinedparam = "";
if (isSourceTopoAll) {
combinedparam = "&mode=combined";
}
String path = "https://%s/api/v2/topology/endpoints?date=%s";
String fullURL = String.format(path, this.endpoint, this.date+combinedparam);
String fullURL = String.format(path, this.endpoint, this.date + combinedparam);
String content = this.requestManager.getResource(fullURL);

if (content.equals("{}")) {
this.shouldRepeat = true;
return;
}
this.data.put(ApiResource.TOPOENDPOINTS, this.apiResponseParser.getJsonData(content, true));

}

public boolean isIsCombined() {
return isCombined;
}
Expand All @@ -600,7 +662,6 @@ public void setIsSourceTopoAll(boolean isSourceTopoAll) {
this.isSourceTopoAll = isSourceTopoAll;
}



public static DateTime convertStringtoDate(String format, String dateStr) throws ParseException {

Expand All @@ -617,5 +678,7 @@ public static String convertDateToString(String format, DateTime date) throws Pa
DateTimeFormatter dtf = DateTimeFormat.forPattern(format);
String dateString = date.toString(dtf);
return dateString;

}

}
45 changes: 38 additions & 7 deletions flink_jobs_v2/ApiResourceManager/src/main/java/argo/amr/RequestManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,11 @@
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.ConnectTimeoutException;
import org.apache.http.conn.HttpHostConnectException;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
Expand All @@ -19,7 +22,6 @@
import org.apache.http.ssl.SSLContextBuilder;

/**
*
* Establish a connection to the given url and request data
*/
public class RequestManager {
Expand Down Expand Up @@ -48,7 +50,7 @@ public RequestManager(String proxy, String token) {
* content (expected in json format)
*
* @param fullURL String containing the full url representation of the
* argo-web-api resource
* argo-web-api resource
* @return A string representation of the resource json content
* @throws ClientProtocolException
* @throws IOException
Expand Down Expand Up @@ -78,9 +80,38 @@ public String getResource(String fullURL) {

content = r.execute().returnContent().asString();
}
} catch (KeyManagementException | NoSuchAlgorithmException | KeyStoreException | IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (HttpHostConnectException | ConnectTimeoutException e) {
// Handle the case where the API is not reachable
content = "{}";
} catch (IOException e) {
// Handle HTTP response errors
if (e instanceof org.apache.http.client.HttpResponseException) {
org.apache.http.client.HttpResponseException httpException = (org.apache.http.client.HttpResponseException) e;
int statusCode = httpException.getStatusCode();

if (statusCode == 404) {
// Specific report does not exist
System.err.println("The report ID does not exist: " + fullURL);
throw new RuntimeException("Path does not exist: " + fullURL);
} else {
// Other HTTP errors
System.err.println("HTTP error: " + statusCode + " - " + httpException.getMessage());
throw new RuntimeException("HTTP error: " + statusCode + " - " + httpException.getMessage());
}
} else {
// If it's a general IOException (non-HTTP related)
System.err.println("General I/O error: " + e.getMessage());
// Throw the IOException with a specific message
try {
throw new IOException("An error occurred during I/O operation: " + e.getMessage());
} catch (IOException ex) {
throw new RuntimeException(ex);
}
}
} catch (KeyManagementException | NoSuchAlgorithmException | KeyStoreException e) {
// Handle SSL-related exceptions
System.err.println("SSL configuration error: " + e.getMessage());
throw new RuntimeException("SSL configuration error: " + e.getMessage(), e);
}

return content;
Expand Down Expand Up @@ -128,6 +159,6 @@ public void setTimeoutSec(int timeoutSec) {
public void setVerify(boolean verify) {
this.verify = verify;
}


}
Loading