From cdb92f2554f0adba9286dfc78d6678a97c12c360 Mon Sep 17 00:00:00 2001 From: Szilard Parrag Date: Wed, 24 Jun 2026 13:45:20 +0200 Subject: [PATCH 1/2] feat: add elasticsearch cloud connector Add a connector that reads logs from Elasticsearch via the elasticsearchlogs receiver and forwards them to Axorouter. - connectors/elasticsearch/config.yaml: ELASTICSEARCH_*-driven receiver (API key or basic auth, search_after pagination, per-index cursor persisted to file_storage) wired through the standard axoflow processors to the otlp_grpc/axorouter exporter - entrypoint.sh: auto-detect the provider from ELASTICSEARCH_* env vars - README + connector README: document the provider and its variables Note: requires the elasticsearchlogs receiver to be present in the axoflow-otel-collector image. Signed-off-by: Szilard Parrag --- README.md | 34 ++++++++++ connectors/elasticsearch/README.md | 94 ++++++++++++++++++++++++++ connectors/elasticsearch/config.yaml | 99 ++++++++++++++++++++++++++++ entrypoint.sh | 2 + 4 files changed, 229 insertions(+) create mode 100644 connectors/elasticsearch/README.md create mode 100644 connectors/elasticsearch/config.yaml diff --git a/README.md b/README.md index 9a69d05..dafe397 100644 --- a/README.md +++ b/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 diff --git a/connectors/elasticsearch/README.md b/connectors/elasticsearch/README.md new file mode 100644 index 0000000..57748f7 --- /dev/null +++ b/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="" \ + --from-literal=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. diff --git a/connectors/elasticsearch/config.yaml b/connectors/elasticsearch/config.yaml new file mode 100644 index 0000000..706798d --- /dev/null +++ b/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] diff --git a/entrypoint.sh b/entrypoint.sh index 55ac568..94db2a3 100755 --- a/entrypoint.sh +++ b/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 From db297bdf859ae662eb02dcb0ed09cf38e497d5af Mon Sep 17 00:00:00 2001 From: Szilard Parrag Date: Wed, 24 Jun 2026 13:55:16 +0200 Subject: [PATCH 2/2] chore: upgrade axoflow-otel-collector version to v0.152.0-axoflow.2 This image includes the elasticsearchlogs receiver required by the elasticsearch connector. Signed-off-by: Szilard Parrag --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 677dd0f..0dae265 100644 --- a/Dockerfile +++ b/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