# Content Streamers The information provided on this page explains _Content Streamers_ and their usage within the Squirro platform. This information is intended for advanced users. > **Note** > > It is not necessary to read this page to ingest data into Squirro. ## Overview In Squirro, a content streamer is a component that provides the necessary functionality for managing the `inputstream`. For example, it provides the ability to: - Enqueue a new batch (data file) to the `inputstream`. - Dequeue a batch from the `inputstream`. - Move a processed batch to the `processed` sub-directory. - Move a failed batch to the `failed` sub-directory. - Postpone the processing of a batch. The `inputstream` is the directory (by default, located at `/var/lib/squirro/inputstream`) that stores the data loaded by a Data Loader plugin and is meant to be transformed and indexed into Squirro by a defined data processing pipeline. ### Types of Content Streamers There are two types of content streamers available: - The `FileSystemStreamer`, which is the default content streamer. - The `QueueFileSystemStreamer`, which builds on top of the `FileSystemStreamer` streamer by adding the notion of a distinct queue for each data source. This streamer relies on the Redis server of the Squirro installation to do its work. ## FileSystemStreamer The `FileSystemStreamer` works by manipulating the `inputstream` directory. When a service wants to enqueue a new batch of data (for example, the `provider` service), it uses the `enqueue` method of the `FileSystemStreamer`. This method checks the current date and hour and creates a new `data dir` of the form `%Y-%m-%d-%H` (e.g., `2023-05-21-18`, which corresponds to `May 05 2023 at 18:00`) if it doesn’t exist already. In this data dir, it adds the batch of data into a new file (called `data file` or simply `batch`) with a filename that adheres to the following form: ``` data_project__source__batch_[_priority_].json ``` Where: - `project_id` is the ID of the project to which the data belongs. - `source_id` is the ID of the data source from which the data was loaded. - `batch_id` is simply a unique ID to distinguish this batch from other batches of the same data source. - `priority_level` is the priority level of the data source that this batch of data came from. It is only present if the data source has a priority level different from the default one (which is `normal`). When a service wants to dequeue a batch from the `inputstream` (for example, the `ingester` service), it uses the `dequeue` method of the `FileSystemStreamer`. ### Example Usage To explain how this method works, consider a simple example. Assume that the state of the `inputstream` directory is as follows: ``` . ├── 2023-05-21-13 │ ├── data_project_tTSjA3BVRlCFBbCIydh27A_source_Fc723hf2S4Waeo9SVClSDw_batch_YP5Ncdu9QxK8BEgfoZ8msQ.json │ └── data_project_tTSjA3BVRlCFBbCIydh27A_source_IrbxJh20TJKxoFkyci4N9w_batch_Pa6H0a1uR4SHqvroTgN0pg.json └── 2023-05-21-14 └── data_project_tTSjA3BVRlCFBbCIydh27A_source_40G0-m0WSnuv4pcm_FfAUA_batch_7T9Nqb-dQcC6QCAfOhWb7g.json ``` In the above `tree` output you can see 3 batches in 2 data dirs (2023-05-21-13, 2023-05-21-14), from 3 data sources (Fc723hf2S4Waeo9SVClSDw, IrbxJh20TJKxoFkyci4N9w, 40G0-m0WSnuv4pcm_FfAUA), with the same priority level (the default one, which is normal; for this reason is missing from the filename), from the same project (tTSjA3BVRlCFBbCIydh27A). Also, assume a single processor is used for processing the `normal` priority batches. Reference: For information on configuring the processors of the `ingester` service, see [Pipeline Prioritization](priorities.md#data-processing-pipeline-priorities). Therefore, the 3 batches will be processed by the same processor. Now, when the processor uses the `dequeue` method of the `FileSystemStreamer`, the following steps take place: 1. The streamer scans the `inputstream` directory and chooses the oldest data dir first. In this case, the 2023-05-21-13 data dir. 2. Then, the streamer scans the selected data dir for available batches and sorts them by modification date. The oldest batch of the selected data dir is picked up first. 3. To signal that this batch is now owned by this processor, it renames the data file by appending the ID of the processor in the end of the filename (_processor_). This is the locking mechanism employed by this type of streamer. 4. If the locking is successful, it extracts some metadata from the batch in order to find the appropriate pipeline workflow for processing the data in this batch. 5. If the data in the batch are processed successfully through the pipeline steps of the selected workflow, the streamer removes the batch from the inputstream directory. ## QueueFileSystemStreamer > **Warning** > > The `QueueFileSystemStreamer` is currently in beta. It is supported and recommended for use on single-node deployments. However, it should not be used on multi-node production deployments, where the `FileSystemStreamer` should continue to be used. Please report any issues or requests related to the `QueueFileSystemStreamer`. The `QueueFileSystemStreamer` (also known as the queue-based filesystem streamer or QFSS) extends the `FileSystemStreamer` by adding a layer where the available batches of each data source are stored in a dedicated queue in Redis. This setup distinguishes the `inputstream` into two layers: the message queue layer (Redis) and the pure storage layer (filesystem directory). ### Enqueue Method When the `enqueue` method of the `QueueFileSystemStreamer` is called, the logic of the `FileSystemStreamer.enqueue` method is invoked first to create the batch in the filesystem (i.e., in the `inputstream` directory). Then, having the absolute path to the enqueued batch, it does the following: 1. Adds the queue of the data source that the batch belongs to in an `index of queues` Redis key (a set) called `qfss:queues_index:`. This serves as a registry of all the available queues for a given priority level. 2. Pushes the absolute path of the batch in the queue of the data source that the batch belongs to. The name of the queue is of the form `qfss:queue:__`. ### Dequeue Method When the `dequeue` method of the `QueueFileSystemStreamer` is called, the following steps occur: 1. The streamer fetches the current `index of queues` and checks decides from which queue to dequeue from. The decision is based on a simple round-robin algorithm which tries to process a batch from a different source and project every time in order to give a fair chance to all the data sources to be processed. 2. Based on the selected queue, the streamer pops one item from it, which is the absolute path of the batch to be processed. The locking mechanism is essentially handled here by Redis, no other streamer is able to pop the same batch from the queue anymore. 3. Finally, having the absolute path to the batch, the streamer invokes part of the `FileSystemStreamer.dequeue` logic to actually dequeue the data from the filesystem and give it to the processor to continue with the processing. ## Configuration The `stream_type` configuration option in the section `[content]` of the `/etc/squirro/common.ini` file, controls which type of content streamer is used by all the services of a Squirro instance. Note: The default value is `filesystem`, which corresponds to the `FileSystemStreamer`. In order to use the `QueueFileSystemStreamer`, you need to change the value of this option to `filesystem-queue`. If the `[content]` section is not present in the `/etc/squirro/common.ini` file, then you are free to add it and define your own value for the `stream_type` option. ### Example Usage The following example shows that the `QueueFileSystemStreamer` is used: ```ini [content] stream_type = filesystem-queue ```