Skip to content
Open
98 changes: 98 additions & 0 deletions foreign/python/apache_iggy.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -108,11 +108,13 @@ __all__ = [
"SendMessage",
"SendMessagesConfirmation",
"SendMessagesResponse",
"SnapshotCompression",
"Stats",
"Stream",
"StreamDetails",
"StreamOptions",
"StreamPermissions",
"SystemSnapshotType",
"TcpConfig",
"TcpReconnectionConfig",
"TokenInfo",
Expand Down Expand Up @@ -1260,6 +1262,36 @@ class IggyClient:
authenticated, the user lacks the permission, or the request
fails.
"""
def snapshot(
self,
snapshot_compression: SnapshotCompression,
snapshot_types: typing.Sequence[SystemSnapshotType],
) -> collections.abc.Awaitable[bytes]:
r"""
Capture a diagnostic snapshot of the server as a compressed ZIP archive.

Requires an authenticated session whose user holds the `manage_servers`
global permission.

Args:
snapshot_compression: Archive compression method as a
`SnapshotCompression` enum.
snapshot_types: Data to include in the snapshot as a list of
`SystemSnapshotType` enums. `SystemSnapshotType.All` cannot be
combined with any other snapshot type.

Returns:
An awaitable that resolves to the ZIP archive `bytes`.

Raises:
TypeError: If `snapshot_compression` is not a `SnapshotCompression`
or `snapshot_types` contains elements other than `SystemSnapshotType`.
ValueError: If `snapshot_types` is empty.
RuntimeError: If the client is not connected, the session is not
authenticated, the user lacks the permission, `SystemSnapshotType.All` is
combined with another snapshot type, the 30-second binary
operation deadline expires, or the request fails.
"""
def describe_options(
self, scope: builtins.str
) -> collections.abc.Awaitable[list[OptionSpec]]:
Expand Down Expand Up @@ -4121,6 +4153,37 @@ class SendMessagesResponse:
receipts for persisted topics.
"""

@typing.final
class SnapshotCompression:
r"""
Compression method for server snapshots.
"""

Stored: typing.ClassVar[SnapshotCompression]
r"""
Store the file as is without compression.
"""
Deflated: typing.ClassVar[SnapshotCompression]
r"""
Compress the file using Deflate.
"""
Bzip2: typing.ClassVar[SnapshotCompression]
r"""
Compress the file using BZIP2.
"""
Zstd: typing.ClassVar[SnapshotCompression]
r"""
Compress the file using ZStandard.
"""
Lzma: typing.ClassVar[SnapshotCompression]
r"""
Compress the file using LZMA.
"""
Xz: typing.ClassVar[SnapshotCompression]
r"""
Compress the file using XZ.
"""

@typing.final
class Stats:
r"""
Expand Down Expand Up @@ -4576,6 +4639,41 @@ class StreamPermissions:
treated as `None`.
"""

@typing.final
class SystemSnapshotType:
r"""
Type of system data to include in a snapshot.
"""

FilesystemOverview: typing.ClassVar[SystemSnapshotType]
r"""
Overview of the filesystem.
"""
ProcessList: typing.ClassVar[SystemSnapshotType]
r"""
List of currently running processes.
"""
ResourceUsage: typing.ClassVar[SystemSnapshotType]
r"""
Resource usage statistics of the system.
"""
Test: typing.ClassVar[SystemSnapshotType]
r"""
Test snapshot type for development purposes.
"""
ServerLogs: typing.ClassVar[SystemSnapshotType]
r"""
Server logs.
"""
ServerConfig: typing.ClassVar[SystemSnapshotType]
r"""
Server configuration.
"""
All: typing.ClassVar[SystemSnapshotType]
r"""
Everything.
"""

@typing.final
class TcpConfig:
r"""
Expand Down
56 changes: 56 additions & 0 deletions foreign/python/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,9 @@ use crate::producer::{IggyProducer, ProducerMode, u32_param as producer_u32_para
use crate::receive_message::{PartitionContext, PollingStrategy, ReceiveMessage};
use crate::resource_options::{OptionKey, StreamOptions, TopicOptions, UserOptions};
use crate::send_message::{SendMessage, SendMessagesResponse as PySendMessagesResponse};
use crate::snapshot::{
SnapshotCompression as PySnapshotCompression, SystemSnapshotType as PySystemSnapshotType,
};
use crate::stats::Stats as PyStats;
use crate::stream::{Stream, StreamDetails};
use crate::topic::{IggyExpiry, MaxTopicSize, Topic, TopicDetails};
Expand Down Expand Up @@ -315,6 +318,59 @@ impl IggyClient {
})
}

/// Capture a diagnostic snapshot of the server as a compressed ZIP archive.
///
/// Requires an authenticated session whose user holds the `manage_servers`
/// global permission.
///
/// Args:
/// snapshot_compression: Archive compression method as a
/// `SnapshotCompression` enum.
/// snapshot_types: Data to include in the snapshot as a list of
/// `SystemSnapshotType` enums. `SystemSnapshotType.All` cannot be
/// combined with any other snapshot type.
///
/// Returns:
/// An awaitable that resolves to the ZIP archive `bytes`.
///
/// Raises:
/// TypeError: If `snapshot_compression` is not a `SnapshotCompression`
/// or `snapshot_types` contains elements other than `SystemSnapshotType`.
/// ValueError: If `snapshot_types` is empty.
/// RuntimeError: If the client is not connected, the session is not
/// authenticated, the user lacks the permission, `SystemSnapshotType.All` is
/// combined with another snapshot type, the 30-second binary
/// operation deadline expires, or the request fails.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[bytes]", imports=("collections.abc")))]
fn snapshot<'a>(
&self,
py: Python<'a>,
snapshot_compression: PySnapshotCompression,
snapshot_types: Vec<PySystemSnapshotType>,
) -> PyResult<Bound<'a, PyAny>> {
if snapshot_types.is_empty() {
return Err(PyValueError::new_err("snapshot_types must not be empty"));
}

let mut parsed_types = Vec::with_capacity(snapshot_types.len());
for snapshot_type in snapshot_types {
let rust_type = SystemSnapshotType::from(snapshot_type);
if !parsed_types.contains(&rust_type) {
parsed_types.push(rust_type);
}
}

let inner = self.inner.clone();
let compression = SnapshotCompression::from(snapshot_compression);
future_into_py(py, async move {
let snapshot = inner
.snapshot(compression, parsed_types)
Comment thread
hubcio marked this conversation as resolved.
.await
.map_err(to_runtime_error)?;
Ok(Python::attach(|py| PyBytes::new(py, &snapshot.0).unbind()))
})
}

/// Describe the option catalog for a resource scope.
///
/// This is the discovery surface for the `options` argument on
Expand Down
4 changes: 4 additions & 0 deletions foreign/python/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ mod producer;
mod receive_message;
mod resource_options;
mod send_message;
mod snapshot;
mod stats;
mod stream;
mod topic;
Expand Down Expand Up @@ -70,6 +71,7 @@ use resource_options::{
encryption_key_fingerprint,
};
use send_message::{SendMessage, SendMessagesConfirmation, SendMessagesResponse};
use snapshot::{SnapshotCompression, SystemSnapshotType};
use stats::{CacheMetrics, CacheMetricsKey, Stats};
use stream::{Stream, StreamDetails};
use topic::{IggyExpiry, MaxTopicSize, Partition, Topic, TopicDetails};
Expand Down Expand Up @@ -146,6 +148,8 @@ fn apache_iggy(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<GlobalPermissions>()?;
m.add_class::<StreamPermissions>()?;
m.add_class::<TopicPermissions>()?;
m.add_class::<SnapshotCompression>()?;
m.add_class::<SystemSnapshotType>()?;
m.add_class::<MetadataChanges>()?;
m.add_class::<MetadataChangeRecord>()?;
m.add_class::<MetadataCurrentItem>()?;
Expand Down
116 changes: 116 additions & 0 deletions foreign/python/src/snapshot.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

use iggy::prelude::{
SnapshotCompression as RustSnapshotCompression, SystemSnapshotType as RustSystemSnapshotType,
};
use pyo3::prelude::*;
use pyo3_stub_gen::derive::gen_stub_pyclass_enum;

/// Compression method for server snapshots.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[gen_stub_pyclass_enum]
#[pyclass(eq, from_py_object)]
pub enum SnapshotCompression {
/// Store the file as is without compression.
Stored,
/// Compress the file using Deflate.
Deflated,
/// Compress the file using BZIP2.
Bzip2,
/// Compress the file using ZStandard.
Zstd,
/// Compress the file using LZMA.
Lzma,
/// Compress the file using XZ.
Xz,
}

impl From<SnapshotCompression> for RustSnapshotCompression {
fn from(compression: SnapshotCompression) -> Self {
match compression {
SnapshotCompression::Stored => RustSnapshotCompression::Stored,
SnapshotCompression::Deflated => RustSnapshotCompression::Deflated,
SnapshotCompression::Bzip2 => RustSnapshotCompression::Bzip2,
SnapshotCompression::Zstd => RustSnapshotCompression::Zstd,
SnapshotCompression::Lzma => RustSnapshotCompression::Lzma,
SnapshotCompression::Xz => RustSnapshotCompression::Xz,
}
}
}

impl From<RustSnapshotCompression> for SnapshotCompression {
fn from(compression: RustSnapshotCompression) -> Self {
match compression {
RustSnapshotCompression::Stored => SnapshotCompression::Stored,
RustSnapshotCompression::Deflated => SnapshotCompression::Deflated,
RustSnapshotCompression::Bzip2 => SnapshotCompression::Bzip2,
RustSnapshotCompression::Zstd => SnapshotCompression::Zstd,
RustSnapshotCompression::Lzma => SnapshotCompression::Lzma,
RustSnapshotCompression::Xz => SnapshotCompression::Xz,
}
}
}

/// Type of system data to include in a snapshot.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[gen_stub_pyclass_enum]
#[pyclass(eq, from_py_object)]
pub enum SystemSnapshotType {
/// Overview of the filesystem.
FilesystemOverview,
/// List of currently running processes.
ProcessList,
/// Resource usage statistics of the system.
ResourceUsage,
/// Test snapshot type for development purposes.
Test,
/// Server logs.
ServerLogs,
/// Server configuration.
ServerConfig,
/// Everything.
All,
}

impl From<SystemSnapshotType> for RustSystemSnapshotType {
fn from(snapshot_type: SystemSnapshotType) -> Self {
match snapshot_type {
SystemSnapshotType::FilesystemOverview => RustSystemSnapshotType::FilesystemOverview,
SystemSnapshotType::ProcessList => RustSystemSnapshotType::ProcessList,
SystemSnapshotType::ResourceUsage => RustSystemSnapshotType::ResourceUsage,
SystemSnapshotType::Test => RustSystemSnapshotType::Test,
SystemSnapshotType::ServerLogs => RustSystemSnapshotType::ServerLogs,
SystemSnapshotType::ServerConfig => RustSystemSnapshotType::ServerConfig,
SystemSnapshotType::All => RustSystemSnapshotType::All,
}
}
}

impl From<RustSystemSnapshotType> for SystemSnapshotType {
fn from(snapshot_type: RustSystemSnapshotType) -> Self {
match snapshot_type {
RustSystemSnapshotType::FilesystemOverview => SystemSnapshotType::FilesystemOverview,
RustSystemSnapshotType::ProcessList => SystemSnapshotType::ProcessList,
RustSystemSnapshotType::ResourceUsage => SystemSnapshotType::ResourceUsage,
RustSystemSnapshotType::Test => SystemSnapshotType::Test,
RustSystemSnapshotType::ServerLogs => SystemSnapshotType::ServerLogs,
RustSystemSnapshotType::ServerConfig => SystemSnapshotType::ServerConfig,
RustSystemSnapshotType::All => SystemSnapshotType::All,
}
}
}
Loading