br_rapp_sdk.monitoring_services.monitoring_types

ServiceModel Objects

class ServiceModel(BaseModel)

Represents a service model to be monitored within an O-RAN monitoring deployment.

Attributes:

• name - The type of service model to monitor. Must be one of the following: "KPM", "MAC", "RLC", "PDCP", "GTP", "SLICE", or "TC".
• periodicity - The monitoring interval, specified in milliseconds. Supported values include: "1_ms", "2_ms", "5_ms", "10_ms", "100_ms", and "1000_ms".
• metrics - An optional list of metric names to monitor for the selected service model. If omitted, all available metrics may be monitored by default.

MonitoringStatements Objects

class MonitoringStatements(BaseModel)

Represents the monitoring statements that define the parameters for monitoring.

Attributes:

• service_models List[ServiceModel] - List of service models to be monitored.
• database Optional[str] - Optional database backend (currently only "SQL" allowed).
• environment_variables Optional[Dict[str, str]] - Key-value pairs of environment variables to set in the monitoring container. Keys are written in standard lowercase YAML/JSON style but will be converted to uppercase with underscores when injected into the container's environment.
• extra_config_annotation Optional[Dict[str, str]] - Extra annotations for configuration.
• profiles Optional[List[str]] - Active profiles in the monitoring deployment.

MonitoringObject Objects

class MonitoringObject(SnakeModel)

Represents a monitoring object containing scope and monitoring statements.

Attributes:

• scope_identifier - The identifier for the scope of the monitoring.
• monitoring_statements - The statements that define the monitoring parameters.

MonitoringObjectInformation Objects

class MonitoringObjectInformation(SnakeModel)

Represents the information of a monitoring object, including its target, type, and the actual monitoring object.

Attributes:

• monitoring_id - The unique identifier of the monitoring.
• monitoring_type_id - The type of the monitoring.

br_rapp_sdk.monitoring_services.monitoring_services

MonitoringServices Objects

class MonitoringServices()

This class provides methods to interact with the monitoring services API in BubbleRAN environment.

MonitoringServices allows you to create, list, and delete monitoring jobs in the BubbleRAN environment.

Attributes:

• kubeconfig_path Optional[str] - Path to the kubeconfig file.
• namespace str - Kubernetes namespace for Monitoring jobs.

Example:

from br_rapp_sdk import MonitoringServices
from br_rapp_sdk.monitoring_services.monitoring_types import MonitoringJob, MonitoringJobSpec

# Initialize the MonitoringServices client
monitoring_services = MonitoringServices("/path/to/kubeconfig")

# list all monitorings
result = monitoring_services.list_monitorings()
if result.status == 'success':
    for monitoring_id, monitoring_obj in result.data.get("items", []):
        print(f"Monitoring ID: {monitoring_id}, Object: {monitoring_obj}")
else:
    print(f"Error listing monitorings: {result.error}")
# Get a specific monitoring job
monitoring_id = MonitoringId("example-monitoring")
result = monitoring_services.get_monitoring(monitoring_id)
if result.status == 'success':
    monitoring_obj = result.data.get("item")
    print(f"Monitoring Job: {monitoring_obj}")
else:
    print(f"Error retrieving monitoring job: {result.error}")

__init__

def __init__(kubeconfig_path: str = None, namespace: str = "trirematics")

Initializes the MonitoringServices client.

Arguments:

• kubeconfig_path str - Path to the kubeconfig file. If None, uses the default kubeconfig.
• namespace str - Kubernetes namespace for Monitoring jobs. Default is "trirematics".

Raises:

• RuntimeError - If kubeconfig cannot be loaded or if there are issues with the API.

list_monitorings

def list_monitorings(
        monitoring_id: Optional[MonitoringId] = None
) -> KubectlOperationResult

This method lists all Monitorings in the Monitoring Services API.

Arguments:

• monitoring_id Optional[MonitoringId] - ID of the monitoring job to filter by. If None, lists all jobs.

Returns:

• KubectlOperationResult - An object representing the result of the operation, containing a list of MonitoringId and MonitoringObjectInformation tuples if successful, or an error message if not.

Example:

from br_rapp_sdk import MonitoringServices
monitoring_services = MonitoringServices()
result = monitoring_services.list_monitorings()
if result.status == 'success':
    for monitoring_id, monitoring_obj in result.data:
        print(f"Monitoring ID: {monitoring_id}")
else:
    print(f"Error listing monitorings: {result.error}")

get_monitoring

def get_monitoring(monitoring_id: MonitoringId) -> KubectlOperationResult

This method retrieves a specific Monitoring job by its ID.

Arguments:

• monitoring_id MonitoringId - ID of the monitoring job to retrieve.

Returns:

• KubectlOperationResult - An object representing the result of the operation, containing the MonitoringObjectInformation if successful, or an error message if not.

Example:

from br_rapp_sdk import MonitoringServices
from br_rapp_sdk.monitoring_services.monitoring_types import MonitoringId
monitoring_services = MonitoringServices()
result = monitoring_services.get_monitoring(MonitoringId("example-monitoring"))
if result.status == 'success':
    monitoring_obj = result.data.get('item')
    print(f"Monitoring Job: {monitoring_obj}")
else:
    print(f"Error retrieving monitoring job: {result.error}")

apply_monitoring

def apply_monitoring(
        monitoring_name: str, monitoring_object: MonitoringObjectInformation
) -> KubectlOperationResult

This method applies a Monitoring job to the Monitoring Services API.

Arguments:

• monitoring_object MonitoringObjectInformation - The Monitoring job object to apply. It should contain the necessary specifications for the monitoring job.

Returns:

• KubectlOperationResult - An object representing the result of the operation, containing the applied MonitoringId if successful, or an error message if not.

Example:

from br_rapp_sdk import MonitoringServices
from br_rapp_sdk.monitoring_services.monitoring_types import MonitoringObjectInformation, MonitoringTypeId, TargetId, MonitoringObject  
monitoring_services = MonitoringServices()
monitoring_job = MonitoringObjectInformation(
    target=TargetId("ric-name.network-name"),
    monitoring_type_id =MonitoringTypeId(name="cm/example"),
    monitoring_object = MonitoringObject( ... )
)
monitoring_name = "example-monitoring"
result = monitoring_services.apply_monitoring(monitoring_name, monitoring_job)
if result.status == 'success':
    print(f"Monitoring Job applied successfully: {result.data.get('monitoring_id')}")
else:
    print(f"Error applying monitoring job: {result.error}")

delete_monitoring

def delete_monitoring(monitoring_id: MonitoringId) -> KubectlOperationResult

This method deletes a specific Monitoring job by its ID.

Arguments:

• monitoring_id MonitoringId - ID of the monitoring job to delete.

Returns:

• KubectlOperationResult - An object representing the result of the operation, containing a success message if successful, or an error message if not.

Example:

from br_rapp_sdk import MonitoringServices
from br_rapp_sdk.monitoring_services.monitoring_types import MonitoringId
monitoring_services = MonitoringServices()
result = monitoring_services.delete_monitoring(MonitoringId("example-monitoring"))
if result.status == 'success':
    print("Monitoring Job deleted successfully.")
else:
    print(f"Error deleting monitoring job: {result.error}")

get_monitoring_status

def get_monitoring_status(
        monitoring_id: MonitoringId) -> KubectlOperationResult

This method retrieves the status of a specific Monitoring job by its ID.

Arguments:

• monitoring_id MonitoringId - ID of the monitoring job to retrieve the status for.

Returns:

• KubectlOperationResult - An object representing the result of the operation, containing the monitoring status if successful, or an error message if not.

Example:

from br_rapp_sdk import MonitoringServices
from br_rapp_sdk.monitoring_services.monitoring_types import MonitoringId
monitoring_services = MonitoringServices()
result = monitoring_services.get_monitoring_status(MonitoringId("example-monitoring"))
if result.status == 'success':
    print(f"Monitoring Job Status: {result.data.get('status')}")
else:
    print(f"Error retrieving monitoring job status: {result.error}")