Merge "Add Cetus Datasource"

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)