predicate) {
- return getContainerStatus(pod).stream().filter(predicate).collect(Collectors.toList());
- }
-
- /**
- * Cancel queue items matching the given pod.
- *
The queue item has to have a task url matching the pod "runUrl"-annotation
- * and the queue item assigned label needs to match the label jenkins/label of the pod.
- *
It uses the current thread context to list item queues,
- * so make sure to be in the right context before calling this method.
- *
- * @param pod The pod to cancel items for.
- * @param reason The reason the item are being cancelled.
- */
- public static void cancelQueueItemFor(Pod pod, String reason) {
- var metadata = pod.getMetadata();
- if (metadata == null) {
- return;
- }
- String podName = metadata.getName();
- String podNamespace = metadata.getNamespace();
- String podDisplayName = podNamespace + "/" + podName;
- var annotations = metadata.getAnnotations();
- if (annotations == null) {
- LOGGER.log(Level.FINE, () -> "Pod " + podDisplayName + " .metadata.annotations is null");
- return;
- }
- var runUrl = annotations.get(PodTemplateStepExecution.POD_ANNOTATION_RUN_URL);
- if (runUrl == null) {
- LOGGER.log(Level.FINE, () -> "Pod " + podDisplayName + " .metadata.annotations.runUrl is null");
- return;
- }
- var labels = metadata.getLabels();
- if (labels == null) {
- LOGGER.log(Level.FINE, () -> "Pod " + podDisplayName + " .metadata.labels is null");
- return;
+ public static void cancelQueueItemFor(ArmadaSlave node, String reason) {
+ var queueItem = node.getItem();
+ if(queueItem != null) {
+ LOGGER.log(Level.FINE, "Canceling queue item \"" + queueItem.task.getDisplayName() + "\"\n" + (!StringUtils.isBlank(reason) ? "due to " + reason : ""));
+ var queue = Jenkins.get().getQueue();
+ queue.cancel(queueItem);
}
- cancelQueueItemFor(runUrl, labels.get(PodTemplate.JENKINS_LABEL), reason, podDisplayName);
}
public static void cancelQueueItemFor(
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/StandardPlannedNodeBuilder.java b/src/main/java/io/armadaproject/jenkins/plugin/StandardPlannedNodeBuilder.java
index 4590508c3..17d14cc65 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/StandardPlannedNodeBuilder.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/StandardPlannedNodeBuilder.java
@@ -17,7 +17,7 @@ public NodeProvisioner.PlannedNode build() {
CompletableFuture f;
String displayName;
try {
- KubernetesSlave agent = KubernetesSlave.builder()
+ ArmadaSlave agent = ArmadaSlave.builder()
.podTemplate(t.isUnwrapped() ? t : cloud.getUnwrappedTemplate(t))
.cloud(cloud)
.build();
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientParameters.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientParameters.java
new file mode 100644
index 000000000..2493d1368
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientParameters.java
@@ -0,0 +1,33 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import java.util.Objects;
+
+public class ArmadaClientParameters {
+ public final String apiUrl;
+ public final int apiPort;
+ public final String queue;
+ public final String namespace;
+ public final String credentialsId;
+ public final ArmadaJobSetStrategy jobSetStrategy;
+
+ public ArmadaClientParameters(String apiUrl, int apiPort, String queue, String namespace, String credentialsId, ArmadaJobSetStrategy jobSetStrategy) {
+ this.apiUrl = apiUrl;
+ this.apiPort = apiPort;
+ this.queue = queue;
+ this.namespace = namespace;
+ this.credentialsId = credentialsId;
+ this.jobSetStrategy = jobSetStrategy;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof ArmadaClientParameters)) return false;
+ ArmadaClientParameters that = (ArmadaClientParameters) o;
+ return apiPort == that.apiPort && Objects.equals(apiUrl, that.apiUrl) && Objects.equals(queue, that.queue) && Objects.equals(namespace, that.namespace) && Objects.equals(credentialsId, that.credentialsId) && Objects.equals(jobSetStrategy, that.jobSetStrategy);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(apiUrl, apiPort, queue, namespace, credentialsId, jobSetStrategy);
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientProvider.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientProvider.java
new file mode 100644
index 000000000..c006610f2
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientProvider.java
@@ -0,0 +1,8 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import io.armadaproject.ArmadaClient;
+import org.jenkinsci.plugins.kubernetes.auth.KubernetesAuthException;
+
+public interface ArmadaClientProvider {
+ ArmadaClient get() throws KubernetesAuthException;
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientUtil.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientUtil.java
new file mode 100644
index 000000000..7b0a0ad22
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaClientUtil.java
@@ -0,0 +1,185 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import api.EventOuterClass;
+import api.SubmitOuterClass;
+import org.apache.commons.lang.StringUtils;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public final class ArmadaClientUtil {
+ private ArmadaClientUtil() {}
+
+ public static final Set TERMINAL_STATES = new HashSet<>(List.of(SubmitOuterClass.JobState.FAILED, SubmitOuterClass.JobState.CANCELLED, SubmitOuterClass.JobState.SUCCEEDED, SubmitOuterClass.JobState.PREEMPTED));
+
+ public static String lookoutUrlForJob(String lookoutBaseUrl, int lookoutPort, String queue, String jobSetId, String jobId) {
+ return StringUtils.stripEnd(lookoutBaseUrl, "/") + ":" + lookoutPort + "/?page=0&f[0][id]=queue&f[0][value][0]=" +
+ queue + "&f[0][match]=anyOf&f[1][id]=jobSet&f[1][value]=" +
+ jobSetId + "&f[1][match]=exact&f[2][id]=jobId&f[2][value]=" +
+ jobId + "&f[2][match]=exact";
+ }
+
+ public static SubmitOuterClass.JobState toJobState(EventOuterClass.EventMessage.EventsCase event) {
+ switch(event) {
+ case RUNNING:
+ return SubmitOuterClass.JobState.RUNNING;
+ case PENDING:
+ return SubmitOuterClass.JobState.PENDING;
+ case QUEUED:
+ return SubmitOuterClass.JobState.QUEUED;
+ case SUCCEEDED:
+ return SubmitOuterClass.JobState.SUCCEEDED;
+ case FAILED:
+ return SubmitOuterClass.JobState.FAILED;
+ case SUBMITTED:
+ return SubmitOuterClass.JobState.SUBMITTED;
+ case LEASED:
+ return SubmitOuterClass.JobState.LEASED;
+ case PREEMPTED:
+ return SubmitOuterClass.JobState.PREEMPTED;
+ case CANCELLED:
+ return SubmitOuterClass.JobState.CANCELLED;
+ default:
+ return SubmitOuterClass.JobState.UNKNOWN;
+ }
+ }
+
+ public static boolean isInFailedState(SubmitOuterClass.JobState jobState) {
+ switch(jobState) {
+ case FAILED:
+ case REJECTED:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ public static boolean isInFailedState(EventOuterClass.EventMessage.EventsCase jobState) {
+ switch(jobState) {
+ case FAILED:
+ case UNABLE_TO_SCHEDULE:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ public static boolean isInTerminalState(SubmitOuterClass.JobState jobState) {
+ switch(jobState) {
+ case FAILED:
+ case CANCELLED:
+ case SUCCEEDED:
+ case PREEMPTED:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ public static boolean isInTerminalState(EventOuterClass.EventMessage.EventsCase eventsCase) {
+ switch(eventsCase) {
+ case FAILED:
+ case CANCELLED:
+ case SUCCEEDED:
+ case PREEMPTED:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ public static ArmadaJobMetadata extractMetadata(EventOuterClass.EventMessage eventMessage) {
+ String jobId = null;
+ String jobSetId = null;
+ String clusterId = null;
+ String podName = null;
+ String reason = null;
+ EventOuterClass.Cause cause = null;
+ switch(eventMessage.getEventsCase()) {
+ case SUBMITTED:
+ jobId = eventMessage.getSubmitted().getJobId();
+ jobSetId = eventMessage.getSubmitted().getJobSetId();
+ break;
+ case QUEUED:
+ jobId = eventMessage.getQueued().getJobId();
+ jobSetId = eventMessage.getQueued().getJobSetId();
+ break;
+ case LEASED:
+ jobId = eventMessage.getLeased().getJobId();
+ jobSetId = eventMessage.getLeased().getJobSetId();
+ break;
+ case LEASE_RETURNED:
+ jobId = eventMessage.getLeaseReturned().getJobId();
+ jobSetId = eventMessage.getLeaseReturned().getJobSetId();
+ break;
+ case LEASE_EXPIRED:
+ jobId = eventMessage.getLeaseExpired().getJobId();
+ jobSetId = eventMessage.getLeaseExpired().getJobSetId();
+ break;
+ case PENDING:
+ jobId = eventMessage.getPending().getJobId();
+ jobSetId = eventMessage.getPending().getJobSetId();
+ break;
+ case RUNNING:
+ var running = eventMessage.getRunning();
+ jobSetId = running.getJobSetId();
+ jobId = running.getJobId();
+ clusterId = running.getClusterId();
+ podName = running.getPodName();
+ break;
+ case UNABLE_TO_SCHEDULE:
+ jobId = eventMessage.getUnableToSchedule().getJobId();
+ jobSetId = eventMessage.getUnableToSchedule().getJobSetId();
+ reason = eventMessage.getUnableToSchedule().getReason();
+ break;
+ case FAILED:
+ jobId = eventMessage.getFailed().getJobId();
+ jobSetId = eventMessage.getFailed().getJobSetId();
+ reason = eventMessage.getFailed().getReason();
+ cause = eventMessage.getFailed().getCause();
+ break;
+ case SUCCEEDED:
+ jobId = eventMessage.getSucceeded().getJobId();
+ jobSetId = eventMessage.getSucceeded().getJobSetId();
+ break;
+ case REPRIORITIZED:
+ jobId = eventMessage.getReprioritized().getJobId();
+ jobSetId = eventMessage.getReprioritized().getJobSetId();
+ break;
+ case CANCELLING:
+ jobId = eventMessage.getCancelling().getJobId();
+ jobSetId = eventMessage.getCancelling().getJobSetId();
+ reason = eventMessage.getCancelling().getReason();
+ break;
+ case CANCELLED:
+ jobId = eventMessage.getCancelled().getJobId();
+ jobSetId = eventMessage.getCancelled().getJobSetId();
+ reason = eventMessage.getCancelled().getReason();
+ break;
+ case UTILISATION:
+ jobId = eventMessage.getUtilisation().getJobId();
+ jobSetId = eventMessage.getUtilisation().getJobSetId();
+ break;
+ case INGRESS_INFO:
+ jobId = eventMessage.getIngressInfo().getJobId();
+ jobSetId = eventMessage.getIngressInfo().getJobSetId();
+ break;
+ case REPRIORITIZING:
+ jobId = eventMessage.getReprioritizing().getJobId();
+ jobSetId = eventMessage.getReprioritizing().getJobSetId();
+ break;
+ case PREEMPTED:
+ jobId = eventMessage.getPreempted().getJobId();
+ jobSetId = eventMessage.getPreempted().getJobSetId();
+ reason = eventMessage.getPreempted().getReason();
+ break;
+ case PREEMPTING:
+ jobId = eventMessage.getPreempting().getJobId();
+ jobSetId = eventMessage.getPreempting().getJobSetId();
+ reason = eventMessage.getPreempted().getReason();
+ }
+
+ return new ArmadaJobMetadata(jobSetId, jobId, podName, clusterId, reason, cause);
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaEventWatcher.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaEventWatcher.java
new file mode 100644
index 000000000..94d9c7b8c
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaEventWatcher.java
@@ -0,0 +1,9 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import api.EventOuterClass;
+
+public interface ArmadaEventWatcher {
+ void onClose();
+
+ void onEvent(EventOuterClass.EventMessage message);
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaGarbageCollection.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaGarbageCollection.java
new file mode 100644
index 000000000..49d20a78b
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaGarbageCollection.java
@@ -0,0 +1,29 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import hudson.Extension;
+import hudson.model.AsyncPeriodicWork;
+import hudson.model.TaskListener;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+public class ArmadaGarbageCollection {
+ @Extension
+ public static final class PeriodicGarbageCollection extends AsyncPeriodicWork {
+ public PeriodicGarbageCollection() {
+ super("Periodic cleanup of armada plugin state and jobs");
+ }
+
+ @Override
+ protected void execute(TaskListener listener) throws IOException, InterruptedException {
+ var state = ArmadaState.getInstance();
+ state.runCleanup();
+ state.save();
+ }
+
+ @Override
+ public long getRecurrencePeriod() {
+ return TimeUnit.MINUTES.toMillis(5);
+ }
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobManager.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobManager.java
new file mode 100644
index 000000000..9dc6bb4a7
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobManager.java
@@ -0,0 +1,239 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import api.EventOuterClass;
+import hudson.model.Saveable;
+import io.armadaproject.ArmadaClient;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import org.apache.commons.lang.StringUtils;
+import org.jenkinsci.plugins.kubernetes.auth.KubernetesAuthException;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class ArmadaJobManager implements Serializable, ArmadaClientProvider, ArmadaEventWatcher {
+ private static final Logger LOGGER = Logger.getLogger(ArmadaJobManager.class.getName());
+
+ private transient final Map eventWatchers;
+ private final Object jobSetIdLock;
+ private final ConcurrentMap jobSetManagers;
+ private final Saveable save;
+
+ private volatile String currentJobSetId;
+ private volatile ArmadaClientParameters parameters;
+
+ public ArmadaJobManager(ArmadaClientParameters parameters, Saveable save) {
+ this(parameters, save, new Object(), new ConcurrentHashMap<>(), null);
+ }
+
+ private ArmadaJobManager(ArmadaClientParameters parameters, Saveable save, Object jobSetIdLock, ConcurrentMap jobSetManagers, String currentJobSetId) {
+ this.parameters = parameters;
+ this.save = save;
+ this.jobSetIdLock = jobSetIdLock;
+ this.jobSetManagers = jobSetManagers;
+ this.currentJobSetId = currentJobSetId;
+ this.eventWatchers = new ConcurrentHashMap<>();
+ }
+
+ public Closeable watchEvents(ArmadaEventWatcher eventWatcher) {
+ final var id = UUID.randomUUID().toString();
+ eventWatchers.put(id, eventWatcher);
+ return () -> {
+ var removed = eventWatchers.remove(id);
+ if(removed != null) {
+ removed.onClose();
+ }
+ };
+ }
+
+ public int getValidity() {
+ return parameters.hashCode();
+ }
+
+ public boolean reconfigure(ArmadaClientParameters parameters) {
+ var current = this.parameters;
+ var changed = false;
+
+ if(!current.jobSetStrategy.equals(parameters.jobSetStrategy)) {
+ changed = true;
+ }
+
+ // changed api url, close all jobset managers/kill all jobs
+ if(!current.apiUrl.equals(parameters.apiUrl)) {
+ jobSetManagers.forEach((k, jsm) -> jsm.close());
+ jobSetManagers.clear();
+ changed = true;
+ } else if(current.apiPort != parameters.apiPort ||
+ !current.queue.equals(parameters.queue) ||
+ !current.namespace.equals(parameters.namespace)||
+ !StringUtils.equals(current.credentialsId, parameters.credentialsId)) {
+ jobSetManagers.forEach((k, jsm) -> jsm.reconfigure(parameters.namespace, parameters.queue));
+ changed = true;
+ }
+
+ if(changed) {
+ this.parameters = parameters;
+ }
+
+ return changed;
+ }
+
+ public ArmadaJobMetadata ensurePod(String existingJobSetId, String existingJobId, Pod pod) {
+ ArmadaJobMetadata result = null;
+ try {
+ if (existingJobSetId != null && existingJobId != null) {
+ result = getJobSetManager(existingJobSetId).ensureJob(pod, existingJobId);
+ } else {
+ result = getJobSetManager(computeJobSetId()).ensureJob(pod, null);
+ }
+ } catch(StatusRuntimeException e) {
+ handleGrpcError(parameters.queue, e);
+ }
+
+ trySave();
+ return result;
+ }
+
+ public void cancelJob(String jobSetId, String jobId) {
+ getJobSetManager(jobSetId).cancelJob(jobId);
+ trySave();
+ }
+
+ public boolean hasFailed(String jobSetId, String jobId) {
+ return getJobSetManager(jobSetId).hasFailed(jobId);
+ }
+
+ public boolean hasTerminated(String jobSetId, String jobId) {
+ return getJobSetManager(jobSetId).hasTerminated(jobId);
+ }
+
+ public ArmadaJobMetadata waitUntilRunning(String jobSetId, String jobId, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+ return getJobSetManager(jobSetId).waitUntilRunning(jobId, timeout, unit);
+ }
+
+ public void waitUntilTerminated(String jobSetId, String jobId, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+ getJobSetManager(jobSetId).waitUntilTerminated(jobId, timeout, unit);
+ }
+
+ public void close() {
+ jobSetManagers.forEach((k, jsm) -> jsm.close());
+ jobSetManagers.clear();
+ eventWatchers.forEach((k, v) -> v.onClose());
+ eventWatchers.clear();
+ }
+
+ @Override
+ public ArmadaClient get() throws KubernetesAuthException {
+ return ArmadaState.createClient(parameters);
+ }
+
+ public void cleanupAbandonedJobSets() {
+ var jobSetIds = new HashSet<>(jobSetManagers.keySet());
+ for(var jobSetId : jobSetIds) {
+ var jobSetManager = getJobSetManager(jobSetId);
+ if(jobSetManager.isAbandoned() && !jobSetManager.hasActiveJobs()) {
+ var removed = jobSetManagers.remove(jobSetId);
+ if(removed != null) {
+ removed.close();
+ }
+ }
+ }
+ }
+
+ public void cleanupAbandonedJobs(HashMap> jobsPerJobSetId) {
+ jobsPerJobSetId.keySet().forEach(jobSetId -> {
+ var jobSetManager = jobSetManagers.get(jobSetId);
+ if(jobSetManager != null) {
+ jobSetManager.cleanupAbandonedJobs(jobsPerJobSetId.get(jobSetId));
+ }
+ });
+ }
+
+ protected void evaluateJobSetId() {
+ computeJobSetId();
+ }
+
+ protected Object readResolve() {
+ return new ArmadaJobManager(parameters, save, jobSetIdLock, jobSetManagers, currentJobSetId);
+ }
+
+ private void handleGrpcError(String queue, StatusRuntimeException e) {
+ var code = e.getStatus().getCode();
+ var message = e.getStatus().getDescription();
+ if((code == Status.Code.PERMISSION_DENIED || code == Status.Code.NOT_FOUND) && StringUtils.contains(message, "queue")) {
+ // if not perms for queue or it's not found
+ // make sure we clean up the jobset event listener to prevent it from trying to continuously connect
+ // unlikely the jobset will exists if we can't submit jobs to the configured queue
+
+ invalidateQueueConfig(queue);
+ }
+ throw e;
+ }
+
+ private void invalidateQueueConfig(String queue) {
+ jobSetManagers.forEach((k, v) -> v.invalidateConfig(queue));
+ }
+
+ private synchronized void trySave() {
+ try {
+ save.save();
+ } catch (IOException e) {
+ LOGGER.log(Level.WARNING, "Unable to save ArmadaJobManager state", e);
+ }
+ }
+
+ private String computeJobSetId() {
+ var currentJobSetId = parameters.jobSetStrategy.getCurrentJobSet();
+ synchronized (jobSetIdLock) {
+ if(!currentJobSetId.equals(this.currentJobSetId)) {
+ this.currentJobSetId = currentJobSetId;
+ abandonExpiredJobSetManagers();
+ }
+ }
+ return currentJobSetId;
+ }
+
+ private void abandonExpiredJobSetManagers() {
+ var toAbandon = new HashSet<>(jobSetManagers.keySet());
+ if(currentJobSetId != null) {
+ toAbandon.remove(currentJobSetId);
+ }
+ for(var jobSetId : toAbandon) {
+ getJobSetManager(jobSetId).abandon();
+ }
+ }
+
+ // this will actually initialize the job set manager and start watching for events
+ private ArmadaJobSetManager getJobSetManager(String jobSet) {
+ var params = parameters;
+ var newJobSetManager = new ArmadaJobSetManager(params.queue, jobSet, params.namespace);
+ var result = jobSetManagers.putIfAbsent(jobSet, newJobSetManager);
+ if(result == null) {
+ result = newJobSetManager;
+ trySave();
+ }
+ result.initialize(this, this);
+ return result;
+ }
+
+ @Override
+ public void onClose() {
+ // ignore
+ }
+
+ @Override
+ public void onEvent(EventOuterClass.EventMessage message) {
+ eventWatchers.forEach((k, v) -> {
+ v.onEvent(message);
+ });
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobMetadata.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobMetadata.java
new file mode 100644
index 000000000..9a14df89a
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobMetadata.java
@@ -0,0 +1,67 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import api.EventOuterClass;
+
+public class ArmadaJobMetadata {
+ private final String jobSetId;
+ private final String jobId;
+ private final String podName;
+ private final String clusterId;
+ private final String reason;
+ private final EventOuterClass.Cause cause;
+
+ public ArmadaJobMetadata(String jobSetId, String jobId, String podName, String clusterId) {
+ this(jobSetId, jobId, podName, clusterId, null, null);
+ }
+
+ public ArmadaJobMetadata(String jobSetId, String jobId, String podName, String clusterId, String reason, EventOuterClass.Cause cause) {
+ this.jobSetId = jobSetId;
+ this.jobId = jobId;
+ this.podName = podName;
+ this.clusterId = clusterId;
+ this.reason = reason;
+ this.cause = cause;
+ }
+
+ public String getJobId() {
+ return jobId;
+ }
+
+ public String getPodName() {
+ return podName;
+ }
+
+ public String getClusterId() {
+ return clusterId;
+ }
+
+ public ArmadaJobMetadata mergeWith(final ArmadaJobMetadata other) {
+ String podName;
+ String clusterId;
+ if (other.podName != null) {
+ podName = other.podName;
+ } else {
+ podName = this.podName;
+ }
+
+ if (other.clusterId != null) {
+ clusterId = other.clusterId;
+ } else {
+ clusterId = this.clusterId;
+ }
+
+ return new ArmadaJobMetadata(jobSetId, jobId, podName, clusterId);
+ }
+
+ public String getJobSetId() {
+ return jobSetId;
+ }
+
+ public String getReason() {
+ return reason;
+ }
+
+ public EventOuterClass.Cause getCause() {
+ return cause;
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobNotifier.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobNotifier.java
new file mode 100644
index 000000000..bdc19a861
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobNotifier.java
@@ -0,0 +1,78 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import api.SubmitOuterClass;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+public class ArmadaJobNotifier {
+ public interface Callback {
+ void accept(ArmadaJobMetadata metadata);
+
+ void error(RuntimeException e);
+
+ void cancelled();
+ }
+
+ private static class Key {
+ private final String jobId;
+ private final SubmitOuterClass.JobState state;
+
+ private Key(String jobId, SubmitOuterClass.JobState state) {
+ this.jobId = jobId;
+ this.state = state;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof Key)) return false;
+ Key key = (Key) o;
+ return Objects.equals(jobId, key.jobId) && state == key.state;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(jobId, state);
+ }
+ }
+
+ private final ConcurrentMap> callbacks = new ConcurrentHashMap<>();
+
+ public void close() {
+ callbacks.forEach((jobId, callbacks) -> callbacks.forEach(Callback::cancelled));
+ callbacks.clear();
+ }
+
+ public void subscribe(String jobId, SubmitOuterClass.JobState state, Callback callback) {
+ var key = new Key(jobId, state);
+ callbacks.computeIfAbsent(key, k -> new CopyOnWriteArrayList<>()).add(callback);
+ }
+
+ public void unsubscribe(String jobId, SubmitOuterClass.JobState state, Callback callback) {
+ var key = new Key(jobId, state);
+ var subscriptions = callbacks.get(key);
+ if(subscriptions != null) {
+ subscriptions.remove(callback);
+ if(subscriptions.isEmpty()) {
+ callbacks.remove(key);
+ }
+ }
+ }
+
+ public void notify(ArmadaJobMetadata metadata, SubmitOuterClass.JobState state, RuntimeException error) {
+ var key = new Key(metadata.getJobId(), state);
+ var list = callbacks.get(key);
+ if (list != null) {
+ list.forEach(callback -> {
+ if (error != null) {
+ callback.error(error);
+ } else {
+ callback.accept(metadata);
+ }
+ });
+ }
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetEventWatcher.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetEventWatcher.java
new file mode 100644
index 000000000..5e5f29f1c
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetEventWatcher.java
@@ -0,0 +1,182 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import api.EventOuterClass;
+import com.sun.istack.NotNull;
+import io.armadaproject.ArmadaClient;
+import io.grpc.Context;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+
+public class ArmadaJobSetEventWatcher implements Runnable {
+ private static final Logger LOGGER = Logger.getLogger(ArmadaJobSetEventWatcher.class.getName());
+
+ public interface ArmadaMessageCallback {
+ void onMessage(EventOuterClass.EventStreamMessage eventStreamMessage);
+ }
+
+ private final Object cancellationLock = new Object();
+ private final AtomicReference innerContextRef = new AtomicReference<>();
+ private final AtomicBoolean reconnect = new AtomicBoolean(false);
+ private final AtomicBoolean waitForReconnect = new AtomicBoolean(false);
+ private final ArmadaClientProvider armadaClientProvider;
+ private final ArmadaMessageCallback messageCallback;
+ private volatile String queue;
+ private final String jobSetId;
+ private final String fromMessageId;
+
+ public ArmadaJobSetEventWatcher(ArmadaClientProvider armadaClientProvider, ArmadaMessageCallback messageCallback, String queue, String jobSetId, String fromMessageId) {
+ this.armadaClientProvider = armadaClientProvider;
+ this.messageCallback = messageCallback;
+ this.queue = queue;
+ this.jobSetId = jobSetId;
+ this.fromMessageId = fromMessageId;
+ }
+
+ public void close() {
+ synchronized (cancellationLock) {
+ cancellationLock.notify();
+ }
+ }
+
+ public void forceReconnect(String queue, boolean invalidConfig) {
+ this.queue = queue;
+ reconnect.set(true);
+ waitForReconnect.set(invalidConfig);
+ var innerContext = innerContextRef.get();
+ if(innerContext != null) {
+ innerContext.cancel(null);
+ }
+ }
+
+ @Override
+ public void run() {
+ try(var cancellableContext = Context.current().withCancellation()) {
+ var cancelThread = new Thread(() -> {
+ try {
+ synchronized (cancellationLock) {
+ cancellationLock.wait();
+ }
+ } catch (InterruptedException e) {
+ // left empty
+ }
+ cancellableContext.cancel(null);
+ });
+ cancelThread.setDaemon(true);
+ cancelThread.start();
+
+ cancellableContext.run(() -> {
+ final AtomicReference fromMessageId = new AtomicReference<>(this.fromMessageId);
+ var requestBuilder = EventOuterClass.JobSetRequest.newBuilder()
+ .setId(jobSetId)
+ .setErrorIfMissing(false)
+ .setWatch(true);
+
+ while(!cancellableContext.isCancelled()) {
+ try(var client = getClient()) {
+ var builder = requestBuilder.setQueue(queue);
+ var msgId = fromMessageId.get();
+ if(msgId != null) {
+ builder = builder.setFromMessageId(msgId);
+ }
+ LOGGER.info("Starting to stream events for queue " + queue + " and jobSetId " + jobSetId);
+
+ final var completed = new CountDownLatch(1);
+ final var error = new AtomicReference();
+ final var innerContext = cancellableContext.withCancellation();
+ innerContextRef.set(innerContext);
+ final var request = builder.build();
+
+ innerContext.run(() -> {
+ if(waitForReconnect.get()) {
+ LOGGER.info("Config invalidated, waiting for reconnect...");
+ var cancelled = new CountDownLatch(1);
+ Context.CancellationListener cancellationListener = context -> {
+ cancelled.countDown();
+ };
+ innerContext.addListener(cancellationListener, runnable -> {
+ cancelled.countDown();
+ });
+ try {
+ cancelled.await();
+ } catch (InterruptedException e) {
+ // left empty
+ }
+ innerContext.removeListener(cancellationListener);
+ } else {
+
+ client.streamEvents(request, new StreamObserver<>() {
+ @Override
+ public void onNext(EventOuterClass.EventStreamMessage eventStreamMessage) {
+ try {
+ messageCallback.onMessage(eventStreamMessage);
+ } catch (Throwable e) {
+ // avoid throwing here by all costs, it kills the grpc cancellable and this whole thing falls apart
+ }
+ fromMessageId.set(eventStreamMessage.getId());
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ var isError = true;
+ if (throwable instanceof StatusRuntimeException) {
+ var status = (StatusRuntimeException) throwable;
+ isError = status.getStatus().getCode() != Status.Code.CANCELLED;
+ }
+ if (isError) {
+ LOGGER.log(Level.SEVERE, "Failed to stream events for queue " + queue + " and jobSetId " + jobSetId, throwable);
+ error.set(throwable);
+ }
+ completed.countDown();
+ }
+
+ @Override
+ public void onCompleted() {
+ LOGGER.info("Finished streaming events for queue " + queue + " and jobSetId " + jobSetId);
+ completed.countDown();
+ }
+ });
+ try {
+ if (reconnect.compareAndExchange(true, false)) {
+ innerContext.cancel(null);
+ }
+ completed.await();
+
+ var lastError = error.get();
+ if (lastError != null) {
+ LOGGER.log(Level.WARNING, "error while listening to armada events", lastError);
+ if (lastError instanceof StatusRuntimeException) {
+ var statusEx = (StatusRuntimeException) lastError;
+ var code = statusEx.getStatus().getCode();
+ LOGGER.info("Failed to watch armada jobset " + jobSetId + " code: " + code + " ...waiting before retrying...");
+ Thread.sleep(5000);
+ }
+ }
+ } catch (InterruptedException e) {
+ LOGGER.log(Level.SEVERE, "Thread interrupted while waiting for stream completion");
+ }
+ LOGGER.info("Finished streaming events for queue " + queue + (!cancellableContext.isCancelled() ? " reconnecting..." : ""));
+ }
+ });
+ }
+ }
+ });
+ }
+ }
+
+ private ArmadaClient getClient() {
+ try {
+ return armadaClientProvider.get();
+ } catch (Throwable e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetManager.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetManager.java
new file mode 100644
index 000000000..46b62657a
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetManager.java
@@ -0,0 +1,388 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import api.EventOuterClass;
+import api.Job;
+import api.SubmitOuterClass;
+import io.fabric8.kubernetes.api.model.Pod;
+import jenkins.metrics.api.Metrics;
+import org.apache.commons.lang.StringUtils;
+import org.jenkinsci.plugins.kubernetes.auth.KubernetesAuthException;
+
+import java.io.Serializable;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import static io.armadaproject.jenkins.plugin.MetricNames.metricNameForPodStatus;
+import static io.armadaproject.jenkins.plugin.job.ArmadaClientUtil.*;
+
+public class ArmadaJobSetManager implements Serializable, ArmadaJobSetEventWatcher.ArmadaMessageCallback {
+ private static final Logger LOGGER = Logger.getLogger(ArmadaJobSetManager.class.getName());
+
+ private final ConcurrentMap knownJobs = new ConcurrentHashMap<>();
+ private final String jobSetId;
+ private volatile String queue;
+ private volatile String namespace;
+ private volatile String lastMessageId = null;
+ private volatile boolean abandoned = false;
+
+ private transient volatile ArmadaJobSetEventWatcher watcher;
+ private transient volatile Thread watcherThread;
+ private transient volatile ArmadaClientProvider clientProvider;
+ private transient volatile ArmadaEventWatcher eventWatcher;
+ private transient volatile ArmadaJobNotifier jobNotifier;
+
+ private static class JobStatus {
+ private final ArmadaJobMetadata metadata;
+ private final SubmitOuterClass.JobState state;
+
+ private JobStatus(ArmadaJobMetadata metadata, SubmitOuterClass.JobState state) {
+ this.metadata = metadata;
+ this.state = state;
+ }
+ }
+
+ public ArmadaJobSetManager(String queue, String jobSetId, String namespace) {
+ if(StringUtils.isEmpty(queue)) {
+ throw new IllegalArgumentException("queue is empty");
+ }
+
+ if(StringUtils.isEmpty(jobSetId)) {
+ throw new IllegalArgumentException("jobSetId is empty");
+ }
+
+ if(StringUtils.isEmpty(namespace)) {
+ throw new IllegalArgumentException("namespace is empty");
+ }
+
+ this.queue = queue;
+ this.jobSetId = jobSetId;
+ this.namespace = namespace;
+ }
+
+ public void abandon() {
+ this.abandoned = true;
+ }
+
+ public boolean isAbandoned() {
+ return this.abandoned;
+ }
+
+ public boolean hasActiveJobs() {
+ if(knownJobs.isEmpty()) {
+ return false;
+ }
+
+ if(knownJobs.values().stream().noneMatch(s -> s.state != SubmitOuterClass.JobState.UNKNOWN && !isInTerminalState(s.state))) {
+ return false;
+ }
+
+ try(var client = clientProvider.get()) {
+ var response = client.getJobStatus(Job.JobStatusRequest.newBuilder().addAllJobIds(knownJobs.keySet()).build());
+ return response.getJobStatesMap().values().stream().anyMatch(s -> s != SubmitOuterClass.JobState.UNKNOWN && !isInTerminalState(s));
+ } catch (KubernetesAuthException e) {
+ LOGGER.log(Level.SEVERE, "Error while querying known job statuses jobset " + jobSetId, e);
+ //return false here, this will make sure everything is closed down if run into an error
+ return false;
+ }
+ }
+
+ public void cancelJob(String jobId) {
+ try(var client = clientProvider.get()) {
+ var response = client.getJobStatus(Job.JobStatusRequest.newBuilder().addJobIds(jobId).build());
+ if (!isInTerminalState(response.getJobStatesMap().get(jobId))) {
+ client.cancelJob(SubmitOuterClass.JobCancelRequest.newBuilder()
+ .setQueue(queue)
+ .setJobSetId(jobSetId)
+ .setJobId(jobId)
+ .build());
+
+ String msg = ("Cancelled job id: " + jobId + " with job set id: "
+ + jobSetId);
+ LOGGER.info(msg);
+ } else {
+ String msg = ("No jobs in a non-terminal state for id: " + jobId + " with job set id: "
+ + jobSetId);
+ LOGGER.log(Level.WARNING, msg);
+ }
+ } catch (KubernetesAuthException e) {
+ LOGGER.log(Level.SEVERE, "Error while cancelling job", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public ArmadaJobMetadata ensureJob(Pod pod, String existingJobId) {
+ if(abandoned) {
+ throw new IllegalStateException("JobSet abandoned");
+ }
+
+ try(var client = clientProvider.get()) {
+ tryRefreshJobState(existingJobId);
+ // if the controller was interrupted after creating the pod but before it connected back, then
+ // the pod might already exist and the creating logic must be skipped.
+ var needsSubmit = existingJobId == null;
+ if(existingJobId != null && knownJobs.containsKey(existingJobId)) {
+ var knownJob = knownJobs.getOrDefault(existingJobId, null);
+ needsSubmit = knownJob == null || isInTerminalState(knownJob.state);
+ if(knownJob != null && isInTerminalState(knownJob.state)) {
+ knownJobs.remove(existingJobId);
+ }
+ } else {
+ needsSubmit = true;
+ }
+
+ if(needsSubmit) {
+ ArmadaMapper armadaMapper = new ArmadaMapper(queue, namespace, jobSetId, pod);
+ var jobSubmitResponse = client.submitJob(armadaMapper.createJobSubmitRequest());
+ var jobId = jobSubmitResponse.getJobResponseItems(0).getJobId();
+ knownJobs.put(jobId, new JobStatus(null, SubmitOuterClass.JobState.UNKNOWN));
+ return new ArmadaJobMetadata(jobSetId, jobId, null, null);
+ }
+
+ return new ArmadaJobMetadata(jobSetId, existingJobId, null, null);
+ } catch (KubernetesAuthException e) {
+ LOGGER.log(Level.SEVERE, "Failed to create job", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void initialize(ArmadaClientProvider clientProvider, ArmadaEventWatcher eventWatcher) {
+ if(watcher == null) {
+ this.eventWatcher = eventWatcher;
+ this.clientProvider = clientProvider;
+ this.jobNotifier = new ArmadaJobNotifier();
+
+ // initialize known job states if we are loading from saved state
+ try(var client = clientProvider.get()) {
+ var request = Job.JobStatusRequest.newBuilder().addAllJobIds(knownJobs.keySet()).build();
+ var response = client.getJobStatus(request);
+ var jobStateMap = response.getJobStatesMap();
+ jobStateMap.forEach((jobId, jobState) -> {
+ if(isInTerminalState(jobState)) {
+ knownJobs.remove(jobId);
+ } else {
+ knownJobs.computeIfPresent(jobId, (k, v) -> new JobStatus(v.metadata, jobState));
+ }
+ });
+ } catch (KubernetesAuthException e) {
+ LOGGER.log(Level.SEVERE, "Failed to set armada job client", e);
+ throw new RuntimeException(e);
+ }
+
+ watcher = new ArmadaJobSetEventWatcher(clientProvider, this, queue, jobSetId, lastMessageId);
+ watcherThread = new Thread(watcher);
+ watcherThread.setDaemon(true);
+ watcherThread.start();
+ }
+ }
+
+ public void cleanupAbandonedJobs(Set agentAssociatedJobIds) {
+ knownJobs.keySet().forEach(j -> {
+ if (!agentAssociatedJobIds.contains(j)) {
+ cancelJob(j);
+ }
+ });
+ }
+
+ public void waitUntilTerminated(String jobId, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+ waitUntilState(jobId, ArmadaClientUtil.TERMINAL_STATES, timeout, unit);
+ }
+
+ public ArmadaJobMetadata waitUntilRunning(String jobId, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+ return waitUntilState(jobId, new HashSet<>(List.of(SubmitOuterClass.JobState.RUNNING)), timeout, unit);
+ }
+
+ public boolean hasFailed(String jobId) {
+ var job = knownJobs.getOrDefault(jobId, null);
+ return job != null && isInFailedState(job.state);
+ }
+
+ public boolean hasTerminated(String jobId) {
+ var job = knownJobs.getOrDefault(jobId, null);
+ return job != null && isInTerminalState(job.state);
+ }
+
+ @Override
+ public void onMessage(EventOuterClass.EventStreamMessage eventStreamMessage) {
+ var message = eventStreamMessage.getMessage();
+ var eventsCase = message.getEventsCase();
+ var jobMetadata = extractMetadata(message);
+
+ var jobId = jobMetadata.getJobId();
+ final var state = toJobState(eventsCase);
+
+ Metrics.metricRegistry().counter(metricNameForPodStatus(state.toString())).inc();
+
+ knownJobs.compute(jobId, (k, v) -> updateJobStatus(k, v, jobMetadata, state));
+ if(isInFailedState(state)) {
+ jobNotifier.notify(new ArmadaJobMetadata(jobSetId, jobId, null, null), state, new RuntimeException("Job " + state));
+ } else {
+ jobNotifier.notify(jobMetadata, state, null);
+ }
+
+ if(isInTerminalState(eventsCase)) {
+ knownJobs.remove(jobId);
+ }
+
+ if(eventWatcher != null) {
+ eventWatcher.onEvent(message);
+ }
+
+ lastMessageId = eventStreamMessage.getId();
+ }
+
+ public void invalidateConfig(String queue) {
+ if(StringUtils.equals(queue, this.queue)) {
+ var currentWatcher = watcher;
+ if(currentWatcher != null) {
+ currentWatcher.forceReconnect(queue, true);
+ }
+ }
+ }
+
+ public void reconfigure(String namespace, String queue) {
+ this.namespace = namespace;
+ this.queue = queue;
+ var currentWatcher = watcher;
+ if(currentWatcher != null) {
+ currentWatcher.forceReconnect(queue, false);
+ }
+ }
+
+ public void close() {
+ var currentWatcher = watcher;
+ watcher = null;
+ if(currentWatcher != null) {
+ currentWatcher.close();
+ try {
+ watcherThread.join();
+ } catch (InterruptedException e) {
+ LOGGER.log(Level.SEVERE, "Interrupted while waiting for watcher to complete", e);
+ }
+ }
+
+ if(jobNotifier != null) {
+ jobNotifier.close();
+ }
+
+ if(clientProvider != null) {
+ try (var client = clientProvider.get()) {
+ client.cancelJob(SubmitOuterClass.JobCancelRequest.newBuilder()
+ .setQueue(queue)
+ .setJobSetId(jobSetId)
+ .addAllJobIds(knownJobs.keySet())
+ .build());
+ } catch (KubernetesAuthException e) {
+ LOGGER.log(Level.SEVERE, "Error while cancelling jobs", e);
+ }
+ }
+ }
+
+ private static JobStatus updateJobStatus(String k, JobStatus v, ArmadaJobMetadata jobMetadata, SubmitOuterClass.JobState state) {
+ return new JobStatus(v != null && v.metadata != null ? v.metadata.mergeWith(jobMetadata) : jobMetadata,
+ state != SubmitOuterClass.JobState.UNKNOWN ? state : (v == null ? SubmitOuterClass.JobState.UNKNOWN : v.state));
+ }
+
+ private ArmadaJobMetadata waitUntilState(String jobId, Set states, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
+ var job = knownJobs.getOrDefault(jobId, null);
+ if(job == null) {
+ throw new IllegalArgumentException("Unknown job: " + jobId);
+ }
+
+ // check if we already know the job is in required state
+ if(states.contains(job.state)) {
+ return job.metadata;
+ }
+
+
+ final ArmadaJobMetadata preWaitMetadata = job.metadata;
+ final AtomicReference metadata = new AtomicReference<>(job.metadata);
+ final AtomicBoolean cancelled = new AtomicBoolean(false);
+ final AtomicReference error = new AtomicReference<>();
+ final CountDownLatch latch = new CountDownLatch(1);
+ ArmadaJobNotifier.Callback callback = new ArmadaJobNotifier.Callback() {
+ @Override
+ public void accept(ArmadaJobMetadata newMetadata) {
+ metadata.set(newMetadata);
+ latch.countDown();
+ }
+
+ @Override
+ public void error(RuntimeException e) {
+ error.set(e);
+ latch.countDown();
+ }
+
+ @Override
+ public void cancelled() {
+ cancelled.set(true);
+ latch.countDown();
+ }
+ };
+ // subscribe to get state events
+ states.forEach(state -> jobNotifier.subscribe(jobId, state, callback));
+
+ var timedOut = false;
+ // check again if it maybe already in required state before waiting
+ // if the job terminated it has already been removed
+ job = knownJobs.getOrDefault(jobId, null);
+ if(job == null) {
+ states.forEach(state -> jobNotifier.unsubscribe(jobId, state, callback));
+ if(!states.stream().anyMatch(ArmadaClientUtil::isInTerminalState)) {
+ throw new IllegalArgumentException("Job already terminated: " + jobId);
+ }
+ return preWaitMetadata;
+ } else if(!states.contains(job.state)) {
+ timedOut = !latch.await(timeout, unit);
+ } else {
+ states.forEach(state -> jobNotifier.unsubscribe(jobId, state, callback));
+ return job.metadata;
+ }
+
+ if(!cancelled.get()) {
+ states.forEach(state -> jobNotifier.unsubscribe(jobId, state, callback));
+ } else {
+ throw new CancellationException("Armada event watcher cancelled");
+ }
+
+ var err = error.get();
+ if(err != null) {
+ throw err;
+ }
+
+ if(timedOut) {
+ job = knownJobs.get(jobId);
+ if(job == null && states.stream().anyMatch(ArmadaClientUtil::isInTerminalState)) {
+ return preWaitMetadata;
+ }
+ else if(job != null && states.contains(job.state)) {
+ return job.metadata;
+ } else {
+ throw new TimeoutException("Timed out waiting for job: " + jobId);
+ }
+ } else {
+ return metadata.get();
+ }
+ }
+
+ private void tryRefreshJobState(String jobId)
+ {
+ if(jobId != null) {
+ var request = Job.JobStatusRequest.newBuilder().addJobIds(jobId).build();
+ try (var client = clientProvider.get()) {
+ var response = client.getJobStatus(request);
+ var jobStateMap = response.getJobStatesMap();
+ if (jobStateMap.containsKey(jobId)) {
+ knownJobs.compute(jobId, (k, v) -> updateJobStatus(k, v, null, jobStateMap.get(jobId)));
+ }
+ } catch (KubernetesAuthException e) {
+ LOGGER.log(Level.SEVERE, "Error while cancelling job", e);
+ }
+ }
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetStrategy.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetStrategy.java
new file mode 100644
index 000000000..dccab6bfa
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaJobSetStrategy.java
@@ -0,0 +1,5 @@
+package io.armadaproject.jenkins.plugin.job;
+
+public interface ArmadaJobSetStrategy {
+ String getCurrentJobSet();
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaLaunchFailedOfflineCause.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaLaunchFailedOfflineCause.java
new file mode 100644
index 000000000..edb39cbac
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaLaunchFailedOfflineCause.java
@@ -0,0 +1,17 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import hudson.slaves.OfflineCause;
+import org.kohsuke.stapler.export.Exported;
+
+public class ArmadaLaunchFailedOfflineCause extends OfflineCause {
+ public final String description;
+
+ public ArmadaLaunchFailedOfflineCause(String description) {
+ this.description = description;
+ }
+
+ @Exported(name = "description")
+ public String toString() {
+ return this.description;
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaState.java b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaState.java
new file mode 100644
index 000000000..0c2c163c5
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/ArmadaState.java
@@ -0,0 +1,208 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import com.cloudbees.plugins.credentials.common.StandardCredentials;
+import hudson.Extension;
+import hudson.XmlFile;
+import hudson.model.AsyncPeriodicWork;
+import hudson.model.Saveable;
+import hudson.model.TaskListener;
+import hudson.model.listeners.SaveableListener;
+import io.armadaproject.ArmadaClient;
+import io.armadaproject.jenkins.plugin.ArmadaCloud;
+import io.armadaproject.jenkins.plugin.ArmadaComputer;
+import io.armadaproject.jenkins.plugin.ArmadaSlave;
+import jenkins.model.Jenkins;
+import org.jenkinsci.plugins.kubernetes.auth.KubernetesAuthException;
+import org.jenkinsci.plugins.plaincredentials.StringCredentials;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import static io.armadaproject.jenkins.plugin.KubernetesFactoryAdapter.resolveCredentials;
+
+public class ArmadaState implements Saveable, Serializable {
+ private static final Logger LOGGER = Logger.getLogger(ArmadaState.class.getName());
+
+ private static ArmadaState instance;
+ private final ConcurrentMap jobManagers = new ConcurrentHashMap<>();
+
+ @Extension
+ public static final class PeriodicSave extends AsyncPeriodicWork {
+ public PeriodicSave() {
+ super("Periodic save of armada plugin state");
+ }
+
+ @Override
+ protected void execute(TaskListener listener) throws IOException, InterruptedException {
+ ArmadaState.getInstance().save();
+ }
+
+ @Override
+ public long getRecurrencePeriod() {
+ return TimeUnit.SECONDS.toMillis(60);
+ }
+ }
+
+ @Extension(ordinal = 1)
+ public static class SaveableListenerImpl extends SaveableListener {
+ @Override
+ public void onChange(Saveable o, XmlFile file) {
+ if (o instanceof Jenkins) {
+ Jenkins jenkins = (Jenkins) o;
+ getInstance().reconfigure(jenkins.clouds.getAll(ArmadaCloud.class));
+ }
+ super.onChange(o, file);
+ }
+ }
+
+ protected ArmadaState() {
+ LOGGER.info("ArmadaState created");
+ }
+
+ public static ArmadaJobManager getJobManager(ArmadaCloud cloud) {
+ return getInstance().doGetJobManager(cloud);
+ }
+
+ private synchronized void reconfigure(List clouds) {
+ var keys = new HashSet<>(jobManagers.keySet());
+ var changed = false;
+ for (var cloud : clouds) {
+ String displayName = cloud.getDisplayName();
+ var jobManager = getJobManager(cloud);
+ changed = jobManager.reconfigure(toParameters(cloud));
+ keys.remove(displayName);
+ }
+ for(var cloudName : keys) {
+ changed = true;
+ jobManagers.remove(cloudName).close();
+ }
+
+ if(changed) {
+ trySave();
+ }
+ }
+
+ private static ArmadaClientParameters toParameters(ArmadaCloud cloud) {
+ return new ArmadaClientParameters(
+ cloud.getArmadaUrl(),
+ Integer.parseInt(cloud.getArmadaPort()),
+ cloud.getArmadaQueue(),
+ cloud.getArmadaNamespace(),
+ cloud.getArmadaCredentialsId(),
+ cloud.getJobSetStrategy()
+ );
+ }
+
+ private ArmadaJobManager doGetJobManager(ArmadaCloud cloud) {
+ jobManagers.computeIfAbsent(cloud.getDisplayName(), (cloudName) -> new ArmadaJobManager(toParameters(cloud), this));
+ trySave();
+ return jobManagers.get(cloud.getDisplayName());
+ }
+
+ public static ArmadaClient createClient(ArmadaClientParameters params) throws KubernetesAuthException {
+ if(params.credentialsId == null) {
+ return new ArmadaClient(params.apiUrl, params.apiPort);
+ }
+
+ StandardCredentials standardCredentials = resolveCredentials(params.credentialsId);
+ if (!(standardCredentials instanceof StringCredentials)) {
+ throw new KubernetesAuthException("credentials not a string credentials");
+ }
+
+ String secret = ((StringCredentials) standardCredentials).getSecret().getPlainText();
+
+ return new ArmadaClient(params.apiUrl, params.apiPort, secret);
+ }
+
+ @Override
+ public synchronized void save() throws IOException {
+ getConfigFile().write(this);
+ }
+
+ protected void runCleanup() {
+ var activeAgents = new HashMap>();
+ Arrays.stream(Jenkins.get().getComputers()).filter(c -> c instanceof ArmadaComputer).forEach(c -> {
+ var agent = ((ArmadaComputer)c).getNode();
+ activeAgents.computeIfAbsent(agent.getCloudName(), (cloudName) -> new ArrayList<>()).add(agent);
+ });
+
+ jobManagers.forEach((cloudName, jobManager) -> {
+ try {
+ if (activeAgents.containsKey(cloudName)) {
+ jobManager.evaluateJobSetId();
+ var jobsPerJobSet = new HashMap>();
+ activeAgents.get(cloudName).forEach(a -> {
+ var jobSetId = a.getArmadaJobSetId();
+ var jobId = a.getArmadaJobId();
+ if (jobSetId != null && jobId != null) {
+ jobsPerJobSet.computeIfAbsent(jobSetId, (c) -> new HashSet<>()).add(jobId);
+ }
+ });
+ jobManager.cleanupAbandonedJobs(jobsPerJobSet);
+ }
+ } catch(Throwable e) {
+ LOGGER.log(Level.WARNING, "Failed to clean up abandoned jobs", e);
+ }
+
+ jobManager.cleanupAbandonedJobSets();
+ });
+ }
+
+ protected static synchronized ArmadaState getInstance() {
+ if (instance == null) {
+ var configFile = getConfigFile();
+ if (!configFile.exists()) {
+ instance = createNewAndSave();
+ } else {
+
+ try {
+ instance = (ArmadaState) configFile.read();
+ } catch (Throwable e) {
+ try {
+ configFile.delete();
+ } catch (Throwable ex) {
+ LOGGER.log(Level.SEVERE, "Failed to delete config file", ex);
+ throw new RuntimeException(e);
+ }
+ instance = createNewAndSave();
+ }
+ }
+ }
+ return instance;
+ }
+
+ private static ArmadaState createNewAndSave() {
+ ArmadaState result;
+ result = new ArmadaState();
+ try {
+ result.save();
+ } catch(Throwable ex) {
+ LOGGER.log(Level.SEVERE, "Failed to save empty config file", ex);
+ }
+ return result;
+ }
+
+ private void trySave() {
+ try {
+ save();
+ } catch (IOException e) {
+ LOGGER.log(Level.WARNING, "Failed to save armada state", e);
+ }
+ }
+
+ private static XmlFile getConfigFile() {
+ var dir = new File(Jenkins.get().getRootDir(), "armada-plugin");
+ if (!dir.exists()) {
+ //noinspection ResultOfMethodCallIgnored
+ dir.mkdirs();
+ }
+ return new XmlFile(Jenkins.XSTREAM, new File(dir, "job-manager.xml"));
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/job/DailyArmadaJobSetStrategy.java b/src/main/java/io/armadaproject/jenkins/plugin/job/DailyArmadaJobSetStrategy.java
new file mode 100644
index 000000000..ea78eada7
--- /dev/null
+++ b/src/main/java/io/armadaproject/jenkins/plugin/job/DailyArmadaJobSetStrategy.java
@@ -0,0 +1,30 @@
+package io.armadaproject.jenkins.plugin.job;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Objects;
+
+public class DailyArmadaJobSetStrategy implements ArmadaJobSetStrategy {
+ private final String jobSetPrefix;
+
+ public DailyArmadaJobSetStrategy(String jobSetPrefix) {
+ this.jobSetPrefix = jobSetPrefix;
+ }
+
+ @Override
+ public String getCurrentJobSet() {
+ return jobSetPrefix + new SimpleDateFormat("-ddMMyyyy").format(new Date());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof DailyArmadaJobSetStrategy)) return false;
+ DailyArmadaJobSetStrategy that = (DailyArmadaJobSetStrategy) o;
+ return Objects.equals(jobSetPrefix, that.jobSetPrefix);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(jobSetPrefix);
+ }
+}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pipeline/ArmadaDeclarativeAgent.java b/src/main/java/io/armadaproject/jenkins/plugin/pipeline/ArmadaDeclarativeAgent.java
index aa307591a..bc5fa3bb7 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pipeline/ArmadaDeclarativeAgent.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pipeline/ArmadaDeclarativeAgent.java
@@ -9,7 +9,6 @@
import hudson.util.ListBoxModel;
import io.armadaproject.jenkins.plugin.ContainerTemplate;
import io.armadaproject.jenkins.plugin.PodTemplate;
-import io.armadaproject.jenkins.plugin.pod.retention.PodRetention;
import io.armadaproject.jenkins.plugin.pod.yaml.YamlMergeStrategy;
import java.util.Collections;
import java.util.List;
@@ -72,9 +71,6 @@ public class ArmadaDeclarativeAgent extends RetryableDeclarativeAgent containerTemplates;
@@ -287,18 +283,6 @@ public void setSlaveConnectTimeout(int slaveConnectTimeout) {
this.slaveConnectTimeout = slaveConnectTimeout;
}
- public PodRetention getPodRetention() {
- return this.podRetention == null ? ArmadaPodTemplateStep.DescriptorImpl.defaultPodRetention : this.podRetention;
- }
-
- @DataBoundSetter
- public void setPodRetention(@CheckForNull PodRetention podRetention) {
- this.podRetention =
- (podRetention == null || podRetention.equals(ArmadaPodTemplateStep.DescriptorImpl.defaultPodRetention))
- ? null
- : podRetention;
- }
-
public String getYamlFile() {
return yamlFile;
}
@@ -439,9 +423,6 @@ public Map getAsArgs() {
if (slaveConnectTimeout != 0) {
argMap.put("slaveConnectTimeout", slaveConnectTimeout);
}
- if (podRetention != null) {
- argMap.put("podRetention", podRetention);
- }
if (instanceCap > 0 && instanceCap < Integer.MAX_VALUE) {
argMap.put("instanceCap", instanceCap);
}
@@ -468,7 +449,6 @@ public static class DescriptorImpl extends DeclarativeAgentDescriptor serverUrl = new AtomicReference<>();
- try (ArmadaClient armadaClient = kubernetesCloud.connectToArmada()) {
- JobSetRequest jobSetRequest = JobSetRequest.newBuilder()
- .setId(kubernetesCloud.getArmadaJobSetPrefix()
- + kubernetesCloud.getArmadaJobSetId())
- .setQueue(kubernetesCloud.getArmadaQueue())
- .setErrorIfMissing(true)
- .build();
-
- armadaClient.getEvents(jobSetRequest).forEachRemaining(e -> {
- EventMessage message = e.getMessage();
- // FIXME add wait mechanism
- if (message.getRunning().getJobId().equals(kubernetesSlave.getArmadaJobId())) {
- String clusterId = message.getRunning().getClusterId();
- try {
- serverUrl.set(
- ClusterConfigParser.parse(kubernetesCloud.getArmadaClusterConfigPath())
- .get(clusterId));
-
- } catch (Exception ex) {
- throw new RuntimeException("Failed to parse cluster config file", ex);
- }
-
- namespace = message.getRunning().getPodNamespace();
- podName = message.getRunning().getPodName();
- }
- });
- }
-
- return kubernetesCloud.connect(serverUrl.get(), namespace);
+ return getKubernetesSlave().connect();
}
- private KubernetesSlave getKubernetesSlave() throws IOException, InterruptedException {
+ private ArmadaSlave getKubernetesSlave() throws IOException, InterruptedException {
Node node = context.get(Node.class);
- if (!(node instanceof KubernetesSlave)) {
+ if (!(node instanceof ArmadaSlave)) {
throw new AbortException(
String.format("Node is not a Armada node: %s", node != null ? node.getNodeName() : null));
}
- return (KubernetesSlave) node;
+ return (ArmadaSlave) node;
}
}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pipeline/PodTemplateContext.java b/src/main/java/io/armadaproject/jenkins/plugin/pipeline/PodTemplateContext.java
index d685c791f..c2ffb5564 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pipeline/PodTemplateContext.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pipeline/PodTemplateContext.java
@@ -8,18 +8,12 @@
public class PodTemplateContext implements Serializable {
private static final long serialVersionUID = 3065143885759619305L;
- private final String namespace;
private final String name;
- public PodTemplateContext(String namespace, String name) {
- this.namespace = namespace;
+ public PodTemplateContext(String name) {
this.name = name;
}
- public String getNamespace() {
- return namespace;
- }
-
public String getName() {
return name;
}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pipeline/PodTemplateStepExecution.java b/src/main/java/io/armadaproject/jenkins/plugin/pipeline/PodTemplateStepExecution.java
index eff367f8a..0145297d7 100755
--- a/src/main/java/io/armadaproject/jenkins/plugin/pipeline/PodTemplateStepExecution.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pipeline/PodTemplateStepExecution.java
@@ -2,7 +2,6 @@
import static java.util.stream.Collectors.toList;
-import edu.umd.cs.findbugs.annotations.CheckForNull;
import edu.umd.cs.findbugs.annotations.NonNull;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import hudson.AbortException;
@@ -13,7 +12,7 @@
import hudson.slaves.Cloud;
import io.armadaproject.jenkins.plugin.ContainerTemplate;
import io.armadaproject.jenkins.plugin.ArmadaCloud;
-import io.armadaproject.jenkins.plugin.KubernetesFolderProperty;
+import io.armadaproject.jenkins.plugin.ArmadaFolderProperty;
import io.armadaproject.jenkins.plugin.PodAnnotation;
import io.armadaproject.jenkins.plugin.PodImagePullSecret;
import io.armadaproject.jenkins.plugin.PodTemplate;
@@ -91,11 +90,9 @@ public boolean start() throws Exception {
stepName = label;
}
String name = String.format(NAME_FORMAT, stepName, randString);
- String namespace = checkNamespace(cloud, podTemplateContext);
newTemplate = new PodTemplate();
newTemplate.setName(name);
- newTemplate.setNamespace(namespace);
if (step.getInheritFrom() == null) {
newTemplate.setInheritFrom(PodTemplateUtils.emptyToNull(parentTemplates));
@@ -143,7 +140,6 @@ public boolean start() throws Exception {
}
newTemplate.setAgentInjection(step.isAgentInjection());
newTemplate.setAgentContainer(step.getAgentContainer());
- newTemplate.setPodRetention(step.getPodRetention());
if (step.getActiveDeadlineSeconds() != 0) {
newTemplate.setActiveDeadlineSeconds(step.getActiveDeadlineSeconds());
@@ -166,7 +162,7 @@ public boolean start() throws Exception {
cloud.addDynamicTemplate(newTemplate);
BodyInvoker invoker = getContext()
.newBodyInvoker()
- .withContexts(step, new PodTemplateContext(namespace, name))
+ .withContexts(step, new PodTemplateContext(name))
.withCallback(new PodTemplateCallback(newTemplate, cloudName));
if (step.getLabel() == null) {
invoker.withContext(EnvironmentExpander.merge(
@@ -215,25 +211,12 @@ private void checkAccess(Run run, ArmadaCloud armadaCloud) throws AbortExc
ItemGroup parent = job.getParent(); // Get the Parent of the Job (which might be a Folder)
Set allowedClouds = new HashSet<>();
- KubernetesFolderProperty.collectAllowedClouds(allowedClouds, parent);
+ ArmadaFolderProperty.collectAllowedClouds(allowedClouds, parent);
if (!allowedClouds.contains(armadaCloud.name)) {
throw new AbortException(String.format("Not authorized to use Kubernetes cloud: %s", step.getCloud()));
}
}
- private String checkNamespace(
- ArmadaCloud armadaCloud, @CheckForNull PodTemplateContext podTemplateContext) {
- String namespace = null;
- if (!PodTemplateUtils.isNullOrEmpty(step.getNamespace())) {
- namespace = step.getNamespace();
- } else if (podTemplateContext != null && !PodTemplateUtils.isNullOrEmpty(podTemplateContext.getNamespace())) {
- namespace = podTemplateContext.getNamespace();
- } else {
- namespace = armadaCloud.getNamespace();
- }
- return namespace;
- }
-
/**
* Re-inject the dynamic template when resuming the pipeline
*/
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pipeline/SecretsMasker.java b/src/main/java/io/armadaproject/jenkins/plugin/pipeline/SecretsMasker.java
index 40f816d0d..2da77c9fb 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pipeline/SecretsMasker.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pipeline/SecretsMasker.java
@@ -20,8 +20,8 @@
import hudson.Extension;
import hudson.remoting.Channel;
import hudson.util.LogTaskListener;
-import io.armadaproject.jenkins.plugin.KubernetesComputer;
-import io.armadaproject.jenkins.plugin.KubernetesSlave;
+import io.armadaproject.jenkins.plugin.ArmadaComputer;
+import io.armadaproject.jenkins.plugin.ArmadaSlave;
import io.armadaproject.jenkins.plugin.PodTemplate;
import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.EnvVar;
@@ -77,7 +77,7 @@ public OutputStream decorate(OutputStream logger) throws IOException, Interrupte
@Extension
public static final class Factory extends DynamicContext.Typed {
- private final Map> secrets = new WeakHashMap<>();
+ private final Map> secrets = new WeakHashMap<>();
@Override
protected Class type() {
@@ -86,9 +86,9 @@ protected Class type() {
@Override
protected TaskListenerDecorator get(DelegatedContext context) throws IOException, InterruptedException {
- KubernetesComputer c;
+ ArmadaComputer c;
try {
- c = context.get(KubernetesComputer.class);
+ c = context.get(ArmadaComputer.class);
} catch (IOException | InterruptedException x) {
LOGGER.log(Level.FINE, "Unable to look up KubernetesComputer", x);
return null;
@@ -122,13 +122,13 @@ protected TaskListenerDecorator get(DelegatedContext context) throws IOException
}
}
- private static @CheckForNull Set secretsOf(KubernetesComputer c)
+ private static @CheckForNull Set secretsOf(ArmadaComputer c)
throws IOException, InterruptedException {
Channel ch = c.getChannel();
if (ch == null) {
return null;
}
- KubernetesSlave slave = c.getNode();
+ ArmadaSlave slave = c.getNode();
if (slave == null) {
return null;
}
@@ -165,8 +165,7 @@ protected TaskListenerDecorator get(DelegatedContext context) throws IOException
return null;
}
try (OutputStream errs = new LogTaskListener(LOGGER, Level.FINE).getLogger();
- ExecWatch exec = slave.getKubernetesCloud()
- .connect()
+ ExecWatch exec = slave.connect()
.pods()
.inNamespace(slave.getNamespace())
.withName(slave.getPodName())
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Always.java b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Always.java
index 9f74ca14c..e69de29bb 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Always.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Always.java
@@ -1,57 +0,0 @@
-package io.armadaproject.jenkins.plugin.pod.retention;
-
-import hudson.Extension;
-import io.armadaproject.jenkins.plugin.ArmadaCloud;
-import io.fabric8.kubernetes.api.model.Pod;
-import java.io.Serializable;
-import java.util.function.Supplier;
-import org.jenkinsci.Symbol;
-import org.kohsuke.stapler.DataBoundConstructor;
-
-public class Always extends PodRetention implements Serializable {
-
- private static final long serialVersionUID = -3363056751880572952L;
-
- @DataBoundConstructor
- public Always() {}
-
- @Override
- public boolean shouldDeletePod(ArmadaCloud cloud, Supplier pod) {
- return false;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null) {
- return false;
- }
-
- if (obj instanceof Always) {
- return true;
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return this.toString().hashCode();
- }
-
- @Override
- public String toString() {
- return Messages.always();
- }
-
- @Extension
- @Symbol("always")
- public static class DescriptorImpl extends PodRetentionDescriptor {
-
- @Override
- public String getDisplayName() {
- return Messages.always();
- }
- }
-}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Default.java b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Default.java
index 24a07bb4a..e69de29bb 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Default.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Default.java
@@ -1,75 +0,0 @@
-package io.armadaproject.jenkins.plugin.pod.retention;
-
-import hudson.Extension;
-import hudson.model.Descriptor;
-import hudson.model.DescriptorVisibilityFilter;
-import io.fabric8.kubernetes.api.model.Pod;
-import java.io.Serializable;
-import java.util.function.Supplier;
-import io.armadaproject.jenkins.plugin.ArmadaCloud;
-import org.jenkinsci.Symbol;
-import org.kohsuke.stapler.DataBoundConstructor;
-
-public class Default extends PodRetention implements Serializable {
-
- private static final long serialVersionUID = -5209499689925746138L;
-
- @DataBoundConstructor
- public Default() {}
-
- @Override
- public boolean shouldDeletePod(ArmadaCloud cloud, Supplier pod) {
- PodRetention parent = cloud.getPodRetention();
- if (!(parent instanceof Default)) {
- return parent.shouldDeletePod(cloud, pod);
- }
- return true;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null) {
- return false;
- }
- if (obj instanceof Default) {
- return true;
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return this.toString().hashCode();
- }
-
- @Override
- public String toString() {
- return Messages._default();
- }
-
- @Extension
- public static class FilterImpl extends DescriptorVisibilityFilter {
-
- @Override
- @SuppressWarnings("rawtypes")
- public boolean filter(Object context, Descriptor descriptor) {
- if (context instanceof ArmadaCloud.DescriptorImpl && descriptor instanceof DescriptorImpl) {
- return false;
- }
- return true;
- }
- }
-
- @Extension
- @Symbol("default")
- public static class DescriptorImpl extends PodRetentionDescriptor {
-
- @Override
- public String getDisplayName() {
- return Messages._default();
- }
- }
-}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Never.java b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Never.java
index f0670235c..e69de29bb 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Never.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Never.java
@@ -1,56 +0,0 @@
-package io.armadaproject.jenkins.plugin.pod.retention;
-
-import hudson.Extension;
-import io.armadaproject.jenkins.plugin.ArmadaCloud;
-import io.fabric8.kubernetes.api.model.Pod;
-import java.io.Serializable;
-import java.util.function.Supplier;
-import org.jenkinsci.Symbol;
-import org.kohsuke.stapler.DataBoundConstructor;
-
-public class Never extends PodRetention implements Serializable {
-
- private static final long serialVersionUID = -7127652621214283411L;
-
- @DataBoundConstructor
- public Never() {}
-
- @Override
- public boolean shouldDeletePod(ArmadaCloud cloud, Supplier pod) {
- return true;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null) {
- return false;
- }
- if (obj instanceof Never) {
- return true;
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return this.toString().hashCode();
- }
-
- @Override
- public String toString() {
- return Messages.never();
- }
-
- @Extension
- @Symbol("never")
- public static class DescriptorImpl extends PodRetentionDescriptor {
-
- @Override
- public String getDisplayName() {
- return Messages.never();
- }
- }
-}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/OnFailure.java b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/OnFailure.java
index 6c5fede69..e69de29bb 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/OnFailure.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/OnFailure.java
@@ -1,72 +0,0 @@
-package io.armadaproject.jenkins.plugin.pod.retention;
-
-import hudson.Extension;
-import io.armadaproject.jenkins.plugin.ArmadaCloud;
-import io.fabric8.kubernetes.api.model.Pod;
-import java.io.Serializable;
-import java.util.Locale;
-import java.util.function.Supplier;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-import org.jenkinsci.Symbol;
-import org.kohsuke.stapler.DataBoundConstructor;
-
-public class OnFailure extends PodRetention implements Serializable {
-
- private static final long serialVersionUID = 6424267627207206819L;
-
- private static final Logger LOGGER = Logger.getLogger(OnFailure.class.getName());
-
- @DataBoundConstructor
- public OnFailure() {}
-
- @Override
- public boolean shouldDeletePod(ArmadaCloud cloud, Supplier podS) {
- Pod pod = null;
- try {
- pod = podS.get();
- } catch (RuntimeException x) {
- LOGGER.log(Level.WARNING, null, x);
- }
- if (pod == null || pod.getStatus() == null) {
- return false;
- }
- boolean hasErrors =
- pod.getStatus().getPhase().toLowerCase(Locale.getDefault()).matches("(failed|unknown)");
- return !hasErrors;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null) {
- return false;
- }
- if (obj instanceof OnFailure) {
- return true;
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- return this.toString().hashCode();
- }
-
- @Override
- public String toString() {
- return Messages.on_Failure();
- }
-
- @Extension
- @Symbol("onFailure")
- public static class DescriptorImpl extends PodRetentionDescriptor {
-
- @Override
- public String getDisplayName() {
- return Messages.on_Failure();
- }
- }
-}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/PodRetention.java b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/PodRetention.java
index 279e2766a..e69de29bb 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/PodRetention.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/PodRetention.java
@@ -1,50 +0,0 @@
-package io.armadaproject.jenkins.plugin.pod.retention;
-
-import hudson.ExtensionPoint;
-import hudson.model.AbstractDescribableImpl;
-import io.fabric8.kubernetes.api.model.Pod;
-import java.util.function.Supplier;
-import io.armadaproject.jenkins.plugin.ArmadaCloud;
-
-/**
- * PodRetention instances determine if the Kubernetes pod running a Jenkins agent
- * should be deleted after Jenkins terminates the agent.
- *
- * Custom pod retention behavior can be added by extending this class, including a descriptor
- * that extends {@link PodRetentionDescriptor}
- */
-public abstract class PodRetention extends AbstractDescribableImpl implements ExtensionPoint {
-
- /**
- * Returns the default PodRetention for a KubernetesCloud instance.
- *
- * @return the {@link Never} PodRetention strategy.
- */
- public static PodRetention getKubernetesCloudDefault() {
- return new Never();
- }
-
- /**
- * Returns the default PodRetention for a PodTemplate instance.
- *
- * @return the {@link Default} PodRetention strategy.
- */
- public static PodRetention getPodTemplateDefault() {
- return new Default();
- }
-
- /**
- * Determines if a agent pod should be deleted after the Jenkins build completes.
- *
- * @param cloud - the {@link ArmadaCloud} the agent pod belongs to.
- * @param pod - the {@link Pod} running the Jenkins build.
- *
- * @return true if the agent pod should be deleted.
- */
- public abstract boolean shouldDeletePod(ArmadaCloud cloud, Supplier pod);
-
- @Override
- public String toString() {
- return getClass().getSimpleName();
- }
-}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/PodRetentionDescriptor.java b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/PodRetentionDescriptor.java
index 5ebb99fbe..e69de29bb 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/PodRetentionDescriptor.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/PodRetentionDescriptor.java
@@ -1,8 +0,0 @@
-package io.armadaproject.jenkins.plugin.pod.retention;
-
-import hudson.model.Descriptor;
-
-/**
- * A {@link Descriptor} for any {@link PodRetention} implementation.
- */
-public abstract class PodRetentionDescriptor extends Descriptor {}
diff --git a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Reaper.java b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Reaper.java
index 6657ea249..c0f1cc14c 100644
--- a/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Reaper.java
+++ b/src/main/java/io/armadaproject/jenkins/plugin/pod/retention/Reaper.java
@@ -16,10 +16,10 @@
package io.armadaproject.jenkins.plugin.pod.retention;
+import api.EventOuterClass;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;
-import edu.umd.cs.findbugs.annotations.CheckForNull;
import edu.umd.cs.findbugs.annotations.NonNull;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import hudson.Extension;
@@ -35,29 +35,24 @@
import hudson.slaves.ComputerListener;
import hudson.slaves.EphemeralNode;
import hudson.slaves.OfflineCause;
-import io.armadaproject.jenkins.plugin.KubernetesClientProvider;
import io.armadaproject.jenkins.plugin.ArmadaCloud;
-import io.armadaproject.jenkins.plugin.KubernetesComputer;
-import io.armadaproject.jenkins.plugin.KubernetesSlave;
+import io.armadaproject.jenkins.plugin.ArmadaComputer;
+import io.armadaproject.jenkins.plugin.ArmadaSlave;
import io.armadaproject.jenkins.plugin.PodUtils;
+import io.armadaproject.jenkins.plugin.job.ArmadaClientUtil;
+import io.armadaproject.jenkins.plugin.job.ArmadaEventWatcher;
+import io.armadaproject.jenkins.plugin.job.ArmadaJobMetadata;
+import io.armadaproject.jenkins.plugin.job.ArmadaState;
import io.fabric8.kubernetes.api.model.ContainerStateTerminated;
import io.fabric8.kubernetes.api.model.ContainerStateWaiting;
import io.fabric8.kubernetes.api.model.ContainerStatus;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodStatus;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.Watch;
import io.fabric8.kubernetes.client.Watcher;
-import io.fabric8.kubernetes.client.WatcherException;
+
+import java.io.Closeable;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.TimeUnit;
@@ -71,22 +66,15 @@
import org.jenkinsci.plugins.kubernetes.auth.KubernetesAuthException;
/**
- * Checks for deleted pods corresponding to {@link KubernetesSlave} and ensures the node is removed from Jenkins too.
+ * Checks for deleted pods corresponding to {@link ArmadaSlave} and ensures the node is removed from Jenkins too.
* If the pod has been deleted, all of the associated state (running user processes, workspace, etc.) must also be gone;
* so there is no point in retaining this agent definition any further.
- * ({@link KubernetesSlave} is not an {@link EphemeralNode}: it does support running across Jenkins restarts.)
- *
Note that pod retention policies other than the default {@link Never} may disable this system,
- * unless some external process or garbage collection policy results in pod deletion.
+ * ({@link ArmadaSlave} is not an {@link EphemeralNode}: it does support running across Jenkins restarts.)
*/
@Extension
public class Reaper extends ComputerListener {
-
private static final Logger LOGGER = Logger.getLogger(Reaper.class.getName());
- /**
- * Only useful for tests which shutdown Jenkins without terminating the JVM.
- * Close the watch so that we don't end up with spam in logs
- */
@Extension
public static class ReaperShutdownListener extends ItemListener {
@Override
@@ -105,22 +93,20 @@ public static Reaper getInstance() {
*/
private final AtomicBoolean activated = new AtomicBoolean();
- private final Map watchers = new ConcurrentHashMap<>();
+ private final Map watchers = new ConcurrentHashMap<>();
private final LoadingCache> terminationReasons =
Caffeine.newBuilder().expireAfterAccess(1, TimeUnit.DAYS).build(k -> new ConcurrentSkipListSet<>());
@Override
public void preLaunch(Computer c, TaskListener taskListener) throws IOException, InterruptedException {
- if (c instanceof KubernetesComputer) {
+ if (c instanceof ArmadaComputer) {
Timer.get().schedule(this::maybeActivate, 10, TimeUnit.SECONDS);
- // ensure associated cloud is being watched. the watch may have been closed due to exception or
- // failure to register on initial activation.
- KubernetesSlave node = ((KubernetesComputer) c).getNode();
+ var node = ((ArmadaComputer) c).getNode();
if (node != null && !isWatchingCloud(node.getCloudName())) {
try {
- watchCloud(node.getKubernetesCloud());
+ watchCloud(node.getArmadaCloud());
} catch (IllegalStateException ise) {
LOGGER.log(Level.WARNING, ise, () -> "kubernetes cloud not found: " + node.getCloudName());
}
@@ -146,16 +132,16 @@ private void activate() {
}
/**
- * Remove any {@link KubernetesSlave} nodes that reference Pods that don't exist.
+ * Remove any {@link ArmadaSlave} nodes that reference Pods that don't exist.
*/
private void reapAgents() {
Jenkins jenkins = Jenkins.getInstanceOrNull();
if (jenkins != null) {
for (Node n : new ArrayList<>(jenkins.getNodes())) {
- if (!(n instanceof KubernetesSlave)) {
+ if (!(n instanceof ArmadaSlave)) {
continue;
}
- KubernetesSlave ks = (KubernetesSlave) n;
+ ArmadaSlave ks = (ArmadaSlave) n;
if (ks.getLauncher().isLaunchSupported()) {
// Being launched, don't touch it.
continue;
@@ -169,8 +155,7 @@ private void reapAgents() {
// yet we do not want to do an unnamespaced pod list for RBAC reasons.
// Could use a hybrid approach: first list all pods in the configured namespace for all clouds;
// then go back and individually check any unmatched agents with their configured namespace.
- ArmadaCloud cloud = ks.getKubernetesCloud();
- if (cloud.connect().pods().inNamespace(ns).withName(name).get() == null) {
+ if (ks.connect().pods().inNamespace(ns).withName(name).get() == null) {
LOGGER.info(() -> ns + "/" + name
+ " seems to have been deleted, so removing corresponding Jenkins agent");
jenkins.removeNode(ks);
@@ -184,11 +169,6 @@ private void reapAgents() {
}
}
- /**
- * Create watchers for each configured {@link ArmadaCloud} in Jenkins and remove any existing watchers
- * for clouds that have been removed. If a {@link ArmadaCloud} client configuration property has been
- * updated a new watcher will be created to replace the existing one.
- */
private void watchClouds() {
Jenkins jenkins = Jenkins.getInstanceOrNull();
if (jenkins != null) {
@@ -206,62 +186,49 @@ private void watchClouds() {
}
}
- /**
- * Register {@link CloudPodWatcher} for the given cloud if one does not exist or if the existing watcher
- * is no longer valid.
- * @param kc kubernetes cloud to watch
- */
private void watchCloud(@NonNull ArmadaCloud kc) {
// can't use ConcurrentHashMap#computeIfAbsent because CloudPodWatcher will remove itself from the watchers
// map on close. If an error occurs when creating the watch it would create a deadlock situation.
- CloudPodWatcher watcher = new CloudPodWatcher(kc);
+ var watcher = new ArmadaJobWatcher(kc);
if (!isCloudPodWatcherActive(watcher)) {
try {
- KubernetesClient client = kc.connect();
- watcher.watch = client.pods().inNamespace(client.getNamespace()).watch(watcher);
- CloudPodWatcher old = watchers.put(kc.name, watcher);
+ var jobManager = ArmadaState.getJobManager(kc);
+ watcher.watch = jobManager.watchEvents(watcher);
+ var old = watchers.put(kc.name, watcher);
// if another watch slipped in then make sure it stopped
if (old != null) {
old.stop();
}
LOGGER.info(() -> "set up watcher on " + kc.getDisplayName());
- } catch (KubernetesAuthException | IOException | RuntimeException x) {
- LOGGER.log(Level.WARNING, x, () -> "failed to set up watcher on " + kc.getDisplayName());
+ } catch(Throwable t) {
+ LOGGER.log(Level.WARNING, "Failed to set up watcher on " + kc.getDisplayName(), t);
}
}
}
- /**
- * Check if the cloud is watched for Pod events.
- * @param name cloud name
- * @return true if a watcher has been registered for the given cloud
- */
boolean isWatchingCloud(String name) {
return watchers.get(name) != null;
}
- public Map getWatchers() {
- return watchers;
- }
-
/**
* Check if the given cloud pod watcher exists and is still valid. Watchers may become invalid
* of the kubernetes client configuration changes.
* @param watcher watcher to check
* @return true if the provided watcher already exists and is valid, false otherwise
*/
- private boolean isCloudPodWatcherActive(@NonNull CloudPodWatcher watcher) {
- CloudPodWatcher existing = watchers.get(watcher.cloudName);
- return existing != null && existing.clientValidity == watcher.clientValidity;
+ private boolean isCloudPodWatcherActive(@NonNull ArmadaJobWatcher watcher) {
+ var existing = watchers.get(watcher.cloudName);
+ return existing != null && existing.jobManagerValidity == watcher.jobManagerValidity;
}
- private static Optional resolveNode(@NonNull Jenkins jenkins, String namespace, String name) {
+ private static Optional resolveNode(@NonNull Jenkins jenkins, String jobId, String jobSetId) {
return new ArrayList<>(jenkins.getNodes())
.stream()
- .filter(KubernetesSlave.class::isInstance)
- .map(KubernetesSlave.class::cast)
+ .filter(ArmadaSlave.class::isInstance)
+ .map(ArmadaSlave.class::cast)
.filter(ks ->
- Objects.equals(ks.getNamespace(), namespace) && Objects.equals(ks.getPodName(), name))
+ Objects.equals(ks.getArmadaJobId(), jobId) &&
+ Objects.equals(ks.getArmadaJobSetId(), jobSetId))
.findFirst();
}
@@ -269,101 +236,64 @@ private static Optional resolveNode(@NonNull Jenkins jenkins, S
* Stop all watchers
*/
private void closeAllWatchers() {
- // on close each watcher should remove itself from the watchers map (see CloudPodWatcher#onClose)
- watchers.values().forEach(CloudPodWatcher::stop);
+ // on close each watcher should remove itself from the watchers map (see ArmadaJobWatcher#onClose)
+ watchers.values().forEach(ArmadaJobWatcher::stop);
}
- /**
- * Kubernetes pod event watcher for a Kubernetes Cloud. Notifies {@link Listener}
- * extensions on Pod events. The default Kubernetes client watch manager will
- * attempt to reconnect on connection errors. If the watch api returns "410 Gone"
- * then the Watch will close itself with a WatchException and this watcher will
- * deregister itself.
- */
- private class CloudPodWatcher implements Watcher {
+ private class ArmadaJobWatcher implements ArmadaEventWatcher {
private final String cloudName;
- private final int clientValidity;
+ private final int jobManagerValidity;
+ private Closeable watch;
- @CheckForNull
- private Watch watch;
-
- CloudPodWatcher(@NonNull ArmadaCloud cloud) {
+ ArmadaJobWatcher(@NonNull ArmadaCloud cloud) {
this.cloudName = cloud.name;
- this.clientValidity = KubernetesClientProvider.getValidity(cloud);
+ jobManagerValidity = ArmadaState.getJobManager(cloud).getValidity();
}
- @Override
- public void eventReceived(Action action, Pod pod) {
- // don't send bookmark event to listeners as they don't represent change in pod state
- if (action == Action.BOOKMARK) {
- // TODO future enhancement might be to keep track of bookmarks for better reconnect behavior. Would
- // likely have to track based on cloud address/namespace in case cloud was renamed or namespace
- // is changed.
- return;
+ void stop() {
+ if (watch != null) {
+ LOGGER.info("Stopping watch for armada cloud " + cloudName);
+ try {
+ this.watch.close();
+ } catch (IOException e) {
+ }
}
+ }
- // If there was a non-success http response code from watch request
- // or the api returned a Status object the watch manager notifies with
- // an error action and null resource.
- if (action == Action.ERROR && pod == null) {
+ @Override
+ public void onClose() {
+ LOGGER.fine(() -> cloudName + " watcher closed");
+ Reaper.this.watchers.remove(cloudName, this);
+ }
+
+ @Override
+ public void onEvent(EventOuterClass.EventMessage message) {
+ if(!ArmadaClientUtil.isInTerminalState(message.getEventsCase())) {
return;
}
- Jenkins jenkins = Jenkins.getInstanceOrNull();
+ var jenkins = Jenkins.getInstanceOrNull();
if (jenkins == null) {
return;
}
- String ns = pod.getMetadata().getNamespace();
- String name = pod.getMetadata().getName();
- Optional optionalNode = resolveNode(jenkins, ns, name);
- if (!optionalNode.isPresent()) {
+ var metadata = ArmadaClientUtil.extractMetadata(message);
+ var optionalNode = resolveNode(jenkins, metadata.getJobId(), metadata.getJobSetId());
+ if(!optionalNode.isPresent()) {
return;
}
+ var jobSetId = metadata.getJobSetId();
+ var jobId = metadata.getJobId();
Listeners.notify(Listener.class, true, listener -> {
try {
- Set terminationReasons = Reaper.this.terminationReasons.get(
- optionalNode.get().getNodeName());
- listener.onEvent(
- action,
- optionalNode.get(),
- pod,
- terminationReasons != null ? terminationReasons : Collections.emptySet());
+ var terminationReasons = Reaper.this.terminationReasons.get(optionalNode.get().getNodeName());
+ listener.onEvent(optionalNode.get(), metadata, message, terminationReasons);
} catch (Exception x) {
- LOGGER.log(Level.WARNING, "Listener " + listener + " failed for " + ns + "/" + name, x);
+ LOGGER.log(Level.WARNING, "Listener " + listener + " failed for " + jobSetId + "/" + jobId, x);
}
});
}
-
- /**
- * Close the associated {@link Watch} handle. This should be used shutdown/stop the watch. It will cause the
- * watch manager to call this classes {@link #onClose()} method.
- */
- void stop() {
- if (watch != null) {
- LOGGER.info("Stopping watch for kubernetes cloud " + cloudName);
- this.watch.close();
- }
- }
-
- @Override
- public void onClose() {
- LOGGER.fine(() -> cloudName + " watcher closed");
- // remove self from watchers list
- Reaper.this.watchers.remove(cloudName, this);
- }
-
- @Override
- public void onClose(WatcherException e) {
- // usually triggered because of "410 Gone" responses
- // https://kubernetes.io/docs/reference/using-api/api-concepts/#410-gone-responses
- // "Gone" may be returned if the resource version requested is older than the server
- // has retained.
- LOGGER.log(Level.WARNING, e, () -> cloudName + " watcher closed with exception");
- // remove self from watchers list
- Reaper.this.watchers.remove(cloudName, this);
- }
}
/**
@@ -371,10 +301,6 @@ public void onClose(WatcherException e) {
* @param node a {@link Node#getNodeName}
* @return a possibly empty set of {@link ContainerStateTerminated#getReason} or {@link PodStatus#getReason}
*/
- @SuppressFBWarnings(
- value = "NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE",
- justification =
- "Confused by @org.checkerframework.checker.nullness.qual.Nullable on LoadingCache.get? Never null here.")
@NonNull
public Set terminationReasons(@NonNull String node) {
synchronized (terminationReasons) {
@@ -387,256 +313,105 @@ public Set terminationReasons(@NonNull String node) {
*/
public interface Listener extends ExtensionPoint {
- /**
- * Handle Pod event.
- * @param action the kind of event that happened to the referred pod
- * @param node The affected node
- * @param pod The affected pod
- * @param terminationReasons Set of termination reasons
- */
void onEvent(
- @NonNull Watcher.Action action,
- @NonNull KubernetesSlave node,
- @NonNull Pod pod,
+ @NonNull ArmadaSlave node,
+ @NonNull ArmadaJobMetadata metadata,
+ @NonNull EventOuterClass.EventMessage message,
@NonNull Set terminationReasons)
throws IOException, InterruptedException;
}
@Extension
- public static class RemoveAgentOnPodDeleted implements Listener {
+ public static class RemoveAgentOnPodCancelled implements Listener {
@Override
public void onEvent(
- @NonNull Watcher.Action action,
- @NonNull KubernetesSlave node,
- @NonNull Pod pod,
+ @NonNull ArmadaSlave node,
+ @NonNull ArmadaJobMetadata metadata,
+ @NonNull EventOuterClass.EventMessage message,
@NonNull Set terminationReasons)
throws IOException {
- if (action != Watcher.Action.DELETED) {
+ if (!message.hasCancelled()) {
return;
}
- String ns = pod.getMetadata().getNamespace();
- String name = pod.getMetadata().getName();
- LOGGER.info(() -> ns + "/" + name + " was just deleted, so removing corresponding Jenkins agent");
- node.getRunListener().getLogger().printf("Pod %s/%s was just deleted%n", ns, name);
+ String jobSet = node.getArmadaJobSetId();
+ String job = node.getArmadaJobId();
+ LOGGER.info(() -> jobSet + "/" + job + " was just cancelled, so removing corresponding Jenkins agent");
+ node.getRunListener().getLogger().printf("Job %s/%s was just cancelled%n", jobSet, job);
Jenkins.get().removeNode(node);
disconnectComputer(node, new PodOfflineCause(Messages._PodOfflineCause_PodDeleted()));
}
}
@Extension
- public static class TerminateAgentOnContainerTerminated implements Listener {
-
- @Override
- public void onEvent(
- @NonNull Watcher.Action action,
- @NonNull KubernetesSlave node,
- @NonNull Pod pod,
- @NonNull Set terminationReasons)
- throws IOException, InterruptedException {
- if (action != Watcher.Action.MODIFIED) {
- return;
- }
-
- List terminatedContainers = PodUtils.getTerminatedContainers(pod);
- if (!terminatedContainers.isEmpty()) {
- List containers = new ArrayList<>();
- terminatedContainers.forEach(c -> {
- ContainerStateTerminated t = c.getState().getTerminated();
- String containerName = c.getName();
- containers.add(containerName);
- String reason = t.getReason();
- if (reason != null) {
- terminationReasons.add(reason);
- }
- });
- String reason = pod.getStatus().getReason();
- String message = pod.getStatus().getMessage();
- var sb = new StringBuilder()
- .append(pod.getMetadata().getNamespace())
- .append("/")
- .append(pod.getMetadata().getName());
- if (containers.size() > 1) {
- sb.append(" Containers ")
- .append(String.join(",", containers))
- .append(" were terminated.");
- } else {
- sb.append(" Container ")
- .append(String.join(",", containers))
- .append(" was terminated.");
- }
- logAndCleanUp(
- node,
- pod,
- terminationReasons,
- reason,
- message,
- sb,
- node.getRunListener(),
- new PodOfflineCause(Messages._PodOfflineCause_ContainerFailed("ContainerError", containers)));
- }
- }
- }
-
- @Extension
- public static class TerminateAgentOnPodFailed implements Listener {
+ public static class TerminateAgentOnJobFailed implements Listener {
@Override
public void onEvent(
- @NonNull Watcher.Action action,
- @NonNull KubernetesSlave node,
- @NonNull Pod pod,
+ @NonNull ArmadaSlave node,
+ @NonNull ArmadaJobMetadata metadata,
+ @NonNull EventOuterClass.EventMessage message,
@NonNull Set terminationReasons)
throws IOException, InterruptedException {
- if (action != Watcher.Action.MODIFIED) {
+ if (!ArmadaClientUtil.isInFailedState(message.getEventsCase())) {
return;
}
- if ("Failed".equals(pod.getStatus().getPhase())) {
- String reason = pod.getStatus().getReason();
- String message = pod.getStatus().getMessage();
- logAndCleanUp(
- node,
- pod,
- terminationReasons,
- reason,
- message,
- new StringBuilder()
- .append(pod.getMetadata().getNamespace())
- .append("/")
- .append(pod.getMetadata().getName())
- .append(" Pod just failed."),
- node.getRunListener(),
- new PodOfflineCause(Messages._PodOfflineCause_PodFailed(reason, message)));
- }
+ var reason = metadata.getReason();
+ var cause = metadata.getCause();
+ logAndCleanUp(
+ node,
+ terminationReasons,
+ reason,
+ cause,
+ new StringBuilder()
+ .append(metadata.getJobSetId())
+ .append("/")
+ .append(metadata.getJobId())
+ .append(" Job just failed."),
+ node.getRunListener(),
+ new PodOfflineCause(Messages._PodOfflineCause_PodFailed(reason, message)));
}
}
private static void logAndCleanUp(
- KubernetesSlave node,
- Pod pod,
+ ArmadaSlave node,
Set terminationReasons,
String reason,
- String message,
+ EventOuterClass.Cause failCause,
StringBuilder sb,
TaskListener runListener,
- PodOfflineCause cause)
- throws IOException, InterruptedException {
+ PodOfflineCause cause) {
List details = new ArrayList<>();
if (reason != null) {
details.add("Reason: " + reason);
terminationReasons.add(reason);
}
- if (message != null) {
- details.add("Message: " + message);
- }
if (!details.isEmpty()) {
sb.append(" ").append(String.join(", ", details)).append(".");
}
- var evictionCondition = pod.getStatus().getConditions().stream()
- .filter(c -> "EvictionByEvictionAPI".equals(c.getReason()))
- .findFirst();
- if (evictionCondition.isPresent()) {
- sb.append(" Pod was evicted by the Kubernetes Eviction API.");
- terminationReasons.add(evictionCondition.get().getReason());
+ if(failCause != null) {
+ sb.append(" failure cause: ").append(failCause);
+ terminationReasons.add(failCause.name());
}
LOGGER.info(() -> sb + " Removing corresponding node " + node.getNodeName() + " from Jenkins.");
runListener.getLogger().println(sb);
- logLastLinesThenTerminateNode(node, pod, runListener);
- PodUtils.cancelQueueItemFor(pod, "PodFailure");
+ PodUtils.cancelQueueItemFor(node, "PodFailure");
disconnectComputer(node, cause);
}
- private static void logLastLinesThenTerminateNode(KubernetesSlave node, Pod pod, TaskListener runListener)
- throws IOException, InterruptedException {
- try {
- String lines = PodUtils.logLastLines(pod, node.getKubernetesCloud().connect());
- if (lines != null) {
- runListener.getLogger().print(lines);
- }
- } catch (KubernetesAuthException e) {
- LOGGER.log(Level.FINE, e, () -> "Unable to get logs after pod failed event");
- } finally {
- node.terminate();
- }
- }
-
/**
* Disconnect computer associated with the given node. Should be called AFTER terminate so the offline cause
- * takes precedence over the one set by {@link KubernetesSlave#terminate()} (via {@link jenkins.model.Nodes#removeNode(Node)}).
+ * takes precedence over the one set by {@link ArmadaSlave#terminate()} (via {@link jenkins.model.Nodes#removeNode(Node)}).
* @see Computer#disconnect(OfflineCause)
* @param node node to disconnect
* @param cause reason for offline
*/
- private static void disconnectComputer(KubernetesSlave node, OfflineCause cause) {
+ private static void disconnectComputer(ArmadaSlave node, OfflineCause cause) {
Computer computer = node.getComputer();
if (computer != null) {
computer.disconnect(cause);
}
}
- @Extension
- public static class TerminateAgentOnImagePullBackOff implements Listener {
-
- @SuppressFBWarnings(
- value = "MS_SHOULD_BE_FINAL",
- justification = "Allow tests or groovy console to change the value")
- public static long BACKOFF_EVENTS_LIMIT =
- SystemProperties.getInteger(Reaper.class.getName() + ".backoffEventsLimit", 3);
-
- public static final String IMAGE_PULL_BACK_OFF = "ImagePullBackOff";
-
- // For each pod with at least 1 backoff, keep track of the first backoff event for 15 minutes.
- private Cache ttlCache =
- Caffeine.newBuilder().expireAfterWrite(15, TimeUnit.MINUTES).build();
-
- @Override
- public void onEvent(
- @NonNull Watcher.Action action,
- @NonNull KubernetesSlave node,
- @NonNull Pod pod,
- @NonNull Set terminationReasons)
- throws IOException, InterruptedException {
- if (action != Watcher.Action.MODIFIED) {
- return;
- }
-
- List backOffContainers = PodUtils.getContainers(pod, cs -> {
- ContainerStateWaiting waiting = cs.getState().getWaiting();
- return waiting != null
- && waiting.getMessage() != null
- && waiting.getMessage().contains("Back-off pulling image");
- });
-
- if (!backOffContainers.isEmpty()) {
- List images = new ArrayList<>();
- backOffContainers.forEach(cs -> images.add(cs.getImage()));
- var podUid = pod.getMetadata().getUid();
- var backOffNumber = ttlCache.get(podUid, k -> 0);
- ttlCache.put(podUid, ++backOffNumber);
- if (backOffNumber >= BACKOFF_EVENTS_LIMIT) {
- var imagesString = String.join(",", images);
- node.getRunListener()
- .error("Unable to pull container image \"" + imagesString
- + "\". Check if image tag name is spelled correctly.");
- terminationReasons.add(IMAGE_PULL_BACK_OFF);
- PodUtils.cancelQueueItemFor(pod, IMAGE_PULL_BACK_OFF);
- node.terminate();
- disconnectComputer(
- node,
- new PodOfflineCause(
- Messages._PodOfflineCause_ImagePullBackoff(IMAGE_PULL_BACK_OFF, images)));
- } else {
- node.getRunListener()
- .error("Image pull backoff detected, waiting for image to be available. Will wait for "
- + (BACKOFF_EVENTS_LIMIT - backOffNumber)
- + " more events before terminating the node.");
- }
- }
- }
- }
-
- /**
- * {@link SaveableListener} that will update cloud watchers when Jenkins configuration is updated.
- */
@Extension
public static class ReaperSaveableListener extends SaveableListener {
@Override
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/config.jelly b/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/config.jelly
index b01e52c9c..642f12651 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/config.jelly
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/config.jelly
@@ -21,7 +21,7 @@ THE SOFTWARE.
-
+
@@ -29,7 +29,7 @@ THE SOFTWARE.
-
+
@@ -49,7 +49,7 @@ THE SOFTWARE.
-
+
@@ -62,26 +62,14 @@ THE SOFTWARE.
-
-
-
-
-
-
-
-
-
-
-
-
@@ -94,8 +82,6 @@ THE SOFTWARE.
-
-
@@ -129,11 +115,6 @@ THE SOFTWARE.
deleteCaption="${%Delete Pod Label}" />
-
-
-
-
@@ -157,5 +138,4 @@ THE SOFTWARE.
-
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/help-garbageCollection.html b/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/help-garbageCollection.html
index 45f5ce9ea..e69de29bb 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/help-garbageCollection.html
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/help-garbageCollection.html
@@ -1,5 +0,0 @@
-
- Enables garbage collection of orphan pods for this Kubernetes cloud.
-
- When enabled, Jenkins will periodically check for orphan pods that have not been touched for the given timeout period and delete them.
-
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/help-podRetention.html b/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/help-podRetention.html
index 9fa0fd5ff..e69de29bb 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/help-podRetention.html
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/ArmadaCloud/help-podRetention.html
@@ -1,15 +0,0 @@
-
-
- This setting controls how agent pods are retained after the Jenkins build completes.
- The following retention policies are provided:
-
-
- - Never - always delete the agent pod.
- - On Failure - keep the agent pod if it fails during the build.
- - Always - always keep the agent pod.
-
-
- Note: Kubernetes administrators are responsible for managing any kept agent pod.
- These will not be deleted by the Jenkins Kubernetes plugin.
-
-
- X509 PEM encoded certificate. Can be additionally base64 encoded (as provided by Amazon EKS).
-
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/GarbageCollection/config.jelly b/src/main/resources/io/armadaproject/jenkins/plugin/GarbageCollection/config.jelly
index d958f89ff..e69de29bb 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/GarbageCollection/config.jelly
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/GarbageCollection/config.jelly
@@ -1,30 +0,0 @@
-
-
-
-
-
-
-
-
-
-
-
-
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/GarbageCollection/help-namespaces.html b/src/main/resources/io/armadaproject/jenkins/plugin/GarbageCollection/help-namespaces.html
index c9639c424..e69de29bb 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/GarbageCollection/help-namespaces.html
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/GarbageCollection/help-namespaces.html
@@ -1,2 +0,0 @@
-Namespaces to look at for garbage collection, in addition to the default namespace defined for the cloud.
-One namespace per line.
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/Messages.properties b/src/main/resources/io/armadaproject/jenkins/plugin/Messages.properties
index 25161e0f4..f58673d8b 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/Messages.properties
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/Messages.properties
@@ -1,9 +1,9 @@
-offline=Kubernetes agent is going offline
+offline=Armada agent is going offline
NonConfigurableKubernetesCloud.displayName=Kubernetes (predefined settings)
KubernetesSlave.AgentIsProvisionedFromTemplate=Agent {0} is provisioned from template {1}
RFC1123.error=Container Names MUST match RFC 1123 - They can only contain lowercase letters, numbers or dashes: {0}
label.error=Labels must follow required specs - https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#syntax-and-character-set: {0}
-KubernetesFolderProperty.displayName=Kubernetes
+ArmadaFolderProperty.displayName=Armada
KubernetesSlave.HomeWarning=[WARNING] HOME is set to / in the agent container. You may encounter \
troubles when using tools or ssh client. This usually happens if the uid doesn't have any \
entry in /etc/passwd. Please add a user to your Dockerfile or set the HOME environment \
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/config.jelly b/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/config.jelly
index 7547cbea5..3902b7134 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/config.jelly
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/config.jelly
@@ -29,10 +29,6 @@ THE SOFTWARE.
-
-
-
-
@@ -77,8 +73,6 @@ THE SOFTWARE.
-
-
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/help-namespace.html b/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/help-namespace.html
index 6215d744e..e69de29bb 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/help-namespace.html
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/help-namespace.html
@@ -1,3 +0,0 @@
-Namespace in which to schedule the pod.
-
-Leave empty to use the namespace defined at cloud level.
diff --git a/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/help-podRetention.html b/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/help-podRetention.html
index 1a6250e26..e69de29bb 100644
--- a/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/help-podRetention.html
+++ b/src/main/resources/io/armadaproject/jenkins/plugin/PodTemplate/help-podRetention.html
@@ -1,17 +0,0 @@
-
-
- This setting controls how agent pods are retained after the Jenkins build completes for this pod template.
- Values other than "Default" will override the plugin's Pod Retention setting.
- The following retention policies are provided:
-
-
- - Always - always keep the agent pod.
- - Default - use the Pod Retention setting for the plugin.
- - Never - always delete the agent pod.
- - On Failure - keep the agent pod if it fails during the build.
-
-
- Note: Kubernetes administrators are responsible for managing any kept agent pod.
- These will not be deleted by the Jenkins Kubernetes plugin.
-
-