Implement Batch Jobs using Google Drive

package.json

@@ -39,6 +39,7 @@
 	"dependencies": {
 		"@google-cloud/storage": "^7.7.0",
 		"@google/earthengine": "^0.1.385",
+		"@googleapis/drive": "^8.11.0",
 		"@openeo/js-commons": "^1.4.1",
 		"@openeo/js-processgraphs": "^1.3.0",
 		"@seald-io/nedb": "^4.0.4",
@@ -47,6 +48,7 @@
 		"check-disk-space": "^3.4.0",
 		"epsg-index": "^2.0.0",
 		"fs-extra": "^11.2.0",
+		"googleapis-common": "^7.2.0",
 		"luxon": "^3.4.4",
 		"proj4": "^2.10.0",
 		"restify": "^11.1.0"

src/api/collections.js

@@ -48,7 +48,7 @@
 
 		const b = Date.now();
 		const pContext = this.context.processingContext();
-		this.ee = await pContext.connectGee(true);
+		this.ee = await pContext.connect(true);
 		console.log(`Established connection to GEE for STAC (${Date.now()-b} ms)`);
 
 		return num;
@@ -151,7 +151,7 @@
 		const limit = parseInt(req.query.limit, 10) || 10;
 		const offset = parseInt(req.query.offset, 10) || 0;
 
 		// todo: migrate to ee.data.listImages?
 		// Load the collection
 		let ic = this.ee.ImageCollection(id);
 

src/api/files.js

@@ -94,8 +94,7 @@ export default class FilesAPI {
 						size: newFileStat.size,
 						modified: Utils.getISODateTime(newFileStat.mtime)
 					});
-				}
-				catch (e) {
+				} catch (e) {
 					if (this.context.debug) {
 						console.error(e);
 					}

src/api/jobs.js

@@ -54,6 +54,9 @@
 		if (!req.user._id) {
 			throw new Errors.AuthenticationRequired();
 		}
+
+		// Update the task status
+		this.context.processingContext(req.user).startTaskMonitor();
 	}
 
 	async getJobs(req, res) {
@@ -170,7 +173,7 @@
 			await this.storage.updateJobStatus(query, newStatus);
 		}
 
 		// todo: actually stop the processing
 
 		res.send(204);
 	}
@@ -200,6 +203,9 @@
 		}
 
 		const makeStorageUrl = obj => {
+			if (Utils.isUrl(obj.href)) {
+				return obj;
+			}
 			obj.href = API.getUrl("/storage/" + job.token + "/" + obj.href);
 			return obj;
 		};
@@ -246,13 +252,13 @@
 			if (this.storage.isFieldEditable(key)) {
 				switch(key) {
 					case 'process': {
-						const pg = new ProcessGraph(req.body.process, this.context.processingContext(req.user));
+						const pg = new ProcessGraph(req.body.process, this.context.processingContext(req.user, job));
 						pg.allowUndefinedParameters(false);
 						promises.push(pg.validate());
 						break;
 					}
 					default:
 						// ToDo: Validate further data #73
 						// For example, if budget < costs, reject request
 				}
 				data[key] = req.body[key];
@@ -290,7 +296,7 @@
 		pg.allowUndefinedParameters(false);
 		await pg.validate();
 
 		// ToDo: Validate further data #73
 		const data = {
 			title: req.body.title || null,
 			description: req.body.description || null,
@@ -324,7 +330,7 @@
 
 		// const plan = req.body.plan || this.context.plans.default;
 		// const budget = req.body.budget || null;
 		// ToDo: Validate data, handle budget and plan input #73
 		const id = Utils.timeId();
     const log_level = Logs.checkLevel(req.body.log_level, this.context.defaultLogLevel);
 

src/api/services.js

@@ -118,8 +118,8 @@
 			if (this.storage.isFieldEditable(key)) {
 				switch(key) {
 					case 'process': {
-						const pg = new ProcessGraph(req.body.process, this.context.processingContext(req.user));
+						const pg = new ProcessGraph(req.body.process, this.context.processingContext(req.user, service));
 						// ToDo 1.0: Correctly handle service paramaters #79
 						pg.allowUndefinedParameters(false);
 						promises.push(pg.validate());
 						break;
@@ -132,7 +132,7 @@
 						});
 						break;
 					default:
 						// ToDo: Validate further data  #73
 						// For example, if budget < costs, reject request
 				}
 				data[key] = req.body[key];
@@ -189,11 +189,11 @@
 		}
 
 		const pg = new ProcessGraph(req.body.process, this.context.processingContext(req.user));
 		// ToDo 1.0: Correctly handle service paramaters #79
 		pg.allowUndefinedParameters(false);
 		await pg.validate();
 
 		// ToDo: Validate further data  #73
 		const data = {
 			title: req.body.title || null,
 			description: req.body.description || null,

src/api/worker/batchjob.js

@@ -3,6 +3,7 @@ import path from 'path';
 import ProcessGraph from '../../processgraph/processgraph.js';
 import GeeResults from '../../processes/utils/results.js';
 import Utils from '../../utils/utils.js';
+import GDrive from '../../utils/gdrive.js';
 const packageInfo = Utils.require('../../package.json');
 
 export default async function run(config, storage, user, query) {
@@ -19,26 +20,59 @@ export default async function run(config, storage, user, query) {
     await Promise.all(cleanupTasks);
 
     logger.info("Starting batch job");
+
     await storage.updateJobStatus(query, 'running');
 
-    const context = config.processingContext(user);
+    const jobfolder = storage.getJobFolder(job._id);
+    await fse.ensureDir(path.dirname(jobfolder));
+
+    const context = config.processingContext(user, job);
     const pg = new ProcessGraph(job.process, context, logger);
     await pg.execute();
 
-    const computeTasks = pg.getResults().map(async (datacube) => {
-      const response = await GeeResults.retrieve(context, datacube, logger);
-      const params = datacube.getOutputFormatParameters();
-      const filename = (params.name || String(Utils.generateHash())) + GeeResults.getFileExtension(datacube, config);
-      const filepath = storage.getJobFile(job._id, filename);
-      logger.debug("Storing result to: " + filepath);
-      await fse.ensureDir(path.dirname(filepath));
-      await new Promise((resolve, reject) => {
-        const writer = fse.createWriteStream(filepath);
-        response.data.pipe(writer);
-        writer.on('error', reject);
-        writer.on('close', resolve);
-      });
-      return { filepath, datacube };
+    const computeTasks = pg.getResults().map(async (dc) => {
+      const format = config.getOutputFormat(dc.getOutputFormat());
+      const datacube = format.preprocess(GeeResults.BATCH, context, dc, logger);
+
+      if (format.canExport()) {
+        // Ensure early that we have access to the Google Drive API
+        const drive = new GDrive(context.server(), user);
+        await drive.connect();
+        // Start processing
+        const tasks = await format.export(context.ee, dc, context.getResource());
+        storage.addTasks(job, tasks);
+        context.startTaskMonitor();
+        const driveUrls = await new Promise((resolve, reject) => {
+          setInterval(async () => {
+            const updatedJob = await storage.getById(job._id, job.user_id);
+            if (!updatedJob) {
+              reject(new Error("Job was deleted"));
+            }
+            if (['canceled', 'error', 'finished'].includes(updatedJob.status)) {
+              resolve(job.googleDriveResults);
+            }
+          }, 10000);
+        });
+        // Handle Google Drive specifics (permissions and public URLs)
+        const folderName = GDrive.getFolderName(job);
+        await drive.publishFoldersByName(folderName);
+        const files = await drive.getAssetsForFolder(folderName);
+
+        return { files, datacube, links: driveUrls };
+      }
+      else {
+        const response = await format.retrieve(context.ee, dc);
+        const params = datacube.getOutputFormatParameters();
+        const filename = (params.name || String(Utils.generateHash())) + GeeResults.getFileExtension(datacube, config);
+        const filepath = storage.getJobFile(job._id, filename);
+        await new Promise((resolve, reject) => {
+          const writer = fse.createWriteStream(filepath);
+          response.data.pipe(writer);
+          writer.on('error', reject);
+          writer.on('close', resolve);
+        });
+        return { files: [filepath], datacube };
+      }
     });
 
     await Promise.all(computeTasks);
@@ -53,6 +87,7 @@ export default async function run(config, storage, user, query) {
     await fse.writeJSON(stacpath, item, {spaces: 2});
 
     logger.info("Finished");
+    // todo: set to error is any task failed
     storage.updateJobStatus(query, 'finished');
   } catch(e) {
     logger.error(e);
@@ -77,17 +112,11 @@ async function createSTAC(storage, job, results) {
   let startTime = null;
   let endTime = null;
   const extents = [];
-  for(const { filepath, datacube } of results) {
-    const filename = path.basename(filepath);
-    const stat = await fse.stat(filepath);
-    let asset = {
-      href: path.relative(folder, filepath),
+  for(const result of results) {
+    const files = result.files || [];
+    const datacube = result.datacube;
+    const baseAsset = {
       roles: ["data"],
-      type: Utils.extensionToMediaType(filepath),
-      title: filename,
-      "file:size": stat.size,
-      created: stat.birthtime,
-      updated: stat.mtime
     };
 
     if (datacube.hasT()) {
@@ -109,8 +138,8 @@ async function createSTAC(storage, job, results) {
       const extent = datacube.getSpatialExtent();
       let wgs84Extent = extent;
       if (crs !== 4326) {
-        asset["proj:epsg"] = crs;
-        asset["proj:geometry"] = extent;
+        baseAsset["proj:epsg"] = crs;
+        baseAsset["proj:geometry"] = extent;
         wgs84Extent = Utils.projExtent(extent, 4326);
       }
       // Check the coordinates with a delta of 0.0001 or so
@@ -120,8 +149,34 @@ async function createSTAC(storage, job, results) {
       }
     }
 
-    const params = datacube.getOutputFormatParameters();
-    assets[filename] = Object.assign(asset, params.metadata);
+    for (const file of files) {
+      let asset;
+      let filename;
+      if (Utils.isUrl(file)) {
+        let url = new URL(file);
+        filename = path.basename(url.pathname || url.hash.substring(1));
+        asset = {
+          href: file,
+  //      type: Utils.extensionToMediaType(file),
+          title: filename
+        };
+      }
+      else {
+        filename = path.basename(file);
+        const stat = await fse.stat(file);
+        asset = {
+          href: path.relative(folder, file),
+          type: Utils.extensionToMediaType(file),
+          title: filename,
+          "file:size": stat.size,
+          created: stat.birthtime,
+          updated: stat.mtime
+        };
+      }
+
+      const params = datacube.getOutputFormatParameters();
+      assets[filename] = Object.assign(asset, baseAsset, params.metadata);
+    }
   }
   const item = {
     stac_version: packageInfo.stac_version,

src/api/worker/sync.js

@@ -25,7 +25,11 @@ export default async function run(config, user, id, process, log_level) {
   if (pg.getResults().length > 1) {
     logger.warn("Multiple results can't be processed in synchronous mode. Only the result from the result node will be returned.");
   }
-  return await GeeResults.retrieve(context, resultNode.getResult(), logger);
+
+  const dc = resultNode.getResult();
+  const format = config.getOutputFormat(dc.getOutputFormat());
+  const dc2 = format.preprocess(GeeResults.SYNC, context, dc, logger);
+  return await format.retrieve(context.ee, dc2);
 }
 
 export async function getResultLogs(user_id, id, log_level) {

src/api/worker/webservice.js

@@ -9,7 +9,7 @@ export default async function run(config, storage, user, query, xyz) {
 
   try {
     const rect = storage.calculateXYZRect(...xyz);
-    const context = config.processingContext(user);
+    const context = config.processingContext(user, service);
     // Update user id to the user id, which stored the job.
     // See https://github.com/Open-EO/openeo-earthengine-driver/issues/19
     context.setUserId(service.user_id);
@@ -29,7 +29,9 @@ export default async function run(config, storage, user, query, xyz) {
       dc.setOutputFormat('png');
     }
 
-    return await GeeResults.retrieve(context, dc, logger);
+    const format = config.getOutputFormat(dc.getOutputFormat());
+    const dc2 = format.preprocess(GeeResults.SERVICE, context, dc, logger);
+    return await format.retrieve(context.ee, dc2);
   } catch(e) {
     logger.error(e);
     throw e;

src/formats/bitmap.js

@@ -3,6 +3,7 @@ import GeeResults from "../processes/utils/results.js";
 import DataCube from "../datacube/datacube.js";
 import Utils from "../utils/utils.js";
 import FileFormat, { EPSGCODE_PARAMETER, SIZE_PARAMETER } from "./fileformat.js";
+import HttpUtils from "../utils/http.js";
 
 export const EPSGCODE_PARAMETER_BITMAP = Object.assign({}, EPSGCODE_PARAMETER);
 EPSGCODE_PARAMETER_BITMAP.default = 4326;
@@ -96,7 +97,7 @@ export default class BitmapLike extends FileFormat {
     return renderer === 'filmstrip';
   }
 
-  preprocess(context, dc, logger) {
+  preprocess(mode, context, dc, logger) {
     const ee = context.ee;
 		const parameters = dc.getOutputFormatParameters();
 
@@ -175,7 +176,7 @@ export default class BitmapLike extends FileFormat {
           reject('Download URL provided by Google Earth Engine is empty.');
         }
         else {
-          resolve(url);
+          resolve(HttpUtils.stream(url));
         }
       });
     });

src/formats/fileformat.js

@@ -16,7 +16,7 @@ export const SIZE_PARAMETER = {
 
 export const SCALE_PARAMETER = {
   type: 'number',
-  description: 'Scale of the image in meters per pixel.',
+  description: 'Scale of the image in meters per pixel. Defaults to native resolution in batch jobs, and 100 otherwise.',
   default: 100,
   minimum: 1
 };
@@ -81,15 +81,19 @@ export default class FileFormat {
     };
   }
 
-  preprocess(context, dc/*, logger*/) {
+  preprocess(mode, context, dc/*, logger*/) {
     return dc;
   }
 
-  async retrieve(/*ee, dc*/) {
+  async retrieve(/*ee, dc */) {
     throw new Error('Not implemented');
   }
 
-  async export(/*ee, dc*/) {
+  canExport() {
+    return false;
+  }
+
+  async export(/*ee, dc */) {
     throw new Error('Not implemented');
   }
 

src/formats/gif.js

@@ -1,3 +1,4 @@
+import HttpUtils from "../utils/http.js";
 import Utils from "../utils/utils.js";
 import BitmapLike from "./bitmap.js";
 
@@ -53,7 +54,7 @@ export default class GifFormat extends BitmapLike {
           reject('Download URL provided by Google Earth Engine is empty.');
         }
         else {
-          resolve(url);
+          resolve(HttpUtils.stream(url));
         }
       });
     });