636 snapshot and restore2 (#638)
* chore: remove unnecessary whitespace and comment in snapshot_and_restore.py * chore: remove commented-out test line in snapshot_and_restore.py * feat: add snapshot creation, restoration, and listing functionality in Wazuh Indexer * feat: add functionality to identify and filter write indices during snapshot creation * feat: implement snapshot schedule management including creation, listing, updating, and deletion functionalities * feat: add security dependencies for admin access to snapshot and restore endpoints * feat: add snapshots API endpoints and types for snapshot management * feat: implement snapshot and restore functionality with UI components for managing snapshots, schedules, and repositories * feat: enhance snapshot and restore functionality with new UI components and routing * precommit-fixes * lint fixes * feat: remove SnapshotsIcon from Navbar items * feat: update CURRENT_VERSION to 0.1.32
taylor_socfortress committed
Jan 28, 2026 at 15:19 UTC
7810f97f68d5f29c8a55f7feeb74a57233dfcd81
22 files changed
+3176
-34
backend/alembic/env.py
+1
@@ -10,6 +10,7 @@ from alembic import context
10
from app.auth.models.users import User
11
from app.connectors.models import Connectors
12
from app.connectors.wazuh_indexer.models.sigma import SigmaQuery
13
+from app.connectors.wazuh_indexer.models.snapshot_and_restore import SnapshotSchedule
14
15
# from app.integrations.sap_siem.models.sap_siem import SapSiemMultipleLogins
16
from app.customer_provisioning.models.default_settings import (
backend/alembic/versions/8c419faae3e5_add_index_snapshots_table.py
new
+48
@@ -0,0 +1,48 @@
1
+"""Add index snapshots table
2
+
3
+Revision ID: 8c419faae3e5
4
+Revises: 2b39bb1f528f
5
+Create Date: 2026-01-28 13:22:53.153175
6
+
7
+"""
8
+from typing import Sequence
9
+from typing import Union
10
+
11
+import sqlalchemy as sa
12
+
13
+from alembic import op
14
+
15
+# revision identifiers, used by Alembic.
16
+revision: str = "8c419faae3e5"
17
+down_revision: Union[str, None] = "2b39bb1f528f"
18
+branch_labels: Union[str, Sequence[str], None] = None
19
+depends_on: Union[str, Sequence[str], None] = None
20
+
21
+
22
+def upgrade() -> None:
23
+ # ### commands auto generated by Alembic - please adjust! ###
24
+ op.create_table(
25
+ "index_snapshot_schedules",
26
+ sa.Column("id", sa.Integer(), nullable=False),
27
+ sa.Column("name", sa.String(length=255), nullable=False),
28
+ sa.Column("index_pattern", sa.String(length=255), nullable=False),
29
+ sa.Column("repository", sa.String(length=255), nullable=False),
30
+ sa.Column("enabled", sa.Boolean(), nullable=False),
31
+ sa.Column("snapshot_prefix", sa.String(length=255), nullable=False),
32
+ sa.Column("include_global_state", sa.Boolean(), nullable=False),
33
+ sa.Column("skip_write_indices", sa.Boolean(), nullable=False),
34
+ sa.Column("retention_days", sa.Integer(), nullable=True),
35
+ sa.Column("last_execution_time", sa.DateTime(), nullable=True),
36
+ sa.Column("last_snapshot_name", sa.String(length=255), nullable=True),
37
+ sa.Column("last_execution_status", sa.String(length=255), nullable=True),
38
+ sa.Column("created_at", sa.DateTime(), nullable=False),
39
+ sa.Column("updated_at", sa.DateTime(), nullable=False),
40
+ sa.PrimaryKeyConstraint("id"),
41
+ )
42
+ # ### end Alembic commands ###
43
+
44
+
45
+def downgrade() -> None:
46
+ # ### commands auto generated by Alembic - please adjust! ###
47
+ op.drop_table("index_snapshot_schedules")
48
+ # ### end Alembic commands ###
backend/app/connectors/wazuh_indexer/models/snapshot_and_restore.py
new
+23
@@ -0,0 +1,23 @@
1
+from datetime import datetime
2
+from typing import Optional
3
+
4
+from sqlmodel import Field
5
+from sqlmodel import SQLModel
6
+
7
+
8
+class SnapshotSchedule(SQLModel, table=True):
9
+ __tablename__ = "index_snapshot_schedules"
10
+ id: Optional[int] = Field(default=None, primary_key=True)
11
+ name: str = Field(nullable=False, description="Friendly name for this schedule")
12
+ index_pattern: str = Field(nullable=False, description="Index pattern to snapshot (e.g., wazuh_customer_*)")
13
+ repository: str = Field(nullable=False, description="Repository to store snapshots")
14
+ enabled: bool = Field(default=True, description="Whether this schedule is active")
15
+ snapshot_prefix: str = Field(default="scheduled", description="Prefix for snapshot names")
16
+ include_global_state: bool = Field(default=False, description="Include global cluster state")
17
+ skip_write_indices: bool = Field(default=True, description="Skip indices currently being written to")
18
+ retention_days: Optional[int] = Field(default=None, description="Number of days to retain snapshots (None = forever)")
19
+ last_execution_time: Optional[datetime] = Field(default=None, description="Last time this schedule was executed")
20
+ last_snapshot_name: Optional[str] = Field(default=None, description="Name of the last snapshot created")
21
+ last_execution_status: Optional[str] = Field(default=None, description="Status of the last execution")
22
+ created_at: datetime = Field(default_factory=datetime.utcnow, description="When this schedule was created")
23
+ updated_at: datetime = Field(default_factory=datetime.utcnow, description="When this schedule was last updated")
backend/app/connectors/wazuh_indexer/routes/snapshot_and_restore.py
+350
-8
@@ -1,16 +1,76 @@
1
from typing import Optional
2
3
from fastapi import APIRouter
4
+from fastapi import Depends
5
from fastapi import HTTPException
6
+from fastapi import Path
7
from fastapi import Query
8
+from fastapi import Security
9
from loguru import logger
10
+from sqlalchemy.ext.asyncio import AsyncSession
11
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
+from app.auth.routes.auth import AuthHandler
13
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
14
+ CreateSnapshotRequest,
15
+)
16
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
17
+ CreateSnapshotResponse,
18
+)
19
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
20
+ RestoreSnapshotRequest,
21
+)
22
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
23
+ RestoreSnapshotResponse,
24
+)
25
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
26
+ SnapshotListResponse,
27
+)
28
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
29
+ SnapshotRepositoryListResponse,
30
+)
31
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
32
+ SnapshotScheduleCreate,
33
+)
34
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
35
+ SnapshotScheduleListResponse,
36
+)
37
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
38
+ SnapshotScheduleOperationResponse,
39
+)
40
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
41
+ SnapshotScheduleUpdate,
42
+)
43
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
44
+ SnapshotStatusResponse,
45
+)
46
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import create_snapshot
47
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import (
48
+ create_snapshot_schedule,
49
+)
50
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import (
51
+ delete_snapshot_schedule,
52
+)
53
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import (
54
+ get_snapshot_schedule,
55
+)
56
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import (
57
+ get_snapshot_status,
58
+)
59
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import (
60
+ list_snapshot_repositories,
61
+)
62
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import (
63
+ list_snapshot_schedules,
64
+)
65
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import list_snapshots
66
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import restore_snapshot
67
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import (
68
+ update_snapshot_schedule,
69
+)
70
+from app.db.db_session import get_db
71
72
wazuh_indexer_snapshots_router = APIRouter()
73
+auth_handler = AuthHandler()
74
75
76
@wazuh_indexer_snapshots_router.get(
@@ -18,6 +78,7 @@ wazuh_indexer_snapshots_router = APIRouter()
78
response_model=SnapshotRepositoryListResponse,
79
summary="List Snapshot Repositories",
80
description="Retrieve a list of all configured snapshot repositories in the Wazuh Indexer.",
81
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
82
)
83
async def get_snapshot_repositories() -> SnapshotRepositoryListResponse:
84
"""
@@ -44,6 +105,7 @@ async def get_snapshot_repositories() -> SnapshotRepositoryListResponse:
105
response_model=SnapshotStatusResponse,
106
summary="Get Snapshot Status",
107
description="Retrieve the status of snapshots. Optionally filter by repository and snapshot name.",
108
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
109
)
110
async def get_snapshots_status(
111
repository: Optional[str] = Query(
@@ -65,10 +127,7 @@ async def get_snapshots_status(
127
Returns:
128
SnapshotStatusResponse: Status of the requested snapshots.
129
"""
68
- logger.info(
69
- f"Received request to get snapshot status "
70
- f"(repository={repository}, snapshot={snapshot})",
71
- )
130
+ logger.info(f"Received request to get snapshot status " f"(repository={repository}, snapshot={snapshot},"),
131
132
if snapshot and not repository:
133
raise HTTPException(
@@ -85,3 +144,286 @@ async def get_snapshots_status(
144
)
145
146
return response
147
+
148
+
149
+@wazuh_indexer_snapshots_router.get(
150
+ "/repositories/{repository}/snapshots",
151
+ response_model=SnapshotListResponse,
152
+ summary="List Snapshots",
153
+ description="Retrieve a list of all snapshots in a specific repository.",
154
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
155
+)
156
+async def get_snapshots(
157
+ repository: str = Path(
158
+ ...,
159
+ description="Name of the repository to list snapshots from.",
160
+ ),
161
+) -> SnapshotListResponse:
162
+ """
163
+ List all snapshots in a repository.
164
+
165
+ Args:
166
+ repository: Name of the repository.
167
+
168
+ Returns:
169
+ SnapshotListResponse: List of snapshots in the repository.
170
+ """
171
+ logger.info(f"Received request to list snapshots in repository: {repository}")
172
+
173
+ response = await list_snapshots(repository=repository)
174
+
175
+ if not response.success:
176
+ raise HTTPException(
177
+ status_code=500,
178
+ detail=response.message,
179
+ )
180
+
181
+ return response
182
+
183
+
184
+@wazuh_indexer_snapshots_router.post(
185
+ "/create",
186
+ response_model=CreateSnapshotResponse,
187
+ summary="Create Snapshot",
188
+ description="Create a new snapshot in a repository.",
189
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
190
+)
191
+async def create_snapshot_endpoint(
192
+ request: CreateSnapshotRequest,
193
+) -> CreateSnapshotResponse:
194
+ """
195
+ Create a new snapshot in a repository.
196
+
197
+ Args:
198
+ request: CreateSnapshotRequest containing snapshot parameters.
199
+
200
+ Returns:
201
+ CreateSnapshotResponse: Details of the snapshot creation operation.
202
+ """
203
+ logger.info(
204
+ f"Received request to create snapshot {request.snapshot} " f"in repository {request.repository}",
205
+ )
206
+
207
+ response = await create_snapshot(request=request)
208
+
209
+ if not response.success:
210
+ raise HTTPException(
211
+ status_code=500,
212
+ detail=response.message,
213
+ )
214
+
215
+ return response
216
+
217
+
218
+@wazuh_indexer_snapshots_router.post(
219
+ "/restore",
220
+ response_model=RestoreSnapshotResponse,
221
+ summary="Restore Snapshot",
222
+ description="Restore a snapshot from a repository.",
223
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
224
+)
225
+async def restore_snapshot_endpoint(
226
+ request: RestoreSnapshotRequest,
227
+) -> RestoreSnapshotResponse:
228
+ """
229
+ Restore a snapshot from a repository.
230
+
231
+ Args:
232
+ request: RestoreSnapshotRequest containing restore parameters.
233
+
234
+ Returns:
235
+ RestoreSnapshotResponse: Details of the restoration operation.
236
+ """
237
+ logger.info(
238
+ f"Received request to restore snapshot {request.snapshot} " f"from repository {request.repository}",
239
+ )
240
+
241
+ response = await restore_snapshot(request=request)
242
+
243
+ if not response.success:
244
+ raise HTTPException(
245
+ status_code=500,
246
+ detail=response.message,
247
+ )
248
+
249
+ return response
250
+
251
+
252
+# Snapshot Schedule Routes
253
+@wazuh_indexer_snapshots_router.post(
254
+ "/schedules",
255
+ response_model=SnapshotScheduleOperationResponse,
256
+ summary="Create Snapshot Schedule",
257
+ description="Create a new scheduled snapshot configuration.",
258
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
259
+)
260
+async def create_schedule_endpoint(
261
+ request: SnapshotScheduleCreate,
262
+ session: AsyncSession = Depends(get_db),
263
+) -> SnapshotScheduleOperationResponse:
264
+ """
265
+ Create a new snapshot schedule.
266
+
267
+ Args:
268
+ request: SnapshotScheduleCreate containing schedule parameters.
269
+ session: Database session.
270
+
271
+ Returns:
272
+ SnapshotScheduleOperationResponse: Details of the created schedule.
273
+ """
274
+ logger.info(f"Received request to create snapshot schedule: {request.name}")
275
+
276
+ response = await create_snapshot_schedule(request=request, session=session)
277
+
278
+ if not response.success:
279
+ raise HTTPException(
280
+ status_code=500,
281
+ detail=response.message,
282
+ )
283
+
284
+ return response
285
+
286
+
287
+@wazuh_indexer_snapshots_router.get(
288
+ "/schedules",
289
+ response_model=SnapshotScheduleListResponse,
290
+ summary="List Snapshot Schedules",
291
+ description="Retrieve a list of all configured snapshot schedules.",
292
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
293
+)
294
+async def list_schedules_endpoint(
295
+ enabled_only: bool = Query(
296
+ False,
297
+ description="If true, only return enabled schedules.",
298
+ ),
299
+ session: AsyncSession = Depends(get_db),
300
+) -> SnapshotScheduleListResponse:
301
+ """
302
+ List all snapshot schedules.
303
+
304
+ Args:
305
+ enabled_only: Whether to filter to only enabled schedules.
306
+ session: Database session.
307
+
308
+ Returns:
309
+ SnapshotScheduleListResponse: List of snapshot schedules.
310
+ """
311
+ logger.info("Received request to list snapshot schedules")
312
+
313
+ response = await list_snapshot_schedules(session=session, enabled_only=enabled_only)
314
+
315
+ if not response.success:
316
+ raise HTTPException(
317
+ status_code=500,
318
+ detail=response.message,
319
+ )
320
+
321
+ return response
322
+
323
+
324
+@wazuh_indexer_snapshots_router.get(
325
+ "/schedules/{schedule_id}",
326
+ response_model=SnapshotScheduleOperationResponse,
327
+ summary="Get Snapshot Schedule",
328
+ description="Retrieve a specific snapshot schedule by ID.",
329
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
330
+)
331
+async def get_schedule_endpoint(
332
+ schedule_id: int = Path(..., description="ID of the schedule to retrieve"),
333
+ session: AsyncSession = Depends(get_db),
334
+) -> SnapshotScheduleOperationResponse:
335
+ """
336
+ Get a snapshot schedule by ID.
337
+
338
+ Args:
339
+ schedule_id: ID of the schedule.
340
+ session: Database session.
341
+
342
+ Returns:
343
+ SnapshotScheduleOperationResponse: The requested schedule.
344
+ """
345
+ logger.info(f"Received request to get snapshot schedule ID: {schedule_id}")
346
+
347
+ response = await get_snapshot_schedule(schedule_id=schedule_id, session=session)
348
+
349
+ if not response.success:
350
+ raise HTTPException(
351
+ status_code=404 if "not found" in response.message.lower() else 500,
352
+ detail=response.message,
353
+ )
354
+
355
+ return response
356
+
357
+
358
+@wazuh_indexer_snapshots_router.put(
359
+ "/schedules/{schedule_id}",
360
+ response_model=SnapshotScheduleOperationResponse,
361
+ summary="Update Snapshot Schedule",
362
+ description="Update an existing snapshot schedule.",
363
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
364
+)
365
+async def update_schedule_endpoint(
366
+ schedule_id: int = Path(..., description="ID of the schedule to update"),
367
+ request: SnapshotScheduleUpdate = ...,
368
+ session: AsyncSession = Depends(get_db),
369
+) -> SnapshotScheduleOperationResponse:
370
+ """
371
+ Update a snapshot schedule.
372
+
373
+ Args:
374
+ schedule_id: ID of the schedule to update.
375
+ request: SnapshotScheduleUpdate containing fields to update.
376
+ session: Database session.
377
+
378
+ Returns:
379
+ SnapshotScheduleOperationResponse: The updated schedule.
380
+ """
381
+ logger.info(f"Received request to update snapshot schedule ID: {schedule_id}")
382
+
383
+ response = await update_snapshot_schedule(
384
+ schedule_id=schedule_id,
385
+ request=request,
386
+ session=session,
387
+ )
388
+
389
+ if not response.success:
390
+ raise HTTPException(
391
+ status_code=404 if "not found" in response.message.lower() else 500,
392
+ detail=response.message,
393
+ )
394
+
395
+ return response
396
+
397
+
398
+@wazuh_indexer_snapshots_router.delete(
399
+ "/schedules/{schedule_id}",
400
+ response_model=SnapshotScheduleOperationResponse,
401
+ summary="Delete Snapshot Schedule",
402
+ description="Delete a snapshot schedule.",
403
+ dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])],
404
+)
405
+async def delete_schedule_endpoint(
406
+ schedule_id: int = Path(..., description="ID of the schedule to delete"),
407
+ session: AsyncSession = Depends(get_db),
408
+) -> SnapshotScheduleOperationResponse:
409
+ """
410
+ Delete a snapshot schedule.
411
+
412
+ Args:
413
+ schedule_id: ID of the schedule to delete.
414
+ session: Database session.
415
+
416
+ Returns:
417
+ SnapshotScheduleOperationResponse: Confirmation of deletion.
418
+ """
419
+ logger.info(f"Received request to delete snapshot schedule ID: {schedule_id}")
420
+
421
+ response = await delete_snapshot_schedule(schedule_id=schedule_id, session=session)
422
+
423
+ if not response.success:
424
+ raise HTTPException(
425
+ status_code=404 if "not found" in response.message.lower() else 500,
426
+ detail=response.message,
427
+ )
428
+
429
+ return response
backend/app/connectors/wazuh_indexer/schema/snapshot_and_restore.py
+266
@@ -9,6 +9,7 @@ from pydantic import Field
9
10
class SnapshotRepositorySettings(BaseModel):
11
"""Settings for a snapshot repository."""
12
+
13
location: Optional[str] = Field(None, description="Repository location/path")
14
compress: Optional[bool] = Field(None, description="Whether snapshots are compressed")
15
chunk_size: Optional[str] = Field(None, description="Chunk size for snapshot files")
@@ -19,6 +20,7 @@ class SnapshotRepositorySettings(BaseModel):
20
21
class SnapshotRepository(BaseModel):
22
"""Model for a single snapshot repository."""
23
+
24
name: str = Field(..., description="Name of the repository")
25
type: str = Field(..., description="Type of the repository (fs, s3, etc.)")
26
settings: Dict[str, Any] = Field(default_factory=dict, description="Repository settings")
@@ -26,6 +28,7 @@ class SnapshotRepository(BaseModel):
28
29
class SnapshotRepositoryListResponse(BaseModel):
30
"""Response model for listing snapshot repositories."""
31
+
32
repositories: List[SnapshotRepository] = Field(
33
default_factory=list,
34
description="List of snapshot repositories",
@@ -36,6 +39,7 @@ class SnapshotRepositoryListResponse(BaseModel):
39
40
class SnapshotShardStatus(BaseModel):
41
"""Status of a single shard in a snapshot."""
42
+
43
stage: str = Field(..., description="Current stage of the shard snapshot")
44
total_files: Optional[int] = Field(None, alias="total_file_count", description="Total number of files")
45
total_size_in_bytes: Optional[int] = Field(None, description="Total size in bytes")
@@ -45,6 +49,7 @@ class SnapshotShardStatus(BaseModel):
49
50
class SnapshotIndexStatus(BaseModel):
51
"""Status of an index within a snapshot."""
52
+
53
shards_stats: Dict[str, Any] = Field(default_factory=dict, description="Shard statistics")
54
stats: Dict[str, Any] = Field(default_factory=dict, description="Index statistics")
55
shards: Dict[str, SnapshotShardStatus] = Field(default_factory=dict, description="Individual shard statuses")
@@ -52,6 +57,7 @@ class SnapshotIndexStatus(BaseModel):
57
58
class SnapshotStatus(BaseModel):
59
"""Status of a single snapshot."""
60
+
61
snapshot: str = Field(..., description="Name of the snapshot")
62
repository: str = Field(..., description="Repository containing the snapshot")
63
uuid: Optional[str] = Field(None, description="UUID of the snapshot")
@@ -64,9 +70,269 @@ class SnapshotStatus(BaseModel):
70
71
class SnapshotStatusResponse(BaseModel):
72
"""Response model for snapshot status."""
73
+
74
snapshots: List[SnapshotStatus] = Field(
75
default_factory=list,
76
description="List of snapshot statuses",
77
)
78
success: bool = Field(..., description="Whether the operation was successful")
79
message: str = Field(..., description="Status message")
80
+
81
+
82
+# Models for listing snapshots
83
+class SnapshotInfo(BaseModel):
84
+ """Information about a single snapshot."""
85
+
86
+ snapshot: str = Field(..., description="Name of the snapshot")
87
+ uuid: Optional[str] = Field(None, description="UUID of the snapshot")
88
+ version_id: Optional[int] = Field(None, description="Version ID")
89
+ version: Optional[str] = Field(None, description="OpenSearch version")
90
+ indices: List[str] = Field(default_factory=list, description="List of indices in the snapshot")
91
+ include_global_state: Optional[bool] = Field(None, description="Whether global state is included")
92
+ state: str = Field(..., description="State of the snapshot")
93
+ start_time: Optional[str] = Field(None, description="Start time of the snapshot")
94
+ start_time_in_millis: Optional[int] = Field(None, description="Start time in milliseconds")
95
+ end_time: Optional[str] = Field(None, description="End time of the snapshot")
96
+ end_time_in_millis: Optional[int] = Field(None, description="End time in milliseconds")
97
+ duration_in_millis: Optional[int] = Field(None, description="Duration in milliseconds")
98
+ failures: List[Dict[str, Any]] = Field(default_factory=list, description="List of failures")
99
+ shards: Dict[str, Any] = Field(default_factory=dict, description="Shard information")
100
+
101
+
102
+class SnapshotListResponse(BaseModel):
103
+ """Response model for listing snapshots."""
104
+
105
+ repository: str = Field(..., description="Repository name")
106
+ snapshots: List[SnapshotInfo] = Field(
107
+ default_factory=list,
108
+ description="List of snapshots in the repository",
109
+ )
110
+ success: bool = Field(..., description="Whether the operation was successful")
111
+ message: str = Field(..., description="Status message")
112
+
113
+
114
+# Models for restoring snapshots
115
+class RestoreSnapshotRequest(BaseModel):
116
+ """Request model for restoring a snapshot."""
117
+
118
+ repository: str = Field(..., description="Repository name containing the snapshot")
119
+ snapshot: str = Field(..., description="Name of the snapshot to restore")
120
+ indices: Optional[List[str]] = Field(
121
+ None,
122
+ description="List of indices to restore. If not specified, all indices are restored.",
123
+ )
124
+ ignore_unavailable: Optional[bool] = Field(
125
+ True,
126
+ description="Whether to ignore unavailable indices",
127
+ )
128
+ include_global_state: Optional[bool] = Field(
129
+ False,
130
+ description="Whether to restore the global state",
131
+ )
132
+ rename_pattern: Optional[str] = Field(
133
+ None,
134
+ description="Pattern to match indices to rename",
135
+ example="wazuh_(.+)",
136
+ )
137
+ rename_replacement: Optional[str] = Field(
138
+ None,
139
+ description="Replacement string for renamed indices",
140
+ example="restored_wazuh_$1",
141
+ )
142
+ include_aliases: Optional[bool] = Field(
143
+ True,
144
+ description="Whether to restore aliases",
145
+ )
146
+ partial: Optional[bool] = Field(
147
+ False,
148
+ description="Whether to allow partial restore",
149
+ )
150
+
151
+
152
+class RestoreShardInfo(BaseModel):
153
+ """Information about restored shards."""
154
+
155
+ total: int = Field(..., description="Total number of shards")
156
+ failed: int = Field(..., description="Number of failed shards")
157
+ successful: int = Field(..., description="Number of successful shards")
158
+
159
+
160
+class RestoreIndexInfo(BaseModel):
161
+ """Information about a restored index."""
162
+
163
+ index: str = Field(..., description="Index name")
164
+ shards: RestoreShardInfo = Field(..., description="Shard restoration info")
165
+
166
+
167
+class RestoreSnapshotResponse(BaseModel):
168
+ """Response model for snapshot restoration."""
169
+
170
+ snapshot: str = Field(..., description="Name of the restored snapshot")
171
+ repository: str = Field(..., description="Repository name")
172
+ indices: List[str] = Field(default_factory=list, description="List of restored indices")
173
+ shards: RestoreShardInfo = Field(..., description="Overall shard restoration info")
174
+ success: bool = Field(..., description="Whether the operation was successful")
175
+ message: str = Field(..., description="Status message")
176
+
177
+
178
+# Models for creating snapshots
179
+class IndexWriteStatus(BaseModel):
180
+ """Status of an index regarding write activity."""
181
+
182
+ index_name: str = Field(..., description="Name of the index")
183
+ is_write_index: bool = Field(..., description="Whether this is the current write index")
184
+ index_number: Optional[int] = Field(None, description="Extracted index number from naming convention")
185
+ base_name: Optional[str] = Field(None, description="Base name without the index number")
186
+
187
+
188
+class CreateSnapshotRequest(BaseModel):
189
+ """Request model for creating a snapshot."""
190
+
191
+ repository: str = Field(..., description="Repository name to store the snapshot")
192
+ snapshot: str = Field(..., description="Name of the snapshot to create")
193
+ indices: Optional[List[str]] = Field(
194
+ None,
195
+ description="List of indices to include in the snapshot. If not specified, all indices are included.",
196
+ )
197
+ ignore_unavailable: Optional[bool] = Field(
198
+ False,
199
+ description="Whether to ignore unavailable indices",
200
+ )
201
+ include_global_state: Optional[bool] = Field(
202
+ True,
203
+ description="Whether to include the global cluster state in the snapshot",
204
+ )
205
+ partial: Optional[bool] = Field(
206
+ False,
207
+ description="Whether to allow partial snapshots",
208
+ )
209
+ wait_for_completion: Optional[bool] = Field(
210
+ False,
211
+ description="Whether to wait for the snapshot to complete before returning",
212
+ )
213
+ metadata: Optional[Dict[str, Any]] = Field(
214
+ None,
215
+ description="Custom metadata to attach to the snapshot",
216
+ )
217
+ skip_write_indices: Optional[bool] = Field(
218
+ True,
219
+ description="Whether to skip indices that are currently being written to (Graylog active indices)",
220
+ )
221
+
222
+
223
+class CreateSnapshotResponse(BaseModel):
224
+ """Response model for snapshot creation."""
225
+
226
+ snapshot: str = Field(..., description="Name of the created snapshot")
227
+ repository: str = Field(..., description="Repository name")
228
+ uuid: Optional[str] = Field(None, description="UUID of the snapshot")
229
+ state: Optional[str] = Field(None, description="Current state of the snapshot")
230
+ indices: List[str] = Field(default_factory=list, description="List of indices in the snapshot")
231
+ skipped_write_indices: List[str] = Field(
232
+ default_factory=list,
233
+ description="List of indices skipped because they are currently being written to",
234
+ )
235
+ shards: Optional[RestoreShardInfo] = Field(None, description="Shard information (if wait_for_completion=true)")
236
+ accepted: bool = Field(..., description="Whether the snapshot request was accepted")
237
+ success: bool = Field(..., description="Whether the operation was successful")
238
+ message: str = Field(..., description="Status message")
239
+
240
+
241
+# Models for scheduled snapshots
242
+class SnapshotScheduleCreate(BaseModel):
243
+ """Request model for creating a snapshot schedule."""
244
+
245
+ name: str = Field(..., description="Friendly name for this schedule")
246
+ index_pattern: str = Field(
247
+ ...,
248
+ description="Index pattern to snapshot (e.g., wazuh_customer_*)",
249
+ example="wazuh_customer_*",
250
+ )
251
+ repository: str = Field(..., description="Repository to store snapshots")
252
+ enabled: Optional[bool] = Field(True, description="Whether this schedule is active")
253
+ snapshot_prefix: Optional[str] = Field(
254
+ "scheduled",
255
+ description="Prefix for snapshot names",
256
+ )
257
+ include_global_state: Optional[bool] = Field(
258
+ False,
259
+ description="Include global cluster state",
260
+ )
261
+ skip_write_indices: Optional[bool] = Field(
262
+ True,
263
+ description="Skip indices currently being written to",
264
+ )
265
+ retention_days: Optional[int] = Field(
266
+ None,
267
+ description="Number of days to retain snapshots (None = forever)",
268
+ )
269
+
270
+
271
+class SnapshotScheduleUpdate(BaseModel):
272
+ """Request model for updating a snapshot schedule."""
273
+
274
+ name: Optional[str] = Field(None, description="Friendly name for this schedule")
275
+ index_pattern: Optional[str] = Field(None, description="Index pattern to snapshot")
276
+ repository: Optional[str] = Field(None, description="Repository to store snapshots")
277
+ enabled: Optional[bool] = Field(None, description="Whether this schedule is active")
278
+ snapshot_prefix: Optional[str] = Field(None, description="Prefix for snapshot names")
279
+ include_global_state: Optional[bool] = Field(None, description="Include global cluster state")
280
+ skip_write_indices: Optional[bool] = Field(None, description="Skip indices currently being written to")
281
+ retention_days: Optional[int] = Field(None, description="Number of days to retain snapshots")
282
+
283
+
284
+class SnapshotScheduleResponse(BaseModel):
285
+ """Response model for a snapshot schedule."""
286
+
287
+ id: int = Field(..., description="Schedule ID")
288
+ name: str = Field(..., description="Friendly name for this schedule")
289
+ index_pattern: str = Field(..., description="Index pattern to snapshot")
290
+ repository: str = Field(..., description="Repository to store snapshots")
291
+ enabled: bool = Field(..., description="Whether this schedule is active")
292
+ snapshot_prefix: str = Field(..., description="Prefix for snapshot names")
293
+ include_global_state: bool = Field(..., description="Include global cluster state")
294
+ skip_write_indices: bool = Field(..., description="Skip indices currently being written to")
295
+ retention_days: Optional[int] = Field(None, description="Number of days to retain snapshots")
296
+ last_execution_time: Optional[str] = Field(None, description="Last execution time")
297
+ last_snapshot_name: Optional[str] = Field(None, description="Name of the last snapshot created")
298
+ last_execution_status: Optional[str] = Field(None, description="Status of the last execution")
299
+ created_at: str = Field(..., description="When this schedule was created")
300
+ updated_at: str = Field(..., description="When this schedule was last updated")
301
+
302
+
303
+class SnapshotScheduleListResponse(BaseModel):
304
+ """Response model for listing snapshot schedules."""
305
+
306
+ schedules: List[SnapshotScheduleResponse] = Field(
307
+ default_factory=list,
308
+ description="List of snapshot schedules",
309
+ )
310
+ success: bool = Field(..., description="Whether the operation was successful")
311
+ message: str = Field(..., description="Status message")
312
+
313
+
314
+class SnapshotScheduleOperationResponse(BaseModel):
315
+ """Response model for snapshot schedule operations."""
316
+
317
+ schedule: Optional[SnapshotScheduleResponse] = Field(
318
+ None,
319
+ description="The snapshot schedule",
320
+ )
321
+ success: bool = Field(..., description="Whether the operation was successful")
322
+ message: str = Field(..., description="Status message")
323
+
324
+
325
+class ScheduledSnapshotExecutionResponse(BaseModel):
326
+ """Response model for scheduled snapshot execution."""
327
+
328
+ schedule_id: int = Field(..., description="Schedule ID that was executed")
329
+ schedule_name: str = Field(..., description="Schedule name")
330
+ snapshot_name: Optional[str] = Field(None, description="Name of the created snapshot")
331
+ indices_snapshotted: List[str] = Field(default_factory=list, description="Indices included in snapshot")
332
+ skipped_write_indices: List[str] = Field(default_factory=list, description="Indices skipped")
333
+ already_snapshotted_indices: List[str] = Field(
334
+ default_factory=list,
335
+ description="Indices skipped because they were already snapshotted",
336
+ )
337
+ success: bool = Field(..., description="Whether the execution was successful")
338
+ message: str = Field(..., description="Status message")
backend/app/connectors/wazuh_indexer/services/snapshot_and_restore.py
+1103
-6
@@ -1,13 +1,227 @@
1
+import re
2
+from collections import defaultdict
3
+from datetime import datetime
4
+from datetime import timedelta
5
+from typing import Dict
6
from typing import List
7
from typing import Optional
8
+from typing import Tuple
9
10
from loguru import logger
11
+from sqlalchemy import select
12
+from sqlalchemy.ext.asyncio import AsyncSession
13
14
+from app.connectors.wazuh_indexer.models.snapshot_and_restore import SnapshotSchedule
15
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
16
+ CreateSnapshotRequest,
17
+)
18
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
19
+ CreateSnapshotResponse,
20
+)
21
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import IndexWriteStatus
22
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import RestoreShardInfo
23
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
24
+ RestoreSnapshotRequest,
25
+)
26
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
27
+ RestoreSnapshotResponse,
28
+)
29
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
30
+ ScheduledSnapshotExecutionResponse,
31
+)
32
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotInfo
33
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
34
+ SnapshotListResponse,
35
+)
36
from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotRepository
7
-from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotRepositoryListResponse
37
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
38
+ SnapshotRepositoryListResponse,
39
+)
40
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
41
+ SnapshotScheduleCreate,
42
+)
43
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
44
+ SnapshotScheduleListResponse,
45
+)
46
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
47
+ SnapshotScheduleOperationResponse,
48
+)
49
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
50
+ SnapshotScheduleResponse,
51
+)
52
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
53
+ SnapshotScheduleUpdate,
54
+)
55
from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotStatus
9
-from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotStatusResponse
56
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import (
57
+ SnapshotStatusResponse,
58
+)
59
from app.connectors.wazuh_indexer.utils.universal import create_wazuh_indexer_client
60
+from app.db.db_session import get_db_session
61
+
62
+
63
+def parse_graylog_index_name(index_name: str) -> Tuple[Optional[str], Optional[int]]:
64
+ """
65
+ Parse a Graylog-style index name to extract the base name and index number.
66
+
67
+ Graylog naming convention: {base_name}_{number}
68
+ Examples:
69
+ - wazuh_customer_01 -> ("wazuh_customer", 1)
70
+ - wazuh_00002_307 -> ("wazuh_00002", 307)
71
+ - graylog_0 -> ("graylog", 0)
72
+
73
+ Args:
74
+ index_name: The index name to parse.
75
+
76
+ Returns:
77
+ Tuple of (base_name, index_number) or (None, None) if pattern doesn't match.
78
+ """
79
+ # Match pattern: anything followed by underscore and a number at the end
80
+ pattern = r"^(.+)_(\d+)$"
81
+ match = re.match(pattern, index_name)
82
+
83
+ if match:
84
+ base_name = match.group(1)
85
+ index_number = int(match.group(2))
86
+ return base_name, index_number
87
+
88
+ return None, None
89
+
90
+
91
+def identify_write_indices(index_names: List[str]) -> Dict[str, IndexWriteStatus]:
92
+ """
93
+ Identify which indices are currently being written to based on Graylog naming convention.
94
+
95
+ The index with the highest number for each base name is considered the write index.
96
+
97
+ Args:
98
+ index_names: List of index names to analyze.
99
+
100
+ Returns:
101
+ Dictionary mapping index names to their write status.
102
+ """
103
+ # Group indices by base name
104
+ index_groups: Dict[str, List[Tuple[str, int]]] = defaultdict(list)
105
+
106
+ for index_name in index_names:
107
+ base_name, index_number = parse_graylog_index_name(index_name)
108
+ if base_name is not None and index_number is not None:
109
+ index_groups[base_name].append((index_name, index_number))
110
+
111
+ # Find the highest numbered index for each base name
112
+ write_indices: Dict[str, IndexWriteStatus] = {}
113
+
114
+ for base_name, indices in index_groups.items():
115
+ # Sort by index number to find the highest
116
+ sorted_indices = sorted(indices, key=lambda x: x[1], reverse=True)
117
+ highest_index_name, highest_index_number = sorted_indices[0]
118
+
119
+ # Mark all indices with their write status
120
+ for index_name, index_number in indices:
121
+ is_write_index = index_name == highest_index_name
122
+ write_indices[index_name] = IndexWriteStatus(
123
+ index_name=index_name,
124
+ is_write_index=is_write_index,
125
+ index_number=index_number,
126
+ base_name=base_name,
127
+ )
128
+
129
+ # Handle indices that don't match the Graylog pattern
130
+ for index_name in index_names:
131
+ if index_name not in write_indices:
132
+ write_indices[index_name] = IndexWriteStatus(
133
+ index_name=index_name,
134
+ is_write_index=False, # Assume non-Graylog indices are not write indices
135
+ index_number=None,
136
+ base_name=None,
137
+ )
138
+
139
+ return write_indices
140
+
141
+
142
+async def get_all_indices() -> List[str]:
143
+ """
144
+ Get all index names from the Wazuh Indexer.
145
+
146
+ Returns:
147
+ List of index names.
148
+ """
149
+ try:
150
+ es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
151
+ indices = es_client.indices.get_alias(index="*")
152
+ return list(indices.keys())
153
+ except Exception as e:
154
+ logger.error(f"Failed to get indices: {e}")
155
+ return []
156
+
157
+
158
+async def filter_write_indices(
159
+ requested_indices: Optional[List[str]] = None,
160
+) -> Tuple[List[str], List[str]]:
161
+ """
162
+ Filter out write indices from the requested indices list.
163
+
164
+ Args:
165
+ requested_indices: List of indices to filter. If None, all indices are considered.
166
+
167
+ Returns:
168
+ Tuple of (indices_to_snapshot, skipped_write_indices).
169
+ """
170
+ # Get all indices from the cluster
171
+ all_indices = await get_all_indices()
172
+
173
+ # Identify write indices
174
+ write_status = identify_write_indices(all_indices)
175
+
176
+ # Determine which indices to check
177
+ if requested_indices:
178
+ # Expand wildcards if present
179
+ indices_to_check = []
180
+ for pattern in requested_indices:
181
+ if "*" in pattern:
182
+ # Simple wildcard matching
183
+ regex_pattern = pattern.replace("*", ".*")
184
+ for index_name in all_indices:
185
+ if re.match(f"^{regex_pattern}$", index_name):
186
+ indices_to_check.append(index_name)
187
+ else:
188
+ indices_to_check.append(pattern)
189
+ else:
190
+ indices_to_check = all_indices
191
+
192
+ # Separate write indices from non-write indices
193
+ indices_to_snapshot = []
194
+ skipped_write_indices = []
195
+
196
+ for index_name in indices_to_check:
197
+ status = write_status.get(index_name)
198
+ if status and status.is_write_index:
199
+ skipped_write_indices.append(index_name)
200
+ logger.info(f"Skipping write index: {index_name} " f"(base: {status.base_name}, number: {status.index_number}, "),
201
+ else:
202
+ indices_to_snapshot.append(index_name)
203
+
204
+ return indices_to_snapshot, skipped_write_indices
205
+
206
+
207
+def _schedule_to_response(schedule: SnapshotSchedule) -> SnapshotScheduleResponse:
208
+ """Convert a SnapshotSchedule model to a response model."""
209
+ return SnapshotScheduleResponse(
210
+ id=schedule.id,
211
+ name=schedule.name,
212
+ index_pattern=schedule.index_pattern,
213
+ repository=schedule.repository,
214
+ enabled=schedule.enabled,
215
+ snapshot_prefix=schedule.snapshot_prefix,
216
+ include_global_state=schedule.include_global_state,
217
+ skip_write_indices=schedule.skip_write_indices,
218
+ retention_days=schedule.retention_days,
219
+ last_execution_time=schedule.last_execution_time.isoformat() if schedule.last_execution_time else None,
220
+ last_snapshot_name=schedule.last_snapshot_name,
221
+ last_execution_status=schedule.last_execution_status,
222
+ created_at=schedule.created_at.isoformat(),
223
+ updated_at=schedule.updated_at.isoformat(),
224
+ )
225
226
227
async def list_snapshot_repositories() -> SnapshotRepositoryListResponse:
@@ -66,10 +280,7 @@ async def get_snapshot_status(
280
Returns:
281
SnapshotStatusResponse: Response containing snapshot statuses.
282
"""
69
- logger.info(
70
- f"Fetching snapshot status from Wazuh Indexer "
71
- f"(repository={repository}, snapshot={snapshot})",
72
- )
283
+ logger.info(f"Fetching snapshot status from Wazuh Indexer " f"(repository={repository}, snapshot={snapshot}, "),
284
285
try:
286
es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
@@ -115,3 +326,889 @@ async def get_snapshot_status(
326
success=False,
327
message=f"Failed to get snapshot status: {str(e)}",
328
)
329
+
330
+
331
+async def list_snapshots(repository: str) -> SnapshotListResponse:
332
+ """
333
+ List all snapshots in a repository.
334
+
335
+ Args:
336
+ repository: Name of the repository to list snapshots from.
337
+
338
+ Returns:
339
+ SnapshotListResponse: Response containing list of snapshots.
340
+ """
341
+ logger.info(f"Fetching snapshots from repository: {repository}")
342
+
343
+ try:
344
+ es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
345
+
346
+ # Get all snapshots in the repository
347
+ response = es_client.snapshot.get(
348
+ repository=repository,
349
+ snapshot="_all",
350
+ ignore_unavailable=True,
351
+ )
352
+
353
+ snapshots: List[SnapshotInfo] = []
354
+
355
+ for snap_data in response.get("snapshots", []):
356
+ snapshot_info = SnapshotInfo(
357
+ snapshot=snap_data.get("snapshot", "unknown"),
358
+ uuid=snap_data.get("uuid"),
359
+ version_id=snap_data.get("version_id"),
360
+ version=snap_data.get("version"),
361
+ indices=snap_data.get("indices", []),
362
+ include_global_state=snap_data.get("include_global_state"),
363
+ state=snap_data.get("state", "unknown"),
364
+ start_time=snap_data.get("start_time"),
365
+ start_time_in_millis=snap_data.get("start_time_in_millis"),
366
+ end_time=snap_data.get("end_time"),
367
+ end_time_in_millis=snap_data.get("end_time_in_millis"),
368
+ duration_in_millis=snap_data.get("duration_in_millis"),
369
+ failures=snap_data.get("failures", []),
370
+ shards=snap_data.get("shards", {}),
371
+ )
372
+ snapshots.append(snapshot_info)
373
+
374
+ logger.info(f"Successfully retrieved {len(snapshots)} snapshots from repository {repository}")
375
+
376
+ return SnapshotListResponse(
377
+ repository=repository,
378
+ snapshots=snapshots,
379
+ success=True,
380
+ message=f"Successfully retrieved {len(snapshots)} snapshots from repository {repository}",
381
+ )
382
+
383
+ except Exception as e:
384
+ logger.error(f"Failed to list snapshots from repository {repository}: {e}")
385
+ return SnapshotListResponse(
386
+ repository=repository,
387
+ snapshots=[],
388
+ success=False,
389
+ message=f"Failed to list snapshots: {str(e)}",
390
+ )
391
+
392
+
393
+async def restore_snapshot(request: RestoreSnapshotRequest) -> RestoreSnapshotResponse:
394
+ """
395
+ Restore a snapshot from a repository.
396
+
397
+ Args:
398
+ request: RestoreSnapshotRequest containing restore parameters.
399
+
400
+ Returns:
401
+ RestoreSnapshotResponse: Response containing restoration details.
402
+ """
403
+ logger.info(
404
+ f"Restoring snapshot {request.snapshot} from repository {request.repository}",
405
+ )
406
+
407
+ try:
408
+ es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
409
+
410
+ # Build the restore body
411
+ body = {}
412
+
413
+ if request.indices:
414
+ body["indices"] = ",".join(request.indices)
415
+
416
+ if request.ignore_unavailable is not None:
417
+ body["ignore_unavailable"] = request.ignore_unavailable
418
+
419
+ if request.include_global_state is not None:
420
+ body["include_global_state"] = request.include_global_state
421
+
422
+ if request.rename_pattern:
423
+ body["rename_pattern"] = request.rename_pattern
424
+
425
+ if request.rename_replacement:
426
+ body["rename_replacement"] = request.rename_replacement
427
+
428
+ if request.include_aliases is not None:
429
+ body["include_aliases"] = request.include_aliases
430
+
431
+ if request.partial is not None:
432
+ body["partial"] = request.partial
433
+
434
+ # Restore the snapshot
435
+ response = es_client.snapshot.restore(
436
+ repository=request.repository,
437
+ snapshot=request.snapshot,
438
+ body=body if body else None,
439
+ wait_for_completion=False,
440
+ )
441
+
442
+ # Parse the response
443
+ snapshot_data = response.get("snapshot", {})
444
+ shards_data = snapshot_data.get("shards", {})
445
+
446
+ shards_info = RestoreShardInfo(
447
+ total=shards_data.get("total", 0),
448
+ failed=shards_data.get("failed", 0),
449
+ successful=shards_data.get("successful", 0),
450
+ )
451
+
452
+ restored_indices = snapshot_data.get("indices", [])
453
+
454
+ logger.info(f"Successfully initiated restore of snapshot {request.snapshot} ({len(restored_indices)} indices)")
455
+
456
+ return RestoreSnapshotResponse(
457
+ snapshot=request.snapshot,
458
+ repository=request.repository,
459
+ indices=restored_indices,
460
+ shards=shards_info,
461
+ success=True,
462
+ message=f"Successfully initiated restore of snapshot {request.snapshot}",
463
+ )
464
+
465
+ except Exception as e:
466
+ logger.error(f"Failed to restore snapshot {request.snapshot}: {e}")
467
+ return RestoreSnapshotResponse(
468
+ snapshot=request.snapshot,
469
+ repository=request.repository,
470
+ indices=[],
471
+ shards=RestoreShardInfo(total=0, failed=0, successful=0),
472
+ success=False,
473
+ message=f"Failed to restore snapshot: {str(e)}",
474
+ )
475
+
476
+
477
+async def create_snapshot(request: CreateSnapshotRequest) -> CreateSnapshotResponse:
478
+ """
479
+ Create a snapshot in a repository.
480
+
481
+ Args:
482
+ request: CreateSnapshotRequest containing snapshot parameters.
483
+
484
+ Returns:
485
+ CreateSnapshotResponse: Response containing snapshot creation details.
486
+ """
487
+ logger.info(
488
+ f"Creating snapshot {request.snapshot} in repository {request.repository}",
489
+ )
490
+
491
+ try:
492
+ es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
493
+
494
+ # Filter out write indices if requested
495
+ skipped_write_indices = []
496
+ indices_to_snapshot = request.indices
497
+
498
+ if request.skip_write_indices:
499
+ indices_to_snapshot, skipped_write_indices = await filter_write_indices(
500
+ requested_indices=request.indices,
501
+ )
502
+
503
+ if skipped_write_indices:
504
+ logger.info(
505
+ f"Skipping {len(skipped_write_indices)} write indices: {skipped_write_indices}",
506
+ )
507
+
508
+ if not indices_to_snapshot:
509
+ logger.warning("No indices to snapshot after filtering write indices")
510
+ return CreateSnapshotResponse(
511
+ snapshot=request.snapshot,
512
+ repository=request.repository,
513
+ uuid=None,
514
+ state=None,
515
+ indices=[],
516
+ skipped_write_indices=skipped_write_indices,
517
+ shards=None,
518
+ accepted=False,
519
+ success=False,
520
+ message="No indices to snapshot - all requested indices are currently being written to",
521
+ )
522
+
523
+ # Build the snapshot body
524
+ body = {}
525
+
526
+ if indices_to_snapshot:
527
+ body["indices"] = ",".join(indices_to_snapshot)
528
+
529
+ if request.ignore_unavailable is not None:
530
+ body["ignore_unavailable"] = request.ignore_unavailable
531
+
532
+ if request.include_global_state is not None:
533
+ body["include_global_state"] = request.include_global_state
534
+
535
+ if request.partial is not None:
536
+ body["partial"] = request.partial
537
+
538
+ if request.metadata:
539
+ # Add skipped indices to metadata for reference
540
+ metadata = request.metadata.copy()
541
+ if skipped_write_indices:
542
+ metadata["skipped_write_indices"] = skipped_write_indices
543
+ body["metadata"] = metadata
544
+ elif skipped_write_indices:
545
+ body["metadata"] = {"skipped_write_indices": skipped_write_indices}
546
+
547
+ # Create the snapshot
548
+ response = es_client.snapshot.create(
549
+ repository=request.repository,
550
+ snapshot=request.snapshot,
551
+ body=body if body else None,
552
+ wait_for_completion=request.wait_for_completion or False,
553
+ )
554
+
555
+ # Parse the response based on whether we waited for completion
556
+ if request.wait_for_completion:
557
+ snapshot_data = response.get("snapshot", {})
558
+ shards_data = snapshot_data.get("shards", {})
559
+
560
+ shards_info = RestoreShardInfo(
561
+ total=shards_data.get("total", 0),
562
+ failed=shards_data.get("failed", 0),
563
+ successful=shards_data.get("successful", 0),
564
+ )
565
+
566
+ message = f"Successfully created snapshot {request.snapshot}"
567
+ if skipped_write_indices:
568
+ message += f" (skipped {len(skipped_write_indices)} write indices)"
569
+
570
+ logger.info(message)
571
+
572
+ return CreateSnapshotResponse(
573
+ snapshot=snapshot_data.get("snapshot", request.snapshot),
574
+ repository=request.repository,
575
+ uuid=snapshot_data.get("uuid"),
576
+ state=snapshot_data.get("state"),
577
+ indices=snapshot_data.get("indices", []),
578
+ skipped_write_indices=skipped_write_indices,
579
+ shards=shards_info,
580
+ accepted=True,
581
+ success=True,
582
+ message=message,
583
+ )
584
+ else:
585
+ # When not waiting, we get an accepted response
586
+ accepted = response.get("accepted", False)
587
+
588
+ message = f"Snapshot {request.snapshot} creation initiated"
589
+ if skipped_write_indices:
590
+ message += f" (skipped {len(skipped_write_indices)} write indices)"
591
+
592
+ logger.info(message)
593
+
594
+ return CreateSnapshotResponse(
595
+ snapshot=request.snapshot,
596
+ repository=request.repository,
597
+ uuid=None,
598
+ state="IN_PROGRESS",
599
+ indices=indices_to_snapshot or [],
600
+ skipped_write_indices=skipped_write_indices,
601
+ shards=None,
602
+ accepted=accepted,
603
+ success=accepted,
604
+ message=message if accepted else "Snapshot request was not accepted",
605
+ )
606
+
607
+ except Exception as e:
608
+ logger.error(f"Failed to create snapshot {request.snapshot}: {e}")
609
+ return CreateSnapshotResponse(
610
+ snapshot=request.snapshot,
611
+ repository=request.repository,
612
+ uuid=None,
613
+ state=None,
614
+ indices=[],
615
+ skipped_write_indices=[],
616
+ shards=None,
617
+ accepted=False,
618
+ success=False,
619
+ message=f"Failed to create snapshot: {str(e)}",
620
+ )
621
+
622
+
623
+async def create_snapshot_schedule(
624
+ request: SnapshotScheduleCreate,
625
+ session: AsyncSession,
626
+) -> SnapshotScheduleOperationResponse:
627
+ """
628
+ Create a new snapshot schedule.
629
+
630
+ Args:
631
+ request: SnapshotScheduleCreate containing schedule parameters.
632
+ session: Database session.
633
+
634
+ Returns:
635
+ SnapshotScheduleOperationResponse: Response containing the created schedule.
636
+ """
637
+ logger.info(f"Creating snapshot schedule: {request.name}")
638
+
639
+ try:
640
+ schedule = SnapshotSchedule(
641
+ name=request.name,
642
+ index_pattern=request.index_pattern,
643
+ repository=request.repository,
644
+ enabled=request.enabled if request.enabled is not None else True,
645
+ snapshot_prefix=request.snapshot_prefix or "scheduled",
646
+ include_global_state=request.include_global_state if request.include_global_state is not None else False,
647
+ skip_write_indices=request.skip_write_indices if request.skip_write_indices is not None else True,
648
+ retention_days=request.retention_days,
649
+ )
650
+
651
+ session.add(schedule)
652
+ await session.commit()
653
+ await session.refresh(schedule)
654
+
655
+ logger.info(f"Successfully created snapshot schedule: {schedule.name} (ID: {schedule.id},)")
656
+
657
+ return SnapshotScheduleOperationResponse(
658
+ schedule=_schedule_to_response(schedule),
659
+ success=True,
660
+ message=f"Successfully created snapshot schedule: {schedule.name}",
661
+ )
662
+
663
+ except Exception as e:
664
+ logger.error(f"Failed to create snapshot schedule: {e}")
665
+ await session.rollback()
666
+ return SnapshotScheduleOperationResponse(
667
+ schedule=None,
668
+ success=False,
669
+ message=f"Failed to create snapshot schedule: {str(e)}",
670
+ )
671
+
672
+
673
+async def list_snapshot_schedules(
674
+ session: AsyncSession,
675
+ enabled_only: bool = False,
676
+) -> SnapshotScheduleListResponse:
677
+ """
678
+ List all snapshot schedules.
679
+
680
+ Args:
681
+ session: Database session.
682
+ enabled_only: If True, only return enabled schedules.
683
+
684
+ Returns:
685
+ SnapshotScheduleListResponse: Response containing list of schedules.
686
+ """
687
+ logger.info("Fetching snapshot schedules")
688
+
689
+ try:
690
+ query = select(SnapshotSchedule)
691
+ if enabled_only:
692
+ query = query.where(SnapshotSchedule.enabled == True)
693
+
694
+ result = await session.execute(query)
695
+ schedules = result.scalars().all()
696
+
697
+ schedule_responses = [_schedule_to_response(s) for s in schedules]
698
+
699
+ logger.info(f"Successfully retrieved {len(schedule_responses)} snapshot schedules")
700
+
701
+ return SnapshotScheduleListResponse(
702
+ schedules=schedule_responses,
703
+ success=True,
704
+ message=f"Successfully retrieved {len(schedule_responses)} snapshot schedules",
705
+ )
706
+
707
+ except Exception as e:
708
+ logger.error(f"Failed to list snapshot schedules: {e}")
709
+ return SnapshotScheduleListResponse(
710
+ schedules=[],
711
+ success=False,
712
+ message=f"Failed to list snapshot schedules: {str(e)}",
713
+ )
714
+
715
+
716
+async def get_snapshot_schedule(
717
+ schedule_id: int,
718
+ session: AsyncSession,
719
+) -> SnapshotScheduleOperationResponse:
720
+ """
721
+ Get a snapshot schedule by ID.
722
+
723
+ Args:
724
+ schedule_id: ID of the schedule to retrieve.
725
+ session: Database session.
726
+
727
+ Returns:
728
+ SnapshotScheduleOperationResponse: Response containing the schedule.
729
+ """
730
+ logger.info(f"Fetching snapshot schedule ID: {schedule_id}")
731
+
732
+ try:
733
+ result = await session.execute(select(SnapshotSchedule).where(SnapshotSchedule.id == schedule_id))
734
+ schedule = result.scalar_one_or_none()
735
+
736
+ if not schedule:
737
+ return SnapshotScheduleOperationResponse(
738
+ schedule=None,
739
+ success=False,
740
+ message=f"Snapshot schedule with ID {schedule_id} not found",
741
+ )
742
+
743
+ return SnapshotScheduleOperationResponse(
744
+ schedule=_schedule_to_response(schedule),
745
+ success=True,
746
+ message=f"Successfully retrieved snapshot schedule: {schedule.name}",
747
+ )
748
+
749
+ except Exception as e:
750
+ logger.error(f"Failed to get snapshot schedule: {e}")
751
+ return SnapshotScheduleOperationResponse(
752
+ schedule=None,
753
+ success=False,
754
+ message=f"Failed to get snapshot schedule: {str(e)}",
755
+ )
756
+
757
+
758
+async def update_snapshot_schedule(
759
+ schedule_id: int,
760
+ request: SnapshotScheduleUpdate,
761
+ session: AsyncSession,
762
+) -> SnapshotScheduleOperationResponse:
763
+ """
764
+ Update a snapshot schedule.
765
+
766
+ Args:
767
+ schedule_id: ID of the schedule to update.
768
+ request: SnapshotScheduleUpdate containing fields to update.
769
+ session: Database session.
770
+
771
+ Returns:
772
+ SnapshotScheduleOperationResponse: Response containing the updated schedule.
773
+ """
774
+ logger.info(f"Updating snapshot schedule ID: {schedule_id}")
775
+
776
+ try:
777
+ result = await session.execute(select(SnapshotSchedule).where(SnapshotSchedule.id == schedule_id))
778
+ schedule = result.scalar_one_or_none()
779
+
780
+ if not schedule:
781
+ return SnapshotScheduleOperationResponse(
782
+ schedule=None,
783
+ success=False,
784
+ message=f"Snapshot schedule with ID {schedule_id} not found",
785
+ )
786
+
787
+ # Update fields if provided
788
+ if request.name is not None:
789
+ schedule.name = request.name
790
+ if request.index_pattern is not None:
791
+ schedule.index_pattern = request.index_pattern
792
+ if request.repository is not None:
793
+ schedule.repository = request.repository
794
+ if request.enabled is not None:
795
+ schedule.enabled = request.enabled
796
+ if request.snapshot_prefix is not None:
797
+ schedule.snapshot_prefix = request.snapshot_prefix
798
+ if request.include_global_state is not None:
799
+ schedule.include_global_state = request.include_global_state
800
+ if request.skip_write_indices is not None:
801
+ schedule.skip_write_indices = request.skip_write_indices
802
+ if request.retention_days is not None:
803
+ schedule.retention_days = request.retention_days
804
+
805
+ schedule.updated_at = datetime.utcnow()
806
+
807
+ await session.commit()
808
+ await session.refresh(schedule)
809
+
810
+ logger.info(f"Successfully updated snapshot schedule: {schedule.name}")
811
+
812
+ return SnapshotScheduleOperationResponse(
813
+ schedule=_schedule_to_response(schedule),
814
+ success=True,
815
+ message=f"Successfully updated snapshot schedule: {schedule.name}",
816
+ )
817
+
818
+ except Exception as e:
819
+ logger.error(f"Failed to update snapshot schedule: {e}")
820
+ await session.rollback()
821
+ return SnapshotScheduleOperationResponse(
822
+ schedule=None,
823
+ success=False,
824
+ message=f"Failed to update snapshot schedule: {str(e)}",
825
+ )
826
+
827
+
828
+async def delete_snapshot_schedule(
829
+ schedule_id: int,
830
+ session: AsyncSession,
831
+) -> SnapshotScheduleOperationResponse:
832
+ """
833
+ Delete a snapshot schedule.
834
+
835
+ Args:
836
+ schedule_id: ID of the schedule to delete.
837
+ session: Database session.
838
+
839
+ Returns:
840
+ SnapshotScheduleOperationResponse: Response indicating success or failure.
841
+ """
842
+ logger.info(f"Deleting snapshot schedule ID: {schedule_id}")
843
+
844
+ try:
845
+ result = await session.execute(select(SnapshotSchedule).where(SnapshotSchedule.id == schedule_id))
846
+ schedule = result.scalar_one_or_none()
847
+
848
+ if not schedule:
849
+ return SnapshotScheduleOperationResponse(
850
+ schedule=None,
851
+ success=False,
852
+ message=f"Snapshot schedule with ID {schedule_id} not found",
853
+ )
854
+
855
+ schedule_name = schedule.name
856
+ await session.delete(schedule)
857
+ await session.commit()
858
+
859
+ logger.info(f"Successfully deleted snapshot schedule: {schedule_name}")
860
+
861
+ return SnapshotScheduleOperationResponse(
862
+ schedule=None,
863
+ success=True,
864
+ message=f"Successfully deleted snapshot schedule: {schedule_name}",
865
+ )
866
+
867
+ except Exception as e:
868
+ logger.error(f"Failed to delete snapshot schedule: {e}")
869
+ await session.rollback()
870
+ return SnapshotScheduleOperationResponse(
871
+ schedule=None,
872
+ success=False,
873
+ message=f"Failed to delete snapshot schedule: {str(e)}",
874
+ )
875
+
876
+
877
+async def get_snapshotted_indices_for_schedule(
878
+ schedule: SnapshotSchedule,
879
+) -> set[str]:
880
+ """
881
+ Get all indices that have already been snapshotted for a given schedule.
882
+
883
+ Args:
884
+ schedule: The snapshot schedule to check.
885
+
886
+ Returns:
887
+ Set of index names that have already been snapshotted.
888
+ """
889
+ logger.info(f"Fetching previously snapshotted indices for schedule: {schedule.name}")
890
+
891
+ snapshotted_indices: set[str] = set()
892
+
893
+ try:
894
+ es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
895
+
896
+ # List all snapshots in the repository
897
+ response = es_client.snapshot.get(
898
+ repository=schedule.repository,
899
+ snapshot="_all",
900
+ ignore_unavailable=True,
901
+ )
902
+
903
+ # Build the prefix pattern for this schedule's snapshots
904
+ prefix = f"{schedule.snapshot_prefix}_{schedule.name}_".lower().replace(" ", "_")
905
+
906
+ for snap_data in response.get("snapshots", []):
907
+ snapshot_name = snap_data.get("snapshot", "")
908
+
909
+ # Only consider snapshots created by this schedule
910
+ if snapshot_name.startswith(prefix):
911
+ indices = snap_data.get("indices", [])
912
+ snapshotted_indices.update(indices)
913
+
914
+ logger.info(f"Found {len(snapshotted_indices)} previously snapshotted indices " f"for schedule {schedule.name}")
915
+
916
+ return snapshotted_indices
917
+
918
+ except Exception as e:
919
+ logger.error(f"Failed to get snapshotted indices for schedule {schedule.name}: {e}")
920
+ return set()
921
+
922
+
923
+async def get_indices_needing_snapshot(
924
+ schedule: SnapshotSchedule,
925
+) -> tuple[list[str], list[str], list[str]]:
926
+ """
927
+ Determine which indices need to be snapshotted based on the schedule's index pattern.
928
+
929
+ This function:
930
+ 1. Gets all indices matching the schedule's pattern
931
+ 2. Filters out write indices (if configured)
932
+ 3. Filters out indices that have already been snapshotted
933
+
934
+ Args:
935
+ schedule: The snapshot schedule.
936
+
937
+ Returns:
938
+ Tuple of (indices_to_snapshot, skipped_write_indices, already_snapshotted_indices).
939
+ """
940
+ logger.info(f"Determining indices needing snapshot for schedule: {schedule.name}")
941
+
942
+ # Get all indices from the cluster
943
+ all_indices = await get_all_indices()
944
+
945
+ # Filter indices matching the schedule's pattern
946
+ matching_indices = []
947
+ pattern = schedule.index_pattern
948
+ if "*" in pattern:
949
+ regex_pattern = pattern.replace("*", ".*")
950
+ for index_name in all_indices:
951
+ if re.match(f"^{regex_pattern}$", index_name):
952
+ matching_indices.append(index_name)
953
+ else:
954
+ if pattern in all_indices:
955
+ matching_indices.append(pattern)
956
+
957
+ logger.info(f"Found {len(matching_indices)} indices matching pattern '{pattern}'")
958
+
959
+ # Identify write indices
960
+ write_status = identify_write_indices(all_indices)
961
+
962
+ # Get previously snapshotted indices
963
+ previously_snapshotted = await get_snapshotted_indices_for_schedule(schedule)
964
+
965
+ # Categorize indices
966
+ indices_to_snapshot = []
967
+ skipped_write_indices = []
968
+ already_snapshotted_indices = []
969
+
970
+ for index_name in matching_indices:
971
+ status = write_status.get(index_name)
972
+
973
+ # Check if it's a write index
974
+ if schedule.skip_write_indices and status and status.is_write_index:
975
+ skipped_write_indices.append(index_name)
976
+ logger.debug(f"Skipping write index: {index_name}")
977
+ continue
978
+
979
+ # Check if already snapshotted
980
+ if index_name in previously_snapshotted:
981
+ already_snapshotted_indices.append(index_name)
982
+ logger.debug(f"Skipping already snapshotted index: {index_name}")
983
+ continue
984
+
985
+ # This index needs to be snapshotted
986
+ indices_to_snapshot.append(index_name)
987
+
988
+ logger.info(
989
+ f"Schedule {schedule.name}: "
990
+ f"{len(indices_to_snapshot)} to snapshot, "
991
+ f"{len(skipped_write_indices)} write indices skipped, "
992
+ f"{len(already_snapshotted_indices)} already snapshotted",
993
+ )
994
+
995
+ return indices_to_snapshot, skipped_write_indices, already_snapshotted_indices
996
+
997
+
998
+async def execute_snapshot_schedule(
999
+ schedule: SnapshotSchedule,
1000
+ session: AsyncSession,
1001
+) -> ScheduledSnapshotExecutionResponse:
1002
+ """
1003
+ Execute a single snapshot schedule.
1004
+
1005
+ Args:
1006
+ schedule: The schedule to execute.
1007
+ session: Database session.
1008
+
1009
+ Returns:
1010
+ ScheduledSnapshotExecutionResponse: Response containing execution details.
1011
+ """
1012
+ logger.info(f"Executing snapshot schedule: {schedule.name} (ID: {schedule.id})")
1013
+
1014
+ try:
1015
+ # Determine which indices need to be snapshotted
1016
+ indices_to_snapshot, skipped_write_indices, already_snapshotted = await get_indices_needing_snapshot(schedule)
1017
+
1018
+ # If no new indices to snapshot, skip this execution
1019
+ if not indices_to_snapshot:
1020
+ message = "No new indices to snapshot"
1021
+ if skipped_write_indices:
1022
+ message += f" ({len(skipped_write_indices)} write indices skipped)"
1023
+ if already_snapshotted:
1024
+ message += f" ({len(already_snapshotted)} already snapshotted)"
1025
+
1026
+ logger.info(f"Schedule {schedule.name}: {message}")
1027
+
1028
+ # Update schedule with execution results
1029
+ schedule.last_execution_time = datetime.utcnow()
1030
+ schedule.last_execution_status = f"SKIPPED: {message}"
1031
+ schedule.updated_at = datetime.utcnow()
1032
+
1033
+ await session.commit()
1034
+
1035
+ return ScheduledSnapshotExecutionResponse(
1036
+ schedule_id=schedule.id,
1037
+ schedule_name=schedule.name,
1038
+ snapshot_name=None,
1039
+ indices_snapshotted=[],
1040
+ skipped_write_indices=skipped_write_indices,
1041
+ success=True, # Not a failure, just nothing to do
1042
+ message=message,
1043
+ )
1044
+
1045
+ # Generate snapshot name with timestamp
1046
+ timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
1047
+ snapshot_name = f"{schedule.snapshot_prefix}_{schedule.name}_{timestamp}".lower().replace(" ", "_")
1048
+
1049
+ # Create the snapshot request with specific indices (not patterns)
1050
+ request = CreateSnapshotRequest(
1051
+ repository=schedule.repository,
1052
+ snapshot=snapshot_name,
1053
+ indices=indices_to_snapshot, # Use the specific indices, not the pattern
1054
+ ignore_unavailable=True,
1055
+ include_global_state=schedule.include_global_state,
1056
+ partial=False,
1057
+ wait_for_completion=False,
1058
+ skip_write_indices=False, # Already filtered above
1059
+ metadata={
1060
+ "schedule_id": schedule.id,
1061
+ "schedule_name": schedule.name,
1062
+ "created_by": "scheduled_job",
1063
+ "skipped_write_indices": skipped_write_indices,
1064
+ "already_snapshotted_count": len(already_snapshotted),
1065
+ },
1066
+ )
1067
+
1068
+ # Execute the snapshot
1069
+ response = await create_snapshot(request)
1070
+
1071
+ # Update schedule with execution results
1072
+ schedule.last_execution_time = datetime.utcnow()
1073
+ schedule.last_snapshot_name = snapshot_name if response.success else None
1074
+ schedule.last_execution_status = "SUCCESS" if response.success else f"FAILED: {response.message}"
1075
+ schedule.updated_at = datetime.utcnow()
1076
+
1077
+ await session.commit()
1078
+
1079
+ if response.success:
1080
+ logger.info(
1081
+ f"Successfully executed schedule {schedule.name}: "
1082
+ f"snapshot={snapshot_name}, "
1083
+ f"indices={len(indices_to_snapshot)}, "
1084
+ f"skipped_write={len(skipped_write_indices)}, "
1085
+ f"already_snapshotted={len(already_snapshotted)}",
1086
+ )
1087
+ else:
1088
+ logger.error(f"Failed to execute schedule {schedule.name}: {response.message}")
1089
+
1090
+ return ScheduledSnapshotExecutionResponse(
1091
+ schedule_id=schedule.id,
1092
+ schedule_name=schedule.name,
1093
+ snapshot_name=snapshot_name if response.success else None,
1094
+ indices_snapshotted=indices_to_snapshot if response.success else [],
1095
+ skipped_write_indices=skipped_write_indices,
1096
+ success=response.success,
1097
+ message=response.message,
1098
+ )
1099
+
1100
+ except Exception as e:
1101
+ logger.error(f"Failed to execute snapshot schedule {schedule.name}: {e}")
1102
+
1103
+ # Update schedule with failure status
1104
+ schedule.last_execution_time = datetime.utcnow()
1105
+ schedule.last_execution_status = f"FAILED: {str(e)}"
1106
+ schedule.updated_at = datetime.utcnow()
1107
+
1108
+ try:
1109
+ await session.commit()
1110
+ except Exception:
1111
+ pass
1112
+
1113
+ return ScheduledSnapshotExecutionResponse(
1114
+ schedule_id=schedule.id,
1115
+ schedule_name=schedule.name,
1116
+ snapshot_name=None,
1117
+ indices_snapshotted=[],
1118
+ skipped_write_indices=[],
1119
+ success=False,
1120
+ message=f"Failed to execute snapshot schedule: {str(e)}",
1121
+ )
1122
+
1123
+
1124
+async def execute_all_enabled_schedules() -> List[ScheduledSnapshotExecutionResponse]:
1125
+ """
1126
+ Execute all enabled snapshot schedules.
1127
+
1128
+ Returns:
1129
+ List of execution responses for each schedule.
1130
+ """
1131
+ logger.info("Executing all enabled snapshot schedules")
1132
+
1133
+ results: List[ScheduledSnapshotExecutionResponse] = []
1134
+
1135
+ async with get_db_session() as session:
1136
+ # Get all enabled schedules
1137
+ result = await session.execute(select(SnapshotSchedule).where(SnapshotSchedule.enabled == True))
1138
+ schedules = result.scalars().all()
1139
+
1140
+ logger.info(f"Found {len(schedules)} enabled snapshot schedules")
1141
+
1142
+ for schedule in schedules:
1143
+ execution_result = await execute_snapshot_schedule(schedule, session)
1144
+ results.append(execution_result)
1145
+
1146
+ successful = sum(1 for r in results if r.success)
1147
+ failed = len(results) - successful
1148
+ logger.info(f"Completed scheduled snapshots: {successful} successful, {failed} failed")
1149
+
1150
+ return results
1151
+
1152
+
1153
+async def cleanup_old_snapshots(
1154
+ schedule: SnapshotSchedule,
1155
+) -> Dict[str, any]:
1156
+ """
1157
+ Clean up old snapshots based on retention policy.
1158
+
1159
+ Args:
1160
+ schedule: The schedule with retention settings.
1161
+
1162
+ Returns:
1163
+ Dictionary with cleanup results.
1164
+ """
1165
+ if not schedule.retention_days:
1166
+ return {"deleted": 0, "message": "No retention policy configured"}
1167
+
1168
+ logger.info(f"Cleaning up snapshots for schedule {schedule.name} " f"(retention: {schedule.retention_days} days)")
1169
+
1170
+ try:
1171
+ es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
1172
+
1173
+ # List snapshots in the repository
1174
+ response = es_client.snapshot.get(
1175
+ repository=schedule.repository,
1176
+ snapshot="_all",
1177
+ ignore_unavailable=True,
1178
+ )
1179
+
1180
+ snapshots_to_delete = []
1181
+ cutoff_time = datetime.utcnow() - timedelta(days=schedule.retention_days)
1182
+ cutoff_millis = int(cutoff_time.timestamp() * 1000)
1183
+
1184
+ # Find snapshots matching this schedule's prefix that are older than retention
1185
+ prefix = f"{schedule.snapshot_prefix}_{schedule.name}_".lower().replace(" ", "_")
1186
+
1187
+ for snap_data in response.get("snapshots", []):
1188
+ snapshot_name = snap_data.get("snapshot", "")
1189
+ end_time_millis = snap_data.get("end_time_in_millis", 0)
1190
+
1191
+ if snapshot_name.startswith(prefix) and end_time_millis < cutoff_millis:
1192
+ snapshots_to_delete.append(snapshot_name)
1193
+
1194
+ # Delete old snapshots
1195
+ deleted_count = 0
1196
+ for snapshot_name in snapshots_to_delete:
1197
+ try:
1198
+ es_client.snapshot.delete(
1199
+ repository=schedule.repository,
1200
+ snapshot=snapshot_name,
1201
+ )
1202
+ deleted_count += 1
1203
+ logger.info(f"Deleted old snapshot: {snapshot_name}")
1204
+ except Exception as e:
1205
+ logger.error(f"Failed to delete snapshot {snapshot_name}: {e}")
1206
+
1207
+ return {
1208
+ "deleted": deleted_count,
1209
+ "message": f"Deleted {deleted_count} snapshots older than {schedule.retention_days} days",
1210
+ }
1211
+
1212
+ except Exception as e:
1213
+ logger.error(f"Failed to cleanup snapshots for schedule {schedule.name}: {e}")
1214
+ return {"deleted": 0, "message": f"Cleanup failed: {str(e)}"}
backend/app/routers/wazuh_indexer.py
+3
-1
@@ -3,7 +3,9 @@ from fastapi import APIRouter
3
from app.connectors.wazuh_indexer.routes.alerts import wazuh_indexer_alerts_router
4
from app.connectors.wazuh_indexer.routes.monitoring import wazuh_indexer_router
5
from app.connectors.wazuh_indexer.routes.sigma import wazuh_indexer_sigma_router
6
-from app.connectors.wazuh_indexer.routes.snapshot_and_restore import wazuh_indexer_snapshots_router
6
+from app.connectors.wazuh_indexer.routes.snapshot_and_restore import (
7
+ wazuh_indexer_snapshots_router,
8
+)
9
10
# Instantiate the APIRouter
11
router = APIRouter()
backend/app/schedulers/scheduler.py
+10
@@ -57,6 +57,9 @@ from app.schedulers.services.invoke_sap_siem import (
57
invoke_sap_siem_integration_suspicious_logins_analysis,
58
)
59
from app.schedulers.services.invoke_sigma_queries import invoke_sigma_queries_collect
60
+from app.schedulers.services.invoke_snapshot_and_restore import (
61
+ invoke_snapshot_schedules,
62
+)
63
from app.schedulers.services.wazuh_index_resize import resize_wazuh_index_fields
64
65
@@ -145,6 +148,12 @@ async def initialize_job_metadata():
148
"function": invoke_alert_creation_collect,
149
"description": "Invokes alert creation collection.",
150
},
151
+ {
152
+ "job_id": "invoke_snapshot_schedules",
153
+ "time_interval": 60,
154
+ "function": invoke_snapshot_schedules,
155
+ "description": "Invokes Index snapshot schedules execution.",
156
+ },
157
# ! Mirgrated SIGMA to VELO ! #
158
# {
159
# "job_id": "invoke_sigma_queries_collect",
@@ -256,6 +265,7 @@ def get_function_by_name(function_name: str):
265
"resize_wazuh_index_fields": resize_wazuh_index_fields,
266
"invoke_alert_creation_collect": invoke_alert_creation_collect,
267
"invoke_sigma_queries_collect": invoke_sigma_queries_collect,
268
+ "invoke_snapshot_schedules": invoke_snapshot_schedules,
269
"invoke_mimecast_integration": invoke_mimecast_integration,
270
"invoke_mimecast_integration_ttp": invoke_mimecast_integration_ttp,
271
"invoke_sap_siem_integration_collection": invoke_sap_siem_integration_collection,
backend/app/schedulers/services/invoke_snapshot_and_restore.py
new
+61
@@ -0,0 +1,61 @@
1
+from datetime import datetime
2
+
3
+from loguru import logger
4
+from sqlalchemy.future import select
5
+
6
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import (
7
+ execute_all_enabled_schedules,
8
+)
9
+from app.db.db_session import get_db_session
10
+from app.schedulers.models.scheduler import JobMetadata
11
+
12
+
13
+async def invoke_snapshot_schedules():
14
+ """
15
+ Scheduled job to execute all enabled snapshot schedules.
16
+ This function is called by the scheduler at configured intervals (e.g., hourly).
17
+ """
18
+ logger.info("Starting scheduled snapshot execution")
19
+
20
+ try:
21
+ results = await execute_all_enabled_schedules()
22
+
23
+ successful = sum(1 for r in results if r.success)
24
+ failed = len(results) - successful
25
+
26
+ if results:
27
+ logger.info(
28
+ f"Scheduled snapshot execution completed: " f"{successful} successful, {failed} failed out of {len(results)} schedules",
29
+ )
30
+
31
+ # Log details for each execution
32
+ for result in results:
33
+ if result.success:
34
+ logger.info(
35
+ f" - {result.schedule_name}: SUCCESS "
36
+ f"(snapshot: {result.snapshot_name}, "
37
+ f"indices: {len(result.indices_snapshotted)}, "
38
+ f"skipped: {len(result.skipped_write_indices)}, ",
39
+ )
40
+ else:
41
+ logger.error(f" - {result.schedule_name}: FAILED - {result.message}")
42
+ else:
43
+ logger.info("No enabled snapshot schedules to execute")
44
+
45
+ # Update job metadata with last success timestamp
46
+ async with get_db_session() as session:
47
+ stmt = select(JobMetadata).where(JobMetadata.job_id == "invoke_snapshot_schedules")
48
+ result = await session.execute(stmt)
49
+ job_metadata = result.scalars().first()
50
+
51
+ if job_metadata:
52
+ job_metadata.last_success = datetime.utcnow()
53
+ session.add(job_metadata)
54
+ await session.commit()
55
+ logger.info("Updated job metadata with the last success timestamp.")
56
+ else:
57
+ logger.warning("JobMetadata for 'invoke_snapshot_schedules' not found.")
58
+
59
+ except Exception as e:
60
+ logger.error(f"Failed to execute scheduled snapshots: {e}")
61
+ raise
backend/app/version/services/version.py
+1
-1
@@ -7,7 +7,7 @@ from loguru import logger
7
from packaging.version import Version
8
9
# Current version - update this with each release
10
-CURRENT_VERSION = "0.1.31"
10
+CURRENT_VERSION = "0.1.32"
11
VERSION_CHECK_URL = "https://api.github.com/repos/socfortress/CoPilot/releases/latest"
12
13
frontend/src/api/endpoints/snapshots.ts
new
+63
@@ -0,0 +1,63 @@
1
+import type {
2
+ CreateSnapshotRequest,
3
+ CreateSnapshotResponse,
4
+ RestoreSnapshotRequest,
5
+ RestoreSnapshotResponse,
6
+ SnapshotListResponse,
7
+ SnapshotRepositoryListResponse,
8
+ SnapshotScheduleCreate,
9
+ SnapshotScheduleListResponse,
10
+ SnapshotScheduleOperationResponse,
11
+ SnapshotScheduleUpdate,
12
+ SnapshotStatusResponse
13
+} from "@/types/snapshots.d"
14
+import { HttpClient } from "../httpClient"
15
+
16
+export default {
17
+ // Repository endpoints
18
+ getRepositories() {
19
+ return HttpClient.get<SnapshotRepositoryListResponse>("/snapshots/repositories")
20
+ },
21
+
22
+ // Snapshot endpoints
23
+ getSnapshotStatus(repository?: string, snapshot?: string) {
24
+ const params = new URLSearchParams()
25
+ if (repository) params.append("repository", repository)
26
+ if (snapshot) params.append("snapshot", snapshot)
27
+ const queryString = params.toString()
28
+ return HttpClient.get<SnapshotStatusResponse>(`/snapshots/status${queryString ? `?${queryString}` : ""}`)
29
+ },
30
+
31
+ listSnapshots(repository: string) {
32
+ return HttpClient.get<SnapshotListResponse>(`/snapshots/repositories/${repository}/snapshots`)
33
+ },
34
+
35
+ createSnapshot(request: CreateSnapshotRequest) {
36
+ return HttpClient.post<CreateSnapshotResponse>("/snapshots/create", request)
37
+ },
38
+
39
+ restoreSnapshot(request: RestoreSnapshotRequest) {
40
+ return HttpClient.post<RestoreSnapshotResponse>("/snapshots/restore", request)
41
+ },
42
+
43
+ // Schedule endpoints
44
+ getSchedules(enabledOnly: boolean = false) {
45
+ return HttpClient.get<SnapshotScheduleListResponse>(`/snapshots/schedules?enabled_only=${enabledOnly}`)
46
+ },
47
+
48
+ getSchedule(scheduleId: number) {
49
+ return HttpClient.get<SnapshotScheduleOperationResponse>(`/snapshots/schedules/${scheduleId}`)
50
+ },
51
+
52
+ createSchedule(request: SnapshotScheduleCreate) {
53
+ return HttpClient.post<SnapshotScheduleOperationResponse>("/snapshots/schedules", request)
54
+ },
55
+
56
+ updateSchedule(scheduleId: number, request: SnapshotScheduleUpdate) {
57
+ return HttpClient.put<SnapshotScheduleOperationResponse>(`/snapshots/schedules/${scheduleId}`, request)
58
+ },
59
+
60
+ deleteSchedule(scheduleId: number) {
61
+ return HttpClient.delete<SnapshotScheduleOperationResponse>(`/snapshots/schedules/${scheduleId}`)
62
+ }
63
+}
frontend/src/api/index.ts
+3
-1
@@ -25,6 +25,7 @@ import sca from "./endpoints/sca"
25
import scheduler from "./endpoints/scheduler"
26
import shuffle from "./endpoints/shuffle"
27
import sigma from "./endpoints/sigma"
28
+import snapshots from "./endpoints/snapshots"
29
import soc from "./endpoints/soc"
30
import stackProvisioning from "./endpoints/stackProvisioning"
31
import sysmonConfig from "./endpoints/sysmonConfig"
@@ -73,5 +74,6 @@ export default {
74
shuffle,
75
copilotMCP,
76
customerPortal,
76
- version
77
+ version,
78
+ snapshots
79
}
frontend/src/app-layouts/common/Navbar/items.tsx
+31
-12
@@ -37,18 +37,37 @@ export default function getItems(): MenuMixedOption[] {
37
icon: renderIcon(OverviewIcon)
38
},
39
{
40
- label: () =>
41
- h(
42
- RouterLink,
43
- {
44
- to: {
45
- name: "Indices"
46
- }
47
- },
48
- { default: () => "Indices" }
49
- ),
50
- key: "Indices",
51
- icon: renderIcon(IndiciesIcon)
40
+ label: "Indices",
41
+ key: "IndicesMenu",
42
+ icon: renderIcon(IndiciesIcon),
43
+ children: [
44
+ {
45
+ label: () =>
46
+ h(
47
+ RouterLink,
48
+ {
49
+ to: {
50
+ name: "Indices"
51
+ }
52
+ },
53
+ { default: () => "Index Management" }
54
+ ),
55
+ key: "Indices"
56
+ },
57
+ {
58
+ label: () =>
59
+ h(
60
+ RouterLink,
61
+ {
62
+ to: {
63
+ name: "Snapshots"
64
+ }
65
+ },
66
+ { default: () => "Snapshot & Restore" }
67
+ ),
68
+ key: "Snapshots"
69
+ }
70
+ ]
71
},
72
{
73
label: "Agents",
frontend/src/components/snapshots/CreateSnapshotForm.vue
new
+121
@@ -0,0 +1,121 @@
1
+<template>
2
+ <n-form ref="formRef" :model="formData" :rules="rules" label-placement="left" label-width="160px">
3
+ <n-form-item label="Snapshot Name" path="snapshot">
4
+ <n-input v-model:value="formData.snapshot" placeholder="Enter snapshot name" />
5
+ </n-form-item>
6
+
7
+ <n-form-item label="Indices" path="indices">
8
+ <n-input
9
+ v-model:value="indicesInput"
10
+ placeholder="Enter index patterns (comma-separated, e.g., wazuh_*)"
11
+ type="textarea"
12
+ :rows="2"
13
+ />
14
+ </n-form-item>
15
+
16
+ <n-form-item label="Skip Write Indices" path="skip_write_indices">
17
+ <n-switch v-model:value="formData.skip_write_indices" />
18
+ <span class="ml-2 text-sm text-gray-500">Skip indices currently being written to</span>
19
+ </n-form-item>
20
+
21
+ <n-form-item label="Include Global State" path="include_global_state">
22
+ <n-switch v-model:value="formData.include_global_state" />
23
+ </n-form-item>
24
+
25
+ <n-form-item label="Ignore Unavailable" path="ignore_unavailable">
26
+ <n-switch v-model:value="formData.ignore_unavailable" />
27
+ </n-form-item>
28
+
29
+ <n-form-item label="Wait for Completion" path="wait_for_completion">
30
+ <n-switch v-model:value="formData.wait_for_completion" />
31
+ <span class="ml-2 text-sm text-gray-500">Wait for snapshot to complete before returning</span>
32
+ </n-form-item>
33
+
34
+ <div class="flex justify-end gap-2 mt-4">
35
+ <n-button @click="$emit('cancel')">Cancel</n-button>
36
+ <n-button type="primary" :loading="loading" @click="handleSubmit">Create Snapshot</n-button>
37
+ </div>
38
+ </n-form>
39
+</template>
40
+
41
+<script setup lang="ts">
42
+import type { FormInst, FormRules } from "naive-ui"
43
+import type { CreateSnapshotRequest } from "@/types/snapshots.d"
44
+import { NButton, NForm, NFormItem, NInput, NSwitch, useMessage } from "naive-ui"
45
+import { ref } from "vue"
46
+import Api from "@/api"
47
+
48
+const props = defineProps<{
49
+ repository: string | null
50
+}>()
51
+
52
+const emit = defineEmits<{
53
+ (e: "success"): void
54
+ (e: "cancel"): void
55
+}>()
56
+
57
+const message = useMessage()
58
+const formRef = ref<FormInst | null>(null)
59
+const loading = ref(false)
60
+const indicesInput = ref("")
61
+
62
+const formData = ref<Omit<CreateSnapshotRequest, "repository" | "indices">>({
63
+ snapshot: "",
64
+ ignore_unavailable: true,
65
+ include_global_state: false,
66
+ partial: false,
67
+ wait_for_completion: false,
68
+ skip_write_indices: true
69
+})
70
+
71
+const rules: FormRules = {
72
+ snapshot: {
73
+ required: true,
74
+ message: "Snapshot name is required",
75
+ trigger: "blur"
76
+ }
77
+}
78
+
79
+async function handleSubmit() {
80
+ if (!props.repository) {
81
+ message.error("Repository is required")
82
+ return
83
+ }
84
+
85
+ try {
86
+ await formRef.value?.validate()
87
+ } catch {
88
+ return
89
+ }
90
+
91
+ loading.value = true
92
+ try {
93
+ const indices = indicesInput.value
94
+ .split(",")
95
+ .map(s => s.trim())
96
+ .filter(s => s.length > 0)
97
+
98
+ const request: CreateSnapshotRequest = {
99
+ repository: props.repository,
100
+ snapshot: formData.value.snapshot,
101
+ indices: indices.length > 0 ? indices : undefined,
102
+ ignore_unavailable: formData.value.ignore_unavailable,
103
+ include_global_state: formData.value.include_global_state,
104
+ partial: formData.value.partial,
105
+ wait_for_completion: formData.value.wait_for_completion,
106
+ skip_write_indices: formData.value.skip_write_indices
107
+ }
108
+
109
+ const response = await Api.snapshots.createSnapshot(request)
110
+ if (response.data.success) {
111
+ emit("success")
112
+ } else {
113
+ message.error(response.data.message)
114
+ }
115
+ } catch (error: any) {
116
+ message.error(error.message || "Failed to create snapshot")
117
+ } finally {
118
+ loading.value = false
119
+ }
120
+}
121
+</script>
frontend/src/components/snapshots/RestoreSnapshotForm.vue
new
+127
@@ -0,0 +1,127 @@
1
+<template>
2
+ <n-form ref="formRef" :model="formData" :rules="rules" label-placement="left" label-width="160px">
3
+ <n-form-item label="Snapshot">
4
+ <n-input :value="snapshot?.snapshot" disabled />
5
+ </n-form-item>
6
+
7
+ <n-form-item label="Indices in Snapshot">
8
+ <n-tag v-for="index in snapshot?.indices.slice(0, 5)" :key="index" size="small" class="mr-1 mb-1">
9
+ {{ index }}
10
+ </n-tag>
11
+ <n-tag v-if="(snapshot?.indices.length || 0) > 5" size="small">
12
+ +{{ (snapshot?.indices.length || 0) - 5 }} more
13
+ </n-tag>
14
+ </n-form-item>
15
+
16
+ <n-form-item label="Indices to Restore" path="indices">
17
+ <n-input
18
+ v-model:value="indicesInput"
19
+ placeholder="Leave empty to restore all indices, or enter specific indices (comma-separated)"
20
+ type="textarea"
21
+ :rows="2"
22
+ />
23
+ </n-form-item>
24
+
25
+ <n-form-item label="Rename Pattern" path="rename_pattern">
26
+ <n-input v-model:value="formData.rename_pattern" placeholder="e.g., (.+)" />
27
+ </n-form-item>
28
+
29
+ <n-form-item label="Rename Replacement" path="rename_replacement">
30
+ <n-input v-model:value="formData.rename_replacement" placeholder="e.g., restored_$1" />
31
+ </n-form-item>
32
+
33
+ <n-form-item label="Include Global State" path="include_global_state">
34
+ <n-switch v-model:value="formData.include_global_state" />
35
+ </n-form-item>
36
+
37
+ <n-form-item label="Include Aliases" path="include_aliases">
38
+ <n-switch v-model:value="formData.include_aliases" />
39
+ </n-form-item>
40
+
41
+ <n-form-item label="Ignore Unavailable" path="ignore_unavailable">
42
+ <n-switch v-model:value="formData.ignore_unavailable" />
43
+ </n-form-item>
44
+
45
+ <n-alert type="warning" class="mb-4">
46
+ <strong>Note:</strong> If an index with the same name already exists, use the rename pattern to restore under
47
+ a different name. Example: Pattern <code>(.+)</code> with replacement <code>restored_$1</code>
48
+ </n-alert>
49
+
50
+ <div class="flex justify-end gap-2 mt-4">
51
+ <n-button @click="$emit('cancel')">Cancel</n-button>
52
+ <n-button type="primary" :loading="loading" @click="handleSubmit">Restore Snapshot</n-button>
53
+ </div>
54
+ </n-form>
55
+</template>
56
+
57
+<script setup lang="ts">
58
+import type { FormInst, FormRules } from "naive-ui"
59
+import type { RestoreSnapshotRequest, SnapshotInfo } from "@/types/snapshots.d"
60
+import { NAlert, NButton, NForm, NFormItem, NInput, NSwitch, NTag, useMessage } from "naive-ui"
61
+import { ref } from "vue"
62
+import Api from "@/api"
63
+
64
+const props = defineProps<{
65
+ repository: string | null
66
+ snapshot: SnapshotInfo | null
67
+}>()
68
+
69
+const emit = defineEmits<{
70
+ (e: "success"): void
71
+ (e: "cancel"): void
72
+}>()
73
+
74
+const message = useMessage()
75
+const formRef = ref<FormInst | null>(null)
76
+const loading = ref(false)
77
+const indicesInput = ref("")
78
+
79
+const formData = ref({
80
+ rename_pattern: "(.+)",
81
+ rename_replacement: "restored_$1",
82
+ include_global_state: false,
83
+ include_aliases: true,
84
+ ignore_unavailable: true,
85
+ partial: false
86
+})
87
+
88
+const rules: FormRules = {}
89
+
90
+async function handleSubmit() {
91
+ if (!props.repository || !props.snapshot) {
92
+ message.error("Repository and snapshot are required")
93
+ return
94
+ }
95
+
96
+ loading.value = true
97
+ try {
98
+ const indices = indicesInput.value
99
+ .split(",")
100
+ .map(s => s.trim())
101
+ .filter(s => s.length > 0)
102
+
103
+ const request: RestoreSnapshotRequest = {
104
+ repository: props.repository,
105
+ snapshot: props.snapshot.snapshot,
106
+ indices: indices.length > 0 ? indices : undefined,
107
+ rename_pattern: formData.value.rename_pattern || undefined,
108
+ rename_replacement: formData.value.rename_replacement || undefined,
109
+ include_global_state: formData.value.include_global_state,
110
+ include_aliases: formData.value.include_aliases,
111
+ ignore_unavailable: formData.value.ignore_unavailable,
112
+ partial: formData.value.partial
113
+ }
114
+
115
+ const response = await Api.snapshots.restoreSnapshot(request)
116
+ if (response.data.success) {
117
+ emit("success")
118
+ } else {
119
+ message.error(response.data.message)
120
+ }
121
+ } catch (error: any) {
122
+ message.error(error.message || "Failed to restore snapshot")
123
+ } finally {
124
+ loading.value = false
125
+ }
126
+}
127
+</script>
frontend/src/components/snapshots/SnapshotList.vue
new
+203
@@ -0,0 +1,203 @@
1
+<template>
2
+ <div class="flex flex-col gap-4">
3
+ <div class="flex items-center justify-between">
4
+ <h2 class="text-lg font-semibold">Snapshots</h2>
5
+ <div class="flex items-center gap-2">
6
+ <n-select
7
+ v-model:value="selectedRepository"
8
+ :options="repositoryOptions"
9
+ placeholder="Select Repository"
10
+ style="width: 200px"
11
+ @update:value="fetchSnapshots"
12
+ />
13
+ <n-button type="primary" :disabled="!selectedRepository" @click="showCreateModal = true">
14
+ <template #icon>
15
+ <Icon :name="AddIcon" :size="16" />
16
+ </template>
17
+ Create Snapshot
18
+ </n-button>
19
+ </div>
20
+ </div>
21
+
22
+ <n-spin :show="loading">
23
+ <n-card>
24
+ <n-data-table
25
+ :columns="columns"
26
+ :data="snapshots"
27
+ :bordered="false"
28
+ :single-line="false"
29
+ size="small"
30
+ :row-key="(row: SnapshotInfo) => row.snapshot"
31
+ />
32
+ </n-card>
33
+ </n-spin>
34
+
35
+ <n-empty v-if="!loading && !selectedRepository" description="Select a repository to view snapshots" />
36
+ <n-empty v-else-if="!loading && snapshots.length === 0" description="No snapshots found in this repository" />
37
+
38
+ <!-- Create Snapshot Modal -->
39
+ <n-modal v-model:show="showCreateModal" preset="dialog" title="Create Snapshot">
40
+ <CreateSnapshotForm
41
+ :repository="selectedRepository"
42
+ @success="onSnapshotCreated"
43
+ @cancel="showCreateModal = false"
44
+ />
45
+ </n-modal>
46
+
47
+ <!-- Restore Snapshot Modal -->
48
+ <n-modal v-model:show="showRestoreModal" preset="dialog" title="Restore Snapshot">
49
+ <RestoreSnapshotForm
50
+ :repository="selectedRepository"
51
+ :snapshot="selectedSnapshot"
52
+ @success="onSnapshotRestored"
53
+ @cancel="showRestoreModal = false"
54
+ />
55
+ </n-modal>
56
+ </div>
57
+</template>
58
+
59
+<script setup lang="ts">
60
+import type { DataTableColumns, SelectOption } from "naive-ui"
61
+import type { SnapshotInfo, SnapshotRepository } from "@/types/snapshots.d"
62
+import { Icon } from "@iconify/vue"
63
+import { NButton, NCard, NDataTable, NEmpty, NModal, NSelect, NSpin, NTag, useMessage } from "naive-ui"
64
+import { computed, h, onMounted, ref } from "vue"
65
+import Api from "@/api"
66
+import CreateSnapshotForm from "./CreateSnapshotForm.vue"
67
+import RestoreSnapshotForm from "./RestoreSnapshotForm.vue"
68
+
69
+const AddIcon = "carbon:add"
70
+
71
+const message = useMessage()
72
+const loading = ref(false)
73
+const repositories = ref<SnapshotRepository[]>([])
74
+const selectedRepository = ref<string | null>(null)
75
+const snapshots = ref<SnapshotInfo[]>([])
76
+const showCreateModal = ref(false)
77
+const showRestoreModal = ref(false)
78
+const selectedSnapshot = ref<SnapshotInfo | null>(null)
79
+
80
+const repositoryOptions = computed<SelectOption[]>(() =>
81
+ repositories.value.map(repo => ({
82
+ label: repo.name,
83
+ value: repo.name
84
+ }))
85
+)
86
+
87
+const columns: DataTableColumns<SnapshotInfo> = [
88
+ {
89
+ title: "Snapshot",
90
+ key: "snapshot",
91
+ sorter: "default"
92
+ },
93
+ {
94
+ title: "State",
95
+ key: "state",
96
+ render(row) {
97
+ const typeMap: Record<string, "success" | "warning" | "error" | "info"> = {
98
+ SUCCESS: "success",
99
+ IN_PROGRESS: "warning",
100
+ PARTIAL: "warning",
101
+ FAILED: "error"
102
+ }
103
+ return h(NTag, { type: typeMap[row.state] || "info", size: "small" }, () => row.state)
104
+ }
105
+ },
106
+ {
107
+ title: "Indices",
108
+ key: "indices",
109
+ render(row) {
110
+ return h("span", {}, `${row.indices.length} indices`)
111
+ }
112
+ },
113
+ {
114
+ title: "Start Time",
115
+ key: "start_time",
116
+ render(row) {
117
+ return row.start_time ? new Date(row.start_time).toLocaleString() : "-"
118
+ }
119
+ },
120
+ {
121
+ title: "End Time",
122
+ key: "end_time",
123
+ render(row) {
124
+ return row.end_time ? new Date(row.end_time).toLocaleString() : "-"
125
+ }
126
+ },
127
+ {
128
+ title: "Duration",
129
+ key: "duration_in_millis",
130
+ render(row) {
131
+ if (!row.duration_in_millis) return "-"
132
+ const seconds = Math.floor(row.duration_in_millis / 1000)
133
+ if (seconds < 60) return `${seconds}s`
134
+ const minutes = Math.floor(seconds / 60)
135
+ return `${minutes}m ${seconds % 60}s`
136
+ }
137
+ },
138
+ {
139
+ title: "Actions",
140
+ key: "actions",
141
+ render(row) {
142
+ return h(
143
+ NButton,
144
+ {
145
+ size: "small",
146
+ type: "primary",
147
+ onClick: () => openRestoreModal(row)
148
+ },
149
+ () => "Restore"
150
+ )
151
+ }
152
+ }
153
+]
154
+
155
+function openRestoreModal(snapshot: SnapshotInfo) {
156
+ selectedSnapshot.value = snapshot
157
+ showRestoreModal.value = true
158
+}
159
+
160
+async function fetchRepositories() {
161
+ try {
162
+ const response = await Api.snapshots.getRepositories()
163
+ if (response.data.success) {
164
+ repositories.value = response.data.repositories
165
+ }
166
+ } catch (error: any) {
167
+ message.error(error.message || "Failed to fetch repositories")
168
+ }
169
+}
170
+
171
+async function fetchSnapshots() {
172
+ if (!selectedRepository.value) return
173
+
174
+ loading.value = true
175
+ try {
176
+ const response = await Api.snapshots.listSnapshots(selectedRepository.value)
177
+ if (response.data.success) {
178
+ snapshots.value = response.data.snapshots
179
+ } else {
180
+ message.error(response.data.message)
181
+ }
182
+ } catch (error: any) {
183
+ message.error(error.message || "Failed to fetch snapshots")
184
+ } finally {
185
+ loading.value = false
186
+ }
187
+}
188
+
189
+function onSnapshotCreated() {
190
+ showCreateModal.value = false
191
+ fetchSnapshots()
192
+ message.success("Snapshot creation initiated")
193
+}
194
+
195
+function onSnapshotRestored() {
196
+ showRestoreModal.value = false
197
+ message.success("Snapshot restoration initiated")
198
+}
199
+
200
+onMounted(() => {
201
+ fetchRepositories()
202
+})
203
+</script>
frontend/src/components/snapshots/SnapshotRepositories.vue
new
+95
@@ -0,0 +1,95 @@
1
+<template>
2
+ <div class="flex flex-col gap-4">
3
+ <n-alert type="info" :show-icon="true">
4
+ <template #header>Repository Registration Required</template>
5
+ Snapshot repositories must be manually registered in your Wazuh Indexer cluster.
6
+ <n-a href="https://docs.opensearch.org/2.19/tuning-your-cluster/availability-and-recovery/snapshots/snapshot-restore/" target="_blank">
7
+ View the documentation
8
+ </n-a>
9
+ for instructions on how to register a snapshot repository.
10
+ </n-alert>
11
+
12
+ <div class="flex items-center justify-between">
13
+ <h2 class="text-lg font-semibold">Snapshot Repositories</h2>
14
+ <n-button type="primary" :loading="loading" @click="fetchRepositories">
15
+ <template #icon>
16
+ <Icon :name="RefreshIcon" :size="16" />
17
+ </template>
18
+ Refresh
19
+ </n-button>
20
+ </div>
21
+
22
+ <n-spin :show="loading">
23
+ <n-card>
24
+ <n-data-table
25
+ :columns="columns"
26
+ :data="repositories"
27
+ :bordered="false"
28
+ :single-line="false"
29
+ size="small"
30
+ />
31
+ </n-card>
32
+ </n-spin>
33
+
34
+ <n-empty v-if="!loading && repositories.length === 0" description="No snapshot repositories found" />
35
+ </div>
36
+</template>
37
+
38
+<script setup lang="ts">
39
+import type { DataTableColumns } from "naive-ui"
40
+import type { SnapshotRepository } from "@/types/snapshots.d"
41
+import { Icon } from "@iconify/vue"
42
+import { NA, NAlert, NButton, NCard, NDataTable, NEmpty, NSpin, useMessage } from "naive-ui"
43
+import { h, onMounted, ref } from "vue"
44
+import Api from "@/api"
45
+
46
+const RefreshIcon = "carbon:refresh"
47
+
48
+const message = useMessage()
49
+const loading = ref(false)
50
+const repositories = ref<SnapshotRepository[]>([])
51
+
52
+const columns: DataTableColumns<SnapshotRepository> = [
53
+ {
54
+ title: "Name",
55
+ key: "name",
56
+ sorter: "default"
57
+ },
58
+ {
59
+ title: "Type",
60
+ key: "type",
61
+ sorter: "default"
62
+ },
63
+ {
64
+ title: "Settings",
65
+ key: "settings",
66
+ render(row) {
67
+ return h(
68
+ "code",
69
+ { class: "text-xs" },
70
+ JSON.stringify(row.settings, null, 2)
71
+ )
72
+ }
73
+ }
74
+]
75
+
76
+async function fetchRepositories() {
77
+ loading.value = true
78
+ try {
79
+ const response = await Api.snapshots.getRepositories()
80
+ if (response.data.success) {
81
+ repositories.value = response.data.repositories
82
+ } else {
83
+ message.error(response.data.message)
84
+ }
85
+ } catch (error: any) {
86
+ message.error(error.message || "Failed to fetch repositories")
87
+ } finally {
88
+ loading.value = false
89
+ }
90
+}
91
+
92
+onMounted(() => {
93
+ fetchRepositories()
94
+})
95
+</script>
frontend/src/components/snapshots/SnapshotScheduleForm.vue
new
+190
@@ -0,0 +1,190 @@
1
+<template>
2
+ <n-form ref="formRef" :model="formData" :rules="rules" label-placement="left" label-width="160px">
3
+ <n-form-item label="Name" path="name">
4
+ <n-input v-model:value="formData.name" placeholder="Enter a friendly name for this schedule" />
5
+ </n-form-item>
6
+
7
+ <n-form-item label="Index Pattern" path="index_pattern">
8
+ <n-input v-model:value="formData.index_pattern" placeholder="e.g., wazuh_customer_*" />
9
+ <template #feedback>
10
+ Use wildcards (*) to match multiple indices. Example: wazuh_customer_*
11
+ </template>
12
+ </n-form-item>
13
+
14
+ <n-form-item label="Repository" path="repository">
15
+ <n-select
16
+ v-model:value="formData.repository"
17
+ :options="repositoryOptions"
18
+ placeholder="Select a repository"
19
+ :loading="loadingRepositories"
20
+ />
21
+ </n-form-item>
22
+
23
+ <n-form-item label="Snapshot Prefix" path="snapshot_prefix">
24
+ <n-input v-model:value="formData.snapshot_prefix" placeholder="e.g., scheduled" />
25
+ <template #feedback>
26
+ Prefix for generated snapshot names. Full name: {prefix}_{schedule_name}_{timestamp}
27
+ </template>
28
+ </n-form-item>
29
+
30
+ <n-form-item label="Enabled" path="enabled">
31
+ <n-switch v-model:value="formData.enabled" />
32
+ </n-form-item>
33
+
34
+ <n-form-item label="Skip Write Indices" path="skip_write_indices">
35
+ <n-switch v-model:value="formData.skip_write_indices" />
36
+ <span class="ml-2 text-sm text-gray-500">Skip indices currently being written to (recommended)</span>
37
+ </n-form-item>
38
+
39
+ <n-form-item label="Include Global State" path="include_global_state">
40
+ <n-switch v-model:value="formData.include_global_state" />
41
+ </n-form-item>
42
+
43
+ <n-form-item label="Retention (Days)" path="retention_days">
44
+ <n-input-number
45
+ v-model:value="formData.retention_days"
46
+ :min="1"
47
+ :max="365"
48
+ placeholder="Leave empty for no retention limit"
49
+ clearable
50
+ style="width: 100%"
51
+ />
52
+ <template #feedback>
53
+ Automatically delete snapshots older than this many days. Leave empty to keep forever.
54
+ </template>
55
+ </n-form-item>
56
+
57
+ <div class="flex justify-end gap-2 mt-4">
58
+ <n-button @click="$emit('cancel')">Cancel</n-button>
59
+ <n-button type="primary" :loading="loading" @click="handleSubmit">
60
+ {{ isEditing ? "Update Schedule" : "Create Schedule" }}
61
+ </n-button>
62
+ </div>
63
+ </n-form>
64
+</template>
65
+
66
+<script setup lang="ts">
67
+import type { FormInst, FormRules, SelectOption } from "naive-ui"
68
+import type { SnapshotRepository, SnapshotScheduleCreate, SnapshotScheduleResponse } from "@/types/snapshots.d"
69
+import { NButton, NForm, NFormItem, NInput, NInputNumber, NSelect, NSwitch, useMessage } from "naive-ui"
70
+import { computed, onMounted, ref, watch } from "vue"
71
+import Api from "@/api"
72
+
73
+const props = defineProps<{
74
+ schedule?: SnapshotScheduleResponse | null
75
+}>()
76
+
77
+const emit = defineEmits<{
78
+ (e: "success"): void
79
+ (e: "cancel"): void
80
+}>()
81
+
82
+const message = useMessage()
83
+const formRef = ref<FormInst | null>(null)
84
+const loading = ref(false)
85
+const loadingRepositories = ref(false)
86
+const repositories = ref<SnapshotRepository[]>([])
87
+
88
+const isEditing = computed(() => !!props.schedule)
89
+
90
+const formData = ref<SnapshotScheduleCreate>({
91
+ name: "",
92
+ index_pattern: "",
93
+ repository: "",
94
+ enabled: true,
95
+ snapshot_prefix: "scheduled",
96
+ include_global_state: false,
97
+ skip_write_indices: true,
98
+ retention_days: null
99
+})
100
+
101
+const repositoryOptions = computed<SelectOption[]>(() =>
102
+ repositories.value.map(repo => ({
103
+ label: repo.name,
104
+ value: repo.name
105
+ }))
106
+)
107
+
108
+const rules: FormRules = {
109
+ name: {
110
+ required: true,
111
+ message: "Name is required",
112
+ trigger: "blur"
113
+ },
114
+ index_pattern: {
115
+ required: true,
116
+ message: "Index pattern is required",
117
+ trigger: "blur"
118
+ },
119
+ repository: {
120
+ required: true,
121
+ message: "Repository is required",
122
+ trigger: "change"
123
+ }
124
+}
125
+
126
+watch(
127
+ () => props.schedule,
128
+ newSchedule => {
129
+ if (newSchedule) {
130
+ formData.value = {
131
+ name: newSchedule.name,
132
+ index_pattern: newSchedule.index_pattern,
133
+ repository: newSchedule.repository,
134
+ enabled: newSchedule.enabled,
135
+ snapshot_prefix: newSchedule.snapshot_prefix,
136
+ include_global_state: newSchedule.include_global_state,
137
+ skip_write_indices: newSchedule.skip_write_indices,
138
+ retention_days: newSchedule.retention_days
139
+ }
140
+ }
141
+ },
142
+ { immediate: true }
143
+)
144
+
145
+async function fetchRepositories() {
146
+ loadingRepositories.value = true
147
+ try {
148
+ const response = await Api.snapshots.getRepositories()
149
+ if (response.data.success) {
150
+ repositories.value = response.data.repositories
151
+ }
152
+ } catch (error: any) {
153
+ message.error(error.message || "Failed to fetch repositories")
154
+ } finally {
155
+ loadingRepositories.value = false
156
+ }
157
+}
158
+
159
+async function handleSubmit() {
160
+ try {
161
+ await formRef.value?.validate()
162
+ } catch {
163
+ return
164
+ }
165
+
166
+ loading.value = true
167
+ try {
168
+ let response
169
+ if (isEditing.value && props.schedule) {
170
+ response = await Api.snapshots.updateSchedule(props.schedule.id, formData.value)
171
+ } else {
172
+ response = await Api.snapshots.createSchedule(formData.value)
173
+ }
174
+
175
+ if (response.data.success) {
176
+ emit("success")
177
+ } else {
178
+ message.error(response.data.message)
179
+ }
180
+ } catch (error: any) {
181
+ message.error(error.message || "Failed to save schedule")
182
+ } finally {
183
+ loading.value = false
184
+ }
185
+}
186
+
187
+onMounted(() => {
188
+ fetchRepositories()
189
+})
190
+</script>
frontend/src/components/snapshots/SnapshotSchedules.vue
new
+240
@@ -0,0 +1,240 @@
1
+<template>
2
+ <div class="flex flex-col gap-4">
3
+ <div class="flex items-center justify-between">
4
+ <h2 class="text-lg font-semibold">Scheduled Snapshots</h2>
5
+ <div class="flex items-center gap-2">
6
+ <n-button :loading="loading" @click="fetchSchedules">
7
+ <template #icon>
8
+ <Icon :name="RefreshIcon" :size="16" />
9
+ </template>
10
+ Refresh
11
+ </n-button>
12
+ <n-button type="primary" @click="showCreateModal = true">
13
+ <template #icon>
14
+ <Icon :name="AddIcon" :size="16" />
15
+ </template>
16
+ Create Schedule
17
+ </n-button>
18
+ </div>
19
+ </div>
20
+
21
+ <n-spin :show="loading">
22
+ <n-card>
23
+ <n-data-table
24
+ :columns="columns"
25
+ :data="schedules"
26
+ :bordered="false"
27
+ :single-line="false"
28
+ size="small"
29
+ :row-key="(row: SnapshotScheduleResponse) => row.id"
30
+ />
31
+ </n-card>
32
+ </n-spin>
33
+
34
+ <n-empty v-if="!loading && schedules.length === 0" description="No snapshot schedules configured" />
35
+
36
+ <!-- Create Schedule Modal -->
37
+ <n-modal v-model:show="showCreateModal" preset="dialog" title="Create Snapshot Schedule" style="width: 600px">
38
+ <SnapshotScheduleForm
39
+ @success="onScheduleCreated"
40
+ @cancel="showCreateModal = false"
41
+ />
42
+ </n-modal>
43
+
44
+ <!-- Edit Schedule Modal -->
45
+ <n-modal v-model:show="showEditModal" preset="dialog" title="Edit Snapshot Schedule" style="width: 600px">
46
+ <SnapshotScheduleForm
47
+ :schedule="selectedSchedule"
48
+ @success="onScheduleUpdated"
49
+ @cancel="showEditModal = false"
50
+ />
51
+ </n-modal>
52
+ </div>
53
+</template>
54
+
55
+<script setup lang="ts">
56
+import type { DataTableColumns } from "naive-ui"
57
+import type { SnapshotScheduleResponse } from "@/types/snapshots.d"
58
+import { Icon } from "@iconify/vue"
59
+import {
60
+ NButton,
61
+ NCard,
62
+ NDataTable,
63
+ NEmpty,
64
+ NModal,
65
+ NPopconfirm,
66
+ NSpin,
67
+ NSwitch,
68
+ NTag,
69
+ useMessage
70
+} from "naive-ui"
71
+import { h, onMounted, ref } from "vue"
72
+import Api from "@/api"
73
+import SnapshotScheduleForm from "./SnapshotScheduleForm.vue"
74
+
75
+const AddIcon = "carbon:add"
76
+const RefreshIcon = "carbon:refresh"
77
+
78
+const message = useMessage()
79
+const loading = ref(false)
80
+const schedules = ref<SnapshotScheduleResponse[]>([])
81
+const showCreateModal = ref(false)
82
+const showEditModal = ref(false)
83
+const selectedSchedule = ref<SnapshotScheduleResponse | null>(null)
84
+
85
+const columns: DataTableColumns<SnapshotScheduleResponse> = [
86
+ {
87
+ title: "Name",
88
+ key: "name",
89
+ sorter: "default"
90
+ },
91
+ {
92
+ title: "Index Pattern",
93
+ key: "index_pattern",
94
+ render(row) {
95
+ return h("code", { class: "text-sm" }, row.index_pattern)
96
+ }
97
+ },
98
+ {
99
+ title: "Repository",
100
+ key: "repository"
101
+ },
102
+ {
103
+ title: "Enabled",
104
+ key: "enabled",
105
+ render(row) {
106
+ return h(NSwitch, {
107
+ value: row.enabled,
108
+ onUpdateValue: (value: boolean) => toggleEnabled(row, value)
109
+ })
110
+ }
111
+ },
112
+ {
113
+ title: "Retention",
114
+ key: "retention_days",
115
+ render(row) {
116
+ return row.retention_days ? `${row.retention_days} days` : "Forever"
117
+ }
118
+ },
119
+ {
120
+ title: "Last Execution",
121
+ key: "last_execution_time",
122
+ render(row) {
123
+ if (!row.last_execution_time) return "-"
124
+ return h("div", { class: "flex flex-col" }, [
125
+ h("span", {}, new Date(row.last_execution_time).toLocaleString()),
126
+ h(
127
+ NTag,
128
+ {
129
+ type: row.last_execution_status?.startsWith("SUCCESS")
130
+? "success" :
131
+ row.last_execution_status?.startsWith("SKIPPED") ? "warning" : "error",
132
+ size: "small",
133
+ class: "mt-1"
134
+ },
135
+ () => row.last_execution_status?.split(":")[0] || "Unknown"
136
+ )
137
+ ])
138
+ }
139
+ },
140
+ {
141
+ title: "Last Snapshot",
142
+ key: "last_snapshot_name",
143
+ render(row) {
144
+ return row.last_snapshot_name || "-"
145
+ }
146
+ },
147
+ {
148
+ title: "Actions",
149
+ key: "actions",
150
+ width: 150,
151
+ render(row) {
152
+ return h("div", { class: "flex gap-2" }, [
153
+ h(
154
+ NButton,
155
+ {
156
+ size: "small",
157
+ onClick: () => openEditModal(row)
158
+ },
159
+ () => "Edit"
160
+ ),
161
+ h(
162
+ NPopconfirm,
163
+ {
164
+ onPositiveClick: () => deleteSchedule(row)
165
+ },
166
+ {
167
+ trigger: () => h(NButton, { size: "small", type: "error" }, () => "Delete"),
168
+ default: () => "Are you sure you want to delete this schedule?"
169
+ }
170
+ )
171
+ ])
172
+ }
173
+ }
174
+]
175
+
176
+function openEditModal(schedule: SnapshotScheduleResponse) {
177
+ selectedSchedule.value = schedule
178
+ showEditModal.value = true
179
+}
180
+
181
+async function toggleEnabled(schedule: SnapshotScheduleResponse, enabled: boolean) {
182
+ try {
183
+ const response = await Api.snapshots.updateSchedule(schedule.id, { enabled })
184
+ if (response.data.success) {
185
+ message.success(`Schedule ${enabled ? "enabled" : "disabled"}`)
186
+ fetchSchedules()
187
+ } else {
188
+ message.error(response.data.message)
189
+ }
190
+ } catch (error: any) {
191
+ message.error(error.message || "Failed to update schedule")
192
+ }
193
+}
194
+
195
+async function deleteSchedule(schedule: SnapshotScheduleResponse) {
196
+ try {
197
+ const response = await Api.snapshots.deleteSchedule(schedule.id)
198
+ if (response.data.success) {
199
+ message.success("Schedule deleted")
200
+ fetchSchedules()
201
+ } else {
202
+ message.error(response.data.message)
203
+ }
204
+ } catch (error: any) {
205
+ message.error(error.message || "Failed to delete schedule")
206
+ }
207
+}
208
+
209
+async function fetchSchedules() {
210
+ loading.value = true
211
+ try {
212
+ const response = await Api.snapshots.getSchedules()
213
+ if (response.data.success) {
214
+ schedules.value = response.data.schedules
215
+ } else {
216
+ message.error(response.data.message)
217
+ }
218
+ } catch (error: any) {
219
+ message.error(error.message || "Failed to fetch schedules")
220
+ } finally {
221
+ loading.value = false
222
+ }
223
+}
224
+
225
+function onScheduleCreated() {
226
+ showCreateModal.value = false
227
+ fetchSchedules()
228
+ message.success("Schedule created successfully")
229
+}
230
+
231
+function onScheduleUpdated() {
232
+ showEditModal.value = false
233
+ fetchSchedules()
234
+ message.success("Schedule updated successfully")
235
+}
236
+
237
+onMounted(() => {
238
+ fetchSchedules()
239
+})
240
+</script>
frontend/src/router/index.ts
+21
-5
@@ -20,11 +20,27 @@ const router = createRouter({
20
meta: { title: "Overview", auth: true, roles: RouteRole.All }
21
},
22
{
23
- path: "/indices",
24
- name: "Indices",
25
- component: () => import("@/views/Indices.vue"),
26
- meta: { title: "Indices", auth: true, roles: RouteRole.All }
27
- },
23
+ path: "/indices",
24
+ redirect: "/indices/management",
25
+ meta: {
26
+ auth: true,
27
+ roles: RouteRole.All
28
+ },
29
+ children: [
30
+ {
31
+ path: "management",
32
+ name: "Indices",
33
+ component: () => import("@/views/Indices.vue"),
34
+ meta: { title: "Index Management" }
35
+ },
36
+ {
37
+ path: "snapshots",
38
+ name: "Snapshots",
39
+ component: () => import("@/views/Snapshots.vue"),
40
+ meta: { title: "Snapshot & Restore" }
41
+ }
42
+ ]
43
+ },
44
{
45
path: "/connectors",
46
name: "Connectors",
frontend/src/types/snapshots.d.ts
new
+181
@@ -0,0 +1,181 @@
1
+export interface SnapshotRepository {
2
+ name: string
3
+ type: string
4
+ settings: Record<string, any>
5
+}
6
+
7
+export interface SnapshotRepositoryListResponse {
8
+ repositories: SnapshotRepository[]
9
+ success: boolean
10
+ message: string
11
+}
12
+
13
+export interface SnapshotShardStatus {
14
+ stage: string
15
+ total_files?: number
16
+ total_size_in_bytes?: number
17
+ processed_files?: number
18
+ processed_size_in_bytes?: number
19
+}
20
+
21
+export interface SnapshotIndexStatus {
22
+ shards_stats: Record<string, any>
23
+ stats: Record<string, any>
24
+ shards: Record<string, SnapshotShardStatus>
25
+}
26
+
27
+export interface SnapshotStatus {
28
+ snapshot: string
29
+ repository: string
30
+ uuid?: string
31
+ state: string
32
+ include_global_state?: boolean
33
+ shards_stats: Record<string, any>
34
+ stats: Record<string, any>
35
+ indices: Record<string, SnapshotIndexStatus>
36
+}
37
+
38
+export interface SnapshotStatusResponse {
39
+ snapshots: SnapshotStatus[]
40
+ success: boolean
41
+ message: string
42
+}
43
+
44
+export interface SnapshotInfo {
45
+ snapshot: string
46
+ uuid?: string
47
+ version_id?: number
48
+ version?: string
49
+ indices: string[]
50
+ include_global_state?: boolean
51
+ state: string
52
+ start_time?: string
53
+ start_time_in_millis?: number
54
+ end_time?: string
55
+ end_time_in_millis?: number
56
+ duration_in_millis?: number
57
+ failures: Record<string, any>[]
58
+ shards: Record<string, any>
59
+}
60
+
61
+export interface SnapshotListResponse {
62
+ repository: string
63
+ snapshots: SnapshotInfo[]
64
+ success: boolean
65
+ message: string
66
+}
67
+
68
+export interface RestoreSnapshotRequest {
69
+ repository: string
70
+ snapshot: string
71
+ indices?: string[]
72
+ ignore_unavailable?: boolean
73
+ include_global_state?: boolean
74
+ rename_pattern?: string
75
+ rename_replacement?: string
76
+ include_aliases?: boolean
77
+ partial?: boolean
78
+}
79
+
80
+export interface RestoreShardInfo {
81
+ total: number
82
+ failed: number
83
+ successful: number
84
+}
85
+
86
+export interface RestoreSnapshotResponse {
87
+ snapshot: string
88
+ repository: string
89
+ indices: string[]
90
+ shards: RestoreShardInfo
91
+ success: boolean
92
+ message: string
93
+}
94
+
95
+export interface CreateSnapshotRequest {
96
+ repository: string
97
+ snapshot: string
98
+ indices?: string[]
99
+ ignore_unavailable?: boolean
100
+ include_global_state?: boolean
101
+ partial?: boolean
102
+ wait_for_completion?: boolean
103
+ metadata?: Record<string, any>
104
+ skip_write_indices?: boolean
105
+}
106
+
107
+export interface CreateSnapshotResponse {
108
+ snapshot: string
109
+ repository: string
110
+ uuid?: string
111
+ state?: string
112
+ indices: string[]
113
+ skipped_write_indices: string[]
114
+ shards?: RestoreShardInfo
115
+ accepted: boolean
116
+ success: boolean
117
+ message: string
118
+}
119
+
120
+// Snapshot Schedule Types
121
+export interface SnapshotScheduleCreate {
122
+ name: string
123
+ index_pattern: string
124
+ repository: string
125
+ enabled?: boolean
126
+ snapshot_prefix?: string
127
+ include_global_state?: boolean
128
+ skip_write_indices?: boolean
129
+ retention_days?: number | null
130
+}
131
+
132
+export interface SnapshotScheduleUpdate {
133
+ name?: string
134
+ index_pattern?: string
135
+ repository?: string
136
+ enabled?: boolean
137
+ snapshot_prefix?: string
138
+ include_global_state?: boolean
139
+ skip_write_indices?: boolean
140
+ retention_days?: number | null
141
+}
142
+
143
+export interface SnapshotScheduleResponse {
144
+ id: number
145
+ name: string
146
+ index_pattern: string
147
+ repository: string
148
+ enabled: boolean
149
+ snapshot_prefix: string
150
+ include_global_state: boolean
151
+ skip_write_indices: boolean
152
+ retention_days?: number | null
153
+ last_execution_time?: string | null
154
+ last_snapshot_name?: string | null
155
+ last_execution_status?: string | null
156
+ created_at: string
157
+ updated_at: string
158
+}
159
+
160
+export interface SnapshotScheduleListResponse {
161
+ schedules: SnapshotScheduleResponse[]
162
+ success: boolean
163
+ message: string
164
+}
165
+
166
+export interface SnapshotScheduleOperationResponse {
167
+ schedule?: SnapshotScheduleResponse | null
168
+ success: boolean
169
+ message: string
170
+}
171
+
172
+export interface ScheduledSnapshotExecutionResponse {
173
+ schedule_id: number
174
+ schedule_name: string
175
+ snapshot_name?: string | null
176
+ indices_snapshotted: string[]
177
+ skipped_write_indices: string[]
178
+ already_snapshotted_indices: string[]
179
+ success: boolean
180
+ message: string
181
+}
frontend/src/views/Snapshots.vue
new
+35
@@ -0,0 +1,35 @@
1
+<template>
2
+ <div class="page-wrapper flex flex-col gap-6 p-6">
3
+ <div class="flex items-center justify-between">
4
+ <h1 class="text-2xl font-bold">Snapshot & Restore</h1>
5
+ </div>
6
+
7
+ <n-tabs v-model:value="activeTab" type="line" animated>
8
+ <n-tab-pane name="repositories" tab="Repositories">
9
+ <SnapshotRepositories />
10
+ </n-tab-pane>
11
+ <n-tab-pane name="snapshots" tab="Snapshots">
12
+ <SnapshotList />
13
+ </n-tab-pane>
14
+ <n-tab-pane name="schedules" tab="Scheduled Snapshots">
15
+ <SnapshotSchedules />
16
+ </n-tab-pane>
17
+ </n-tabs>
18
+ </div>
19
+</template>
20
+
21
+<script setup lang="ts">
22
+import { NTabPane, NTabs } from "naive-ui"
23
+import { ref } from "vue"
24
+import SnapshotList from "@/components/snapshots/SnapshotList.vue"
25
+import SnapshotRepositories from "@/components/snapshots/SnapshotRepositories.vue"
26
+import SnapshotSchedules from "@/components/snapshots/SnapshotSchedules.vue"
27
+
28
+const activeTab = ref("repositories")
29
+</script>
30
+
31
+<style scoped>
32
+.page-wrapper {
33
+ min-height: 100%;
34
+}
35
+</style>