Remove MongoJack and consolidate MongoDB utils

build.gradle

@@ -147,10 +147,7 @@ dependencies {
  }
 
  // Database driver.
- implementation 'org.mongodb:mongo-java-driver:3.11.0'
-
- // Legacy system for storing Java objects, this functionality is now provided by the MongoDB driver itself.
- implementation 'org.mongojack:mongojack:2.10.1'
+ implementation 'org.mongodb:mongodb-driver-sync:4.7.2'
 
  // JSON serialization and deserialization from and to Java objects
  implementation 'com.fasterxml.jackson.core:jackson-core:2.10.3'

src/main/java/com/conveyal/analysis/BackendMain.java

@@ -2,7 +2,6 @@
 
 import com.conveyal.analysis.components.BackendComponents;
 import com.conveyal.analysis.components.LocalBackendComponents;
-import com.conveyal.analysis.persistence.Persistence;
 import com.conveyal.r5.SoftwareVersion;
 import com.conveyal.r5.analyst.PointSetCache;
 import com.conveyal.r5.analyst.WorkerCategory;
@@ -44,7 +43,6 @@ private static void startServerInternal (BackendComponents components, TaskActio
  // TODO migrate to non-static Components.
  // TODO remove the static ApiMain abstraction layer. We do not use it anywhere but in handling GraphQL queries.
  // TODO we could move this to something like BackendComponents.initialize()
- Persistence.initializeStatically(components.config);
  PointSetCache.initializeStatically(components.fileStorage);
 
  // TODO handle this via components without explicit "if (offline)"

src/main/java/com/conveyal/analysis/components/BackendComponents.java

@@ -83,20 +83,20 @@ public List<HttpController> standardHttpControllers () {
  return Lists.newArrayList(
  // These handlers are at paths beginning with /api
  // and therefore subject to authentication and authorization.
- new GtfsController(gtfsCache),
- new BundleController(this),
+ new GtfsController(database, gtfsCache),
+ new BundleController(database, fileStorage, taskScheduler),
  new OpportunityDatasetController(fileStorage, taskScheduler, censusExtractor, database),
- new RegionalAnalysisController(broker, fileStorage),
+ new RegionalAnalysisController(broker, database, fileStorage),
  new AggregationAreaController(fileStorage, database, taskScheduler),
  // This broker controller registers at least one handler at URL paths beginning with /internal, which
  // is exempted from authentication and authorization, but should be hidden from the world
  // outside the cluster by the reverse proxy. Perhaps we should serve /internal on a separate
  // port so they can't be accidentally exposed by the reverse proxy. It could even be a separate
  // InternalHttpApi component with its own spark service, renaming this ExternalHttpApi.
- new BrokerController(broker, eventBus),
+ new BrokerController(broker, database, eventBus),
  new UserActivityController(taskScheduler),
  new DataSourceController(fileStorage, database, taskScheduler, censusExtractor),
- new WorkerProxyController(broker)
+ new WorkerProxyController(broker, database)
  );
  }
 

src/main/java/com/conveyal/analysis/components/LocalBackendComponents.java

@@ -34,14 +34,13 @@ public LocalBackendComponents () {
  authentication = new LocalAuthentication();
  // TODO add nested LocalWorkerComponents here, to reuse some components, and pass it into the LocalWorkerLauncher?
  workerLauncher = new LocalWorkerLauncher(config, fileStorage, gtfsCache, osmCache);
- broker = new Broker(config, fileStorage, eventBus, workerLauncher);
+ broker = new Broker(config, eventBus, workerLauncher);
  censusExtractor = new SeamlessCensusGridExtractor(config);
  // Instantiate the HttpControllers last, when all the components except the HttpApi are already created.
  List<HttpController> httpControllers = standardHttpControllers();
  httpControllers.add(new LocalFilesController(fileStorage));
  httpApi = new HttpApi(fileStorage, authentication, eventBus, config, httpControllers);
  // compute = new LocalCompute();
- // persistence = persistence(local_Mongo)
  eventBus.addHandlers(new ErrorLogger());
  }
 

src/main/java/com/conveyal/analysis/components/broker/Broker.java

@@ -6,17 +6,11 @@
 import com.conveyal.analysis.components.eventbus.EventBus;
 import com.conveyal.analysis.components.eventbus.RegionalAnalysisEvent;
 import com.conveyal.analysis.components.eventbus.WorkerEvent;
-import com.conveyal.analysis.models.RegionalAnalysis;
 import com.conveyal.analysis.results.MultiOriginAssembler;
-import com.conveyal.analysis.util.JsonUtil;
-import com.conveyal.file.FileStorage;
-import com.conveyal.file.FileStorageKey;
-import com.conveyal.file.FileUtils;
 import com.conveyal.r5.analyst.WorkerCategory;
 import com.conveyal.r5.analyst.cluster.RegionalTask;
 import com.conveyal.r5.analyst.cluster.RegionalWorkResult;
 import com.conveyal.r5.analyst.cluster.WorkerStatus;
-import com.conveyal.r5.analyst.scenario.Scenario;
 import com.conveyal.r5.util.ExceptionUtils;
 import com.google.common.collect.ListMultimap;
 import com.google.common.collect.MultimapBuilder;
@@ -27,8 +21,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -42,7 +34,6 @@
 import static com.conveyal.analysis.components.eventbus.WorkerEvent.Action.REQUESTED;
 import static com.conveyal.analysis.components.eventbus.WorkerEvent.Role.REGIONAL;
 import static com.conveyal.analysis.components.eventbus.WorkerEvent.Role.SINGLE_POINT;
-import static com.conveyal.file.FileCategory.BUNDLES;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
@@ -93,7 +84,6 @@ public interface Config {
  private Config config;
 
  // Component Dependencies
- private final FileStorage fileStorage;
  private final EventBus eventBus;
  private final WorkerLauncher workerLauncher;
 
@@ -143,9 +133,8 @@ public interface Config {
  public TObjectLongMap<WorkerCategory> recentlyRequestedWorkers =
  TCollections.synchronizedMap(new TObjectLongHashMap<>());
 
- public Broker (Config config, FileStorage fileStorage, EventBus eventBus, WorkerLauncher workerLauncher) {
+ public Broker(Config config, EventBus eventBus, WorkerLauncher workerLauncher) {
  this.config = config;
- this.fileStorage = fileStorage;
  this.eventBus = eventBus;
  this.workerLauncher = workerLauncher;
  }
@@ -154,89 +143,35 @@ public Broker (Config config, FileStorage fileStorage, EventBus eventBus, Worker
  * Enqueue a set of tasks for a regional analysis.
  * Only a single task is passed in, which the broker will expand into all the individual tasks for a regional job.
  */
- public synchronized void enqueueTasksForRegionalJob (RegionalAnalysis regionalAnalysis) {
-
- // Make a copy of the regional task inside the RegionalAnalysis, replacing the scenario with a scenario ID.
- RegionalTask templateTask = templateTaskFromRegionalAnalysis(regionalAnalysis);
-
- LOG.info("Enqueuing tasks for job {} using template task.", templateTask.jobId);
- if (findJob(templateTask.jobId) != null) {
- LOG.error("Someone tried to enqueue job {} but it already exists.", templateTask.jobId);
- throw new RuntimeException("Enqueued duplicate job " + templateTask.jobId);
+ public synchronized void enqueueTasksForRegionalJob(Job job, MultiOriginAssembler assembler) {
+ // Once the assembler has been created, enqueue the job.
+ LOG.info("Enqueuing tasks for job {} using template task.", job.jobId);
+ if (findJob(job.jobId) != null) {
+ LOG.error("Someone tried to enqueue job {} but it already exists.", job.jobId);
+ throw new RuntimeException("Enqueued duplicate job " + job.jobId);
  }
- WorkerTags workerTags = WorkerTags.fromRegionalAnalysis(regionalAnalysis);
- Job job = new Job(templateTask, workerTags);
  jobs.put(job.workerCategory, job);
 
- // Register the regional job so results received from multiple workers can be assembled into one file.
- // TODO encapsulate MultiOriginAssemblers in a new Component
- // Note: if this fails with an exception we'll have a job enqueued, possibly being processed, with no assembler.
- // That is not catastrophic, but the user may need to recognize and delete the stalled regional job.
- MultiOriginAssembler assembler = new MultiOriginAssembler(regionalAnalysis, job, fileStorage);
- resultAssemblers.put(templateTask.jobId, assembler);
+ resultAssemblers.put(job.jobId, assembler);
 
  if (config.testTaskRedelivery()) {
  // This is a fake job for testing, don't confuse the worker startup code below with null graph ID.
  return;
  }
 
  if (workerCatalog.noWorkersAvailable(job.workerCategory, config.offline())) {
- createOnDemandWorkerInCategory(job.workerCategory, workerTags);
+ createOnDemandWorkerInCategory(job.workerCategory, job.workerTags);
  } else {
  // Workers exist in this category, clear out any record that we're waiting for one to start up.
  recentlyRequestedWorkers.remove(job.workerCategory);
  }
- eventBus.send(new RegionalAnalysisEvent(templateTask.jobId, STARTED).forUser(workerTags.user, workerTags.group));
- }
-
- /**
- * The single RegionalTask object represents a lot of individual accessibility tasks at many different origin
- * points, typically on a grid. Before passing that RegionalTask on to the Broker (which distributes tasks to
- * workers and tracks progress), we remove the details of the scenario, substituting the scenario's unique ID
- * to save time and bandwidth. This avoids repeatedly sending the scenario details to the worker in every task,
- * as they are often quite voluminous. The workers will fetch the scenario once from S3 and cache it based on
- * its ID only. We protectively clone this task because we're going to null out its scenario field, and don't
- * want to affect the original object which contains all the scenario details.
- * TODO Why is all this detail added after the Persistence call?
- * We don't want to store all the details added below in Mongo?
- */
- private RegionalTask templateTaskFromRegionalAnalysis (RegionalAnalysis regionalAnalysis) {
- RegionalTask templateTask = regionalAnalysis.request.clone();
- // First replace the inline scenario with a scenario ID, storing the scenario for retrieval by workers.
- Scenario scenario = templateTask.scenario;
- templateTask.scenarioId = scenario.id;
- // Null out the scenario in the template task, avoiding repeated serialization to the workers as massive JSON.
- templateTask.scenario = null;
- String fileName = String.format("%s_%s.json", regionalAnalysis.bundleId, scenario.id);
- FileStorageKey fileStorageKey = new FileStorageKey(BUNDLES, fileName);
- try {
- File localScenario = FileUtils.createScratchFile("json");
- JsonUtil.objectMapper.writeValue(localScenario, scenario);
- // FIXME this is using a network service in a method called from a synchronized broker method.
- // Move file into storage before entering the synchronized block.
- fileStorage.moveIntoStorage(fileStorageKey, localScenario);
- } catch (IOException e) {
- LOG.error("Error storing scenario for retrieval by workers.", e);
- }
- // Fill in all the fields in the template task that will remain the same across all tasks in a job.
- // I am not sure why we are re-setting all these fields, it seems like they are already set when the task is
- // initialized by AnalysisRequest.populateTask. But we'd want to thoroughly check that assumption before
- // eliminating or moving these lines.
- templateTask.jobId = regionalAnalysis._id;
- templateTask.graphId = regionalAnalysis.bundleId;
- templateTask.workerVersion = regionalAnalysis.workerVersion;
- templateTask.height = regionalAnalysis.height;
- templateTask.width = regionalAnalysis.width;
- templateTask.north = regionalAnalysis.north;
- templateTask.west = regionalAnalysis.west;
- templateTask.zoom = regionalAnalysis.zoom;
- return templateTask;
+ eventBus.send(new RegionalAnalysisEvent(job.jobId, STARTED).forUser(job.workerTags.user, job.workerTags.group));
  }
 
  /**
  * Create on-demand worker for a given job.
  */
- public void createOnDemandWorkerInCategory(WorkerCategory category, WorkerTags workerTags){
+ public void createOnDemandWorkerInCategory(WorkerCategory category, WorkerTags workerTags) {
  createWorkersInCategory(category, workerTags, 1, 0);
  }
 

src/main/java/com/conveyal/analysis/components/broker/Job.java

@@ -128,23 +128,42 @@ private RegionalTask makeOneTask (int taskNumber) {
  */
  public final Set<String> errors = new HashSet();
 
- public Job (RegionalTask templateTask, WorkerTags workerTags) {
- this.jobId = templateTask.jobId;
- this.templateTask = templateTask;
- this.workerCategory = new WorkerCategory(templateTask.graphId, templateTask.workerVersion);
- this.nTasksCompleted = 0;
- this.nextTaskToDeliver = 0;
-
- if (templateTask.originPointSetKey != null) {
- checkNotNull(templateTask.originPointSet);
- this.nTasksTotal = templateTask.originPointSet.featureCount();
+ public Job(RegionalTask task, WorkerTags workerTags) {
+ templateTask = templateTaskFromRegionalTask(task);
+ jobId = templateTask.jobId;
+ workerCategory = new WorkerCategory(templateTask.graphId, templateTask.workerVersion);
+ nTasksCompleted = 0;
+ nextTaskToDeliver = 0;
+ nTasksTotal = getTasksTotal(templateTask);
+ completedTasks = new BitSet(nTasksTotal);
+ this.workerTags = workerTags;
+ }
+
+ public static int getTasksTotal(RegionalTask task) {
+ if (task.originPointSetKey != null) {
+ checkNotNull(task.originPointSet);
+ return task.originPointSet.featureCount();
  } else {
- this.nTasksTotal = templateTask.width * templateTask.height;
+ return task.width * task.height;
  }
+ }
 
- this.completedTasks = new BitSet(nTasksTotal);
- this.workerTags = workerTags;
-
+ /**
+ * The single RegionalTask object represents a lot of individual accessibility tasks at many different origin
+ * points, typically on a grid. Before passing that RegionalTask on to the Broker (which distributes tasks to
+ * workers and tracks progress), we remove the details of the scenario, substituting the scenario's unique ID
+ * to save time and bandwidth. This avoids repeatedly sending the scenario details to the worker in every task,
+ * as they are often quite voluminous. The workers will fetch the scenario once from S3 and cache it based on
+ * its ID only. We protectively clone this task because we're going to null out its scenario field, and don't
+ * want to affect the original object which contains all the scenario details.
+ */
+ private static RegionalTask templateTaskFromRegionalTask(RegionalTask task) {
+ RegionalTask templateTask = task.clone();
+ // First replace the inline scenario with a scenario ID, storing the scenario for retrieval by workers.
+ templateTask.scenarioId = templateTask.scenario.id;
+ // Null out the scenario in the template task, avoiding repeated serialization to the workers as massive JSON.
+ templateTask.scenario = null;
+ return templateTask;
  }
 
  public boolean markTaskCompleted(int taskId) {

src/main/java/com/conveyal/analysis/components/broker/RedeliveryTest.java

@@ -1,13 +1,16 @@
 package com.conveyal.analysis.components.broker;
 
+import com.conveyal.analysis.UserPermissions;
 import com.conveyal.analysis.components.BackendComponents;
 import com.conveyal.analysis.components.LocalBackendComponents;
 import com.conveyal.analysis.models.RegionalAnalysis;
+import com.conveyal.analysis.results.MultiOriginAssembler;
 import com.conveyal.r5.analyst.cluster.RegionalTask;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.UUID;
 
 /**
@@ -64,9 +67,11 @@ private static void sendFakeJob(Broker broker) {
  templateTask.height = 1;
  templateTask.width = N_TASKS_PER_JOB;
  templateTask.scenarioId = "FAKE";
- RegionalAnalysis regionalAnalysis = new RegionalAnalysis();
+ RegionalAnalysis regionalAnalysis = new RegionalAnalysis(new UserPermissions("[email protected]", false, "testing"), "test");
  regionalAnalysis.request = templateTask;
- broker.enqueueTasksForRegionalJob(regionalAnalysis);
+ var job = new Job(templateTask, WorkerTags.fromRegionalAnalysis(regionalAnalysis));
+ var assembler = new MultiOriginAssembler(job, new ArrayList<>());
+ broker.enqueueTasksForRegionalJob(job, assembler);
  }
 
  public static String compactUUID() {