Skip to content
This repository was archived by the owner on Jun 10, 2021. It is now read-only.

Commit a7456e6

Browse files
VladislavNikolovVladislavNikolov
authored
Merge pull request #236 from rndsolutions/refactor-service-result
Service Result refactored
2 parents 1050e06 + 24a5154 commit a7456e6

File tree

85 files changed

+1155
-1064
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

85 files changed

+1155
-1064
lines changed

‎Server/src/main/java/net/hawkengine/core/HawkServer.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ public class HawkServer {
2222
private Thread pipelinePreparer;
2323
private Thread jobAssigner;
2424
private Thread materialTracker;
25-
private EndpointFinder endpointFinder;
2625
private DataImporter dataImporter;
2726

2827
public HawkServer() {
@@ -32,7 +31,6 @@ public HawkServer() {
3231
this.pipelinePreparer = new Thread(new PipelinePreparer());
3332
this.jobAssigner = new Thread(new JobAssigner());
3433
this.materialTracker = new Thread(new MaterialTracker());
35-
this.endpointFinder = new EndpointFinder();
3634
this.dataImporter = new DataImporter();
3735
}
3836

‎Server/src/main/java/net/hawkengine/core/materialhandler/MaterialTracker.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ public void run() {
6969
if (isPipelineUpdated) {
7070
pipeline.setMaterialsUpdated(true);
7171
ServiceResult result = this.pipelineService.update(pipeline);
72-
EndpointConnector.passResultToEndpoint(PipelineService.class.getSimpleName(), "update", result);
72+
// EndpointConnector.passResultToEndpoint(PipelineService.class.getSimpleName(), "update", result);
7373
}
7474
}
7575

‎Server/src/main/java/net/hawkengine/core/pipelinescheduler/JobAssigner.java‎

Lines changed: 12 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -2,60 +2,40 @@
22

33
import net.hawkengine.core.ServerConfiguration;
44
import net.hawkengine.core.utilities.constants.LoggerMessages;
5-
import net.hawkengine.model.*;
6-
import net.hawkengine.model.enums.StageStatus;
5+
import net.hawkengine.model.Agent;
76
import net.hawkengine.services.AgentService;
8-
import net.hawkengine.services.PipelineService;
97
import net.hawkengine.services.interfaces.IAgentService;
10-
import net.hawkengine.services.interfaces.IPipelineService;
11-
import net.hawkengine.ws.EndpointConnector;
128
import org.apache.log4j.Logger;
139

1410
import java.util.List;
1511

1612
public class JobAssigner implements Runnable {
17-
private IAgentService agentService;
18-
private IPipelineService pipelineService;
13+
private static final Logger LOGGER = Logger.getLogger(PipelinePreparer.class.getName());
14+
private static final int POLL_INTERVAL = ServerConfiguration.getConfiguration().getPipelineSchedulerPollInterval() * 1000;
15+
1916
private JobAssignerService jobAssignerService;
2017
private StatusUpdaterService statusUpdaterService;
21-
private static final Logger LOGGER = Logger.getLogger(PipelinePreparer.class.getName());
18+
private IAgentService agentService;
2219

2320
public JobAssigner() {
24-
this.agentService = new AgentService();
25-
this.pipelineService = new PipelineService();
2621
this.jobAssignerService = new JobAssignerService();
2722
this.statusUpdaterService = new StatusUpdaterService();
23+
this.agentService = new AgentService();
2824
}
2925

3026
@Override
3127
public void run() {
3228
LOGGER.info(String.format(LoggerMessages.WORKER_STARTED, this.getClass().getSimpleName()));
3329
try {
3430
while (true) {
35-
this.statusUpdaterService.updateStatuses();
36-
List<Agent> agents = (List<Agent>) this.agentService.getAllAssignableAgents().getObject();
37-
List<Pipeline> pipelines = (List<Pipeline>) this.pipelineService.getAllPreparedPipelinesInProgress().getObject();
31+
List<Agent> agents = (List<Agent>) this.agentService.getAll().getObject();
3832

39-
for (Pipeline pipeline : pipelines) {
40-
for (Stage stage : pipeline.getStages()) {
41-
if (stage.getStatus() == StageStatus.IN_PROGRESS) {
42-
for (Job job : stage.getJobs()) {
43-
if (agents.size() != 0) {
44-
Agent agent = this.jobAssignerService.assignAgentToJob(job, agents);
45-
if (agent != null) {
46-
ServiceResult result = this.agentService.update(agent);
47-
EndpointConnector.passResultToEndpoint(AgentService.class.getSimpleName(), "update", result);
48-
}
49-
}
50-
}
51-
}
52-
}
53-
54-
ServiceResult result = this.pipelineService.update(pipeline);
55-
EndpointConnector.passResultToEndpoint(PipelineService.class.getSimpleName(), "update", result);
56-
}
33+
this.statusUpdaterService.updateStatuses();
34+
this.jobAssignerService.checkUnassignedJobs(agents);
35+
this.jobAssignerService.checkAwaitingJobs(agents);
36+
this.jobAssignerService.assignJobs(agents);
5737

58-
Thread.sleep(ServerConfiguration.getConfiguration().getMaterialTrackerPollInterval() * 1000);
38+
Thread.sleep(POLL_INTERVAL);
5939
}
6040
} catch (InterruptedException e) {
6141
e.printStackTrace();

‎Server/src/main/java/net/hawkengine/core/pipelinescheduler/JobAssignerService.java‎

Lines changed: 87 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -1,88 +1,115 @@
11
package net.hawkengine.core.pipelinescheduler;
22

3-
import net.hawkengine.model.Agent;
4-
import net.hawkengine.model.Job;
3+
import net.hawkengine.model.*;
54
import net.hawkengine.model.enums.JobStatus;
5+
import net.hawkengine.model.enums.NotificationType;
6+
import net.hawkengine.model.enums.StageStatus;
7+
import net.hawkengine.model.enums.Status;
8+
import net.hawkengine.services.AgentService;
9+
import net.hawkengine.services.PipelineService;
10+
import net.hawkengine.services.interfaces.IAgentService;
11+
import net.hawkengine.services.interfaces.IPipelineService;
12+
import net.hawkengine.ws.EndpointConnector;
613
import org.apache.log4j.Logger;
714

8-
import java.util.ArrayList;
915
import java.util.List;
16+
import java.util.stream.Collectors;
1017

1118
public class JobAssignerService {
1219
private static final Logger LOGGER = Logger.getLogger(JobAssignerService.class.getName());
1320

14-
public Agent assignAgentToJob(Job job, List<Agent> agents) {
15-
Agent result = null;
16-
if (job.getStatus() == JobStatus.SCHEDULED) {
17-
Agent assignedAgent = agents.stream().filter(a -> a.getId().equals(job.getAssignedAgentId())).findFirst().orElse(null);
18-
result = assignedAgent;
19-
boolean isEligible = this.isAgentEligibleForJob(job, assignedAgent);
20-
if (!isEligible) {
21-
job.setStatus(JobStatus.AWAITING);
22-
assignedAgent.setAssigned(false);
23-
result = assignedAgent;
24-
LOGGER.info(String.format("Job %s unassigned from Agent %s", job.getJobDefinitionName(), assignedAgent.getName()));
25-
}
26-
}
27-
28-
if (job.getStatus() == JobStatus.AWAITING) {
29-
List<Agent> eligibleAgents = this.getEligibleAgentsForJob(job, agents);
30-
Agent agentForJob = this.pickMostSuitableAgent(eligibleAgents);
31-
if (agentForJob != null) {
32-
job.setAssignedAgentId(agentForJob.getId());
33-
job.setStatus(JobStatus.SCHEDULED);
34-
agentForJob.setAssigned(true);
35-
result = agentForJob;
36-
LOGGER.info(String.format("Job %s assigned to Agent %s", job.getJobDefinitionName(), agentForJob.getName()));
37-
}
38-
}
21+
private IAgentService agentService;
22+
private IPipelineService pipelineService;
23+
private JobAssignerUtilities jobAssignerUtilities;
3924

40-
return result;
25+
public JobAssignerService() {
26+
this.agentService = new AgentService();
27+
this.pipelineService = new PipelineService();
28+
this.jobAssignerUtilities = new JobAssignerUtilities();
4129
}
4230

43-
public List<Agent> getEligibleAgentsForJob(Job job, List<Agent> agents) {
44-
List<Agent> eligibleAgents = new ArrayList<>();
45-
for (Agent agent : agents) {
46-
boolean isEligible = this.isAgentEligibleForJob(job, agent);
31+
public void checkUnassignedJobs(List<Agent> agents) {
32+
List<Agent> filteredAgents = agents.stream().filter(a -> a.isConnected() && a.isEnabled()).collect(Collectors.toList());
33+
List<Pipeline> pipelinesInProgress = (List<Pipeline>) this.pipelineService.getAllPreparedPipelinesInProgress().getObject();
34+
for (Pipeline pipeline : pipelinesInProgress) {
35+
boolean isSetToAwaiting = false;
36+
Stage stageInProgress = pipeline.getStages().stream().filter(s -> s.getStatus() == StageStatus.IN_PROGRESS).findFirst().orElse(null);
37+
if (stageInProgress == null) {
38+
continue;
39+
}
4740

48-
if (isEligible) {
49-
eligibleAgents.add(agent);
41+
for (Job job : stageInProgress.getJobs()) {
42+
if (job.getStatus() == JobStatus.UNASSIGNED) {
43+
boolean hasAssignableAgent = this.jobAssignerUtilities.hasAssignableAgent(job, filteredAgents);
44+
if (!hasAssignableAgent) {
45+
job.setStatus(JobStatus.AWAITING);
46+
isSetToAwaiting = true;
47+
LOGGER.info(String.format("Job %s has no assignable Agents.", job.getJobDefinitionName()));
48+
}
49+
}
5050
}
51-
}
5251

53-
return eligibleAgents;
52+
if (isSetToAwaiting) {
53+
stageInProgress.setStatus(StageStatus.AWAITING);
54+
pipeline.setStatus(Status.AWAITING);
55+
this.pipelineService.update(pipeline);
56+
String message = String.format("Pipeline %s set to AWAITING.", pipeline.getPipelineDefinitionName());
57+
LOGGER.info(message);
58+
ServiceResult notification = new ServiceResult(null, NotificationType.WARNING, message);
59+
EndpointConnector.passResultToEndpoint("NotificationService", "sendMessage", notification);
60+
}
61+
}
5462
}
5563

56-
public Agent pickMostSuitableAgent(List<Agent> agents) {
57-
Agent agentForJob = null;
58-
if (agents.size() == 1) {
59-
agentForJob = agents.get(0);
60-
} else if (agents.size() > 1) {
61-
int numberOfResources = Integer.MAX_VALUE;
62-
for (Agent agent : agents) {
63-
if (agent.getResources().size() < numberOfResources) {
64-
numberOfResources = agent.getResources().size();
65-
agentForJob = agent;
64+
public void checkAwaitingJobs(List<Agent> agents) {
65+
List<Agent> filteredAgents = agents.stream().filter(a -> a.isConnected() && a.isEnabled()).collect(Collectors.toList());
66+
List<Pipeline> awaitingPipelines = (List<Pipeline>) this.pipelineService.getAllPreparedAwaitingPipelines().getObject();
67+
for (Pipeline pipeline : awaitingPipelines) {
68+
Stage awaitingStage = pipeline.getStages().stream().filter(s -> s.getStatus() == StageStatus.AWAITING).findFirst().orElse(null);
69+
if (awaitingStage == null) {
70+
continue;
71+
}
72+
73+
for (Job job : awaitingStage.getJobs()) {
74+
if (job.getStatus() == JobStatus.AWAITING) {
75+
boolean hasAssignableAgent = this.jobAssignerUtilities.hasAssignableAgent(job, filteredAgents);
76+
if (hasAssignableAgent) {
77+
job.setStatus(JobStatus.UNASSIGNED);
78+
LOGGER.info(String.format("Job %s set back to IN_PROGRESS.", job.getJobDefinitionName()));
79+
}
6680
}
6781
}
68-
}
6982

70-
return agentForJob;
83+
boolean hasAwaitingJobs = awaitingStage.getJobs().stream().anyMatch(j -> j.getStatus() == JobStatus.AWAITING);
84+
if (!hasAwaitingJobs) {
85+
awaitingStage.setStatus(StageStatus.IN_PROGRESS);
86+
pipeline.setStatus(Status.IN_PROGRESS);
87+
this.pipelineService.update(pipeline);
88+
LOGGER.info(String.format("Pipeline %s set back to IN_PROGRESS.", pipeline.getPipelineDefinitionName()));
89+
}
90+
}
7191
}
7292

73-
public boolean isAgentEligibleForJob(Job job, Agent agent) {
74-
boolean isEligible = true;
75-
if ((agent == null) || !agent.isConnected() || !agent.isEnabled() || agent.isRunning() || agent.isAssigned()) {
76-
isEligible = false;
77-
} else {
78-
for (String resource : job.getResources()) {
79-
if (!(agent.getResources().contains(resource))) {
80-
isEligible = false;
81-
break;
93+
public void assignJobs(List<Agent> agents) {
94+
List<Agent> filteredAgents = agents.stream().filter(a -> a.isConnected() && a.isEnabled() && !a.isRunning() && !a.isAssigned()).collect(Collectors.toList());
95+
List<Pipeline> pipelines = (List<Pipeline>) this.pipelineService.getAllPreparedPipelinesInProgress().getObject();
96+
for (Pipeline pipeline : pipelines) {
97+
for (Stage stage : pipeline.getStages()) {
98+
if (stage.getStatus() == StageStatus.IN_PROGRESS) {
99+
for (Job job : stage.getJobs()) {
100+
if (filteredAgents.size() != 0) {
101+
Agent agent = this.jobAssignerUtilities.assignAgentToJob(job, filteredAgents);
102+
if (agent != null) {
103+
ServiceResult result = this.agentService.update(agent);
104+
EndpointConnector.passResultToEndpoint(AgentService.class.getSimpleName(), "update", result);
105+
}
106+
}
107+
}
82108
}
83109
}
84-
}
85110

86-
return isEligible;
111+
this.pipelineService.update(pipeline);
112+
// EndpointConnector.passResultToEndpoint(PipelineService.class.getSimpleName(), "update", result);
113+
}
87114
}
88115
}

‎Server/src/main/java/net/hawkengine/core/pipelinescheduler/JobAssignerUtilities.java‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
package net.hawkengine.core.pipelinescheduler;
2+
3+
import net.hawkengine.model.Agent;
4+
import net.hawkengine.model.Job;
5+
import net.hawkengine.model.enums.JobStatus;
6+
import org.apache.log4j.Logger;
7+
8+
import java.util.ArrayList;
9+
import java.util.List;
10+
11+
public class JobAssignerUtilities {
12+
private static final Logger LOGGER = Logger.getLogger(JobAssignerUtilities.class.getName());
13+
14+
public Agent assignAgentToJob(Job job, List<Agent> agents) {
15+
Agent result = null;
16+
if (job.getStatus() == JobStatus.ASSIGNED) {
17+
Agent assignedAgent = agents.stream().filter(a -> a.getId().equals(job.getAssignedAgentId())).findFirst().orElse(null);
18+
result = assignedAgent;
19+
boolean isEligible = this.isAgentEligibleForJob(job, assignedAgent);
20+
if (!isEligible) {
21+
job.setStatus(JobStatus.UNASSIGNED);
22+
assignedAgent.setAssigned(false);
23+
result = assignedAgent;
24+
LOGGER.info(String.format("Job %s unassigned from Agent %s", job.getJobDefinitionName(), assignedAgent.getName()));
25+
}
26+
}
27+
28+
if (job.getStatus() == JobStatus.UNASSIGNED) {
29+
List<Agent> eligibleAgents = this.getEligibleAgentsForJob(job, agents);
30+
Agent agentForJob = this.pickMostSuitableAgent(eligibleAgents);
31+
if (agentForJob != null) {
32+
job.setAssignedAgentId(agentForJob.getId());
33+
job.setStatus(JobStatus.ASSIGNED);
34+
agentForJob.setAssigned(true);
35+
result = agentForJob;
36+
LOGGER.info(String.format("Job %s assigned to Agent %s", job.getJobDefinitionName(), agentForJob.getName()));
37+
}
38+
}
39+
40+
return result;
41+
}
42+
43+
public List<Agent> getEligibleAgentsForJob(Job job, List<Agent> agents) {
44+
List<Agent> eligibleAgents = new ArrayList<>();
45+
for (Agent agent : agents) {
46+
boolean isEligible = this.isAgentEligibleForJob(job, agent);
47+
48+
if (isEligible) {
49+
eligibleAgents.add(agent);
50+
}
51+
}
52+
53+
return eligibleAgents;
54+
}
55+
56+
public Agent pickMostSuitableAgent(List<Agent> agents) {
57+
Agent agentForJob = null;
58+
int numberOfResources = Integer.MAX_VALUE;
59+
for (Agent agent : agents) {
60+
if (agent.getResources().size() < numberOfResources) {
61+
numberOfResources = agent.getResources().size();
62+
agentForJob = agent;
63+
}
64+
}
65+
66+
return agentForJob;
67+
}
68+
69+
70+
// public Agent pickMostSuitableAgent(List<Agent> agents) {
71+
// Agent agentForJob = null;
72+
// if (agents.size() == 1) {
73+
// agentForJob = agents.get(0);
74+
// } else if (agents.size() > 1) {
75+
// int numberOfResources = Integer.MAX_VALUE;
76+
// for (Agent agent : agents) {
77+
// if (agent.getResources().size() < numberOfResources) {
78+
// numberOfResources = agent.getResources().size();
79+
// agentForJob = agent;
80+
// }
81+
// }
82+
// }
83+
//
84+
// return agentForJob;
85+
// }
86+
87+
public boolean isAgentEligibleForJob(Job job, Agent agent) {
88+
boolean isEligible = true;
89+
if ((agent == null) || !agent.isConnected() || !agent.isEnabled() || agent.isRunning() || agent.isAssigned()) {
90+
isEligible = false;
91+
} else {
92+
for (String resource : job.getResources()) {
93+
if (!(agent.getResources().contains(resource))) {
94+
isEligible = false;
95+
break;
96+
}
97+
}
98+
}
99+
100+
return isEligible;
101+
}
102+
103+
public boolean hasAssignableAgent(Job job, List<Agent> agents) {
104+
boolean hasAssignableAgent = true;
105+
for (Agent agent : agents) {
106+
for (String resource : job.getResources()) {
107+
if (!(agent.getResources().contains(resource))) {
108+
hasAssignableAgent = false;
109+
}
110+
}
111+
112+
if (hasAssignableAgent) {
113+
return true;
114+
}
115+
}
116+
117+
return false;
118+
}
119+
}

0 commit comments

Comments
 (0)