feat: implement snapshot status retrieval functionality in Wazuh Indexer
taylorwalton committed
Jan 28, 2026 at 11:58 UTC
46892048299ed5da2307fa14224dc671b2c8c52f
3 files changed
+159
backend/app/connectors/wazuh_indexer/routes/snapshot_and_restore.py
+53
@@ -1,8 +1,13 @@
1
+from typing import Optional
2
+
3
from fastapi import APIRouter
4
from fastapi import HTTPException
5
+from fastapi import Query
6
from loguru import logger
7
8
from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotRepositoryListResponse
9
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotStatusResponse
10
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import get_snapshot_status
11
from app.connectors.wazuh_indexer.services.snapshot_and_restore import list_snapshot_repositories
12
13
wazuh_indexer_snapshots_router = APIRouter()
@@ -32,3 +37,51 @@ async def get_snapshot_repositories() -> SnapshotRepositoryListResponse:
37
)
38
39
return response
40
+
41
+
42
+@wazuh_indexer_snapshots_router.get(
43
+ "/status",
44
+ response_model=SnapshotStatusResponse,
45
+ summary="Get Snapshot Status",
46
+ description="Retrieve the status of snapshots. Optionally filter by repository and snapshot name.",
47
+)
48
+async def get_snapshots_status(
49
+ repository: Optional[str] = Query(
50
+ None,
51
+ description="Repository name to filter by. If not provided, all repositories are queried.",
52
+ ),
53
+ snapshot: Optional[str] = Query(
54
+ None,
55
+ description="Snapshot name to filter by. Requires repository to be specified.",
56
+ ),
57
+) -> SnapshotStatusResponse:
58
+ """
59
+ Get the status of snapshots in the Wazuh Indexer.
60
+
61
+ Args:
62
+ repository: Optional repository name to filter by.
63
+ snapshot: Optional snapshot name to filter by.
64
+
65
+ Returns:
66
+ SnapshotStatusResponse: Status of the requested snapshots.
67
+ """
68
+ logger.info(
69
+ f"Received request to get snapshot status "
70
+ f"(repository={repository}, snapshot={snapshot})",
71
+ )
72
+
73
+ if snapshot and not repository:
74
+ raise HTTPException(
75
+ status_code=400,
76
+ detail="Repository must be specified when filtering by snapshot name.",
77
+ )
78
+
79
+ response = await get_snapshot_status(repository=repository, snapshot=snapshot)
80
+
81
+ if not response.success:
82
+ raise HTTPException(
83
+ status_code=500,
84
+ detail=response.message,
85
+ )
86
+
87
+ return response
backend/app/connectors/wazuh_indexer/schema/snapshot_and_restore.py
+38
@@ -32,3 +32,41 @@ class SnapshotRepositoryListResponse(BaseModel):
32
)
33
success: bool = Field(..., description="Whether the operation was successful")
34
message: str = Field(..., description="Status message")
35
+
36
+
37
+class SnapshotShardStatus(BaseModel):
38
+ """Status of a single shard in a snapshot."""
39
+ stage: str = Field(..., description="Current stage of the shard snapshot")
40
+ total_files: Optional[int] = Field(None, alias="total_file_count", description="Total number of files")
41
+ total_size_in_bytes: Optional[int] = Field(None, description="Total size in bytes")
42
+ processed_files: Optional[int] = Field(None, alias="processed_file_count", description="Processed files count")
43
+ processed_size_in_bytes: Optional[int] = Field(None, alias="done_size_in_bytes", description="Processed size in bytes")
44
+
45
+
46
+class SnapshotIndexStatus(BaseModel):
47
+ """Status of an index within a snapshot."""
48
+ shards_stats: Dict[str, Any] = Field(default_factory=dict, description="Shard statistics")
49
+ stats: Dict[str, Any] = Field(default_factory=dict, description="Index statistics")
50
+ shards: Dict[str, SnapshotShardStatus] = Field(default_factory=dict, description="Individual shard statuses")
51
+
52
+
53
+class SnapshotStatus(BaseModel):
54
+ """Status of a single snapshot."""
55
+ snapshot: str = Field(..., description="Name of the snapshot")
56
+ repository: str = Field(..., description="Repository containing the snapshot")
57
+ uuid: Optional[str] = Field(None, description="UUID of the snapshot")
58
+ state: str = Field(..., description="Current state of the snapshot")
59
+ include_global_state: Optional[bool] = Field(None, description="Whether global state is included")
60
+ shards_stats: Dict[str, Any] = Field(default_factory=dict, description="Shard statistics")
61
+ stats: Dict[str, Any] = Field(default_factory=dict, description="Snapshot statistics")
62
+ indices: Dict[str, SnapshotIndexStatus] = Field(default_factory=dict, description="Index statuses")
63
+
64
+
65
+class SnapshotStatusResponse(BaseModel):
66
+ """Response model for snapshot status."""
67
+ snapshots: List[SnapshotStatus] = Field(
68
+ default_factory=list,
69
+ description="List of snapshot statuses",
70
+ )
71
+ success: bool = Field(..., description="Whether the operation was successful")
72
+ message: str = Field(..., description="Status message")
backend/app/connectors/wazuh_indexer/services/snapshot_and_restore.py
+68
@@ -1,9 +1,12 @@
1
from typing import List
2
+from typing import Optional
3
4
from loguru import logger
5
6
from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotRepository
7
from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotRepositoryListResponse
8
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotStatus
9
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotStatusResponse
10
from app.connectors.wazuh_indexer.utils.universal import create_wazuh_indexer_client
11
12
@@ -47,3 +50,68 @@ async def list_snapshot_repositories() -> SnapshotRepositoryListResponse:
50
success=False,
51
message=f"Failed to list snapshot repositories: {str(e)}",
52
)
53
+
54
+
55
+async def get_snapshot_status(
56
+ repository: Optional[str] = None,
57
+ snapshot: Optional[str] = None,
58
+) -> SnapshotStatusResponse:
59
+ """
60
+ Get the status of snapshots in the Wazuh Indexer (OpenSearch).
61
+
62
+ Args:
63
+ repository: Optional repository name to filter by.
64
+ snapshot: Optional snapshot name to filter by (requires repository).
65
+
66
+ Returns:
67
+ SnapshotStatusResponse: Response containing snapshot statuses.
68
+ """
69
+ logger.info(
70
+ f"Fetching snapshot status from Wazuh Indexer "
71
+ f"(repository={repository}, snapshot={snapshot})",
72
+ )
73
+
74
+ try:
75
+ es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
76
+
77
+ # Build the request parameters
78
+ repo_param = repository if repository else "_all"
79
+ snapshot_param = snapshot if snapshot else "_all"
80
+
81
+ # Get snapshot status using the _snapshot/_status API
82
+ response = es_client.snapshot.status(
83
+ repository=repo_param,
84
+ snapshot=snapshot_param,
85
+ ignore_unavailable=True,
86
+ )
87
+
88
+ snapshots: List[SnapshotStatus] = []
89
+
90
+ for snap_data in response.get("snapshots", []):
91
+ snapshot_status = SnapshotStatus(
92
+ snapshot=snap_data.get("snapshot", "unknown"),
93
+ repository=snap_data.get("repository", "unknown"),
94
+ uuid=snap_data.get("uuid"),
95
+ state=snap_data.get("state", "unknown"),
96
+ include_global_state=snap_data.get("include_global_state"),
97
+ shards_stats=snap_data.get("shards_stats", {}),
98
+ stats=snap_data.get("stats", {}),
99
+ indices=snap_data.get("indices", {}),
100
+ )
101
+ snapshots.append(snapshot_status)
102
+
103
+ logger.info(f"Successfully retrieved status for {len(snapshots)} snapshots")
104
+
105
+ return SnapshotStatusResponse(
106
+ snapshots=snapshots,
107
+ success=True,
108
+ message=f"Successfully retrieved status for {len(snapshots)} snapshots",
109
+ )
110
+
111
+ except Exception as e:
112
+ logger.error(f"Failed to get snapshot status: {e}")
113
+ return SnapshotStatusResponse(
114
+ snapshots=[],
115
+ success=False,
116
+ message=f"Failed to get snapshot status: {str(e)}",
117
+ )