Job progress...

Kai Reinhard

04.58.2019 6ff74e6e78e27fbcb751dcf20c9e61dafb78caed

Job progress...

 borgbutler-core/src/main/java/de/micromata/borgbutler/BorgCommands.java

@@ -53,7 +53,7 @@
        BorgCommand command = new BorgCommand()
                .setRepoConfig(repoConfig)
                .setCommand("info")
                .setParams("--json")
                .setParams("--json") // --progress has no effect.
                .setDescription("Loading info of repo '" + repoConfig.getDisplayName() + "'.");
        JobResult<String> jobResult = execute(command).getResult();
        if (jobResult.getStatus() != JobResult.Status.OK) {
@@ -86,7 +86,7 @@
        BorgCommand command = new BorgCommand()
                .setRepoConfig(repoConfig)
                .setCommand("list")
                .setParams("--json")
                .setParams("--json") // --progress has no effect.
                .setDescription("Loading list of archives of repo '" + repoConfig.getDisplayName() + "'.");
        JobResult<String> jobResult = execute(command).getResult();
        if (jobResult.getStatus() != JobResult.Status.OK) {
@@ -128,7 +128,7 @@
                .setRepoConfig(repoConfig)
                .setCommand("info")
                .setArchive(archive.getName())
                .setParams("--json")
                .setParams("--json", "--progress")
                .setDescription("Loading info of archive '" + archive.getName() + "' of repo '" + repoConfig.getDisplayName() + "'.");
        JobResult<String> jobResult = execute(command).getResult();
        if (jobResult.getStatus() != JobResult.Status.OK) {
@@ -172,12 +172,16 @@
                .setParams("--json-lines")
                .setDescription("Loading list of files of archive '" + archive.getName() + "' of repo '" + repoConfig.getDisplayName() + "'.");
        // The returned job might be an already queued or running one!
        final ProgressMessage progressMessage = new ProgressMessage()
                .setMessage("Getting file list...")
                .setCurrent(0);
        BorgJob<List<BorgFilesystemItem>> job = BorgQueueExecutor.getInstance().execute(new BorgJob<List<BorgFilesystemItem>>(command) {
            @Override
            protected void processStdOutLine(String line, int level) {
                BorgFilesystemItem item = JsonUtils.fromJson(BorgFilesystemItem.class, line);
                item.setMtime(DateUtils.format(item.getMtime()));
                payload.add(item);
                setProgressMessage(progressMessage.incrementCurrent());
            }
        });
        job.payload = new ArrayList<>();

 borgbutler-core/src/main/java/de/micromata/borgbutler/BorgJob.java

@@ -2,8 +2,13 @@

import de.micromata.borgbutler.config.BorgRepoConfig;
import de.micromata.borgbutler.config.ConfigurationHandler;
import de.micromata.borgbutler.data.Archive;
import de.micromata.borgbutler.jobs.AbstractCommandLineJob;
import de.micromata.borgbutler.json.JsonUtils;
import de.micromata.borgbutler.json.borg.ProgressMessage;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.exec.CommandLine;
import org.apache.commons.exec.environment.EnvironmentUtils;
import org.apache.commons.lang3.StringUtils;
@@ -17,19 +22,29 @@
 * A queue is important because Borg doesn't support parallel calls for one repository.
 * For each repository one single queue is allocated.
 */
public class BorgJob<T> extends AbstractCommandLineJob {
public class BorgJob<T> extends AbstractCommandLineJob implements Cloneable {
    private Logger log = LoggerFactory.getLogger(BorgJob.class);
    @Getter
    private BorgCommand command;
    /**
     * Some jobs may store here the result of the command (e. g. {@link BorgCommands#listArchiveContent(BorgRepoConfig, Archive)}).
     */
    @Getter
    protected T payload;

    @Getter
    @Setter(AccessLevel.PROTECTED)
    private ProgressMessage progressMessage;

    public BorgJob(BorgCommand command) {
        this.command = command;
        setWorkingDirectory(command.getWorkingDir());
        setDescription(command.getDescription());
    }

    private BorgJob() {
    }

    @Override
    protected CommandLine buildCommandLine() {
        CommandLine commandLine = new CommandLine(ConfigurationHandler.getConfiguration().getBorgCommand());
@@ -52,6 +67,24 @@
        return commandLine;
    }

    protected void processStdErrLine(String line, int level) {
        log.info(line);
        try {
            if (StringUtils.startsWith(line, "{\"message")) {
                ProgressMessage message = JsonUtils.fromJson(ProgressMessage.class, line);
                if (message != null) {
                    log.info(JsonUtils.toJson(progressMessage));
                    progressMessage = message;
                    return;
                }
            }
            errorOutputStream.write(line.getBytes());
            errorOutputStream.write("\n".getBytes());
        } catch (IOException ex) {
            log.error(ex.getMessage(), ex);
        }
    }

    @Override
    protected Map<String, String> getEnvironment() throws IOException {
        BorgRepoConfig repoConfig = command.getRepoConfig();
@@ -70,4 +103,17 @@
        }
        return env;
    }

    @Override
    public BorgJob<?> clone() {
        BorgJob<?> clone = new BorgJob<>();
        clone.setTitle(getTitle());
        clone.setExecuteStarted(isExecuteStarted());
        clone.setCommandLineAsString(getCommandLineAsString());
        clone.setCancelledRequested(isCancelledRequested());
        clone.setStatus(getStatus());
        clone.setWorkingDirectory(getWorkingDirectory());
        clone.setDescription(getDescription());
        return clone;
    }
}

 borgbutler-core/src/main/java/de/micromata/borgbutler/BorgQueueExecutor.java

@@ -1,11 +1,14 @@
package de.micromata.borgbutler;

import de.micromata.borgbutler.config.BorgRepoConfig;
import de.micromata.borgbutler.jobs.AbstractJob;
import de.micromata.borgbutler.jobs.JobQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
@@ -23,10 +26,34 @@
    // key is the repo name.
    private Map<String, JobQueue<String>> queueMap = new HashMap<>();

    private JobQueue<String> getQueue(BorgRepoConfig config) {
    /**
     * For displaying purposes.
     *
     * @param repo
     * @return A list of all jobs of the queue (as copies).
     */
    public List<BorgJob<?>> getJobListCopy(String repo) {
        JobQueue<String> origQueue = getQueue(repo);
        List<BorgJob<?>> jobList = new ArrayList<>();
        for (AbstractJob<String> origJob : origQueue.getQueue()) {
            if (!(origJob instanceof BorgJob)) {
                log.error("Oups, only BorgJobs are supported. Ignoring unexpected job: " + origJob.getClass());
                continue;
            }
            BorgJob<?> borgJob = ((BorgJob<?>) origJob).clone();
            jobList.add(borgJob);
        }
        return jobList;
    }

    private JobQueue<String> getQueue(String repo) {
        return queueMap.get(getQueueName(repo));
    }

    private JobQueue<String> ensureAndGetQueue(String repo) {
        synchronized (queueMap) {
            String queueName = config != null ? config.getRepo() : "--NO_REPO--";
            JobQueue<String> queue = queueMap.get(queueName);
            String queueName = getQueueName(repo);
            JobQueue<String> queue = getQueue(queueName);
            if (queue == null) {
                queue = new JobQueue<>();
                queueMap.put(queueName, queue);
@@ -35,13 +62,22 @@
        }
    }

    private JobQueue<String> ensureAndGetQueue(BorgRepoConfig config) {
        return ensureAndGetQueue(config != null ? config.getRepo() : null);
    }

    private String getQueueName(String repo) {
        return repo != null ? repo : "--NO_REPO--";
    }

    public BorgJob<Void> execute(BorgCommand command) {
        BorgJob<Void> job = new BorgJob<Void>(command);
        return execute(job);
    }

    @SuppressWarnings("unchecked")
    public <T> BorgJob<T> execute(BorgJob<T> job) {
        return (BorgJob<T>)getQueue(job.getCommand().getRepoConfig()).append(job);
        return (BorgJob<T>) ensureAndGetQueue(job.getCommand().getRepoConfig()).append(job);
    }

    private BorgQueueExecutor() {

 borgbutler-core/src/main/java/de/micromata/borgbutler/jobs/AbstractCommandLineJob.java

@@ -1,6 +1,7 @@
package de.micromata.borgbutler.jobs;

import de.micromata.borgbutler.config.Definitions;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.exec.*;
@@ -22,14 +23,18 @@
    private Logger log = LoggerFactory.getLogger(AbstractCommandLineJob.class);
    private ExecuteWatchdog watchdog;
    @Getter
    @Setter(AccessLevel.PROTECTED)
    private boolean executeStarted;
    private CommandLine commandLine;
    /**
     * The command line as string. This property is also used as ID for detecting multiple borg calls.
     */
    @Setter(AccessLevel.PROTECTED)
    private String commandLineAsString;
    @Getter
    @Setter
    private File workingDirectory;
    @Getter
    @Setter
    private String description;
    protected ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

 borgbutler-core/src/main/java/de/micromata/borgbutler/jobs/AbstractJob.java

@@ -17,19 +17,14 @@
    @Getter
    @Setter
    private boolean cancelledRequested;

    @Getter
    @Setter(AccessLevel.PACKAGE)
    @Setter(AccessLevel.PROTECTED)
    private Status status;
    @Getter
    @Setter
    private String title;
    @Getter
    @Setter
    private String statusText;
    @Getter(AccessLevel.PACKAGE)
    @Setter(AccessLevel.PACKAGE)

    private Future<JobResult<T>> future;

    public void cancel() {
@@ -52,6 +47,7 @@

    /**
     * Waits for and gets the result.
     *
     * @return
     */
    public JobResult<T> getResult() {

 borgbutler-core/src/main/java/de/micromata/borgbutler/jobs/JobQueue.java

@@ -1,5 +1,6 @@
package de.micromata.borgbutler.jobs;

import lombok.Getter;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -12,6 +13,7 @@
public class JobQueue<T> {
    private static final int MAX_DONE_JOBS_SIZE = 50;
    private Logger log = LoggerFactory.getLogger(JobQueue.class);
    @Getter
    private List<AbstractJob<T>> queue = new ArrayList<>();
    private List<AbstractJob<T>> doneJobs = new LinkedList<>();
    private ExecutorService executorService = Executors.newSingleThreadExecutor();

 borgbutler-core/src/main/java/de/micromata/borgbutler/json/borg/ProgressMessage.java

New file
@@ -0,0 +1,55 @@
package de.micromata.borgbutler.json.borg;

import lombok.Getter;
import lombok.Setter;

/**
 * Output of borg option <tt>--progress</tt>.
 * See https://borgbackup.readthedocs.io/en/stable/internals/frontends.html,
 */
public class ProgressMessage {
    // {"message": "Calculating statistics...   0%", "current": 1, "total": 2497, "info": null, "operation": 1, "msgid": null, "type": "progress_percent", "finished": false, "time": 1546640510.116256}
    /**
     * e. g. Calculating statistics...   5%
     */
    @Getter
    @Setter
    private String message;
    /**
     * Current counter of total.
     */
    @Getter
    @Setter
    private long current;
    @Getter
    private long total;
    /**
     * Array that describes the current item, may be null, contents depend on msgid.
     */
    @Getter
    private String[] info;
    /**
     * unique, opaque integer ID of the operation.
     */
    @Getter
    private int operation;
    @Getter
    private int msgid;
    /**
     * e. g. progress_percent
     */
    @Getter
    private String type;
    @Getter
    private boolean finished;
    /**
     * Unix timestamp (float).
     */
    @Getter
    private double time;

    public ProgressMessage incrementCurrent() {
        ++current;
        return this;
    }
}

 borgbutler-server/src/main/java/de/micromata/borgbutler/server/rest/JobsRest.java

New file
@@ -0,0 +1,48 @@
package de.micromata.borgbutler.server.rest;

import de.micromata.borgbutler.BorgJob;
import de.micromata.borgbutler.BorgQueueExecutor;
import de.micromata.borgbutler.data.Archive;
import de.micromata.borgbutler.data.Repository;
import de.micromata.borgbutler.json.JsonUtils;
import de.micromata.borgbutler.server.rest.queue.JsonJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;
import java.util.ArrayList;
import java.util.List;

@Path("/jobs")
public class JobsRest {
    private static Logger log = LoggerFactory.getLogger(JobsRest.class);

    @GET
    @Produces(MediaType.APPLICATION_JSON)
    /**
     *
     * @param repo Name of repository ({@link Repository#getName()}.
     * @param prettyPrinter If true then the json output will be in pretty format.
     * @return Job queue as json string.
     * @see JsonUtils#toJson(Object, boolean)
     */
    public String getJobs(@QueryParam("repo") String repoName,
                          @QueryParam("prettyPrinter") boolean prettyPrinter) {
        BorgQueueExecutor borgQueueExecutor = BorgQueueExecutor.getInstance();
        List<BorgJob<?>> borgJobList = borgQueueExecutor.getJobListCopy(repoName);
        List<JsonJob> jobList = new ArrayList<>(borgJobList.size());
        for (BorgJob<?> borgJob : borgJobList) {
            JsonJob job = new JsonJob();
            job.setTitle(borgJob.getTitle());
            // job.setProgressText(borgJob.get);
            job.setStatus(borgJob.getStatus());
            job.setCancelledRequested(borgJob.isCancelledRequested());
        }
        Archive archive = null;
        return JsonUtils.toJson(archive, prettyPrinter);
    }
}

 borgbutler-server/src/main/java/de/micromata/borgbutler/server/rest/queue/JsonJob.java

New file
@@ -0,0 +1,20 @@
package de.micromata.borgbutler.server.rest.queue;

import de.micromata.borgbutler.jobs.AbstractJob;
import lombok.Getter;
import lombok.Setter;

public class JsonJob {
    @Getter
    @Setter
    private boolean cancelledRequested;
    @Getter
    @Setter
    private AbstractJob.Status status;
    @Getter
    @Setter
    private String title;
    @Getter
    @Setter
    private String progressText;
}

 borgbutler-server/src/main/java/de/micromata/borgbutler/server/rest/queue/JsonJobQueue.java

New file
@@ -0,0 +1,15 @@
package de.micromata.borgbutler.server.rest.queue;

import lombok.Getter;
import lombok.Setter;

import java.util.List;

public class JsonJobQueue {
    @Getter
    @Setter
    private String title;
    @Getter
    @Setter
    private List<JsonJob> jobs;
}