Add Cetus Datasource

Cetus is a based on openstack solution of k8s on openstack, and it can be
operated through kesytone to automatically create multiple instances, and
automatically deploy multiple k8s clusters on instances. Cetus Datasource
include Cetus entities(cetus.cluster, cetus.pod).

Implements: blueprint add-cetus-datasource
Change-Id: I2057c901c8e28e1d61c506dcd8790d404817aaa8

releasenotes/notes/cetus-datasource-0c8297005ac6ca92.yaml

@@ -0,0 +1,14 @@
+---
+features:
+  - A new ``Cetus Datasource`` has been introduced to include Cetus entities
+    (cluster and pod) in Vitrage Entity Graph. Cetus is a self-developed
+    openstack solution of k8s on openstack. It can automatically create
+    multiple instances, and automatically deploy multiple k8s clusters on
+    instances. Cetus mainly represents the self-developed openstack project
+    and the multi-cluster k8s project, so it can be operated through openstack
+    authentication access. Cetus mainly includes cetus.cluster, cetus.pod,
+    cetus.node corresponding to k8s cluster, pod, node, among cetus.node is
+    vm instance or bm instance in openstack, so only includes cetus.cluster
+    and cetus.pod in the cetus datasource. At this point, Cetus entities are
+    extracted using PULL approach, based on periodical snapshot-query to Cetus
+    API for the current list of Cetus entities.

vitrage/datasources/cetus/cetus_driver_base.py

@@ -0,0 +1,98 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, 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.
+
+from oslo_config import cfg
+from oslo_log import log
+import requests
+
+from vitrage.keystone_client import get_auth_token
+from vitrage.keystone_client import get_client
+from vitrage.keystone_client import get_service_catalog
+
+from vitrage.datasources.driver_base import DriverBase
+
+CONF = cfg.CONF
+LOG = log.getLogger(__name__)
+
+
+class CetusDriverBase(DriverBase):
+
+    def __init__(self):
+        super(CetusDriverBase, self).__init__()
+        self.cetus_url = None
+        self._client = None
+
+    @property
+    def client(self):
+        if not self._client:
+            self._client = self._get_cetus_cli()
+        return self._client
+
+    @staticmethod
+    def _get_cetus_cli():
+        token = get_auth_token(get_client())
+        headers = dict()
+        headers['Content-Type'] = 'application/json'
+        headers["X-Auth-Token"] = token
+        session = requests.Session()
+        session.headers = headers
+        return session
+
+    @staticmethod
+    def _get_cetus_url(service_name='cetusv1'):
+        """modify service_name to get service endpoint url"""
+
+        services = get_service_catalog(get_client())
+        for service in services.catalog:
+            if service["name"] == service_name:
+                urls = [endpoint["url"] for endpoint in service["endpoints"]]
+                return urls[0] if urls else None
+        return None
+
+    def get_data(self, url):
+        try:
+            self.cetus_url = self._get_cetus_url()
+            url = '{}/{}'.format(self.cetus_url, url)
+            res = self.client.get(url)
+            return res.json()
+        except Exception as e:
+            LOG.error("Couldn't access cetus service:" + str(e))
+
+    def list_clusters(self):
+        url = "v1/clusters"
+        res = self.get_data(url)
+        return res
+
+    def list_cluster_pods(self, cid):
+        if not cid:
+            return dict()
+        url = "v1/clusters/{}/pods".format(cid)
+        res = self.get_data(url)
+        return res
+
+    def list_all(self):
+        pods = []
+        clusters = self.list_clusters()
+        for cluster in clusters.get('items', []):
+            cluster_id = cluster.get("cluster_id", "")
+            cluster_pods = self.list_cluster_pods(cluster_id)
+            cluster_pods = cluster_pods.get("items", [])
+            for pod in cluster_pods:
+                pod["cluster"] = cluster
+                pods.append(pod)
+        return pods
+
+    @staticmethod
+    def should_delete_outdated_entities():
+        return True

vitrage/datasources/cetus/cluster/__init__.py

@@ -0,0 +1,39 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, 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.
+
+from oslo_config import cfg
+from vitrage.common.constants import DatasourceOpts as DSOpts
+from vitrage.common.constants import UpdateMethod
+
+CETUS_CLUSTER_DATASOURCE = 'cetus.cluster'
+
+OPTS = [
+    cfg.StrOpt(DSOpts.TRANSFORMER,
+               default='vitrage.datasources.cetus.cluster.transformer.'
+                       'ClusterTransformer',
+               help='Cetus Cluster transformer class path',
+               required=True),
+    cfg.StrOpt(DSOpts.DRIVER,
+               default='vitrage.datasources.cetus.cluster.driver.'
+                       'ClusterDriver',
+               help='Cetus Cluster driver class path',
+               required=True),
+    cfg.StrOpt(DSOpts.UPDATE_METHOD,
+               default=UpdateMethod.NONE,
+               help='None: updates only via Vitrage periodic snapshots.'
+                    'Pull: updates every [changes_interval] seconds.'
+                    'Push: updates by getting notifications from the'
+                    ' datasource itself.',
+               required=True)
+]

vitrage/datasources/cetus/cluster/driver.py

@@ -0,0 +1,45 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,  software
+# distributed under the License is distributed on an "AS IS" BASIS, 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.
+
+from vitrage.datasources.cetus.cetus_driver_base import CetusDriverBase
+from vitrage.datasources.cetus.cluster import CETUS_CLUSTER_DATASOURCE
+
+
+class ClusterDriver(CetusDriverBase):
+
+    def get_all(self, datasource_action):
+        return self.make_pickleable(self._prepare_entities(
+            self.list_all()),
+            CETUS_CLUSTER_DATASOURCE,
+            datasource_action)
+
+    @staticmethod
+    def _prepare_entities(pods):
+        clusters_dict = {}
+        for pod in pods:
+
+            cluster = {
+                "name": pod['cluster']["name"],
+                "id": pod['cluster']["cluster_id"],
+                "status": pod['cluster']["status"],
+            }
+
+            node_id = pod["metadata"]["labels"]["instance_id"]
+            cluster_id = pod['cluster']["cluster_id"]
+            if cluster_id not in clusters_dict:
+                clusters_dict[cluster_id] = cluster
+                clusters_dict[cluster_id]['nodes'] = []
+            if node_id not in clusters_dict[cluster_id]['nodes']:
+                clusters_dict[cluster_id]['nodes'].append(node_id)
+        return [c for c in clusters_dict.values()]

vitrage/datasources/cetus/cluster/transformer.py

@@ -0,0 +1,102 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, 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.
+
+from oslo_config import cfg
+from oslo_log import log as logging
+
+from vitrage.common.constants import DatasourceProperties as DSProps
+from vitrage.common.constants import EdgeLabel
+from vitrage.common.constants import EntityCategory
+from vitrage.common.constants import VertexProperties as VProps
+
+from vitrage.datasources.resource_transformer_base import \
+    ResourceTransformerBase
+from vitrage.datasources.transformer_base import extract_field_value
+import vitrage.graph.utils as graph_utils
+
+from vitrage.datasources import NOVA_INSTANCE_DATASOURCE
+from vitrage.datasources import transformer_base as tbase
+
+from vitrage.datasources.cetus.cluster import CETUS_CLUSTER_DATASOURCE
+from vitrage.datasources.cetus.properties import CetusClusterProperties\
+    as CetusProp
+
+CONF = cfg.CONF
+LOG = logging.getLogger(__name__)
+
+
+class ClusterTransformer(ResourceTransformerBase):
+
+    def _create_vertex(self, entity_event, state, node_name):
+        metadata = {
+            VProps.NAME: node_name
+        }
+
+        entity_key = self._create_entity_key(entity_event)
+        vitrage_sample_timestamp = entity_event[DSProps.SAMPLE_DATE]
+        update_timestamp = self._format_update_timestamp(
+            extract_field_value(entity_event, DSProps.SAMPLE_DATE),
+            vitrage_sample_timestamp)
+
+        return graph_utils.create_vertex(
+            entity_key,
+            vitrage_category=EntityCategory.RESOURCE,
+            vitrage_type=CETUS_CLUSTER_DATASOURCE,
+            vitrage_sample_timestamp=vitrage_sample_timestamp,
+            update_timestamp=update_timestamp,
+            entity_id=extract_field_value(entity_event, CetusProp.ID),
+            entity_state=state,
+            metadata=metadata
+        )
+
+    def _create_snapshot_entity_vertex(self, entity_event):
+        node_name = extract_field_value(entity_event, CetusProp.NAME)
+        state = extract_field_value(entity_event, CetusProp.STATUS)
+        return self._create_vertex(entity_event, state, node_name)
+
+    def _create_update_entity_vertex(self, entity_event):
+        node_name = extract_field_value(entity_event, CetusProp.NAME)
+        state = extract_field_value(entity_event, CetusProp.STATUS)
+        return self._create_vertex(entity_event, state, node_name)
+
+    def _create_snapshot_neighbors(self, entity_event):
+        return self._create_cluster_neighbors(entity_event)
+
+    def _create_update_neighbors(self, entity_event):
+        return self._create_cluster_neighbors(entity_event)
+
+    def _create_entity_key(self, event):
+        instance_id = extract_field_value(event, CetusProp.ID)
+        key_fields = self._key_values(
+            CETUS_CLUSTER_DATASOURCE, instance_id)
+        key = tbase.build_key(key_fields)
+        return key
+
+    def get_vitrage_type(self):
+        return CETUS_CLUSTER_DATASOURCE
+
+    def _create_cluster_neighbors(self, entity_event):
+        neighbors = []
+        nodes = extract_field_value(entity_event, CetusProp.NODES)
+
+        for node in nodes:
+            node_neighbor = self._create_neighbor(
+                entity_event,
+                node,
+                NOVA_INSTANCE_DATASOURCE,
+                EdgeLabel.CONTAINS,
+                is_entity_source=True)
+            neighbors.append(node_neighbor)
+
+        return neighbors

vitrage/datasources/cetus/pod/__init__.py

@@ -0,0 +1,39 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, 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.
+
+from oslo_config import cfg
+from vitrage.common.constants import DatasourceOpts as DSOpts
+from vitrage.common.constants import UpdateMethod
+
+CETUS_POD_DATASOURCE = 'cetus.pod'
+
+OPTS = [
+    cfg.StrOpt(DSOpts.TRANSFORMER,
+               default='vitrage.datasources.cetus.pod.transformer.'
+                       'PodTransformer',
+               help='Cetus Pod transformer class path',
+               required=True),
+    cfg.StrOpt(DSOpts.DRIVER,
+               default='vitrage.datasources.cetus.pod.driver.'
+                       'PodDriver',
+               help='Cetus Pod driver class path',
+               required=True),
+    cfg.StrOpt(DSOpts.UPDATE_METHOD,
+               default=UpdateMethod.NONE,
+               help='None: updates only via Vitrage periodic snapshots.'
+                    'Pull: updates every [changes_interval] seconds.'
+                    'Push: updates by getting notifications from the'
+                    ' datasource itself.',
+               required=True)
+]

vitrage/datasources/cetus/pod/driver.py

@@ -0,0 +1,41 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,  software
+# distributed under the License is distributed on an "AS IS" BASIS, 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.
+
+from vitrage.datasources.cetus.cetus_driver_base import CetusDriverBase
+from vitrage.datasources.cetus.pod import CETUS_POD_DATASOURCE
+
+
+class PodDriver(CetusDriverBase):
+
+    def get_all(self, datasource_action):
+        return self.make_pickleable(self._prepare_entities(
+            self.list_all()),
+            CETUS_POD_DATASOURCE,
+            datasource_action)
+
+    @staticmethod
+    def _prepare_entities(pods):
+        pods_dict = {}
+        for pod in pods:
+
+            p = {
+                "name": pod["metadata"]["name"],
+                "id": pod["metadata"]["uid"],
+                "status": pod["status"]["phase"],
+                "node": pod["metadata"]["labels"]["instance_id"]
+            }
+
+            p_id = pod["metadata"]["uid"]
+            pods_dict[p_id] = p
+        return [p for p in pods_dict.values()]

vitrage/datasources/cetus/pod/transformer.py

@@ -0,0 +1,101 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, 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.
+
+from oslo_config import cfg
+from oslo_log import log as logging
+
+from vitrage.common.constants import DatasourceProperties as DSProps
+from vitrage.common.constants import EdgeLabel
+from vitrage.common.constants import EntityCategory
+from vitrage.common.constants import VertexProperties as VProps
+
+from vitrage.datasources.resource_transformer_base import \
+    ResourceTransformerBase
+from vitrage.datasources.transformer_base import extract_field_value
+import vitrage.graph.utils as graph_utils
+
+from vitrage.datasources import NOVA_INSTANCE_DATASOURCE
+from vitrage.datasources import transformer_base as tbase
+
+from vitrage.datasources.cetus.pod import CETUS_POD_DATASOURCE
+from vitrage.datasources.cetus.properties import CetusPodProperties \
+    as CetusProp
+
+CONF = cfg.CONF
+LOG = logging.getLogger(__name__)
+
+
+class PodTransformer(ResourceTransformerBase):
+
+    def _create_vertex(self, entity_event, state, node_name):
+        metadata = {
+            VProps.NAME: node_name
+        }
+
+        entity_key = self._create_entity_key(entity_event)
+        vitrage_sample_timestamp = entity_event[DSProps.SAMPLE_DATE]
+        update_timestamp = self._format_update_timestamp(
+            extract_field_value(entity_event, DSProps.SAMPLE_DATE),
+            vitrage_sample_timestamp)
+
+        return graph_utils.create_vertex(
+            entity_key,
+            vitrage_category=EntityCategory.RESOURCE,
+            vitrage_type=CETUS_POD_DATASOURCE,
+            vitrage_sample_timestamp=vitrage_sample_timestamp,
+            update_timestamp=update_timestamp,
+            entity_id=extract_field_value(entity_event, CetusProp.ID),
+            entity_state=state,
+            metadata=metadata
+        )
+
+    def _create_snapshot_entity_vertex(self, entity_event):
+        node_name = extract_field_value(entity_event, CetusProp.NAME)
+        state = extract_field_value(entity_event, CetusProp.STATUS)
+        return self._create_vertex(entity_event, state, node_name)
+
+    def _create_update_entity_vertex(self, entity_event):
+        node_name = extract_field_value(entity_event, CetusProp.NAME)
+        state = extract_field_value(entity_event, CetusProp.STATUS)
+        return self._create_vertex(entity_event, state, node_name)
+
+    def _create_snapshot_neighbors(self, entity_event):
+        return self._create_pod_neighbors(entity_event)
+
+    def _create_update_neighbors(self, entity_event):
+        return self._create_pod_neighbors(entity_event)
+
+    def _create_entity_key(self, event):
+        instance_id = extract_field_value(event, CetusProp.ID)
+        key_fields = self._key_values(
+            CETUS_POD_DATASOURCE, instance_id)
+        key = tbase.build_key(key_fields)
+        return key
+
+    def get_vitrage_type(self):
+        return CETUS_POD_DATASOURCE
+
+    def _create_pod_neighbors(self, entity_event):
+        neighbors = []
+        node = extract_field_value(entity_event, CetusProp.NODE)
+
+        node_neighbor = self._create_neighbor(
+            entity_event,
+            node,
+            NOVA_INSTANCE_DATASOURCE,
+            EdgeLabel.CONTAINS,
+            is_entity_source=False,
+        )
+        neighbors.append(node_neighbor)
+        return neighbors

vitrage/datasources/cetus/properties.py

@@ -0,0 +1,28 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, 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.
+
+
+class CetusBaseProperties(object):
+    ID = 'id'
+    NAME = 'name'
+    PROJECT_ID = 'project_id'
+    STATUS = 'status'
+
+
+class CetusPodProperties(CetusBaseProperties):
+    NODE = 'node'
+
+
+class CetusClusterProperties(CetusBaseProperties):
+    NODES = 'nodes'

vitrage/tests/unit/datasources/cetus/test_cetus_cluster_transformer.py

@@ -0,0 +1,196 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, 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 datetime
+
+from oslo_config import cfg
+from oslo_log import log as logging
+
+from testtools import matchers
+
+from vitrage.common.constants import DatasourceAction
+from vitrage.common.constants import DatasourceOpts as DSOpts
+from vitrage.common.constants import DatasourceProperties as DSProps
+from vitrage.common.constants import EdgeLabel
+from vitrage.common.constants import EntityCategory
+from vitrage.common.constants import UpdateMethod
+from vitrage.common.constants import VertexProperties as VProps
+from vitrage.datasources.cetus.cluster import CETUS_CLUSTER_DATASOURCE
+from vitrage.datasources.cetus.cluster.transformer import ClusterTransformer
+from vitrage.datasources.nova.instance import NOVA_INSTANCE_DATASOURCE
+from vitrage.datasources import transformer_base as tbase
+from vitrage.datasources.transformer_base import TransformerBase
+from vitrage.tests import base
+
+from vitrage.utils.datetime import format_utcnow
+
+LOG = logging.getLogger(__name__)
+
+events = [
+    {
+        'name': 'prom',
+        'id': 'c-6v5gr',
+        'status': 'active',
+        'nodes': [
+            '85ac015b-ec84-4d3c-a56e-d97fafff6a2a',
+            'c241c602-8c7b-44f8-8275-50b83a42787e'
+        ]
+    }
+]
+
+
+class TestCetusClusterTransformer(base.BaseTest):
+    OPTS = [
+        cfg.StrOpt(DSOpts.UPDATE_METHOD,
+                   default=UpdateMethod.PULL),
+    ]
+
+    # noinspection PyAttributeOutsideInit,PyPep8Naming
+    @classmethod
+    def setUpClass(cls):
+        super(TestCetusClusterTransformer, cls).setUpClass()
+        cls.transformers = {}
+        cls.conf = cfg.ConfigOpts()
+        cls.conf.register_opts(cls.OPTS, group=CETUS_CLUSTER_DATASOURCE)
+        cls.transformers[CETUS_CLUSTER_DATASOURCE] = ClusterTransformer(
+            cls.transformers)
+
+    def test_create_placeholder_vertex(self):
+        LOG.debug('Cetus cluster transformer test: Test create placeholder '
+                  'vertex')
+
+        # Test setup
+        cluster_id = "cluster123"
+        timestamp = datetime.datetime.utcnow()
+        cluster_transformer = self.transformers[CETUS_CLUSTER_DATASOURCE]
+
+        # Test action
+        properties = {
+            VProps.ID: cluster_id,
+            VProps.VITRAGE_TYPE: CETUS_CLUSTER_DATASOURCE,
+            VProps.VITRAGE_CATEGORY: EntityCategory.RESOURCE,
+            VProps.VITRAGE_SAMPLE_TIMESTAMP: timestamp
+        }
+        placeholder = \
+            cluster_transformer.create_neighbor_placeholder_vertex(
+                **properties)
+
+        # Test assertions
+        observed_uuid = placeholder.vertex_id
+        expected_key = tbase.build_key(cluster_transformer._key_values(
+            CETUS_CLUSTER_DATASOURCE,
+            cluster_id))
+        expected_uuid = \
+            TransformerBase.uuid_from_deprecated_vitrage_id(expected_key)
+        self.assertEqual(expected_uuid, observed_uuid)
+
+        observed_time = placeholder.get(VProps.VITRAGE_SAMPLE_TIMESTAMP)
+        self.assertEqual(timestamp, observed_time)
+
+        observed_subtype = placeholder.get(VProps.VITRAGE_TYPE)
+        self.assertEqual(CETUS_CLUSTER_DATASOURCE, observed_subtype)
+
+        observed_entity_id = placeholder.get(VProps.ID)
+        self.assertEqual(cluster_id, observed_entity_id)
+
+        observed_vitrage_category = placeholder.get(VProps.VITRAGE_CATEGORY)
+        self.assertEqual(EntityCategory.RESOURCE, observed_vitrage_category)
+
+        vitrage_is_placeholder = placeholder.get(
+            VProps.VITRAGE_IS_PLACEHOLDER)
+        self.assertTrue(vitrage_is_placeholder)
+
+    def test_key_values(self):
+
+        LOG.debug('Test key values')
+
+        # Test setup
+        cluster_id = "cluster123456"
+        cluster_transformer = self.transformers[CETUS_CLUSTER_DATASOURCE]
+        # Test action
+        observed_key_fields = cluster_transformer._key_values(
+            CETUS_CLUSTER_DATASOURCE,
+            cluster_id)
+
+        # Test assertions
+        self.assertEqual(EntityCategory.RESOURCE, observed_key_fields[0])
+        self.assertEqual(CETUS_CLUSTER_DATASOURCE, observed_key_fields[1])
+        self.assertEqual(cluster_id, observed_key_fields[2])
+
+    def test_snapshot_event_transform(self):
+        sample_timestamp = format_utcnow()
+        for event in events:
+            event[DSProps.DATASOURCE_ACTION] = DatasourceAction.SNAPSHOT
+            event[DSProps.SAMPLE_DATE] = sample_timestamp
+            wrapper = self.transformers[CETUS_CLUSTER_DATASOURCE].transform(
+                event)
+            vertex = wrapper.vertex
+
+            self._validate_vertex_props(vertex, event)
+            neighbors = wrapper.neighbors
+            self.assertThat(neighbors, matchers.HasLength(2))
+            self._validate_neighbors(neighbors, vertex.vertex_id)
+
+    def _validate_neighbors(self, neighbors, instance_vertex_id):
+        node_neighbors_counter = 0
+        for neighbor in neighbors:
+            self._validate_node_neighbor(neighbor,
+                                         instance_vertex_id)
+            node_neighbors_counter += 1
+
+        self.assertEqual(2, node_neighbors_counter)
+
+    def _validate_node_neighbor(self, node_neighbor, cluster_vertex_id):
+
+        node_vertex = node_neighbor.vertex
+        vitrage_type = node_vertex[VProps.VITRAGE_TYPE]
+        self.assertEqual(NOVA_INSTANCE_DATASOURCE, vitrage_type)
+        vitrage_is_deleted = node_vertex[VProps.VITRAGE_IS_DELETED]
+        self.assertFalse(vitrage_is_deleted)
+        vitrage_is_placeholder = node_vertex[VProps.VITRAGE_IS_PLACEHOLDER]
+        self.assertTrue(vitrage_is_placeholder)
+
+        # Validate neighbor edge
+        edge = node_neighbor.edge
+        self.assertEqual(edge.target_id, node_neighbor.vertex.vertex_id)
+        self.assertEqual(edge.source_id, cluster_vertex_id)
+        self.assertEqual(edge.label, EdgeLabel.CONTAINS)
+
+    def _validate_vertex_props(self, vertex, event):
+
+        extract_value = tbase.extract_field_value
+
+        expected_id = extract_value(event, 'id')
+        observed_id = vertex[VProps.ID]
+        self.assertEqual(expected_id, observed_id)
+
+        self.assertEqual(EntityCategory.RESOURCE,
+                         vertex[VProps.VITRAGE_CATEGORY])
+
+        self.assertEqual(CETUS_CLUSTER_DATASOURCE,
+                         vertex[VProps.VITRAGE_TYPE])
+
+        expected_timestamp = event[DSProps.SAMPLE_DATE]
+        observed_timestamp = vertex[VProps.VITRAGE_SAMPLE_TIMESTAMP]
+        self.assertEqual(expected_timestamp, observed_timestamp)
+
+        expected_name = extract_value(event, 'name')
+        observed_name = vertex[VProps.NAME]
+        self.assertEqual(expected_name, observed_name)
+
+        vitrage_is_placeholder = vertex[VProps.VITRAGE_IS_PLACEHOLDER]
+        self.assertFalse(vitrage_is_placeholder)
+
+        vitrage_is_deleted = vertex[VProps.VITRAGE_IS_DELETED]
+        self.assertFalse(vitrage_is_deleted)

vitrage/tests/unit/datasources/cetus/test_cetus_pod_transformer.py

@@ -0,0 +1,191 @@
+# Copyright 2020 - Inspur - Qitao
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, 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 datetime
+
+from oslo_config import cfg
+from oslo_log import log as logging
+
+from testtools import matchers
+
+from vitrage.common.constants import DatasourceAction
+from vitrage.common.constants import DatasourceOpts as DSOpts
+from vitrage.common.constants import DatasourceProperties as DSProps
+from vitrage.common.constants import EntityCategory
+from vitrage.common.constants import UpdateMethod
+from vitrage.common.constants import VertexProperties as VProps
+from vitrage.datasources.cetus.pod import CETUS_POD_DATASOURCE
+from vitrage.datasources.cetus.pod.transformer import PodTransformer
+from vitrage.datasources.nova.instance import NOVA_INSTANCE_DATASOURCE
+from vitrage.datasources import transformer_base as tbase
+from vitrage.datasources.transformer_base import TransformerBase
+from vitrage.tests import base
+
+from vitrage.utils.datetime import format_utcnow
+
+LOG = logging.getLogger(__name__)
+
+events = [
+    {
+        'name': 'app01-3232323232',
+        'id': '0903912039123',
+        'status': 'Pending',
+        'node': 'd7f8b66d-5221-4b4c-ade8-416a6a7bc661'
+    }, {
+        'name': 'app02-3232323232',
+        'id': '0903912039122',
+        'status': 'Pending',
+        'node': '85ac015b-ec84-4d3c-a56e-d97fafff6a2a'
+    }]
+
+
+class TestCetusClusterTransformer(base.BaseTest):
+    OPTS = [
+        cfg.StrOpt(DSOpts.UPDATE_METHOD,
+                   default=UpdateMethod.PULL),
+    ]
+
+    # noinspection PyAttributeOutsideInit,PyPep8Naming
+    @classmethod
+    def setUpClass(cls):
+        super(TestCetusClusterTransformer, cls).setUpClass()
+        cls.transformers = {}
+        cls.conf = cfg.ConfigOpts()
+        cls.conf.register_opts(cls.OPTS, group=CETUS_POD_DATASOURCE)
+        cls.transformers[CETUS_POD_DATASOURCE] = PodTransformer(
+            cls.transformers)
+
+    def test_create_placeholder_vertex(self):
+        LOG.debug('Cetus pod transformer test: Test create placeholder '
+                  'vertex')
+
+        # Test setup
+        pod_id = "pod123"
+        timestamp = datetime.datetime.utcnow()
+        pod_transformer = self.transformers[CETUS_POD_DATASOURCE]
+
+        # Test action
+        properties = {
+            VProps.ID: pod_id,
+            VProps.VITRAGE_TYPE: CETUS_POD_DATASOURCE,
+            VProps.VITRAGE_CATEGORY: EntityCategory.RESOURCE,
+            VProps.VITRAGE_SAMPLE_TIMESTAMP: timestamp
+        }
+        placeholder = \
+            pod_transformer.create_neighbor_placeholder_vertex(**properties)
+
+        # Test assertions
+        observed_uuid = placeholder.vertex_id
+        expected_key = tbase.build_key(pod_transformer._key_values(
+            CETUS_POD_DATASOURCE,
+            pod_id))
+        expected_uuid = \
+            TransformerBase.uuid_from_deprecated_vitrage_id(expected_key)
+        self.assertEqual(expected_uuid, observed_uuid)
+
+        observed_time = placeholder.get(VProps.VITRAGE_SAMPLE_TIMESTAMP)
+        self.assertEqual(timestamp, observed_time)
+
+        observed_subtype = placeholder.get(VProps.VITRAGE_TYPE)
+        self.assertEqual(CETUS_POD_DATASOURCE, observed_subtype)
+
+        observed_entity_id = placeholder.get(VProps.ID)
+        self.assertEqual(pod_id, observed_entity_id)
+
+        observed_vitrage_category = placeholder.get(VProps.VITRAGE_CATEGORY)
+        self.assertEqual(EntityCategory.RESOURCE, observed_vitrage_category)
+
+        vitrage_is_placeholder = placeholder.get(
+            VProps.VITRAGE_IS_PLACEHOLDER)
+        self.assertTrue(vitrage_is_placeholder)
+
+    def test_key_values(self):
+
+        LOG.debug('Test key values')
+
+        # Test setup
+        pod_id = "pod123456"
+        pod_transformer = self.transformers[CETUS_POD_DATASOURCE]
+        # Test action
+        observed_key_fields = pod_transformer._key_values(
+            CETUS_POD_DATASOURCE,
+            pod_id)
+
+        # Test assertions
+        self.assertEqual(EntityCategory.RESOURCE, observed_key_fields[0])
+        self.assertEqual(CETUS_POD_DATASOURCE, observed_key_fields[1])
+        self.assertEqual(pod_id, observed_key_fields[2])
+
+    def test_snapshot_event_transform(self):
+        sample_timestamp = format_utcnow()
+        for event in events:
+            event[DSProps.DATASOURCE_ACTION] = DatasourceAction.SNAPSHOT
+            event[DSProps.SAMPLE_DATE] = sample_timestamp
+            wrapper = self.transformers[CETUS_POD_DATASOURCE].transform(event)
+            vertex = wrapper.vertex
+            self._validate_vertex_props(vertex, event)
+            neighbors = wrapper.neighbors
+            self.assertThat(neighbors, matchers.HasLength(1))
+            self._validate_neighbors(neighbors, vertex.vertex_id)
+
+    def _validate_neighbors(self, neighbors, instance_vertex_id):
+        node_neighbors_counter = 0
+        for neighbor in neighbors:
+            node_neighbors_counter += 1
+            self._validate_node_neighbor(neighbor, instance_vertex_id)
+
+        self.assertEqual(1, node_neighbors_counter)
+
+    def _validate_node_neighbor(self, node_neighbor, instance_vertex_id):
+
+        self.assertEqual(node_neighbor.vertex[VProps.VITRAGE_ID],
+                         node_neighbor.vertex.vertex_id)
+        self.assertFalse(node_neighbor.vertex[VProps.VITRAGE_IS_DELETED])
+
+        self.assertTrue(node_neighbor.vertex[VProps.VITRAGE_IS_PLACEHOLDER])
+        self.assertEqual(NOVA_INSTANCE_DATASOURCE,
+                         node_neighbor.vertex[VProps.VITRAGE_TYPE])
+
+        # Validate neighbor edge
+        edge = node_neighbor.edge
+        self.assertEqual(edge.source_id, node_neighbor.vertex.vertex_id)
+        self.assertEqual(edge.target_id, instance_vertex_id)
+
+    def _validate_vertex_props(self, vertex, event):
+
+        extract_value = tbase.extract_field_value
+
+        expected_id = extract_value(event, 'id')
+        observed_id = vertex[VProps.ID]
+        self.assertEqual(expected_id, observed_id)
+
+        self.assertEqual(EntityCategory.RESOURCE,
+                         vertex[VProps.VITRAGE_CATEGORY])
+
+        self.assertEqual(CETUS_POD_DATASOURCE,
+                         vertex[VProps.VITRAGE_TYPE])
+
+        expected_timestamp = event[DSProps.SAMPLE_DATE]
+        observed_timestamp = vertex[VProps.VITRAGE_SAMPLE_TIMESTAMP]
+        self.assertEqual(expected_timestamp, observed_timestamp)
+
+        expected_name = extract_value(event, 'name')
+        observed_name = vertex[VProps.NAME]
+        self.assertEqual(expected_name, observed_name)
+
+        vitrage_is_placeholder = vertex[VProps.VITRAGE_IS_PLACEHOLDER]
+        self.assertFalse(vitrage_is_placeholder)
+
+        vitrage_is_deleted = vertex[VProps.VITRAGE_IS_DELETED]
+        self.assertFalse(vitrage_is_deleted)