OPSAPS-32613 [DR] (PART VI) Add python APIs for hdfs s3 replication

1. added class ApiHdfsCloudReplicationArguments;
2. changed class ApiReplicationSchedule by adding field 'hdfsCloudArguments';
3. changed to allow cloud schedule creation;
4. added unit tests.

Author: Lei Shi

python/src/cm_api/endpoints/services.py

@@ -1473,9 +1473,12 @@ def create_replication_schedule(self,
       alertOnFail=alert_on_fail, alertOnAbort=alert_on_abort)
 
     if self.type == 'HDFS':
-      if not isinstance(arguments, ApiHdfsReplicationArguments):
+      if isinstance(arguments, ApiHdfsCloudReplicationArguments):
+        schedule.hdfsCloudArguments = arguments
+      elif isinstance(arguments, ApiHdfsReplicationArguments):
+        schedule.hdfsArguments = arguments
+      else:
         raise TypeError, 'Unexpected type for HDFS replication argument.'
-      schedule.hdfsArguments = arguments
     elif self.type == 'HIVE':
       if not isinstance(arguments, ApiHiveReplicationArguments):
         raise TypeError, 'Unexpected type for Hive replication argument.'

python/src/cm_api/endpoints/types.py

@@ -708,6 +708,18 @@ class ApiHdfsReplicationArguments(BaseApiObject):
     'exclusionFilters'          : None,
   }
 
+class ApiHdfsCloudReplicationArguments(ApiHdfsReplicationArguments):
+  @classmethod
+  def _get_attributes(cls):
+    if not cls.__dict__.has_key('_ATTRIBUTES'):
+      attrs = {
+        'cloudAccount'  : None,
+        'cloudService'    : None,
+      }
+      attrs.update(ApiHdfsReplicationArguments._get_attributes())
+      cls._ATTRIBUTES = attrs
+    return cls._ATTRIBUTES
+
 class ApiHdfsReplicationResult(BaseApiObject):
   _ATTRIBUTES = {
     'progress'            : ROAttr(),
@@ -797,6 +809,7 @@ class ApiReplicationSchedule(BaseApiObject):
     'paused'          : None,
     'hdfsArguments'   : Attr(ApiHdfsReplicationArguments),
     'hiveArguments'   : Attr(ApiHiveReplicationArguments),
+    'hdfsCloudArguments'   : Attr(ApiHdfsCloudReplicationArguments),
     'alertOnStart'    : None,
     'alertOnSuccess'  : None,
     'alertOnFail'     : None,

python/src/cm_api_tests/test_replication.py

@@ -72,6 +72,54 @@ def test_hdfs_arguments(self):
     self.assertEquals('DYNAMIC', args.replicationStrategy)
     self.assertFalse(args.preserveXAttrs)
 
+  def test_hdfs_cloud_arguments(self):
+    RAW = '''{
+      "sourceService" : {
+        "peerName" : "vst2",
+        "clusterName" : "Cluster 1 - CDH4",
+        "serviceName" : "HDFS-1"
+      },
+      "sourcePath" : "/data",
+      "destinationPath" : "/copy/data2",
+      "mapreduceServiceName" : "MAPREDUCE-1",
+      "schedulerPoolName" : "medium",
+      "userName" : "systest",
+      "dryRun" : false,
+      "abortOnError" : true,
+      "removeMissingFiles" : false,
+      "preserveReplicationCount" : true,
+      "preserveBlockSize" : true,
+      "preservePermissions" : false,
+      "skipTrash" : false,
+      "replicationStrategy" : "DYNAMIC",
+      "logPath" : "/tmp",
+      "bandwidthPerMap" : "20",
+      "preserveXAttrs" : false,
+      "exclusionFilters" : ["ac"],
+      "cloudAccount" : "someTestAccount",
+      "cloudService" : "TARGET"
+    }'''
+    args = utils.deserialize(RAW, ApiHdfsCloudReplicationArguments)
+    self.assertEquals('vst2', args.sourceService.peerName)
+    self.assertEquals('Cluster 1 - CDH4', args.sourceService.clusterName)
+    self.assertEquals('HDFS-1', args.sourceService.serviceName)
+    self.assertEquals('/data', args.sourcePath)
+    self.assertEquals('/copy/data2', args.destinationPath)
+    self.assertEquals('MAPREDUCE-1', args.mapreduceServiceName)
+    self.assertEquals('medium', args.schedulerPoolName)
+    self.assertEquals('systest', args.userName)
+    self.assertFalse(args.dryRun)
+    self.assertTrue(args.abortOnError)
+    self.assertFalse(args.removeMissingFiles)
+    self.assertTrue(args.preserveBlockSize)
+    self.assertFalse(args.preservePermissions)
+    self.assertTrue(args.preserveReplicationCount)
+    self.assertFalse(args.skipTrash)
+    self.assertEquals('DYNAMIC', args.replicationStrategy)
+    self.assertFalse(args.preserveXAttrs)
+    self.assertEquals('someTestAccount', args.cloudAccount)
+    self.assertEquals('TARGET', args.cloudService)
+
   def test_hive_arguments(self):
     RAW = '''{
       "sourceService" : {
@@ -321,5 +369,53 @@ def test_replication_crud(self):
         retdata=return_sched.to_json_dict())
     service.delete_replication_schedule(1)
 
+  def test_hdfs_cloud_replication_crud(self):
+    service = ApiService(self.resource, 'hdfs1', 'HDFS')
+    service.__dict__['clusterRef'] = ApiClusterRef(self.resource, clusterName='cluster1')
+
+    hdfs_args = ApiHdfsCloudReplicationArguments(self.resource)
+    hdfs_args.sourceService = ApiServiceRef('cluster2', 'hdfs2')
+    hdfs_args.sourcePath = '/src'
+    hdfs_args.destinationPath = 's3a://somebucket/dst'
+    hdfs_args.cloudAccount = 'someTestAccount'
+    hdfs_args.cloudService = 'TARGET'
+
+    return_sched = ApiReplicationSchedule(self.resource,
+        interval=2, intervalUnit='DAY')
+    return_sched.hdfsCloudArguments = hdfs_args
+    return_sched.__dict__['id'] = 1
+    return_list = ApiList([ return_sched ]).to_json_dict()
+
+    self.resource.expect("POST",
+        "/clusters/cluster1/services/hdfs1/replications",
+        retdata=return_list)
+
+    sched = service.create_replication_schedule(
+        None, None, 'DAY', 2, True, hdfs_args, alert_on_fail=True)
+    self.assertEqual(return_sched.intervalUnit, sched.intervalUnit)
+    self.assertEqual(return_sched.interval, sched.interval)
+    self.assertIsInstance(sched.hdfsCloudArguments, ApiHdfsCloudReplicationArguments)
+
+    self.resource.expect("GET",
+        "/clusters/cluster1/services/hdfs1/replications",
+        retdata=return_list)
+    service.get_replication_schedules()
+
+    self.resource.expect("GET",
+        "/clusters/cluster1/services/hdfs1/replications/1",
+        retdata=return_sched.to_json_dict())
+    service.get_replication_schedule(1)
+
+    self.resource.expect("PUT",
+        "/clusters/cluster1/services/hdfs1/replications/1",
+        data=return_sched,
+        retdata=return_sched.to_json_dict())
+    service.update_replication_schedule(1, return_sched)
+
+    self.resource.expect("DELETE",
+        "/clusters/cluster1/services/hdfs1/replications/1",
+        retdata=return_sched.to_json_dict())
+    service.delete_replication_schedule(1)
+
 if __name__ == '__main__':
     unittest.main()