.. _ingestion: 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 :ref:`running_commands_k8s` or :ref:`running_commands_swarm` 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: .. code-block:: bash 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: Harvesting ^^^^^^^^^^^^^ The following command executes a ``redis-cli lpush`` command to trigger a new harvesting operation on a "Kompsat2" harvester .. code-block:: bash redis-cli lpush harvester_queue '{"name":"Kompsat2"}' To check a certain queue, do the following: .. code-block:: bash $ redis-cli lrange harvester_queue 0 -1 {'name':'Kompsat2'} For configuration details, see :ref:`harvester_configuration`. 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. Preprocessing ~~~~~~~~~~~~~ In this section, all command examples are assumed to be run from within a running preprocessor container. .. code-block:: bash preprocess \ --config-file /config.yml \ {STAC_ITEM_JSON_STRING} 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. 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: .. code-block:: bash 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: .. code-block:: bash manage.py collection insert Multiple products can be inserted in one pass by providing more than one identifier. The reverse command excludes a product from a collection: .. code-block:: bash manage.py collection exclude Again, multiple products can be excluded in a single call. Product Handling ~~~~~~~~~~~~~~~~ Registration Products can be registered using the EOxServer CLI tools as well. .. code-block:: bash 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: .. code-block:: bash 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 deregistered when no longer in use. A product can be deregistered using its identifier: .. code-block:: bash manage.py product deregister "${product_id}" Preprocessing vs registration ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 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 of ingestion ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 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 .. code-block:: SQL 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: .. code-block:: bash 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: .. code-block:: SQL 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: .. code-block:: SQL "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). Ingestor and sftp - PRISM specific ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 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..ops.yml for ``ingestor`` service needs to be set to ``python3 filedaemon.py``: .. code-block:: yaml ingestor: command: ["python3", "/filedaemon.py"]