Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
comparing upstreams for each service
typo
- Loading branch information
1 parent
5d2f676
commit d70ea9b
Showing
5 changed files
with
50 additions
and
73 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
75 changes: 30 additions & 45 deletions
75
...tyService/src/main/java/com/hubspot/singularity/scheduler/SingularityUpstreamChecker.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,80 +1,65 @@ | ||
| package com.hubspot.singularity.scheduler; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.Collection; | ||
| import java.util.List; | ||
|
|
||
| import javax.inject.Inject; | ||
| import javax.inject.Singleton; | ||
|
|
||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import com.google.common.base.Optional; | ||
| import com.google.common.collect.Lists; | ||
| import com.hubspot.baragon.models.UpstreamInfo; | ||
| import com.hubspot.mesos.protos.MesosParameter; | ||
| import com.hubspot.singularity.LoadBalancerRequestType; | ||
| import com.hubspot.singularity.LoadBalancerRequestType.LoadBalancerRequestId; | ||
| import com.hubspot.singularity.SingularityRequest; | ||
| import com.hubspot.singularity.SingularityTask; | ||
| import com.hubspot.singularity.SingularityTaskId; | ||
| import com.hubspot.singularity.data.RequestManager; | ||
| import com.hubspot.singularity.data.TaskManager; | ||
| import com.hubspot.singularity.helpers.MesosUtils; | ||
| import com.hubspot.singularity.hooks.LoadBalancerClient; | ||
| import com.hubspot.singularity.hooks.LoadBalancerClientImpl; | ||
|
|
||
| @Singleton | ||
| public class SingularityUpstreamChecker { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(LoadBalancerClient.class); | ||
|
|
||
| private final LoadBalancerClientImpl lbClient; | ||
| private final TaskManager taskManager; | ||
| private final RequestManager requestManager; | ||
|
|
||
| @Inject | ||
| public SingularityUpstreamChecker(LoadBalancerClientImpl lbClient, TaskManager taskManager) { | ||
| public SingularityUpstreamChecker(LoadBalancerClientImpl lbClient, TaskManager taskManager, RequestManager requestManager) { | ||
| this.lbClient = lbClient; | ||
| this.taskManager = taskManager; | ||
| this.requestManager = requestManager; | ||
| } | ||
|
|
||
| private UpstreamInfo taskToUpstreamInfo (SingularityTask task, Optional<String> loadBalancerUpstreamGroup) { | ||
| UpstreamInfo upstreamInfo = null; //TODO: what is a better way to do this? We cannot initialize an empty upstream | ||
| final Optional<Long> maybeLoadBalancerPort = MesosUtils.getPortByIndex(lbClient.getMesosProtosUtils().toResourceList(task.getMesosTask().getResources()), task.getTaskRequest().getDeploy().getLoadBalancerPortIndex().or(0)); | ||
| if (maybeLoadBalancerPort.isPresent()) { | ||
| final String host = task.getHostname(); | ||
| final long port = maybeLoadBalancerPort.get(); | ||
| String upstream = String.format("%s:%d", host, port); | ||
| final Optional<String> requestId = Optional.of(task.getTaskRequest().getRequest().getId()); | ||
| Optional<String> group = loadBalancerUpstreamGroup; | ||
| if (lbClient.getTaskLabelForLoadBalancerUpstreamGroup().isPresent()) { | ||
| for (MesosParameter label : task.getMesosTask().getLabels().getLabels()) { | ||
| if (label.hasKey() && label.getKey().equals(lbClient.getTaskLabelForLoadBalancerUpstreamGroup().get()) && label.hasValue()) { | ||
| group = Optional.of(label.getValue()); | ||
| break; | ||
| } | ||
| } | ||
| private List<SingularityTask> getActiveSingularityTasksForRequest (String requestId) { | ||
| List<SingularityTask> activeSingularityTasksForRequest = new ArrayList<>(); | ||
| for (SingularityTaskId taskId: taskManager.getActiveTaskIdsForRequest(requestId)){ | ||
| Optional<SingularityTask> maybeTask = taskManager.getTask(taskId); | ||
| if (maybeTask.isPresent()) { | ||
| activeSingularityTasksForRequest.add(maybeTask.get()); | ||
| } | ||
| upstreamInfo = new UpstreamInfo(upstream, requestId, task.getRackId(), Optional.<String>absent(), group); | ||
| } else { | ||
| LOG.warn("Task {} is missing port, not passed to upstreams({})", task.getTaskId(), task); | ||
| } | ||
| return upstreamInfo; | ||
| return activeSingularityTasksForRequest; | ||
| } | ||
|
|
||
| public Collection<UpstreamInfo> getUpstreamsFromActiveTasks() { | ||
| final List<UpstreamInfo> upstreams = Lists.newArrayListWithCapacity(taskManager.getNumActiveTasks()); | ||
| for (SingularityTask task: taskManager.getActiveTasks()){ | ||
| final Optional<String> loadBalancerUpstreamGroup = task.getTaskRequest().getDeploy().getLoadBalancerUpstreamGroup(); | ||
| final UpstreamInfo upstream = taskToUpstreamInfo(task, loadBalancerUpstreamGroup); | ||
| if (upstream.equals(null)){ | ||
| upstreams.add(upstream); | ||
| } | ||
| } | ||
| return upstreams; | ||
| private Collection<UpstreamInfo> getUpstreamsFromActiveTasks(String requestId, Optional<String> loadBalancerUpstreamGroup) { | ||
| return lbClient.tasksToUpstreams(getActiveSingularityTasksForRequest(requestId), requestId, loadBalancerUpstreamGroup); | ||
| } | ||
|
|
||
| private void syncUpstreams( ) { | ||
| Collection<UpstreamInfo> upstreamsInBaragon = lbClient.getUpstreams(); | ||
| Collection<UpstreamInfo> upstreamsInSingularity = getUpstreamsFromActiveTasks(); | ||
| upstreamsInBaragon.removeAll(upstreamsInSingularity); | ||
| for (UpstreamInfo upstream: upstreamsInBaragon){ | ||
| private void syncUpstreamsForService(SingularityRequest request, Optional<String> loadBalancerUpstreamGroup) { | ||
| String requestId = request.getId(); | ||
| Collection<UpstreamInfo> upstreamsInBaragonForRequest = lbClient.getBaragonUpstreamsForRequest(requestId); | ||
| Collection<UpstreamInfo> upstreamsInSingularityForRequest = getUpstreamsFromActiveTasks(requestId, loadBalancerUpstreamGroup); | ||
| upstreamsInBaragonForRequest.removeAll(upstreamsInSingularityForRequest); | ||
| LoadBalancerRequestId loadBalancerRequestId = new LoadBalancerRequestId(requestId, LoadBalancerRequestType.REMOVE, Optional.absent()); | ||
| for (UpstreamInfo upstream: upstreamsInBaragonForRequest){ | ||
| //TODO: remove the upstream from Baragon | ||
| } | ||
| } | ||
|
|
||
| public void syncUpstreams() { | ||
| // TODO: check through the active requests and run the method above | ||
|
|
||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters