feat(logging): add fluent-bit log shipping (#3431)

* feat(logging): add fluent-bit log shipping

Implements #3430.

This PR is partially implemented using Cursor.

* Fix pyright errors by using try/except/else pattern for optional imports

* refactor(fluentbit): cleanup protocol lambdas and address codex comments

* feat(fluentbit): validate next_token format and raise ServerClientError for malformed tokens

* chore(fluentbit): address quick comments

* feat(fluentbit): add tag prefix support to HTTPFluentBitWriter

Author: DragonStuff

docs/docs/guides/server-deployment.md

@@ -159,7 +159,7 @@ $ DSTACK_DATABASE_URL=postgresql+asyncpg://user:password@db-host:5432/dstack dst
 
 By default, `dstack` stores workload logs locally in `~/.dstack/server/projects/<project_name>/logs`.
 For multi-replica server deployments, it's required to store logs externally.
-`dstack` supports storing logs using AWS CloudWatch or GCP Logging.
+`dstack` supports storing logs using AWS CloudWatch, GCP Logging, or Fluent-bit with Elasticsearch / Opensearch.
 
 ### AWS CloudWatch
 
@@ -222,6 +222,78 @@ To store logs using GCP Logging, set the `DSTACK_SERVER_GCP_LOGGING_PROJECT` env
 
     </div>
 
+### Fluent-bit
+
+To store logs using Fluent-bit, set the `DSTACK_SERVER_FLUENTBIT_HOST` environment variable.
+Fluent-bit supports two modes depending on how you want to access logs.
+
+=== "Full mode"
+
+    Logs are shipped to Fluent-bit and can be read back through the dstack UI and CLI via Elasticsearch or OpenSearch.
+    Use this mode when you want a complete integration with log viewing in dstack:
+
+    ```shell
+    $ DSTACK_SERVER_FLUENTBIT_HOST=fluentbit.example.com \
+      DSTACK_SERVER_ELASTICSEARCH_HOST=https://elasticsearch.example.com:9200 \
+      dstack server
+    ```
+
+=== "Ship-only mode"
+
+    Logs are forwarded to Fluent-bit but cannot be read through `dstack`. 
+    The dstack UI/CLI will show empty logs. Use this mode when:
+
+    - You have an existing logging infrastructure (Kibana, Grafana, Datadog, etc.)
+    - You only need to forward logs without reading them back through dstack
+    - You want to reduce operational complexity by not running Elasticsearch/OpenSearch
+
+    ```shell
+    $ DSTACK_SERVER_FLUENTBIT_HOST=fluentbit.example.com \
+      dstack server
+    ```
+
+??? info "Additional configuration"
+    The following optional environment variables can be used to customize the Fluent-bit integration:
+
+    **Fluent-bit settings:**
+
+    - `DSTACK_SERVER_FLUENTBIT_PORT` – The Fluent-bit port. Defaults to `24224`.
+    - `DSTACK_SERVER_FLUENTBIT_PROTOCOL` – The protocol to use: `forward` or `http`. Defaults to `forward`.
+    - `DSTACK_SERVER_FLUENTBIT_TAG_PREFIX` – The tag prefix for logs. Defaults to `dstack`.
+
+    **Elasticsearch/OpenSearch settings (for full mode only):**
+
+    - `DSTACK_SERVER_ELASTICSEARCH_HOST` – The Elasticsearch/OpenSearch host for reading logs. If not set, runs in ship-only mode.
+    - `DSTACK_SERVER_ELASTICSEARCH_INDEX` – The Elasticsearch/OpenSearch index pattern. Defaults to `dstack-logs`.
+    - `DSTACK_SERVER_ELASTICSEARCH_API_KEY` – The Elasticsearch/OpenSearch API key for authentication.
+
+??? info "Fluent-bit configuration"
+    Configure Fluent-bit to receive logs and forward them to Elasticsearch or OpenSearch. Example configuration:
+
+    ```ini
+    [INPUT]
+        Name        forward
+        Listen      0.0.0.0
+        Port        24224
+
+    [OUTPUT]
+        Name            es
+        Match           dstack.*
+        Host            elasticsearch.example.com
+        Port            9200
+        Index           dstack-logs
+        Suppress_Type_Name On
+    ```
+
+??? info "Required dependencies"
+    To use Fluent-bit log storage, install the `fluentbit` extras:
+
+    ```shell
+    $ pip install "dstack[all]" -U
+    # or
+    $ pip install "dstack[fluentbit]" -U
+    ```
+
 ## File storage
 
 When using  [files](../concepts/dev-environments.md#files) or [repos](../concepts/dev-environments.md#repos), `dstack` uploads local files and diffs to the server so that you can have access to them within runs. By default, the files are stored in the DB and each upload is limited to 2MB. You can configure an object storage to be used for uploads and increase the default limit by setting the `DSTACK_SERVER_CODE_UPLOAD_LIMIT` environment variable
@@ -426,8 +498,10 @@ If a deployment is stuck due to a deadlock when applying DB migrations, try scal
 
 ??? info "Can I run multiple replicas of dstack server?"
 
-    Yes, you can if you configure `dstack` to use [PostgreSQL](#postgresql) and [AWS CloudWatch](#aws-cloudwatch).
+    Yes, you can if you configure `dstack` to use [PostgreSQL](#postgresql) and an external log storage
+    such as [AWS CloudWatch](#aws-cloudwatch), [GCP Logging](#gcp-logging), or [Fluent-bit](#fluent-bit).
 
 ??? info "Does dstack server support blue-green or rolling deployments?"
 
-    Yes, it does if you configure `dstack` to use [PostgreSQL](#postgresql) and [AWS CloudWatch](#aws-cloudwatch).
+    Yes, it does if you configure `dstack` to use [PostgreSQL](#postgresql) and an external log storage
+    such as [AWS CloudWatch](#aws-cloudwatch), [GCP Logging](#gcp-logging), or [Fluent-bit](#fluent-bit).

docs/docs/reference/environment-variables.md

@@ -113,6 +113,13 @@ For more details on the options below, refer to the [server deployment](../guide
 - `DSTACK_SERVER_CLOUDWATCH_LOG_GROUP`{ #DSTACK_SERVER_CLOUDWATCH_LOG_GROUP } – The CloudWatch Logs group for storing workloads logs. If not set, the default file-based log storage is used.
 - `DSTACK_SERVER_CLOUDWATCH_LOG_REGION`{ #DSTACK_SERVER_CLOUDWATCH_LOG_REGION } – The CloudWatch Logs region. Defaults to `None`.
 - `DSTACK_SERVER_GCP_LOGGING_PROJECT`{ #DSTACK_SERVER_GCP_LOGGING_PROJECT } – The GCP Logging project for storing workloads logs. If not set, the default file-based log storage is used.
+- `DSTACK_SERVER_FLUENTBIT_HOST`{ #DSTACK_SERVER_FLUENTBIT_HOST } – The Fluent-bit host for log forwarding. If set, enables Fluent-bit log storage.
+- `DSTACK_SERVER_FLUENTBIT_PORT`{ #DSTACK_SERVER_FLUENTBIT_PORT } – The Fluent-bit port. Defaults to `24224`.
+- `DSTACK_SERVER_FLUENTBIT_PROTOCOL`{ #DSTACK_SERVER_FLUENTBIT_PROTOCOL } – The protocol to use: `forward` or `http`. Defaults to `forward`.
+- `DSTACK_SERVER_FLUENTBIT_TAG_PREFIX`{ #DSTACK_SERVER_FLUENTBIT_TAG_PREFIX } – The tag prefix for logs. Defaults to `dstack`.
+- `DSTACK_SERVER_ELASTICSEARCH_HOST`{ #DSTACK_SERVER_ELASTICSEARCH_HOST } – The Elasticsearch/OpenSearch host for reading logs back through dstack. Optional; if not set, Fluent-bit runs in ship-only mode (logs are forwarded but not readable through dstack UI/CLI).
+- `DSTACK_SERVER_ELASTICSEARCH_INDEX`{ #DSTACK_SERVER_ELASTICSEARCH_INDEX } – The Elasticsearch/OpenSearch index pattern. Defaults to `dstack-logs`.
+- `DSTACK_SERVER_ELASTICSEARCH_API_KEY`{ #DSTACK_SERVER_ELASTICSEARCH_API_KEY } – The Elasticsearch/OpenSearch API key for authentication.
 - `DSTACK_ENABLE_PROMETHEUS_METRICS`{ #DSTACK_ENABLE_PROMETHEUS_METRICS } — Enables Prometheus metrics collection and export.
 - `DSTACK_DEFAULT_SERVICE_CLIENT_MAX_BODY_SIZE`{ #DSTACK_DEFAULT_SERVICE_CLIENT_MAX_BODY_SIZE } – Request body size limit for services running with a gateway, in bytes. Defaults to 64 MiB.
 - `DSTACK_SERVICE_CLIENT_TIMEOUT`{ #DSTACK_SERVICE_CLIENT_TIMEOUT } – Timeout in seconds for HTTP requests sent from the in-server proxy and gateways to service replicas. Defaults to 60.

pyproject.toml

@@ -215,6 +215,11 @@ nebius = [
     "nebius>=0.3.4,<0.4; python_version >= '3.10'",
     "dstack[server]",
 ]
+fluentbit = [
+    "fluent-logger>=0.10.0",
+    "elasticsearch>=8.0.0",
+    "dstack[server]",
+]
 all = [
-    "dstack[gateway,server,aws,azure,gcp,verda,kubernetes,lambda,nebius,oci]",
+    "dstack[gateway,server,aws,azure,gcp,verda,kubernetes,lambda,nebius,oci,fluentbit]",
 ]

src/dstack/_internal/server/services/logs/__init__.py

@@ -8,6 +8,7 @@
 from dstack._internal.server.schemas.logs import PollLogsRequest
 from dstack._internal.server.schemas.runner import LogEvent as RunnerLogEvent
 from dstack._internal.server.services.logs import aws as aws_logs
+from dstack._internal.server.services.logs import fluentbit as fluentbit_logs
 from dstack._internal.server.services.logs import gcp as gcp_logs
 from dstack._internal.server.services.logs.base import (
     LogStorage,
@@ -57,6 +58,29 @@ def get_log_storage() -> LogStorage:
                 logger.debug("Using GCP Logs storage")
         else:
             logger.error("Cannot use GCP Logs storage: GCP deps are not installed")
+    elif settings.SERVER_FLUENTBIT_HOST:
+        if fluentbit_logs.FLUENTBIT_AVAILABLE:
+            try:
+                _log_storage = fluentbit_logs.FluentBitLogStorage(
+                    host=settings.SERVER_FLUENTBIT_HOST,
+                    port=settings.SERVER_FLUENTBIT_PORT,
+                    protocol=settings.SERVER_FLUENTBIT_PROTOCOL,
+                    tag_prefix=settings.SERVER_FLUENTBIT_TAG_PREFIX,
+                    es_host=settings.SERVER_ELASTICSEARCH_HOST,
+                    es_index=settings.SERVER_ELASTICSEARCH_INDEX,
+                    es_api_key=settings.SERVER_ELASTICSEARCH_API_KEY,
+                )
+            except LogStorageError as e:
+                logger.error("Failed to initialize Fluent-bit Logs storage: %s", e)
+            except Exception:
+                logger.exception("Got exception when initializing Fluent-bit Logs storage")
+            else:
+                if settings.SERVER_ELASTICSEARCH_HOST:
+                    logger.debug("Using Fluent-bit Logs storage with Elasticsearch/OpenSearch")
+                else:
+                    logger.debug("Using Fluent-bit Logs storage in ship-only mode")
+        else:
+            logger.error("Cannot use Fluent-bit Logs storage: fluent-logger is not installed")
     if _log_storage is None:
         _log_storage = FileLogStorage()
         logger.debug("Using file-based storage")