Refactor Monitoring

CHANGELOG.md

@@ -4,39 +4,34 @@
 
 ### Added
 
-- [API] Added `lithops.concurrent.futures`, a `concurrent.futures`-compatible executor interface (`submit`, eager `map`, stdlib `wait`/`as_completed`) backed by Lithops. See issue #1427.
-- [Docs] Renamed the native executor documentation from "Futures API" to "Core API".
+- [API] Added `lithops.concurrent.futures`, a `concurrent.futures`-compatible executor interface backed by Lithops.
 - [Tests] Added a unit test suite for all non-backend modules (18 files, 876 tests).
-- [Core] Added a `log_prefix()` helper for uniform log prefixes across core and backends.
-- [Core] Added a cache of serialized functions to avoid re-uploading the same function.
-- [Core] Added `ShutdownSafeStreamHandler` to avoid tracebacks when logging on a closed stream.
-- [Localhost] Added `localhost/utils.py` with helpers shared by the v1 and v2 backends.
-- [AWS Batch] Added `instance_types` config option for EC2/SPOT compute environments.
+- [Monitoring] Added Redis, AWS SQS (`aws_sqs`), GCP Pub/Sub (`gcp_pubsub`) and Azure Queue Storage (`azure_queue`) monitoring backends.
 
 ### Changed
 
 - [Worker] Replaced the `multiprocessing` Manager queue of the worker pool with a POSIX pipe.
 - [Core] Results under 8KB now travel in the call status instead of a separate storage object.
-- [Core] Reorganised all non-backend modules for readability, with no behaviour change.
+- [Core] Added a cache of serialized functions to avoid re-uploading the same function.
+- [Monitoring] Reorganised job monitoring as pluggable backends.
 - [Core] `wait()` now returns two empty lists for empty input instead of `None`.
-- [Core] `verify_args()` now raises a single message instead of a tuple.
-- [Monitoring] The RabbitMQ queues of a call status now travel with the job.
 - [CLI] `job list`, `worker list`, `image delete` and `image list` now reject unknown flags.
-- [CLI] `lithops clean --all` no longer shadows the `all` builtin.
-- [CLI] `lithops clean` now empties the local temp directory instead of removing it, and leaves the pending cleaner requests of the other processes alone.
+- [CLI] `lithops clean` now empties the local temp directory instead of removing it.
 - [Storage] `CloudFileProxy.walk()` now yields nothing for a missing path, like `os.walk`.
 - [Storage] `cloud_open()` now raises `ValueError` on an unsupported mode.
 - [Joblib] Capped the shared-argument upload and download pools at 32 threads.
 - [Joblib] `lithops_args` is now applied to the pool that runs the batches.
 - [Standalone] `docker login` now reads the password from stdin and quotes its arguments.
+- [AWS Batch] Allow to set `instance_types` config option for EC2/SPOT compute environments.
 
 ### Fixed
 
 - [Core] Fixed `wait()` on futures another executor invoked, which crashed with an `AttributeError` in `JobMonitor.is_alive()` and, once past it, watched the wrong storage prefix and never returned.
 - [Chaining] Fixed pickling a `FuturesList` detaching the list being pickled from its executor.
 - [Chaining] Fixed a list or a slice of futures of a previous job not being recognised as a chain, which failed with an argument binding error instead.
 - [Chaining] `extra_args` now raises at submit time instead of letting every activation of the chained job fail on a missing argument.
-- [Localhost] Fixed a deadlock on a `map` after `wait()` and `get_result()`, caused by stale work queue sentinels.
+- [Monitoring] Redis, RabbitMQ and SQS now delete their queues only in ``cleanup()``, and keep the monitor thread until ``stop()``.
+- [Monitoring] Status lines (Pending/Running/Done) are logged on start, every 30s, and when the job finishes, not on every activation.
 - [Localhost] Fixed a partial `clear()` tearing down the consumers, tasks and latches of other jobs.
 - [Localhost] Fixed a task starting after `stop()`, leaving a process nobody kills.
 - [Localhost] Fixed the v2 job manager spinning a core while an invocation was queueing.
@@ -53,11 +48,11 @@
 - [Core] Fixed `find_free_port()` setting `SO_REUSEADDR` after the bind.
 - [Core] Fixed module inspection crashing on a function whose `__module__` is `None`.
 - [Core] Fixed a hand-built `FuturesList` raising `AttributeError` instead of creating its executor.
-- [Core] Fixed the cleaner skipping requests and two cleaners racing for the pid file.
-- [Core] Fixed `lithops clean` deleting the local temp directory of the jobs running at the same time on the same machine.
-- [Core] Fixed the cleaner reading a request another process was still writing.
-- [Core] Fixed the cleaner looping forever on a request it could not read or classify.
-- [Core] Fixed the cleaner lock surviving a killed cleaner and blocking every later one.
+- [Cleaner] Fixed the cleaner skipping requests and two cleaners racing for the pid file.
+- [Cleaner] Fixed `lithops clean` deleting the local temp directory of the jobs running at the same time on the same machine.
+- [Cleaner] Fixed the cleaner reading a request another process was still writing.
+- [Cleaner] Fixed the cleaner looping forever on a request it could not read or classify.
+- [Cleaner] Fixed the cleaner lock surviving a killed cleaner and blocking every later one.
 - [Monitoring] Fixed a nested executor publishing statuses to a queue nobody declares.
 - [Monitoring] Fixed the failed RabbitMQ publishes being dropped with nothing in the log.
 - [Worker] Fixed the memory monitor reporting a peak of zero where usage cannot be read.

config/README.md

@@ -154,7 +154,7 @@ if __name__ == '__main__':
 | lithops | backend             | aws_lambda   | no        | Compute backend implementation. `localhost` is the default if no config or config file is provided. |
 | lithops | storage             | aws_s3       | no        | Storage backend implementation. `localhost` is the default if no config or config file is provided. |
 | lithops | data_cleaner        | True         | no        | If True, automatically deletes temporary data written to `storage_bucket/lithops.jobs`.          |
-| lithops | monitoring          | storage      | no        | Monitoring system implementation. Options: **storage** or **rabbitmq**.                           |
+| lithops | monitoring          | storage      | no        | Monitoring backend. Built-in: **storage**, **rabbitmq**, **redis**, **aws_sqs**, **gcp_pubsub**, **azure_queue**. |
 | lithops | monitoring_interval | 2            | no        | Interval in seconds for monitoring checks when using **storage** monitoring.                      |
 | lithops | data_limit          | 4            | no        | Maximum size (in MB) for iterator data chunks. Set to False for unlimited size.                   |
 | lithops | execution_timeout   | 1800         | no        | Maximum execution time in seconds for functions. Functions exceeding this time are terminated. Can also be set per call via the `timeout` parameter. |

config/config_template.yaml

@@ -23,7 +23,7 @@
     #backend: aws_lambda                  # Default: aws_lambda
     #storage: aws_s3                      # Default: aws_s3
     #data_cleaner: True                   # Clean intermediate data after execution. Default: True
-    #monitoring: storage                  # One of: storage, rabbitmq. Default: storage
+    #monitoring: storage                  # Monitoring backend name. Built-in: storage, rabbitmq, redis, aws_sqs, gcp_pubsub, azure_queue. Default: storage
     #monitoring_interval: 2               # Polling interval (seconds) when monitoring=storage. Default: 2
     #data_limit: 4                        # Max iterdata aggregated size (MiB) before using storage. Default: 4
     #execution_timeout: 1800              # Per-call timeout in seconds. Default: 1800
@@ -561,6 +561,35 @@
     #amqp_url: amqp://<USER>:<PASSWORD>@<HOST>:<PORT>/<VHOST>
 
 
+# =============================================================================
+# AWS SQS – optional overrides when `lithops.monitoring=aws_sqs`. Credentials
+# and region are taken from the `aws` section if this one is empty.
+# =============================================================================
+#aws_sqs:
+    #region: <REGION>                     # Mandatory here or under `aws`
+    #access_key_id: <ACCESS_KEY_ID>
+    #secret_access_key: <SECRET_ACCESS_KEY>
+    #session_token: <SESSION_TOKEN>
+
+
+# =============================================================================
+# GCP Pub/Sub – optional overrides when `lithops.monitoring=gcp_pubsub`.
+# Project and credentials are taken from the `gcp` section if this one is empty.
+# =============================================================================
+#gcp_pubsub:
+    #project_name: <GCP_PROJECT_ID>       # Mandatory here or under `gcp`
+    #credentials_path: <ABSOLUTE_PATH_TO_SERVICE_ACCOUNT_JSON>
+
+
+# =============================================================================
+# Azure Queue Storage – optional overrides when `lithops.monitoring=azure_queue`.
+# Account name and key are taken from `azure_storage` if this one is empty.
+# =============================================================================
+#azure_queue:
+    #storage_account_name: <STORAGE_ACCOUNT>
+    #storage_account_key: <STORAGE_ACCOUNT_KEY>
+
+
 # =============================================================================
 # Prometheus – pushed metrics when `lithops.telemetry=True`
 # =============================================================================

docs/source/design.rst

@@ -72,7 +72,7 @@ directly to a module or class in the ``lithops`` package:
      - Backend-specific dispatch. ``FaaSInvoker`` performs concurrent per-call
        invocations; ``BatchInvoker`` submits a single batch/job for many tasks.
    * - ``JobMonitor``
-     - ``lithops/monitor.py``
+     - ``lithops/monitoring/``
      - Tracks completion through either ``StorageMonitor`` (polling) or
        ``RabbitmqMonitor`` (push notifications).
    * - ``InternalStorage``

docs/source/monitoring.rst

@@ -1,36 +1,217 @@
 Monitoring
 ==========
 
-By default, Lithops uses the storage backend to monitor function activations: each function activation stores a file
-named *{id}/status.json* in Object Storage when it finishes its execution. This file contains statistics about the
-execution, including whether the function activation ran successfully or not. Based on these files, the default
-monitoring approach polls the Object Storage every few seconds to determine which function activations have finished
-and which have not.
+While a job runs, the client has to know which of its function activations have started, which have finished, and
+whether each one succeeded. Lithops calls this *monitoring*, and it is what makes ``wait()``, ``get_result()`` and the
+progress bar work.
 
-Since this default approach can slow down total application execution time due to the number of requests it has to
-make against the object storage, Lithops also integrates a RabbitMQ service to monitor function activations in real time.
-With RabbitMQ, the content of the *{id}/status.json* file is sent through a queue. This speeds up total application
-execution time since Lithops only needs one connection to the messaging service to monitor all function activations.
-We currently support the AMQP protocol.
+Monitoring is a *channel*, not a storage provider: it only decides how call statuses travel back to the client. Your
+compute and storage backends are unaffected by the choice, and so are your results.
 
-.. note:: The RabbitMQ server has to be accessible from both the client and the functions. For example, it could be deployed in a cloud server with a public IP address and with the AMQP port open (5672).
 
-To enable Lithops to use this service, add the *AMQP_URL* key into the *rabbitmq* section in
-the configuration, for example:
+How it works
+------------
+
+Every activation reports twice: once when it starts, and once when it finishes. The final report carries the execution
+statistics and whether the call raised an exception.
+
+Lithops can carry those reports in two ways.
+
+**Storage polling** (the default). Each activation writes its status into the storage backend the executor already
+uses, under a key of its own:
+
+.. code::
+
+    lithops.jobs/<executor_id>-<job_id>/<call_id>/<activation_id>.init   # the call started
+    lithops.jobs/<executor_id>-<job_id>/<call_id>/status.json            # the call finished
+
+The client lists those keys every couple of seconds to work out what has happened. It needs no extra infrastructure,
+which is why it is the default. The cost is that a job with many activations means many requests against the object
+storage, and a status is only noticed on the next poll.
+
+**Message services.** The same status is published to a queue or a topic instead. The client holds one connection and
+is notified as each message arrives, so a status shows up as soon as it is sent and the number of activations does not
+change how much the client has to ask for.
+
+.. note::
+    A message service has to be reachable **from both the client and the functions**. This is the usual reason a
+    message backend works locally but not in the cloud, or the other way around.
+
+
+Which one to use
+----------------
+
+.. list-table::
+   :header-rows: 1
+   :widths: 18 34 22 26
+
+   * - ``monitoring``
+     - What it needs
+     - Extra dependency
+     - Created per executor
+   * - ``storage``
+     - nothing, reuses your storage backend
+     - none
+     - nothing
+   * - ``rabbitmq``
+     - a RabbitMQ broker
+     - none
+     - one queue
+   * - ``redis``
+     - a Redis server
+     - ``lithops[redis]``
+     - one list
+   * - ``aws_sqs``
+     - an AWS account
+     - ``lithops[aws]``
+     - one queue
+   * - ``gcp_pubsub``
+     - a GCP project
+     - ``lithops[gcp]``
+     - one topic and subscription
+   * - ``azure_queue``
+     - an Azure storage account
+     - ``lithops[azure]``
+     - one queue
+
+**Start with** ``storage``. It works everywhere and needs nothing set up. Move to a message backend when one of these
+actually shows up in your timings: jobs of many thousands of activations, where the storage requests add up; or short
+functions, where waiting for the next poll is a noticeable part of the total.
+
+If you already run one of these services for something else, using it here costs you nothing extra.
+
+
+What to expect
+--------------
+
+**Your results are never at risk.** Whichever backend is configured, the final status of every call is also written to
+the object storage. If a message is lost, Lithops notices and reads the status back from there, so the job still
+finishes normally. A message backend is a faster path for the same information, not a different source of truth.
+
+**Resources are cleaned up on exit.** A message backend creates one queue, topic or list per executor, named after the
+executor id. It is created before the first function is invoked and deleted when the executor shuts down — when the
+``with`` block ends, or on interpreter exit. A process killed hard enough to skip that leaves the resource behind, and
+it has to be removed by hand.
+
+**Nested executors work.** A function may create a ``FunctionExecutor`` of its own. Its call statuses reach every
+executor up the chain, so a client waiting on the outer job still sees the progress of the inner one.
+
+**A worker retries.** If publishing a status fails, the worker tries again a few times, backing off, before giving up
+and logging an error. The status is still in the object storage either way.
+
+
+Configuration
+-------------
+
+Select the backend in the ``lithops`` section of your config file:
 
 .. code:: yaml
 
+    lithops:
+        monitoring: storage          # storage | rabbitmq | redis | aws_sqs | gcp_pubsub | azure_queue
+
+or per executor:
+
+.. code:: python
+
+    fexec = lithops.FunctionExecutor(monitoring='rabbitmq')
+
+``monitoring_interval`` sets how often the client polls, in seconds. It is used **only by the** ``storage`` **backend**;
+message backends are event-driven and ignore it.
+
+.. code:: yaml
+
+    lithops:
+        monitoring_interval: 2
+
+.. note::
+    The default is ``2``, except with the localhost storage backend, where it is ``0.1`` because polling a local
+    directory is cheap.
+
+Each backend then reads its own section. Where a matching cloud provider section already exists, it is used as the
+default, so you rarely have to repeat credentials.
+
+
+RabbitMQ
+~~~~~~~~
+
+.. code:: yaml
+
+    lithops:
+        monitoring: rabbitmq
+
     rabbitmq:
-        amqp_url: <AMQP_URL>  # amqp://
+        amqp_url: <AMQP_URL>  # amqp://<USER>:<PASSWORD>@<HOST>:<PORT>/<VHOST>
+
+``amqp_url`` is mandatory. The broker has to be reachable from your functions as well as from the client — for
+instance a cloud server with a public IP and the AMQP port (5672) open.
+
+The same section is used by the Kubernetes backend with ``rabbitmq_executor: True`` and by the Singularity backend.
 
-In addition, activate the monitoring service by setting ``monitoring: rabbitmq`` in the configuration (Lithops section):
+
+Redis
+~~~~~
 
 .. code:: yaml
 
     lithops:
-       monitoring: rabbitmq
+        monitoring: redis
 
+    redis:
+        host: <REDIS_HOST>
+        #port: 6379
+        #username: <USERNAME>
+        #password: <PASSWORD>
 
-.. code:: python
+``host`` is mandatory. This is the same section as the Redis storage backend, so a deployment that already stores data
+in Redis needs no new keys.
+
+
+AWS SQS
+~~~~~~~
+
+.. code:: yaml
+
+    lithops:
+        monitoring: aws_sqs
+
+    aws:
+        region: <REGION>
+        #access_key_id: <ACCESS_KEY_ID>
+        #secret_access_key: <SECRET_ACCESS_KEY>
+
+``region`` is mandatory. Credentials may instead come from the environment or an instance role. Add an ``aws_sqs``
+section only if you need to override what is in ``aws``.
+
+
+GCP Pub/Sub
+~~~~~~~~~~~
+
+.. code:: yaml
+
+    lithops:
+        monitoring: gcp_pubsub
+
+    gcp:
+        project_name: <GCP_PROJECT_ID>
+        #credentials_path: <ABSOLUTE_PATH_TO_SERVICE_ACCOUNT_JSON>
+
+``project_name`` is mandatory, but it is read from the service account JSON when a credentials file is given.
+``credentials_path`` falls back to ``GOOGLE_APPLICATION_CREDENTIALS``. Add a ``gcp_pubsub`` section only if you need to
+override what is in ``gcp``.
+
+
+Azure Queue Storage
+~~~~~~~~~~~~~~~~~~~
+
+.. code:: yaml
+
+    lithops:
+        monitoring: azure_queue
+
+    azure_storage:
+        storage_account_name: <STORAGE_ACCOUNT>
+        storage_account_key: <STORAGE_ACCOUNT_KEY>
 
-    fexec = lithops.FunctionExecutor(monitoring='rabbitmq')
+Both keys are mandatory. Add an ``azure_queue`` section only if you need to override what is in ``azure_storage``.
+Azure only accepts lowercase queue names, so Lithops adjusts the name it derives from the executor id.

examples/failed_futures.py

@@ -1,9 +1,10 @@
 """
-Simple Lithops example using the map method.
-In this example the map() method will launch one
-map function for each entry in 'iterdata'. Finally
-it will print the results for each invocation with
-fexec.get_result()
+Simple Lithops example showing what happens when a function fails.
+
+Some of the functions below raise an exception. Passing
+throw_except=False to get_result() makes Lithops hand back what it has
+instead of re-raising, so the successful calls can be told apart from
+the failed ones.
 """
 import lithops
 import time

examples/ibmcos_arg.py

@@ -1,6 +1,6 @@
 """
 Simple Lithops example using the 'ibm_cos' parameter, which is
-a ready-to-use ibm_boto3.CLient() instance.
+a ready-to-use ibm_boto3.Client() instance.
 """
 import lithops
 

examples/joblib_backend.py

@@ -1,3 +1,10 @@
+"""
+Simple Lithops example using the joblib backend.
+
+Registering Lithops with joblib makes Parallel() run its jobs as cloud
+functions, so code already written against joblib needs no changes
+beyond the parallel_backend() line.
+"""
 import joblib
 from joblib import Parallel, delayed
 from lithops.util.joblib import register_lithops