Data Ingestion

This section details the data ingestion and later 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.

Via the Redis, the ingestion can be triggered and observed. In order to eventually start the preprocessing of a product, its path on the configured object storage has to be pushed onto the preprocess_queue, as will be explained in detail in this chapter.

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.

When doing one off commands, it is maybe more convenient to execute it on a running service. For this, the docker ps command can be used to select the identifier of the running docker container of the redis service.

container_id=$(docker ps -qf "name=<stack-name>_redis")

With this identifier, a command can be issued:

docker exec -it $container_id redis-cli ...

When performing more than one command, it can be simpler to open a shell on the service instead:

docker exec -it $container_id bash

As the container ID may change (for example when the replica is restarted) it is better to retrieve it for every command instead of relying on a variable:

docker exec -it $(docker ps -qf "name=<stack-name>_redis")

For the sake of brevity, the subsequent commands in this chapter can be used with either of the above techniques and will just print the final commands that are run inside the redis container.

Note

For the VS, only the List and Set Redis data types are really used.

Sets are an unordered collection of string elements. In the VS it is used to denote that an element is part of a particular group, e.g: being preprocessed, or having failed registration.

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 in order to add an element to be processed as soon as possible instead of waiting before all other elements before it are processed.

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

For a more concrete example the following command executes a redis-cli lpush command to add a new path of an object to preprocess on the preprocess_queue:

redis-cli lpush preprocess_queue "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"

Usually, with a preprocessor service running and no other items in the preprocess_queue this value will be immediately popped from the list and processed. For the sake of demonstration this command would print the contents of the preprocess_queue:

$ redis-cli lrange preprocess_queue 0 -1
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

Now that the product is being preprocessed, it should be visible in the preprocessing_set. As the name indicates, this is using the Set datatype, thus requiring the SMEMBERS subcommand to list:

$ redis-cli smembers preprocessing_set 0 -1
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

Once the preprocessing of the product is finished, the preprocessor will remove the currently worked on path from the preprocessing_set and add it either to the preprocess-success_set or the preprocess-failure_set depending on whether the processing succeeded or not. They can be inspected using the same SMEMBERS subcommand with one of set names as a parameter.

Additionally, upon success, the preprocessor places the same product path on the register_queue, where it can be inspected with the following command.

$ redis-cli lrange register_queue 0 -1
/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

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

redis-cli lpush register_queue "/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"

Very similar to the preprocessing, during the registration the product path is added to the registering_set, afterwards the path is placed to either the register-success_set or register-failure_set. Again, these queues or sets can be inspected by the LRANGE or SMEMBERS subcommands respectively.

Ingestor and sftp

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 from “live” sources. Ingestor service, together optionally with sftp service allow data ingestion to be initiated by external means.

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 path to the object storage with product to be ingested. It then parses this information and internally puts it into configured redis queue (preprocess or register).

  • Alternative: Listening for newly added Browse Report or Availability Report files on a configured path on a file system via inotify.

These Browse Report files need to be in an agreed XML schema to be correctly handled. Sftp service enables a 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:
  environment:
    REDIS_PREPROCESS_MD_QUEUE_KEY: "preprocess_queue" # to override md_queue (json) and instead use (string)
  command:
    ["python3", "/filedaemon.py"]

Direct Data Management

Sometimes it is necessary to directly interact with the preprocessor or registrar. The following section shows what tasks on the preprocessor and registrar can be accomplished.

Warning

This approach is not recommended for everyday use, as it circumvents the Redis sets to track what products have been registered and where the registration failed.

Preprocessing

In this section all command examples are assumed to be run from within a running preprocessor container. To open a shell on a preprocessor, the following command can be used.

docker exec -it $(docker ps -qf "name=<stack-name>_preprocessor") bash

The preprocessor 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 the Redis queue for new items, which will be preprocessed one by one. The second mode is to run the preprocessor 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.

preprocess \
    --config-file /preprocessor_config.yml \
    --validate \
    --use-dir /tmp \
    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

In order to preprocess a ngEO Ingest Browse Report, an additonal --browse-report parameter needs to be added:

preprocess \
    --config-file /preprocessor_config.yml \
    --browse-report \
    --use-dir /tmp \
    browse_report_test1.json

In this “one-off” mode, the item will not be placed in the resulting set (preprocessing_set, preprocess-success_set, and preprocess-failure_set).

Registration Handling

For all intents and purposes in this section it is assumed, that the operator is logged into a shell on the registrar service. This can be achieved via the following command (assuming at least one registrar replica is running):


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

alias manage.py='python3 /var/www/pvs/dev/pvs_instance/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.

Product Handling

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 to the console and can be used to put it into a collection. Additionally, it is necessary to add a 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
Deregistration

Products and coverages need to be derigestered when no longer in use. A product can be deregistered using its identifier:

manage.py product deregister "${product_id}"

The contained coverage must also be deregistered manually:

manage.py coverage deregister "${product_id}_coverage"

Preprocessing vs registration

The preprocessing step aims to ensure that cloud optimized GeoTIFF (COG) files are created in order to significantly speed up the viewing of large amount 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 for most of the times, 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 allowing access to the metadata and catalog functions, while justifying slower rendering times can be preferred.

Monitoring ingestion

Monitoring ingestion can be done on production system easily via Kibana using its query language KQL. Kibana in Discover mode shows 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.

In order to watch for successful registrations or preprocessing campaigns, simply search for

"<stack-name>_registrar" AND "Successfully"

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 preprocessing following search query can be used:

"<stack-name>_preprocessor" AND "ERROR" AND NOT "Target.replace"

Preprocessor and registrar by default run in mode, where they skip already registered/preprocessed products. This KQL query does not list errors like “file is already in target storage”.

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

"emg-pdas_registrar" AND "data26/0000171398/PL00_DOV_MS_L3A_20190313T075450_20190313T075450_PLA_000000_D5E9.DIR.tar"

Then click on an arrow on 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).

It is also advisable to filter the logs per container (showing only logs from that registrar/preprocessor container, that has selected surrounding documents).

Querying for ingestor logs allows to see if while using the ingestor push ingestion mode, the XML was parsed correctly.

"<stack-name>_ingestor"

Next chapter Access describes used authorization and authentication concepts and lines out how the external access to individual components and service as such is configured.