Add retry mechanism and error handling for Elasticsearch connection failures

[tanmayjoddar]

Currently experiencing issues where the application fails to start when Elasticsearch is not immediately available during container startup. This creates problems in docker-compose environments where services start in parallel.

The ingestion module should have proper retry logic with exponential backoff when connecting to Elasticsearch. Right now if ES isn't ready, the whole service can crash or hang.

Proposed changes:

• Add connection retry decorator for ES operations
• Implement exponential backoff (start at 1s, max 30s)
• Add proper logging for connection attempts
• Make retry attempts configurable via environment variable
• Handle connection pool exhaustion gracefully

This would make the deployment more robust and prevent the startup race conditions we're seeing.

[tanmayjoddar]

I’ve opened a draft PR that addresses this issue with a proposed retry mechanism and backoff strategy. Sharing early so we can iterate on the approach if needed.

[Lorygold]

the ingestion is the main process, don't know if a retry mechanism makes sense, because if Elasticsearch is broken, it will remain broken and BuffaLogs can't start.

Anyway, additional logging is okay.

What do you think about it?

Ps. In any case, the variables should be defined in the ingestion.json config file, not as env variables

[tanmayjoddar]

Thanks for the feedback — that makes sense, especially considering ingestion is the main process.

You're right that if Elasticsearch is actually broken, retries won't help and BuffaLogs should fail. The intent behind the retry was only to handle temporary startup delays (e.g. docker-compose race conditions where ES is still initializing for ~30–60s), not to mask real failures.

To align better with this:

• I can limit retries to a short, bounded startup window (e.g. max ~60s), after which ingestion fails normally.

• Or, if you prefer, I can drop retries entirely and focus only on clearer, more explicit logging for connection failures.

Agreed on configuration placement — I'll move everything from env vars into ingestion.json, consistent with the existing ingestion configs.

Let me know which direction you'd prefer:

Option A: Short, startup-only retry + improved logging

• Limit retries to a strict, bounded startup window (e.g. max ~60s)
• After that, ingestion fails normally
• Add clearer, explicit logging so failures are obvious

Option B: Improved logging only, no retry logic

• Drop retry logic entirely
• Focus only on clearer and more explicit connection failure logs

Happy to adjust the draft PR accordingly.

[Lorygold]

I agree on adding retry/backoff logic for Elasticsearch connections, I'll propose some guidelines.

• Ingestion: To keep the implementation consistent with BuffaLogs’ architecture, I think the retry configuration should live in ingestion.json, since it directly affects ingestion behavior and is already the single source of truth for ingestion targets and logic. There should be the same retry dictionary inside each ingestion source (with the same default values for each key). Let me know here which structure for this dict you propose.

• Retry behavior: Retry only on transient errors (connection errors, timeouts, ES not ready). Fail fast on configuration or non-recoverable errors. Use exponential backoff with an upper bound and optional jitter. Avoid infinite retries. Enforce a maximum backoff time.

• Logging: Log each retry attempt with attempt number and error cause. Log explicitly when max retries are exceeded.

• Unit tests and Doc: Add unit tests covering retry/backoff logic. Document the retry configuration with a minimal ingestion.json example

[tanmayjoddar]

Thanks for the guidelines. I propose the following structure for ingestion.json ---

{
 "active_ingestion": "elasticsearch",
 "elasticsearch": {
   "url": "http://elasticsearch:9200/",
   "username": "foobar",
   "password": "bar",
   "timeout": 90,
   "retry": {
     "enabled": true,
     "max_retries": 10,
     "initial_backoff": 1,
     "max_backoff": 30,
     "max_elapsed_time": 60,
     "jitter": true
   },
   "indexes": "cloud-*,fw-proxy-*",
   "bucket_size": 10000,
   "custom_mapping": {
     "@timestamp": "timestamp",
     "_id": "id",
     "_index": "index",
     "user.name": "username",
     "source.ip": "ip",
     "user_agent.original": "agent",
     "source.as.organization.name": "organization",
     "source.geo.country_name": "country",
     "source.geo.location.lat": "lat",
     "source.geo.location.lon": "lon",
     "source.intelligence_category": "intelligence_category"
   },
   "__custom_fields__": ["url", "username", "password", "timeout", "indexes"]
 },
 "opensearch": {
   "url": "http://opensearch:9200/",
   "username": "foobar",
   "password": "bar",
   "timeout": 90,
   "retry": {
     "enabled": true,
     "max_retries": 10,
     "initial_backoff": 1,
     "max_backoff": 30,
     "max_elapsed_time": 60,
     "jitter": true
   },
   "indexes": "cloud-*,fw-proxy-*",
   "bucket_size": 10000,
   "custom_mapping": {
     "@timestamp": "timestamp",
     "_id": "id",
     "_index": "index",
     "user.name": "username",
     "source.ip": "ip",
     "user_agent.original": "agent",
     "source.as.organization.name": "organization",
     "source.geo.country_name": "country",
     "source.geo.location.lat": "lat",
     "source.geo.location.lon": "lon",
     "source.intelligence_category": "intelligence_category"
   },
   "__custom_fields__": ["url", "username", "password", "timeout", "indexes"]
 },
 "splunk": {
   "host": "splunk",
   "port": 8089,
   "scheme": "http",
   "username": "foobar",
   "password": "bar",
   "timeout": 90,
   "retry": {
     "enabled": true,
     "max_retries": 10,
     "initial_backoff": 1,
     "max_backoff": 30,
     "max_elapsed_time": 60,
     "jitter": true
   },
   "indexes": "main",
   "bucket_size": 10000,
   "custom_mapping": {
     "user.name": "username",
     "@timestamp": "timestamp",
     "source.ip": "ip",
     "source.geo.country_name": "country",
     "source.geo.location.lat": "lat",
     "source.geo.location.lon": "lon",
     "user_agent.original": "agent",
     "source.as.organization.name": "organization",
     "_id": "id",
     "index": "index",
     "source.intelligence_category": "intelligence_category"
   },
   "__custom_fields__": ["host", "port", "scheme", "username", "password", "timeout", "indexes"]
 }
}

Key definitions:

• enabled — Toggle retry on/off per source
• max_retries — Maximum number of retry attempts (10)
• initial_backoff — Starting delay in seconds (1s)
• max_backoff — Cap on delay between retries (30s)
• max_elapsed_time — Total timeout in seconds (60s)
• jitter — Randomize backoff to avoid thundering herd

Implementation notes:

• The retry loop will stop when either max_retries or max_elapsed_time is reached (whichever comes first)
• Retry configuration is a behavioral setting parsed separately from __custom_fields__ (which remains connection/auth/mapping fields)
• This keeps retry configuration centralized in ingestion.json as the single source of truth

However, I should note: my current draft PR hasn't fully implemented this yet — I've been using env vars and haven't moved config to ingestion.json.

Once you approve this structure, I'll refactor the code to match these requirements completely.

Shall I proceed?

[Lorygold]

ok, that's fine.
Remember then that the config values have to be read by the upper-level Ingestion Class, since it's the same structure for each ingestion source.

Please test it locally with the Elastichsearch container (generating data and running the buffalogs detection) and attach screenshots, thank you

[tanmayjoddar]

Got it, thanks.

I’ll refactor retry config parsing into the upper-level Ingestion class, apply it consistently across ingestion sources, and test locally with Elasticsearch.

I’ll attach screenshots once the PR is updated.

[tanmayjoddar]

Context / clarification

While iterating on the retry mechanism, two intermediate commits were auto-linked to this issue during development.

Those commits are not part of the final branch history and were superseded during cleanup.

This PR contains the final, clean implementation with:

• Predicate-based retry logic
• Exception logging as discussed
• Resolved Docker cross-platform startup issue