Add Redis service layer and Elasticsearch log indexing/search; update config and monitoring APIs#77
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
…lement-elasticsearch-for-logs-and-analytics
fuzziecoder
merged commit a467c0d
into
codex/fix-remaining-issues-and-raise-pr
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| } | ||
|
|
||
| response = await self._client.search(index=settings.ELASTICSEARCH_LOGS_INDEX, body=body) | ||
| return [hit.get("_source", {}) for hit in response.get("hits", {}).get("hits", [])] |
There was a problem hiding this comment.
🔴 response.get() fails on ObjectApiResponse in elasticsearch-py 8.x
In pipeline/backend/core/elasticsearch_client.py:143, the code calls response.get("hits", {}).get("hits", []) on the return value of self._client.search(). In elasticsearch-py 8.15.1, search() returns an ObjectApiResponse, which does not implement a .get() method. Its __getattr__ tries self.body["get"] — looking up "get" as a key in the response dict (not as a method) — which raises KeyError, converted to AttributeError.
Root Cause and Impact
ObjectApiResponse provides __getitem__ (delegating to self.body[item]) and __getattr__ (also delegating to self.body[item]), but it does not inherit from dict or collections.abc.Mapping, so the dict .get() method is not available.
When response.get("hits", {}) is called:
- Python looks up
geton theObjectApiResponseclass hierarchy — not found - Falls back to
__getattr__("get") __getattr__doesself.body["get"]where body is{"hits": {...}, "took": ..., ...}"get"is not a key in the response body →KeyError→ re-raised asAttributeError
The correct approach is response["hits"]["hits"] or response.body.get("hits", {}).get("hits", []).
Impact: Every call to search_logs() — including the /monitoring/logs/search API endpoint — will crash with an AttributeError, making log search completely non-functional.
| return [hit.get("_source", {}) for hit in response.get("hits", {}).get("hits", [])] | |
| return [hit.get("_source", {}) for hit in response["hits"]["hits"]] |
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Motivation
Description
pipeline/backend/config.py, addelasticsearchtopipeline/backend/requirements.txt, and register anelasticsearchservice inpipeline/docker-compose.yml.pipeline/backend/core/elasticsearch_client.pythat initializes the client, ensures an execution-log index, exposesindex_execution_logandsearch_logs, and integrates with app lifecycle and health checks.pipeline/backend/core/redis_services.pywithset_session,get_session,check_rate_limit,enqueue_job, anddequeue_job, and expose the Redis client via aclientproperty onRedisStateManagerfor specialized operations.pipeline/backend/api/routes/monitoring.pywith endpoints for log search, session cache set/get, rate-limit checks, and background job enqueue/dequeue, and add best-effort async Elasticsearch indexing of logs inpipeline/backend/db/crud.py.Testing
python -m compileall pipeline/backendto validate code compiles successfully, which completed without errors.pytest -q pipeline/backend/tests/test_monitoring_engine.py, which passed (1 passed).Codex Task