Skip to content
Snippets Groups Projects
Select Git revision
  • 0addb9c70dfaadf04998fb6d74a2c8b7d86d68bc
  • master default
  • feat/331-bitnami-kafka-moved-to-bitnamilegacy-kafka
  • feat/330-tid-pcep-component
  • feat/tid-newer-pcep-component
  • feat/policy-refactor
  • feat/314-tid-new-service-for-ipowdm-configuration-fron-orchestrator-to-ipowdm-controller
  • feat/329-ecoc-2025-hack-your-research
  • develop protected
  • feat/116-ubi-updates-in-telemetry-backend-to-support-p4-in-band-network-telemetry
  • feat/320-cttc-ietf-simap-basic-support-with-kafka-yang-push
  • feat/310-cttc-implement-nbi-connector-to-interface-with-osm-client
  • feat/307-update-python-version-service
  • feat/292-cttc-implement-integration-test-for-ryu-openflow
  • cnit_tapi
  • feat/327-tid-new-service-to-ipowdm-controller-to-manage-transceivers-configuration-on-external-agent
  • cnit-p2mp-premerge
  • feat/325-tid-nbi-e2e-to-manage-e2e-path-computation
  • feat/326-tid-external-management-of-devices-telemetry-nbi
  • openroadm-flex-grid
  • CTTC-IMPLEMENT-NBI-CONNECTOR-NOS-ZTP
  • feat/324-tid-nbi-ietf_l3vpn-deploy-fail
  • v5.0.0 protected
  • v4.0.0 protected
  • demo-dpiab-eucnc2024
  • v3.0.0 protected
  • v2.1.0 protected
  • v2.0.0 protected
  • v1.0.0 protected
29 results

l3_centralizedattackdetectorServiceServicerImpl.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    InterdomainClient.py 2.96 KiB
    import grpc, logging
    from common.tools.client.RetryDecorator import retry, delay_exponential
    from common.tools.grpc.Tools import grpc_message_to_json_string
    from interdomain.proto.context_pb2 import AuthenticationResult, Slice, SliceId, SliceStatus, TeraFlowController
    from interdomain.proto.interdomain_pb2_grpc import InterdomainServiceStub
    
    LOGGER = logging.getLogger(__name__)
    MAX_RETRIES = 15
    DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0)
    RETRY_DECORATOR = retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')
    
    class InterdomainClient:
        def __init__(self, address, port):
            self.endpoint = '{:s}:{:s}'.format(str(address), str(port))
            LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))
            self.channel = None
            self.stub = None
            self.connect()
            LOGGER.debug('Channel created')
    
        def connect(self):
            self.channel = grpc.insecure_channel(self.endpoint)
            self.stub = InterdomainServiceStub(self.channel)
    
        def close(self):
            if self.channel is not None: self.channel.close()
            self.channel = None
            self.stub = None
    
        @RETRY_DECORATOR
        def RequestSlice(self, request : Slice) -> SliceId:
            LOGGER.debug('RequestSlice request: {:s}'.format(grpc_message_to_json_string(request)))
            response = self.stub.RequestSlice(request)
            LOGGER.debug('RequestSlice result: {:s}'.format(grpc_message_to_json_string(response)))
            return response
    
        @RETRY_DECORATOR
        def Authenticate(self, request : TeraFlowController) -> AuthenticationResult:
            LOGGER.debug('Authenticate request: {:s}'.format(grpc_message_to_json_string(request)))
            response = self.stub.Authenticate(request)
            LOGGER.debug('Authenticate result: {:s}'.format(grpc_message_to_json_string(response)))
            return response
    
        @RETRY_DECORATOR
        def LookUpSlice(self, request : Slice) -> SliceId:
            LOGGER.debug('LookUpSlice request: {:s}'.format(grpc_message_to_json_string(request)))
            response = self.stub.LookUpSlice(request)
            LOGGER.debug('LookUpSlice result: {:s}'.format(grpc_message_to_json_string(response)))
            return response
    
        @RETRY_DECORATOR
        def OrderSliceFromCatalog(self, request : Slice) -> SliceStatus:
            LOGGER.debug('OrderSliceFromCatalog request: {:s}'.format(grpc_message_to_json_string(request)))
            response = self.stub.OrderSliceFromCatalog(request)
            LOGGER.debug('OrderSliceFromCatalog result: {:s}'.format(grpc_message_to_json_string(response)))
            return response
    
        @RETRY_DECORATOR
        def CreateSliceAndAddToCatalog(self, request : Slice) -> SliceStatus:
            LOGGER.debug('CreateSliceAndAddToCatalog request: {:s}'.format(grpc_message_to_json_string(request)))
            response = self.stub.CreateSliceAndAddToCatalog(request)
            LOGGER.debug('CreateSliceAndAddToCatalog result: {:s}'.format(grpc_message_to_json_string(response)))
            return response