Skip to content
Snippets Groups Projects
Commit c56e0def authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

DLT connector: 

- factorized recording methods
- implemented delete of record
parent ff7d1879
No related branches found
No related tags found
2 merge requests!54Release 2.0.0,!36Performance Evaluation Framework + Helper Tools
Hide whitespace changes
Inline Side-by-side
src/dlt/connector/service/DltConnectorServiceServicerImpl.py
+ 69
− 103
View file @ c56e0def
...@@ -13,8 +13,9 @@ ...@@ -13,8 +13,9 @@
# limitations under the License. # limitations under the License.
import grpc, logging import grpc, logging
from typing import Optional
from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
from common.proto.context_pb2 import DeviceId, Empty, LinkId, ServiceId, SliceId, TopologyId from common.proto.context_pb2 import Empty, TopologyId
from common.proto.dlt_connector_pb2 import DltDeviceId, DltLinkId, DltServiceId, DltSliceId from common.proto.dlt_connector_pb2 import DltDeviceId, DltLinkId, DltServiceId, DltSliceId
from common.proto.dlt_connector_pb2_grpc import DltConnectorServiceServicer from common.proto.dlt_connector_pb2_grpc import DltConnectorServiceServicer
from common.proto.dlt_gateway_pb2 import DltRecord, DltRecordId, DltRecordOperationEnum, DltRecordTypeEnum from common.proto.dlt_gateway_pb2 import DltRecord, DltRecordId, DltRecordOperationEnum, DltRecordTypeEnum
...@@ -42,34 +43,15 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer): ...@@ -42,34 +43,15 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer):
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def RecordDevice(self, request : DltDeviceId, context : grpc.ServicerContext) -> Empty: def RecordDevice(self, request : DltDeviceId, context : grpc.ServicerContext) -> Empty:
context_client = ContextClient() data_json = None
device = context_client.GetDevice(request.device_id) if not request.delete:
context_client = ContextClient()
dltgateway_client = DltGatewayClient() device = context_client.GetDevice(request.device_id)
data_json = grpc_message_to_json_string(device)
dlt_record_id = DltRecordId()
dlt_record_id.domain_uuid.uuid = request.topology_id.topology_uuid.uuid self._record_entity(
dlt_record_id.type = DltRecordTypeEnum.DLTRECORDTYPE_DEVICE request.topology_id.topology_uuid.uuid, DltRecordTypeEnum.DLTRECORDTYPE_DEVICE,
dlt_record_id.record_uuid.uuid = device.device_id.device_uuid.uuid request.device_id.device_uuid.uuid, request.delete, data_json)
LOGGER.info('[RecordDevice] sent dlt_record_id = {:s}'.format(grpc_message_to_json_string(dlt_record_id)))
dlt_record = dltgateway_client.GetFromDlt(dlt_record_id)
LOGGER.info('[RecordDevice] recv dlt_record = {:s}'.format(grpc_message_to_json_string(dlt_record)))
exists = record_exists(dlt_record)
LOGGER.info('[RecordDevice] exists = {:s}'.format(str(exists)))
dlt_record = DltRecord()
dlt_record.record_id.CopyFrom(dlt_record_id)
dlt_record.operation = \
DltRecordOperationEnum.DLTRECORDOPERATION_UPDATE \
if exists else \
DltRecordOperationEnum.DLTRECORDOPERATION_ADD
dlt_record.data_json = grpc_message_to_json_string(device)
LOGGER.info('[RecordDevice] sent dlt_record = {:s}'.format(grpc_message_to_json_string(dlt_record)))
dlt_record_status = dltgateway_client.RecordToDlt(dlt_record)
LOGGER.info('[RecordDevice] recv dlt_record_status = {:s}'.format(grpc_message_to_json_string(dlt_record_status)))
return Empty() return Empty()
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
...@@ -78,34 +60,15 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer): ...@@ -78,34 +60,15 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer):
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def RecordLink(self, request : DltLinkId, context : grpc.ServicerContext) -> Empty: def RecordLink(self, request : DltLinkId, context : grpc.ServicerContext) -> Empty:
context_client = ContextClient() data_json = None
link = context_client.GetLink(request.link_id) if not request.delete:
context_client = ContextClient()
dltgateway_client = DltGatewayClient() link = context_client.GetLink(request.link_id)
data_json = grpc_message_to_json_string(link)
dlt_record_id = DltRecordId()
dlt_record_id.domain_uuid.uuid = request.topology_id.topology_uuid.uuid self._record_entity(
dlt_record_id.type = DltRecordTypeEnum.DLTRECORDTYPE_LINK request.topology_id.topology_uuid.uuid, DltRecordTypeEnum.DLTRECORDTYPE_LINK,
dlt_record_id.record_uuid.uuid = link.link_id.link_uuid.uuid request.link_id.link_uuid.uuid, request.delete, data_json)
LOGGER.info('[RecordLink] sent dlt_record_id = {:s}'.format(grpc_message_to_json_string(dlt_record_id)))
dlt_record = dltgateway_client.GetFromDlt(dlt_record_id)
LOGGER.info('[RecordLink] recv dlt_record = {:s}'.format(grpc_message_to_json_string(dlt_record)))
exists = record_exists(dlt_record)
LOGGER.info('[RecordLink] exists = {:s}'.format(str(exists)))
dlt_record = DltRecord()
dlt_record.record_id.CopyFrom(dlt_record_id)
dlt_record.operation = \
DltRecordOperationEnum.DLTRECORDOPERATION_UPDATE \
if exists else \
DltRecordOperationEnum.DLTRECORDOPERATION_ADD
dlt_record.data_json = grpc_message_to_json_string(link)
LOGGER.info('[RecordLink] sent dlt_record = {:s}'.format(grpc_message_to_json_string(dlt_record)))
dlt_record_status = dltgateway_client.RecordToDlt(dlt_record)
LOGGER.info('[RecordLink] recv dlt_record_status = {:s}'.format(grpc_message_to_json_string(dlt_record_status)))
return Empty() return Empty()
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
...@@ -114,34 +77,15 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer): ...@@ -114,34 +77,15 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer):
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def RecordService(self, request : DltServiceId, context : grpc.ServicerContext) -> Empty: def RecordService(self, request : DltServiceId, context : grpc.ServicerContext) -> Empty:
context_client = ContextClient() data_json = None
service = context_client.GetService(request.service_id) if not request.delete:
context_client = ContextClient()
dltgateway_client = DltGatewayClient() service = context_client.GetService(request.service_id)
data_json = grpc_message_to_json_string(service)
dlt_record_id = DltRecordId()
dlt_record_id.domain_uuid.uuid = request.topology_id.topology_uuid.uuid self._record_entity(
dlt_record_id.type = DltRecordTypeEnum.DLTRECORDTYPE_SERVICE request.topology_id.topology_uuid.uuid, DltRecordTypeEnum.DLTRECORDTYPE_SERVICE,
dlt_record_id.record_uuid.uuid = service.service_id.service_uuid.uuid request.service_id.service_uuid.uuid, request.delete, data_json)
LOGGER.info('[RecordService] sent dlt_record_id = {:s}'.format(grpc_message_to_json_string(dlt_record_id)))
dlt_record = dltgateway_client.GetFromDlt(dlt_record_id)
LOGGER.info('[RecordService] recv dlt_record = {:s}'.format(grpc_message_to_json_string(dlt_record)))
exists = record_exists(dlt_record)
LOGGER.info('[RecordService] exists = {:s}'.format(str(exists)))
dlt_record = DltRecord()
dlt_record.record_id.CopyFrom(dlt_record_id)
dlt_record.operation = \
DltRecordOperationEnum.DLTRECORDOPERATION_UPDATE \
if exists else \
DltRecordOperationEnum.DLTRECORDOPERATION_ADD
dlt_record.data_json = grpc_message_to_json_string(service)
LOGGER.info('[RecordService] sent dlt_record = {:s}'.format(grpc_message_to_json_string(dlt_record)))
dlt_record_status = dltgateway_client.RecordToDlt(dlt_record)
LOGGER.info('[RecordService] recv dlt_record_status = {:s}'.format(grpc_message_to_json_string(dlt_record_status)))
return Empty() return Empty()
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
...@@ -150,32 +94,54 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer): ...@@ -150,32 +94,54 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer):
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER) @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def RecordSlice(self, request : DltSliceId, context : grpc.ServicerContext) -> Empty: def RecordSlice(self, request : DltSliceId, context : grpc.ServicerContext) -> Empty:
context_client = ContextClient() data_json = None
slice_ = context_client.GetSlice(request.slice_id) if not request.delete:
context_client = ContextClient()
slice_ = context_client.GetSlice(request.slice_id)
data_json = grpc_message_to_json_string(slice_)
self._record_entity(
request.topology_id.topology_uuid.uuid, DltRecordTypeEnum.DLTRECORDTYPE_SLICE,
request.slice_id.slice_uuid.uuid, request.delete, data_json)
return Empty()
def _record_entity(
self, dlt_domain_uuid : str, dlt_record_type : DltRecordTypeEnum, dlt_record_uuid : str, delete : bool,
data_json : Optional[str] = None
) -> None:
dltgateway_client = DltGatewayClient() dltgateway_client = DltGatewayClient()
dlt_record_id = DltRecordId() dlt_record_id = DltRecordId()
dlt_record_id.domain_uuid.uuid = request.topology_id.topology_uuid.uuid dlt_record_id.domain_uuid.uuid = dlt_domain_uuid # pylint: disable=no-member
dlt_record_id.type = DltRecordTypeEnum.DLTRECORDTYPE_SLICE dlt_record_id.type = dlt_record_type
dlt_record_id.record_uuid.uuid = slice_.slice_id.slice_uuid.uuid dlt_record_id.record_uuid.uuid = dlt_record_uuid # pylint: disable=no-member
LOGGER.info('[RecordSlice] sent dlt_record_id = {:s}'.format(grpc_message_to_json_string(dlt_record_id))) str_dlt_record_id = grpc_message_to_json_string(dlt_record_id)
LOGGER.debug('[_record_entity] sent dlt_record_id = {:s}'.format(str_dlt_record_id))
dlt_record = dltgateway_client.GetFromDlt(dlt_record_id) dlt_record = dltgateway_client.GetFromDlt(dlt_record_id)
LOGGER.info('[RecordSlice] recv dlt_record = {:s}'.format(grpc_message_to_json_string(dlt_record))) str_dlt_record = grpc_message_to_json_string(dlt_record)
LOGGER.debug('[_record_entity] recv dlt_record = {:s}'.format(str_dlt_record))
exists = record_exists(dlt_record) exists = record_exists(dlt_record)
LOGGER.info('[RecordSlice] exists = {:s}'.format(str(exists))) LOGGER.debug('[_record_entity] exists = {:s}'.format(str(exists)))
dlt_record = DltRecord() dlt_record = DltRecord()
dlt_record.record_id.CopyFrom(dlt_record_id) dlt_record.record_id.CopyFrom(dlt_record_id) # pylint: disable=no-member
dlt_record.operation = \ if delete and exists:
DltRecordOperationEnum.DLTRECORDOPERATION_UPDATE \ dlt_record.operation = DltRecordOperationEnum.DLTRECORDOPERATION_DELETE
if exists else \ elif not delete and exists:
DltRecordOperationEnum.DLTRECORDOPERATION_ADD dlt_record.operation = DltRecordOperationEnum.DLTRECORDOPERATION_UPDATE
if data_json is None: raise Exception('data_json must be provided when updating')
dlt_record.data_json = grpc_message_to_json_string(slice_) dlt_record.data_json = data_json
LOGGER.info('[RecordSlice] sent dlt_record = {:s}'.format(grpc_message_to_json_string(dlt_record))) elif not delete and not exists:
dlt_record.operation = DltRecordOperationEnum.DLTRECORDOPERATION_ADD
if data_json is None: raise Exception('data_json must be provided when adding')
dlt_record.data_json = data_json
else:
return
str_dlt_record = grpc_message_to_json_string(dlt_record)
LOGGER.debug('[_record_entity] sent dlt_record = {:s}'.format(str_dlt_record))
dlt_record_status = dltgateway_client.RecordToDlt(dlt_record) dlt_record_status = dltgateway_client.RecordToDlt(dlt_record)
LOGGER.info('[RecordSlice] recv dlt_record_status = {:s}'.format(grpc_message_to_json_string(dlt_record_status))) str_dlt_record_status = grpc_message_to_json_string(dlt_record_status)
return Empty() LOGGER.debug('[_record_entity] recv dlt_record_status = {:s}'.format(str_dlt_record_status))
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment