feat: add elasticsearch cloud connector

Dockerfile

@@ -1,4 +1,4 @@
-FROM ghcr.io/axoflow/axoflow-otel-collector/axoflow-otel-collector:0.150.0-axoflow.2 AS axo-otelcol
+FROM ghcr.io/axoflow/axoflow-otel-collector/axoflow-otel-collector:0.152.0-axoflow.2 AS axo-otelcol
 
 FROM alpine:3.21 AS base
 

README.md

@@ -9,6 +9,7 @@ You can find guides per connector:
 - [Azure connector](./connectors/azure/README.md#quickstart)
 - [AWS connector](./connectors/aws/README.md#quickstart)
 - [Kafka connector](./connectors/kafka/README.md#quickstart)
+- [Elasticsearch connector](./connectors/elasticsearch/README.md#quickstart)
 
 ## Environment Variables
 
@@ -107,6 +108,39 @@ You can find guides per connector:
 | `CROWDSTRIKE_TLS_MAX_VERSION` | No | - | Maximum TLS version to use. |
 | `CROWDSTRIKE_TLS_INCLUDE_SYSTEM_CA_CERTS_POOL` | No | `false` | Include system CA certs along with provided CA. |
 
+### Elasticsearch Provider
+
+| Variable | Required | Default | Description |
+|----------|----------|---------|-------------|
+| `ELASTICSEARCH_ENDPOINT` | No | `http://localhost:9200` | Elasticsearch base URL. |
+| `ELASTICSEARCH_API_KEY` | No* | - | Elasticsearch API key (base64 `id:api_key`). Use this **or** username/password. |
+| `ELASTICSEARCH_USERNAME` | No* | - | Username for HTTP basic auth. Must be set together with the password. |
+| `ELASTICSEARCH_PASSWORD` | No* | - | Password for HTTP basic auth. Must be set together with the username. |
+| `ELASTICSEARCH_INDEX` | No | `logs-*` | Index or data-stream pattern to read. |
+| `ELASTICSEARCH_TIMESTAMP_FIELD` | No | `@timestamp` | Document field used as the time cursor. |
+| `ELASTICSEARCH_PAGE_SIZE` | No | `1000` | Documents per `_search` request. |
+| `ELASTICSEARCH_BATCH_LIMIT` | No | `0` | Max documents fetched per poll cycle (`0` = no limit). |
+| `ELASTICSEARCH_POLL_INTERVAL` | No | `30s` | How often a new search cycle starts. |
+| `ELASTICSEARCH_INITIAL_DELAY` | No | `1s` | Delay before the first poll after startup. |
+| `ELASTICSEARCH_START_AT` | No | `end` | Where to begin on a fresh start: `beginning` or `end`. |
+| `ELASTICSEARCH_INITIAL_LOOKBACK` | No | `1h` | When `start_at=end`, how far back from now to begin. |
+
+#### TLS Settings
+
+| Variable | Required | Default | Description |
+|----------|----------|---------|-------------|
+| `ELASTICSEARCH_TLS_INSECURE` | No | `false` | Disable TLS security (insecure). |
+| `ELASTICSEARCH_TLS_INSECURE_SKIP_VERIFY` | No | `false` | Skip TLS certificate verification. |
+| `ELASTICSEARCH_TLS_CA_FILE` | No | - | Path to a CA certificate file. |
+| `ELASTICSEARCH_TLS_CA_PEM` | No | - | PEM-encoded CA certificate. |
+| `ELASTICSEARCH_TLS_CERT_FILE` | No | - | Path to a client certificate file. |
+| `ELASTICSEARCH_TLS_CERT_PEM` | No | - | PEM-encoded client certificate. |
+| `ELASTICSEARCH_TLS_KEY_FILE` | No | - | Path to a client private key file. |
+| `ELASTICSEARCH_TLS_KEY_PEM` | No | - | PEM-encoded client private key. |
+| `ELASTICSEARCH_TLS_MIN_VERSION` | No | `1.2` | Minimum TLS version to use. |
+| `ELASTICSEARCH_TLS_MAX_VERSION` | No | - | Maximum TLS version to use. |
+| `ELASTICSEARCH_TLS_INCLUDE_SYSTEM_CA_CERTS_POOL` | No | `false` | Include system CA certs along with provided CA. |
+
 
 ## Usage
 

connectors/elasticsearch/README.md

@@ -0,0 +1,94 @@
+
+# Elasticsearch Logs Receiver
+
+This directory contains the Axoflow Elasticsearch connector, which collects logs from an
+Elasticsearch cluster by polling the `_search` API and forwards them to Axorouter.
+
+It paginates with `search_after` over a stable sort and persists a per-index cursor to disk, so it
+resumes where it left off across restarts.
+
+## Quickstart
+
+Make sure the required environment variables are set before running the connector.
+
+### Authentication with an API key
+
+```bash
+UUID_FULL=$(uuidgen 2>/dev/null || cat /proc/sys/kernel/random/uuid 2>/dev/null || python3 -c "import uuid; print(uuid.uuid4())")
+AXOCLOUDCONNECTOR_DEVICE_ID=$(echo "$UUID_FULL" | cut -d'-' -f1)
+
+docker run \
+        --rm \
+        -v "${STORAGE_DIRECTORY}":"${STORAGE_DIRECTORY}" \
+        -e ELASTICSEARCH_ENDPOINT="${ELASTICSEARCH_ENDPOINT}" \
+        -e ELASTICSEARCH_API_KEY="${ELASTICSEARCH_API_KEY}" \
+        -e ELASTICSEARCH_INDEX="${ELASTICSEARCH_INDEX}" \
+        -e AXOROUTER_ENDPOINT="${AXOROUTER_ENDPOINT}" \
+        -e STORAGE_DIRECTORY="${STORAGE_DIRECTORY}" \
+        -e AXOCLOUDCONNECTOR_DEVICE_ID="${AXOCLOUDCONNECTOR_DEVICE_ID}" \
+        ghcr.io/axoflow/axocloudconnectors:latest
+```
+
+### Authentication with username / password
+
+```bash
+UUID_FULL=$(uuidgen 2>/dev/null || cat /proc/sys/kernel/random/uuid 2>/dev/null || python3 -c "import uuid; print(uuid.uuid4())")
+AXOCLOUDCONNECTOR_DEVICE_ID=$(echo "$UUID_FULL" | cut -d'-' -f1)
+
+docker run \
+        --rm \
+        -v "${STORAGE_DIRECTORY}":"${STORAGE_DIRECTORY}" \
+        -e ELASTICSEARCH_ENDPOINT="${ELASTICSEARCH_ENDPOINT}" \
+        -e ELASTICSEARCH_USERNAME="${ELASTICSEARCH_USERNAME}" \
+        -e ELASTICSEARCH_PASSWORD="${ELASTICSEARCH_PASSWORD}" \
+        -e ELASTICSEARCH_INDEX="${ELASTICSEARCH_INDEX}" \
+        -e AXOROUTER_ENDPOINT="${AXOROUTER_ENDPOINT}" \
+        -e STORAGE_DIRECTORY="${STORAGE_DIRECTORY}" \
+        -e AXOCLOUDCONNECTOR_DEVICE_ID="${AXOCLOUDCONNECTOR_DEVICE_ID}" \
+        ghcr.io/axoflow/axocloudconnectors:latest
+```
+
+## Deploy with Helm-chart
+
+```bash
+make minikube-cluster
+make docker-build
+make minikube-load-image
+
+kubectl create namespace cloudconnectors
+kubectl create secret generic elasticsearch \
+  --from-literal=endpoint="<YOUR-ELASTICSEARCH-ENDPOINT>" \
+  --from-literal=api-key="<YOUR-ELASTICSEARCH-API-KEY>" \
+  --namespace cloudconnectors \
+  --dry-run=client -o yaml | kubectl apply -f -
+
+UUID_FULL=$(uuidgen 2>/dev/null || cat /proc/sys/kernel/random/uuid 2>/dev/null || python3 -c "import uuid; print(uuid.uuid4())")
+AXOCLOUDCONNECTOR_DEVICE_ID=$(echo "$UUID_FULL" | cut -d'-' -f1)
+
+helm upgrade --install --wait --namespace cloudconnectors cloudconnectors ./charts/cloudconnectors \
+  --set image.repository="axocloudconnectors" \
+  --set image.tag="dev" \
+  --set 'env[0].name=AXOROUTER_ENDPOINT' \
+  --set 'env[0].value=axorouter.axoflow-local.svc.cluster.local:4317' \
+  --set 'env[1].name=AXOCLOUDCONNECTOR_DEVICE_ID' \
+  --set "env[1].value=${AXOCLOUDCONNECTOR_DEVICE_ID}" \
+  --set 'env[2].name=ELASTICSEARCH_ENDPOINT' \
+  --set 'env[2].valueFrom.secretKeyRef.name=elasticsearch' \
+  --set 'env[2].valueFrom.secretKeyRef.key=endpoint' \
+  --set 'env[3].name=ELASTICSEARCH_API_KEY' \
+  --set 'env[3].valueFrom.secretKeyRef.name=elasticsearch' \
+  --set 'env[3].valueFrom.secretKeyRef.key=api-key' \
+  --set 'env[4].name=ELASTICSEARCH_INDEX' \
+  --set 'env[4].value=logs-*'
+```
+
+## Notes
+
+- `search_after` requires a stable, deterministic sort. The connector defaults to
+  `[{"@timestamp": asc}, {"_seq_no": asc}]`; the last entry must be unique per document. If your
+  documents use a different time field, set `ELASTICSEARCH_TIMESTAMP_FIELD` and adjust the `sort` in
+  `config.yaml` accordingly.
+- With `ELASTICSEARCH_START_AT=end` (the default) and `ELASTICSEARCH_INITIAL_LOOKBACK=0`, documents
+  whose timestamp precedes startup but that are indexed afterwards (indexing lag or clock skew) can be
+  missed. The default lookback of `1h` covers that window; increase it if needed, or use
+  `ELASTICSEARCH_START_AT=beginning` to read all history.

connectors/elasticsearch/config.yaml

@@ -0,0 +1,99 @@
+exporters:
+  otlp_grpc/axorouter:
+    endpoint: ${env:AXOROUTER_ENDPOINT}
+    retry_on_failure:
+      enabled: true
+      max_elapsed_time: 0
+    sending_queue:
+      enabled: true
+      storage: file_storage
+    tls:
+      insecure: ${env:AXOROUTER_TLS_INSECURE:-false}
+      ca_file: ${env:AXOROUTER_TLS_CA_FILE}
+      ca_pem: ${env:AXOROUTER_TLS_CA_PEM}
+      cert_file: ${env:AXOROUTER_TLS_CERT_FILE}
+      cert_pem: ${env:AXOROUTER_TLS_CERT_PEM}
+      key_file: ${env:AXOROUTER_TLS_KEY_FILE}
+      key_pem: ${env:AXOROUTER_TLS_KEY_PEM}
+      min_version: ${env:AXOROUTER_TLS_MIN_VERSION:-1.2}
+      max_version: ${env:AXOROUTER_TLS_MAX_VERSION}
+      include_system_ca_certs_pool: ${env:AXOROUTER_TLS_INCLUDE_SYSTEM_CA_CERTS_POOL:-false}
+      insecure_skip_verify: ${env:AXOROUTER_TLS_INSECURE_SKIP_VERIFY:-false}
+
+processors:
+  resource/axoflow_device_id:
+    attributes:
+      - key: "com.axoflow.device_id"
+        action: insert
+        value: "${env:AXOCLOUDCONNECTOR_DEVICE_ID}"
+
+  resourcedetection/system:
+    detectors: ["system", "env"]
+    system:
+      hostname_sources: ["dns", "os", "cname", "lookup"]
+      resource_attributes:
+        host.name:
+          enabled: true
+        host.ip:
+          enabled: true
+        host.id:
+          enabled: true
+
+  resource/axoflow: # Provider specific!
+    attributes:
+      - key: "com.axoflow.vendor"
+        action: insert
+        value: "elastic"
+      - key: "com.axoflow.product"
+        action: insert
+        value: "elasticsearch"
+
+receivers: # Provider specific!
+  elasticsearchlogs:
+    endpoint: ${env:ELASTICSEARCH_ENDPOINT:-http://localhost:9200}
+    api_key: ${env:ELASTICSEARCH_API_KEY}
+    username: ${env:ELASTICSEARCH_USERNAME}
+    password: ${env:ELASTICSEARCH_PASSWORD}
+    indices:
+      - ${env:ELASTICSEARCH_INDEX:-logs-*}
+    timestamp_field: ${env:ELASTICSEARCH_TIMESTAMP_FIELD:-@timestamp}
+    # Stable sort for search_after pagination. The last entry must be unique per document; _seq_no is
+    # monotonic and sortable without requiring fielddata (unlike _id). Adjust if your schema differs.
+    sort:
+      - "@timestamp": asc
+      - "_seq_no": asc
+    page_size: ${env:ELASTICSEARCH_PAGE_SIZE:-1000}
+    batch_limit: ${env:ELASTICSEARCH_BATCH_LIMIT:-0}
+    poll_interval: ${env:ELASTICSEARCH_POLL_INTERVAL:-30s}
+    initial_delay: ${env:ELASTICSEARCH_INITIAL_DELAY:-1s}
+    start_at: ${env:ELASTICSEARCH_START_AT:-end}
+    initial_lookback: ${env:ELASTICSEARCH_INITIAL_LOOKBACK:-1h}
+    storage: file_storage
+    tls:
+      insecure: ${env:ELASTICSEARCH_TLS_INSECURE:-false}
+      insecure_skip_verify: ${env:ELASTICSEARCH_TLS_INSECURE_SKIP_VERIFY:-false}
+      ca_file: ${env:ELASTICSEARCH_TLS_CA_FILE}
+      ca_pem: ${env:ELASTICSEARCH_TLS_CA_PEM}
+      cert_file: ${env:ELASTICSEARCH_TLS_CERT_FILE}
+      cert_pem: ${env:ELASTICSEARCH_TLS_CERT_PEM}
+      key_file: ${env:ELASTICSEARCH_TLS_KEY_FILE}
+      key_pem: ${env:ELASTICSEARCH_TLS_KEY_PEM}
+      min_version: ${env:ELASTICSEARCH_TLS_MIN_VERSION:-1.2}
+      max_version: ${env:ELASTICSEARCH_TLS_MAX_VERSION}
+      include_system_ca_certs_pool: ${env:ELASTICSEARCH_TLS_INCLUDE_SYSTEM_CA_CERTS_POOL:-false}
+
+extensions:
+  health_check:
+    endpoint: ${env:POD_IP}:13133
+  file_storage:
+    directory: ${env:STORAGE_DIRECTORY}
+    create_directory: true
+
+service:
+  extensions: [health_check, file_storage]
+  pipelines:
+    logs:
+      receivers: [elasticsearchlogs]
+      processors:
+        [resource/axoflow_device_id, resourcedetection/system, resource/axoflow]
+      exporters: [otlp_grpc/axorouter]

entrypoint.sh

@@ -12,6 +12,7 @@ detect_provider() {
     env | grep -q "^AWS_" && provider="$provider aws" && count=$((count + 1))
     env | grep -q "^KAFKA_" && provider="$provider kafka" && count=$((count + 1))
     env | grep -q "^CROWDSTRIKE_" && provider="$provider crowdstrike" && count=$((count + 1))
+    env | grep -q "^ELASTICSEARCH_" && provider="$provider elasticsearch" && count=$((count + 1))
     # env | grep -q "^GCP_" && provider="$provider gcp" && count=$((count + 1))
 
     if [ "$count" -gt 1 ]; then
@@ -33,5 +34,6 @@ echo "   - Azure (AZURE_*)"
 echo "   - AWS (AWS_*)"
 echo "   - Kafka (KAFKA_*)"
 echo "   - Crowdstrike (CROWDSTRIKE_*)"
+echo "   - Elasticsearch (ELASTICSEARCH_*)"
 # echo "   - GCP (GCP_*)"
 exit 1