Data Ingestion

This section details the data ingestion and management in the VS.

Redis Queues

The central synchronization component in the VS is the redis key-value store. It provides various queues, which the services are listening to. For operators, it provides a high-level interface through which data products can be registered and managed.

As the Redis store is not publicly accessible from outside of the stack. So to interact with it, the operator has to run a command from one of the services. Conveniently, the service running Redis also has the redis-cli tool installed that lets users interact with the store.

Please see Running commands in VS services or Running commands in VS services to learn how commands can be run in Kubernetes and Docker Swarm respectively.

Note

For the VS, only the List Redis data type is used.

Lists are used as a task queue. It is possible to add items to either end of the queue, but by convention items are pushed on the “left” and popped from the “right” end of the list resulting in a first-in-first-out (FIFO) queue. It is entirely possible to push elements to the “right” end as well, and an operator may want to do so to add an element to be processed as soon as possible instead of waiting before all other elements before it is processed.

The full list of available commands can be found for Lists .

If an operator wants to trigger only the re-registration of a product without preprocessing the STAC Item needs to be pushed to this queue:

redis-cli lpush register_queue '{"type": "Feature", "stac_version": "1.0.0", "id": "urn:eop:DOVE:MULTISPECTRAL_4m:20180811_081455_1054_3be7", "properties": {"product_type": "DOV_MS_L3A", "start_datetime": "2018-08-11T08:14:55Z", "end_datetime": "2018-08-11T08:14:55Z", "datetime": "2022-10-27T11:53:45Z"}, "geometry": null, "links": [], "assets": {"gsc_metadata": {"href": "OA/PL00/1.0/00/urn:eop:DOVE:MULTISPECTRAL_4m:20180811_081455_1054_3be7/0001/PL00_DOV_MS_L3A_20180811T081455_20180811T081455_TOU_1234_3be7.DIMA.tar/GSC#CR#ESA#VHR_IMAGE_2018#20190706#165304.xml", "type": "application/xml", "title": "GSC Metadata file", "description": "GSC metadata file from source archive", "roles": ["metadata"]}, "ms": {"href": "OA/PL00/1.0/00/urn:eop:DOVE:MULTISPECTRAL_4m:20180811_081455_1054_3be7/0001/PL00_DOV_MS_L3A_20180811T081455_20180811T081455_TOU_1234_3be7.DIMA.tar/IMG_PL00_DOV_MS_L3A_20180811T081455_20180811T081455_TOU_1234_3be7_R1C1.TIF", "type": "image/tiff; application=geotiff; profile=cloud-optimized", "title": "Preprocessed image", "description": "Product image converted into a COG", "raster:bands": [{"nodata": 0.0, "data_type": "uint16"}], "eo:bands": [{"name": "band1", "common_name": "band1"}], "roles": ["data"]}}, "stac_extensions": []}'

STAC Item assets should either contain the full URL or full path to the asset with a protocol prefix (eg. https://, s3://) or if a storage.container, storage.endpoint_url or storage.root_directory is configured, then it can also contain only the object name.

Example asset paths:

s3 storage.bucket defined:

OA/PL00/1.0/00/urn:eop:DOVE:MULTISPECTRAL_4m:20180811_081455_1054_3be7/0001/PL00_DOV_MS_L3A_20180811T081455_20180811T081455_TOU_1234_3be7.DIMA.tar/GSC#CR#ESA#VHR_IMAGE_2018#20190706#165304.xml

s3 storage.bucket not defined:

s3://test-data/OA/PL00/1.0/00/urn:eop:DOVE:MULTISPECTRAL_4m:20180811_081455_1054_3be7/0001/PL00_DOV_MS_L3A_20180811T081455_20180811T081455_TOU_1234_3be7.DIMA.tar/GSC#CR#ESA#VHR_IMAGE_2018#20190706#165304.xml

Harvesting

The following command executes a redis-cli lpush command to trigger a new harvesting operation on a “Kompsat2” harvester

redis-cli lpush harvester_queue '{"name":"Kompsat2"}'

To check a certain queue, do the following:

$ redis-cli lrange harvester_queue 0 -1
{'name':'Kompsat2'}

For configuration details, see Harvester configuration - harvester.

Direct Data Management

Sometimes it is necessary to interact with the services for ingestion directly.

Harvester, registrar, and preprocessor services can be used in two modes. The first (and default mode when used as a service) is to be run as a daemon: it listens to a Redis queue for new items, which will be triggered one by one. The second mode is to run the service in a “one-off” mode: instead of pulling an item from the queue, it is passed as a command line argument, which is then processed normally.

In this section, all command examples are assumed to be run from within a running preprocessor container.

preprocess \
    --config-file /config.yml \
    {STAC_ITEM_JSON_STRING}

For all intents and purposes in this section, it is assumed, that the operator is logged into a shell on the registrar service.

The content of the shared registrar/renderer database can be managed using the registrar’s instance manage.py script. For brevity, the following bash alias is assumed:

alias manage.py='python3 $INSTANCE_DIR/manage.py'

A collection is a grouping of earth observation products, accessible as a single entity via various service endpoints. Depending on the configuration, multiple collections are created when the service is set up. They can be listed using the collection list command.

New collections can be created using the collection create command. This can refer to a Collection Type, which will restrict the collection in terms of insertable products: only products of an allowed Product Type can be added. Detailed information about the available Collection management commands can be found in the CLI documentation.

Collections can be deleted, without affecting the contained products.

Warning

As some other services have fixed configurations and depend on specific collections, deleting said collections without a replacement can lead to configuration inconsistencies and ultimately service disruptions.

In certain scenarios, it may be useful to add specific products to or exclude them from a collection. For this, the Product identifier needs to be known. To find out the Product identifier, either a query of the existing collection via OpenSearch or the CLI command id list can be used.

When the identifier is obtained, the following management command inserts a product into a collection:

manage.py collection insert <collection-id> <product-id>

Multiple products can be inserted in one pass by providing more than one identifier.

The reverse command excludes a product from a collection:

manage.py collection exclude <collection-id> <product-id>

Again, multiple products can be excluded in a single call.

Registration

Products can be registered using the EOxServer CLI tools as well.

manage.py product register \
    --metadata-file data25 /OA/PL00/1.0/00/urn:eop:DOVE:MULTISPECTRAL_4m:20180811_081455_1054_3be7/0001/PL00_DOV_MS_L3A_20180811T081455_20180811T081455_TOU_1234_3be7.DIMA.tar/metadata.xml \
    --print-identifier \
    --type PL00

The identifier of the newly registered product is printed on the console and can be used to put it into a collection. Additionally, it is necessary to add coverage to it, which can be registered like:

manage.py coverage register \
    -d data25 /OA/PL00/1.0/00/urn:eop:DOVE:MULTISPECTRAL_4m:20180811_081455_1054_3be7/0001/PL00_DOV_MS_L3A_20180811T081455_20180811T081455_TOU_1234_3be7.DIMA.tar/some.tif \
    -m data25 /OA/PL00/1.0/00/urn:eop:DOVE:MULTISPECTRAL_4m:20180811_081455_1054_3be7/0001/PL00_DOV_MS_L3A_20180811T081455_20180811T081455_TOU_1234_3be7.DIMA.tar/metadata.xml \
    --identifier "${product_id}_coverage" \
    --type RGBNir

Products and coverages need to be deregistered when no longer in use. A

product can be deregistered using its identifier:

manage.py product deregister "${product_id}"

The preprocessing step aims to ensure that cloud-optimized GeoTIFF (COG) files are created to significantly speed up the viewing of a large volume of data in lower zoom levels. There are several cases, where such preprocessing is not necessary or wanted.

If data are already in COGs and in favorable projection, which will be presented to the user most of the time, direct registration should be used. This means paths to individual products will be pushed directly to the register_queue.
Also for cases, where preprocessing step would take too much time, direct registration allows access to the metadata and catalog functions, while justifying slower rendering times can be preferred.

Monitoring ingestion can be done on the production system easily via Kibana using its query language KQL. Kibana in Discover mode shows a time histogram of individual entries, which makes it easy to visually infer the ingestion progress in time. These queries can be saved for later use and more importantly to set up alerts and statistics on these saved queries.

To watch for successful registrations or preprocessing campaigns, simply search for

event: is one of Successfully registered Product, Successfully replaced Product

Example of such a query, filtering data for one day into the past from now:

https://kibana.pdas.prism.eox.at/app/discover#/?_g=(filters:!(),refreshInterval:(pause:!t,value:0),time:(from:now-1d,to:now))&_a=(columns:!(log,container_name),filters:!(),index:'57007c50-f270-11ea-8728-ab85b3e61ad6',interval:auto,query:(language:kuery,query:'"emg-pdas_registrar"%20AND%20"Successfully"'),sort:!())

stack-name, kibana-url and elasticsearch-index-id needs to be substituted with valid values.

For failures in registration, a query would look like this:

container_name: *registrar
exception: exists

For checking the status of individual product ingestion (for example to find out why it failed), it can be searched for its identifier and then list surrounding documents and filter them by docker container name. An example query would be:

"urn:eop:PNE:PMS__3_0.3m:ACQ_PNEO4_02520705839470"

Then click on an arrow on the left border of the individual log message (document) to display more details -> View surrounding documents link appears, which lists other logs close in time to this one (default 5 before and 5 after).

Triggering preprocessing and registration via pushing to the Redis queues is very convenient for single ingestion campaigns, but not optimal for continuous ingestion of new products by external sources via the push approach.

Ingestor service, together optionally with sftp service allows data ingestion to be initiated by an external trigger.

Ingestor can work in two modes:

Default: Exposing a simple / endpoint, and listening for POST requests containing data with either a Browse Report XML, Browse Report, or a string with the path to the object storage with a product to be ingested. It then parses this information and internally pushes it into configured Redis queue.
Alternative: Listening for newly added Browse Report or Availability Report files on a configured path on a file system via inotify.

The Browse Report files need to be in an agreed XML schema to be correctly handled.

Sftp service enables secure access to a configured folder via sftp, while this folder can be mounted to other vs services. This way, Ingestor can listen for newly created files by the sftp access.

If the filedaemon alternative mode should be used, INOTIFY_WATCH_DIR environment variable needs to be set and a command used in the docker-compose.<stack>.ops.yml for ingestor service needs to be set to python3 filedaemon.py:

ingestor:
  command:
    ["python3", "/filedaemon.py"]