Skip to content

Commit

Permalink
resolve conflicts
Browse files Browse the repository at this point in the history 
Signed-off-by: kusumachalasani <[email protected]>
  • Loading branch information
@kusumachalasani
kusumachalasani committed Dec 10, 2024
1 parent baae100 commit c3c0c7d
Showing 1 changed file with 10 additions and 15 deletions.
25 changes: 10 additions & 15 deletions src/main/java/com/autotune/analyzer/workerimpl/BulkJobManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
*******************************************************************************/
package com.autotune.analyzer.workerimpl;


import com.autotune.analyzer.kruizeObject.RecommendationSettings;
import com.autotune.analyzer.serviceObjects.*;
import com.autotune.analyzer.utils.AnalyzerConstants;
Expand All @@ -31,6 +30,7 @@
import com.autotune.utils.Utils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.gson.Gson;
import io.micrometer.core.instrument.Timer;
import org.apache.http.conn.ConnectTimeoutException;
import org.json.JSONObject;
import org.slf4j.Logger;
Expand All @@ -50,15 +50,13 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static com.autotune.operator.KruizeDeploymentInfo.bulk_thread_pool_size;
import static com.autotune.utils.KruizeConstants.KRUIZE_BULK_API.*;
import static com.autotune.utils.KruizeConstants.KRUIZE_BULK_API.NotificationConstants.*;


/**
* The `run` method processes bulk input to create experiments and generates resource optimization recommendations.
* It handles the creation of experiment names based on various data source components, makes HTTP POST requests
Expand Down Expand Up @@ -123,7 +121,7 @@ private static Map<String, String> parseLabelString(String labelString) {
public void run() {
String statusValue = "failure";
MetricsConfig.activeJobs.incrementAndGet();
io.micrometer.core.instrument.Timer.Sample timerRunJob = Timer.start(MetricsConfig.meterRegistry());
Timer.Sample timerRunJob = Timer.start(MetricsConfig.meterRegistry());
DataSourceMetadataInfo metadataInfo = null;
DataSourceManager dataSourceManager = new DataSourceManager();
DataSourceInfo datasource = null;
Expand Down Expand Up @@ -170,26 +168,22 @@ public void run() {
GenericRestApiClient apiClient = new GenericRestApiClient(finalDatasource);
apiClient.setBaseURL(KruizeDeploymentInfo.experiments_url);
GenericRestApiClient.HttpResponseWrapper responseCode;
boolean experiment_exists = false;
boolean expriment_exists = false;
try {
responseCode = apiClient.callKruizeAPI("[" + new Gson().toJson(apiObject) + "]");
LOGGER.debug("API Response code: {}", responseCode);
if (responseCode.getStatusCode() == HttpURLConnection.HTTP_CREATED) {
experiment_exists = true;
} else if (responseCode.getStatusCode() == HttpURLConnection.HTTP_CONFLICT) {
experiment_exists = true;
} else {
expriment_exists = true;
} else if (responseCode.getStatusCode() == HttpURLConnection.HTTP_CONFLICT) {
experiment_exists = true;
expriment_exists = true;
} else {
jobData.setProcessed_experiments(jobData.getProcessed_experiments() + 1);
experiment.setNotification(new BulkJobStatus.Notification(BulkJobStatus.NotificationType.ERROR, responseCode.getResponseBody().toString(), responseCode.getStatusCode()));
}
} catch (Exception e) {
e.printStackTrace();
experiment.setNotification(new BulkJobStatus.Notification(BulkJobStatus.NotificationType.ERROR, e.getMessage(), HttpURLConnection.HTTP_BAD_REQUEST));
} finally {
if (!experiment_exists) {
if (!expriment_exists) {
LOGGER.info("Processing experiment {}", jobData.getProcessed_experiments());
jobData.setProcessed_experiments(jobData.getProcessed_experiments() + 1);
}
Expand All @@ -198,9 +192,9 @@ public void run() {
setFinalJobStatus(COMPLETED, null, null, finalDatasource);
}
}
}
}

if (experiment_exists) {
if (expriment_exists) {
generateExecutor.submit(() -> {
// send request to generateRecommendations API
GenericRestApiClient recommendationApiClient = new GenericRestApiClient(finalDatasource);
Expand Down Expand Up @@ -242,6 +236,7 @@ public void run() {
});
}
} finally {
// Shutdown createExecutor and wait for it to finish
createExecutor.shutdown();
while (!createExecutor.isTerminated()) {
try {
Expand All @@ -252,6 +247,7 @@ public void run() {
}
}

// Shutdown generateExecutor and wait for it to finish
generateExecutor.shutdown();
while (!generateExecutor.isTerminated()) {
try {
Expand Down Expand Up @@ -512,4 +508,3 @@ public String frameExperimentName(String labelString, DataSourceCluster dataSour
return experimentName;
}
}

0 comments on commit c3c0c7d

Please sign in to comment.