Building Intelligent RAN: O-RAN and RIC Architecture Deep Dive

┌─────────────────────────────────────────────────────────────────────────┐
│              TRADITIONAL RAN vs O-RAN ARCHITECTURE                       │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  TRADITIONAL RAN (Integrated, Proprietary):                             │
│  ──────────────────────────────────────────                             │
│                                                                          │
│  ┌────────────────────────────────────────────────────────────┐         │
│  │                                                             │         │
│  │              VENDOR-SPECIFIC BASE STATION                   │         │
│  │                                                             │         │
│  │  ┌──────────────────────────────────────────────────────┐  │         │
│  │  │  BBU (Baseband Unit)                                 │  │         │
│  │  │  • L2/L3 processing                                  │  │         │
│  │  │  • Scheduling, handover                              │  │         │
│  │  │  • Proprietary software                              │  │         │
│  │  └──────────────────┬───────────────────────────────────┘  │         │
│  │                     │ Proprietary interface                │         │
│  │                     ▼                                       │         │
│  │  ┌──────────────────────────────────────────────────────┐  │         │
│  │  │  RRU (Remote Radio Unit)                             │  │         │
│  │  │  • RF processing                                     │  │         │
│  │  │  • Proprietary hardware                              │  │         │
│  │  └──────────────────────────────────────────────────────┘  │         │
│  │                                                             │         │
│  └─────────────────────────────────────────────────────────────┘         │
│                                                                          │
│  ❌ Single vendor lock-in                                               │
│  ❌ Expensive, specialized hardware                                     │
│  ❌ Slow innovation cycles                                              │
│  ❌ Limited programmability                                             │
│                                                                          │
│  ─────────────────────────────────────────────────────────────────────  │
│                                                                          │
│  O-RAN (Disaggregated, Open):                                           │
│  ─────────────────────────────                                          │
│                                                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    SMO (Service Management                       │   │
│  │                   & Orchestration Framework)                     │   │
│  │                                                                  │   │
│  │  ┌────────────────────────────────────────────────────────────┐ │   │
│  │  │         NON-REAL-TIME RIC                                  │ │   │
│  │  │         (rApps: >1s latency)                               │ │   │
│  │  │  • Traffic prediction                                      │ │   │
│  │  │  • Network slicing orchestration                           │ │   │
│  │  │  • Policy management                                       │ │   │
│  │  └──────────────────┬─────────────────────────────────────────┘ │   │
│  │                     │ A1 interface (policies, ML models)        │   │
│  │                     │ O1 interface (management)                 │   │
│  └─────────────────────┼───────────────────────────────────────────┘   │
│                        │                                               │
│                        ▼                                               │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │               NEAR-REAL-TIME RIC                                 │   │
│  │               (xApps: 10ms-1s latency)                          │   │
│  │                                                                  │   │
│  │  • Load balancing                                               │   │
│  │  • Beamforming optimization                                     │   │
│  │  • Handover control                                             │   │
│  │  • QoS management                                               │   │
│  └──────────────────┬───────────────────────────────────────────────┘   │
│                     │ E2 interface (control, telemetry)                │
│                     ▼                                                   │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                   O-RAN COMPONENTS                               │   │
│  │                                                                  │   │
│  │  ┌──────────────┐          ┌──────────────┐                    │   │
│  │  │   O-CU-CP    │          │   O-CU-UP    │                    │   │
│  │  │  (Control)   │◄────────►│  (User Data) │                    │   │
│  │  └──────┬───────┘          └──────┬───────┘                    │   │
│  │         │ F1-C                    │ F1-U                        │   │
│  │         └─────────┬───────────────┘                             │   │
│  │                   ▼                                              │   │
│  │         ┌─────────────────────┐                                 │   │
│  │         │       O-DU           │                                 │   │
│  │         │  (Distributed Unit)  │                                 │   │
│  │         └──────────┬───────────┘                                 │   │
│  │                    │ Open Fronthaul                              │   │
│  │                    ▼                                              │   │
│  │         ┌─────────────────────┐                                 │   │
│  │         │       O-RU           │                                 │   │
│  │         │   (Radio Unit)       │                                 │   │
│  │         └──────────────────────┘                                 │   │
│  │                                                                  │   │
│  └──────────────────────────────────────────────────────────────────┘   │
│                                                                          │
│  ✅ Multi-vendor interoperability                                       │
│  ✅ COTS hardware (cost reduction)                                      │
│  ✅ Rapid innovation (xApps/rApps)                                      │
│  ✅ AI-native intelligence (RIC)                                        │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────┐
│              NEAR-REAL-TIME RIC ARCHITECTURE                             │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                        xApps Layer                               │   │
│  │                                                                  │   │
│  │  ┌───────────┐  ┌───────────┐  ┌───────────┐  ┌───────────┐   │   │
│  │  │  xApp 1   │  │  xApp 2   │  │  xApp 3   │  │  xApp N   │   │   │
│  │  │   Load    │  │   Beam    │  │ Handover  │  │   QoS     │   │   │
│  │  │ Balancing │  │  Forming  │  │  Control  │  │  Manager  │   │   │
│  │  └─────┬─────┘  └─────┬─────┘  └─────┬─────┘  └─────┬─────┘   │   │
│  │        │              │              │              │          │   │
│  └────────┼──────────────┼──────────────┼──────────────┼──────────┘   │
│           │              │              │              │              │
│           └──────────────┴──────────────┴──────────────┘              │
│                                 │                                      │
│                                 ▼                                      │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    RIC Platform Services                         │   │
│  │                                                                  │   │
│  │  ┌───────────────────────────────────────────────────────────┐  │   │
│  │  │  Subscription Manager                                     │  │   │
│  │  │  • xApps register interest in specific E2 messages       │  │   │
│  │  │  • Route messages to subscribed xApps                    │  │   │
│  │  └───────────────────────────────────────────────────────────┘  │   │
│  │                                                                  │   │
│  │  ┌───────────────────────────────────────────────────────────┐  │   │
│  │  │  SDL (Shared Data Layer)                                  │  │   │
│  │  │  • Redis-based storage for xApp data sharing             │  │   │
│  │  │  • Namespaces for isolation                              │  │   │
│  │  │  • Key-value and timeseries storage                      │  │   │
│  │  └───────────────────────────────────────────────────────────┘  │   │
│  │                                                                  │   │
│  │  ┌───────────────────────────────────────────────────────────┐  │   │
│  │  │  Conflict Resolution                                      │  │   │
│  │  │  • Arbitrate when multiple xApps send conflicting cmds   │  │   │
│  │  │  • Priority-based or policy-based resolution             │  │   │
│  │  └───────────────────────────────────────────────────────────┘  │   │
│  │                                                                  │   │
│  │  ┌───────────────────────────────────────────────────────────┐  │   │
│  │  │  xApp SDK & Lifecycle Management                         │  │   │
│  │  │  • xApp onboarding, deployment, updates                  │  │   │
│  │  │  • Health monitoring, auto-restart                       │  │   │
│  │  │  • Resource allocation (CPU, memory)                     │  │   │
│  │  └───────────────────────────────────────────────────────────┘  │   │
│  │                                                                  │   │
│  └──────────────────────┬───────────────────────────────────────────┘   │
│                         │                                               │
│                         ▼                                               │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    E2 Termination                                │   │
│  │                                                                  │   │
│  │  • E2AP (Application Protocol) handler                          │   │
│  │  • E2SM (Service Model) implementations                         │   │
│  │    - E2SM-KPM: Key Performance Measurements                     │   │
│  │    - E2SM-RC: RAN Control                                       │   │
│  │    - E2SM-NI: Network Interface                                 │   │
│  │  • SCTP connection management                                   │   │
│  │                                                                  │   │
│  └──────────────────────┬───────────────────────────────────────────┘   │
│                         │ E2 Interface                                  │
│                         ▼                                               │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    RAN Nodes (O-DU, O-CU)                        │   │
│  └──────────────────────────────────────────────────────────────────┘   │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

E2 Interface = E2AP (Application Protocol) + E2SM (Service Models)

# Conceptual E2AP message structure
class E2SetupRequest:
    """RAN node announces capabilities to RIC"""
    def __init__(self, ran_function_list, e2_node_id):
        self.ran_function_list = ran_function_list  # Supported E2SMs
        self.e2_node_id = e2_node_id                # Unique identifier
        self.plmn_id = plmn_id                      # Operator ID

class E2SetupResponse:
    """RIC acknowledges and provides connection info"""
    def __init__(self, transaction_id, accepted_functions):
        self.transaction_id = transaction_id
        self.accepted_functions = accepted_functions

class RICSubscriptionRequest:
    """xApp requests telemetry subscription"""
    def __init__(self, request_id, ran_function_id, event_triggers, actions):
        self.request_id = request_id
        self.ran_function_id = ran_function_id      # Which E2SM
        self.event_triggers = event_triggers        # When to send reports
        self.actions = actions                      # What data to include

class RICIndication:
    """RAN sends telemetry to RIC"""
    def __init__(self, request_id, indication_header, indication_message):
        self.request_id = request_id
        self.indication_header = indication_header  # Metadata
        self.indication_message = indication_message  # Actual KPIs

class RICControlRequest:
    """RIC sends control command"""
    def __init__(self, request_id, ran_function_id, control_header, control_message):
        self.request_id = request_id
        self.ran_function_id = ran_function_id
        self.control_header = control_header        # Metadata
        self.control_message = control_message      # Control parameters

# E2SM-KPM Indication Message structure
class E2SM_KPM_IndicationMessage:
    def __init__(self, measurement_records):
        self.measurement_records = measurement_records

class MeasurementRecord:
    def __init__(self, cell_id, timestamp, kpis):
        self.cell_id = cell_id
        self.timestamp = timestamp
        self.kpis = {
            'dl_prb_usage': 0.75,         # 75% utilization
            'ul_prb_usage': 0.60,         # 60% utilization
            'num_active_ues': 45,
            'avg_dl_throughput_mbps': 150.3,
            'avg_ul_throughput_mbps': 75.1,
            'packet_loss_rate': 0.001,    # 0.1%
        }

# xApp processes KPM data
def process_kpm_indication(indication_msg):
    """xApp callback for KPM indications"""
    for record in indication_msg.measurement_records:
        cell_id = record.cell_id
        prb_usage = record.kpis['dl_prb_usage']

        if prb_usage > 0.85:  # Overload threshold
            # Trigger load balancing action
            trigger_load_balancing(cell_id)

# E2SM-RC Control Message
class E2SM_RC_ControlMessage:
    def __init__(self, control_action, target, parameters):
        self.control_action = control_action  # e.g., "UPDATE_HANDOVER_PARAMS"
        self.target = target                  # e.g., specific UE or cell
        self.parameters = parameters          # Action-specific params

# Example: xApp sends handover control
def optimize_handover(ue_id, source_cell, target_cell):
    """xApp sends handover optimization command"""
    control_msg = E2SM_RC_ControlMessage(
        control_action="HANDOVER",
        target={"ue_id": ue_id, "source_cell": source_cell},
        parameters={
            "target_cell": target_cell,
            "handover_type": "SEAMLESS",
            "target_pci": 123,  # Physical Cell ID
            "offset": 3         # Handover offset (dB)
        }
    )

    ric_sdk.send_control_request(
        ran_function_id=E2SM_RC_ID,
        control_message=control_msg
    )

import logging
from ricxappframe.xapp_frame import Xapp, RMRXapp
from ricxappframe.xapp_sdl import SDLWrapper

# xApp configuration
xapp_name = "load-balancer"
logger = logging.getLogger(xapp_name)

class LoadBalancingXApp(RMRXapp):
    """Production-grade load balancing xApp"""

    def __init__(self):
        super().__init__(
            default_handler=self.default_handler,
            config_handler=self.config_change_handler
        )

        # Initialize shared data layer
        self.sdl = SDLWrapper(namespace=xapp_name)

        # Configuration
        self.high_load_threshold = 0.85  # 85% PRB usage
        self.low_load_threshold = 0.30   # 30% PRB usage
        self.handover_hysteresis = 0.10

        # State: cell load tracking
        self.cell_loads = {}  # {cell_id: prb_utilization}
        self.neighbor_graph = {}  # {cell_id: [neighbor_ids]}

        logger.info(f"{xapp_name} initialized")

    def start(self):
        """Start xApp lifecycle"""
        # Subscribe to E2SM-KPM indications
        self.subscribe_kpm()

        # Start periodic tasks
        self.run()

    def subscribe_kpm(self):
        """Subscribe to KPM reports from all cells"""
        subscription_req = {
            'ran_function_id': 2,  # E2SM-KPM
            'event_triggers': {
                'type': 'PERIODIC',
                'period_ms': 1000  # 1 second reporting
            },
            'actions': [{
                'action_id': 1,
                'action_type': 'REPORT',
                'measurement_types': [
                    'DL_PRB_Usage',
                    'UL_PRB_Usage',
                    'Num_Active_UEs'
                ]
            }]
        }

        # Send subscription via RIC platform
        self.subscribe(subscription_req)
        logger.info("Subscribed to KPM reports")

    def default_handler(self, summary, sbuf):
        """Handle incoming E2 messages"""
        # Parse E2 indication
        indication = self.parse_indication(sbuf)

        if indication['type'] == 'KPM_INDICATION':
            self.handle_kpm_indication(indication)

    def handle_kpm_indication(self, indication):
        """Process KPM measurement reports"""
        for measurement in indication['measurements']:
            cell_id = measurement['cell_id']
            prb_usage = measurement['metrics']['DL_PRB_Usage']
            num_ues = measurement['metrics']['Num_Active_UEs']

            # Update cell load state
            self.cell_loads[cell_id] = prb_usage

            # Store in shared data layer (for other xApps)
            self.sdl.set(f"cell:{cell_id}:load", prb_usage)
            self.sdl.set(f"cell:{cell_id}:num_ues", num_ues)

            logger.debug(f"Cell {cell_id}: PRB={prb_usage:.2f}, UEs={num_ues}")

            # Check if action needed
            if prb_usage > self.high_load_threshold:
                self.handle_overload(cell_id)

    def handle_overload(self, overloaded_cell):
        """Trigger load balancing for overloaded cell"""
        logger.warning(f"Cell {overloaded_cell} overloaded: "
                      f"{self.cell_loads[overloaded_cell]:.2f}")

        # Find best target cell
        target_cell = self.find_best_target(overloaded_cell)

        if target_cell is not None:
            # Send control action to adjust handover parameters
            self.adjust_handover_params(overloaded_cell, target_cell)
        else:
            logger.warning(f"No suitable target for {overloaded_cell}")

    def find_best_target(self, source_cell):
        """Select best neighbor cell for offloading"""
        neighbors = self.neighbor_graph.get(source_cell, [])
        if not neighbors:
            return None

        # Find least loaded neighbor
        best_target = None
        min_load = float('inf')

        for neighbor in neighbors:
            load = self.cell_loads.get(neighbor, 1.0)
            if load < self.low_load_threshold and load < min_load:
                min_load = load
                best_target = neighbor

        return best_target

    def adjust_handover_params(self, source_cell, target_cell):
        """Send E2SM-RC control to modify handover parameters"""
        # Reduce handover threshold for target cell
        # (make handovers to target easier)
        control_msg = {
            'ran_function_id': 3,  # E2SM-RC
            'control_action': 'UPDATE_HANDOVER_PARAMS',
            'target': {
                'source_cell': source_cell,
                'target_cell': target_cell
            },
            'parameters': {
                'A3_offset': -3,  # Reduce by 3 dB (easier handover)
                'time_to_trigger': 160,  # ms
                'hysteresis': self.handover_hysteresis * 10  # Convert to 0.5dB units
            }
        }

        # Send control request
        self.send_control_request(control_msg)

        logger.info(f"Adjusted handover: {source_cell} → {target_cell}")

        # Store action in SDL for monitoring
        self.sdl.set(f"action:{source_cell}:{target_cell}:timestamp",
                     time.time())

    def config_change_handler(self, config):
        """Handle configuration updates from Non-RT RIC"""
        logger.info(f"Configuration updated: {config}")
        self.high_load_threshold = config.get('high_load_threshold', 0.85)
        self.low_load_threshold = config.get('low_load_threshold', 0.30)

    def stop(self):
        """Cleanup on shutdown"""
        logger.info(f"{xapp_name} shutting down")
        self.sdl.close()

# Entry point
if __name__ == "__main__":
    xapp = LoadBalancingXApp()
    xapp.start()

# Kubernetes Helm chart for xApp deployment
apiVersion: v1
kind: ConfigMap
metadata:
  name: load-balancer-xapp-config
data:
  config.json: |
    {
      "xapp_name": "load-balancer",
      "version": "1.0.0",
      "high_load_threshold": 0.85,
      "low_load_threshold": 0.30,
      "rmr": {
        "protPort": "tcp:4560",
        "maxSize": 2072,
        "numWorkers": 1
      }
    }
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: load-balancer-xapp
spec:
  replicas: 1
  selector:
    matchLabels:
      app: load-balancer-xapp
  template:
    metadata:
      labels:
        app: load-balancer-xapp
    spec:
      containers:
      - name: load-balancer
        image: oran/load-balancer-xapp:1.0.0
        ports:
        - containerPort: 4560
          name: rmr
        env:
        - name: CONFIG_FILE
          value: "/config/config.json"
        - name: SDL_NAMESPACE
          value: "load-balancer"
        resources:
          limits:
            cpu: "2"
            memory: "2Gi"
          requests:
            cpu: "1"
            memory: "1Gi"
        volumeMounts:
        - name: config
          mountPath: /config
      volumes:
      - name: config
        configMap:
          name: load-balancer-xapp-config

import torch
import torch.nn as nn
import numpy as np
from datetime import datetime, timedelta

class TrafficForecastingRApp:
    """rApp for traffic prediction and proactive resource management"""

    def __init__(self, non_rt_ric_sdk):
        self.ric = non_rt_ric_sdk
        self.model = TrafficForecaster(
            input_dim=5,  # Features: hour, day_of_week, holiday, historical_load, events
            hidden_dim=128,
            num_layers=2,
            horizon=96  # Predict next 24 hours (15min intervals)
        )

    def collect_training_data(self, days=30):
        """Fetch historical traffic data from RIC"""
        end_time = datetime.now()
        start_time = end_time - timedelta(days=days)

        # Query historical KPIs
        traffic_data = self.ric.query_historical_data(
            metric='prb_utilization',
            start_time=start_time,
            end_time=end_time,
            aggregation='15min',
            per_cell=True
        )

        return traffic_data

    def train_model(self):
        """Train forecasting model"""
        # Collect data
        data = self.collect_training_data(days=30)

        # Prepare features and targets
        X, y = self.prepare_features(data)

        # Train/val split
        split_idx = int(0.8 * len(X))
        X_train, X_val = X[:split_idx], X[split_idx:]
        y_train, y_val = y[:split_idx], y[split_idx:]

        # Training loop
        optimizer = torch.optim.Adam(self.model.parameters(), lr=1e-3)
        criterion = nn.MSELoss()

        for epoch in range(100):
            self.model.train()
            optimizer.zero_grad()

            y_pred = self.model(X_train)
            loss = criterion(y_pred, y_train)

            loss.backward()
            optimizer.step()

            # Validation
            if epoch % 10 == 0:
                self.model.eval()
                with torch.no_grad():
                    val_pred = self.model(X_val)
                    val_loss = criterion(val_pred, y_val)
                    print(f"Epoch {epoch}, Train Loss: {loss.item():.4f}, "
                          f"Val Loss: {val_loss.item():.4f}")

        # Evaluate and deploy if good
        mape = self.compute_mape(y_val, val_pred)
        if mape < 0.15:  # <15% error
            self.deploy_model()
        else:
            print(f"Model performance insufficient (MAPE={mape:.2f}), not deploying")

    def deploy_model(self):
        """Deploy trained model to Near-RT RIC"""
        # Export model
        model_path = "/tmp/traffic_forecaster.onnx"
        dummy_input = torch.randn(1, 5)
        torch.onnx.export(self.model, dummy_input, model_path)

        # Upload to RIC
        self.ric.deploy_model(
            model_path=model_path,
            target='near_rt_ric',
            model_id='traffic_forecast_v1.5',
            target_xapps=['capacity-manager', 'energy-optimizer']
        )

        print("Model deployed successfully")

    def generate_predictions(self):
        """Generate traffic forecasts and send policies to Near-RT RIC"""
        # Get current state
        current_data = self.ric.get_current_state()

        # Prepare features
        features = self.extract_features(current_data)

        # Predict next 24 hours
        self.model.eval()
        with torch.no_grad():
            predictions = self.model(features)  # (num_cells, 96)

        # Analyze predictions and send policies
        for cell_id, pred in predictions.items():
            # Find peak hour
            peak_idx = np.argmax(pred)
            peak_time = datetime.now() + timedelta(minutes=15*peak_idx)
            peak_load = pred[peak_idx]

            if peak_load > 0.9:  # Predicted overload
                # Send proactive policy to Near-RT RIC
                self.ric.send_policy(
                    interface='A1',
                    policy_type='CAPACITY_EXPANSION',
                    target_cell=cell_id,
                    parameters={
                        'activate_at': peak_time.isoformat(),
                        'duration_minutes': 60,
                        'actions': [
                            'activate_carrier_aggregation',
                            'enable_aggressive_load_balancing'
                        ]
                    }
                )

                print(f"Cell {cell_id}: Peak load {peak_load:.2f} at {peak_time}, "
                      f"policy sent")

    def run(self):
        """Main rApp loop"""
        # Initial training
        self.train_model()

        # Periodic tasks
        while True:
            # Generate predictions every hour
            self.generate_predictions()

            # Retrain weekly
            if datetime.now().weekday() == 0:  # Monday
                self.train_model()

            time.sleep(3600)  # 1 hour

class TrafficForecaster(nn.Module):
    """LSTM-based traffic forecasting model"""
    def __init__(self, input_dim, hidden_dim, num_layers, horizon):
        super().__init__()
        self.lstm = nn.LSTM(input_dim, hidden_dim, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_dim, horizon)

    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        prediction = self.fc(lstm_out[:, -1, :])
        return prediction

┌─────────────────────────────────────────────────────────────────────────┐
│              PRODUCTION RIC DEPLOYMENT                                   │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  CLOUD TIER (Non-RT RIC):                                               │
│  ──────────────────────────                                             │
│  • Multi-region Kubernetes cluster                                      │
│  • High availability: 3+ replicas per rApp                              │
│  • Auto-scaling based on load                                           │
│  • S3/GCS for model storage                                             │
│  • Kafka/Pulsar for event streaming                                     │
│  • Grafana/Prometheus for monitoring                                    │
│                                                                          │
│  Hardware: 8-16 vCPUs, 32-64GB RAM per rApp                            │
│            1-4 GPUs for training (A100, H100)                           │
│                                                                          │
│  EDGE TIER (Near-RT RIC):                                               │
│  ───────────────────────                                                │
│  • Regional edge datacenters (co-located with mobile core)              │
│  • Kubernetes for xApp orchestration                                    │
│  • Low-latency storage (Redis, NVMe SSD)                                │
│  • Dedicated GPU for ML inference (T4, A10, L4)                         │
│  • High-bandwidth connection to RAN (25-100 Gbps)                       │
│                                                                          │
│  Hardware: 16-32 vCPUs, 64-128GB RAM                                    │
│            1-2 inference GPUs                                            │
│                                                                          │
│  LATENCY REQUIREMENTS:                                                   │
│  • Near-RT RIC ↔ O-DU/O-CU: <10ms                                      │
│  • xApp processing: <100ms                                              │
│  • E2 message round-trip: <1s                                           │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

# Network policy for xApp isolation
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: xapp-isolation
spec:
  podSelector:
    matchLabels:
      app: xapp
  policyTypes:
  - Ingress
  - Egress
  ingress:
  - from:
    - podSelector:
        matchLabels:
          app: ric-platform
    ports:
    - protocol: TCP
      port: 4560  # RMR
  egress:
  - to:
    - podSelector:
        matchLabels:
          app: sdl-redis
    ports:
    - protocol: TCP
      port: 6379
  - to:
    - podSelector:
        matchLabels:
          app: e2-termination
    ports:
    - protocol: SCTP
      port: 36421

# Prometheus metrics for xApp
from prometheus_client import Counter, Histogram, Gauge

# Metrics
e2_messages_received = Counter('xapp_e2_messages_received_total',
                                'Total E2 messages received',
                                ['message_type'])
control_actions_sent = Counter('xapp_control_actions_sent_total',
                                'Total control actions sent',
                                ['action_type', 'target_cell'])
processing_latency = Histogram('xapp_processing_latency_seconds',
                                'Message processing latency')
current_load = Gauge('xapp_cell_load', 'Current cell PRB utilization',
                     ['cell_id'])

# Instrument xApp code
@processing_latency.time()
def handle_kpm_indication(indication):
    e2_messages_received.labels(message_type='KPM_INDICATION').inc()

    for measurement in indication['measurements']:
        cell_id = measurement['cell_id']
        prb_usage = measurement['metrics']['DL_PRB_Usage']

        current_load.labels(cell_id=cell_id).set(prb_usage)

        if prb_usage > threshold:
            send_control_action(cell_id)
            control_actions_sent.labels(
                action_type='LOAD_BALANCE',
                target_cell=cell_id
            ).inc()

# GitOps pipeline for xApp updates
apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: load-balancer-xapp
spec:
  project: default
  source:
    repoURL: https://github.com/operator/xapps
    targetRevision: HEAD
    path: load-balancer
  destination:
    server: https://kubernetes.default.svc
    namespace: ric
  syncPolicy:
    automated:
      prune: true
      selfHeal: true
    syncOptions:
    - CreateNamespace=true