Add support for CDE (#35)

curtishoward · acsjumpi · web-flow · commit 1c37151730e6 · 2021-10-25T11:58:55.000-04:00
* Initial CDE deployment support
* cdpy changes in de implementation
* add DE service failure states

Signed-off-by: Curtis Howard &lt;curtis@cloudera.com&gt;
Co-authored-by: Alan Silva &lt;alan.silva@cloudera.com&gt;
diff --git a/src/cdpy/cdpy.py b/src/cdpy/cdpy.py
@@ -10,6 +10,7 @@
 from cdpy.datahub import CdpyDatahub
 from cdpy.datalake import CdpyDatalake
 from cdpy.ml import CdpyMl
+from cdpy.de import CdpyDe
 from cdpy.opdb import CdpyOpdb
 from cdpy.dw import CdpyDw
 from cdpy.df import CdpyDf
@@ -24,6 +25,8 @@ def __init__(self, *args, **kwargs):
         self.datahub = CdpyDatahub(*args, **kwargs)
         self.datalake = CdpyDatalake(*args, **kwargs)
         self.ml = CdpyMl(*args, **kwargs)
+        self.de = CdpyDe(*args, **kwargs)
         self.opdb = CdpyOpdb(*args, **kwargs)
         self.dw = CdpyDw(*args, **kwargs)
         self.df = CdpyDf(*args, **kwargs)
+        self.de = CdpyDe(*args, **kwargs)
diff --git a/src/cdpy/common.py b/src/cdpy/common.py
@@ -216,16 +216,18 @@ def _warning_format(message, category, filename, lineno, line=None):
             'RUNNING',
             'installation:finished',  # ML
             'Running',  # DW
-            'GOOD_HEALTH'  # DF
+            'GOOD_HEALTH',  # DF
+            'ClusterCreationCompleted' #DE
         ]
 
         self.STOPPED_STATES = [
             'EXTERNAL_DATABASE_STOP_IN_PROGRESS',
             'STOP_IN_PROGRESS',
             'STOPPED',
             'ENV_STOPPED',
-            'Stopped',  # DW
-            'NOT_ENABLED'  # DF
+            'Stopped', # DW
+            'NOT_ENABLED',  # DF
+            'ClusterDeletionCompleted' # DE
         ]
 
         self.FAILED_STATES = [
@@ -239,7 +241,18 @@ def _warning_format(message, category, filename, lineno, line=None):
             'installation:failed',  # ML
             'provision:failed',  # ML
             'deprovision:failed',  # ML
-            'BAD_HEALTH'  # DF
+            'BAD_HEALTH',  # DF
+            # DE (all possibly intermediate failure states, until CDE exposes a higher-level summary state)
+            'ClusterChartInstallationFailed', 'ClusterDNSCreationFailed', 'ClusterDNSDeletionFailed',
+            'ClusterIngressCreationFailed', 'ClusterProvisioningFailed', 'DBProvisioningFailed',
+            'FSMountTargetsCreationFailed', 'FSProvisioningFailed', 'ClusterTLSCertCreationFailed',
+            'ClusterServiceMeshProvisioningFailed', 'ClusterMonitoringConfigurationFailed',
+            'ClusterChartDeletionFailed', 'ClusterDeletionFailed', 'ClusterNamespaceDeletionFailed',
+            'DBDeletionFailed', 'FSMountTargetsDeletionFailed', 'FSDeletionFailed',
+            'ClusterTLSCertDeletionFailed', 'ClusterServiceMeshDeletionFailed',
+            'ClusterAccessGroupCreationFailed', 'ClusterAccessGroupDeletionFailed',
+            'ClusterUserSyncCheckFailed', 'ClusterCreationFailed', 'ClusterDeleteFromDBFailed',
+            'ClusterMaintenanceFailed', 'ClusterTLSCertRenewalFailed'
         ]
 
         self.REMOVABLE_STATES = [
@@ -248,6 +261,7 @@ def _warning_format(message, category, filename, lineno, line=None):
             'installation:failed', 'deprovision:failed', 'installation:finished', 'modify:finished',  # ML
             'Error', 'Running', 'Stopped', 'Deleting',  # DW
             'GOOD_HEALTH', 'CONCERNING_HEALTH', 'BAD_HEALTH',  # DF
+            'ClusterCreationCompleted' #DE
         ]
 
         # common regex patterns
@@ -443,9 +457,9 @@ def wait_for_state(self, describe_func, params: dict, field: Union[str, None, li
                     current_status = self._get_path(current, field)
                 else:  # field not provided, therefore seek default status fields to check for failures
                     default_status_fields = [
-                        ['status'],  # Datalake, DW, OpDB, Datahub
+                        ['status'],  # Datalake, DW, OpDB, Datahub, DE
                         ['instanceStatus'],   # ML
-                        ['status', 'state'],  # DF
+                        ['status', 'state'],  # DF, DE
                     ]
                     possible_status = [
                         self._get_path(current, x) for x in default_status_fields
diff --git a/src/cdpy/de.py b/src/cdpy/de.py
@@ -0,0 +1,97 @@
+# -*- coding: utf-8 -*-
+
+from cdpy.common import CdpSdkBase, Squelch, CdpcliWrapper
+
+class CdpyDe(CdpSdkBase):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    def describe_vc(self, cluster_id, vc_id):
+        return self.sdk.call(
+            svc='de', func='describe_vc', ret_field='vc', squelch=[
+                Squelch('NOT_FOUND'), Squelch('INVALID_ARGUMENT')
+            ],
+            clusterId=cluster_id,
+            vcId=vc_id
+        )
+
+    def list_vcs(self, cluster_id):
+         return self.sdk.call(
+            svc='de', func='list_vcs', ret_field='vcs', squelch=[
+                Squelch(value='NOT_FOUND', default=list()),
+                Squelch(field='status_code', value='504', default=list(),
+                        warning="No VCS in this Cluster"),
+            ],
+            clusterId=cluster_id
+        )
+
+    def create_vc(self, name, cluster_id, cpu_requests, memory_requests,
+                  chart_value_overrides=None, runtime_spot_component=None):
+        return self.sdk.call(
+            svc='de', func='create_vc', ret_field='Vc',
+            name=name,
+            clusterId=cluster_id,
+            cpuRequests=cpu_requests,
+            memoryRequests=memory_requests,
+            chartValueOverrides=chart_value_overrides,
+            runtimeSpotComponent=runtime_spot_component
+        )
+
+    def delete_vc(self, cluster_id, vc_id):
+        return self.sdk.call(
+            svc='de', func='delete_vc', ret_field='status', squelch=[Squelch('NOT_FOUND')],
+            clusterId=cluster_id, vcId=vc_id
+        )
+
+    def describe_service(self, cluster_id):
+        return self.sdk.call(
+            svc='de', func='describe_service', ret_field='service', squelch=[
+                Squelch('NOT_FOUND'), Squelch('INVALID_ARGUMENT')
+            ],
+            clusterId=cluster_id,
+        )
+
+    def list_services(self, env=None, remove_deleted=False):
+        services = self.sdk.call(
+            svc='de', func='list_services', ret_field='services', squelch=[
+                Squelch(value='NOT_FOUND', default=list())], removeDeleted=remove_deleted
+        )
+        return [s for s in services if env is None or s['environmentName'] == env]
+
+    def enable_service(self, name, env, instance_type, minimum_instances, maximum_instances, 
+            initial_instances=None, minimum_spot_instances=None, maximum_spot_instances=None, 
+            initial_spot_instances=None, chart_value_overrides=None, enable_public_endpoint=False,
+            enable_workload_analytics=False, root_volume_size=None, skip_validation=False,
+            tags=None, use_ssd=False, whitelist_ips=None):
+        return self.sdk.call(
+            svc='de', func='enable_service', ret_field='service',
+            name=name,
+            env=env,
+            instanceType=instance_type,
+            minimumInstances=minimum_instances,
+            maximumInstances=maximum_instances,
+            initialInstances=initial_instances,
+            minimumSpotInstances=minimum_spot_instances,
+            maximumSpotInstances=maximum_spot_instances,
+            initialSpotInstances=initial_spot_instances,
+            chartValueOverrides=chart_value_overrides,
+            enablePublicEndpoint=enable_public_endpoint,
+            enableWorkloadAnalytics=enable_workload_analytics,
+            rootVolumeSize=root_volume_size,
+            skipValidation=skip_validation,
+            tags=tags,
+            useSsd=use_ssd,
+            whitelistIps=whitelist_ips
+        )
+
+    def disable_service(self, cluster_id, force=False):
+        return self.sdk.call(
+            svc='de', func='disable_service', ret_field='status', squelch=[Squelch('NOT_FOUND')], 
+            clusterId=cluster_id, force=force
+        )
+
+    def get_kubeconfig(self, cluster_id):
+        return self.sdk.call(
+            svc='de', func='get_kubeconfig', ret_field='kubeconfig', squelch=[Squelch('NOT_FOUND')],
+            clusterId=cluster_id
+        )
