DLT:

- Extended DltGatewayClient to support configuration through env vars
- Added DltEventsCollector to DltConnector
- Implemented RecordDevice in DltConnectorServicer

src/common/Constants.py

@@ -30,8 +30,10 @@ DEFAULT_HTTP_BIND_ADDRESS = '0.0.0.0'
 DEFAULT_METRICS_PORT = 9192
 
 # Default context and topology UUIDs
-DEFAULT_CONTEXT_UUID = 'admin'
-DEFAULT_TOPOLOGY_UUID = 'admin'
+DEFAULT_CONTEXT_UUID     = 'admin'
+DEFAULT_TOPOLOGY_UUID    = 'admin'      # contains the detailed local topology
+DOMAINS_TOPOLOGY_UUID    = 'domains'    # contains the abstracted domains (abstracted local + abstracted remotes)
+AGGREGATED_TOPOLOGY_UUID = 'aggregated' # contains the aggregated view (detailed local + abstracted remotes)
 
 # Default service names
 class ServiceNameEnum(Enum):
@@ -50,7 +52,7 @@ class ServiceNameEnum(Enum):
     WEBUI         = 'webui'
 
     # Used for test and debugging only
-    DLT_GATEWAY   = 'dlt-gateway'
+    DLT_GATEWAY   = 'dltgateway'
 
 # Default gRPC service ports
 DEFAULT_SERVICE_GRPC_PORTS = {

src/dlt/connector/Config.py

@@ -11,3 +11,14 @@
 # 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.
+
+import os
+
+DEFAULT_DLT_GATEWAY_HOST = '127.0.0.1'
+DEFAULT_DLT_GATEWAY_PORT = '50051'
+
+# Find IP:port of gateway container as follows:
+# - first check env vars DLT_GATEWAY_HOST & DLT_GATEWAY_PORT
+# - if not set, use DEFAULT_DLT_GATEWAY_HOST & DEFAULT_DLT_GATEWAY_PORT
+DLT_GATEWAY_HOST = str(os.environ.get('DLT_GATEWAY_HOST', DEFAULT_DLT_GATEWAY_HOST))
+DLT_GATEWAY_PORT = int(os.environ.get('DLT_GATEWAY_PORT', DEFAULT_DLT_GATEWAY_PORT))

src/dlt/connector/client/DltGatewayClient.py

@@ -14,14 +14,13 @@
 
 from typing import Iterator
 import grpc, logging
-from common.Constants import ServiceNameEnum
-from common.Settings import get_service_host, get_service_port_grpc
 from common.proto.context_pb2 import Empty, TeraFlowController
 from common.proto.dlt_gateway_pb2 import (
     DltPeerStatus, DltPeerStatusList, DltRecord, DltRecordEvent, DltRecordId, DltRecordStatus, DltRecordSubscription)
 from common.proto.dlt_gateway_pb2_grpc import DltGatewayServiceStub
 from common.tools.client.RetryDecorator import retry, delay_exponential
 from common.tools.grpc.Tools import grpc_message_to_json_string
+from dlt.connector.Config import DLT_GATEWAY_HOST, DLT_GATEWAY_PORT
 
 LOGGER = logging.getLogger(__name__)
 MAX_RETRIES = 15
@@ -30,8 +29,8 @@ RETRY_DECORATOR = retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION,
 
 class DltGatewayClient:
     def __init__(self, host=None, port=None):
-        if not host: host = get_service_host(ServiceNameEnum.DLT)
-        if not port: port = get_service_port_grpc(ServiceNameEnum.DLT)
+        if not host: host = DLT_GATEWAY_HOST
+        if not port: port = DLT_GATEWAY_PORT
         self.endpoint = '{:s}:{:s}'.format(str(host), str(port))
         LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))
         self.channel = None

src/dlt/connector/service/DltConnectorServiceServicerImpl.py

@@ -12,10 +12,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import grpc, logging
+import copy, grpc, logging
+from typing import Optional
+from common.Constants import DEFAULT_CONTEXT_UUID
+from common.proto.dlt_gateway_pb2 import DltRecord, DltRecordId, DltRecordOperationEnum, DltRecordTypeEnum
 from common.rpc_method_wrapper.Decorator import create_metrics, safe_and_metered_rpc_method
 from common.proto.context_pb2 import DeviceId, Empty, ServiceId, SliceId
 from common.proto.dlt_connector_pb2_grpc import DltConnectorServiceServicer
+from common.tools.grpc.Tools import grpc_message_to_json_string
+from context.client.ContextClient import ContextClient
+from dlt.connector.client.DltGatewayClient import DltGatewayClient
 
 LOGGER = logging.getLogger(__name__)
 
@@ -31,8 +37,26 @@ METRICS = create_metrics(SERVICE_NAME, METHOD_NAMES)
 class DltConnectorServiceServicerImpl(DltConnectorServiceServicer):
     def __init__(self):
         LOGGER.debug('Creating Servicer...')
+        self._own_domain_uuid : Optional[str] = None
         LOGGER.debug('Servicer Created')
 
+    @property
+    def own_domain_id(self) -> str:
+        if self._own_domain_uuid is not None: return self._own_domain_uuid
+
+        context_client = ContextClient()
+        existing_context_ids = context_client.ListContextIds(Empty())
+        existing_context_uuids = {context_id.context_uuid.uuid for context_id in existing_context_ids.context_ids}
+
+        # Detect local context name (will be used as abstracted device name); exclude DEFAULT_CONTEXT_UUID
+        existing_non_admin_context_uuids = copy.deepcopy(existing_context_uuids)
+        existing_non_admin_context_uuids.discard(DEFAULT_CONTEXT_UUID)
+        if len(existing_non_admin_context_uuids) != 1:
+            MSG = 'Unable to identify own domain name. Existing Contexts({:s})'
+            raise Exception(MSG.format(str(existing_context_uuids)))
+        self._own_domain_uuid = existing_non_admin_context_uuids.pop()
+        return self._own_domain_uuid
+
     @safe_and_metered_rpc_method(METRICS, LOGGER)
     def RecordAll(self, request : Empty, context : grpc.ServicerContext) -> Empty:
         return Empty()
@@ -43,6 +67,33 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer):
 
     @safe_and_metered_rpc_method(METRICS, LOGGER)
     def RecordDevice(self, request : DeviceId, context : grpc.ServicerContext) -> Empty:
+        context_client = ContextClient()
+        device = context_client.GetDevice(request)
+
+        dltgateway_client = DltGatewayClient()
+
+        dlt_record_id = DltRecordId()
+        dlt_record_id.domain_uuid.uuid = self.own_domain_id
+        dlt_record_id.type             = DltRecordTypeEnum.DLTRECORDTYPE_DEVICE
+        dlt_record_id.record_uuid.uuid = device.device_id.device_uuid.uuid
+
+        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 = False
+
+        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()
 
     @safe_and_metered_rpc_method(METRICS, LOGGER)

src/dlt/connector/service/__main__.py

@@ -18,6 +18,8 @@ from common.Constants import ServiceNameEnum
 from common.Settings import (
     ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, get_env_var_name, get_log_level, get_metrics_port,
     wait_for_environment_variables)
+from dlt.connector.client.DltEventsCollector import DltEventsCollector
+from dlt.connector.client.DltGatewayClient import DltGatewayClient
 from .DltConnectorService import DltConnectorService
 
 terminate = threading.Event()
@@ -48,6 +50,10 @@ def main():
     metrics_port = get_metrics_port()
     start_http_server(metrics_port)
 
+    dlt_gateway_client = DltGatewayClient()
+    dlt_events_collector = DltEventsCollector(dlt_gateway_client, log_events_received=True)
+    dlt_events_collector.start()
+
     # Starting DLT connector service
     grpc_service = DltConnectorService()
     grpc_service.start()
@@ -56,6 +62,7 @@ def main():
     while not terminate.wait(timeout=0.1): pass
 
     LOGGER.info('Terminating...')
+    dlt_events_collector.stop()
     grpc_service.stop()
 
     LOGGER.info('Bye')